Skip to main content

sysknife_daemon/
transactions.rs

1use crate::audit_chain::{
2    self, AuditEventKind, AuditKey, ChainContent, ChainIdentity, ChainRow, EventContent, EventRow,
3    VerifyOutcome, CURRENT_KEY_ID,
4};
5use crate::audit_watermark::emit_chain_tip_watermark;
6use crate::auth::CallerPrincipal;
7use rusqlite::{params, Connection, TransactionBehavior};
8use serde::{de::DeserializeOwned, Deserialize, Serialize};
9use std::path::{Path, PathBuf};
10use std::sync::Arc;
11use subtle::ConstantTimeEq;
12use sysknife_types::{CallerRole, JobState, PreviewEnvelope, RiskLevel, TransactionRecord};
13use uuid::Uuid;
14
15/// Lifetime of an approval receipt / preview, in minutes.
16///
17/// A transaction can be approved, its receipt claimed for execution, or it is
18/// swept as stale, only while `created_at` is within this window. It is the
19/// single source of truth for the TTL: both the SQLite (`julianday`) and the
20/// PostgreSQL (`INTERVAL`) backends interpolate this constant into their SQL,
21/// so the two engines can never disagree on the window. 15 minutes balances
22/// operator usability (time to run `sysknife approve` in a terminal) against
23/// the exposure window of a single-use bearer receipt. Prose in `SECURITY.md`
24/// cites the same "15-minute" value; keep them in sync if this changes.
25pub(crate) const APPROVAL_RECEIPT_TTL_MINUTES: i64 = 15;
26
27/// One structured audit-log row for the history IPC.
28///
29/// Unlike [`TransactionRecord`] (which crosses the proto boundary and omits the
30/// creation timestamp), this is a serde-only DTO carried over the JSON daemon
31/// wire. It exists so programmatic clients (the MCP `sysknife_history` tool)
32/// get typed `risk_level` and `created_at` instead of re-parsing formatted
33/// text. `created_at` is the ISO-8601 timestamp stored at insert time.
34#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
35pub struct JobHistoryEntry {
36    pub transaction_id: String,
37    pub action_name: String,
38    pub risk_level: RiskLevel,
39    pub status: JobState,
40    pub summary: String,
41    pub created_at: String,
42}
43
44/// Data provided by the caller when recording a new transaction.
45///
46/// The initial `status` is always `Queued` — it is not caller-controllable.
47/// Hardcoding this in the store prevents callers from bypassing the state
48/// machine by recording a transaction in a terminal state.
49#[derive(Clone, Debug, PartialEq, Eq)]
50pub struct NewTransaction {
51    pub request_id: String,
52    pub request_hash: String,
53    pub action_name: String,
54    pub risk_level: RiskLevel,
55    /// Human-readable description of the planned action.
56    ///
57    /// **Chain-hashed at INSERT; intentionally not in the mutable field set.**
58    ///
59    /// `summary` is captured in [`crate::audit_chain::ChainContent`] and
60    /// baked into `chain_hash = ed25519_sign(canonical(fields) || prev_hash, key)`
61    /// at the moment the row is written. After that point the stored signature
62    /// is a one-time commitment.
63    ///
64    /// **Do not add an `update_summary` API** (or any equivalent that modifies
65    /// this field in an existing row). Any such change will cause
66    /// `sysknife audit verify` to report `VerifyOutcome::Broken` for the
67    /// modified row, because the signature will no longer verify against the
68    /// stored `chain_hash`.
69    ///
70    /// If a correction is genuinely needed, use one of the two safe strategies
71    /// documented on [`crate::audit_chain::ChainContent`]:
72    /// 1. Insert a corrective row that references the original `transaction_id`.
73    /// 2. Extend the chain protocol with a dedicated amendment record type.
74    pub summary: String,
75    pub warnings: Vec<String>,
76    /// Role the daemon resolved for the connection that asked for this action.
77    ///
78    /// Chain-hashed at INSERT alongside `summary`. Before this field existed the
79    /// signed record could say what was authorised but not which privilege tier
80    /// asked; which *account* asked is `caller_principal` below. Not
81    /// caller-supplied over the wire: the dispatcher passes the role
82    /// it resolved from `SO_PEERCRED` (or the vsock token), never a value from
83    /// the request body.
84    pub caller_role: CallerRole,
85    /// Which account asked, as opposed to what its role permitted.
86    ///
87    /// Chain-hashed at INSERT alongside `caller_role`. The role alone cannot
88    /// separate two members of `sysknife-admin`, so a trail built only from it
89    /// answers "an Admin did this" and stops there. Like the role, this is
90    /// resolved by the dispatcher from the connection, never read from the
91    /// request body. See [`crate::auth::CallerPrincipal`].
92    pub caller_principal: CallerPrincipal,
93}
94
95/// One forward-only SQLite schema step, applied in `version` order and
96/// recorded in `schema_migrations`.
97///
98/// This mirrors the Postgres `MIGRATIONS` list deliberately. The SQLite path
99/// used to be a single `CREATE TABLE IF NOT EXISTS` batch plus a hardcoded
100/// `version > 1` guard, which had no way to express "add a column to an
101/// existing database" — `IF NOT EXISTS` skips a table that is already there,
102/// columns and all. Adding the caller-identity columns needed a real step.
103struct SqliteMigration {
104    version: i64,
105    name: &'static str,
106    sql: &'static str,
107}
108
109/// Schema history for the SQLite backend.
110///
111/// Migration 1 covers the append-tamper-evident hash chain (see
112/// `audit_chain.rs` for the full threat model — note that truncation of the
113/// tail is NOT detected by this chain alone; that requires the signed
114/// checkpoints anchored to an append-only sink (`checkpoint_sink`), with the
115/// journald watermark as a lighter best-effort complement):
116///   seq             — monotonic ordering, 1-indexed
117///   key_id          — identifies the key generation (forward-compatible with
118///                     epoch rotation in a follow-up issue)
119///   chain_hash      — ed25519_sign(ROW_DOMAIN || canonical(immutable_fields)
120///                     || prev_chain_hash, key)
121///   prev_chain_hash — chain_hash of the previous row, "" for the first row
122///
123/// `status` is intentionally absent from the chain content — it is mutable.
124/// The chain protects the *authorisation decision* captured at insert time,
125/// not the live execution state.
126const SQLITE_MIGRATIONS: &[SqliteMigration] = &[
127    SqliteMigration {
128        version: 1,
129        name: "initial_audit_schema",
130        sql: r#"
131            CREATE TABLE IF NOT EXISTS transactions (
132                transaction_id TEXT PRIMARY KEY,
133                request_id TEXT NOT NULL,
134                request_hash TEXT NOT NULL,
135                action_name TEXT NOT NULL,
136                risk_level TEXT NOT NULL,
137                status TEXT NOT NULL,
138                approval_id TEXT,
139                summary TEXT NOT NULL,
140                warnings_json TEXT NOT NULL,
141                created_at TEXT NOT NULL DEFAULT (datetime('now')),
142                seq INTEGER NOT NULL UNIQUE,
143                key_id TEXT NOT NULL,
144                chain_hash TEXT NOT NULL,
145                prev_chain_hash TEXT NOT NULL DEFAULT ''
146            );
147
148            CREATE TABLE IF NOT EXISTS transaction_previews (
149                transaction_id TEXT PRIMARY KEY,
150                preview_json TEXT NOT NULL
151            );
152
153            CREATE TABLE IF NOT EXISTS transaction_approvals (
154                transaction_id TEXT PRIMARY KEY,
155                receipt_digest TEXT NOT NULL,
156                approved_at TEXT NOT NULL DEFAULT (datetime('now')),
157                consumed_at TEXT
158            );
159
160            CREATE INDEX IF NOT EXISTS transactions_seq_idx ON transactions(seq);
161        "#,
162    },
163    // Caller identity in the signed content, and the approval-event chain.
164    //
165    // The three new `transactions` columns are nullable with a legacy default
166    // on purpose: rows written by an earlier binary were signed over an
167    // encoding that has no such fields, and backfilling them with empty
168    // strings would change every historical message and report the whole chain
169    // as Broken. `chain_version` selects the encoding per row — see
170    // `audit_chain::ChainIdentity`.
171    SqliteMigration {
172        version: 2,
173        name: "caller_identity_and_approval_events",
174        sql: r#"
175            ALTER TABLE transactions ADD COLUMN chain_version INTEGER NOT NULL DEFAULT 1;
176            ALTER TABLE transactions ADD COLUMN caller_role TEXT;
177            ALTER TABLE transactions ADD COLUMN event_tip TEXT;
178
179            CREATE TABLE IF NOT EXISTS audit_events (
180                seq INTEGER PRIMARY KEY,
181                key_id TEXT NOT NULL,
182                kind TEXT NOT NULL,
183                transaction_id TEXT NOT NULL,
184                receipt_digest TEXT NOT NULL,
185                created_at TEXT NOT NULL,
186                chain_hash TEXT NOT NULL,
187                prev_chain_hash TEXT NOT NULL DEFAULT ''
188            );
189
190            CREATE INDEX IF NOT EXISTS audit_events_transaction_idx
191                ON audit_events(transaction_id);
192        "#,
193    },
194    // Nullable for the same reason the v2 columns are: rows written before this
195    // step were signed over an encoding with no principal field, so backfilling
196    // any value would rewrite their message and report the chain as Broken.
197    // `chain_version` keeps selecting the encoding per row.
198    SqliteMigration {
199        version: 3,
200        name: "caller_principal",
201        sql: r#"
202            ALTER TABLE transactions ADD COLUMN caller_principal TEXT;
203        "#,
204    },
205];
206
207/// Column list for every `ChainRow` read, kept next to the mapper below.
208///
209/// The two read paths (`fetch_chain_rows`, `fetch_chain_row`) used to repeat
210/// both the SELECT and a positional `row.get(n)` block. Adding a column meant
211/// editing four places in step, and a mismatch between the two would surface
212/// as a verification failure rather than a compile error.
213const CHAIN_ROW_COLUMNS: &str = "seq, key_id, transaction_id, request_id, request_hash, \
214     action_name, risk_level, summary, approval_id, warnings_json, \
215     created_at, prev_chain_hash, chain_hash, chain_version, caller_role, event_tip, \
216     caller_principal";
217
218fn chain_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChainRow> {
219    Ok(ChainRow {
220        seq: row.get::<_, i64>(0)? as u64,
221        key_id: row.get(1)?,
222        transaction_id: row.get(2)?,
223        request_id: row.get(3)?,
224        request_hash: row.get(4)?,
225        action_name: row.get(5)?,
226        risk_level: deserialize_field(&row.get::<_, String>(6)?).map_err(|e| {
227            rusqlite::Error::FromSqlConversionFailure(6, rusqlite::types::Type::Text, Box::new(e))
228        })?,
229        summary: row.get(7)?,
230        approval_id: row.get(8)?,
231        warnings_json: row.get(9)?,
232        created_at: row.get(10)?,
233        prev_chain_hash: row.get(11)?,
234        chain_hash: row.get(12)?,
235        chain_version: row.get::<_, i64>(13)? as u32,
236        caller_role: row.get(14)?,
237        event_tip: row.get(15)?,
238        caller_principal: row.get(16)?,
239    })
240}
241
242fn event_row_from_sqlite(row: &rusqlite::Row<'_>) -> rusqlite::Result<EventRow> {
243    Ok(EventRow {
244        seq: row.get::<_, i64>(0)? as u64,
245        key_id: row.get(1)?,
246        kind: row.get(2)?,
247        transaction_id: row.get(3)?,
248        receipt_digest: row.get(4)?,
249        created_at: row.get(5)?,
250        prev_chain_hash: row.get(6)?,
251        chain_hash: row.get(7)?,
252    })
253}
254
255/// Run `query_row`, treating "no row matched" as `Ok(None)` instead of an
256/// error. Three call sites used to spell out
257/// `.map(Some).or_else(|e| match e { QueryReturnedNoRows => Ok(None), other => Err(other) })`
258/// by hand; sharing this mapping means a future call site cannot drop the
259/// `QueryReturnedNoRows` arm and turn a routine "not found" into a hard error.
260fn query_optional<T>(
261    conn: &Connection,
262    sql: &str,
263    params: impl rusqlite::Params,
264    f: impl FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
265) -> rusqlite::Result<Option<T>> {
266    match conn.query_row(sql, params, f) {
267        Ok(v) => Ok(Some(v)),
268        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
269        Err(e) => Err(e),
270    }
271}
272
273#[derive(Clone, Debug)]
274pub struct TransactionStore {
275    path: PathBuf,
276    /// Ed25519 signing key used to compute the forward audit chain on insert.
277    /// `None` only for read-only callers that never write rows.
278    audit_key: Option<Arc<AuditKey>>,
279}
280
281#[derive(Clone, Debug, PartialEq)]
282pub struct RecordedPreviewedTransaction {
283    pub transaction: TransactionRecord,
284    pub preview: PreviewEnvelope,
285}
286
287struct InsertedTransaction {
288    record: TransactionRecord,
289    seq: u64,
290    chain_hash: String,
291}
292
293#[derive(Debug, thiserror::Error)]
294pub enum TransactionStoreError {
295    #[error("io error: {0}")]
296    Io(#[from] std::io::Error),
297
298    #[error("sqlite error: {0}")]
299    Sqlite(#[from] rusqlite::Error),
300
301    #[error("json error: {0}")]
302    Json(#[from] serde_json::Error),
303
304    #[error("database invariant violation: {0}")]
305    DatabaseInvariant(String),
306
307    #[error("transaction not found: {0}")]
308    NotFound(String),
309
310    #[error("invalid transition from {from:?} to {to:?}")]
311    InvalidTransition { from: JobState, to: JobState },
312
313    #[error("audit chain misconfiguration: {0}")]
314    AuditChainMissing(&'static str),
315
316    /// `update_status`'s compare-and-swap `UPDATE ... WHERE status = <observed>`
317    /// matched zero rows: another writer changed `status` between our read and
318    /// our write. This is a lost race, not a validation failure — the caller
319    /// (see `dispatcher::update_terminal_status`) already retries on any error
320    /// and re-reads the current status, so surfacing this distinctly (rather
321    /// than silently overwriting a transition we never validated) is safe to
322    /// retry.
323    #[error("transaction {0} status changed concurrently; retry")]
324    ConcurrentStatusChange(String),
325}
326
327impl TransactionStore {
328    /// Open the store with **no audit chain key**. Inserts will fail with
329    /// `AuditChainMissing` — only suitable for read-only callers (e.g.
330    /// `sysknife audit verify` which loads the key separately).
331    pub fn open_read_only(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
332        let path = path.as_ref().to_path_buf();
333        if let Some(parent) = path.parent() {
334            ensure_private_dir(parent)?;
335        }
336
337        let store = Self {
338            path,
339            audit_key: None,
340        };
341        store.initialize()?;
342        Ok(store)
343    }
344
345    /// Open the store and bind it to an audit chain key. Every insert
346    /// computes a forward Ed25519-signed chain hash linked to the previous row.
347    ///
348    /// The key path defaults to `<db_dir>/audit-key` so dev/test runs with
349    /// per-tempdir databases are fully isolated. Production deployments
350    /// override with `SYSKNIFE_AUDIT_KEY_PATH=/etc/sysknife/audit-key`.
351    pub fn open(path: impl AsRef<Path>) -> Result<Self, TransactionStoreError> {
352        let db_path = path.as_ref();
353        let key_path = std::env::var("SYSKNIFE_AUDIT_KEY_PATH")
354            .map(PathBuf::from)
355            .unwrap_or_else(|_| {
356                db_path
357                    .parent()
358                    .unwrap_or_else(|| Path::new("."))
359                    .join("audit-key")
360            });
361        let key = AuditKey::load_or_generate(&key_path).map_err(|e| {
362            TransactionStoreError::Io(std::io::Error::new(
363                std::io::ErrorKind::InvalidData,
364                format!("audit key load failed: {e}"),
365            ))
366        })?;
367        Self::open_with_key(path, Arc::new(key))
368    }
369
370    /// Open the store with an explicit audit key. Used by tests and by
371    /// production code paths that want to inject a key from a specific path.
372    pub fn open_with_key(
373        path: impl AsRef<Path>,
374        audit_key: Arc<AuditKey>,
375    ) -> Result<Self, TransactionStoreError> {
376        let path = path.as_ref().to_path_buf();
377        if let Some(parent) = path.parent() {
378            ensure_private_dir(parent)?;
379        }
380
381        let store = Self {
382            path,
383            audit_key: Some(audit_key),
384        };
385        store.initialize()?;
386        restrict_db_file_mode(&store.path);
387        Ok(store)
388    }
389
390    /// Shared body of [`record`](Self::record) and
391    /// [`record_previewed`](Self::record_previewed): check for a writable
392    /// store, open the write transaction, insert the transaction row, run
393    /// `in_tx` inside that same transaction (so e.g. a preview insert commits
394    /// atomically with its transaction row), commit, and emit the chain-tip
395    /// watermark. `in_tx` runs between the insert and the commit — never
396    /// after — so nothing it does can land outside the transaction it needs
397    /// to be atomic with.
398    fn record_with(
399        &self,
400        transaction: NewTransaction,
401        in_tx: impl FnOnce(&Connection, &InsertedTransaction) -> Result<(), TransactionStoreError>,
402    ) -> Result<InsertedTransaction, TransactionStoreError> {
403        let key = self
404            .audit_key
405            .as_ref()
406            .ok_or(TransactionStoreError::AuditChainMissing(
407                "this TransactionStore was opened read-only; cannot record",
408            ))?;
409        let mut conn = self.connection()?;
410        // IMMEDIATE acquires the write lock at BEGIN, so the read of
411        // `next_seq_and_prev_hash` is consistent with the eventual INSERT.
412        // Default DEFERRED would let two writers both read the same prev_hash
413        // and then race to INSERT — the loser hits SQLITE_BUSY.
414        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
415        let transaction_id = Uuid::new_v4().to_string();
416        let inserted = Self::insert_transaction(&tx, key, &transaction_id, transaction)?;
417        in_tx(&tx, &inserted)?;
418        tx.commit()?;
419        emit_chain_tip_watermark(inserted.seq, &inserted.chain_hash);
420        Ok(inserted)
421    }
422
423    pub fn record(
424        &self,
425        transaction: NewTransaction,
426    ) -> Result<TransactionRecord, TransactionStoreError> {
427        let inserted = self.record_with(transaction, |_, _| Ok(()))?;
428        Ok(inserted.record)
429    }
430
431    pub fn record_previewed(
432        &self,
433        transaction: NewTransaction,
434        preview: PreviewEnvelope,
435    ) -> Result<RecordedPreviewedTransaction, TransactionStoreError> {
436        let inserted = self.record_with(transaction, |conn, inserted| {
437            Self::insert_preview(conn, &inserted.record.transaction_id, &preview)
438        })?;
439        Ok(RecordedPreviewedTransaction {
440            transaction: inserted.record,
441            preview,
442        })
443    }
444
445    pub fn get(
446        &self,
447        transaction_id: &str,
448    ) -> Result<Option<TransactionRecord>, TransactionStoreError> {
449        let conn = self.connection()?;
450        let mut stmt = conn.prepare(
451            "SELECT
452                transaction_id,
453                request_id,
454                request_hash,
455                action_name,
456                risk_level,
457                status,
458                approval_id,
459                summary,
460                warnings_json
461             FROM transactions
462             WHERE transaction_id = ?1",
463        )?;
464        let mut rows = stmt.query(params![transaction_id])?;
465        if let Some(row) = rows.next()? {
466            Ok(Some(row_to_record(row)?))
467        } else {
468            Ok(None)
469        }
470    }
471
472    pub fn get_preview(
473        &self,
474        transaction_id: &str,
475    ) -> Result<Option<PreviewEnvelope>, TransactionStoreError> {
476        let conn = self.connection()?;
477        let mut stmt = conn.prepare(
478            "SELECT preview_json
479             FROM transaction_previews
480             WHERE transaction_id = ?1",
481        )?;
482        let mut rows = stmt.query(params![transaction_id])?;
483        if let Some(row) = rows.next()? {
484            let preview_json: String = row.get(0)?;
485            Ok(Some(serde_json::from_str(&preview_json)?))
486        } else {
487            Ok(None)
488        }
489    }
490
491    pub fn update_status(
492        &self,
493        transaction_id: &str,
494        new_status: JobState,
495    ) -> Result<(), TransactionStoreError> {
496        let conn = self.connection()?;
497
498        // Read the current status so we can validate the transition.
499        let current_status: String = conn
500            .query_row(
501                "SELECT status FROM transactions WHERE transaction_id = ?1",
502                params![transaction_id],
503                |row| row.get(0),
504            )
505            .map_err(|e| match e {
506                rusqlite::Error::QueryReturnedNoRows => {
507                    TransactionStoreError::NotFound(transaction_id.to_string())
508                }
509                other => TransactionStoreError::Sqlite(other),
510            })?;
511
512        let current: JobState = deserialize_field(&current_status)?;
513        if !crate::jobs::allowed_transition(&current, &new_status) {
514            return Err(TransactionStoreError::InvalidTransition {
515                from: current,
516                to: new_status,
517            });
518        }
519
520        // Compare-and-swap: only write if `status` still equals the value we
521        // just validated the transition against. Without the `AND status =
522        // ?3` guard this was a check-then-act race — a concurrent writer
523        // (e.g. two calls racing to move the same job to two different
524        // terminal states) could flip `status` between our SELECT and this
525        // UPDATE, and the loser would silently report success for a
526        // transition it never actually validated against the row's real
527        // prior value. `rows_affected == 0` means we lost that race (rows are
528        // never deleted, so it cannot mean the row vanished); the caller
529        // (`dispatcher::update_terminal_status`) already retries on any error
530        // and re-checks the final status, so failing closed here is safe.
531        let rows_affected = conn.execute(
532            "UPDATE transactions SET status = ?1 WHERE transaction_id = ?2 AND status = ?3",
533            params![
534                serialize_field(&new_status)?,
535                transaction_id,
536                current_status
537            ],
538        )?;
539        if rows_affected == 0 {
540            return Err(TransactionStoreError::ConcurrentStatusChange(
541                transaction_id.to_string(),
542            ));
543        }
544        Ok(())
545    }
546
547    /// Attach one immutable approval receipt digest to a fresh queued preview.
548    pub fn approve_transaction(
549        &self,
550        transaction_id: &str,
551    ) -> Result<Option<String>, TransactionStoreError> {
552        let key = self
553            .audit_key
554            .as_ref()
555            .ok_or(TransactionStoreError::AuditChainMissing(
556                "this TransactionStore was opened read-only; cannot approve",
557            ))?;
558        let Some(record) = self.get(transaction_id)? else {
559            return Ok(None);
560        };
561        let receipt = key.approval_receipt(transaction_id, &record.request_hash);
562        let receipt_digest = audit_chain::approval_receipt_digest(&receipt);
563        let Some(committed_digest) = record.approval_id.as_deref() else {
564            return Err(TransactionStoreError::DatabaseInvariant(format!(
565                "transaction {transaction_id} has no signed approval commitment"
566            )));
567        };
568        if !bool::from(receipt_digest.as_bytes().ct_eq(committed_digest.as_bytes())) {
569            return Err(TransactionStoreError::DatabaseInvariant(format!(
570                "transaction {transaction_id} approval commitment does not match its signed preview"
571            )));
572        }
573
574        // The approval row and its chained event commit together: an approval
575        // that is not in the event chain is exactly the gap this chain closes.
576        let mut conn = self.connection()?;
577        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
578        let queued_json = serialize_field(&JobState::Queued)?;
579        let rows_affected = tx.execute(
580            &format!(
581                "INSERT INTO transaction_approvals (transaction_id, receipt_digest) \
582                 SELECT transaction_id, ?1 FROM transactions \
583                 WHERE transaction_id = ?2 \
584                   AND status = ?3 \
585                   AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
586                   AND NOT EXISTS ( \
587                       SELECT 1 FROM transaction_approvals WHERE transaction_id = ?2 \
588                   )"
589            ),
590            params![receipt_digest, transaction_id, queued_json],
591        )?;
592        if rows_affected > 0 {
593            Self::append_event(
594                &tx,
595                key,
596                AuditEventKind::ApprovalGranted,
597                transaction_id,
598                &receipt_digest,
599            )?;
600        }
601        tx.commit()?;
602        Ok((rows_affected > 0).then_some(receipt))
603    }
604
605    /// Remove an approval that was persisted but could not be delivered to the
606    /// caller. Consumed receipts are never revocable.
607    pub fn revoke_unconsumed_approval(
608        &self,
609        transaction_id: &str,
610    ) -> Result<bool, TransactionStoreError> {
611        let key = self
612            .audit_key
613            .as_ref()
614            .ok_or(TransactionStoreError::AuditChainMissing(
615                "this TransactionStore was opened read-only; cannot revoke",
616            ))?;
617        let mut conn = self.connection()?;
618        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
619        let revoked = Self::revoke_unconsumed_approval_in_tx(&tx, key, transaction_id)?;
620        tx.commit()?;
621        Ok(revoked)
622    }
623
624    fn revoke_unconsumed_approval_in_tx(
625        conn: &Connection,
626        key: &AuditKey,
627        transaction_id: &str,
628    ) -> Result<bool, TransactionStoreError> {
629        // Capture the digest before the DELETE: the event has to name which
630        // receipt was retracted, and after the delete there is nothing to name.
631        let digest: Option<String> = query_optional(
632            conn,
633            "SELECT receipt_digest FROM transaction_approvals \
634             WHERE transaction_id = ?1 AND consumed_at IS NULL",
635            params![transaction_id],
636            |row| row.get(0),
637        )?;
638        let rows_affected = conn.execute(
639            "DELETE FROM transaction_approvals \
640             WHERE transaction_id = ?1 AND consumed_at IS NULL",
641            params![transaction_id],
642        )?;
643        if rows_affected > 0 {
644            let digest = digest.ok_or_else(|| {
645                TransactionStoreError::DatabaseInvariant(format!(
646                    "revoked an approval for {transaction_id} that had no receipt digest"
647                ))
648            })?;
649            Self::append_event(
650                conn,
651                key,
652                AuditEventKind::ApprovalRevoked,
653                transaction_id,
654                &digest,
655            )?;
656        }
657        Ok(rows_affected > 0)
658    }
659
660    /// Atomically consume an approved receipt and transition Queued to Running.
661    pub fn claim_approved_for_execution(
662        &self,
663        transaction_id: &str,
664        receipt_digest: &str,
665    ) -> Result<bool, TransactionStoreError> {
666        let key = self
667            .audit_key
668            .as_ref()
669            .ok_or(TransactionStoreError::AuditChainMissing(
670                "this TransactionStore was opened read-only; cannot claim",
671            ))?;
672        let mut conn = self.connection()?;
673        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
674        let queued_json = serialize_field(&JobState::Queued)?;
675        let running_json = serialize_field(&JobState::Running)?;
676        let rows_affected = tx.execute(
677            &format!(
678                "UPDATE transactions SET status = ?1 \
679                 WHERE transaction_id = ?2 \
680                   AND status = ?3 \
681                   AND julianday(created_at) > julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes') \
682                   AND EXISTS ( \
683                       SELECT 1 FROM transaction_approvals \
684                       WHERE transaction_id = ?2 \
685                         AND receipt_digest = ?4 \
686                         AND consumed_at IS NULL \
687                   )"
688            ),
689            params![running_json, transaction_id, queued_json, receipt_digest],
690        )?;
691        if rows_affected > 0 {
692            tx.execute(
693                "UPDATE transaction_approvals \
694                 SET consumed_at = datetime('now') \
695                 WHERE transaction_id = ?1 AND consumed_at IS NULL",
696                params![transaction_id],
697            )?;
698            Self::append_event(
699                &tx,
700                key,
701                AuditEventKind::ApprovalConsumed,
702                transaction_id,
703                receipt_digest,
704            )?;
705        }
706        tx.commit()?;
707        Ok(rows_affected > 0)
708    }
709
710    /// Cancel all `Queued` transactions whose `created_at` timestamp is older
711    /// than the 15-minute TTL window.  Returns the number of rows affected.
712    ///
713    /// **State-machine safety:** the WHERE clause pins `status = Queued`
714    /// before applying `Queued → Canceled`, which is the only legal
715    /// transition reachable from `Queued` other than `Running`. A row that
716    /// has progressed to `Running` (or any terminal state) in between the
717    /// TTL match and our UPDATE is skipped because the predicate no longer
718    /// matches it. This makes the bulk SQL semantically equivalent to
719    /// fetching each candidate, building a `JobStateMachine`, and calling
720    /// `cancel()` on it — but in a single statement so we don't race ourselves
721    /// when many rows expire at once. The
722    /// `cleanup_stale_queued_does_not_clobber_running_rows` regression test
723    /// in `tests/coverage_gaps.rs` pins this guarantee.
724    pub fn cleanup_stale_queued(&self) -> Result<u64, TransactionStoreError> {
725        let key = self
726            .audit_key
727            .as_ref()
728            .ok_or(TransactionStoreError::AuditChainMissing(
729                "this TransactionStore was opened read-only; cannot clean up",
730            ))?;
731        let mut conn = self.connection()?;
732        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
733        let canceled_json = serialize_field(&JobState::Canceled)?;
734        let queued_json = serialize_field(&JobState::Queued)?;
735        let stale_ids: Vec<String> = {
736            let mut statement = tx.prepare(&format!(
737                "SELECT transaction_id FROM transactions \
738                 WHERE status = ?1 \
739                   AND julianday(created_at) <= julianday('now', '-{APPROVAL_RECEIPT_TTL_MINUTES} minutes')"
740            ))?;
741            let stale_ids = statement
742                .query_map(params![queued_json], |row| row.get(0))?
743                .collect::<Result<_, _>>()?;
744            stale_ids
745        };
746        let mut canceled = 0;
747        for transaction_id in stale_ids {
748            let rows_affected = tx.execute(
749                "UPDATE transactions SET status = ?1 \
750                 WHERE transaction_id = ?2 AND status = ?3",
751                params![canceled_json, transaction_id, queued_json],
752            )?;
753            if rows_affected > 0 {
754                Self::revoke_unconsumed_approval_in_tx(&tx, key, &transaction_id)?;
755                canceled += rows_affected;
756            }
757        }
758        tx.commit()?;
759        Ok(canceled as u64)
760    }
761
762    /// Cancel one still-`Queued` transaction (`Queued → Canceled`). Returns
763    /// `true` iff a queued row was transitioned.
764    ///
765    /// Option A semantics: the `status = ?3` (Queued) guard means a `Running`
766    /// transaction — one whose privileged action is already executing — is
767    /// never cancelled, so we never leave a half-applied root mutation behind
768    /// a `Canceled` record. Missing or already-terminal transactions return
769    /// `false`.
770    pub fn cancel_queued(&self, transaction_id: &str) -> Result<bool, TransactionStoreError> {
771        let key = self
772            .audit_key
773            .as_ref()
774            .ok_or(TransactionStoreError::AuditChainMissing(
775                "this TransactionStore was opened read-only; cannot cancel",
776            ))?;
777        let mut conn = self.connection()?;
778        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
779        let canceled_json = serialize_field(&JobState::Canceled)?;
780        let queued_json = serialize_field(&JobState::Queued)?;
781        let rows_affected = tx.execute(
782            "UPDATE transactions SET status = ?1 \
783             WHERE transaction_id = ?2 AND status = ?3",
784            params![canceled_json, transaction_id, queued_json],
785        )?;
786        if rows_affected > 0 {
787            Self::revoke_unconsumed_approval_in_tx(&tx, key, transaction_id)?;
788        }
789        tx.commit()?;
790        Ok(rows_affected > 0)
791    }
792
793    /// List transactions with optional filters, ordered by newest first.
794    ///
795    /// - `limit`: max number of rows (capped at 100)
796    /// - `status_filter`: if set, only return rows matching this status
797    ///   (must be a valid `JobState` variant: `"succeeded"`, `"failed"`,
798    ///   `"queued"`, `"running"`, `"canceled"`, `"rolled_back"`, `"needs_reboot"`)
799    /// - `action_filter`: if set, only return rows with this exact action name
800    /// - `since_hours`: if set, only return rows created within the last N hours
801    pub fn list_transactions(
802        &self,
803        limit: u32,
804        status_filter: Option<&str>,
805        action_filter: Option<&str>,
806        since_hours: Option<u32>,
807    ) -> Result<Vec<TransactionRecord>, TransactionStoreError> {
808        let conn = self.connection()?;
809        let (filter_sql, param_values) =
810            Self::build_history_filter(limit, status_filter, action_filter, since_hours)?;
811        let sql = format!(
812            "SELECT transaction_id, request_id, request_hash, action_name, \
813             risk_level, status, approval_id, summary, warnings_json \
814             FROM transactions WHERE 1=1{filter_sql}"
815        );
816        let params_ref: Vec<&dyn rusqlite::types::ToSql> =
817            param_values.iter().map(|b| b.as_ref()).collect();
818
819        let mut stmt = conn.prepare(&sql)?;
820        let rows = stmt.query_map(params_ref.as_slice(), |row| Ok(row_to_record(row)))?;
821
822        let mut results = Vec::new();
823        for row in rows {
824            results.push(row??);
825        }
826        Ok(results)
827    }
828
829    /// Structured history for programmatic clients (the MCP `sysknife_history`
830    /// tool). Unlike [`list_transactions`](Self::list_transactions) it selects
831    /// `created_at` and returns [`JobHistoryEntry`] so `risk_level` and
832    /// `created_at` reach the caller typed, without text re-parsing.
833    pub fn list_history(
834        &self,
835        limit: u32,
836        status_filter: Option<&str>,
837        action_filter: Option<&str>,
838        since_hours: Option<u32>,
839    ) -> Result<Vec<JobHistoryEntry>, TransactionStoreError> {
840        let conn = self.connection()?;
841        let (filter_sql, param_values) =
842            Self::build_history_filter(limit, status_filter, action_filter, since_hours)?;
843        let sql = format!(
844            "SELECT transaction_id, action_name, risk_level, status, summary, created_at \
845             FROM transactions WHERE 1=1{filter_sql}"
846        );
847        let params_ref: Vec<&dyn rusqlite::types::ToSql> =
848            param_values.iter().map(|b| b.as_ref()).collect();
849
850        let mut stmt = conn.prepare(&sql)?;
851        let rows = stmt.query_map(params_ref.as_slice(), |row| {
852            Ok((|| {
853                Ok::<JobHistoryEntry, TransactionStoreError>(JobHistoryEntry {
854                    transaction_id: row.get(0)?,
855                    action_name: row.get(1)?,
856                    risk_level: deserialize_field(&row.get::<_, String>(2)?)?,
857                    status: deserialize_field(&row.get::<_, String>(3)?)?,
858                    summary: row.get(4)?,
859                    created_at: row.get(5)?,
860                })
861            })())
862        })?;
863
864        let mut results = Vec::new();
865        for row in rows {
866            results.push(row??);
867        }
868        Ok(results)
869    }
870
871    /// Build the shared `WHERE`/`ORDER BY`/`LIMIT` suffix (after `WHERE 1=1`)
872    /// and its bound parameters for the history queries, so
873    /// [`list_transactions`](Self::list_transactions) and
874    /// [`list_history`](Self::list_history) cannot filter differently.
875    fn build_history_filter(
876        limit: u32,
877        status_filter: Option<&str>,
878        action_filter: Option<&str>,
879        since_hours: Option<u32>,
880    ) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), TransactionStoreError> {
881        let mut sql = String::new();
882        let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
883
884        if let Some(status) = status_filter {
885            // Validate against known JobState variants to avoid silent empty
886            // results from typos (e.g. "success" instead of "succeeded").
887            // deserialize_field returns serde_json::Error → TransactionStoreError::Json.
888            let validated: JobState = deserialize_field(&format!("\"{status}\""))?;
889            let status_json = serialize_field(&validated)?;
890            sql.push_str(" AND status = ?");
891            param_values.push(Box::new(status_json));
892        }
893
894        if let Some(action) = action_filter {
895            sql.push_str(" AND action_name = ?");
896            param_values.push(Box::new(action.to_string()));
897        }
898
899        if let Some(hours) = since_hours {
900            sql.push_str(" AND julianday(created_at) > julianday('now', '-' || ? || ' hours')");
901            param_values.push(Box::new(hours));
902        }
903
904        sql.push_str(" ORDER BY seq DESC LIMIT ?");
905        param_values.push(Box::new(limit.min(100)));
906
907        Ok((sql, param_values))
908    }
909
910    fn connection(&self) -> Result<Connection, TransactionStoreError> {
911        let conn = Connection::open(&self.path)?;
912        // Concurrency tuning :
913        //   - WAL journal mode lets readers proceed concurrently with writers.
914        //   - busy_timeout=5000ms makes a contending writer block instead of
915        //     immediately returning SQLITE_BUSY. Without it, two concurrent
916        //     `record()` calls (one of the two daemon use cases the dispatcher
917        //     supports) had a 100% second-writer failure rate.
918        //   - synchronous=NORMAL is the WAL-recommended setting; FULL is
919        //     overkill for an audit log that's already append-only by design,
920        //     and OFF risks losing the latest transactions on a crash.
921        //   - foreign_keys=ON for parity with future schema changes.
922        conn.pragma_update(None, "journal_mode", "WAL")?;
923        conn.pragma_update(None, "synchronous", "NORMAL")?;
924        conn.pragma_update(None, "foreign_keys", "ON")?;
925        conn.busy_timeout(std::time::Duration::from_secs(5))?;
926        Ok(conn)
927    }
928
929    fn initialize(&self) -> Result<(), TransactionStoreError> {
930        let mut conn = self.connection()?;
931        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
932        tx.execute_batch(
933            "CREATE TABLE IF NOT EXISTS schema_migrations (\
934                 version INTEGER PRIMARY KEY,\
935                 name TEXT NOT NULL,\
936                 applied_at TEXT NOT NULL DEFAULT (datetime('now'))\
937             );",
938        )?;
939        let current: i64 = tx.query_row(
940            "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
941            [],
942            |row| row.get(0),
943        )?;
944        let latest = SQLITE_MIGRATIONS.last().map_or(0, |m| m.version);
945        if current > latest {
946            return Err(TransactionStoreError::DatabaseInvariant(format!(
947                "sqlite schema version {current} is newer than this binary supports ({latest})"
948            )));
949        }
950        for migration in SQLITE_MIGRATIONS.iter().filter(|m| m.version > current) {
951            tx.execute_batch(migration.sql)?;
952            tx.execute(
953                "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
954                params![migration.version, migration.name],
955            )?;
956        }
957        tx.commit()?;
958        Ok(())
959    }
960
961    /// Return all rows in seq order with the chain fields needed for verify.
962    pub fn fetch_chain_rows(&self) -> Result<Vec<ChainRow>, TransactionStoreError> {
963        let conn = self.connection()?;
964        let mut stmt = conn.prepare(&format!(
965            "SELECT {CHAIN_ROW_COLUMNS} FROM transactions ORDER BY seq ASC"
966        ))?;
967        let rows = stmt.query_map([], chain_row_from_sqlite)?;
968        let mut out = Vec::new();
969        for row in rows {
970            out.push(row?);
971        }
972        Ok(out)
973    }
974
975    /// Return every approval event in seq order.
976    pub fn fetch_event_rows(&self) -> Result<Vec<EventRow>, TransactionStoreError> {
977        let conn = self.connection()?;
978        let mut stmt = conn.prepare(
979            "SELECT seq, key_id, kind, transaction_id, receipt_digest, \
980                    created_at, prev_chain_hash, chain_hash \
981             FROM audit_events ORDER BY seq ASC",
982        )?;
983        let rows = stmt.query_map([], event_row_from_sqlite)?;
984        let mut out = Vec::new();
985        for row in rows {
986            out.push(row?);
987        }
988        Ok(out)
989    }
990
991    /// Walk the audit chain with `key` and report integrity status.
992    pub fn verify_audit_chain(
993        &self,
994        key: &AuditKey,
995    ) -> Result<VerifyOutcome, TransactionStoreError> {
996        let rows = self.fetch_chain_rows()?;
997        Ok(audit_chain::verify_chain(key, &rows))
998    }
999
1000    /// Walk the approval-event chain with `key`.
1001    pub fn verify_event_chain(
1002        &self,
1003        key: &AuditKey,
1004    ) -> Result<VerifyOutcome, TransactionStoreError> {
1005        let rows = self.fetch_event_rows()?;
1006        Ok(audit_chain::verify_event_chain(key, &rows))
1007    }
1008
1009    /// Check that every event tip committed by a transaction row is still
1010    /// present in the event chain. See [`audit_chain::verify_event_binding`].
1011    pub fn verify_event_binding(
1012        &self,
1013    ) -> Result<audit_chain::BindingOutcome, TransactionStoreError> {
1014        let tx_rows = self.fetch_chain_rows()?;
1015        let event_rows = self.fetch_event_rows()?;
1016        Ok(audit_chain::verify_event_binding(&tx_rows, &event_rows))
1017    }
1018
1019    /// Verify the chain with only the hex-encoded Ed25519 **public** key. The
1020    /// auditor path: proves the chain without the private key and cannot forge.
1021    pub fn verify_audit_chain_with_pubkey(
1022        &self,
1023        verifying_key_hex: &str,
1024    ) -> Result<VerifyOutcome, TransactionStoreError> {
1025        let rows = self.fetch_chain_rows()?;
1026        Ok(audit_chain::verify_chain_with_pubkey(
1027            verifying_key_hex,
1028            &rows,
1029        ))
1030    }
1031
1032    /// Fetch a single row's chain metadata by `transaction_id`. Used by the
1033    /// audit-log forwarder to construct an `AuditEvent` after insert.
1034    pub fn fetch_chain_row(
1035        &self,
1036        transaction_id: &str,
1037    ) -> Result<Option<ChainRow>, TransactionStoreError> {
1038        let conn = self.connection()?;
1039        let mut stmt = conn.prepare(&format!(
1040            "SELECT {CHAIN_ROW_COLUMNS} FROM transactions WHERE transaction_id = ?1"
1041        ))?;
1042        let mut rows = stmt.query(params![transaction_id])?;
1043        if let Some(row) = rows.next()? {
1044            Ok(Some(chain_row_from_sqlite(row)?))
1045        } else {
1046            Ok(None)
1047        }
1048    }
1049
1050    /// Allocate the next monotonic `seq` and fetch the previous row's
1051    /// `chain_hash`. Caller must hold a transaction so the (seq, prev_hash)
1052    /// pair is consistent against concurrent writers.
1053    fn next_seq_and_prev_hash(conn: &Connection) -> Result<(u64, String), TransactionStoreError> {
1054        let prev: Option<(i64, String)> = query_optional(
1055            conn,
1056            "SELECT seq, chain_hash FROM transactions ORDER BY seq DESC LIMIT 1",
1057            [],
1058            |row| Ok((row.get(0)?, row.get(1)?)),
1059        )?;
1060        Ok(match prev {
1061            Some((seq, hash)) => ((seq as u64) + 1, hash),
1062            None => (1, String::new()),
1063        })
1064    }
1065
1066    /// `chain_hash` of the last approval event, or `None` when no event has
1067    /// ever been recorded.
1068    fn event_chain_tip(conn: &Connection) -> Result<Option<String>, TransactionStoreError> {
1069        Ok(query_optional(
1070            conn,
1071            "SELECT chain_hash FROM audit_events ORDER BY seq DESC LIMIT 1",
1072            [],
1073            |row| row.get(0),
1074        )?)
1075    }
1076
1077    /// Append one signed approval event. Must be called inside a DB
1078    /// transaction that also performs the state change being recorded: an
1079    /// event committed without its state change (or the reverse) would be a
1080    /// trail that disagrees with reality.
1081    fn append_event(
1082        conn: &Connection,
1083        key: &AuditKey,
1084        kind: AuditEventKind,
1085        transaction_id: &str,
1086        receipt_digest: &str,
1087    ) -> Result<(), TransactionStoreError> {
1088        let prev_chain_hash = Self::event_chain_tip(conn)?.unwrap_or_default();
1089        let seq: i64 = conn.query_row(
1090            "SELECT COALESCE(MAX(seq), 0) + 1 FROM audit_events",
1091            [],
1092            |row| row.get(0),
1093        )?;
1094        let created_at: String =
1095            conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1096                row.get(0)
1097            })?;
1098        let key_id = CURRENT_KEY_ID.to_string();
1099        let chain_hash = key.event_hash(
1100            &EventContent {
1101                seq: seq as u64,
1102                key_id: &key_id,
1103                kind,
1104                transaction_id,
1105                receipt_digest,
1106                created_at: &created_at,
1107            },
1108            &prev_chain_hash,
1109        );
1110        conn.execute(
1111            "INSERT INTO audit_events (
1112                seq, key_id, kind, transaction_id, receipt_digest,
1113                created_at, chain_hash, prev_chain_hash
1114            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
1115            params![
1116                seq,
1117                key_id,
1118                kind.as_str(),
1119                transaction_id,
1120                receipt_digest,
1121                created_at,
1122                chain_hash,
1123                prev_chain_hash,
1124            ],
1125        )?;
1126        Ok(())
1127    }
1128
1129    fn insert_transaction(
1130        conn: &Connection,
1131        key: &AuditKey,
1132        transaction_id: &str,
1133        transaction: NewTransaction,
1134    ) -> Result<InsertedTransaction, TransactionStoreError> {
1135        let request_id = transaction.request_id;
1136        let request_hash = transaction.request_hash;
1137        let action_name = transaction.action_name;
1138        let risk_level = transaction.risk_level;
1139        // Initial status is always Queued — not caller-controllable.
1140        let status = JobState::Queued;
1141        // Always the store's own commitment over this transaction. It used to
1142        // accept a caller-supplied `Some(..)` and use it verbatim; production
1143        // always passed `None`, so the invariant held only because every call
1144        // site remembered to. The value is chain-hashed and then treated as
1145        // authoritative evidence that an approval happened, which is not
1146        // something a caller may hand us.
1147        let approval_id = Some(key.approval_commitment(transaction_id, &request_hash));
1148        let summary = transaction.summary;
1149        let warnings = transaction.warnings;
1150        let warnings_json = serde_json::to_string(&warnings)?;
1151
1152        // Allocate the next seq + previous chain hash inside the same DB
1153        // transaction so concurrent writers can't race.
1154        let (seq, prev_chain_hash) = Self::next_seq_and_prev_hash(conn)?;
1155
1156        // SQLite's `datetime('now')` (default for the column) is computed at
1157        // INSERT time, but we need the same value to compute the chain hash
1158        // before the row exists. Compute it ourselves and pin it.
1159        let created_at: String =
1160            conn.query_row("SELECT strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", [], |row| {
1161                row.get(0)
1162            })?;
1163
1164        let key_id = CURRENT_KEY_ID.to_string();
1165        // Bind this row to the approval-event chain as it stands right now.
1166        // Read inside the caller's DB transaction so a concurrent event append
1167        // cannot land between the read and the insert.
1168        let event_tip = Self::event_chain_tip(conn)?.unwrap_or_default();
1169        let caller_role = transaction.caller_role.as_str();
1170        let caller_principal = transaction.caller_principal.as_signed_str();
1171        // Build the identity once and derive the stored version from it, so the
1172        // chain_version column is provably the version whose message was signed.
1173        // A constant here let the two drift the moment a new encoding landed.
1174        let identity = ChainIdentity::V3 {
1175            caller_role,
1176            event_tip: &event_tip,
1177            caller_principal: &caller_principal,
1178        };
1179        let chain_hash = key.chain_hash(
1180            &ChainContent {
1181                seq,
1182                key_id: &key_id,
1183                transaction_id,
1184                request_id: &request_id,
1185                request_hash: &request_hash,
1186                action_name: &action_name,
1187                risk_level,
1188                summary: &summary,
1189                approval_id: approval_id.as_deref(),
1190                warnings_json: &warnings_json,
1191                created_at: &created_at,
1192                identity,
1193            },
1194            &prev_chain_hash,
1195        );
1196
1197        let record = TransactionRecord {
1198            transaction_id: transaction_id.to_string(),
1199            request_id: request_id.clone(),
1200            request_hash: request_hash.clone(),
1201            action_name: action_name.clone(),
1202            risk_level,
1203            status,
1204            approval_id: approval_id.clone(),
1205            summary: summary.clone(),
1206            warnings: warnings.clone(),
1207        };
1208
1209        conn.execute(
1210            "INSERT INTO transactions (
1211                transaction_id,
1212                request_id,
1213                request_hash,
1214                action_name,
1215                risk_level,
1216                status,
1217                approval_id,
1218                summary,
1219                warnings_json,
1220                created_at,
1221                seq,
1222                key_id,
1223                chain_hash,
1224                prev_chain_hash,
1225                chain_version,
1226                caller_role,
1227                event_tip,
1228                caller_principal
1229            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)",
1230            params![
1231                transaction_id,
1232                request_id,
1233                request_hash,
1234                action_name,
1235                serialize_field(&risk_level)?,
1236                serialize_field(&status)?,
1237                approval_id,
1238                summary,
1239                warnings_json,
1240                created_at,
1241                seq as i64,
1242                key_id,
1243                chain_hash,
1244                prev_chain_hash,
1245                identity.version() as i64,
1246                caller_role,
1247                event_tip,
1248                caller_principal,
1249            ],
1250        )?;
1251
1252        Ok(InsertedTransaction {
1253            record,
1254            seq,
1255            chain_hash,
1256        })
1257    }
1258
1259    fn insert_preview(
1260        conn: &Connection,
1261        transaction_id: &str,
1262        preview: &PreviewEnvelope,
1263    ) -> Result<(), TransactionStoreError> {
1264        conn.execute(
1265            "INSERT INTO transaction_previews (transaction_id, preview_json)
1266             VALUES (?1, ?2)",
1267            params![transaction_id, serde_json::to_string(preview)?],
1268        )?;
1269        Ok(())
1270    }
1271}
1272
1273/// Create `dir` and any missing parents with mode `0o700` (rwx owner only).
1274///
1275/// If the directory already exists, its mode is left untouched — the operator
1276/// or packaging spec (`sysknife-tmpfiles.conf`) owns existing-directory policy.
1277/// If the directory must be created here (e.g. dev contributor's first daemon
1278/// run), we never produce a world-readable audit-log directory.
1279fn ensure_private_dir(dir: &Path) -> std::io::Result<()> {
1280    use std::os::unix::fs::DirBuilderExt;
1281    if dir.exists() {
1282        return Ok(());
1283    }
1284    std::fs::DirBuilder::new()
1285        .recursive(true)
1286        .mode(0o700)
1287        .create(dir)
1288}
1289
1290/// Narrow the audit database (and its WAL sidecars) to mode `0o600`.
1291///
1292/// SQLite creates its files `0o644 & ~umask`. The 0700 parent directory hides
1293/// them from other users today, but the file mode travels: a backup, a `cp`, or
1294/// an operator who widens the directory exposes every action name, parameter,
1295/// and approving principal the daemon has recorded. SQLite also derives the
1296/// `-wal`/`-shm` modes from the main file, so leaving the main file permissive
1297/// leaks the most recent (not-yet-checkpointed) transactions as well.
1298///
1299/// Tightening only, and never fatal: the directory is the primary control, and
1300/// a `chmod` that fails because the file belongs to another user must not stop
1301/// a daemon that previously started fine. It is loud, because a security
1302/// property that quietly stops holding is worse than one that never held.
1303fn restrict_db_file_mode(db_path: &Path) {
1304    use std::os::unix::fs::PermissionsExt;
1305
1306    // `-shm` is recreated per connection; `-wal` persists. Both inherit from the
1307    // main file at creation, so fixing the main file covers the common case and
1308    // these two cover a database that already exists with a wider mode.
1309    let sidecars = [
1310        db_path.to_path_buf(),
1311        with_suffix(db_path, "-wal"),
1312        with_suffix(db_path, "-shm"),
1313    ];
1314    for path in sidecars {
1315        let Ok(metadata) = std::fs::metadata(&path) else {
1316            continue; // Absent sidecar: nothing to narrow.
1317        };
1318        let mode = metadata.permissions().mode() & 0o777;
1319        if mode & 0o077 == 0 {
1320            continue; // Already owner-only or tighter.
1321        }
1322        if let Err(err) = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) {
1323            eprintln!(
1324                "[sysknife-daemon] WARNING: {} is mode {mode:o} and could not be \
1325                 narrowed to 0600 ({err}); it is readable by anyone who can reach \
1326                 it. The parent directory should be 0700 — check it.",
1327                path.display()
1328            );
1329        }
1330    }
1331}
1332
1333/// `path` with `suffix` appended to the file name (not the extension), matching
1334/// how SQLite names `daemon.sqlite-wal` beside `daemon.sqlite`.
1335fn with_suffix(path: &Path, suffix: &str) -> PathBuf {
1336    let mut name = path.as_os_str().to_os_string();
1337    name.push(suffix);
1338    PathBuf::from(name)
1339}
1340
1341fn row_to_record(row: &rusqlite::Row) -> Result<TransactionRecord, TransactionStoreError> {
1342    Ok(TransactionRecord {
1343        transaction_id: row.get(0)?,
1344        request_id: row.get(1)?,
1345        request_hash: row.get(2)?,
1346        action_name: row.get(3)?,
1347        risk_level: deserialize_field(&row.get::<_, String>(4)?)?,
1348        status: deserialize_field(&row.get::<_, String>(5)?)?,
1349        approval_id: row.get(6)?,
1350        summary: row.get(7)?,
1351        warnings: serde_json::from_str(&row.get::<_, String>(8)?)?,
1352    })
1353}
1354
1355fn serialize_field<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
1356    serde_json::to_string(value)
1357}
1358
1359fn deserialize_field<T: DeserializeOwned>(value: &str) -> Result<T, serde_json::Error> {
1360    serde_json::from_str(value)
1361}
1362
1363#[cfg(test)]
1364mod tests {
1365    use super::*;
1366    use crate::audit_chain::CHAIN_VERSION_CURRENT;
1367    use crate::audit_chain::CHAIN_VERSION_LEGACY;
1368    use std::os::unix::fs::PermissionsExt;
1369    use tempfile::tempdir;
1370
1371    /// Open a TransactionStore with a deterministic test key. Avoids the
1372    /// XDG/`/etc` lookup in `TransactionStore::open` so tests don't share
1373    /// state with the dev environment.
1374    fn test_store(path: impl AsRef<Path>) -> TransactionStore {
1375        let key = Arc::new(AuditKey::from_bytes(vec![0x42; 32]));
1376        TransactionStore::open_with_key(path, key).unwrap()
1377    }
1378
1379    // ── Schema migration ─────────────────────────────────────────────────
1380
1381    /// Create a database at exactly schema version 1 — the shape a v0.2.12
1382    /// daemon left behind — and write one row using the encoding that binary
1383    /// signed. Building the fixture from `SQLITE_MIGRATIONS[0]` rather than a
1384    /// copied DDL string means it stays honest if migration 1 is ever edited.
1385    fn legacy_v1_database(path: &Path, key: &AuditKey) -> String {
1386        let conn = Connection::open(path).unwrap();
1387        conn.execute_batch(
1388            "CREATE TABLE schema_migrations (
1389                 version INTEGER PRIMARY KEY,
1390                 name TEXT NOT NULL,
1391                 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1392             );",
1393        )
1394        .unwrap();
1395        conn.execute_batch(SQLITE_MIGRATIONS[0].sql).unwrap();
1396        conn.execute(
1397            "INSERT INTO schema_migrations (version, name) VALUES (1, ?1)",
1398            params![SQLITE_MIGRATIONS[0].name],
1399        )
1400        .unwrap();
1401
1402        let transaction_id = "tx-legacy";
1403        let created_at = "2026-04-24T12:00:00.000Z";
1404        let chain_hash = key.chain_hash(
1405            &ChainContent {
1406                seq: 1,
1407                key_id: CURRENT_KEY_ID,
1408                transaction_id,
1409                request_id: "req-legacy",
1410                request_hash: "hash-legacy",
1411                action_name: "UpdateSystem",
1412                risk_level: RiskLevel::High,
1413                summary: "Upgrade the system",
1414                approval_id: None,
1415                warnings_json: "[]",
1416                created_at,
1417                identity: ChainIdentity::LegacyV1,
1418            },
1419            "",
1420        );
1421        conn.execute(
1422            "INSERT INTO transactions (
1423                transaction_id, request_id, request_hash, action_name, risk_level,
1424                status, approval_id, summary, warnings_json, created_at,
1425                seq, key_id, chain_hash, prev_chain_hash
1426            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '')",
1427            params![
1428                transaction_id,
1429                "req-legacy",
1430                "hash-legacy",
1431                "UpdateSystem",
1432                serialize_field(&RiskLevel::High).unwrap(),
1433                serialize_field(&JobState::Queued).unwrap(),
1434                "Upgrade the system",
1435                "[]",
1436                created_at,
1437                CURRENT_KEY_ID,
1438                chain_hash,
1439            ],
1440        )
1441        .unwrap();
1442        transaction_id.to_string()
1443    }
1444
1445    /// A database at schema version 2 holding one v2 row: the shape every host
1446    /// running v0.2.13 through v0.2.16 has on disk right now.
1447    ///
1448    /// Built by replaying `SQLITE_MIGRATIONS[0..=1]` rather than hardcoded DDL,
1449    /// so editing those migrations cannot leave this fixture describing a schema
1450    /// that never existed.
1451    fn v2_database(path: &Path, key: &AuditKey) -> String {
1452        let conn = Connection::open(path).unwrap();
1453        conn.execute_batch(
1454            "CREATE TABLE schema_migrations (
1455                 version INTEGER PRIMARY KEY,
1456                 name TEXT NOT NULL,
1457                 applied_at TEXT NOT NULL DEFAULT (datetime('now'))
1458             );",
1459        )
1460        .unwrap();
1461        for migration in &SQLITE_MIGRATIONS[0..=1] {
1462            conn.execute_batch(migration.sql).unwrap();
1463            conn.execute(
1464                "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
1465                params![migration.version, migration.name],
1466            )
1467            .unwrap();
1468        }
1469
1470        let transaction_id = "tx-v2";
1471        let created_at = "2026-07-01T12:00:00.000Z";
1472        let chain_hash = key.chain_hash(
1473            &ChainContent {
1474                seq: 1,
1475                key_id: CURRENT_KEY_ID,
1476                transaction_id,
1477                request_id: "req-v2",
1478                request_hash: "hash-v2",
1479                action_name: "UpdateSystem",
1480                risk_level: RiskLevel::High,
1481                summary: "Upgrade the system",
1482                approval_id: None,
1483                warnings_json: "[]",
1484                created_at,
1485                identity: ChainIdentity::V2 {
1486                    caller_role: "admin",
1487                    event_tip: "",
1488                },
1489            },
1490            "",
1491        );
1492        conn.execute(
1493            "INSERT INTO transactions (
1494                transaction_id, request_id, request_hash, action_name, risk_level,
1495                status, approval_id, summary, warnings_json, created_at,
1496                seq, key_id, chain_hash, prev_chain_hash,
1497                chain_version, caller_role, event_tip
1498            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, ?7, ?8, ?9, 1, ?10, ?11, '', 2, 'admin', '')",
1499            params![
1500                transaction_id,
1501                "req-v2",
1502                "hash-v2",
1503                "UpdateSystem",
1504                serialize_field(&RiskLevel::High).unwrap(),
1505                serialize_field(&JobState::Queued).unwrap(),
1506                "Upgrade the system",
1507                "[]",
1508                created_at,
1509                CURRENT_KEY_ID,
1510                chain_hash,
1511            ],
1512        )
1513        .unwrap();
1514        transaction_id.to_string()
1515    }
1516
1517    /// The upgrade every deployed host actually performs: v2 on disk, then v3
1518    /// rows appended by the new daemon, in one chain.
1519    ///
1520    /// The v1 fixture already covers the older jump, but nothing covered this
1521    /// one, and it is the only one with installed users. If migration 3 ever
1522    /// backfilled a principal, or the v2 dispatch drifted, this is where a real
1523    /// operator's audit log would start reporting as tampered.
1524    #[test]
1525    fn a_v2_database_keeps_verifying_and_accepts_v3_rows() {
1526        let dir = tempdir().unwrap();
1527        let db_path = dir.path().join("tx.db");
1528        let key = AuditKey::from_bytes(vec![0x42; 32]);
1529        let old_transaction = v2_database(&db_path, &key);
1530
1531        let store = test_store(&db_path);
1532
1533        let version: i64 = store
1534            .connection()
1535            .unwrap()
1536            .query_row("SELECT MAX(version) FROM schema_migrations", [], |row| {
1537                row.get(0)
1538            })
1539            .unwrap();
1540        assert_eq!(version, 3, "opening a v2 database must apply migration 3");
1541
1542        let rows = store.fetch_chain_rows().unwrap();
1543        assert_eq!(rows.len(), 1);
1544        assert_eq!(rows[0].transaction_id, old_transaction);
1545        assert_eq!(rows[0].chain_version, audit_chain::CHAIN_VERSION_V2);
1546        assert_eq!(
1547            rows[0].caller_principal, None,
1548            "migration 3 must not invent a principal for a row signed without one"
1549        );
1550        assert!(matches!(
1551            store.verify_audit_chain(&key).unwrap(),
1552            audit_chain::VerifyOutcome::Intact { rows_checked: 1 }
1553        ));
1554
1555        // Append under the new encoding and re-verify the mixed chain.
1556        let mut fresh = queued_transaction();
1557        fresh.caller_principal = CallerPrincipal::Uid(4242);
1558        store.record(fresh).unwrap();
1559
1560        let rows = store.fetch_chain_rows().unwrap();
1561        assert_eq!(rows.len(), 2);
1562        assert_eq!(rows[1].chain_version, audit_chain::CHAIN_VERSION_CURRENT);
1563        assert_eq!(rows[1].caller_principal.as_deref(), Some("uid:4242"));
1564        assert_eq!(
1565            rows[1].prev_chain_hash, rows[0].chain_hash,
1566            "the v3 row must chain onto the v2 row, not start a new chain"
1567        );
1568        assert!(matches!(
1569            store.verify_audit_chain(&key).unwrap(),
1570            audit_chain::VerifyOutcome::Intact { rows_checked: 2 }
1571        ));
1572
1573        // The census, taken over rows that made a real round trip through SQLite
1574        // rather than hand-built fixtures. This is the only place that can catch
1575        // the SQL side going wrong: a `SELECT` that dropped `caller_principal`, or
1576        // a mapping that coerced `NULL` to `""`, would move every legacy row out
1577        // of "names nobody" and into "names an account", restoring the exact
1578        // misreport this census exists to prevent, with every in-memory test
1579        // still green.
1580        let census = audit_chain::AttributionCensus::of(&rows);
1581        assert_eq!(census.rows(), 2);
1582        assert_eq!(census.named(), 1, "only the v3 row names uid:4242");
1583        assert_eq!(
1584            census.not_recorded(),
1585            1,
1586            "the v2 row's encoding never signed a principal"
1587        );
1588        assert_eq!(census.attribution_failed(), 0);
1589        assert_eq!(census.unattested(), 0);
1590        assert_eq!(census.unnamed(), 1);
1591    }
1592
1593    #[test]
1594    fn a_database_written_before_the_migration_still_verifies_after_upgrading() {
1595        // The migration contract, end to end through the store. An operator
1596        // upgrading the daemon must not see their existing audit log start
1597        // reporting as tampered — that would make a real compromise
1598        // indistinguishable from a routine upgrade.
1599        let dir = tempdir().unwrap();
1600        let db_path = dir.path().join("tx.db");
1601        let key = AuditKey::from_bytes(vec![0x42; 32]);
1602        legacy_v1_database(&db_path, &key);
1603
1604        let store = test_store(&db_path);
1605        assert_eq!(
1606            store.verify_audit_chain(&key).unwrap(),
1607            VerifyOutcome::Intact { rows_checked: 1 }
1608        );
1609        let rows = store.fetch_chain_rows().unwrap();
1610        assert_eq!(rows[0].chain_version, CHAIN_VERSION_LEGACY);
1611        assert_eq!(rows[0].caller_role, None);
1612    }
1613
1614    #[test]
1615    fn rows_appended_after_the_migration_chain_onto_legacy_rows() {
1616        // Mixed-generation chain written through the real store, not a
1617        // hand-built fixture: the new row's prev_chain_hash must link to the
1618        // legacy row, and both encodings must verify in one walk.
1619        let dir = tempdir().unwrap();
1620        let db_path = dir.path().join("tx.db");
1621        let key = AuditKey::from_bytes(vec![0x42; 32]);
1622        legacy_v1_database(&db_path, &key);
1623
1624        let store = test_store(&db_path);
1625        store.record(queued_transaction()).unwrap();
1626
1627        let rows = store.fetch_chain_rows().unwrap();
1628        assert_eq!(rows.len(), 2);
1629        assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
1630        assert_eq!(rows[1].chain_version, CHAIN_VERSION_CURRENT);
1631        assert_eq!(
1632            store.verify_audit_chain(&key).unwrap(),
1633            VerifyOutcome::Intact { rows_checked: 2 }
1634        );
1635    }
1636
1637    #[test]
1638    fn a_schema_newer_than_this_binary_is_refused() {
1639        let dir = tempdir().unwrap();
1640        let db_path = dir.path().join("tx.db");
1641        let key = AuditKey::from_bytes(vec![0x42; 32]);
1642        legacy_v1_database(&db_path, &key);
1643        let conn = Connection::open(&db_path).unwrap();
1644        conn.execute(
1645            "INSERT INTO schema_migrations (version, name) VALUES (99, 'from the future')",
1646            [],
1647        )
1648        .unwrap();
1649        drop(conn);
1650
1651        let err = TransactionStore::open_with_key(&db_path, Arc::new(key)).unwrap_err();
1652        assert!(
1653            matches!(err, TransactionStoreError::DatabaseInvariant(ref m) if m.contains("99")),
1654            "expected a refusal naming the unsupported version, got {err:?}"
1655        );
1656    }
1657
1658    // ── caller identity + approval events ────────────────────────────────
1659
1660    #[test]
1661    fn the_signed_row_records_which_role_asked() {
1662        let dir = tempdir().unwrap();
1663        let store = test_store(dir.path().join("tx.db"));
1664        let mut tx = queued_transaction();
1665        tx.caller_role = CallerRole::Admin;
1666        store.record(tx).unwrap();
1667
1668        let rows = store.fetch_chain_rows().unwrap();
1669        assert_eq!(rows[0].caller_role.as_deref(), Some("admin"));
1670    }
1671
1672    #[test]
1673    fn approving_consuming_and_revoking_each_append_a_chained_event() {
1674        let dir = tempdir().unwrap();
1675        let key = AuditKey::from_bytes(vec![0x42; 32]);
1676        let store = test_store(dir.path().join("tx.db"));
1677
1678        // Approve then consume.
1679        let a = store.record(queued_transaction()).unwrap();
1680        let receipt = store
1681            .approve_transaction(&a.transaction_id)
1682            .unwrap()
1683            .expect("queued transaction approves");
1684        let digest = audit_chain::approval_receipt_digest(&receipt);
1685        assert!(store
1686            .claim_approved_for_execution(&a.transaction_id, &digest)
1687            .unwrap());
1688
1689        // Approve then revoke.
1690        let b = store.record(queued_transaction()).unwrap();
1691        store.approve_transaction(&b.transaction_id).unwrap();
1692        assert!(store.revoke_unconsumed_approval(&b.transaction_id).unwrap());
1693
1694        let events = store.fetch_event_rows().unwrap();
1695        let kinds: Vec<&str> = events.iter().map(|e| e.kind.as_str()).collect();
1696        assert_eq!(
1697            kinds,
1698            vec![
1699                "approval_granted",
1700                "approval_consumed",
1701                "approval_granted",
1702                "approval_revoked",
1703            ]
1704        );
1705        assert_eq!(
1706            store.verify_event_chain(&key).unwrap(),
1707            VerifyOutcome::Intact { rows_checked: 4 }
1708        );
1709    }
1710
1711    #[test]
1712    fn a_failed_approval_appends_no_event() {
1713        // Only real state changes get chained. A no-op approve (already
1714        // approved) writing an event would make the trail claim something that
1715        // did not happen.
1716        let dir = tempdir().unwrap();
1717        let store = test_store(dir.path().join("tx.db"));
1718        let tx = store.record(queued_transaction()).unwrap();
1719        store.approve_transaction(&tx.transaction_id).unwrap();
1720        store.approve_transaction(&tx.transaction_id).unwrap();
1721        assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
1722
1723        // Same for a revoke with nothing to revoke.
1724        assert!(!store.revoke_unconsumed_approval("no-such-tx").unwrap());
1725        assert_eq!(store.fetch_event_rows().unwrap().len(), 1);
1726    }
1727
1728    #[test]
1729    fn deleting_the_record_that_an_approval_happened_no_longer_goes_unnoticed() {
1730        // The finding. Before the event chain, `transaction_approvals` was a
1731        // plain table: DELETE the row and `audit verify` still said Intact, so
1732        // the fact that a privileged action had been approved could be erased
1733        // without a trace. Now the deletion has to survive two independent
1734        // checks.
1735        let dir = tempdir().unwrap();
1736        let db_path = dir.path().join("tx.db");
1737        let key = AuditKey::from_bytes(vec![0x42; 32]);
1738        let store = test_store(&db_path);
1739
1740        let a = store.record(queued_transaction()).unwrap();
1741        store.approve_transaction(&a.transaction_id).unwrap();
1742        // A later transaction commits to the event tip, which is what carries
1743        // the binding into the checkpoint-anchored transaction chain.
1744        store.record(queued_transaction()).unwrap();
1745
1746        assert_eq!(
1747            store.verify_event_chain(&key).unwrap(),
1748            VerifyOutcome::Intact { rows_checked: 1 }
1749        );
1750        assert_eq!(
1751            store.verify_event_binding().unwrap(),
1752            audit_chain::BindingOutcome::Consistent {
1753                bindings_checked: 1
1754            }
1755        );
1756
1757        // Erase the approval and its event.
1758        let conn = Connection::open(&db_path).unwrap();
1759        conn.execute("DELETE FROM transaction_approvals", [])
1760            .unwrap();
1761        conn.execute("DELETE FROM audit_events", []).unwrap();
1762        drop(conn);
1763
1764        // The event chain alone is now empty and walks clean — deleting the
1765        // only event leaves nothing to contradict. The transaction row's
1766        // committed tip is what catches it.
1767        assert_eq!(
1768            store.verify_event_chain(&key).unwrap(),
1769            VerifyOutcome::Intact { rows_checked: 0 }
1770        );
1771        match store.verify_event_binding().unwrap() {
1772            audit_chain::BindingOutcome::MissingEvent {
1773                transaction_seq, ..
1774            } => assert_eq!(transaction_seq, 2),
1775            other => panic!("expected MissingEvent, got {other:?}"),
1776        }
1777    }
1778
1779    #[test]
1780    fn a_read_only_store_cannot_revoke_or_claim() {
1781        // Both paths now append to the event chain, so both need the signing
1782        // key. Refusing loudly beats silently skipping the event.
1783        let dir = tempdir().unwrap();
1784        let db_path = dir.path().join("tx.db");
1785        let store = test_store(&db_path);
1786        store.record(queued_transaction()).unwrap();
1787
1788        let read_only = TransactionStore::open_read_only(&db_path).unwrap();
1789        assert!(matches!(
1790            read_only.revoke_unconsumed_approval("tx"),
1791            Err(TransactionStoreError::AuditChainMissing(_))
1792        ));
1793        assert!(matches!(
1794            read_only.claim_approved_for_execution("tx", "digest"),
1795            Err(TransactionStoreError::AuditChainMissing(_))
1796        ));
1797    }
1798
1799    // ── Audit chain integration tests ────────────────────────────────────
1800
1801    /// The store must persist the principal the dispatcher resolved, under the
1802    /// v3 tag, or the chain claims an encoding whose evidence is missing. Reads
1803    /// the columns directly rather than through the verifier, so a mismatch
1804    /// between what was signed and what was stored is visible.
1805    #[test]
1806    fn record_persists_the_caller_principal_under_the_current_encoding() {
1807        let dir = tempdir().unwrap();
1808        let store = test_store(dir.path().join("tx.db"));
1809        let mut tx = queued_transaction();
1810        tx.caller_principal = CallerPrincipal::Uid(4242);
1811        store.record(tx).unwrap();
1812
1813        let conn = store.connection().unwrap();
1814        let (version, role, principal): (i64, Option<String>, Option<String>) = conn
1815            .query_row(
1816                "SELECT chain_version, caller_role, caller_principal FROM transactions",
1817                [],
1818                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
1819            )
1820            .unwrap();
1821
1822        assert_eq!(version, i64::from(audit_chain::CHAIN_VERSION_CURRENT));
1823        assert_eq!(role.as_deref(), Some("dev"));
1824        assert_eq!(
1825            principal.as_deref(),
1826            Some("uid:4242"),
1827            "the row must name the account the dispatcher resolved"
1828        );
1829
1830        // And the signature must agree with those columns.
1831        let key = AuditKey::from_bytes(vec![0x42; 32]);
1832        assert!(matches!(
1833            store.verify_audit_chain(&key).unwrap(),
1834            audit_chain::VerifyOutcome::Intact { .. }
1835        ));
1836    }
1837
1838    #[test]
1839    fn record_writes_audit_chain_columns() {
1840        let dir = tempdir().unwrap();
1841        let store = test_store(dir.path().join("tx.db"));
1842        let _record = store.record(queued_transaction()).unwrap();
1843
1844        let conn = store.connection().unwrap();
1845        let (seq, key_id, chain_hash, prev): (i64, String, String, String) = conn
1846            .query_row(
1847                "SELECT seq, key_id, chain_hash, prev_chain_hash FROM transactions",
1848                [],
1849                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
1850            )
1851            .unwrap();
1852        assert_eq!(seq, 1, "first row gets seq=1");
1853        assert_eq!(key_id, audit_chain::CURRENT_KEY_ID);
1854        assert_eq!(prev, "", "first row has empty prev_chain_hash");
1855        assert_eq!(
1856            chain_hash.len(),
1857            audit_chain::HASH_HEX_LEN,
1858            "chain_hash is a hex-encoded Ed25519 signature"
1859        );
1860    }
1861
1862    #[test]
1863    fn sequential_records_produce_chained_hashes() {
1864        let dir = tempdir().unwrap();
1865        let store = test_store(dir.path().join("tx.db"));
1866        store.record(queued_transaction()).unwrap();
1867        store.record(queued_transaction()).unwrap();
1868        store.record(queued_transaction()).unwrap();
1869
1870        let rows = store.fetch_chain_rows().unwrap();
1871        assert_eq!(rows.len(), 3);
1872        assert_eq!(rows[0].seq, 1);
1873        assert_eq!(rows[1].seq, 2);
1874        assert_eq!(rows[2].seq, 3);
1875        // Each row's prev_chain_hash matches the previous row's chain_hash.
1876        assert_eq!(rows[1].prev_chain_hash, rows[0].chain_hash);
1877        assert_eq!(rows[2].prev_chain_hash, rows[1].chain_hash);
1878    }
1879
1880    /// T3 — concurrent `record()` keeps the chain intact and seqs contiguous.
1881    ///
1882    /// The store guarantees this via `BEGIN IMMEDIATE` on every record:
1883    /// the immediate write lock means `next_seq_and_prev_hash` is read
1884    /// inside the same SQLite transaction that does the INSERT, so two
1885    /// records cannot both observe `seq=N` and produce two rows with the
1886    /// same chain hash.  Drive 8 worker threads × 10 records each
1887    /// through the same store and assert (a) `verify_audit_chain` returns
1888    /// Intact { rows_checked: 80 } and (b) the seq column is contiguous
1889    /// 1..=80.  A regression that drops `BEGIN IMMEDIATE` or substitutes
1890    /// a non-locking read fails one of these on every run.
1891    #[test]
1892    fn concurrent_record_keeps_chain_intact_and_seqs_contiguous() {
1893        const WORKERS: usize = 8;
1894        const RECORDS_PER_WORKER: usize = 10;
1895        const TOTAL: usize = WORKERS * RECORDS_PER_WORKER;
1896
1897        let dir = tempdir().unwrap();
1898        let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
1899
1900        let mut handles = Vec::with_capacity(WORKERS);
1901        for w in 0..WORKERS {
1902            let store = std::sync::Arc::clone(&store);
1903            handles.push(std::thread::spawn(move || {
1904                for r in 0..RECORDS_PER_WORKER {
1905                    let tx = NewTransaction {
1906                        request_id: format!("worker-{w}-record-{r}"),
1907                        request_hash: format!("hash-{w}-{r}"),
1908                        action_name: "GetSystemState".to_string(),
1909                        risk_level: RiskLevel::Low,
1910                        summary: format!("worker {w} record {r}"),
1911                        warnings: vec![],
1912                        caller_role: CallerRole::Dev,
1913                        // Distinct per worker: with every row claiming the same
1914                        // account, the chain could attribute a row to the wrong
1915                        // caller under contention and still verify.
1916                        caller_principal: CallerPrincipal::Uid(3000 + w as u32),
1917                    };
1918                    store
1919                        .record(tx)
1920                        .expect("record must succeed under contention");
1921                }
1922            }));
1923        }
1924        for h in handles {
1925            h.join().expect("worker thread did not panic");
1926        }
1927
1928        // (a) chain must be intact end-to-end.
1929        let key = AuditKey::from_bytes(vec![0x42; 32]);
1930        let outcome = store.verify_audit_chain(&key).unwrap();
1931        match outcome {
1932            VerifyOutcome::Intact { rows_checked } => {
1933                assert_eq!(
1934                    rows_checked, TOTAL as u64,
1935                    "expected {TOTAL} rows checked, got {rows_checked}"
1936                );
1937            }
1938            other => panic!("chain must be Intact under concurrent writes; got {other:?}"),
1939        }
1940
1941        // (a2) every row must still name the worker that wrote it. The summary
1942        // carries the worker id independently of the principal, so a row whose
1943        // principal drifted onto another worker's insert is detectable.
1944        {
1945            let conn = store.connection().unwrap();
1946            let mut stmt = conn
1947                .prepare("SELECT summary, caller_principal FROM transactions")
1948                .unwrap();
1949            let pairs: Vec<(String, Option<String>)> = stmt
1950                .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
1951                .unwrap()
1952                .map(|r| r.unwrap())
1953                .collect();
1954            assert_eq!(pairs.len(), TOTAL);
1955            for (summary, principal) in pairs {
1956                let worker: u32 = summary
1957                    .split_whitespace()
1958                    .nth(1)
1959                    .and_then(|w| w.parse().ok())
1960                    .unwrap_or_else(|| panic!("unexpected summary {summary:?}"));
1961                assert_eq!(
1962                    principal.as_deref(),
1963                    Some(format!("uid:{}", 3000 + worker).as_str()),
1964                    "row {summary:?} was attributed to the wrong account"
1965                );
1966            }
1967        }
1968
1969        // (b) seq must be a contiguous run 1..=TOTAL with no gaps and no duplicates.
1970        let conn = store.connection().unwrap();
1971        let mut stmt = conn
1972            .prepare("SELECT seq FROM transactions ORDER BY seq ASC")
1973            .unwrap();
1974        let seqs: Vec<i64> = stmt
1975            .query_map([], |row| row.get::<_, i64>(0))
1976            .unwrap()
1977            .map(|r| r.unwrap())
1978            .collect();
1979        assert_eq!(seqs.len(), TOTAL, "row count mismatch");
1980        for (i, s) in seqs.iter().enumerate() {
1981            assert_eq!(
1982                *s,
1983                (i as i64) + 1,
1984                "seq column must be contiguous 1..={TOTAL}; saw {s} at position {i}"
1985            );
1986        }
1987    }
1988
1989    #[test]
1990    fn verify_audit_chain_intact_after_inserts() {
1991        let dir = tempdir().unwrap();
1992        let store = test_store(dir.path().join("tx.db"));
1993        for _ in 0..3 {
1994            store.record(queued_transaction()).unwrap();
1995        }
1996        let key = AuditKey::from_bytes(vec![0x42; 32]);
1997        let outcome = store.verify_audit_chain(&key).unwrap();
1998        assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
1999    }
2000
2001    #[test]
2002    fn verify_audit_chain_with_pubkey_intact_after_inserts() {
2003        let dir = tempdir().unwrap();
2004        let store = test_store(dir.path().join("tx.db"));
2005        for _ in 0..3 {
2006            store.record(queued_transaction()).unwrap();
2007        }
2008        // Auditor path: verify with ONLY the public key, no private key.
2009        let key = AuditKey::from_bytes(vec![0x42; 32]);
2010        let outcome = store
2011            .verify_audit_chain_with_pubkey(&key.verifying_key_hex())
2012            .unwrap();
2013        assert!(matches!(outcome, VerifyOutcome::Intact { rows_checked: 3 }));
2014    }
2015
2016    #[test]
2017    fn verify_audit_chain_with_wrong_pubkey_is_broken() {
2018        let dir = tempdir().unwrap();
2019        let store = test_store(dir.path().join("tx.db"));
2020        store.record(queued_transaction()).unwrap();
2021        // A different keypair's public key must not validate the chain.
2022        let other = AuditKey::from_bytes(vec![0x99; 32]);
2023        let outcome = store
2024            .verify_audit_chain_with_pubkey(&other.verifying_key_hex())
2025            .unwrap();
2026        assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2027    }
2028
2029    #[test]
2030    fn verify_detects_tampered_summary() {
2031        let dir = tempdir().unwrap();
2032        let store = test_store(dir.path().join("tx.db"));
2033        let tx = store.record(queued_transaction()).unwrap();
2034
2035        // Tamper: modify the summary field directly via SQL — simulates an
2036        // attacker with database write access who skips the daemon code path.
2037        let conn = store.connection().unwrap();
2038        conn.execute(
2039            "UPDATE transactions SET summary = ?1 WHERE transaction_id = ?2",
2040            params!["EVIL CHANGE", tx.transaction_id],
2041        )
2042        .unwrap();
2043
2044        let key = AuditKey::from_bytes(vec![0x42; 32]);
2045        let outcome = store.verify_audit_chain(&key).unwrap();
2046        assert!(matches!(outcome, VerifyOutcome::Broken { .. }));
2047    }
2048
2049    #[test]
2050    fn status_update_does_not_break_chain() {
2051        // Status is mutable; the chain protects only immutable fields.
2052        let dir = tempdir().unwrap();
2053        let store = test_store(dir.path().join("tx.db"));
2054        let tx = store.record(queued_transaction()).unwrap();
2055        store
2056            .update_status(&tx.transaction_id, JobState::Running)
2057            .unwrap();
2058        store
2059            .update_status(&tx.transaction_id, JobState::Succeeded)
2060            .unwrap();
2061
2062        let key = AuditKey::from_bytes(vec![0x42; 32]);
2063        let outcome = store.verify_audit_chain(&key).unwrap();
2064        assert!(
2065            matches!(outcome, VerifyOutcome::Intact { rows_checked: 1 }),
2066            "status mutation must not break the chain (status not in chain content): {outcome:?}"
2067        );
2068    }
2069
2070    #[test]
2071    fn open_read_only_rejects_record() {
2072        let dir = tempdir().unwrap();
2073        let key_path = dir.path().join("audit-key");
2074        std::fs::write(&key_path, vec![0x42; 32]).unwrap();
2075        std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)).unwrap();
2076
2077        let store = TransactionStore::open_read_only(dir.path().join("tx.db")).unwrap();
2078        let result = store.record(queued_transaction());
2079        assert!(matches!(
2080            result,
2081            Err(TransactionStoreError::AuditChainMissing(_))
2082        ));
2083    }
2084
2085    #[test]
2086    fn ensure_private_dir_creates_with_0700_mode() {
2087        let root = tempdir().unwrap();
2088        let target = root.path().join("a/b/c");
2089        ensure_private_dir(&target).unwrap();
2090        assert!(target.is_dir());
2091        let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2092        assert_eq!(mode, 0o700, "leaf dir must be 0o700, got {mode:o}");
2093    }
2094
2095    /// The audit database is the record of every privileged action taken on the
2096    /// host: action names, parameters, and who approved them. SQLite creates it
2097    /// `0644 & ~umask`, which the 0700 parent hides today — but only today. A
2098    /// backup, a `cp`, or an operator who widens the directory carries the file
2099    /// mode with it, and SQLite derives the `-wal`/`-shm` sidecar modes from the
2100    /// main file, so a permissive main file leaks the most recent transactions
2101    /// through the WAL too.
2102    #[test]
2103    fn the_audit_database_is_created_owner_only() {
2104        let dir = tempdir().unwrap();
2105        let db = dir.path().join("daemon.sqlite");
2106        let store = TransactionStore::open(&db).unwrap();
2107        // Write a row so the WAL sidecar exists and can be checked as well.
2108        store
2109            .record(NewTransaction {
2110                request_id: "mode-probe".to_string(),
2111                request_hash: "hash-mode-probe".to_string(),
2112                action_name: "GetSystemState".to_string(),
2113                risk_level: RiskLevel::Low,
2114                summary: "probe".to_string(),
2115                warnings: vec![],
2116                caller_role: CallerRole::Dev,
2117                caller_principal: CallerPrincipal::Uid(3000),
2118            })
2119            .unwrap();
2120
2121        for path in [db.clone(), db.with_extension("sqlite-wal")] {
2122            if !path.exists() {
2123                continue;
2124            }
2125            let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
2126            assert_eq!(
2127                mode,
2128                0o600,
2129                "{} must be owner-only, got {mode:o}",
2130                path.display()
2131            );
2132        }
2133    }
2134
2135    #[test]
2136    fn ensure_private_dir_is_idempotent_and_does_not_widen_existing_mode() {
2137        let root = tempdir().unwrap();
2138        let target = root.path().join("preexisting");
2139        std::fs::create_dir(&target).unwrap();
2140        std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o755)).unwrap();
2141        ensure_private_dir(&target).unwrap();
2142        // Existing directory: we don't touch its mode.
2143        let mode = std::fs::metadata(&target).unwrap().permissions().mode() & 0o777;
2144        assert_eq!(mode, 0o755);
2145    }
2146
2147    #[test]
2148    fn open_creates_parent_with_private_mode() {
2149        let root = tempdir().unwrap();
2150        let db_path = root.path().join("nested/dirs/daemon.sqlite");
2151        let _store = test_store(&db_path);
2152        let parent = db_path.parent().unwrap();
2153        let mode = std::fs::metadata(parent).unwrap().permissions().mode() & 0o777;
2154        assert_eq!(mode, 0o700);
2155    }
2156
2157    fn queued_transaction() -> NewTransaction {
2158        NewTransaction {
2159            request_id: "req-1".to_string(),
2160            request_hash: "hash-abc".to_string(),
2161            action_name: "UpdateSystem".to_string(),
2162            risk_level: RiskLevel::High,
2163            summary: "Upgrade the system".to_string(),
2164            warnings: vec![],
2165            caller_role: CallerRole::Dev,
2166            caller_principal: CallerPrincipal::Uid(1000),
2167        }
2168    }
2169
2170    #[test]
2171    fn update_status_transitions_queued_to_running() {
2172        let dir = tempdir().unwrap();
2173        let store = test_store(dir.path().join("tx.db"));
2174        let tx = store.record(queued_transaction()).unwrap();
2175
2176        store
2177            .update_status(&tx.transaction_id, JobState::Running)
2178            .unwrap();
2179
2180        let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2181        assert_eq!(updated.status, JobState::Running);
2182    }
2183
2184    #[test]
2185    fn update_status_transitions_running_to_succeeded() {
2186        let dir = tempdir().unwrap();
2187        let store = test_store(dir.path().join("tx.db"));
2188        let tx = store.record(queued_transaction()).unwrap();
2189
2190        store
2191            .update_status(&tx.transaction_id, JobState::Running)
2192            .unwrap();
2193        store
2194            .update_status(&tx.transaction_id, JobState::Succeeded)
2195            .unwrap();
2196
2197        let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2198        assert_eq!(updated.status, JobState::Succeeded);
2199    }
2200
2201    #[test]
2202    fn update_status_for_unknown_id_returns_not_found() {
2203        let dir = tempdir().unwrap();
2204        let store = test_store(dir.path().join("tx.db"));
2205
2206        let result = store.update_status("does-not-exist", JobState::Running);
2207        assert!(
2208            matches!(result, Err(TransactionStoreError::NotFound(ref id)) if id == "does-not-exist"),
2209            "expected NotFound, got: {result:?}"
2210        );
2211    }
2212
2213    #[test]
2214    fn update_status_leaves_other_fields_intact() {
2215        let dir = tempdir().unwrap();
2216        let store = test_store(dir.path().join("tx.db"));
2217        let tx = store.record(queued_transaction()).unwrap();
2218
2219        store
2220            .update_status(&tx.transaction_id, JobState::Running)
2221            .unwrap();
2222        store
2223            .update_status(&tx.transaction_id, JobState::Failed)
2224            .unwrap();
2225
2226        let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2227        assert_eq!(updated.action_name, "UpdateSystem");
2228        assert_eq!(updated.risk_level, RiskLevel::High);
2229        assert_eq!(updated.status, JobState::Failed);
2230    }
2231
2232    /// `update_status` must be a compare-and-swap, not a check-then-act race.
2233    /// Bring a transaction to `Running`, then fire two conflicting terminal
2234    /// transitions (`Succeeded` and `Failed`) from concurrent threads. Both
2235    /// read `Running` and both pass `allowed_transition`, but only one may
2236    /// actually apply — the loser must see an explicit error, never a silent
2237    /// double-write. Before the `AND status = ?<observed>` guard, both calls
2238    /// would unconditionally UPDATE and both would return `Ok(())`, hiding the
2239    /// fact that one of them clobbered a transition it never validated.
2240    #[test]
2241    fn update_status_is_atomic_under_concurrent_conflicting_transitions() {
2242        let dir = tempdir().unwrap();
2243        let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2244        let tx = store.record(queued_transaction()).unwrap();
2245        store
2246            .update_status(&tx.transaction_id, JobState::Running)
2247            .unwrap();
2248
2249        let store_a = std::sync::Arc::clone(&store);
2250        let id_a = tx.transaction_id.clone();
2251        let succeed = std::thread::spawn(move || store_a.update_status(&id_a, JobState::Succeeded));
2252
2253        let store_b = std::sync::Arc::clone(&store);
2254        let id_b = tx.transaction_id.clone();
2255        let fail = std::thread::spawn(move || store_b.update_status(&id_b, JobState::Failed));
2256
2257        let succeed_result = succeed.join().unwrap();
2258        let fail_result = fail.join().unwrap();
2259
2260        // Exactly one of the two conflicting transitions may win.
2261        let oks = [succeed_result.is_ok(), fail_result.is_ok()]
2262            .into_iter()
2263            .filter(|ok| *ok)
2264            .count();
2265        assert_eq!(
2266            oks, 1,
2267            "exactly one concurrent transition should succeed: succeed={succeed_result:?} \
2268             fail={fail_result:?}"
2269        );
2270
2271        // The loser must be refused outright, not silently succeed and clobber
2272        // the winner. *How* it is refused depends on whether the two threads
2273        // actually overlapped, and both outcomes are correct:
2274        //
2275        //   ConcurrentStatusChange — they overlapped, so the loser read
2276        //       `Running` and its compare-and-swap found the row already moved.
2277        //   InvalidTransition      — they serialised, so the loser read the
2278        //       winner's committed terminal status and rejected the transition
2279        //       before reaching the CAS at all.
2280        //
2281        // Asserting only the first made this test depend on thread
2282        // interleaving; it passes on a many-core dev machine and fails on a
2283        // 2-core CI runner, where the first thread routinely finishes before
2284        // the second starts. The invariant under test is "exactly one wins and
2285        // the loser does not clobber it", which both errors satisfy.
2286        let loser = if succeed_result.is_ok() {
2287            &fail_result
2288        } else {
2289            &succeed_result
2290        };
2291        match loser {
2292            Err(TransactionStoreError::ConcurrentStatusChange(id)) => {
2293                assert_eq!(id, &tx.transaction_id);
2294            }
2295            Err(TransactionStoreError::InvalidTransition { from, to }) => {
2296                // The winner's terminal state must be what the loser saw.
2297                let winner = if succeed_result.is_ok() {
2298                    JobState::Succeeded
2299                } else {
2300                    JobState::Failed
2301                };
2302                assert_eq!(
2303                    *from, winner,
2304                    "the loser must have observed the winner's committed status"
2305                );
2306                assert_ne!(*to, winner, "the loser must be the other transition");
2307            }
2308            other => panic!(
2309                "the losing transition must be refused with either \
2310                 ConcurrentStatusChange or InvalidTransition, got: {other:?}"
2311            ),
2312        }
2313
2314        // The stored status must match whichever transition actually won —
2315        // never a mix, and never left at `Running`.
2316        let final_status = store.get(&tx.transaction_id).unwrap().unwrap().status;
2317        if succeed_result.is_ok() {
2318            assert_eq!(final_status, JobState::Succeeded);
2319        } else {
2320            assert_eq!(final_status, JobState::Failed);
2321        }
2322    }
2323
2324    #[test]
2325    fn revoke_unconsumed_approval_removes_an_unused_receipt() {
2326        // Revocation is the operator's "actually, no" between approving and
2327        // executing. Nothing covered it at all.
2328        let dir = tempdir().unwrap();
2329        let store = test_store(dir.path().join("tx.db"));
2330        let tx = store.record(queued_transaction()).unwrap();
2331        let receipt = store
2332            .approve_transaction(&tx.transaction_id)
2333            .unwrap()
2334            .expect("approved");
2335        let digest = audit_chain::approval_receipt_digest(&receipt);
2336
2337        assert!(
2338            store
2339                .revoke_unconsumed_approval(&tx.transaction_id)
2340                .unwrap(),
2341            "an unconsumed approval must be revocable"
2342        );
2343        assert!(
2344            !store
2345                .claim_approved_for_execution(&tx.transaction_id, &digest)
2346                .unwrap(),
2347            "a revoked receipt must no longer be claimable"
2348        );
2349    }
2350
2351    #[test]
2352    fn revoke_cannot_retract_an_approval_that_was_already_used() {
2353        // The documented guarantee: once a receipt has been consumed, the
2354        // execution it authorised is a fact and revocation must not rewrite
2355        // history to say otherwise.
2356        let dir = tempdir().unwrap();
2357        let store = test_store(dir.path().join("tx.db"));
2358        let tx = store.record(queued_transaction()).unwrap();
2359        let receipt = store
2360            .approve_transaction(&tx.transaction_id)
2361            .unwrap()
2362            .expect("approved");
2363        let digest = audit_chain::approval_receipt_digest(&receipt);
2364        assert!(store
2365            .claim_approved_for_execution(&tx.transaction_id, &digest)
2366            .unwrap());
2367
2368        assert!(
2369            !store
2370                .revoke_unconsumed_approval(&tx.transaction_id)
2371                .unwrap(),
2372            "a consumed approval must not be revocable"
2373        );
2374        assert_eq!(
2375            store.get(&tx.transaction_id).unwrap().unwrap().status,
2376            JobState::Running,
2377            "a failed revocation must not disturb the running transaction"
2378        );
2379    }
2380
2381    #[test]
2382    fn approve_refuses_a_transaction_that_is_no_longer_queued() {
2383        // Every existing approve test starts from a freshly-recorded Queued
2384        // row. Approving something already running (or finished) would mint a
2385        // receipt for work that is past the point of approval.
2386        let dir = tempdir().unwrap();
2387        let store = test_store(dir.path().join("tx.db"));
2388        let tx = store.record(queued_transaction()).unwrap();
2389        let receipt = store
2390            .approve_transaction(&tx.transaction_id)
2391            .unwrap()
2392            .expect("approved");
2393        let digest = audit_chain::approval_receipt_digest(&receipt);
2394        assert!(store
2395            .claim_approved_for_execution(&tx.transaction_id, &digest)
2396            .unwrap());
2397
2398        // Running.
2399        assert!(
2400            store
2401                .approve_transaction(&tx.transaction_id)
2402                .unwrap()
2403                .is_none(),
2404            "a Running transaction must not be approvable"
2405        );
2406
2407        // Terminal.
2408        store
2409            .update_status(&tx.transaction_id, JobState::Succeeded)
2410            .unwrap();
2411        assert!(
2412            store
2413                .approve_transaction(&tx.transaction_id)
2414                .unwrap()
2415                .is_none(),
2416            "a completed transaction must not be approvable"
2417        );
2418    }
2419
2420    #[test]
2421    fn only_one_of_two_concurrent_claims_can_execute() {
2422        // `claim_approved_for_execution` is the interlock that stops one
2423        // approval being spent twice. `update_status` has a concurrency test;
2424        // this path had none, and a double claim means the same approved
2425        // action runs twice.
2426        let dir = tempdir().unwrap();
2427        let store = std::sync::Arc::new(test_store(dir.path().join("tx.db")));
2428        let tx = store.record(queued_transaction()).unwrap();
2429        let receipt = store
2430            .approve_transaction(&tx.transaction_id)
2431            .unwrap()
2432            .expect("approved");
2433        let digest = audit_chain::approval_receipt_digest(&receipt);
2434
2435        let mut handles = Vec::new();
2436        for _ in 0..2 {
2437            let store = std::sync::Arc::clone(&store);
2438            let id = tx.transaction_id.clone();
2439            let digest = digest.clone();
2440            handles.push(std::thread::spawn(move || {
2441                store.claim_approved_for_execution(&id, &digest)
2442            }));
2443        }
2444        let claims: Vec<bool> = handles
2445            .into_iter()
2446            .map(|h| h.join().unwrap().unwrap_or(false))
2447            .collect();
2448
2449        assert_eq!(
2450            claims.iter().filter(|c| **c).count(),
2451            1,
2452            "exactly one claim may win: {claims:?}"
2453        );
2454        assert_eq!(
2455            store.get(&tx.transaction_id).unwrap().unwrap().status,
2456            JobState::Running
2457        );
2458    }
2459
2460    #[test]
2461    fn cancel_queued_cancels_a_queued_transaction_once() {
2462        let dir = tempdir().unwrap();
2463        let store = test_store(dir.path().join("tx.db"));
2464        let tx = store.record(queued_transaction()).unwrap();
2465
2466        assert!(
2467            store.cancel_queued(&tx.transaction_id).unwrap(),
2468            "a queued transaction is cancelable"
2469        );
2470        assert_eq!(
2471            store.get(&tx.transaction_id).unwrap().unwrap().status,
2472            JobState::Canceled
2473        );
2474        assert!(
2475            !store.cancel_queued(&tx.transaction_id).unwrap(),
2476            "an already-canceled transaction is not cancelable again"
2477        );
2478        assert!(
2479            !store.cancel_queued("no-such-transaction").unwrap(),
2480            "a missing transaction is not cancelable"
2481        );
2482        assert!(
2483            store.fetch_event_rows().unwrap().is_empty(),
2484            "canceling an unapproved transaction must not append an approval event"
2485        );
2486    }
2487
2488    #[test]
2489    fn cancel_queued_revokes_an_unconsumed_approval_and_appends_event() {
2490        let dir = tempdir().unwrap();
2491        let store = test_store(dir.path().join("tx.db"));
2492        let tx = store.record(queued_transaction()).unwrap();
2493        let receipt = store
2494            .approve_transaction(&tx.transaction_id)
2495            .unwrap()
2496            .expect("queued transaction is approvable");
2497        let digest = audit_chain::approval_receipt_digest(&receipt);
2498
2499        assert!(store.cancel_queued(&tx.transaction_id).unwrap());
2500        assert_eq!(
2501            store.get(&tx.transaction_id).unwrap().unwrap().status,
2502            JobState::Canceled
2503        );
2504        assert!(
2505            !store
2506                .claim_approved_for_execution(&tx.transaction_id, &digest)
2507                .unwrap(),
2508            "canceling must revoke the unconsumed receipt"
2509        );
2510        assert_eq!(
2511            store
2512                .fetch_event_rows()
2513                .unwrap()
2514                .iter()
2515                .map(|event| event.kind.as_str())
2516                .collect::<Vec<_>>(),
2517            vec!["approval_granted", "approval_revoked"]
2518        );
2519    }
2520
2521    #[test]
2522    fn cancel_queued_refuses_a_running_transaction() {
2523        // Option A: never cancel an in-flight privileged action. Once a
2524        // transaction is claimed (Running), cancel must refuse and leave it.
2525        let dir = tempdir().unwrap();
2526        let store = test_store(dir.path().join("tx.db"));
2527        let tx = store.record(queued_transaction()).unwrap();
2528        let receipt = store
2529            .approve_transaction(&tx.transaction_id)
2530            .unwrap()
2531            .expect("approved");
2532        let digest = audit_chain::approval_receipt_digest(&receipt);
2533        assert!(store
2534            .claim_approved_for_execution(&tx.transaction_id, &digest)
2535            .unwrap());
2536
2537        assert!(
2538            !store.cancel_queued(&tx.transaction_id).unwrap(),
2539            "a running transaction must not be cancelable"
2540        );
2541        assert_eq!(
2542            store.get(&tx.transaction_id).unwrap().unwrap().status,
2543            JobState::Running,
2544            "cancel must not disturb a running transaction"
2545        );
2546    }
2547
2548    #[test]
2549    fn approved_receipt_is_required_and_consumed_once() {
2550        let dir = tempdir().unwrap();
2551        let store = test_store(dir.path().join("tx.db"));
2552        let tx = store.record(queued_transaction()).unwrap();
2553
2554        assert!(
2555            !store
2556                .claim_approved_for_execution(&tx.transaction_id, "digest-a")
2557                .unwrap(),
2558            "an unapproved preview must not execute"
2559        );
2560        let receipt = store
2561            .approve_transaction(&tx.transaction_id)
2562            .unwrap()
2563            .expect("first approval must succeed");
2564        let digest = audit_chain::approval_receipt_digest(&receipt);
2565        assert_eq!(tx.approval_id.as_deref(), Some(digest.as_str()));
2566        assert!(
2567            store
2568                .approve_transaction(&tx.transaction_id)
2569                .unwrap()
2570                .is_none(),
2571            "approval is immutable once issued"
2572        );
2573        assert!(
2574            !store
2575                .claim_approved_for_execution(&tx.transaction_id, "wrong-digest")
2576                .unwrap(),
2577            "a forged receipt must not execute"
2578        );
2579        assert!(
2580            store
2581                .claim_approved_for_execution(&tx.transaction_id, &digest)
2582                .unwrap(),
2583            "the exact approved receipt must execute"
2584        );
2585        assert!(
2586            !store
2587                .claim_approved_for_execution(&tx.transaction_id, &digest)
2588                .unwrap(),
2589            "the receipt must be one-time"
2590        );
2591    }
2592
2593    #[test]
2594    fn approval_commitment_is_covered_by_the_signed_chain() {
2595        let dir = tempdir().unwrap();
2596        let store = test_store(dir.path().join("tx.db"));
2597        let tx = store.record(queued_transaction()).unwrap();
2598        assert!(tx.approval_id.is_some());
2599
2600        store
2601            .connection()
2602            .unwrap()
2603            .execute(
2604                "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
2605                params![tx.transaction_id],
2606            )
2607            .unwrap();
2608
2609        let key = AuditKey::from_bytes(vec![0x42; 32]);
2610        assert!(matches!(
2611            store.verify_audit_chain(&key).unwrap(),
2612            VerifyOutcome::Broken { .. }
2613        ));
2614    }
2615
2616    #[test]
2617    fn approve_rejects_a_forged_commitment_at_runtime() {
2618        // Defense-in-depth: even before the chain-verify pass runs, approving a
2619        // transaction whose stored commitment was tampered must fail closed via
2620        // the constant-time check in `approve_transaction`, not issue a receipt.
2621        let dir = tempdir().unwrap();
2622        let store = test_store(dir.path().join("tx.db"));
2623        let tx = store.record(queued_transaction()).unwrap();
2624
2625        store
2626            .connection()
2627            .unwrap()
2628            .execute(
2629                "UPDATE transactions SET approval_id = 'forged' WHERE transaction_id = ?1",
2630                params![tx.transaction_id],
2631            )
2632            .unwrap();
2633
2634        let err = store
2635            .approve_transaction(&tx.transaction_id)
2636            .expect_err("a forged commitment must be rejected, not approved");
2637        assert!(
2638            matches!(err, TransactionStoreError::DatabaseInvariant(_)),
2639            "forged commitment must surface as a DatabaseInvariant, got {err:?}"
2640        );
2641    }
2642
2643    #[test]
2644    fn stale_iso_timestamp_cannot_be_approved() {
2645        let dir = tempdir().unwrap();
2646        let store = test_store(dir.path().join("tx.db"));
2647        let tx = store.record(queued_transaction()).unwrap();
2648        let conn = store.connection().unwrap();
2649        conn.execute(
2650            "UPDATE transactions \
2651             SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2652             WHERE transaction_id = ?1",
2653            params![tx.transaction_id],
2654        )
2655        .unwrap();
2656
2657        assert!(
2658            store
2659                .approve_transaction(&tx.transaction_id)
2660                .unwrap()
2661                .is_none(),
2662            "a production-format timestamp outside the TTL must not be approved"
2663        );
2664    }
2665
2666    #[test]
2667    fn stale_iso_timestamp_cannot_be_claimed_for_execution() {
2668        // The TTL is enforced at *execute* (claim) time too, not only at approve
2669        // time. An approval that ages past the 15-minute window between approve
2670        // and execute must not be claimable — the predicate is duplicated at
2671        // both sites, so both need coverage.
2672        let dir = tempdir().unwrap();
2673        let store = test_store(dir.path().join("tx.db"));
2674        let tx = store.record(queued_transaction()).unwrap();
2675        let receipt = store
2676            .approve_transaction(&tx.transaction_id)
2677            .unwrap()
2678            .expect("a fresh approval succeeds");
2679        let digest = audit_chain::approval_receipt_digest(&receipt);
2680
2681        // Age the row past the TTL window *after* the approval was issued.
2682        let conn = store.connection().unwrap();
2683        conn.execute(
2684            "UPDATE transactions \
2685             SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2686             WHERE transaction_id = ?1",
2687            params![tx.transaction_id],
2688        )
2689        .unwrap();
2690
2691        assert!(
2692            !store
2693                .claim_approved_for_execution(&tx.transaction_id, &digest)
2694                .unwrap(),
2695            "an approval aged past the TTL must not be claimable at execute time"
2696        );
2697        assert_eq!(
2698            store.get(&tx.transaction_id).unwrap().unwrap().status,
2699            JobState::Queued,
2700            "a TTL-expired claim must leave the transaction Queued, never Running"
2701        );
2702    }
2703
2704    #[test]
2705    fn cleanup_stale_queued_cancels_old_records() {
2706        let dir = tempdir().unwrap();
2707        let store = test_store(dir.path().join("tx.db"));
2708
2709        // Create two transactions: one fresh, one stale.
2710        let fresh = store.record(queued_transaction()).unwrap();
2711        let stale = store.record(queued_transaction()).unwrap();
2712
2713        // Backdate the stale one.
2714        let conn = store.connection().unwrap();
2715        conn.execute(
2716            "UPDATE transactions \
2717             SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2718             WHERE transaction_id = ?1",
2719            params![stale.transaction_id],
2720        )
2721        .unwrap();
2722
2723        let canceled = store.cleanup_stale_queued().unwrap();
2724        assert_eq!(canceled, 1, "only the stale record should be canceled");
2725
2726        // The stale record should now be Canceled.
2727        let stale_record = store.get(&stale.transaction_id).unwrap().unwrap();
2728        assert_eq!(stale_record.status, JobState::Canceled);
2729
2730        // The fresh record should still be Queued.
2731        let fresh_record = store.get(&fresh.transaction_id).unwrap().unwrap();
2732        assert_eq!(fresh_record.status, JobState::Queued);
2733    }
2734
2735    #[test]
2736    fn cleanup_stale_queued_revokes_every_unconsumed_approval() {
2737        let dir = tempdir().unwrap();
2738        let store = test_store(dir.path().join("tx.db"));
2739        let transactions: Vec<_> = (0..3)
2740            .map(|_| {
2741                let transaction = store.record(queued_transaction()).unwrap();
2742                store
2743                    .approve_transaction(&transaction.transaction_id)
2744                    .unwrap()
2745                    .expect("stale transaction is approvable");
2746                transaction
2747            })
2748            .collect();
2749
2750        let conn = store.connection().unwrap();
2751        for transaction in &transactions {
2752            conn.execute(
2753                "UPDATE transactions \
2754                 SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-20 minutes') \
2755                 WHERE transaction_id = ?1",
2756                params![transaction.transaction_id],
2757            )
2758            .unwrap();
2759        }
2760
2761        assert_eq!(store.cleanup_stale_queued().unwrap(), 3);
2762        for transaction in &transactions {
2763            assert_eq!(
2764                store
2765                    .get(&transaction.transaction_id)
2766                    .unwrap()
2767                    .unwrap()
2768                    .status,
2769                JobState::Canceled
2770            );
2771            assert!(!store
2772                .revoke_unconsumed_approval(&transaction.transaction_id)
2773                .unwrap());
2774        }
2775        assert_eq!(
2776            store
2777                .fetch_event_rows()
2778                .unwrap()
2779                .iter()
2780                .map(|event| event.kind.as_str())
2781                .collect::<Vec<_>>(),
2782            vec![
2783                "approval_granted",
2784                "approval_granted",
2785                "approval_granted",
2786                "approval_revoked",
2787                "approval_revoked",
2788                "approval_revoked"
2789            ]
2790        );
2791    }
2792
2793    // ── State-machine validation tests ──────────────────────────────────────
2794
2795    #[test]
2796    fn update_status_rejects_queued_to_succeeded() {
2797        let dir = tempdir().unwrap();
2798        let store = test_store(dir.path().join("tx.db"));
2799        let tx = store.record(queued_transaction()).unwrap();
2800
2801        let result = store.update_status(&tx.transaction_id, JobState::Succeeded);
2802        assert!(
2803            matches!(
2804                result,
2805                Err(TransactionStoreError::InvalidTransition {
2806                    from: JobState::Queued,
2807                    to: JobState::Succeeded,
2808                })
2809            ),
2810            "Queued -> Succeeded must be rejected (must go through Running first): {result:?}"
2811        );
2812    }
2813
2814    #[test]
2815    fn update_status_rejects_succeeded_to_running() {
2816        let dir = tempdir().unwrap();
2817        let store = test_store(dir.path().join("tx.db"));
2818        let tx = store.record(queued_transaction()).unwrap();
2819
2820        store
2821            .update_status(&tx.transaction_id, JobState::Running)
2822            .unwrap();
2823        store
2824            .update_status(&tx.transaction_id, JobState::Succeeded)
2825            .unwrap();
2826
2827        let result = store.update_status(&tx.transaction_id, JobState::Running);
2828        assert!(
2829            matches!(
2830                result,
2831                Err(TransactionStoreError::InvalidTransition {
2832                    from: JobState::Succeeded,
2833                    to: JobState::Running,
2834                })
2835            ),
2836            "Succeeded -> Running must be rejected (terminal state): {result:?}"
2837        );
2838    }
2839
2840    #[test]
2841    fn update_status_accepts_running_to_failed() {
2842        let dir = tempdir().unwrap();
2843        let store = test_store(dir.path().join("tx.db"));
2844        let tx = store.record(queued_transaction()).unwrap();
2845
2846        store
2847            .update_status(&tx.transaction_id, JobState::Running)
2848            .unwrap();
2849        store
2850            .update_status(&tx.transaction_id, JobState::Failed)
2851            .unwrap();
2852
2853        let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2854        assert_eq!(updated.status, JobState::Failed);
2855    }
2856
2857    #[test]
2858    fn update_status_accepts_running_to_rolled_back() {
2859        let dir = tempdir().unwrap();
2860        let store = test_store(dir.path().join("tx.db"));
2861        let tx = store.record(queued_transaction()).unwrap();
2862
2863        store
2864            .update_status(&tx.transaction_id, JobState::Running)
2865            .unwrap();
2866        store
2867            .update_status(&tx.transaction_id, JobState::RolledBack)
2868            .unwrap();
2869
2870        let updated = store.get(&tx.transaction_id).unwrap().unwrap();
2871        assert_eq!(updated.status, JobState::RolledBack);
2872    }
2873
2874    // ── list_transactions tests ───────────────────────────────────────────
2875
2876    #[test]
2877    fn list_transactions_returns_empty_for_fresh_store() {
2878        let dir = tempdir().unwrap();
2879        let store = test_store(dir.path().join("tx.db"));
2880        let results = store.list_transactions(10, None, None, None).unwrap();
2881        assert!(results.is_empty());
2882    }
2883
2884    #[test]
2885    fn list_transactions_returns_all_records_ordered_by_newest_first() {
2886        let dir = tempdir().unwrap();
2887        let store = test_store(dir.path().join("tx.db"));
2888        store.record(queued_transaction()).unwrap();
2889
2890        let mut second = queued_transaction();
2891        second.action_name = "GetDiskUsage".to_string();
2892        second.risk_level = RiskLevel::Low;
2893        store.record(second).unwrap();
2894
2895        let results = store.list_transactions(10, None, None, None).unwrap();
2896        assert_eq!(results.len(), 2);
2897        // Most recent first (GetDiskUsage was recorded second).
2898        assert_eq!(results[0].action_name, "GetDiskUsage");
2899        assert_eq!(results[1].action_name, "UpdateSystem");
2900    }
2901
2902    #[test]
2903    fn list_history_populates_created_at_and_risk_level() {
2904        let dir = tempdir().unwrap();
2905        let store = test_store(dir.path().join("tx.db"));
2906        store.record(queued_transaction()).unwrap();
2907
2908        let entries = store.list_history(10, None, None, None).unwrap();
2909        assert_eq!(entries.len(), 1);
2910        let entry = &entries[0];
2911        assert_eq!(entry.action_name, "UpdateSystem");
2912        assert_eq!(entry.risk_level, RiskLevel::High);
2913        assert_eq!(entry.status, JobState::Queued);
2914        assert!(
2915            !entry.created_at.is_empty(),
2916            "created_at must be populated from the stored row, not left blank"
2917        );
2918    }
2919
2920    #[test]
2921    fn list_history_applies_the_same_filters_as_list_transactions() {
2922        let dir = tempdir().unwrap();
2923        let store = test_store(dir.path().join("tx.db"));
2924        store.record(queued_transaction()).unwrap();
2925        let mut low = queued_transaction();
2926        low.action_name = "GetDiskUsage".to_string();
2927        low.risk_level = RiskLevel::Low;
2928        store.record(low).unwrap();
2929
2930        let only = store
2931            .list_history(10, None, Some("GetDiskUsage"), None)
2932            .unwrap();
2933        assert_eq!(only.len(), 1);
2934        assert_eq!(only[0].action_name, "GetDiskUsage");
2935    }
2936
2937    #[test]
2938    fn list_transactions_respects_limit() {
2939        let dir = tempdir().unwrap();
2940        let store = test_store(dir.path().join("tx.db"));
2941        for _ in 0..5 {
2942            store.record(queued_transaction()).unwrap();
2943        }
2944        let results = store.list_transactions(3, None, None, None).unwrap();
2945        assert_eq!(results.len(), 3);
2946    }
2947
2948    #[test]
2949    fn list_transactions_filters_by_status() {
2950        let dir = tempdir().unwrap();
2951        let store = test_store(dir.path().join("tx.db"));
2952        let tx = store.record(queued_transaction()).unwrap();
2953        store
2954            .update_status(&tx.transaction_id, JobState::Running)
2955            .unwrap();
2956        store
2957            .update_status(&tx.transaction_id, JobState::Succeeded)
2958            .unwrap();
2959
2960        // Add another that stays Queued.
2961        store.record(queued_transaction()).unwrap();
2962
2963        let succeeded = store
2964            .list_transactions(10, Some("succeeded"), None, None)
2965            .unwrap();
2966        assert_eq!(succeeded.len(), 1);
2967        assert_eq!(succeeded[0].status, JobState::Succeeded);
2968
2969        let queued = store
2970            .list_transactions(10, Some("queued"), None, None)
2971            .unwrap();
2972        assert_eq!(queued.len(), 1);
2973        assert_eq!(queued[0].status, JobState::Queued);
2974    }
2975
2976    #[test]
2977    fn list_transactions_filters_by_action_name() {
2978        let dir = tempdir().unwrap();
2979        let store = test_store(dir.path().join("tx.db"));
2980        store.record(queued_transaction()).unwrap(); // UpdateSystem
2981
2982        let mut disk = queued_transaction();
2983        disk.action_name = "GetDiskUsage".to_string();
2984        store.record(disk).unwrap();
2985
2986        let results = store
2987            .list_transactions(10, None, Some("GetDiskUsage"), None)
2988            .unwrap();
2989        assert_eq!(results.len(), 1);
2990        assert_eq!(results[0].action_name, "GetDiskUsage");
2991    }
2992
2993    #[test]
2994    fn list_transactions_filters_by_since_hours() {
2995        let dir = tempdir().unwrap();
2996        let store = test_store(dir.path().join("tx.db"));
2997
2998        // Record a transaction and backdate it to 48 hours ago.
2999        let old = store.record(queued_transaction()).unwrap();
3000        let conn = store.connection().unwrap();
3001        conn.execute(
3002            "UPDATE transactions \
3003             SET created_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-48 hours') \
3004             WHERE transaction_id = ?1",
3005            params![old.transaction_id],
3006        )
3007        .unwrap();
3008
3009        // Record a fresh transaction.
3010        store.record(queued_transaction()).unwrap();
3011
3012        // since_hours=24 should only return the fresh one.
3013        let results = store.list_transactions(10, None, None, Some(24)).unwrap();
3014        assert_eq!(results.len(), 1);
3015
3016        // since_hours=72 should return both.
3017        let results = store.list_transactions(10, None, None, Some(72)).unwrap();
3018        assert_eq!(results.len(), 2);
3019    }
3020
3021    #[test]
3022    fn list_transactions_rejects_invalid_status_filter() {
3023        let dir = tempdir().unwrap();
3024        let store = test_store(dir.path().join("tx.db"));
3025        store.record(queued_transaction()).unwrap();
3026        let result = store.list_transactions(10, Some("bogus"), None, None);
3027        assert!(result.is_err(), "invalid status filter should return error");
3028    }
3029
3030    // ── Audit watermark sink tests ────────────────────────────────────────
3031    //
3032    // Each test below installs a `WatermarkSink` via `install_test_sink`.
3033    // `cargo nextest` runs every test in its own process, so the `OnceLock`
3034    // that backs the sink is always unset at the start of each test.
3035
3036    /// W1 — `record()` emits exactly one watermark per chain entry.
3037    #[test]
3038    fn record_emits_one_watermark_per_entry() {
3039        let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3040        crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3041
3042        let dir = tempdir().unwrap();
3043        let store = test_store(dir.path().join("tx.db"));
3044        store.record(queued_transaction()).unwrap();
3045
3046        let calls = crate::audit_watermark::take_watermarks(&sink);
3047        assert_eq!(calls.len(), 1, "expected exactly 1 watermark per record()");
3048    }
3049
3050    /// W2 — `record_previewed()` emits exactly one watermark.
3051    #[test]
3052    fn record_previewed_emits_one_watermark() {
3053        let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3054        crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3055
3056        let dir = tempdir().unwrap();
3057        let store = test_store(dir.path().join("tx.db"));
3058        let preview = PreviewEnvelope {
3059            summary: "Upgrade the system".to_string(),
3060            risk_level: RiskLevel::High,
3061            current_state: serde_json::Value::Null,
3062            proposed_change: serde_json::Value::Null,
3063            expected_side_effects: vec![],
3064            reboot_required: false,
3065            rollback_available: false,
3066            warnings: vec![],
3067            request_hash: sysknife_types::RequestHash::from("hash-abc".to_string()),
3068        };
3069        store
3070            .record_previewed(queued_transaction(), preview)
3071            .unwrap();
3072
3073        let calls = crate::audit_watermark::take_watermarks(&sink);
3074        assert_eq!(
3075            calls.len(),
3076            1,
3077            "expected exactly 1 watermark per record_previewed()"
3078        );
3079    }
3080
3081    /// W3 — watermark seq and chain_hash_hex match the stored chain row.
3082    #[test]
3083    fn watermark_seq_and_hash_match_chain_row() {
3084        let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3085        crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3086
3087        let dir = tempdir().unwrap();
3088        let store = test_store(dir.path().join("tx.db"));
3089        store.record(queued_transaction()).unwrap();
3090
3091        let rows = store.fetch_chain_rows().unwrap();
3092        assert_eq!(rows.len(), 1);
3093        let row = &rows[0];
3094
3095        let calls = crate::audit_watermark::take_watermarks(&sink);
3096        assert_eq!(calls.len(), 1);
3097        assert_eq!(calls[0].seq, row.seq, "watermark seq must match chain row");
3098        assert_eq!(
3099            calls[0].chain_hash_hex, row.chain_hash,
3100            "watermark chain_hash_hex must match stored chain_hash"
3101        );
3102    }
3103
3104    /// W4 — N records produce N watermarks, one per entry, in seq order.
3105    #[test]
3106    fn multiple_records_produce_one_watermark_each() {
3107        let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3108        crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3109
3110        let dir = tempdir().unwrap();
3111        let store = test_store(dir.path().join("tx.db"));
3112        for _ in 0..3 {
3113            store.record(queued_transaction()).unwrap();
3114        }
3115
3116        let calls = crate::audit_watermark::take_watermarks(&sink);
3117        assert_eq!(calls.len(), 3, "one watermark per record call");
3118        assert_eq!(calls[0].seq, 1);
3119        assert_eq!(calls[1].seq, 2);
3120        assert_eq!(calls[2].seq, 3);
3121    }
3122
3123    /// W5 — a failed SQL INSERT (unique-constraint violation via a crafted
3124    /// duplicate seq) must NOT emit a watermark, because the row was never
3125    /// committed to the chain.
3126    ///
3127    /// We simulate this by calling `insert_transaction` directly on an already-
3128    /// committed connection with duplicate seq. In practice this cannot happen
3129    /// through the public API (BEGIN IMMEDIATE + seq allocation inside the same
3130    /// DB transaction prevents races), but the unit test validates the ordering
3131    /// invariant: the watermark is emitted AFTER `tx.commit()` succeeds, so a
3132    /// rolled-back transaction emits nothing.
3133    ///
3134    /// Strategy: install the sink, then verify that a store that has never had
3135    /// `record()` called on it emits zero watermarks.
3136    #[test]
3137    fn no_watermark_emitted_before_any_record() {
3138        let sink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
3139        crate::audit_watermark::install_test_sink(std::sync::Arc::clone(&sink));
3140
3141        let dir = tempdir().unwrap();
3142        let _store = test_store(dir.path().join("tx.db"));
3143
3144        // No record() called — sink must be empty.
3145        let calls = crate::audit_watermark::take_watermarks(&sink);
3146        assert!(
3147            calls.is_empty(),
3148            "no watermark must be emitted without a record() call"
3149        );
3150    }
3151}