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

fix(ticket): separate retained responsibility from unfinished work

Browse Source 
T-715: end assignment work identities atomically on done/close without deleting responsibility. Reopen never reactivates old authority. Schema 82 preserves legacy closed assignments, historical removal plans and Worker attribution; active work drives admission, auth/routing, retention and cleanup while independent lifecycle/attachment fences remain.

Validation: root cargo check; cargo test -p yoi-workspace-server (630 lib + 11 bin), -p merge-request (20), -p worker (947); worker-runtime restore/removal filters (2); Deno ticket responsibility/runtime UI contracts (12); cargo fmt --all -- --check; git diff --check HEAD.
This commit is contained in:
Hare
2026-10-08 15:35:16 +09:00
parent b0d8c7b786
commit b6de908e9f
16 changed files with 1799 additions and 122 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
@@ -146,7 +146,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,
@@ -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)
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
+562 -34
View File
@@ -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<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,
@@ -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<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,
@@ -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();
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(*)