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        -- Deferred verdicts live in their OWN table, keyed by trace_id. They are NEVER folded
84        -- into the sealed, hashed `traces.body`, so a late outcome can't alter a past record and
85        -- the tamper-evident chain stays valid. They are merged onto a trace only on read.
86        CREATE TABLE IF NOT EXISTS deferred_verdicts (
87            id INTEGER PRIMARY KEY AUTOINCREMENT,
88            trace_id TEXT NOT NULL,
89            gate_id TEXT NOT NULL,
90            verdict TEXT NOT NULL,
91            score REAL,
92            reported_at TEXT NOT NULL,
93            reporter TEXT NOT NULL
94        );
95        CREATE INDEX IF NOT EXISTS deferred_trace_idx ON deferred_verdicts(trace_id);",
96    )?;
97    Ok(())
98}
99
100/// The writer's main loop: runs on a blocking-pool thread for the lifetime of the store.
101/// Never panics on a bad trace — logs and drops it, so one malformed record can't wedge the
102/// whole audit pipeline.
103fn writer_loop(conn: Connection, mut rx: mpsc::Receiver<Trace>) {
104    let mut head = match current_head(&conn) {
105        Ok(head) => head,
106        Err(err) => {
107            tracing::error!(%err, "trace writer: failed to load chain head, stopping");
108            return;
109        }
110    };
111
112    while let Some(mut trace) = rx.blocking_recv() {
113        trace.prev_hash = head.clone();
114        let hash = match trace.hash() {
115            Ok(hash) => hash,
116            Err(err) => {
117                tracing::error!(%err, trace_id = %trace.trace_id, "trace writer: failed to hash trace, dropping");
118                continue;
119            }
120        };
121        if let Err(err) = insert(&conn, &trace, &hash) {
122            tracing::error!(%err, trace_id = %trace.trace_id, "trace writer: failed to persist trace, dropping");
123            continue;
124        }
125        head = hash;
126    }
127}
128
129fn current_head(conn: &Connection) -> Result<String, StoreError> {
130    let mut stmt = conn.prepare("SELECT hash FROM traces ORDER BY seq DESC LIMIT 1")?;
131    let mut rows = stmt.query([])?;
132    match rows.next()? {
133        Some(row) => Ok(row.get(0)?),
134        None => Ok(GENESIS_HASH.to_owned()),
135    }
136}
137
138fn insert(conn: &Connection, trace: &Trace, hash: &str) -> Result<(), StoreError> {
139    let body = serde_json::to_string(trace)?;
140    conn.execute(
141        "INSERT INTO traces (trace_id, ts, prev_hash, hash, tenant, session, body)
142         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
143        rusqlite::params![
144            trace.trace_id.to_string(),
145            trace.ts.to_string(),
146            trace.prev_hash,
147            hash,
148            trace.tenant_id,
149            trace.session_id,
150            body,
151        ],
152    )?;
153    Ok(())
154}
155
156/// Load every trace from the database in insertion (chain) order — used by tests and
157/// operators to verify the hash chain with [`firstpass_core::verify_chain`].
158///
159/// # Errors
160/// Returns [`StoreError::Sqlite`] on a database error, or [`StoreError::Json`] if a stored
161/// row is not valid trace JSON.
162pub fn load_all_traces(db_path: impl AsRef<Path>) -> Result<Vec<Trace>, StoreError> {
163    let conn = connect(db_path.as_ref())?;
164    let mut stmt = conn.prepare("SELECT body FROM traces ORDER BY seq ASC")?;
165    let rows = stmt.query_map([], |row| row.get::<_, String>(0))?;
166    let mut traces = Vec::new();
167    for row in rows {
168        traces.push(serde_json::from_str(&row?)?);
169    }
170    Ok(traces)
171}
172
173/// Whether a trace with `trace_id` exists — used to reject feedback for unknown traces.
174///
175/// # Errors
176/// Returns [`StoreError::Sqlite`] on a database error.
177pub fn trace_exists(db_path: impl AsRef<Path>, trace_id: &str) -> Result<bool, StoreError> {
178    let conn = connect(db_path.as_ref())?;
179    let n: i64 = conn.query_row(
180        "SELECT COUNT(1) FROM traces WHERE trace_id = ?1",
181        [trace_id],
182        |row| row.get(0),
183    )?;
184    Ok(n > 0)
185}
186
187/// Append a deferred verdict for `trace_id` (a downstream outcome or async gate result). This
188/// writes ONLY to the `deferred_verdicts` table; the sealed trace and its hash are untouched, so
189/// the audit chain remains verifiable (SPEC §8.3.4 — the outcome-feedback loop).
190///
191/// # Errors
192/// Returns [`StoreError::Sqlite`] on a database error.
193pub fn append_deferred(
194    db_path: impl AsRef<Path>,
195    trace_id: &str,
196    v: &DeferredVerdict,
197) -> Result<(), StoreError> {
198    let conn = connect(db_path.as_ref())?;
199    conn.execute(
200        "INSERT INTO deferred_verdicts (trace_id, gate_id, verdict, score, reported_at, reporter)
201         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
202        rusqlite::params![
203            trace_id,
204            v.gate_id,
205            v.verdict.as_str(),
206            v.score.map(Score::value),
207            v.reported_at.to_string(),
208            v.reporter,
209        ],
210    )?;
211    Ok(())
212}
213
214/// Load the deferred verdicts recorded for `trace_id`, oldest first. Malformed stored rows are
215/// skipped (logged), never fatal — a corrupt late outcome must not break reading the trace.
216///
217/// # Errors
218/// Returns [`StoreError::Sqlite`] on a database error.
219pub fn load_deferred(
220    db_path: impl AsRef<Path>,
221    trace_id: &str,
222) -> Result<Vec<DeferredVerdict>, StoreError> {
223    let conn = connect(db_path.as_ref())?;
224    let mut stmt = conn.prepare(
225        "SELECT gate_id, verdict, score, reported_at, reporter
226         FROM deferred_verdicts WHERE trace_id = ?1 ORDER BY id ASC",
227    )?;
228    let rows = stmt.query_map([trace_id], |row| {
229        Ok((
230            row.get::<_, String>(0)?,
231            row.get::<_, String>(1)?,
232            row.get::<_, Option<f64>>(2)?,
233            row.get::<_, String>(3)?,
234            row.get::<_, String>(4)?,
235        ))
236    })?;
237
238    let mut out = Vec::new();
239    for row in rows {
240        let (gate_id, verdict_s, score, reported_s, reporter) = row?;
241        let verdict = match verdict_s.as_str() {
242            "pass" => Verdict::Pass,
243            "fail" => Verdict::Fail,
244            "abstain" => Verdict::Abstain,
245            other => {
246                tracing::warn!(verdict = %other, %trace_id, "skipping deferred row with bad verdict");
247                continue;
248            }
249        };
250        let Ok(reported_at) = reported_s.parse::<jiff::Timestamp>() else {
251            tracing::warn!(%trace_id, "skipping deferred row with bad timestamp");
252            continue;
253        };
254        out.push(DeferredVerdict {
255            gate_id,
256            verdict,
257            score: score.and_then(|s| Score::new(s).ok()),
258            reported_at,
259            reporter,
260        });
261    }
262    Ok(out)
263}
264
265/// Load a single trace by id with its deferred verdicts merged into `deferred` — the **view**
266/// for display/inspection. This is deliberately separate from [`load_all_traces`]: merging
267/// deferred verdicts changes the record, so a merged trace must NOT be fed to `verify_chain`
268/// (chain verification always runs on the sealed bodies from [`load_all_traces`]).
269///
270/// # Errors
271/// Returns [`StoreError::Sqlite`] / [`StoreError::Json`] on database or decode errors.
272pub fn load_trace_view(
273    db_path: impl AsRef<Path>,
274    trace_id: &str,
275) -> Result<Option<Trace>, StoreError> {
276    let conn = connect(db_path.as_ref())?;
277    let body: Option<String> = conn
278        .query_row(
279            "SELECT body FROM traces WHERE trace_id = ?1",
280            [trace_id],
281            |row| row.get(0),
282        )
283        .ok();
284    let Some(body) = body else { return Ok(None) };
285    let mut trace: Trace = serde_json::from_str(&body)?;
286    trace.deferred = load_deferred(db_path, trace_id)?;
287    Ok(Some(trace))
288}
289
290#[cfg(test)]
291mod tests {
292    use firstpass_core::{
293        Attempt, Features, FinalOutcome, GENESIS_HASH, PolicyRef, RequestInfo, ServedFrom,
294        TaskKind, Verdict, verify_chain,
295    };
296
297    use super::*;
298
299    fn sample_trace(tenant: &str, session: &str) -> Trace {
300        let attempt = Attempt {
301            rung: 0,
302            model: "claude-haiku-4-5".to_owned(),
303            provider: "anthropic".to_owned(),
304            in_tokens: 10,
305            out_tokens: 5,
306            cost_usd: 0.001,
307            latency_ms: 12,
308            gates: vec![],
309            verdict: Verdict::Pass,
310        };
311        let mut trace = Trace {
312            trace_id: uuid::Uuid::now_v7(),
313            prev_hash: GENESIS_HASH.to_owned(),
314            tenant_id: tenant.to_owned(),
315            session_id: session.to_owned(),
316            ts: jiff::Timestamp::now(),
317            mode: firstpass_core::Mode::Observe,
318            policy: PolicyRef {
319                id: "observe-passthrough@v0".to_owned(),
320                explore: false,
321            },
322            request: RequestInfo {
323                api: "anthropic.messages".to_owned(),
324                prompt_hash: "deadbeef".to_owned(),
325                features: Features::new(TaskKind::Other),
326            },
327            attempts: vec![attempt],
328            deferred: Vec::new(),
329            final_: FinalOutcome {
330                served_rung: Some(0),
331                served_from: ServedFrom::Attempt,
332                total_cost_usd: 0.001,
333                gate_cost_usd: 0.0,
334                total_latency_ms: 12,
335                escalations: 0,
336                counterfactual_baseline_usd: 0.001,
337                savings_usd: 0.0,
338            },
339        };
340        trace.recompute_savings();
341        trace
342    }
343
344    #[tokio::test]
345    async fn writer_assigns_prev_hash_and_forms_a_valid_chain() {
346        let db_path =
347            std::env::temp_dir().join(format!("firstpass-store-test-{}.db", uuid::Uuid::now_v7()));
348        let (tx, handle) = open(&db_path).unwrap();
349
350        tx.try_send(sample_trace("tenant-a", "session-1")).unwrap();
351        tx.try_send(sample_trace("tenant-a", "session-1")).unwrap();
352        drop(tx);
353        handle.await.unwrap();
354
355        let traces = load_all_traces(&db_path).unwrap();
356        assert_eq!(traces.len(), 2);
357        assert_eq!(traces[0].prev_hash, GENESIS_HASH);
358        assert_eq!(traces[1].prev_hash, traces[0].hash().unwrap());
359        verify_chain(&traces, GENESIS_HASH).unwrap();
360
361        let _ = std::fs::remove_file(&db_path);
362    }
363
364    #[tokio::test]
365    async fn deferred_verdicts_attach_on_read_without_breaking_the_chain() {
366        let db_path = std::env::temp_dir().join(format!(
367            "firstpass-deferred-test-{}.db",
368            uuid::Uuid::now_v7()
369        ));
370        let (tx, handle) = open(&db_path).unwrap();
371        let t0 = sample_trace("acme", "run-1");
372        let t1 = sample_trace("acme", "run-1");
373        let (id0, id1) = (t0.trace_id.to_string(), t1.trace_id.to_string());
374        tx.try_send(t0).unwrap();
375        tx.try_send(t1).unwrap();
376        drop(tx);
377        handle.await.unwrap();
378
379        // A downstream outcome arrives for the first trace (e.g. "tests passed an hour later").
380        let dv = DeferredVerdict {
381            gate_id: "tests".to_owned(),
382            verdict: Verdict::Pass,
383            score: Some(Score::new(1.0).unwrap()),
384            reported_at: jiff::Timestamp::now(),
385            reporter: "ci".to_owned(),
386        };
387        append_deferred(&db_path, &id0, &dv).unwrap();
388        assert!(trace_exists(&db_path, &id0).unwrap());
389        assert!(!trace_exists(&db_path, "no-such-trace").unwrap());
390
391        // The view surfaces the deferred verdict...
392        let view = load_trace_view(&db_path, &id0).unwrap().unwrap();
393        assert_eq!(view.deferred.len(), 1);
394        assert_eq!(view.deferred[0].gate_id, "tests");
395        assert_eq!(view.deferred[0].verdict, Verdict::Pass);
396        // ...the second trace has none.
397        assert!(
398            load_trace_view(&db_path, &id1)
399                .unwrap()
400                .unwrap()
401                .deferred
402                .is_empty()
403        );
404
405        // THE INVARIANT: the sealed bodies are untouched, so the chain still verifies. A late
406        // outcome can never alter a past decision's hash.
407        let traces = load_all_traces(&db_path).unwrap();
408        assert!(
409            traces.iter().all(|t| t.deferred.is_empty()),
410            "sealed records stay deferred-free"
411        );
412        verify_chain(&traces, GENESIS_HASH).unwrap();
413
414        let _ = std::fs::remove_file(&db_path);
415    }
416}
firstpass_proxy/store.rs

firstpass_proxy/
store.rs
