Skip to main content

chio_store_sqlite/receipt_store/
evidence_retention.rs

1use super::support::{
2    checkpoint_error_to_receipt_store, ensure_checkpoint_transparency_guards,
3    ensure_chio_receipt_verified, load_claim_tree_canonical_bytes_range,
4    load_persisted_checkpoint_row, parse_persisted_checkpoint_row,
5    verify_checkpoint_chain_integrity,
6};
7use super::*;
8
9impl SqliteReceiptStore {
10    pub fn append_chio_receipt_returning_seq(
11        &self,
12        receipt: &ChioReceipt,
13    ) -> Result<u64, ReceiptStoreError> {
14        ensure_chio_receipt_verified(receipt)?;
15        let raw_json = serde_json::to_string(receipt)?;
16        let attribution = extract_receipt_attribution(receipt);
17        let mut connection = self.connection()?;
18        let tx = connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
19        let mut subject_key = attribution.subject_key;
20        let mut issuer_key = attribution.issuer_key;
21        if subject_key.is_none() || issuer_key.is_none() {
22            if let Some((lineage_subject_key, lineage_issuer_key)) = tx
23                .query_row(
24                    "SELECT subject_key, issuer_key FROM capability_lineage WHERE capability_id = ?1",
25                    params![receipt.capability_id.as_str()],
26                    |row| {
27                        Ok((
28                            row.get::<_, Option<String>>(0)?,
29                            row.get::<_, Option<String>>(1)?,
30                        ))
31                    },
32                )
33                .optional()?
34            {
35                if subject_key.is_none() {
36                    subject_key = lineage_subject_key;
37                }
38                if issuer_key.is_none() {
39                    issuer_key = lineage_issuer_key;
40                }
41            }
42        }
43        // Phase 1.5: tenant_id is populated directly from the signed
44        // receipt body. The evaluate path derived it from the session's
45        // enterprise_identity; we carry it through to a dedicated column
46        // so the tenant-scoped WHERE clause can filter without having
47        // to json_extract on every query.
48        let tenant_id = receipt.tenant_id.clone();
49        let inserted = tx.execute(
50            r#"
51            INSERT INTO chio_tool_receipts (
52                receipt_id,
53                timestamp,
54                capability_id,
55                subject_key,
56                issuer_key,
57                grant_index,
58                tool_server,
59                tool_name,
60                decision_kind,
61                policy_hash,
62                content_hash,
63                tenant_id,
64                raw_json
65            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
66            ON CONFLICT(receipt_id) DO NOTHING
67            "#,
68            params![
69                receipt.id,
70                sqlite_i64(receipt.timestamp, "receipt timestamp")?,
71                receipt.capability_id,
72                subject_key,
73                issuer_key,
74                attribution.grant_index.map(i64::from),
75                receipt.tool_server,
76                receipt.tool_name,
77                decision_kind(&receipt.decision),
78                receipt.policy_hash,
79                receipt.content_hash,
80                tenant_id,
81                raw_json,
82            ],
83        )?;
84        if inserted == 0 {
85            tx.commit()?;
86            return Ok(0);
87        }
88        let seq = tx.last_insert_rowid().max(0) as u64;
89        tx.commit()?;
90        Ok(seq)
91    }
92
93    /// Store a signed KernelCheckpoint in the kernel_checkpoints table.
94    pub fn store_checkpoint(&self, checkpoint: &KernelCheckpoint) -> Result<(), ReceiptStoreError> {
95        let connection = self.connection()?;
96        ensure_checkpoint_transparency_guards(&connection)?;
97
98        chio_kernel::checkpoint::validate_checkpoint(checkpoint)
99            .map_err(checkpoint_error_to_receipt_store)?;
100        if let Some(existing) =
101            load_persisted_checkpoint_row(&connection, checkpoint.body.checkpoint_seq)?
102        {
103            let existing = parse_persisted_checkpoint_row(existing)?;
104            if existing == *checkpoint {
105                return Ok(());
106            }
107            return Err(ReceiptStoreError::Conflict(format!(
108                "checkpoint {} already exists with different content",
109                checkpoint.body.checkpoint_seq
110            )));
111        }
112
113        match verify_checkpoint_chain_integrity(&connection)? {
114            Some(predecessor) => {
115                if checkpoint.body.checkpoint_seq <= predecessor.body.checkpoint_seq {
116                    return Err(ReceiptStoreError::Conflict(format!(
117                        "checkpoint {} must be appended after existing checkpoint {}",
118                        checkpoint.body.checkpoint_seq, predecessor.body.checkpoint_seq
119                    )));
120                }
121                chio_kernel::checkpoint::validate_checkpoint_predecessor(&predecessor, checkpoint)
122                    .map_err(|error| {
123                        ReceiptStoreError::Conflict(format!(
124                            "checkpoint predecessor continuity violation: {error}"
125                        ))
126                    })?;
127            }
128            None if checkpoint.body.checkpoint_seq != 1 => {
129                return Err(ReceiptStoreError::Conflict(format!(
130                    "checkpoint {} cannot initialize an empty checkpoint log",
131                    checkpoint.body.checkpoint_seq
132                )));
133            }
134            None => {}
135        }
136
137        let statement_json = serde_json::to_string(&checkpoint.body)?;
138        connection.execute(
139            r#"
140            INSERT INTO kernel_checkpoints (
141                checkpoint_seq, batch_start_seq, batch_end_seq, tree_size,
142                merkle_root, issued_at, statement_json, signature, kernel_key
143            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
144            "#,
145            params![
146                sqlite_i64(checkpoint.body.checkpoint_seq, "checkpoint_seq")?,
147                sqlite_i64(checkpoint.body.batch_start_seq, "batch_start_seq")?,
148                sqlite_i64(checkpoint.body.batch_end_seq, "batch_end_seq")?,
149                sqlite_i64(checkpoint.body.tree_size as u64, "tree_size")?,
150                checkpoint.body.merkle_root.to_hex(),
151                sqlite_i64(checkpoint.body.issued_at, "issued_at")?,
152                statement_json,
153                checkpoint.signature.to_hex(),
154                checkpoint.body.kernel_key.to_hex(),
155            ],
156        )?;
157
158        let stored = load_persisted_checkpoint_row(&connection, checkpoint.body.checkpoint_seq)?
159            .ok_or_else(|| {
160                ReceiptStoreError::Conflict(format!(
161                    "checkpoint {} was not visible after persistence",
162                    checkpoint.body.checkpoint_seq
163                ))
164            })?;
165        let stored = parse_persisted_checkpoint_row(stored)?;
166        if stored != *checkpoint {
167            return Err(ReceiptStoreError::Conflict(format!(
168                "checkpoint {} persisted with conflicting contents",
169                checkpoint.body.checkpoint_seq
170            )));
171        }
172        Ok(())
173    }
174
175    /// Load a KernelCheckpoint by its checkpoint_seq.
176    pub fn load_checkpoint_by_seq(
177        &self,
178        checkpoint_seq: u64,
179    ) -> Result<Option<KernelCheckpoint>, ReceiptStoreError> {
180        let connection = self.connection()?;
181        ensure_checkpoint_transparency_guards(&connection)?;
182        load_persisted_checkpoint_row(&connection, checkpoint_seq)?
183            .map(parse_persisted_checkpoint_row)
184            .transpose()
185    }
186
187    /// Return canonical JSON bytes for receipts with seq in [start_seq, end_seq], ordered by seq.
188    ///
189    /// Uses RFC 8785 canonical JSON for deterministic Merkle leaf hashing.
190    pub fn receipts_canonical_bytes_range(
191        &self,
192        start_seq: u64,
193        end_seq: u64,
194    ) -> Result<Vec<(u64, Vec<u8>)>, ReceiptStoreError> {
195        let connection = self.connection()?;
196        load_claim_tree_canonical_bytes_range(&connection, start_seq, end_seq)
197    }
198
199    /// Return the current on-disk size of the database in bytes.
200    ///
201    /// Uses `PRAGMA page_count` and `PRAGMA page_size` to compute the size
202    /// without requiring a filesystem stat, which is consistent in WAL mode.
203    pub fn db_size_bytes(&self) -> Result<u64, ReceiptStoreError> {
204        let page_count: i64 = self
205            .connection()?
206            .query_row("PRAGMA page_count", [], |row| row.get(0))?;
207        let page_size: i64 = self
208            .connection()?
209            .query_row("PRAGMA page_size", [], |row| row.get(0))?;
210        Ok((page_count.max(0) as u64) * (page_size.max(0) as u64))
211    }
212
213    /// Return the Unix timestamp (seconds) of the oldest receipt in the live
214    /// database, or `None` if there are no receipts.
215    pub fn oldest_receipt_timestamp(&self) -> Result<Option<u64>, ReceiptStoreError> {
216        let ts = self.connection()?.query_row(
217            "SELECT MIN(timestamp) FROM chio_tool_receipts",
218            [],
219            |row| row.get::<_, Option<i64>>(0),
220        )?;
221        Ok(ts.map(|t| t.max(0) as u64))
222    }
223
224    /// Return the oldest live receipt timestamp for a tenant.
225    pub fn oldest_receipt_timestamp_for_tenant(
226        &self,
227        tenant_id: &str,
228    ) -> Result<Option<u64>, ReceiptStoreError> {
229        let ts = self.connection()?.query_row(
230            "SELECT MIN(timestamp) FROM chio_tool_receipts WHERE tenant_id = ?1",
231            params![tenant_id],
232            |row| row.get::<_, Option<i64>>(0),
233        )?;
234        Ok(ts.map(|t| t.max(0) as u64))
235    }
236
237    /// Archive all receipts with `timestamp < cutoff_unix_secs` to an external
238    /// SQLite file, then delete them from the live database.
239    ///
240    /// Checkpoint rows whose entire batch (`batch_end_seq`) falls within the
241    /// archived receipt range are also copied to the archive. Partial batches
242    /// are never archived to avoid breaking inclusion proofs.
243    ///
244    /// Returns the number of receipt rows deleted from the live database.
245    pub fn archive_receipts_before(
246        &mut self,
247        cutoff_unix_secs: u64,
248        archive_path: &str,
249    ) -> Result<u64, ReceiptStoreError> {
250        self.archive_receipts_before_scoped(cutoff_unix_secs, archive_path, None)
251    }
252
253    /// Archive receipts for a single tenant without deleting other tenants'
254    /// evidence that may have a longer retention window.
255    pub fn archive_receipts_before_for_tenant(
256        &mut self,
257        cutoff_unix_secs: u64,
258        archive_path: &str,
259        tenant_id: &str,
260    ) -> Result<u64, ReceiptStoreError> {
261        self.archive_receipts_before_scoped(cutoff_unix_secs, archive_path, Some(tenant_id))
262    }
263
264    fn archive_receipts_before_scoped(
265        &mut self,
266        cutoff_unix_secs: u64,
267        archive_path: &str,
268        tenant_id: Option<&str>,
269    ) -> Result<u64, ReceiptStoreError> {
270        // Escape single quotes in the path to safely embed it in an ATTACH statement.
271        let escaped_path = archive_path.replace('\'', "''");
272
273        // Attach the archive database.
274        self.connection()?
275            .execute_batch(&format!("ATTACH DATABASE '{escaped_path}' AS archive"))?;
276
277        // Create archive tables with the same schema as the main database.
278        self.connection()?.execute_batch(
279            r#"
280            CREATE TABLE IF NOT EXISTS archive.chio_tool_receipts (
281                seq INTEGER PRIMARY KEY AUTOINCREMENT,
282                receipt_id TEXT NOT NULL UNIQUE,
283                timestamp INTEGER NOT NULL,
284                capability_id TEXT NOT NULL,
285                subject_key TEXT,
286                issuer_key TEXT,
287                grant_index INTEGER,
288                tool_server TEXT NOT NULL,
289                tool_name TEXT NOT NULL,
290                decision_kind TEXT NOT NULL,
291                policy_hash TEXT NOT NULL,
292                content_hash TEXT NOT NULL,
293                raw_json TEXT NOT NULL,
294                tenant_id TEXT
295            );
296
297            CREATE TABLE IF NOT EXISTS archive.chio_child_receipts (
298                seq INTEGER PRIMARY KEY AUTOINCREMENT,
299                receipt_id TEXT NOT NULL UNIQUE,
300                timestamp INTEGER NOT NULL,
301                session_id TEXT NOT NULL,
302                parent_request_id TEXT NOT NULL,
303                request_id TEXT NOT NULL,
304                operation_kind TEXT NOT NULL,
305                terminal_state TEXT NOT NULL,
306                policy_hash TEXT NOT NULL,
307                outcome_hash TEXT NOT NULL,
308                raw_json TEXT NOT NULL
309            );
310
311            CREATE TABLE IF NOT EXISTS archive.kernel_checkpoints (
312                id INTEGER PRIMARY KEY AUTOINCREMENT,
313                checkpoint_seq INTEGER NOT NULL UNIQUE,
314                batch_start_seq INTEGER NOT NULL,
315                batch_end_seq INTEGER NOT NULL,
316                tree_size INTEGER NOT NULL,
317                merkle_root TEXT NOT NULL,
318                issued_at INTEGER NOT NULL,
319                statement_json TEXT NOT NULL,
320                signature TEXT NOT NULL,
321                kernel_key TEXT NOT NULL
322            );
323
324            CREATE TABLE IF NOT EXISTS archive.capability_lineage (
325                capability_id        TEXT PRIMARY KEY,
326                subject_key          TEXT NOT NULL,
327                issuer_key           TEXT NOT NULL,
328                issued_at            INTEGER NOT NULL,
329                expires_at           INTEGER NOT NULL,
330                grants_json          TEXT NOT NULL,
331                delegation_depth     INTEGER NOT NULL DEFAULT 0,
332                parent_capability_id TEXT
333            );
334            "#,
335        )?;
336
337        let cutoff = cutoff_unix_secs as i64;
338
339        // Copy qualifying receipts to the archive (ignore duplicates from prior runs).
340        match tenant_id {
341            Some(tenant_id) => {
342                self.connection()?.execute(
343                    "INSERT OR IGNORE INTO archive.chio_tool_receipts \
344                     SELECT * FROM main.chio_tool_receipts WHERE timestamp < ?1 AND tenant_id = ?2",
345                    params![cutoff, tenant_id],
346                )?;
347                self.connection()?.execute(
348                    "INSERT OR IGNORE INTO archive.chio_child_receipts \
349                     SELECT child.* \
350                     FROM main.chio_child_receipts child \
351                     WHERE child.timestamp < ?1 \
352                       AND EXISTS ( \
353                           SELECT 1 \
354                           FROM main.receipt_lineage_statements lineage \
355                           INNER JOIN main.chio_tool_receipts parent \
356                               ON parent.receipt_id = lineage.parent_receipt_id \
357                           WHERE lineage.receipt_id = child.receipt_id \
358                             AND parent.tenant_id = ?2 \
359                       )",
360                    params![cutoff, tenant_id],
361                )?;
362                self.connection()?.execute(
363                    "INSERT OR IGNORE INTO archive.capability_lineage
364                     SELECT DISTINCT cl.*
365                     FROM main.capability_lineage cl
366                     INNER JOIN main.chio_tool_receipts r ON r.capability_id = cl.capability_id
367                     WHERE r.timestamp < ?1 AND r.tenant_id = ?2",
368                    params![cutoff, tenant_id],
369                )?;
370            }
371            None => {
372                self.connection()?.execute(
373                    "INSERT OR IGNORE INTO archive.chio_tool_receipts \
374                     SELECT * FROM main.chio_tool_receipts WHERE timestamp < ?1",
375                    params![cutoff],
376                )?;
377                self.connection()?.execute(
378                    "INSERT OR IGNORE INTO archive.chio_child_receipts \
379                     SELECT * FROM main.chio_child_receipts WHERE timestamp < ?1",
380                    params![cutoff],
381                )?;
382                self.connection()?.execute(
383                    "INSERT OR IGNORE INTO archive.capability_lineage
384                     SELECT DISTINCT cl.*
385                     FROM main.capability_lineage cl
386                     INNER JOIN main.chio_tool_receipts r ON r.capability_id = cl.capability_id
387                     WHERE r.timestamp < ?1",
388                    params![cutoff],
389                )?;
390            }
391        }
392
393        // Find the maximum seq among archived receipts (for checkpoint filtering).
394        let max_archived_seq: Option<i64> = match tenant_id {
395            Some(tenant_id) => self.connection()?.query_row(
396                "SELECT MAX(seq) FROM main.chio_tool_receipts WHERE timestamp < ?1 AND tenant_id = ?2",
397                params![cutoff, tenant_id],
398                |row| row.get(0),
399            )?,
400            None => self.connection()?.query_row(
401                "SELECT MAX(seq) FROM main.chio_tool_receipts WHERE timestamp < ?1",
402                params![cutoff],
403                |row| row.get(0),
404            )?,
405        };
406
407        if let Some(max_seq) = max_archived_seq {
408            // Copy checkpoint rows whose full batch is covered by the archived receipts.
409            // Never archive a checkpoint whose batch_end_seq exceeds the max archived seq
410            // because that would leave a partial batch in the archive.
411            self.connection()?.execute(
412                "INSERT OR IGNORE INTO archive.kernel_checkpoints \
413                 SELECT * FROM main.kernel_checkpoints WHERE batch_end_seq <= ?1",
414                params![max_seq],
415            )?;
416
417            // Verify that every checkpoint covering the archived range is now present
418            // in the archive. If any checkpoint failed to transfer, refuse to delete the
419            // receipts from the live database to preserve inclusion-proof integrity.
420            let live_count: i64 = self.connection()?.query_row(
421                "SELECT COUNT(*) FROM main.kernel_checkpoints WHERE batch_end_seq <= ?1",
422                params![max_seq],
423                |row| row.get(0),
424            )?;
425            let archive_count: i64 = self.connection()?.query_row(
426                "SELECT COUNT(*) FROM archive.kernel_checkpoints WHERE batch_end_seq <= ?1",
427                params![max_seq],
428                |row| row.get(0),
429            )?;
430            if archive_count < live_count {
431                // Detach the archive before returning the error to avoid leaving
432                // the database in an attached state.
433                let _ = self.connection()?.execute_batch("DETACH DATABASE archive");
434                return Err(ReceiptStoreError::Canonical(format!(
435                    "checkpoint co-archival incomplete: {live_count} checkpoints in live, \
436                     only {archive_count} transferred to archive; aborting receipt deletion \
437                     to preserve inclusion-proof integrity"
438                )));
439            }
440        }
441
442        // Delete archived receipts from the live database.
443        let deleted = match tenant_id {
444            Some(tenant_id) => {
445                // Delete linked child receipts before deleting their tenant parent receipts,
446                // because the tenant association is derived through receipt_lineage_statements.
447                self.connection()?.execute(
448                    "DELETE FROM main.chio_child_receipts \
449                     WHERE rowid IN ( \
450                         SELECT child.rowid \
451                         FROM main.chio_child_receipts child \
452                         WHERE child.timestamp < ?1 \
453                           AND EXISTS ( \
454                               SELECT 1 \
455                               FROM main.receipt_lineage_statements lineage \
456                               INNER JOIN main.chio_tool_receipts parent \
457                                   ON parent.receipt_id = lineage.parent_receipt_id \
458                               WHERE lineage.receipt_id = child.receipt_id \
459                                 AND parent.tenant_id = ?2 \
460                           ) \
461                    )",
462                    params![cutoff, tenant_id],
463                )?;
464                self.connection()?.execute(
465                    "DELETE FROM main.chio_tool_receipts WHERE timestamp < ?1 AND tenant_id = ?2",
466                    params![cutoff, tenant_id],
467                )? as u64
468            }
469            None => {
470                let deleted = self.connection()?.execute(
471                    "DELETE FROM main.chio_tool_receipts WHERE timestamp < ?1",
472                    params![cutoff],
473                )? as u64;
474                self.connection()?.execute(
475                    "DELETE FROM main.chio_child_receipts WHERE timestamp < ?1",
476                    params![cutoff],
477                )?;
478                deleted
479            }
480        };
481
482        // Detach the archive and checkpoint WAL.
483        self.connection()?
484            .execute_batch("DETACH DATABASE archive")?;
485        self.connection()?
486            .execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")?;
487
488        Ok(deleted)
489    }
490
491    /// Check time and size thresholds and archive receipts if either is exceeded.
492    ///
493    /// - Time threshold: receipts older than `config.retention_days` days are archived.
494    /// - Size threshold: if `db_size_bytes()` exceeds `config.max_size_bytes`, receipts
495    ///   older than the median timestamp are archived (removes roughly half the receipts).
496    ///
497    /// Returns the number of receipt rows archived (0 if no threshold was exceeded).
498    pub fn rotate_if_needed(&mut self, config: &RetentionConfig) -> Result<u64, ReceiptStoreError> {
499        // Check time threshold.
500        let now = SystemTime::now()
501            .duration_since(UNIX_EPOCH)
502            .map(|d| d.as_secs())
503            .unwrap_or(0);
504        let time_cutoff = now.saturating_sub(config.retention_days.saturating_mul(86_400));
505        let tenant_id = config.tenant_id.as_deref();
506        let oldest = match tenant_id {
507            Some(tenant_id) => self.oldest_receipt_timestamp_for_tenant(tenant_id)?,
508            None => self.oldest_receipt_timestamp()?,
509        };
510
511        if let Some(oldest_ts) = oldest {
512            if oldest_ts < time_cutoff {
513                return match tenant_id {
514                    Some(tenant_id) => self.archive_receipts_before_for_tenant(
515                        time_cutoff,
516                        &config.archive_path,
517                        tenant_id,
518                    ),
519                    None => self.archive_receipts_before(time_cutoff, &config.archive_path),
520                };
521            }
522        }
523
524        // Check size threshold.
525        let size = self.db_size_bytes()?;
526        if size > config.max_size_bytes {
527            // Use the median timestamp as the cutoff to archive roughly half the receipts.
528            let median_cutoff: Option<i64> = match tenant_id {
529                Some(tenant_id) => self
530                    .connection()?
531                    .query_row(
532                        r#"
533                        SELECT timestamp FROM chio_tool_receipts
534                        WHERE tenant_id = ?1
535                        ORDER BY timestamp
536                        LIMIT 1
537                        OFFSET (SELECT COUNT(*) FROM chio_tool_receipts WHERE tenant_id = ?1) / 2
538                        "#,
539                        params![tenant_id],
540                        |row| row.get(0),
541                    )
542                    .optional()?,
543                None => self
544                    .connection()?
545                    .query_row(
546                        r#"
547                        SELECT timestamp FROM chio_tool_receipts
548                        ORDER BY timestamp
549                        LIMIT 1
550                        OFFSET (SELECT COUNT(*) FROM chio_tool_receipts) / 2
551                        "#,
552                        [],
553                        |row| row.get(0),
554                    )
555                    .optional()?,
556            };
557
558            if let Some(cutoff) = median_cutoff {
559                return match tenant_id {
560                    Some(tenant_id) => self.archive_receipts_before_for_tenant(
561                        cutoff.max(0) as u64,
562                        &config.archive_path,
563                        tenant_id,
564                    ),
565                    None => {
566                        self.archive_receipts_before(cutoff.max(0) as u64, &config.archive_path)
567                    }
568                };
569            }
570        }
571
572        Ok(0)
573    }
574
575    /// Internal implementation for `query_receipts` (called from `receipt_query` module).
576    ///
577    /// Requires access to the private `connection` field, so it lives here in `receipt_store`.
578    pub(crate) fn query_receipts_impl(
579        &self,
580        query: &ReceiptQuery,
581    ) -> Result<ReceiptQueryResult, ReceiptStoreError> {
582        // Validate the `outcome` filter against the known decision_kind values.
583        // Silently accepting unknown values would return zero results and could
584        // mask caller bugs; fail explicitly instead.
585        const VALID_OUTCOMES: &[&str] = &["allow", "deny", "cancelled", "incomplete"];
586        if let Some(outcome) = query.outcome.as_deref() {
587            if !VALID_OUTCOMES.contains(&outcome) {
588                return Err(ReceiptStoreError::InvalidOutcome(format!(
589                    "unknown outcome filter {:?}; valid values are: allow, deny, cancelled, incomplete",
590                    outcome
591                )));
592            }
593        }
594
595        let limit = query.limit.clamp(1, MAX_QUERY_LIMIT);
596
597        // Phase 1.5 multi-tenant receipt isolation: compute the tenant
598        // WHERE fragment. Three modes:
599        //
600        //   * `tenant_filter = None`           -> "1=1" (admin/compat).
601        //   * `tenant_filter = Some(id)` w/ strict_tenant_isolation=true
602        //     -> `tenant_id = ?X` (legacy rows hidden).
603        //   * `tenant_filter = Some(id)` w/ strict_tenant_isolation=false
604        //     -> `tenant_id = ?X OR tenant_id IS NULL` so legacy
605        //     pre-1.5 receipts stay visible during explicit
606        //     compatibility mode.
607        //
608        // Bound parameter ?12 carries the tenant string when present.
609        // When `tenant_filter = None`, `?12 IS NULL` makes the fragment
610        // a tautology and no rows are removed.
611        let tenant_fragment = match (
612            query.tenant_filter.as_deref(),
613            self.strict_tenant_isolation_enabled(),
614        ) {
615            (None, _) => "(?12 IS NULL)",
616            (Some(_), true) => "(r.tenant_id = ?12)",
617            (Some(_), false) => "(r.tenant_id = ?12 OR r.tenant_id IS NULL)",
618        };
619
620        // Both queries share the same filter parameters.
621        // Parameters:
622        //   ?1  capability_id
623        //   ?2  tool_server
624        //   ?3  tool_name
625        //   ?4  outcome (decision_kind)
626        //   ?5  since (timestamp >=, inclusive)
627        //   ?6  until (timestamp <=, inclusive)
628        //   ?7  min_cost (json_extract cost_charged >=)
629        //   ?8  max_cost (json_extract cost_charged <=)
630        //   ?9  agent_subject (receipt subject_key, falling back to capability_lineage)
631        //   ?12 tenant_filter (tenant_id exact match or NULL fallback)
632        // Data query also uses:
633        //   ?10 cursor (seq >, exclusive)
634        //   ?11 limit
635        //
636        // When agent_subject is None, the LEFT JOIN produces NULL for cl.subject_key,
637        // and the (?9 IS NULL OR ...) guard passes -- no rows are filtered out.
638        let data_sql = format!(
639            r#"
640            SELECT r.seq, r.raw_json
641            FROM chio_tool_receipts r
642            LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
643            WHERE (?1 IS NULL OR r.capability_id = ?1)
644              AND (?2 IS NULL OR r.tool_server = ?2)
645              AND (?3 IS NULL OR r.tool_name = ?3)
646              AND (?4 IS NULL OR r.decision_kind = ?4)
647              AND (?5 IS NULL OR r.timestamp >= ?5)
648              AND (?6 IS NULL OR r.timestamp <= ?6)
649              AND (?7 IS NULL OR CAST(json_extract(r.raw_json, '$.metadata.financial.cost_charged') AS INTEGER) >= ?7)
650              AND (?8 IS NULL OR CAST(json_extract(r.raw_json, '$.metadata.financial.cost_charged') AS INTEGER) <= ?8)
651              AND (?9 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?9)
652              AND {tenant_fragment}
653              AND (?10 IS NULL OR r.seq > ?10)
654            ORDER BY r.seq ASC
655            LIMIT ?11
656        "#
657        );
658
659        // Count query uses identical WHERE clause but no cursor and no LIMIT.
660        // total_count reflects the full filtered set regardless of pagination.
661        let count_sql = format!(
662            r#"
663            SELECT COUNT(*)
664            FROM chio_tool_receipts r
665            LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
666            WHERE (?1 IS NULL OR r.capability_id = ?1)
667              AND (?2 IS NULL OR r.tool_server = ?2)
668              AND (?3 IS NULL OR r.tool_name = ?3)
669              AND (?4 IS NULL OR r.decision_kind = ?4)
670              AND (?5 IS NULL OR r.timestamp >= ?5)
671              AND (?6 IS NULL OR r.timestamp <= ?6)
672              AND (?7 IS NULL OR CAST(json_extract(r.raw_json, '$.metadata.financial.cost_charged') AS INTEGER) >= ?7)
673              AND (?8 IS NULL OR CAST(json_extract(r.raw_json, '$.metadata.financial.cost_charged') AS INTEGER) <= ?8)
674              AND (?9 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?9)
675              AND {tenant_fragment}
676        "#
677        );
678
679        let cap_id = query.capability_id.as_deref();
680        let tool_srv = query.tool_server.as_deref();
681        let tool_nm = query.tool_name.as_deref();
682        let outcome = query.outcome.as_deref();
683        let since = query.since.map(|v| v as i64);
684        let until = query.until.map(|v| v as i64);
685        let min_cost = query.min_cost.map(|v| v as i64);
686        let max_cost = query.max_cost.map(|v| v as i64);
687        let agent_sub = query.agent_subject.as_deref();
688        let tenant = query.tenant_filter.as_deref();
689        // Convert cursor to signed i64 for SQLite. SQLite AUTOINCREMENT seq
690        // values are bounded by i64::MAX; a cursor above that can never be
691        // exceeded. Convert with a checked cast: on overflow return an empty
692        // receipts page (the cursor excludes everything) while still reporting
693        // the correct total_count for the uncursored filter set.
694        let cursor_i64: Option<i64> = match query.cursor {
695            None => None,
696            Some(c) => match i64::try_from(c) {
697                Ok(v) => Some(v),
698                Err(_) => {
699                    // cursor > i64::MAX: no AUTOINCREMENT seq can exceed it.
700                    // Run only the count query (no cursor applied) and return empty.
701                    // ?10 and ?11 (cursor/limit) are not used in the count query
702                    // but must still bind placeholders if we reuse `params!`;
703                    // the count SQL uses only ?1..=?9 and ?12, so we need to
704                    // bind ?10 and ?11 as NULL / 0 to keep indexes stable.
705                    let total_count: u64 = self
706                        .connection()?
707                        .query_row(
708                            &count_sql,
709                            params![
710                                cap_id,
711                                tool_srv,
712                                tool_nm,
713                                outcome,
714                                since,
715                                until,
716                                min_cost,
717                                max_cost,
718                                agent_sub,
719                                // ?10, ?11 unused in count_sql but bound so ?12
720                                // resolves to the tenant filter.
721                                None::<i64>,
722                                0i64,
723                                tenant,
724                            ],
725                            |row| row.get::<_, i64>(0),
726                        )
727                        .map(|n| n.max(0) as u64)?;
728                    return Ok(ReceiptQueryResult {
729                        receipts: Vec::new(),
730                        total_count,
731                        next_cursor: None,
732                    });
733                }
734            },
735        };
736
737        // Execute data query.
738        let connection = self.connection()?;
739        let mut stmt = connection.prepare(&data_sql)?;
740        let rows = stmt.query_map(
741            params![
742                cap_id,
743                tool_srv,
744                tool_nm,
745                outcome,
746                since,
747                until,
748                min_cost,
749                max_cost,
750                agent_sub,
751                cursor_i64,
752                limit as i64,
753                tenant,
754            ],
755            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
756        )?;
757
758        let mut receipts = Vec::new();
759        for row in rows {
760            let (seq, raw_json) = row?;
761            let seq = seq.max(0) as u64;
762            let receipt =
763                decode_verified_chio_receipt(&raw_json, "persisted tool receipt", Some(seq))?;
764            receipts.push(StoredToolReceipt { seq, receipt });
765        }
766
767        // Execute count query (same filters, no cursor, no limit).
768        let total_count: u64 = self
769            .connection()?
770            .query_row(
771                &count_sql,
772                params![
773                    cap_id,
774                    tool_srv,
775                    tool_nm,
776                    outcome,
777                    since,
778                    until,
779                    min_cost,
780                    max_cost,
781                    agent_sub,
782                    // ?10, ?11 unused in count_sql; bound to keep ?12 stable.
783                    None::<i64>,
784                    0i64,
785                    tenant,
786                ],
787                |row| row.get::<_, i64>(0),
788            )
789            .map(|n| n.max(0) as u64)?;
790
791        // next_cursor is Some(last_seq) when the page is full (more results may exist).
792        let next_cursor = if receipts.len() == limit {
793            receipts.last().map(|r| r.seq)
794        } else {
795            None
796        };
797
798        Ok(ReceiptQueryResult {
799            receipts,
800            total_count,
801            next_cursor,
802        })
803    }
804}