feat(ticket): expose case tags and independent assignment handling

This commit is contained in:
2026-10-12 07:33:11 +09:00
parent 56795fb92c
commit 9b0d42fdbc
79 changed files with 9326 additions and 713 deletions
@@ -124,12 +124,14 @@ WITH eligible AS (
SELECT a.ticket_id,a.assignment_id,'ticket-event:' || e.event_index,a.tracking_json,e.event_index
FROM eligible a JOIN typed_ticket_events e ON e.workspace_id=a.workspace_id AND e.ticket_id=a.ticket_id
WHERE e.event_index>a.after_ticket_event_index
-- External content/answer inputs, not case/routing/history or own self-report.
-- Ticket supports typed extension events; an answer need not be named comment.
AND e.kind NOT IN ('create','close','state_changed','status_changed','history')
-- Continuing request intent is explicit: fenced assignment answers or item content edits.
-- Ordinary comments and all case/routing/handling history are not new requests.
AND e.kind IN ('assignment_answer','item_edit')
AND (e.kind<>'assignment_answer' OR EXISTS(SELECT 1 FROM typed_ticket_event_attributes attr WHERE attr.workspace_id=e.workspace_id AND attr.ticket_id=e.ticket_id AND attr.event_index=e.event_index AND attr.key='orchestrator_assignment_id' AND attr.value=a.assignment_id))
AND e.kind NOT LIKE 'assignment_tracking%'
AND e.kind NOT LIKE 'assignment_delivery%'
AND COALESCE(e.state_field,'') NOT IN ('case','routing','assignment_tracking')
AND NOT EXISTS(SELECT 1 FROM typed_ticket_event_attributes attr WHERE attr.workspace_id=e.workspace_id AND attr.ticket_id=e.ticket_id AND attr.event_index=e.event_index AND attr.key='orchestrator_assignment_id' AND attr.value<>a.assignment_id)
)
SELECT i.ticket_id,i.assignment_id,i.input_event_id,
json_extract(i.tracking_json,'$.waiting_reason'),json_extract(i.tracking_json,'$.next_action'),d.payload_json
@@ -378,3 +380,85 @@ pub(super) fn save_operation_fence(
}
Ok(())
}
// Same input selection as delivery, plus immutable delivered history. Never
// SELECT payload/source/content into the public diagnostic projection.
const DIAGNOSTIC_INPUTS_SQL: &str = r#"
WITH inputs AS (
SELECT o.input_event_id,-1 AS input_order FROM ticket_assignment_delivery_origins o
WHERE o.workspace_id=?1 AND o.assignment_id=?3
UNION
SELECT d.input_event_id,CASE WHEN d.input_event_id LIKE 'ticket-event:%' THEN CAST(substr(d.input_event_id,14) AS INTEGER) ELSE -1 END
FROM ticket_assignment_deliveries d WHERE d.workspace_id=?1 AND d.assignment_id=?3
UNION
SELECT 'ticket-event:' || e.event_index,e.event_index
FROM ticket_assignment_delivery_origins o JOIN typed_ticket_events e ON e.workspace_id=o.workspace_id AND e.ticket_id=o.ticket_id
WHERE o.workspace_id=?1 AND o.ticket_id=?2 AND o.assignment_id=?3 AND e.event_index>o.after_ticket_event_index
AND e.kind IN ('assignment_answer','item_edit')
AND (e.kind<>'assignment_answer' OR EXISTS(SELECT 1 FROM typed_ticket_event_attributes attr WHERE attr.workspace_id=e.workspace_id AND attr.ticket_id=e.ticket_id AND attr.event_index=e.event_index AND attr.key='orchestrator_assignment_id' AND attr.value=?3))
AND e.kind NOT LIKE 'assignment_tracking%' AND e.kind NOT LIKE 'assignment_delivery%'
AND COALESCE(e.state_field,'') NOT IN ('case','routing','assignment_tracking')
-- Retain historical input identity after done/release. A later delegation's
-- origin bounds ordinary events; an explicit answer's attribute is authoritative.
AND e.event_index<=COALESCE((SELECT MIN(later.after_ticket_event_index) FROM ticket_assignment_delivery_origins later
JOIN ticket_worker_assignments la ON la.workspace_id=later.workspace_id AND la.assignment_id=later.assignment_id
JOIN ticket_worker_assignments original ON original.workspace_id=o.workspace_id AND original.assignment_id=o.assignment_id
WHERE later.workspace_id=o.workspace_id AND later.ticket_id=o.ticket_id AND la.rowid>original.rowid),9223372036854775807)
AND NOT EXISTS(SELECT 1 FROM typed_ticket_event_attributes attr WHERE attr.workspace_id=e.workspace_id AND attr.ticket_id=e.ticket_id AND attr.event_index=e.event_index AND attr.key='orchestrator_assignment_id' AND attr.value<>?3)
)
"#;
pub(super) fn diagnostics_page(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
assignment_id: &str,
after_input_event_id: Option<&str>,
after_start_operation_id: Option<&str>,
limit: usize,
) -> Result<server_api::TicketAssignmentDiagnosticsPage> {
if !(1..=100).contains(&limit) {
return Err(Error::InvalidInput(
"Diagnostic page limit must be 1..100".into(),
));
}
store.with_conn(|conn| {
// One read snapshot for current assignment, evidence and cursor validation.
let tx = conn.unchecked_transaction()?;
let assignment = ticket::sqlite_ticket_assignment_tracking(&tx,workspace_id,ticket_id,assignment_id)
.map_err(|e| Error::Store(e.to_string()))?.ok_or_else(|| Error::InvalidInput("Assignment does not belong to this Ticket".into()))?;
let known: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_assignment_delivery_origins WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3)",params![workspace_id,ticket_id,assignment_id],|r| r.get(0))?;
let actionable = match require_current(&tx,workspace_id,ticket_id,assignment_id) {
Ok(()) => true,
Err(Error::TicketAssignmentConflict(_)) => false,
Err(error) => return Err(error),
};
let after = match after_input_event_id {
None => -2,
Some(id) => tx.query_row(&format!("{DIAGNOSTIC_INPUTS_SQL} SELECT input_order FROM inputs WHERE input_event_id=?5"),params![workspace_id,ticket_id,assignment_id,actionable,id],|r| r.get::<_,i64>(0)).optional()?.ok_or_else(|| Error::InvalidInput("Input cursor does not belong to this assignment".into()))?,
};
let sql = format!("{DIAGNOSTIC_INPUTS_SQL} SELECT i.input_event_id,d.delivery_id,EXISTS(SELECT 1 FROM ticket_assignment_delivery_requests req WHERE req.workspace_id=?1 AND req.delivery_id=d.delivery_id),EXISTS(SELECT 1 FROM ticket_assignment_delivery_acknowledgements ack WHERE ack.workspace_id=?1 AND ack.delivery_id=d.delivery_id) FROM inputs i LEFT JOIN ticket_assignment_deliveries d ON d.workspace_id=?1 AND d.assignment_id=?3 AND d.input_event_id=i.input_event_id WHERE i.input_order>?5 ORDER BY i.input_order LIMIT ?6");
let mut stmt = tx.prepare(&sql)?;
let mut inputs = stmt.query_map(params![workspace_id,ticket_id,assignment_id,actionable,after,(limit+1) as i64],|r| {
let delivery_id: Option<String> = r.get(1)?;
let prepared: bool = r.get(2)?;
let ack: bool = r.get(3)?;
use server_api::TicketAssignmentInputDeliveryState::*;
let notification = if ack { RuntimeAccepted } else if prepared { RequestPersisted } else if delivery_id.is_some() { Reserved } else { NotReserved };
Ok(server_api::TicketAssignmentInputDiagnostic { input_event_id:r.get(0)?, delivery_id,notification,acceptance_outcome_unknown:prepared && !ack })
})?.collect::<std::result::Result<Vec<_>,_>>()?;
let more = inputs.len()>limit;
inputs.truncate(limit);
let next_input_event_id = if more { inputs.last().map(|i| i.input_event_id.clone()) } else { None };
let after_start = match after_start_operation_id {
None => 0,
Some(id) => tx.query_row("SELECT f.rowid FROM ticket_assignment_operation_fences f JOIN ticket_assignment_operations o ON o.workspace_id=f.workspace_id AND o.operation_id=f.operation_id WHERE f.workspace_id=?1 AND f.orchestrator_assignment_id=?3 AND o.ticket_id=?2 AND f.operation_id=?4",params![workspace_id,ticket_id,assignment_id,id],|r| r.get::<_,i64>(0)).optional()?.ok_or_else(|| Error::InvalidInput("Start cursor does not belong to this assignment".into()))?,
};
let mut stmt = tx.prepare("SELECT f.operation_id,o.claim_state,o.assignment_id,o.runtime_id,o.worker_id FROM ticket_assignment_operation_fences f JOIN ticket_assignment_operations o ON o.workspace_id=f.workspace_id AND o.operation_id=f.operation_id WHERE f.workspace_id=?1 AND o.ticket_id=?2 AND f.orchestrator_assignment_id=?3 AND f.rowid>?4 ORDER BY f.rowid LIMIT ?5")?;
let mut starts = stmt.query_map(params![workspace_id,ticket_id,assignment_id,after_start,(limit+1) as i64],|r| Ok(server_api::TicketAssignmentStartDiagnostic { operation_id:r.get(0)?,claim_state:r.get(1)?,assignment_id:r.get(2)?,runtime_id:r.get(3)?,worker_id:r.get(4)?,actual_processing_unknown:true }))?.collect::<std::result::Result<Vec<_>,_>>()?;
let more = starts.len()>limit;
starts.truncate(limit);
let next_start_operation_id = if more { starts.last().map(|i| i.operation_id.clone()) } else { None };
Ok(server_api::TicketAssignmentDiagnosticsPage { assignment_id:assignment_id.into(),current:assignment.current,input_origin_known:known,actionable,inputs,next_input_event_id,starts,next_start_operation_id })
})
}
@@ -11,6 +11,27 @@ fn delivery_store_external_event(
}).unwrap();
}
fn delivery_store_answer(store: &SqliteWorkspaceStore, key: &str) {
let current = store
.list_current_ticket_assignment_tracking("w", "tracking")
.unwrap()
.remove(0);
store
.answer_ticket_assignment(
"w",
"tracking",
&ticket::TicketAssignmentAnswer {
assignment_id: current.assignment_id,
expected_last_event_id: current.last_event_id.unwrap(),
operation_key: key.into(),
body: "Explicit continuing answer".into(),
},
&assignment_answer_source(),
"fixture-actor",
)
.unwrap();
}
fn delivery_store_reserve_initial(
store: &SqliteWorkspaceStore,
source: &TicketAssignmentTrackingSourceContext,
@@ -60,11 +81,18 @@ async fn assignment_delivery_waiting_self_reports_do_not_busy_loop_and_answers_a
Some("assignment_tracking"),
);
delivery_store_external_event(&store, 2, "comment", None);
assert!(
store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.is_empty()
);
delivery_store_answer(&store, "answer-after-comment");
let pending = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].input_event_id, "ticket-event:2");
assert_eq!(pending[0].input_event_id, "ticket-event:3");
assert_eq!(
pending[0].waiting_reason.as_deref(),
Some("Waiting for owner answer")
@@ -74,7 +102,7 @@ async fn assignment_delivery_waiting_self_reports_do_not_busy_loop_and_answers_a
Some("Read answer and decide")
);
let answer = store
.reserve_ticket_assignment_delivery("w", "a", "ticket-event:2", &source)
.reserve_ticket_assignment_delivery("w", "a", "ticket-event:3", &source)
.unwrap();
store
.ack_ticket_assignment_delivery("w", &answer.delivery_id, &source)
@@ -85,16 +113,16 @@ async fn assignment_delivery_waiting_self_reports_do_not_busy_loop_and_answers_a
.unwrap()
.is_empty()
);
// Typed extension inputs must not disappear merely because they are not
// comments; own tracking/history events remain excluded.
delivery_store_external_event(&store, 3, "assignment_tracking_updated", None);
delivery_store_external_event(&store, 4, "history", None);
delivery_store_external_event(&store, 5, "owner_answer", None);
// Explicit fenced answers remain inputs; own tracking/history and ordinary
// comments remain excluded.
delivery_store_external_event(&store, 4, "assignment_tracking_updated", None);
delivery_store_external_event(&store, 5, "history", None);
delivery_store_answer(&store, "next-explicit-answer");
let pending = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].input_event_id, "ticket-event:5");
assert_eq!(pending[0].input_event_id, "ticket-event:6");
let saved = store
.get_ticket_assignment_tracking("w", "tracking", "a")
.unwrap()
@@ -792,3 +820,216 @@ async fn assignment_startup_unfinished_dependencies_remain_diagnostics_not_store
})
.unwrap();
}
#[tokio::test]
async fn assignment_diagnostics_keep_persisted_notification_acceptance_and_history_distinct_without_payload()
{
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let initial = store
.list_pending_ticket_assignment_deliveries("w", 10)
.unwrap()
.remove(0);
let page = store
.page_ticket_assignment_diagnostics("w", "tracking", "a", None, None, 1)
.unwrap();
assert!(page.current && page.actionable && page.input_origin_known);
assert_eq!(
page.inputs[0].notification,
server_api::TicketAssignmentInputDeliveryState::NotReserved
);
let initial = store
.reserve_ticket_assignment_delivery(
"w",
"a",
&initial.input_event_id,
&tracking_store_source(),
)
.unwrap();
assert_eq!(
store
.page_ticket_assignment_diagnostics("w", "tracking", "a", None, None, 1)
.unwrap()
.inputs[0]
.notification,
server_api::TicketAssignmentInputDeliveryState::Reserved
);
store
.prepare_ticket_assignment_delivery_content(
"w",
&initial.delivery_id,
"SECRET RENDERED PAYLOAD",
&tracking_store_source(),
)
.unwrap();
let pending = store
.page_ticket_assignment_diagnostics("w", "tracking", "a", None, None, 1)
.unwrap();
assert_eq!(
pending.inputs[0].notification,
server_api::TicketAssignmentInputDeliveryState::RequestPersisted
);
assert!(pending.inputs[0].acceptance_outcome_unknown);
store
.ack_ticket_assignment_delivery("w", &initial.delivery_id, &tracking_store_source())
.unwrap();
delivery_store_answer(&store, "diagnostic-answer");
delivery_store_external_event(&store, 1, "history", None);
let accepted = store
.page_ticket_assignment_diagnostics("w", "tracking", "a", None, None, 1)
.unwrap();
assert_eq!(
accepted.inputs[0].notification,
server_api::TicketAssignmentInputDeliveryState::RuntimeAccepted
);
assert!(!accepted.inputs[0].acceptance_outcome_unknown);
assert_eq!(
accepted.next_input_event_id.as_deref(),
Some(initial.input_event_id.as_str())
);
let next = store
.page_ticket_assignment_diagnostics(
"w",
"tracking",
"a",
accepted.next_input_event_id.as_deref(),
None,
1,
)
.unwrap();
assert_eq!(next.inputs[0].input_event_id, "ticket-event:0");
assert!(next.next_input_event_id.is_none());
let wire = serde_json::to_string(&accepted).unwrap();
for forbidden in [
"SECRET",
"tracking-binding",
"authenticated-worker",
"payload_json",
"source_json",
"content",
] {
assert!(!wire.contains(forbidden), "{wire}");
}
store
.update_ticket_assignment_tracking(
"w",
"tracking",
&tracking_store_input("a", "event-assign-a", "done"),
&tracking_store_source(),
)
.unwrap();
let history = store
.page_ticket_assignment_diagnostics("w", "tracking", "a", None, None, 100)
.unwrap();
assert!(history.current && !history.actionable);
assert_eq!(history.inputs.len(), 2); // retained inputs are history, not fresh work
assert_eq!(
store
.page_ticket_assignment_diagnostics(
"w",
"tracking",
"a",
accepted.next_input_event_id.as_deref(),
None,
100
)
.unwrap()
.inputs[0]
.input_event_id,
"ticket-event:0"
);
assert!(
store
.page_ticket_assignment_diagnostics(
"w",
"tracking",
"a",
Some("ticket-event:999"),
None,
1
)
.is_err()
);
assert!(
store
.page_ticket_assignment_diagnostics("other", "tracking", "a", None, None, 1)
.is_err()
);
}
#[tokio::test]
async fn assignment_diagnostics_unknown_retained_relationship_does_not_invent_input_or_acceptance()
{
let (_dir, store) = tracking_store_fixture().await;
store.with_conn(|conn| {
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','tracking','legacy-unknown','orchestrator','workspace_agent','workspace-orchestrator','legacy','past')",[])?;
conn.execute("UPDATE ticket_worker_assignments SET tracking_json=NULL,tracking_updated_at=NULL,tracking_migration_diagnostic='Unknown historical relationship' WHERE assignment_id='legacy-unknown'",[])?;
conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','tracking','legacy-unknown','orchestrator','workspace_agent','workspace-orchestrator','past')",[])?;
Ok(())
}).unwrap();
delivery_store_external_event(&store, 0, "comment", None);
let page = store
.page_ticket_assignment_diagnostics("w", "tracking", "legacy-unknown", None, None, 100)
.unwrap();
assert!(page.current);
assert!(!page.actionable && !page.input_origin_known);
assert!(page.inputs.is_empty() && page.starts.is_empty());
assert!(
store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn assignment_diagnostic_start_cursor_pages_existing_claims_without_input_causality_or_processing_inference()
{
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
delivery_store_start_reserve(&store, "z-first-start", "a", &source).unwrap();
store
.fail_ticket_assignment_operation("w", "z-first-start", "External resource refused start")
.unwrap();
delivery_store_start_reserve(&store, "a-second-start", "a", &source).unwrap();
let first = store
.page_ticket_assignment_diagnostics("w", "tracking", "a", None, None, 1)
.unwrap();
assert_eq!(first.starts.len(), 1);
assert_eq!(first.starts[0].operation_id, "z-first-start");
assert_eq!(first.starts[0].claim_state, "failed");
assert!(first.starts[0].actual_processing_unknown);
assert_eq!(
first.next_start_operation_id.as_deref(),
Some("z-first-start")
);
let second = store
.page_ticket_assignment_diagnostics(
"w",
"tracking",
"a",
None,
first.next_start_operation_id.as_deref(),
1,
)
.unwrap();
assert_eq!(second.starts[0].operation_id, "a-second-start");
assert_eq!(second.starts[0].claim_state, "pending");
assert!(second.starts[0].actual_processing_unknown);
assert!(second.next_start_operation_id.is_none());
assert!(
store
.page_ticket_assignment_diagnostics(
"w",
"tracking",
"a",
None,
Some("foreign-operation"),
1
)
.is_err()
);
let wire = serde_json::to_string(&first).unwrap();
assert!(!wire.contains("tracking-binding") && !wire.contains("External resource"));
}
@@ -171,3 +171,45 @@ pub(super) fn history_page(
})
})
}
pub(super) fn assignments_page(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
after_assignment_id: Option<&str>,
limit: usize,
) -> Result<ticket::TicketAssignmentsPage> {
store.with_conn(|conn| {
let tx = conn.unchecked_transaction()?;
ticket::sqlite_ticket_assignments_page(
&tx,
workspace_id,
ticket_id,
after_assignment_id,
limit,
)
.map_err(|e| Error::InvalidInput(e.to_string()))
})
}
pub(super) fn answer(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
input: &ticket::TicketAssignmentAnswer,
source: &TicketOrchestratorAssignmentSourceContext,
actor: &str,
) -> Result<ticket::TicketAssignmentAnswerReceipt> {
store.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
require_assignment_source(&tx,workspace_id,source)?;
// Replay retains its saved result after assignment replacement; new input
// must still be admitted by the exact current operational delegation.
let replay: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM typed_ticket_assignment_tracking_operations WHERE workspace_id=?1 AND operation_key=?2)",params![workspace_id,input.operation_key],|r| r.get(0))?;
if !replay { assignment_delivery::require_current(&tx,workspace_id,ticket_id,&input.assignment_id)?; }
let source = serde_json::to_value(source).map_err(|e| Error::Store(e.to_string()))?;
let result = ticket::sqlite_answer_ticket_assignment_in_transaction(&tx,workspace_id,ticket_id,input,&source,actor).map_err(|e| Error::TicketAssignmentConflict(e.to_string()))?;
tx.commit()?;
Ok(result)
})
}
@@ -458,7 +458,10 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
vec![
(91, "workspace schema baseline".into()),
(92, "Ticket-owned assignment tracking".into()),
(93, "Durable assignment deliveries and source-bound startup fences".into())
(
93,
"Durable assignment deliveries and source-bound startup fences".into()
)
]
);
for (table, rows) in retained.iter().zip(original.iter()) {
@@ -1485,3 +1488,337 @@ async fn orchestrator_clear_receipt_compares_nullable_reason_as_full_payload_not
.unwrap()
);
}
fn assignment_answer_input(key: &str) -> ticket::TicketAssignmentAnswer {
ticket::TicketAssignmentAnswer {
assignment_id: "answer-a".into(),
expected_last_event_id: "event-answer-assign".into(),
operation_key: key.into(),
body: "Proceed with option B".into(),
}
}
fn assignment_answer_source() -> TicketOrchestratorAssignmentSourceContext {
TicketOrchestratorAssignmentSourceContext::Runtime {
runtime_id: "r".into(),
worker_id: "worker".into(),
agent_key: "workspace-orchestrator".into(),
authentication_binding_id: "tracking-binding".into(),
}
}
#[tokio::test]
async fn assignment_answer_replay_returns_saved_event_without_new_assignment_or_status_change() {
let (dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "answer-a", "answer-assign");
let input = assignment_answer_input("answer-key");
let before = store
.get_ticket_assignment_tracking("w", "tracking", "answer-a")
.unwrap()
.unwrap();
let saved = store
.answer_ticket_assignment(
"w",
"tracking",
&input,
&assignment_answer_source(),
"authenticated-worker:r/worker",
)
.unwrap();
drop(store);
let store = SqliteWorkspaceStore::open(dir.path().join("workspace.db")).unwrap();
assert_eq!(
store
.answer_ticket_assignment(
"w",
"tracking",
&input,
&assignment_answer_source(),
"authenticated-worker:r/worker"
)
.unwrap(),
saved
);
assert_eq!(
store
.get_ticket_assignment_tracking("w", "tracking", "answer-a")
.unwrap()
.unwrap(),
before
);
let mut changed = input.clone();
changed.body = "Different answer".into();
assert!(
store
.answer_ticket_assignment(
"w",
"tracking",
&changed,
&assignment_answer_source(),
"authenticated-worker:r/worker"
)
.is_err()
);
let pending = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap();
assert_eq!(pending.len(), 2);
assert!(
pending
.iter()
.any(|p| p.input_event_id == saved.input_event_id)
);
store
.with_conn(|conn| {
assert_eq!(
conn.query_row(
"SELECT count(*) FROM typed_ticket_events WHERE kind='assignment_answer'",
[],
|r| r.get::<_, i64>(0)
)?,
1
);
assert_eq!(
conn.query_row(
"SELECT value FROM typed_ticket_event_attributes WHERE key='operation_key'",
[],
|r| r.get::<_, String>(0)
)?,
"answer-key"
);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn assignment_answer_receipt_failure_rolls_back_body_audit_and_ticket_touch() {
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "answer-a", "answer-assign");
store.with_conn(|conn| {
conn.execute_batch("CREATE TRIGGER reject_answer_receipt BEFORE INSERT ON typed_ticket_assignment_tracking_operations BEGIN SELECT RAISE(ABORT,'receipt failed'); END;")?;
Ok(())
}).unwrap();
let before: String = store
.with_conn(|conn| {
Ok(conn.query_row(
"SELECT updated_at FROM typed_tickets WHERE ticket_id='tracking'",
[],
|r| r.get(0),
)?)
})
.unwrap();
assert!(
store
.answer_ticket_assignment(
"w",
"tracking",
&assignment_answer_input("failed-key"),
&assignment_answer_source(),
"authenticated-worker:r/worker"
)
.is_err()
);
store
.with_conn(|conn| {
assert_eq!(
conn.query_row(
"SELECT count(*) FROM typed_ticket_events WHERE kind='assignment_answer'",
[],
|r| r.get::<_, i64>(0)
)?,
0
);
assert_eq!(
conn.query_row(
"SELECT count(*) FROM typed_ticket_event_attributes WHERE key='operation_key'",
[],
|r| r.get::<_, i64>(0)
)?,
0
);
assert_eq!(
conn.query_row(
"SELECT updated_at FROM typed_tickets WHERE ticket_id='tracking'",
[],
|r| r.get::<_, String>(0)
)?,
before
);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn assignment_answer_new_input_rejects_replaced_or_changed_handling_but_exact_retry_is_saved()
{
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "answer-a", "answer-assign");
let input = assignment_answer_input("saved-answer");
let saved = store
.answer_ticket_assignment(
"w",
"tracking",
&input,
&assignment_answer_source(),
"actor",
)
.unwrap();
let mut update = tracking_store_input("answer-a", "event-answer-assign", "next-status");
update.tracking.status = TicketAssignmentTrackingStatus::Waiting;
store
.update_ticket_assignment_tracking("w", "tracking", &update, &tracking_store_source())
.unwrap();
assert!(
store
.answer_ticket_assignment(
"w",
"tracking",
&assignment_answer_input("stale-answer"),
&assignment_answer_source(),
"actor"
)
.is_err()
);
let mut replacement = manual_worker_assignment("tracking", "answer-b");
replacement.role = TicketAssignmentRole::Orchestrator;
replacement.principal = TicketAssignmentPrincipal::WorkspaceAgent {
agent_key: "workspace-orchestrator".into(),
};
store
.set_current_ticket_role_assignment(
&replacement,
Some("answer-a"),
"replace-answer",
"replace-answer-key",
true,
)
.unwrap();
assert!(
store
.answer_ticket_assignment(
"w",
"tracking",
&assignment_answer_input("replaced-answer"),
&assignment_answer_source(),
"actor"
)
.is_err()
);
assert_eq!(
store
.answer_ticket_assignment(
"w",
"tracking",
&input,
&assignment_answer_source(),
"actor"
)
.unwrap(),
saved
);
store
.with_conn(|conn| {
conn.execute(
"UPDATE worker_singleton_owners SET worker_id='replacement' WHERE workspace_id='w'",
[],
)?;
Ok(())
})
.unwrap();
assert!(
store
.answer_ticket_assignment(
"w",
"tracking",
&input,
&assignment_answer_source(),
"actor"
)
.is_err()
);
}
#[tokio::test]
async fn assignment_identity_pages_keep_release_and_last_status_separate_and_reject_foreign_cursor()
{
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "z-first", "first");
let mut replacement = manual_worker_assignment("tracking", "a-second");
replacement.role = TicketAssignmentRole::Orchestrator;
replacement.principal = TicketAssignmentPrincipal::WorkspaceAgent {
agent_key: "workspace-orchestrator".into(),
};
store
.set_current_ticket_role_assignment(
&replacement,
Some("z-first"),
"replacement-event",
"replacement-key",
true,
)
.unwrap();
let page = store
.page_ticket_assignments("w", "tracking", None, 1)
.unwrap();
assert_eq!(page.next_assignment_id.as_deref(), Some("z-first"));
assert!(!page.items[0].current);
assert_eq!(
page.items[0].tracking.as_ref().unwrap().status,
TicketAssignmentTrackingStatus::Waiting
);
assert_eq!(
page.items[0].release.as_ref().unwrap().action,
ticket::TicketAssignmentReleaseAction::Replaced
);
let next = store
.page_ticket_assignments("w", "tracking", page.next_assignment_id.as_deref(), 1)
.unwrap();
assert_eq!(next.items[0].assignment_id, "a-second");
assert!(next.items[0].current);
assert!(next.next_assignment_id.is_none());
assert!(
store
.page_ticket_assignments("other", "tracking", Some("z-first"), 1)
.is_err()
);
assert!(
store
.page_ticket_assignments("w", "tracking", None, 101)
.is_err()
);
}
#[tokio::test]
async fn assignment_answer_concurrent_exact_retry_on_two_connections_creates_one_event_and_receipt()
{
let (dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "answer-a", "answer-assign");
let path = dir.path().join("workspace.db");
let first = SqliteWorkspaceStore::open(&path).unwrap();
let second = SqliteWorkspaceStore::open(&path).unwrap();
let barrier = std::sync::Barrier::new(2);
let results = std::thread::scope(|scope| {
let run = |store: &SqliteWorkspaceStore| {
barrier.wait();
store
.answer_ticket_assignment(
"w",
"tracking",
&assignment_answer_input("concurrent-answer"),
&assignment_answer_source(),
"actor",
)
.unwrap()
};
let a = scope.spawn(move || run(&first));
let b = scope.spawn(move || run(&second));
(a.join().unwrap(), b.join().unwrap())
});
assert_eq!(results.0, results.1);
store.with_conn(|conn| {
assert_eq!(conn.query_row("SELECT count(*) FROM typed_ticket_events WHERE kind='assignment_answer'",[],|r| r.get::<_,i64>(0))?,1);
assert_eq!(conn.query_row("SELECT count(*) FROM typed_ticket_assignment_tracking_operations WHERE operation_key='concurrent-answer'",[],|r| r.get::<_,i64>(0))?,1);
Ok(())
}).unwrap();
}
+110 -2
View File
@@ -405,6 +405,54 @@ impl SqliteWorkspaceAuthority {
.collect::<Vec<_>>();
predicates.push(format!("t.workflow_state IN ({})", states.join(",")));
}
if let Some(reason) = &query.closure_reason {
let encoded =
serde_json::to_string(reason).map_err(|e| Error::InvalidInput(e.to_string()))?;
let value = bind(SqlValue::Text(encoded));
predicates.push(format!("t.workflow_state='closed' AND EXISTS (SELECT 1 FROM typed_ticket_case_metadata m JOIN typed_ticket_case_history h ON h.workspace_id=m.workspace_id AND h.ticket_id=m.ticket_id AND h.operation_key=m.closure_operation_key WHERE m.workspace_id=t.workspace_id AND m.ticket_id=t.ticket_id AND json_extract(h.judgment_json,'$.action.reason')=json_extract({value},'$'))"));
}
if query.assigned_principal.is_some() || query.handling_status.is_some() {
let mut assignment = vec![
"a.workspace_id=t.workspace_id".to_string(),
"a.ticket_id=t.ticket_id".to_string(),
];
if let Some(principal) = &query.assigned_principal {
match principal {
ticket::TicketAssignmentPrincipalSnapshot::WorkspaceAgent { agent_key } => {
let value = bind(SqlValue::Text(agent_key.clone()));
assignment.push(format!(
"a.principal_kind='workspace_agent' AND a.principal_id={value}"
));
}
ticket::TicketAssignmentPrincipalSnapshot::User { account_id } => {
let value = bind(SqlValue::Text(account_id.clone()));
assignment.push(format!(
"a.principal_kind='user' AND a.principal_id={value}"
));
}
ticket::TicketAssignmentPrincipalSnapshot::Worker {
runtime_id,
worker_id,
} => {
let runtime = bind(SqlValue::Text(runtime_id.clone()));
let worker = bind(SqlValue::Text(worker_id.clone()));
assignment.push(format!("a.principal_kind='worker' AND a.runtime_id={runtime} AND a.worker_id={worker}"));
}
}
}
if let Some(status) = &query.handling_status {
let value = match status {
ticket::TicketAssignmentTrackingStatus::Waiting => "waiting",
ticket::TicketAssignmentTrackingStatus::Inprogress => "inprogress",
ticket::TicketAssignmentTrackingStatus::Done => "done",
};
let value = bind(SqlValue::Text(value.into()));
assignment.push(format!(
"a.role='orchestrator' AND json_extract(a.tracking_json,'$.status')={value}"
));
}
predicates.push(format!("EXISTS (SELECT 1 FROM ticket_current_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.assignment_id=c.assignment_id WHERE {})",assignment.join(" AND ")));
}
for id in &query.tag_ids {
let id = bind(SqlValue::Text(id.clone()));
predicates.push(format!("EXISTS (SELECT 1 FROM typed_ticket_tags l JOIN workspace_tags tag ON tag.workspace_id=l.workspace_id AND tag.tag_id=l.tag_id WHERE l.workspace_id=t.workspace_id AND l.ticket_id=t.ticket_id AND l.tag_id={id} AND tag.deleted=0)"));
@@ -426,7 +474,9 @@ impl SqliteWorkspaceAuthority {
"(lower(t.title) LIKE {p} OR lower(t.body) LIKE {p} OR EXISTS (
SELECT 1 FROM typed_ticket_events e
WHERE e.workspace_id=t.workspace_id AND e.ticket_id=t.ticket_id
AND lower(e.body) LIKE {p}))",
AND lower(e.body) LIKE {p}) OR EXISTS (
SELECT 1 FROM typed_ticket_case_history h WHERE h.workspace_id=t.workspace_id AND h.ticket_id=t.ticket_id
AND lower(CAST(json_extract(h.judgment_json,'$.action.reason') AS TEXT)) LIKE {p}))",
p = pattern
));
}
@@ -1121,6 +1171,7 @@ impl TicketAuthority for SqliteWorkspaceAuthority {
let summary = ticket_summary_from_ticket(&authoritative)?;
let authoritative_body = authoritative.document.body.clone();
let authoritative_events = authoritative.events.clone();
let authoritative_case_history = authoritative.case_history.clone();
let detail = self.ticket_detail_from_ticket(
authoritative,
TicketShowRequest {
@@ -1134,6 +1185,7 @@ impl TicketAuthority for SqliteWorkspaceAuthority {
&detail,
authoritative_body.as_str(),
&authoritative_events,
&authoritative_case_history,
&query,
);
let item = ticket_query_item(
@@ -1141,6 +1193,7 @@ impl TicketAuthority for SqliteWorkspaceAuthority {
&detail,
authoritative_body.as_str(),
&authoritative_events,
&authoritative_case_history,
&query,
);
last_scanned_cursor = Some(make_ticket_cursor(&item, sort, &fingerprint));
@@ -1993,6 +2046,33 @@ fn ticket_evidence_summary(
}
fn validate_ticket_query(query: &TicketQueryRequest) -> Result<()> {
if let Some(ticket::TicketClosureReason::Other(text)) = &query.closure_reason
&& (text.trim().is_empty() || text.len() > 1024 || text.contains('\0'))
{
return Err(Error::InvalidInput(
"Closure reason Other must be nonblank and at most 1024 bytes without NUL".into(),
));
}
if let Some(principal) = &query.assigned_principal {
let ids: Vec<&str> = match principal {
ticket::TicketAssignmentPrincipalSnapshot::User { account_id } => vec![account_id],
ticket::TicketAssignmentPrincipalSnapshot::WorkspaceAgent { agent_key } => {
vec![agent_key]
}
ticket::TicketAssignmentPrincipalSnapshot::Worker {
runtime_id,
worker_id,
} => vec![runtime_id, worker_id],
};
if ids
.iter()
.any(|id| id.trim().is_empty() || id.len() > 1024 || id.chars().any(char::is_control))
{
return Err(Error::InvalidInput(
"Assigned principal requires bounded stable identities".into(),
));
}
}
for ids in [&query.tag_ids, &query.tag_ids_any, &query.tag_ids_none] {
if ids.len() > 100 || ids.iter().any(|id| id.is_empty() || id.len() > 128) {
return Err(Error::InvalidInput(
@@ -2181,11 +2261,27 @@ fn ticket_attention_matches(
}
}
fn closure_reason_text(action: &ticket::TicketCaseAction) -> Option<String> {
let reason = match action {
ticket::TicketCaseAction::Close { reason }
| ticket::TicketCaseAction::CorrectClosure { reason } => reason,
ticket::TicketCaseAction::Reopen => return None,
};
Some(match reason {
ticket::TicketClosureReason::Completed => "completed".into(),
ticket::TicketClosureReason::NotPlanned => "not_planned".into(),
ticket::TicketClosureReason::Duplicate => "duplicate".into(),
ticket::TicketClosureReason::Unknown => "unknown".into(),
ticket::TicketClosureReason::Other(text) => format!("other {text}"),
})
}
fn ticket_matches_query(
summary: &TicketSummary,
detail: &TicketDetail,
authoritative_body: &str,
authoritative_events: &[TicketEvent],
authoritative_case_history: &[ticket::TicketCaseHistoryEntry],
query: &TicketQueryRequest,
) -> bool {
let has_tag = |id: &String| detail.tags.iter().any(|tag| &tag.id == id);
@@ -2311,7 +2407,10 @@ fn ticket_matches_query(
}
query.query.as_ref().is_none_or(|text| {
let needle = text.to_lowercase();
summary.title.to_lowercase().contains(&needle)
authoritative_case_history.iter().any(|entry| {
closure_reason_text(&entry.action)
.is_some_and(|reason| reason.to_lowercase().contains(&needle))
}) || summary.title.to_lowercase().contains(&needle)
|| authoritative_body.to_lowercase().contains(&needle)
|| authoritative_events
.iter()
@@ -2324,6 +2423,7 @@ fn ticket_query_item(
detail: &TicketDetail,
authoritative_body: &str,
authoritative_events: &[TicketEvent],
authoritative_case_history: &[ticket::TicketCaseHistoryEntry],
query: &TicketQueryRequest,
) -> TicketQueryItem {
let mut matched_fields = Vec::new();
@@ -2331,6 +2431,14 @@ fn ticket_query_item(
let mut matching_event = None;
if let Some(text) = query.query.as_ref() {
let needle = text.to_lowercase();
if let Some(reason) = authoritative_case_history
.iter()
.filter_map(|entry| closure_reason_text(&entry.action))
.find(|reason| reason.to_lowercase().contains(&needle))
{
matched_fields.push("closure_reason".into());
snippet = Some(reason);
}
if summary.title.to_lowercase().contains(&needle) {
matched_fields.push("title".to_string());
snippet = Some(summary.title.clone());
+192 -6
View File
@@ -3097,7 +3097,9 @@ fn is_server_global_forward(path: &str) -> bool {
}
fn is_server_static_forward(path: &str) -> bool {
!path.starts_with("/api/") && !path.starts_with("/internal/")
!path.starts_with("/api/")
&& !path.starts_with("/internal/")
&& canonical_ticket_operation_workspace_id(path).is_none()
}
async fn serve_server_static_shell(api: &WorkspaceServerApi, path: &str) -> Response {
@@ -3107,7 +3109,7 @@ async fn serve_server_static_shell(api: &WorkspaceServerApi, path: &str) -> Resp
static_file_or_spa_response(static_root, path).await
}
// Canonical case/tracking routes are deliberately not /api/w aliases. Recognize
// Canonical Ticket operation routes are deliberately not /api/w aliases. Recognize
// only these operations so authentication/scope are identical without exposing CRUD.
fn canonical_ticket_operation_workspace_id(path: &str) -> Option<&str> {
let mut segments = path.strip_prefix("/workspaces/")?.split('/');
@@ -3119,10 +3121,10 @@ fn canonical_ticket_operation_workspace_id(path: &str) -> Option<&str> {
&& resource == "tickets"
&& !ticket_id.is_empty()
&& match operation {
"case" => segments.next().is_none(),
"case" | "assignment-diagnostics" | "assignment-answer" => segments.next().is_none(),
"assignment-tracking" => match segments.next() {
None => true,
Some("history") => segments.next().is_none(),
Some("history" | "assignments") => segments.next().is_none(),
_ => false,
},
_ => false,
@@ -6614,6 +6616,18 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::ticket_case_update(
service.clone(),
))
.merge(server_api::server_api_axum::ticket_assignments_page(
service.clone(),
))
.merge(server_api::server_api_axum::ticket_assignments_page_browser(service.clone()))
.merge(server_api::server_api_axum::ticket_assignment_diagnostics(
service.clone(),
))
.merge(server_api::server_api_axum::ticket_assignment_diagnostics_browser(service.clone()))
.merge(server_api::server_api_axum::ticket_assignment_answer(
service.clone(),
))
.merge(server_api::server_api_axum::ticket_assignment_answer_browser(service.clone()))
.merge(server_api::server_api_axum::ticket_assignment_tracking_get(
service.clone(),
))
@@ -10566,6 +10580,132 @@ impl server_api::ServerApi for ServerApiContractService {
Ok(response)
}
async fn ticket_assignments_page(
&self,
_context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
query: server_api::TicketAssignmentsQuery,
) -> std::result::Result<server_api::TicketAssignmentsPage, server_api::RepositoryApiError>
{
let api = self.workspace_api()?;
validate_workspace_scope(api, &workspace_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let ticket_id = resolve_workspace_ticket_reference(api, &workspace_id, &ticket_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
api.store
.page_ticket_assignments(
&workspace_id,
&ticket_id,
query.after_assignment_id.as_deref(),
query.limit.unwrap_or(30),
)
.map(server_api::TicketAssignmentsPage)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)
}
async fn ticket_assignments_page_browser(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
query: server_api::TicketAssignmentsQuery,
) -> std::result::Result<server_api::TicketAssignmentsPage, server_api::RepositoryApiError>
{
self.ticket_assignments_page(context, workspace_id, ticket_id, query)
.await
}
async fn ticket_assignment_diagnostics(
&self,
_context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
query: server_api::TicketAssignmentDiagnosticsQuery,
) -> std::result::Result<
server_api::TicketAssignmentDiagnosticsPage,
server_api::RepositoryApiError,
> {
let api = self.workspace_api()?;
validate_workspace_scope(api, &workspace_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let ticket_id = resolve_workspace_ticket_reference(api, &workspace_id, &ticket_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
api.store
.page_ticket_assignment_diagnostics(
&workspace_id,
&ticket_id,
&query.assignment_id,
query.after_input_event_id.as_deref(),
query.after_start_operation_id.as_deref(),
query.limit.unwrap_or(30),
)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)
}
async fn ticket_assignment_diagnostics_browser(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
query: server_api::TicketAssignmentDiagnosticsQuery,
) -> std::result::Result<
server_api::TicketAssignmentDiagnosticsPage,
server_api::RepositoryApiError,
> {
self.ticket_assignment_diagnostics(context, workspace_id, ticket_id, query)
.await
}
async fn ticket_assignment_answer(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
request: server_api::TicketAssignmentAnswerRequest,
) -> std::result::Result<
server_api::TicketAssignmentAnswerResponse,
server_api::RepositoryApiError,
> {
let api = self.workspace_api()?;
validate_workspace_scope(api, &workspace_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let ticket_id = resolve_workspace_ticket_reference(api, &workspace_id, &ticket_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let (actor, source) =
authenticated_assignment_source(api, &contract_request_headers(&context)?, &context)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let receipt = api
.store
.answer_ticket_assignment(&workspace_id, &ticket_id, &request.0, &source, &actor)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
// The entire event/audit/receipt transaction has committed. Response loss
// retries the saved receipt; delivery scan deduplicates accepted inputs.
if let Some(orchestrator) = find_workspace_orchestrator(api) {
dispatch_orchestrator_assignment_attention(api, &orchestrator.worker);
}
Ok(server_api::TicketAssignmentAnswerResponse(receipt))
}
async fn ticket_assignment_answer_browser(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
request: server_api::TicketAssignmentAnswerRequest,
) -> std::result::Result<
server_api::TicketAssignmentAnswerResponse,
server_api::RepositoryApiError,
> {
self.ticket_assignment_answer(context, workspace_id, ticket_id, request)
.await
}
async fn ticket_assignment_tracking_get(
&self,
_context: server_api::ServerRequestContext,
@@ -48195,6 +48335,42 @@ mod tests {
}
}
async fn author_notification_test_answer(api: &WorkspaceApi, ticket_id: &str, key: &str) {
let current = api
.store
.list_current_ticket_assignment_tracking(TEST_WORKSPACE_ID, ticket_id)
.unwrap()
.remove(0);
let response = build_router(api.clone())
.oneshot(
Request::builder()
.method("POST")
.uri(format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{ticket_id}/assignment-answer"
))
.header(
"authorization",
format!("Bearer api-token-{TEST_WORKSPACE_ID}"),
)
.header(CONTENT_TYPE, "application/json")
.body(Body::from(
json!({
"assignment_id": current.assignment_id,
"expected_last_event_id": current.last_event_id.unwrap(),
"operation_key": key,
"body": "Explicit continuing input"
})
.to_string(),
))
.unwrap(),
)
.await
.unwrap();
let status = response.status();
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(status, StatusCode::OK, "{}", String::from_utf8_lossy(&body));
}
#[tokio::test]
async fn orchestrator_ticket_notifications_project_authoritative_post_mutation_state() {
let dir = tempfile::tempdir().unwrap();
@@ -48350,8 +48526,15 @@ mod tests {
.unwrap();
}
assert_eq!(
execution.take_inputs().len(),
1,
"ordinary history must not create unfenced continuing input"
);
seed_test_api_token(api.store.as_ref(), TEST_WORKSPACE_ID);
author_notification_test_answer(&api, &ticket.id, "notification-explicit-answer").await;
let inputs = execution.take_inputs();
assert_eq!(inputs.len(), 4, "Assign input plus three external events");
assert_eq!(inputs.len(), 1, "one fenced continuing answer");
let unique = inputs
.iter()
.map(|(_, content)| content.clone())
@@ -50357,11 +50540,13 @@ mod tests {
assert!(!serialized.contains("queued update"));
assert!(!serialized.contains("Ticket notification:"));
seed_test_api_token(api.store.as_ref(), TEST_WORKSPACE_ID);
author_notification_test_answer(&api, &ticket_ref.id, "notification-saved-answer").await;
let notifications = execution.take_inputs();
assert_eq!(
notifications.len(),
2,
"initial delegation plus external comment"
"initial delegation plus fenced answer; ordinary comment is history"
);
assert_eq!(
notifications[0].0.worker_id.to_string(),
@@ -50406,6 +50591,7 @@ mod tests {
api.authority.ticket(&ticket_ref.id).unwrap().state.as_str(),
TicketState::Open.as_str()
);
author_notification_test_answer(&api, &ticket_ref.id, "notification-failed-answer").await;
let warnings = TICKET_NOTIFICATION_DELIVERY_WARNING_CAPTURE
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
@@ -309,7 +309,11 @@ async fn orchestrator_attention_unknown_response_retry_keeps_id_and_payload() {
"existing"
);
let retry = execution.take_inputs();
assert_eq!(retry.len(), 2, "acknowledged retry must also drain the changed backlog");
assert_eq!(
retry.len(),
2,
"acknowledged retry must also drain the changed backlog"
);
assert_eq!(retry[0].1, pending.content);
assert!(retry[1].1.contains("Changed backlog"));
let ids = execution.take_input_request_ids();
@@ -760,11 +764,27 @@ async fn real_attention_worker_start_restore(lose_response: bool) {
assert_eq!(receipts().len(), 1);
assert_eq!(commits(), 1);
// Only a new external input earns another delivery, not restore itself.
browser_ticket_backend(&api)
seed_test_api_token(api.store.as_ref(), "attention-answer");
let observed = api
.store
.list_current_ticket_assignment_tracking(TEST_WORKSPACE_ID, &ticket.id)
.unwrap()
.add_event(
ticket.id.clone().into(),
ticket::NewTicketEvent::new(ticket::TicketEventKind::Comment, "New owner answer"),
.remove(0);
api.store
.answer_ticket_assignment(
TEST_WORKSPACE_ID,
&ticket.id,
&ticket::TicketAssignmentAnswer {
assignment_id: observed.assignment_id,
expected_last_event_id: observed.last_event_id.unwrap(),
operation_key: "attention-owner-answer".into(),
body: "New owner answer".into(),
},
&crate::store::TicketOrchestratorAssignmentSourceContext::Account {
account_id: "account-attention-answer".into(),
user_id: "user-attention-answer".into(),
},
"fixture-account",
)
.unwrap();
maybe_dispatch_orchestrator_turn_end(&api, &worker.worker_id, Some(S::Running), S::Idle);
@@ -163,6 +163,41 @@ async fn delivery_self_report(
serde_json::from_slice(&bytes).unwrap()
}
async fn delivery_external_answer(api: &WorkspaceApi, ticket: &str, text: &str, key: &str) {
let current = api
.store
.list_current_ticket_assignment_tracking(TEST_WORKSPACE_ID, ticket)
.unwrap()
.remove(0);
delivery_http(api,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{ticket}/assignment-answer"),Some(json!({"assignment_id":current.assignment_id,"expected_last_event_id":current.last_event_id,"operation_key":key,"body":text})),StatusCode::OK).await;
}
fn delivery_commit_answer_without_scan(api: &WorkspaceApi, ticket: &str, text: &str, key: &str) {
seed_test_api_token(api.store.as_ref(), "direct-answer");
let current = api
.store
.list_current_ticket_assignment_tracking(TEST_WORKSPACE_ID, ticket)
.unwrap()
.remove(0);
api.store
.answer_ticket_assignment(
TEST_WORKSPACE_ID,
ticket,
&ticket::TicketAssignmentAnswer {
assignment_id: current.assignment_id,
expected_last_event_id: current.last_event_id.unwrap(),
operation_key: key.into(),
body: text.into(),
},
&crate::store::TicketOrchestratorAssignmentSourceContext::Account {
account_id: "account-direct-answer".into(),
user_id: "user-direct-answer".into(),
},
"fixture-account",
)
.unwrap();
}
fn delivery_external_comment(api: &WorkspaceApi, ticket: &str, text: &str) {
browser_ticket_backend(api)
.unwrap()
@@ -212,7 +247,18 @@ async fn assignment_waiting_self_report_does_not_loop_but_external_answer_is_del
);
let waiting = delivery_tracking(&api, &ticket.id, assignment);
assert_eq!(serde_json::to_value(&waiting).unwrap(), reported);
delivery_external_comment(&api, &ticket.id, "Human answer: choose option B");
delivery_external_comment(&api, &ticket.id, "History-only annotation while waiting");
assert!(
execution.take_inputs().is_empty(),
"ordinary waiting comment became fresh work"
);
delivery_external_answer(
&api,
&ticket.id,
"Human answer: choose option B",
"choose-B",
)
.await;
let answer = execution.take_inputs();
assert_eq!(
answer.len(),
@@ -369,7 +415,10 @@ async fn assignment_clear_and_reassign_discard_old_unknown_delivery_without_stop
assert!(inputs[0].1.contains(new_assignment));
let ids = execution.take_input_request_ids();
assert_eq!(ids.len(), 1);
assert_ne!(ids[0], old_ids[0], "new assignment reused stale input identity");
assert_ne!(
ids[0], old_ids[0],
"new assignment reused stale input identity"
);
assert_eq!(
delivery_assign(&api, &ticket.id, "assign-old").await,
assigned
@@ -404,7 +453,8 @@ async fn assignment_or_answer_committed_during_empty_scan_is_delivered_without_a
&ticket.id,
assigned["assignment"]["assignment_id"].as_str().unwrap(),
);
delivery_self_report(&api, &identity, &worker, &current, "waiting", "wait-answer").await;
delivery_self_report(&api, &identity, &worker, &current, "waiting", "wait-answer")
.await;
execution.take_inputs();
execution.take_input_request_ids();
}
@@ -422,29 +472,33 @@ async fn assignment_or_answer_committed_during_empty_scan_is_delivered_without_a
true,
|state, pending| {
check_orchestrator_assignment_attention(
&caller_api, &caller_worker, state, pending,
&caller_api,
&caller_worker,
state,
pending,
);
if checks == 0 {
// The production scan has read an empty snapshot but
// still owns its reservation. Race a real HTTP commit.
entered_tx.send(()).unwrap();
release_rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
release_rx
.recv_timeout(std::time::Duration::from_secs(5))
.unwrap();
}
checks += 1;
},
);
assert_eq!(checks, 2, "committed input must defer a scan");
});
entered_rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
entered_rx
.recv_timeout(std::time::Duration::from_secs(5))
.unwrap();
if answer {
futures::executor::block_on(delivery_http(
futures::executor::block_on(delivery_external_answer(
&api,
"POST",
&format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{}/thread-events", ticket.id),
Some(serde_json::to_value(NewTicketEvent::new(
TicketEventKind::Comment, "Human answer: go ahead",
)).unwrap()),
StatusCode::NO_CONTENT,
&ticket.id,
"Human answer: go ahead",
"racing-answer",
));
} else {
futures::executor::block_on(delivery_assign(&api, &ticket.id, "racing-assign"));
@@ -456,7 +510,11 @@ async fn assignment_or_answer_committed_during_empty_scan_is_delivered_without_a
});
let inputs = execution.take_inputs();
assert_eq!(inputs.len(), 1, "idle Orchestrator was stranded");
assert!(inputs[0].1.contains(ticket.resource_key.as_deref().unwrap()));
assert!(
inputs[0]
.1
.contains(ticket.resource_key.as_deref().unwrap())
);
if answer {
assert!(inputs[0].1.contains("Need the human's choice"));
}
@@ -476,15 +534,26 @@ async fn assignment_failed_delivery_and_its_reentrant_idle_do_not_schedule_retry
execution.before_input_returns(move |worker| {
use protocol::subscription::SubscriptionWorkerState as S;
maybe_dispatch_orchestrator_turn_end(
&callback_api, &worker.worker_id.to_string(), Some(S::Running), S::Idle,
&callback_api,
&worker.worker_id.to_string(),
Some(S::Running),
S::Idle,
);
});
delivery_assign(&api, &ticket.id, "assign-failed").await;
assert_eq!(execution.take_inputs().len(), 2, "idle must not add retries beyond the bounded send");
assert_eq!(
execution.take_inputs().len(),
2,
"idle must not add retries beyond the bounded send"
);
let ids = execution.take_input_request_ids();
assert_eq!(ids.len(), 2);
assert_eq!(ids[0], ids[1]);
assert_eq!(attention_pending(&api).len(), 1, "failure must retain durable input");
assert_eq!(
attention_pending(&api).len(),
1,
"failure must retain durable input"
);
delivery_stop(&api, &worker);
}
@@ -796,8 +865,22 @@ async fn delivery_wait_idle(
"duplicate canonical delivery: {:?}",
delivery_commits(api, worker)
);
let entries = delivery_session_entries(api, worker);
let last_notification = entries.iter().rposition(|entry| matches!(
entry,
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, session_store::SystemItem::Notification { .. })
));
let last_completed = entries.iter().rposition(|entry| matches!(entry, LogEntry::RunCompleted { .. }));
// Runtime acceptance and the saved notification precede processing.
// Initial Idle can therefore be observed before any model request.
// Require a committed run outcome after the latest input, not a
// guessed delay or Worker liveness as proof of consumption.
let processed = last_notification.zip(last_completed)
.is_some_and(|(input, completed)| completed > input);
let summary = api.runtime.worker(worker).unwrap();
if commits == expected_commits
&& processed
&& summary
.worker_state
.as_ref()
@@ -962,16 +1045,12 @@ async fn assignment_real_worker_response_loss_restore_and_concurrent_scans_commi
waiting
);
let before = execution.attempts.lock().unwrap().len();
browser_ticket_backend(&reopened)
.unwrap()
.add_event(
ticket.id.clone().into(),
ticket::NewTicketEvent::new(
ticket::TicketEventKind::Comment,
"External answer after server restart",
),
)
.unwrap();
delivery_commit_answer_without_scan(
&reopened,
&ticket.id,
"External answer after server restart",
"answer-after-restart",
);
let barrier = Arc::new(std::sync::Barrier::new(8));
std::thread::scope(|scope| {
let joins = (0..8)
@@ -68,9 +68,13 @@ async fn runtime_and_account_assign_retries_return_saved_result_after_clear_and_
})).unwrap();
tracking_response(
app.clone(),
Request::builder().method("PUT").uri(&worker_path).header(CONTENT_TYPE, "application/json")
Request::builder()
.method("PUT")
.uri(&worker_path)
.header(CONTENT_TYPE, "application/json")
.header("authorization", format!("Bearer {token}"))
.body(Body::from(worker_input)).unwrap(),
.body(Body::from(worker_input))
.unwrap(),
StatusCode::OK,
)
.await;
@@ -127,7 +131,10 @@ async fn runtime_and_account_assign_retries_return_saved_result_after_clear_and_
StatusCode::OK,
)
.await;
assert_ne!(fresh["assignment"]["assignment_id"].as_str(), Some(assignment));
assert_ne!(
fresh["assignment"]["assignment_id"].as_str(),
Some(assignment)
);
assert_eq!(
api.store
.get_ticket_assignment_tracking(TEST_WORKSPACE_ID, &ticket, assignment)
@@ -146,9 +153,18 @@ async fn runtime_and_account_assign_retries_return_saved_result_after_clear_and_
.unwrap(),
active
);
assert_eq!(api.store.get_current_ticket_role_assignment(TEST_WORKSPACE_ID, &ticket,
TicketAssignmentRole::Orchestrator).unwrap().unwrap().assignment_id,
fresh["assignment"]["assignment_id"].as_str().unwrap());
assert_eq!(
api.store
.get_current_ticket_role_assignment(
TEST_WORKSPACE_ID,
&ticket,
TicketAssignmentRole::Orchestrator
)
.unwrap()
.unwrap()
.assignment_id,
fresh["assignment"]["assignment_id"].as_str().unwrap()
);
}
}
@@ -1494,3 +1510,550 @@ async fn assignment_signed_runtime_without_worker_cannot_self_assign_service_rol
.is_empty()
);
}
#[tokio::test]
async fn public_assignment_answer_authenticates_source_replays_saved_result_and_rejects_read_reassign_race()
{
let directory = tempfile::tempdir().unwrap();
let (api, _, identity, source, ticket) = tracking_fixture(directory.path()).await;
let current = assign_tracking(&api, &identity, &source, &ticket).await;
let token = seed_test_api_token(api.store.as_ref(), "answer-owner");
let app = build_router(api.clone());
let path = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{ticket}/assignment-answer");
let body = serde_json::json!({"assignment_id":current.assignment_id,"expected_last_event_id":current.last_event_id,"operation_key":"answer-op","body":"Choose the bounded option"});
let account_request = |body: &serde_json::Value| {
Request::builder()
.method("POST")
.uri(&path)
.header(CONTENT_TYPE, "application/json")
.header("authorization", format!("Bearer {token}"))
.body(Body::from(serde_json::to_vec(body).unwrap()))
.unwrap()
};
tracking_response(
app.clone(),
Request::builder()
.method("POST")
.uri(&path)
.header(CONTENT_TYPE, "application/json")
.body(Body::from(serde_json::to_vec(&body).unwrap()))
.unwrap(),
StatusCode::UNAUTHORIZED,
)
.await;
let saved = tracking_response(app.clone(), account_request(&body), StatusCode::OK).await;
assert_eq!(saved["assignment_id"], current.assignment_id);
let backend = browser_ticket_backend(&api).unwrap();
let authored = backend.show(ticket.clone().into()).unwrap();
assert_eq!(
authored
.events
.iter()
.filter(|e| e.kind.as_str() == "assignment_answer")
.count(),
1
);
let event = authored
.events
.iter()
.find(|e| e.kind.as_str() == "assignment_answer")
.unwrap();
assert!(
event
.author
.as_deref()
.unwrap()
.starts_with("workspace-user:")
);
assert_eq!(event.body.as_str(), "Choose the bounded option");
assert_eq!(
api.store
.get_ticket_assignment_tracking(TEST_WORKSPACE_ID, &ticket, &current.assignment_id)
.unwrap()
.unwrap(),
current
);
assert_eq!(
tracking_response(app.clone(), account_request(&body), StatusCode::OK).await,
saved
);
let mut changed = body.clone();
changed["body"] = "Changed payload".into();
tracking_response(app.clone(), account_request(&changed), StatusCode::CONFLICT).await;
let mut spoof = body.clone();
spoof["actor"] = "owner".into();
tracking_response(
app.clone(),
account_request(&spoof),
StatusCode::UNPROCESSABLE_ENTITY,
)
.await;
let assignment_path =
format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{ticket}/assignments/orchestrator");
tracking_response(app.clone(),runtime_source_request(&identity,Some(&source.worker_id),"PUT",&assignment_path,serde_json::to_vec(&serde_json::json!({"operation_id":"replace-after-read","expected_assignment_id":current.assignment_id,"principal":{"kind":"workspace_agent","agent_key":"workspace-orchestrator"}})).unwrap()),StatusCode::OK).await;
let mut stale = body.clone();
stale["operation_key"] = "fresh-stale-answer".into();
tracking_response(app.clone(), account_request(&stale), StatusCode::CONFLICT).await;
assert_eq!(
tracking_response(app, account_request(&body), StatusCode::OK).await,
saved
);
assert_eq!(
backend
.show(ticket.into())
.unwrap()
.events
.iter()
.filter(|e| e.kind.as_str() == "assignment_answer")
.count(),
1
);
}
#[tokio::test]
async fn public_assignment_pages_and_diagnostics_require_scoped_read_and_reject_foreign_cursors() {
let directory = tempfile::tempdir().unwrap();
let (api, _, identity, source, ticket) = tracking_fixture(directory.path()).await;
let current = assign_tracking(&api, &identity, &source, &ticket).await;
let token = seed_test_api_token(api.store.as_ref(), "diagnostics-reader");
let app = build_router(api.clone());
for prefix in [
format!("/api/w/{TEST_WORKSPACE_ID}"),
format!("/workspaces/{TEST_WORKSPACE_ID}"),
] {
let path = format!("{prefix}/tickets/{ticket}/assignment-tracking/assignments?limit=1");
tracking_response(
app.clone(),
Request::builder().uri(&path).body(Body::empty()).unwrap(),
StatusCode::UNAUTHORIZED,
)
.await;
let request = |path: String| {
Request::builder()
.uri(path)
.header("authorization", format!("Bearer {token}"))
.body(Body::empty())
.unwrap()
};
let page = tracking_response(app.clone(), request(path), StatusCode::OK).await;
assert_eq!(page["items"][0]["assignment_id"], current.assignment_id);
assert!(page["next_assignment_id"].is_null());
let path = format!(
"{prefix}/tickets/{ticket}/assignment-diagnostics?assignment_id={}&limit=1",
current.assignment_id
);
let diagnostic =
tracking_response(app.clone(), request(path.clone()), StatusCode::OK).await;
assert_eq!(diagnostic["assignment_id"], current.assignment_id);
assert_eq!(diagnostic["input_origin_known"], true);
assert!(diagnostic.get("payload_json").is_none());
tracking_response(
app.clone(),
request(format!("{path}&after_input_event_id=ticket-event:999")),
StatusCode::BAD_REQUEST,
)
.await;
tracking_response(app.clone(),request(format!("{prefix}/tickets/{ticket}/assignment-tracking/assignments?after_assignment_id=foreign")),StatusCode::BAD_REQUEST).await;
tracking_response(
app.clone(),
request(format!(
"{prefix}/tickets/{ticket}/assignment-diagnostics?assignment_id={}&limit=101",
current.assignment_id
)),
StatusCode::BAD_REQUEST,
)
.await;
}
let path = format!(
"/api/w/another-workspace/tickets/{ticket}/assignment-diagnostics?assignment_id={}",
current.assignment_id
);
let response = app
.oneshot(
Request::builder()
.uri(path)
.header("authorization", format!("Bearer {token}"))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert!(!response.status().is_success());
}
#[tokio::test]
async fn ticket_query_principal_and_handling_filter_before_single_item_page_without_worker_inference()
{
let directory = tempfile::tempdir().unwrap();
let (api, _, identity, source, ticket) = tracking_fixture(directory.path()).await;
let current = assign_tracking(&api, &identity, &source, &ticket).await;
tracking_response(
build_router(api.clone()),
signed_tracking_update(
&identity,
Some(&source.worker_id),
&ticket,
&tracking_update(&current, "query-done"),
),
StatusCode::OK,
)
.await;
let backend = browser_ticket_backend(&api).unwrap();
for n in 0..8 {
backend
.create(ticket::NewTicket::new(format!("Unassigned newer {n}")))
.unwrap();
}
let query:server_api::TicketQueryRequest=serde_json::from_value(serde_json::json!({"handling_status":"done","assigned_principal":{"kind":"workspace_agent","agent_key":"workspace-orchestrator"},"limit":1})).unwrap();
let found = api.authority.query_tickets(query).unwrap();
assert_eq!(found.items.len(), 1);
assert_eq!(found.items[0].id, ticket);
let query:server_api::TicketQueryRequest=serde_json::from_value(serde_json::json!({"handling_status":"waiting","assigned_principal":{"kind":"workspace_agent","agent_key":"workspace-orchestrator"},"limit":1})).unwrap();
assert!(api.authority.query_tickets(query).unwrap().items.is_empty());
let query:server_api::TicketQueryRequest=serde_json::from_value(serde_json::json!({"assigned_principal":{"kind":"worker","runtime_id":source.runtime_id,"worker_id":source.worker_id},"limit":1})).unwrap();
assert!(api.authority.query_tickets(query).unwrap().items.is_empty());
}
#[test]
fn canonical_assignment_operations_are_explicit_scoped_api_routes_not_static_shell_paths() {
for suffix in [
"case",
"assignment-tracking",
"assignment-tracking/history",
"assignment-tracking/assignments",
"assignment-diagnostics",
"assignment-answer",
] {
let path = format!("/workspaces/workspace-a/tickets/ticket-a/{suffix}");
assert_eq!(
canonical_ticket_operation_workspace_id(&path),
Some("workspace-a")
);
assert_eq!(scoped_workspace_id(&path), Some("workspace-a"));
assert!(!is_server_static_forward(&path));
assert!(!is_server_global_forward(&path));
assert!(canonical_ticket_operation_workspace_id(&format!("{path}/extra")).is_none());
}
for invalid in [
"/workspaces//tickets/t/assignment-answer",
"/workspaces/w/tickets//assignment-diagnostics",
"/workspaces/w/tickets/t",
"/workspaces/w/tickets/t/unknown",
"/workspaces/w/tickets/t/assignment-tracking/unknown",
"/workspaces/w/workers/t/assignment-answer",
] {
assert!(
canonical_ticket_operation_workspace_id(invalid).is_none(),
"{invalid}"
);
}
}
#[tokio::test]
async fn catalog_assignment_routes_apply_cookie_workspace_auth_and_origin_to_browser_and_canonical_paths()
{
let directory = tempfile::tempdir().unwrap();
let (api, _, identity, source, ticket) = tracking_fixture(directory.path()).await;
let current = assign_tracking(&api, &identity, &source, &ticket).await;
seed_test_api_token(api.store.as_ref(), "assignment-cookie");
api.store
.create_browser_session(&BrowserSessionRecord {
token_hash: crate::auth::token_hash("assignment-browser-session"),
session_id: "assignment-session".into(),
user_id: "user-assignment-cookie".into(),
created_at: "2026-01-01T00:00:00Z".into(),
expires_at: "2099-01-01T00:00:00Z".into(),
revoked_at: None,
})
.unwrap();
let AuthConfig::Passkey {
origin,
cookie_name,
..
} = &api.config.auth;
let cookie = format!("{cookie_name}=assignment-browser-session");
let catalog = WorkspaceServerApi::new(api.config.clone(), api.store.clone());
catalog
.apis
.lock()
.await
.insert(TEST_WORKSPACE_ID.into(), api.clone());
catalog
.routers
.lock()
.await
.insert(TEST_WORKSPACE_ID.into(), build_inner_router(api.clone()));
let app = workspace_server_router(catalog);
let body = serde_json::json!({"assignment_id":current.assignment_id,"expected_last_event_id":current.last_event_id,"operation_key":"cookie-answer","body":"Authorized browser continuation"});
let mut receipt = None;
for prefix in [
format!("/api/w/{TEST_WORKSPACE_ID}"),
format!("/workspaces/{TEST_WORKSPACE_ID}"),
] {
let identity_path =
format!("{prefix}/tickets/{ticket}/assignment-tracking/assignments?limit=1");
let diagnostic_path = format!(
"{prefix}/tickets/{ticket}/assignment-diagnostics?assignment_id={}",
current.assignment_id
);
for path in [identity_path, diagnostic_path] {
tracking_response(
app.clone(),
Request::builder().uri(&path).body(Body::empty()).unwrap(),
StatusCode::UNAUTHORIZED,
)
.await;
tracking_response(
app.clone(),
Request::builder()
.uri(path)
.header(axum::http::header::COOKIE, &cookie)
.body(Body::empty())
.unwrap(),
StatusCode::OK,
)
.await;
}
let answer_path = format!("{prefix}/tickets/{ticket}/assignment-answer");
let request = |with_origin: bool| {
let mut request = Request::builder()
.method("POST")
.uri(&answer_path)
.header(CONTENT_TYPE, "application/json")
.header(axum::http::header::COOKIE, &cookie);
if with_origin {
request = request.header(ORIGIN, origin);
}
request
.body(Body::from(serde_json::to_vec(&body).unwrap()))
.unwrap()
};
tracking_response(app.clone(), request(false), StatusCode::FORBIDDEN).await;
let saved = tracking_response(app.clone(), request(true), StatusCode::OK).await;
if let Some(previous) = &receipt {
assert_eq!(&saved, previous);
} else {
receipt = Some(saved);
}
let cross = answer_path.replace(TEST_WORKSPACE_ID, "foreign-workspace");
tracking_response(
app.clone(),
Request::builder()
.method("POST")
.uri(cross)
.header(CONTENT_TYPE, "application/json")
.header(axum::http::header::COOKIE, &cookie)
.header(ORIGIN, origin)
.body(Body::from(serde_json::to_vec(&body).unwrap()))
.unwrap(),
StatusCode::NOT_FOUND,
)
.await;
}
let events = browser_ticket_backend(&api)
.unwrap()
.show(ticket.into())
.unwrap()
.events;
let answers = events
.iter()
.filter(|event| event.kind.as_str() == "assignment_answer")
.collect::<Vec<_>>();
assert_eq!(answers.len(), 1);
assert!(
answers[0]
.author
.as_deref()
.unwrap()
.contains("account-assignment-cookie")
);
}
#[tokio::test]
async fn canonical_answer_uses_actual_runtime_binding_and_rejects_tampered_source_or_binding_replay()
{
let directory = tempfile::tempdir().unwrap();
let (api, _, identity, source, ticket) = tracking_fixture(directory.path()).await;
let current = assign_tracking(&api, &identity, &source, &ticket).await;
let path = format!("/workspaces/{TEST_WORKSPACE_ID}/tickets/{ticket}/assignment-answer");
let input = serde_json::json!({"assignment_id":current.assignment_id,"expected_last_event_id":current.last_event_id,"operation_key":"runtime-answer","body":"Runtime-authenticated continuation"});
let request = || {
runtime_source_request(
&identity,
Some(&source.worker_id),
"POST",
&path,
serde_json::to_vec(&input).unwrap(),
)
};
let app = build_router(api.clone());
let mut unsigned = request();
unsigned
.headers_mut()
.remove(worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER);
tracking_response(app.clone(), unsigned, StatusCode::UNAUTHORIZED).await;
let mut tampered = request();
*tampered.body_mut() = Body::from("{}");
tracking_response(app.clone(), tampered, StatusCode::UNAUTHORIZED).await;
let attacker = RuntimeIdentityMaterial::generate(&source.runtime_id).unwrap();
tracking_response(
app.clone(),
runtime_source_request(
&attacker,
Some(&source.worker_id),
"POST",
&path,
serde_json::to_vec(&input).unwrap(),
),
StatusCode::UNAUTHORIZED,
)
.await;
tracking_response(
app.clone(),
runtime_source_request(
&identity,
None,
"POST",
&path,
serde_json::to_vec(&input).unwrap(),
),
StatusCode::FORBIDDEN,
)
.await;
let saved = tracking_response(app.clone(), request(), StatusCode::OK).await;
assert_eq!(
tracking_response(app.clone(), request(), StatusCode::OK).await,
saved
);
let conn = rusqlite::Connection::open(&api.config.database_path).unwrap();
let binding:String=conn.query_row("SELECT binding_id FROM workspace_runtime_bindings WHERE workspace_id=?1 AND runtime_id=?2",rusqlite::params![TEST_WORKSPACE_ID,source.runtime_id],|r| r.get(0)).unwrap();
let receipt_source:String=conn.query_row("SELECT source_context_json FROM typed_ticket_assignment_tracking_operations WHERE workspace_id=?1 AND operation_key='runtime-answer'",[TEST_WORKSPACE_ID],|r| r.get(0)).unwrap();
let receipt_source: serde_json::Value = serde_json::from_str(&receipt_source).unwrap();
assert_eq!(
receipt_source["source"]["authentication_binding_id"],
binding
);
assert_eq!(receipt_source["source"]["worker_id"], source.worker_id);
conn.execute("UPDATE workspace_runtime_bindings SET binding_id='new-answer-binding' WHERE workspace_id=?1 AND runtime_id=?2",rusqlite::params![TEST_WORKSPACE_ID,source.runtime_id]).unwrap();
tracking_response(app.clone(), request(), StatusCode::CONFLICT).await;
let foreign = path.replace(TEST_WORKSPACE_ID, "foreign-workspace");
let response = app
.oneshot(runtime_source_request(
&identity,
Some(&source.worker_id),
"POST",
&foreign,
serde_json::to_vec(&input).unwrap(),
))
.await
.unwrap();
assert!(!response.status().is_success());
assert_eq!(
browser_ticket_backend(&api)
.unwrap()
.show(ticket.into())
.unwrap()
.events
.iter()
.filter(|event| event.kind.as_str() == "assignment_answer")
.count(),
1
);
}
#[tokio::test]
async fn closure_reason_filter_matches_current_judgment_before_paging_and_text_search_retains_old_reasons()
{
let directory = tempfile::tempdir().unwrap();
let api = test_api(directory.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let reference = backend
.create(ticket::NewTicket::new("Reason classification"))
.unwrap();
let mutate = |key: &str, action: ticket::TicketCaseAction| {
let observed = backend.show(reference.id.clone().into()).unwrap();
backend
.update_case(
&reference.id,
ticket::TicketCaseUpdate {
expected_content_digest: ticket::ticket_content_digest(&observed),
expected_state: observed.meta.workflow_state,
expected_closure_operation_key: observed.closure_operation_key,
operation_key: key.into(),
action,
explanation: "Case decision explanation, no reason name here".into(),
references: vec![],
author: Some("human".into()),
},
)
.unwrap();
};
mutate(
"close-not-planned",
ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::NotPlanned,
},
);
for n in 0..8 {
backend
.create(ticket::NewTicket::new(format!("Newer unrelated {n}")))
.unwrap();
}
let query = |value: serde_json::Value| {
api.authority
.query_tickets(serde_json::from_value(value).unwrap())
.unwrap()
};
assert_eq!(
query(json!({"closure_reason":"not_planned","limit":1})).items[0].id,
reference.id
);
mutate(
"correct-duplicate",
ticket::TicketCaseAction::CorrectClosure {
reason: ticket::TicketClosureReason::Duplicate,
},
);
assert!(
query(json!({"closure_reason":"not_planned","limit":1}))
.items
.is_empty()
);
assert_eq!(
query(json!({"closure_reason":"duplicate","limit":1})).items[0].id,
reference.id
);
assert_eq!(
query(json!({"query":"not_planned","limit":1})).items[0].matched_fields,
vec!["closure_reason"]
);
mutate("reopen", ticket::TicketCaseAction::Reopen);
assert!(
query(json!({"closure_reason":"duplicate","limit":1}))
.items
.is_empty()
);
assert_eq!(
query(json!({"query":"duplicate","limit":1})).items[0].id,
reference.id
);
mutate(
"close-custom",
ticket::TicketCaseAction::Close {
reason: ticket::TicketClosureReason::Other("Superseded plan".into()),
},
);
assert_eq!(
query(json!({"closure_reason":{"other":"Superseded plan"},"limit":1})).items[0].id,
reference.id
);
assert!(
query(json!({"closure_reason":{"other":"Other plan"},"limit":1}))
.items
.is_empty()
);
assert_eq!(
query(json!({"query":"Superseded plan","limit":1})).items[0].id,
reference.id
);
}
+68
View File
@@ -1889,6 +1889,30 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
) -> Result<Option<TicketRoleAssignmentRecord>>;
/// Retained responsibility, including ended work and removed principals.
/// Use the active variants for admission, routing, authorization or removal.
fn page_ticket_assignments(
&self,
workspace_id: &str,
ticket_id: &str,
after_assignment_id: Option<&str>,
limit: usize,
) -> Result<ticket::TicketAssignmentsPage>;
fn page_ticket_assignment_diagnostics(
&self,
workspace_id: &str,
ticket_id: &str,
assignment_id: &str,
after_input_event_id: Option<&str>,
after_start_operation_id: Option<&str>,
limit: usize,
) -> Result<server_api::TicketAssignmentDiagnosticsPage>;
fn answer_ticket_assignment(
&self,
workspace_id: &str,
ticket_id: &str,
input: &ticket::TicketAssignmentAnswer,
source: &TicketOrchestratorAssignmentSourceContext,
actor: &str,
) -> Result<ticket::TicketAssignmentAnswerReceipt>;
fn update_ticket_assignment_tracking(
&self,
workspace_id: &str,
@@ -7872,6 +7896,50 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn page_ticket_assignments(
&self,
workspace_id: &str,
ticket_id: &str,
after_assignment_id: Option<&str>,
limit: usize,
) -> Result<ticket::TicketAssignmentsPage> {
assignment_tracking::assignments_page(
self,
workspace_id,
ticket_id,
after_assignment_id,
limit,
)
}
fn page_ticket_assignment_diagnostics(
&self,
workspace_id: &str,
ticket_id: &str,
assignment_id: &str,
after_input_event_id: Option<&str>,
after_start_operation_id: Option<&str>,
limit: usize,
) -> Result<server_api::TicketAssignmentDiagnosticsPage> {
assignment_delivery::diagnostics_page(
self,
workspace_id,
ticket_id,
assignment_id,
after_input_event_id,
after_start_operation_id,
limit,
)
}
fn answer_ticket_assignment(
&self,
workspace_id: &str,
ticket_id: &str,
input: &ticket::TicketAssignmentAnswer,
source: &TicketOrchestratorAssignmentSourceContext,
actor: &str,
) -> Result<ticket::TicketAssignmentAnswerReceipt> {
assignment_tracking::answer(self, workspace_id, ticket_id, input, source, actor)
}
fn update_ticket_assignment_tracking(
&self,
workspace_id: &str,