diff --git a/crates/merge-request/src/lib.rs b/crates/merge-request/src/lib.rs index 78a40bbd5..3afda0d9a 100644 --- a/crates/merge-request/src/lib.rs +++ b/crates/merge-request/src/lib.rs @@ -979,7 +979,7 @@ impl MergeRequestStore { } let assignment_is_current: bool = transaction.query_row( "SELECT EXISTS( - SELECT 1 FROM ticket_current_worker_assignments + SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3 )", params![i.auth.workspace_id, i.ticket_id, i.auth.assignment_id], @@ -1174,7 +1174,7 @@ impl MergeRequestStore { } let assignment_is_current: bool = transaction.query_row( "SELECT EXISTS( - SELECT 1 FROM ticket_current_worker_assignments + SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3 )", params![ @@ -1291,20 +1291,8 @@ impl MergeRequestStore { "Ticket state changed during completion".into(), )); } - let released = transaction.execute( - "DELETE FROM ticket_current_worker_assignments - WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3", - params![ - input.auth.workspace_id, - input.ticket_id, - input.auth.assignment_id - ], - )?; - if released != 1 { - return Err(MergeRequestError::Unauthorized( - "Ticket completion assignment changed while releasing it".into(), - )); - } + // Terminal state ends unfinished work through the Workspace authority's + // lifecycle trigger. Keep the responsibility identity and its history. let event = TicketCompletionEvent { operation_id: input.operation_id, ticket_id: input.ticket_id.clone(), diff --git a/crates/merge-request/tests/store.rs b/crates/merge-request/tests/store.rs index b73f656eb..183e7c677 100644 --- a/crates/merge-request/tests/store.rs +++ b/crates/merge-request/tests/store.rs @@ -39,7 +39,7 @@ fn fixture() -> (tempfile::TempDir, MergeRequestStore) { let d = tempfile::tempdir().unwrap(); let p = d.path().join("db"); let c = Connection::open(&p).unwrap(); - c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO repositories VALUES('W','R2');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t');").unwrap(); + c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO repositories VALUES('W','R2');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t');CREATE VIEW ticket_active_worker_assignments AS SELECT current.* FROM ticket_current_worker_assignments current JOIN typed_tickets ticket ON ticket.workspace_id=current.workspace_id AND ticket.ticket_id=current.ticket_id WHERE ticket.workflow_state NOT IN ('done','closed');").unwrap(); drop(c); let a = Assignments(Arc::new(Mutex::new(CurrentAssignment { assignment_id: "A".into(), @@ -250,7 +250,9 @@ fn selectors_thread_and_completion_have_no_revision_or_commit_api() { ) .unwrap(); assert_eq!(state, "done"); - assert!(!assignment_exists); + assert!(assignment_exists); + let active: bool = connection.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id='W' AND ticket_id='T')", [], |row| row.get(0)).unwrap(); + assert!(!active); let replayed_completion = s .complete_ticket(CompleteTicket { ticket_id: "T".into(), diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 789229581..b88574510 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -875,7 +875,7 @@ impl WorkerOperation { "Restore a stopped Backend/Runtime Worker session in the current Workspace. Supply expected_observation_token from WorkerList; a conflict requires refreshing and deliberate new intent. New requests default to the durable Tool call ID. For explicit recovery of an unknown/pending outcome, supply the reported request_id and the original token unchanged; never create a new intent or automatically retry." } Self::Remove => { - "Remove an eligible stopped, unassigned, non-internal Worker. Supply a bounded reason; Backend validation and retention are authoritative." + "Remove an eligible stopped, non-internal Worker with no unfinished Ticket work. Terminal Ticket responsibility is retained and does not require unassignment. Supply a bounded reason; Backend validation and retention are authoritative." } } } diff --git a/crates/worker/src/feature/builtin/merge_request.rs b/crates/worker/src/feature/builtin/merge_request.rs index 88460b2d7..b1d1e786c 100644 --- a/crates/worker/src/feature/builtin/merge_request.rs +++ b/crates/worker/src/feature/builtin/merge_request.rs @@ -1149,7 +1149,7 @@ pub fn description(n: &str) -> Option<&'static str> { "Record Orchestrator-owned integration for the explicitly addressed Merge Request without completing the Ticket or releasing its assignment.", ), "CompleteTicket" => Some( - "Complete the Ticket only after reviewing its current item revision and exact linked Merge Request result set; this releases the current assignment atomically.", + "Complete the Ticket only after reviewing its current item revision and exact linked Merge Request result set; this ends unfinished work atomically while retaining responsibility and assignment history.", ), "ReviewMergeRequest" => Some( "Submit the injected Reviewer capability result for its captured exact source ref; source movement cancels it, while target-only movement does not.", diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index 1b72b8415..6ef4aa501 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -1027,7 +1027,7 @@ mod tests { } #[test] - fn builtin_orchestrator_cleanup_policy_renders_with_common_includes() { + fn builtin_orchestrator_cleanup_retains_terminal_responsibility_and_blocks_unfinished_work() { let rendered = PromptCatalog::builtins_only() .unwrap() .render_name("role.orchestrator", Value::UNDEFINED) @@ -1040,7 +1040,21 @@ mod tests { assert!(rendered.contains("call `WorkerStop`")); assert!(rendered.contains("call `WorkerRemove`")); assert!(rendered.contains("only then call `WorkdirDelete`")); - assert!(rendered.contains("`CurrentAssignment` means unassign and reread")); + assert!(rendered.contains("must have no unfinished Ticket work")); + assert!(rendered.contains("does not require unassignment before cleanup")); + assert!(rendered.contains("`UnfinishedWork` (`unfinished_work` on the wire)")); + assert!(rendered.contains("requires explicit work end or reassignment")); + assert!(rendered.contains("idle or stopped status does not end unfinished work")); + for stale_policy in [ + "no current Ticket assignment", + "unassign it through the available orchestration authority", + "`CurrentAssignment`", + ] { + assert!( + !rendered.contains(stale_policy), + "stale cleanup policy: {stale_policy}" + ); + } assert!(!rendered.contains("{% include")); } } diff --git a/crates/workspace-server/src/assignment_work.sql b/crates/workspace-server/src/assignment_work.sql new file mode 100644 index 000000000..72de3169c --- /dev/null +++ b/crates/workspace-server/src/assignment_work.sql @@ -0,0 +1,100 @@ +-- Responsibility is durable. A release only ends the operational authority of +-- this assignment identity; reopening a Ticket must not revive that authority. +CREATE TABLE ticket_assignment_work_releases ( + workspace_id TEXT NOT NULL, + assignment_id TEXT NOT NULL, + released_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, assignment_id), + FOREIGN KEY (workspace_id, assignment_id) + REFERENCES ticket_worker_assignments(workspace_id, assignment_id) ON DELETE CASCADE +); +CREATE VIEW ticket_active_worker_assignments AS + SELECT current.* FROM ticket_current_worker_assignments AS current + JOIN typed_tickets AS ticket + ON ticket.workspace_id = current.workspace_id AND ticket.ticket_id = current.ticket_id + WHERE ticket.workflow_state NOT IN ('done', 'closed') + AND NOT EXISTS (SELECT 1 FROM ticket_assignment_work_releases AS released + WHERE released.workspace_id = current.workspace_id + AND released.assignment_id = current.assignment_id); +CREATE TRIGGER ticket_terminal_releases_work +AFTER UPDATE OF workflow_state ON typed_tickets +WHEN NEW.workflow_state IN ('done', 'closed') +BEGIN + INSERT OR IGNORE INTO ticket_assignment_work_releases + SELECT workspace_id, assignment_id, COALESCE(NEW.updated_at, updated_at) + FROM ticket_current_worker_assignments + WHERE workspace_id = NEW.workspace_id AND ticket_id = NEW.ticket_id; +END; +CREATE TRIGGER ticket_terminal_assignment_releases_work_insert +AFTER INSERT ON ticket_current_worker_assignments +WHEN EXISTS (SELECT 1 FROM typed_tickets WHERE workspace_id = NEW.workspace_id + AND ticket_id = NEW.ticket_id AND workflow_state IN ('done', 'closed')) +BEGIN + INSERT OR IGNORE INTO ticket_assignment_work_releases + VALUES (NEW.workspace_id, NEW.assignment_id, NEW.updated_at); +END; +CREATE TRIGGER ticket_terminal_assignment_releases_work_update +AFTER UPDATE ON ticket_current_worker_assignments +WHEN EXISTS (SELECT 1 FROM typed_tickets WHERE workspace_id = NEW.workspace_id + AND ticket_id = NEW.ticket_id AND workflow_state IN ('done', 'closed')) +BEGIN + INSERT OR IGNORE INTO ticket_assignment_work_releases + VALUES (NEW.workspace_id, NEW.assignment_id, NEW.updated_at); +END; +CREATE TRIGGER ticket_active_worker_role_insert +BEFORE INSERT ON ticket_current_worker_assignments +WHEN NEW.principal_kind = 'worker' + AND EXISTS (SELECT 1 FROM typed_tickets WHERE workspace_id = NEW.workspace_id + AND ticket_id = NEW.ticket_id AND workflow_state NOT IN ('done', 'closed')) + AND NOT EXISTS (SELECT 1 FROM ticket_assignment_work_releases + WHERE workspace_id = NEW.workspace_id AND assignment_id = NEW.assignment_id) + AND EXISTS (SELECT 1 FROM ticket_active_worker_assignments + WHERE workspace_id = NEW.workspace_id AND role = NEW.role + AND runtime_id = NEW.runtime_id AND worker_id = NEW.worker_id + AND assignment_id != NEW.assignment_id) +BEGIN SELECT RAISE(ABORT, 'Worker already has unfinished Ticket work in this role'); END; +CREATE TRIGGER ticket_active_worker_role_update +BEFORE UPDATE ON ticket_current_worker_assignments +WHEN NEW.principal_kind = 'worker' + AND EXISTS (SELECT 1 FROM typed_tickets WHERE workspace_id = NEW.workspace_id + AND ticket_id = NEW.ticket_id AND workflow_state NOT IN ('done', 'closed')) + AND NOT EXISTS (SELECT 1 FROM ticket_assignment_work_releases + WHERE workspace_id = NEW.workspace_id AND assignment_id = NEW.assignment_id) + AND EXISTS (SELECT 1 FROM ticket_active_worker_assignments + WHERE workspace_id = NEW.workspace_id AND role = NEW.role + AND runtime_id = NEW.runtime_id AND worker_id = NEW.worker_id + AND assignment_id != OLD.assignment_id) +BEGIN SELECT RAISE(ABORT, 'Worker already has unfinished Ticket work in this role'); END; +-- Keep archived responsibilities valid after Worker removal, but never allow +-- newly installed identities to refer to a removed / removal-reserved Worker. +CREATE TRIGGER ticket_active_worker_principal_insert +BEFORE INSERT ON ticket_current_worker_assignments +WHEN NEW.principal_kind = 'worker' AND ( + NOT EXISTS (SELECT 1 FROM worker_registry WHERE workspace_id = NEW.workspace_id + AND runtime_id = NEW.runtime_id AND worker_id = NEW.worker_id) + OR EXISTS (SELECT 1 FROM worker_removal_operations WHERE workspace_id = NEW.workspace_id + AND runtime_id = NEW.runtime_id AND worker_id = NEW.worker_id + AND state IN ('executing', 'failed', 'succeeded'))) +BEGIN SELECT RAISE(ABORT, 'Worker is missing or removal is reserved'); END; +CREATE TRIGGER ticket_active_worker_principal_update +BEFORE UPDATE ON ticket_current_worker_assignments +WHEN NEW.principal_kind = 'worker' AND ( + NOT EXISTS (SELECT 1 FROM worker_registry WHERE workspace_id = NEW.workspace_id + AND runtime_id = NEW.runtime_id AND worker_id = NEW.worker_id) + OR EXISTS (SELECT 1 FROM worker_removal_operations WHERE workspace_id = NEW.workspace_id + AND runtime_id = NEW.runtime_id AND worker_id = NEW.worker_id + AND state IN ('executing', 'failed', 'succeeded'))) +BEGIN SELECT RAISE(ABORT, 'Worker is missing or removal is reserved'); END; +CREATE TRIGGER ticket_unfinished_work_blocks_worker_delete +BEFORE DELETE ON worker_registry +WHEN EXISTS (SELECT 1 FROM ticket_active_worker_assignments + WHERE workspace_id = OLD.workspace_id AND runtime_id = OLD.runtime_id + AND worker_id = OLD.worker_id) +BEGIN SELECT RAISE(ABORT, 'Worker has unfinished Ticket work'); END; +CREATE TRIGGER ticket_unfinished_work_blocks_worker_move +BEFORE UPDATE OF runtime_id, worker_id, workspace_id ON worker_registry +WHEN (OLD.runtime_id != NEW.runtime_id OR OLD.worker_id != NEW.worker_id OR OLD.workspace_id != NEW.workspace_id) + AND EXISTS (SELECT 1 FROM ticket_active_worker_assignments + WHERE workspace_id = OLD.workspace_id AND runtime_id = OLD.runtime_id + AND worker_id = OLD.worker_id) +BEGIN SELECT RAISE(ABORT, 'Worker has unfinished Ticket work'); END; diff --git a/crates/workspace-server/src/authority.rs b/crates/workspace-server/src/authority.rs index 8a60ffe40..2ef3830e2 100644 --- a/crates/workspace-server/src/authority.rs +++ b/crates/workspace-server/src/authority.rs @@ -146,7 +146,7 @@ impl merge_request::AssignmentSource for AuthorityMergeRequestSource { ticket_id: &str, ) -> std::result::Result, String> { self.store - .get_current_ticket_coder_assignment(workspace_id, ticket_id) + .get_active_ticket_worker_assignment(workspace_id, ticket_id) .map(|assignment| { assignment.map(|assignment| merge_request::CurrentAssignment { assignment_id: assignment.assignment_id, @@ -831,10 +831,15 @@ impl SqliteWorkspaceAuthority { }) }) .transpose()?; - let has_orchestrator = role_assignments + // Responsibility remains visible after work ends; action eligibility must + // use only active unfinished work, including after a Ticket is reopened. + let active_role_assignments = self + .store + .list_active_ticket_role_assignments(&self.workspace_id, id)?; + let has_orchestrator = active_role_assignments .iter() .any(|assignment| assignment.role == TicketAssignmentRole::Orchestrator); - let has_coder = role_assignments + let has_coder = active_role_assignments .iter() .any(|assignment| assignment.role == TicketAssignmentRole::Coder); let targets = ticket @@ -860,7 +865,7 @@ impl SqliteWorkspaceAuthority { for ticket_id in &dependency_check.queue_tickets { let assignments = self .store - .list_current_ticket_role_assignments(&self.workspace_id, ticket_id)?; + .list_active_ticket_role_assignments(&self.workspace_id, ticket_id)?; if !assignments .iter() .any(|assignment| assignment.role == TicketAssignmentRole::Orchestrator) diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index 64275a3ab..3e9db8685 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -384,9 +384,6 @@ CREATE TABLE ticket_current_worker_assignments ( FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE, - FOREIGN KEY (workspace_id, runtime_id, worker_id) - REFERENCES worker_registry(workspace_id, runtime_id, worker_id) - ON DELETE RESTRICT, CHECK( (principal_kind = 'worker' AND runtime_id IS NOT NULL AND worker_id IS NOT NULL AND principal_id IS NULL) OR @@ -1219,9 +1216,6 @@ CREATE INDEX ticket_current_principal_idx CREATE UNIQUE INDEX ticket_current_singleton_role_idx ON ticket_current_worker_assignments(workspace_id, ticket_id, role) WHERE role IN ('orchestrator', 'coder'); -CREATE UNIQUE INDEX ticket_current_worker_role_idx - ON ticket_current_worker_assignments(workspace_id, role, runtime_id, worker_id) - WHERE principal_kind = '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 diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index bb36636f8..34cf2c671 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -69,7 +69,7 @@ pub struct WorkerRemovalPlanRequest { #[serde(tag = "kind", rename_all = "snake_case")] pub enum WorkerRemovalBlocker { Hold, - CurrentAssignment { + UnfinishedWork { assignment_id: String, ticket_id: String, }, @@ -229,8 +229,8 @@ impl SqliteWorkspaceStore { }; let mut blockers=Vec::new(); if worker.retention_state=="pinned" { blockers.push(WorkerRemovalBlocker::Hold); } - if let Some((assignment_id,ticket_id))=tx.query_row("SELECT a.assignment_id,a.ticket_id FROM ticket_current_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::CurrentAssignment{assignment_id,ticket_id}); + 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}); } let fp=fingerprint(req,&worker.updated_at,inv,&policy,&blockers)?; let plan_id=stable("wrp",&fp); let operation_id=stable("wro",&fp); @@ -276,10 +276,10 @@ 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_current_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)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?; if assigned{ - mark_stale(&tx,&plan,"current assignment added")?;tx.commit()?; - return Err(stale_error(&plan,"current assignment added")); + mark_stale(&tx,&plan,"unfinished work added")?;tx.commit()?; + return Err(stale_error(&plan,"unfinished work added")); } let now=Utc::now().to_rfc3339(); tx.execute("UPDATE worker_removal_operations SET state='executing',updated_at=?1 WHERE operation_id=?2",params![now,plan.operation_id])?; @@ -462,8 +462,8 @@ 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_current_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))?; - if assigned { return Err(StoreError::InvalidInput(format!("stale:{}:current assignment 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))?; + 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{ if plan.archive_id.as_deref()!=Some(&a.archive_id)||a.workspace_id!=workspace_id||a.source_runtime_id!=plan.worker.runtime_id||a.source_worker_id.to_string()!=plan.worker.worker_id||a.policy_id!=plan.policy_id||a.policy_revision!=plan.policy_revision{return Err(StoreError::InvalidInput("archive manifest mismatch".into()));} @@ -1427,29 +1427,60 @@ mod tests { ).map_err(StoreError::from)).unwrap(); assert_eq!(historical, 1); } + fn install_current_assignment( + s: &SqliteWorkspaceStore, + assignment_id: &str, + ) -> crate::Result<()> { + let assignment = TicketCoderAssignmentRecord { + workspace_id: "w".into(), + ticket_id: "ticket".into(), + assignment_id: assignment_id.into(), + worker: req().worker, + assigned_by: "test".into(), + assigned_at: "t".into(), + }; + s.set_current_ticket_coder_assignment( + &assignment, + None, + &format!("event-{assignment_id}"), + &format!("operation-{assignment_id}"), + false, + ) + .map(|_| ()) + } + + fn seed_current_assignment(s: &SqliteWorkspaceStore) { + install_current_assignment(s, "assignment").unwrap(); + } + + fn reopened_terminal_assignment_store(terminal_state: &str) -> SqliteWorkspaceStore { + let store = setup(); + seed_current_assignment(&store); + store + .with_conn(|conn| { + conn.execute( + "UPDATE typed_tickets SET workflow_state=?1 + WHERE workspace_id='w' AND ticket_id='ticket'", + [terminal_state], + )?; + conn.execute( + "UPDATE typed_tickets SET workflow_state='planning' + WHERE workspace_id='w' AND ticket_id='ticket'", + [], + )?; + Ok(()) + }) + .unwrap(); + store + } + #[test] fn assignment_and_orphan_are_authoritative() { let s = setup(); - s.with_conn(|c| { - let stable_worker_id = worker_id().to_string(); - c.execute( - "INSERT INTO ticket_worker_assignments(\ - workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at\ - ) VALUES('w','ticket','assignment','r',?1,'test','t')", - [&stable_worker_id], - )?; - c.execute( - "INSERT INTO ticket_current_worker_assignments(\ - workspace_id,ticket_id,assignment_id,runtime_id,worker_id,updated_at\ - ) VALUES('w','ticket','assignment','r',?1,'t')", - [&stable_worker_id], - )?; - Ok(()) - }) - .unwrap(); + seed_current_assignment(&s); let p = s.plan_worker_removal(&req(), &inv()).unwrap(); assert!( - matches!(&p.blockers[..],[WorkerRemovalBlocker::CurrentAssignment{assignment_id,ticket_id}] if assignment_id=="assignment"&&ticket_id=="ticket") + matches!(&p.blockers[..],[WorkerRemovalBlocker::UnfinishedWork{assignment_id,ticket_id}] if assignment_id=="assignment"&&ticket_id=="ticket") ); let runtime_only = WorkerRetentionInventory { workspace_id: "w".into(), @@ -1501,6 +1532,355 @@ mod tests { .unwrap(); assert_eq!(unchanged, 2); } + #[test] + fn unfinished_work_blocker_uses_explicit_wire_kind() { + let wire = serde_json::json!({ + "kind": "unfinished_work", + "assignment_id": "assignment", + "ticket_id": "ticket", + }); + let blocker: WorkerRemovalBlocker = serde_json::from_value(wire.clone()).unwrap(); + assert_eq!(serde_json::to_value(blocker).unwrap(), wire); + } + + #[test] + fn terminal_retained_assignment_allows_worker_removal_even_after_reopen() { + for (status, workflow_state) in [("closed", "closed"), ("open", "done")] { + for reopen in [false, true] { + let store = setup(); + seed_current_assignment(&store); + store + .with_conn(|conn| { + conn.execute( + "UPDATE typed_tickets SET status=?1, workflow_state=?2 + WHERE workspace_id='w' AND ticket_id='ticket'", + params![status, workflow_state], + )?; + if reopen { + conn.execute( + "UPDATE typed_tickets SET status='open', workflow_state='planning' + WHERE workspace_id='w' AND ticket_id='ticket'", + [], + )?; + } + Ok(()) + }) + .unwrap(); + store + .update_worker_retention_policy( + "w", + 1, + &WorkerRetentionPolicyUpdate { + policy_id: "purge".into(), + session_disposition: SessionDisposition::Purge, + metadata_disposition: MetadataDisposition::Tombstone, + archive_retention: ArchiveRetention::Forever, + diagnostics_disposition: DiagnosticsDisposition::Purge, + diagnostics_retention_seconds: None, + }, + ) + .unwrap(); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + assert!( + plan.blockers.is_empty(), + "{status}/{workflow_state}, reopened={reopen}" + ); + store + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + let result = WorkerRetentionExecutionResult { + operation_id: plan.operation_id.clone(), + input_fingerprint: plan.input_fingerprint.clone(), + expected_worker_revision: plan.worker_revision.clone(), + worker_id: worker_id(), + session_disposition: plan.session_disposition, + diagnostics_disposition: plan.diagnostics_disposition, + archive: None, + source_removed: true, + diagnostics_retained: false, + }; + let committed = store + .commit_worker_removal( + "w", + &plan.operation_id, + &plan.input_fingerprint, + &result, + ) + .unwrap(); + assert_eq!(committed.plan.state, WorkerRemovalPlanState::Succeeded); + assert!( + store + .get_worker_registry("w", &plan.worker) + .unwrap() + .is_none() + ); + assert!(store.worker_tombstone("w", &plan.worker).unwrap().is_some()); + let (retained, historical): (i64, i64) = store + .with_conn(|conn| { + Ok(( + conn.query_row( + "SELECT COUNT(*) FROM ticket_current_worker_assignments + WHERE workspace_id='w' AND assignment_id='assignment'", + [], + |row| row.get(0), + )?, + conn.query_row( + "SELECT COUNT(*) FROM ticket_worker_assignments + WHERE workspace_id='w' AND assignment_id='assignment'", + [], + |row| row.get(0), + )?, + )) + }) + .unwrap(); + assert_eq!( + (retained, historical), + (1, 1), + "assignment history must survive removal" + ); + } + } + } + + #[test] + fn unfinished_work_added_after_planning_blocks_preparation() { + let store = setup(); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + seed_current_assignment(&store); + let error = store + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap_err(); + assert!( + matches!(error, WorkerRetentionError::StalePlan { reason, .. } + if reason == "unfinished work added") + ); + assert!( + store + .get_worker_registry("w", &plan.worker) + .unwrap() + .is_some() + ); + } + + #[test] + fn distinct_assignment_after_terminal_end_blocks_preparation() { + for terminal_state in ["done", "closed"] { + let store = reopened_terminal_assignment_store(terminal_state); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + assert!(plan.blockers.is_empty()); + install_current_assignment(&store, "new-assignment").unwrap(); + let blocked = store.plan_worker_removal(&req(), &inv()).unwrap(); + assert_eq!( + blocked.blockers, + vec![WorkerRemovalBlocker::UnfinishedWork { + assignment_id: "new-assignment".into(), + ticket_id: "ticket".into(), + }] + ); + let replay = store.plan_worker_removal(&req(), &inv()).unwrap(); + assert_eq!(replay.plan_id, blocked.plan_id); + assert_eq!(replay.input_fingerprint, blocked.input_fingerprint); + assert_ne!(blocked.plan_id, plan.plan_id); + let error = store + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap_err(); + assert!( + matches!(error, WorkerRetentionError::StalePlan { reason, .. } + if reason == "unfinished work added") + ); + assert!( + store + .get_worker_registry("w", &plan.worker) + .unwrap() + .is_some() + ); + } + } + + #[test] + fn prepared_removal_fences_distinct_assignment_after_terminal_end() { + let store = reopened_terminal_assignment_store("done"); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + store + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + let error = install_current_assignment(&store, "new-assignment").unwrap_err(); + assert!( + matches!(error, StoreError::TicketAssignmentConflict(message) + if message.contains("being retained or has been removed")) + ); + let (current, history): (i64, i64) = store + .with_conn(|conn| { + Ok(( + conn.query_row( + "SELECT COUNT(*) FROM ticket_current_worker_assignments + WHERE workspace_id='w' AND ticket_id='ticket'", + [], + |row| row.get(0), + )?, + conn.query_row( + "SELECT COUNT(*) FROM ticket_worker_assignments + WHERE workspace_id='w' AND assignment_id='new-assignment'", + [], + |row| row.get(0), + )?, + )) + }) + .unwrap(); + assert_eq!(current, 1, "the ended responsibility must remain retained"); + assert_eq!(history, 0, "the rejected new identity must roll back"); + } + + #[test] + fn commit_rechecks_distinct_assignment_independently_of_admission_fence() { + let store = reopened_terminal_assignment_store("done"); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + store + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + // Fault injection is confined to this in-memory authority fixture: use + // direct SQLite writes to bypass the public API removal fence, and drop + // both principal admission triggers to prove commit's independent guard. + store + .with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + tx.execute_batch( + "DROP TRIGGER ticket_active_worker_principal_insert; + DROP TRIGGER ticket_active_worker_principal_update;", + )?; + tx.execute( + "INSERT INTO ticket_worker_assignments ( + workspace_id, ticket_id, assignment_id, runtime_id, worker_id, + assigned_by, assigned_at + ) VALUES ('w', 'ticket', 'new-assignment', 'r', ?1, 'test', 'new')", + [worker_id().to_string()], + )?; + 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' + AND assignment_id='assignment' + AND EXISTS ( + SELECT 1 FROM ticket_assignment_work_releases + WHERE workspace_id='w' AND assignment_id='assignment' + )", + [], + )?; + if replaced != 1 { + return Err(StoreError::InvalidInput( + "fault injection requires one ended current assignment".into(), + )); + } + tx.commit()?; + Ok(()) + }) + .unwrap(); + let result = WorkerRetentionExecutionResult { + operation_id: plan.operation_id.clone(), + input_fingerprint: plan.input_fingerprint.clone(), + expected_worker_revision: plan.worker_revision.clone(), + worker_id: worker_id(), + session_disposition: plan.session_disposition, + diagnostics_disposition: plan.diagnostics_disposition, + archive: None, + source_removed: true, + diagnostics_retained: false, + }; + let error = store + .commit_worker_removal("w", &plan.operation_id, &plan.input_fingerprint, &result) + .unwrap_err(); + assert!( + matches!(error, WorkerRetentionError::StalePlan { reason, .. } + if reason == "unfinished work added") + ); + assert!( + store + .get_worker_registry("w", &plan.worker) + .unwrap() + .is_some() + ); + assert!(store.worker_tombstone("w", &plan.worker).unwrap().is_none()); + let (state, active): (String, String) = store + .with_conn(|conn| { + Ok(( + conn.query_row( + "SELECT state FROM worker_removal_operations WHERE operation_id=?1", + [&plan.operation_id], + |row| row.get(0), + )?, + conn.query_row( + "SELECT assignment_id FROM ticket_active_worker_assignments + WHERE workspace_id='w' AND ticket_id='ticket'", + [], + |row| row.get(0), + )?, + )) + }) + .unwrap(); + assert_eq!( + state, "executing", + "rejected commit must remain recoverable" + ); + assert_eq!( + active, "new-assignment", + "the ended identity must not reactivate" + ); + } + + #[test] + fn terminal_assignment_does_not_bypass_hold_added_before_commit() { + let store = setup(); + seed_current_assignment(&store); + store + .with_conn(|conn| { + conn.execute( + "UPDATE typed_tickets SET workflow_state='done' + WHERE workspace_id='w' AND ticket_id='ticket'", + [], + )?; + Ok(()) + }) + .unwrap(); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + store + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + store + .with_conn(|conn| { + conn.execute( + "UPDATE worker_registry SET retention_state='pinned' + WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1", + [worker_id().to_string()], + )?; + Ok(()) + }) + .unwrap(); + let result = WorkerRetentionExecutionResult { + operation_id: plan.operation_id.clone(), + input_fingerprint: plan.input_fingerprint.clone(), + expected_worker_revision: plan.worker_revision.clone(), + worker_id: worker_id(), + session_disposition: plan.session_disposition, + diagnostics_disposition: plan.diagnostics_disposition, + archive: None, + source_removed: true, + diagnostics_retained: false, + }; + let error = store + .commit_worker_removal("w", &plan.operation_id, &plan.input_fingerprint, &result) + .unwrap_err(); + assert!( + matches!(error, WorkerRetentionError::StalePlan { reason, .. } + if reason == "hold added") + ); + assert!( + store + .get_worker_registry("w", &plan.worker) + .unwrap() + .is_some() + ); + } + #[test] fn concurrent_plan_converges_and_purge_omits_tombstone() { let s = std::sync::Arc::new(setup()); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 445a7566d..2e6b64acb 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -5697,7 +5697,7 @@ fn ticket_target_capabilities_for_workdir( } let Some(assignment) = api .store - .get_current_ticket_role_assignment_for_worker(&api.config.workspace_id, worker)? + .get_active_ticket_role_assignment_for_worker(&api.config.workspace_id, worker)? .filter(|assignment| assignment.role == TicketAssignmentRole::Coder) else { return Ok(source_capabilities); @@ -13177,7 +13177,7 @@ async fn scoped_set_ticket_assignment( } if api .store - .get_current_ticket_role_assignment( + .get_active_ticket_role_assignment( &workspace_id, &ticket.id, TicketAssignmentRole::Coder, @@ -13310,7 +13310,7 @@ async fn scoped_cancel_ticket_implementation( .into()); } - let current = api.store.get_current_ticket_role_assignment( + let current = api.store.get_active_ticket_role_assignment( &path.workspace_id, &ticket.id, TicketAssignmentRole::Coder, @@ -13479,7 +13479,7 @@ fn validate_ticket_assignment_spawn( if let Some(current) = api .store - .get_current_ticket_coder_assignment(&api.config.workspace_id, &assignment.ticket_id)? + .get_active_ticket_worker_assignment(&api.config.workspace_id, &assignment.ticket_id)? { let replay_matches = api .store @@ -13563,7 +13563,7 @@ fn assign_ticket_worker_from_lifecycle( if operation.action == "assign" && operation.ticket_id == assignment.ticket_id && operation.worker.as_ref() == Some(&worker) - && let Some(current) = api.store.get_current_ticket_coder_assignment( + && let Some(current) = api.store.get_active_ticket_worker_assignment( &api.config.workspace_id, &assignment.ticket_id, )? @@ -14137,7 +14137,7 @@ async fn execute_ticket_rest_operation( for ticket_id in candidates { if api .store - .get_current_ticket_role_assignment( + .get_active_ticket_role_assignment( workspace_id, &ticket_id, TicketAssignmentRole::Coder, @@ -14626,7 +14626,7 @@ impl merge_request::AssignmentSource for MergeRequestAssignmentSource { ticket_id: &str, ) -> std::result::Result, String> { self.store - .get_current_ticket_coder_assignment(workspace_id, ticket_id) + .get_active_ticket_worker_assignment(workspace_id, ticket_id) .map(|value| { value.map(|assignment| merge_request::CurrentAssignment { assignment_id: assignment.assignment_id, @@ -15328,7 +15328,7 @@ async fn scoped_list_merge_requests( ) { (Some(selector), Some(ticket_id)) => match api .store - .get_current_ticket_coder_assignment(&workspace_id, ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, ticket_id)? { Some(assignment) => match observe_published_source_ref( &api, @@ -15452,7 +15452,7 @@ async fn scoped_show_merge_request( })?; let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, ticket_id)?; + .get_active_ticket_worker_assignment(&workspace_id, ticket_id)?; let source = match (mr.selector_from.as_deref(), assignment.as_ref()) { (Some(selector), Some(assignment)) => { match observe_published_source_ref( @@ -15526,7 +15526,7 @@ async fn scoped_merge_request_readiness( let ticket_id = merge_request_ticket_id(&mr)?; let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)?; + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)?; let (current_subject_ref, source_blocker) = match (mr.selector_from.as_deref(), assignment) { (Some(selector), Some(assignment)) => match observe_published_source_ref( &api, @@ -15586,9 +15586,9 @@ async fn scoped_open_merge_request( let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?; let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? .ok_or_else(|| { - Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into()) + Error::TicketAssignmentConflict("Ticket has no active Coder work assignment".into()) })?; if assignment.worker.runtime_id != source.runtime_id || assignment.worker.worker_id != source.worker_id @@ -15685,9 +15685,9 @@ async fn scoped_repair_merge_request_selector( let ticket_id = merge_request_ticket_id(&mr)?; let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? .ok_or_else(|| { - Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into()) + Error::TicketAssignmentConflict("Ticket has no active Coder work assignment".into()) })?; let resolved_subject_ref = observe_published_source_ref( &api, @@ -15757,9 +15757,9 @@ async fn scoped_register_merge_request_review_capability( } let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? .ok_or_else(|| { - Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into()) + Error::TicketAssignmentConflict("Ticket has no active Coder work assignment".into()) })?; if assignment.worker.runtime_id != source.runtime_id || assignment.worker.worker_id != source.worker_id @@ -15867,9 +15867,9 @@ async fn scoped_submit_merge_request_review( let mr = store.get_by_id(&workspace_id, &merge_request_id)?; let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? .ok_or_else(|| { - Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into()) + Error::TicketAssignmentConflict("Ticket has no active Coder work assignment".into()) })?; let current_subject_ref = match mr.state { merge_request::MergeRequestState::Merged => merged_result_source_ref(&mr)?, @@ -15929,9 +15929,9 @@ async fn scoped_revoke_merge_request_review( let ticket_id = merge_request_ticket_id(&mr)?; let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? .ok_or_else(|| { - Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into()) + Error::TicketAssignmentConflict("Ticket has no active Coder work assignment".into()) })?; if assignment.worker.runtime_id != source.runtime_id || assignment.worker.worker_id != source.worker_id @@ -15999,9 +15999,9 @@ async fn scoped_complete_merge_request( } let assignment = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? .ok_or_else(|| { - Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into()) + Error::TicketAssignmentConflict("Ticket has no active Coder work assignment".into()) })?; let selector = mr .selector_from @@ -16065,9 +16065,12 @@ async fn scoped_complete_ticket( require_workspace_access(&workspace_id, &api)?; let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?; require_online_workspace_orchestrator_source(&api, &source)?; + // No active assignment is expected after completion. The store checks an + // exact recorded-operation replay before live authority, including its actor + // and evidence fingerprint; an empty identity cannot authorize new completion. let assignment_id = api .store - .get_current_ticket_coder_assignment(&workspace_id, &ticket_id)? + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? .map(|assignment| assignment.assignment_id) .unwrap_or_default(); let event = merge_request_store(&api, &workspace_id)?.complete_ticket( @@ -16409,7 +16412,7 @@ fn active_orchestrator_assignment( workspace_id: &str, ticket_id: &str, ) -> Result> { - let assignment = api.store.get_current_ticket_role_assignment( + let assignment = api.store.get_active_ticket_role_assignment( workspace_id, ticket_id, TicketAssignmentRole::Orchestrator, @@ -16452,7 +16455,7 @@ fn worker_ticket_source_context( ) -> WorkerTicketSourceContext { let assignment = ticket.and_then(|ticket| { api.store - .get_current_ticket_coder_assignment(workspace_id, &ticket.meta.id) + .get_active_ticket_worker_assignment(workspace_id, &ticket.meta.id) .ok() .flatten() }); @@ -16595,7 +16598,7 @@ fn notify_ticket_recipients( let mut recipients = Vec::new(); if let Some(assignment) = api .store - .get_current_ticket_coder_assignment(workspace_id, ticket_id) + .get_active_ticket_worker_assignment(workspace_id, ticket_id) .ok() .flatten() { @@ -21178,7 +21181,7 @@ fn worker_retention_error_response( crate::retention::WorkerRetentionError::Blocked(_) => worker_remove_error_response( StatusCode::CONFLICT, "worker_removal_blocked", - "Worker removal is blocked by current assignment, hold, or retention policy", + "Worker removal is blocked by unfinished Ticket work, hold, or retention policy", ), crate::retention::WorkerRetentionError::Invalid(_) => worker_remove_error_response( StatusCode::BAD_REQUEST, @@ -24300,15 +24303,15 @@ fn build_runtime_cleanup_plan( let links = api .store .list_worker_workdir_links(&api.config.workspace_id, &record.worker)?; - let current_assignment = api.store.get_current_ticket_role_assignment_for_worker( + let active_assignment = api.store.get_active_ticket_role_assignment_for_worker( &api.config.workspace_id, &record.worker, )?; let is_running = live_running_worker_ids.contains(&record.worker); let pinned = record.retention_state == "pinned"; - let blocking_reason = if let Some(assignment) = current_assignment { + let blocking_reason = if let Some(assignment) = active_assignment { Some(format!( - "worker has current Ticket assignment `{}` (`{}`)", + "worker has unfinished work for Ticket `{}` (`{}`)", assignment.ticket_id, assignment.role.as_str() )) @@ -24524,13 +24527,13 @@ async fn execute_runtime_cleanup_with_context( ); if let Some(assignment) = api .store - .get_current_ticket_role_assignment_for_worker(&api.config.workspace_id, &worker)? + .get_active_ticket_role_assignment_for_worker(&api.config.workspace_id, &worker)? { return Err(cleanup_api_error( runtime_id, "workspace_cleanup_worker_assigned", &format!( - "Worker is assigned to Ticket `{}` as `{}` and cannot be deleted", + "Worker has unfinished work for Ticket `{}` as `{}` and cannot be deleted", assignment.ticket_id, assignment.role.as_str() ), @@ -46949,6 +46952,444 @@ mod tests { } } + async fn seed_live_test_coder_assignment( + api: &WorkspaceApi, + ticket_id: &str, + assignment_id: &str, + ) -> RuntimeWorkerRef { + let Json(created) = create_workspace_worker( + State(api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + display_name: "Assignment lifecycle Coder".to_string(), + singleton_key: None, + profile: Some("builtin:coder".to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachments: Vec::new(), + feature_connections: Default::default(), + control_operation_id: None, + }), + ) + .await + .unwrap(); + let worker = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + api.store + .set_current_ticket_coder_assignment( + &TicketCoderAssignmentRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + ticket_id: ticket_id.to_string(), + assignment_id: assignment_id.to_string(), + worker: worker.clone(), + assigned_by: "test".to_string(), + assigned_at: TEST_CREATED_AT.to_string(), + }, + None, + &format!("event-{assignment_id}"), + &format!("operation-{assignment_id}"), + false, + ) + .unwrap(); + worker + } + + #[tokio::test] + async fn complete_ticket_api_replays_done_operation_without_reviving_assignment_authority() { + let dir = tempfile::tempdir().unwrap(); + let api = test_api(dir.path()).await; + let backend = browser_ticket_backend(&api).unwrap(); + let mut input = ticket::NewTicket::new("Completed operation replay"); + input.workflow_state = Some(TicketWorkflowState::InProgress); + let ticket = backend.create(input).unwrap(); + let coder = seed_live_test_coder_assignment(&api, &ticket.id, "completed-coder").await; + let Json(started) = scoped_start_workspace_orchestrator( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + ) + .await + .unwrap(); + let orchestrator = started.worker.unwrap(); + let orchestrator = RuntimeWorkerRef::new(orchestrator.runtime_id, orchestrator.worker_id); + let recorded = merge_request::TicketCompletionEvent { + operation_id: "recorded-ticket-completion".to_string(), + ticket_id: ticket.id.clone(), + item_revision: api.authority.ticket(&ticket.id).unwrap().item_revision, + merge_request_ids: vec!["merged-result".to_string()], + requirement_approval_event_id: "requirement-approval".to_string(), + completed_by: merge_request::WorkerIdentity { + runtime_id: orchestrator.runtime_id.clone(), + worker_id: orchestrator.worker_id.clone(), + }, + created_at: Utc::now(), + }; + // Seed the persisted completion boundary, not the review/integration flow: + // this test proves API replay authorization against an already-recorded result. + let conn = rusqlite::Connection::open(&api.config.database_path).unwrap(); + crate::store::configure_sqlite(&conn).unwrap(); + conn.execute( + "UPDATE typed_tickets SET workflow_state = 'done' + WHERE workspace_id = ?1 AND ticket_id = ?2", + rusqlite::params![TEST_WORKSPACE_ID, ticket.id], + ) + .unwrap(); + backend + .with_event_attributes(BTreeMap::from([ + ("operation_id".to_string(), recorded.operation_id.clone()), + ( + "ticket_completion".to_string(), + serde_json::to_string(&recorded).unwrap(), + ), + ])) + .add_event( + TicketIdOrSlug::Id(ticket.id.clone()), + NewTicketEvent::new(TicketEventKind::StateChanged, "Recorded completion"), + ) + .unwrap(); + let mut former_actor_record = recorded.clone(); + former_actor_record.operation_id = "former-actor-completion".to_string(); + former_actor_record.completed_by = merge_request::WorkerIdentity { + runtime_id: coder.runtime_id.clone(), + worker_id: coder.worker_id.clone(), + }; + browser_ticket_backend(&api) + .unwrap() + .with_event_attributes(BTreeMap::from([ + ( + "operation_id".to_string(), + former_actor_record.operation_id.clone(), + ), + ( + "ticket_completion".to_string(), + serde_json::to_string(&former_actor_record).unwrap(), + ), + ])) + .add_event( + TicketIdOrSlug::Id(ticket.id.clone()), + NewTicketEvent::new(TicketEventKind::StateChanged, "Former actor completion"), + ) + .unwrap(); + let event_count = browser_ticket_backend(&api) + .unwrap() + .show(TicketIdOrSlug::Id(ticket.id.clone())) + .unwrap() + .events + .len(); + let request = || server_api::CompleteTicketRequest { + operation_id: recorded.operation_id.clone(), + item_revision: recorded.item_revision.clone(), + merge_request_ids: recorded.merge_request_ids.clone(), + requirement_approval_event_id: recorded.requirement_approval_event_id.clone(), + }; + let Json(replayed) = scoped_complete_ticket( + State(api.clone()), + ticket_check_headers(&orchestrator), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + Json(request()), + ) + .await + .unwrap(); + assert_eq!( + serde_json::to_value(replayed).unwrap(), + serde_json::to_value(public_ticket_completion_event(recorded.clone())).unwrap() + ); + for field in ["operation", "revision", "mr_set", "approval", "actor"] { + let mut changed = request(); + match field { + "operation" => changed.operation_id = "new-completion-operation".to_string(), + "revision" => changed.item_revision = "different-revision".to_string(), + "mr_set" => changed.merge_request_ids = vec!["different-result".to_string()], + "approval" => { + changed.requirement_approval_event_id = "different-approval".to_string(); + } + "actor" => changed.operation_id = former_actor_record.operation_id.clone(), + _ => unreachable!(), + } + let response = scoped_complete_ticket( + State(api.clone()), + ticket_check_headers(&orchestrator), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + Json(changed), + ) + .await + .unwrap_err() + .into_response(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8_lossy(&bytes); + assert_eq!(status, StatusCode::CONFLICT, "field={field}: {body}"); + let reason = if field == "operation" { + "Ticket must be inprogress" + } else { + "Ticket completion operation fingerprint mismatch" + }; + assert!(body.contains(reason), "field={field}: {body}"); + } + assert!( + api.store + .get_active_ticket_role_assignment_for_worker(TEST_WORKSPACE_ID, &coder) + .unwrap() + .is_none() + ); + assert_eq!( + api.store + .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &ticket.id) + .unwrap() + .unwrap() + .assignment_id, + "completed-coder" + ); + let after = browser_ticket_backend(&api) + .unwrap() + .show(TicketIdOrSlug::Id(ticket.id)) + .unwrap(); + assert_eq!(after.meta.workflow_state, TicketWorkflowState::Done); + assert_eq!(after.events.len(), event_count); + } + + #[tokio::test] + async fn close_and_reopen_api_preserves_responsibility_but_requires_fresh_routing_identity() { + let dir = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(dir.path()).await; + let mut input = ticket::NewTicket::new("Close and reopen responsibility"); + input.workflow_state = Some(TicketWorkflowState::InProgress); + set_test_ticket_target(&mut input, "test-repository", "develop"); + let ticket = browser_ticket_backend(&api).unwrap().create(input).unwrap(); + let old_coder = seed_live_test_coder_assignment(&api, &ticket.id, "old-coder").await; + assign_test_orchestrator(&api, &ticket.id); + let old_orchestrator_id = format!("orchestrator-{}", ticket.id); + let Json(started) = scoped_start_workspace_orchestrator( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + ) + .await + .unwrap(); + let orchestrator = started.worker.unwrap(); + let orchestrator = RuntimeWorkerRef::new(orchestrator.runtime_id, orchestrator.worker_id); + execution.take_inputs(); + let status = scoped_close_ticket_record( + State(api.clone()), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + HeaderMap::new(), + Json(MarkdownText::new("Normal close ends implementation work")), + ) + .await + .unwrap(); + assert_eq!(status, StatusCode::NO_CONTENT); + assert!(execution.take_inputs().is_empty()); + + for state in ["closed", "planning"] { + if state == "planning" { + // No public reopen operation exists. Seed a reopened authority + // record after normal API close; the durable release must survive. + let conn = rusqlite::Connection::open(&api.config.database_path).unwrap(); + crate::store::configure_sqlite(&conn).unwrap(); + conn.execute( + "UPDATE typed_tickets SET workflow_state = 'planning', status = 'open' + WHERE workspace_id = ?1 AND ticket_id = ?2", + rusqlite::params![TEST_WORKSPACE_ID, ticket.id], + ) + .unwrap(); + } + let path = || { + AxumPath(ScopedRecordPath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + id: ticket.id.clone(), + }) + }; + let Json(detail) = scoped_get_ticket(State(api.clone()), path()).await.unwrap(); + assert_eq!(detail.state, state); + assert_eq!(detail.current_coder.unwrap().assignment_id, "old-coder"); + assert!( + detail + .assignments + .iter() + .any(|assignment| assignment.assignment_id == old_orchestrator_id) + ); + assert_eq!( + detail.action_eligibility.can_assign_orchestrator, + state == "planning" + ); + let Json(assignments) = scoped_list_ticket_assignments(State(api.clone()), path()) + .await + .unwrap(); + assert!( + assignments + .assignments + .iter() + .any(|assignment| assignment.assignment_id == "old-coder") + ); + assert!( + api.store + .list_active_ticket_role_assignments(TEST_WORKSPACE_ID, &ticket.id) + .unwrap() + .is_empty(), + "state={state}" + ); + let status = scoped_add_ticket_thread_event( + State(api.clone()), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + HeaderMap::new(), + Json(NewTicketEvent::new( + TicketEventKind::Comment, + "Browser activity", + )), + ) + .await + .unwrap(); + assert_eq!(status, StatusCode::NO_CONTENT); + assert!(execution.take_inputs().is_empty(), "state={state}"); + scoped_add_ticket_thread_event( + State(api.clone()), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + ticket_check_headers(&old_coder), + Json(NewTicketEvent::new( + TicketEventKind::Comment, + "Former Coder activity is ordinary Worker activity", + )), + ) + .await + .unwrap(); + let activity = browser_ticket_backend(&api) + .unwrap() + .show(TicketIdOrSlug::Id(ticket.id.clone())) + .unwrap(); + let attributes = &activity.events.last().unwrap().attributes; + assert_eq!( + attributes.get("source_actor_role").map(String::as_str), + Some("worker"), + "state={state}" + ); + assert!( + !attributes.contains_key("source_assignment_id"), + "state={state}" + ); + assert!(execution.take_inputs().is_empty(), "state={state}"); + let response = scoped_open_merge_request( + State(api.clone()), + ticket_check_headers(&old_coder), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + Json(server_api::OpenMergeRequestRequest { + repository_key: "test-repository".to_string(), + selector_from: "work/old-coder".to_string(), + selector_to: "develop".to_string(), + summary: "Old responsibility is not implementation authority".to_string(), + }), + ) + .await + .unwrap_err() + .into_response(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8_lossy(&bytes); + assert_eq!(status, StatusCode::CONFLICT, "state={state}: {body}"); + assert!( + body.contains("Ticket has no active Coder work assignment"), + "state={state}: {body}" + ); + } + execute_ticket_rest_operation( + &api, + TEST_WORKSPACE_ID, + HeaderMap::new(), + TicketBackendOperation::MarkReady { + id: TicketIdOrSlug::Id(ticket.id.clone()), + request: ticket::TicketMarkReady { + operation_key: "reopened-ready".to_string(), + reason: Some("Fresh implementation cycle".to_string()), + author: None, + intake_summary: None, + }, + }, + ) + .await + .unwrap(); + let response = scoped_queue_ticket_record( + State(api.clone()), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + HeaderMap::new(), + ) + .await + .unwrap_err() + .into_response(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8_lossy(&bytes); + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert!( + body.contains("Queue requires role=orchestrator assignment"), + "{body}" + ); + assert!(execution.take_inputs().is_empty()); + let Json(assigned) = scoped_set_ticket_assignment( + State(api.clone()), + AxumPath(( + TEST_WORKSPACE_ID.to_string(), + ticket.id.clone(), + "orchestrator".to_string(), + )), + Json(server_api::SetTicketRoleAssignmentRequest { + operation_id: "fresh-orchestrator-cycle".to_string(), + principal: server_api::TicketAssignmentPrincipal::WorkspaceAgent { + agent_key: "workspace-orchestrator".to_string(), + }, + expected_assignment_id: None, + }), + ) + .await + .unwrap(); + let fresh_id = assigned.assignment.unwrap().assignment_id; + assert_ne!(fresh_id, old_orchestrator_id); + let Json(detail) = scoped_get_ticket( + State(api.clone()), + AxumPath(ScopedRecordPath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + id: ticket.id.clone(), + }), + ) + .await + .unwrap(); + assert!(detail.action_eligibility.can_queue); + assert_eq!(detail.current_coder.unwrap().assignment_id, "old-coder"); + let _ = scoped_queue_ticket_record( + State(api.clone()), + AxumPath((TEST_WORKSPACE_ID.to_string(), ticket.id.clone())), + HeaderMap::new(), + ) + .await + .unwrap(); + let queued = browser_ticket_backend(&api) + .unwrap() + .show(TicketIdOrSlug::Id(ticket.id.clone())) + .unwrap(); + assert_eq!(queued.meta.workflow_state, TicketWorkflowState::Queued); + assert_eq!( + queued + .events + .iter() + .rev() + .find_map(|event| event.attributes.get("orchestrator_assignment_id")), + Some(&fresh_id) + ); + let inputs = execution.take_inputs(); + assert_eq!(inputs.len(), 1); + assert_eq!(inputs[0].0.worker_id.to_string(), orchestrator.worker_id); + assert_eq!( + inputs[0].1, + ticket_notification_content(ticket.resource_key.as_deref().unwrap(), "ready", "queued") + ); + assert!( + api.store + .get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket.id) + .unwrap() + .is_none() + ); + } + struct ManualCoderAssignmentFixture { _workspace: tempfile::TempDir, api: WorkspaceApi, @@ -53701,7 +54142,7 @@ mod tests { } #[tokio::test] - async fn cleanup_blocks_assigned_worker_before_runtime_deletion() { + async fn cleanup_blocks_unfinished_work_before_runtime_deletion() { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let api = test_api(workspace.path()).await; @@ -53717,7 +54158,7 @@ mod tests { .unwrap(); assert_eq!( candidate.blocking_reason.as_deref(), - Some("worker has current Ticket assignment `ticket-assigned` (`coder`)") + Some("worker has unfinished work for Ticket `ticket-assigned` (`coder`)") ); let request = ExecuteRuntimeCleanupRequest { expected_plan_revision: plan.revision.clone(), @@ -53750,6 +54191,93 @@ mod tests { ); } + #[tokio::test] + async fn cleanup_preview_ignores_terminal_ticket_responsibility() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let worker_id = seed_cleanup_worker(&api, 3, "normal"); + let ticket_id = "terminal-responsibility"; + seed_cleanup_worker_assignment(&api, &worker_id, ticket_id); + let worker = RuntimeWorkerRef::new("runtime-test", &worker_id); + let conn = rusqlite::Connection::open(&api.config.database_path).unwrap(); + crate::store::configure_sqlite(&conn).unwrap(); + + for (state, status) in [("done", "open"), ("closed", "closed")] { + conn.execute( + "UPDATE typed_tickets SET workflow_state = ?1, status = ?2 + WHERE workspace_id = ?3 AND ticket_id = ?4", + rusqlite::params![state, status, api.config.workspace_id, ticket_id], + ) + .unwrap(); + assert!( + api.store + .get_current_ticket_role_assignment_for_worker( + &api.config.workspace_id, + &worker, + ) + .unwrap() + .is_some(), + "terminal responsibility must remain visible: {state}" + ); + let plan = build_runtime_cleanup_plan(&api, "runtime-test") + .unwrap_or_else(|err| panic!("cleanup plan: {}", err.error)); + let candidate = plan + .workers + .iter() + .find(|candidate| candidate.worker_id == worker_id) + .unwrap(); + assert_eq!(candidate.blocking_reason, None, "state={state}"); + } + } + + #[tokio::test] + async fn merge_request_assignment_authority_rejects_terminal_ticket_responsibility() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let worker_id = seed_cleanup_worker(&api, 3, "normal"); + let ticket_id = "terminal-mr-responsibility"; + seed_cleanup_worker_assignment(&api, &worker_id, ticket_id); + let source = MergeRequestAssignmentSource { + store: api.store.clone(), + }; + assert!( + merge_request::AssignmentSource::current_assignment( + &source, + &api.config.workspace_id, + ticket_id, + ) + .unwrap() + .is_some() + ); + let conn = rusqlite::Connection::open(&api.config.database_path).unwrap(); + crate::store::configure_sqlite(&conn).unwrap(); + for (state, status) in [("done", "open"), ("closed", "closed")] { + conn.execute( + "UPDATE typed_tickets SET workflow_state = ?1, status = ?2 + WHERE workspace_id = ?3 AND ticket_id = ?4", + rusqlite::params![state, status, api.config.workspace_id, ticket_id], + ) + .unwrap(); + assert!( + api.store + .get_current_ticket_coder_assignment(&api.config.workspace_id, ticket_id) + .unwrap() + .is_some(), + "terminal responsibility must remain visible: {state}" + ); + assert!( + merge_request::AssignmentSource::current_assignment( + &source, + &api.config.workspace_id, + ticket_id, + ) + .unwrap() + .is_none(), + "terminal responsibility must not authorize MR operations: {state}" + ); + } + } + #[tokio::test] async fn cleanup_execution_retains_dirty_and_requires_fresh_provider_not_found() { let workspace = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 13d19cb7f..f9bd8e96c 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -28,7 +28,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 81; +const LATEST_SCHEMA_VERSION: i64 = 82; const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs"; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; @@ -283,6 +283,11 @@ const MIGRATIONS: &[Migration] = &[ name: "Backend Job immutable resource serialization", apply: migrate_backend_job_resources_v80_to_v81, }, + Migration { + version: 82, + name: "Ticket responsibility and unfinished work separation", + apply: migrate_assignment_work_v81_to_v82, + }, ]; #[derive(Clone, Copy)] @@ -1853,6 +1858,26 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { workspace_id: &str, operation_id: &str, ) -> Result<()>; + /// Operational authority/occupancy only. Terminal state releases these + /// identities durably; reopening does not reactivate retained responsibility. + fn list_active_ticket_role_assignments( + &self, + workspace_id: &str, + ticket_id: &str, + ) -> Result>; + fn get_active_ticket_role_assignment_for_worker( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result>; + fn get_active_ticket_role_assignment( + &self, + workspace_id: &str, + ticket_id: &str, + role: TicketAssignmentRole, + ) -> Result>; + /// Retained responsibility, including ended work and removed principals. + /// Use the active variants for admission, routing, authorization or removal. fn list_current_ticket_role_assignments( &self, workspace_id: &str, @@ -1907,6 +1932,11 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { occurred_at: &str, reason: &str, ) -> Result; + fn get_active_ticket_worker_assignment( + &self, + workspace_id: &str, + ticket_id: &str, + ) -> Result>; fn get_current_ticket_coder_assignment( &self, workspace_id: &str, @@ -8026,6 +8056,69 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn list_active_ticket_role_assignments( + &self, + workspace_id: &str, + ticket_id: &str, + ) -> Result> { + self.with_conn(|conn| { + let sql = active_ticket_role_assignment_select_sql( + "WHERE current.workspace_id = ?1 AND current.ticket_id = ?2 \ + ORDER BY current.role, a.assigned_at, a.assignment_id", + ); + let mut stmt = conn.prepare(&sql)?; + let rows = stmt.query_map( + params![workspace_id, ticket_id], + read_ticket_role_assignment_record, + )?; + rows.collect::>>() + .map_err(Into::into) + }) + } + + fn get_active_ticket_role_assignment( + &self, + workspace_id: &str, + ticket_id: &str, + role: TicketAssignmentRole, + ) -> Result> { + self.with_conn(|conn| { + let sql = active_ticket_role_assignment_select_sql( + "WHERE current.workspace_id = ?1 AND current.ticket_id = ?2 AND current.role = ?3 \ + ORDER BY a.assigned_at, a.assignment_id LIMIT 1", + ); + Ok(conn + .query_row( + &sql, + params![workspace_id, ticket_id, role.as_str()], + read_ticket_role_assignment_record, + ) + .optional()?) + }) + } + + fn get_active_ticket_role_assignment_for_worker( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result> { + self.with_conn(|conn| { + let sql = active_ticket_role_assignment_select_sql( + "WHERE current.workspace_id = ?1 \ + AND current.principal_kind = 'worker' \ + AND current.runtime_id = ?2 AND current.worker_id = ?3 \ + ORDER BY a.assigned_at, a.assignment_id LIMIT 1", + ); + Ok(conn + .query_row( + &sql, + params![workspace_id, worker.runtime_id, worker.worker_id], + read_ticket_role_assignment_record, + ) + .optional()?) + }) + } + fn list_current_ticket_role_assignments( &self, workspace_id: &str, @@ -8203,12 +8296,15 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } Some(expected.clone()) } else if record.role.is_singleton() && !current.is_empty() { - return Err(Error::TicketAssignmentConflict(format!( - "Ticket `{}` already has an active `{}` assignment `{}`", - record.ticket_id, - record.role.as_str(), - current[0].assignment_id - ))); + let active: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id=?1 AND assignment_id=?2)", + params![record.workspace_id, current[0].assignment_id], |row| row.get(0))?; + if active { + return Err(Error::TicketAssignmentConflict(format!( + "Ticket {} already has unfinished {} work; use explicit reassign", + record.ticket_id, record.role.as_str()))); + } + Some(current[0].clone()) } else { None }; @@ -8434,7 +8530,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { )); } let conflicting: i64 = tx.query_row( - "SELECT COUNT(*) FROM ticket_current_worker_assignments + "SELECT COUNT(*) FROM ticket_active_worker_assignments WHERE workspace_id = ?1 AND ticket_id = ?2 AND role IN ('orchestrator', 'coder')", params![record.workspace_id, record.ticket_id], @@ -8447,6 +8543,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore { )); } + let previous: Option = tx.query_row( + "SELECT assignment_id FROM ticket_current_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND role='coder'", + params![record.workspace_id, record.ticket_id], |row| row.get(0)).optional()?; + tx.execute("DELETE FROM ticket_current_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND role='coder'", + params![record.workspace_id, record.ticket_id])?; tx.execute( "INSERT INTO ticket_assignment_operations ( workspace_id, operation_id, action, ticket_id, role, principal_kind, @@ -8478,10 +8579,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore { tx.execute( "INSERT INTO ticket_worker_assignment_events ( workspace_id, ticket_id, role, event_id, action, assignment_id, - actor, created_at, operation_id - ) VALUES (?1, ?2, 'coder', ?3, 'assigned', ?4, ?5, ?6, ?7)", + actor, created_at, operation_id, previous_assignment_id + ) VALUES (?1, ?2, 'coder', ?3, ?8, ?4, ?5, ?6, ?7, ?9)", params![record.workspace_id, record.ticket_id, event_id, record.assignment_id, - record.assigned_by, record.assigned_at, operation_id], + record.assigned_by, record.assigned_at, operation_id, + if previous.is_some() { "reassigned" } else { "assigned" }, previous], )?; let event_index: i64 = tx.query_row( "SELECT COALESCE(MAX(event_index), -1) + 1 FROM typed_ticket_events @@ -8649,6 +8751,27 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn get_active_ticket_worker_assignment( + &self, + workspace_id: &str, + ticket_id: &str, + ) -> Result> { + self.with_conn(|conn| { + conn.query_row( + current_ticket_worker_assignment_select_sql() + .replace( + "ticket_current_worker_assignments", + "ticket_active_worker_assignments", + ) + .as_str(), + params![workspace_id, ticket_id], + read_ticket_worker_assignment_record, + ) + .optional() + .map_err(Error::from) + }) + } + fn get_current_ticket_coder_assignment( &self, workspace_id: &str, @@ -8753,7 +8876,10 @@ impl ControlPlaneStore for SqliteWorkspaceStore { read_ticket_worker_assignment_record, ) .optional()?; - if previous.is_some() && !allow_reassign { + let previous_active: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND role='coder')", + params![record.workspace_id, record.ticket_id], |row| row.get(0))?; + if previous_active && !allow_reassign { return Err(Error::TicketAssignmentConflict(format!( "Ticket {} is already assigned; use the explicit reassign operation", record.ticket_id @@ -8790,7 +8916,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { record.assigned_at, ], )?; - let current_write = if allow_reassign { + let current_write = if previous.is_some() { tx.execute( r#"UPDATE ticket_current_worker_assignments SET assignment_id = ?3, runtime_id = ?4, worker_id = ?5, updated_at = ?6 @@ -10633,7 +10759,7 @@ fn runtime_removal_preflight( ), ( "active_worker_assignment", - "SELECT EXISTS(SELECT 1 FROM ticket_current_worker_assignments WHERE runtime_id = ?1)", + "SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments WHERE runtime_id = ?1)", ), ( "active_workdir", @@ -11743,6 +11869,13 @@ fn read_ticket_role_assignment_by_id( .optional()?) } +fn active_ticket_role_assignment_select_sql(where_clause: &str) -> String { + ticket_role_assignment_select_sql(where_clause).replace( + "ticket_current_worker_assignments", + "ticket_active_worker_assignments", + ) +} + fn ticket_role_assignment_select_sql(where_clause: &str) -> String { format!( "SELECT a.workspace_id, a.ticket_id, a.assignment_id, a.role, \ @@ -11982,7 +12115,7 @@ fn read_assignment_operation( fn map_assignment_constraint(error: rusqlite::Error, ticket_id: &str, worker_id: &str) -> Error { if matches!(error, rusqlite::Error::SqliteFailure(_, _)) { Error::TicketAssignmentConflict(format!( - "Ticket {ticket_id} or Worker {worker_id} already has a current assignment" + "Ticket {ticket_id} has a responsibility conflict or Worker {worker_id} has unfinished work" )) } else { Error::Sqlite(error) @@ -15128,6 +15261,99 @@ fn migrate_archive_observe_grants_v75_to_v76(conn: &Connection) -> Result<()> { Ok(()) } +fn migrate_assignment_work_v81_to_v82(conn: &Connection) -> Result<()> { + if current_schema_version(conn)? != 81 { + return Err(Error::Store( + "Ticket work separation requires schema 81".into(), + )); + } + let foreign_keys = + conn.pragma_query_value(None, "foreign_keys", |row| row.get::<_, bool>(0))?; + let legacy = + conn.pragma_query_value(None, "legacy_alter_table", |row| row.get::<_, bool>(0))?; + conn.execute_batch("PRAGMA foreign_keys = OFF; PRAGMA legacy_alter_table = ON;")?; + let result = (|| { + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + let dependents = { + let mut stmt = tx.prepare("SELECT sql FROM sqlite_schema WHERE tbl_name='ticket_current_worker_assignments' AND type IN ('index','trigger') AND sql IS NOT NULL AND name != 'ticket_current_worker_role_idx'")?; + let rows = stmt.query_map([], |row| row.get::<_, String>(0))?; + rows.collect::>>()? + }; + let schema = include_str!("latest_schema.sql"); + let start = schema + .find("CREATE TABLE ticket_current_worker_assignments (") + .ok_or_else(|| Error::Store("Missing assignment schema".into()))?; + let end = schema[start..] + .find("CREATE TABLE ticket_worker_assignment_events") + .ok_or_else(|| Error::Store("Missing assignment schema end".into()))? + + start; + tx.execute_batch(&schema[start..end].replace( + "CREATE TABLE ticket_current_worker_assignments", + "CREATE TABLE ticket_current_worker_assignments_v82", + ))?; + tx.execute_batch("INSERT INTO ticket_current_worker_assignments_v82 SELECT * FROM ticket_current_worker_assignments; + DROP TABLE ticket_current_worker_assignments; + ALTER TABLE ticket_current_worker_assignments_v82 RENAME TO ticket_current_worker_assignments;")?; + for sql in dependents { + tx.execute_batch(&sql)?; + } + tx.execute_batch(include_str!("assignment_work.sql"))?; + // Existing done/closed + current-assignment data is preserved, and a + // durable release prevents a later reopen from resurrecting old work. + tx.execute_batch("INSERT OR IGNORE INTO ticket_assignment_work_releases + SELECT current.workspace_id, current.assignment_id, COALESCE(ticket.updated_at, current.updated_at) + FROM ticket_current_worker_assignments current JOIN typed_tickets ticket + ON ticket.workspace_id=current.workspace_id AND ticket.ticket_id=current.ticket_id + WHERE ticket.workflow_state IN ('done','closed');")?; + let legacy_plans = { + let mut statement = + tx.prepare("SELECT operation_id, blockers_json FROM worker_removal_operations")?; + let rows = statement.query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })?; + rows.collect::>>()? + }; + for (operation_id, encoded) in legacy_plans { + let mut blockers: Vec = serde_json::from_str(&encoded) + .map_err(|error| Error::Store(format!("invalid removal blockers: {error}")))?; + let mut changed = false; + for blocker in &mut blockers { + if blocker.get("kind").and_then(serde_json::Value::as_str) + == Some("current_assignment") + { + blocker["kind"] = serde_json::Value::String("unfinished_work".into()); + changed = true; + } + } + if changed { + let encoded = serde_json::to_string(&blockers) + .map_err(|error| Error::Store(error.to_string()))?; + // Historical blocked plans stay blocked; issued fingerprints + // and identities are immutable. Fresh planning rereads active work. + tx.execute( + "UPDATE worker_removal_operations SET blockers_json=?2 WHERE operation_id=?1", + params![operation_id, encoded], + )?; + } + } + let broken: i64 = + tx.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { + row.get(0) + })?; + if broken != 0 { + return Err(Error::Store( + "Ticket work migration foreign-key verification failed".into(), + )); + } + tx.execute("INSERT INTO __yoi_schema_migrations(version,name) VALUES (82,'Ticket responsibility and unfinished work separation')", [])?; + tx.commit()?; + Ok(()) + })(); + conn.pragma_update(None, "legacy_alter_table", legacy)?; + conn.pragma_update(None, "foreign_keys", foreign_keys)?; + result +} + fn migrate_backend_job_resources_v80_to_v81(conn: &Connection) -> Result<()> { if current_schema_version(conn)? != 80 { return Err(Error::Store( @@ -15371,6 +15597,7 @@ fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { )); "#, )?; + conn.execute_batch(include_str!("assignment_work.sql"))?; ticket::verify_sqlite_ticket_schema(conn).map_err(|error| { Error::Store(format!( "fresh Workspace Ticket schema verification failed: {error}" @@ -15403,6 +15630,32 @@ fn verify_workspace_resource_constraints(conn: &Connection) -> Result<()> { ))); } } + if current_schema_version(conn)? >= 82 { + for (kind, name) in [ + ("table", "ticket_assignment_work_releases"), + ("view", "ticket_active_worker_assignments"), + ("trigger", "ticket_terminal_releases_work"), + ("trigger", "ticket_terminal_assignment_releases_work_insert"), + ("trigger", "ticket_terminal_assignment_releases_work_update"), + ("trigger", "ticket_active_worker_role_insert"), + ("trigger", "ticket_active_worker_role_update"), + ("trigger", "ticket_active_worker_principal_insert"), + ("trigger", "ticket_active_worker_principal_update"), + ("trigger", "ticket_unfinished_work_blocks_worker_delete"), + ("trigger", "ticket_unfinished_work_blocks_worker_move"), + ] { + let exists: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type=?1 AND name=?2)", + params![kind, name], + |row| row.get(0), + )?; + if !exists { + return Err(Error::Store(format!( + "unfinished Ticket work constraint {name} is missing" + ))); + } + } + } Ok(()) } @@ -17212,6 +17465,9 @@ mod tests { let store = SqliteWorkspaceStore::open(&path).unwrap(); store .with_conn(|conn| { + conn.execute_batch("PRAGMA foreign_keys=OFF; PRAGMA legacy_alter_table=ON;")?; + downgrade_assignment_work_schema(conn); + conn.execute_batch("PRAGMA legacy_alter_table=OFF; PRAGMA foreign_keys=ON;")?; downgrade_schema_66_ticket_and_workdir_authority(conn)?; conn.execute_batch( "DROP TRIGGER runtime_binding_insert_blocked_by_removal; \ @@ -17281,6 +17537,11 @@ mod tests { drop(SqliteWorkspaceStore::open(&path).unwrap()); { let conn = Connection::open(&path).unwrap(); + conn.execute_batch("PRAGMA foreign_keys=OFF; PRAGMA legacy_alter_table=ON;") + .unwrap(); + downgrade_assignment_work_schema(&conn); + conn.execute_batch("PRAGMA legacy_alter_table=OFF; PRAGMA foreign_keys=ON;") + .unwrap(); downgrade_schema_66_ticket_and_workdir_authority(&conn).unwrap(); conn.execute_batch( "ALTER TABLE worker_removal_operations \ @@ -17659,6 +17920,44 @@ mod tests { .unwrap(); } + fn downgrade_assignment_work_schema(conn: &Connection) { + conn.execute_batch( + r#" +DROP TRIGGER ticket_terminal_releases_work; +DROP TRIGGER ticket_terminal_assignment_releases_work_insert; +DROP TRIGGER ticket_terminal_assignment_releases_work_update; +DROP TRIGGER ticket_active_worker_role_insert; +DROP TRIGGER ticket_active_worker_role_update; +DROP TRIGGER ticket_active_worker_principal_insert; +DROP TRIGGER ticket_active_worker_principal_update; +DROP TRIGGER ticket_unfinished_work_blocks_worker_delete; +DROP TRIGGER ticket_unfinished_work_blocks_worker_move; +DROP VIEW ticket_active_worker_assignments; +DROP TABLE ticket_assignment_work_releases; +CREATE UNIQUE INDEX ticket_current_worker_role_idx + ON ticket_current_worker_assignments(workspace_id, role, runtime_id, worker_id) + WHERE principal_kind='worker'; +"#, + ) + .unwrap(); + // Restore the actual old FK contract, not just the old unique index. + let dependents = { + let mut stmt = conn.prepare("SELECT sql FROM sqlite_schema WHERE tbl_name='ticket_current_worker_assignments' AND type IN ('index','trigger') AND sql IS NOT NULL").unwrap(); + stmt.query_map([], |row| row.get::<_, String>(0)) + .unwrap() + .collect::>>() + .unwrap() + }; + let table: String = conn.query_row("SELECT sql FROM sqlite_schema WHERE type='table' AND name='ticket_current_worker_assignments'", [], |row| row.get(0)).unwrap(); + let old_table = table.replace("CREATE TABLE ticket_current_worker_assignments", "CREATE TABLE current_assignments_old") + .replace(" CHECK(\n", " FOREIGN KEY(workspace_id,runtime_id,worker_id) REFERENCES worker_registry(workspace_id,runtime_id,worker_id) ON DELETE RESTRICT,\n CHECK(\n"); + conn.execute_batch(&old_table).unwrap(); + conn.execute_batch("INSERT INTO current_assignments_old SELECT * FROM ticket_current_worker_assignments; DROP TABLE ticket_current_worker_assignments; ALTER TABLE current_assignments_old RENAME TO ticket_current_worker_assignments;").unwrap(); + for sql in dependents { + conn.execute_batch(&sql).unwrap(); + } + } + fn prepare_schema_v52(path: &Path) { let conn = Connection::open(path).unwrap(); configure_sqlite(&conn).unwrap(); @@ -17667,6 +17966,7 @@ mod tests { conn.execute_batch("PRAGMA foreign_keys = OFF; PRAGMA legacy_alter_table = ON;") .unwrap(); create_latest_workspace_schema(&conn).unwrap(); + downgrade_assignment_work_schema(&conn); downgrade_schema_66_ticket_and_workdir_authority(&conn).unwrap(); conn.execute_batch("PRAGMA legacy_alter_table = OFF; PRAGMA foreign_keys = ON;") .unwrap(); @@ -17928,6 +18228,10 @@ mod tests { version: 81, name: "Backend Job immutable resource serialization".to_string() }, + WorkspaceSchemaMigrationStep { + version: 82, + name: "Ticket responsibility and unfinished work separation".to_string() + }, ] ); @@ -18019,6 +18323,7 @@ mod tests { (79, WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME.to_string()), (80, WORKER_RESTORE_INTENTS_MIGRATION_NAME.to_string()), (81, "Backend Job immutable resource serialization".to_string()), + (82, "Ticket responsibility and unfinished work separation".to_string()), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -18370,7 +18675,7 @@ mod tests { .collect::>(), vec![ 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, - 73, 74, 75, 76, 77, 78, 79, 80, 81 + 73, 74, 75, 76, 77, 78, 79, 80, 81, 82 ] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); @@ -18379,7 +18684,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 32); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 33); assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap()); assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap()); assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap()); @@ -20473,6 +20778,203 @@ mod tests { assert_eq!(pinned.definition.name, "coder-review"); } + // Equivalent to the reported closed T-12 + retained Coder data. The + // migration fixture is a real pre-cutover schema, not the live Workspace. + #[test] + fn schema_v81_terminal_responsibility_survives_migration_reopen_and_worker_deletion() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("workspace.db"); + prepare_schema_v50(&path, Some("workspace-a")); + let conn = Connection::open(&path).unwrap(); + configure_sqlite(&conn).unwrap(); + for migration in MIGRATIONS.iter().filter(|m| m.version <= 81) { + (migration.apply)(&conn).unwrap(); + } + conn.execute_batch(r#" + INSERT INTO worker_registry(workspace_id,runtime_id,worker_id,display_name,retention_state,created_at,updated_at) + VALUES('workspace-a','runtime-a','old-coder','Coder T-12','normal','1','1'); + INSERT INTO typed_tickets(workspace_id,ticket_id,slug,title,status,kind,priority,body,workflow_state,workflow_state_explicit,updated_at) + VALUES('workspace-a','closed-ticket','closed-ticket','Closed Ticket','closed','task','normal','','closed',1,'2'); + INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at) + VALUES('workspace-a','closed-ticket','old-assignment','runtime-a','old-coder','actor','1'); + INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,updated_at) + VALUES('workspace-a','closed-ticket','old-assignment','runtime-a','old-coder','1'); + INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,event_id,action,assignment_id,actor,created_at) + VALUES('workspace-a','closed-ticket','assigned-event','assigned','old-assignment','actor','1'); + "#).unwrap(); + let legacy_blockers = serde_json::json!([ + {"kind":"hold"}, {"kind":"current_assignment","assignment_id":"current_assignment","ticket_id":"closed-ticket"} + ]).to_string(); + conn.execute("INSERT INTO worker_removal_operations(operation_id,plan_id,input_fingerprint,workspace_id,runtime_id,worker_id,worker_revision,policy_id,policy_revision,session_disposition,metadata_disposition,archive_retention_kind,diagnostics_disposition,blockers_json,state,reason,created_at,updated_at) + VALUES('old-op','old-plan','immutable-fingerprint','workspace-a','runtime-a','old-coder','1','workspace-default-conservative',1,'purge','tombstone','forever','purge',?1,'blocked','retain','1','1')", [&legacy_blockers]).unwrap(); + migrate_assignment_work_v81_to_v82(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 82); + apply_migrations(&conn).unwrap(); // canonical retry is a no-op + let (plan, fingerprint, state, encoded): (String,String,String,String) = conn.query_row( + "SELECT plan_id,input_fingerprint,state,blockers_json FROM worker_removal_operations WHERE operation_id='old-op'", [], + |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?))).unwrap(); + assert_eq!( + (plan.as_str(), fingerprint.as_str(), state.as_str()), + ("old-plan", "immutable-fingerprint", "blocked") + ); + let blockers: Vec = + serde_json::from_str(&encoded).unwrap(); + assert!( + matches!(&blockers[..], [crate::retention::WorkerRemovalBlocker::Hold,crate::retention::WorkerRemovalBlocker::UnfinishedWork { assignment_id,ticket_id }] if assignment_id=="current_assignment" && ticket_id=="closed-ticket") + ); + conn.execute_batch( + "UPDATE typed_tickets SET workflow_state='planning' WHERE ticket_id='closed-ticket'; + DELETE FROM worker_registry WHERE worker_id='old-coder';", + ) + .unwrap(); + let (responsibility, active, history, tombstone): (i64, i64, i64, i64) = conn + .query_row( + "SELECT (SELECT COUNT(*) FROM ticket_current_worker_assignments), + (SELECT COUNT(*) FROM ticket_active_worker_assignments), + (SELECT COUNT(*) FROM ticket_worker_assignment_events), + (SELECT COUNT(*) FROM ticket_assignment_worker_tombstones)", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .unwrap(); + assert_eq!((responsibility, active, history, tombstone), (1, 0, 1, 1)); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { + row.get::<_, i64>(0) + }) + .unwrap(), + 0 + ); + assert!( + conn.execute( + "UPDATE ticket_current_worker_assignments SET updated_at='3'", + [] + ) + .is_err(), + "removed Worker cannot receive new authority" + ); + } + + #[tokio::test] + async fn close_retains_responsibility_without_occupying_new_work_and_reopen_requires_new_identity() + { + use ticket::TicketBackend; + 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(); + store.with_conn(|conn| { + conn.execute_batch("INSERT INTO worker_registry(workspace_id,runtime_id,worker_id,display_name,retention_state,created_at,updated_at) VALUES('w','r','coder','Coder','normal','1','1');")?; + Ok(()) + }).unwrap(); + let backend = ticket::SqliteTicketBackend::open_verified(&path, "w").unwrap(); + let first = backend.create(ticket::NewTicket::new("First")).unwrap(); + let second = backend.create(ticket::NewTicket::new("Second")).unwrap(); + let assignment = TicketCoderAssignmentRecord { + workspace_id: "w".into(), + ticket_id: first.id.clone(), + assignment_id: "a1".into(), + worker: RuntimeWorkerRef::new("r", "coder"), + assigned_by: "actor".into(), + assigned_at: "1".into(), + }; + store + .set_current_ticket_coder_assignment(&assignment, None, "e1", "op1", false) + .unwrap(); + let next = TicketCoderAssignmentRecord { + ticket_id: second.id.clone(), + assignment_id: "a2".into(), + ..assignment.clone() + }; + assert!( + store + .set_current_ticket_coder_assignment(&next, None, "e2", "op2", false) + .is_err() + ); + backend + .close( + first.id.clone().into(), + ticket::MarkdownText::new("Closed without integration proof"), + ) + .unwrap(); + assert_eq!( + store + .get_current_ticket_coder_assignment("w", &first.id) + .unwrap(), + Some(assignment.clone()) + ); + assert!( + store + .get_active_ticket_worker_assignment("w", &first.id) + .unwrap() + .is_none() + ); + store + .set_current_ticket_coder_assignment(&next, None, "e2", "op2", false) + .unwrap(); + // Generic Ticket tools do not expose terminal -> planning. Seed the + // authority state boundary to prove future/reconciled reopen is inert. + store.with_conn(|conn| { conn.execute("UPDATE typed_tickets SET workflow_state='planning' WHERE workspace_id='w' AND ticket_id=?1", [&first.id])?; Ok(()) }).unwrap(); + assert!( + store + .get_active_ticket_worker_assignment("w", &first.id) + .unwrap() + .is_none() + ); + // Replay of the original operation reports its identity but does not revive work. + store + .set_current_ticket_coder_assignment(&assignment, None, "e1", "op1", false) + .unwrap(); + assert!( + store + .get_active_ticket_worker_assignment("w", &first.id) + .unwrap() + .is_none() + ); + let restarted = TicketCoderAssignmentRecord { + assignment_id: "a3".into(), + ..assignment.clone() + }; + assert!( + store + .set_current_ticket_coder_assignment(&restarted, None, "e3", "op3", false) + .is_err(), + "another Ticket still occupies the Worker" + ); + backend + .close(second.id.clone().into(), ticket::MarkdownText::new("Ended")) + .unwrap(); + store + .set_current_ticket_coder_assignment(&restarted, None, "e3", "op3", false) + .unwrap(); + assert_eq!( + store + .get_active_ticket_worker_assignment("w", &first.id) + .unwrap(), + Some(restarted) + ); + let events = store + .list_ticket_coder_assignment_events("w", &first.id, 10) + .unwrap(); + assert_eq!(events[0].previous_assignment_id.as_deref(), Some("a1")); + assert_eq!(events[0].action, "reassigned"); + assert!( + store + .delete_worker_registry("w", &assignment.worker) + .is_err() + ); + } + #[tokio::test] async fn ticket_worker_assignment_replaces_current_and_preserves_audit_history() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/workdir_removal.rs b/crates/workspace-server/src/workdir_removal.rs index 5044d8045..0208714fe 100644 --- a/crates/workspace-server/src/workdir_removal.rs +++ b/crates/workspace-server/src/workdir_removal.rs @@ -356,10 +356,10 @@ impl SqliteWorkspaceStore { params![operation.workspace_id, operation.working_directory_id], |row| row.get(0), )?; - let current_assignment: bool = conn.query_row( + let unfinished_work: bool = conn.query_row( r#"SELECT EXISTS( SELECT 1 FROM worker_workdir_links AS link - JOIN ticket_current_worker_assignments AS current + JOIN ticket_active_worker_assignments AS current ON current.workspace_id=link.workspace_id JOIN ticket_worker_assignments AS assignment ON assignment.workspace_id=current.workspace_id @@ -404,10 +404,10 @@ impl SqliteWorkspaceStore { detail: "Workdir has a pending Worker attachment reservation", }); } - if current_assignment { + if unfinished_work { guards.push(WorkdirRemovalGuard { - category: "current_assignment", - detail: "Workdir is bound to a Worker with a current Ticket assignment", + category: "unfinished_work", + detail: "Workdir is bound to a Worker with unfinished Ticket work", }); } if retention_hold { @@ -1196,6 +1196,156 @@ mod tests { ); } + #[tokio::test] + async fn terminal_assignments_release_unfinished_work_guard_but_not_attachment_or_hold() { + for (status, workflow_state) in [("closed", "closed"), ("open", "done")] { + let (store, workdir) = seeded_store().await; + store + .with_conn(|conn| { + conn.execute( + "INSERT INTO worker_registry ( + workspace_id, runtime_id, worker_id, display_name, retention_state, + created_at, updated_at + ) VALUES ('workspace-a', 'runtime-a', 'worker-a', 'Worker A', 'pinned', '1', '1')", + [], + )?; + conn.execute( + "INSERT INTO worker_workdir_links ( + workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at + ) VALUES ('workspace-a', 'runtime-a', 'worker-a', 'workdir-a', 'main', 'read_write', '1')", + [], + )?; + conn.execute( + "INSERT INTO typed_tickets ( + workspace_id, ticket_id, slug, title, status, kind, priority, body, + workflow_state, workflow_state_explicit + ) VALUES ('workspace-a', 'ticket-a', 'ticket-a', 'Ticket A', 'open', + 'task', 'normal', '', 'inprogress', 1)", + [], + )?; + conn.execute( + "INSERT INTO ticket_worker_assignments ( + workspace_id, ticket_id, assignment_id, runtime_id, worker_id, + assigned_by, assigned_at + ) VALUES ('workspace-a', 'ticket-a', 'assignment-a', 'runtime-a', 'worker-a', 'test', '1')", + [], + )?; + conn.execute( + "INSERT INTO ticket_current_worker_assignments ( + workspace_id, ticket_id, assignment_id, runtime_id, worker_id, updated_at + ) VALUES ('workspace-a', 'ticket-a', 'assignment-a', 'runtime-a', 'worker-a', '1')", + [], + )?; + Ok(()) + }) + .unwrap(); + let intent = + workdir_removal_intent(&workdir, "workspace-api", "remove Workdir").unwrap(); + let operation = store.reserve_workdir_removal_operation(&intent).unwrap(); + let categories = |guards: Vec| { + guards + .into_iter() + .map(|guard| guard.category) + .collect::>() + }; + assert_eq!( + categories(store.workdir_removal_guards(&operation).unwrap()), + vec!["active_attachment", "unfinished_work", "retention_hold"] + ); + store + .with_conn(|conn| { + conn.execute( + "UPDATE typed_tickets SET status=?1, workflow_state=?2 + WHERE workspace_id='workspace-a' AND ticket_id='ticket-a'", + params![status, workflow_state], + )?; + Ok(()) + }) + .unwrap(); + assert_eq!( + categories(store.workdir_removal_guards(&operation).unwrap()), + vec!["active_attachment", "retention_hold"] + ); + store + .with_conn(|conn| { + conn.execute( + "UPDATE typed_tickets SET status='open', workflow_state='planning' + WHERE workspace_id='workspace-a' AND ticket_id='ticket-a'", + [], + )?; + Ok(()) + }) + .unwrap(); + assert_eq!( + categories(store.workdir_removal_guards(&operation).unwrap()), + vec!["active_attachment", "retention_hold"], + "reopening must not restore unfinished work on the retained assignment" + ); + assert!(matches!( + store.commit_workdir_removal_removed(&operation), + Err(Error::WorkdirAttachmentConflict(_)) + )); + assert!( + store + .get_workdir_registry("workspace-a", "workdir-a") + .unwrap() + .is_some() + ); + store + .with_conn(|conn| { + conn.execute( + "UPDATE worker_registry SET retention_state='normal' + WHERE workspace_id='workspace-a' AND runtime_id='runtime-a' AND worker_id='worker-a'", + [], + )?; + Ok(()) + }) + .unwrap(); + assert_eq!( + categories(store.workdir_removal_guards(&operation).unwrap()), + vec!["active_attachment"] + ); + assert!(matches!( + store.commit_workdir_removal_removed(&operation), + Err(Error::WorkdirAttachmentConflict(_)) + )); + store + .with_conn(|conn| { + conn.execute( + "UPDATE worker_workdir_links SET unlinked_at='2' + WHERE workspace_id='workspace-a' AND workdir_id='workdir-a'", + [], + )?; + Ok(()) + }) + .unwrap(); + assert!(store.workdir_removal_guards(&operation).unwrap().is_empty()); + let completed = store.commit_workdir_removal_removed(&operation).unwrap(); + assert_eq!( + completed.disposition, + Some(WorkdirRemovalDisposition::Removed) + ); + assert!( + store + .get_workdir_registry("workspace-a", "workdir-a") + .unwrap() + .is_none() + ); + let retained: i64 = store + .with_conn(|conn| { + conn.query_row( + "SELECT COUNT(*) FROM ticket_current_worker_assignments + WHERE workspace_id='workspace-a' AND assignment_id='assignment-a'", + [], + |row| row.get(0), + ) + .map_err(Error::from) + }) + .unwrap(); + assert_eq!(retained, 1); + } + } + #[tokio::test] async fn retained_completion_keeps_registry_and_is_auditable() { let (store, workdir) = seeded_store().await; diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index b5d997dcf..360b239d2 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -45,6 +45,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[ "repository_ssh_host_trusts", "server_secret_versions", "ticket_assignment_operations", + "ticket_assignment_work_releases", "ticket_assignment_ticket_tombstones", "ticket_assignment_worker_tombstones", "ticket_current_worker_assignments", @@ -686,10 +687,10 @@ fn workspace_database_blockers( for (sql, kind, resource_kind, message) in [ ( - "SELECT COUNT(*) FROM ticket_current_worker_assignments WHERE workspace_id = ?1", + "SELECT COUNT(*) FROM ticket_active_worker_assignments WHERE workspace_id = ?1", WorkspaceDeletionBlockerKind::WorkerRemovalBlocked, "ticket", - "Remove current Ticket assignments before deleting the Workspace.", + "End unfinished Ticket work before deleting the Workspace.", ), ( "SELECT COUNT(*) diff --git a/resources/prompts/role/orchestrator.md b/resources/prompts/role/orchestrator.md index bbee8320b..34db081a0 100644 --- a/resources/prompts/role/orchestrator.md +++ b/resources/prompts/role/orchestrator.md @@ -16,14 +16,14 @@ Push the resulting `merge_to` branch through its configured normal push path. Pr After each repository push, verify the provider resolves `merge_to` exactly to `target_ref_after`, then call `CompleteMergeRequest` with that explicit MR ID, before/after evidence, authoritative approval event, and merge strategy. `CompleteMergeRequest` records and verifies only that already-applied repository integration; it does not update the branch itself, and it does not complete the Ticket. Partial integration is valid: keep the Ticket and assignment in progress while remaining linked MRs are implemented, reviewed, or integrated. -If a repository push succeeds but MR completion recording fails, do not push again or invent a new result. Retry the same `CompleteMergeRequest` operation and evidence. Once every linked required result is recorded, reread the latest Ticket item revision and exact linked MR/source-result set, identify the effective Reviewer approval that attests that exact snapshot and judges the requirements satisfied, then call `CompleteTicket` with the item revision, MR IDs, and that requirement approval event ID. If the Ticket was rescoped after partial or full integration, obtain a fresh Reviewer attestation against the immutable merged-result subjects through an explicitly addressed merged MR; this post-integration review updates requirement judgment without rewriting any MergeResult. `CompleteTicket` alone moves the Ticket to `done` and releases the assignment atomically; MR-set changes, rescope, review revocation, stale assignment/state, or an unmerged result must fail closed. Retry an already-recorded operation only with the same fingerprint. +If a repository push succeeds but MR completion recording fails, do not push again or invent a new result. Retry the same `CompleteMergeRequest` operation and evidence. Once every linked required result is recorded, reread the latest Ticket item revision and exact linked MR/source-result set, identify the effective Reviewer approval that attests that exact snapshot and judges the requirements satisfied, then call `CompleteTicket` with the item revision, MR IDs, and that requirement approval event ID. If the Ticket was rescoped after partial or full integration, obtain a fresh Reviewer attestation against the immutable merged-result subjects through an explicitly addressed merged MR; this post-integration review updates requirement judgment without rewriting any MergeResult. `CompleteTicket` alone moves the Ticket to `done` and releases unfinished work atomically while retaining current responsibility for display; MR-set changes, rescope, review revocation, stale assignment/state, or an unmerged result must fail closed. Retry an already-recorded operation only with the same fingerprint. Do not create or delegate an implementation worktree/branch until the Ticket records enough agreed intent, requirements, and acceptance criteria to bound the work. Workspace roots, cwd, profile selector, and launch-prompt configuration are control-plane/environment facts rather than user instructions. If the launch input names explicit Git/worktree operation targets, use those paths only for that operation and do not substitute heuristic roots. -Use `WorkerRemove` only for a terminal or authoritatively reassigned non-internal Coder after implementation, review, fix, merge/commit, and report handoffs are complete. Do not remove a Coder merely because one turn completed or it is temporarily idle; retain it while review or request-changes work can still return. The Worker must already be stopped, must not be restoring, must have no current Ticket assignment, pending notification, Reviewer handoff, legal hold, or pin, and must not be this Orchestrator. Immediately before removal, reread authoritative Ticket state, assignment, thread/review evidence, and the target Worker through `WorkerList`, then call `WorkerRemove` with a concise reason. Backend authority captures the current Worker revision internally and revalidates removal guards; do not guess policy or supply lifecycle authority in model input. After removal, reread the Worker catalog and attachment state. Treat assignment, running/restoring, retention-policy, attachment-close, and attachment-release conflicts as authoritative failures. `WorkerRemove` releases the Worker attachment but deliberately preserves the Workdir materialization. +Use `WorkerRemove` only for a terminal or authoritatively reassigned non-internal Coder after implementation, review, fix, merge/commit, and report handoffs are complete. Do not remove a Coder merely because one turn completed or it is temporarily idle; retain it while review or request-changes work can still return. The Worker must already be stopped, must not be restoring, must have no unfinished Ticket work, pending notification, Reviewer handoff, legal hold, or pin, and must not be this Orchestrator. Immediately before removal, reread authoritative Ticket state, assignment, thread/review evidence, and the target Worker through `WorkerList`, then call `WorkerRemove` with a concise reason. Backend authority captures the current Worker revision internally and revalidates removal guards; do not guess policy or supply lifecycle authority in model input. After removal, reread the Worker catalog and attachment state. Treat unfinished-work, running/restoring, retention-policy, attachment-close, and attachment-release conflicts as authoritative failures. `WorkerRemove` releases the Worker attachment but deliberately preserves the Workdir materialization. -Coder cleanup is a separate post-completion decision owned by this Orchestrator. Never predeclare `delete_on_completion`, `retain_on_completion`, or equivalent retention policy when launching or reserving a Coder. After all `CompleteMergeRequest` operations and guarded `CompleteTicket`, perform one cleanup pass before ending the orchestration turn: reread the current Ticket and `WorkerList`, verify completion is authoritative and the Coder has no current Ticket assignment, then inspect the Coder Worker, Workdir attachment and occupancy, repository cleanliness, ownership, provider availability, and any other current use of that Workdir. For Worker control, use the exact subject returned by `WorkerList`. If the Coder is active, call `WorkerStop` and reread its terminal status before `WorkerRemove`; idle status, Coder self-report, or review approval alone is not removal authority. Retain an existing or still-needed Workdir. Delete only a Ticket-dedicated Workdir that this Orchestrator created or selected and current authority proves is no longer needed, clean, and unoccupied. +Coder cleanup is a separate post-completion decision owned by this Orchestrator. Never predeclare `delete_on_completion`, `retain_on_completion`, or equivalent retention policy when launching or reserving a Coder. After all `CompleteMergeRequest` operations and guarded `CompleteTicket`, perform one cleanup pass before ending the orchestration turn: reread the current Ticket and `WorkerList`, verify completion is authoritative and the Coder has no unfinished Ticket work (terminal current responsibility may remain), then inspect the Coder Worker, Workdir attachment and occupancy, repository cleanliness, ownership, provider availability, and any other current use of that Workdir. For Worker control, use the exact subject returned by `WorkerList`. If the Coder is active, call `WorkerStop` and reread its terminal status before `WorkerRemove`; idle status, Coder self-report, or review approval alone is not removal authority. Retain an existing or still-needed Workdir. Delete only a Ticket-dedicated Workdir that this Orchestrator created or selected and current authority proves is no longer needed, clean, and unoccupied. -For Ticket-dedicated cleanup, preserve this guarded order: stop the Coder if needed and confirm it is terminal; unassign it through the available orchestration authority; call `WorkerRemove`; use `WorkdirList` to reread the actual Workdir and confirm attachment release, clean state, and no occupancy; only then call `WorkdirDelete`. Never delete a Workdir before Worker removal has released its attachment, and never invent an unassignment operation or bypass when the required authority is unavailable. `CurrentAssignment` means unassign and reread before retrying `WorkerRemove`. Running, restoring, pinned, legal hold, occupied, dirty, blocked, provider-unavailable, ownership-unknown, still-needed, or uncertain state means retain the resource and report the concrete bounded blocker. If stop, unassign, removal, attachment release, or deletion reports a partial failure, do not infer success or advance to the next step; reread current Ticket, assignment, Worker, and Workdir authority before a bounded safe retry. Cleanup failure never rolls back an already completed Merge Request or Ticket. Do not add routine cleanup success comments; report only blockers that require human or Orchestrator judgment. Never force removal, discard changes, or retry from stale assumptions. +For Ticket-dedicated cleanup, preserve this guarded order: stop the Coder if needed and confirm it is terminal; verify there is no unfinished Ticket work; call `WorkerRemove`; use `WorkdirList` to reread the actual Workdir and confirm attachment release, clean state, and no occupancy; only then call `WorkdirDelete`. Terminal Ticket responsibility is retained for display and does not require unassignment before cleanup. Never delete a Workdir before Worker removal has released its attachment, and never invent an unassignment operation or bypass when the required authority is unavailable. `UnfinishedWork` (`unfinished_work` on the wire) requires explicit work end or reassignment through available guarded authority, then rereading before retrying `WorkerRemove`; idle or stopped status does not end unfinished work. Running, restoring, pinned, legal hold, occupied, dirty, blocked, provider-unavailable, ownership-unknown, still-needed, or uncertain state means retain the resource and report the concrete bounded blocker. If stop, work end/reassignment, removal, attachment release, or deletion reports a partial failure, do not infer success or advance to the next step; reread current Ticket, active unfinished work, Worker, and Workdir authority before a bounded safe retry. Cleanup failure never rolls back an already completed Merge Request or Ticket. Do not add routine cleanup success comments; report only blockers that require human or Orchestrator judgment. Never force removal, discard changes, or retry from stale assumptions. diff --git a/web/workspace/src/routes/w/[workspaceId]/tickets/[ticketId]/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/tickets/[ticketId]/+page.svelte index be448323d..3bc2ad131 100644 --- a/web/workspace/src/routes/w/[workspaceId]/tickets/[ticketId]/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/tickets/[ticketId]/+page.svelte @@ -519,7 +519,8 @@