Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

merge(develop): retain active-work guards with immutable Ticket evidence

Browse Source 
T-716: incorporate T-715 (b6de908e) into the existing Ticket source branch only. Resolve MR read/readiness conflicts by keeping persisted merged-result evaluation and active assignment selection for open source observation. Preserve responsibility records and terminal work releases. Update temporary API regression fixtures to distinguish retained identity from active work.

Narrow validation: merge-request store 21 tests; Workspace Server Ticket evidence API 5 tests, including done/merged remote evidence without active assignment and unavailable open source with retained released responsibility.
This commit is contained in:
Hare
2026-10-08 16:20:38 +09:00
parent 5424280348 b6de908e9f
commit 14fea520df
17 changed files with 1829 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/assignment_work.sql
+100
View File
@@ -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;
crates/workspace-server/src/authority.rs
+9 -4
View File
@@ -148,7 +148,7 @@ impl merge_request::AssignmentSource for AuthorityMergeRequestSource {
ticket_id: &str,
) -> std::result::Result<Option<merge_request::CurrentAssignment>, 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,
@@ -771,10 +771,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
@@ -800,7 +805,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)
crates/workspace-server/src/latest_schema.sql
-6
View File
@@ -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
crates/workspace-server/src/retention.rs
+406 -26
View File
@@ -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());
crates/workspace-server/src/server.rs
+563 -35
View File
@@ -5701,7 +5701,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);
@@ -13181,7 +13181,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,
@@ -13314,7 +13314,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,
@@ -13483,7 +13483,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
@@ -13567,7 +13567,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,
)?
@@ -14141,7 +14141,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,
@@ -14630,7 +14630,7 @@ impl merge_request::AssignmentSource for MergeRequestAssignmentSource {
ticket_id: &str,
) -> std::result::Result<Option<merge_request::CurrentAssignment>, 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,
@@ -14664,7 +14664,7 @@ impl TicketMergeRevisionSource for RuntimeTicketMergeRevisionSource {
let assignment = self
.api
.store
.get_current_ticket_coder_assignment(self.api.workspace_id(), ticket_id)
.get_active_ticket_worker_assignment(self.api.workspace_id(), ticket_id)
.map_err(|error| merge_ref_diagnostic(error.into()))?
.ok_or_else(|| MergeRequestRefDiagnostic {
code: "source_ref_runtime_unavailable".into(),
@@ -15356,7 +15356,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,
@@ -15532,7 +15532,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, target) = if let Some(refs) = integrated_refs {
refs
} else {
@@ -15611,7 +15611,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) =
if mr.state == merge_request::MergeRequestState::Merged {
(
@@ -15681,9 +15681,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
@@ -15780,9 +15780,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,
@@ -15852,9 +15852,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
@@ -15962,9 +15962,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)?,
@@ -16024,9 +16024,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
@@ -16094,9 +16094,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
@@ -16160,9 +16160,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(
@@ -16504,7 +16507,7 @@ fn active_orchestrator_assignment(
workspace_id: &str,
ticket_id: &str,
) -> Result<Option<TicketRoleAssignmentRecord>> {
let assignment = api.store.get_current_ticket_role_assignment(
let assignment = api.store.get_active_ticket_role_assignment(
workspace_id,
ticket_id,
TicketAssignmentRole::Orchestrator,
@@ -16547,7 +16550,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()
});
@@ -16690,7 +16693,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()
{
@@ -21273,7 +21276,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,
@@ -24395,15 +24398,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()
))
@@ -24619,13 +24622,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()
),
@@ -47071,6 +47074,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,
@@ -53823,7 +54264,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;
@@ -53839,7 +54280,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(),
@@ -53872,6 +54313,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();
crates/workspace-server/src/server/tests/ticket_evidence_tests.rs
+29 -5
View File
@@ -313,10 +313,18 @@ async fn done_merged_remote_ticket_uses_immutable_result_in_show_query_list_and_
fixture
.api
.store
.get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
.unwrap()
.is_none()
);
assert!(
fixture
.api
.store
.get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
.unwrap()
.is_some()
);
let shown = fixture.show().await;
assert_eq!(shown.state, "done");
@@ -508,14 +516,30 @@ async fn open_source_without_runtime_is_typed_unavailable_not_stale_after_rescop
let fixture = EvidenceApiFixture::new().await;
let approval = fixture.approve("open-review");
fixture.rescope();
// Simulate a retired Coder assignment, leaving an open MR and its review
// history. Only this temporary authority fixture is changed.
// End this assignment identity's work without erasing its responsibility.
// A retained Coder must not silently supply a Runtime for an open source.
let connection = Connection::open(&fixture.api.config.database_path).unwrap();
assert_eq!(connection.execute(
"DELETE FROM ticket_current_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2",
params![TEST_WORKSPACE_ID, fixture.ticket_id],
"INSERT INTO ticket_assignment_work_releases(workspace_id,assignment_id,released_at) VALUES(?1,?2,?3)",
params![TEST_WORKSPACE_ID, fixture.auth.assignment_id, evidence_time().to_rfc3339()],
).unwrap(), 1);
drop(connection);
assert!(
fixture
.api
.store
.get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
.unwrap()
.is_some()
);
assert!(
fixture
.api
.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
.unwrap()
.is_none()
);
let shown = fixture.show().await;
let summary = &shown.merge_requests[0];
crates/workspace-server/src/store.rs
+519 -17
View File
@@ -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<Vec<TicketRoleAssignmentRecord>>;
fn get_active_ticket_role_assignment_for_worker(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
) -> Result<Option<TicketRoleAssignmentRecord>>;
fn get_active_ticket_role_assignment(
&self,
workspace_id: &str,
ticket_id: &str,
role: TicketAssignmentRole,
) -> Result<Option<TicketRoleAssignmentRecord>>;
/// 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<bool>;
fn get_active_ticket_worker_assignment(
&self,
workspace_id: &str,
ticket_id: &str,
) -> Result<Option<TicketCoderAssignmentRecord>>;
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<Vec<TicketRoleAssignmentRecord>> {
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::<rusqlite::Result<Vec<_>>>()
.map_err(Into::into)
})
}
fn get_active_ticket_role_assignment(
&self,
workspace_id: &str,
ticket_id: &str,
role: TicketAssignmentRole,
) -> Result<Option<TicketRoleAssignmentRecord>> {
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<Option<TicketRoleAssignmentRecord>> {
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<String> = 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<Option<TicketCoderAssignmentRecord>> {
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::<rusqlite::Result<Vec<_>>>()?
};
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::<rusqlite::Result<Vec<_>>>()?
};
for (operation_id, encoded) in legacy_plans {
let mut blockers: Vec<serde_json::Value> = 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::<rusqlite::Result<Vec<_>>>()
.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<_>>(),
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<crate::retention::WorkerRemovalBlocker> =
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();
crates/workspace-server/src/workdir_removal.rs
+155 -5
View File
@@ -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<WorkdirRemovalGuard>| {
guards
.into_iter()
.map(|guard| guard.category)
.collect::<Vec<_>>()
};
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;
crates/workspace-server/src/workspace_deletion.rs
+3 -2
View File
@@ -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(*)