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..87ec4e2 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, TaskEventRow, TaskRow, + TierTokens, UsageDelta, UsageLedger, VirtualKey, }; use postgres::types::ToSql; use postgres::{Client, NoTls, Row, Transaction}; @@ -232,7 +232,23 @@ 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. +/// +/// 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 ( @@ -337,6 +353,130 @@ 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); + +-- 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 — @@ -1344,6 +1484,206 @@ 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 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()) @@ -1352,6 +1692,103 @@ 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"; + +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 +/// 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..802f305 100644 --- a/store-postgres/src/tests.rs +++ b/store-postgres/src/tests.rs @@ -6,7 +6,10 @@ //! `postgres:16`. use super::*; -use busbar_api::{CredentialMeta, CredentialSecret, ModelTokensDelta, SecretForm, TierTokensDelta}; +use busbar_api::{ + CredentialMeta, CredentialSecret, McpCallRecord, ModelTokensDelta, SecretForm, TaskEventRow, + TaskRow, 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 +2132,719 @@ 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]); +} + +// ── 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); +}