fix: simplify Ticket reporting and speed up browser lists

Remove model-supplied assignment and retry identities from implementation reports; retain Host idempotency, authentication and notification evidence.

Add a SQL-only browser projection and decouple auxiliary page loads. Preserve authoritative MR evidence queries. Include regression tests and component benchmarks, documenting the remaining no-match search cost.
This commit is contained in:
2026-10-16 01:58:15 +09:00
parent 9f1cbbd385
commit b67aa2de7d
42 changed files with 3197 additions and 152 deletions
+1 -1
View File
@@ -29,7 +29,7 @@ server-api.workspace = true
project-record.workspace = true
reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "native-tls"] }
ring.workspace = true
rusqlite.workspace = true
rusqlite = { workspace = true, features = ["functions"] }
serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true
serde_yaml.workspace = true
@@ -471,7 +471,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
"Common notification routing and durable Worker delivery".into()
),
(96,"Backend Job common notification routing".into()),
(97, "MR notifications and authenticated implementation results".into())
(97, "MR notifications and authenticated implementation results".into()),
(98, "Assignment-independent implementation reports".into())
]
);
for (table, rows) in retained.iter().zip(original.iter()) {
+3
View File
@@ -50,6 +50,9 @@ use crate::store::{
};
use crate::{Error, Result};
#[path = "ticket_browse.rs"]
mod ticket_browse;
const DETAIL_BODY_LIMIT: usize = 64 * 1024;
const TICKET_EVENT_LIMIT: usize = 100;
const TICKET_EVENT_BODY_LIMIT: usize = 16 * 1024;
@@ -1,6 +1,18 @@
//! Implementation results are authenticated facts, not Orchestrator answers or handling updates.
//! The write lock fences the original implementation assignment and its pinned delegation.
//! Replays return the original result after reassignment, but cannot authorize a new result.
//! Reports belong to their authenticated Workspace Worker and Ticket, not an assignment.
//! Replays preserve the original result and notification recipient even after reassignment.
fn decode_result_request(raw: &str) -> Result<server_api::TicketImplementationResultRequest> {
// Retained pre-v98 requests keep their original bytes for audit. Assignment
// fields are historical metadata, not inputs or authorization for a new report.
let mut value: serde_json::Value =
serde_json::from_str(raw).map_err(|e| Error::Store(e.to_string()))?;
if let Some(object) = value.as_object_mut() {
object.remove("assignment_id");
object.remove("orchestrator_assignment_id");
}
serde_json::from_value(value).map_err(|e| Error::Store(e.to_string()))
}
use super::*;
impl SqliteWorkspaceStore {
@@ -13,7 +25,7 @@ impl SqliteWorkspaceStore {
self.with_conn(|conn| {
let saved: Option<(String,String)> = conn.query_row("SELECT request_json,created_at FROM ticket_implementation_results WHERE workspace_id=?1 AND ticket_id=?2 AND result_id=?3",params![workspace,ticket,result],|r| Ok((r.get(0)?,r.get(1)?))).optional()?;
let (request,created_at) = saved.ok_or_else(|| Error::InvalidInput("Implementation result not found in this Ticket".into()))?;
Ok(server_api::TicketImplementationResultReceipt { result_id: result.into(), ticket_id: ticket.into(), request: serde_json::from_str(&request).map_err(|e| Error::Store(e.to_string()))?, created_at })
Ok(server_api::TicketImplementationResultReceipt { result_id: result.into(), ticket_id: ticket.into(), request: decode_result_request(&request)?, created_at })
})
}
pub(crate) fn record_implementation_result(
@@ -24,14 +36,7 @@ impl SqliteWorkspaceStore {
worker: &RuntimeWorkerRef,
binding: &str,
) -> Result<server_api::TicketImplementationResultReceipt> {
for (name, value) in [
("operation_key", &request.operation_key),
("assignment_id", &request.assignment_id),
(
"orchestrator_assignment_id",
&request.orchestrator_assignment_id,
),
] {
for (name, value) in [("operation_key", &request.operation_key)] {
if value.trim().is_empty() || value.len() > 1024 || value.chars().any(char::is_control)
{
return Err(Error::InvalidInput(format!(
@@ -50,23 +55,19 @@ impl SqliteWorkspaceStore {
let encoded = serde_json::to_string(request).map_err(|e| Error::Store(e.to_string()))?;
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let authenticated: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM workspace_runtime_bindings WHERE workspace_id=?1 AND runtime_id=?2 AND binding_id=?3 AND revoked_at IS NULL)", params![workspace,worker.runtime_id,binding], |r| r.get(0))?;
let authenticated: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM workspace_runtime_bindings b JOIN worker_registry w ON w.workspace_id=b.workspace_id AND w.runtime_id=b.runtime_id WHERE b.workspace_id=?1 AND b.runtime_id=?2 AND b.binding_id=?3 AND b.revoked_at IS NULL AND w.worker_id=?4)", params![workspace,worker.runtime_id,binding,worker.worker_id], |r| r.get(0))?;
if !authenticated {
return Err(Error::WorkspacePermissionDenied("Implementation result requires its authenticated Runtime binding".into()));
}
let saved: Option<(String,String,String,String,String,String)> = tx.query_row("SELECT result_id,ticket_id,request_json,runtime_id,worker_id,created_at FROM ticket_implementation_results WHERE workspace_id=?1 AND operation_key=?2",params![workspace,request.operation_key], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?))).optional()?;
if let Some((result_id,saved_ticket,saved_request,runtime,source,created_at)) = saved {
if saved_ticket != ticket || saved_request != encoded || runtime != worker.runtime_id || source != worker.worker_id {
if saved_ticket != ticket || decode_result_request(&saved_request)? != *request || runtime != worker.runtime_id || source != worker.worker_id {
return Err(Error::TicketAssignmentConflict("Implementation result operation key was used for different input or source".into()));
}
return Ok(server_api::TicketImplementationResultReceipt { result_id, ticket_id: ticket.into(), request: request.clone(), created_at });
}
let owns: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_current_worker_assignments c JOIN ticket_implementation_delegations d ON d.workspace_id=c.workspace_id AND d.assignment_id=c.assignment_id WHERE c.workspace_id=?1 AND c.ticket_id=?2 AND c.role='worker' AND c.assignment_id=?3 AND c.principal_kind='worker' AND c.runtime_id=?4 AND c.worker_id=?5 AND d.orchestrator_assignment_id=?6 AND NOT EXISTS(SELECT 1 FROM ticket_assignment_work_releases r WHERE r.workspace_id=c.workspace_id AND r.assignment_id=c.assignment_id))",params![workspace,ticket,request.assignment_id,worker.runtime_id,worker.worker_id,request.orchestrator_assignment_id],|r|r.get(0))?;
if !owns {
return Err(Error::WorkspacePermissionDenied("Result source is not the current implementation Worker for this delegation; legacy unknown work requires explicit assignment confirmation".into()));
}
// A delayed new result is rejected rather than retargeted to a new Orchestrator.
notification_routing::require_current(&tx,workspace,ticket,&request.orchestrator_assignment_id,"orchestrator")?;
let exists: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM typed_tickets WHERE workspace_id=?1 AND ticket_id=?2)",params![workspace,ticket],|r|r.get(0))?;
if !exists { return Err(Error::InvalidInput("Result Ticket not found in this Workspace".into())); }
if let Some(mr) = &request.merge_request {
// References identify retained evidence, not a model's assertion that a branch
// is approved/current. Integration must resolve Git and reread the latest MR.
@@ -75,7 +76,7 @@ impl SqliteWorkspaceStore {
}
let result_id = uuid::Uuid::now_v7().to_string();
let created_at = chrono::Utc::now().to_rfc3339();
tx.execute("INSERT INTO ticket_implementation_results VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)",params![workspace,ticket,request.operation_key,result_id,request.assignment_id,request.orchestrator_assignment_id,worker.runtime_id,worker.worker_id,encoded,created_at])?;
tx.execute("INSERT INTO ticket_implementation_results(workspace_id,ticket_id,operation_key,result_id,runtime_id,worker_id,request_json,created_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",params![workspace,ticket,request.operation_key,result_id,worker.runtime_id,worker.worker_id,encoded,created_at])?;
tx.commit()?;
Ok(server_api::TicketImplementationResultReceipt { result_id, ticket_id: ticket.into(), request: request.clone(), created_at })
})
@@ -0,0 +1,21 @@
-- v98: result reports are Worker/Ticket facts, not assignment-completion operations.
-- Preserve legacy request bytes and already-captured notification evidence verbatim.
DROP TRIGGER notification_implementation_result;
DROP TRIGGER implementation_delegation_capture;
CREATE TABLE ticket_implementation_results_v98 (
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, operation_key TEXT NOT NULL,
result_id TEXT NOT NULL,
runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, request_json TEXT NOT NULL CHECK(json_valid(request_json)),
created_at TEXT NOT NULL,
PRIMARY KEY(workspace_id,operation_key), UNIQUE(workspace_id,result_id),
FOREIGN KEY(workspace_id,ticket_id) REFERENCES typed_tickets(workspace_id,ticket_id)
);
INSERT INTO ticket_implementation_results_v98
SELECT workspace_id,ticket_id,operation_key,result_id,runtime_id,worker_id,request_json,created_at
FROM ticket_implementation_results;
DROP TABLE ticket_implementation_results;
ALTER TABLE ticket_implementation_results_v98 RENAME TO ticket_implementation_results;
CREATE TRIGGER implementation_results_no_update BEFORE UPDATE ON ticket_implementation_results BEGIN SELECT RAISE(ABORT,'Implementation results are immutable'); END;
CREATE TRIGGER implementation_results_no_delete BEFORE DELETE ON ticket_implementation_results BEGIN SELECT RAISE(ABORT,'Implementation results are immutable'); END;
-- ticket_implementation_delegations remains read-only historical evidence only.
-- It is no longer populated or consulted by result reporting.
@@ -3,14 +3,198 @@
fn implementation_result_input() -> server_api::TicketImplementationResultRequest {
server_api::TicketImplementationResultRequest {
operation_key: "result-operation".into(),
assignment_id: "implementation".into(),
orchestrator_assignment_id: "delegation".into(),
kind: server_api::TicketImplementationResultKind::Result,
body: "Research complete; no repository was required".into(),
merge_request: None,
}
}
fn implementation_result_business_state(conn: &Connection) -> Vec<Vec<Vec<rusqlite::types::Value>>> {
[
"typed_tickets",
"typed_ticket_events",
"typed_ticket_event_attributes",
"ticket_worker_assignments",
"ticket_current_worker_assignments",
"ticket_worker_assignment_events",
"typed_ticket_assignment_tracking_history",
"ticket_assignment_work_releases",
"worker_control_grants",
"ticket_implementation_delegations",
]
.map(|table| fixture_query_rows(conn, &format!("SELECT * FROM {table} ORDER BY rowid")))
.into_iter()
.collect()
}
fn implementation_result_notification_evidence(
conn: &Connection,
) -> Vec<Vec<Vec<rusqlite::types::Value>>> {
[
"notification_sources",
"notification_decisions",
"notification_requests",
"notification_failures",
"notification_acknowledgements",
"notification_assignment_targets",
"ticket_assignment_deliveries",
"ticket_assignment_delivery_requests",
"ticket_assignment_delivery_acknowledgements",
]
.map(|table| stored_fixture_rows(conn, table))
.into_iter()
.collect()
}
#[tokio::test]
async fn implementation_result_without_any_assignment_is_authenticated_and_never_retargeted() {
let (dir, store) = tracking_store_fixture().await;
let worker = RuntimeWorkerRef::new("r", "replacement");
let input = implementation_result_input();
let before = store.with_conn(|conn| {
// Retention holds are not reporting permissions.
conn.execute("UPDATE worker_registry SET retention_state='pinned' WHERE workspace_id='w' AND worker_id='replacement'", [])?;
assert_eq!(current_schema_version(conn)?, 98);
assert_eq!(conn.query_row("SELECT count(*) FROM ticket_worker_assignments", [], |r| r.get::<_, i64>(0))?, 0);
assert_eq!(conn.query_row("SELECT count(*) FROM pragma_table_info('ticket_implementation_results') WHERE name IN ('assignment_id','orchestrator_assignment_id')", [], |r| r.get::<_, i64>(0))?, 0);
assert_eq!(conn.query_row("SELECT count(*) FROM sqlite_master WHERE type='trigger' AND name='implementation_delegation_capture'", [], |r| r.get::<_, i64>(0))?, 0);
Ok(implementation_result_business_state(conn))
}).unwrap();
let receipt = store
.record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
.unwrap();
assert_eq!(store.with_conn(|conn| Ok(implementation_result_business_state(conn))).unwrap(), before);
assert_eq!(store.implementation_result("w", "tracking", &receipt.result_id).unwrap(), receipt);
// A second, independently authenticated Worker cannot reuse this Workspace operation key.
let other = RuntimeWorkerRef::new("r", "worker");
assert!(store.record_implementation_result("w", "tracking", &input, &other, "tracking-binding").is_err());
let mut other_input = input.clone();
other_input.operation_key = "other-worker-result".into();
store.record_implementation_result("w", "tracking", &other_input, &other, "tracking-binding").unwrap();
seed_manual_start_ticket(&store, "other-ticket", "open");
assert!(store.record_implementation_result("w", "other-ticket", &input, &worker, "tracking-binding").is_err());
let mut missing = input.clone();
missing.operation_key = "missing-ticket-result".into();
assert!(store.record_implementation_result("w", "missing", &missing, &worker, "tracking-binding").is_err());
assert!(!store.list_pending_ticket_assignment_deliveries("w", 100).unwrap().iter()
.any(|p| p.input_event_id.starts_with("implementation-result:")));
let evidence = store.with_conn(|conn| {
let source: (String, String, String) = conn.query_row(
"SELECT s.assignment_id,d.disposition,d.reason FROM notification_sources s JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.input_event_id=?1",
[format!("implementation-result:{}", receipt.result_id)], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
assert_eq!(source, ("".into(), "suppress".into(), "no_recipient".into()));
Ok((
fixture_query_rows(conn, "SELECT * FROM notification_sources WHERE kind='implementation_result' ORDER BY sequence"),
fixture_query_rows(conn, "SELECT d.* FROM notification_decisions d JOIN notification_sources s ON s.sequence=d.source_sequence WHERE s.kind='implementation_result' ORDER BY s.sequence"),
))
}).unwrap();
// A later Orchestrator must not acquire previously recipient-less reports.
tracking_store_assign(&store, "late-delegation", "assign-late");
assert_eq!(store.record_implementation_result("w", "tracking", &input, &worker, "tracking-binding").unwrap(), receipt);
assert!(!store.list_pending_ticket_assignment_deliveries("w", 100).unwrap().iter()
.any(|p| p.input_event_id.starts_with("implementation-result:")));
store.with_conn(|conn| {
assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE kind='implementation_result'", [], |r| r.get::<_, i64>(0))?, 2);
// Assignment creation has its own evidence; old report evidence stays byte-for-byte intact.
assert_eq!(fixture_query_rows(conn, "SELECT * FROM notification_sources WHERE kind='implementation_result' ORDER BY sequence"), evidence.0);
assert_eq!(fixture_query_rows(conn, "SELECT d.* FROM notification_decisions d JOIN notification_sources s ON s.sequence=d.source_sequence WHERE s.kind='implementation_result' ORDER BY s.sequence"), evidence.1);
Ok(())
}).unwrap();
let reopened = SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db")).unwrap();
assert_eq!(reopened.record_implementation_result("w", "tracking", &input, &worker, "tracking-binding").unwrap(), receipt);
assert_eq!(reopened.implementation_result("w", "tracking", &receipt.result_id).unwrap(), receipt);
}
#[tokio::test]
async fn implementation_result_requires_live_binding_and_registered_worker_even_for_replay() {
for change in ["revoked", "binding-changed", "unregistered", "foreign-worker", "foreign-binding"] {
let (_dir, store) = tracking_store_fixture().await;
store.upsert_workspace(&WorkspaceRecord {
workspace_id: "other".into(), owner_account_id: "owner-account".into(),
display_name: "Other".into(), state: "active".into(),
created_at: "1".into(), updated_at: "1".into(),
}).await.unwrap();
let worker = RuntimeWorkerRef::new("r", "replacement");
let input = implementation_result_input();
store.record_implementation_result("w", "tracking", &input, &worker, "tracking-binding").unwrap();
store.with_conn(|conn| {
match change {
"revoked" => { conn.execute("UPDATE workspace_runtime_bindings SET revoked_at='2' WHERE workspace_id='w'", [])?; }
"binding-changed" => { conn.execute("UPDATE workspace_runtime_bindings SET binding_id='new-binding' WHERE workspace_id='w'", [])?; }
"unregistered" => { conn.execute("DELETE FROM worker_registry WHERE workspace_id='w' AND worker_id='replacement'", [])?; }
"foreign-worker" => { conn.execute("UPDATE worker_registry SET workspace_id='other' WHERE workspace_id='w' AND worker_id='replacement'", [])?; }
"foreign-binding" => { conn.execute("UPDATE workspace_runtime_bindings SET workspace_id='other' WHERE workspace_id='w'", [])?; }
_ => unreachable!(),
}
Ok(())
}).unwrap();
let mut fresh = input.clone();
fresh.operation_key = "fresh-after-auth-change".into();
for request in [&input, &fresh] {
assert!(matches!(store.record_implementation_result("w", "tracking", request, &worker, "tracking-binding"),
Err(Error::WorkspacePermissionDenied(_))), "{change}: {request:?}");
}
store.with_conn(|conn| {
assert_eq!(conn.query_row("SELECT count(*) FROM ticket_implementation_results", [], |r| r.get::<_, i64>(0))?, 1);
assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE kind='implementation_result'", [], |r| r.get::<_, i64>(0))?, 1);
Ok(())
}).unwrap();
}
}
#[tokio::test]
async fn implementation_result_late_reports_do_not_change_ticket_assignment_or_handling_state() {
for situation in ["never-assigned", "released", "reassigned", "closed", "no-orchestrator", "assigned-before-orchestrator"] {
let (_dir, store) = tracking_store_fixture().await;
if situation != "assigned-before-orchestrator" {
tracking_store_assign(&store, "delegation", "assign");
}
let worker = if situation == "never-assigned" {
RuntimeWorkerRef::new("r", "replacement")
} else {
routing_worker_assign(&store, "implementation").worker
};
match situation {
"never-assigned" => {}
"released" => {
assert!(store.clear_current_ticket_role_assignment("w", "tracking", TicketAssignmentRole::Worker,
"implementation", "release-event", "release-op", "human", "3", Some("Released")).unwrap());
}
"reassigned" => {
store.set_current_ticket_role_assignment(&manual_worker_assignment("tracking", "replacement-implementation"),
Some("implementation"), "reassign-event", "reassign-op", true).unwrap();
}
"closed" => {
store.with_conn(|conn| {
conn.execute("UPDATE typed_tickets SET workflow_state='closed' WHERE workspace_id='w' AND ticket_id='tracking'", [])?;
Ok(())
}).unwrap();
}
"no-orchestrator" => {
assert!(store.clear_current_ticket_role_assignment("w", "tracking", TicketAssignmentRole::Orchestrator,
"delegation", "release-orchestrator-event", "release-orchestrator-op", "human", "3", Some("Returned to human")).unwrap());
}
"assigned-before-orchestrator" => { tracking_store_assign(&store, "late-delegation", "assign-late"); }
_ => unreachable!(),
}
let before = store.with_conn(|conn| Ok(implementation_result_business_state(conn))).unwrap();
let input = implementation_result_input();
let receipt = store.record_implementation_result("w", "tracking", &input, &worker, "tracking-binding").unwrap();
assert_eq!(store.record_implementation_result("w", "tracking", &input, &worker, "tracking-binding").unwrap(), receipt);
store.with_conn(|conn| {
assert_eq!(implementation_result_business_state(conn), before, "{situation}");
let expected = match situation {
"no-orchestrator" => "",
"assigned-before-orchestrator" => "late-delegation",
_ => "delegation",
};
assert_eq!(conn.query_row("SELECT assignment_id FROM notification_sources WHERE kind='implementation_result'", [], |r| r.get::<_, String>(0))?, expected);
assert_eq!(conn.query_row("SELECT count(*) FROM ticket_implementation_delegations", [], |r| r.get::<_, i64>(0))?, 0, "new assignments must not create historical delegation evidence");
Ok(())
}).unwrap();
}
}
#[tokio::test]
async fn implementation_result_replay_is_atomic_and_cannot_be_retargeted() {
let (_dir, store) = tracking_store_fixture().await;
@@ -53,6 +237,11 @@ async fn implementation_result_replay_is_atomic_and_cannot_be_retargeted() {
.record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
.is_err()
);
let mut changed_kind = input.clone();
changed_kind.kind = server_api::TicketImplementationResultKind::Blocker;
assert!(store
.record_implementation_result("w", "tracking", &changed_kind, &worker, "tracking-binding")
.is_err());
for (workspace, ticket, source, binding) in [
("other", "tracking", worker.clone(), "tracking-binding"),
("w", "other", worker.clone(), "tracking-binding"),
@@ -107,25 +296,37 @@ async fn implementation_result_replay_is_atomic_and_cannot_be_retargeted() {
.unwrap(),
receipt
);
changed.operation_key = "late-result".into();
assert!(
store
.record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
.is_err()
);
changed.orchestrator_assignment_id = "replacement-delegation".into();
assert!(
store
.record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
.is_err()
);
assert!(
!store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.iter()
.any(|p| p.input_event_id.starts_with("implementation-result:"))
.any(|p| p.input_event_id == format!("implementation-result:{}", receipt.result_id))
);
changed.operation_key = "late-result".into();
let late = store
.record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
.unwrap();
assert_ne!(late.result_id, receipt.result_id);
store.with_conn(|conn| {
for (result, assignment) in [
(&receipt.result_id, "delegation"),
(&late.result_id, "replacement-delegation"),
] {
assert_eq!(conn.query_row(
"SELECT assignment_id FROM notification_sources WHERE input_event_id=?1 AND role='orchestrator'",
[format!("implementation-result:{result}")], |r| r.get::<_, String>(0))?, assignment);
assert_eq!(conn.query_row(
"SELECT count(*) FROM notification_sources WHERE input_event_id=?1",
[format!("implementation-result:{result}")], |r| r.get::<_, i64>(0))?, 1);
}
Ok(())
}).unwrap();
assert!(store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap()
.iter()
.any(|p| p.input_event_id == format!("implementation-result:{}", late.result_id)));
}
#[tokio::test]
@@ -166,6 +367,137 @@ async fn implementation_result_rejects_forged_evidence_and_rolls_back_without_no
.unwrap();
}
#[tokio::test]
async fn implementation_result_v97_to_v98_preserves_legacy_bytes_receipts_and_notification_evidence() {
let (dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "delegation", "assign");
let worker = routing_worker_assign(&store, "implementation").worker;
// Reconstruct schema97 from a fresh schema98 fixture, including its historical capture.
store.with_conn(|conn| {
conn.execute_batch("DROP TRIGGER notification_mr_event;
DROP TRIGGER notification_mr_source;
DROP TRIGGER notification_implementation_result;
DROP TRIGGER IF EXISTS implementation_delegation_capture;
DROP TABLE ticket_implementation_results;
DROP TABLE ticket_implementation_delegations;
DROP TABLE merge_request_source_observations;
UPDATE __yoi_schema_migrations SET version=96 WHERE version=98;")?;
migrate_business_notifications_v96_to_v97(conn)?;
assert_eq!(current_schema_version(conn)?, 97);
Ok(())
}).unwrap();
let input = implementation_result_input();
let mut second = input.clone();
second.operation_key = "legacy-second-operation".into();
second.body = "A second retained report".into();
let requests = [input.clone(), second];
let raw_requests = requests.iter().map(|request| {
let mut legacy = serde_json::to_value(request).unwrap();
legacy["assignment_id"] = serde_json::json!("implementation");
legacy["orchestrator_assignment_id"] = serde_json::json!("delegation");
format!(" \n{}\n ", serde_json::to_string_pretty(&legacy).unwrap())
}).collect::<Vec<_>>();
store.with_conn(|conn| {
for (index, request) in requests.iter().enumerate() {
conn.execute("INSERT INTO ticket_implementation_results(workspace_id,ticket_id,operation_key,result_id,assignment_id,orchestrator_assignment_id,runtime_id,worker_id,request_json,created_at)
VALUES('w','tracking',?1,?2,'implementation','delegation','r','replacement',?3,'4')",
params![request.operation_key, format!("legacy-result-{index}"), raw_requests[index]])?;
}
Ok(())
}).unwrap();
// Populate persisted request, target, ACK and failure evidence, not just source rows.
let pending = store.list_pending_ticket_assignment_deliveries("w", 100).unwrap();
for index in 0..requests.len() {
let input_event = format!("implementation-result:legacy-result-{index}");
assert!(pending.iter().any(|p| p.input_event_id == input_event));
let reserved = store.reserve_ticket_assignment_delivery("w", "delegation", &input_event, &tracking_store_source()).unwrap();
store.prepare_ticket_assignment_delivery_content("w", &reserved.delivery_id, "Retained legacy report notification", &tracking_store_source()).unwrap();
store.ack_ticket_assignment_delivery("w", &reserved.delivery_id, &tracking_store_source()).unwrap();
store.with_conn(|conn| {
conn.execute("INSERT INTO notification_failures(workspace_id,delivery_id,code,recorded_at) VALUES('w',?1,'legacy-failure','4')", [&reserved.delivery_id])?;
Ok(())
}).unwrap();
}
routing_ack_worker_pending(&store, &TicketWorkerNotificationTarget {
worker: worker.clone(),
authentication_binding_id: "tracking-binding".into(),
});
let (results, business, evidence) = store.with_conn(|conn| {
Ok((
fixture_query_rows(conn, "SELECT workspace_id,ticket_id,operation_key,result_id,runtime_id,worker_id,request_json,created_at FROM ticket_implementation_results ORDER BY rowid"),
implementation_result_business_state(conn),
implementation_result_notification_evidence(conn),
))
}).unwrap();
assert!(evidence.iter().all(|rows| !rows.is_empty()), "fixture must retain nonempty source, decision, request, target, ACK and failure evidence");
store.with_conn(|conn| {
// The migration must roll back its trigger drops and leave every byte replayable.
conn.execute_batch("CREATE TABLE ticket_implementation_results_v98(collision INTEGER);")?;
assert!(migrate_result_reports_v97_to_v98(conn).is_err());
assert_eq!(current_schema_version(conn)?, 97);
assert_eq!(conn.query_row("SELECT count(*) FROM sqlite_master WHERE type='trigger' AND name IN ('notification_implementation_result','implementation_delegation_capture')", [], |r| r.get::<_, i64>(0))?, 2);
assert_eq!(conn.query_row("SELECT count(*) FROM pragma_table_info('ticket_implementation_results') WHERE name IN ('assignment_id','orchestrator_assignment_id')", [], |r| r.get::<_, i64>(0))?, 2);
assert_eq!(fixture_query_rows(conn, "SELECT workspace_id,ticket_id,operation_key,result_id,runtime_id,worker_id,request_json,created_at FROM ticket_implementation_results ORDER BY rowid"), results);
assert_eq!(implementation_result_business_state(conn), business);
assert_eq!(implementation_result_notification_evidence(conn), evidence);
conn.execute_batch("DROP TABLE ticket_implementation_results_v98;")?;
migrate_result_reports_v97_to_v98(conn)?;
assert_eq!(current_schema_version(conn)?, 98);
assert!(migrate_result_reports_v97_to_v98(conn).is_err(), "migration rejects the wrong starting version");
assert_eq!(fixture_query_rows(conn, "SELECT * FROM ticket_implementation_results ORDER BY rowid"), results);
assert_eq!(implementation_result_business_state(conn), business);
assert_eq!(implementation_result_notification_evidence(conn), evidence);
assert_eq!(conn.query_row("SELECT count(*) FROM pragma_table_info('ticket_implementation_results') WHERE name IN ('assignment_id','orchestrator_assignment_id')", [], |r| r.get::<_, i64>(0))?, 0);
assert_eq!(conn.query_row("SELECT count(*) FROM sqlite_master WHERE type='trigger' AND name='implementation_delegation_capture'", [], |r| r.get::<_, i64>(0))?, 0);
assert!(conn.execute("UPDATE ticket_implementation_results SET created_at='changed'", []).is_err());
assert!(conn.execute("DELETE FROM ticket_implementation_results", []).is_err());
assert!(conn.execute("UPDATE ticket_implementation_delegations SET orchestrator_assignment_id='delegation'", []).is_err());
assert!(conn.execute("DELETE FROM ticket_implementation_delegations", []).is_err());
assert!(conn.execute("INSERT INTO ticket_implementation_results SELECT workspace_id,'missing','missing-key','missing-result',runtime_id,worker_id,request_json,created_at FROM ticket_implementation_results LIMIT 1", []).is_err(), "Ticket foreign key remains enforced");
assert!(conn.execute("INSERT INTO ticket_implementation_results SELECT workspace_id,ticket_id,'duplicate-result-key',result_id,runtime_id,worker_id,request_json,created_at FROM ticket_implementation_results LIMIT 1", []).is_err(), "result identity remains unique");
assert!(conn.execute("INSERT INTO ticket_implementation_results SELECT workspace_id,ticket_id,operation_key,'duplicate-key-result',runtime_id,worker_id,request_json,created_at FROM ticket_implementation_results LIMIT 1", []).is_err(), "operation key remains unique");
assert_eq!(conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |r| r.get::<_, i64>(0))?, 0);
Ok(())
}).unwrap();
let reopened = SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db")).unwrap();
for (index, request) in requests.iter().enumerate() {
let expected = server_api::TicketImplementationResultReceipt {
result_id: format!("legacy-result-{index}"), ticket_id: "tracking".into(),
request: request.clone(), created_at: "4".into(),
};
assert_eq!(reopened.implementation_result("w", "tracking", &expected.result_id).unwrap(), expected);
assert_eq!(reopened.record_implementation_result("w", "tracking", request, &worker, "tracking-binding").unwrap(), expected);
}
reopened.with_conn(|conn| {
assert_eq!(fixture_query_rows(conn, "SELECT * FROM ticket_implementation_results ORDER BY rowid"), results);
assert_eq!(implementation_result_notification_evidence(conn), evidence, "reading and replaying legacy JSON must never recapture notifications");
Ok(())
}).unwrap();
tracking_store_assign(&reopened, "new-delegation", "assign-new");
routing_worker_assign(&reopened, "new-implementation");
reopened.with_conn(|conn| {
assert_eq!(stored_fixture_rows(conn, "ticket_implementation_delegations"), business[9], "historical delegation evidence must not grow");
for index in 0..requests.len() {
assert_eq!(conn.query_row("SELECT assignment_id FROM notification_sources WHERE input_event_id=?1", [format!("implementation-result:legacy-result-{index}")], |r| r.get::<_, String>(0))?, "delegation");
}
Ok(())
}).unwrap();
for request in &requests {
reopened.record_implementation_result("w", "tracking", request, &worker, "tracking-binding").unwrap();
}
let mut fresh = input;
fresh.operation_key = "new-schema98-report".into();
let receipt = reopened.record_implementation_result("w", "tracking", &fresh, &worker, "tracking-binding").unwrap();
reopened.with_conn(|conn| {
assert_eq!(conn.query_row("SELECT assignment_id FROM notification_sources WHERE input_event_id=?1", [format!("implementation-result:{}", receipt.result_id)], |r| r.get::<_, String>(0))?, "new-delegation");
for index in 0..requests.len() {
assert_eq!(conn.query_row("SELECT request_json FROM ticket_implementation_results WHERE result_id=?1", [format!("legacy-result-{index}")], |r| r.get::<_, String>(0))?, raw_requests[index]);
assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE input_event_id=?1", [format!("implementation-result:legacy-result-{index}")], |r| r.get::<_, i64>(0))?, 1);
}
Ok(())
}).unwrap();
}
#[tokio::test]
async fn mr_notification_upgrade_recovers_events_not_comments_and_retains_unknown_legacy_results() {
let (_dir, store) = tracking_store_fixture().await;
@@ -175,9 +507,9 @@ async fn mr_notification_upgrade_recovers_events_not_comments_and_retains_unknow
store.with_conn(|conn| {
conn.execute("INSERT INTO ticket_assignment_operation_fences VALUES('w','operation-known-implementation','delegation','r','worker','workspace-orchestrator','tracking-binding',?1)",[serde_json::to_string(&tracking_store_source()).unwrap()])?;
// Reconstruct the prior composition before its new business capture existed.
conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER implementation_delegation_capture;
conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER IF EXISTS implementation_delegation_capture;
DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations;
UPDATE __yoi_schema_migrations SET version=96 WHERE version=97;")?;
UPDATE __yoi_schema_migrations SET version=96 WHERE version=98;")?;
conn.execute_batch("INSERT INTO merge_requests VALUES('w','historical-mr','main','open','refs/heads/work','refs/heads/main','3','3');
INSERT INTO merge_request_ticket_relations VALUES('w','historical-mr','tracking','implements','3');
INSERT INTO merge_request_thread_events VALUES('w','historical-mr','approved-event',1,'review','{\"subject_ref\":\"source-a\",\"decision\":\"approve\"}',NULL,'3');
@@ -233,6 +565,9 @@ async fn implementation_result_notification_failure_rolls_back_and_concurrent_re
tracking_store_assign(&store, "delegation", "assign");
let worker = routing_worker_assign(&store, "implementation").worker;
let input = implementation_result_input();
let before = store
.with_conn(|conn| Ok(implementation_result_business_state(conn)))
.unwrap();
store.with_conn(|conn| {
conn.execute_batch("CREATE TRIGGER reject_result_notification BEFORE INSERT ON notification_sources WHEN NEW.kind='implementation_result' BEGIN SELECT RAISE(ABORT,'injected capture failure'); END;")?;
Ok(())
@@ -252,6 +587,8 @@ async fn implementation_result_notification_failure_rolls_back_and_concurrent_re
)?,
0
);
assert_eq!(implementation_result_business_state(conn), before);
assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE kind='implementation_result'", [], |r| r.get::<_, i64>(0))?, 0);
conn.execute_batch("DROP TRIGGER reject_result_notification;")?;
Ok(())
})
@@ -324,9 +661,9 @@ async fn mr_event_actors_preserve_self_suppression_in_capture_and_backfill() {
];
store.with_conn(|conn| {
if historical {
conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER implementation_delegation_capture;
conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER IF EXISTS implementation_delegation_capture;
DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations;
UPDATE __yoi_schema_migrations SET version=96 WHERE version=97;")?;
UPDATE __yoi_schema_migrations SET version=96 WHERE version=98;")?;
}
conn.execute_batch("INSERT INTO merge_requests VALUES('w','actor-mr','main','open','work','main','3','3');
INSERT INTO merge_request_ticket_relations VALUES('w','actor-mr','tracking','implements','3');")?;
+2
View File
@@ -26,6 +26,8 @@ pub mod server;
pub mod skills;
pub mod store;
pub mod subjektiv;
#[cfg(test)]
mod ticket_browse_performance_tests;
mod ticket_item_checker;
pub mod workdir_create_operations;
mod workdir_removal;
@@ -375,6 +375,23 @@ pub(super) fn install_business_events(conn: &Connection) -> Result<()> {
Ok(())
}
/// New reports notify the delegation present when the report is saved. Assignment
/// identity only fences this saved notification against later retargeting; it is
/// neither a caller input nor permission to publish a report.
pub(super) fn upgrade_result_reports(conn: &Connection) -> Result<()> {
conn.execute_batch(include_str!("implementation_result_reports.sql"))?;
conn.execute_batch(&format!("CREATE TRIGGER notification_implementation_result AFTER INSERT ON ticket_implementation_results BEGIN
INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,actor,event_index,facts_json)
SELECT NEW.workspace_id,NEW.ticket_id,'implementation-result:' || NEW.result_id,'orchestrator',COALESCE(a.assignment_id,''),'implementation_result','worker:' || NEW.runtime_id || '/' || NEW.worker_id,NULL,
json_set({FACTS_SQL},'$.result_id',NEW.result_id,'$.merge_request_id',json_extract(NEW.request_json,'$.merge_request.merge_request_id'))
FROM typed_tickets t
LEFT JOIN ticket_current_worker_assignments c ON c.workspace_id=t.workspace_id AND c.ticket_id=t.ticket_id AND c.role='orchestrator'
LEFT JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.assignment_id=c.assignment_id
WHERE t.workspace_id=NEW.workspace_id AND t.ticket_id=NEW.ticket_id;
END;"))?;
Ok(())
}
impl SqliteWorkspaceStore {
/// Record a provider-resolved source observation, not approval. Git has no local
/// transactional event log: transitions are captured when the existing MR paths
@@ -438,12 +455,14 @@ pub(super) fn verify(conn: &Connection) -> Result<()> {
] {
names.push(name.into());
}
if current_schema_version(conn)? >= 96 {
if current_schema_version(conn)? == 97 {
names.push("implementation_delegation_capture".into());
}
if current_schema_version(conn)? >= 97 {
names.extend(
[
"ticket_implementation_delegations",
"ticket_implementation_results",
"implementation_delegation_capture",
"implementation_results_no_update",
"implementation_results_no_delete",
"implementation_delegations_no_update",
@@ -845,6 +845,7 @@ async fn notification_routing_fresh_and_upgraded_schema_and_guards_match() {
migrate_notification_routing_v94_to_v95(&conn).unwrap();
migrate_job_notifications_v95_to_v96(&conn).unwrap();
migrate_business_notifications_v96_to_v97(&conn).unwrap();
migrate_result_reports_v97_to_v98(&conn).unwrap();
let snapshot = |conn: &Connection| -> Vec<(String, String)> {
conn.prepare("SELECT name,sql FROM sqlite_schema WHERE name LIKE 'notification_%' OR name LIKE 'ticket_assignment_content%' OR name IN ('ticket_assignment_title_input','ticket_assignment_initial_post_input') ORDER BY name").unwrap().query_map([],|r|Ok((r.get(0)?,r.get(1)?))).unwrap().collect::<rusqlite::Result<_>>().unwrap()
};
+24
View File
@@ -6671,6 +6671,7 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::ticket_create_record(
service.clone(),
))
.merge(server_api::server_api_axum::ticket_browse(service.clone()))
.merge(server_api::server_api_axum::ticket_query(service.clone()))
.merge(server_api::server_api_axum::ticket_get_alias(
service.clone(),
@@ -10148,6 +10149,25 @@ impl server_api::ServerApi for ServerApiContractService {
Ok(server_api::TicketRecordRef(response))
}
async fn ticket_browse(
&self,
workspace_id: String,
request: server_api::TicketBrowseQueryRequest,
) -> std::result::Result<server_api::TicketBrowseQueryResponse, server_api::RepositoryApiError>
{
let api = self.workspace_api()?.clone();
api.store
.db_executor()
.run(move || {
validate_workspace_scope(&api, &workspace_id)?;
api.authority
.query_ticket_browse(request)
.map_err(ApiError::from)
})
.await
.map_err(ApiError::into_repository_api_error)
}
async fn ticket_query(
&self,
workspace_id: String,
@@ -35868,6 +35888,10 @@ mod tests {
include!("server_ticket_tags_tests.rs");
}
mod ticket_browse_tests {
include!("server_ticket_browse_tests.rs");
}
mod api_logging_tests;
mod auth_logging_tests;
mod drive_api_tests;
@@ -18,6 +18,48 @@ const EVIDENCE_MR: &str = "ticket-evidence-mr";
const EVIDENCE_SOURCE: &str = "immutable-approved-source";
const EVIDENCE_TARGET: &str = "immutable-target-result";
#[tokio::test]
async fn ticket_browse_does_not_observe_sources_but_authoritative_query_and_show_still_do() {
struct CountingObserver(Arc<std::sync::atomic::AtomicUsize>);
impl crate::authority::TicketSourceCommitResolver for CountingObserver {
fn resolve_source_commit(
&self,
_ticket: &str,
_repository: &str,
_selector: &str,
) -> std::result::Result<String, MergeRequestRefDiagnostic> {
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(EVIDENCE_SOURCE.into())
}
}
let mut fixture = EvidenceApiFixture::new().await;
let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
fixture.api.authority = fixture
.api
.authority
.clone()
.with_source_commit_resolver(Arc::new(CountingObserver(calls.clone())));
let response = request_json(
build_inner_router(fixture.api.clone()),
"POST",
&format!("/api/w/{TEST_WORKSPACE_ID}/tickets/browse"),
Some(json!({"limit":30})),
StatusCode::OK,
)
.await;
assert_eq!(response["items"][0]["id"], fixture.ticket_id);
assert!(response["items"][0].get("evidence").is_none());
assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 0);
fixture
.api
.authority
.query_tickets(Default::default())
.unwrap();
assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
fixture.api.authority.ticket(&fixture.ticket_id).unwrap();
assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2);
}
// Deliberately freeze review dates: freshness must follow the exact item
// content digest and source/result snapshot, not a comparison of wall-clock dates.
fn evidence_time() -> chrono::DateTime<Utc> {
@@ -1673,7 +1673,7 @@ async fn mr_open_retains_source_before_first_review_and_observed_result() {
"/workspaces/{TEST_WORKSPACE_ID}/tickets/{}/implementation-results",
ticket.id
);
let mut result = json!({"operation_key":"observed-result","assignment_id":"opening-coder","orchestrator_assignment_id":format!("orchestrator-{}",ticket.id),"kind":"handoff","body":"Published new source; review not requested yet.","merge_request":{"merge_request_id":mr,"subject_ref":"forged-source"}});
let mut result = json!({"operation_key":"observed-result","kind":"handoff","body":"Published new source; review not requested yet.","merge_request":{"merge_request_id":mr,"subject_ref":"forged-source"}});
call(
"POST",
result_path.clone(),
@@ -43,7 +43,7 @@ async fn assert_result_diagnostics(api: &WorkspaceApi, recipient: &RuntimeWorker
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn implementation_result_signed_http_reaches_persistent_orchestrator_once_without_mr() {
async fn implementation_result_signed_http_reaches_persistent_orchestrator_without_assignment_or_mr() {
let dir = tempfile::tempdir().unwrap();
let (mut api, execution, _) = delivery_real_fixture(dir.path(), false).await;
let identity = RuntimeIdentityMaterial::generate(EMBEDDED_WORKER_RUNTIME_ID).unwrap();
@@ -66,22 +66,14 @@ async fn implementation_result_signed_http_reaches_persistent_orchestrator_once_
.await;
let coder = spawn_ticket_check_source(&api, "research-coder");
seed_worker_source_member(&api, &coder.runtime_id, &coder.worker_id);
delivery_assign_worker(&api, &ticket.id, &coder, "research-implementation", None);
// Reporting does not require an implementation assignment.
recover_ticket_notifications(&api);
let before = delivery_tracking(&api, &ticket.id, delegation);
let input = json!({"operation_key":"research-result","assignment_id":"research-implementation","orchestrator_assignment_id":delegation,"kind":"result","body":"Documented findings; no repository or MR was required.","merge_request":null});
// The registered Orchestrator is not the implementation assignee and cannot forge this result.
result_signed_http(
&api,
&identity,
&orchestrator,
&ticket.id,
&input,
StatusCode::FORBIDDEN,
)
.await;
let input = json!({"operation_key":"research-result","kind":"result","body":"Documented findings; no repository or MR was required.","merge_request":null});
let receipt =
result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
// Another authenticated Worker may report, but cannot replay this Workers receipt.
result_signed_http(&api, &identity, &orchestrator, &ticket.id, &input, StatusCode::CONFLICT).await;
delivery_wait_idle(&api, &orchestrator, 2).await;
assert_eq!(
result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await,
@@ -239,7 +231,7 @@ async fn mr_result_recovery_and_handoff_are_distinct_persistent_inputs() {
assert_eq!(pending["source_checks"][0]["captured"], true);
recover_ticket_notifications(&api);
delivery_wait_idle(&api, &orchestrator, 3).await;
let input = json!({"operation_key":"approved-handoff","assignment_id":"mr-implementation","orchestrator_assignment_id":delegation,"kind":"handoff","body":"Implementation and validation complete; integration remains separate.","merge_request":{"merge_request_id":"result-mr","subject_ref":"source-a"}});
let input = json!({"operation_key":"approved-handoff","kind":"handoff","body":"Implementation and validation complete; integration remains separate.","merge_request":{"merge_request_id":"result-mr","subject_ref":"source-a"}});
let receipt =
result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
delivery_wait_idle(&api, &orchestrator, 4).await;
@@ -0,0 +1,72 @@
use super::*;
#[tokio::test]
async fn ticket_browse_http_projects_only_list_fields_and_rejects_evidence_queries() {
let temp = tempfile::tempdir().unwrap();
let api = test_api(temp.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let created = backend
.create(ticket::NewTicket::new("Lightweight browse HTTP"))
.unwrap();
let path = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/browse");
request_json(
build_router(api.clone()),
"POST",
&path,
Some(json!({})),
StatusCode::UNAUTHORIZED,
)
.await;
let app = build_inner_router(api);
let response = request_json(
app.clone(),
"POST",
&path,
Some(json!({"states":["open"],"query":"Lightweight browse HTTP","limit":30})),
StatusCode::OK,
)
.await;
let parsed: server_api::TicketBrowseQueryResponse =
serde_json::from_value(response.clone()).unwrap();
assert_eq!(parsed.items.len(), 1);
assert_eq!(parsed.items[0].id, created.id);
assert_eq!(parsed.record_authority, "workspace-sqlite");
for absent in [
"evidence",
"merge_requests",
"body",
"events",
"content_digest",
] {
assert!(response["items"][0].get(absent).is_none(), "{absent}");
}
// Browse cannot silently weaken an evidence query into a metadata filter.
request_json(
app.clone(),
"POST",
&path,
Some(json!({"review_status":"approved"})),
StatusCode::UNPROCESSABLE_ENTITY,
)
.await;
request_json(
app.clone(),
"POST",
&path,
Some(json!({"sort":"relevance"})),
StatusCode::BAD_REQUEST,
)
.await;
let foreign = app
.oneshot(
Request::builder()
.method("POST")
.uri("/api/w/other-workspace/tickets/browse")
.header("content-type", "application/json")
.body(Body::from("{}"))
.unwrap(),
)
.await
.unwrap();
assert!(!foreign.status().is_success());
}
+37 -2
View File
@@ -84,7 +84,7 @@ pub use ticket::{
};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 97;
const LATEST_SCHEMA_VERSION: i64 = 98;
const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
@@ -377,6 +377,11 @@ const MIGRATIONS: &[Migration] = &[
name: "MR notifications and authenticated implementation results",
apply: migrate_business_notifications_v96_to_v97,
},
Migration {
version: 98,
name: "Assignment-independent implementation reports",
apply: migrate_result_reports_v97_to_v98,
},
];
#[derive(Clone, Copy)]
@@ -14286,6 +14291,32 @@ fn migrate_business_notifications_v96_to_v97(conn: &Connection) -> Result<()> {
Ok(())
}
fn migrate_result_reports_v97_to_v98(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 97 {
return Err(Error::Store(
"Result reports migration requires schema 97".into(),
));
}
let tx = conn.unchecked_transaction()?;
notification_routing::upgrade_result_reports(&tx)?;
tx.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(98,?1)",
["Assignment-independent implementation reports"],
)?;
verify_workspace_resource_constraints(&tx)?;
let violations: i64 =
tx.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |r| {
r.get(0)
})?;
if violations != 0 {
return Err(Error::Store(
"Result reports migration foreign-key violation".into(),
));
}
tx.commit()?;
Ok(())
}
fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
create_workspace_schema_structure(conn)?;
assignment_delivery::install(conn)?;
@@ -14293,6 +14324,7 @@ fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
notification_routing::install(conn)?;
notification_routing::jobs::install(conn)?;
notification_routing::install_business_events(conn)?;
notification_routing::upgrade_result_reports(conn)?;
ticket::verify_sqlite_ticket_schema(conn).map_err(|error| {
Error::Store(format!(
"fresh Workspace Ticket schema verification failed: {error}"
@@ -17527,6 +17559,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
WorkspaceSchemaMigrationStep {version:95, name:"Common notification routing and durable Worker delivery".into()},
WorkspaceSchemaMigrationStep {version:96, name:"Backend Job common notification routing".into()},
WorkspaceSchemaMigrationStep {version:97, name:"MR notifications and authenticated implementation results".into()},
WorkspaceSchemaMigrationStep {version:98, name:"Assignment-independent implementation reports".into()},
]
);
@@ -17634,6 +17667,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(95, "Common notification routing and durable Worker delivery".to_string()),
(96, "Backend Job common notification routing".to_string()),
(97, "MR notifications and authenticated implementation results".to_string()),
(98, "Assignment-independent implementation reports".to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -20255,7 +20289,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(94, "Canonical Ticket posts".into()),
(95, "Common notification routing and durable Worker delivery".into()),
(96, "Backend Job common notification routing".into()),
(97, "MR notifications and authenticated implementation results".into())]);
(97, "MR notifications and authenticated implementation results".into()),
(98, "Assignment-independent implementation reports".into())]);
for (table, rows) in unchanged.iter().zip(before.iter()) {
assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
}
@@ -0,0 +1,375 @@
//! SQL-only browser projection. No Ticket detail, dependency graph, MR/provider
//! observations, or per-ticket reads. Full QueryTicket/show retain their own semantics.
use super::{Error, RECORD_SOURCE_WORKSPACE_SQLITE, Result, SqliteWorkspaceAuthority};
use rusqlite::{Connection, params_from_iter, types::Value};
use server_api::{
BrowserTicketState, QueryPage, TicketBrowseQueryItem, TicketBrowseQueryRequest,
TicketBrowseQueryResponse,
};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
#[cfg(test)]
#[path = "ticket_browse_tests.rs"]
mod tests;
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct BrowseCursor {
scope: String,
updated_at: Option<String>,
id: String,
}
impl SqliteWorkspaceAuthority {
/// A filtered page of display fields in one SQLite read snapshot. All matching
/// predicates execute before LIMIT; unrelated live evidence is never observed.
pub fn query_ticket_browse(
&self,
mut query: TicketBrowseQueryRequest,
) -> Result<TicketBrowseQueryResponse> {
if query
.sort
.as_deref()
.is_some_and(|sort| sort != "updated_desc")
{
return Err(Error::InvalidInput(
"Ticket browse supports only updated_desc".into(),
));
}
query.sort = Some("updated_desc".into());
query.states.sort_by_key(|state| state.as_str());
query.states.dedup();
query.tag_ids.sort();
query.tag_ids.dedup();
if query.tag_ids.len() > 100 || query.tag_ids.iter().any(|id| id.trim().is_empty()) {
return Err(Error::InvalidInput(
"Ticket browse requires at most 100 nonblank tag IDs".into(),
));
}
query.query = query
.query
.filter(|text| !text.is_empty())
.map(|text| text.to_lowercase());
let cursor = query.cursor.take();
let limit = query.limit.take().unwrap_or(50).clamp(1, 100);
let digest = Sha256::digest(
serde_json::to_vec(&(&self.workspace_id, &query))
.map_err(|e| Error::InvalidInput(e.to_string()))?,
);
let scope = format!(
"ticket-browse-v1:{}",
digest
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
);
let after = cursor
.map(|cursor| -> Result<BrowseCursor> {
let cursor: BrowseCursor = serde_json::from_str(&cursor)
.map_err(|_| Error::InvalidInput("Invalid Ticket browse cursor".into()))?;
if cursor.scope != scope || cursor.id.is_empty() {
return Err(Error::InvalidInput(
"Ticket browse cursor does not match Workspace, filters or sort".into(),
));
}
Ok(cursor)
})
.transpose()?;
let (sql, values) = page_sql(&self.workspace_id, &query, after.as_ref(), limit);
self.store.with_conn_mut(|conn| {
if query.query.is_some() {
// SQLite lower() only folds ASCII. Use the same Unicode lowercasing
// as the bound needle, without post-filtering a limited page.
conn.create_scalar_function(
"ticket_browse_contains",
2,
rusqlite::functions::FunctionFlags::SQLITE_UTF8
| rusqlite::functions::FunctionFlags::SQLITE_DETERMINISTIC
| rusqlite::functions::FunctionFlags::SQLITE_INNOCUOUS,
|context| {
let raw_text = context.get_raw(0);
if raw_text == rusqlite::types::ValueRef::Null {
return Ok(false);
}
let text = raw_text
.as_str()
.map_err(|error| rusqlite::Error::UserFunctionError(Box::new(error)))?;
let needle = context
.get_raw(1)
.as_str()
.map_err(|error| rusqlite::Error::UserFunctionError(Box::new(error)))?;
Ok(text.to_lowercase().contains(needle))
},
)?;
}
let tx = conn.transaction()?;
let mut items = {
let mut statement = tx.prepare(&sql)?;
let rows = statement.query_map(params_from_iter(values.iter()), |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<String>>(4)?,
))
})?;
rows.map(|row| {
let (id, key, title, state, updated_at) = row?;
let resource_key =
key.ok_or_else(|| Error::Store(format!("missing resource key for {id}")))?;
let state = match state.as_str() {
"open" => BrowserTicketState::Open,
"closed" => BrowserTicketState::Closed,
_ => {
return Err(Error::Store(format!("invalid Ticket case state {state}")));
}
};
Ok(TicketBrowseQueryItem {
id,
resource_key,
title,
state,
updated_at,
tags: vec![],
current_closure: None,
assignment_tracking: vec![],
})
})
.collect::<Result<Vec<_>>>()?
};
let has_more = items.len() > limit;
items.truncate(limit);
enrich_page(&tx, &self.workspace_id, &mut items)?;
let next_cursor = if has_more {
items
.last()
.map(|item| {
serde_json::to_string(&BrowseCursor {
scope,
updated_at: item.updated_at.clone(),
id: item.id.clone(),
})
.map_err(|e| Error::Store(e.to_string()))
})
.transpose()?
} else {
None
};
tx.commit()?;
Ok(TicketBrowseQueryResponse {
page: QueryPage {
limit,
returned: items.len(),
has_more,
next_cursor,
sort: "updated_desc".into(),
source_limit: None,
source_truncated: false,
},
items,
record_authority: RECORD_SOURCE_WORKSPACE_SQLITE.into(),
})
})
}
}
fn page_sql(
workspace: &str,
query: &TicketBrowseQueryRequest,
after: Option<&BrowseCursor>,
limit: usize,
) -> (String, Vec<Value>) {
let mut values = vec![Value::Text(workspace.into())];
let mut predicates = vec!["t.workspace_id=?1".to_string()];
let mut bind = |value: String| {
values.push(Value::Text(value));
format!("?{}", values.len())
};
if !query.states.is_empty() {
let states = query
.states
.iter()
.map(|state| bind(state.as_str().into()))
.collect::<Vec<_>>();
predicates.push(format!("t.workflow_state IN ({})", states.join(",")));
}
if let Some(reason) = &query.closure_reason {
let reason = bind(serde_json::to_string(reason).expect("closure reason serializes"));
predicates.push(format!("t.workflow_state='closed' AND EXISTS (
SELECT 1 FROM typed_ticket_case_metadata m JOIN typed_ticket_case_history h
ON h.workspace_id=m.workspace_id AND h.ticket_id=m.ticket_id AND h.operation_key=m.closure_operation_key
WHERE m.workspace_id=t.workspace_id AND m.ticket_id=t.ticket_id
AND json_extract(h.judgment_json,'$.action.reason')=json_extract({reason},'$'))"));
}
if query.assigned_principal.is_some() || query.handling_status.is_some() {
let mut terms = vec![
"c.workspace_id=t.workspace_id".to_string(),
"c.ticket_id=t.ticket_id".into(),
];
if let Some(principal) = &query.assigned_principal {
use ticket::TicketAssignmentPrincipalSnapshot::*;
terms.push(match principal {
WorkspaceAgent { agent_key } => format!(
"a.principal_kind='workspace_agent' AND a.principal_id={}",
bind(agent_key.clone())
),
User { account_id } => format!(
"a.principal_kind='user' AND a.principal_id={}",
bind(account_id.clone())
),
Worker {
runtime_id,
worker_id,
} => format!(
"a.principal_kind='worker' AND a.runtime_id={} AND a.worker_id={}",
bind(runtime_id.clone()),
bind(worker_id.clone())
),
});
}
if let Some(status) = &query.handling_status {
let status = match status {
ticket::TicketAssignmentTrackingStatus::Waiting => "waiting",
ticket::TicketAssignmentTrackingStatus::Inprogress => "inprogress",
ticket::TicketAssignmentTrackingStatus::Done => "done",
};
terms.push(format!(
"a.role='orchestrator' AND json_extract(a.tracking_json,'$.status')={}",
bind(status.into())
));
}
predicates.push(format!("EXISTS (SELECT 1 FROM ticket_current_worker_assignments c
JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id AND a.role=c.role
WHERE {})", terms.join(" AND ")));
}
for tag in &query.tag_ids {
let tag = bind(tag.clone());
predicates.push(format!("EXISTS (SELECT 1 FROM typed_ticket_tags l
JOIN workspace_tags tag ON tag.workspace_id=l.workspace_id AND tag.tag_id=l.tag_id AND tag.deleted=0
WHERE l.workspace_id=t.workspace_id AND l.ticket_id=t.ticket_id AND l.tag_id={tag})"));
}
if let Some(text) = &query.query {
// Literal substring search: %, _ and backslash are not wildcards.
// Initial comment (-1) is the authoritative body.
let needle = bind(text.clone());
predicates.push(format!("(ticket_browse_contains(t.title,{needle}) OR EXISTS (
SELECT 1 FROM typed_ticket_events e WHERE e.workspace_id=t.workspace_id AND e.ticket_id=t.ticket_id
AND e.archived=0 AND e.event_index>=-1 AND ticket_browse_contains(e.body,{needle}))
OR EXISTS (SELECT 1 FROM typed_ticket_case_history h
WHERE h.workspace_id=t.workspace_id AND h.ticket_id=t.ticket_id
AND ticket_browse_contains(CASE json_type(h.judgment_json,'$.action.reason')
WHEN 'object' THEN 'other ' || json_extract(h.judgment_json,'$.action.reason.other')
ELSE json_extract(h.judgment_json,'$.action.reason') END,{needle})))"));
}
if let Some(after) = after {
let id = bind(after.id.clone());
if let Some(time) = &after.updated_at {
let time = bind(time.clone());
predicates.push(format!("(t.updated_at<{time} OR t.updated_at IS NULL OR (t.updated_at={time} AND t.ticket_id>{id}))"));
} else {
predicates.push(format!("(t.updated_at IS NULL AND t.ticket_id>{id})"));
}
}
values.push(Value::Integer((limit + 1) as i64));
let sql = format!("SELECT t.ticket_id,k.resource_key,t.title,t.workflow_state,t.updated_at
FROM typed_tickets t LEFT JOIN workspace_resource_keys k
ON k.workspace_id=t.workspace_id AND k.resource_kind='ticket' AND k.resource_id=t.ticket_id
WHERE {} ORDER BY t.updated_at DESC,t.ticket_id ASC LIMIT ?{}", predicates.join(" AND "), values.len());
(sql, values)
}
fn enrich_page(
conn: &Connection,
workspace: &str,
items: &mut [TicketBrowseQueryItem],
) -> Result<()> {
if items.is_empty() {
return Ok(());
}
let positions: HashMap<String, usize> = items
.iter()
.enumerate()
.map(|(i, item)| (item.id.clone(), i))
.collect();
let mut values = vec![Value::Text(workspace.into())];
values.extend(items.iter().map(|item| Value::Text(item.id.clone())));
let ids = (2..=values.len())
.map(|i| format!("?{i}"))
.collect::<Vec<_>>()
.join(",");
let mut statement = conn.prepare(&format!("SELECT l.ticket_id,tag.tag_id,tag.name,tag.last_operation_key
FROM typed_ticket_tags l JOIN workspace_tags tag ON tag.workspace_id=l.workspace_id AND tag.tag_id=l.tag_id
WHERE l.workspace_id=?1 AND l.ticket_id IN ({ids}) AND tag.deleted=0 ORDER BY l.ticket_id,tag.tag_id"))?;
let mut rows = statement.query(params_from_iter(values.iter()))?;
while let Some(row) = rows.next()? {
let id: String = row.get(0)?;
items[positions[&id]].tags.push(ticket::WorkspaceTag {
workspace_id: workspace.into(),
id: row.get(1)?,
name: row.get(2)?,
last_operation_key: row.get(3)?,
deleted: false,
});
}
let mut statement = conn.prepare(&format!("SELECT t.ticket_id,h.judgment_json FROM typed_tickets t
JOIN typed_ticket_case_metadata m ON m.workspace_id=t.workspace_id AND m.ticket_id=t.ticket_id
JOIN typed_ticket_case_history h ON h.workspace_id=m.workspace_id AND h.ticket_id=m.ticket_id AND h.operation_key=m.closure_operation_key
WHERE t.workspace_id=?1 AND t.ticket_id IN ({ids}) AND t.workflow_state='closed'"))?;
let mut rows = statement.query(params_from_iter(values.iter()))?;
while let Some(row) = rows.next()? {
let id: String = row.get(0)?;
let entry: ticket::TicketCaseHistoryEntry = decode_json(&row.get::<_, String>(1)?)?;
let reason = match entry.action {
ticket::TicketCaseAction::Close { reason }
| ticket::TicketCaseAction::CorrectClosure { reason } => reason,
ticket::TicketCaseAction::Reopen => {
return Err(Error::Store("Closure key names a reopen receipt".into()));
}
};
items[positions[&id]].current_closure = Some(ticket::TicketClosureJudgment {
operation_key: entry.operation_key,
reason,
explanation: entry.explanation,
references: entry.references,
author: entry.author,
at: entry.at,
});
}
// Current Orchestrator projections only, exactly as TicketQueryItem. Legacy
// unknown handling retains its real Assign observation without inventing status.
let mut statement = conn.prepare(&format!("SELECT a.ticket_id,json_object(
'workspace_id',a.workspace_id,'ticket_id',a.ticket_id,'assignment_id',a.assignment_id,'role',a.role,
'principal',json(CASE a.principal_kind
WHEN 'workspace_agent' THEN json_object('kind','workspace_agent','agent_key',a.principal_id)
WHEN 'user' THEN json_object('kind','user','account_id',a.principal_id)
WHEN 'worker' THEN json_object('kind','worker','runtime_id',a.runtime_id,'worker_id',a.worker_id) END),
'assigned_by',a.assigned_by,'assigned_at',a.assigned_at,'release',NULL,'current',json('true'),
'tracking',json(a.tracking_json),'last_event_id',CASE
WHEN a.tracking_json IS NULL AND a.last_tracking_event_id IS NULL THEN (
SELECT e.event_id FROM ticket_worker_assignment_events e
WHERE e.workspace_id=a.workspace_id AND e.ticket_id=a.ticket_id AND e.assignment_id=a.assignment_id
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=a.workspace_id AND h.ticket_id=a.ticket_id AND h.assignment_id=a.assignment_id)
ORDER BY e.rowid LIMIT 1)
ELSE a.last_tracking_event_id END,
'updated_at',a.tracking_updated_at,'migration_diagnostic',a.tracking_migration_diagnostic)
FROM ticket_current_worker_assignments c JOIN ticket_worker_assignments a
ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id AND a.role=c.role
WHERE c.workspace_id=?1 AND c.ticket_id IN ({ids}) AND c.role='orchestrator'
ORDER BY a.ticket_id,a.assignment_id"))?;
let mut rows = statement.query(params_from_iter(values.iter()))?;
while let Some(row) = rows.next()? {
let id: String = row.get(0)?;
items[positions[&id]]
.assignment_tracking
.push(decode_json(&row.get::<_, String>(1)?)?);
}
Ok(())
}
fn decode_json<T: serde::de::DeserializeOwned>(json: &str) -> Result<T> {
serde_json::from_str(json).map_err(|e| Error::Store(e.to_string()))
}
@@ -0,0 +1,173 @@
//! Reproducible component benchmark: disposable SQLite, real authority methods, no HTTP/Runtime.
//! Run with `cargo test -p yoi-workspace-server ticket_browse_component_performance --lib -- --ignored --nocapture`.
use crate::authority::{SqliteWorkspaceAuthority, TicketAuthority};
use crate::store::{ControlPlaneStore, SqliteWorkspaceStore, WorkspaceRecord};
use rusqlite::{Connection, params};
use std::time::Instant;
use ticket::{SqliteTicketBackend, TicketBackend};
async fn fixture(
tickets: usize,
events_per_ticket: usize,
) -> (tempfile::TempDir, SqliteWorkspaceAuthority) {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("benchmark.db");
let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "bench".into(),
owner_account_id: "owner-account".into(),
display_name: "Component benchmark".into(),
state: "active".into(),
created_at: "2026-10-01T00:00:00Z".into(),
updated_at: "2026-10-01T00:00:00Z".into(),
})
.await
.unwrap();
let backend = SqliteTicketBackend::open_verified(&path, "bench").unwrap();
let seed = backend
.create(ticket::NewTicket::new("Benchmark seed"))
.unwrap();
let mut conn = Connection::open(&path).unwrap();
conn.pragma_update(None, "foreign_keys", true).unwrap();
let ticket_columns: Vec<String> = conn
.prepare("PRAGMA table_info(typed_tickets)")
.unwrap()
.query_map([], |row| row.get(1))
.unwrap()
.map(Result::unwrap)
.collect();
let expressions = ticket_columns
.iter()
.map(|name| match name.as_str() {
"ticket_id" | "slug" => "?1".to_owned(),
"title" => "?2".to_owned(),
"created_at" | "updated_at" => "'2026-10-01T00:00:00Z'".to_owned(),
_ => format!("\"{name}\""),
})
.collect::<Vec<_>>()
.join(",");
let clone_ticket = format!(
"INSERT INTO typed_tickets SELECT {expressions} FROM typed_tickets WHERE workspace_id='bench' AND ticket_id=?3"
);
let tx = conn.transaction().unwrap();
{
let mut insert_ticket = tx.prepare(&clone_ticket).unwrap();
let mut insert_key = tx.prepare("INSERT INTO workspace_resource_keys(workspace_id,resource_kind,resource_id,sequence,resource_key,allocated_at) VALUES('bench','ticket',?1,?2,?3,'2026-10-01T00:00:00Z')").unwrap();
let mut insert_event = tx.prepare("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body,event_ref,archived) VALUES('bench',?1,?2,'comment',?3,?4,0)").unwrap();
let mut insert_attribute = tx.prepare("INSERT INTO typed_ticket_event_attributes(workspace_id,ticket_id,event_index,key,value) VALUES('bench',?1,?2,'bench','value')").unwrap();
let mut insert_reference = tx.prepare("INSERT INTO typed_ticket_event_references(workspace_id,ticket_id,event_index,ordinal,kind,target) VALUES('bench',?1,?2,0,'ticket',?3)").unwrap();
let mut insert_relation = tx.prepare("INSERT INTO typed_ticket_relations(workspace_id,ticket_id,kind,target,author,at) VALUES('bench',?1,'related',?2,'owner-account','2026-10-01T00:00:00Z')").unwrap();
let body = "component benchmark text ".repeat(40);
for i in 0..tickets {
let id = format!("bench-{i:06}");
insert_ticket
.execute(params![id, format!("Ticket {i:06}"), seed.id])
.unwrap();
insert_key
.execute(params![id, i + 2, format!("T-{}", i + 2)])
.unwrap();
insert_relation.execute(params![id, seed.id]).unwrap();
for e in -1..events_per_ticket as i64 {
let text = if e == events_per_ticket as i64 - 1 {
format!("search-needle {body}")
} else {
body.clone()
};
insert_event
.execute(params![id, e, text, format!("{id}-event-{e}")])
.unwrap();
insert_attribute.execute(params![id, e]).unwrap();
insert_reference.execute(params![id, e, seed.id]).unwrap();
}
}
}
tx.commit().unwrap();
drop(conn);
let authority = SqliteWorkspaceAuthority::new(path, "bench").unwrap();
(dir, authority)
}
fn query(text: Option<&str>) -> server_api::TicketQueryRequest {
server_api::TicketQueryRequest {
states: vec![ticket::TicketState::Open],
sort: Some("updated_desc".into()),
limit: Some(30),
query: text.map(str::to_owned),
..Default::default()
}
}
#[tokio::test]
#[ignore = "component timing benchmark; no production services"]
async fn ticket_browse_component_performance() {
let tickets = std::env::var("BROWSE_BENCH_TICKETS")
.ok()
.map(|v| v.parse().unwrap())
.unwrap_or(1_000);
let events = std::env::var("BROWSE_BENCH_EVENTS")
.ok()
.map(|v| v.parse().unwrap())
.unwrap_or(100);
let (_dir, authority) = fixture(tickets, events).await;
for text in [
None,
Some("search-needle"),
Some("Ticket 00099"),
Some("absent-needle"),
] {
let request = query(text);
let browse = server_api::TicketBrowseQueryRequest {
states: request.states.clone(),
sort: request.sort.clone(),
limit: request.limit,
query: request.query.clone(),
..Default::default()
};
// Warm both paths on the same database; exclude fixture generation.
authority.query_tickets(request.clone()).unwrap();
authority.query_ticket_browse(browse.clone()).unwrap();
let mut old_times = Vec::new();
let mut new_times = Vec::new();
let mut old_bytes = 0;
let mut new_bytes = 0;
// Alternate paths to reduce bias from unrelated machine activity. Include
// complete response JSON serialization, but not parity assertions, in timings.
for _ in 0..3 {
let started = Instant::now();
let old = authority.query_tickets(request.clone()).unwrap();
old_bytes = serde_json::to_vec(&old).unwrap().len();
old_times.push(started.elapsed().as_secs_f64() * 1000.0);
let started = Instant::now();
let new = authority.query_ticket_browse(browse.clone()).unwrap();
new_bytes = serde_json::to_vec(&new).unwrap().len();
new_times.push(started.elapsed().as_secs_f64() * 1000.0);
let projected = old
.items
.into_iter()
.map(|item| server_api::TicketBrowseQueryItem {
id: item.id,
resource_key: item.resource_key,
title: item.title,
state: item.state,
updated_at: item.updated_at,
tags: item.tags,
current_closure: item.current_closure,
assignment_tracking: item.assignment_tracking,
})
.collect::<Vec<_>>();
assert_eq!(projected, new.items);
assert!(new.items.len() <= 30);
if text == Some("absent-needle") {
assert!(new.items.is_empty());
}
assert_eq!(old.page.has_more, new.page.has_more);
}
old_times.sort_by(f64::total_cmp);
new_times.sort_by(f64::total_cmp);
println!(
"COMPARE tickets={tickets} events_each={events} search={text:?} samples=3 old_ms={old_times:?} new_ms={new_times:?} median_speedup={:.1} old_bytes={old_bytes} new_bytes={new_bytes}",
old_times[1] / new_times[1]
);
}
}
@@ -0,0 +1,620 @@
use super::*;
use crate::authority::TicketAuthority;
use crate::store::{ControlPlaneStore, SqliteWorkspaceStore, WorkspaceRecord};
use rusqlite::params;
use ticket::{SqliteTicketBackend, TicketBackend, TicketClosureReason, TicketState};
struct Fixture {
_dir: tempfile::TempDir,
authority: SqliteWorkspaceAuthority,
backend: SqliteTicketBackend,
}
impl Fixture {
async fn new() -> Self {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("browse.db");
let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
for workspace in ["w", "foreign"] {
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: workspace.into(),
owner_account_id: "owner-account".into(),
display_name: workspace.into(),
state: "active".into(),
created_at: "1".into(),
updated_at: "1".into(),
})
.await
.unwrap();
}
Self {
authority: SqliteWorkspaceAuthority::new(&path, "w").unwrap(),
backend: SqliteTicketBackend::open_verified(&path, "w").unwrap(),
_dir: dir,
}
}
fn ticket(&self, title: &str, time: Option<&str>) -> String {
let id = self
.backend
.create(ticket::NewTicket::new(title))
.unwrap()
.id;
self.authority.store.with_conn(|conn| {
conn.execute("UPDATE typed_tickets SET updated_at=?1 WHERE workspace_id='w' AND ticket_id=?2", params![time,id])?;
Ok(())
}).unwrap();
id
}
fn run(&self, query: TicketBrowseQueryRequest) -> TicketBrowseQueryResponse {
self.authority.query_ticket_browse(query).unwrap()
}
fn tag(&self, ticket: &str, tag: &str) {
self.authority.store.with_conn(|conn| {
conn.execute("INSERT OR IGNORE INTO workspace_tags(workspace_id,tag_id,name,normalized_name,last_operation_key,deleted) VALUES('w',?1,?1,?1,?1,0)",[tag])?;
conn.execute("INSERT INTO typed_ticket_tags(workspace_id,ticket_id,tag_id) VALUES('w',?1,?2)",params![ticket,tag])?;
Ok(())
}).unwrap();
}
fn closure(&self, id: &str, operation: &str, reason: TicketClosureReason, closed: bool) {
let entry = ticket::TicketCaseHistoryEntry {
operation_key: operation.into(),
previous_closure_operation_key: None,
action: ticket::TicketCaseAction::Close { reason },
explanation: "test judgment".into(),
references: vec![],
author: "owner-account".into(),
at: "1".into(),
};
self.authority.store.with_conn(|conn| {
conn.execute("INSERT INTO typed_ticket_case_history(workspace_id,ticket_id,ordinal,operation_key,request_fingerprint,judgment_json)
SELECT 'w',?1,coalesce(max(ordinal),0)+1,?2,'fixture',?3 FROM typed_ticket_case_history WHERE workspace_id='w' AND ticket_id=?1",
params![id,operation,serde_json::to_string(&entry).unwrap()])?;
conn.execute("INSERT INTO typed_ticket_case_metadata(workspace_id,ticket_id,closure_operation_key) VALUES('w',?1,?2)
ON CONFLICT(workspace_id,ticket_id) DO UPDATE SET closure_operation_key=excluded.closure_operation_key",params![id,operation])?;
conn.execute("UPDATE typed_tickets SET workflow_state=?1 WHERE workspace_id='w' AND ticket_id=?2",params![if closed {"closed"} else {"open"},id])?;
Ok(())
}).unwrap();
}
fn assign(&self, id: &str, assignment: &str, status: Option<&str>, current: bool) {
self.authority.store.with_conn(|conn| {
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at)
VALUES('w',?1,?2,'orchestrator','workspace_agent','workspace-orchestrator','owner-account','1')",params![id,assignment])?;
conn.execute("UPDATE ticket_worker_assignments SET tracking_json=?1,tracking_updated_at=?2 WHERE workspace_id='w' AND assignment_id=?3",
params![status.map(|s| serde_json::json!({"status":s,"waiting_reason":null,"progress":null,"next_action":null,"result":null}).to_string()),status.map(|_|"1"),assignment])?;
if current {
conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,role,assignment_id,principal_kind,principal_id,updated_at)
VALUES('w',?1,'orchestrator',?2,'workspace_agent','workspace-orchestrator','1')",params![id,assignment])?;
}
Ok(())
}).unwrap();
}
fn event(&self, id: &str, index: i64, body: &str, archived: bool) {
self.authority.store.with_conn(|conn| {
if index == -1 {
conn.execute("UPDATE typed_ticket_events SET body=?1 WHERE workspace_id='w' AND ticket_id=?2 AND event_index=-1",params![body,id])?;
} else {
conn.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body,event_ref,archived)
VALUES('w',?1,?2,'comment',?3,?4,?5)",params![id,index,body,format!("{id}-event-{index}"),archived])?;
}
Ok(())
}).unwrap();
}
}
fn agent() -> ticket::TicketAssignmentPrincipalSnapshot {
ticket::TicketAssignmentPrincipalSnapshot::WorkspaceAgent {
agent_key: "workspace-orchestrator".into(),
}
}
fn ids(response: &TicketBrowseQueryResponse) -> Vec<String> {
response.items.iter().map(|item| item.id.clone()).collect()
}
#[tokio::test]
async fn ticket_browse_unicode_search_uses_same_case_mapping_before_pagination() {
let f = Fixture::new().await;
f.ticket("newer nonmatch", Some("9"));
let title = f.ticket("ÉCOLE", Some("4"));
let body = f.ticket("body", Some("3"));
f.event(&body, -1, "ÜBER", false);
let thread = f.ticket("thread", Some("2"));
f.event(&thread, 0, "ΜΉΝΥΜΑ", false);
let closure = f.ticket("closure", Some("1"));
f.closure(
&closure,
"unicode-close",
TicketClosureReason::Other("ÖFFEN".into()),
true,
);
let special = f.ticket("special", Some("0"));
f.event(&special, 0, "ascii\0ÜNIT Kelvin", false);
for (needle, expected) in [
("ünit", &special),
("kelvin", &special),
("ÉCOLE", &title),
("école", &title),
("ÜBER", &body),
("über", &body),
("ΜΉΝΥΜΑ", &thread),
("μήνυμα", &thread),
("ÖFFEN", &closure),
("öffen", &closure),
] {
let response = f.run(TicketBrowseQueryRequest {
query: Some(needle.into()),
limit: Some(1),
..Default::default()
});
assert_eq!(ids(&response), vec![expected.clone()], "{needle}");
assert!(!response.page.has_more);
}
}
#[tokio::test]
async fn ticket_browse_keyset_ties_null_dates_and_scoped_canonical_cursors() {
let f = Fixture::new().await;
let mut tied = vec![
f.ticket("tie 1", Some("2")),
f.ticket("tie 2", Some("2")),
f.ticket("tie 3", Some("2")),
];
tied.sort();
let newer = f.ticket("new", Some("3"));
let older = f.ticket("old", Some("1"));
let empty = f.ticket("empty", Some(""));
let mut nulls = vec![f.ticket("null 1", None), f.ticket("null 2", None)];
nulls.sort();
let mut expected = vec![newer];
expected.extend(tied);
expected.extend([older, empty]);
expected.extend(nulls);
let mut query = TicketBrowseQueryRequest {
states: vec![TicketState::Closed, TicketState::Open, TicketState::Open],
limit: Some(2),
..Default::default()
};
let first = f.run(query.clone());
assert_eq!(first.page.returned, 2);
assert!(first.page.has_more);
assert!(!first.page.source_truncated);
assert_eq!(first.page.source_limit, None);
let cursor = first.page.next_cursor.clone();
let mut seen = ids(&first);
query.states = vec![TicketState::Open, TicketState::Closed];
query.sort = Some("updated_desc".into());
query.limit = Some(3);
query.cursor = cursor.clone();
loop {
let page = f.run(query.clone());
seen.extend(ids(&page));
if !page.page.has_more {
assert!(page.page.next_cursor.is_none());
break;
}
assert_eq!(page.items.len(), 3);
query.cursor = page.page.next_cursor;
}
assert_eq!(seen, expected);
let changed = TicketBrowseQueryRequest {
cursor: cursor.clone(),
states: vec![TicketState::Closed],
..Default::default()
};
assert!(f.authority.query_ticket_browse(changed).is_err());
let foreign =
SqliteWorkspaceAuthority::new(f._dir.path().join("browse.db"), "foreign").unwrap();
assert!(
foreign
.query_ticket_browse(TicketBrowseQueryRequest {
cursor: cursor.clone(),
states: vec![TicketState::Open, TicketState::Closed],
..Default::default()
})
.is_err()
);
assert!(
f.authority
.query_tickets(server_api::TicketQueryRequest {
cursor,
..Default::default()
})
.is_err()
);
for sort in ["title", "relevance", "created_desc"] {
assert!(
f.authority
.query_ticket_browse(TicketBrowseQueryRequest {
sort: Some(sort.into()),
..Default::default()
})
.is_err()
);
}
assert!(
f.authority
.query_ticket_browse(TicketBrowseQueryRequest {
cursor: Some("v1:garbage".into()),
..Default::default()
})
.is_err()
);
}
#[tokio::test]
async fn ticket_browse_all_filters_are_applied_before_limit_and_batch_fields_match() {
let f = Fixture::new().await;
for i in 0..8 {
f.ticket(&format!("irrelevant {i}"), Some("9"));
}
let one = f.ticket("Needle one", Some("2"));
let two = f.ticket("Needle two", Some("1"));
let partial = f.ticket("Needle partial", Some("8"));
for id in [&one, &two, &partial] {
f.tag(id, "a");
f.closure(
id,
&format!("close-{id}"),
TicketClosureReason::Other("custom reason".into()),
true,
);
f.assign(id, &format!("assignment-{id}"), Some("done"), true);
}
f.tag(&one, "b");
f.tag(&two, "b");
let query = TicketBrowseQueryRequest {
query: Some("NEEDLE".into()),
states: vec![TicketState::Closed],
tag_ids: vec!["b".into(), "a".into(), "a".into()],
assigned_principal: Some(agent()),
handling_status: Some(ticket::TicketAssignmentTrackingStatus::Done),
closure_reason: Some(TicketClosureReason::Other("custom reason".into())),
limit: Some(1),
..Default::default()
};
let first = f.run(query.clone());
assert_eq!(ids(&first), vec![one.clone()]);
assert!(first.page.has_more);
assert_eq!(
first.items[0]
.tags
.iter()
.map(|tag| tag.id.as_str())
.collect::<Vec<_>>(),
vec!["a", "b"]
);
let detail = f.authority.ticket(&one).unwrap();
assert_eq!(
first.items[0].assignment_tracking,
detail.assignment_tracking
);
assert_eq!(first.items[0].current_closure, detail.current_closure);
let second = f.run(TicketBrowseQueryRequest {
cursor: first.page.next_cursor,
tag_ids: vec!["a".into(), "b".into()],
query: Some("needle".into()),
..query.clone()
});
assert_eq!(ids(&second), vec![two]);
assert!(!second.page.has_more);
for missing in ["missing", "deleted"] {
assert!(
f.run(TicketBrowseQueryRequest {
tag_ids: vec![missing.into()],
..query.clone()
})
.items
.is_empty()
);
}
f.authority
.store
.with_conn(|conn| {
conn.execute(
"UPDATE workspace_tags SET deleted=1 WHERE workspace_id='w' AND tag_id='b'",
[],
)?;
Ok(())
})
.unwrap();
assert!(f.run(query).items.is_empty());
}
#[tokio::test]
async fn ticket_browse_assignment_current_missing_unknown_and_cleared_semantics() {
let f = Fixture::new().await;
let unassigned = f.ticket("unassigned", Some("1"));
let unknown = f.ticket("unknown", Some("1"));
let cleared = f.ticket("cleared", Some("1"));
let current = f.ticket("current", Some("1"));
f.assign(&unknown, "unknown", None, true);
f.authority.store.with_conn(|conn| {
conn.execute("INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,actor,created_at) VALUES('w',?1,'orchestrator','legacy-assign','assigned','unknown','owner-account','1')",[&unknown])?;
Ok(())
}).unwrap();
f.assign(&cleared, "cleared", Some("done"), true);
f.authority.store.with_conn(|conn| {
conn.execute("DELETE FROM ticket_current_worker_assignments WHERE workspace_id='w' AND assignment_id='cleared'",[])?; Ok(())
}).unwrap();
f.assign(&current, "historical", Some("done"), false);
f.assign(&current, "current", Some("waiting"), true);
let all = f.run(Default::default());
assert_eq!(all.items.len(), 4);
for id in [&unassigned, &cleared] {
assert!(
all.items
.iter()
.find(|item| &item.id == id)
.unwrap()
.assignment_tracking
.is_empty()
);
}
assert!(
all.items
.iter()
.find(|item| item.id == unknown)
.unwrap()
.assignment_tracking[0]
.tracking
.is_none()
);
let unknown_projection = &all
.items
.iter()
.find(|item| item.id == unknown)
.unwrap()
.assignment_tracking;
assert_eq!(
unknown_projection[0].last_event_id.as_deref(),
Some("legacy-assign")
);
assert_eq!(
*unknown_projection,
f.authority.ticket_assignment_tracking(&unknown).unwrap()
);
let mut assigned = ids(&f.run(TicketBrowseQueryRequest {
assigned_principal: Some(agent()),
..Default::default()
}));
assigned.sort();
let mut expected = vec![current.clone(), unknown];
expected.sort();
assert_eq!(assigned, expected);
assert!(
f.run(TicketBrowseQueryRequest {
handling_status: Some(ticket::TicketAssignmentTrackingStatus::Done),
..Default::default()
})
.items
.is_empty()
);
assert_eq!(
ids(&f.run(TicketBrowseQueryRequest {
handling_status: Some(ticket::TicketAssignmentTrackingStatus::Waiting),
..Default::default()
})),
vec![current]
);
for principal in [
ticket::TicketAssignmentPrincipalSnapshot::User {
account_id: "missing".into(),
},
ticket::TicketAssignmentPrincipalSnapshot::Worker {
runtime_id: "missing".into(),
worker_id: "missing".into(),
},
] {
assert!(
f.run(TicketBrowseQueryRequest {
assigned_principal: Some(principal),
..Default::default()
})
.items
.is_empty()
);
}
}
#[tokio::test]
async fn ticket_browse_search_literal_title_initial_thread_and_closure_history() {
let f = Fixture::new().await;
let title = f.ticket("Literal %_ title", Some("1"));
let body = f.ticket("body", Some("1"));
f.event(&body, -1, "Initial %_ BODY", false);
let thread = f.ticket("thread", Some("1"));
f.event(&thread, 80, "Thread %_ needle", false);
let historical = f.ticket("historical", Some("1"));
f.closure(
&historical,
"old",
TicketClosureReason::Other("%_ custom".into()),
true,
);
f.closure(&historical, "new", TicketClosureReason::Completed, false);
let archived = f.ticket("archived", Some("1"));
f.event(&archived, 90, "Archived %_", true);
f.ticket("anything", Some("1"));
let mut expected = vec![title, body, thread, historical.clone()];
expected.sort();
let result = f.run(TicketBrowseQueryRequest {
query: Some("%_".into()),
..Default::default()
});
assert_eq!(ids(&result), expected);
assert!(
result
.items
.iter()
.find(|item| item.id == historical)
.unwrap()
.current_closure
.is_none()
);
assert_eq!(
ids(&f.run(TicketBrowseQueryRequest {
query: Some("other %_ custom".into()),
..Default::default()
})),
vec![historical]
);
assert!(
f.run(TicketBrowseQueryRequest {
closure_reason: Some(TicketClosureReason::Other("%_ custom".into())),
..Default::default()
})
.items
.is_empty()
);
assert!(
f.run(TicketBrowseQueryRequest {
query: Some("archived %_".into()),
..Default::default()
})
.items
.is_empty()
);
}
#[tokio::test]
async fn ticket_browse_current_closure_not_prior_or_reopened_and_workspace_isolation() {
let f = Fixture::new().await;
let id = f.ticket("same identity", Some("1"));
f.closure(&id, "previous", TicketClosureReason::Completed, true);
f.closure(&id, "current", TicketClosureReason::NotPlanned, true);
assert!(
f.run(TicketBrowseQueryRequest {
closure_reason: Some(TicketClosureReason::Completed),
..Default::default()
})
.items
.is_empty()
);
assert_eq!(
ids(&f.run(TicketBrowseQueryRequest {
closure_reason: Some(TicketClosureReason::NotPlanned),
..Default::default()
})),
vec![id.clone()]
);
f.authority.store.with_conn(|conn| {
let columns: Vec<String> = conn.prepare("PRAGMA table_info(typed_tickets)")?.query_map([],|row| row.get(1))?.collect::<rusqlite::Result<_>>()?;
let expressions = columns.iter().map(|name| match name.as_str() { "workspace_id" => "'foreign'".into(), "title" => "'foreign secret'".into(), _ => format!("\"{name}\"") }).collect::<Vec<_>>().join(",");
conn.execute(&format!("INSERT INTO typed_tickets SELECT {expressions} FROM typed_tickets WHERE workspace_id='w' AND ticket_id=?1"),[&id])?;
conn.execute("INSERT INTO workspace_resource_keys(workspace_id,resource_kind,resource_id,sequence,resource_key,allocated_at) VALUES('foreign','ticket',?1,1,'T-1','1')",[&id])?;
conn.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body,event_ref,archived) VALUES('foreign',?1,-1,'comment','foreign secret','foreign-event',0)",[&id])?;
conn.execute("INSERT INTO workspace_tags(workspace_id,tag_id,name,normalized_name,last_operation_key,deleted) VALUES('foreign','foreign-tag','Foreign','foreign','tag-op',0)",[])?;
conn.execute("INSERT INTO typed_ticket_tags(workspace_id,ticket_id,tag_id) VALUES('foreign',?1,'foreign-tag')",[&id])?;
Ok(())
}).unwrap();
let local = f.run(Default::default());
assert_eq!(local.items.len(), 1);
assert!(local.items[0].tags.is_empty());
assert!(
f.run(TicketBrowseQueryRequest {
query: Some("foreign secret".into()),
..Default::default()
})
.items
.is_empty()
);
assert!(
f.run(TicketBrowseQueryRequest {
tag_ids: vec!["foreign-tag".into()],
..Default::default()
})
.items
.is_empty()
);
let foreign =
SqliteWorkspaceAuthority::new(f._dir.path().join("browse.db"), "foreign").unwrap();
let foreign = foreign.query_ticket_browse(Default::default()).unwrap();
assert_eq!(foreign.items[0].title, "foreign secret");
assert!(foreign.items[0].current_closure.is_none());
assert_eq!(foreign.items[0].tags[0].id, "foreign-tag");
}
#[tokio::test]
async fn ticket_browse_user_worker_principals_are_exact_current_identity_not_handling() {
let f = Fixture::new().await;
let id = f.ticket("current roles", Some("1"));
f.assign(&id, "orchestrator", Some("waiting"), true);
f.authority.store.with_conn(|conn| {
conn.execute("INSERT INTO worker_registry(workspace_id,runtime_id,worker_id,display_name,retention_state,created_at,updated_at) VALUES('w','runtime','worker','Fixture','normal','1','1')",[])?;
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w',?1,'owner','owner','user','owner-account','owner-account','1')",[&id])?;
conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w',?1,'owner','owner','user','owner-account','1')",[&id])?;
conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,runtime_id,worker_id,assigned_by,assigned_at) VALUES('w',?1,'worker','worker','worker','runtime','worker','owner-account','1')",[&id])?;
conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,runtime_id,worker_id,updated_at) VALUES('w',?1,'worker','worker','worker','runtime','worker','1')",[&id])?;
Ok(())
}).unwrap();
for principal in [
ticket::TicketAssignmentPrincipalSnapshot::User {
account_id: "owner-account".into(),
},
ticket::TicketAssignmentPrincipalSnapshot::Worker {
runtime_id: "runtime".into(),
worker_id: "worker".into(),
},
] {
let query = TicketBrowseQueryRequest {
assigned_principal: Some(principal),
..Default::default()
};
let result = f.run(query.clone());
assert_eq!(ids(&result), vec![id.clone()]);
assert_eq!(result.items[0].assignment_tracking.len(), 1);
assert!(
f.run(TicketBrowseQueryRequest {
handling_status: Some(ticket::TicketAssignmentTrackingStatus::Waiting),
..query
})
.items
.is_empty()
);
}
assert!(
f.run(TicketBrowseQueryRequest {
assigned_principal: Some(ticket::TicketAssignmentPrincipalSnapshot::Worker {
runtime_id: "different-runtime".into(),
worker_id: "worker".into()
}),
..Default::default()
})
.items
.is_empty()
);
}
#[tokio::test]
async fn ticket_browse_plain_page_sql_uses_recent_index_without_history_or_graph() {
let f = Fixture::new().await;
f.ticket("first", Some("1"));
let (sql, values) = page_sql("w", &TicketBrowseQueryRequest::default(), None, 30);
for forbidden in [
"typed_ticket_events",
"typed_ticket_case_history",
"typed_ticket_relations",
"merge_request",
"tracking_history",
] {
assert!(!sql.contains(forbidden), "{sql}");
}
f.authority
.store
.with_conn(|conn| {
let plans: Vec<String> = conn
.prepare(&format!("EXPLAIN QUERY PLAN {sql}"))?
.query_map(params_from_iter(values.iter()), |row| row.get(3))?
.collect::<rusqlite::Result<_>>()?;
assert!(
plans
.iter()
.any(|plan| plan.contains("typed_tickets_workspace_updated")),
"{plans:?}"
);
assert!(
!plans.iter().any(|plan| plan.contains("TEMP B-TREE")),
"{plans:?}"
);
Ok(())
})
.unwrap();
}