From e2f6dfb225d4ad621344a30a43ff1150e729de2d Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Sun, 9 Aug 2026 07:57:20 -0700 Subject: [PATCH 1/5] store: the MCP tool-call log is durable, and a restart is what proves it The MCP plane's call evidence rode the admin audit ring: an in-memory, size-bounded working set shared with admin mutations. Two things follow, and both are bad. Data-plane tool calls arrive at request rate, so a busy afternoon evicts every admin row from the ring -- the question of who changed a registration becomes unanswerable exactly when an incident makes somebody ask. And nothing survived a restart at all. So: `mcp_calls`, its own table, a different population from `audit_log` on purpose. The chain is scoped to the PRINCIPAL -- (principal, seq) is the primary key -- because a global chain would serialise every caller's calls behind one append and would make one caller's evidence unverifiable without possessing every other caller's rows. Opaque `body` plus only the columns a query needs. `principal` and `ts` are the index columns (scoped read; the retention sweep's age key). The chain columns -- seq, prev_hash, hash -- are REAL columns rather than buried in the body, because durability here is established by READING THE CHAIN BACK and verifying it, and a digest reachable only by decoding a blob can be neither constrained nor indexed. `version` and `expires_at` are carried now and written by nothing: adding a column to a populated table later is a rewrite, carrying it from the first migration is free. The store never computes or recomputes a digest. It persists what it was handed, verbatim, and returns it verbatim -- a digest a store could recompute is one a compromised store could forge consistently. A record landing on an occupied (principal, seq) is settled the way the contract settles it: byte-identical is the at-least-once retry and succeeds, DIFFERENT is a forked or tampered log and is an error. Overwriting would destroy precisely the case worth reporting. A trigger enforces that a persisted record is never rewritten, so the property survives an operator with the sqlite3 CLI; there is deliberately no matching delete guard, because retention has to remain possible and bounded-retention is not an integrity claim. v6 -> v7 is purely additive: the table is new, so the unconditional CREATE TABLE IF NOT EXISTS is the whole migration and no backfill arm exists. The test that carries this is not a unit test against a live handle -- that cannot tell a backend that wrote to disk from one holding a HashMap behind the same trait. It writes a chain, DROPS the store (closing every connection and its WAL), reopens the same file, and asserts the chain still links from the bytes that came back off disk. Run before the methods existed, against the trait's accept-and-keep-nothing defaults, it fails with "got 0 records back after reopening the file"; that is the behaviour this backend replaces. `.busbar-ref` moves off the 1.5.3 pin, which predates the contract, onto the `dev` commit that carries it. CI already builds against the same-named core branch, so CI and release now name the same line without a hand-held pin. --- .busbar-ref | 2 +- store-sqlite/src/lib.rs | 188 ++++++++++++++++++++++++++- store-sqlite/src/tests.rs | 258 +++++++++++++++++++++++++++++++++++++- 3 files changed, 442 insertions(+), 6 deletions(-) diff --git a/.busbar-ref b/.busbar-ref index 9b9106f..499c0cb 100644 --- a/.busbar-ref +++ b/.busbar-ref @@ -1 +1 @@ -c8780349cf66d09b891478d50766b89dc1ff224c 1.5.3 +5a4f0195e29ef8f17a8abcffa308fb797c797edf 1.5.3 diff --git a/store-sqlite/src/lib.rs b/store-sqlite/src/lib.rs index bd32f97..4bc8aa6 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, TierTokens, UsageDelta, + UsageLedger, VirtualKey, }; use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior}; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -47,7 +47,13 @@ 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. +const SCHEMA_VERSION: i64 = 7; const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS store_meta ( @@ -179,6 +185,41 @@ 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); + -- 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 +231,13 @@ 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; "; /// Apply the fixed pragma set to a connection, in the documented order (`busy_timeout` FIRST: @@ -1262,6 +1310,138 @@ 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) + } +} + +/// 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..1de2796 100644 --- a/store-sqlite/src/tests.rs +++ b/store-sqlite/src/tests.rs @@ -2,7 +2,9 @@ // Copyright (C) 2026 Busbar Inc and contributors use super::*; -use busbar_api::{AuditRecord, ModelTokensDelta, Store, TierTokensDelta, VirtualKey}; +use busbar_api::{ + AuditRecord, McpCallRecord, ModelTokensDelta, Store, TierTokensDelta, VirtualKey, +}; use rusqlite::TransactionBehavior; fn sample_key(id: &str, generation: &str) -> VirtualKey { @@ -1201,3 +1203,257 @@ 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"); +} From e89c66174731cff66bbf6f5a621df17654babc0f Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Sun, 9 Aug 2026 13:14:58 -0700 Subject: [PATCH 2/5] store: an in-flight A2A task survives a restart, and a restart is what proves it A2A is asynchronous by design. A task spans turns, can sit interrupted waiting on a human, and can outlive the process that started it. The engine has had a seam for that since 1.5.6 -- `put_task`/`get_task`/`list_tasks`/ `purge_tasks_before`/`append_task_event`/`list_task_events` -- and this backend implemented none of them, so the trait's defaults applied: accept the write, return `Ok(())`, keep nothing. Every in-flight task was lost on every deploy, and nothing anywhere reported it, because the return value of a write is not evidence that anything was stored. So: `tasks` and `task_events`, their own tables. Every field is a real column rather than the opaque `body` that `mcp_calls` uses, and the difference is not inconsistency. A call record carries open-ended narration no query 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. `put_task` UPSERTS by `task_id`: the engine writes through on every state transition, so a second write for one task must replace the row rather than leave a second one behind. `append_task_event` also upserts, and this is the one place the task contract genuinely DIFFERS from `append_mcp_call`'s. That method treats a different record at an occupied sequence as a forked log and refuses it. A task event is specified to upsert on `(task_id, seq)` so 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 have been wrong in a way that looks right, so `task_events` deliberately has no no-update trigger to match `mcp_calls`'s. Retention drops TERMINAL rows only, and the terminal set is named as a closed list rather than derived by negation: a state token this build has never heard of reads as not-terminal, so a store compiled before a state existed cannot delete a task it does not understand. An interrupt 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. A purged task takes its provenance chain with it, via a trigger rather than a foreign key. The cascade is load-bearing -- `purge_tasks_before` is the only retention method the contract gives this data, so leaving the events behind would leave `task_events` with no bound anywhere in the trait. A trigger rather than an FK because an FK would also impose an ORDER on the writes, and the engine is under no such obligation: a `task.submitted` event and the first `put_task` are two independent write-throughs. A `u64` past `i64::MAX` is refused rather than mangled, the same guard `append_audit` already applies to its own seq, factored out now that five fields across two methods need it. Here the value that would silently change is the ARTIFACT CURSOR -- how much of a stream has been relayed -- so a wrapped one either replays delivered artifacts or skips undelivered ones, with no error ever reported. v7 -> v8 is purely additive: two new tables, no backfill, nothing dropped. The test that carries this is not a unit test against a live handle -- that cannot tell a backend that wrote to disk from one holding a HashMap behind the same trait, nor either of those from the defaults. It writes a task, DROPS the store (closing every connection and its WAL), reopens the same file, and reads the row back off disk. Run before the methods existed, against the accept-and-keep-nothing defaults, all seven fail, the headline being "got None back after reopening the file"; that is the behaviour this backend replaces. --- store-sqlite/src/lib.rs | 281 +++++++++++++++++++++++++- store-sqlite/src/tests.rs | 407 +++++++++++++++++++++++++++++++++++++- 2 files changed, 684 insertions(+), 4 deletions(-) diff --git a/store-sqlite/src/lib.rs b/store-sqlite/src/lib.rs index 4bc8aa6..2fc0477 100644 --- a/store-sqlite/src/lib.rs +++ b/store-sqlite/src/lib.rs @@ -9,8 +9,8 @@ use busbar_api::{ AuditRecord, CredentialMeta, CredentialSecret, McpCallRecord, MeteringDelta, MeteringRow, - ModelTokens, ScopeRef, SecretForm, Store, StoreError, StoreResult, TierTokens, UsageDelta, - UsageLedger, VirtualKey, + ModelTokens, ScopeRef, SecretForm, Store, StoreError, StoreResult, TaskEventRow, TaskRow, + TierTokens, UsageDelta, UsageLedger, VirtualKey, }; use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior}; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -53,7 +53,17 @@ impl IntoStoreResult for Result { /// 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. -const SCHEMA_VERSION: i64 = 7; +/// +/// 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 ( @@ -220,6 +230,65 @@ CREATE TABLE IF NOT EXISTS mcp_calls ( -- 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 @@ -238,6 +307,11 @@ CREATE TRIGGER IF NOT EXISTS audit_log_no_delete BEFORE DELETE ON audit_log -- 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: @@ -1396,6 +1470,207 @@ impl Store for SqliteStore { .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`, diff --git a/store-sqlite/src/tests.rs b/store-sqlite/src/tests.rs index 1de2796..15ff37e 100644 --- a/store-sqlite/src/tests.rs +++ b/store-sqlite/src/tests.rs @@ -3,7 +3,8 @@ use super::*; use busbar_api::{ - AuditRecord, McpCallRecord, ModelTokensDelta, Store, TierTokensDelta, VirtualKey, + AuditRecord, McpCallRecord, ModelTokensDelta, Store, TaskEventRow, TaskRow, TierTokensDelta, + VirtualKey, }; use rusqlite::TransactionBehavior; @@ -1457,3 +1458,407 @@ fn mcp_calls_rejects_a_direct_update_but_allows_the_retention_delete() { .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); +} From 6ece705d79e60486d58a403c0e882e5d6ba6da38 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 11 Aug 2026 10:54:40 -0700 Subject: [PATCH 3/5] .busbar-ref: record the busbar version as 1.6.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owner's call, 2026-08-11: the release carrying MCP and A2A is 1.6.0, not 1.5.5. Field 2 was stale (1.5.3 / 1.5.5) and is now the real number. Field 1 (the engine SHA) is deliberately UNCHANGED — a separate change owns the pin. --- .busbar-ref | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.busbar-ref b/.busbar-ref index 499c0cb..35a5507 100644 --- a/.busbar-ref +++ b/.busbar-ref @@ -1 +1 @@ -5a4f0195e29ef8f17a8abcffa308fb797c797edf 1.5.3 +5a4f0195e29ef8f17a8abcffa308fb797c797edf 1.6.0 From 7b101a883b113113cc246311bd63e84de3eb8728 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 11 Aug 2026 10:59:10 -0700 Subject: [PATCH 4/5] test(abi): a task written through the REAL plugin path, and a reload is what proves it Every existing test of the ten task / call-log methods calls SqliteStore directly, in-process, and none of them can see the failure that matters in production. `busbar_api::Store` defaults all ten 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 -- the only way it ever runs -- loses every in-flight A2A task and every tool-call record while reporting success. So this 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 three 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 dlopens again over the same file and reads everything back, retention counts included. A single-row round trip cannot tell a relayed method from a lucky default; a multi-row one across a reload cannot be faked by either. RED against busbar dev @ afde0814 (`list_tasks` answers 0 of 3), GREEN against fix/abi-a2a-task-methods @ 3f1bc096, both watched. The red is the truthful report that durable tasks do not yet work through the only path that ships, and it clears the moment the engine-side ABI relay lands on the ref this repo builds against. --- store-sqlite-plugin/tests/e2e.rs | 215 ++++++++++++++++++++++++++++++- 1 file changed, 214 insertions(+), 1 deletion(-) diff --git a/store-sqlite-plugin/tests/e2e.rs b/store-sqlite-plugin/tests/e2e.rs index f036891..0805482 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; @@ -722,3 +724,214 @@ 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 Some(path) = plugin_path() else { + eprintln!("skip: store-sqlite-plugin cdylib not built (run cargo test/build first)"); + return; + }; + 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); +} From 0ee025153fde33ef23d0cc66bb1991a5bb4de7c2 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:56:30 -0700 Subject: [PATCH 5/5] test(abi): load the cdylib this build produced, and refuse a stale one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The over-the-ABI durability test looked for the plugin cdylib in target//, which only `cargo build` ever refreshes. Under `cargo test` cargo emits the artifact into target//deps/ and never uplifts it, so the test read whatever an earlier build happened to leave behind — or, on a fresh clone, nothing at all, in which case it printed "skip:" and reported GREEN with zero coverage of the ten task/call-log methods. Both outcomes are lies about durability. A cdylib older than an ABI change answers every write Ok(()) and every read empty, which is byte-for-byte the unrelayed-seam defect this test exists to catch; and an artifact newer than a regression reports green while the shipped ABI is broken. Proven, not argued: with a regressed plugin in the tree and a good cdylib in target/debug/, the old lookup PASSED and this one FAILS. plugin_path() now reads only deps/ — the output of the same build graph that produced this test binary — panics instead of skipping when it is absent, and asserts the artifact is no older than any workspace src/**/*.rs, failing with a message that says STALE ARTIFACT and explicitly not a durability verdict. Note the dev-dependency edge used elsewhere for this does NOT make the artifact exist: cargo satisfies a dev-dep with the rlib. Verified by deleting the cdylib and re-running. --- store-sqlite-plugin/tests/e2e.rs | 134 +++++++++++++++++++++++-------- 1 file changed, 102 insertions(+), 32 deletions(-) diff --git a/store-sqlite-plugin/tests/e2e.rs b/store-sqlite-plugin/tests/e2e.rs index 0805482..759f0cb 100644 --- a/store-sqlite-plugin/tests/e2e.rs +++ b/store-sqlite-plugin/tests/e2e.rs @@ -83,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. @@ -176,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(); @@ -449,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"); @@ -686,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. @@ -750,10 +823,7 @@ fn load_and_exercise_sqlite_plugin_bad_config_fails_over_abi() { /// 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 Some(path) = plugin_path() else { - eprintln!("skip: store-sqlite-plugin cdylib not built (run cargo test/build first)"); - return; - }; + 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();