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

feat: separate Ticket case state and immutable closure judgments

Browse Source
This commit is contained in:
Hare
2026-10-11 01:41:46 +09:00
parent 6a12b6cc34
commit ba3ebaa68c
65 changed files with 7625 additions and 6844 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/authority.rs
+96 -115
View File
@@ -7,11 +7,11 @@ use merge_request::{
MergeRequest, MergeRequestState, MergeRequestStore, MergeRequestThreadEvent, ReviewDecision,
};
use project_record::{allocate_record_id, unix_epoch_millis_now};
use rusqlite::{params_from_iter, types::Value as SqlValue};
use rusqlite::{OptionalExtension, params, params_from_iter, types::Value as SqlValue};
use ticket::{
SqliteTicketBackend, SqliteTicketListCursor, SqliteTicketListItem, SqliteTicketListPageQuery,
TicketBackend, TicketEvent, TicketIdOrSlug, TicketWorkflowState, TicketWorkspaceActionPriority,
TicketBackend, TicketEvent, TicketIdOrSlug, TicketState, TicketWorkspaceActionPriority,
project_ticket_workspace_item,
};
@@ -282,7 +282,7 @@ impl SqliteWorkspaceAuthority {
id: ticket.id,
resource_key: ticket.resource_key,
title: ticket.title,
state: ticket.state,
state: ticket.state.as_str().into(),
})
.collect::<Vec<_>>();
let resources = self
@@ -387,7 +387,7 @@ impl SqliteWorkspaceAuthority {
let states = query
.states
.iter()
.map(|state| bind(SqlValue::Text(state.clone())))
.map(|state| bind(SqlValue::Text(state.as_str().into())))
.collect::<Vec<_>>();
predicates.push(format!("t.workflow_state IN ({})", states.join(",")));
}
@@ -438,14 +438,7 @@ impl SqliteWorkspaceAuthority {
WHERE relation.workspace_id=t.workspace_id
AND ((relation.ticket_id=t.ticket_id AND relation.kind='depends_on')
OR (relation.target=t.ticket_id AND relation.kind='blocks'))
AND blocker.workflow_state NOT IN ('done','closed'))";
let active_blocker = "EXISTS (SELECT 1 FROM typed_ticket_relations relation
JOIN typed_tickets blocker ON blocker.workspace_id=relation.workspace_id
AND blocker.ticket_id=CASE WHEN relation.ticket_id=t.ticket_id THEN relation.target ELSE relation.ticket_id END
WHERE relation.workspace_id=t.workspace_id
AND ((relation.ticket_id=t.ticket_id AND relation.kind='depends_on')
OR (relation.target=t.ticket_id AND relation.kind='blocks'))
AND blocker.workflow_state NOT IN ('queued','inprogress','done','closed'))";
AND blocker.workflow_state = 'open')";
if !query.event_kinds.is_empty() {
let event_kinds = query
.event_kinds
@@ -459,17 +452,13 @@ impl SqliteWorkspaceAuthority {
// a primary-MR/latest-request or implementation-report heuristic loses candidates.
for attention in &query.attention {
match attention.as_str() {
"done_not_closed" => predicates.push("t.workflow_state='done'".into()),
"blocked" => predicates.push(blocker.into()),
"unblocked" => predicates.push(format!("NOT {blocker}")),
"ready" => predicates.push(format!("t.workflow_state='ready' AND NOT {blocker}")),
_ => {} // live evidence is evaluated after observation
}
}
let rank_expression = match sort {
TicketQuerySort::Priority => format!(
"CASE WHEN t.workflow_state='ready' AND NOT {active_blocker} THEN 0 WHEN t.workflow_state IN ('queued','inprogress') THEN 1 ELSE 2 END"
),
TicketQuerySort::Priority => "0".to_string(),
TicketQuerySort::Relevance => {
if let Some(text) = query.query.as_deref().filter(|text| !text.is_empty()) {
let pattern = bind(SqlValue::Text(format!("%{}%", text.to_lowercase())));
@@ -807,24 +796,13 @@ impl SqliteWorkspaceAuthority {
}
let action_blockers = Vec::new();
let action_eligibility = TicketActionEligibility {
can_assign_orchestrator: matches!(
ticket.meta.workflow_state,
TicketWorkflowState::Planning | TicketWorkflowState::Ready
) && !has_orchestrator
&& !has_worker,
can_unassign_orchestrator: has_orchestrator
&& matches!(
ticket.meta.workflow_state,
TicketWorkflowState::Planning | TicketWorkflowState::Ready
),
can_queue: true,
can_start_manual_worker: matches!(
ticket.meta.workflow_state,
TicketWorkflowState::Planning
| TicketWorkflowState::Ready
| TicketWorkflowState::Queued
| TicketWorkflowState::InProgress
) && !has_worker
// Case openness is not readiness. These are explicit operations;
// occupancy remains an independent resource constraint.
can_assign_orchestrator: !has_orchestrator && !has_worker,
can_unassign_orchestrator: has_orchestrator,
can_queue: ticket.meta.workflow_state == TicketState::Open,
can_start_manual_worker: ticket.meta.workflow_state == TicketState::Open
&& !has_worker
&& !has_blockers,
queue_tickets: dependency_check.queue_tickets.clone(),
blockers: action_blockers,
@@ -916,12 +894,16 @@ impl SqliteWorkspaceAuthority {
id: ticket.meta.id,
resource_key,
title: ticket.meta.title,
state: ticket.meta.workflow_state.as_str().to_string(),
state: ticket.meta.workflow_state.into(),
readiness: ticket.meta.readiness,
priority: ticket.meta.priority,
created_at: ticket.meta.created_at,
updated_at: ticket.meta.updated_at,
content_digest,
closure_operation_key: ticket.closure_operation_key.clone(),
current_closure: ticket.current_closure.clone(),
closure_history: ticket.case_history.clone(),
legacy_handoff: ticket.legacy_workflow_snapshot.clone(),
queued_by: ticket.meta.queued_by,
queued_at: ticket.meta.queued_at,
targets,
@@ -963,6 +945,44 @@ impl SqliteWorkspaceAuthority {
}
}
impl SqliteWorkspaceAuthority {
/// Bounded list projection: read only one immutable closure record and the
/// migration handoff, never the full body or event history for every row.
fn populate_ticket_summary_case_fields(&self, summary: &mut TicketSummary) -> Result<()> {
self.store.with_conn(|conn| {
summary.closure_operation_key = conn.query_row(
"SELECT closure_operation_key FROM typed_ticket_case_metadata WHERE workspace_id=?1 AND ticket_id=?2",
params![self.workspace_id, summary.id], |row| row.get(0)
).optional()?.flatten();
if summary.state == server_api::BrowserTicketState::Closed {
if let Some(key) = &summary.closure_operation_key {
let json: Option<String> = conn.query_row(
"SELECT judgment_json FROM typed_ticket_case_history WHERE workspace_id=?1 AND ticket_id=?2 AND operation_key=?3",
params![self.workspace_id, summary.id, key], |row| row.get(0)
).optional()?;
if let Some(json) = json {
let entry: ticket::TicketCaseHistoryEntry = serde_json::from_str(&json).map_err(|e| Error::Store(e.to_string()))?;
let reason = match entry.action {
ticket::TicketCaseAction::Close { reason } | ticket::TicketCaseAction::CorrectClosure { reason } => reason,
ticket::TicketCaseAction::Reopen => return Err(Error::Store("Closure key names a reopen receipt".into())),
};
summary.current_closure = Some(ticket::TicketClosureJudgment {
operation_key: entry.operation_key, reason, explanation: entry.explanation,
references: entry.references, author: entry.author, at: entry.at,
});
}
}
}
let snapshot: Option<String> = conn.query_row(
"SELECT snapshot_json FROM typed_ticket_legacy_workflow_snapshot WHERE workspace_id=?1 AND ticket_id=?2",
params![self.workspace_id, summary.id], |row| row.get(0)
).optional()?;
summary.legacy_handoff = snapshot.map(|json| serde_json::from_str(&json).map_err(|e| Error::Store(e.to_string()))).transpose()?;
Ok(())
})
}
}
impl TicketAuthority for SqliteWorkspaceAuthority {
fn list_tickets(&self, limit: usize) -> Result<ProjectRecordList<TicketSummary>> {
let projection = self.ticket_backend.list_workspace_projection(limit)?;
@@ -970,24 +990,9 @@ impl TicketAuthority for SqliteWorkspaceAuthority {
.items
.into_iter()
.map(|item| {
let projection =
project_ticket_workspace_item(&item.summary, &item.relation_blockers, None);
let resource_key = item.summary.resource_key.clone().ok_or_else(|| {
Error::Store(format!("missing resource key for {}", item.summary.id))
})?;
Ok::<_, Error>(TicketSummary {
resource_key,
id: item.summary.id,
title: item.summary.title,
state: item.summary.workflow_state.as_str().to_string(),
priority: item.summary.priority,
updated_at: item.summary.updated_at,
queued_by: item.summary.queued_by,
queued_at: item.summary.queued_at,
workspace_action_priority: workspace_action_priority_name(projection.priority)
.to_string(),
record_source: "sqlite_yoi_ticket".to_string(),
})
let mut summary = ticket_summary_from_sqlite_item(item)?;
self.populate_ticket_summary_case_fields(&mut summary)?;
Ok(summary)
})
.collect::<Result<Vec<_>>>()?;
Ok(ProjectRecordList {
@@ -1005,7 +1010,7 @@ impl TicketAuthority for SqliteWorkspaceAuthority {
let parsed_states = states
.iter()
.map(|state| {
TicketWorkflowState::parse(state).ok_or_else(|| {
TicketState::parse(state).ok_or_else(|| {
Error::InvalidRecordId(format!("unsupported ticket state `{state}`"))
})
})
@@ -1029,7 +1034,11 @@ impl TicketAuthority for SqliteWorkspaceAuthority {
let items = page
.items
.into_iter()
.map(ticket_summary_from_sqlite_item)
.map(|item| {
let mut summary = ticket_summary_from_sqlite_item(item)?;
self.populate_ticket_summary_case_fields(&mut summary)?;
Ok(summary)
})
.collect::<Result<Vec<_>>>()?;
let next_cursor = page
.next
@@ -1945,13 +1954,6 @@ fn ticket_evidence_summary(
}
fn validate_ticket_query(query: &TicketQueryRequest) -> Result<()> {
for state in &query.states {
if ticket::TicketWorkflowState::parse(state).is_none() {
return Err(Error::InvalidRecordId(format!(
"unsupported Ticket workflow state `{state}`"
)));
}
}
for evidence in &query.evidence {
if !matches!(
evidence.as_str(),
@@ -1962,25 +1964,6 @@ fn validate_ticket_query(query: &TicketQueryRequest) -> Result<()> {
)));
}
}
for attention in &query.attention {
if !matches!(
attention.as_str(),
"done_not_closed"
| "unresolved_review"
| "missing_commit"
| "blocked"
| "unblocked"
| "ready"
| "awaiting_review"
| "unresolved_changes"
| "stale_after_rescope"
| "missing_evidence"
) {
return Err(Error::InvalidRecordId(format!(
"unsupported Ticket attention filter `{attention}`"
)));
}
}
if let Some(status) = query.review_status.as_deref()
&& !matches!(
status,
@@ -2125,19 +2108,17 @@ fn ticket_evidence_matches(evidence: &TicketEvidenceSummary, filter: &str) -> bo
}
fn ticket_attention_matches(
state: &str,
_state: &str,
evidence: &TicketEvidenceSummary,
is_blocked: bool,
authoritative_events: &[TicketEvent],
attention: &str,
) -> bool {
match attention {
"done_not_closed" => state == "done",
"unresolved_review" => evidence.unresolved_request_changes,
"missing_commit" => !evidence.has_commit,
"blocked" => is_blocked,
"unblocked" => !is_blocked,
"ready" => state == "ready" && !is_blocked,
"awaiting_review" => {
evidence.has_review_request && evidence.review_status.as_deref() == Some("pending")
}
@@ -2209,11 +2190,11 @@ fn ticket_matches_query(
}
if !query.attention.iter().all(|attention| {
ticket_attention_matches(
&summary.state,
summary.state.as_str(),
&detail.evidence,
!detail.relations.blockers.is_empty(),
authoritative_events,
attention,
attention.as_str(),
)
}) {
return false;
@@ -2323,6 +2304,9 @@ fn ticket_query_item(
created_at: detail.created_at.clone(),
updated_at: summary.updated_at,
content_digest: detail.content_digest.clone(),
closure_operation_key: detail.closure_operation_key.clone(),
current_closure: detail.current_closure.clone(),
legacy_handoff: detail.legacy_handoff.clone(),
workspace_action_priority: summary.workspace_action_priority,
matched_fields,
snippet: snippet.map(|value| truncate_body(&value, 512).0),
@@ -2682,10 +2666,14 @@ fn ticket_summary_from_ticket(ticket: &ticket::Ticket) -> Result<TicketSummary>
queued_at: ticket.meta.queued_at.clone(),
updated_at: ticket.meta.updated_at.clone(),
};
ticket_summary_from_sqlite_item(SqliteTicketListItem {
let mut summary = ticket_summary_from_sqlite_item(SqliteTicketListItem {
summary,
relation_blockers: ticket.relations.blockers.clone(),
})
})?;
summary.closure_operation_key = ticket.closure_operation_key.clone();
summary.current_closure = ticket.current_closure.clone();
summary.legacy_handoff = ticket.legacy_workflow_snapshot.clone();
Ok(summary)
}
fn ticket_summary_from_sqlite_item(item: SqliteTicketListItem) -> Result<TicketSummary> {
@@ -2699,13 +2687,16 @@ fn ticket_summary_from_sqlite_item(item: SqliteTicketListItem) -> Result<TicketS
resource_key,
id: item.summary.id,
title: item.summary.title,
state: item.summary.workflow_state.as_str().to_string(),
state: item.summary.workflow_state.into(),
priority: item.summary.priority,
updated_at: item.summary.updated_at,
queued_by: item.summary.queued_by,
queued_at: item.summary.queued_at,
workspace_action_priority: workspace_action_priority_name(projection.priority).to_string(),
record_source: "sqlite_yoi_ticket".to_string(),
closure_operation_key: None,
current_closure: None,
legacy_handoff: None,
})
}
@@ -3222,14 +3213,7 @@ mod tests {
.create(ticket::NewTicket::new("Explicit work"))
.unwrap();
let authority = SqliteWorkspaceAuthority::new(&path, "w").unwrap();
for (state, allowed) in [
("planning", true),
("ready", true),
("queued", true),
("inprogress", true),
("done", false),
("closed", false),
] {
for (state, allowed) in [("open", true), ("closed", false)] {
conn.execute(
"UPDATE typed_tickets SET workflow_state=?1 WHERE ticket_id=?2",
rusqlite::params![state, ticket.id],
@@ -3246,7 +3230,7 @@ mod tests {
);
}
conn.execute(
"UPDATE typed_tickets SET workflow_state='planning' WHERE ticket_id=?1",
"UPDATE typed_tickets SET workflow_state='open' WHERE ticket_id=?1",
[&ticket.id],
)
.unwrap();
@@ -3302,7 +3286,7 @@ mod tests {
)
.unwrap();
conn.execute(
"UPDATE typed_tickets SET workflow_state='planning' WHERE ticket_id=?1",
"UPDATE typed_tickets SET workflow_state='open' WHERE ticket_id=?1",
[&ticket.id],
)
.unwrap();
@@ -3353,11 +3337,11 @@ INSERT INTO typed_tickets (
) VALUES
('workspace-test', '00000000001J2', '00000000001J2', 'Read bridge', 'open', 'ticket', 'P2',
'Ticket body. Deep Ticket marker. This body is intentionally long enough for bounded projection tests.',
'2026-01-01T00:00:00Z', '2026-01-03T00:00:00Z', 'ready', 1),
'2026-01-01T00:00:00Z', '2026-01-03T00:00:00Z', 'open', 1),
('workspace-test', '00000000001J5', '00000000001J5', 'Second ticket', 'open', 'ticket', 'P2',
'Ticket body', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z', 'queued', 1),
'Ticket body', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z', 'open', 1),
('workspace-test', '00000000001J6', '00000000001J6', 'Third ticket', 'open', 'ticket', 'P2',
'Ticket body', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z', 'planning', 1);
'Ticket body', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z', 'open', 1);
INSERT INTO typed_ticket_events (
workspace_id, ticket_id, event_index, kind, author, at, heading, body
@@ -3496,11 +3480,8 @@ VALUES ('workspace-test', 'ticket', 4);
assert_eq!(tickets.record_authority, "workspace-sqlite");
assert_eq!(tickets.items[0].record_source, "sqlite_yoi_ticket");
assert_eq!(tickets.items[0].id, "00000000001J2");
assert_eq!(tickets.items[0].state, "ready");
assert_eq!(
tickets.items[0].workspace_action_priority,
"ready_for_queue"
);
assert_eq!(tickets.items[0].state.as_str(), "open");
assert_eq!(tickets.items[0].workspace_action_priority, "background");
let ticket_by_key = authority.ticket(&tickets.items[0].resource_key).unwrap();
assert_eq!(ticket_by_key.id, tickets.items[0].id);
@@ -3516,9 +3497,9 @@ VALUES ('workspace-test', 'ticket', 4);
let ticket_query = authority
.query_tickets(TicketQueryRequest {
query: Some("Deep Ticket marker".to_string()),
states: vec!["ready".to_string()],
states: vec![server_api::BrowserTicketState::Open],
linked_objective_id: Some("00000000001J3".to_string()),
attention: vec!["missing_commit".to_string()],
attention: vec![server_api::TicketAttention::MissingCommit],
limit: Some(1),
..TicketQueryRequest::default()
})
@@ -3617,7 +3598,7 @@ VALUES ('workspace-test', 'ticket', 4);
assert_eq!(incoming_relation.items[0].id, "00000000001J5");
let summary_page = authority
.list_ticket_page(TicketListProjectionRequest {
states: vec!["planning".to_string(), "ready".to_string()],
states: vec!["open".to_string()],
limit: Some(1),
cursor: None,
})
@@ -3625,7 +3606,7 @@ VALUES ('workspace-test', 'ticket', 4);
assert_eq!(summary_page.items.len(), 1);
assert!(summary_page.page.has_more);
let mismatched_summary_cursor = authority.list_ticket_page(TicketListProjectionRequest {
states: vec!["done".to_string()],
states: vec!["closed".to_string()],
limit: Some(1),
cursor: summary_page.page.next_cursor,
});
@@ -3694,7 +3675,7 @@ VALUES ('workspace-test', 'ticket', 4);
let objective = authority.objective("00000000001J3").unwrap();
assert!(objective.body.contains("Objective body"));
assert_eq!(objective.linked_ticket_summaries[0].id, "00000000001J2");
assert_eq!(objective.linked_ticket_summaries[0].state, "ready");
assert_eq!(objective.linked_ticket_summaries[0].state, "open");
let linked_ticket_key = objective.linked_ticket_summaries[0].resource_key.clone();
let objective_query = authority
.query_objectives(ObjectiveQueryRequest {
@@ -3783,8 +3764,8 @@ INSERT INTO typed_tickets (
workspace_id, ticket_id, slug, title, status, kind, priority, body,
workflow_state, workflow_state_explicit
) VALUES
('workspace-test', '00000000001J2', 'ticket-j2', 'Ticket J2', 'open', 'task', 'normal', '', 'planning', 1),
('workspace-test', '00000000001J3', 'ticket-j3', 'Ticket J3', 'open', 'task', 'normal', '', 'planning', 1);
('workspace-test', '00000000001J2', 'ticket-j2', 'Ticket J2', 'open', 'task', 'normal', '', 'open', 1),
('workspace-test', '00000000001J3', 'ticket-j3', 'Ticket J3', 'open', 'task', 'normal', '', 'open', 1);
INSERT INTO workspace_resource_keys (
workspace_id, resource_kind, resource_id, sequence, resource_key, allocated_at
) VALUES
crates/workspace-server/src/latest_schema.sql
+111
View File
@@ -1689,3 +1689,114 @@ CREATE TABLE repository_secret_legacy_receipts (
FOREIGN KEY (workspace_id, operation_id)
REFERENCES repository_secret_operations(workspace_id, operation_id) ON DELETE CASCADE
);
-- T715: responsibility is retained; closing irreversibly releases an assignment identity.
CREATE TABLE IF NOT EXISTS 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
);
-- Ticket case assignment work constraints (schema 90).
DROP VIEW IF EXISTS ticket_active_worker_assignments;
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 = 'open'
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);
DROP TRIGGER IF EXISTS ticket_terminal_releases_work;
CREATE TRIGGER ticket_terminal_releases_work
AFTER UPDATE OF workflow_state ON typed_tickets
WHEN NEW.workflow_state = '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;
DROP TRIGGER IF EXISTS ticket_terminal_assignment_releases_work_insert;
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 = 'closed')
BEGIN
INSERT OR IGNORE INTO ticket_assignment_work_releases
VALUES (NEW.workspace_id, NEW.assignment_id, NEW.updated_at);
END;
DROP TRIGGER IF EXISTS ticket_terminal_assignment_releases_work_update;
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 = 'closed')
BEGIN
INSERT OR IGNORE INTO ticket_assignment_work_releases
VALUES (NEW.workspace_id, NEW.assignment_id, NEW.updated_at);
END;
DROP TRIGGER IF EXISTS ticket_active_worker_role_insert;
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 = 'open')
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;
DROP TRIGGER IF EXISTS ticket_active_worker_role_update;
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 = 'open')
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.
DROP TRIGGER IF EXISTS ticket_active_worker_principal_insert;
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;
DROP TRIGGER IF EXISTS ticket_active_worker_principal_update;
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;
DROP TRIGGER IF EXISTS ticket_unfinished_work_blocks_worker_delete;
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;
DROP TRIGGER IF EXISTS ticket_unfinished_work_blocks_worker_move;
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/legacy_evidence_column_tests.rs
+1 -1
View File
@@ -48,7 +48,7 @@ fn evidence_column_cutover_preserves_all_values_and_current_schema_names() {
let upgraded = SqliteWorkspaceStore::open(&path).unwrap();
upgraded
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 89);
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
assert_eq!(evidence_rows(conn, "artifacts"), artifacts);
assert_eq!(
evidence_rows(conn, "repository_secret_legacy_receipts"),
crates/workspace-server/src/retention.rs
+133 -131
View File
@@ -1136,7 +1136,7 @@ mod tests {
)?;
c.execute(
"INSERT INTO typed_tickets (workspace_id, ticket_id, slug, title, status, kind, priority, body, workflow_state, workflow_state_explicit) \
VALUES ('w', 'ticket', 'ticket', 'Ticket', 'open', 'task', 'normal', '', 'planning', 1)",
VALUES ('w', 'ticket', 'ticket', 'Ticket', 'open', 'task', 'normal', '', 'open', 1)",
[],
)?;
Ok(())
@@ -1398,7 +1398,7 @@ mod tests {
})
.unwrap();
s.with_conn(|conn| {
conn.execute("INSERT INTO typed_tickets(workspace_id,ticket_id,slug,title,status,kind,priority,body,workflow_state,workflow_state_explicit) VALUES('w','ticket-old','ticket-old','Old Ticket','open','task','normal','','planning',1)", [])?;
conn.execute("INSERT INTO typed_tickets(workspace_id,ticket_id,slug,title,status,kind,priority,body,workflow_state,workflow_state_explicit) VALUES('w','ticket-old','ticket-old','Old Ticket','open','task','normal','','open',1)", [])?;
conn.execute("INSERT INTO worker_registry(workspace_id,worker_id,runtime_id,display_name,profile,retention_state,created_at,updated_at) VALUES('w','1','r','old worker','builtin:coder','normal','created','2026-08-24T00:00:00Z')", [])?;
conn.execute("INSERT INTO worker_create_reservations(workspace_id,allocation_key,worker_id,runtime_id,create_fingerprint,state,created_at,updated_at) VALUES('w','allocation-old',?1,'r','fingerprint','created','created','created')", [worker_id().to_string()])?;
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at) VALUES('w','ticket-old','assignment-old','r','1','test','t')", [])?;
@@ -1522,24 +1522,38 @@ mod tests {
install_current_assignment(s, "assignment").unwrap();
}
fn reopened_terminal_assignment_store(terminal_state: &str) -> SqliteWorkspaceStore {
fn reopened_terminal_assignment_store() -> SqliteWorkspaceStore {
let store = setup();
seed_current_assignment(&store);
store
.with_conn(|conn| {
conn.execute(
"UPDATE typed_tickets SET workflow_state=?1
"UPDATE typed_tickets SET status='closed', workflow_state='closed'
WHERE workspace_id='w' AND ticket_id='ticket'",
[terminal_state],
[],
)?;
conn.execute(
"UPDATE typed_tickets SET workflow_state='planning'
"UPDATE typed_tickets SET status='open', workflow_state='open'
WHERE workspace_id='w' AND ticket_id='ticket'",
[],
)?;
Ok(())
})
.unwrap();
// Narrow lifecycle-schema evidence: closing durably releases work,
// while reopening retains routing without reviving that assignment.
let released: i64 = store
.with_conn(|conn| {
conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_work_releases
WHERE workspace_id='w' AND assignment_id='assignment'",
[],
|row| row.get(0),
)
.map_err(StoreError::from)
})
.unwrap();
assert_eq!(released, 1);
store
}
@@ -1646,103 +1660,93 @@ mod tests {
#[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
for reopen in [false, true] {
let store = setup();
seed_current_assignment(&store);
store
.with_conn(|conn| {
conn.execute(
"UPDATE typed_tickets SET status='closed', workflow_state='closed'
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'
[],
)?;
if reopen {
conn.execute(
"UPDATE typed_tickets SET status='open', workflow_state='open'
WHERE workspace_id='w' AND ticket_id='ticket'",
[],
)?;
}
Ok(())
})
.unwrap();
store
.update_worker_retention_policy(
"w",
&store
.worker_retention_policy("w")
.unwrap()
.unwrap()
.policy_digest,
&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(),
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)
[],
)?;
}
Ok(())
})
.unwrap();
store
.update_worker_retention_policy(
"w",
&store
.worker_retention_policy("w")
.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
.unwrap()
.policy_digest,
&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(), "closed case, 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(),
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
[],
|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"
);
}
[],
|row| row.get(0),
)?,
))
})
.unwrap();
assert_eq!(
(retained, historical),
(1, 1),
"assignment history must survive removal"
);
}
}
@@ -1768,42 +1772,40 @@ mod tests {
#[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, .. }
let store = reopened_terminal_assignment_store();
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()
);
}
);
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 store = reopened_terminal_assignment_store();
let plan = store.plan_worker_removal(&req(), &inv()).unwrap();
store
.prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint)
@@ -1837,7 +1839,7 @@ mod tests {
#[test]
fn commit_rechecks_distinct_assignment_independently_of_admission_fence() {
let store = reopened_terminal_assignment_store("done");
let store = reopened_terminal_assignment_store();
let plan = store.plan_worker_removal(&req(), &inv()).unwrap();
store
.prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint)
@@ -1938,7 +1940,7 @@ mod tests {
store
.with_conn(|conn| {
conn.execute(
"UPDATE typed_tickets SET workflow_state='done'
"UPDATE typed_tickets SET status='closed', workflow_state='closed'
WHERE workspace_id='w' AND ticket_id='ticket'",
[],
)?;
crates/workspace-server/src/server.rs
+334 -647
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/server/tests/api_logging_tests.rs
+29 -5
View File
@@ -131,8 +131,7 @@ async fn generated_cancellation_errors_preserve_response_and_log_details() {
})
.unwrap();
let backend = browser_ticket_backend(&api).unwrap();
let mut input = ticket::NewTicket::new("Cancellation logging");
input.workflow_state = Some(TicketWorkflowState::InProgress);
let input = ticket::NewTicket::new("Cancellation logging");
let ticket = backend.create(input).unwrap();
api.store
.set_current_ticket_worker_assignment(
@@ -197,8 +196,8 @@ async fn generated_cancellation_errors_preserve_response_and_log_details() {
"{diagnostics}"
);
assert_eq!(
api.authority.ticket(&ticket.id).unwrap().state,
"inprogress"
api.authority.ticket(&ticket.id).unwrap().state.as_str(),
"open"
);
// Errors without diagnostics still carry a useful kind and message.
@@ -213,7 +212,32 @@ async fn generated_cancellation_errors_preserve_response_and_log_details() {
*execution.cancel_failure.lock().unwrap() = None;
let (status, body, events) = request_with_logs(app, request("body-secret")).await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["state"], "ready");
assert_eq!(body["state"], "open");
assert!(
api.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket.id)
.unwrap()
.is_none()
);
assert!(
api.store
.get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket.id)
.unwrap()
.is_none()
);
let assignment_history = api
.store
.list_ticket_worker_assignment_events(TEST_WORKSPACE_ID, &ticket.id, 10)
.unwrap();
assert!(
assignment_history
.iter()
.any(|event| event.action == "unassigned"
&& event.previous_assignment_id.as_deref() == Some("logging-assignment"))
);
let after = backend.show(ticket.id.into()).unwrap();
assert!(after.case_history.is_empty());
assert!(after.current_closure.is_none());
assert!(
events.is_empty(),
"successful cancellation logged as failure: {events:?}"
crates/workspace-server/src/server/tests/ticket_evidence_tests.rs
+353 -93
View File
@@ -47,7 +47,6 @@ impl EvidenceApiFixture {
assert!(!temp.path().join(".git").exists());
let backend = browser_ticket_backend(&api).unwrap();
let mut input = ticket::NewTicket::new("T716 remote immutable evidence");
input.workflow_state = Some(TicketWorkflowState::InProgress);
input.targets = vec![ticket::TicketTarget {
repository_key: "test-repository".into(),
ref_selector: Some("develop".into()),
@@ -191,12 +190,16 @@ impl EvidenceApiFixture {
.unwrap()
}
fn completion(&self, _approval: &ReviewEvent) -> ticket::TicketCompletion {
ticket::TicketCompletion {
operation_key: "evidence-ticket-completion".into(),
fn closure(&self) -> ticket::TicketCaseUpdate {
ticket::TicketCaseUpdate {
operation_key: "evidence-ticket-closure".into(),
expected_content_digest: self.content_digest(),
expected_state: TicketWorkflowState::InProgress,
reason: "Implementation judged complete independently of MR attestation".into(),
expected_state: ticket::TicketState::Open,
expected_closure_operation_key: None,
action: ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::Completed,
},
explanation: "Implementation judged complete independently of MR attestation".into(),
references: Vec::new(),
author: Some("fixture".into()),
}
@@ -297,14 +300,14 @@ impl EvidenceApiFixture {
}
#[tokio::test]
async fn done_merged_remote_ticket_uses_immutable_result_in_show_query_list_and_paged_detail() {
async fn closed_merged_remote_ticket_uses_immutable_result_in_show_query_list_and_paged_detail() {
let fixture = EvidenceApiFixture::new().await;
let approval = fixture.approve("integration");
let merge = fixture.integrate(&approval);
let content_digest = fixture.content_digest();
let completion = fixture
.backend
.complete(&fixture.ticket_id, fixture.completion(&approval))
.update_case(&fixture.ticket_id, fixture.closure())
.unwrap();
assert_eq!(ticket::ticket_content_digest(&completion), content_digest);
assert!(
@@ -325,7 +328,7 @@ async fn done_merged_remote_ticket_uses_immutable_result_in_show_query_list_and_
);
let shown = fixture.show().await;
assert_eq!(shown.state, "done");
assert_eq!(shown.state.as_str(), "closed");
assert_eq!(shown.content_digest, content_digest);
assert!(
shown.evidence.complete_for_integration,
@@ -395,15 +398,15 @@ async fn latest_postmerge_approval_requires_exact_content_and_stored_source_resu
.contains(&"review_after_rescope".into())
);
fixture.assert_filters(false, true, true).await;
let mut outdated_completion = fixture.completion(&integration);
outdated_completion.expected_content_digest = integration.ticket_content_digest.clone();
let mut outdated_closure = fixture.closure();
outdated_closure.expected_content_digest = integration.ticket_content_digest.clone();
assert!(
fixture
.backend
.complete(&fixture.ticket_id, outdated_completion)
.update_case(&fixture.ticket_id, outdated_closure)
.is_err()
);
assert_eq!(fixture.show().await.state, "inprogress");
assert_eq!(fixture.show().await.state.as_str(), "open");
let latest = fixture.approve("latest-content");
assert_eq!(latest.created_at, integration.created_at);
@@ -467,7 +470,7 @@ async fn latest_postmerge_approval_requires_exact_content_and_stored_source_resu
assert!(wrong.evidence.approved_current_subject);
assert!(!wrong.evidence.complete_for_integration);
fixture.assert_filters(false, true, true).await;
assert_eq!(fixture.show().await.state, "inprogress");
assert_eq!(fixture.show().await.state.as_str(), "open");
let repaired = fixture.approve("exact-snapshot");
fixture.assert_filters(true, false, false).await;
@@ -488,10 +491,10 @@ async fn latest_postmerge_approval_requires_exact_content_and_stored_source_resu
assert_eq!(detail.target.resolved_ref.as_deref(), Some(EVIDENCE_TARGET));
let completion = fixture
.backend
.complete(&fixture.ticket_id, fixture.completion(&repaired))
.update_case(&fixture.ticket_id, fixture.closure())
.unwrap();
assert_eq!(ticket::ticket_content_digest(&completion), content_digest);
assert_eq!(fixture.show().await.state, "done");
assert_eq!(fixture.show().await.state.as_str(), "closed");
fixture.assert_filters(true, false, false).await;
}
@@ -645,14 +648,18 @@ async fn closing_ticket_with_open_unreviewed_mr_does_not_change_its_proof_or_sta
.unwrap();
fixture
.backend
.update_state(
.update_case(
&fixture.ticket_id,
ticket::TicketStateUpdate {
ticket::TicketCaseUpdate {
operation_key: "close-with-open-mr".into(),
expected_content_digest: fixture.content_digest(),
expected_state: TicketWorkflowState::InProgress,
state: TicketWorkflowState::Closed,
reason: "The request was withdrawn; no integration or approval is asserted".into(),
expected_state: ticket::TicketState::Open,
expected_closure_operation_key: None,
action: ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::NotPlanned,
},
explanation: "The request was withdrawn; no integration or approval is asserted"
.into(),
references: Vec::new(),
author: Some("workspace-user".into()),
},
@@ -667,13 +674,13 @@ async fn closing_ticket_with_open_unreviewed_mr_does_not_change_its_proof_or_sta
serde_json::to_value(before).unwrap()
);
let shown = fixture.show().await;
assert_eq!(shown.state, "closed");
assert_eq!(shown.state.as_str(), "closed");
assert!(!shown.evidence.approved_current_subject);
assert_eq!(shown.merge_requests[0].state, "open");
}
#[tokio::test]
async fn mrless_completion_is_a_ticket_judgment_not_approved_or_missing_merge_evidence() {
async fn mrless_completed_closure_is_a_ticket_judgment_not_approved_or_missing_merge_evidence() {
let temp = tempfile::tempdir().unwrap();
let api = test_api_with_remote_repository(temp.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
@@ -682,13 +689,17 @@ async fn mrless_completion_is_a_ticket_judgment_not_approved_or_missing_merge_ev
.unwrap();
let ticket = backend.show(reference.id.clone().into()).unwrap();
backend
.complete(
.update_case(
&reference.id,
ticket::TicketCompletion {
ticket::TicketCaseUpdate {
operation_key: "research-result".into(),
expected_content_digest: ticket::ticket_content_digest(&ticket),
expected_state: TicketWorkflowState::Planning,
reason: "Answer recorded in the Ticket thread".into(),
expected_state: ticket::TicketState::Open,
expected_closure_operation_key: None,
action: ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::Completed,
},
explanation: "Answer recorded in the Ticket thread".into(),
references: Vec::new(),
author: Some("workspace-user".into()),
},
@@ -703,7 +714,7 @@ async fn mrless_completion_is_a_ticket_judgment_not_approved_or_missing_merge_ev
.await,
)
.unwrap();
assert_eq!(shown.state, "done");
assert_eq!(shown.state.as_str(), "closed");
assert!(!shown.evidence.has_merge_request);
assert!(!shown.evidence.approved_current_subject);
assert!(!shown.evidence.complete_for_integration);
@@ -728,8 +739,22 @@ async fn mrless_completion_is_a_ticket_judgment_not_approved_or_missing_merge_ev
}
}
// Exercise the generated public API through the router, then reread the real
// Ticket authority. No live Runtime, repository provider, or Worker process runs.
fn case_request(before: &ticket::Ticket, operation_key: &str, action: Value) -> Value {
json!({
"operation_key": operation_key,
"expected_content_digest": ticket::ticket_content_digest(before),
"expected_state": before.meta.workflow_state,
"expected_closure_operation_key": before.closure_operation_key,
"action": action,
"explanation": "Explicit user case judgment",
"references": [{"kind": "ticket", "target": "T-42"}],
})
}
#[tokio::test]
async fn all_public_state_and_close_routes_require_explicit_cas_and_operation_receipts() {
async fn case_route_rejects_missing_cas_client_author_and_removed_workflow_routes() {
let temp = tempfile::tempdir().unwrap();
let api = test_api_with_remote_repository(temp.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
@@ -738,84 +763,319 @@ async fn all_public_state_and_close_routes_require_explicit_cas_and_operation_re
.unwrap();
let before = backend.show(reference.id.clone().into()).unwrap();
let app = build_inner_router(api);
let path = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{}", reference.id);
for suffix in ["workflow-state", "state-changes", "state-fields/state"] {
let path = format!(
"/workspaces/{TEST_WORKSPACE_ID}/tickets/{}/case",
reference.id
);
let request = case_request(
&before,
"close",
json!({"action": "close", "reason": "completed"}),
);
for field in [
"operation_key",
"expected_content_digest",
"expected_state",
"action",
"explanation",
] {
let mut missing = request.clone();
missing.as_object_mut().unwrap().remove(field);
request_json(
app.clone(),
"POST",
&format!("{path}/{suffix}"),
Some(json!({
"from":"planning","to":"done","reason":"old unguarded request","body":""
})),
&path,
Some(missing),
StatusCode::UNPROCESSABLE_ENTITY,
)
.await;
}
for (field, value) in [
("author", json!("spoofed-client-author")),
("expected_state", json!("done")),
("action", json!({"action": "close", "reason": "invented"})),
] {
let mut invalid = request.clone();
invalid[field] = value;
request_json(
app.clone(),
"POST",
&path,
Some(invalid),
StatusCode::UNPROCESSABLE_ENTITY,
)
.await;
}
let legacy = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{}", reference.id);
for suffix in [
"workflow-state",
"state-changes",
"state-fields/state",
"workflow/close",
"complete",
] {
// A surviving GET catch-all can produce 405 rather than 404 for an
// absent POST route. Neither status authorizes a legacy mutation.
let response = app
.clone()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("{legacy}/{suffix}"))
.header(CONTENT_TYPE, "application/json")
.body(Body::from(request.to_string()))
.unwrap(),
)
.await
.unwrap();
assert!(
matches!(
response.status(),
StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED
),
"legacy POST {suffix}: {}",
response.status()
);
}
request_json(
app,
"POST",
&format!(
"/workspaces/019d0000-0000-7000-8000-0000000000bb/tickets/{}/case",
reference.id
),
Some(request),
StatusCode::NOT_FOUND,
)
.await;
assert_eq!(backend.show(reference.id.into()).unwrap(), before);
}
#[tokio::test]
async fn case_route_preserves_corrected_history_replay_and_reopen_aba_guard_without_mr() {
let temp = tempfile::tempdir().unwrap();
let api = test_api_with_remote_repository(temp.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let reference = backend
.create(ticket::NewTicket::new(
"Case judgments without repository work",
))
.unwrap();
let before = backend.show(reference.id.clone().into()).unwrap();
let digest = ticket::ticket_content_digest(&before);
let app = build_inner_router(api);
let path = format!(
"/workspaces/{TEST_WORKSPACE_ID}/tickets/{}/case",
reference.id
);
let detail_path = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{}", reference.id);
let close = case_request(
&before,
"guarded-close",
json!({"action": "close", "reason": "completed"}),
);
let response = request_json(
app.clone(),
"POST",
&path,
Some(close.clone()),
StatusCode::OK,
)
.await;
let summary: server_api::TicketSummary = serde_json::from_value(response).unwrap();
assert_eq!(summary.0.workflow_state, ticket::TicketState::Closed);
let closed = backend.show(reference.id.clone().into()).unwrap();
assert_eq!(
closed.current_closure.as_ref().unwrap().author,
"workspace-user"
);
assert_eq!(
closed.current_closure.as_ref().unwrap().reason,
ticket::TicketClosureReason::Completed
);
assert_eq!(closed.case_history.len(), 1);
assert_eq!(
closed.case_history[0].references,
closed.current_closure.as_ref().unwrap().references
);
let correction = case_request(
&closed,
"correct-closure",
json!({"action": "correct_closure", "reason": {"other": "Answer superseded by user decision"}}),
);
request_json(
app.clone(),
"POST",
&format!("{path}/workflow/close"),
Some(json!("unguarded close")),
StatusCode::UNPROCESSABLE_ENTITY,
&path,
Some(correction.clone()),
StatusCode::OK,
)
.await;
let close = json!({
"operation_key":"guarded-close","expected_content_digest":ticket::ticket_content_digest(&before),
"expected_state":"planning","reason":"No further work required","author":"spoofed-client-author"
});
let response = app
.clone()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("{path}/workflow/close"))
.header(CONTENT_TYPE, "application/json")
.body(Body::from(close.to_string()))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NO_CONTENT);
let closed = backend.show(reference.id.clone().into()).unwrap();
assert_eq!(closed.meta.workflow_state, TicketWorkflowState::Closed);
let receipt = closed.events.last().unwrap();
assert_eq!(receipt.author.as_deref(), Some("workspace-user"));
assert_eq!(receipt.from.as_deref(), Some("planning"));
let corrected = backend.show(reference.id.clone().into()).unwrap();
assert_eq!(corrected.case_history.len(), 2);
assert_eq!(corrected.case_history[0], closed.case_history[0]);
assert_eq!(
receipt.attributes.get("operation_key"),
Some(&"guarded-close".to_string())
corrected.current_closure.as_ref().unwrap().reason,
ticket::TicketClosureReason::Other("Answer superseded by user decision".into())
);
let reopen = json!({
"operation_key":"reopen","expected_content_digest":ticket::ticket_content_digest(&closed),
"expected_state":"closed","state":"planning","reason":"A new question arrived"
});
let response = app
.clone()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("{path}/workflow-state"))
.header(CONTENT_TYPE, "application/json")
.body(Body::from(reopen.to_string()))
.unwrap(),
)
.await
assert_eq!(
corrected.closure_operation_key.as_deref(),
Some("correct-closure")
);
let detail: TicketDetail =
serde_json::from_value(get_json(app.clone(), &detail_path).await).unwrap();
assert_eq!(detail.current_closure, corrected.current_closure);
assert_eq!(detail.closure_history, corrected.case_history);
assert_eq!(
detail.closure_operation_key,
corrected.closure_operation_key
);
assert!(detail.legacy_handoff.is_none());
assert!(detail.merge_requests.is_empty());
// Same state and content are insufficient after correcting a closure.
let stale_correction = case_request(
&closed,
"stale-correction",
json!({"action": "correct_closure", "reason": "duplicate"}),
);
request_json(
app.clone(),
"POST",
&path,
Some(stale_correction),
StatusCode::CONFLICT,
)
.await;
let reopen = case_request(&corrected, "reopen", json!({"action": "reopen"}));
request_json(
app.clone(),
"POST",
&path,
Some(reopen.clone()),
StatusCode::OK,
)
.await;
let reopened = backend.show(reference.id.clone().into()).unwrap();
assert_eq!(reopened.meta.workflow_state, ticket::TicketState::Open);
assert!(reopened.current_closure.is_none());
assert_eq!(
reopened.closure_operation_key,
corrected.closure_operation_key
);
assert_eq!(reopened.case_history.len(), 3);
assert_eq!(ticket::ticket_content_digest(&reopened), digest);
// Exact replay returns current authority, not the old closed response.
for replay in [close.clone(), correction, reopen] {
let result = request_json(app.clone(), "POST", &path, Some(replay), StatusCode::OK).await;
let summary: server_api::TicketSummary = serde_json::from_value(result).unwrap();
assert_eq!(summary.0.workflow_state, ticket::TicketState::Open);
assert_eq!(backend.show(reference.id.clone().into()).unwrap(), reopened);
}
let mut changed_replay = close.clone();
changed_replay["explanation"] = json!("Different judgment under the same operation key");
request_json(
app.clone(),
"POST",
&path,
Some(changed_replay),
StatusCode::CONFLICT,
)
.await;
let mut aba_close = close;
aba_close["operation_key"] = json!("stale-open-close");
request_json(
app.clone(),
"POST",
&path,
Some(aba_close),
StatusCode::CONFLICT,
)
.await;
assert_eq!(backend.show(reference.id.clone().into()).unwrap(), reopened);
let second_close = case_request(
&reopened,
"second-close",
json!({"action": "close", "reason": "not_planned"}),
);
request_json(
app.clone(),
"POST",
&path,
Some(second_close),
StatusCode::OK,
)
.await;
let final_ticket = backend.show(reference.id.into()).unwrap();
assert_eq!(
final_ticket.meta.workflow_state,
ticket::TicketState::Closed
);
assert_eq!(final_ticket.case_history.len(), 4);
assert_eq!(final_ticket.case_history[..3], reopened.case_history);
assert_eq!(
final_ticket.closure_operation_key.as_deref(),
Some("second-close")
);
assert_eq!(ticket::ticket_content_digest(&final_ticket), digest);
}
#[tokio::test]
async fn case_route_rejects_stale_content_state_and_invalid_actions_without_audit_side_effects() {
let temp = tempfile::tempdir().unwrap();
let api = test_api_with_remote_repository(temp.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let reference = backend
.create(ticket::NewTicket::new("CAS decision"))
.unwrap();
assert_eq!(response.status(), StatusCode::NO_CONTENT);
// Exact replay of the old close is a receipt lookup, not another close.
let response = app
.oneshot(
Request::builder()
.method("POST")
.uri(format!("{path}/workflow/close"))
.header(CONTENT_TYPE, "application/json")
.body(Body::from(close.to_string()))
.unwrap(),
let before = backend.show(reference.id.clone().into()).unwrap();
let app = build_inner_router(api);
let path = format!(
"/workspaces/{TEST_WORKSPACE_ID}/tickets/{}/case",
reference.id
);
let close = case_request(
&before,
"close",
json!({"action": "close", "reason": "unknown"}),
);
for (field, value) in [
("expected_content_digest", json!("stale-content")),
("expected_state", json!("closed")),
("action", json!({"action": "reopen"})),
(
"action",
json!({"action": "correct_closure", "reason": "duplicate"}),
),
] {
let mut invalid = close.clone();
invalid[field] = value;
request_json(
app.clone(),
"POST",
&path,
Some(invalid),
StatusCode::CONFLICT,
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NO_CONTENT);
let reopened = backend.show(reference.id.into()).unwrap();
assert_eq!(reopened.meta.workflow_state, TicketWorkflowState::Planning);
assert_eq!(reopened.events.len(), closed.events.len() + 1);
.await;
assert_eq!(backend.show(reference.id.clone().into()).unwrap(), before);
}
request_json(app.clone(), "POST", &path, Some(close), StatusCode::OK).await;
let closed = backend.show(reference.id.clone().into()).unwrap();
let duplicate_close = case_request(
&closed,
"duplicate-close",
json!({"action": "close", "reason": "completed"}),
);
request_json(
app,
"POST",
&path,
Some(duplicate_close),
StatusCode::CONFLICT,
)
.await;
assert_eq!(backend.show(reference.id.into()).unwrap(), closed);
}
crates/workspace-server/src/server/tests/ticket_worker_claim_tests.rs
+24 -33
View File
@@ -42,17 +42,7 @@ async fn explicit_existing_worker_claim_accepts_no_resources_and_preserves_assig
},
)
.unwrap();
backend
.set_workflow_state(
fixture.ticket_id.clone().into(),
TicketStateChange::new(
"ready",
"planning",
"Explicit investigation",
"No repository resources required",
),
)
.unwrap();
let before = backend.show(fixture.ticket_id.clone().into()).unwrap();
let request = manual_worker_assignment_request(&fixture, "claim-without-resources");
assert!(request.workdir_bindings.is_empty());
let Json(first) = claim(&fixture, request.clone()).await.unwrap();
@@ -67,8 +57,16 @@ async fn explicit_existing_worker_claim_accepts_no_resources_and_preserves_assig
.authority
.ticket(&fixture.ticket_id)
.unwrap()
.state,
"inprogress"
.state
.as_str(),
"open"
);
let after = backend.show(fixture.ticket_id.clone().into()).unwrap();
assert_eq!(after.meta.workflow_state, before.meta.workflow_state);
assert_eq!(after.case_history, before.case_history);
assert_eq!(
ticket::ticket_content_digest(&after),
ticket::ticket_content_digest(&before)
);
assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 1);
assert!(
@@ -282,8 +280,9 @@ async fn ambiguous_existing_worker_restore_keeps_claim_pending_until_accepted_re
.authority
.ticket(&fixture.ticket_id)
.unwrap()
.state,
"ready"
.state
.as_str(),
"open"
);
assert!(
fixture
@@ -313,8 +312,9 @@ async fn ambiguous_existing_worker_restore_keeps_claim_pending_until_accepted_re
.authority
.ticket(&fixture.ticket_id)
.unwrap()
.state,
"inprogress"
.state
.as_str(),
"open"
);
assert_eq!(*fixture.runtime.restore_attempts.lock().unwrap(), 3);
assert_eq!(
@@ -435,8 +435,7 @@ async fn pending_manual_worker_claim_reconciles_same_runtime_restore_and_commits
.is_empty()
);
let backend = browser_ticket_backend(&api).unwrap();
let mut input = ticket::NewTicket::new("Manual investigation without resources");
input.workflow_state = Some(TicketWorkflowState::Ready);
let input = ticket::NewTicket::new("Manual investigation without resources");
let ticket_id = backend.create(input).unwrap().id;
backend
.edit_item(
@@ -447,17 +446,6 @@ async fn pending_manual_worker_claim_reconciles_same_runtime_restore_and_commits
},
)
.unwrap();
backend
.set_workflow_state(
ticket_id.clone().into(),
TicketStateChange::new(
"ready",
"planning",
"Explicit investigation",
"No repository resources required",
),
)
.unwrap();
let admitted_content_digest = api.authority.ticket(&ticket_id).unwrap().content_digest;
let request = server_api::SetTicketRoleAssignmentRequest {
operation_id: "claim-runtime-reconciliation".into(),
@@ -503,7 +491,10 @@ async fn pending_manual_worker_claim_reconciles_same_runtime_restore_and_commits
assert!(pending.assignment_id.is_none());
assert_eq!(pending.worker, Some(worker.clone()));
assert!(pending.request_fingerprint.is_some());
assert_eq!(api.authority.ticket(&ticket_id).unwrap().state, "planning");
assert_eq!(
api.authority.ticket(&ticket_id).unwrap().state.as_str(),
"open"
);
assert!(
api.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket_id)
@@ -601,8 +592,8 @@ async fn pending_manual_worker_claim_reconciles_same_runtime_restore_and_commits
);
assert_eq!(committed.request_fingerprint, pending.request_fingerprint);
assert_eq!(
api.authority.ticket(&ticket_id).unwrap().state,
"inprogress"
api.authority.ticket(&ticket_id).unwrap().state.as_str(),
"open"
);
assert_eq!(
api.store
crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs
+204 -38
View File
@@ -288,17 +288,36 @@ impl LaunchFixture {
fixture
}
fn ticket(&self, state: TicketWorkflowState) -> String {
let mut input =
ticket::NewTicket::new(format!("Explicit Worker launch from {}", state.as_str()));
input.workflow_state = Some(state);
fn ticket(&self) -> String {
browser_ticket_backend(&self.api)
.unwrap()
.create(input)
.create(ticket::NewTicket::new(
"Explicit Worker launch on an open case",
))
.unwrap()
.id
}
fn case_action(&self, ticket_id: &str, operation_key: &str, action: ticket::TicketCaseAction) {
let backend = browser_ticket_backend(&self.api).unwrap();
let before = backend.show(ticket_id.into()).unwrap();
backend
.update_case(
ticket_id,
ticket::TicketCaseUpdate {
expected_content_digest: ticket::ticket_content_digest(&before),
expected_state: before.meta.workflow_state,
expected_closure_operation_key: before.closure_operation_key,
operation_key: operation_key.into(),
action,
explanation: "Explicit fixture case judgment".into(),
references: Vec::new(),
author: Some("fixture".into()),
},
)
.unwrap();
}
fn payload(&self, ticket_id: &str, operation_id: &str, profile: &str, flow: bool) -> Value {
self.backend.publish_profile(&self.api, profile);
let mut segments = vec![
@@ -382,24 +401,40 @@ impl LaunchFixture {
Some(current.assignment_id.as_str())
);
assert_eq!(
self.api.authority.ticket(ticket_id).unwrap().state,
"inprogress"
self.api.authority.ticket(ticket_id).unwrap().state.as_str(),
"open"
);
current
}
}
#[tokio::test(flavor = "multi_thread")]
async fn embedded_explicit_worker_starts_each_active_state_without_targets_flow_or_orchestrator() {
async fn embedded_explicit_worker_starts_open_cases_without_phase_transitions_or_queue_requirement()
{
let fixture = LaunchFixture::new(false).await;
for state in [
TicketWorkflowState::Planning,
TicketWorkflowState::Ready,
TicketWorkflowState::Queued,
TicketWorkflowState::InProgress,
] {
let ticket_id = fixture.ticket(state);
let operation_id = format!("explicit-{}", state.as_str());
for context in ["unqueued", "queued", "reopened"] {
let ticket_id = fixture.ticket();
let backend = browser_ticket_backend(&fixture.api).unwrap();
if context == "queued" {
backend
.queue_ready(ticket_id.clone().into(), "fixture")
.unwrap();
} else if context == "reopened" {
fixture.case_action(
&ticket_id,
"fixture-close",
ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::NotPlanned,
},
);
fixture.case_action(
&ticket_id,
"fixture-reopen",
ticket::TicketCaseAction::Reopen,
);
}
let before = backend.show(ticket_id.clone().into()).unwrap();
let operation_id = format!("explicit-{context}");
assert!(
fixture
.api
@@ -414,8 +449,16 @@ async fn embedded_explicit_worker_starts_each_active_state_without_targets_flow_
);
let payload = fixture.payload(&ticket_id, &operation_id, "builtin:companion", false);
let (status, body) = fixture.post(TEST_WORKSPACE_ID, payload).await;
assert_eq!(status, StatusCode::OK, "{}: {body}", state.as_str());
assert_eq!(status, StatusCode::OK, "{context}: {body}");
let assignment = fixture.assert_assigned(&ticket_id, &operation_id, &body);
let after = backend.show(ticket_id.clone().into()).unwrap();
assert_eq!(after.meta.workflow_state, before.meta.workflow_state);
assert_eq!(after.case_history, before.case_history);
assert_eq!(after.closure_operation_key, before.closure_operation_key);
assert_eq!(
ticket::ticket_content_digest(&after),
ticket::ticket_content_digest(&before)
);
assert_eq!(body["worker"]["profile"], "builtin:companion", "{body}");
let creates = fixture.backend.creates_for(&assignment.worker.worker_id);
assert_eq!(creates.len(), 1);
@@ -434,7 +477,7 @@ async fn embedded_explicit_worker_starts_each_active_state_without_targets_flow_
async fn real_runtime_http_worker_launch_preserves_optional_flow_profile_submit_and_assignment() {
let fixture = LaunchFixture::new(true).await;
for (profile, flow) in [("builtin:intake", false), ("builtin:coder", true)] {
let ticket_id = fixture.ticket(TicketWorkflowState::Planning);
let ticket_id = fixture.ticket();
let operation_id = format!("real-http-{flow}");
let payload = fixture.payload(&ticket_id, &operation_id, profile, flow);
let expected_submit = payload["initial_submit"].clone();
@@ -510,7 +553,7 @@ async fn worker_launch_uses_published_project_profile_and_workspace_default_on_r
.commit_evaluated_workspace_config(TEST_WORKSPACE_ID, &candidate)
.unwrap();
for explicit in [true, false] {
let ticket_id = fixture.ticket(TicketWorkflowState::Ready);
let ticket_id = fixture.ticket();
let operation_id = format!("project-profile-{explicit}");
let mut payload = fixture.payload(&ticket_id, &operation_id, "project:research", false);
if !explicit {
@@ -532,21 +575,35 @@ async fn worker_launch_uses_published_project_profile_and_workspace_default_on_r
}
#[tokio::test(flavor = "multi_thread")]
async fn completed_or_closed_ticket_launch_rejects_before_runtime_or_assignment_reservation() {
async fn closed_ticket_launch_rejects_before_runtime_or_assignment_reservation() {
let fixture = LaunchFixture::new(false).await;
for state in [TicketWorkflowState::Done, TicketWorkflowState::Closed] {
let ticket_id = fixture.ticket(state);
let operation_id = format!("terminal-{}", state.as_str());
for (label, reason) in [
("completed", ticket::TicketClosureReason::Completed),
("not-planned", ticket::TicketClosureReason::NotPlanned),
] {
let ticket_id = fixture.ticket();
fixture.case_action(
&ticket_id,
"fixture-close",
ticket::TicketCaseAction::Close { reason },
);
let operation_id = format!("terminal-{label}");
let payload = fixture.payload(&ticket_id, &operation_id, "builtin:companion", false);
let (status, body) = fixture.post(TEST_WORKSPACE_ID, payload).await;
assert_eq!(status, StatusCode::CONFLICT, "{body}");
assert!(
body["message"].as_str().unwrap().contains(state.as_str()),
body["message"].as_str().unwrap().contains("closed"),
"{body}"
);
assert_eq!(
fixture.api.authority.ticket(&ticket_id).unwrap().state,
state.as_str()
fixture
.api
.authority
.ticket(&ticket_id)
.unwrap()
.state
.as_str(),
"closed"
);
assert!(
fixture
@@ -572,7 +629,7 @@ async fn completed_or_closed_ticket_launch_rejects_before_runtime_or_assignment_
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_exact_launch_replays_keep_one_worker_and_committed_assignment() {
let fixture = Arc::new(LaunchFixture::new(false).await);
let ticket_id = fixture.ticket(TicketWorkflowState::Planning);
let ticket_id = fixture.ticket();
let payload = fixture.payload(&ticket_id, "concurrent-launch", "builtin:companion", false);
let barrier = Arc::new(tokio::sync::Barrier::new(5));
let mut tasks = Vec::new();
@@ -617,7 +674,7 @@ async fn concurrent_exact_launch_replays_keep_one_worker_and_committed_assignmen
#[tokio::test(flavor = "multi_thread")]
async fn undispatched_launch_rejection_settles_claim_and_allows_corrected_operation() {
let fixture = LaunchFixture::new(false).await;
let ticket_id = fixture.ticket(TicketWorkflowState::Planning);
let ticket_id = fixture.ticket();
let mut rejected = fixture.payload(&ticket_id, "invalid-resources", "builtin:companion", false);
rejected["workdir_attachments"] =
json!([{"alias":"checkout","working_directory_id":"missing-workdir"}]);
@@ -650,7 +707,7 @@ async fn undispatched_launch_rejection_settles_claim_and_allows_corrected_operat
#[tokio::test(flavor = "multi_thread")]
async fn worker_launch_retry_reuses_assignment_and_does_not_repeat_initial_submit() {
let fixture = LaunchFixture::new(false).await;
let ticket_id = fixture.ticket(TicketWorkflowState::Ready);
let ticket_id = fixture.ticket();
let payload = fixture.payload(&ticket_id, "worker-retry", "builtin:companion", false);
let (status, first) = fixture.post(TEST_WORKSPACE_ID, payload.clone()).await;
assert_eq!(status, StatusCode::OK, "{first}");
@@ -703,7 +760,7 @@ async fn worker_launch_retry_reuses_assignment_and_does_not_repeat_initial_submi
async fn generic_ticket_worker_unknown_input_outcome_preserves_assignment_and_launch_retry_identity()
{
let fixture = LaunchFixture::new(false).await;
let ticket_id = fixture.ticket(TicketWorkflowState::Planning);
let ticket_id = fixture.ticket();
let payload = fixture.payload(
&ticket_id,
"worker-unknown-input",
@@ -766,7 +823,7 @@ async fn generic_ticket_worker_unknown_input_outcome_preserves_assignment_and_la
#[tokio::test(flavor = "multi_thread")]
async fn scoped_worker_launch_rejects_foreign_workspace_and_unpublished_profile_with_body() {
let fixture = LaunchFixture::new(false).await;
let ticket_id = fixture.ticket(TicketWorkflowState::Planning);
let ticket_id = fixture.ticket();
let payload = fixture.payload(&ticket_id, "scope-rejected", "builtin:companion", false);
let (status, body) = fixture
.post("019d0000-0000-7000-8000-0000000000bb", payload.clone())
@@ -925,10 +982,10 @@ async fn read_only_ticket_targets_allow_extra_external_grants_without_widening_c
}
#[tokio::test(flavor = "multi_thread")]
async fn worker_assignment_finalize_failure_compensates_real_embedded_runtime_and_preserves_planning()
async fn worker_assignment_finalize_failure_compensates_real_embedded_runtime_and_preserves_open_case()
{
let fixture = LaunchFixture::new(false).await;
let ticket_id = fixture.ticket(TicketWorkflowState::Planning);
let ticket_id = fixture.ticket();
rusqlite::Connection::open(&fixture.api.config.database_path)
.unwrap()
.execute_batch(
@@ -987,8 +1044,14 @@ async fn worker_assignment_finalize_failure_compensates_real_embedded_runtime_an
.is_none()
);
assert_eq!(
fixture.api.authority.ticket(&ticket_id).unwrap().state,
"planning"
fixture
.api
.authority
.ticket(&ticket_id)
.unwrap()
.state
.as_str(),
"open"
);
let counts: (i64, i64) = fixture.api.config_store.with_conn(|conn| Ok((
conn.query_row("SELECT COUNT(*) FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", rusqlite::params![TEST_WORKSPACE_ID, fixture.runtime_id, worker_id], |row| row.get(0))?,
@@ -1000,7 +1063,7 @@ async fn worker_assignment_finalize_failure_compensates_real_embedded_runtime_an
#[tokio::test(flavor = "multi_thread")]
async fn choosing_reviewer_profile_does_not_register_a_reviewer_child_or_grant_mr_review() {
let fixture = LaunchFixture::new(false).await;
let ticket_id = fixture.ticket(TicketWorkflowState::Planning);
let ticket_id = fixture.ticket();
let payload = fixture.payload(&ticket_id, "ordinary-reviewer", "builtin:reviewer", false);
let (status, body) = fixture.post(TEST_WORKSPACE_ID, payload).await;
assert_eq!(status, StatusCode::OK, "{body}");
@@ -1236,7 +1299,110 @@ async fn uncertain_launch_replay_after_target_edit_keeps_claim_and_blocks_compet
.authority
.ticket(&fixture.ticket_id)
.unwrap()
.state,
"ready"
.state
.as_str(),
"open"
);
}
// Direct policy composition is intentional: this protects routing interest,
// independently of the transport/Runtime launch assertions above.
#[tokio::test(flavor = "multi_thread")]
async fn orchestrator_interest_requires_queue_receipt_or_unfinished_work_not_just_open_state() {
let fixture = LaunchFixture::new(false).await;
let backend = browser_ticket_backend(&fixture.api).unwrap();
let queued_id = fixture.ticket();
assign_test_orchestrator(&fixture.api, &queued_id);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &queued_id, "open")
.unwrap()
.is_none()
);
backend
.queue_ready(queued_id.clone().into(), "fixture")
.unwrap();
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &queued_id, "open")
.unwrap()
.is_some()
);
fixture.case_action(
&queued_id,
"queue-case-close",
ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::Completed,
},
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &queued_id, "closed")
.unwrap()
.is_none()
);
fixture.case_action(
&queued_id,
"queue-case-reopen",
ticket::TicketCaseAction::Reopen,
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &queued_id, "open")
.unwrap()
.is_none()
);
let active_id = fixture.ticket();
assign_test_orchestrator(&fixture.api, &active_id);
let payload = fixture.payload(
&active_id,
"routing-active-work",
"builtin:companion",
false,
);
let (status, body) = fixture.post(TEST_WORKSPACE_ID, payload).await;
assert_eq!(status, StatusCode::OK, "{body}");
fixture.assert_assigned(&active_id, "routing-active-work", &body);
let active = backend.show(active_id.clone().into()).unwrap();
assert!(
active
.events
.iter()
.all(|event| !event.attributes.contains_key("queue_root_ticket"))
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &active_id, "open")
.unwrap()
.is_some()
);
fixture.case_action(
&active_id,
"active-case-close",
ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::NotPlanned,
},
);
fixture.case_action(
&active_id,
"active-case-reopen",
ticket::TicketCaseAction::Reopen,
);
assert!(
fixture
.api
.store
.get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &active_id)
.unwrap()
.is_some()
);
assert!(
fixture
.api
.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &active_id)
.unwrap()
.is_none()
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &active_id, "open")
.unwrap()
.is_none()
);
}
crates/workspace-server/src/store.rs
+674 -144
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/workdir_removal.rs
+107 -95
View File
@@ -1266,9 +1266,8 @@ 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
let (store, workdir) = seeded_store().await;
store
.with_conn(|conn| {
conn.execute(
"INSERT INTO worker_registry (
@@ -1288,7 +1287,7 @@ mod tests {
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)",
'task', 'normal', '', 'open', 1)",
[],
)?;
conn.execute(
@@ -1307,59 +1306,58 @@ mod tests {
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
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='closed', workflow_state='closed'
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'
[],
)?;
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='open'
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()
);
[],
)?;
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'
@@ -1369,49 +1367,63 @@ mod tests {
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'
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
[],
)?;
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);
}
[],
|row| row.get(0),
)
.map_err(Error::from)
})
.unwrap();
assert_eq!(retained, 1);
let released: i64 = store
.with_conn(|conn| {
conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_work_releases
WHERE workspace_id='workspace-a' AND assignment_id='assignment-a'",
[],
|row| row.get(0),
)
.map_err(Error::from)
})
.unwrap();
assert_eq!(
released, 1,
"reopening and cleanup must preserve the work release"
);
}
#[tokio::test]
crates/workspace-server/src/workspace_deletion.rs
+215 -2
View File
@@ -52,6 +52,9 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
"ticket_current_worker_assignments",
"ticket_worker_assignment_events",
"ticket_worker_assignments",
"typed_ticket_case_history",
"typed_ticket_case_metadata",
"typed_ticket_legacy_workflow_snapshot",
"typed_ticket_artifacts",
"typed_ticket_event_attributes",
"typed_ticket_event_references",
@@ -457,6 +460,26 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
}
let workspace_id = operation.response.workspace_id.clone();
// Case judgments and legacy snapshots are immutable under ordinary
// Ticket operations, including Ticket deletion and FK cascades. Only
// this owner-authorized whole-Workspace purge may discard them.
// SQLite transactional DDL holds the writer lock: no concurrent writer
// can observe the suspended guards, and any error rolls back both
// data and schema. Restore the exact schema-owned definitions before
// commit; update guards remain installed throughout.
let mut case_delete_guards = Vec::new();
for (table, trigger) in [
("typed_ticket_case_history", "typed_ticket_case_history_no_delete"),
("typed_ticket_legacy_workflow_snapshot", "typed_ticket_legacy_workflow_no_delete"),
] {
let definition: String = tx.query_row(
"SELECT sql FROM sqlite_master WHERE type='trigger' AND name=?1 AND tbl_name=?2",
params![trigger, table],
|row| row.get(0),
)?;
tx.execute_batch(&format!("DROP TRIGGER \"{trigger}\";"))?;
case_delete_guards.push(definition);
}
for table in WORKSPACE_DELETION_PURGE_TABLES {
tx.execute(
&format!("DELETE FROM \"{table}\" WHERE workspace_id = ?1"),
@@ -472,6 +495,9 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
"Workspace disappeared before deletion finalized".to_string(),
));
}
for definition in case_delete_guards {
tx.execute_batch(&definition)?;
}
let now = Utc::now().to_rfc3339();
tx.execute(
"UPDATE workspace_deletion_operations
@@ -874,8 +900,10 @@ mod tests {
use tempfile::tempdir;
fn setup() -> (SqliteWorkspaceStore, String, String) {
let dir = tempdir().expect("tempdir");
let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).expect("store");
seed_workspaces(SqliteWorkspaceStore::in_memory().expect("store"))
}
fn seed_workspaces(store: SqliteWorkspaceStore) -> (SqliteWorkspaceStore, String, String) {
store.with_conn(|conn| {
let now = Utc::now().to_rfc3339();
conn.execute(
@@ -931,6 +959,191 @@ mod tests {
.expect("purge inventory");
}
#[test]
fn workspace_purge_preserves_other_cases_and_restores_immutability_after_commit_or_rollback() {
use ticket::TicketBackend;
for inject_failure in [false, true] {
let dir = tempdir().expect("tempdir");
let path = dir.path().join("server.db");
let (store, owner, workspace_id) =
seed_workspaces(SqliteWorkspaceStore::open(&path).expect("store"));
for workspace in ["workspace-a", "workspace-b"] {
let backend = ticket::SqliteTicketBackend::open_verified(&path, workspace).unwrap();
let initial = backend
.create_with_snapshot(ticket::NewTicket::new("Case evidence"))
.unwrap()
.1;
backend
.update_case(
&initial.meta.id,
ticket::TicketCaseUpdate {
expected_content_digest: ticket::ticket_content_digest(&initial),
expected_state: initial.meta.workflow_state,
expected_closure_operation_key: initial.closure_operation_key,
operation_key: "close".into(),
action: ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::Completed,
},
explanation: "Requested result satisfied".into(),
references: vec![],
author: Some("owner".into()),
},
)
.unwrap();
// Seed only the opaque retained snapshot for the deletion boundary;
// migration and legacy phase interpretation are tested by the store.
let snapshot = ticket::TicketLegacyWorkflowSnapshot {
workflow_state: "closed".into(),
workflow_state_explicit: true,
status: "closed".into(),
queued_by: None,
queued_at: None,
resolution: None,
};
store
.with_conn(|conn| {
conn.execute(
"INSERT INTO typed_ticket_legacy_workflow_snapshot
(workspace_id,ticket_id,snapshot_json) VALUES (?1,?2,?3)",
params![
workspace,
initial.meta.id,
serde_json::to_string(&snapshot).unwrap()
],
)?;
Ok(())
})
.unwrap();
}
let assert_immutable = |workspace: &str| {
store.with_conn(|conn| {
for table in ["typed_ticket_case_history", "typed_ticket_legacy_workflow_snapshot"] {
assert!(conn.execute(&format!("DELETE FROM {table} WHERE workspace_id=?1"), [workspace]).is_err());
assert!(conn.execute(&format!("UPDATE {table} SET ticket_id=ticket_id WHERE workspace_id=?1"), [workspace]).is_err());
}
assert!(conn.execute("DELETE FROM typed_tickets WHERE workspace_id=?1", [workspace]).is_err(),
"ordinary Ticket deletion must not cascade away case evidence");
Ok(())
}).unwrap();
};
assert_immutable("workspace-a");
assert_immutable("workspace-b");
let preflight = store
.workspace_deletion_preflight(&owner, &workspace_id)
.unwrap();
let operation = store
.reserve_workspace_deletion(
&owner,
&workspace_id,
&WorkspaceDeletionRequest {
operation_id: "delete-cases".into(),
expected_workspace_updated_at: preflight.expected_workspace_updated_at,
confirmation: "Alpha".into(),
},
)
.unwrap()
.operation;
let tables = [
"typed_ticket_case_history",
"typed_ticket_case_metadata",
"typed_ticket_legacy_workflow_snapshot",
];
let other_before: Vec<String> = store
.with_conn(|conn| {
tables
.iter()
.map(|table| {
conn.query_row(
&format!("SELECT * FROM {table} WHERE workspace_id='workspace-b'"),
[],
|row| row.get(row.as_ref().column_count() - 1),
)
.map_err(Into::into)
})
.collect()
})
.unwrap();
if inject_failure {
store
.with_conn(|conn| {
conn.execute_batch(
"CREATE TRIGGER reject_workspace_purge BEFORE DELETE ON workspaces
BEGIN SELECT RAISE(ABORT, 'injected purge failure'); END;",
)?;
Ok(())
})
.unwrap();
assert!(
store
.finalize_workspace_deletion(&operation.operation_id)
.is_err()
);
assert_immutable("workspace-a");
assert_immutable("workspace-b");
store
.with_conn(|conn| {
for table in tables {
let count: i64 = conn.query_row(
&format!(
"SELECT COUNT(*) FROM {table} WHERE workspace_id='workspace-a'"
),
[],
|row| row.get(0),
)?;
assert_eq!(count, 1, "rollback must restore {table}");
}
conn.execute_batch("DROP TRIGGER reject_workspace_purge;")?;
Ok(())
})
.unwrap();
assert_ne!(
store
.workspace_deletion_operation_for_recovery(&operation.operation_id)
.unwrap()
.unwrap()
.state,
WorkspaceDeletionState::Succeeded
);
}
let completed = store
.finalize_workspace_deletion(&operation.operation_id)
.unwrap();
assert_eq!(completed.state, WorkspaceDeletionState::Succeeded);
assert_eq!(
store
.finalize_workspace_deletion(&operation.operation_id)
.unwrap(),
completed
);
assert_immutable("workspace-b");
store
.with_conn(|conn| {
for (table, before) in tables.iter().zip(&other_before) {
let count: i64 = conn.query_row(
&format!(
"SELECT COUNT(*) FROM {table} WHERE workspace_id='workspace-a'"
),
[],
|row| row.get(0),
)?;
assert_eq!(count, 0, "purge must remove {table}");
let after: String = conn.query_row(
&format!("SELECT * FROM {table} WHERE workspace_id='workspace-b'"),
[],
|row| row.get(row.as_ref().column_count() - 1),
)?;
assert_eq!(
&after, before,
"other Workspace {table} evidence must remain unchanged"
);
}
Ok(())
})
.unwrap();
}
}
#[test]
fn deletion_is_idempotent_and_removes_workspace_scoped_rows() {
let (store, owner, workspace_id) = setup();