diff --git a/.busbar-ref b/.busbar-ref index 9b9106f..35a5507 100644 --- a/.busbar-ref +++ b/.busbar-ref @@ -1 +1 @@ -c8780349cf66d09b891478d50766b89dc1ff224c 1.5.3 +5a4f0195e29ef8f17a8abcffa308fb797c797edf 1.6.0 diff --git a/store-sqlite-plugin/tests/e2e.rs b/store-sqlite-plugin/tests/e2e.rs index f036891..759f0cb 100644 --- a/store-sqlite-plugin/tests/e2e.rs +++ b/store-sqlite-plugin/tests/e2e.rs @@ -44,7 +44,9 @@ //! /api/v1/admin/keys`, and independently verifies both landed in the real on-disk file with a //! second `SqliteStore::open` that never touches the plugin/ABI/admin-API/loader. -use busbar_api::{ModelTokens, Store, TierTokens, UsageLedger}; +use busbar_api::{ + McpCallRecord, ModelTokens, Store, TaskEventRow, TaskRow, TierTokens, UsageLedger, +}; use busbar_store_sqlite::SqliteStore; use std::path::PathBuf; use std::process::Command; @@ -81,23 +83,105 @@ impl Drop for ScratchDir { } } -/// Locate the built `busbar-store-sqlite-plugin` cdylib in the target dir (mirrors the loader's own -/// `sqlite_plugin_path` helper in the monorepo). -fn plugin_path() -> Option { - let candidate = (|| { - let exe = std::env::current_exe().ok()?; // .../target//deps/e2e- - let profile_dir = exe.parent()?.parent()?; // .../target/ - let name = busbar_plugin_loader::plugin_library_filename("busbar_store_sqlite_plugin"); - let candidate = profile_dir.join(&name); - candidate.exists().then_some(candidate) - })(); - if candidate.is_none() && std::env::var_os("CI").is_some() { - panic!( - "the store-sqlite-plugin cdylib is not built under CI: `cargo test` must build it. \ - Refusing to silently skip the only over-the-ABI coverage of the durable sqlite store path." - ); +/// Locate the cdylib THIS `cargo test` invocation just built — never a leftover artifact. +/// +/// This looks in `target//deps/`, NOT `target//`, and that distinction is the +/// whole point of this function. +/// +/// `cargo` emits the lib target's cdylib into `deps/` as part of the very build graph that produces +/// this test binary (this package's lib unit is compiled with BOTH declared crate-types — see +/// `[lib] crate-type = ["cdylib", "rlib"]` in Cargo.toml — so `deps/libbusbar_store_sqlite_plugin.dylib` +/// is by construction up to date with the source tree being tested). It only *uplifts* a copy to +/// `target//` for `cargo build`, NEVER for `cargo test`. So the old lookup in +/// `target//` read an artifact that nothing in the test's dependency graph ever refreshes: +/// whatever some earlier `cargo build` happened to leave there, from any commit, or nothing at all. +/// +/// Both failure modes of that are lies about durability, and the second is the dangerous one: +/// * NOTHING there -> the test used to `return` with a "skip:" line and report GREEN, which is +/// how `cargo test --workspace` on a fresh clone reported success with ZERO over-the-ABI +/// coverage of the ten task/call-log methods. +/// * STALE artifact -> a cdylib older than the ABI relay answers every write `Ok(())` and every +/// read empty, which is BYTE-FOR-BYTE the signature of the unrelayed-seam defect this file +/// exists to catch (that defect was real: `DynStore`'s `impl Store` overrode 24 methods, none +/// of them task methods, so `put_task` took the accept-and-keep-nothing trait default). RED on +/// a stale artifact is indistinguishable from RED on the real bug; and an artifact that happens +/// to be NEWER than a regression reports GREEN while the shipped ABI is broken. +/// +/// Same hazard, and the same reasoning, as the engine's `crates/busbar/Cargo.toml` dev-dependency on +/// `busbar-store-example-plugin`: put the cdylib in the graph so the test cannot judge a stale one. +/// Here the plugin's lib IS this package, so the graph edge already exists — what was missing was +/// looking at the artifact that edge produces. +/// +/// Panics rather than skipping. A missing cdylib under `cargo test` means the build graph changed +/// shape, and the only honest report is a failure, not a silent pass. +/// The newest mtime across every workspace crate's `src/` — "how fresh must a cdylib be to be the +/// one this source tree describes". +/// +/// Deliberately ONLY `src/**/*.rs` of each workspace member: editing a `tests/` file or a +/// `[dev-dependencies]` line recompiles the test binary but NOT the lib, so including those would +/// fail a perfectly current cdylib. +fn newest_source_mtime() -> std::time::SystemTime { + fn walk(dir: &std::path::Path, newest: &mut std::time::SystemTime) { + let Ok(rd) = std::fs::read_dir(dir) else { return }; + for e in rd.flatten() { + let p = e.path(); + if p.is_dir() { + walk(&p, newest); + } else if p.extension().is_some_and(|x| x == "rs") { + if let Ok(m) = e.metadata().and_then(|m| m.modified()) { + if m > *newest { + *newest = m; + } + } + } + } + } + let ws_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("the plugin crate always sits under the workspace root"); + let mut newest = std::time::SystemTime::UNIX_EPOCH; + for e in std::fs::read_dir(ws_root).into_iter().flatten().flatten() { + let src = e.path().join("src"); + if src.is_dir() { + walk(&src, &mut newest); + } } - candidate + newest +} + +fn plugin_path() -> PathBuf { + let exe = std::env::current_exe().expect("current_exe"); // .../target//deps/e2e- + let deps_dir = exe.parent().expect("the test binary always lives in deps/"); + let name = busbar_plugin_loader::plugin_library_filename("busbar_store_sqlite_plugin"); + let fresh = deps_dir.join(&name); + assert!( + fresh.exists(), + "the store-sqlite-plugin cdylib is not at {}, where cargo emits it for the same build that \ + produced this test binary. Refusing to fall back to target// (an artifact only \ + `cargo build` refreshes) or to skip: judging a stale cdylib is exactly how an unrelayed \ + plugin ABI reads as green.", + fresh.display() + ); + // FRESHNESS, ASSERTED — not assumed. Under `cargo test` the artifact above is rebuilt by the + // same graph that built this binary (proven: delete it, re-run, cargo re-emits it). But this + // test binary can also be executed DIRECTLY out of `deps/`, where nothing rebuilds anything, + // and a stale cdylib there produces empty reads — indistinguishable from the unrelayed-ABI + // defect. So compare it against the sources and fail with a message that says STALE ARTIFACT, + // explicitly NOT a durability verdict. + let built = std::fs::metadata(&fresh) + .and_then(|m| m.modified()) + .expect("cdylib mtime"); + let newest_src = newest_source_mtime(); + assert!( + built >= newest_src, + "STALE ARTIFACT — THIS IS NOT A DURABILITY FAILURE. {} predates this workspace's sources, \ + so it cannot answer for the code in the tree; a pre-change cdylib returns empty for every \ + read, which reads exactly like an unrelayed plugin ABI. Run `cargo build -p {}` (or just \ + `cargo test`, which rebuilds it) and re-run.", + fresh.display(), + "busbar-store-sqlite-plugin" + ); + fresh } /// Every `env:` secret-ref name a config text references, in first-seen order, de-duplicated. @@ -174,10 +258,7 @@ fn build_real_binaries() -> (PathBuf, PathBuf) { /// full close + reopen, using an independent connection that never touches the plugin/ABI/loader. #[test] fn load_and_exercise_sqlite_plugin_via_file_drop() { - let Some(so_path) = plugin_path() else { - eprintln!("skip: store-sqlite-plugin cdylib not built"); - return; - }; + let so_path = plugin_path(); let (busbar_bin, pack_bin) = build_real_binaries(); @@ -447,10 +528,7 @@ fn wait_for_admin_ready( /// sqlite }` and is the one whose admin API mints the key/credential that lands in the real file. #[test] fn install_sqlite_plugin_via_admin_api_and_verify_persistence() { - let Some(so_path) = plugin_path() else { - eprintln!("skip: store-sqlite-plugin cdylib not built"); - return; - }; + let so_path = plugin_path(); let (busbar_bin, pack_bin) = build_real_binaries(); let work = ScratchDir::create("admin-api-install"); @@ -684,10 +762,7 @@ fn install_sqlite_plugin_via_admin_api_and_verify_persistence() { /// the C ABI as a clean `Err`, never a panic or a silently-succeeded load. #[test] fn load_and_exercise_sqlite_plugin_bad_config_fails_over_abi() { - let Some(path) = plugin_path() else { - eprintln!("skip: store-sqlite-plugin cdylib not built (run cargo test/build first)"); - return; - }; + let path = plugin_path(); // Malformed JSON: the plugin's own `open()` config parsing must reject it, surfaced intact // across the ABI. @@ -722,3 +797,211 @@ fn load_and_exercise_sqlite_plugin_bad_config_fails_over_abi() { "a failed open must not have created the parent directory or file" ); } + +/// THE DURABILITY PROOF FOR THE TEN TASK / CALL-LOG METHODS, OVER THE REAL PLUGIN PATH. +/// +/// Every other test of these methods in this repo calls `SqliteStore` DIRECTLY, in-process, and none +/// of them can see the failure that actually matters in production. `busbar_api::Store` DEFAULTS all +/// ten of `put_task`/`get_task`/`list_tasks`/`purge_tasks_before`/`append_task_event`/ +/// `list_task_events`/`append_mcp_call`/`list_mcp_calls`/`list_mcp_call_principals`/ +/// `purge_mcp_calls_before` to accept-and-keep-nothing, so a plugin seam that does not RELAY them +/// silently substitutes those defaults: every write returns `Ok`, every read answers empty, and a +/// deployment running this backend as a plugin — which is the ONLY way it ever runs — loses every +/// in-flight A2A task and every tool-call record while reporting success. +/// +/// So this test goes through `busbar_plugin_loader::load_store`: a REAL `dlopen` of the packed +/// cdylib, the real C ABI, the real `DynStore`. It writes AT ARITY > 1 (three tasks across two +/// states, three events on one task and one on another, three call records for one principal and +/// one for a second), DROPS the handle — which unloads the library — then `dlopen`s AGAIN over the +/// same file and reads everything back. A single-row round trip would not distinguish a relayed +/// method from a lucky default; a multi-row one over a restart cannot be faked by either. +/// +/// EXPECT THIS TEST TO BE RED until the engine-side ABI relay for these ten methods is on the +/// busbar ref this repo builds against (`busbar-plugin-abi`'s `StoreRequest`/`StoreResponse` +/// variants, the SDK dispatch and the `DynStore` overrides). THAT IS THE POINT: red here is the +/// truthful report that durable tasks do not yet work through the only path that ships, and the +/// alternative — no coverage at all — is how the seam stayed silently broken. +#[test] +fn tasks_and_call_log_survive_an_unload_and_reload_over_the_real_plugin_abi() { + let path = plugin_path(); + let scratch = ScratchDir::create("abi-durable"); + let db_path = scratch.join("tasks.db"); + let cfg = serde_json::json!({ "db_path": db_path.to_str().unwrap() }).to_string(); + + let task = |id: &str, state: &str, updated_at: u64| TaskRow { + task_id: id.to_string(), + context_id: format!("ctx-{id}"), + principal: "vk_abi".to_string(), + direction: "inbound".to_string(), + state: state.to_string(), + agent_id: "planner".to_string(), + artifact_cursor: 7, + push_callback: "https://example.test/push".to_string(), + created_at: 1_000, + updated_at, + }; + let event = |task_id: &str, seq: u64, prev: &str, hash: &str| TaskEventRow { + task_id: task_id.to_string(), + seq, + ts: 1_000 + seq, + kind: "task.working".to_string(), + context_id: format!("ctx-{task_id}"), + principal: "vk_abi".to_string(), + agent_id: "planner".to_string(), + state: "working".to_string(), + request_id: format!("req-{seq}"), + prev_hash: prev.to_string(), + hash: hash.to_string(), + }; + let call = |principal: &str, seq: u64, prev: &str, hash: &str| McpCallRecord { + principal: principal.to_string(), + seq, + ts: 2_000 + seq, + server: "srv".to_string(), + tool: "srv_read_file".to_string(), + outcome: "dispatched".to_string(), + reason: String::new(), + tool_digest: format!("sha256:tool{seq}"), + pin_generation: 3, + request_id: format!("req-{seq}"), + prev_hash: prev.to_string(), + hash: hash.to_string(), + }; + + { + // BOOT 1 — a real dlopen of the cdylib; every call below crosses the C ABI. + let store = busbar_plugin_loader::load_store(&path, &cfg) + .expect("the sqlite plugin must load over the real ABI"); + for (id, state, updated) in [ + ("t_alpha", "working", 10_u64), + ("t_beta", "input-required", 20), + ("t_gamma", "completed", 30), + ] { + store.put_task(&task(id, state, updated)).expect("put_task"); + } + for (seq, prev, hash) in [(1_u64, "", "e1"), (2, "e1", "e2"), (3, "e2", "e3")] { + store + .append_task_event(&event("t_alpha", seq, prev, hash)) + .expect("append_task_event"); + } + store + .append_task_event(&event("t_beta", 1, "", "b1")) + .expect("append_task_event"); + for (seq, prev, hash) in [(1_u64, "", "h1"), (2, "h1", "h2"), (3, "h2", "h3")] { + store + .append_mcp_call(&call("vk_abi", seq, prev, hash)) + .expect("append_mcp_call"); + } + store + .append_mcp_call(&call("vk_other", 1, "", "o1")) + .expect("append_mcp_call"); + // Dropping the boxed store drops the loader's `Library` handle: the dylib is UNLOADED, so + // nothing this process still holds can be answering the reads below. + drop(store); + } + + // BOOT 2 — a second, independent dlopen over the same file. + let store = busbar_plugin_loader::load_store(&path, &cfg) + .expect("the sqlite plugin must load again over the real ABI"); + + let tasks = store.list_tasks().expect("list_tasks"); + assert_eq!( + tasks.len(), + 3, + "all three tasks must survive the unload/reload over the plugin ABI; got {} back, which is \ + the accept-and-keep-nothing shape of the trait default that an unrelayed seam substitutes", + tasks.len() + ); + let beta = store + .get_task("t_beta") + .expect("get_task") + .expect("the interrupted task must be readable by id after a reload"); + assert_eq!(beta.state, "input-required"); + assert_eq!( + beta.artifact_cursor, 7, + "the artifact cursor must round-trip" + ); + assert_eq!(beta.push_callback, "https://example.test/push"); + assert_eq!(beta.context_id, "ctx-t_beta"); + + let events = store.list_task_events("t_alpha").expect("list_task_events"); + assert_eq!( + events.iter().map(|e| e.seq).collect::>(), + vec![1, 2, 3], + "the per-task provenance chain must come back oldest-first and complete" + ); + for w in events.windows(2) { + assert_eq!( + w[1].prev_hash, w[0].hash, + "the chain must still link after the reload: seq {} carries prev_hash {:?} but seq {} \ + persisted hash {:?}", + w[1].seq, w[1].prev_hash, w[0].seq, w[0].hash + ); + } + assert_eq!( + store + .list_task_events("t_beta") + .expect("list_task_events") + .len(), + 1, + "one task's events must not leak into another's chain" + ); + + let calls = store.list_mcp_calls("vk_abi").expect("list_mcp_calls"); + assert_eq!( + calls.iter().map(|c| c.seq).collect::>(), + vec![1, 2, 3], + "the per-principal call chain must survive the reload in chain order" + ); + assert_eq!(calls[2].tool_digest, "sha256:tool3"); + assert_eq!(calls[2].request_id, "req-3"); + assert_eq!(calls[1].pin_generation, 3); + assert_eq!( + store + .list_mcp_calls("vk_other") + .expect("list_mcp_calls") + .len(), + 1, + "one principal's chain must not carry another's records" + ); + let mut principals = store + .list_mcp_call_principals() + .expect("list_mcp_call_principals"); + principals.sort(); + assert_eq!( + principals, + vec!["vk_abi".to_string(), "vk_other".to_string()], + "the boot enumeration must name every principal holding records, exactly once each" + ); + + // Retention crosses the ABI too, count and all — and both purges are checked for the number + // they ACTUALLY removed, because a relay that dropped the return value would read as 0. + assert_eq!( + store.purge_mcp_calls_before(2_002).expect("purge"), + 2, + "both records at ts 2001 go (one per principal); the one sitting exactly at the cutoff stays" + ); + assert_eq!( + store + .list_mcp_calls("vk_abi") + .expect("list_mcp_calls") + .len(), + 2 + ); + assert!(store + .list_mcp_calls("vk_other") + .expect("list_mcp_calls") + .is_empty()); + assert_eq!( + store.purge_tasks_before(25).expect("purge"), + 0, + "no TERMINAL task is older than the cutoff: t_alpha and t_beta are active and must never be \ + swept no matter how old" + ); + assert_eq!( + store.purge_tasks_before(31).expect("purge"), + 1, + "the one completed task at updated_at 30 is the only row retention may drop" + ); + assert_eq!(store.list_tasks().expect("list_tasks").len(), 2); +} diff --git a/store-sqlite/src/lib.rs b/store-sqlite/src/lib.rs index bd32f97..2fc0477 100644 --- a/store-sqlite/src/lib.rs +++ b/store-sqlite/src/lib.rs @@ -8,9 +8,9 @@ //! Depends only on the `busbar-api` contract (plus rusqlite), never on the engine. use busbar_api::{ - AuditRecord, CredentialMeta, CredentialSecret, MeteringDelta, MeteringRow, ModelTokens, - ScopeRef, SecretForm, Store, StoreError, StoreResult, TierTokens, UsageDelta, UsageLedger, - VirtualKey, + AuditRecord, CredentialMeta, CredentialSecret, McpCallRecord, MeteringDelta, MeteringRow, + ModelTokens, ScopeRef, SecretForm, Store, StoreError, StoreResult, TaskEventRow, TaskRow, + TierTokens, UsageDelta, UsageLedger, VirtualKey, }; use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior}; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -47,7 +47,23 @@ impl IntoStoreResult for Result { /// one-time backfill of `billable_requests` for any row where a v5-era write left it at 0 despite /// a nonzero `requests` (see the `version < 6` block in `migrate`, and /// `governance::state::hydrate_budgets` in busbarAI core for the boot-time bug this closes). -const SCHEMA_VERSION: i64 = 6; +/// +/// v7: the durable MCP TOOL-CALL LOG (`mcp_calls`). PURELY ADDITIVE and needs no backfill block — +/// the table is new, so `SCHEMA`'s own `CREATE TABLE IF NOT EXISTS` (executed unconditionally on +/// every open) is the entire migration. Nothing is dropped and no existing row is touched: a v6 +/// database crossing to v7 gains an empty table and keeps everything else, which is why there is no +/// `version < 7` arm in `migrate` to match the `version < 6` one. +/// +/// v8: the durable A2A TASK STORE (`tasks`, `task_events`). Additive on the same terms as v7 — two +/// new tables, no backfill, nothing dropped — so again no `version < 8` arm exists. +const SCHEMA_VERSION: i64 = 8; + +/// The task states that are TERMINAL, and therefore the only ones retention may drop. Named as a +/// closed set rather than derived by negation on purpose: an unrecognised state token — one a newer +/// engine emits and this build has never heard of — must read as NOT terminal, so a store compiled +/// before a state existed cannot delete a task it does not understand. The wrong half to guess on is +/// the deleting half. +const TERMINAL_TASK_STATES: [&str; 4] = ["completed", "failed", "canceled", "rejected"]; const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS store_meta ( @@ -179,6 +195,100 @@ CREATE TABLE IF NOT EXISTS audit_log ( ) STRICT; CREATE INDEX IF NOT EXISTS audit_resource_seq_idx ON audit_log (resource, seq); +-- The DURABLE MCP TOOL-CALL LOG. A DIFFERENT POPULATION from audit_log, kept in its own table on +-- purpose: audit_log is the low-rate admin MUTATION log whose engine-side working set is a bounded +-- ring, while a tool call is data-plane traffic at request rate. Pouring one into the other means a +-- busy afternoon of tool calls evicts every admin row from the ring, so the question of who changed +-- a registration becomes unanswerable exactly when an incident makes somebody ask. +-- +-- The chain is scoped to the PRINCIPAL, which is why (principal, seq) is the primary key and not a +-- global counter: a global chain would serialise every caller's tool calls behind one append, and +-- would make one caller's evidence unverifiable without possessing every other caller's rows. +-- +-- SHAPE: opaque `body` + only the columns a query actually needs. `principal` and `ts` are the index +-- columns (scoped read, and the retention sweep's age key). The CHAIN COLUMNS -- seq, prev_hash, +-- hash -- are REAL columns rather than being buried in `body`, because the engine establishes +-- durability by READING THE CHAIN BACK and verifying it; a digest reachable only by decoding an +-- opaque blob forces a deserialise per verify and cannot be constrained or indexed by the database. +-- `body` carries exactly the fields no query filters on. The store NEVER computes or recomputes a +-- digest -- it persists what it was handed, verbatim, and returns it verbatim. +CREATE TABLE IF NOT EXISTS mcp_calls ( + principal TEXT NOT NULL, + seq INTEGER NOT NULL, + ts INTEGER NOT NULL, + prev_hash TEXT NOT NULL, + hash TEXT NOT NULL, + body TEXT NOT NULL, + -- Carried now, written by nothing yet, and that is deliberate: adding a column to a populated + -- table later is a rewrite, whereas carrying it from the first migration is free. `version` is + -- the compare-and-swap slot an optimistic-concurrency write would test; `expires_at` is the + -- per-row sweep deadline. Retention today goes by `ts` (see purge_mcp_calls_before). + expires_at INTEGER, + version INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (principal, seq) +) STRICT; +-- The retention sweep's access path: purge_mcp_calls_before deletes by `ts` across every principal. +CREATE INDEX IF NOT EXISTS mcp_calls_ts_idx ON mcp_calls (ts); + +-- THE DURABLE A2A TASK STORE. An A2A task spans turns, can sit interrupted waiting on a human, and +-- can outlive the process that started it, so an in-memory task table loses every in-flight task on +-- restart -- the difference between a resume that is real and one that is nominal. +-- +-- SHAPE: every field is a REAL COLUMN, unlike mcp_calls's opaque `body`, and the difference is not +-- inconsistency. A call record carries open-ended narration that no query ever filters on, so it +-- pays for a blob and saves the columns. A task row is a small fixed state record where every field +-- is part of what a resume reads back, `state` and `updated_at` are the retention sweep's own +-- predicates, and there is no field left over to make opaque. A blob here would buy a deserialise +-- per read and nothing else. +CREATE TABLE IF NOT EXISTS tasks ( + task_id TEXT NOT NULL PRIMARY KEY, + context_id TEXT NOT NULL, + principal TEXT NOT NULL, + direction TEXT NOT NULL, + state TEXT NOT NULL, + agent_id TEXT NOT NULL, + artifact_cursor INTEGER NOT NULL, + push_callback TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +) STRICT; +-- The retention sweep's access path (purge_tasks_before filters on state + updated_at), and the +-- boot rehydrate's (which reads every row and partitions on state). +CREATE INDEX IF NOT EXISTS tasks_state_updated_idx ON tasks (state, updated_at); + +-- PER-TASK PROVENANCE, hash-chained WITHIN a task. Per-task rather than one global chain because +-- tasks are concurrent and long-lived: a global chain would serialise every task transition behind +-- one append and would make one task's provenance unverifiable without possessing every other +-- tenant's events. +-- +-- The chain columns -- seq, prev_hash, hash -- are REAL columns for the same reason they are in +-- mcp_calls: the engine establishes durability by reading the chain back and verifying it, and a +-- digest reachable only by decoding a blob can be neither constrained nor indexed. This store NEVER +-- computes or recomputes a digest; it persists what it was handed and returns it verbatim. +-- +-- The cascade on task deletion is load-bearing: `purge_tasks_before` is the ONLY retention method +-- the contract gives this data, so a purge that left the events behind would leave `task_events` +-- with no bound anywhere in the trait. It is a TRIGGER rather than a foreign key, deliberately. A +-- real FK would also impose an ORDER on the writes -- no event could be appended before its task row +-- existed -- and the engine is under no such obligation: a `task.submitted` event and the first +-- `put_task` are two independent write-throughs. A trigger gives the cascade without inventing an +-- ordering constraint the contract never stated, and unlike `foreign_keys` it is not a per-connection +-- pragma that defaults OFF in the sqlite3 CLI. +CREATE TABLE IF NOT EXISTS task_events ( + task_id TEXT NOT NULL, + seq INTEGER NOT NULL, + ts INTEGER NOT NULL, + kind TEXT NOT NULL, + context_id TEXT NOT NULL, + principal TEXT NOT NULL, + agent_id TEXT NOT NULL, + state TEXT NOT NULL, + request_id TEXT NOT NULL, + prev_hash TEXT NOT NULL, + hash TEXT NOT NULL, + PRIMARY KEY (task_id, seq) +) STRICT; + -- Pragma-independent integrity triggers: survive an operator opening the file with the sqlite3 CLI -- (where foreign_keys defaults OFF) or a build with SQLITE_OMIT_FOREIGN_KEY. CREATE TRIGGER IF NOT EXISTS keys_guard_hard_delete BEFORE DELETE ON keys FOR EACH ROW @@ -190,6 +300,18 @@ CREATE TRIGGER IF NOT EXISTS audit_log_no_update BEFORE UPDATE ON audit_log BEGIN SELECT RAISE(ABORT, 'audit_log is append-only'); END; CREATE TRIGGER IF NOT EXISTS audit_log_no_delete BEFORE DELETE ON audit_log BEGIN SELECT RAISE(ABORT, 'audit_log is append-only'); END; +-- mcp_calls is append-only in the sense that MATTERS: a persisted record is never REWRITTEN, so a +-- stored digest can never be quietly restated. There is deliberately NO no-delete counterpart (as +-- audit_log has), because this table has a retention sweep and a blanket delete guard would make +-- purge_mcp_calls_before impossible to honour -- bounded retention and never-rewritten are +-- different properties, and only the second one is an integrity claim. +CREATE TRIGGER IF NOT EXISTS mcp_calls_no_update BEFORE UPDATE ON mcp_calls + BEGIN SELECT RAISE(ABORT, 'mcp_calls is append-only; a persisted call record is never rewritten'); END; +-- task_events follows its task out of the database. There is deliberately no no-update trigger to +-- match mcp_calls's: the task-event contract says a store MUST UPSERT on (task_id, seq) -- a +-- write-through is idempotent on replay -- so forbidding the rewrite here would forbid the contract. +CREATE TRIGGER IF NOT EXISTS tasks_cascade_events AFTER DELETE ON tasks FOR EACH ROW + BEGIN DELETE FROM task_events WHERE task_id = OLD.task_id; END; "; /// Apply the fixed pragma set to a connection, in the documented order (`busy_timeout` FIRST: @@ -1262,6 +1384,339 @@ impl Store for SqliteStore { let rows = stmt.query_map([], |r| r.get::<_, String>(0)).store()?; rows.collect::, _>>().store() } + + fn append_mcp_call(&self, record: &McpCallRecord) -> StoreResult<()> { + let body = mcp_call_body(record); + let mut conn = self.lock_writer(); + // IMMEDIATE, and the existence check shares the transaction with the insert: the two must be + // one atomic step or a concurrent writer could land between them and turn a fork into a + // silent accept. + let tx = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .store()?; + let existing: Option<(i64, String, String, String)> = tx + .query_row( + "SELECT ts, prev_hash, hash, body FROM mcp_calls WHERE principal = ?1 AND seq = ?2", + params![record.principal, record.seq as i64], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)), + ) + .optional() + .store()?; + if let Some((ts, prev_hash, hash, stored_body)) = existing { + // BYTE-IDENTICAL is the at-least-once retry and is success. DIFFERENT is a forked or + // tampered log and is an error: overwriting would destroy exactly the case worth + // reporting, and this store never restates a digest it was handed. + if ts == record.ts as i64 + && prev_hash == record.prev_hash + && hash == record.hash + && stored_body == body + { + return Ok(()); + } + // The message names the sequence and nothing else — it must not echo stored content + // (or caller content) back to whoever provoked it. + return Err(StoreError(format!( + "mcp call log fork: a different record is already persisted at sequence {} for this principal", + record.seq + ))); + } + tx.execute( + "INSERT INTO mcp_calls (principal, seq, ts, prev_hash, hash, body) \ + VALUES (?1,?2,?3,?4,?5,?6)", + params![ + record.principal, + record.seq as i64, + record.ts as i64, + record.prev_hash, + record.hash, + body, + ], + ) + .store()?; + tx.commit().store()?; + Ok(()) + } + + fn list_mcp_calls(&self, principal: &str) -> StoreResult> { + let conn = self.lock_reader(); + let mut stmt = conn + .prepare( + "SELECT principal, seq, ts, prev_hash, hash, body FROM mcp_calls \ + WHERE principal = ?1 ORDER BY seq", + ) + .store()?; + let rows = stmt.query_map([principal], row_to_mcp_call).store()?; + rows.collect::, _>>().store() + } + + fn list_mcp_call_principals(&self) -> StoreResult> { + let conn = self.lock_reader(); + let mut stmt = conn + .prepare("SELECT DISTINCT principal FROM mcp_calls ORDER BY principal") + .store()?; + let rows = stmt.query_map([], |r| r.get::<_, String>(0)).store()?; + rows.collect::, _>>().store() + } + + fn purge_mcp_calls_before(&self, before: u64) -> StoreResult { + // STRICTLY less-than, matching the contract's wording: a row exactly at the cutoff is kept. + // `execute` returns the rows actually removed, so the count reported is one performed. + let removed = self + .lock_writer() + .execute( + "DELETE FROM mcp_calls WHERE ts < ?1", + params![i64::try_from(before).unwrap_or(i64::MAX)], + ) + .store()?; + Ok(removed as u64) + } + + fn put_task(&self, task: &TaskRow) -> StoreResult<()> { + // Refused rather than mangled, for the reason `append_audit` refuses its own out-of-range + // seq: `as i64` wraps a `u64` past `i64::MAX` negative and the read clamps it back, so the + // row read back would not be the row written — and here the value that silently changes is + // the ARTIFACT CURSOR, i.e. how much of a stream has been relayed. A wrapped cursor either + // replays delivered artifacts or skips undelivered ones, and does it without an error ever + // having been reported. + let cursor = as_storable_i64("put_task", "artifact_cursor", task.artifact_cursor)?; + let created = as_storable_i64("put_task", "created_at", task.created_at)?; + let updated = as_storable_i64("put_task", "updated_at", task.updated_at)?; + // FULL sync. The trait's own framing is that this is the difference between a resume that + // is real and one that is nominal, and a task state transition acknowledged to a caller and + // then lost to a power cut a second later is exactly the failure the durability exists to + // stop — the same reasoning that escalates `append_audit` and the revocation paths. + let mut conn = self.lock_writer(); + with_full_sync(&mut conn, |conn| { + // UPSERT BY task_id: the engine writes through on EVERY state transition, so a second + // write for one task must replace the row, never append a second one for the same id. + conn.execute( + "INSERT INTO tasks (task_id, context_id, principal, direction, state, agent_id, \ + artifact_cursor, push_callback, created_at, updated_at) \ + VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10) \ + ON CONFLICT(task_id) DO UPDATE SET \ + context_id=excluded.context_id, principal=excluded.principal, \ + direction=excluded.direction, state=excluded.state, agent_id=excluded.agent_id, \ + artifact_cursor=excluded.artifact_cursor, \ + push_callback=excluded.push_callback, created_at=excluded.created_at, \ + updated_at=excluded.updated_at", + params![ + task.task_id, + task.context_id, + task.principal, + task.direction, + task.state, + task.agent_id, + cursor, + task.push_callback, + created, + updated, + ], + ) + .store()?; + Ok(()) + }) + } + + fn get_task(&self, task_id: &str) -> StoreResult> { + // No principal filter, deliberately: the contract puts the caller-scoping check engine-side, + // because an authorization check living in the backend is one an unauthorized reader + // bypasses by configuring a different backend. + let conn = self.lock_reader(); + conn.query_row( + "SELECT task_id, context_id, principal, direction, state, agent_id, artifact_cursor, \ + push_callback, created_at, updated_at FROM tasks WHERE task_id = ?1", + [task_id], + row_to_task, + ) + .optional() + .store() + } + + fn list_tasks(&self) -> StoreResult> { + // UNFILTERED, terminal rows included. The boot rehydrate wants the active rows, the + // retention sweep wants the terminal ones and the scoped listing wants one principal's; a + // store that pre-filtered for any one of those would break the other two. + let conn = self.lock_reader(); + let mut stmt = conn + .prepare( + "SELECT task_id, context_id, principal, direction, state, agent_id, \ + artifact_cursor, push_callback, created_at, updated_at FROM tasks ORDER BY task_id", + ) + .store()?; + let rows = stmt.query_map([], row_to_task).store()?; + rows.collect::, _>>().store() + } + + fn purge_tasks_before(&self, before: u64) -> StoreResult { + // TERMINAL ONLY, and strictly older than the cutoff. An interrupted task waiting on a human + // is exactly the row that legitimately sits still for a long time; compacting it is losing + // the work, not reclaiming space. The IN list is the closed terminal set, so a state token + // this build does not recognise is never dropped. + // + // `execute` returns the rows actually removed, so the count reported is one performed. The + // task's provenance chain goes with it via the tasks_cascade_events trigger. + let placeholders = (1..=TERMINAL_TASK_STATES.len()) + .map(|i| format!("?{}", i + 1)) + .collect::>() + .join(","); + let sql = format!("DELETE FROM tasks WHERE updated_at < ?1 AND state IN ({placeholders})"); + let mut args: Vec> = + vec![Box::new(i64::try_from(before).unwrap_or(i64::MAX))]; + for s in TERMINAL_TASK_STATES { + args.push(Box::new(s)); + } + let removed = self + .lock_writer() + .execute(&sql, rusqlite::params_from_iter(args.iter())) + .store()?; + Ok(removed as u64) + } + + fn append_task_event(&self, event: &TaskEventRow) -> StoreResult<()> { + let seq = as_storable_i64("append_task_event", "seq", event.seq)?; + let ts = as_storable_i64("append_task_event", "ts", event.ts)?; + // UPSERT ON (task_id, seq), and this is where the task-event contract genuinely DIFFERS from + // `append_mcp_call`'s: that one treats an occupied slot holding a different record as a fork + // and refuses it, while this one is specified to upsert so that the engine's write-through + // is idempotent on replay — "rejecting or duplicating a replayed seq breaks the chain the + // engine will verify on read". Copying the call log's fork check here would be wrong in a + // way that looks right. + // + // FULL sync for the same reason `put_task` uses it: this is the tamper-evidence record of a + // transition, and a chain with a hole where a crash landed is a chain that fails to verify. + let mut conn = self.lock_writer(); + with_full_sync(&mut conn, |conn| { + conn.execute( + "INSERT INTO task_events (task_id, seq, ts, kind, context_id, principal, agent_id, \ + state, request_id, prev_hash, hash) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11) \ + ON CONFLICT(task_id, seq) DO UPDATE SET \ + ts=excluded.ts, kind=excluded.kind, context_id=excluded.context_id, \ + principal=excluded.principal, agent_id=excluded.agent_id, state=excluded.state, \ + request_id=excluded.request_id, prev_hash=excluded.prev_hash, \ + hash=excluded.hash", + params![ + event.task_id, + seq, + ts, + event.kind, + event.context_id, + event.principal, + event.agent_id, + event.state, + event.request_id, + event.prev_hash, + event.hash, + ], + ) + .store()?; + Ok(()) + }) + } + + fn list_task_events(&self, task_id: &str) -> StoreResult> { + // Oldest-first by seq — the order the engine's chain verifier reads, and the scope is the + // one task, because the chain is per-task. + let conn = self.lock_reader(); + let mut stmt = conn + .prepare( + "SELECT task_id, seq, ts, kind, context_id, principal, agent_id, state, \ + request_id, prev_hash, hash FROM task_events WHERE task_id = ?1 ORDER BY seq", + ) + .store()?; + let rows = stmt.query_map([task_id], row_to_task_event).store()?; + rows.collect::, _>>().store() + } +} + +/// Reject a `u64` that SQLite's signed 64-bit integer cannot hold, naming the method and the field. +/// `as i64` would wrap it negative and the read would clamp it back to something else again, so the +/// row read back would not be the row written — and nothing would ever have reported an error. The +/// same guard `append_audit` applies to its own `seq`/`ts`, factored out because the task store has +/// five such fields across two methods. +fn as_storable_i64(method: &str, field: &str, v: u64) -> StoreResult { + i64::try_from(v).map_err(|_| { + StoreError(format!( + "{method}: {field} {v} exceeds the storable range (i64::MAX); refusing to store a row \ + that would not read back as itself" + )) + }) +} + +fn row_to_task(r: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(TaskRow { + task_id: r.get(0)?, + context_id: r.get(1)?, + principal: r.get(2)?, + direction: r.get(3)?, + state: r.get(4)?, + agent_id: r.get(5)?, + artifact_cursor: r.get::<_, i64>(6)?.max(0) as u64, + push_callback: r.get(7)?, + created_at: r.get::<_, i64>(8)?.max(0) as u64, + updated_at: r.get::<_, i64>(9)?.max(0) as u64, + }) +} + +fn row_to_task_event(r: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(TaskEventRow { + task_id: r.get(0)?, + seq: r.get::<_, i64>(1)?.max(0) as u64, + ts: r.get::<_, i64>(2)?.max(0) as u64, + kind: r.get(3)?, + context_id: r.get(4)?, + principal: r.get(5)?, + agent_id: r.get(6)?, + state: r.get(7)?, + request_id: r.get(8)?, + prev_hash: r.get(9)?, + hash: r.get(10)?, + }) +} + +/// The non-indexed payload of a call record, as stored in `mcp_calls.body`. `principal`, `seq`, +/// `ts`, `prev_hash` and `hash` are deliberately NOT duplicated here: they are real columns, and a +/// value stored in two places is a value that can disagree with itself. `serde_json`'s object keys +/// are ordered, so this encoding is deterministic — which is what makes the byte-comparison in +/// `append_mcp_call`'s replay check meaningful. +fn mcp_call_body(record: &McpCallRecord) -> String { + serde_json::json!({ + "server": record.server, + "tool": record.tool, + "outcome": record.outcome, + "reason": record.reason, + "tool_digest": record.tool_digest, + "pin_generation": record.pin_generation, + "request_id": record.request_id, + }) + .to_string() +} + +/// Rebuild a record from its columns plus its opaque body. The CHAIN comes from the columns, which +/// is the point of their being columns: what the engine verifies is what the database holds in a +/// field it can constrain, not a value recovered by decoding a blob. +fn row_to_mcp_call(r: &rusqlite::Row<'_>) -> rusqlite::Result { + let body: String = r.get(5)?; + let v: serde_json::Value = serde_json::from_str(&body).map_err(|e| { + rusqlite::Error::FromSqlConversionFailure(5, rusqlite::types::Type::Text, Box::new(e)) + })?; + let s = |k: &str| v.get(k).and_then(|x| x.as_str()).unwrap_or("").to_string(); + Ok(McpCallRecord { + principal: r.get(0)?, + seq: r.get::<_, i64>(1)? as u64, + ts: r.get::<_, i64>(2)? as u64, + prev_hash: r.get(3)?, + hash: r.get(4)?, + server: s("server"), + tool: s("tool"), + outcome: s("outcome"), + reason: s("reason"), + tool_digest: s("tool_digest"), + pin_generation: v + .get("pin_generation") + .and_then(|x| x.as_u64()) + .unwrap_or(0), + request_id: s("request_id"), + }) } fn put_credential_inner( diff --git a/store-sqlite/src/tests.rs b/store-sqlite/src/tests.rs index 033f74f..15ff37e 100644 --- a/store-sqlite/src/tests.rs +++ b/store-sqlite/src/tests.rs @@ -2,7 +2,10 @@ // Copyright (C) 2026 Busbar Inc and contributors use super::*; -use busbar_api::{AuditRecord, ModelTokensDelta, Store, TierTokensDelta, VirtualKey}; +use busbar_api::{ + AuditRecord, McpCallRecord, ModelTokensDelta, Store, TaskEventRow, TaskRow, TierTokensDelta, + VirtualKey, +}; use rusqlite::TransactionBehavior; fn sample_key(id: &str, generation: &str) -> VirtualKey { @@ -1201,3 +1204,661 @@ fn append_audit_refuses_a_seq_it_cannot_store_faithfully() { s.append_audit(&rec) .expect("an identical retry at the boundary must not read as a forked chain"); } +// ── THE DURABLE MCP TOOL-CALL LOG ──────────────────────────────────────────────────────────── +// +// The property under test is not "the write returned Ok" — the trait's default `append_mcp_call` +// returns `Ok(())` and keeps nothing, so a write's return value is worthless as evidence of +// durability. The only honest way to know a deployment has durable call evidence is to READ IT +// BACK, and the only honest way to know it survives a deploy is to read it back THROUGH A RESTART. + +fn sample_call(principal: &str, seq: u64, ts: u64, prev_hash: &str, hash: &str) -> McpCallRecord { + McpCallRecord { + principal: principal.to_string(), + seq, + ts, + server: "srv".to_string(), + tool: "srv_read_file".to_string(), + outcome: "dispatched".to_string(), + reason: String::new(), + tool_digest: format!("sha256:tool{seq}"), + pin_generation: 3, + request_id: format!("req-{seq}"), + prev_hash: prev_hash.to_string(), + hash: hash.to_string(), + } +} + +/// THE TEST THAT MATTERS. A unit test against a live handle proves nothing here: it cannot +/// distinguish a backend that wrote to disk from one that kept the rows in a HashMap behind the +/// same trait. So this drops the store entirely — closing every SQLite connection and its WAL — +/// reopens the same FILE, and verifies the per-principal hash chain still links from the bytes that +/// came back off disk. +#[test] +fn an_mcp_call_chain_survives_dropping_the_store_and_reopening_the_file() { + let dir = tempdir(); + let file = dir.join("calls.db"); + let path = file.to_str().unwrap().to_string(); + + // Write a 3-long chain, then let every connection close. + { + let s = SqliteStore::open(&path, 5000).unwrap(); + s.append_mcp_call(&sample_call("vk_a", 1, 100, "", "h1")) + .unwrap(); + s.append_mcp_call(&sample_call("vk_a", 2, 200, "h1", "h2")) + .unwrap(); + s.append_mcp_call(&sample_call("vk_a", 3, 300, "h2", "h3")) + .unwrap(); + drop(s); + } + + // A genuinely new store over the same file — nothing carried over in memory. + let reopened = SqliteStore::open(&path, 5000).unwrap(); + let got = reopened.list_mcp_calls("vk_a").unwrap(); + + assert_eq!( + got.len(), + 3, + "the call log must survive a restart; got {} records back after reopening the file, which \ + is the accept-and-keep-nothing behaviour this backend exists to replace", + got.len() + ); + + // The chain must LINK, read back off disk — not merely be non-empty. + assert_eq!( + got[0].prev_hash, "", + "seq 1 opens the chain with an empty prev_hash" + ); + for w in got.windows(2) { + assert_eq!( + w[1].prev_hash, w[0].hash, + "the per-principal chain must still link after a restart: seq {} carries prev_hash {:?} \ + but seq {} persisted hash {:?}", + w[1].seq, w[1].prev_hash, w[0].seq, w[0].hash + ); + } + // Ordering is by seq, and the non-indexed payload must round-trip verbatim too. + assert_eq!(got.iter().map(|r| r.seq).collect::>(), vec![1, 2, 3]); + assert_eq!(got[2].tool_digest, "sha256:tool3"); + assert_eq!(got[2].request_id, "req-3"); + assert_eq!(got[1].tool, "srv_read_file"); + assert_eq!(got[1].pin_generation, 3); + + let _ = std::fs::remove_dir_all(&dir); +} + +/// The boot enumeration: a restart has to resume a chain for a principal this process has not yet +/// seen, so the store must be able to name every principal holding records — across a restart. +#[test] +fn mcp_call_principals_are_enumerable_after_a_restart() { + let dir = tempdir(); + let file = dir.join("principals.db"); + let path = file.to_str().unwrap().to_string(); + { + let s = SqliteStore::open(&path, 5000).unwrap(); + s.append_mcp_call(&sample_call("vk_a", 1, 100, "", "a1")) + .unwrap(); + s.append_mcp_call(&sample_call("vk_b", 1, 100, "", "b1")) + .unwrap(); + s.append_mcp_call(&sample_call("vk_a", 2, 101, "a1", "a2")) + .unwrap(); + drop(s); + } + let reopened = SqliteStore::open(&path, 5000).unwrap(); + let mut principals = reopened.list_mcp_call_principals().unwrap(); + principals.sort(); + assert_eq!( + principals, + vec!["vk_a".to_string(), "vk_b".to_string()], + "every principal holding records must be enumerable after a restart, exactly once each" + ); + // A scoped read returns only its own principal's chain — the chain scope is the principal. + assert_eq!(reopened.list_mcp_calls("vk_a").unwrap().len(), 2); + assert_eq!(reopened.list_mcp_calls("vk_b").unwrap().len(), 1); + assert!( + reopened + .list_mcp_calls("vk_nonexistent") + .unwrap() + .is_empty(), + "a principal with no records reads back empty, not an error" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Retention must ACTUALLY DELETE and report a real count — a purge that returns a number it did +/// not perform is worse than one that reports nothing purged. +#[test] +fn purge_mcp_calls_before_deletes_and_returns_a_real_count() { + let s = SqliteStore::open_in_memory().unwrap(); + s.append_mcp_call(&sample_call("vk_a", 1, 100, "", "h1")) + .unwrap(); + s.append_mcp_call(&sample_call("vk_a", 2, 200, "h1", "h2")) + .unwrap(); + s.append_mcp_call(&sample_call("vk_a", 3, 300, "h2", "h3")) + .unwrap(); + s.append_mcp_call(&sample_call("vk_b", 1, 150, "", "b1")) + .unwrap(); + + // Strictly older than `before`, across every principal. ts=300 and ts=200 stay. + let purged = s.purge_mcp_calls_before(200).unwrap(); + assert_eq!( + purged, 2, + "purge must return the number of rows it actually removed (ts=100 and ts=150), not a guess" + ); + assert_eq!( + s.list_mcp_calls("vk_a") + .unwrap() + .iter() + .map(|r| r.seq) + .collect::>(), + vec![2, 3], + "the rows at or after the cutoff must remain" + ); + assert!( + s.list_mcp_calls("vk_b").unwrap().is_empty(), + "a principal whose every row aged out reads back empty" + ); + // `before` is STRICTLY less-than: a row exactly at the cutoff is kept. + assert_eq!( + s.purge_mcp_calls_before(200).unwrap(), + 0, + "re-running the same purge removes nothing; ts=200 sits exactly at the cutoff and is kept" + ); + // And the count is real: purging past everything clears the rest. + assert_eq!(s.purge_mcp_calls_before(1_000).unwrap(), 2); + assert!(s.list_mcp_calls("vk_a").unwrap().is_empty()); +} + +/// A record arriving on a `(principal, seq)` that already has one is settled the way the contract +/// settles it: BYTE-IDENTICAL is the retry and succeeds; DIFFERENT is a forked or tampered log and +/// is an error. Overwriting would destroy the second case instead of reporting it. +#[test] +fn a_replayed_mcp_call_is_idempotent_but_a_forked_one_is_refused() { + let s = SqliteStore::open_in_memory().unwrap(); + let rec = sample_call("vk_a", 1, 100, "", "h1"); + s.append_mcp_call(&rec).unwrap(); + + s.append_mcp_call(&rec) + .expect("an identical replay is the at-least-once retry and must succeed"); + assert_eq!( + s.list_mcp_calls("vk_a").unwrap().len(), + 1, + "a replay must not duplicate the row" + ); + + // Same (principal, seq), different digest — the fork case. + let forked = sample_call("vk_a", 1, 100, "", "DIFFERENT"); + let err = s + .append_mcp_call(&forked) + .expect_err("a different record at an occupied (principal, seq) is a fork and must error"); + assert!( + !format!("{err}").contains("DIFFERENT"), + "the error must not echo stored content back" + ); + assert_eq!( + s.list_mcp_calls("vk_a").unwrap()[0].hash, + "h1", + "the refused fork must not have overwritten the record already on record" + ); + + // A differing non-indexed payload field is a fork too, not a silent accept. + let mut tampered = sample_call("vk_a", 1, 100, "", "h1"); + tampered.tool = "srv_other_tool".to_string(); + s.append_mcp_call(&tampered) + .expect_err("a payload that differs under an identical digest is a fork and must error"); +} + +/// The v6 -> v7 crossing is additive: a real v6 database gains `mcp_calls` and keeps every row it +/// already had. Regression cover for the legacy-drop path reaching a live database. +#[test] +fn migrate_v6_to_v7_adds_the_call_log_without_wiping_data() { + let dir = tempdir(); + let file = dir.join("v6.db"); + { + let conn = Connection::open(&file).unwrap(); + conn.execute_batch(SCHEMA).unwrap(); + conn.execute("DROP TABLE mcp_calls", []).unwrap(); + conn.execute( + "INSERT INTO keys (id, name, key_group, allowed_pools, labels, enabled, \ + generation_hash, created_at, updated_at, expires_at, deleted_at, revision) \ + VALUES ('vk_v6', 'n', NULL, NULL, '{}', 1, 'g1', 0, 0, NULL, NULL, 0)", + [], + ) + .unwrap(); + conn.pragma_update(None, "user_version", 6i64).unwrap(); + } + let s = SqliteStore::open(file.to_str().unwrap(), 5000) + .expect("a v6 database must migrate additively to v7"); + assert!( + s.get_key("vk_v6").unwrap().is_some(), + "a real v6 key must survive the v6->v7 crossing" + ); + s.append_mcp_call(&sample_call("vk_v6", 1, 10, "", "h1")) + .expect("the newly created mcp_calls table must be writable after the migration"); + assert_eq!(s.list_mcp_calls("vk_v6").unwrap().len(), 1); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A persisted record is never REWRITTEN. Enforced by a trigger so it survives an operator opening +/// the file with the sqlite3 CLI, not merely by the write path being careful. +#[test] +fn mcp_calls_rejects_a_direct_update_but_allows_the_retention_delete() { + let s = SqliteStore::open_in_memory().unwrap(); + s.append_mcp_call(&sample_call("vk_a", 1, 100, "", "h1")) + .unwrap(); + let err = s + .lock_writer() + .execute( + "UPDATE mcp_calls SET hash = 'forged' WHERE principal = 'vk_a'", + [], + ) + .expect_err("a direct UPDATE must be refused by the append-only trigger"); + assert!(format!("{err}").contains("append-only")); + // DELETE is deliberately NOT guarded — retention has to be able to do its job. + s.lock_writer() + .execute("DELETE FROM mcp_calls WHERE principal = 'vk_a'", []) + .expect("retention must remain possible; only rewriting is forbidden"); +} + +// ── THE DURABLE A2A TASK STORE ─────────────────────────────────────────────────────────────── +// +// A2A is async by design: a task spans turns, can sit interrupted waiting on a human, and can +// outlive the process that started it. So the property under test is not "put_task returned Ok" — +// the trait's default `put_task` returns `Ok(())` and keeps nothing, and `get_task` answers `None` +// for everything, which is a backend that accepts every in-flight task and loses all of them on the +// next deploy. The only honest proof is to READ THE TASK BACK THROUGH A RESTART. + +fn sample_task(task_id: &str, state: &str, updated_at: u64) -> TaskRow { + TaskRow { + task_id: task_id.to_string(), + context_id: format!("ctx-{task_id}"), + principal: "vk_a".to_string(), + direction: "inbound".to_string(), + state: state.to_string(), + agent_id: "planner".to_string(), + artifact_cursor: 7, + push_callback: "https://example.test/push".to_string(), + created_at: 100, + updated_at, + } +} + +fn sample_event(task_id: &str, seq: u64, kind: &str, prev_hash: &str, hash: &str) -> TaskEventRow { + TaskEventRow { + task_id: task_id.to_string(), + seq, + // Saturating: the out-of-range test deliberately passes `u64::MAX` as `seq`, and a helper + // that panicked on its own arithmetic would hide the behaviour under test. + ts: seq.saturating_add(100), + kind: kind.to_string(), + context_id: format!("ctx-{task_id}"), + principal: "vk_a".to_string(), + agent_id: "planner".to_string(), + state: "working".to_string(), + request_id: format!("req-{seq}"), + prev_hash: prev_hash.to_string(), + hash: hash.to_string(), + } +} + +/// THE TEST THAT MATTERS, and it is deliberately not a unit test against a live handle: a live +/// handle cannot tell a backend that wrote to disk from one keeping a HashMap behind the same trait, +/// and it cannot tell either of those from the trait's accept-and-keep-nothing defaults if the +/// defaults happen to be exercised through the same handle that "wrote". So this DROPS the store — +/// closing every SQLite connection and its WAL — reopens the same FILE, and reads the task back off +/// disk. Against the unimplemented state it fails on the very first assertion. +#[test] +fn an_in_flight_task_survives_dropping_the_store_and_reopening_the_file() { + let dir = tempdir(); + let file = dir.join("tasks.db"); + let path = file.to_str().unwrap().to_string(); + + { + let s = SqliteStore::open(&path, 5000).unwrap(); + s.put_task(&sample_task("t-1", "working", 200)).unwrap(); + // The write-through on a state transition REPLACES the row rather than appending a second + // one — an interrupted task waiting on a human is what a restart has to find. + let mut interrupted = sample_task("t-1", "input-required", 300); + interrupted.artifact_cursor = 12; + s.put_task(&interrupted).unwrap(); + s.put_task(&sample_task("t-2", "submitted", 210)).unwrap(); + drop(s); + } + + let reopened = SqliteStore::open(&path, 5000).unwrap(); + let got = reopened.get_task("t-1").unwrap().expect( + "an in-flight task must survive a restart; got None back after reopening the file, \ + which is the accept-and-keep-nothing default this backend exists to replace", + ); + + // Every field a resume reads has to come back verbatim — not merely a row with the right id. + assert_eq!(got.state, "input-required", "the LAST state must win"); + assert_eq!( + got.artifact_cursor, 12, + "the artifact cursor is where a resubscribe resumes; a stale one replays or loses the gap" + ); + assert_eq!( + got.context_id, "ctx-t-1", + "the resume key is the context id" + ); + assert_eq!(got.principal, "vk_a"); + assert_eq!(got.direction, "inbound"); + assert_eq!(got.agent_id, "planner"); + assert_eq!(got.push_callback, "https://example.test/push"); + assert_eq!(got.created_at, 100); + assert_eq!(got.updated_at, 300); + + // UPSERT, not append: two writes for one task_id leave ONE row. + let mut all = reopened.list_tasks().unwrap(); + all.sort_by(|a, b| a.task_id.cmp(&b.task_id)); + assert_eq!( + all.iter().map(|t| t.task_id.as_str()).collect::>(), + vec!["t-1", "t-2"], + "put_task upserts by task_id; a second write for the same id must replace, never append" + ); + + assert!( + reopened.get_task("t-nonexistent").unwrap().is_none(), + "an unknown task id reads back None, not an error" + ); + + let _ = std::fs::remove_dir_all(&dir); +} + +/// `list_tasks` is deliberately UNFILTERED. The boot rehydrate wants the active rows, the retention +/// sweep wants the terminal ones and the scoped listing wants one principal's; a store that +/// pre-filtered for any one of those would break the other two. Pinned across a restart because the +/// boot rehydrate is precisely the caller that only ever sees the post-restart answer. +#[test] +fn list_tasks_returns_every_row_including_terminal_ones_after_a_restart() { + let dir = tempdir(); + let file = dir.join("list.db"); + let path = file.to_str().unwrap().to_string(); + { + let s = SqliteStore::open(&path, 5000).unwrap(); + s.put_task(&sample_task("t-active", "working", 200)) + .unwrap(); + s.put_task(&sample_task("t-waiting", "input-required", 201)) + .unwrap(); + s.put_task(&sample_task("t-done", "completed", 202)) + .unwrap(); + s.put_task(&sample_task("t-failed", "failed", 203)).unwrap(); + drop(s); + } + let reopened = SqliteStore::open(&path, 5000).unwrap(); + let mut ids = reopened + .list_tasks() + .unwrap() + .into_iter() + .map(|t| t.task_id) + .collect::>(); + ids.sort(); + assert_eq!( + ids, + vec!["t-active", "t-done", "t-failed", "t-waiting"], + "list_tasks is unfiltered: terminal rows are returned too, and every row survives a restart" + ); +} + +/// The per-task provenance chain, read back off disk. Per-TASK rather than one global chain, so the +/// scope of a read is one task and the links have to hold within it. +#[test] +fn a_task_event_chain_survives_a_restart_and_still_links() { + let dir = tempdir(); + let file = dir.join("events.db"); + let path = file.to_str().unwrap().to_string(); + { + let s = SqliteStore::open(&path, 5000).unwrap(); + s.append_task_event(&sample_event("t-1", 1, "task.submitted", "", "e1")) + .unwrap(); + s.append_task_event(&sample_event("t-1", 2, "task.working", "e1", "e2")) + .unwrap(); + s.append_task_event(&sample_event("t-1", 3, "task.interrupted", "e2", "e3")) + .unwrap(); + // A second task's chain is independent — it must not leak into the first one's read. + s.append_task_event(&sample_event("t-2", 1, "task.submitted", "", "f1")) + .unwrap(); + drop(s); + } + let reopened = SqliteStore::open(&path, 5000).unwrap(); + let got = reopened.list_task_events("t-1").unwrap(); + assert_eq!( + got.len(), + 3, + "the provenance chain must survive a restart; got {} events back after reopening the file, \ + which is the accept-and-keep-nothing default this backend exists to replace", + got.len() + ); + assert_eq!( + got.iter().map(|e| e.seq).collect::>(), + vec![1, 2, 3], + "oldest-first by seq, which is the order the chain verifier reads" + ); + assert_eq!(got[0].prev_hash, "", "seq 1 opens the chain"); + for w in got.windows(2) { + assert_eq!( + w[1].prev_hash, w[0].hash, + "the per-task chain must still link after a restart: seq {} carries prev_hash {:?} but \ + seq {} persisted hash {:?}", + w[1].seq, w[1].prev_hash, w[0].seq, w[0].hash + ); + } + // Every field round-trips, including the join key that is deliberately NOT chained. + assert_eq!(got[2].kind, "task.interrupted"); + assert_eq!(got[2].request_id, "req-3"); + assert_eq!(got[1].context_id, "ctx-t-1"); + assert_eq!(got[1].principal, "vk_a"); + assert_eq!(got[1].agent_id, "planner"); + assert_eq!(got[1].state, "working"); + assert_eq!(got[1].ts, 102); + // The scope of a read is one task. + assert_eq!(reopened.list_task_events("t-2").unwrap().len(), 1); + assert!( + reopened.list_task_events("t-unknown").unwrap().is_empty(), + "a task with no events reads back empty, not an error" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A replayed `(task_id, seq)` UPSERTS. This is where the task-event contract genuinely DIFFERS +/// from `append_mcp_call`'s, and a backend that copied the call log's fork check would be wrong in a +/// way that looks right: the contract says a store "must upsert on that pair — the write-through is +/// idempotent on replay, and rejecting or duplicating a replayed `seq` breaks the chain the engine +/// will verify on read". So neither a duplicate row nor an error, on either an identical replay or a +/// corrected one. +#[test] +fn a_replayed_task_event_upserts_rather_than_duplicating_or_erroring() { + let s = SqliteStore::open_in_memory().unwrap(); + let e = sample_event("t-1", 1, "task.submitted", "", "e1"); + s.append_task_event(&e).unwrap(); + s.append_task_event(&e) + .expect("an identical replay must succeed, not be rejected as a fork"); + assert_eq!( + s.list_task_events("t-1").unwrap().len(), + 1, + "a replay must not duplicate the row" + ); + + // A rewritten event at the same seq REPLACES, per the contract's "must upsert on that pair". + let mut corrected = sample_event("t-1", 1, "task.submitted", "", "e1-corrected"); + corrected.state = "submitted".to_string(); + s.append_task_event(&corrected).unwrap(); + let got = s.list_task_events("t-1").unwrap(); + assert_eq!(got.len(), 1, "an upsert replaces; it does not append"); + assert_eq!(got[0].hash, "e1-corrected"); + assert_eq!(got[0].state, "submitted"); +} + +/// Retention drops TERMINAL rows only, strictly older than the cutoff, and returns a count it +/// actually performed. An interrupted task waiting on a human is exactly the row that legitimately +/// sits still for a long time; compacting it is losing the work, not reclaiming space. +#[test] +fn purge_tasks_before_drops_only_terminal_rows_and_returns_a_real_count() { + let s = SqliteStore::open_in_memory().unwrap(); + s.put_task(&sample_task("t-old-done", "completed", 100)) + .unwrap(); + s.put_task(&sample_task("t-old-failed", "failed", 100)) + .unwrap(); + s.put_task(&sample_task("t-old-canceled", "canceled", 100)) + .unwrap(); + s.put_task(&sample_task("t-old-rejected", "rejected", 100)) + .unwrap(); + // Old, and NOT terminal — never dropped, no matter how old. + s.put_task(&sample_task("t-old-waiting", "input-required", 100)) + .unwrap(); + s.put_task(&sample_task("t-old-auth", "auth-required", 100)) + .unwrap(); + s.put_task(&sample_task("t-old-working", "working", 100)) + .unwrap(); + s.put_task(&sample_task("t-old-submitted", "submitted", 100)) + .unwrap(); + // Terminal but at the cutoff exactly, and terminal but newer — both kept. + s.put_task(&sample_task("t-at-cutoff", "completed", 200)) + .unwrap(); + s.put_task(&sample_task("t-new-done", "completed", 300)) + .unwrap(); + + let purged = s.purge_tasks_before(200).unwrap(); + assert_eq!( + purged, 4, + "only the four TERMINAL rows strictly older than the cutoff go, and the count must be one \ + actually performed rather than a guess" + ); + let mut left = s + .list_tasks() + .unwrap() + .into_iter() + .map(|t| t.task_id) + .collect::>(); + left.sort(); + assert_eq!( + left, + vec![ + "t-at-cutoff", + "t-new-done", + "t-old-auth", + "t-old-submitted", + "t-old-waiting", + "t-old-working", + ], + "an active or interrupted task is never dropped by retention, and `before` is strictly \ + less-than so a row exactly at the cutoff is kept" + ); + assert_eq!( + s.purge_tasks_before(200).unwrap(), + 0, + "re-running the same purge removes nothing" + ); +} + +/// Retention has to bound the EVENT table too. The trait offers no `purge_task_events_before`, so if +/// purging a task left its provenance behind, `task_events` would grow without any bound the +/// contract provides a way to apply. Dropping a task therefore drops the chain that belongs to it — +/// and drops nothing belonging to any other task. +#[test] +fn purging_a_task_takes_its_provenance_chain_with_it_and_no_other() { + let s = SqliteStore::open_in_memory().unwrap(); + s.put_task(&sample_task("t-gone", "completed", 100)) + .unwrap(); + s.put_task(&sample_task("t-stays", "working", 100)).unwrap(); + s.append_task_event(&sample_event("t-gone", 1, "task.submitted", "", "g1")) + .unwrap(); + s.append_task_event(&sample_event("t-gone", 2, "task.completed", "g1", "g2")) + .unwrap(); + s.append_task_event(&sample_event("t-stays", 1, "task.submitted", "", "s1")) + .unwrap(); + + assert_eq!(s.purge_tasks_before(200).unwrap(), 1); + assert!( + s.list_task_events("t-gone").unwrap().is_empty(), + "the purged task's events go with it; otherwise task_events grows unbounded, because the \ + contract offers no other way to purge them" + ); + assert_eq!( + s.list_task_events("t-stays").unwrap().len(), + 1, + "another task's chain must be untouched by that purge" + ); +} + +/// A `seq`/`ts`/`artifact_cursor` past `i64::MAX` cannot be stored faithfully — `as i64` wraps it +/// negative and the read clamps back — so the row read back would not be the row written. Refused +/// outright, exactly as `append_audit` refuses it, rather than silently mangled. +#[test] +fn the_task_store_refuses_values_it_cannot_store_faithfully() { + let s = SqliteStore::open_in_memory().unwrap(); + + let mut t = sample_task("t-1", "working", 200); + t.artifact_cursor = u64::MAX; + let err = s + .put_task(&t) + .expect_err("an artifact cursor past i64::MAX must be refused, not wrapped"); + assert!( + err.0.contains("storable range"), + "the refusal must say why: {}", + err.0 + ); + assert!( + s.get_task("t-1").unwrap().is_none(), + "a refused write must leave nothing behind" + ); + + let mut e = sample_event("t-1", u64::MAX, "task.submitted", "", "e1"); + assert!(s + .append_task_event(&e) + .expect_err("a seq past i64::MAX must be refused") + .0 + .contains("storable range")); + e.seq = 1; + e.ts = u64::MAX; + assert!(s + .append_task_event(&e) + .expect_err("a ts past i64::MAX must be refused") + .0 + .contains("storable range")); + + // The boundary itself is storable and round-trips exactly. + t.artifact_cursor = i64::MAX as u64; + s.put_task(&t).expect("i64::MAX is in range"); + assert_eq!( + s.get_task("t-1").unwrap().unwrap().artifact_cursor, + i64::MAX as u64 + ); +} + +/// The v7 -> v8 crossing is additive: a real v7 database gains `tasks` and `task_events` and keeps +/// every row it already had. Regression cover for the pre-v5 drop-and-recreate path reaching a live +/// database on a version bump it has no business touching. +#[test] +fn migrate_v7_to_v8_adds_the_task_store_without_wiping_data() { + let dir = tempdir(); + let file = dir.join("v7.db"); + { + let conn = Connection::open(&file).unwrap(); + conn.execute_batch(SCHEMA).unwrap(); + conn.execute("DROP TRIGGER tasks_cascade_events", []) + .unwrap(); + conn.execute("DROP TABLE task_events", []).unwrap(); + conn.execute("DROP TABLE tasks", []).unwrap(); + conn.execute( + "INSERT INTO keys (id, name, key_group, allowed_pools, labels, enabled, \ + generation_hash, created_at, updated_at, expires_at, deleted_at, revision) \ + VALUES ('vk_v7', 'n', NULL, NULL, '{}', 1, 'g1', 0, 0, NULL, NULL, 0)", + [], + ) + .unwrap(); + conn.pragma_update(None, "user_version", 7i64).unwrap(); + } + let s = SqliteStore::open(file.to_str().unwrap(), 5000) + .expect("a v7 database must migrate additively to v8"); + assert!( + s.get_key("vk_v7").unwrap().is_some(), + "a real v7 key must survive the v7->v8 crossing" + ); + s.put_task(&sample_task("t-1", "working", 200)) + .expect("the newly created tasks table must be writable after the migration"); + s.append_task_event(&sample_event("t-1", 1, "task.submitted", "", "e1")) + .expect("the newly created task_events table must be writable after the migration"); + assert!(s.get_task("t-1").unwrap().is_some()); + assert_eq!(s.list_task_events("t-1").unwrap().len(), 1); + let _ = std::fs::remove_dir_all(&dir); +}