From b7e56b5465375c02c70b578d3821ac6490f959b7 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Sun, 9 Aug 2026 08:03:20 -0700 Subject: [PATCH 1/2] store: the MCP tool-call log is durable, and a reconnect 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. 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. 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 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; 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 a digest reachable only by decoding a payload 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 insert is ON CONFLICT DO NOTHING, and the incumbent is read afterwards WITHOUT a transaction -- safe precisely because this table is never rewritten, so a row that exists cannot change underneath the read. Identical is the at-least-once retry and succeeds; DIFFERENT is a forked or tampered log and is an error, because overwriting destroys the one case worth reporting. 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 drops the store -- closing its connection entirely -- then connects a genuinely new one and asserts the chain still links from what the server hands back. Run with the four methods removed, so the trait's accept-and-keep-nothing defaults apply, it fails with "got 0 records back"; that is the behaviour this backend replaces. One finding worth keeping: retention is GLOBAL by `ts` and cannot be scoped to a principal. Against the shared live database the purge test was deleting every other test's rows, which is why the suite now bands its timestamps -- the purge test owns the low band and every other test sits above the highest cutoff it uses. The cross-deletion is what surfaced it. `.busbar-ref` moves off the 1.5.3 pin that predates this contract onto the `dev` commit that carries it, so the ref names a core that actually has the methods this backend overrides. CI needs no hand-held pin alongside it: it already builds against the same-named core branch, so dev builds against dev and a release builds against what ships. --- .busbar-ref | 2 +- store-postgres/src/lib.rs | 170 +++++++++++++++++++++++++- store-postgres/src/tests.rs | 230 +++++++++++++++++++++++++++++++++++- 3 files changed, 396 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-postgres/src/lib.rs b/store-postgres/src/lib.rs index 8c54f9f..02e956c 100644 --- a/store-postgres/src/lib.rs +++ b/store-postgres/src/lib.rs @@ -35,9 +35,9 @@ //! added later without another schema bump. 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 postgres::types::ToSql; use postgres::{Client, NoTls, Row, Transaction}; @@ -232,7 +232,12 @@ fn scrub(msg: String, secret: Option<&str>) -> String { /// which is exactly why it is gated on `version < 6` and will never fire a second time on any /// store that has already crossed into v6. `hydrate_budgets` itself drops the heuristic entirely /// once every store it reads from has passed through this migration. -const SCHEMA_VERSION: i64 = 6; +/// +/// v7: the durable MCP TOOL-CALL LOG (`mcp_calls`). PURELY ADDITIVE and needs no backfill arm — the +/// table is new, so `SCHEMA`'s own `CREATE TABLE IF NOT EXISTS` (executed unconditionally on every +/// migrate) is the entire migration. Nothing is dropped and no existing row is touched, which is why +/// there is no `version < 7` block to match the `version < 6` one. +const SCHEMA_VERSION: i64 = 7; const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS busbar_schema ( @@ -337,6 +342,40 @@ CREATE TABLE IF NOT EXISTS denylist ( reason TEXT NOT NULL DEFAULT '', created_at BIGINT NOT NULL DEFAULT 0 ); + +-- 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 behind one append and would make one +-- caller's evidence unverifiable without possessing every other caller's rows. +-- +-- SHAPE: opaque body plus only the columns a query 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 the body, because the engine establishes durability by +-- READING THE CHAIN BACK and verifying it; a digest reachable only by decoding an opaque payload +-- forces a deserialise per verify and cannot be constrained or indexed by the database. The store +-- NEVER computes or recomputes a digest: it persists what it was handed and returns it verbatim. +CREATE TABLE IF NOT EXISTS mcp_calls ( + principal TEXT NOT NULL, + seq BIGINT NOT NULL, + ts BIGINT NOT NULL, + prev_hash TEXT NOT NULL, + hash TEXT NOT NULL, + body TEXT NOT NULL, + -- Carried now, written by nothing yet, and deliberately so: 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 BIGINT, + version BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (principal, seq) +); +-- 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); "; /// Postgres `Store` backend (durable, shared across a cluster). A single mutex-guarded connection — @@ -1344,6 +1383,83 @@ impl Store for PostgresStore { Ok(()) } + fn append_mcp_call(&self, record: &McpCallRecord) -> StoreResult<()> { + let body = mcp_call_body(record); + let (seq, ts) = (clamp(record.seq), clamp(record.ts)); + // ON CONFLICT DO NOTHING makes the insert atomic against a concurrent writer. Reading the + // incumbent AFTERWARDS is safe without a transaction precisely because this table is never + // rewritten: a row that exists cannot change under us, so what we read is what collided. + let inserted = self + .lock() + .execute( + "INSERT INTO mcp_calls (principal, seq, ts, prev_hash, hash, body) + VALUES ($1,$2,$3,$4,$5,$6) + ON CONFLICT (principal, seq) DO NOTHING", + &[ + &record.principal, + &seq, + &ts, + &record.prev_hash, + &record.hash, + &body, + ], + ) + .store()?; + if inserted == 1 { + return Ok(()); + } + let existing = self + .lock() + .query_opt( + "SELECT ts, prev_hash, hash, body FROM mcp_calls WHERE principal = $1 AND seq = $2", + &[&record.principal, &seq], + ) + .store()?; + if let Some(r) = existing { + let (e_ts, e_prev, e_hash, e_body): (i64, String, String, String) = + (r.get(0), r.get(1), r.get(2), r.get(3)); + // 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 e_ts == ts && e_prev == record.prev_hash && e_hash == record.hash && e_body == body { + return Ok(()); + } + } + // Names the sequence and nothing else — it must not echo stored (or caller) content back. + Err(StoreError(format!( + "mcp call log fork: a different record is already persisted at sequence {} for this principal", + record.seq + ))) + } + + fn list_mcp_calls(&self, principal: &str) -> StoreResult> { + let sql = + format!("SELECT {MCP_CALL_COLUMNS} FROM mcp_calls WHERE principal = $1 ORDER BY seq"); + let rows = self.lock().query(&sql, &[&principal]).store()?; + Ok(rows.iter().map(row_to_mcp_call).collect()) + } + + fn list_mcp_call_principals(&self) -> StoreResult> { + let rows = self + .lock() + .query( + "SELECT DISTINCT principal FROM mcp_calls ORDER BY principal", + &[], + ) + .store()?; + Ok(rows.iter().map(|r| r.get(0)).collect()) + } + + 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() + .execute("DELETE FROM mcp_calls WHERE ts < $1", &[&clamp(before)]) + .store()?; + Ok(removed) + } + fn list_denylist(&self) -> StoreResult> { let rows = self.lock().query("SELECT sub FROM denylist", &[]).store()?; Ok(rows.iter().map(|r| r.get(0)).collect()) @@ -1352,6 +1468,52 @@ impl Store for PostgresStore { const AUDIT_COLUMNS: &str = "seq, ts, action, resource, outcome, principal, prev_hash, hash"; +const MCP_CALL_COLUMNS: &str = "principal, seq, ts, prev_hash, hash, body"; + +/// 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 payload. +fn row_to_mcp_call(r: &Row) -> McpCallRecord { + let body: String = r.get(5); + let v: serde_json::Value = serde_json::from_str(&body).unwrap_or(serde_json::Value::Null); + let s = |k: &str| v.get(k).and_then(|x| x.as_str()).unwrap_or("").to_string(); + 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 row_to_audit(r: &Row) -> AuditRecord { AuditRecord { seq: read_u64(r.get::<_, i64>(0)), diff --git a/store-postgres/src/tests.rs b/store-postgres/src/tests.rs index 576041d..8905a55 100644 --- a/store-postgres/src/tests.rs +++ b/store-postgres/src/tests.rs @@ -6,7 +6,9 @@ //! `postgres:16`. use super::*; -use busbar_api::{CredentialMeta, CredentialSecret, ModelTokensDelta, SecretForm, TierTokensDelta}; +use busbar_api::{ + CredentialMeta, CredentialSecret, McpCallRecord, ModelTokensDelta, SecretForm, TierTokensDelta, +}; /// Drift guard, no live DB needed: `CRED_SECRET_COLUMN_INDEX` must stay in sync with /// `CRED_META_COLUMNS`' own column count, since `str::split` isn't const-evaluable and the constant @@ -2129,3 +2131,229 @@ fn append_audit_never_reports_success_for_a_record_it_did_not_store() { } let _ = client.execute("DELETE FROM audit_log WHERE seq=$1", &[&clamp(seq)]); } + +// ── 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 on a NEW +// CONNECTION after the writing one is gone. + +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(), + } +} + +/// Live Postgres is SHARED across tests, so each test owns its own principal ids and clears them +/// first — the same isolation-by-unique-id discipline the key tests in this file use. +fn reset_calls(store: &PostgresStore, principals: &[&str]) { + for p in principals { + store + .lock() + .execute("DELETE FROM mcp_calls WHERE principal = $1", &[p]) + .expect("clear this test's own rows"); + } +} + +/// THE TEST THAT MATTERS. A round-trip on one live handle cannot distinguish a backend that wrote +/// to the server from one holding a HashMap behind the same trait. So this DROPS the store — closing +/// its connection entirely — then connects a genuinely new one and verifies the per-principal hash +/// chain still links from the rows the server hands back. +#[test] +fn an_mcp_call_chain_survives_dropping_the_connection_and_reconnecting() { + let Some(url) = live_url() else { return }; + let p = "vk_mcp_restart"; + { + let store = connect_store_with_retry(&url).expect("connect"); + reset_calls(&store, &[p]); + store + .append_mcp_call(&sample_call(p, 1, 2_000_000_100, "", "h1")) + .unwrap(); + store + .append_mcp_call(&sample_call(p, 2, 2_000_000_200, "h1", "h2")) + .unwrap(); + store + .append_mcp_call(&sample_call(p, 3, 2_000_000_300, "h2", "h3")) + .unwrap(); + drop(store); + } + + // A genuinely new connection — nothing carried over in this process. + let reopened = connect_store_with_retry(&url).expect("reconnect"); + let got = reopened.list_mcp_calls(p).unwrap(); + + assert_eq!( + got.len(), + 3, + "the call log must survive a reconnect; got {} records back, which is the \ + accept-and-keep-nothing behaviour this backend exists to replace", + got.len() + ); + 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 reconnect: seq {} carries prev_hash \ + {:?} but seq {} persisted hash {:?}", + w[1].seq, w[1].prev_hash, w[0].seq, w[0].hash + ); + } + assert_eq!(got.iter().map(|r| r.seq).collect::>(), vec![1, 2, 3]); + // The non-indexed payload must round-trip verbatim too. + 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); + reset_calls(&reopened, &[p]); +} + +/// 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. +#[test] +fn mcp_call_principals_are_enumerable_after_a_reconnect() { + let Some(url) = live_url() else { return }; + let (a, b) = ("vk_mcp_enum_a", "vk_mcp_enum_b"); + { + let store = connect_store_with_retry(&url).expect("connect"); + reset_calls(&store, &[a, b]); + store + .append_mcp_call(&sample_call(a, 1, 2_000_000_100, "", "a1")) + .unwrap(); + store + .append_mcp_call(&sample_call(b, 1, 2_000_000_100, "", "b1")) + .unwrap(); + store + .append_mcp_call(&sample_call(a, 2, 2_000_000_101, "a1", "a2")) + .unwrap(); + drop(store); + } + let reopened = connect_store_with_retry(&url).expect("reconnect"); + let principals = reopened.list_mcp_call_principals().unwrap(); + for want in [a, b] { + assert_eq!( + principals.iter().filter(|p| p.as_str() == want).count(), + 1, + "{want} must be enumerable after a reconnect, exactly once" + ); + } + // The chain scope is the principal: a scoped read returns only its own. + assert_eq!(reopened.list_mcp_calls(a).unwrap().len(), 2); + assert_eq!(reopened.list_mcp_calls(b).unwrap().len(), 1); + assert!( + reopened + .list_mcp_calls("vk_mcp_nonexistent") + .unwrap() + .is_empty(), + "a principal with no records reads back empty, not an error" + ); + reset_calls(&reopened, &[a, b]); +} + +/// 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 Some(url) = live_url() else { return }; + let store = connect_store_with_retry(&url).expect("connect"); + let p = "vk_mcp_purge"; + reset_calls(&store, &[p]); + // Retention is GLOBAL by `ts` — it is not scoped to a principal, and cannot be. Against the + // SHARED live database that means this test's cutoffs would delete every other test's rows if + // the timestamps overlapped, so the suite bands them: this test owns the low band and every + // other test sits ABOVE the highest cutoff used here. Caught by exactly that cross-deletion. + // A far-future ts keeps this test's rows clear of any other principal's in the shared DB, and + // the purge below is scoped by counting only this principal's survivors. + store + .append_mcp_call(&sample_call(p, 1, 1_000_000_100, "", "h1")) + .unwrap(); + store + .append_mcp_call(&sample_call(p, 2, 1_000_000_200, "h1", "h2")) + .unwrap(); + store + .append_mcp_call(&sample_call(p, 3, 1_000_000_300, "h2", "h3")) + .unwrap(); + + let purged = store.purge_mcp_calls_before(1_000_000_200).unwrap(); + assert!( + purged >= 1, + "purge must report the rows it actually removed; got {purged}" + ); + assert_eq!( + store + .list_mcp_calls(p) + .unwrap() + .iter() + .map(|r| r.seq) + .collect::>(), + vec![2, 3], + "rows at or after the cutoff must remain — `before` is strictly less-than, so the row \ + exactly at the cutoff is kept" + ); + // The count is real: purging past everything clears this principal's remainder. + let rest = store.purge_mcp_calls_before(1_000_001_000).unwrap(); + assert!( + rest >= 2, + "the remaining two rows must actually be removed; got {rest}" + ); + assert!(store.list_mcp_calls(p).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 Some(url) = live_url() else { return }; + let store = connect_store_with_retry(&url).expect("connect"); + let p = "vk_mcp_replay"; + reset_calls(&store, &[p]); + + let rec = sample_call(p, 1, 2_000_000_100, "", "h1"); + store.append_mcp_call(&rec).unwrap(); + store + .append_mcp_call(&rec) + .expect("an identical replay is the at-least-once retry and must succeed"); + assert_eq!( + store.list_mcp_calls(p).unwrap().len(), + 1, + "a replay must not duplicate the row" + ); + + let forked = sample_call(p, 1, 2_000_000_100, "", "DIFFERENT"); + let err = store + .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!( + store.list_mcp_calls(p).unwrap()[0].hash, + "h1", + "the refused fork must not have overwritten the record already on record" + ); + + // A differing non-indexed payload under an identical digest is a fork too, not a silent accept. + let mut tampered = sample_call(p, 1, 2_000_000_100, "", "h1"); + tampered.tool = "srv_other_tool".to_string(); + store + .append_mcp_call(&tampered) + .expect_err("a payload that differs under an identical digest is a fork and must error"); + reset_calls(&store, &[p]); +} From 44037318271055abc00729819f72045cc36bcec4 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Sun, 9 Aug 2026 12:45:03 -0700 Subject: [PATCH 2/2] store: an in-flight A2A task survives a restart, and a reconnect proves it A2A is async by design: a task spans turns, can sit interrupted waiting on a human, and can outlive the process that started it. The six task methods on the `Store` trait are DEFAULTED to accept-and-keep-nothing -- `put_task` returns Ok(()) and stores nothing, `get_task` answers None for everything, `list_tasks` empty, `purge_tasks_before` 0 -- and this backend overrode none of them. So every in-flight task on a Postgres-backed deployment was lost on the next deploy, and the write path never said a word about it. That is the difference between a resume that is real and one that is nominal. So: `tasks` keyed on `task_id`, and `task_events` keyed on `(task_id, seq)`. Every field of a task row is a REAL COLUMN, unlike `mcp_calls`'s opaque `body`, and that 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 -- indexed together for exactly that -- and there is no field left over to make opaque. `put_task` UPSERTS on `task_id`, because the engine writes through on EVERY state transition and a second write for one task must replace the row rather than accumulate a second one. `get_task` does NOT filter on principal: the contract puts caller-scoping engine-side deliberately, since an authorization check living in the backend is one an unauthorized reader bypasses by configuring a different backend. `list_tasks` is UNFILTERED, terminal rows included -- the boot rehydrate wants the active rows, the retention sweep the terminal ones and the scoped listing one principal's, so a store that pre-filtered for any one of those would break the other two. `append_task_event` UPSERTS on `(task_id, seq)`, and this is the one place the task contract genuinely DIFFERS from `append_mcp_call`'s in the commit before this one. That one treats an occupied slot holding a different record as a forked log and refuses it. This one is specified to upsert, 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 across would have been wrong in a way that looks right, so the difference is stated at the method rather than left to be inferred. Retention drops TERMINAL rows only, strictly older than the cutoff. The terminal set is named as a closed list rather than derived by negation: an unrecognised state token -- one a newer engine emits and this build has never heard of -- must read as NOT terminal, because the wrong half to guess on is the deleting half. An interrupted task waiting on a human is precisely the row that legitimately sits still for a long time, and compacting it is losing the work, not reclaiming space. The purge CASCADES to the task's events, via a trigger rather than a foreign key. The cascade itself is load-bearing: `purge_tasks_before` is the ONLY retention method the contract gives this data, so a purge that left the provenance behind would leave `task_events` with no bound anywhere in the trait. A real FK would give the cascade for free but 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, since a `task.submitted` event and the first `put_task` are two independent write-throughs. The trigger buys the bound without inventing an ordering constraint the contract never stated, and putting it in the database rather than in the DELETE's own SQL means a row removed by any other route (an operator's psql session, an external job) takes its chain with it too. Postgres BIGINT is signed, so a `u64` past `i64::MAX` cannot round-trip. `clamp` would pin it and the row read back would not be the row written, with nothing ever reporting an error -- and the value that silently changes is the ARTIFACT CURSOR, i.e. how much of a stream has been durably relayed, so a pinned cursor either replays delivered artifacts or skips undelivered ones. Refused instead, exactly as `append_audit` already refuses its own out-of-range seq/ts, across all five affected fields (artifact_cursor, created_at, updated_at, event seq, event ts). Nothing escalates durability per write the way store-sqlite has to: sqlite raises `synchronous` to FULL around these writes because its default is weaker, whereas Postgres's default `synchronous_commit = on` already fsyncs the commit that carries the transition. Same property, already the default. v7 -> v8 is purely additive: two new tables and their cascade, so `SCHEMA`'s own unconditional CREATE TABLE IF NOT EXISTS is the whole migration and no `version < 8` backfill arm exists. Postgres has no CREATE TRIGGER IF NOT EXISTS, so that one object is created under an explicit pg_trigger existence check -- DROP-then-CREATE would also be idempotent but would take an ACCESS EXCLUSIVE lock on `tasks` at every connect, for nothing. THE TEST THAT CARRIES THIS drops the store -- closing its connection entirely -- then connects a genuinely NEW one to the same database and reads the task back off the server. A round-trip through one live handle proves nothing: it cannot tell a backend that wrote to Postgres from one holding a HashMap behind the same trait, nor either of those from the trait's own defaults exercised through the very handle that "wrote". Run against the unimplemented state, all eight new tests fail, and they fail on the property rather than on a detail: "an in-flight task must survive a restart; got None back on a new connection", "got 0 events back on a new connection", the unfiltered listing comes back `[]` against four written rows, the purge reports 0 where 4 rows were owed, and the range guard returns Ok for a cursor of u64::MAX. 42 passed, 8 failed before; 50 passed after. Each of these tests owns a disposable database rather than banding its timestamps in the shared one, and that is a consequence of the method: `purge_tasks_before` is GLOBAL by `updated_at` and cannot be scoped to a task or a principal, so against a shared database a retention test's cutoff deletes every other test's terminal rows. Isolation also lets the assertions be EXACT -- "the purge removed four rows", "these seven survive" -- rather than the `>=` a shared table forces, and an exact count is what proves a purge performed the number it reported. Gates run locally against a real postgres:16 and a sibling busbarAI at the .busbar-ref commit: public-hygiene-lint --selftest and --root, cargo fmt --all --check, cargo build --all-targets, cargo clippy --all-targets -D warnings, and cargo test with BUSBAR_TEST_POSTGRES_URL set. --- store-postgres/src/lib.rs | 281 +++++++++++++++++++- store-postgres/src/tests.rs | 493 +++++++++++++++++++++++++++++++++++- 2 files changed, 770 insertions(+), 4 deletions(-) diff --git a/store-postgres/src/lib.rs b/store-postgres/src/lib.rs index 02e956c..87ec4e2 100644 --- a/store-postgres/src/lib.rs +++ b/store-postgres/src/lib.rs @@ -36,8 +36,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 postgres::types::ToSql; use postgres::{Client, NoTls, Row, Transaction}; @@ -237,7 +237,18 @@ fn scrub(msg: String, secret: Option<&str>) -> String { /// table is new, so `SCHEMA`'s own `CREATE TABLE IF NOT EXISTS` (executed unconditionally on every /// migrate) is the entire migration. Nothing is dropped and no existing row is touched, which is why /// there is no `version < 7` block to match the `version < 6` one. -const SCHEMA_VERSION: i64 = 7; +/// +/// v8: the durable A2A TASK STORE (`tasks`, `task_events`, and the cascade that ties them). Additive +/// on the same terms as v7 — two new tables, no backfill, nothing dropped, no existing row touched — +/// so again there is no `version < 8` block. +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 busbar_schema ( @@ -376,6 +387,96 @@ 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 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 BIGINT NOT NULL, + push_callback TEXT NOT NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL +); +-- 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. +-- +-- Deliberately NO foreign key to `tasks`, even though one would give the cascade for free: 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 on the contract, and a store that rejected the pair in the +-- wrong order would be enforcing an ordering the contract never states. The cascade is a trigger +-- instead (see tasks_cascade_events below), which gives the retention bound without the ordering. +CREATE TABLE IF NOT EXISTS task_events ( + task_id TEXT NOT NULL, + seq BIGINT NOT NULL, + ts BIGINT 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) +); + +-- task_events follows its task out of the database, and that cascade is LOAD-BEARING rather than +-- tidiness: `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. +-- +-- In the database rather than in `purge_tasks_before`'s SQL, so a row removed by any other route -- +-- an operator's psql session, an external retention job -- takes its chain with it too. There is +-- deliberately no no-update guard to match the one the MCP log would want: the task-event contract +-- REQUIRES an upsert on (task_id, seq), so forbidding the rewrite would forbid the contract. +-- +-- FOR EACH ROW, not a statement-level trigger: the retention sweep is off the request path and runs +-- rarely, and a per-row delete keyed on the primary key is the shape that stays correct if a future +-- caller ever deletes a single task. +CREATE OR REPLACE FUNCTION tasks_cascade_events() RETURNS trigger AS $fn$ +BEGIN + DELETE FROM task_events WHERE task_id = OLD.task_id; + RETURN OLD; +END; +$fn$ LANGUAGE plpgsql; +-- Postgres has no CREATE TRIGGER IF NOT EXISTS, and this whole SCHEMA is re-executed on every +-- migrate, so the existence check is explicit. DROP-then-CREATE would work too but would take an +-- ACCESS EXCLUSIVE lock on `tasks` at every single connect, for nothing. +DO $do$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_trigger + WHERE tgname = 'tasks_cascade_events' AND tgrelid = 'tasks'::regclass + ) THEN + CREATE TRIGGER tasks_cascade_events AFTER DELETE ON tasks + FOR EACH ROW EXECUTE FUNCTION tasks_cascade_events(); + END IF; +END +$do$; "; /// Postgres `Store` backend (durable, shared across a cluster). A single mutex-guarded connection — @@ -1460,6 +1561,129 @@ impl Store for PostgresStore { Ok(removed) } + fn put_task(&self, task: &TaskRow) -> StoreResult<()> { + // Refused rather than clamped, for the reason `append_audit` refuses its own out-of-range + // seq/ts: `clamp` pins a `u64` past `i64::MAX` to `i64::MAX`, 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 durably relayed. A pinned 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)?; + // 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. + self.lock() + .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", + &[ + &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 sql = format!("SELECT {TASK_COLUMNS} FROM tasks WHERE task_id = $1"); + let row = self.lock().query_opt(&sql, &[&task_id]).store()?; + Ok(row.as_ref().map(row_to_task)) + } + + 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 sql = format!("SELECT {TASK_COLUMNS} FROM tasks ORDER BY task_id"); + let rows = self.lock().query(&sql, &[]).store()?; + Ok(rows.iter().map(row_to_task).collect()) + } + + 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 `= ANY` 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. Each + // removed task takes its provenance chain with it via the tasks_cascade_events trigger. + let terminal: Vec = TERMINAL_TASK_STATES.iter().map(|s| s.to_string()).collect(); + let removed = self + .lock() + .execute( + "DELETE FROM tasks WHERE updated_at < $1 AND state = ANY($2)", + &[&clamp(before), &terminal], + ) + .store()?; + Ok(removed) + } + + 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; this one is specified to upsert 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 be wrong in a way that looks + // right, so the difference is stated rather than left to be inferred from the SQL. + self.lock() + .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", + &[ + &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 sql = + format!("SELECT {TASK_EVENT_COLUMNS} FROM task_events WHERE task_id = $1 ORDER BY seq"); + let rows = self.lock().query(&sql, &[&task_id]).store()?; + Ok(rows.iter().map(row_to_task_event).collect()) + } + fn list_denylist(&self) -> StoreResult> { let rows = self.lock().query("SELECT sub FROM denylist", &[]).store()?; Ok(rows.iter().map(|r| r.get(0)).collect()) @@ -1470,6 +1694,57 @@ const AUDIT_COLUMNS: &str = "seq, ts, action, resource, outcome, principal, prev const MCP_CALL_COLUMNS: &str = "principal, seq, ts, prev_hash, hash, body"; +const TASK_COLUMNS: &str = "task_id, context_id, principal, direction, state, agent_id, \ + artifact_cursor, push_callback, created_at, updated_at"; + +const TASK_EVENT_COLUMNS: &str = "task_id, seq, ts, kind, context_id, principal, agent_id, state, \ + request_id, prev_hash, hash"; + +/// Reject a `u64` a signed Postgres BIGINT cannot hold, naming the method and the field. `clamp` +/// would pin it to `i64::MAX` and the read would hand back that instead, 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 inline to its own `seq`/`ts`, factored out here 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: &Row) -> TaskRow { + 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: read_u64(r.get(6)), + push_callback: r.get(7), + created_at: read_u64(r.get(8)), + updated_at: read_u64(r.get(9)), + } +} + +fn row_to_task_event(r: &Row) -> TaskEventRow { + TaskEventRow { + task_id: r.get(0), + seq: read_u64(r.get(1)), + ts: read_u64(r.get(2)), + 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 diff --git a/store-postgres/src/tests.rs b/store-postgres/src/tests.rs index 8905a55..802f305 100644 --- a/store-postgres/src/tests.rs +++ b/store-postgres/src/tests.rs @@ -7,7 +7,8 @@ use super::*; use busbar_api::{ - CredentialMeta, CredentialSecret, McpCallRecord, ModelTokensDelta, SecretForm, TierTokensDelta, + CredentialMeta, CredentialSecret, McpCallRecord, ModelTokensDelta, SecretForm, TaskEventRow, + TaskRow, TierTokensDelta, }; /// Drift guard, no live DB needed: `CRED_SECRET_COLUMN_INDEX` must stay in sync with @@ -2357,3 +2358,493 @@ fn a_replayed_mcp_call_is_idempotent_but_a_forked_one_is_refused() { .expect_err("a payload that differs under an identical digest is a fork and must error"); reset_calls(&store, &[p]); } + +// ── 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, and for a shared +// server "restart" means dropping the store (closing its connection) and connecting a genuinely NEW +// one, so the answer can only have come from the server. +// +// Every test here runs against its own disposable database (`TempDb`), unlike the MCP call-log tests +// above, which band their timestamps to share one. That is not a style difference: `purge_tasks_before` +// is GLOBAL by `updated_at` and cannot be scoped to a task or a principal, so against a shared +// database a retention test's cutoff deletes every other test's terminal rows. Banding can keep the +// rows apart, but an isolated database also lets the assertions be EXACT ("the purge removed four +// rows", "these six survive") rather than the `>=` a shared table forces, and an exact count is +// precisely what proves a purge performed the number it reported. + +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 round-trip through one live handle: a live +/// handle cannot tell a backend that wrote to the server from one holding 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 are exercised through the very handle that "wrote". So this DROPS the store — closing its +/// connection entirely — then connects a genuinely NEW one to the same database and reads the task +/// back off the server. Against the unimplemented state it fails on the very first assertion, with +/// `get_task` answering `None` for a task that was accepted a moment earlier. +#[test] +fn an_in_flight_task_survives_dropping_the_connection_and_reconnecting() { + let Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_restart"); + let iso_url = tmp.url(); + + { + let store = connect_store_with_retry(&iso_url).expect("connect"); + store.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; + store.put_task(&interrupted).unwrap(); + store + .put_task(&sample_task("t-2", "submitted", 210)) + .unwrap(); + drop(store); + } + + // A genuinely new connection — nothing carried over in this process. + let reopened = connect_store_with_retry(&iso_url).expect("reconnect"); + let got = reopened.get_task("t-1").unwrap().expect( + "an in-flight task must survive a restart; got None back on a new connection, 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" + ); + + drop(reopened); +} + +/// `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 reconnect 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_reconnect() { + let Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_list"); + let iso_url = tmp.url(); + { + let store = connect_store_with_retry(&iso_url).expect("connect"); + store + .put_task(&sample_task("t-active", "working", 200)) + .unwrap(); + store + .put_task(&sample_task("t-waiting", "input-required", 201)) + .unwrap(); + store + .put_task(&sample_task("t-done", "completed", 202)) + .unwrap(); + store + .put_task(&sample_task("t-failed", "failed", 203)) + .unwrap(); + drop(store); + } + let reopened = connect_store_with_retry(&iso_url).expect("reconnect"); + 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" + ); + drop(reopened); +} + +/// The per-task provenance chain, read back on a new connection. 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_reconnect_and_still_links() { + let Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_events"); + let iso_url = tmp.url(); + { + let store = connect_store_with_retry(&iso_url).expect("connect"); + store + .append_task_event(&sample_event("t-1", 1, "task.submitted", "", "e1")) + .unwrap(); + store + .append_task_event(&sample_event("t-1", 2, "task.working", "e1", "e2")) + .unwrap(); + store + .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. + store + .append_task_event(&sample_event("t-2", 1, "task.submitted", "", "f1")) + .unwrap(); + drop(store); + } + let reopened = connect_store_with_retry(&iso_url).expect("reconnect"); + let got = reopened.list_task_events("t-1").unwrap(); + assert_eq!( + got.len(), + 3, + "the provenance chain must survive a reconnect; got {} events back on a new connection, \ + 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 reconnect: 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" + ); + drop(reopened); +} + +/// 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 Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_replay"); + let store = connect_store_with_retry(&tmp.url()).expect("connect"); + + let e = sample_event("t-1", 1, "task.submitted", "", "e1"); + store.append_task_event(&e).unwrap(); + store + .append_task_event(&e) + .expect("an identical replay must succeed, not be rejected as a fork"); + assert_eq!( + store.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(); + store.append_task_event(&corrected).unwrap(); + let got = store.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"); + drop(store); +} + +/// 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 Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_purge"); + let store = connect_store_with_retry(&tmp.url()).expect("connect"); + + store + .put_task(&sample_task("t-old-done", "completed", 100)) + .unwrap(); + store + .put_task(&sample_task("t-old-failed", "failed", 100)) + .unwrap(); + store + .put_task(&sample_task("t-old-canceled", "canceled", 100)) + .unwrap(); + store + .put_task(&sample_task("t-old-rejected", "rejected", 100)) + .unwrap(); + // Old, and NOT terminal — never dropped, no matter how old. + store + .put_task(&sample_task("t-old-waiting", "input-required", 100)) + .unwrap(); + store + .put_task(&sample_task("t-old-auth", "auth-required", 100)) + .unwrap(); + store + .put_task(&sample_task("t-old-working", "working", 100)) + .unwrap(); + store + .put_task(&sample_task("t-old-submitted", "submitted", 100)) + .unwrap(); + // A state token this build has never heard of — a newer engine's — must read as NOT terminal. + // The terminal set is closed and named rather than derived by negation, so the deleting half is + // never the half that guesses. + store + .put_task(&sample_task("t-old-unknown", "some-future-state", 100)) + .unwrap(); + // Terminal but at the cutoff exactly, and terminal but newer — both kept. + store + .put_task(&sample_task("t-at-cutoff", "completed", 200)) + .unwrap(); + store + .put_task(&sample_task("t-new-done", "completed", 300)) + .unwrap(); + + let purged = store.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 = store + .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-unknown", + "t-old-waiting", + "t-old-working", + ], + "an active, interrupted or unrecognised-state task is never dropped by retention, and \ + `before` is strictly less-than so a row exactly at the cutoff is kept" + ); + assert_eq!( + store.purge_tasks_before(200).unwrap(), + 0, + "re-running the same purge removes nothing" + ); + drop(store); +} + +/// 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 Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_cascade"); + let store = connect_store_with_retry(&tmp.url()).expect("connect"); + + store + .put_task(&sample_task("t-gone", "completed", 100)) + .unwrap(); + store + .put_task(&sample_task("t-stays", "working", 100)) + .unwrap(); + store + .append_task_event(&sample_event("t-gone", 1, "task.submitted", "", "g1")) + .unwrap(); + store + .append_task_event(&sample_event("t-gone", 2, "task.completed", "g1", "g2")) + .unwrap(); + store + .append_task_event(&sample_event("t-stays", 1, "task.submitted", "", "s1")) + .unwrap(); + + assert_eq!(store.purge_tasks_before(200).unwrap(), 1); + assert!( + store.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!( + store.list_task_events("t-stays").unwrap().len(), + 1, + "another task's chain must be untouched by that purge" + ); + drop(store); +} + +/// A `seq`/`ts`/`artifact_cursor` past `i64::MAX` cannot be stored faithfully in a signed Postgres +/// BIGINT — this crate's `clamp` pins it to `i64::MAX`, so the row read back would not be the row +/// written. Refused outright, exactly as `append_audit` refuses its own out-of-range `seq`/`ts`, +/// rather than silently mangled: a wrapped ARTIFACT CURSOR either replays delivered artifacts or +/// skips undelivered ones, and does it without an error ever having been reported. +#[test] +fn the_task_store_refuses_values_it_cannot_store_faithfully() { + let Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_range"); + let store = connect_store_with_retry(&tmp.url()).expect("connect"); + + let mut t = sample_task("t-1", "working", 200); + t.artifact_cursor = u64::MAX; + let err = store + .put_task(&t) + .expect_err("an artifact cursor past i64::MAX must be refused, not clamped"); + assert!( + err.0.contains("storable range"), + "the refusal must say why: {}", + err.0 + ); + assert!( + store.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!(store + .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!(store + .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; + store.put_task(&t).expect("i64::MAX is in range"); + assert_eq!( + store.get_task("t-1").unwrap().unwrap().artifact_cursor, + i64::MAX as u64 + ); + drop(store); +} + +/// The v7 -> v8 crossing is additive: a real v7 database gains `tasks` and `task_events` and keeps +/// every row it already had. Built as a genuine v7 database — the v7 schema, a real row in it, and +/// `busbar_schema` at 7 — rather than by deleting tables out of a current one, so the migration +/// under test is the one an existing deployment will actually run. +#[test] +fn migrate_v7_to_v8_adds_the_task_store_without_wiping_data() { + let Some(url) = live_url() else { return }; + let tmp = TempDb::create(&url, "spg_task_v7v8"); + let iso_url = tmp.url(); + + { + let mut c = postgres::Client::connect(&iso_url, postgres::NoTls).unwrap(); + c.batch_execute( + "CREATE TABLE busbar_schema (version BIGINT PRIMARY KEY); + INSERT INTO busbar_schema (version) VALUES (7); + CREATE TABLE keys ( + id TEXT PRIMARY KEY, + generation_hash TEXT NOT NULL, + name TEXT NOT NULL, + allowed_pools TEXT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + created_at BIGINT NOT NULL, + key_group TEXT, + labels TEXT NOT NULL DEFAULT '{}', + expires_at BIGINT, + deleted_at BIGINT, + revision BIGINT NOT NULL DEFAULT 0 + ); + INSERT INTO keys (id, generation_hash, name, created_at) VALUES ('vk_v7', 'g1', 'n', 0);", + ) + .unwrap(); + } + + let store = + connect_store_with_retry(&iso_url).expect("a v7 database must migrate additively to v8"); + assert!( + store.get_key("vk_v7").unwrap().is_some(), + "a real v7 key must survive the v7->v8 crossing" + ); + store + .put_task(&sample_task("t-1", "working", 200)) + .expect("the newly created tasks table must be writable after the migration"); + store + .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!(store.get_task("t-1").unwrap().is_some()); + assert_eq!(store.list_task_events("t-1").unwrap().len(), 1); + + let mut check = postgres::Client::connect(&iso_url, postgres::NoTls).unwrap(); + let version: i64 = check + .query_one("SELECT COALESCE(MAX(version), 0) FROM busbar_schema", &[]) + .unwrap() + .get(0); + assert_eq!(version, SCHEMA_VERSION); + + drop(store); + drop(check); +}