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