firstpass_proxy/
store.rs

1//! Tamper-evident trace storage: a background SQLite writer fed by a bounded channel.
2//!
3//! The hot request path never blocks on disk: it `try_send`s a [`Trace`] down a bounded channel
4//! and returns immediately (dropping the trace if the writer has fallen far enough behind to fill
5//! the buffer — bounded memory over a guaranteed write). A single background task owns the SQLite
6//! connection, assigns
7//! each trace's `prev_hash` from the current chain head, computes its `hash`, and appends it
8//! (SPEC §9: the hash chain is only meaningful if every writer agrees on chain order, which a
9//! single writer task guarantees for free).
10
11use std::path::Path;
12use std::time::Duration;
13
14use firstpass_core::{DeferredVerdict, GENESIS_HASH, Score, Trace, Verdict};
15use rusqlite::Connection;
16use tokio::sync::mpsc;
17use tokio::task::JoinHandle;
18
19/// Open a connection with WAL + a busy timeout, so the background writer and short-lived
20/// feedback/read connections can share the file without "database is locked" errors.
21fn connect(db_path: impl AsRef<Path>) -> Result<Connection, StoreError> {
22    let conn = Connection::open(db_path.as_ref())?;
23    conn.busy_timeout(Duration::from_secs(5))?;
24    Ok(conn)
25}
26
27/// Errors from the trace store.
28#[derive(Debug, thiserror::Error)]
29pub enum StoreError {
30    /// The SQLite database could not be opened, migrated, or queried.
31    #[error("sqlite error: {0}")]
32    Sqlite(#[from] rusqlite::Error),
33    /// A trace could not be hashed or (de)serialized.
34    #[error("trace error: {0}")]
35    Trace(#[from] firstpass_core::Error),
36    /// A stored row was not valid trace JSON.
37    #[error("json error: {0}")]
38    Json(#[from] serde_json::Error),
39}
40
41/// Sending half of the trace channel; cheap to clone, safe to share across request handlers.
42/// Sending is fire-and-forget via `try_send` — the hot path never awaits the writer, and a bounded
43/// buffer means a stalled writer sheds load (drops traces) instead of growing memory without limit.
44pub type TraceSender = mpsc::Sender<Trace>;
45
46/// Trace buffer depth. Deep enough to absorb normal write bursts; bounded so a wedged writer (disk
47/// stall) can't OOM the process — excess traces are dropped with a warning, not queued forever.
48pub const TRACE_CHANNEL_CAP: usize = 8192;
49
50/// Open (creating if needed) the SQLite trace database, migrate its schema, and spawn the
51/// background writer task.
52///
53/// Returns a [`TraceSender`] for the hot path and the writer's [`JoinHandle`]. The writer
54/// exits cleanly once every clone of the sender is dropped.
55///
56/// # Errors
57/// Returns [`StoreError::Sqlite`] if the database cannot be opened or migrated.
58pub fn open(db_path: impl AsRef<Path>) -> Result<(TraceSender, JoinHandle<()>), StoreError> {
59    let conn = connect(db_path.as_ref())?;
60    migrate(&conn)?;
61
62    let (tx, rx) = mpsc::channel::<Trace>(TRACE_CHANNEL_CAP);
63    let handle = tokio::task::spawn_blocking(move || writer_loop(conn, rx));
64    Ok((tx, handle))
65}
66
67fn migrate(conn: &Connection) -> Result<(), StoreError> {
68    // WAL: lets the background writer and short-lived feedback/read connections share the file
69    // concurrently. `journal_mode` is persisted in the file header once set by any connection.
70    conn.pragma_update(None, "journal_mode", "WAL")?;
71    conn.execute_batch(
72        "CREATE TABLE IF NOT EXISTS traces (
73            seq INTEGER PRIMARY KEY AUTOINCREMENT,
74            trace_id TEXT NOT NULL,
75            ts TEXT NOT NULL,
76            prev_hash TEXT NOT NULL,
77            hash TEXT NOT NULL,
78            tenant TEXT NOT NULL,
79            session TEXT NOT NULL,
80            body TEXT NOT NULL
81        );
82        CREATE INDEX IF NOT EXISTS traces_session_idx ON traces(session);
83        -- Tenant-scoped reads (ADR 0004 §D3) filter on `tenant`; index it with `seq` so a
84        -- per-tenant scan stays ordered and cheap. Existing rows keep their `tenant` value.
85        CREATE INDEX IF NOT EXISTS traces_tenant_seq_idx ON traces(tenant, seq);
86        -- Deferred verdicts live in their OWN table, keyed by trace_id. They are NEVER folded
87        -- into the sealed, hashed `traces.body`, so a late outcome can't alter a past record and
88        -- the tamper-evident chain stays valid. They are merged onto a trace only on read.
89        CREATE TABLE IF NOT EXISTS deferred_verdicts (
90            id INTEGER PRIMARY KEY AUTOINCREMENT,
91            trace_id TEXT NOT NULL,
92            gate_id TEXT NOT NULL,
93            verdict TEXT NOT NULL,
94            score REAL,
95            reported_at TEXT NOT NULL,
96            reporter TEXT NOT NULL
97        );
98        CREATE INDEX IF NOT EXISTS deferred_trace_idx ON deferred_verdicts(trace_id);",
99    )?;
100    Ok(())
101}
102
103/// The writer's main loop: runs on a blocking-pool thread for the lifetime of the store.
104/// Never panics on a bad trace — logs and drops it, so one malformed record can't wedge the
105/// whole audit pipeline.
106fn writer_loop(conn: Connection, mut rx: mpsc::Receiver<Trace>) {
107    let mut head = match current_head(&conn) {
108        Ok(head) => head,
109        Err(err) => {
110            tracing::error!(%err, "trace writer: failed to load chain head, stopping");
111            return;
112        }
113    };
114
115    while let Some(mut trace) = rx.blocking_recv() {
116        trace.prev_hash = head.clone();
117        let hash = match trace.hash() {
118            Ok(hash) => hash,
119            Err(err) => {
120                tracing::error!(%err, trace_id = %trace.trace_id, "trace writer: failed to hash trace, dropping");
121                continue;
122            }
123        };
124        if let Err(err) = insert(&conn, &trace, &hash) {
125            tracing::error!(%err, trace_id = %trace.trace_id, "trace writer: failed to persist trace, dropping");
126            continue;
127        }
128        head = hash;
129    }
130}
131
132fn current_head(conn: &Connection) -> Result<String, StoreError> {
133    let mut stmt = conn.prepare("SELECT hash FROM traces ORDER BY seq DESC LIMIT 1")?;
134    let mut rows = stmt.query([])?;
135    match rows.next()? {
136        Some(row) => Ok(row.get(0)?),
137        None => Ok(GENESIS_HASH.to_owned()),
138    }
139}
140
141fn insert(conn: &Connection, trace: &Trace, hash: &str) -> Result<(), StoreError> {
142    let body = serde_json::to_string(trace)?;
143    conn.execute(
144        "INSERT INTO traces (trace_id, ts, prev_hash, hash, tenant, session, body)
145         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
146        rusqlite::params![
147            trace.trace_id.to_string(),
148            trace.ts.to_string(),
149            trace.prev_hash,
150            hash,
151            trace.tenant_id,
152            trace.session_id,
153            body,
154        ],
155    )?;
156    Ok(())
157}
158
159/// Load every trace from the database in insertion (chain) order — used by tests and
160/// operators to verify the hash chain with [`firstpass_core::verify_chain`].
161///
162/// **Operator-wide** read: every trace across ALL tenants, in `seq` order.
163///
164/// This deliberately crosses tenant boundaries and must stay reserved for operator-scoped work
165/// where a global view is intrinsic — namely verifying the single hash-chain, which spans every
166/// tenant's traces in one sequence (ADR 0004 §D3). For tenant-facing reads use
167/// [`load_tenant_traces`].
168///
169/// # Errors
170/// Returns [`StoreError::Sqlite`] on a database error, or [`StoreError::Json`] if a stored
171/// row is not valid trace JSON.
172pub fn load_all_traces(db_path: impl AsRef<Path>) -> Result<Vec<Trace>, StoreError> {
173    let conn = connect(db_path.as_ref())?;
174    let mut stmt = conn.prepare("SELECT body FROM traces ORDER BY seq ASC")?;
175    let rows = stmt.query_map([], |row| row.get::<_, String>(0))?;
176    let mut traces = Vec::new();
177    for row in rows {
178        traces.push(serde_json::from_str(&row?)?);
179    }
180    Ok(traces)
181}
182
183/// **Tenant-scoped** read: only traces owned by `tenant`, in `seq` order (ADR 0004 §D3). Tenant A
184/// can never see tenant B's traces through this path.
185///
186/// # Errors
187/// Returns [`StoreError::Sqlite`] on a database error, or [`StoreError::Json`] if a stored
188/// row is not valid trace JSON.
189pub fn load_tenant_traces(
190    db_path: impl AsRef<Path>,
191    tenant: &str,
192) -> Result<Vec<Trace>, StoreError> {
193    let conn = connect(db_path.as_ref())?;
194    let mut stmt = conn.prepare("SELECT body FROM traces WHERE tenant = ?1 ORDER BY seq ASC")?;
195    let rows = stmt.query_map([tenant], |row| row.get::<_, String>(0))?;
196    let mut traces = Vec::new();
197    for row in rows {
198        traces.push(serde_json::from_str(&row?)?);
199    }
200    Ok(traces)
201}
202
203/// Whether a trace with `trace_id` exists **and is owned by `tenant`** — used to reject feedback
204/// for unknown traces and, crucially, to deny cross-tenant feedback (ADR 0004 §D3/§D4). A trace
205/// owned by another tenant is indistinguishable from a non-existent one here, so the caller can
206/// return a `404` with no existence oracle.
207///
208/// # Errors
209/// Returns [`StoreError::Sqlite`] on a database error.
210pub fn trace_exists(
211    db_path: impl AsRef<Path>,
212    tenant: &str,
213    trace_id: &str,
214) -> Result<bool, StoreError> {
215    let conn = connect(db_path.as_ref())?;
216    let n: i64 = conn.query_row(
217        "SELECT COUNT(1) FROM traces WHERE tenant = ?1 AND trace_id = ?2",
218        [tenant, trace_id],
219        |row| row.get(0),
220    )?;
221    Ok(n > 0)
222}
223
224/// Append a deferred verdict for `trace_id` (a downstream outcome or async gate result). This
225/// writes ONLY to the `deferred_verdicts` table; the sealed trace and its hash are untouched, so
226/// the audit chain remains verifiable (SPEC §8.3.4 — the outcome-feedback loop).
227///
228/// # Errors
229/// Returns [`StoreError::Sqlite`] on a database error.
230pub fn append_deferred(
231    db_path: impl AsRef<Path>,
232    trace_id: &str,
233    v: &DeferredVerdict,
234) -> Result<(), StoreError> {
235    let conn = connect(db_path.as_ref())?;
236    conn.execute(
237        "INSERT INTO deferred_verdicts (trace_id, gate_id, verdict, score, reported_at, reporter)
238         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
239        rusqlite::params![
240            trace_id,
241            v.gate_id,
242            v.verdict.as_str(),
243            v.score.map(Score::value),
244            v.reported_at.to_string(),
245            v.reporter,
246        ],
247    )?;
248    Ok(())
249}
250
251/// Load the deferred verdicts recorded for `trace_id`, oldest first. Malformed stored rows are
252/// skipped (logged), never fatal — a corrupt late outcome must not break reading the trace.
253///
254/// # Errors
255/// Returns [`StoreError::Sqlite`] on a database error.
256pub fn load_deferred(
257    db_path: impl AsRef<Path>,
258    trace_id: &str,
259) -> Result<Vec<DeferredVerdict>, StoreError> {
260    let conn = connect(db_path.as_ref())?;
261    let mut stmt = conn.prepare(
262        "SELECT gate_id, verdict, score, reported_at, reporter
263         FROM deferred_verdicts WHERE trace_id = ?1 ORDER BY id ASC",
264    )?;
265    let rows = stmt.query_map([trace_id], |row| {
266        Ok((
267            row.get::<_, String>(0)?,
268            row.get::<_, String>(1)?,
269            row.get::<_, Option<f64>>(2)?,
270            row.get::<_, String>(3)?,
271            row.get::<_, String>(4)?,
272        ))
273    })?;
274
275    let mut out = Vec::new();
276    for row in rows {
277        let (gate_id, verdict_s, score, reported_s, reporter) = row?;
278        let verdict = match verdict_s.as_str() {
279            "pass" => Verdict::Pass,
280            "fail" => Verdict::Fail,
281            "abstain" => Verdict::Abstain,
282            other => {
283                tracing::warn!(verdict = %other, %trace_id, "skipping deferred row with bad verdict");
284                continue;
285            }
286        };
287        let Ok(reported_at) = reported_s.parse::<jiff::Timestamp>() else {
288            tracing::warn!(%trace_id, "skipping deferred row with bad timestamp");
289            continue;
290        };
291        out.push(DeferredVerdict {
292            gate_id,
293            verdict,
294            score: score.and_then(|s| Score::new(s).ok()),
295            reported_at,
296            reporter,
297        });
298    }
299    Ok(out)
300}
301
302/// Load a single trace by id **scoped to `tenant`**, with its deferred verdicts merged into
303/// `deferred` — the **view** for display/inspection (ADR 0004 §D3). A trace owned by another
304/// tenant returns `None`, exactly like a missing one, so an inspecting agent can never read across
305/// tenants. This is deliberately separate from [`load_all_traces`]: merging deferred verdicts
306/// changes the record, so a merged trace must NOT be fed to `verify_chain` (chain verification
307/// always runs on the sealed bodies from [`load_all_traces`]).
308///
309/// # Errors
310/// Returns [`StoreError::Sqlite`] / [`StoreError::Json`] on database or decode errors.
311pub fn load_trace_view(
312    db_path: impl AsRef<Path>,
313    tenant: &str,
314    trace_id: &str,
315) -> Result<Option<Trace>, StoreError> {
316    let conn = connect(db_path.as_ref())?;
317    let body: Option<String> = conn
318        .query_row(
319            "SELECT body FROM traces WHERE tenant = ?1 AND trace_id = ?2",
320            [tenant, trace_id],
321            |row| row.get(0),
322        )
323        .ok();
324    let Some(body) = body else { return Ok(None) };
325    let mut trace: Trace = serde_json::from_str(&body)?;
326    trace.deferred = load_deferred(db_path, trace_id)?;
327    Ok(Some(trace))
328}
329
330#[cfg(test)]
331mod tests {
332    use firstpass_core::{
333        Attempt, Features, FinalOutcome, GENESIS_HASH, PolicyRef, RequestInfo, ServedFrom,
334        TaskKind, Verdict, verify_chain,
335    };
336
337    use super::*;
338
339    fn sample_trace(tenant: &str, session: &str) -> Trace {
340        let attempt = Attempt {
341            rung: 0,
342            model: "claude-haiku-4-5".to_owned(),
343            provider: "anthropic".to_owned(),
344            in_tokens: 10,
345            out_tokens: 5,
346            cost_usd: 0.001,
347            latency_ms: 12,
348            gates: vec![],
349            verdict: Verdict::Pass,
350        };
351        let mut trace = Trace {
352            trace_id: uuid::Uuid::now_v7(),
353            prev_hash: GENESIS_HASH.to_owned(),
354            tenant_id: tenant.to_owned(),
355            session_id: session.to_owned(),
356            ts: jiff::Timestamp::now(),
357            mode: firstpass_core::Mode::Observe,
358            policy: PolicyRef {
359                id: "observe-passthrough@v0".to_owned(),
360                explore: false,
361            },
362            request: RequestInfo {
363                api: "anthropic.messages".to_owned(),
364                prompt_hash: "deadbeef".to_owned(),
365                features: Features::new(TaskKind::Other),
366            },
367            attempts: vec![attempt],
368            deferred: Vec::new(),
369            final_: FinalOutcome {
370                served_rung: Some(0),
371                served_from: ServedFrom::Attempt,
372                total_cost_usd: 0.001,
373                gate_cost_usd: 0.0,
374                total_latency_ms: 12,
375                escalations: 0,
376                counterfactual_baseline_usd: 0.001,
377                savings_usd: 0.0,
378            },
379        };
380        trace.recompute_savings();
381        trace
382    }
383
384    #[tokio::test]
385    async fn writer_assigns_prev_hash_and_forms_a_valid_chain() {
386        let db_path =
387            std::env::temp_dir().join(format!("firstpass-store-test-{}.db", uuid::Uuid::now_v7()));
388        let (tx, handle) = open(&db_path).unwrap();
389
390        tx.try_send(sample_trace("tenant-a", "session-1")).unwrap();
391        tx.try_send(sample_trace("tenant-a", "session-1")).unwrap();
392        drop(tx);
393        handle.await.unwrap();
394
395        let traces = load_all_traces(&db_path).unwrap();
396        assert_eq!(traces.len(), 2);
397        assert_eq!(traces[0].prev_hash, GENESIS_HASH);
398        assert_eq!(traces[1].prev_hash, traces[0].hash().unwrap());
399        verify_chain(&traces, GENESIS_HASH).unwrap();
400
401        let _ = std::fs::remove_file(&db_path);
402    }
403
404    /// D7 cross-tenant isolation, at the store layer: with rows for tenants A and B, every
405    /// tenant-scoped read for A returns only A's data, and vice-versa. The operator-wide
406    /// [`load_all_traces`] still sees both (for chain verification).
407    #[tokio::test]
408    async fn tenant_scoped_reads_never_cross_the_boundary() {
409        let db_path =
410            std::env::temp_dir().join(format!("firstpass-isolation-{}.db", uuid::Uuid::now_v7()));
411        let (tx, handle) = open(&db_path).unwrap();
412
413        // Two traces for A, one for B.
414        let a0 = sample_trace("tenant-a", "sa-0");
415        let a1 = sample_trace("tenant-a", "sa-1");
416        let b0 = sample_trace("tenant-b", "sb-0");
417        let (a0_id, a1_id, b0_id) = (
418            a0.trace_id.to_string(),
419            a1.trace_id.to_string(),
420            b0.trace_id.to_string(),
421        );
422        tx.try_send(a0).unwrap();
423        tx.try_send(b0).unwrap();
424        tx.try_send(a1).unwrap();
425        drop(tx);
426        handle.await.unwrap();
427
428        // Scoped list: A sees exactly its two, B sees exactly its one.
429        let a_traces = load_tenant_traces(&db_path, "tenant-a").unwrap();
430        assert_eq!(a_traces.len(), 2, "A must see only A's traces");
431        assert!(a_traces.iter().all(|t| t.tenant_id == "tenant-a"));
432        let b_traces = load_tenant_traces(&db_path, "tenant-b").unwrap();
433        assert_eq!(b_traces.len(), 1, "B must see only B's trace");
434        assert!(b_traces.iter().all(|t| t.tenant_id == "tenant-b"));
435
436        // A can prove its own trace exists but cannot see B's, and vice-versa.
437        assert!(trace_exists(&db_path, "tenant-a", &a0_id).unwrap());
438        assert!(!trace_exists(&db_path, "tenant-a", &b0_id).unwrap());
439        assert!(trace_exists(&db_path, "tenant-b", &b0_id).unwrap());
440        assert!(!trace_exists(&db_path, "tenant-b", &a1_id).unwrap());
441
442        // The view is likewise scoped: cross-tenant reads are indistinguishable from a miss.
443        assert!(
444            load_trace_view(&db_path, "tenant-a", &a0_id)
445                .unwrap()
446                .is_some()
447        );
448        assert!(
449            load_trace_view(&db_path, "tenant-a", &b0_id)
450                .unwrap()
451                .is_none()
452        );
453        assert!(
454            load_trace_view(&db_path, "tenant-b", &a0_id)
455                .unwrap()
456                .is_none()
457        );
458
459        // A tenant that owns nothing sees nothing.
460        assert!(load_tenant_traces(&db_path, "ghost").unwrap().is_empty());
461
462        // Operator-wide read still spans both, and the global chain stays valid.
463        let all = load_all_traces(&db_path).unwrap();
464        assert_eq!(all.len(), 3);
465        verify_chain(&all, GENESIS_HASH).unwrap();
466
467        let _ = std::fs::remove_file(&db_path);
468    }
469
470    #[tokio::test]
471    async fn deferred_verdicts_attach_on_read_without_breaking_the_chain() {
472        let db_path = std::env::temp_dir().join(format!(
473            "firstpass-deferred-test-{}.db",
474            uuid::Uuid::now_v7()
475        ));
476        let (tx, handle) = open(&db_path).unwrap();
477        let t0 = sample_trace("acme", "run-1");
478        let t1 = sample_trace("acme", "run-1");
479        let (id0, id1) = (t0.trace_id.to_string(), t1.trace_id.to_string());
480        tx.try_send(t0).unwrap();
481        tx.try_send(t1).unwrap();
482        drop(tx);
483        handle.await.unwrap();
484
485        // A downstream outcome arrives for the first trace (e.g. "tests passed an hour later").
486        let dv = DeferredVerdict {
487            gate_id: "tests".to_owned(),
488            verdict: Verdict::Pass,
489            score: Some(Score::new(1.0).unwrap()),
490            reported_at: jiff::Timestamp::now(),
491            reporter: "ci".to_owned(),
492        };
493        append_deferred(&db_path, &id0, &dv).unwrap();
494        assert!(trace_exists(&db_path, "acme", &id0).unwrap());
495        assert!(!trace_exists(&db_path, "acme", "no-such-trace").unwrap());
496        // Cross-tenant: the same real trace id is invisible to a different tenant.
497        assert!(!trace_exists(&db_path, "other-tenant", &id0).unwrap());
498
499        // The view surfaces the deferred verdict...
500        let view = load_trace_view(&db_path, "acme", &id0).unwrap().unwrap();
501        assert_eq!(view.deferred.len(), 1);
502        assert_eq!(view.deferred[0].gate_id, "tests");
503        assert_eq!(view.deferred[0].verdict, Verdict::Pass);
504        // ...the second trace has none.
505        assert!(
506            load_trace_view(&db_path, "acme", &id1)
507                .unwrap()
508                .unwrap()
509                .deferred
510                .is_empty()
511        );
512        // Cross-tenant: another tenant cannot read this trace's view at all.
513        assert!(
514            load_trace_view(&db_path, "other-tenant", &id0)
515                .unwrap()
516                .is_none()
517        );
518
519        // THE INVARIANT: the sealed bodies are untouched, so the chain still verifies. A late
520        // outcome can never alter a past decision's hash.
521        let traces = load_all_traces(&db_path).unwrap();
522        assert!(
523            traces.iter().all(|t| t.deferred.is_empty()),
524            "sealed records stay deferred-free"
525        );
526        verify_chain(&traces, GENESIS_HASH).unwrap();
527
528        let _ = std::fs::remove_file(&db_path);
529    }
530}
firstpass_proxy/store.rs

firstpass_proxy/
store.rs
