feat(backend): compose Assign routing with bounded async callers

This commit is contained in:
2026-10-12 06:57:34 +09:00
28 changed files with 5572 additions and 1445 deletions
+14
View File
@@ -1977,6 +1977,7 @@ pub trait ServerApi {
)]
async fn runtime_worker_create_alias(
&self,
#[extension] context: ServerRequestContext,
#[path] runtime_id: String,
#[body] request: RuntimeWorkerSpawnRequest,
) -> Result<RuntimeWorkerSpawnResponse, RepositoryApiError>;
@@ -2006,6 +2007,7 @@ pub trait ServerApi {
)]
async fn runtime_worker_create(
&self,
#[extension] context: ServerRequestContext,
#[path] workspace_id: String,
#[path] runtime_id: String,
#[body] request: RuntimeWorkerSpawnRequest,
@@ -7940,6 +7942,10 @@ pub struct TicketWorkerAttachmentBinding {
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct SetTicketRoleAssignmentRequest {
/// Observed delegation required for Orchestrator Ticket starts; manual starts omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub orchestrator_assignment_id: Option<String>,
pub operation_id: String,
pub principal: TicketAssignmentPrincipal,
pub expected_assignment_id: Option<String>,
@@ -8580,6 +8586,10 @@ pub enum RuntimeWorkerSpawnAcceptanceRequirement {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct RuntimeWorkerTicketAssignmentRequest {
/// Observed delegation required for Orchestrator Ticket starts; manual starts omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub orchestrator_assignment_id: Option<String>,
pub ticket_id: String,
pub operation_id: String,
}
@@ -9909,6 +9919,10 @@ pub struct BrowserWorkerWorkingDirectorySelection {
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct CreateWorkspaceWorkerTicketAssignmentRequest {
/// Observed delegation required for Orchestrator Ticket starts; manual starts omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub orchestrator_assignment_id: Option<String>,
pub ticket_id: String,
pub operation_id: String,
}
@@ -473,6 +473,9 @@ pub struct WorkerLifecycleWorkdirAttachment {
#[derive(Debug, Clone)]
pub struct WorkerLifecycleSpawnRequest {
/// Observed delegation required for Orchestrator Ticket starts.
pub orchestrator_assignment_id: Option<String>,
pub runtime_id: String,
/// Ordered alias-keyed selections forwarded as canonical Workdir attachments.
pub workdir_attachments: Vec<WorkerLifecycleWorkdirAttachment>,
@@ -494,6 +497,7 @@ fn workspace_worker_create_request(
let ticket_assignment = match (request.ticket_id, request.operation_id) {
(Some(ticket_id), Some(operation_id)) => {
Some(CreateWorkspaceWorkerTicketAssignmentRequest {
orchestrator_assignment_id: request.orchestrator_assignment_id,
ticket_id,
operation_id,
})
@@ -729,6 +733,9 @@ struct WorkerSpawnInput {
/// neither the profile nor an initial Flow segment grants review authority.
#[serde(default)]
ticket_id: Option<String>,
/// Observed current Orchestrator delegation, required for a Ticket start by the Orchestrator.
#[serde(default)]
orchestrator_assignment_id: Option<String>,
#[serde(default)]
display_name: Option<String>,
/// Normal typed initial user submission delivered after spawn. An empty
@@ -913,6 +920,10 @@ impl Tool for WorkspaceWorkerTool {
.transpose()?;
self.control
.spawn_worker(WorkerLifecycleSpawnRequest {
orchestrator_assignment_id: input
.orchestrator_assignment_id
.map(|id| authority_id(&id, "orchestrator_assignment_id"))
.transpose()?,
runtime_id: authority_id(&input.runtime_id, "runtime_id")?,
workdir_attachments: vec![WorkerLifecycleWorkdirAttachment {
alias: INITIAL_WORKDIR_ALIAS.to_string(),
@@ -1895,6 +1906,7 @@ mod tests {
#[test]
fn worker_lifecycle_spawn_request_preserves_canonical_attachment_collection() {
let request = workspace_worker_create_request(WorkerLifecycleSpawnRequest {
orchestrator_assignment_id: None,
runtime_id: "runtime-1".to_string(),
workdir_attachments: vec![
WorkerLifecycleWorkdirAttachment {
@@ -1954,6 +1966,7 @@ mod tests {
#[test]
fn worker_lifecycle_spawn_serializes_ticket_assignment_without_workdirs_or_flow() {
let request = workspace_worker_create_request(WorkerLifecycleSpawnRequest {
orchestrator_assignment_id: Some("delegation-742".into()),
runtime_id: "runtime-1".to_string(),
workdir_attachments: Vec::new(),
profile: "project:ticket-worker".to_string(),
@@ -1971,6 +1984,7 @@ mod tests {
value["ticket_assignment"],
serde_json::json!({
"ticket_id": "T-719",
"orchestrator_assignment_id": "delegation-742",
"operation_id": "spawn-ticket-worker:T-719:call-1",
})
);
@@ -21,7 +21,7 @@ use crate::feature::{
const FEATURE_ID: &str = "orchestration";
const TOOL_NAME: &str = "SpawnTicketWorker";
const TOOL_DESCRIPTION: &str = "Spawn and atomically assign a generic Ticket Worker for the natural-language initial request. Profile is optional (defaults to builtin:ticket-worker); Flow and repository/Workdir attachments are optional. Results and return choices follow user intent, not a universal approval or MR gate. Backend validates claims and derives attachment access from Ticket targets; acceptance follows spawn, initial input, assignment, and resource finalization. Profile/Flow selection grants neither trusted review nor merge authority.";
const TOOL_DESCRIPTION: &str = "Spawn and atomically assign a generic Ticket Worker for the natural-language initial request. Profile is optional (defaults to builtin:ticket-worker); Flow and repository/Workdir attachments are optional. Results and return choices follow user intent, not a universal approval or MR gate. Orchestrator starts must supply the observed orchestrator_assignment_id; stale delegations cannot start new work. Assign is enough (no Queue/ready); unresolved dependencies are diagnostics for partial work and never start implicitly. Backend validates claims and derives attachment access from Ticket targets; acceptance follows spawn, initial input, assignment, and resource finalization. Profile/Flow selection grants neither trusted review nor merge authority.";
#[derive(Debug, Default)]
pub struct OrchestrationFeature;
@@ -78,6 +78,9 @@ struct SpawnTicketWorkerWorkdirInput {
#[serde(deny_unknown_fields)]
struct SpawnTicketWorkerInput {
ticket_id: String,
/// Observed current delegation identity. Required by Backend for an Orchestrator start; manual callers omit it.
#[serde(default)]
orchestrator_assignment_id: Option<String>,
runtime_id: String,
/// Optional registered profile selector; omitted/null selects builtin:ticket-worker.
/// Backend resolves the profile and validates claims, not natural-language satisfaction.
@@ -132,6 +135,10 @@ impl Tool for SpawnTicketWorkerTool {
let response = self
.worker_service
.spawn(WorkerLifecycleSpawnRequest {
orchestrator_assignment_id: input
.orchestrator_assignment_id
.map(|id| authority_id(id, "orchestrator_assignment_id"))
.transpose()?,
runtime_id,
workdir_attachments,
profile,
@@ -307,6 +314,7 @@ mod tests {
let service = Arc::new(RecordingService::default());
let mut input = input();
input["flow"] = serde_json::json!("project:investigation");
input["orchestrator_assignment_id"] = serde_json::json!("delegation-742");
input["workdir_attachments"] = serde_json::json!([
{
"alias": "checkout",
@@ -327,6 +335,10 @@ mod tests {
assert_eq!(requests.len(), 1);
let request = &requests[0];
assert_eq!(request.runtime_id, "runtime-1");
assert_eq!(
request.orchestrator_assignment_id.as_deref(),
Some("delegation-742")
);
assert_eq!(request.profile, "project:ticket-worker");
assert_eq!(request.ticket_id.as_deref(), Some("00001KZXN51C7"));
assert_eq!(
@@ -475,6 +487,16 @@ mod tests {
"initial_request must not be empty",
),
("flow", serde_json::json!(" "), "flow must not be empty"),
(
"orchestrator_assignment_id",
serde_json::json!("http://invalid/id"),
"orchestrator_assignment_id must be an authority id",
),
(
"orchestrator_assignment_id",
serde_json::json!(" "),
"orchestrator_assignment_id must not be empty",
),
(
"runtime_id",
serde_json::json!("runtime/1"),
+52 -4
View File
@@ -156,11 +156,15 @@ impl WorkerPrompt {
];
}
/// Model-visible queued Ticket projection shared by Server and TUI backlog attention paths.
/// Model-visible Ticket projection shared by Server and TUI attention paths.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct OrchestratorQueueAttentionTicket {
resource_key: String,
title: String,
assignment_id: Option<String>,
input_event_id: Option<String>,
waiting_reason: Option<String>,
next_action: Option<String>,
}
impl OrchestratorQueueAttentionTicket {
@@ -175,8 +179,28 @@ impl OrchestratorQueueAttentionTicket {
Ok(Self {
resource_key,
title: bounded_queue_attention_text(&title.into(), 240),
assignment_id: None,
input_event_id: None,
waiting_reason: None,
next_action: None,
})
}
/// Committed delegation/input identity, not a handling-status transition.
/// The receiving Worker must reread current authority before a new action.
pub fn with_assignment_input(
mut self,
assignment_id: &str,
input_event_id: &str,
waiting_reason: Option<&str>,
next_action: Option<&str>,
) -> Self {
self.assignment_id = Some(bounded_queue_attention_text(assignment_id, 1024));
self.input_event_id = Some(bounded_queue_attention_text(input_event_id, 1024));
self.waiting_reason = waiting_reason.map(|text| bounded_queue_attention_text(text, 1024));
self.next_action = next_action.map(|text| bounded_queue_attention_text(text, 1024));
self
}
}
/// Shared model-visible context for every Orchestrator backlog attention renderer.
@@ -200,7 +224,7 @@ impl OrchestratorQueueAttentionContext {
}
}
/// Prompt-catalog entries that must share the same backlog-attention body contract.
/// Prompt-catalog entries sharing a bounded Ticket resource-key projection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OrchestratorQueueAttentionPrompt {
Server,
@@ -841,8 +865,10 @@ mod tests {
.orchestrator_queue_attention(OrchestratorQueueAttentionPrompt::Tui, &context)
.unwrap();
assert_eq!(server, tui);
assert!(server.starts_with("Queued Tickets require attention:"));
assert!(server.starts_with("Assigned Tickets have unreceived input:"));
// TUI's final operation model is a separate downstream scope; both
// renderers still consume the same bounded resource-key projection.
assert!(tui.starts_with("Queued Tickets require attention:"));
assert!(server.contains("- T-1 — Ticket 1 with control "));
assert!(!server.contains("T-21"));
assert!(server.contains("were omitted from this notice: 1"));
@@ -859,6 +885,28 @@ mod tests {
}
}
#[test]
fn assignment_attention_carries_input_identity_and_bounded_waiting_context() {
let ticket = OrchestratorQueueAttentionTicket::new("T-742", "Partial work")
.unwrap()
.with_assignment_input(
"assignment-a",
"answer-event",
Some("human\nanswer"),
Some(&"x".repeat(2048)),
);
let context = OrchestratorQueueAttentionContext::new(vec![ticket]);
let output = PromptCatalog::builtins_only()
.unwrap()
.orchestrator_queue_attention(OrchestratorQueueAttentionPrompt::Server, &context)
.unwrap();
assert!(output.contains("Delegation: assignment-a; input: answer-event."));
assert!(output.contains("Waiting reason: human answer"));
assert!(output.contains(&format!("Next action: {}…", "x".repeat(1024))));
assert!(!output.contains(&"x".repeat(1025)));
assert!(output.contains("no ready/Queue step is required"));
}
#[test]
fn queue_attention_prompt_omits_truncation_text_for_complete_list() {
let catalog = PromptCatalog::builtins_only().unwrap();
@@ -0,0 +1,380 @@
//! Durable operational delivery and startup fencing, independent of handling.
use super::*;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TicketAssignmentDelivery {
pub ticket_id: String,
pub assignment_id: String,
pub input_event_id: String,
pub delivery_id: String,
pub request_id: String,
pub waiting_reason: Option<String>,
pub next_action: Option<String>,
}
pub(super) const SCHEMA_SQL: &str = r#"
CREATE TABLE ticket_assignment_delivery_origins (
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, assignment_id TEXT NOT NULL,
input_event_id TEXT NOT NULL, after_ticket_event_index INTEGER NOT NULL,
PRIMARY KEY(workspace_id,assignment_id),
FOREIGN KEY(workspace_id,ticket_id,assignment_id) REFERENCES ticket_worker_assignments(workspace_id,ticket_id,assignment_id)
);
CREATE TABLE ticket_assignment_deliveries (
workspace_id TEXT NOT NULL, assignment_id TEXT NOT NULL, input_event_id TEXT NOT NULL,
delivery_id TEXT NOT NULL, payload_json TEXT NOT NULL CHECK(json_valid(payload_json)),
reserved_source_json TEXT NOT NULL CHECK(json_valid(reserved_source_json)),
PRIMARY KEY(workspace_id,assignment_id,input_event_id), UNIQUE(workspace_id,delivery_id),
FOREIGN KEY(workspace_id,assignment_id) REFERENCES ticket_worker_assignments(workspace_id,assignment_id)
);
CREATE TABLE ticket_assignment_delivery_requests (
workspace_id TEXT NOT NULL, delivery_id TEXT NOT NULL, content TEXT NOT NULL,
PRIMARY KEY(workspace_id,delivery_id),
FOREIGN KEY(workspace_id,delivery_id) REFERENCES ticket_assignment_deliveries(workspace_id,delivery_id)
);
CREATE TRIGGER ticket_assignment_delivery_request_no_update BEFORE UPDATE ON ticket_assignment_delivery_requests
BEGIN SELECT RAISE(ABORT,'Assignment delivery request is immutable'); END;
CREATE TRIGGER ticket_assignment_delivery_request_no_delete BEFORE DELETE ON ticket_assignment_delivery_requests
BEGIN SELECT RAISE(ABORT,'Assignment delivery request is immutable'); END;
CREATE TABLE ticket_assignment_delivery_acknowledgements (
workspace_id TEXT NOT NULL, delivery_id TEXT NOT NULL,
source_json TEXT NOT NULL CHECK(json_valid(source_json)), PRIMARY KEY(workspace_id,delivery_id),
FOREIGN KEY(workspace_id,delivery_id) REFERENCES ticket_assignment_deliveries(workspace_id,delivery_id)
);
CREATE TABLE ticket_assignment_operation_fences (
workspace_id TEXT NOT NULL, operation_id TEXT NOT NULL, orchestrator_assignment_id TEXT NOT NULL,
runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, agent_key TEXT NOT NULL,
authentication_binding_id TEXT NOT NULL, source_json TEXT NOT NULL CHECK(json_valid(source_json)),
PRIMARY KEY(workspace_id,operation_id),
FOREIGN KEY(workspace_id,operation_id) REFERENCES ticket_assignment_operations(workspace_id,operation_id),
FOREIGN KEY(workspace_id,orchestrator_assignment_id) REFERENCES ticket_worker_assignments(workspace_id,assignment_id)
);
CREATE TRIGGER ticket_assignment_delivery_origin_insert
AFTER INSERT ON typed_ticket_assignment_tracking_history
WHEN json_extract(NEW.source_context_json,'$.kind')='assignment'
BEGIN
INSERT INTO ticket_assignment_delivery_origins
VALUES(NEW.workspace_id,NEW.ticket_id,NEW.assignment_id,'assignment-event:' || NEW.event_id,
COALESCE((SELECT MAX(event_index) FROM typed_ticket_events
WHERE workspace_id=NEW.workspace_id AND ticket_id=NEW.ticket_id),-1));
END;
CREATE TRIGGER ticket_assignment_delivery_payload_no_update BEFORE UPDATE ON ticket_assignment_deliveries
BEGIN SELECT RAISE(ABORT,'Assignment delivery payload is immutable'); END;
CREATE TRIGGER ticket_assignment_delivery_payload_no_delete BEFORE DELETE ON ticket_assignment_deliveries
BEGIN SELECT RAISE(ABORT,'Assignment delivery payload is immutable'); END;
CREATE TRIGGER ticket_assignment_delivery_ack_no_update BEFORE UPDATE ON ticket_assignment_delivery_acknowledgements
BEGIN SELECT RAISE(ABORT,'Assignment delivery acknowledgement is immutable'); END;
CREATE TRIGGER ticket_assignment_delivery_ack_no_delete BEFORE DELETE ON ticket_assignment_delivery_acknowledgements
BEGIN SELECT RAISE(ABORT,'Assignment delivery acknowledgement is immutable'); END;
CREATE TRIGGER ticket_assignment_operation_fence_no_update BEFORE UPDATE ON ticket_assignment_operation_fences
BEGIN SELECT RAISE(ABORT,'Assignment startup fence is immutable'); END;
CREATE TRIGGER ticket_assignment_operation_fence_no_delete BEFORE DELETE ON ticket_assignment_operation_fences
BEGIN SELECT RAISE(ABORT,'Assignment startup fence is immutable'); END;
CREATE TRIGGER ticket_assignment_delivery_origin_no_update BEFORE UPDATE ON ticket_assignment_delivery_origins
BEGIN SELECT RAISE(ABORT,'Assignment input origin is immutable'); END;
CREATE TRIGGER ticket_assignment_delivery_origin_no_delete BEFORE DELETE ON ticket_assignment_delivery_origins
BEGIN SELECT RAISE(ABORT,'Assignment input origin is immutable'); END;
"#;
/// Additive Workspace schema only. Native pre-upgrade assignments use their
/// committed initial audit time to exclude earlier Ticket inputs; ambiguous
/// legacy relationships are NOT treated as fresh delegations. Migration inputs
/// retain the exact Queue fence already reconciled by Ticket schema 11.
pub(super) fn install(conn: &Connection) -> Result<()> {
conn.execute_batch(SCHEMA_SQL)?;
conn.execute_batch(r#"
INSERT INTO ticket_assignment_delivery_origins
SELECT h.workspace_id,h.ticket_id,h.assignment_id,'assignment-event:' || h.event_id,
COALESCE((SELECT MAX(e.event_index) FROM typed_ticket_events e
WHERE e.workspace_id=h.workspace_id AND e.ticket_id=h.ticket_id AND e.at<h.at),-1)
FROM typed_ticket_assignment_tracking_history h
WHERE json_extract(h.source_context_json,'$.kind')='assignment';
INSERT INTO ticket_assignment_delivery_origins
SELECT h.workspace_id,h.ticket_id,h.assignment_id,'assignment-event:' || h.event_id,
(SELECT MAX(e.event_index) FROM typed_ticket_events e
JOIN typed_ticket_event_attributes f ON f.workspace_id=e.workspace_id AND f.ticket_id=e.ticket_id AND f.event_index=e.event_index
WHERE e.workspace_id=h.workspace_id AND e.ticket_id=h.ticket_id
AND f.key='orchestrator_assignment_id' AND f.value=h.assignment_id
AND (e.to_state='queued' OR (e.state_field='routing' AND e.reason='queued')))
FROM typed_ticket_assignment_tracking_history h
WHERE json_extract(h.source_context_json,'$.kind')='migration'
AND EXISTS(SELECT 1 FROM typed_ticket_events e JOIN typed_ticket_event_attributes f
ON f.workspace_id=e.workspace_id AND f.ticket_id=e.ticket_id AND f.event_index=e.event_index
WHERE e.workspace_id=h.workspace_id AND e.ticket_id=h.ticket_id
AND f.key='orchestrator_assignment_id' AND f.value=h.assignment_id
AND (e.to_state='queued' OR (e.state_field='routing' AND e.reason='queued')));
"#)?;
Ok(())
}
const INPUTS_SQL: &str = r#"
WITH eligible AS (
SELECT a.*,o.input_event_id,o.after_ticket_event_index
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
JOIN ticket_assignment_delivery_origins o ON o.workspace_id=a.workspace_id AND o.assignment_id=a.assignment_id
JOIN typed_tickets t ON t.workspace_id=a.workspace_id AND t.ticket_id=a.ticket_id
WHERE a.workspace_id=?1 AND a.role='orchestrator' AND a.principal_kind='workspace_agent'
AND a.principal_id='workspace-orchestrator' AND t.workflow_state='open'
AND json_extract(a.tracking_json,'$.status') IN ('waiting','inprogress')
AND NOT EXISTS(SELECT 1 FROM ticket_assignment_work_releases r
WHERE r.workspace_id=a.workspace_id AND r.assignment_id=a.assignment_id)
), inputs AS (
SELECT ticket_id,assignment_id,input_event_id,tracking_json,-1 AS input_order FROM eligible
UNION ALL
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')
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')
)
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
FROM inputs i
LEFT JOIN ticket_assignment_deliveries d ON d.workspace_id=?1 AND d.assignment_id=i.assignment_id AND d.input_event_id=i.input_event_id
WHERE NOT EXISTS(SELECT 1 FROM ticket_assignment_delivery_acknowledgements ack
WHERE ack.workspace_id=?1 AND ack.delivery_id=d.delivery_id)
AND (?3 IS NULL OR i.assignment_id=?3) AND (?4 IS NULL OR i.input_event_id=?4)
"#;
fn decode_payload(payload: &str) -> Result<TicketAssignmentDelivery> {
serde_json::from_str(payload).map_err(|e| Error::Store(e.to_string()))
}
pub(super) fn list_in_tx(
conn: &Connection,
workspace_id: &str,
limit: usize,
) -> Result<Vec<TicketAssignmentDelivery>> {
inputs_in_tx(conn, workspace_id, limit, None, None)
}
fn inputs_in_tx(
conn: &Connection,
workspace_id: &str,
limit: usize,
assignment_id: Option<&str>,
input_event_id: Option<&str>,
) -> Result<Vec<TicketAssignmentDelivery>> {
let sql = format!("{INPUTS_SQL} ORDER BY i.ticket_id,i.assignment_id,i.input_order LIMIT ?2");
let mut statement = conn.prepare(&sql)?;
let rows = statement.query_map(
params![workspace_id, limit as i64, assignment_id, input_event_id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
))
},
)?;
rows.map(|row| {
let (ticket_id, assignment_id, input_event_id, waiting_reason, next_action, payload) = row?;
if let Some(payload) = payload {
return decode_payload(&payload);
}
// Hash the identities, not mutable handling. Scope is Workspace-qualified.
let identities = serde_json::to_string(&(workspace_id, &assignment_id, &input_event_id))
.map_err(|e| Error::Store(e.to_string()))?;
let digest = Sha256::digest(identities.as_bytes());
let hex = digest
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
let delivery_id = format!("assignment-delivery:{hex}");
Ok(TicketAssignmentDelivery {
ticket_id,
assignment_id,
input_event_id,
request_id: delivery_id.clone(),
delivery_id,
waiting_reason,
next_action,
})
})
.collect()
}
fn source_guard(
conn: &Connection,
workspace_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<()> {
assignment_tracking::require_runtime_source(
conn,
workspace_id,
&source.runtime_id,
&source.worker_id,
&source.agent_key,
&source.authentication_binding_id,
)
}
pub(super) fn require_current(
conn: &Connection,
workspace_id: &str,
ticket_id: &str,
assignment_id: &str,
) -> Result<()> {
let current:bool=conn.query_row(r#"SELECT 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
JOIN typed_tickets t ON t.workspace_id=a.workspace_id AND t.ticket_id=a.ticket_id
WHERE c.workspace_id=?1 AND c.ticket_id=?2 AND c.assignment_id=?3 AND c.role='orchestrator'
AND a.principal_kind='workspace_agent' AND a.principal_id='workspace-orchestrator'
AND t.workflow_state='open' AND json_extract(a.tracking_json,'$.status') IN ('waiting','inprogress')
AND EXISTS(SELECT 1 FROM ticket_assignment_delivery_origins o WHERE o.workspace_id=a.workspace_id AND o.assignment_id=a.assignment_id)
AND NOT EXISTS(SELECT 1 FROM ticket_assignment_work_releases r WHERE r.workspace_id=a.workspace_id AND r.assignment_id=a.assignment_id))"#,params![workspace_id,ticket_id,assignment_id],|r|r.get(0))?;
if !current {
return Err(Error::TicketAssignmentConflict(
"Assignment is no longer a current actionable delegation".into(),
));
}
Ok(())
}
pub(super) fn reserve(
store: &SqliteWorkspaceStore,
workspace_id: &str,
assignment_id: &str,
input_event_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<TicketAssignmentDelivery> {
store.with_conn_mut(|conn| {
let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
source_guard(&tx,workspace_id,source)?;
let ticket_id:String=tx.query_row("SELECT ticket_id FROM ticket_worker_assignments WHERE workspace_id=?1 AND assignment_id=?2",params![workspace_id,assignment_id],|r|r.get(0))?;
require_current(&tx,workspace_id,&ticket_id,assignment_id)?;
let saved:Option<String>=tx.query_row("SELECT payload_json FROM ticket_assignment_deliveries WHERE workspace_id=?1 AND assignment_id=?2 AND input_event_id=?3",params![workspace_id,assignment_id,input_event_id],|r|r.get(0)).optional()?;
if let Some(saved)=saved {let result=decode_payload(&saved)?;tx.commit()?;return Ok(result);}
// A caller cannot fabricate an input. Restrict the same common query by
// exact assignment/input identity rather than trusting a scan snapshot.
let candidates=inputs_in_tx(&tx,workspace_id,1,Some(assignment_id),Some(input_event_id))?;
let delivery=candidates.into_iter().find(|d|d.assignment_id==assignment_id && d.input_event_id==input_event_id)
.ok_or_else(||Error::TicketAssignmentConflict("Assignment input is not pending".into()))?;
tx.execute("INSERT INTO ticket_assignment_deliveries VALUES(?1,?2,?3,?4,?5,?6)",params![workspace_id,assignment_id,input_event_id,delivery.delivery_id,serde_json::to_string(&delivery).map_err(|e|Error::Store(e.to_string()))?,serde_json::to_string(source).map_err(|e|Error::Store(e.to_string()))?])?;
tx.commit()?;
Ok(delivery)
})
}
/// Persist the exact rendered Runtime request before I/O. A losing renderer or
/// retry receives the original bytes, not a new payload under the same ID.
pub(super) fn prepare_content(
store: &SqliteWorkspaceStore,
workspace_id: &str,
delivery_id: &str,
content: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<String> {
if content.is_empty() || content.len() > 262144 || content.contains('\0') {
return Err(Error::InvalidInput(
"Assignment delivery content must be nonempty, NUL-free and at most 262144 bytes"
.into(),
));
}
store.with_conn_mut(|conn| {
let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
source_guard(&tx,workspace_id,source)?;
let payload:String=tx.query_row("SELECT payload_json FROM ticket_assignment_deliveries WHERE workspace_id=?1 AND delivery_id=?2",params![workspace_id,delivery_id],|r|r.get(0))?;
let delivery=decode_payload(&payload)?;
require_current(&tx,workspace_id,&delivery.ticket_id,&delivery.assignment_id)?;
let saved:Option<String>=tx.query_row("SELECT content FROM ticket_assignment_delivery_requests WHERE workspace_id=?1 AND delivery_id=?2",params![workspace_id,delivery_id],|r|r.get(0)).optional()?;
let result=if let Some(saved)=saved {saved} else {
tx.execute("INSERT INTO ticket_assignment_delivery_requests VALUES(?1,?2,?3)",params![workspace_id,delivery_id,content])?;
content.to_owned()
};
tx.commit()?;
Ok(result)
})
}
pub(super) fn ack(
store: &SqliteWorkspaceStore,
workspace_id: &str,
delivery_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<()> {
store.with_conn_mut(|conn| {
let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
source_guard(&tx,workspace_id,source)?;
let payload:String=tx.query_row("SELECT payload_json FROM ticket_assignment_deliveries WHERE workspace_id=?1 AND delivery_id=?2",params![workspace_id,delivery_id],|r|r.get(0))?;
let delivery=decode_payload(&payload)?;
require_current(&tx,workspace_id,&delivery.ticket_id,&delivery.assignment_id)?;
let source_json=serde_json::to_string(source).map_err(|e|Error::Store(e.to_string()))?;
let saved:Option<String>=tx.query_row("SELECT source_json FROM ticket_assignment_delivery_acknowledgements WHERE workspace_id=?1 AND delivery_id=?2",params![workspace_id,delivery_id],|r|r.get(0)).optional()?;
if let Some(saved)=saved {
if saved!=source_json {return Err(Error::TicketAssignmentConflict("Delivery acknowledgement source differs from saved receipt".into()));}
} else {tx.execute("INSERT INTO ticket_assignment_delivery_acknowledgements VALUES(?1,?2,?3)",params![workspace_id,delivery_id,source_json])?;}
tx.commit()?;
Ok(())
})
}
/// Must run before receipt replay as well as every bind/finalize write. Absence
/// is the existing manual reservation contract, not inferred delegation.
pub(super) fn operation_guard(
conn: &Connection,
workspace_id: &str,
operation_id: &str,
) -> Result<()> {
let fence=conn.query_row("SELECT o.ticket_id,f.orchestrator_assignment_id,f.runtime_id,f.worker_id,f.agent_key,f.authentication_binding_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 f.operation_id=?2",params![workspace_id,operation_id],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?,r.get::<_,String>(3)?,r.get::<_,String>(4)?,r.get::<_,String>(5)?))).optional()?;
if let Some((ticket_id, assignment_id, runtime_id, worker_id, agent_key, binding_id)) = fence {
assignment_tracking::require_runtime_source(
conn,
workspace_id,
&runtime_id,
&worker_id,
&agent_key,
&binding_id,
)?;
require_current(conn, workspace_id, &ticket_id, &assignment_id)?;
}
Ok(())
}
pub(super) fn save_operation_fence(
conn: &Connection,
workspace_id: &str,
operation_id: &str,
ticket_id: &str,
assignment_id: &str,
source: &TicketAssignmentTrackingSourceContext,
existing: bool,
) -> Result<()> {
source_guard(conn, workspace_id, source)?;
require_current(conn, workspace_id, ticket_id, assignment_id)?;
let source_json = serde_json::to_string(source).map_err(|e| Error::Store(e.to_string()))?;
let saved:Option<(String,String)>=conn.query_row("SELECT orchestrator_assignment_id,source_json FROM ticket_assignment_operation_fences WHERE workspace_id=?1 AND operation_id=?2",params![workspace_id,operation_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
if let Some(saved) = saved {
if saved != (assignment_id.to_owned(), source_json) {
return Err(Error::TicketAssignmentConflict(
"Startup operation assignment/source differs from saved fence".into(),
));
}
} else {
if existing {
return Err(Error::TicketAssignmentConflict(
"Cannot add a delegation fence to an existing manual operation".into(),
));
}
conn.execute(
"INSERT INTO ticket_assignment_operation_fences VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",
params![
workspace_id,
operation_id,
assignment_id,
source.runtime_id,
source.worker_id,
source.agent_key,
source.authentication_binding_id,
source_json
],
)?;
}
Ok(())
}
@@ -0,0 +1,794 @@
// Included in store::tests: exercise persisted transaction boundaries, not Runtime I/O.
fn delivery_store_external_event(
store: &SqliteWorkspaceStore,
index: i64,
kind: &str,
state_field: Option<&str>,
) {
store.with_conn(|conn| {
conn.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,author,at,state_field,body) VALUES('w','tracking',?1,?2,'user','2026-10-12T00:00:00Z',?3,'Answer arrived')",params![index,kind,state_field])?;
Ok(())
}).unwrap();
}
fn delivery_store_reserve_initial(
store: &SqliteWorkspaceStore,
source: &TicketAssignmentTrackingSourceContext,
) -> TicketAssignmentDelivery {
let inputs = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap();
assert_eq!(inputs.len(), 1);
store
.reserve_ticket_assignment_delivery(
"w",
&inputs[0].assignment_id,
&inputs[0].input_event_id,
source,
)
.unwrap()
}
#[tokio::test]
async fn assignment_delivery_waiting_self_reports_do_not_busy_loop_and_answers_are_pending() {
let (_directory, store) = tracking_store_fixture().await;
// Existing history before Assign must not become a new consideration input.
delivery_store_external_event(&store, 0, "comment", None);
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
let initial = delivery_store_reserve_initial(&store, &source);
store
.ack_ticket_assignment_delivery("w", &initial.delivery_id, &source)
.unwrap();
let mut update = tracking_store_input("a", "event-assign-a", "waiting-a");
update.tracking.status = TicketAssignmentTrackingStatus::Waiting;
update.tracking.waiting_reason = Some("Waiting for owner answer".into());
update.tracking.next_action = Some("Read answer and decide".into());
store
.update_ticket_assignment_tracking("w", "tracking", &update, &source)
.unwrap();
assert!(
store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.is_empty()
);
delivery_store_external_event(
&store,
1,
"assignment_tracking",
Some("assignment_tracking"),
);
delivery_store_external_event(&store, 2, "comment", None);
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].waiting_reason.as_deref(),
Some("Waiting for owner answer")
);
assert_eq!(
pending[0].next_action.as_deref(),
Some("Read answer and decide")
);
let answer = store
.reserve_ticket_assignment_delivery("w", "a", "ticket-event:2", &source)
.unwrap();
store
.ack_ticket_assignment_delivery("w", &answer.delivery_id, &source)
.unwrap();
assert!(
store
.list_pending_ticket_assignment_deliveries("w", 100)
.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);
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");
let saved = store
.get_ticket_assignment_tracking("w", "tracking", "a")
.unwrap()
.unwrap();
assert_eq!(
saved.tracking.unwrap().status,
TicketAssignmentTrackingStatus::Waiting
);
}
#[tokio::test]
async fn assignment_delivery_response_loss_returns_immutable_payload_after_progress_changes_and_reopen()
{
let (directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
let initial = delivery_store_reserve_initial(&store, &source);
let mut update = tracking_store_input("a", "event-assign-a", "waiting-a");
update.tracking.status = TicketAssignmentTrackingStatus::Waiting;
update.tracking.waiting_reason = Some("New reason not part of original delivery".into());
store
.update_ticket_assignment_tracking("w", "tracking", &update, &source)
.unwrap();
drop(store);
let store = SqliteWorkspaceStore::open(directory.path().join("workspace.db")).unwrap();
assert_eq!(
store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap(),
vec![initial.clone()]
);
assert_eq!(
store
.reserve_ticket_assignment_delivery("w", "a", &initial.input_event_id, &source)
.unwrap(),
initial
);
store
.ack_ticket_assignment_delivery("w", &initial.delivery_id, &source)
.unwrap();
store
.ack_ticket_assignment_delivery("w", &initial.delivery_id, &source)
.unwrap();
assert!(
store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.is_empty()
);
store
.with_conn(|conn| {
assert!(
conn.execute(
"UPDATE ticket_assignment_deliveries SET payload_json='{}'",
[]
)
.is_err()
);
assert!(
conn.execute(
"DELETE FROM ticket_assignment_delivery_acknowledgements",
[]
)
.is_err()
);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn assignment_delivery_simultaneous_reservations_converge_to_one_receipt() {
let (directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let pending = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.remove(0);
let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
let threads = (0..2)
.map(|_| {
let barrier = barrier.clone();
let path = directory.path().join("workspace.db");
let input = pending.input_event_id.clone();
std::thread::spawn(move || {
let store = SqliteWorkspaceStore::open(path).unwrap();
barrier.wait();
store
.reserve_ticket_assignment_delivery("w", "a", &input, &tracking_store_source())
.unwrap()
})
})
.collect::<Vec<_>>();
let receipts = threads
.into_iter()
.map(|t| t.join().unwrap())
.collect::<Vec<_>>();
assert_eq!(receipts[0], receipts[1]);
store
.with_conn(|conn| {
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_deliveries",
[],
|r| r.get(0),
)?;
assert_eq!(count, 1);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn assignment_delivery_done_close_reopen_and_unknown_history_do_not_revive_delegation() {
for change in ["done", "close-reopen", "unknown"] {
let (_directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
let initial = delivery_store_reserve_initial(&store, &source);
match change {
"done" => {
store
.update_ticket_assignment_tracking(
"w",
"tracking",
&tracking_store_input("a", "event-assign-a", "done-a"),
&source,
)
.unwrap();
}
"close-reopen" => {
store.with_conn(|conn| {conn.execute("UPDATE typed_tickets SET workflow_state='closed' WHERE ticket_id='tracking'",[])?;conn.execute("UPDATE typed_tickets SET workflow_state='open' WHERE ticket_id='tracking'",[])?;Ok(())}).unwrap();
}
_ => {
store.with_conn(|conn| {conn.execute("UPDATE ticket_worker_assignments SET tracking_json=NULL,last_tracking_event_id=NULL WHERE assignment_id='a'",[])?;Ok(())}).unwrap();
}
}
delivery_store_external_event(&store, 1, "comment", None);
assert!(
store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.is_empty(),
"{change}"
);
assert!(
store
.reserve_ticket_assignment_delivery("w", "a", &initial.input_event_id, &source)
.is_err(),
"{change}"
);
}
}
#[tokio::test]
async fn assignment_delivery_clear_reassign_fences_scanned_and_reserved_inputs() {
let (_directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
let old = delivery_store_reserve_initial(&store, &source);
store
.clear_current_ticket_role_assignment(
"w",
"tracking",
TicketAssignmentRole::Orchestrator,
"a",
"clear-a",
"clear-a",
"owner",
"2",
None,
)
.unwrap();
let new = tracking_store_assign(&store, "b", "assign-b");
assert_eq!(new.assignment_id, "b");
assert!(
store
.reserve_ticket_assignment_delivery("w", "a", &old.input_event_id, &source)
.is_err()
);
assert!(
store
.ack_ticket_assignment_delivery("w", &old.delivery_id, &source)
.is_err()
);
let pending = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].assignment_id, "b");
assert_ne!(pending[0].request_id, old.request_id);
store
.with_conn(|conn| {
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM ticket_worker_assignments WHERE role='orchestrator'",
[],
|r| r.get(0),
)?;
assert_eq!(count, 2);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn assignment_delivery_old_worker_or_authenticated_binding_cannot_reserve_or_ack() {
for change in ["replacement", "revoke", "binding-change"] {
let (_directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
let old = delivery_store_reserve_initial(&store, &source);
store.with_conn(|conn| {
match change {
"replacement" => {conn.execute("UPDATE worker_singleton_owners SET worker_id='replacement' WHERE workspace_id='w'",[])?;},
"revoke" => {conn.execute("UPDATE workspace_runtime_bindings SET revoked_at='2' WHERE workspace_id='w'",[])?;},
_ => {conn.execute("UPDATE workspace_runtime_bindings SET binding_id='new-binding' WHERE workspace_id='w'",[])?;},
} Ok(())
}).unwrap();
assert!(
matches!(
store.reserve_ticket_assignment_delivery("w", "a", &old.input_event_id, &source),
Err(Error::WorkspacePermissionDenied(_))
),
"{change}"
);
assert!(
matches!(
store.ack_ticket_assignment_delivery("w", &old.delivery_id, &source),
Err(Error::WorkspacePermissionDenied(_))
),
"{change}"
);
if change == "replacement" {
let mut replacement = source.clone();
replacement.worker_id = "replacement".into();
assert_eq!(
store
.reserve_ticket_assignment_delivery("w", "a", &old.input_event_id, &replacement)
.unwrap(),
old
);
store
.ack_ticket_assignment_delivery("w", &old.delivery_id, &replacement)
.unwrap();
assert_eq!(
store
.list_current_ticket_assignment_tracking("w", "tracking")
.unwrap()
.len(),
1
);
}
}
}
fn delivery_store_start_reserve(
store: &SqliteWorkspaceStore,
key: &str,
assignment: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<bool> {
store.reserve_ticket_assignment_operation_for_orchestrator(
"w",
key,
"tracking",
"r",
Some("replacement"),
"start-input",
"3",
assignment,
source,
)
}
#[tokio::test]
async fn assignment_startup_clear_reassign_fences_reservation_replay_bind_and_finalize() {
for change in ["clear", "reassign", "close-reopen"] {
let (_directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
assert!(delivery_store_start_reserve(&store, "start-a", "a", &source).unwrap());
assert!(!delivery_store_start_reserve(&store, "start-a", "a", &source).unwrap());
if change == "close-reopen" {
store.with_conn(|conn| {conn.execute("UPDATE typed_tickets SET workflow_state='closed' WHERE ticket_id='tracking'",[])?;conn.execute("UPDATE typed_tickets SET workflow_state='open' WHERE ticket_id='tracking'",[])?;Ok(())}).unwrap();
} else {
// Clear/reassign must not be blocked by a pending startup claim:
// assignment authority can invalidate it without Runtime side effects.
store
.clear_current_ticket_role_assignment(
"w",
"tracking",
TicketAssignmentRole::Orchestrator,
"a",
"clear-a",
"clear-a",
"owner",
"4",
None,
)
.unwrap();
if change == "reassign" {
tracking_store_assign(&store, "b", "assign-b");
}
}
assert!(
delivery_store_start_reserve(&store, "start-a", "a", &source).is_err(),
"{change}"
);
assert!(
delivery_store_start_reserve(&store, "start-new", "a", &source).is_err(),
"{change}"
);
assert!(
store
.validate_ticket_assignment_operation_fence("w", "start-a")
.is_err(),
"{change}"
);
assert!(
store
.bind_ticket_assignment_operation_worker("w", "start-a", "replacement")
.is_err(),
"{change}"
);
let record = TicketWorkerAssignmentRecord {
workspace_id: "w".into(),
ticket_id: "tracking".into(),
assignment_id: "target-a".into(),
worker: RuntimeWorkerRef::new("r", "replacement"),
assigned_by: "owner".into(),
assigned_at: "4".into(),
};
assert!(
store
.set_current_ticket_worker_assignment(
&record,
None,
"worker-event",
"start-a",
false
)
.is_err(),
"{change}"
);
assert!(
store
.get_active_ticket_role_assignment("w", "tracking", TicketAssignmentRole::Worker)
.unwrap()
.is_none()
);
}
}
#[tokio::test]
async fn assignment_startup_replaced_worker_cannot_start_or_replay_saved_operations() {
let (_directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
delivery_store_start_reserve(&store, "start-a", "a", &source).unwrap();
store
.with_conn(|conn| {
conn.execute(
"UPDATE worker_singleton_owners SET worker_id='replacement' WHERE workspace_id='w'",
[],
)?;
Ok(())
})
.unwrap();
assert!(matches!(
delivery_store_start_reserve(&store, "start-a", "a", &source),
Err(Error::WorkspacePermissionDenied(_))
));
assert!(matches!(
delivery_store_start_reserve(&store, "start-new", "a", &source),
Err(Error::WorkspacePermissionDenied(_))
));
assert!(matches!(
store.validate_ticket_assignment_operation_fence("w", "start-a"),
Err(Error::WorkspacePermissionDenied(_))
));
assert!(matches!(
store.bind_ticket_assignment_operation_worker("w", "start-a", "replacement"),
Err(Error::WorkspacePermissionDenied(_))
));
// Even the new Worker cannot adopt the old sender's startup operation key.
let mut replacement = source.clone();
replacement.worker_id = "replacement".into();
assert!(delivery_store_start_reserve(&store, "start-a", "a", &replacement).is_err());
}
#[tokio::test]
async fn assignment_startup_saved_fence_cannot_be_bypassed_or_attached_to_manual_receipt() {
let (_directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
delivery_store_start_reserve(&store, "start-a", "a", &source).unwrap();
assert!(
store
.reserve_ticket_assignment_operation(
"w",
"start-a",
"tracking",
"r",
Some("replacement"),
"start-input",
"3"
)
.is_err()
);
store.with_conn(|conn| {assert!(conn.execute("UPDATE ticket_assignment_operation_fences SET orchestrator_assignment_id='different'",[]).is_err());Ok(())}).unwrap();
store
.fail_ticket_assignment_operation("w", "start-a", "explicit cancellation")
.unwrap();
store
.reserve_ticket_assignment_operation(
"w",
"manual",
"tracking",
"r",
Some("replacement"),
"start-input",
"3",
)
.unwrap();
assert!(delivery_store_start_reserve(&store, "manual", "a", &source).is_err());
}
#[tokio::test]
async fn assignment_delivery_rendered_content_is_immutable_across_restart_and_concurrent_renderers()
{
let (directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
let initial = delivery_store_reserve_initial(&store, &source);
assert_eq!(
store
.prepare_ticket_assignment_delivery_content(
"w",
&initial.delivery_id,
"original rendered prompt",
&source
)
.unwrap(),
"original rendered prompt"
);
drop(store);
let store = SqliteWorkspaceStore::open(directory.path().join("workspace.db")).unwrap();
assert_eq!(
store
.prepare_ticket_assignment_delivery_content(
"w",
&initial.delivery_id,
"different title/handling/config prompt",
&source
)
.unwrap(),
"original rendered prompt"
);
store
.with_conn(|conn| {
assert!(
conn.execute(
"UPDATE ticket_assignment_delivery_requests SET content='changed'",
[]
)
.is_err()
);
assert!(
conn.execute("DELETE FROM ticket_assignment_delivery_requests", [])
.is_err()
);
Ok(())
})
.unwrap();
store
.clear_current_ticket_role_assignment(
"w",
"tracking",
TicketAssignmentRole::Orchestrator,
"a",
"clear-a",
"clear-a",
"owner",
"4",
None,
)
.unwrap();
assert!(
store
.prepare_ticket_assignment_delivery_content(
"w",
&initial.delivery_id,
"original rendered prompt",
&source
)
.is_err()
);
}
#[test]
fn assignment_delivery_schema92_upgrade_uses_reconciled_migration_evidence_not_legacy_history() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("schema92.db");
// A genuine retained Ticket-10/Workspace-91 fixture goes through its actual
// applied tracking migration; no current DB markers or columns are lowered.
let conn = tracking_schema91_fixture(&path);
migrate_ticket_tracking_v91_to_v92(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 92);
let assignments = stored_fixture_rows(&conn, "ticket_worker_assignments");
let operations = stored_fixture_rows(&conn, "ticket_assignment_operations");
let tracking = stored_fixture_rows(&conn, "typed_ticket_assignment_tracking_history");
drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap();
let pending = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap();
assert_eq!(
pending
.iter()
.map(|d| d.ticket_id.as_str())
.collect::<Vec<_>>(),
vec!["pending", "progress"]
);
assert_eq!(
pending[0].input_event_id,
"assignment-event:migration-tracking:a-pending"
);
store
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 93);
assert_eq!(
stored_fixture_rows(conn, "ticket_worker_assignments"),
assignments
);
assert_eq!(
stored_fixture_rows(conn, "ticket_assignment_operations"),
operations
);
assert_eq!(
stored_fixture_rows(conn, "typed_ticket_assignment_tracking_history"),
tracking
);
let violations: i64 =
conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| {
r.get(0)
})?;
assert_eq!(violations, 0);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn assignment_startup_simultaneous_reservations_admit_runtime_effects_once() {
let (directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
let threads = (0..2)
.map(|_| {
let barrier = barrier.clone();
let path = directory.path().join("workspace.db");
std::thread::spawn(move || {
let store = SqliteWorkspaceStore::open(path).unwrap();
barrier.wait();
delivery_store_start_reserve(&store, "start-a", "a", &tracking_store_source())
.unwrap()
})
})
.collect::<Vec<_>>();
let admitted = threads
.into_iter()
.map(|t| t.join().unwrap())
.collect::<Vec<_>>();
assert_eq!(admitted.iter().filter(|result| **result).count(), 1);
store
.with_conn(|conn| {
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_operation_fences",
[],
|r| r.get(0),
)?;
assert_eq!(count, 1);
Ok(())
})
.unwrap();
}
#[test]
fn assignment_delivery_fresh_and_upgraded_operational_schemas_are_identical() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("retained.db");
let conn = tracking_schema91_fixture(&path);
migrate_ticket_tracking_v91_to_v92(&conn).unwrap();
drop(conn);
let upgraded = SqliteWorkspaceStore::open(&path).unwrap();
let fresh = SqliteWorkspaceStore::in_memory().unwrap();
let objects = |store: &SqliteWorkspaceStore| {
store.with_conn(|conn| {
let mut statement=conn.prepare("SELECT type,name,sql FROM sqlite_schema WHERE name LIKE 'ticket_assignment_delivery_%' OR name LIKE 'ticket_assignment_operation_fence%' ORDER BY name")?;
Ok(statement.query_map([],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?)))?.collect::<rusqlite::Result<Vec<_>>>()?)
}).unwrap()
};
assert_eq!(objects(&fresh), objects(&upgraded));
}
#[tokio::test]
async fn assignment_delivery_missing_immutable_request_guard_refuses_startup() {
let (directory, store) = tracking_store_fixture().await;
store
.with_conn(|conn| {
conn.execute(
"DROP TRIGGER ticket_assignment_delivery_request_no_update",
[],
)?;
Ok(())
})
.unwrap();
drop(store);
assert!(
matches!(SqliteWorkspaceStore::open(directory.path().join("workspace.db")),Err(Error::Store(message)) if message.contains("ticket_assignment_delivery_request_no_update"))
);
}
#[tokio::test]
async fn assignment_delivery_fabricated_inputs_and_foreign_workspace_cannot_create_receipts() {
let (_directory, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
assert!(
store
.reserve_ticket_assignment_delivery("w", "a", "ticket-event:9999", &source)
.is_err()
);
assert!(
store
.reserve_ticket_assignment_delivery(
"other",
"a",
"assignment-event:event-assign-a",
&source
)
.is_err()
);
store
.with_conn(|conn| {
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_deliveries",
[],
|r| r.get(0),
)?;
assert_eq!(count, 0);
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn assignment_startup_unfinished_dependencies_remain_diagnostics_not_store_admission_gates() {
let (_directory, store) = manual_worker_start_store().await;
seed_manual_start_ticket(&store, "partial", "open");
seed_manual_start_ticket(&store, "dependency", "open");
store.with_conn(|conn| {
conn.execute("INSERT INTO typed_ticket_relations(workspace_id,ticket_id,kind,target,author,at) VALUES('w','partial','depends_on','dependency','owner','1')",[])?;
Ok(())
}).unwrap();
let assignment = manual_worker_assignment("partial", "partial-assignment");
store
.start_ticket_with_worker_assignment(
&assignment,
"partial-event",
"partial-operation",
None,
)
.unwrap();
assert!(
store
.get_active_ticket_role_assignment("w", "partial", TicketAssignmentRole::Worker)
.unwrap()
.is_some()
);
assert!(
store
.get_active_ticket_role_assignment("w", "dependency", TicketAssignmentRole::Worker)
.unwrap()
.is_none()
);
store
.with_conn(|conn| {
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM typed_ticket_relations WHERE kind='depends_on'",
[],
|r| r.get(0),
)?;
assert_eq!(count, 1);
Ok(())
})
.unwrap();
}
@@ -439,12 +439,12 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
.iter()
.map(|m| m.version)
.collect::<Vec<_>>(),
vec![92]
vec![92, 93]
);
let store = SqliteWorkspaceStore::open(&path).unwrap();
store
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 92);
assert_eq!(current_schema_version(conn)?, 93);
assert_eq!(
conn.query_row(
"SELECT applied_at FROM ticket_schema_migrations WHERE version=10",
@@ -457,7 +457,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
workspace_schema_migration_history(conn)?,
vec![
(91, "workspace schema baseline".into()),
(92, "Ticket-owned assignment tracking".into())
(92, "Ticket-owned assignment tracking".into()),
(93, "Durable assignment deliveries and source-bound startup fences".into())
]
);
for (table, rows) in retained.iter().zip(original.iter()) {
@@ -595,7 +596,7 @@ fn tracking_store_schema91_closed_unknown_assignment_uses_assign_audit_for_first
.is_empty()
);
store.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 92);
assert_eq!(current_schema_version(conn)?, 93);
assert_eq!(conn.query_row("SELECT MAX(version) FROM ticket_schema_migrations", [], |row| row.get::<_,i64>(0))?, 11);
assert_eq!(conn.query_row("SELECT workflow_state FROM typed_tickets WHERE ticket_id='historical-done'", [], |row| row.get::<_,String>(0))?, "closed");
let stored: (Option<String>, Option<String>) = conn.query_row("SELECT tracking_json,last_tracking_event_id FROM ticket_worker_assignments WHERE assignment_id='a-historical-done'", [], |row| Ok((row.get(0)?,row.get(1)?)))?;
+88 -8
View File
@@ -783,12 +783,12 @@ impl SqliteWorkspaceAuthority {
})
})
.transpose()?;
// Responsibility remains visible after work ends; action eligibility must
// use only active unfinished work, including after a Ticket is reopened.
// Work start eligibility uses unfinished work. Delegation authoring and
// clearing use the retained current pointer, including after close.
let active_role_assignments = self
.store
.list_active_ticket_role_assignments(&self.workspace_id, id)?;
let has_orchestrator = active_role_assignments
let has_orchestrator = role_assignments
.iter()
.any(|assignment| assignment.role == TicketAssignmentRole::Orchestrator);
let has_worker = active_role_assignments
@@ -811,7 +811,6 @@ impl SqliteWorkspaceAuthority {
})
.collect::<Result<Vec<_>>>()?;
let write_repository_keys = ticket_write_target_repository_keys(&targets);
let has_blockers = !ticket.relations.blockers.is_empty();
let mut assignment_diagnostics = Vec::new();
if let Some(legacy_assignee) = ticket
.meta
@@ -827,12 +826,18 @@ impl SqliteWorkspaceAuthority {
let action_eligibility = TicketActionEligibility {
// Case openness is not readiness. These are explicit operations;
// occupancy remains an independent resource constraint.
can_assign_orchestrator: !has_orchestrator && !has_worker,
// Delegation neither occupies nor transfers existing Worker work.
// Start eligibility below separately retains unfinished-work guards.
// A fresh operation explicitly delegates/reissues even to the same
// principal. Expected-assignment CAS (when supplied) and source
// authority are checked by the mutation, not by Ticket phase.
can_assign_orchestrator: true,
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,
// Unfinished dependencies remain relation diagnostics, not a blanket
// start refusal. Backend start/assign still validates authorization,
// resource grants, reservations and actual unfinished work.
can_start_manual_worker: ticket.meta.workflow_state == TicketState::Open && !has_worker,
queue_tickets: dependency_check.queue_tickets.clone(),
blockers: action_blockers,
};
@@ -3232,6 +3237,81 @@ mod tests {
);
}
#[tokio::test]
async fn unfinished_dependencies_remain_diagnostic_without_disabling_manual_start() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("workspace.db");
let store = SqliteWorkspaceStore::open(&path).unwrap();
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "w".into(),
owner_account_id: "owner-account".into(),
display_name: "W".into(),
state: "active".into(),
created_at: "1".into(),
updated_at: "1".into(),
})
.await
.unwrap();
let backend = SqliteTicketBackend::open_verified(&path, "w").unwrap();
let dependent = backend
.create(ticket::NewTicket::new(
"Partial investigation without resources",
))
.unwrap();
let prerequisite = backend
.create(ticket::NewTicket::new("Unfinished prerequisite"))
.unwrap();
backend
.add_ticket_relation(
dependent.id.clone().into(),
ticket::NewTicketRelation {
kind: ticket::TicketRelationKind::DependsOn,
target: prerequisite.id.clone(),
note: Some("Partial work can proceed".into()),
author: Some("owner-account".into()),
},
)
.unwrap();
let authority = SqliteWorkspaceAuthority::new(&path, "w").unwrap();
let detail = authority.ticket(&dependent.id).unwrap();
assert!(detail.targets.is_empty());
assert!(detail.action_eligibility.can_start_manual_worker);
assert!(detail.action_eligibility.blockers.is_empty());
assert_eq!(detail.relations.blockers.len(), 1);
assert_eq!(
detail.relations.blockers[0].blocking_ticket,
prerequisite.id
);
assert_eq!(detail.relations.outgoing[0].kind, "depends_on");
let blocked = authority
.query_tickets(TicketQueryRequest {
attention: vec![server_api::TicketAttention::Blocked],
..TicketQueryRequest::default()
})
.unwrap();
assert_eq!(blocked.items.len(), 1);
assert_eq!(blocked.items[0].id, dependent.id);
assert_eq!(blocked.items[0].unresolved_blocker_count, 1);
let prerequisite_detail = authority.ticket(&prerequisite.id).unwrap();
assert!(prerequisite_detail.assignments.is_empty());
assert_eq!(prerequisite_detail.state.as_str(), "open");
rusqlite::Connection::open(&path)
.unwrap()
.execute(
"UPDATE typed_tickets SET workflow_state='closed' WHERE ticket_id=?1",
[&dependent.id],
)
.unwrap();
assert!(
!authority
.ticket(&dependent.id)
.unwrap()
.action_eligibility
.can_start_manual_worker
);
}
#[tokio::test]
async fn manual_worker_eligibility_accepts_explicit_states_without_reviving_released_work() {
let dir = tempfile::tempdir().unwrap();
+5
View File
@@ -453,6 +453,10 @@ pub struct WorkerSpawnWorkingDirectoryRequest {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WorkerTicketAssignmentRequest {
/// Observed delegation required for Orchestrator Ticket starts; manual starts omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub orchestrator_assignment_id: Option<String>,
pub ticket_id: String,
pub operation_id: String,
}
@@ -7653,6 +7657,7 @@ mod tests {
fn worker_spawn_idempotency_fingerprint_covers_canonical_initial_submit() {
let mut request = embedded_spawn_request();
request.ticket_assignment = Some(WorkerTicketAssignmentRequest {
orchestrator_assignment_id: None,
ticket_id: "00001KVZSGT0Q".to_string(),
operation_id: "operation-1".to_string(),
});
File diff suppressed because it is too large Load Diff
@@ -448,6 +448,7 @@ async fn pending_manual_worker_claim_reconciles_same_runtime_restore_and_commits
.unwrap();
let admitted_content_digest = api.authority.ticket(&ticket_id).unwrap().content_digest;
let request = server_api::SetTicketRoleAssignmentRequest {
orchestrator_assignment_id: None,
operation_id: "claim-runtime-reconciliation".into(),
principal: server_api::TicketAssignmentPrincipal::Worker {
runtime_id: worker.runtime_id.clone(),
@@ -1287,23 +1287,33 @@ async fn uncertain_launch_replay_after_target_edit_keeps_claim_and_blocks_compet
// 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() {
async fn assignment_inputs_replace_queue_interest_without_reviving_released_work() {
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")
fixture
.api
.store
.list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100)
.unwrap()
.is_none()
.iter()
.any(|input| input.ticket_id == queued_id)
== true
);
backend
.queue_ready(queued_id.clone().into(), "fixture")
.unwrap();
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &queued_id, "open")
fixture
.api
.store
.list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100)
.unwrap()
.is_some()
.iter()
.any(|input| input.ticket_id == queued_id)
== true
);
fixture.case_action(
&queued_id,
@@ -1313,9 +1323,14 @@ async fn orchestrator_interest_requires_queue_receipt_or_unfinished_work_not_jus
},
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &queued_id, "closed")
fixture
.api
.store
.list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100)
.unwrap()
.is_none()
.iter()
.any(|input| input.ticket_id == queued_id)
== false
);
fixture.case_action(
&queued_id,
@@ -1323,9 +1338,14 @@ async fn orchestrator_interest_requires_queue_receipt_or_unfinished_work_not_jus
ticket::TicketCaseAction::Reopen,
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &queued_id, "open")
fixture
.api
.store
.list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100)
.unwrap()
.is_none()
.iter()
.any(|input| input.ticket_id == queued_id)
== false
);
let active_id = fixture.ticket();
@@ -1347,9 +1367,14 @@ async fn orchestrator_interest_requires_queue_receipt_or_unfinished_work_not_jus
.all(|event| !event.attributes.contains_key("queue_root_ticket"))
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &active_id, "open")
fixture
.api
.store
.list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100)
.unwrap()
.is_some()
.iter()
.any(|input| input.ticket_id == active_id)
== true
);
fixture.case_action(
&active_id,
@@ -1380,8 +1405,13 @@ async fn orchestrator_interest_requires_queue_receipt_or_unfinished_work_not_jus
.is_none()
);
assert!(
orchestrator_interested(&fixture.api, TEST_WORKSPACE_ID, &active_id, "open")
fixture
.api
.store
.list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100)
.unwrap()
.is_none()
.iter()
.any(|input| input.ticket_id == active_id)
== false
);
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,752 @@
// Include from server.rs's `tests` module as a child module:
// mod assignment_start_tests { include!("server_assignment_start_tests.rs"); }
// These exercise the authenticated router with the existing in-process Runtime
// adapter; they do not start product processes or claim model/E2E coverage.
use super::*;
struct StartSource {
identity: RuntimeIdentityMaterial,
worker: RuntimeWorkerRef,
}
async fn start_source(api: &mut WorkspaceApi, runtime: &WorkdirlessFixtureRuntime) -> StartSource {
let runtime_id = WorkdirlessFixtureRuntime::RUNTIME_ID;
let identity = RuntimeIdentityMaterial::generate(runtime_id).unwrap();
configure_runtime_request_auth(api, &identity, runtime_id);
let memory = api
.config_store
.get_workspace_memory_settings(TEST_WORKSPACE_ID)
.unwrap();
let reservation = api
.config_store
.reserve_worker_create(
TEST_WORKSPACE_ID,
runtime_id,
"start-orchestrator",
"sha256:start-orchestrator",
Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY),
&memory,
)
.unwrap();
let mut summary = WorkdirlessFixtureRuntime::worker_summary(
WorkerCreateBinding {
worker_id: reservation.worker_id,
create_fingerprint: reservation.create_fingerprint,
},
&WorkerSpawnRequest {
requested_worker_name: Some("Start Orchestrator".into()),
singleton_key: Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.into()),
intent: WorkerSpawnIntent::WorkspaceCompanion,
acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
expected_segments: 0,
},
profile: ProfileSelector::Builtin("builtin:orchestrator".into()),
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
resolved_workspace_api: None,
resolved_memory_settings: None,
resolved_subjektiv_attached: false,
resolved_control_operation: None,
},
);
summary.tags = vec!["orchestrator".into(), "singleton".into()];
let worker = summary.worker.clone();
runtime.workers.lock().unwrap().push(summary);
seed_worker_source_member(api, runtime_id, &worker.worker_id);
api.config_store
.complete_worker_create_reservation(TEST_WORKSPACE_ID, worker.worker_id.parse().unwrap())
.unwrap();
assert_eq!(find_workspace_orchestrator(api).unwrap().worker, worker);
StartSource { identity, worker }
}
fn add_unfinished_start_dependency(api: &WorkspaceApi, ticket_id: &str) -> String {
let backend = browser_ticket_backend(api).unwrap();
let dependency = backend
.create(ticket::NewTicket::new("Unfinished prerequisite"))
.unwrap();
backend
.add_ticket_relation(
ticket_id.into(),
ticket::NewTicketRelation {
kind: ticket::TicketRelationKind::DependsOn,
target: dependency.id.clone(),
note: Some("Partial work is intentional".into()),
author: Some("owner".into()),
},
)
.unwrap();
dependency.id
}
fn start_http_request(
source: Option<&StartSource>,
token: &str,
method: &str,
path: &str,
body: Value,
) -> Request<Body> {
let body = serde_json::to_vec(&body).unwrap();
match source {
Some(source) => runtime_source_request(
&source.identity,
Some(&source.worker.worker_id),
method,
path,
body,
),
None => Request::builder()
.method(method)
.uri(path)
.header(CONTENT_TYPE, "application/json")
.header("authorization", format!("Bearer {token}"))
.body(Body::from(body))
.unwrap(),
}
}
async fn start_http_response(
api: &WorkspaceApi,
request: Request<Body>,
expected: StatusCode,
) -> Value {
let response = build_router(api.clone()).oneshot(request).await.unwrap();
let status = response.status();
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(status, expected, "{}", String::from_utf8_lossy(&body));
serde_json::from_slice(&body)
.unwrap_or_else(|_| serde_json::json!({"message": String::from_utf8_lossy(&body)}))
}
fn observed_start_delegation(
api: &WorkspaceApi,
ticket_id: &str,
source: Option<&StartSource>,
) -> Option<String> {
source.map(|_| {
api.store
.get_current_ticket_role_assignment(
TEST_WORKSPACE_ID,
ticket_id,
TicketAssignmentRole::Orchestrator,
)
.unwrap()
.unwrap()
.assignment_id
})
}
fn start_worker_body(
ticket_id: Option<&str>,
attachments: Value,
delegation: Option<&str>,
) -> Value {
let mut body = serde_json::json!({
"runtime_id": WorkdirlessFixtureRuntime::RUNTIME_ID,
"display_name": "Partial investigation",
"profile": "builtin:intake",
"initial_submit": [],
"workdir_attachments": attachments,
"feature_connections": {},
"control_operation_id": "partial-control-start"
});
if let Some(ticket_id) = ticket_id {
body["ticket_assignment"] = serde_json::json!({
"ticket_id": ticket_id,
"operation_id": "partial-start"
});
if let Some(delegation) = delegation {
body["ticket_assignment"]["orchestrator_assignment_id"] =
Value::String(delegation.into());
}
}
body
}
fn assert_start_dependency_diagnostic_only(api: &WorkspaceApi, ticket: &str, dependency: &str) {
let Json(detail) = browser_ticket_detail(api, ticket).unwrap();
assert_eq!(detail.relations.blockers.len(), 1);
assert_eq!(detail.relations.blockers[0].blocking_ticket, dependency);
assert!(!detail.action_eligibility.can_start_manual_worker);
assert!(detail.current_worker.is_some());
let Json(prerequisite) = browser_ticket_detail(api, dependency).unwrap();
assert_eq!(prerequisite.state.as_str(), "open");
assert!(prerequisite.assignments.is_empty());
assert!(prerequisite.assignment_tracking.is_empty());
}
#[tokio::test]
async fn unfinished_dependencies_allow_manual_and_orchestrator_resource_free_start_and_assign() {
for orchestrator in [false, true] {
for existing in [false, true] {
let directory = tempfile::tempdir().unwrap();
let mut api = test_api(directory.path()).await;
register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
let runtime = WorkdirlessFixtureRuntime::default();
api.runtime.register_or_replace(runtime.clone());
let source = if orchestrator {
Some(start_source(&mut api, &runtime).await)
} else {
None
};
let token = seed_test_api_token(api.store.as_ref(), "partial-start-owner");
let ticket = browser_ticket_backend(&api)
.unwrap()
.create(ticket::NewTicket::new(
"Partial resource-free investigation",
))
.unwrap();
let dependency = add_unfinished_start_dependency(&api, &ticket.id);
if orchestrator {
assign_test_orchestrator(&api, &ticket.id);
}
let delegation = observed_start_delegation(&api, &ticket.id, source.as_ref());
assert!(
browser_ticket_detail(&api, &ticket.id)
.unwrap()
.0
.action_eligibility
.can_start_manual_worker
);
let create_path = if orchestrator {
format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers")
} else {
format!("/api/w/{TEST_WORKSPACE_ID}/workers")
};
let created = start_http_response(
&api,
start_http_request(
source.as_ref(),
&token,
"POST",
&create_path,
start_worker_body(
(!existing).then_some(ticket.id.as_str()),
serde_json::json!([]),
delegation.as_deref(),
),
),
StatusCode::OK,
)
.await;
let worker_id = created["worker_id"].as_str().unwrap();
assert_eq!(created["worker"]["profile"], "builtin:intake");
if existing {
let path = format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/assignments/worker",
ticket.id
);
let mut body = serde_json::json!({
"operation_id": "partial-existing-assign",
"principal": {
"kind": "worker",
"runtime_id": WorkdirlessFixtureRuntime::RUNTIME_ID,
"worker_id": worker_id
},
"workdir_bindings": []
});
if let Some(delegation) = &delegation {
body["orchestrator_assignment_id"] = Value::String(delegation.clone());
}
let assigned = start_http_response(
&api,
start_http_request(source.as_ref(), &token, "PUT", &path, body.clone()),
StatusCode::OK,
)
.await;
let replay = start_http_response(
&api,
start_http_request(source.as_ref(), &token, "PUT", &path, body),
StatusCode::OK,
)
.await;
assert_eq!(assigned, replay);
}
let worker = RuntimeWorkerRef::new(WorkdirlessFixtureRuntime::RUNTIME_ID, worker_id);
assert!(
api.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
.unwrap()
.is_empty()
);
assert_start_dependency_diagnostic_only(&api, &ticket.id, &dependency);
// Partial work never permits a second unfinished Worker.
let mut competing = start_worker_body(
Some(&ticket.id),
serde_json::json!([]),
delegation.as_deref(),
);
competing["ticket_assignment"]["operation_id"] =
Value::String("competing-start".into());
competing["control_operation_id"] = Value::String("competing-control-start".into());
let spawn_count = runtime.spawn_requests().len();
start_http_response(
&api,
start_http_request(source.as_ref(), &token, "POST", &create_path, competing),
StatusCode::CONFLICT,
)
.await;
assert_eq!(runtime.spawn_requests().len(), spawn_count);
if !orchestrator {
assert!(
api.store
.get_current_ticket_role_assignment(
TEST_WORKSPACE_ID,
&ticket.id,
TicketAssignmentRole::Orchestrator
)
.unwrap()
.is_none()
);
}
}
}
}
#[tokio::test]
async fn unfinished_dependencies_allow_read_only_start_and_assign_without_write_or_command_grants()
{
for orchestrator in [false, true] {
for existing in [false, true] {
let mut fixture = manual_worker_assignment_fixture().await;
let source = if orchestrator {
Some(start_source(&mut fixture.api, &fixture.runtime).await)
} else {
None
};
let api = &fixture.api;
// Spawn finalization observes Runtime Workdir identity; do not use
// the adapter's unrelated default Repository id for this checkout.
fixture.runtime.set_workdir_repository(
&fixture.docs_workdir_id,
&test_repository_id_by_key(api, "docs"),
);
let backend = browser_ticket_backend(api).unwrap();
// Keep only the reference target. Partial investigation needs no writable Repository.
backend
.edit_item(
fixture.ticket_id.clone().into(),
ticket::TicketItemEdit {
targets: Some(ticket::TicketTargetsEdit::Set {
targets: vec![test_ticket_target(
"docs",
"develop",
TicketTargetAccess::ReadOnly,
)],
}),
..Default::default()
},
)
.unwrap();
let dependency = add_unfinished_start_dependency(api, &fixture.ticket_id);
if orchestrator {
assign_test_orchestrator(api, &fixture.ticket_id);
}
let delegation = observed_start_delegation(api, &fixture.ticket_id, source.as_ref());
let token = seed_test_api_token(api.store.as_ref(), "read-only-start-owner");
let worker = if existing {
// Drop the unrelated checkout so the entire attachment set is reference-only.
let main = api
.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
.unwrap()
.into_iter()
.find(|link| link.workdir_id == fixture.main_workdir_id)
.unwrap();
api.store
.detach_worker_workdir_connection(
TEST_WORKSPACE_ID,
&fixture.worker,
&main.alias,
&main.connection_id,
&now_registry_timestamp(),
)
.unwrap();
sync_runtime_worker_workdir_attachments(api, &fixture.worker).unwrap();
let mut body = serde_json::to_value(manual_worker_assignment_request(
&fixture,
"read-only-existing",
))
.unwrap();
if let Some(delegation) = &delegation {
body["orchestrator_assignment_id"] = Value::String(delegation.clone());
}
body["workdir_bindings"] = serde_json::json!([{
"alias": "docs", "working_directory_id": fixture.docs_workdir_id,
"connection_id": api.store.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker).unwrap().into_iter().find(|link| link.workdir_id == fixture.docs_workdir_id && link.unlinked_at.is_none()).unwrap().connection_id
}]);
let path = format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/assignments/worker",
fixture.ticket_id
);
start_http_response(
api,
start_http_request(source.as_ref(), &token, "PUT", &path, body),
StatusCode::OK,
)
.await;
fixture.worker.clone()
} else {
// This matrix tests read-only start, not sharing an occupied
// Workdir. Release the fixture candidate's reference attachment.
let docs = api
.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
.unwrap()
.into_iter()
.find(|link| {
link.workdir_id == fixture.docs_workdir_id && link.unlinked_at.is_none()
})
.unwrap();
api.store
.detach_worker_workdir_connection(
TEST_WORKSPACE_ID,
&fixture.worker,
&docs.alias,
&docs.connection_id,
&now_registry_timestamp(),
)
.unwrap();
sync_runtime_worker_workdir_attachments(api, &fixture.worker).unwrap();
let path = if orchestrator {
format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers")
} else {
format!("/api/w/{TEST_WORKSPACE_ID}/workers")
};
let body = start_worker_body(
Some(&fixture.ticket_id),
serde_json::json!([{
"alias": "docs", "working_directory_id": fixture.docs_workdir_id
}]),
delegation.as_deref(),
);
let created = start_http_response(
api,
start_http_request(source.as_ref(), &token, "POST", &path, body),
StatusCode::OK,
)
.await;
assert_eq!(created["worker"]["profile"], "builtin:intake");
RuntimeWorkerRef::new(
WorkdirlessFixtureRuntime::RUNTIME_ID,
created["worker_id"].as_str().unwrap(),
)
};
let links = api
.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
.unwrap()
.into_iter()
.filter(|link| link.unlinked_at.is_none())
.collect::<Vec<_>>();
assert_eq!(links.len(), 1);
assert_eq!(links[0].workdir_id, fixture.docs_workdir_id);
assert_eq!(
links[0].capabilities,
workdir::WorkdirSessionCapabilities::READ_ONLY
);
if existing {
let attachments = fixture.runtime.logical_attachments(&worker.worker_id);
assert_eq!(attachments.len(), 1);
assert_eq!(
attachments[0].capabilities,
workdir::WorkdirSessionCapabilities::READ_ONLY
);
} else {
// This adapter records spawn inputs separately from the
// replacement-attachment calls used by existing-Worker assign.
let request = fixture.runtime.spawn_requests().last().unwrap().clone();
assert_eq!(request.resolved_workdir_attachments.len(), 1);
assert_eq!(
request.resolved_workdir_attachments[0].working_directory_id,
fixture.docs_workdir_id
);
assert_eq!(
request.resolved_workdir_attachments[0].capabilities,
workdir::WorkdirSessionCapabilities::READ_ONLY
);
}
assert_start_dependency_diagnostic_only(api, &fixture.ticket_id, &dependency);
}
}
}
#[tokio::test]
async fn unfinished_dependencies_do_not_bypass_http_auth_closed_or_resource_binding_guards() {
let fixture = manual_worker_assignment_fixture().await;
let api = &fixture.api;
add_unfinished_start_dependency(api, &fixture.ticket_id);
let token = seed_test_api_token(api.store.as_ref(), "guard-start-owner");
let path = format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/assignments/worker",
fixture.ticket_id
);
let valid =
serde_json::to_value(manual_worker_assignment_request(&fixture, "guarded-assign")).unwrap();
let denied = start_http_response(
api,
start_http_request(None, "invalid-token", "PUT", &path, valid.clone()),
StatusCode::UNAUTHORIZED,
)
.await;
assert!(denied["message"].is_string());
let mut stale = valid.clone();
stale["workdir_bindings"][0]["connection_id"] = Value::String("stale-connection".into());
let denied = start_http_response(
api,
start_http_request(None, &token, "PUT", &path, stale),
StatusCode::CONFLICT,
)
.await;
assert!(denied["message"].as_str().unwrap().contains("connection"));
assert!(
api.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
.unwrap()
.is_none()
);
// The existing case/work-release trigger is separate from relation diagnostics.
rusqlite::Connection::open(&api.config.database_path).unwrap().execute(
"UPDATE typed_tickets SET workflow_state='closed' WHERE workspace_id=?1 AND ticket_id=?2",
rusqlite::params![TEST_WORKSPACE_ID, fixture.ticket_id],
).unwrap();
let denied = start_http_response(
api,
start_http_request(None, &token, "PUT", &path, valid),
StatusCode::CONFLICT,
)
.await;
assert!(
denied["message"].as_str().unwrap().contains("open")
|| denied["message"].as_str().unwrap().contains("closed")
);
assert!(
api.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id)
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn orchestrator_start_and_assign_reject_missing_stale_cleared_or_reissued_delegation() {
for existing in [false, true] {
for invalidation in ["missing", "stale", "clear", "reissue"] {
let directory = tempfile::tempdir().unwrap();
let mut api = test_api(directory.path()).await;
register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
let runtime = WorkdirlessFixtureRuntime::default();
api.runtime.register_or_replace(runtime.clone());
let source = start_source(&mut api, &runtime).await;
let token = seed_test_api_token(api.store.as_ref(), "delegation-owner");
let ticket = browser_ticket_backend(&api)
.unwrap()
.create(ticket::NewTicket::new("Observed delegation start"))
.unwrap();
assign_test_orchestrator(&api, &ticket.id);
let observed = observed_start_delegation(&api, &ticket.id, Some(&source)).unwrap();
let worker = if existing {
Some(
start_http_response(
&api,
start_http_request(
None,
&token,
"POST",
&format!("/api/w/{TEST_WORKSPACE_ID}/workers"),
start_worker_body(None, serde_json::json!([]), None),
),
StatusCode::OK,
)
.await,
)
} else {
None
};
if matches!(invalidation, "clear" | "reissue") {
let clear = format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/assignments/orchestrator?assignment_id={observed}&operation_id=clear-observed",
ticket.id
);
start_http_response(
&api,
start_http_request(None, &token, "DELETE", &clear, Value::Null),
StatusCode::OK,
)
.await;
if invalidation == "reissue" {
let path = format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/assignments/orchestrator",
ticket.id
);
start_http_response(&api, start_http_request(None, &token, "PUT", &path,
serde_json::json!({"operation_id":"reissue-observed", "principal":{"kind":"workspace_agent", "agent_key":"workspace-orchestrator"}})), StatusCode::OK).await;
assert_ne!(
observed_start_delegation(&api, &ticket.id, Some(&source)).unwrap(),
observed
);
}
}
let submitted = match invalidation {
"missing" => None,
"stale" => Some("unrelated-assignment"),
_ => Some(observed.as_str()),
};
let (method, path, body) = if let Some(worker) = worker {
let mut body = serde_json::json!({
"operation_id":"invalid-existing-start",
"principal":{"kind":"worker", "runtime_id":WorkdirlessFixtureRuntime::RUNTIME_ID, "worker_id":worker["worker_id"]},
"workdir_bindings":[]
});
if let Some(submitted) = submitted {
body["orchestrator_assignment_id"] = Value::String(submitted.into());
}
(
"PUT",
format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/assignments/worker",
ticket.id
),
body,
)
} else {
(
"POST",
format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers"),
start_worker_body(Some(&ticket.id), serde_json::json!([]), submitted),
)
};
let spawns = runtime.spawn_requests().len();
let denied = start_http_response(
&api,
start_http_request(Some(&source), &token, method, &path, body),
StatusCode::CONFLICT,
)
.await;
assert!(
denied["message"].as_str().unwrap().contains("assignment")
|| denied["message"].as_str().unwrap().contains("delegation"),
"{invalidation}: {denied}"
);
assert_eq!(runtime.spawn_requests().len(), spawns);
assert!(
api.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket.id)
.unwrap()
.is_none()
);
let operation = if existing {
"invalid-existing-start"
} else {
"partial-start"
};
assert!(
api.store
.get_ticket_assignment_operation(TEST_WORKSPACE_ID, operation)
.unwrap()
.is_none()
);
}
}
}
#[tokio::test]
async fn replaced_orchestrator_cannot_start_or_assign_under_retained_current_delegation() {
for existing in [false, true] {
let directory = tempfile::tempdir().unwrap();
let mut api = test_api(directory.path()).await;
register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
let runtime = WorkdirlessFixtureRuntime::default();
api.runtime.register_or_replace(runtime.clone());
let source = start_source(&mut api, &runtime).await;
let token = seed_test_api_token(api.store.as_ref(), "obsolete-worker-owner");
let ticket = browser_ticket_backend(&api)
.unwrap()
.create(ticket::NewTicket::new(
"Retained delegation after Worker replacement",
))
.unwrap();
assign_test_orchestrator(&api, &ticket.id);
let delegation = observed_start_delegation(&api, &ticket.id, Some(&source)).unwrap();
let worker = if existing {
Some(
start_http_response(
&api,
start_http_request(
None,
&token,
"POST",
&format!("/api/w/{TEST_WORKSPACE_ID}/workers"),
start_worker_body(None, serde_json::json!([]), None),
),
StatusCode::OK,
)
.await,
)
} else {
None
};
// Backend ownership changes independently of the retained assignment row.
let replacement =
RuntimeWorkerRef::new(&source.worker.runtime_id, WorkerId::now_v7().to_string());
seed_worker_source_member(&api, &replacement.runtime_id, &replacement.worker_id);
let mut summary = runtime.workers.lock().unwrap()[0].clone();
summary.worker = replacement.clone();
runtime.workers.lock().unwrap().push(summary);
rusqlite::Connection::open(&api.config.database_path).unwrap().execute(
"UPDATE worker_singleton_owners SET worker_id=?1,generation=generation+1 WHERE workspace_id=?2 AND singleton_key=?3",
rusqlite::params![replacement.worker_id, TEST_WORKSPACE_ID, crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY],
).unwrap();
assert_eq!(
find_workspace_orchestrator(&api).unwrap().worker,
replacement
);
assert_eq!(
observed_start_delegation(&api, &ticket.id, Some(&source)).unwrap(),
delegation
);
let (method, path, body) = if let Some(worker) = worker {
(
"PUT",
format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/assignments/worker",
ticket.id
),
serde_json::json!({
"operation_id":"obsolete-existing-start", "orchestrator_assignment_id":delegation,
"principal":{"kind":"worker", "runtime_id":WorkdirlessFixtureRuntime::RUNTIME_ID, "worker_id":worker["worker_id"]},
"workdir_bindings":[]
}),
)
} else {
(
"POST",
format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers"),
start_worker_body(Some(&ticket.id), serde_json::json!([]), Some(&delegation)),
)
};
let spawns = runtime.spawn_requests().len();
let denied = start_http_response(
&api,
start_http_request(Some(&source), &token, method, &path, body),
// The authenticated Worker-source middleware fences the replaced
// singleton before the start/assign handler is admitted.
StatusCode::UNAUTHORIZED,
)
.await;
assert!(denied["message"].is_string());
assert_eq!(runtime.spawn_requests().len(), spawns);
assert!(
api.store
.get_active_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket.id)
.unwrap()
.is_none()
);
}
}
@@ -68,7 +68,9 @@ async fn runtime_and_account_assign_retries_return_saved_result_after_clear_and_
})).unwrap();
tracking_response(
app.clone(),
request("PUT", &worker_path, worker_input),
Request::builder().method("PUT").uri(&worker_path).header(CONTENT_TYPE, "application/json")
.header("authorization", format!("Bearer {token}"))
.body(Body::from(worker_input)).unwrap(),
StatusCode::OK,
)
.await;
@@ -117,15 +119,15 @@ async fn runtime_and_account_assign_retries_return_saved_result_after_clear_and_
.await,
saved
);
// Only a genuinely new operation is refused; a receipt never restores
// the cleared current pointer, initializes tracking, or starts work.
tracking_response(
// Receipt replay does not recreate a delegation or transfer manual work.
assert_eq!(counts(), before);
let fresh = tracking_response(
app,
request("PUT", &path, input("new-assign")),
StatusCode::CONFLICT,
StatusCode::OK,
)
.await;
assert_eq!(counts(), before);
assert_ne!(fresh["assignment"]["assignment_id"].as_str(), Some(assignment));
assert_eq!(
api.store
.get_ticket_assignment_tracking(TEST_WORKSPACE_ID, &ticket, assignment)
@@ -144,16 +146,9 @@ async fn runtime_and_account_assign_retries_return_saved_result_after_clear_and_
.unwrap(),
active
);
assert!(
api.store
.get_current_ticket_role_assignment(
TEST_WORKSPACE_ID,
&ticket,
TicketAssignmentRole::Orchestrator
)
.unwrap()
.is_none()
);
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());
}
}
+321 -134
View File
@@ -27,8 +27,11 @@ use crate::backend_job::{
use crate::workspace_deletion::WorkspaceDeletionStore;
use crate::{Error, Result};
#[path = "assignment_delivery.rs"]
mod assignment_delivery;
#[path = "assignment_tracking.rs"]
mod assignment_tracking;
pub use assignment_delivery::TicketAssignmentDelivery;
pub use assignment_tracking::TicketOrchestratorAssignmentSourceContext;
pub use ticket::{
TicketAssignmentTracking, TicketAssignmentTrackingHistoryEntry,
@@ -37,7 +40,7 @@ pub use ticket::{
};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 92;
const LATEST_SCHEMA_VERSION: i64 = 93;
const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
@@ -305,6 +308,11 @@ const MIGRATIONS: &[Migration] = &[
name: "Ticket-owned assignment tracking",
apply: migrate_ticket_tracking_v91_to_v92,
},
Migration {
version: 93,
name: "Durable assignment deliveries and source-bound startup fences",
apply: migrate_assignment_delivery_v92_to_v93,
},
];
#[derive(Clone, Copy)]
@@ -1808,6 +1816,52 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
request_fingerprint: &str,
created_at: &str,
) -> Result<bool>;
/// Common scan for committed current delegation inputs, not handling status.
fn list_pending_ticket_assignment_deliveries(
&self,
workspace_id: &str,
limit: usize,
) -> Result<Vec<TicketAssignmentDelivery>>;
/// Frozen payload/request identity for exact response-loss retry.
fn reserve_ticket_assignment_delivery(
&self,
workspace_id: &str,
assignment_id: &str,
input_event_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<TicketAssignmentDelivery>;
/// Freeze exact rendered Runtime request bytes before dispatch.
fn prepare_ticket_assignment_delivery_content(
&self,
workspace_id: &str,
delivery_id: &str,
content: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<String>;
fn ack_ticket_assignment_delivery(
&self,
workspace_id: &str,
delivery_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<()>;
fn reserve_ticket_assignment_operation_for_orchestrator(
&self,
workspace_id: &str,
operation_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: Option<&str>,
request_fingerprint: &str,
created_at: &str,
orchestrator_assignment_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<bool>;
/// Recovery/pre-dispatch check; write boundaries also revalidate this fence.
fn validate_ticket_assignment_operation_fence(
&self,
workspace_id: &str,
operation_id: &str,
) -> Result<()>;
fn bind_ticket_assignment_operation_worker(
&self,
workspace_id: &str,
@@ -7841,92 +7895,90 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
request_fingerprint: &str,
created_at: &str,
) -> Result<bool> {
validate_non_empty("operation_id", operation_id)?;
validate_non_empty("request_fingerprint", request_fingerprint)?;
self.reserve_ticket_assignment_operation_with_fence(
workspace_id,
operation_id,
ticket_id,
runtime_id,
worker_id,
request_fingerprint,
created_at,
None,
)
}
fn reserve_ticket_assignment_operation_for_orchestrator(
&self,
workspace_id: &str,
operation_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: Option<&str>,
request_fingerprint: &str,
created_at: &str,
orchestrator_assignment_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<bool> {
self.reserve_ticket_assignment_operation_with_fence(
workspace_id,
operation_id,
ticket_id,
runtime_id,
worker_id,
request_fingerprint,
created_at,
Some((orchestrator_assignment_id, source)),
)
}
fn validate_ticket_assignment_operation_fence(
&self,
workspace_id: &str,
operation_id: &str,
) -> Result<()> {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some(existing) = read_assignment_operation(&tx, workspace_id, operation_id)? {
if existing.action == "assign"
&& existing.ticket_id == ticket_id
&& existing.runtime_id.as_deref() == Some(runtime_id)
&& (worker_id.is_none()
|| existing
.worker
.as_ref()
.map(|worker| worker.worker_id.as_str())
== worker_id)
&& existing.expected_assignment_id.is_none()
&& existing.request_fingerprint.as_deref() == Some(request_fingerprint)
{
tx.commit()?;
return Ok(false);
}
return Err(Error::TicketAssignmentConflict(format!(
"assignment operation {operation_id} was already used with different input"
)));
}
let worker = worker_id.map(|id| RuntimeWorkerRef::new(runtime_id, id));
ensure_no_competing_ticket_claim(
&tx,
workspace_id,
ticket_id,
worker.as_ref(),
operation_id,
)?;
if let Some(worker) = &worker {
ensure_worker_assignment_available(&tx, workspace_id, worker)?;
let occupied: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments
WHERE workspace_id=?1 AND role='worker' AND runtime_id=?2 AND worker_id=?3)",
params![workspace_id, worker.runtime_id, worker.worker_id],
|row| row.get(0),
)?;
if occupied {
return Err(Error::TicketAssignmentConflict(
"Worker has unfinished Ticket work".into(),
));
}
}
let state: String = tx.query_row(
"SELECT workflow_state FROM typed_tickets WHERE workspace_id=?1 AND ticket_id=?2",
params![workspace_id, ticket_id],
|row| row.get(0),
)?;
if state != "open" {
return Err(Error::TicketAssignmentConflict(format!(
"Ticket `{ticket_id}` cannot claim work in state `{state}`"
)));
}
let occupied: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments
WHERE workspace_id=?1 AND ticket_id=?2 AND role='worker')",
params![workspace_id, ticket_id],
|row| row.get(0),
)?;
if occupied {
return Err(Error::TicketAssignmentConflict(
"Ticket has unfinished Worker work".into(),
));
}
tx.execute(
"INSERT INTO ticket_assignment_operations (
workspace_id,operation_id,action,ticket_id,runtime_id,worker_id,assignment_id,
expected_assignment_id,created_at,request_fingerprint,claim_state)
VALUES(?1,?2,'assign',?3,?4,?5,NULL,NULL,?6,?7,'pending')",
params![
workspace_id,
operation_id,
ticket_id,
runtime_id,
worker_id,
created_at,
request_fingerprint
],
)?;
assignment_delivery::operation_guard(&tx, workspace_id, operation_id)?;
tx.commit()?;
Ok(true)
Ok(())
})
}
fn list_pending_ticket_assignment_deliveries(
&self,
workspace_id: &str,
limit: usize,
) -> Result<Vec<TicketAssignmentDelivery>> {
if !(1..=1000).contains(&limit) {
return Err(Error::InvalidInput(
"Assignment delivery scan limit must be 1..=1000".into(),
));
}
self.with_conn(|conn| assignment_delivery::list_in_tx(conn, workspace_id, limit))
}
fn reserve_ticket_assignment_delivery(
&self,
workspace_id: &str,
assignment_id: &str,
input_event_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<TicketAssignmentDelivery> {
assignment_delivery::reserve(self, workspace_id, assignment_id, input_event_id, source)
}
fn prepare_ticket_assignment_delivery_content(
&self,
workspace_id: &str,
delivery_id: &str,
content: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<String> {
assignment_delivery::prepare_content(self, workspace_id, delivery_id, content, source)
}
fn ack_ticket_assignment_delivery(
&self,
workspace_id: &str,
delivery_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<()> {
assignment_delivery::ack(self, workspace_id, delivery_id, source)
}
fn bind_ticket_assignment_operation_worker(
&self,
@@ -7936,6 +7988,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<()> {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
assignment_delivery::operation_guard(&tx,workspace_id,operation_id)?;
let receipt = read_assignment_operation(&tx, workspace_id, operation_id)?
.ok_or_else(|| Error::TicketAssignmentConflict(format!("missing Ticket claim `{operation_id}`")))?;
// Exact binding replay is safe even after assignment finalization.
@@ -8279,6 +8332,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
assignment_delivery::operation_guard(&tx,&record.workspace_id,operation_id)?;
if let Some(worker) = record.principal.worker() {
ensure_worker_assignment_available(&tx, &record.workspace_id, &worker)?;
}
@@ -8334,29 +8388,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
"manual Worker assignment requires an open Ticket; current state is `{state}`"
)));
}
let unresolved_blockers: i64 = tx.query_row(
"SELECT COUNT(*)
FROM typed_ticket_relations AS relation
JOIN typed_tickets AS related
ON related.workspace_id = relation.workspace_id
AND related.ticket_id = CASE
WHEN relation.ticket_id = ?2 THEN relation.target
ELSE relation.ticket_id
END
WHERE relation.workspace_id = ?1
AND (
(relation.ticket_id = ?2 AND relation.kind = 'depends_on')
OR (relation.target = ?2 AND relation.kind = 'blocks')
)
AND related.workflow_state = 'open'",
params![record.workspace_id, record.ticket_id],
|row| row.get(0),
)?;
if unresolved_blockers != 0 {
return Err(Error::TicketAssignmentConflict(
"manual Worker assignment is blocked by unresolved Ticket relations".to_string(),
));
}
// Relations remain diagnostics; they do not reserve work or block partial starts.
let conflicting: i64 = tx.query_row(
"SELECT COUNT(*) FROM ticket_active_worker_assignments
WHERE workspace_id = ?1 AND ticket_id = ?2
@@ -8578,6 +8610,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<TicketWorkerAssignmentUpdate> {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
assignment_delivery::operation_guard(&tx,&record.workspace_id,operation_id)?;
let removal_blocks_assignment: bool = tx.query_row(
"SELECT EXISTS(
SELECT 1 FROM worker_removal_operations
@@ -11603,6 +11636,119 @@ fn ensure_worker_assignment_available(
}
impl SqliteWorkspaceStore {
fn reserve_ticket_assignment_operation_with_fence(
&self,
workspace_id: &str,
operation_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: Option<&str>,
request_fingerprint: &str,
created_at: &str,
fence: Option<(&str, &TicketAssignmentTrackingSourceContext)>,
) -> Result<bool> {
validate_non_empty("operation_id", operation_id)?;
validate_non_empty("request_fingerprint", request_fingerprint)?;
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
assignment_delivery::operation_guard(&tx,workspace_id,operation_id)?;
if fence.is_none() {
let fenced:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_assignment_operation_fences WHERE workspace_id=?1 AND operation_id=?2)",params![workspace_id,operation_id],|r|r.get(0))?;
if fenced {return Err(Error::TicketAssignmentConflict("Source-bound reservation requires authenticated Orchestrator context".into()));}
}
if let Some(existing) = read_assignment_operation(&tx, workspace_id, operation_id)? {
if existing.action == "assign"
&& existing.ticket_id == ticket_id
&& existing.runtime_id.as_deref() == Some(runtime_id)
&& (worker_id.is_none()
|| existing
.worker
.as_ref()
.map(|worker| worker.worker_id.as_str())
== worker_id)
&& existing.expected_assignment_id.is_none()
&& existing.request_fingerprint.as_deref() == Some(request_fingerprint)
{
if let Some((assignment_id,source))=fence {
assignment_delivery::save_operation_fence(&tx,workspace_id,operation_id,ticket_id,assignment_id,source,true)?;
}
tx.commit()?;
return Ok(false);
}
return Err(Error::TicketAssignmentConflict(format!(
"assignment operation {operation_id} was already used with different input"
)));
}
if let Some((assignment_id,source))=fence {
assignment_tracking::require_runtime_source(&tx,workspace_id,&source.runtime_id,&source.worker_id,&source.agent_key,&source.authentication_binding_id)?;
assignment_delivery::require_current(&tx,workspace_id,ticket_id,assignment_id)?;
}
let worker = worker_id.map(|id| RuntimeWorkerRef::new(runtime_id, id));
ensure_no_competing_ticket_claim(
&tx,
workspace_id,
ticket_id,
worker.as_ref(),
operation_id,
)?;
if let Some(worker) = &worker {
ensure_worker_assignment_available(&tx, workspace_id, worker)?;
let occupied: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments
WHERE workspace_id=?1 AND role='worker' AND runtime_id=?2 AND worker_id=?3)",
params![workspace_id, worker.runtime_id, worker.worker_id],
|row| row.get(0),
)?;
if occupied {
return Err(Error::TicketAssignmentConflict(
"Worker has unfinished Ticket work".into(),
));
}
}
let state: String = tx.query_row(
"SELECT workflow_state FROM typed_tickets WHERE workspace_id=?1 AND ticket_id=?2",
params![workspace_id, ticket_id],
|row| row.get(0),
)?;
if state != "open" {
return Err(Error::TicketAssignmentConflict(format!(
"Ticket `{ticket_id}` cannot claim work in state `{state}`"
)));
}
let occupied: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments
WHERE workspace_id=?1 AND ticket_id=?2 AND role='worker')",
params![workspace_id, ticket_id],
|row| row.get(0),
)?;
if occupied {
return Err(Error::TicketAssignmentConflict(
"Ticket has unfinished Worker work".into(),
));
}
tx.execute(
"INSERT INTO ticket_assignment_operations (
workspace_id,operation_id,action,ticket_id,runtime_id,worker_id,assignment_id,
expected_assignment_id,created_at,request_fingerprint,claim_state)
VALUES(?1,?2,'assign',?3,?4,?5,NULL,NULL,?6,?7,'pending')",
params![
workspace_id,
operation_id,
ticket_id,
runtime_id,
worker_id,
created_at,
request_fingerprint
],
)?;
if let Some((assignment_id,source))=fence {
assignment_delivery::save_operation_fence(&tx,workspace_id,operation_id,ticket_id,assignment_id,source,false)?;
}
tx.commit()?;
Ok(true)
})
}
fn set_ticket_role_assignment_with_source(
&self,
record: &TicketRoleAssignmentRecord,
@@ -11664,6 +11810,7 @@ impl SqliteWorkspaceStore {
};
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
assignment_delivery::operation_guard(&tx,&record.workspace_id,operation_id)?;
if let Some(source)=source {assignment_tracking::require_assignment_source(&tx,&record.workspace_id,source)?;}
if let Some(worker) = record.principal.worker() {
ensure_worker_assignment_available(&tx, &record.workspace_id, &worker)?;
@@ -11699,24 +11846,13 @@ impl SqliteWorkspaceStore {
return Ok(persisted);
}
// This is a constraint on a new delegation, not on a saved result.
// Authenticate the source and resolve its full-input receipt first;
// work that began after the original Assign must not block replay.
if record.role == TicketAssignmentRole::Orchestrator {
let active_worker: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments
WHERE workspace_id=?1 AND ticket_id=?2 AND role='worker')",
params![record.workspace_id, record.ticket_id],
|row| row.get(0),
)?;
if active_worker {
return Err(Error::TicketAssignmentConflict(
"Orchestrator assignment conflicts with an active Worker assignment".into(),
));
}
// Delegation has no Runtime/resource occupancy. It can invalidate
// an in-flight startup fence; existing actual Worker/claim resource
// exclusion is checked independently at the next work admission.
if record.role != TicketAssignmentRole::Orchestrator {
ensure_no_competing_ticket_claim(&tx, &record.workspace_id, &record.ticket_id,
record.principal.worker().as_ref(), operation_id)?;
}
ensure_no_competing_ticket_claim(&tx, &record.workspace_id, &record.ticket_id,
record.principal.worker().as_ref(), operation_id)?;
let current_sql = ticket_role_assignment_select_sql(
"WHERE current.workspace_id = ?1 AND current.ticket_id = ?2 AND current.role = ?3 \
ORDER BY a.assigned_at, a.assignment_id",
@@ -12059,13 +12195,15 @@ fn clear_current_ticket_role_assignment_with_source_in_tx(
}
return Ok(true);
}
ensure_no_competing_ticket_claim(
tx,
workspace_id,
ticket_id,
current.principal.worker().as_ref(),
operation_id,
)?;
if role != TicketAssignmentRole::Orchestrator {
ensure_no_competing_ticket_claim(
tx,
workspace_id,
ticket_id,
current.principal.worker().as_ref(),
operation_id,
)?;
}
let is_current:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_current_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND role=?3 AND assignment_id=?4)",params![workspace_id,ticket_id,role.as_str(),assignment_id],|row|row.get(0))?;
if !is_current {
return Ok(false);
@@ -13700,8 +13838,22 @@ fn migrate_ticket_tracking_v91_to_v92(conn: &Connection) -> Result<()> {
Ok(())
}
fn migrate_assignment_delivery_v92_to_v93(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 92 {
return Err(Error::Store(
"Assignment delivery migration requires schema 92".into(),
));
}
let tx = conn.unchecked_transaction()?;
assignment_delivery::install(&tx)?;
tx.execute("INSERT INTO __yoi_schema_migrations(version,name) VALUES(93,'Durable assignment deliveries and source-bound startup fences')",[])?;
tx.commit()?;
Ok(())
}
fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
create_workspace_schema_structure(conn)?;
assignment_delivery::install(conn)?;
ticket::verify_sqlite_ticket_schema(conn).map_err(|error| {
Error::Store(format!(
"fresh Workspace Ticket schema verification failed: {error}"
@@ -13895,6 +14047,37 @@ fn verify_workspace_resource_constraints(conn: &Connection) -> Result<()> {
}
}
}
if current_schema_version(conn)? >= 93 {
for (kind, name) in [
("table", "ticket_assignment_delivery_origins"),
("table", "ticket_assignment_deliveries"),
("table", "ticket_assignment_delivery_requests"),
("table", "ticket_assignment_delivery_acknowledgements"),
("table", "ticket_assignment_operation_fences"),
("trigger", "ticket_assignment_delivery_origin_insert"),
("trigger", "ticket_assignment_delivery_origin_no_update"),
("trigger", "ticket_assignment_delivery_origin_no_delete"),
("trigger", "ticket_assignment_delivery_payload_no_update"),
("trigger", "ticket_assignment_delivery_payload_no_delete"),
("trigger", "ticket_assignment_delivery_request_no_update"),
("trigger", "ticket_assignment_delivery_request_no_delete"),
("trigger", "ticket_assignment_delivery_ack_no_update"),
("trigger", "ticket_assignment_delivery_ack_no_delete"),
("trigger", "ticket_assignment_operation_fence_no_update"),
("trigger", "ticket_assignment_operation_fence_no_delete"),
] {
let present: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type=?1 AND name=?2)",
params![kind, name],
|row| row.get(0),
)?;
if !present {
return Err(Error::Store(format!(
"Assignment delivery constraint {name} is missing"
)));
}
}
}
Ok(())
}
@@ -14648,6 +14831,7 @@ fn table_columns(conn: &Connection, table_name: &str) -> Result<Vec<String>> {
#[cfg(test)]
mod tests {
include!("assignment_tracking_tests.rs");
include!("assignment_delivery_tests.rs");
include!("store_workspace_config_tests.rs");
fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse {
@@ -16857,6 +17041,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
name: "Ticket open/closed cases and retained routing evidence".to_string()
},
WorkspaceSchemaMigrationStep {version:92, name:"Ticket-owned assignment tracking".into()},
WorkspaceSchemaMigrationStep {version:93, name:"Durable assignment deliveries and source-bound startup fences".into()},
]
);
@@ -16959,6 +17144,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(90, "Runtime binding connection configuration and historical verification archive".to_string()),
(91, "Ticket open/closed cases and retained routing evidence".to_string()),
(92, "Ticket-owned assignment tracking".to_string()),
(93, "Durable assignment deliveries and source-bound startup fences".to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -19556,9 +19742,9 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.map(|step| step.version)
.collect::<Vec<_>>(),
if version == 89 {
vec![90, 91, 92]
vec![90, 91, 92, 93]
} else {
vec![91, 92]
vec![91, 92, 93]
}
);
drop(conn);
@@ -19573,7 +19759,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
assert_eq!(workspace_schema_migration_history(conn)?.into_iter().filter(|(version, _)| *version >= 90).collect::<Vec<_>>(),
vec![(90, "Runtime binding connection configuration and historical verification archive".into()),
(91, "Ticket open/closed cases and retained routing evidence".into()),
(92, "Ticket-owned assignment tracking".into())]);
(92, "Ticket-owned assignment tracking".into()),
(93, "Durable assignment deliveries and source-bound startup fences".into())]);
for (table, rows) in unchanged.iter().zip(before.iter()) {
assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
}
@@ -20291,7 +20478,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
}
#[tokio::test]
async fn pending_claims_fence_all_assignment_writers_and_worker_binding_until_owner_commits() {
async fn pending_claims_fence_work_writers_but_allow_delegation_clear() {
let (_dir, store) = manual_worker_start_store().await;
seed_manual_start_ticket(&store, "first", "open");
seed_manual_start_ticket(&store, "second", "open");
@@ -20343,7 +20530,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
"2",
None
)
.is_err()
.unwrap()
);
assert!(
store
@@ -265,7 +265,7 @@ fn assert_retained_classification_upgrade(workspace_version: i64, component: i64
drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap();
store.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?,92);
assert_eq!(current_schema_version(conn)?,93);
assert_eq!(conn.query_row("SELECT MAX(version) FROM ticket_schema_migrations",[],|row|row.get::<_,i64>(0))?,11);
assert_eq!(fixture_auth_rows(conn),auth_before);
for (table, rows) in preserved_tables.iter().zip(&preserved) {
@@ -493,7 +493,7 @@ fn workspace90_cases9_add_classification_without_reinterpreting_reasoned_history
let store = SqliteWorkspaceStore::open(&path).unwrap();
store
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 92);
assert_eq!(current_schema_version(conn)?,93);
assert_eq!(
conn.query_row(
"SELECT MAX(version) FROM ticket_schema_migrations",
@@ -45,6 +45,11 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
"repository_ssh_host_trust_keys",
"repository_ssh_host_trusts",
"server_secret_objects",
"ticket_assignment_delivery_acknowledgements",
"ticket_assignment_delivery_requests",
"ticket_assignment_deliveries",
"ticket_assignment_delivery_origins",
"ticket_assignment_operation_fences",
"ticket_assignment_operations",
"ticket_assignment_work_releases",
"ticket_assignment_ticket_tombstones",
@@ -479,6 +484,11 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
("ticket_worker_assignments", "typed_ticket_assignment_identity_no_delete"),
("ticket_worker_assignment_events", "typed_ticket_assignment_events_no_delete"),
("ticket_assignment_operations", "typed_ticket_assignment_receipt_no_delete"),
("ticket_assignment_deliveries", "ticket_assignment_delivery_payload_no_delete"),
("ticket_assignment_delivery_requests", "ticket_assignment_delivery_request_no_delete"),
("ticket_assignment_delivery_acknowledgements", "ticket_assignment_delivery_ack_no_delete"),
("ticket_assignment_delivery_origins", "ticket_assignment_delivery_origin_no_delete"),
("ticket_assignment_operation_fences", "ticket_assignment_operation_fence_no_delete"),
("typed_ticket_assignment_tracking_history", "typed_ticket_assignment_tracking_history_no_delete"),
("typed_ticket_assignment_tracking_operations", "typed_ticket_assignment_tracking_operations_no_delete"),
("typed_ticket_case_history", "typed_ticket_case_history_no_delete"),
+4 -2
View File
@@ -45,8 +45,10 @@ worker using the original Tokio handle; do not build a replacement runtime for
cached services and do not submit to and await the Host lane from inside itself.
Control/session guards needed by an accepted mutation belong to that actual job.
Backend Job cleanup is admitted before its durable cleanup claim. Rejection leaves
Pending/Failed ownership recoverable instead of abandoning a Running claim. Ticket
Backend Job cleanup has a third finite lane (two workers, 16 queued jobs), so a
long Run teardown wait cannot starve ordinary Host control/stop requests. It is
admitted before its durable cleanup claim. Rejection leaves Pending/Failed
ownership recoverable instead of abandoning an Executing claim. Ticket
checker dispatch and timeout processing use bounded admission; overload leaves the
saved reservation/deadline diagnosable. They are not automatically retried after an
unknown DB outcome. Runtime teardown is still its own lifecycle fence, not proof
@@ -0,0 +1,60 @@
# Assign-driven Orchestrator delivery and start admission (T-742)
## Separate authorities
[Ticket assignment tracking](ticket-assignment-tracking.md) remains the sole delegation/handling authority (Ticket component schema 11). Workspace schema 93 adds **operational** input origins, delivery reservations, exact rendered requests, acceptance acknowledgements, and assignment/source-bound start fences. These tables do not create another assignment identity or handling status.
Assign commits the current pointer, initial waiting, assignment event and source-bound operation receipt together. Its committed tracking-history insertion records an initial input origin in that same transaction. Delivery runs only after commit. Runtime acceptance, running/idle/stop, MR changes, and delivery acknowledgement never update `waiting`/`inprogress`/`done`.
A delivery acknowledgement means the Runtime accepted the immutable Notify request. It is not an assertion that the model considered the input, satisfied the Ticket, or started execution. Runtime notification/activation receipts provide the corresponding durable session boundary. Self-reported progress/result remains a separate explicit mutation.
## Common input scan and delivery
Mutation notification, Backend startup, Orchestrator create (including already-online), accepted restore, replacement, and running→idle all use `list_pending_ticket_assignment_deliveries`. There is no parallel Queue/open-state routing predicate. Openness is a new-work safety boundary, not delegation intent.
The scan selects current `workspace_agent:workspace-orchestrator` assignments with actionable handling and an explicit committed input origin. It excludes released identities, unknown legacy relationships, done/history and closed cases. External content/answer events after the origin are new inputs; handling self-reports and case/routing history are not. Waiting alone is never a resend reason. Close releases operational work; reopen alone does not revive the old delegation. Explicit reissue creates a new assignment and initial input.
Schema 92→93 preserves original assignment/event/Queue evidence. Native Assign history establishes its input origin; migrated tracking contributes an input only where its already-reconciled assignment-specific Queue fence is present. Ambiguous/historical rows remain diagnostic data, not fresh requests. Earlier migrations and Ticket schema 11 are unchanged.
Delivery order:
1. Resolve the currently registered online Worker and active Runtime binding from Backend authority (a stopped Worker is not automatically restored).
2. Reserve the exact `assignment_id`/`input_event_id` under IMMEDIATE, rechecking current delegation, singleton and binding.
3. Render using the reservation's metadata, including waiting reason and next action. Persist the **entire rendered content**, not just its metadata or a title fingerprint.
4. Send the saved content with the saved `request_id`, outside the transaction. Response loss retains the same request/payload. Concurrent reservations converge on the same identity, and Runtime request receipts deduplicate actual acceptance.
5. Acknowledge accepted delivery under another IMMEDIATE transaction after rechecking current authority. The acknowledgement never changes handling.
Process-local attention locks prevent concurrent/reentrant send loops; they are not backlog authority. A committed-input/startup/Restore trigger arriving during a reserved scan is coalesced into an input recheck and drained on reservation release, including when that scan read an empty snapshot. Discarding or acknowledging a pending retry continues the common scan in the same reservation; a successful retry cannot consume the trigger for newer input. Reentrant idle observations caused by a send do not schedule another retry of a failed send. These liveness repairs change no DB transaction or public API boundary. Losing that cache does not lose a delegation or resend acknowledged input. Replacement does not create assignment rows; only pending, unacknowledged inputs are recovered. Accepted history is not replayed into new work merely because a Worker was exchanged.
## Public start/action handoff for T-743
Existing canonical tracking query/history/update endpoints, Tools/WIP, Ticket query/detail, and retained assignment projections remain available; see the tracking document. Model-visible notices include the committed delegation/input identity and bounded waiting/next-action context. The Server attention template is updated here; final Web/TUI/all-Prompt operation alignment belongs to T-743.
Start request contracts now carry optional `orchestrator_assignment_id`:
- `CreateWorkspaceWorkerRequest.ticket_assignment` (including Worker control spawn).
- `RuntimeWorkerSpawnRequest.ticket_assignment`.
- `SetTicketRoleAssignmentRequest` for an existing Worker.
- `SpawnTicketWorker` and generic `WorkerSpawn` tool input/lifecycle forwarding.
An authenticated current Orchestrator **must** supply its observed delegation identity. Backend checks it with the actual `ServerRequestContext.runtime_source.authentication_binding_id`; it never derives the source from model role text or reacquires another binding as authentication proof. Missing/stale/cleared/reissued identities and obsolete Workers cannot admit work. Manual/generic authorized starts omit the delegation field and do not implicitly assign the Orchestrator. New starts and existing Worker assignments share the same reservation/start fence.
The fenced reservation extends the existing Ticket operation receipt and Worker allocation/recovery identity. Reserve/retry, bind and finalization validate the saved assignment/source under IMMEDIATE. Recovery/pre-dispatch checks use the same fence. Clear/reassign can invalidate a reservation but do not stop, transfer or clean up already-running work. Unknown external outcomes keep their reservation/resource fences for explicit reconciliation rather than permitting a second start. Runtime I/O is never inside a database transaction.
Unfinished dependencies remain query/relation diagnostics and do not disable manual/Orchestrator new starts or existing Worker assignment. They are never implicitly started. Case closure, authenticated permissions, actual unfinished work/claims, resource/grant conflicts, target containment and recovery remain independent guards. Repository/Workdir-free work, read-only work and caller-selected published Profiles are preserved.
Action projection permits delegation independently of existing Worker occupancy; actual Worker start eligibility still excludes unfinished Worker work and closed cases. `can_start_manual_worker` no longer includes an unresolved-dependency predicate. `can_assign_orchestrator` permits an explicit fresh Assign/reissue operation even when a current row exists; callers can supply the observed expected assignment for CAS. `can_unassign_orchestrator` uses the retained current pointer, not active-work or case-state filtering. These authoring actions are not Worker transfer. Final browser controls must use these Backend decisions rather than recreating old phase/dependency gates.
Delivery diagnostics are available through the typed Store scan/reservation/ack records and structured `ticket_assignment_delivery_pending` warning (`assignment_id`, `input_event_id`, `delivery_id`, recipient and error category, never content). Existing `ticket_notification_delivery_failed` remains for recipient failures. No browser acknowledgement/status-edit endpoint is introduced: successful delivery is not a user handling update. T-743 can consume the operational query independently of handling without inventing a third status in tracking.
## DB caller/transaction handoff for T-748 and protocol boundaries
Store callers are in `store.rs`; operational SQL/mutations are in `assignment_delivery.rs`. The public Store methods are `list_pending_ticket_assignment_deliveries`, `reserve_ticket_assignment_delivery`, `prepare_ticket_assignment_delivery_content`, `ack_ticket_assignment_delivery`, `reserve_ticket_assignment_operation_for_orchestrator`, and `validate_ticket_assignment_operation_fence`. Existing bind/finalization writes call the saved-operation guard inside their own IMMEDIATE transaction.
Operational tables: `ticket_assignment_delivery_origins`, `ticket_assignment_deliveries`, `ticket_assignment_delivery_requests`, `ticket_assignment_delivery_acknowledgements`, `ticket_assignment_operation_fences`. Workspace explicit purge inventory includes their scoped rows and temporarily suspends/restores their immutable-delete guards in the existing guarded purge transaction. Ordinary operations cannot delete or rewrite receipts.
T-741 set/clear/update/history remain unchanged in transaction ownership: actual source binding/singleton checks, receipt comparison, current/event CAS, audit and result must stay together. A future DB blocking adapter must move the whole synchronous transaction as one unit, not split its queries around awaits. Server Runtime method admission and `ServerRuntimeMutationSourceContext.authentication_binding_id` are not changed. T-739 close/reopen and T-715 identity-specific work releases remain execution boundaries.
## Validation limits
Local Store tests, authenticated HTTP tests, real embedded Controller/session tests, and loopback remote Runtime HTTP tests prove persistence, concurrency, source/assignment fencing, payload replay, resource contracts and activation dedupe. They do not prove model judgment quality or final Web/TUI/all-Prompt behavior (T-743/T-744). No live DB migration, environment update, target integration, Ticket conclusion or Runtime Worker/Workdir cleanup is part of this handoff.
+22 -1
View File
@@ -1606,6 +1606,13 @@
"operation_id": {
"type": "string"
},
"orchestrator_assignment_id": {
"description": "Observed delegation required for Orchestrator Ticket starts; manual starts omit it.",
"type": [
"string",
"null"
]
},
"ticket_id": {
"type": "string"
}
@@ -8953,6 +8960,13 @@
"operation_id": {
"type": "string"
},
"orchestrator_assignment_id": {
"description": "Observed delegation required for Orchestrator Ticket starts; manual starts omit it.",
"type": [
"string",
"null"
]
},
"ticket_id": {
"type": "string"
}
@@ -9304,6 +9318,13 @@
"operation_id": {
"type": "string"
},
"orchestrator_assignment_id": {
"description": "Observed delegation required for Orchestrator Ticket starts; manual starts omit it.",
"type": [
"string",
"null"
]
},
"principal": {
"$ref": "#/components/schemas/TicketAssignmentPrincipal"
},
@@ -16813,7 +16834,7 @@
"info": {
"title": "Yoi Server API",
"version": "0.1.0",
"x-yoi-source-digest": "sha256:982cc7edc8fd8655a14e943b9d0022df56e941107def02251186d8ddebfd6774"
"x-yoi-source-digest": "sha256:8c8254cbe4ca8675f3dfffdf6e0eeecfda1c9213b285f571704648774efa497b"
},
"openapi": "3.1.0",
"paths": {
@@ -1,8 +1,17 @@
Queued Tickets require attention:
Assigned Tickets have unreceived input:
{% for ticket in tickets -%}
- {{ ticket.resource_key }} {{ separator }} {{ ticket.title }}
{% if ticket.assignment_id -%}
Delegation: {{ ticket.assignment_id }}; input: {{ ticket.input_event_id }}.
{% endif -%}
{% if ticket.waiting_reason -%}
Waiting reason: {{ ticket.waiting_reason }}
{% endif -%}
{% if ticket.next_action -%}
Next action: {{ ticket.next_action }}
{% endif -%}
{% endfor -%}
{% if omitted_ticket_count > 0 -%}
Additional queued Tickets were omitted from this notice: {{ omitted_ticket_count }}. Re-query current Ticket authority for the complete set.
Additional assigned Tickets were omitted from this notice: {{ omitted_ticket_count }}. Re-query current Ticket authority for the complete set.
{% endif -%}
Reread the current Ticket state before acting. Preserve the human queue gate and current assignment, dependency, Worker, and Workdir authority; do not create duplicate work.
Reread the current Ticket state before acting, including current assignment tracking and new input. Assign is delegation; no ready/Queue step is required. Handling status is your self-report, not delivery or Worker state. Waiting alone and done history are not new work. Dependencies are diagnostics for partial work, not a blanket start gate; never implicitly start dependencies. Preserve current assignment identity, authenticated Worker, unfinished work, and Workdir/resource authority; do not create duplicate work.
@@ -1,6 +1,6 @@
// @generated by yoi-companion-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:982cc7edc8fd8655a14e943b9d0022df56e941107def02251186d8ddebfd6774)
// Input contract: 0.1.0 (sha256:8c8254cbe4ca8675f3dfffdf6e0eeecfda1c9213b285f571704648774efa497b)
// Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/companion-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check
@@ -1,6 +1,6 @@
// @generated by yoi-repository-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:982cc7edc8fd8655a14e943b9d0022df56e941107def02251186d8ddebfd6774)
// Input contract: 0.1.0 (sha256:8c8254cbe4ca8675f3dfffdf6e0eeecfda1c9213b285f571704648774efa497b)
// Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output
// Output: web/workspace/src/lib/generated/repository-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check
@@ -1,6 +1,6 @@
// @generated by yoi-runtime-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:982cc7edc8fd8655a14e943b9d0022df56e941107def02251186d8ddebfd6774)
// Input contract: 0.1.0 (sha256:8c8254cbe4ca8675f3dfffdf6e0eeecfda1c9213b285f571704648774efa497b)
// Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/runtime-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check
@@ -151,7 +151,7 @@ export type RuntimeWorkerSpawnWorkingDirectoryRequest = { repository_key: string
export type RuntimeWorkerSummary = { diagnostics?: Array<Diagnostic>; display_name?: string; host_id: string; implementation: WorkerImplementationSummary; label: string; last_seen_at?: string | null; pinned?: boolean; profile?: string | null; retention_state?: string; runtime_id: string; singleton_key?: string | null; state: string; tags?: Array<string>; workdir_attachments?: Array<unknown>; worker_id: string; worker_state?: unknown; workspace: WorkerWorkspaceSummary; };
export type RuntimeWorkerTicketAssignmentRequest = { operation_id: string; ticket_id: string; };
export type RuntimeWorkerTicketAssignmentRequest = { operation_id: string; orchestrator_assignment_id?: string | null; ticket_id: string; };
export type Segment = { content: string; kind: "text"; } | { chars: number; content: string; id: number; kind: "paste"; lines: number; } | { artifact: PasteArtifactRef; kind: "paste_artifact"; } | { file: UploadedFileRef; kind: "uploaded_file"; } | { kind: "file_ref"; path: string; } | { invocation: FeatureInvocation; kind: "feature_invoke"; } | { kind: "flow"; selector: string; } | { kind: "unknown"; };
@@ -1,6 +1,6 @@
// @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:982cc7edc8fd8655a14e943b9d0022df56e941107def02251186d8ddebfd6774)
// Input contract: 0.1.0 (sha256:8c8254cbe4ca8675f3dfffdf6e0eeecfda1c9213b285f571704648774efa497b)
// Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/ticket-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check
@@ -83,7 +83,7 @@ export type ReviewDecision = "approve" | "request_changes";
export type ReviewFinding = { body: string; code?: string | null; line?: number | null; path?: string | null; severity: FindingSeverity; };
export type SetTicketRoleAssignmentRequest = { expected_assignment_id?: string | null; operation_id: string; principal: TicketAssignmentPrincipal; workdir_bindings?: Array<TicketWorkerAttachmentBinding>; };
export type SetTicketRoleAssignmentRequest = { expected_assignment_id?: string | null; operation_id: string; orchestrator_assignment_id?: string | null; principal: TicketAssignmentPrincipal; workdir_bindings?: Array<TicketWorkerAttachmentBinding>; };
export type TicketActionEligibility = { blockers: Array<string>; can_assign_orchestrator: boolean; can_queue: boolean; can_start_manual_worker: boolean; can_unassign_orchestrator: boolean; queue_tickets: Array<string>; };
@@ -1,6 +1,6 @@
// @generated by yoi-worker-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:982cc7edc8fd8655a14e943b9d0022df56e941107def02251186d8ddebfd6774)
// Input contract: 0.1.0 (sha256:8c8254cbe4ca8675f3dfffdf6e0eeecfda1c9213b285f571704648774efa497b)
// Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/worker-launch-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check
@@ -13,7 +13,7 @@ export type BrowserWorkspaceOrchestratorResponse = { diagnostics: Array<Diagnost
export type CreateWorkspaceWorkerRequest = { control_operation_id?: string | null; display_name: string; feature_connections: WorkspaceWorkerFeatureConnectionsRequest; initial_submit?: Array<Segment>; profile?: string | null; runtime_id: string; singleton_key?: string | null; ticket_assignment?: CreateWorkspaceWorkerTicketAssignmentRequest | null; workdir_attachments?: Array<BrowserWorkerWorkingDirectorySelection>; };
export type CreateWorkspaceWorkerTicketAssignmentRequest = { operation_id: string; ticket_id: string; };
export type CreateWorkspaceWorkerTicketAssignmentRequest = { operation_id: string; orchestrator_assignment_id?: string | null; ticket_id: string; };
export type Diagnostic = { code: string; message: string; severity: DiagnosticSeverity; };