feat(ticket): persist Orchestrator assignment tracking and audit
This commit is contained in:
@@ -0,0 +1,105 @@
|
||||
-- Ticket owns assignment identities, current pointers, audit and operation receipts.
|
||||
CREATE TABLE IF NOT EXISTS ticket_assignment_operations (
|
||||
workspace_id TEXT NOT NULL,
|
||||
operation_id TEXT NOT NULL,
|
||||
action TEXT NOT NULL CHECK (action IN ('assign', 'reassign', 'unassign')),
|
||||
ticket_id TEXT NOT NULL,
|
||||
role TEXT NOT NULL DEFAULT 'worker'
|
||||
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
|
||||
principal_kind TEXT NOT NULL DEFAULT 'worker'
|
||||
CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
|
||||
principal_id TEXT,
|
||||
runtime_id TEXT,
|
||||
worker_id TEXT,
|
||||
assignment_id TEXT,
|
||||
expected_assignment_id TEXT,
|
||||
created_at TEXT NOT NULL,
|
||||
request_fingerprint TEXT,
|
||||
claim_state TEXT NOT NULL DEFAULT 'committed'
|
||||
CHECK(claim_state IN ('pending', 'failed', 'committed')),
|
||||
failure_reason TEXT,
|
||||
binding_recovery_json TEXT,
|
||||
PRIMARY KEY (workspace_id, operation_id),
|
||||
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
CHECK(claim_state = 'committed' OR assignment_id IS NULL),
|
||||
CHECK((claim_state = 'failed') = (failure_reason IS NOT NULL))
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS ticket_current_worker_assignments (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
role TEXT NOT NULL DEFAULT 'worker'
|
||||
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
|
||||
assignment_id TEXT NOT NULL,
|
||||
principal_kind TEXT NOT NULL DEFAULT 'worker'
|
||||
CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
|
||||
principal_id TEXT,
|
||||
runtime_id TEXT,
|
||||
worker_id TEXT,
|
||||
updated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, role, assignment_id),
|
||||
UNIQUE (workspace_id, assignment_id),
|
||||
FOREIGN KEY (workspace_id, ticket_id, role, assignment_id)
|
||||
REFERENCES ticket_worker_assignments(workspace_id, ticket_id, role, assignment_id)
|
||||
ON DELETE CASCADE,
|
||||
FOREIGN KEY (workspace_id, ticket_id)
|
||||
REFERENCES typed_tickets(workspace_id, ticket_id)
|
||||
ON DELETE CASCADE,
|
||||
CHECK(
|
||||
(principal_kind = 'worker' AND runtime_id IS NOT NULL AND worker_id IS NOT NULL AND principal_id IS NULL)
|
||||
OR
|
||||
(principal_kind != 'worker' AND runtime_id IS NULL AND worker_id IS NULL AND principal_id IS NOT NULL AND length(trim(principal_id)) > 0)
|
||||
)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS ticket_worker_assignment_events (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
role TEXT NOT NULL DEFAULT 'worker'
|
||||
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
|
||||
event_id TEXT NOT NULL,
|
||||
action TEXT NOT NULL CHECK (action IN ('assigned', 'reassigned', 'unassigned')),
|
||||
assignment_id TEXT,
|
||||
previous_assignment_id TEXT,
|
||||
actor TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
operation_id TEXT,
|
||||
reason TEXT,
|
||||
PRIMARY KEY (workspace_id, event_id),
|
||||
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS ticket_worker_assignments (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
assignment_id TEXT NOT NULL,
|
||||
role TEXT NOT NULL DEFAULT 'worker'
|
||||
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
|
||||
principal_kind TEXT NOT NULL DEFAULT 'worker'
|
||||
CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
|
||||
principal_id TEXT,
|
||||
runtime_id TEXT,
|
||||
worker_id TEXT,
|
||||
assigned_by TEXT NOT NULL,
|
||||
assigned_at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, assignment_id),
|
||||
UNIQUE (workspace_id, ticket_id, assignment_id),
|
||||
UNIQUE (workspace_id, ticket_id, role, assignment_id),
|
||||
UNIQUE (workspace_id, ticket_id, role, assignment_id, principal_kind, principal_id, runtime_id, worker_id),
|
||||
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
CHECK(
|
||||
(principal_kind = 'worker' AND runtime_id IS NOT NULL AND worker_id IS NOT NULL AND principal_id IS NULL)
|
||||
OR
|
||||
(principal_kind != 'worker' AND runtime_id IS NULL AND worker_id IS NULL AND principal_id IS NOT NULL AND length(trim(principal_id)) > 0)
|
||||
)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_ticket_assignment_operations_ticket
|
||||
ON ticket_assignment_operations(workspace_id, ticket_id, role, created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_ticket_worker_assignment_events_ticket
|
||||
ON ticket_worker_assignment_events(workspace_id, ticket_id, role, created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_ticket_worker_assignments_principal
|
||||
ON ticket_worker_assignments(workspace_id, role, principal_kind, principal_id, runtime_id, worker_id, assigned_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_ticket_worker_assignments_ticket
|
||||
ON ticket_worker_assignments(workspace_id, ticket_id, role, assigned_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS ticket_current_principal_idx
|
||||
ON ticket_current_worker_assignments(workspace_id, principal_kind, principal_id, runtime_id, worker_id);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS ticket_current_singleton_role_idx
|
||||
ON ticket_current_worker_assignments(workspace_id, ticket_id, role)
|
||||
WHERE role IN ('orchestrator', 'worker');
|
||||
@@ -0,0 +1,900 @@
|
||||
//! Ticket-owned assignment tracking. Runtime execution, grants and notification
|
||||
//! delivery are intentionally not tracking status. The host resolves the current
|
||||
//! workspace-agent Worker and authenticates it before constructing source context.
|
||||
use super::*;
|
||||
|
||||
/// Used by the Workspace host to compose its stricter principal constraints.
|
||||
/// Historical physical names remain to preserve immutable IDs and receipts.
|
||||
pub const TICKET_ASSIGNMENT_SCHEMA_SQL: &str = include_str!("assignment_schema.sql");
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum TicketAssignmentTrackingStatus {
|
||||
Waiting,
|
||||
Inprogress,
|
||||
Done,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct TicketAssignmentTracking {
|
||||
pub status: TicketAssignmentTrackingStatus,
|
||||
pub waiting_reason: Option<String>,
|
||||
pub progress: Option<String>,
|
||||
pub next_action: Option<String>,
|
||||
pub result: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct TicketAssignmentTrackingUpdate {
|
||||
pub assignment_id: String,
|
||||
pub expected_last_event_id: String,
|
||||
pub operation_key: String,
|
||||
pub tracking: TicketAssignmentTracking,
|
||||
}
|
||||
|
||||
/// Backend-owned authenticated context, never decoded from model input. Agent
|
||||
/// principal remains stable across Worker exchange; exact replay binds to the
|
||||
/// authenticated concrete sender and its binding, not merely a claimed role.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct TicketAssignmentTrackingSourceContext {
|
||||
pub actor: String,
|
||||
pub runtime_id: String,
|
||||
pub worker_id: String,
|
||||
pub agent_key: String,
|
||||
pub authentication_binding_id: String,
|
||||
}
|
||||
|
||||
/// Enrichment is separate from the established assignment record. `None`
|
||||
/// tracking means a retained historical relationship with insufficient delegation
|
||||
/// evidence, not waiting/done. Removal/replacement does not alter last tracking.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
pub struct TicketAssignmentTrackingProjection {
|
||||
pub workspace_id: String,
|
||||
pub ticket_id: String,
|
||||
pub assignment_id: String,
|
||||
pub role: String,
|
||||
pub principal: TicketAssignmentPrincipalSnapshot,
|
||||
pub assigned_by: String,
|
||||
pub assigned_at: String,
|
||||
pub release: Option<TicketAssignmentRelease>,
|
||||
pub current: bool,
|
||||
pub tracking: Option<TicketAssignmentTracking>,
|
||||
/// Tracking event, or the existing immutable Assign audit observation for a
|
||||
/// retained current relationship with no tracking yet. This baseline does
|
||||
/// not imply a handling status, delegation intent, or tracking timestamp.
|
||||
pub last_event_id: Option<String>,
|
||||
pub updated_at: Option<String>,
|
||||
pub migration_diagnostic: Option<String>,
|
||||
}
|
||||
|
||||
/// Immutable principal as stored at Assign, not live Worker availability.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
pub enum TicketAssignmentPrincipalSnapshot {
|
||||
User {
|
||||
account_id: String,
|
||||
},
|
||||
Worker {
|
||||
runtime_id: String,
|
||||
worker_id: String,
|
||||
},
|
||||
WorkspaceAgent {
|
||||
agent_key: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum TicketAssignmentReleaseAction {
|
||||
Unassigned,
|
||||
Replaced,
|
||||
}
|
||||
|
||||
/// Release is an independent immutable assignment audit event. Last tracking
|
||||
/// remains unchanged, including unknown legacy tracking; it is never made done.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
pub struct TicketAssignmentRelease {
|
||||
pub action: TicketAssignmentReleaseAction,
|
||||
pub event_id: String,
|
||||
pub operation_key: Option<String>,
|
||||
pub actor: String,
|
||||
pub at: String,
|
||||
pub reason: Option<String>,
|
||||
pub replacement_assignment_id: Option<String>,
|
||||
pub last_tracking: Option<TicketAssignmentTracking>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
pub struct TicketAssignmentTrackingHistoryEntry {
|
||||
pub event_id: String,
|
||||
pub operation_key: String,
|
||||
pub tracking: TicketAssignmentTracking,
|
||||
pub source_context: serde_json::Value,
|
||||
pub at: String,
|
||||
}
|
||||
|
||||
/// Bounded append-order page. Cursor is an immutable event identity, not a
|
||||
/// lexical event-ID comparison or a transferable offset into another assignment.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
pub struct TicketAssignmentTrackingHistoryPage {
|
||||
pub items: Vec<TicketAssignmentTrackingHistoryEntry>,
|
||||
pub next_event_id: Option<String>,
|
||||
}
|
||||
|
||||
fn valid_identity(value: &str) -> bool {
|
||||
!value.trim().is_empty() && value.len() <= 1024 && !value.chars().any(char::is_control)
|
||||
}
|
||||
|
||||
fn validate_update(
|
||||
update: &TicketAssignmentTrackingUpdate,
|
||||
source: &TicketAssignmentTrackingSourceContext,
|
||||
) -> Result<()> {
|
||||
if [
|
||||
&update.assignment_id,
|
||||
&update.expected_last_event_id,
|
||||
&update.operation_key,
|
||||
&source.actor,
|
||||
&source.runtime_id,
|
||||
&source.worker_id,
|
||||
&source.authentication_binding_id,
|
||||
]
|
||||
.into_iter()
|
||||
.any(|value| !valid_identity(value))
|
||||
|| source.agent_key != "workspace-orchestrator"
|
||||
{
|
||||
return Err(TicketError::Conflict(
|
||||
"tracking requires bounded identities and authenticated workspace-orchestrator source"
|
||||
.into(),
|
||||
));
|
||||
}
|
||||
for text in [
|
||||
&update.tracking.waiting_reason,
|
||||
&update.tracking.progress,
|
||||
&update.tracking.next_action,
|
||||
&update.tracking.result,
|
||||
]
|
||||
.into_iter()
|
||||
.flatten()
|
||||
{
|
||||
if text.len() > 16 * 1024 || text.contains('\0') {
|
||||
return Err(TicketError::Conflict(
|
||||
"tracking fields must be at most 16384 bytes without NUL".into(),
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Reads responsibility independently of case state and operational work release.
|
||||
pub fn sqlite_ticket_assignment_tracking(
|
||||
conn: &Connection,
|
||||
workspace_id: &str,
|
||||
ticket_id: &str,
|
||||
assignment_id: &str,
|
||||
) -> Result<Option<TicketAssignmentTrackingProjection>> {
|
||||
let row = conn.query_row("SELECT a.tracking_json, a.last_tracking_event_id, a.tracking_updated_at, a.tracking_migration_diagnostic,
|
||||
EXISTS(SELECT 1 FROM ticket_current_worker_assignments c WHERE c.workspace_id=a.workspace_id AND c.assignment_id=a.assignment_id),
|
||||
a.role, a.principal_kind, a.principal_id, a.runtime_id, a.worker_id, a.assigned_by, a.assigned_at
|
||||
FROM ticket_worker_assignments a WHERE a.workspace_id=?1 AND a.ticket_id=?2 AND a.assignment_id=?3 AND a.role='orchestrator'",
|
||||
params![workspace_id,ticket_id,assignment_id], |r| Ok((r.get::<_,Option<String>>(0)?,r.get::<_,Option<String>>(1)?,r.get::<_,Option<String>>(2)?,r.get::<_,Option<String>>(3)?,r.get::<_,bool>(4)?, r.get::<_,String>(5)?, r.get::<_,String>(6)?,r.get::<_,Option<String>>(7)?,r.get::<_,Option<String>>(8)?,r.get::<_,Option<String>>(9)?,r.get::<_,String>(10)?,r.get::<_,String>(11)?)))
|
||||
.optional().map_err(sqlite_err)?;
|
||||
let Some((
|
||||
json,
|
||||
last_event_id,
|
||||
updated_at,
|
||||
migration_diagnostic,
|
||||
current,
|
||||
role,
|
||||
kind,
|
||||
principal_id,
|
||||
runtime_id,
|
||||
worker_id,
|
||||
assigned_by,
|
||||
assigned_at,
|
||||
)) = row
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
let tracking: Option<TicketAssignmentTracking> = json
|
||||
.map(|value| serde_json::from_str(&value).map_err(sqlite_err))
|
||||
.transpose()?;
|
||||
// Legacy migration may retain current responsibility without evidence of
|
||||
// handling status. Its real Assign audit is still an immutable observation
|
||||
// against which the first explicit self-report can CAS, even after close.
|
||||
// Never manufacture a tracking event/status or repair a missing audit. Once
|
||||
// tracking exists, only its event identity is a valid observation.
|
||||
let last_event_id = if current && tracking.is_none() && last_event_id.is_none() {
|
||||
conn.query_row(
|
||||
"SELECT e.event_id FROM ticket_worker_assignment_events e
|
||||
WHERE e.workspace_id=?1 AND e.ticket_id=?2 AND e.assignment_id=?3
|
||||
AND e.role='orchestrator' AND e.action IN ('assigned','reassigned')
|
||||
AND NOT EXISTS (SELECT 1 FROM typed_ticket_assignment_tracking_history h
|
||||
WHERE h.workspace_id=?1 AND h.ticket_id=?2 AND h.assignment_id=?3)
|
||||
ORDER BY e.rowid LIMIT 1",
|
||||
params![workspace_id, ticket_id, assignment_id],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()
|
||||
.map_err(sqlite_err)?
|
||||
} else {
|
||||
last_event_id
|
||||
};
|
||||
let missing =
|
||||
|| TicketError::Conflict("stored assignment principal identity is incomplete".into());
|
||||
let principal = match kind.as_str() {
|
||||
"workspace_agent" => TicketAssignmentPrincipalSnapshot::WorkspaceAgent {
|
||||
agent_key: principal_id.ok_or_else(missing)?,
|
||||
},
|
||||
"user" => TicketAssignmentPrincipalSnapshot::User {
|
||||
account_id: principal_id.ok_or_else(missing)?,
|
||||
},
|
||||
"worker" => TicketAssignmentPrincipalSnapshot::Worker {
|
||||
runtime_id: runtime_id.ok_or_else(missing)?,
|
||||
worker_id: worker_id.ok_or_else(missing)?,
|
||||
},
|
||||
_ => return Err(missing()),
|
||||
};
|
||||
let release = conn.query_row("SELECT action,event_id,operation_id,actor,created_at,reason,assignment_id FROM ticket_worker_assignment_events
|
||||
WHERE workspace_id=?1 AND ticket_id=?2 AND ((action='unassigned' AND (assignment_id=?3 OR previous_assignment_id=?3)) OR (action='reassigned' AND previous_assignment_id=?3)) ORDER BY rowid DESC LIMIT 1",
|
||||
params![workspace_id,ticket_id,assignment_id], |r| Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,Option<String>>(2)?,r.get::<_,String>(3)?,r.get::<_,String>(4)?,r.get::<_,Option<String>>(5)?,r.get::<_,Option<String>>(6)?))).optional().map_err(sqlite_err)?
|
||||
.map(|(action,event_id,operation_key,actor,at,reason,replacement)| TicketAssignmentRelease {
|
||||
action: if action=="unassigned" {TicketAssignmentReleaseAction::Unassigned} else {TicketAssignmentReleaseAction::Replaced},
|
||||
event_id,operation_key,actor,at,reason,replacement_assignment_id: if action=="reassigned" {replacement} else {None}, last_tracking: tracking.clone(),
|
||||
});
|
||||
Ok(Some(TicketAssignmentTrackingProjection {
|
||||
workspace_id: workspace_id.into(),
|
||||
ticket_id: ticket_id.into(),
|
||||
assignment_id: assignment_id.into(),
|
||||
role,
|
||||
principal,
|
||||
assigned_by,
|
||||
assigned_at,
|
||||
release,
|
||||
current,
|
||||
tracking,
|
||||
last_event_id,
|
||||
updated_at,
|
||||
migration_diagnostic,
|
||||
}))
|
||||
}
|
||||
|
||||
pub fn sqlite_current_ticket_assignment_tracking(
|
||||
conn: &Connection,
|
||||
workspace_id: &str,
|
||||
ticket_id: &str,
|
||||
) -> Result<Vec<TicketAssignmentTrackingProjection>> {
|
||||
let ids = conn.prepare("SELECT assignment_id FROM ticket_current_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND role='orchestrator' ORDER BY assignment_id")
|
||||
.map_err(sqlite_err)?.query_map(params![workspace_id,ticket_id],|r|r.get::<_,String>(0)).map_err(sqlite_err)?
|
||||
.collect::<std::result::Result<Vec<_>,_>>().map_err(sqlite_err)?;
|
||||
ids.into_iter()
|
||||
.map(|id| {
|
||||
sqlite_ticket_assignment_tracking(conn, workspace_id, ticket_id, &id)?
|
||||
.ok_or_else(|| TicketError::Conflict("current assignment row is missing".into()))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn sqlite_ticket_assignment_tracking_history(
|
||||
conn: &Connection,
|
||||
workspace_id: &str,
|
||||
ticket_id: &str,
|
||||
assignment_id: &str,
|
||||
) -> Result<Vec<TicketAssignmentTrackingHistoryEntry>> {
|
||||
let mut stmt=conn.prepare("SELECT h.event_id,h.operation_key,h.tracking_json,h.source_context_json,h.at FROM typed_ticket_assignment_tracking_history h
|
||||
WHERE h.workspace_id=?1 AND h.ticket_id=?2 AND h.assignment_id=?3 ORDER BY h.rowid").map_err(sqlite_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![workspace_id, ticket_id, assignment_id], |r| {
|
||||
Ok((
|
||||
r.get::<_, String>(0)?,
|
||||
r.get::<_, String>(1)?,
|
||||
r.get::<_, String>(2)?,
|
||||
r.get::<_, String>(3)?,
|
||||
r.get::<_, String>(4)?,
|
||||
))
|
||||
})
|
||||
.map_err(sqlite_err)?;
|
||||
rows.map(|r| {
|
||||
let (event_id, operation_key, json, source, at) = r.map_err(sqlite_err)?;
|
||||
Ok(TicketAssignmentTrackingHistoryEntry {
|
||||
event_id,
|
||||
operation_key,
|
||||
tracking: serde_json::from_str(&json).map_err(sqlite_err)?,
|
||||
source_context: serde_json::from_str(&source).map_err(sqlite_err)?,
|
||||
at,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// DB-bounded history pagination. Immutable append-only rows provide a stable
|
||||
/// cursor position; the scope index also orders its implicit rowids. No caller
|
||||
/// needs to load, deserialize or truncate the lifetime assignment history.
|
||||
pub fn sqlite_ticket_assignment_tracking_history_page(
|
||||
conn: &Connection,
|
||||
workspace_id: &str,
|
||||
ticket_id: &str,
|
||||
assignment_id: &str,
|
||||
after_event_id: Option<&str>,
|
||||
limit: usize,
|
||||
) -> Result<TicketAssignmentTrackingHistoryPage> {
|
||||
if !(1..=100).contains(&limit) {
|
||||
return Err(TicketError::Conflict(
|
||||
"tracking history limit must be 1..100".into(),
|
||||
));
|
||||
}
|
||||
let exists:bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM ticket_worker_assignments WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3 AND role='orchestrator')",params![workspace_id,ticket_id,assignment_id],|r|r.get(0)).map_err(sqlite_err)?;
|
||||
if !exists {
|
||||
return Err(TicketError::Conflict(
|
||||
"tracking assignment is not in this Workspace/Ticket".into(),
|
||||
));
|
||||
}
|
||||
let after_rowid:Option<i64>=after_event_id.map(|event| conn.query_row("SELECT rowid FROM typed_ticket_assignment_tracking_history WHERE workspace_id=?1 AND ticket_id=?2 AND assignment_id=?3 AND event_id=?4",params![workspace_id,ticket_id,assignment_id,event],|r|r.get::<_,i64>(0)).optional().map_err(sqlite_err)?.ok_or_else(||TicketError::Conflict("tracking history cursor is not in this exact Workspace/Ticket/assignment".into()))).transpose()?;
|
||||
let after_clause = if after_rowid.is_some() {
|
||||
"AND h.rowid>?4"
|
||||
} else {
|
||||
"AND ?4 IS NULL"
|
||||
};
|
||||
let sql = format!(
|
||||
"SELECT h.event_id,h.operation_key,h.tracking_json,h.source_context_json,h.at FROM typed_ticket_assignment_tracking_history h WHERE h.workspace_id=?1 AND h.ticket_id=?2 AND h.assignment_id=?3 {after_clause} ORDER BY h.rowid LIMIT ?5"
|
||||
);
|
||||
let mut stmt = conn.prepare(&sql).map_err(sqlite_err)?;
|
||||
let rows = stmt
|
||||
.query_map(
|
||||
params![
|
||||
workspace_id,
|
||||
ticket_id,
|
||||
assignment_id,
|
||||
after_rowid,
|
||||
(limit + 1) as i64
|
||||
],
|
||||
|r| {
|
||||
Ok((
|
||||
r.get::<_, String>(0)?,
|
||||
r.get::<_, String>(1)?,
|
||||
r.get::<_, String>(2)?,
|
||||
r.get::<_, String>(3)?,
|
||||
r.get::<_, String>(4)?,
|
||||
))
|
||||
},
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
let mut items = rows
|
||||
.map(|r| {
|
||||
let (event_id, operation_key, json, source, at) = r.map_err(sqlite_err)?;
|
||||
Ok(TicketAssignmentTrackingHistoryEntry {
|
||||
event_id,
|
||||
operation_key,
|
||||
tracking: serde_json::from_str(&json).map_err(sqlite_err)?,
|
||||
source_context: serde_json::from_str(&source).map_err(sqlite_err)?,
|
||||
at,
|
||||
})
|
||||
})
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
let has_more = items.len() > limit;
|
||||
items.truncate(limit);
|
||||
let next_event_id = if has_more {
|
||||
items.last().map(|entry| entry.event_id.clone())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Ok(TicketAssignmentTrackingHistoryPage {
|
||||
items,
|
||||
next_event_id,
|
||||
})
|
||||
}
|
||||
|
||||
/// Host owns BEGIN IMMEDIATE/commit. Receipt lookup precedes current/CAS checks;
|
||||
/// exact replay returns the immutable operation result even after reassignment.
|
||||
/// Case closed and work-release records are deliberately absent from this guard.
|
||||
pub fn sqlite_update_ticket_assignment_tracking_in_transaction(
|
||||
conn: &Connection,
|
||||
workspace_id: &str,
|
||||
ticket_id: &str,
|
||||
update: &TicketAssignmentTrackingUpdate,
|
||||
source: &TicketAssignmentTrackingSourceContext,
|
||||
) -> Result<TicketAssignmentTrackingProjection> {
|
||||
if conn.is_autocommit() {
|
||||
return Err(TicketError::Conflict(
|
||||
"tracking update requires a host write transaction".into(),
|
||||
));
|
||||
}
|
||||
validate_update(update, source)?;
|
||||
let input_json = serde_json::to_string(&(ticket_id, update)).map_err(sqlite_err)?;
|
||||
let source_json = serde_json::to_string(source).map_err(sqlite_err)?;
|
||||
let receipt=conn.query_row("SELECT input_json,source_context_json,result_json FROM typed_ticket_assignment_tracking_operations WHERE workspace_id=?1 AND operation_key=?2",params![workspace_id,update.operation_key],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?))).optional().map_err(sqlite_err)?;
|
||||
if let Some((input, context, result)) = receipt {
|
||||
if input != input_json || context != source_json {
|
||||
return Err(TicketError::Conflict(
|
||||
"tracking operation key already used for different input or authenticated source"
|
||||
.into(),
|
||||
));
|
||||
}
|
||||
return serde_json::from_str(&result).map_err(sqlite_err);
|
||||
}
|
||||
let current =
|
||||
sqlite_ticket_assignment_tracking(conn, workspace_id, ticket_id, &update.assignment_id)?
|
||||
.ok_or_else(|| {
|
||||
TicketError::Conflict(
|
||||
"tracking assignment not found in this Workspace/Ticket".into(),
|
||||
)
|
||||
})?;
|
||||
if !current.current
|
||||
|| current.last_event_id.as_deref() != Some(update.expected_last_event_id.as_str())
|
||||
{
|
||||
return Err(TicketError::Conflict(
|
||||
"tracking assignment is removed/replaced or last event changed".into(),
|
||||
));
|
||||
}
|
||||
let principal_matches:bool=conn.query_row("SELECT principal_kind='workspace_agent' AND principal_id=?3 FROM ticket_worker_assignments WHERE workspace_id=?1 AND assignment_id=?2",params![workspace_id,update.assignment_id,source.agent_key],|r|r.get(0)).map_err(sqlite_err)?;
|
||||
if !principal_matches {
|
||||
return Err(TicketError::Conflict(
|
||||
"authenticated agent does not own this assignment".into(),
|
||||
));
|
||||
}
|
||||
let event_id=format!("tracking-{}",allocate_record_id(unix_epoch_millis_now().unwrap_or(0), |candidate| conn.query_row("SELECT EXISTS(SELECT 1 FROM typed_ticket_assignment_tracking_history WHERE workspace_id=?1 AND event_id=?2)",params![workspace_id,format!("tracking-{candidate}")],|r|r.get::<_,bool>(0)).unwrap_or(true)).map_err(sqlite_err)?);
|
||||
let at = Utc::now().to_rfc3339();
|
||||
let tracking_json = serde_json::to_string(&update.tracking).map_err(sqlite_err)?;
|
||||
conn.execute("UPDATE ticket_worker_assignments SET tracking_json=?3,last_tracking_event_id=?4,tracking_updated_at=?5 WHERE workspace_id=?1 AND assignment_id=?2",params![workspace_id,update.assignment_id,tracking_json,event_id,at]).map_err(sqlite_err)?;
|
||||
conn.execute(
|
||||
"INSERT INTO typed_ticket_assignment_tracking_history VALUES (?1,?2,?3,?4,?5,?6,?7,?8)",
|
||||
params![
|
||||
workspace_id,
|
||||
ticket_id,
|
||||
update.assignment_id,
|
||||
event_id,
|
||||
update.operation_key,
|
||||
tracking_json,
|
||||
source_json,
|
||||
at
|
||||
],
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
let result =
|
||||
sqlite_ticket_assignment_tracking(conn, workspace_id, ticket_id, &update.assignment_id)?
|
||||
.ok_or_else(|| TicketError::Conflict("tracking assignment disappeared".into()))?;
|
||||
conn.execute(
|
||||
"INSERT INTO typed_ticket_assignment_tracking_operations VALUES (?1,?2,?3,?4,?5,?6)",
|
||||
params![
|
||||
workspace_id,
|
||||
update.operation_key,
|
||||
ticket_id,
|
||||
input_json,
|
||||
source_json,
|
||||
serde_json::to_string(&result).map_err(sqlite_err)?
|
||||
],
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
pub(crate) fn install_assignment_core(conn: &Connection) -> Result<()> {
|
||||
// Standalone Ticket stores do not own Workspace records. The Workspace host
|
||||
// composes the unmodified SQL and adds its principal/resource constraints.
|
||||
let sql = if conn
|
||||
.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name='workspaces')",
|
||||
[],
|
||||
|r| r.get::<_, bool>(0),
|
||||
)
|
||||
.map_err(sqlite_err)?
|
||||
{
|
||||
TICKET_ASSIGNMENT_SCHEMA_SQL.to_owned()
|
||||
} else {
|
||||
TICKET_ASSIGNMENT_SCHEMA_SQL.replace("FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,", "").replace(",\n FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE\n", "\n")
|
||||
};
|
||||
conn.execute_batch(&sql).map_err(sqlite_err)
|
||||
}
|
||||
|
||||
pub(crate) fn migrate_assignment_tracking(conn: &Connection) -> Result<()> {
|
||||
install_assignment_core(conn)?;
|
||||
conn.execute_batch(include_str!("assignment_tracking_schema.sql"))
|
||||
.map_err(sqlite_err)?;
|
||||
reconcile_legacy_tracking(conn)
|
||||
}
|
||||
|
||||
fn reconcile_legacy_tracking(conn: &Connection) -> Result<()> {
|
||||
let rows=conn.prepare("SELECT a.workspace_id,a.ticket_id,a.assignment_id,s.snapshot_json,
|
||||
EXISTS(SELECT 1 FROM ticket_current_worker_assignments c WHERE c.workspace_id=a.workspace_id AND c.assignment_id=a.assignment_id)
|
||||
FROM ticket_worker_assignments a LEFT JOIN typed_ticket_legacy_workflow_snapshot s ON s.workspace_id=a.workspace_id AND s.ticket_id=a.ticket_id WHERE a.role='orchestrator'").map_err(sqlite_err)?
|
||||
.query_map([],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?,r.get::<_,Option<String>>(3)?,r.get::<_,bool>(4)?))).map_err(sqlite_err)?
|
||||
.collect::<std::result::Result<Vec<_>,_>>().map_err(sqlite_err)?;
|
||||
for (workspace, ticket, assignment, snapshot, current) in rows {
|
||||
let original = snapshot
|
||||
.as_deref()
|
||||
.map(serde_json::from_str::<TicketLegacyWorkflowSnapshot>)
|
||||
.transpose()
|
||||
.map_err(sqlite_err)?;
|
||||
// Read the latest routing intention, not just the latest matching
|
||||
// fence: a newer mismatched Queue must not revive an older delegation.
|
||||
let queue=conn.query_row("SELECT e.event_index,e.at,(e.state_field='routing' AND e.reason='queued'),f.value FROM typed_ticket_events e LEFT 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 AND f.key='orchestrator_assignment_id'
|
||||
WHERE e.workspace_id=?1 AND e.ticket_id=?2 AND (e.to_state='queued' OR (e.state_field='routing' AND e.reason='queued')) ORDER BY e.event_index DESC LIMIT 1",params![workspace,ticket],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,Option<String>>(1)?,r.get::<_,Option<bool>>(2)?.unwrap_or(false),r.get::<_,Option<String>>(3)?))).optional().map_err(sqlite_err)?;
|
||||
let original_active = original
|
||||
.as_ref()
|
||||
.is_some_and(|s| matches!(s.workflow_state.as_str(), "queued" | "inprogress"));
|
||||
if let Some((index, at, typed_queue, _)) = queue.filter(|(_, _, typed_queue, fence)| {
|
||||
current
|
||||
&& (original_active || *typed_queue)
|
||||
&& fence.as_deref() == Some(assignment.as_str())
|
||||
}) {
|
||||
// A case judgment is earlier only when its immutable operation key
|
||||
// identifies a case event before Queue. Missing correspondence is
|
||||
// ambiguous; neither timestamp order nor Worker state repairs it.
|
||||
let later_case:bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM typed_ticket_case_history h WHERE h.workspace_id=?1 AND h.ticket_id=?2 AND h.request_fingerprint!='migration-v10'
|
||||
AND NOT EXISTS(SELECT 1 FROM typed_ticket_event_attributes operation JOIN typed_ticket_events e ON e.workspace_id=operation.workspace_id AND e.ticket_id=operation.ticket_id AND e.event_index=operation.event_index
|
||||
WHERE operation.workspace_id=h.workspace_id AND operation.ticket_id=h.ticket_id AND operation.key='operation_key' AND operation.value=h.operation_key AND e.state_field='case' AND e.event_index<?3))",params![workspace,ticket,index],|r|r.get(0)).map_err(sqlite_err)?;
|
||||
let superseded:bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2 AND event_index>?3 AND (state_field='case' OR to_state IN ('planning','ready','done','closed') OR (to_state='open' AND COALESCE(state_field,'')!='routing')))",params![workspace,ticket,index],|r|r.get(0)).map_err(sqlite_err)?;
|
||||
if !later_case && !superseded {
|
||||
let status = if !typed_queue
|
||||
&& original
|
||||
.as_ref()
|
||||
.is_some_and(|s| s.workflow_state == "inprogress")
|
||||
{
|
||||
TicketAssignmentTrackingStatus::Inprogress
|
||||
} else {
|
||||
TicketAssignmentTrackingStatus::Waiting
|
||||
};
|
||||
let tracking = TicketAssignmentTracking {
|
||||
status,
|
||||
waiting_reason: Some(
|
||||
"Migrated fenced legacy delegation; Orchestrator confirmation pending"
|
||||
.into(),
|
||||
),
|
||||
progress: None,
|
||||
next_action: None,
|
||||
result: None,
|
||||
};
|
||||
let event_id = format!("migration-tracking:{assignment}");
|
||||
let at = at.unwrap_or_else(|| Utc::now().to_rfc3339());
|
||||
let json = serde_json::to_string(&tracking).map_err(sqlite_err)?;
|
||||
conn.execute("UPDATE ticket_worker_assignments SET tracking_json=?3,last_tracking_event_id=?4,tracking_updated_at=?5,tracking_migration_diagnostic=?6 WHERE workspace_id=?1 AND assignment_id=?2",params![workspace,assignment,json,event_id,at,format!("Retained original workflow snapshot and latest Queue event {index} (typed routing={typed_queue}) fenced to this current assignment")]).map_err(sqlite_err)?;
|
||||
conn.execute("INSERT INTO typed_ticket_assignment_tracking_history VALUES (?1,?2,?3,?4,?4,?5,'{\"kind\":\"migration\"}',?6)",params![workspace,ticket,assignment,event_id,json,at]).map_err(sqlite_err)?;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
conn.execute("UPDATE ticket_worker_assignments SET tracking_migration_diagnostic='Historical assignment retained without tracking: absent/ambiguous or superseded original workflow and exact Queue fence; no delegation or done inferred' WHERE workspace_id=?1 AND assignment_id=?2",params![workspace,assignment]).map_err(sqlite_err)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn verify_assignment_tracking_schema(conn: &Connection) -> Result<()> {
|
||||
// Object names alone would accept same-named no-op triggers after drift.
|
||||
let canonical = Connection::open_in_memory().map_err(sqlite_err)?;
|
||||
canonical
|
||||
.execute_batch(TICKET_ASSIGNMENT_SCHEMA_SQL)
|
||||
.map_err(sqlite_err)?;
|
||||
canonical
|
||||
.execute_batch(include_str!("assignment_tracking_schema.sql"))
|
||||
.map_err(sqlite_err)?;
|
||||
let normalize = |sql: String| {
|
||||
// SQL keyword/format differences are harmless; literal case and spaces
|
||||
// are not (for example changing the principal or role in a guard).
|
||||
let mut result = String::new();
|
||||
let mut quoted = false;
|
||||
let mut chars = sql.chars().peekable();
|
||||
while let Some(ch) = chars.next() {
|
||||
if ch == '\'' {
|
||||
result.push(ch);
|
||||
if quoted && chars.peek() == Some(&'\'') {
|
||||
result.push(chars.next().unwrap());
|
||||
} else {
|
||||
quoted = !quoted;
|
||||
}
|
||||
} else if quoted {
|
||||
result.push(ch);
|
||||
} else if !ch.is_whitespace() {
|
||||
result.push(ch.to_ascii_lowercase());
|
||||
}
|
||||
}
|
||||
result.replace("ifnotexists", "")
|
||||
};
|
||||
for name in [
|
||||
"typed_ticket_assignment_tracking_history",
|
||||
"typed_ticket_assignment_tracking_operations",
|
||||
"typed_ticket_assignment_tracking_initialize",
|
||||
"typed_ticket_assignment_tracking_assigned",
|
||||
"typed_ticket_assignment_tracking_history_no_update",
|
||||
"typed_ticket_assignment_tracking_history_no_delete",
|
||||
"typed_ticket_assignment_tracking_operations_no_update",
|
||||
"typed_ticket_assignment_tracking_operations_no_delete",
|
||||
"ticket_current_singleton_role_idx",
|
||||
"typed_ticket_assignment_identity_no_update",
|
||||
"typed_ticket_assignment_identity_no_delete",
|
||||
"typed_ticket_assignment_events_no_update",
|
||||
"typed_ticket_assignment_events_no_delete",
|
||||
"typed_ticket_assignment_receipt_no_update",
|
||||
"typed_ticket_assignment_receipt_no_delete",
|
||||
"typed_ticket_assignment_current_principal_insert",
|
||||
"typed_ticket_assignment_current_principal_update",
|
||||
"typed_ticket_assignment_orchestrator_principal_insert",
|
||||
"typed_ticket_assignment_released_tracking_no_update",
|
||||
"typed_ticket_assignment_parent_insert",
|
||||
"typed_ticket_assignment_role_principal_insert",
|
||||
"typed_ticket_assignment_audit_parent_insert",
|
||||
"typed_ticket_assignment_operation_parent_insert",
|
||||
] {
|
||||
let expected: String = canonical
|
||||
.query_row(
|
||||
"SELECT sql FROM sqlite_schema WHERE name=?1",
|
||||
[name],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
let actual: Option<String> = conn
|
||||
.query_row(
|
||||
"SELECT sql FROM sqlite_schema WHERE name=?1",
|
||||
[name],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.optional()
|
||||
.map_err(sqlite_err)?;
|
||||
if actual.map(&normalize).as_deref() != Some(normalize(expected).as_str()) {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket assignment tracking constraint {name} differs from canonical schema"
|
||||
)));
|
||||
}
|
||||
}
|
||||
for (table, columns) in [
|
||||
(
|
||||
"ticket_worker_assignments",
|
||||
vec![
|
||||
"workspace_id",
|
||||
"ticket_id",
|
||||
"assignment_id",
|
||||
"role",
|
||||
"principal_kind",
|
||||
"principal_id",
|
||||
"runtime_id",
|
||||
"worker_id",
|
||||
"assigned_by",
|
||||
"assigned_at",
|
||||
"tracking_json",
|
||||
"last_tracking_event_id",
|
||||
"tracking_updated_at",
|
||||
"tracking_migration_diagnostic",
|
||||
],
|
||||
),
|
||||
(
|
||||
"ticket_current_worker_assignments",
|
||||
vec![
|
||||
"workspace_id",
|
||||
"ticket_id",
|
||||
"role",
|
||||
"assignment_id",
|
||||
"principal_kind",
|
||||
"principal_id",
|
||||
"runtime_id",
|
||||
"worker_id",
|
||||
"updated_at",
|
||||
],
|
||||
),
|
||||
(
|
||||
"ticket_worker_assignment_events",
|
||||
vec![
|
||||
"workspace_id",
|
||||
"ticket_id",
|
||||
"role",
|
||||
"event_id",
|
||||
"action",
|
||||
"assignment_id",
|
||||
"previous_assignment_id",
|
||||
"actor",
|
||||
"created_at",
|
||||
"operation_id",
|
||||
"reason",
|
||||
],
|
||||
),
|
||||
(
|
||||
"ticket_assignment_operations",
|
||||
vec![
|
||||
"workspace_id",
|
||||
"operation_id",
|
||||
"action",
|
||||
"ticket_id",
|
||||
"role",
|
||||
"principal_kind",
|
||||
"principal_id",
|
||||
"runtime_id",
|
||||
"worker_id",
|
||||
"assignment_id",
|
||||
"expected_assignment_id",
|
||||
"created_at",
|
||||
"request_fingerprint",
|
||||
"claim_state",
|
||||
"failure_reason",
|
||||
"binding_recovery_json",
|
||||
],
|
||||
),
|
||||
(
|
||||
"typed_ticket_assignment_tracking_history",
|
||||
vec![
|
||||
"workspace_id",
|
||||
"ticket_id",
|
||||
"assignment_id",
|
||||
"event_id",
|
||||
"operation_key",
|
||||
"tracking_json",
|
||||
"source_context_json",
|
||||
"at",
|
||||
],
|
||||
),
|
||||
(
|
||||
"typed_ticket_assignment_tracking_operations",
|
||||
vec![
|
||||
"workspace_id",
|
||||
"operation_key",
|
||||
"ticket_id",
|
||||
"input_json",
|
||||
"source_context_json",
|
||||
"result_json",
|
||||
],
|
||||
),
|
||||
] {
|
||||
let actual = conn
|
||||
.prepare(&format!("PRAGMA table_info({table})"))
|
||||
.map_err(sqlite_err)?
|
||||
.query_map([], |r| r.get::<_, String>(1))
|
||||
.map_err(sqlite_err)?
|
||||
.collect::<std::result::Result<Vec<_>, _>>()
|
||||
.map_err(sqlite_err)?;
|
||||
if actual != columns {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket-owned assignment schema {table} has unexpected columns"
|
||||
)));
|
||||
}
|
||||
// Verify type, nullability, defaults and key positions, not just names.
|
||||
let shape = |db: &Connection| -> Result<Vec<(String, String, i64, Option<String>, i64)>> {
|
||||
db.prepare(&format!("PRAGMA table_info({table})"))
|
||||
.map_err(sqlite_err)?
|
||||
.query_map([], |r| {
|
||||
Ok((r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?))
|
||||
})
|
||||
.map_err(sqlite_err)?
|
||||
.collect::<std::result::Result<Vec<_>, _>>()
|
||||
.map_err(sqlite_err)
|
||||
};
|
||||
if shape(conn)? != shape(&canonical)? {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket-owned assignment schema {table} type/default/key constraints differ"
|
||||
)));
|
||||
}
|
||||
let foreign_keys = |db: &Connection| -> Result<BTreeSet<(String, String, String, String)>> {
|
||||
db.prepare(&format!("PRAGMA foreign_key_list({table})"))
|
||||
.map_err(sqlite_err)?
|
||||
.query_map([], |r| {
|
||||
Ok((
|
||||
r.get::<_, String>(2)?,
|
||||
r.get::<_, String>(3)?,
|
||||
r.get::<_, String>(4)?,
|
||||
r.get::<_, String>(6)?,
|
||||
))
|
||||
})
|
||||
.map_err(sqlite_err)?
|
||||
.filter_map(|r| match r {
|
||||
Ok(key) if key.0 == "workspaces" => None,
|
||||
other => Some(other),
|
||||
})
|
||||
.collect::<std::result::Result<BTreeSet<_>, _>>()
|
||||
.map_err(sqlite_err)
|
||||
};
|
||||
if !foreign_keys(&canonical)?.is_subset(&foreign_keys(conn)?) {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket-owned assignment schema {table} lacks canonical foreign keys"
|
||||
)));
|
||||
}
|
||||
let expected_sql: String = canonical
|
||||
.query_row(
|
||||
"SELECT sql FROM sqlite_schema WHERE name=?1",
|
||||
[table],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
let actual_sql: String = conn
|
||||
.query_row(
|
||||
"SELECT sql FROM sqlite_schema WHERE name=?1",
|
||||
[table],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
let expected_sql = normalize(expected_sql);
|
||||
let actual_sql = normalize(actual_sql);
|
||||
let mut remainder = expected_sql.as_str();
|
||||
while let Some(start) = remainder.find("check(") {
|
||||
let check = &remainder[start..];
|
||||
let mut depth = 0i64;
|
||||
let mut end = None;
|
||||
for (index, ch) in check.char_indices().skip(5) {
|
||||
if ch == '(' {
|
||||
depth += 1;
|
||||
} else if ch == ')' {
|
||||
depth -= 1;
|
||||
if depth == 0 {
|
||||
end = Some(index + 1);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
let end = end.ok_or_else(|| {
|
||||
TicketError::Sqlite("canonical assignment CHECK is malformed".into())
|
||||
})?;
|
||||
if !actual_sql.contains(&check[..end]) {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket-owned assignment schema {table} CHECK constraint differs"
|
||||
)));
|
||||
}
|
||||
remainder = &check[end..];
|
||||
}
|
||||
let index_keys = |db: &Connection| -> Result<BTreeSet<(bool, bool, Vec<String>)>> {
|
||||
let indices = db
|
||||
.prepare(&format!("PRAGMA index_list({table})"))
|
||||
.map_err(sqlite_err)?
|
||||
.query_map([], |r| {
|
||||
Ok((
|
||||
r.get::<_, String>(1)?,
|
||||
r.get::<_, bool>(2)?,
|
||||
r.get::<_, bool>(4)?,
|
||||
))
|
||||
})
|
||||
.map_err(sqlite_err)?
|
||||
.collect::<std::result::Result<Vec<_>, _>>()
|
||||
.map_err(sqlite_err)?;
|
||||
indices
|
||||
.into_iter()
|
||||
.map(|(name, unique, partial)| {
|
||||
let columns = db
|
||||
.prepare(&format!("PRAGMA index_info({name})"))
|
||||
.map_err(sqlite_err)?
|
||||
.query_map([], |r| r.get::<_, String>(2))
|
||||
.map_err(sqlite_err)?
|
||||
.collect::<std::result::Result<Vec<_>, _>>()
|
||||
.map_err(sqlite_err)?;
|
||||
Ok((unique, partial, columns))
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
if !index_keys(&canonical)?.is_subset(&index_keys(conn)?) {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket-owned assignment schema {table} lacks canonical key/index constraints"
|
||||
)));
|
||||
}
|
||||
let violations: i64 = conn
|
||||
.query_row(
|
||||
&format!("SELECT COUNT(*) FROM pragma_foreign_key_check('{table}')"),
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
if violations != 0 {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket-owned assignment schema {table} foreign key violations"
|
||||
)));
|
||||
}
|
||||
}
|
||||
for name in [
|
||||
"typed_ticket_assignment_tracking_initialize",
|
||||
"typed_ticket_assignment_tracking_assigned",
|
||||
"typed_ticket_assignment_tracking_history_no_update",
|
||||
"typed_ticket_assignment_tracking_history_no_delete",
|
||||
"typed_ticket_assignment_tracking_operations_no_update",
|
||||
"typed_ticket_assignment_tracking_operations_no_delete",
|
||||
] {
|
||||
let exists: bool = conn
|
||||
.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type='trigger' AND name=?1)",
|
||||
[name],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
if !exists {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket assignment tracking constraint {name} missing"
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "assignment_tracking_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,128 @@
|
||||
-- Applied only by component 10 -> 11, never by request handlers.
|
||||
ALTER TABLE ticket_worker_assignments ADD COLUMN tracking_json TEXT
|
||||
CHECK(tracking_json IS NULL OR (role = 'orchestrator' AND json_valid(tracking_json) AND json_type(tracking_json)='object'
|
||||
AND COALESCE(json_extract(tracking_json, '$.status') IN ('waiting', 'inprogress', 'done'), 0)));
|
||||
ALTER TABLE ticket_worker_assignments ADD COLUMN last_tracking_event_id TEXT;
|
||||
ALTER TABLE ticket_worker_assignments ADD COLUMN tracking_updated_at TEXT;
|
||||
ALTER TABLE ticket_worker_assignments ADD COLUMN tracking_migration_diagnostic TEXT;
|
||||
CREATE TABLE typed_ticket_assignment_tracking_history (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
assignment_id TEXT NOT NULL,
|
||||
event_id TEXT NOT NULL,
|
||||
operation_key TEXT NOT NULL,
|
||||
tracking_json TEXT NOT NULL CHECK(json_valid(tracking_json)),
|
||||
source_context_json TEXT NOT NULL CHECK(json_valid(source_context_json)),
|
||||
at TEXT NOT NULL,
|
||||
PRIMARY KEY(workspace_id, event_id),
|
||||
UNIQUE(workspace_id, assignment_id, operation_key),
|
||||
FOREIGN KEY(workspace_id, ticket_id, assignment_id)
|
||||
REFERENCES ticket_worker_assignments(workspace_id, ticket_id, assignment_id)
|
||||
);
|
||||
CREATE TABLE typed_ticket_assignment_tracking_operations (
|
||||
workspace_id TEXT NOT NULL,
|
||||
operation_key TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
input_json TEXT NOT NULL,
|
||||
source_context_json TEXT NOT NULL,
|
||||
result_json TEXT NOT NULL,
|
||||
PRIMARY KEY(workspace_id, operation_key)
|
||||
);
|
||||
CREATE TRIGGER typed_ticket_assignment_tracking_initialize
|
||||
AFTER INSERT ON ticket_worker_assignments WHEN NEW.role = 'orchestrator'
|
||||
BEGIN
|
||||
UPDATE ticket_worker_assignments SET tracking_json = '{"status":"waiting","waiting_reason":"Initial judgment pending","progress":null,"next_action":null,"result":null}', tracking_updated_at = NEW.assigned_at
|
||||
WHERE workspace_id = NEW.workspace_id AND assignment_id = NEW.assignment_id;
|
||||
END;
|
||||
CREATE TRIGGER typed_ticket_assignment_tracking_assigned
|
||||
AFTER INSERT ON ticket_worker_assignment_events
|
||||
WHEN NEW.role = 'orchestrator' AND NEW.action IN ('assigned', 'reassigned')
|
||||
BEGIN
|
||||
UPDATE ticket_worker_assignments SET last_tracking_event_id = NEW.event_id, tracking_updated_at = NEW.created_at
|
||||
WHERE workspace_id = NEW.workspace_id AND assignment_id = NEW.assignment_id AND tracking_json IS NOT NULL;
|
||||
INSERT INTO typed_ticket_assignment_tracking_history
|
||||
SELECT workspace_id, ticket_id, assignment_id, NEW.event_id, COALESCE(NEW.operation_id, NEW.event_id), tracking_json,
|
||||
json_object('actor', NEW.actor, 'kind', 'assignment'), NEW.created_at
|
||||
FROM ticket_worker_assignments
|
||||
WHERE workspace_id = NEW.workspace_id AND assignment_id = NEW.assignment_id AND tracking_json IS NOT NULL;
|
||||
END;
|
||||
CREATE TRIGGER typed_ticket_assignment_tracking_history_no_update
|
||||
BEFORE UPDATE ON typed_ticket_assignment_tracking_history
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment tracking history is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_tracking_history_no_delete
|
||||
BEFORE DELETE ON typed_ticket_assignment_tracking_history
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment tracking history is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_tracking_operations_no_update
|
||||
BEFORE UPDATE ON typed_ticket_assignment_tracking_operations
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment tracking receipts are immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_tracking_operations_no_delete
|
||||
BEFORE DELETE ON typed_ticket_assignment_tracking_operations
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment tracking receipts are immutable'); END;
|
||||
-- Original assignment identity/principal/audit is append-only. Tracking is the
|
||||
-- only mutable assignment value; startup claims are mutable only while pending.
|
||||
CREATE TRIGGER typed_ticket_assignment_identity_no_update
|
||||
BEFORE UPDATE OF workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,runtime_id,worker_id,assigned_by,assigned_at ON ticket_worker_assignments
|
||||
WHEN OLD.workspace_id IS NOT NEW.workspace_id OR OLD.ticket_id IS NOT NEW.ticket_id
|
||||
OR OLD.assignment_id IS NOT NEW.assignment_id OR OLD.role IS NOT NEW.role
|
||||
OR OLD.principal_kind IS NOT NEW.principal_kind OR OLD.principal_id IS NOT NEW.principal_id
|
||||
OR OLD.runtime_id IS NOT NEW.runtime_id OR OLD.worker_id IS NOT NEW.worker_id
|
||||
OR OLD.assigned_by IS NOT NEW.assigned_by OR OLD.assigned_at IS NOT NEW.assigned_at
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment identity is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_identity_no_delete BEFORE DELETE ON ticket_worker_assignments
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment identity is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_events_no_update BEFORE UPDATE ON ticket_worker_assignment_events
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment audit is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_events_no_delete BEFORE DELETE ON ticket_worker_assignment_events
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment audit is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_receipt_no_update BEFORE UPDATE ON ticket_assignment_operations
|
||||
WHEN OLD.claim_state != 'pending'
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment terminal receipt is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_receipt_no_delete BEFORE DELETE ON ticket_assignment_operations
|
||||
WHEN OLD.claim_state != 'pending'
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket assignment terminal receipt is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_current_principal_insert BEFORE INSERT ON ticket_current_worker_assignments
|
||||
WHEN NOT EXISTS (SELECT 1 FROM ticket_worker_assignments a WHERE a.workspace_id=NEW.workspace_id
|
||||
AND a.ticket_id=NEW.ticket_id AND a.assignment_id=NEW.assignment_id AND a.role=NEW.role
|
||||
AND a.principal_kind=NEW.principal_kind AND a.principal_id IS NEW.principal_id
|
||||
AND a.runtime_id IS NEW.runtime_id AND a.worker_id IS NEW.worker_id)
|
||||
OR EXISTS (SELECT 1 FROM ticket_worker_assignment_events e WHERE e.workspace_id=NEW.workspace_id
|
||||
AND ((e.action='unassigned' AND (e.assignment_id=NEW.assignment_id OR e.previous_assignment_id=NEW.assignment_id))
|
||||
OR (e.action='reassigned' AND e.previous_assignment_id=NEW.assignment_id)))
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket current assignment must match an unreleased immutable identity'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_current_principal_update BEFORE UPDATE ON ticket_current_worker_assignments
|
||||
WHEN NOT EXISTS (SELECT 1 FROM ticket_worker_assignments a WHERE a.workspace_id=NEW.workspace_id
|
||||
AND a.ticket_id=NEW.ticket_id AND a.assignment_id=NEW.assignment_id AND a.role=NEW.role
|
||||
AND a.principal_kind=NEW.principal_kind AND a.principal_id IS NEW.principal_id
|
||||
AND a.runtime_id IS NEW.runtime_id AND a.worker_id IS NEW.worker_id)
|
||||
OR EXISTS (SELECT 1 FROM ticket_worker_assignment_events e WHERE e.workspace_id=NEW.workspace_id
|
||||
AND ((e.action='unassigned' AND (e.assignment_id=NEW.assignment_id OR e.previous_assignment_id=NEW.assignment_id))
|
||||
OR (e.action='reassigned' AND e.previous_assignment_id=NEW.assignment_id)))
|
||||
BEGIN SELECT RAISE(ABORT, 'Ticket current assignment must match an unreleased immutable identity'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_orchestrator_principal_insert BEFORE INSERT ON ticket_worker_assignments
|
||||
WHEN NEW.role='orchestrator' AND (NEW.principal_kind!='workspace_agent' OR NEW.principal_id!='workspace-orchestrator')
|
||||
BEGIN SELECT RAISE(ABORT, 'Orchestrator assignment requires the stable Workspace agent principal'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_released_tracking_no_update
|
||||
BEFORE UPDATE OF tracking_json,last_tracking_event_id,tracking_updated_at ON ticket_worker_assignments
|
||||
WHEN EXISTS (SELECT 1 FROM ticket_worker_assignment_events e WHERE e.workspace_id=OLD.workspace_id
|
||||
AND ((e.action='unassigned' AND (e.assignment_id=OLD.assignment_id OR e.previous_assignment_id=OLD.assignment_id))
|
||||
OR (e.action='reassigned' AND e.previous_assignment_id=OLD.assignment_id)))
|
||||
BEGIN SELECT RAISE(ABORT, 'Released assignment tracking is immutable'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_parent_insert BEFORE INSERT ON ticket_worker_assignments
|
||||
WHEN NOT EXISTS (SELECT 1 FROM typed_tickets t WHERE t.workspace_id=NEW.workspace_id AND t.ticket_id=NEW.ticket_id)
|
||||
BEGIN SELECT RAISE(ABORT, 'New assignment requires a Ticket in the same Workspace'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_role_principal_insert BEFORE INSERT ON ticket_worker_assignments
|
||||
WHEN (NEW.role='owner' AND NEW.principal_kind!='user')
|
||||
OR (NEW.role='worker' AND NEW.principal_kind='workspace_agent')
|
||||
OR (NEW.principal_kind='workspace_agent' AND NEW.principal_id!='workspace-orchestrator')
|
||||
BEGIN SELECT RAISE(ABORT, 'Assignment role and principal are incompatible'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_audit_parent_insert BEFORE INSERT ON ticket_worker_assignment_events
|
||||
WHEN (NEW.assignment_id IS NOT NULL AND NOT EXISTS (SELECT 1 FROM ticket_worker_assignments a WHERE a.workspace_id=NEW.workspace_id AND a.ticket_id=NEW.ticket_id AND a.role=NEW.role AND a.assignment_id=NEW.assignment_id))
|
||||
OR (NEW.previous_assignment_id IS NOT NULL AND NOT EXISTS (SELECT 1 FROM ticket_worker_assignments a WHERE a.workspace_id=NEW.workspace_id AND a.ticket_id=NEW.ticket_id AND a.role=NEW.role AND a.assignment_id=NEW.previous_assignment_id))
|
||||
BEGIN SELECT RAISE(ABORT, 'Assignment audit must match the same immutable Ticket role'); END;
|
||||
CREATE TRIGGER typed_ticket_assignment_operation_parent_insert BEFORE INSERT ON ticket_assignment_operations
|
||||
WHEN NOT EXISTS (SELECT 1 FROM typed_tickets t WHERE t.workspace_id=NEW.workspace_id AND t.ticket_id=NEW.ticket_id)
|
||||
BEGIN SELECT RAISE(ABORT, 'New assignment operation requires a Ticket in the same Workspace'); END;
|
||||
|
||||
-- Equality scope keys leave SQLite's implicit rowid as stable append ordering.
|
||||
CREATE INDEX typed_ticket_assignment_tracking_history_scope
|
||||
ON typed_ticket_assignment_tracking_history(workspace_id,ticket_id,assignment_id);
|
||||
@@ -0,0 +1,730 @@
|
||||
use super::*;
|
||||
|
||||
fn db() -> Connection {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
conn.execute_batch("PRAGMA foreign_keys=ON").unwrap();
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
seed_ticket(&conn, "w", "t");
|
||||
conn
|
||||
}
|
||||
|
||||
fn seed_ticket(conn: &Connection, workspace: &str, ticket: &str) {
|
||||
conn.execute("INSERT INTO typed_tickets(workspace_id,ticket_id,slug,title,status,kind,priority,body,workflow_state,workflow_state_explicit) VALUES (?1,?2,?2,'Title','open','task','P2','body','open',1)",params![workspace,ticket]).unwrap();
|
||||
}
|
||||
|
||||
fn assign(conn: &Connection, id: &str, event: &str) {
|
||||
let previous: Option<String> = conn.query_row("SELECT assignment_id FROM ticket_current_worker_assignments WHERE workspace_id='w' AND ticket_id='t' AND role='orchestrator'",[],|r|r.get(0)).optional().unwrap();
|
||||
conn.execute_batch("BEGIN IMMEDIATE").unwrap();
|
||||
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES ('w','t',?1,'orchestrator','workspace_agent','workspace-orchestrator','author','2026-10-10T00:00:00Z')",[id]).unwrap();
|
||||
conn.execute("DELETE FROM ticket_current_worker_assignments WHERE workspace_id='w' AND ticket_id='t' AND role='orchestrator'",[]).unwrap();
|
||||
conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES ('w','t',?1,'orchestrator','workspace_agent','workspace-orchestrator','now')",[id]).unwrap();
|
||||
conn.execute("INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,previous_assignment_id,actor,created_at,operation_id) VALUES ('w','t','orchestrator',?2,?4,?1,?3,'author','now',?2)",params![id,event,previous,if previous.is_some(){"reassigned"}else{"assigned"}]).unwrap();
|
||||
conn.execute_batch("COMMIT").unwrap();
|
||||
}
|
||||
|
||||
fn source() -> TicketAssignmentTrackingSourceContext {
|
||||
TicketAssignmentTrackingSourceContext {
|
||||
actor: "orchestrator".into(),
|
||||
runtime_id: "runtime".into(),
|
||||
worker_id: "worker".into(),
|
||||
agent_key: "workspace-orchestrator".into(),
|
||||
authentication_binding_id: "binding".into(),
|
||||
}
|
||||
}
|
||||
fn update(
|
||||
id: &str,
|
||||
event: &str,
|
||||
key: &str,
|
||||
status: TicketAssignmentTrackingStatus,
|
||||
) -> TicketAssignmentTrackingUpdate {
|
||||
TicketAssignmentTrackingUpdate {
|
||||
assignment_id: id.into(),
|
||||
expected_last_event_id: event.into(),
|
||||
operation_key: key.into(),
|
||||
tracking: TicketAssignmentTracking {
|
||||
status,
|
||||
waiting_reason: None,
|
||||
progress: Some("Considering request".into()),
|
||||
next_action: None,
|
||||
result: Some("Returned alternatives".into()),
|
||||
},
|
||||
}
|
||||
}
|
||||
fn apply(
|
||||
conn: &Connection,
|
||||
input: &TicketAssignmentTrackingUpdate,
|
||||
source: &TicketAssignmentTrackingSourceContext,
|
||||
) -> Result<TicketAssignmentTrackingProjection> {
|
||||
conn.execute_batch("BEGIN IMMEDIATE").unwrap();
|
||||
let result =
|
||||
sqlite_update_ticket_assignment_tracking_in_transaction(conn, "w", "t", input, source);
|
||||
conn.execute_batch(if result.is_ok() { "COMMIT" } else { "ROLLBACK" })
|
||||
.unwrap();
|
||||
result
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assignment_initializes_waiting_and_audit_atomically() {
|
||||
let conn = db();
|
||||
assign(&conn, "a", "assigned-a");
|
||||
let projection = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(projection.current);
|
||||
assert_eq!(projection.last_event_id.as_deref(), Some("assigned-a"));
|
||||
assert_eq!(
|
||||
projection.tracking.unwrap().status,
|
||||
TicketAssignmentTrackingStatus::Waiting
|
||||
);
|
||||
let history = sqlite_ticket_assignment_tracking_history(&conn, "w", "t", "a").unwrap();
|
||||
assert_eq!(history.len(), 1);
|
||||
assert_eq!(history[0].event_id, "assigned-a");
|
||||
conn.execute_batch("BEGIN IMMEDIATE").unwrap();
|
||||
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES ('w','t','rolled-back','orchestrator','workspace_agent','workspace-orchestrator','author','now')",[]).unwrap();
|
||||
conn.execute_batch("ROLLBACK").unwrap();
|
||||
assert!(
|
||||
sqlite_ticket_assignment_tracking(&conn, "w", "t", "rolled-back")
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn direct_done_close_reopen_and_removal_keep_last_result_and_exact_replay() {
|
||||
let conn = db();
|
||||
assign(&conn, "a", "assigned-a");
|
||||
let input = update(
|
||||
"a",
|
||||
"assigned-a",
|
||||
"done",
|
||||
TicketAssignmentTrackingStatus::Done,
|
||||
);
|
||||
let result = apply(&conn, &input, &source()).unwrap();
|
||||
assert_eq!(
|
||||
conn.query_row("SELECT workflow_state FROM typed_tickets", [], |r| r
|
||||
.get::<_, String>(0))
|
||||
.unwrap(),
|
||||
"open"
|
||||
);
|
||||
conn.execute("UPDATE typed_tickets SET workflow_state='closed'", [])
|
||||
.unwrap();
|
||||
let input2 = update(
|
||||
"a",
|
||||
result.last_event_id.as_deref().unwrap(),
|
||||
"correct-after-close",
|
||||
TicketAssignmentTrackingStatus::Waiting,
|
||||
);
|
||||
let corrected = apply(&conn, &input2, &source()).unwrap();
|
||||
assert_eq!(
|
||||
conn.query_row("SELECT workflow_state FROM typed_tickets", [], |r| r
|
||||
.get::<_, String>(0))
|
||||
.unwrap(),
|
||||
"closed"
|
||||
);
|
||||
conn.execute("UPDATE typed_tickets SET workflow_state='open'", [])
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap(),
|
||||
corrected
|
||||
);
|
||||
assign(&conn, "b", "assigned-b");
|
||||
assert_eq!(apply(&conn, &input, &source()).unwrap(), result);
|
||||
assert_eq!(apply(&conn, &input2, &source()).unwrap(), corrected);
|
||||
assert!(
|
||||
apply(
|
||||
&conn,
|
||||
&update(
|
||||
"a",
|
||||
corrected.last_event_id.as_deref().unwrap(),
|
||||
"old-new",
|
||||
TicketAssignmentTrackingStatus::Done
|
||||
),
|
||||
&source()
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
let old = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(!old.current);
|
||||
assert_eq!(old.tracking, corrected.tracking);
|
||||
let b = sqlite_ticket_assignment_tracking(&conn, "w", "t", "b")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
b.tracking.unwrap().status,
|
||||
TicketAssignmentTrackingStatus::Waiting
|
||||
);
|
||||
conn.execute("DELETE FROM ticket_current_worker_assignments", [])
|
||||
.unwrap();
|
||||
assert!(
|
||||
apply(
|
||||
&conn,
|
||||
&update(
|
||||
"b",
|
||||
"assigned-b",
|
||||
"removed-new",
|
||||
TicketAssignmentTrackingStatus::Done
|
||||
),
|
||||
&source()
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
sqlite_current_ticket_assignment_tracking(&conn, "w", "t")
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
assert_eq!(
|
||||
sqlite_ticket_assignment_tracking_history(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.len(),
|
||||
3
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tracking_cas_and_receipts_bind_full_payload_and_authenticated_source() {
|
||||
let conn = db();
|
||||
assign(&conn, "a", "assigned-a");
|
||||
let input = update(
|
||||
"a",
|
||||
"assigned-a",
|
||||
"update",
|
||||
TicketAssignmentTrackingStatus::Inprogress,
|
||||
);
|
||||
let result = apply(&conn, &input, &source()).unwrap();
|
||||
assert_eq!(apply(&conn, &input, &source()).unwrap(), result);
|
||||
let mut changed = input.clone();
|
||||
changed.tracking.result = Some("Different result".into());
|
||||
assert!(apply(&conn, &changed, &source()).is_err());
|
||||
changed = input.clone();
|
||||
changed.operation_key = "fresh-stale".into();
|
||||
assert!(apply(&conn, &changed, &source()).is_err());
|
||||
for change in 0..5 {
|
||||
let mut sender = source();
|
||||
match change {
|
||||
0 => sender.actor = "other".into(),
|
||||
1 => sender.runtime_id = "other".into(),
|
||||
2 => sender.worker_id = "replacement".into(),
|
||||
3 => sender.authentication_binding_id = "replacement-binding".into(),
|
||||
_ => sender.agent_key = "other-agent".into(),
|
||||
};
|
||||
assert!(apply(&conn, &input, &sender).is_err());
|
||||
}
|
||||
let mut other = source();
|
||||
other.worker_id = "replacement".into();
|
||||
// Backend authenticated the replacement as the current stable agent. A new
|
||||
// operation can proceed on the same assignment without starting a new row.
|
||||
let next = update(
|
||||
"a",
|
||||
result.last_event_id.as_deref().unwrap(),
|
||||
"replacement-update",
|
||||
TicketAssignmentTrackingStatus::Done,
|
||||
);
|
||||
apply(&conn, &next, &other).unwrap();
|
||||
assert_eq!(
|
||||
conn.query_row("SELECT COUNT(*) FROM ticket_worker_assignments", [], |r| {
|
||||
r.get::<_, i64>(0)
|
||||
})
|
||||
.unwrap(),
|
||||
1
|
||||
);
|
||||
assert!(
|
||||
sqlite_ticket_assignment_tracking(&conn, "other", "t", "a")
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
conn.execute_batch("BEGIN IMMEDIATE").unwrap();
|
||||
assert!(
|
||||
sqlite_update_ticket_assignment_tracking_in_transaction(
|
||||
&conn,
|
||||
"other",
|
||||
"t",
|
||||
&input,
|
||||
&source()
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
conn.execute_batch("ROLLBACK").unwrap();
|
||||
assert!(
|
||||
conn.execute(
|
||||
"UPDATE typed_ticket_assignment_tracking_history SET at='tamper'",
|
||||
[]
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
conn.execute(
|
||||
"DELETE FROM typed_ticket_assignment_tracking_operations",
|
||||
[]
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tracking_reopens_database_without_restarting_or_reassigning_workers() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let path = directory.path().join("ticket.db");
|
||||
{
|
||||
let conn = Connection::open(&path).unwrap();
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
seed_ticket(&conn, "w", "t");
|
||||
assign(&conn, "a", "assigned-a");
|
||||
apply(
|
||||
&conn,
|
||||
&update(
|
||||
"a",
|
||||
"assigned-a",
|
||||
"done",
|
||||
TicketAssignmentTrackingStatus::Done,
|
||||
),
|
||||
&source(),
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
let conn = Connection::open(&path).unwrap();
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
assert_eq!(
|
||||
sqlite_current_ticket_assignment_tracking(&conn, "w", "t").unwrap()[0]
|
||||
.tracking
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.status,
|
||||
TicketAssignmentTrackingStatus::Done
|
||||
);
|
||||
assert_eq!(
|
||||
sqlite_ticket_assignment_tracking_history(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.len(),
|
||||
2
|
||||
);
|
||||
}
|
||||
|
||||
fn legacy(original: &str, fenced: bool, later_case: bool, reassigned: bool) -> Connection {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
conn.execute_batch(include_str!("latest_schema.sql"))
|
||||
.unwrap();
|
||||
conn.execute_batch("CREATE TABLE ticket_schema_migrations(version INTEGER PRIMARY KEY,name TEXT NOT NULL,applied_at TEXT NOT NULL); INSERT INTO ticket_schema_migrations VALUES(10,'ticket schema baseline','original-time');").unwrap();
|
||||
install_assignment_core(&conn).unwrap();
|
||||
seed_ticket(&conn, "w", "t");
|
||||
assign(&conn, "a", "assigned-a");
|
||||
let snapshot = TicketLegacyWorkflowSnapshot {
|
||||
workflow_state: original.into(),
|
||||
workflow_state_explicit: true,
|
||||
status: original.into(),
|
||||
queued_by: Some("author".into()),
|
||||
queued_at: Some("original-queue-time".into()),
|
||||
resolution: None,
|
||||
};
|
||||
if original != "no-snapshot" {
|
||||
conn.execute(
|
||||
"INSERT INTO typed_ticket_legacy_workflow_snapshot VALUES('w','t',?1)",
|
||||
[serde_json::to_string(&snapshot).unwrap()],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
conn.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,to_state,at,body) VALUES('w','t',1,'state_changed','queued','original-queue-time','Queue')",[]).unwrap();
|
||||
if fenced {
|
||||
conn.execute("INSERT INTO typed_ticket_event_attributes VALUES('w','t',1,'orchestrator_assignment_id','a')",[]).unwrap();
|
||||
}
|
||||
if later_case {
|
||||
conn.execute("INSERT INTO typed_ticket_case_history VALUES('w','t',0,'closed-later','case-fingerprint','{}')",[]).unwrap();
|
||||
}
|
||||
if reassigned {
|
||||
assign(&conn, "b", "assigned-b");
|
||||
}
|
||||
conn
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migration_preserves_original_records_and_only_reconciles_current_fenced_delegation() {
|
||||
for (original, fenced, later_case, reassigned, expected) in [
|
||||
(
|
||||
"queued",
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
Some(TicketAssignmentTrackingStatus::Waiting),
|
||||
),
|
||||
(
|
||||
"inprogress",
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
Some(TicketAssignmentTrackingStatus::Inprogress),
|
||||
),
|
||||
("queued", false, false, false, None),
|
||||
("queued", true, true, false, None),
|
||||
("inprogress", true, false, true, None),
|
||||
("ready", true, false, false, None),
|
||||
("done", true, false, false, None),
|
||||
("closed", true, false, false, None),
|
||||
] {
|
||||
let conn = legacy(original, fenced, later_case, reassigned);
|
||||
let original_snapshot: String = conn
|
||||
.query_row(
|
||||
"SELECT snapshot_json FROM typed_ticket_legacy_workflow_snapshot",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
let original_event: String = conn
|
||||
.query_row(
|
||||
"SELECT event_id FROM ticket_worker_assignment_events ORDER BY event_id LIMIT 1",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
assert_eq!(
|
||||
conn.query_row(
|
||||
"SELECT applied_at FROM ticket_schema_migrations WHERE version=10",
|
||||
[],
|
||||
|r| r.get::<_, String>(0)
|
||||
)
|
||||
.unwrap(),
|
||||
"original-time"
|
||||
);
|
||||
assert_eq!(
|
||||
conn.query_row(
|
||||
"SELECT MAX(version) FROM ticket_schema_migrations",
|
||||
[],
|
||||
|r| r.get::<_, i64>(0)
|
||||
)
|
||||
.unwrap(),
|
||||
11
|
||||
);
|
||||
let value = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
value.tracking.as_ref().map(|t| t.status),
|
||||
expected,
|
||||
"{original}/{fenced}/{later_case}/{reassigned}"
|
||||
);
|
||||
assert!(value.migration_diagnostic.is_some());
|
||||
assert_eq!(
|
||||
conn.query_row(
|
||||
"SELECT snapshot_json FROM typed_ticket_legacy_workflow_snapshot",
|
||||
[],
|
||||
|r| r.get::<_, String>(0)
|
||||
)
|
||||
.unwrap(),
|
||||
original_snapshot
|
||||
);
|
||||
assert_eq!(
|
||||
conn.query_row(
|
||||
"SELECT event_id FROM ticket_worker_assignment_events ORDER BY event_id LIMIT 1",
|
||||
[],
|
||||
|r| r.get::<_, String>(0)
|
||||
)
|
||||
.unwrap(),
|
||||
original_event
|
||||
);
|
||||
let history = sqlite_ticket_assignment_tracking_history(&conn, "w", "t", "a").unwrap();
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
assert_eq!(
|
||||
sqlite_ticket_assignment_tracking_history(&conn, "w", "t", "a").unwrap(),
|
||||
history
|
||||
);
|
||||
if reassigned {
|
||||
assert!(
|
||||
sqlite_ticket_assignment_tracking(&conn, "w", "t", "b")
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.tracking
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tracking_schema_rejects_drift_and_duplicate_current_singleton_roles() {
|
||||
let conn = db();
|
||||
assign(&conn, "a", "assigned-a");
|
||||
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','b','orchestrator','workspace_agent','workspace-orchestrator','author','now')",[]).unwrap();
|
||||
assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','b','orchestrator','workspace_agent','workspace-orchestrator','now')",[]).is_err());
|
||||
conn.execute_batch("DROP TRIGGER typed_ticket_assignment_tracking_history_no_update; CREATE TRIGGER typed_ticket_assignment_tracking_history_no_update BEFORE UPDATE ON typed_ticket_assignment_tracking_history BEGIN SELECT 1; END;").unwrap();
|
||||
assert!(
|
||||
verify_sqlite_ticket_schema(&conn)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("differs from canonical schema")
|
||||
);
|
||||
assert!(migrate_sqlite_ticket_schema(&conn).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tracking_input_wire_contract_rejects_unknown_fields_and_statuses() {
|
||||
let input = update(
|
||||
"a",
|
||||
"event",
|
||||
"operation",
|
||||
TicketAssignmentTrackingStatus::Inprogress,
|
||||
);
|
||||
let mut value = serde_json::to_value(&input).unwrap();
|
||||
assert_eq!(value["tracking"]["status"], "inprogress");
|
||||
value["claimed_role"] = serde_json::json!("orchestrator");
|
||||
assert!(serde_json::from_value::<TicketAssignmentTrackingUpdate>(value).is_err());
|
||||
let mut value = serde_json::to_value(&input).unwrap();
|
||||
value["tracking"]["status"] = serde_json::json!("running");
|
||||
assert!(serde_json::from_value::<TicketAssignmentTrackingUpdate>(value).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standalone_assignment_identity_audit_and_terminal_receipts_are_immutable_but_pending_claims_can_finish()
|
||||
{
|
||||
let conn = db();
|
||||
assign(&conn, "a", "assigned-a");
|
||||
conn.execute("INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,role,principal_kind,principal_id,assignment_id,created_at,request_fingerprint) VALUES('w','assigned-op','assign','t','orchestrator','workspace_agent','workspace-orchestrator','a','original-time','original-fingerprint')",[]).unwrap();
|
||||
for sql in [
|
||||
"UPDATE ticket_worker_assignments SET assignment_id='changed' WHERE assignment_id='a'",
|
||||
"UPDATE ticket_worker_assignments SET assigned_by='forged' WHERE assignment_id='a'",
|
||||
"UPDATE ticket_worker_assignments SET assigned_at='forged-time' WHERE assignment_id='a'",
|
||||
"UPDATE ticket_worker_assignments SET role='owner',principal_kind='user',principal_id='user' WHERE assignment_id='a'",
|
||||
"DELETE FROM ticket_worker_assignments WHERE assignment_id='a'",
|
||||
"UPDATE ticket_worker_assignment_events SET actor='forged'",
|
||||
"UPDATE ticket_worker_assignment_events SET event_id='forged'",
|
||||
"DELETE FROM ticket_worker_assignment_events",
|
||||
"UPDATE ticket_assignment_operations SET request_fingerprint='forged' WHERE operation_id='assigned-op'",
|
||||
"UPDATE ticket_assignment_operations SET created_at='forged' WHERE operation_id='assigned-op'",
|
||||
"DELETE FROM ticket_assignment_operations WHERE operation_id='assigned-op'",
|
||||
"UPDATE ticket_current_worker_assignments SET principal_id='another-agent'",
|
||||
"UPDATE ticket_worker_assignments SET tracking_json='{}' WHERE assignment_id='a'",
|
||||
] {
|
||||
assert!(
|
||||
conn.execute(sql, []).is_err(),
|
||||
"deny direct mutation: {sql}"
|
||||
);
|
||||
}
|
||||
// The immutable original principal also fences insertion of a mismatched
|
||||
// current row, even in a standalone Ticket store without host constraints.
|
||||
conn.execute("DELETE FROM ticket_current_worker_assignments", [])
|
||||
.unwrap();
|
||||
assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','a','orchestrator','workspace_agent','different-agent','now')",[]).is_err());
|
||||
conn.execute("INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,role,principal_kind,runtime_id,worker_id,created_at,request_fingerprint,claim_state) VALUES('w','pending','assign','t','worker','worker','runtime','pending-worker','claim-time','claim-fingerprint','pending')",[]).unwrap();
|
||||
conn.execute("UPDATE ticket_assignment_operations SET worker_id='resolved-worker',binding_recovery_json='{}' WHERE operation_id='pending'",[]).unwrap();
|
||||
conn.execute("UPDATE ticket_assignment_operations SET claim_state='failed',failure_reason='Admission failed' WHERE operation_id='pending'",[]).unwrap();
|
||||
assert!(conn.execute("UPDATE ticket_assignment_operations SET failure_reason='rewrite' WHERE operation_id='pending'",[]).is_err());
|
||||
assert!(
|
||||
conn.execute(
|
||||
"DELETE FROM ticket_assignment_operations WHERE operation_id='pending'",
|
||||
[]
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
verify_sqlite_ticket_schema(&conn).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn release_projection_preserves_actor_time_reason_and_optional_legacy_tracking() {
|
||||
let conn = db();
|
||||
assign(&conn, "a", "assigned-a");
|
||||
let result = apply(
|
||||
&conn,
|
||||
&update(
|
||||
"a",
|
||||
"assigned-a",
|
||||
"result",
|
||||
TicketAssignmentTrackingStatus::Done,
|
||||
),
|
||||
&source(),
|
||||
)
|
||||
.unwrap();
|
||||
assign(&conn, "b", "assigned-b");
|
||||
let old = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let release = old.release.as_ref().unwrap();
|
||||
assert_eq!(release.action, TicketAssignmentReleaseAction::Replaced);
|
||||
assert_eq!(release.replacement_assignment_id.as_deref(), Some("b"));
|
||||
assert_eq!(release.last_tracking, result.tracking);
|
||||
assert_eq!(release.actor, "author");
|
||||
assert_eq!(release.event_id, "assigned-b");
|
||||
conn.execute("DELETE FROM ticket_current_worker_assignments", [])
|
||||
.unwrap();
|
||||
conn.execute("INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,previous_assignment_id,actor,created_at,operation_id,reason) VALUES('w','t','orchestrator','unassign-b','unassigned','b','human','original-unassign-time','unassign-op','Withdrawn')",[]).unwrap();
|
||||
let removed = sqlite_ticket_assignment_tracking(&conn, "w", "t", "b")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let release = removed.release.unwrap();
|
||||
assert_eq!(release.action, TicketAssignmentReleaseAction::Unassigned);
|
||||
assert_eq!(release.at, "original-unassign-time");
|
||||
assert_eq!(release.reason.as_deref(), Some("Withdrawn"));
|
||||
assert_eq!(
|
||||
release.last_tracking.as_ref().unwrap().status,
|
||||
TicketAssignmentTrackingStatus::Waiting
|
||||
);
|
||||
assert!(conn.execute("UPDATE ticket_worker_assignments SET tracking_json='{\"status\":\"done\"}' WHERE assignment_id='b'",[]).is_err());
|
||||
assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','b','orchestrator','workspace_agent','workspace-orchestrator','now')",[]).is_err());
|
||||
let legacy = legacy("done", true, false, false);
|
||||
migrate_sqlite_ticket_schema(&legacy).unwrap();
|
||||
legacy
|
||||
.execute("DELETE FROM ticket_current_worker_assignments", [])
|
||||
.unwrap();
|
||||
legacy.execute("INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,previous_assignment_id,actor,created_at,operation_id) VALUES('w','t','orchestrator','unassign-legacy','unassigned','a','human','legacy-unassign-time','legacy-unassign-op')",[]).unwrap();
|
||||
let original = sqlite_ticket_assignment_tracking(&legacy, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(original.tracking.is_none());
|
||||
assert!(original.release.unwrap().last_tracking.is_none());
|
||||
}
|
||||
|
||||
fn routing_queue(conn: &Connection, index: i64, fence: &str) {
|
||||
conn.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,from_state,to_state,at,state_field,reason,body) VALUES('w','t',?1,'state_changed','open','open','typed-queue-time','routing','queued','Explicit component-10 Queue')",[index]).unwrap();
|
||||
conn.execute("INSERT INTO typed_ticket_event_attributes VALUES('w','t',?1,'orchestrator_assignment_id',?2)",params![index,fence]).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migration_recognizes_typed_case_neutral_queue_and_never_revives_an_older_mismatched_queue() {
|
||||
for original in ["ready", "no-snapshot"] {
|
||||
let conn = legacy(original, true, false, false);
|
||||
routing_queue(&conn, 2, "a");
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
let assignment = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
assignment.tracking.as_ref().unwrap().status,
|
||||
TicketAssignmentTrackingStatus::Waiting
|
||||
);
|
||||
assert_eq!(assignment.updated_at.as_deref(), Some("typed-queue-time"));
|
||||
}
|
||||
let conn = legacy("queued", true, false, false);
|
||||
routing_queue(&conn, 2, "mismatched-assignment");
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
assert!(
|
||||
sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.tracking
|
||||
.is_none()
|
||||
);
|
||||
let conn = legacy("no-snapshot", true, false, false);
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
assert!(
|
||||
sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.tracking
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migration_accepts_new_explicit_queue_after_ordered_case_judgments_but_not_after_a_later_case() {
|
||||
for queue_after_case in [true, false] {
|
||||
let conn = legacy("queued", true, false, false);
|
||||
for (ordinal, index, action, state) in [
|
||||
(
|
||||
0,
|
||||
2,
|
||||
TicketCaseAction::Close {
|
||||
reason: TicketClosureReason::Unknown,
|
||||
},
|
||||
"closed",
|
||||
),
|
||||
(1, 3, TicketCaseAction::Reopen, "open"),
|
||||
] {
|
||||
let key = format!("case-{ordinal}");
|
||||
let judgment = TicketCaseHistoryEntry {
|
||||
operation_key: key.clone(),
|
||||
previous_closure_operation_key: if ordinal == 1 {
|
||||
Some("case-0".into())
|
||||
} else {
|
||||
None
|
||||
},
|
||||
action,
|
||||
explanation: "Original case decision".into(),
|
||||
references: vec![],
|
||||
author: "human".into(),
|
||||
at: format!("case-time-{ordinal}"),
|
||||
};
|
||||
conn.execute("INSERT INTO typed_ticket_case_history VALUES('w','t',?1,?2,'original-fingerprint',?3)",params![ordinal,key,serde_json::to_string(&judgment).unwrap()]).unwrap();
|
||||
conn.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,state_field,to_state,at,body) VALUES('w','t',?1,'state_changed','case',?2,?3,'Original case decision')",params![index,state,judgment.at]).unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO typed_ticket_event_attributes VALUES('w','t',?1,'operation_key',?2)",
|
||||
params![index, key],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
if queue_after_case {
|
||||
routing_queue(&conn, 4, "a");
|
||||
}
|
||||
let original_history: i64 = conn
|
||||
.query_row("SELECT COUNT(*) FROM typed_ticket_case_history", [], |r| {
|
||||
r.get(0)
|
||||
})
|
||||
.unwrap();
|
||||
migrate_sqlite_ticket_schema(&conn).unwrap();
|
||||
let assignment = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(assignment.tracking.is_some(), queue_after_case);
|
||||
assert_eq!(
|
||||
conn.query_row("SELECT COUNT(*) FROM typed_ticket_case_history", [], |r| {
|
||||
r.get::<_, i64>(0)
|
||||
})
|
||||
.unwrap(),
|
||||
original_history
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standalone_assignment_constraints_reject_cross_ticket_principals_or_orphan_audit_and_operations()
|
||||
{
|
||||
let conn = db();
|
||||
assign(&conn, "a", "assigned-a");
|
||||
for sql in [
|
||||
"INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('other','t','cross-workspace','orchestrator','workspace_agent','workspace-orchestrator','author','now')",
|
||||
"INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','bad-orchestrator','orchestrator','user','user','author','now')",
|
||||
"INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,runtime_id,worker_id,assigned_by,assigned_at) VALUES('w','t','bad-owner','owner','worker','r','worker','author','now')",
|
||||
"INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','bad-worker','worker','workspace_agent','workspace-orchestrator','author','now')",
|
||||
"INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,assigned_by,assigned_at) VALUES('w','t','missing-worker-identity','worker','worker','author','now')",
|
||||
"INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,actor,created_at) VALUES('other','t','orchestrator','cross-audit','assigned','a','author','now')",
|
||||
"INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,actor,created_at) VALUES('w','missing','orchestrator','cross-ticket-audit','assigned','a','author','now')",
|
||||
"INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,created_at) VALUES('other','orphan-operation','assign','t','now')",
|
||||
"INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at,tracking_json) VALUES('w','t','bad-status','orchestrator','workspace_agent','workspace-orchestrator','author','now','{\"status\":\"running\"}')",
|
||||
] {
|
||||
assert!(conn.execute(sql, []).is_err(), "reject {sql}");
|
||||
}
|
||||
assert_eq!(
|
||||
conn.query_row("SELECT COUNT(*) FROM ticket_worker_assignments", [], |r| {
|
||||
r.get::<_, i64>(0)
|
||||
})
|
||||
.unwrap(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
conn.query_row(
|
||||
"SELECT COUNT(*) FROM ticket_worker_assignment_events",
|
||||
[],
|
||||
|r| r.get::<_, i64>(0)
|
||||
)
|
||||
.unwrap(),
|
||||
1
|
||||
);
|
||||
verify_sqlite_ticket_schema(&conn).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn schema_verification_does_not_normalize_away_case_sensitive_authority_literals() {
|
||||
let conn = db();
|
||||
let sql:String=conn.query_row("SELECT sql FROM sqlite_schema WHERE name='typed_ticket_assignment_orchestrator_principal_insert'",[],|r|r.get(0)).unwrap();
|
||||
conn.execute_batch("DROP TRIGGER typed_ticket_assignment_orchestrator_principal_insert")
|
||||
.unwrap();
|
||||
conn.execute_batch(&sql.replace("NEW.role='orchestrator'", "NEW.role='Orchestrator'"))
|
||||
.unwrap();
|
||||
assert!(
|
||||
verify_sqlite_ticket_schema(&conn)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("differs from canonical schema")
|
||||
);
|
||||
}
|
||||
@@ -20,9 +20,11 @@ use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest, Sha256};
|
||||
use thiserror::Error;
|
||||
|
||||
mod assignment_tracking;
|
||||
pub mod config;
|
||||
mod sqlite_schema;
|
||||
mod tags;
|
||||
pub use assignment_tracking::*;
|
||||
pub use tags::{
|
||||
TagListRequest, TicketTagSetRequest, WorkspaceTag, WorkspaceTagAction, WorkspaceTagMutation,
|
||||
};
|
||||
|
||||
@@ -7,7 +7,7 @@ use crate::{Result, TicketError, sqlite_err};
|
||||
|
||||
const MIGRATION_TABLE: &str = "ticket_schema_migrations";
|
||||
const MAX_SCHEMA_DIAGNOSTICS: usize = 32;
|
||||
const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 10;
|
||||
const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 11;
|
||||
const BASELINE_MIGRATION_NAME: &str = "ticket schema baseline";
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
@@ -307,10 +307,11 @@ pub fn migrate_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Composes retained component 6/7, tags-only 8, or cases-only 9 into component 10.
|
||||
/// The host owns the surrounding transaction (for example Workspace 90→91).
|
||||
/// Composes retained 6/7, tags-only 8 or cases-only 9 through component 10, then
|
||||
/// appends the 10→11 assignment migration without replacing the applied 10 marker.
|
||||
/// The host owns the surrounding transaction (for example Workspace 91→92).
|
||||
/// A savepoint rolls back this component on failure, without committing host work.
|
||||
/// Current component 10 is verified only; immutable judgments are never reapplied.
|
||||
/// Current component 11 is verified only; immutable judgments are never reapplied.
|
||||
pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> Result<()> {
|
||||
if connection.is_autocommit() {
|
||||
return Err(TicketError::Sqlite(
|
||||
@@ -334,22 +335,30 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
|
||||
create_latest_ticket_schema(connection)?;
|
||||
connection
|
||||
.execute(
|
||||
"INSERT INTO ticket_schema_migrations VALUES (10, ?1, ?2)",
|
||||
"INSERT INTO ticket_schema_migrations VALUES (11, ?1, ?2)",
|
||||
params![BASELINE_MIGRATION_NAME, chrono::Utc::now().to_rfc3339()],
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
} else {
|
||||
let version = match applied.iter().next() {
|
||||
Some((&version, name))
|
||||
if applied.len() == 1
|
||||
&& name == BASELINE_MIGRATION_NAME
|
||||
&& matches!(version, 6..=10) =>
|
||||
{
|
||||
version
|
||||
}
|
||||
_ => {
|
||||
validate_applied_migrations(&applied)?;
|
||||
unreachable!()
|
||||
let version = if applied
|
||||
== BTreeMap::from([
|
||||
(10, BASELINE_MIGRATION_NAME.to_owned()),
|
||||
(11, "Ticket-owned assignment tracking".to_owned()),
|
||||
]) {
|
||||
11
|
||||
} else {
|
||||
match applied.iter().next() {
|
||||
Some((&version, name))
|
||||
if applied.len() == 1
|
||||
&& name == BASELINE_MIGRATION_NAME
|
||||
&& matches!(version, 6..=11) =>
|
||||
{
|
||||
version
|
||||
}
|
||||
_ => {
|
||||
validate_applied_migrations(&applied)?;
|
||||
unreachable!()
|
||||
}
|
||||
}
|
||||
};
|
||||
if version == 6 {
|
||||
@@ -370,9 +379,9 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
|
||||
}
|
||||
TicketSchemaShape::Tags8 => migrate_legacy_cases(connection)?,
|
||||
TicketSchemaShape::Cases9 => add_tag_schema(connection)?,
|
||||
TicketSchemaShape::Composed10 => {}
|
||||
TicketSchemaShape::Composed10 | TicketSchemaShape::Tracking11 => {}
|
||||
}
|
||||
if version != 10 {
|
||||
if version < 10 {
|
||||
connection
|
||||
.execute(
|
||||
"UPDATE ticket_schema_migrations SET version = 10, applied_at = ?1",
|
||||
@@ -380,6 +389,10 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
}
|
||||
if version < 11 {
|
||||
crate::assignment_tracking::migrate_assignment_tracking(connection)?;
|
||||
connection.execute("INSERT INTO ticket_schema_migrations VALUES (11, 'Ticket-owned assignment tracking', ?1)", params![chrono::Utc::now().to_rfc3339()]).map_err(sqlite_err)?;
|
||||
}
|
||||
}
|
||||
verify_sqlite_ticket_schema(connection)
|
||||
})();
|
||||
@@ -402,6 +415,7 @@ enum TicketSchemaShape {
|
||||
Tags8,
|
||||
Cases9,
|
||||
Composed10,
|
||||
Tracking11,
|
||||
}
|
||||
impl TicketSchemaShape {
|
||||
fn from_version(version: i64) -> Self {
|
||||
@@ -410,6 +424,7 @@ impl TicketSchemaShape {
|
||||
8 => Self::Tags8,
|
||||
9 => Self::Cases9,
|
||||
10 => Self::Composed10,
|
||||
11 => Self::Tracking11,
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
@@ -419,19 +434,20 @@ impl TicketSchemaShape {
|
||||
Self::Tags8 => 8,
|
||||
Self::Cases9 => 9,
|
||||
Self::Composed10 => 10,
|
||||
Self::Tracking11 => 11,
|
||||
}
|
||||
}
|
||||
fn tags(self) -> bool {
|
||||
matches!(self, Self::Tags8 | Self::Composed10)
|
||||
matches!(self, Self::Tags8 | Self::Composed10 | Self::Tracking11)
|
||||
}
|
||||
fn cases(self) -> bool {
|
||||
matches!(self, Self::Cases9 | Self::Composed10)
|
||||
matches!(self, Self::Cases9 | Self::Composed10 | Self::Tracking11)
|
||||
}
|
||||
}
|
||||
|
||||
/// Verifies the current Ticket-owned SQLite schema without executing DDL.
|
||||
pub fn verify_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
|
||||
verify_ticket_schema_shape(connection, TicketSchemaShape::Composed10)
|
||||
verify_ticket_schema_shape(connection, TicketSchemaShape::Tracking11)
|
||||
}
|
||||
|
||||
/// Retained Workspace schema-66 migration produces the pre-classification v7
|
||||
@@ -469,7 +485,12 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
|
||||
Ok(applied) => {
|
||||
let history = if applied
|
||||
== BTreeMap::from([(shape.version(), BASELINE_MIGRATION_NAME.to_owned())])
|
||||
{
|
||||
|| (matches!(shape, TicketSchemaShape::Tracking11)
|
||||
&& applied
|
||||
== BTreeMap::from([
|
||||
(10, BASELINE_MIGRATION_NAME.to_owned()),
|
||||
(11, "Ticket-owned assignment tracking".to_owned()),
|
||||
])) {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(TicketError::Sqlite(format!(
|
||||
@@ -769,6 +790,13 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
|
||||
collect_foreign_key_check_diagnostics(connection, table, &mut diagnostics);
|
||||
}
|
||||
|
||||
if matches!(shape, TicketSchemaShape::Tracking11) {
|
||||
if let Err(error) =
|
||||
crate::assignment_tracking::verify_assignment_tracking_schema(connection)
|
||||
{
|
||||
push_diagnostic(&mut diagnostics, error.to_string());
|
||||
}
|
||||
}
|
||||
if diagnostics.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
@@ -788,7 +816,8 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
|
||||
fn create_latest_ticket_schema(connection: &Connection) -> Result<()> {
|
||||
connection
|
||||
.execute_batch(include_str!("latest_schema.sql"))
|
||||
.map_err(sqlite_err)
|
||||
.map_err(sqlite_err)?;
|
||||
crate::assignment_tracking::migrate_assignment_tracking(connection)
|
||||
}
|
||||
|
||||
/// Retained 7/8 case migration: leave tags and historical receipts untouched.
|
||||
@@ -1040,7 +1069,13 @@ fn validate_applied_migrations(applied: &BTreeMap<i64, String>) -> Result<()> {
|
||||
LATEST_SQLITE_TICKET_SCHEMA_VERSION,
|
||||
BASELINE_MIGRATION_NAME.to_string(),
|
||||
)]);
|
||||
if applied == &expected {
|
||||
if applied == &expected
|
||||
|| applied
|
||||
== &BTreeMap::from([
|
||||
(10, BASELINE_MIGRATION_NAME.to_owned()),
|
||||
(11, "Ticket-owned assignment tracking".to_owned()),
|
||||
])
|
||||
{
|
||||
Ok(())
|
||||
} else {
|
||||
Err(TicketError::Sqlite(format!(
|
||||
@@ -1541,7 +1576,10 @@ mod tests {
|
||||
assert!(!migrated_attributes.contains_key("ref_selector"));
|
||||
assert_eq!(
|
||||
load_applied_migrations(&connection).unwrap(),
|
||||
BTreeMap::from([(10, BASELINE_MIGRATION_NAME.to_owned())])
|
||||
BTreeMap::from([
|
||||
(10, BASELINE_MIGRATION_NAME.to_owned()),
|
||||
(11, "Ticket-owned assignment tracking".to_owned())
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1683,7 +1721,7 @@ mod tests {
|
||||
|
||||
let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
|
||||
assert!(error.to_string().contains(
|
||||
"migration history must contain only the canonical version 10 baseline marker"
|
||||
"migration history must contain only the canonical version 11 baseline marker"
|
||||
));
|
||||
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 2);
|
||||
}
|
||||
@@ -1705,7 +1743,7 @@ mod tests {
|
||||
|
||||
let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
|
||||
assert!(error.to_string().contains(
|
||||
"migration history must contain only the canonical version 10 baseline marker"
|
||||
"migration history must contain only the canonical version 11 baseline marker"
|
||||
));
|
||||
assert!(!table_exists(&connection, "typed_tickets").unwrap());
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user