feat(ticket): generalize Worker launch and durable resource claims

crates/workspace-server/src/authority.rs

@@ -750,9 +750,9 @@ impl SqliteWorkspaceAuthority {
                 assigned_at: assignment.assigned_at,
             })
             .collect::<Vec<_>>();
-        let current_coder = role_assignments
+        let current_worker = role_assignments
             .iter()
-            .find(|assignment| assignment.role == TicketAssignmentRole::Coder)
+            .find(|assignment| assignment.role == TicketAssignmentRole::Worker)
             .and_then(|assignment| {
                 assignment
                     .principal
@@ -779,9 +779,9 @@ impl SqliteWorkspaceAuthority {
         let has_orchestrator = active_role_assignments
             .iter()
             .any(|assignment| assignment.role == TicketAssignmentRole::Orchestrator);
-        let has_coder = active_role_assignments
+        let has_worker = active_role_assignments
             .iter()
-            .any(|assignment| assignment.role == TicketAssignmentRole::Coder);
+            .any(|assignment| assignment.role == TicketAssignmentRole::Worker);
         let targets = ticket
             .meta
             .targets
@@ -799,7 +799,6 @@ impl SqliteWorkspaceAuthority {
             })
             .collect::<Result<Vec<_>>>()?;
         let write_repository_keys = ticket_write_target_repository_keys(&targets);
-        let has_target = !write_repository_keys.is_empty();
         let has_blockers = !ticket.relations.blockers.is_empty();
         let mut assignment_diagnostics = Vec::new();
         if let Some(legacy_assignee) = ticket
@@ -818,17 +817,20 @@ impl SqliteWorkspaceAuthority {
                 ticket.meta.workflow_state,
                 TicketWorkflowState::Planning | TicketWorkflowState::Ready
             ) && !has_orchestrator
-                && !has_coder,
+                && !has_worker,
             can_unassign_orchestrator: has_orchestrator
                 && matches!(
                     ticket.meta.workflow_state,
                     TicketWorkflowState::Planning | TicketWorkflowState::Ready
                 ),
             can_queue: true,
-            can_start_manual_coder: ticket.meta.workflow_state == TicketWorkflowState::Ready
-                && !has_orchestrator
-                && !has_coder
-                && has_target
+            can_start_manual_worker: matches!(
+                ticket.meta.workflow_state,
+                TicketWorkflowState::Planning
+                    | TicketWorkflowState::Ready
+                    | TicketWorkflowState::Queued
+                    | TicketWorkflowState::InProgress
+            ) && !has_worker
                 && !has_blockers,
             queue_tickets: dependency_check.queue_tickets.clone(),
             blockers: action_blockers,
@@ -960,7 +962,7 @@ impl SqliteWorkspaceAuthority {
             linked_objectives,
             implementation_reports,
             assignments,
-            current_coder,
+            current_worker,
             assignment_diagnostics,
             action_eligibility,
             merge_requests,
@@ -2353,7 +2355,7 @@ fn ticket_query_item(
         evidence: detail.evidence.clone(),
         merge_requests: detail.merge_requests.clone(),
         merge_request: detail.merge_request.clone(),
-        current_coder: detail.current_coder.clone(),
+        current_worker: detail.current_worker.clone(),
     }
 }
 
@@ -3193,6 +3195,143 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn manual_worker_eligibility_accepts_explicit_states_without_reviving_released_work() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("workspace.db");
+        let store = SqliteWorkspaceStore::open(&path).unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: "w".into(),
+                owner_account_id: "owner-account".into(),
+                display_name: "W".into(),
+                state: "active".into(),
+                created_at: "1".into(),
+                updated_at: "1".into(),
+            })
+            .await
+            .unwrap();
+        let conn = rusqlite::Connection::open(&path).unwrap();
+        conn.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
+        store
+            .upsert_worker_registry(&crate::store::WorkerRegistryRecord {
+                workspace_id: "w".into(),
+                worker: worker_runtime::identity::RuntimeWorkerRef::new("runtime", "worker"),
+                display_name: "Worker".into(),
+                profile: Some("builtin:default".into()),
+                retention_state: "normal".into(),
+                transcript_ref: None,
+                session_ref: None,
+                summary_ref: None,
+                diagnostics_ref: None,
+                created_at: "1".into(),
+                updated_at: "1".into(),
+            })
+            .unwrap();
+        let backend = SqliteTicketBackend::open_verified(&path, "w").unwrap();
+        let ticket = backend
+            .create(ticket::NewTicket::new("Explicit work"))
+            .unwrap();
+        let authority = SqliteWorkspaceAuthority::new(&path, "w").unwrap();
+        for (state, allowed) in [
+            ("planning", true),
+            ("ready", true),
+            ("queued", true),
+            ("inprogress", true),
+            ("done", false),
+            ("closed", false),
+        ] {
+            conn.execute(
+                "UPDATE typed_tickets SET workflow_state=?1 WHERE ticket_id=?2",
+                rusqlite::params![state, ticket.id],
+            )
+            .unwrap();
+            assert_eq!(
+                authority
+                    .ticket(&ticket.id)
+                    .unwrap()
+                    .action_eligibility
+                    .can_start_manual_worker,
+                allowed,
+                "{state}"
+            );
+        }
+        conn.execute(
+            "UPDATE typed_tickets SET workflow_state='planning' WHERE ticket_id=?1",
+            [&ticket.id],
+        )
+        .unwrap();
+        let record = crate::store::TicketRoleAssignmentRecord {
+            workspace_id: "w".into(),
+            ticket_id: ticket.id.clone(),
+            assignment_id: "assignment".into(),
+            role: TicketAssignmentRole::Worker,
+            principal: TicketAssignmentPrincipal::Worker {
+                runtime_id: "runtime".into(),
+                worker_id: "worker".into(),
+            },
+            assigned_by: "owner-account".into(),
+            assigned_at: "2".into(),
+        };
+        let orchestrator = crate::store::TicketRoleAssignmentRecord {
+            role: TicketAssignmentRole::Orchestrator,
+            assignment_id: "orchestrator".into(),
+            principal: TicketAssignmentPrincipal::WorkspaceAgent {
+                agent_key: "workspace-orchestrator".into(),
+            },
+            ..record.clone()
+        };
+        store
+            .set_current_ticket_role_assignment(
+                &orchestrator,
+                None,
+                "orchestrator-event",
+                "orchestrator-op",
+                false,
+            )
+            .unwrap();
+        assert!(
+            authority
+                .ticket(&ticket.id)
+                .unwrap()
+                .action_eligibility
+                .can_start_manual_worker
+        );
+        store
+            .start_ticket_with_worker_assignment(&record, "event", "operation", None)
+            .unwrap();
+        assert!(
+            !authority
+                .ticket(&ticket.id)
+                .unwrap()
+                .action_eligibility
+                .can_start_manual_worker
+        );
+        conn.execute(
+            "UPDATE typed_tickets SET workflow_state='closed' WHERE ticket_id=?1",
+            [&ticket.id],
+        )
+        .unwrap();
+        conn.execute(
+            "UPDATE typed_tickets SET workflow_state='planning' WHERE ticket_id=?1",
+            [&ticket.id],
+        )
+        .unwrap();
+        let detail = authority.ticket(&ticket.id).unwrap();
+        assert!(detail.action_eligibility.can_start_manual_worker);
+        assert_eq!(detail.assignments.len(), 2);
+        let worker = detail
+            .assignments
+            .iter()
+            .find(|a| a.role == "worker")
+            .unwrap();
+        assert_eq!(worker.assignment_id, record.assignment_id);
+        assert_eq!(
+            detail.current_worker.unwrap().assignment_id,
+            record.assignment_id
+        );
+    }
+
     #[tokio::test]
     async fn sqlite_workspace_authority_reads_sqlite_records_without_filesystem_authority() {
         let dir = tempfile::tempdir().unwrap();

crates/workspace-server/src/hosts.rs

@@ -484,7 +484,7 @@ pub(crate) fn worker_spawn_idempotency(
     let Some(assignment) = request.ticket_assignment.as_ref() else {
         return Ok(None);
     };
-    let encoded = serde_json::to_vec(request)
+    let encoded = serde_json::to_vec(&(request, request.resolved_workdir_attachments.as_slice()))
         .map_err(|error| format!("serialize Worker spawn idempotency input: {error}"))?;
     Ok(Some((
         assignment.operation_id.clone(),
@@ -578,7 +578,7 @@ pub enum WorkerSpawnIntent {
 pub enum TicketWorkerRole {
     Intake,
     Orchestrator,
-    Coder,
+    Worker,
     Reviewer,
 }
 
@@ -6456,7 +6456,7 @@ fn worker_spawn_intent_label(intent: &WorkerSpawnIntent) -> &'static str {
         WorkerSpawnIntent::TicketRole { role, .. } => match role {
             TicketWorkerRole::Intake => "ticket_intake",
             TicketWorkerRole::Orchestrator => "ticket_orchestrator",
-            TicketWorkerRole::Coder => "ticket_coder",
+            TicketWorkerRole::Worker => "ticket_worker",
             TicketWorkerRole::Reviewer => "ticket_reviewer",
         },
     }
@@ -6521,6 +6521,27 @@ mod tests {
     use std::sync::{Arc, Mutex};
     use std::thread;
 
+    #[test]
+    fn ticket_worker_spawn_intent_uses_worker_wire_role() {
+        let intent = WorkerSpawnIntent::TicketRole {
+            ticket_id: "ticket".into(),
+            role: TicketWorkerRole::Worker,
+        };
+        let encoded = json!({"kind": "ticket_role", "ticket_id": "ticket", "role": "worker"});
+        assert_eq!(serde_json::to_value(&intent).unwrap(), encoded);
+        assert_eq!(
+            serde_json::from_value::<WorkerSpawnIntent>(encoded).unwrap(),
+            intent
+        );
+        assert!(
+            serde_json::from_value::<WorkerSpawnIntent>(
+                json!({"kind": "ticket_role", "ticket_id": "ticket", "role": "coder"})
+            )
+            .is_err()
+        );
+        assert_eq!(worker_spawn_intent_label(&intent), "ticket_worker");
+    }
+
     #[test]
     fn remote_restore_transport_failure_requires_reconciliation() {
         let provider = RemoteWorkerRuntime::new(
@@ -7538,7 +7559,7 @@ mod tests {
         WorkerSpawnRequest {
             intent: WorkerSpawnIntent::TicketRole {
                 ticket_id: "00001KVZSGT0Q".to_string(),
-                role: TicketWorkerRole::Coder,
+                role: TicketWorkerRole::Worker,
             },
             requested_worker_name: None,
             singleton_key: None,
@@ -7827,7 +7848,7 @@ mod tests {
                 WorkerSpawnRequest {
                     intent: WorkerSpawnIntent::TicketRole {
                         ticket_id: "00001KVZSGT0Q".to_string(),
-                        role: TicketWorkerRole::Coder,
+                        role: TicketWorkerRole::Worker,
                     },
                     requested_worker_name: Some("friendly-name-is-not-authority".to_string()),
                     singleton_key: None,
@@ -7939,7 +7960,7 @@ mod tests {
                 WorkerSpawnRequest {
                     intent: WorkerSpawnIntent::TicketRole {
                         ticket_id: "00001KVZSGT0Q".to_string(),
-                        role: TicketWorkerRole::Coder,
+                        role: TicketWorkerRole::Worker,
                     },
                     requested_worker_name: None,
                     singleton_key: None,

crates/workspace-server/src/latest_schema.sql

@@ -335,8 +335,8 @@ CREATE TABLE ticket_assignment_operations (
             operation_id TEXT NOT NULL,
             action TEXT NOT NULL CHECK (action IN ('assign', 'reassign', 'unassign')),
             ticket_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             principal_kind TEXT NOT NULL DEFAULT 'worker'
                 CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
             principal_id TEXT,
@@ -346,8 +346,14 @@ CREATE TABLE ticket_assignment_operations (
             expected_assignment_id TEXT,
             created_at TEXT NOT NULL,
             request_fingerprint TEXT,
+            claim_state TEXT NOT NULL DEFAULT 'committed'
+                CHECK(claim_state IN ('pending', 'failed', 'committed')),
+            failure_reason TEXT,
+            binding_recovery_json TEXT,
             PRIMARY KEY (workspace_id, operation_id),
-            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
+            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+            CHECK(claim_state = 'committed' OR assignment_id IS NULL),
+            CHECK((claim_state = 'failed') = (failure_reason IS NOT NULL))
         );
 CREATE TABLE ticket_assignment_ticket_tombstones (
     workspace_id TEXT NOT NULL,
@@ -367,8 +373,8 @@ CREATE TABLE ticket_assignment_worker_tombstones (
 CREATE TABLE ticket_current_worker_assignments (
             workspace_id TEXT NOT NULL,
             ticket_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             assignment_id TEXT NOT NULL,
             principal_kind TEXT NOT NULL DEFAULT 'worker'
                 CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
@@ -393,8 +399,8 @@ CREATE TABLE ticket_current_worker_assignments (
 CREATE TABLE ticket_worker_assignment_events (
             workspace_id TEXT NOT NULL,
             ticket_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             event_id TEXT NOT NULL,
             action TEXT NOT NULL CHECK (action IN ('assigned', 'reassigned', 'unassigned')),
             assignment_id TEXT,
@@ -410,8 +416,8 @@ CREATE TABLE ticket_worker_assignments (
             workspace_id TEXT NOT NULL,
             ticket_id TEXT NOT NULL,
             assignment_id TEXT NOT NULL,
-            role TEXT NOT NULL DEFAULT 'coder'
-                CHECK(role IN ('orchestrator', 'coder', 'owner', 'contributor')),
+            role TEXT NOT NULL DEFAULT 'worker'
+                CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
             principal_kind TEXT NOT NULL DEFAULT 'worker'
                 CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
             principal_id TEXT,
@@ -1215,7 +1221,7 @@ CREATE INDEX ticket_current_principal_idx
             ON ticket_current_worker_assignments(workspace_id, principal_kind, principal_id, runtime_id, worker_id);
 CREATE UNIQUE INDEX ticket_current_singleton_role_idx
             ON ticket_current_worker_assignments(workspace_id, ticket_id, role)
-            WHERE role IN ('orchestrator', 'coder');
+            WHERE role IN ('orchestrator', 'worker');
 CREATE INDEX typed_ticket_events_workspace_kind_ticket
             ON typed_ticket_events(workspace_id, kind, ticket_id, event_index);
 CREATE UNIQUE INDEX ux_worker_workdir_attachment_reservation_id

crates/workspace-server/src/retention.rs

@@ -69,6 +69,10 @@ pub struct WorkerRemovalPlanRequest {
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum WorkerRemovalBlocker {
     Hold,
+    PendingTicketClaim {
+        operation_id: String,
+        ticket_id: String,
+    },
     UnfinishedWork {
         assignment_id: String,
         ticket_id: String,
@@ -232,6 +236,9 @@ impl SqliteWorkspaceStore {
             if let Some((assignment_id,ticket_id))=tx.query_row("SELECT a.assignment_id,a.ticket_id FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3",params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()? {
                 blockers.push(WorkerRemovalBlocker::UnfinishedWork{assignment_id,ticket_id});
             }
+            if let Some((operation_id,ticket_id))=tx.query_row("SELECT operation_id,ticket_id FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign')",params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()? {
+                blockers.push(WorkerRemovalBlocker::PendingTicketClaim{operation_id,ticket_id});
+            }
             let fp=fingerprint(req,&worker.updated_at,inv,&policy,&blockers)?;
             let plan_id=stable("wrp",&fp); let operation_id=stable("wro",&fp);
             let archive_id=(policy.session_disposition==SessionDisposition::Archive).then(||stable("wra",&fp));
@@ -276,7 +283,7 @@ impl SqliteWorkspaceStore {
                 mark_stale(&tx,&plan,"hold added")?;tx.commit()?;
                 return Err(stale_error(&plan,"hold added"));
             }
-            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?;
+            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?;
             if assigned{
                 mark_stale(&tx,&plan,"unfinished work added")?;tx.commit()?;
                 return Err(stale_error(&plan,"unfinished work added"));
@@ -462,7 +469,7 @@ impl SqliteWorkspaceStore {
             let worker=load_worker(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput("Worker missing before commit".into()))?;
             if worker.updated_at!=plan.worker_revision{return Err(StoreError::InvalidInput(format!("stale:{}:Worker revision changed",plan.plan_id)));}
             if worker.retention_state=="pinned" { return Err(StoreError::InvalidInput(format!("stale:{}:hold added",plan.plan_id))); }
-            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|row|row.get(0))?;
+            let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|row|row.get(0))?;
             if assigned { return Err(StoreError::InvalidInput(format!("stale:{}:unfinished work added",plan.plan_id))); }
             let now=Utc::now().to_rfc3339();
             if let Some(a)=&result.archive{
@@ -1092,7 +1099,7 @@ fn parse_state(v: &str) -> rusqlite::Result<WorkerRemovalPlanState> {
 mod tests {
     use super::*;
     use crate::store::{
-        ControlPlaneStore, TicketCoderAssignmentRecord, WorkerCatalogChange, WorkerRegistryRecord,
+        ControlPlaneStore, TicketWorkerAssignmentRecord, WorkerCatalogChange, WorkerRegistryRecord,
     };
     use worker_runtime::identity::WorkerId;
     fn worker_id() -> WorkerId {
@@ -1431,7 +1438,7 @@ mod tests {
         s: &SqliteWorkspaceStore,
         assignment_id: &str,
     ) -> crate::Result<()> {
-        let assignment = TicketCoderAssignmentRecord {
+        let assignment = TicketWorkerAssignmentRecord {
             workspace_id: "w".into(),
             ticket_id: "ticket".into(),
             assignment_id: assignment_id.into(),
@@ -1439,7 +1446,7 @@ mod tests {
             assigned_by: "test".into(),
             assigned_at: "t".into(),
         };
-        s.set_current_ticket_coder_assignment(
+        s.set_current_ticket_worker_assignment(
             &assignment,
             None,
             &format!("event-{assignment_id}"),
@@ -1474,6 +1481,38 @@ mod tests {
         store
     }
 
+    #[test]
+    fn pending_ticket_claim_blocks_removal_and_is_rechecked_after_planning() {
+        let s = setup();
+        let old = s.plan_worker_removal(&req(), &inv()).unwrap();
+        s.reserve_ticket_assignment_operation(
+            "w",
+            "claim",
+            "ticket",
+            "r",
+            Some(&worker_id().to_string()),
+            "intent",
+            "t",
+        )
+        .unwrap();
+        let blocked = s.plan_worker_removal(&req(), &inv()).unwrap();
+        assert!(
+            matches!(&blocked.blockers[..], [WorkerRemovalBlocker::PendingTicketClaim{operation_id,ticket_id}] if operation_id=="claim" && ticket_id=="ticket")
+        );
+        assert!(
+            s.begin_worker_removal("w", &old.plan_id, &old.input_fingerprint)
+                .is_err()
+        );
+        s.fail_ticket_assignment_operation("w", "claim", "No Runtime effects dispatched")
+            .unwrap();
+        assert!(
+            s.plan_worker_removal(&req(), &inv())
+                .unwrap()
+                .blockers
+                .is_empty()
+        );
+    }
+
     #[test]
     fn assignment_and_orphan_are_authoritative() {
         let s = setup();
@@ -1758,7 +1797,7 @@ mod tests {
                 let replaced = tx.execute(
                     "UPDATE ticket_current_worker_assignments
                      SET assignment_id='new-assignment', updated_at='new'
-                     WHERE workspace_id='w' AND ticket_id='ticket' AND role='coder'
+                     WHERE workspace_id='w' AND ticket_id='ticket' AND role='worker'
                        AND assignment_id='assignment'
                        AND EXISTS (
                            SELECT 1 FROM ticket_assignment_work_releases
@@ -2088,7 +2127,7 @@ mod tests {
         ).map_err(StoreError::from)).unwrap();
         assert_eq!(revision, "rev1");
 
-        let assignment = TicketCoderAssignmentRecord {
+        let assignment = TicketWorkerAssignmentRecord {
             workspace_id: "w".into(),
             ticket_id: "new-ticket".into(),
             assignment_id: "new-assignment".into(),
@@ -2101,7 +2140,7 @@ mod tests {
         };
         assert!(
             store
-                .set_current_ticket_coder_assignment(
+                .set_current_ticket_worker_assignment(
                     &assignment,
                     None,
                     "event",

crates/workspace-server/src/server/tests/ticket_evidence_tests.rs

@@ -74,8 +74,8 @@ impl EvidenceApiFixture {
             .unwrap();
         let assignment_id = "evidence-assignment".to_string();
         api.store
-            .set_current_ticket_coder_assignment(
-                &TicketCoderAssignmentRecord {
+            .set_current_ticket_worker_assignment(
+                &TicketWorkerAssignmentRecord {
                     workspace_id: TEST_WORKSPACE_ID.into(),
                     ticket_id: ticket_id.clone(),
                     assignment_id: assignment_id.clone(),
@@ -319,7 +319,7 @@ async fn done_merged_remote_ticket_uses_immutable_result_in_show_query_list_and_
         fixture
             .api
             .store
-            .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
+            .get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
             .unwrap()
             .is_some()
     );
@@ -512,7 +512,7 @@ async fn open_source_without_runtime_is_typed_unavailable_not_stale_after_rescop
         fixture
             .api
             .store
-            .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
+            .get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
             .unwrap()
             .is_some()
     );

crates/workspace-server/src/server/tests/ticket_worker_claim_tests.rs

@@ -0,0 +1,657 @@
+use super::*;
+
+async fn claim(
+    fixture: &ManualCoderAssignmentFixture,
+    request: server_api::SetTicketRoleAssignmentRequest,
+) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
+    scoped_set_ticket_assignment(
+        State(fixture.api.clone()),
+        AxumPath((
+            TEST_WORKSPACE_ID.into(),
+            fixture.ticket_id.clone(),
+            "worker".into(),
+        )),
+        Json(request),
+    )
+    .await
+}
+
+#[tokio::test]
+async fn explicit_existing_worker_claim_accepts_no_resources_and_preserves_assignment_on_retry() {
+    let fixture = manual_worker_assignment_fixture().await;
+    for id in [&fixture.main_workdir_id, &fixture.docs_workdir_id] {
+        fixture
+            .api
+            .store
+            .detach_worker_workdir(
+                TEST_WORKSPACE_ID,
+                &fixture.worker,
+                Some(id),
+                TEST_CREATED_AT,
+            )
+            .unwrap();
+    }
+    sync_runtime_worker_workdir_attachments(&fixture.api, &fixture.worker).unwrap();
+    let backend = browser_ticket_backend(&fixture.api).unwrap();
+    backend
+        .edit_item(
+            fixture.ticket_id.clone().into(),
+            TicketItemEdit {
+                targets: Some(TicketTargetsEdit::Clear),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    backend
+        .set_workflow_state(
+            fixture.ticket_id.clone().into(),
+            TicketStateChange::new(
+                "ready",
+                "planning",
+                "Explicit investigation",
+                "No repository resources required",
+            ),
+        )
+        .unwrap();
+    let request = manual_worker_assignment_request(&fixture, "claim-without-resources");
+    assert!(request.workdir_bindings.is_empty());
+    let Json(first) = claim(&fixture, request.clone()).await.unwrap();
+    let Json(replay) = claim(&fixture, request).await.unwrap();
+    assert_eq!(
+        first.assignment.unwrap().assignment_id,
+        replay.assignment.unwrap().assignment_id
+    );
+    assert_eq!(
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .state,
+        "inprogress"
+    );
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 1);
+    assert!(
+        fixture
+            .runtime
+            .logical_attachments(&fixture.worker.worker_id)
+            .is_empty()
+    );
+}
+
+#[tokio::test]
+async fn explicit_existing_worker_claim_narrows_read_only_targets_without_orchestrator_conflict() {
+    let fixture = manual_worker_assignment_fixture().await;
+    let backend = browser_ticket_backend(&fixture.api).unwrap();
+    backend
+        .edit_item(
+            fixture.ticket_id.clone().into(),
+            TicketItemEdit {
+                targets: Some(TicketTargetsEdit::Set {
+                    targets: vec![
+                        test_ticket_target(
+                            "test-repository",
+                            "develop",
+                            TicketTargetAccess::ReadOnly,
+                        ),
+                        test_ticket_target("docs", "develop", TicketTargetAccess::ReadOnly),
+                    ],
+                }),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    assign_test_orchestrator(&fixture.api, &fixture.ticket_id);
+    let request = manual_worker_assignment_request(&fixture, "claim-read-only");
+    let _ = claim(&fixture, request).await.unwrap();
+    for link in fixture
+        .api
+        .store
+        .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
+        .unwrap()
+        .into_iter()
+        .filter(|link| link.unlinked_at.is_none())
+    {
+        assert_eq!(
+            link.capabilities,
+            workdir::WorkdirSessionCapabilities::READ_ONLY
+        );
+        let workdir = fixture
+            .api
+            .store
+            .get_workdir_registry(TEST_WORKSPACE_ID, &link.workdir_id)
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            effective_worker_workdir_capabilities(
+                &fixture.api,
+                &fixture.worker,
+                &workdir,
+                link.capabilities
+            )
+            .unwrap(),
+            workdir::WorkdirSessionCapabilities::READ_ONLY
+        );
+    }
+    assert!(
+        fixture
+            .runtime
+            .logical_attachments(&fixture.worker.worker_id)
+            .iter()
+            .all(|a| a.capabilities == workdir::WorkdirSessionCapabilities::READ_ONLY)
+    );
+    assert!(
+        require_ticket_read_write_target(
+            &fixture.api,
+            &fixture.ticket_id,
+            &test_repository_id(&fixture.api),
+            "develop"
+        )
+        .is_err()
+    );
+}
+
+#[tokio::test]
+async fn explicit_existing_worker_claim_requires_connections_instead_of_ambient_attachments() {
+    let fixture = manual_worker_assignment_fixture().await;
+    let mut request = manual_worker_assignment_request(&fixture, "claim-ambient-denied");
+    request.workdir_bindings.clear();
+    let response = claim(&fixture, request).await.unwrap_err().into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("exact live Workdir connections"));
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 0);
+    assert!(
+        fixture
+            .api
+            .store
+            .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-ambient-denied")
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[tokio::test]
+async fn ambiguous_existing_worker_restore_keeps_claim_pending_until_accepted_recovery() {
+    let fixture = manual_worker_assignment_fixture().await;
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    let request = manual_worker_assignment_request(&fixture, "claim-unknown");
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    let operation = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(operation.claim_state, "pending");
+    assert!(operation.assignment_id.is_none());
+    assert_eq!(operation.worker, Some(fixture.worker.clone()));
+    assert!(operation.request_fingerprint.is_some());
+    let recovery = fixture
+        .api
+        .config_store
+        .get_ticket_claim_recovery(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        recovery.item_revision,
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .item_revision
+    );
+    assert_eq!(recovery.selected.len(), request.workdir_bindings.len());
+    for selected in &request.workdir_bindings {
+        assert!(recovery.selected.contains(selected));
+    }
+    assert!(
+        recovery
+            .original_links
+            .iter()
+            .all(|link| link.capabilities == workdir::WorkdirSessionCapabilities::ALL)
+    );
+    assert_eq!(
+        recovery
+            .effective_links
+            .iter()
+            .find(|link| link.alias == "docs")
+            .unwrap()
+            .capabilities,
+        workdir::WorkdirSessionCapabilities::READ_ONLY
+    );
+    // This fake proves the domain transition, not execution-level restore deduplication.
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    let pending = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(pending.request_fingerprint, operation.request_fingerprint);
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 2);
+    assert!(
+        fixture
+            .api
+            .store
+            .reserve_ticket_assignment_operation(
+                TEST_WORKSPACE_ID,
+                "competing-launch",
+                &fixture.ticket_id,
+                &fixture.worker.runtime_id,
+                None,
+                "other-intent",
+                TEST_CREATED_AT
+            )
+            .is_err()
+    );
+    let worker = WorkspaceWorker::resolve(
+        &fixture.api,
+        &fixture.worker.runtime_id,
+        &fixture.worker.worker_id,
+    )
+    .unwrap();
+    let denied = worker
+        .input(
+            &WorkerOperationContext::Backend,
+            WorkerInputRequest {
+                kind: WorkerInputKind::User,
+                content: "Do not run while claim is unresolved".into(),
+                segments: None,
+                submission_request_id: None,
+            },
+        )
+        .await
+        .unwrap_err();
+    assert!(denied.error.to_string().contains("pending Ticket claim"));
+    assert_eq!(
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .state,
+        "ready"
+    );
+    assert!(
+        fixture
+            .api
+            .store
+            .get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
+            .unwrap()
+            .is_none()
+    );
+
+    let Json(accepted) = claim(&fixture, request.clone()).await.unwrap();
+    let assignment = accepted.assignment.unwrap();
+    let committed = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, "claim-unknown")
+        .unwrap()
+        .unwrap();
+    assert_eq!(committed.claim_state, "committed");
+    assert_eq!(
+        committed.assignment_id,
+        Some(assignment.assignment_id.clone())
+    );
+    assert_eq!(
+        fixture
+            .api
+            .authority
+            .ticket(&fixture.ticket_id)
+            .unwrap()
+            .state,
+        "inprogress"
+    );
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 3);
+    assert_eq!(
+        fixture
+            .api
+            .config_store
+            .get_ticket_claim_recovery(TEST_WORKSPACE_ID, "claim-unknown")
+            .unwrap()
+            .unwrap(),
+        recovery
+    );
+    let Json(replay) = claim(&fixture, request).await.unwrap();
+    assert_eq!(
+        replay.assignment.unwrap().assignment_id,
+        assignment.assignment_id
+    );
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 3);
+}
+
+#[tokio::test]
+async fn pending_manual_worker_claim_rejects_changed_bindings_without_restore() {
+    let fixture = manual_worker_assignment_fixture().await;
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    let mut request = manual_worker_assignment_request(&fixture, "claim-changed-bindings");
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    let admitted = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    request.workdir_bindings[0].connection_id = "changed-connection".into();
+    let response = claim(&fixture, request.clone())
+        .await
+        .unwrap_err()
+        .into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("different input"));
+    let pending = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(pending.request_fingerprint, admitted.request_fingerprint);
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 1);
+}
+
+#[tokio::test]
+async fn pending_manual_worker_claim_without_pinned_intent_rejects_replay_without_restore() {
+    let fixture = manual_worker_assignment_fixture().await;
+    *fixture.runtime.next_restore_outcome.lock().unwrap() =
+        Some(server_api::WorkerRestoreState::ReconciliationRequired);
+    let request = manual_worker_assignment_request(&fixture, "claim-missing-intent");
+    assert!(claim(&fixture, request.clone()).await.is_err());
+    // Simulate an interrupted/legacy pending receipt with no exact restore intent.
+    let request_id = format!("manual-worker-restore:{}", request.operation_id);
+    fixture.api.config_store.with_conn(|conn| {
+        let deleted = conn.execute(
+            "DELETE FROM worker_restore_intents WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND request_id=?4",
+            rusqlite::params![TEST_WORKSPACE_ID, fixture.worker.runtime_id, fixture.worker.worker_id, request_id],
+        )?;
+        assert_eq!(deleted, 1);
+        Ok(())
+    }).unwrap();
+    let response = claim(&fixture, request.clone())
+        .await
+        .unwrap_err()
+        .into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("no pinned restore intent"));
+    let pending = fixture
+        .api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 1);
+    assert!(
+        fixture
+            .api
+            .config_store
+            .get_internal_worker_restore_intent(TEST_WORKSPACE_ID, &fixture.worker, &request_id)
+            .unwrap()
+            .is_none()
+    );
+}
+
+#[tokio::test]
+async fn pending_manual_worker_claim_reconciles_same_runtime_restore_and_commits_without_input() {
+    let workspace = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(workspace.path()).await;
+    execution.accept_restores();
+    let worker = spawn_ticket_check_source(&api, "manual-claim-reconciliation");
+    assert!(
+        api.store
+            .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .is_empty()
+    );
+    let backend = browser_ticket_backend(&api).unwrap();
+    let mut input = ticket::NewTicket::new("Manual investigation without resources");
+    input.workflow_state = Some(TicketWorkflowState::Ready);
+    let ticket_id = backend.create(input).unwrap().id;
+    backend
+        .edit_item(
+            ticket_id.clone().into(),
+            TicketItemEdit {
+                targets: Some(TicketTargetsEdit::Clear),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    backend
+        .set_workflow_state(
+            ticket_id.clone().into(),
+            TicketStateChange::new(
+                "ready",
+                "planning",
+                "Explicit investigation",
+                "No repository resources required",
+            ),
+        )
+        .unwrap();
+    let admitted_revision = api.authority.ticket(&ticket_id).unwrap().item_revision;
+    let request = server_api::SetTicketRoleAssignmentRequest {
+        operation_id: "claim-runtime-reconciliation".into(),
+        principal: server_api::TicketAssignmentPrincipal::Worker {
+            runtime_id: worker.runtime_id.clone(),
+            worker_id: worker.worker_id.clone(),
+        },
+        expected_assignment_id: None,
+        workdir_bindings: Vec::new(),
+    };
+    let submit_claim = || {
+        scoped_set_ticket_assignment(
+            State(api.clone()),
+            AxumPath((TEST_WORKSPACE_ID.into(), ticket_id.clone(), "worker".into())),
+            Json(request.clone()),
+        )
+    };
+    let uncertain_restore = || {
+        worker_runtime::execution::WorkerExecutionSpawnResult::Errored(
+            worker_runtime::execution::WorkerExecutionResult::rejected(
+                worker_runtime::execution::WorkerExecutionOperation::Restore,
+                "uncertain manual claim restore",
+            ),
+        )
+    };
+    *execution.restore_failure.lock().unwrap() = Some(uncertain_restore());
+    let response = submit_claim().await.unwrap_err().into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("unknown"));
+    let pending = api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pending.claim_state, "pending");
+    assert!(pending.assignment_id.is_none());
+    assert_eq!(pending.worker, Some(worker.clone()));
+    assert!(pending.request_fingerprint.is_some());
+    assert_eq!(api.authority.ticket(&ticket_id).unwrap().state, "planning");
+    assert!(
+        api.store
+            .get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket_id)
+            .unwrap()
+            .is_none()
+    );
+    let recovery = api
+        .config_store
+        .get_ticket_claim_recovery(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(recovery.item_revision, admitted_revision);
+    assert!(recovery.selected.is_empty());
+    assert!(recovery.original_links.is_empty());
+    assert!(recovery.effective_links.is_empty());
+    let request_id = format!("manual-worker-restore:{}", request.operation_id);
+    let pinned = api
+        .config_store
+        .get_internal_worker_restore_intent(TEST_WORKSPACE_ID, &worker, &request_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(pinned.request_id, request_id);
+    let original = execution.restore_operations.lock().unwrap()[0];
+    assert_eq!(
+        execution.restore_operations.lock().unwrap().as_slice(),
+        &[original]
+    );
+    assert!(execution.reconcile_operations.lock().unwrap().is_empty());
+
+    let handle = WorkspaceWorker::resolve(&api, &worker.runtime_id, &worker.worker_id).unwrap();
+    let denied = handle
+        .input(
+            &WorkerOperationContext::Backend,
+            WorkerInputRequest {
+                kind: WorkerInputKind::User,
+                content: "Do not submit while the manual claim is pending".into(),
+                segments: None,
+                submission_request_id: None,
+            },
+        )
+        .await
+        .unwrap_err();
+    assert!(denied.error.to_string().contains("pending Ticket claim"));
+
+    *execution.restore_failure.lock().unwrap() = Some(uncertain_restore());
+    let response = submit_claim().await.unwrap_err().into_response();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(
+        status,
+        StatusCode::CONFLICT,
+        "{}",
+        String::from_utf8_lossy(&bytes)
+    );
+    assert!(String::from_utf8_lossy(&bytes).contains("pending"));
+    let still_pending = api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(still_pending.claim_state, "pending");
+    assert!(still_pending.assignment_id.is_none());
+    assert_eq!(
+        still_pending.request_fingerprint,
+        pending.request_fingerprint
+    );
+    assert_eq!(
+        api.authority.ticket(&ticket_id).unwrap().item_revision,
+        admitted_revision
+    );
+    assert_eq!(
+        api.config_store
+            .get_internal_worker_restore_intent(TEST_WORKSPACE_ID, &worker, &request_id)
+            .unwrap()
+            .unwrap(),
+        pinned
+    );
+    assert_eq!(
+        execution.reconcile_operations.lock().unwrap().as_slice(),
+        &[original]
+    );
+
+    let Json(accepted) = submit_claim().await.unwrap();
+    let assignment = accepted.assignment.unwrap();
+    assert_eq!(assignment.principal, request.principal);
+    let committed = api
+        .store
+        .get_ticket_assignment_operation(TEST_WORKSPACE_ID, &request.operation_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(committed.claim_state, "committed");
+    assert_eq!(
+        committed.assignment_id,
+        Some(assignment.assignment_id.clone())
+    );
+    assert_eq!(committed.request_fingerprint, pending.request_fingerprint);
+    assert_eq!(
+        api.authority.ticket(&ticket_id).unwrap().state,
+        "inprogress"
+    );
+    assert_eq!(
+        api.store
+            .get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket_id)
+            .unwrap()
+            .unwrap()
+            .assignment_id,
+        assignment.assignment_id
+    );
+    assert_eq!(
+        api.config_store
+            .get_ticket_claim_recovery(TEST_WORKSPACE_ID, &request.operation_id)
+            .unwrap()
+            .unwrap(),
+        recovery
+    );
+    assert!(
+        api.store
+            .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .is_empty()
+    );
+    // The executor logs reconcile_restore in restore_operations as well. There is
+    // exactly one fresh dispatch; both subsequent entries reconcile its same id.
+    assert_eq!(
+        execution.restore_operations.lock().unwrap().as_slice(),
+        &[original, original, original]
+    );
+    assert_eq!(
+        execution.reconcile_operations.lock().unwrap().as_slice(),
+        &[original, original]
+    );
+    let Json(replay) = submit_claim().await.unwrap();
+    assert_eq!(
+        replay.assignment.unwrap().assignment_id,
+        assignment.assignment_id
+    );
+    assert_eq!(execution.restore_operations.lock().unwrap().len(), 3);
+    assert_eq!(execution.reconcile_operations.lock().unwrap().len(), 2);
+    assert!(
+        !execution
+            .protocol_methods()
+            .iter()
+            .any(|(_, method)| matches!(
+                method,
+                protocol::Method::Submit { .. }
+                    | protocol::Method::SubmitTracked { .. }
+                    | protocol::Method::SubmitIfIdle { .. }
+            )),
+        "manual claims and denied input must not generate a Submit"
+    );
+}

crates/workspace-server/src/server/worker_operations.rs

@@ -359,6 +359,8 @@ impl WorkspaceWorker {
             "send_input"
         };
         let _guard = self.authorize_operation(context, permission).await?;
+        let session_lock = current_worker_session_lock(&self.api, &self.identity);
+        let _session_guard = session_lock.lock().await;
         self.ensure_interactive()?;
         if matches!(context, WorkerOperationContext::WorkerControl { .. })
             && request.kind == WorkerInputKind::User
@@ -373,6 +375,19 @@ impl WorkspaceWorker {
     }
 
     fn ensure_interactive(&self) -> ApiResult<()> {
+        if let Some(claim) = self
+            .api
+            .store
+            .get_pending_ticket_assignment_operation_for_worker(
+                self.api.workspace_id(),
+                &self.identity,
+            )?
+        {
+            return Err(Error::TicketAssignmentConflict(format!(
+                "Worker has pending Ticket claim {}; reconcile that exact operation before admitting input",
+                claim.operation_id,
+            )).into());
+        }
         if self
             .api
             .store
@@ -578,6 +593,8 @@ impl WorkspaceWorkerMethodSender {
             .worker
             .authorize_operation(context, "send_input")
             .await?;
+        let session_lock = current_worker_session_lock(&self.worker.api, &self.worker.identity);
+        let _session_guard = session_lock.lock().await;
         self.worker.ensure_interactive()?;
         let WorkerOperationContext::Browser { source, .. } = context else {
             return Err(Error::WorkspacePermissionDenied(

crates/workspace-server/src/server_workspace_config_tests.rs

@@ -129,7 +129,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_signed_editor_routes_cannot_bypass_wip_authority() {
-        let mut fixture = manual_coder_assignment_fixture().await;
+        let mut fixture = manual_worker_assignment_fixture().await;
         let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
         configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
         let api = &fixture.api;
@@ -222,7 +222,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_entrypoint_support_is_exact_and_canonical_mutations_stay_guarded() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -299,7 +299,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_grants_owner_scope_and_signed_routes() {
-        let mut fixture = manual_coder_assignment_fixture().await;
+        let mut fixture = manual_worker_assignment_fixture().await;
         let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
         configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
         let api = &fixture.api;
@@ -615,7 +615,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_readonly_revoke_and_conditional_lifetime_fences() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let read_grant = grant(api, &fixture.worker, Access::ReadOnly).await;
         assert!(
@@ -827,7 +827,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_two_workers_ui_cas_and_identical_virtual_replacement() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let second = second_worker(&fixture).await;
         grant(api, &fixture.worker, Access::ReadWrite).await;
@@ -982,7 +982,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_invalid_bounds_and_failed_save_do_not_corrupt_tree() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1114,7 +1114,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_operation_and_revoke_share_the_detach_session_lock() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let granted = grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1168,7 +1168,7 @@ mod workspace_config_integration {
     }
     #[tokio::test]
     async fn workspace_config_grant_restart_source_immutability_and_owner_acceptance_guard() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         let granted = grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1255,7 +1255,7 @@ mod workspace_config_integration {
 
     #[tokio::test]
     async fn workspace_config_auxiliary_secret_effects_are_unknown_not_false_rollback() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;
@@ -1324,7 +1324,7 @@ mod workspace_config_integration {
     }
     #[tokio::test]
     async fn workspace_config_observe_node_limit_is_complete_or_error_not_silent_truncation() {
-        let fixture = manual_coder_assignment_fixture().await;
+        let fixture = manual_worker_assignment_fixture().await;
         let api = &fixture.api;
         grant(api, &fixture.worker, Access::ReadWrite).await;
         let attached = attach(api, &fixture.worker).await;

crates/workspace-server/src/server_workspace_config_wip_tests.rs

@@ -259,7 +259,7 @@ async fn assert_config_catalog(runtime: &worker::wip::WipRuntime, attached: &Att
 async fn workspace_config_real_selected_invoke_backend_wip_edit_create_delete_import_and_unknown() {
     use worker::feature::builtin::workspace_config::wip::mount_workspace_config_wip;
     use worker::feature::builtin::workspace_config::{CONTENT_ROOT, WorkspaceConfigFeature};
-    let mut fixture = manual_coder_assignment_fixture().await;
+    let mut fixture = manual_worker_assignment_fixture().await;
     let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
     configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
     let granted = grant(&fixture.api, &fixture.worker, Access::ReadWrite).await;
@@ -706,7 +706,7 @@ async fn workspace_config_actual_remote_and_embedded_clients_use_authenticated_w
         decode_runtime_request_source_claims,
     };
     // Remote Runtime-owned HTTP adapter, including its real request signer.
-    let mut fixture = manual_coder_assignment_fixture().await;
+    let mut fixture = manual_worker_assignment_fixture().await;
     let identity = RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id).unwrap();
     configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
     let claims = Arc::new(Mutex::new(Vec::new()));