Skip to main content

sysknife_daemon/
transactions.rs

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