fix: repair chained imports and track blocked evidence dependencies

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1046e-465c-7d43-8c4c-d10f0a1feadd

audit_core/cli.py

@@ -101,6 +101,10 @@ def build_parser() -> argparse.ArgumentParser:
         help="DSN; defaults to AUDIT_CORE_DATABASE_URL",
     )
     migrate_parser.add_argument("--schema", default="audit_core")
+    migrate_parser.add_argument(
+        "--source-id", default=None,
+        help="Stable import identity; retain it if the SQLite file moves (defaults to its absolute path).",
+    )
     migrate_parser.add_argument(
         "--dry-run", action="store_true",
         help="Report what would move without writing anything.",
@@ -233,7 +237,7 @@ def migrate_store(args: argparse.Namespace) -> int:
 
     backend = PostgresAuditBackend(dsn, schema=args.schema)
     try:
-        report = migrate(args.source, backend)
+        report = migrate(args.source, backend, source_id=args.source_id)
     finally:
         backend.close()
     print(report.summary())

audit_core/ingestion.py

@@ -340,7 +340,7 @@ class IngestionApplication:
         obligation; audit-core supplies the numerator and says nothing about
         what the answer means.
         """
-        query = parse_qs(environ.get("QUERY_STRING", ""))
+        query = parse_qs(environ.get("QUERY_STRING", ""), keep_blank_values=True)
         source = (query.get("source") or [""])[0]
         if not source:
             return self._json(
@@ -371,6 +371,13 @@ class IngestionApplication:
             return self._json(
                 start_response, HTTPStatus.BAD_REQUEST, {"error": "tenant_required"}
             )
+        instances = query.get("stream_instance_id")
+        if instances is not None and (len(instances) != 1 or not instances[0].strip()):
+            return self._json(
+                start_response, HTTPStatus.BAD_REQUEST,
+                {"error": "invalid_stream_instance_id"},
+            )
+        stream_instance_id = instances[0] if instances else None
         counter = getattr(self.backend, "event_counts", None)
         if not callable(counter):
             return self._json(
@@ -378,7 +385,10 @@ class IngestionApplication:
                 {"error": "reconciliation_not_supported"},
             )
         try:
-            counts = counter(source, since, until, tenant)
+            if stream_instance_id is None:
+                counts = counter(source, since, until, tenant)
+            else:
+                counts = counter(source, since, until, tenant, stream_instance_id=stream_instance_id)
         except BackendUnavailableError as exc:
             log.error("reconciliation failed: %s", exc)
             return self._json(
@@ -391,6 +401,7 @@ class IngestionApplication:
             "since": since,
             "until": until,
             "counts": counts,
+            **({"stream_instance_id": stream_instance_id} if instances else {}),
             # Said in the response because this is the number most likely to
             # be quoted out of context in someone else's conformance argument.
             "means": (

audit_core/integrity.py

@@ -64,6 +64,29 @@ class ChainReport:
         }
 
 
+def order_chain_rows(rows: Iterable[ChainRow]) -> list[ChainRow]:
+    """Follow stored links, independently of historical acceptance timestamps.
+
+    Keep unreachable/forked rows in the result so verify_rows reports a break
+    rather than silently dropping evidence outside the genesis chain.
+    """
+    remaining = list(rows)
+    successors: dict[str, list[ChainRow]] = {}
+    for row in remaining:
+        successors.setdefault(row.chain_prev, []).append(row)
+    ordered: list[ChainRow] = []
+    used: set[str] = set()
+    previous = GENESIS
+    while len(successors.get(previous, [])) == 1:
+        row = successors[previous][0]
+        if row.event_id in used:
+            break
+        ordered.append(row)
+        used.add(row.event_id)
+        previous = row.chain_hash
+    return ordered + [row for row in remaining if row.event_id not in used]
+
+
 def verify_rows(
     rows: Iterable[ChainRow],
     *,

audit_core/migrate_store.py

@@ -18,9 +18,13 @@ current time.
 
 from __future__ import annotations
 
+import hashlib
 import json
 import sqlite3
 from dataclasses import dataclass, field
+from pathlib import Path
+
+from audit_core.integrity import CHAIN_LOCK_KEY, chain_link
 
 from audit_core.interface import BackendUnavailableError
 
@@ -56,110 +60,172 @@ class MigrationReport:
         return "\n".join(lines)
 
 
-def migrate(sqlite_path: str, backend, *, verify: bool = True) -> MigrationReport:
+def migrate(
+    sqlite_path: str, backend, *, verify: bool = True, source_id: str | None = None
+) -> MigrationReport:
     """Copy every record from ``sqlite_path`` into ``backend``.
 
-    Idempotent: re-running skips events already present with a matching payload
-    hash. An event present with a *different* hash is reported as a conflict
+    Idempotent: re-running skips events with matching hash, time and record.
+    An event present with different historical evidence is reported as a conflict
     and never overwritten — the destination record wins, and a human decides.
     """
+    import psycopg
+
     report = MigrationReport()
+    # Callers moving a source to a new path must retain its explicit source_id.
+    source_id = source_id or str(Path(sqlite_path).resolve())
+    source_id = hashlib.sha256(source_id.encode()).hexdigest()
     source = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True)
     try:
+        source.execute("BEGIN")  # One consistent source snapshot, including counters.
         rows = source.execute(
             "SELECT event_id, payload_hash, accepted_at, correlation_id, tenant, record "
             "FROM events ORDER BY accepted_at, event_id"
         ).fetchall()
         report.events_read = len(rows)
 
-        for event_id, payload_hash, accepted_at, correlation_id, tenant, record in rows:
-            outcome = _import_event(
-                backend, event_id, payload_hash, accepted_at, correlation_id,
-                tenant, json.loads(record),
-            )
-            if outcome == "written":
-                report.events_written += 1
-            elif outcome == "present":
-                report.events_already_present += 1
-            else:
-                report.conflicts.append(outcome)
+        # Same serialization boundary as native accept(), including side-table
+        # receipts. A retry cannot double-count after an interrupted import.
+        with backend.pool.connection() as conn, conn.transaction():
+            conn.execute("SELECT pg_advisory_xact_lock(%s)", (CHAIN_LOCK_KEY,))
+            for event_id, payload_hash, accepted_at, correlation_id, tenant, record in rows:
+                outcome = _import_event(
+                    backend, conn, event_id, payload_hash, accepted_at, correlation_id,
+                    tenant, json.loads(record),
+                )
+                if outcome == "written":
+                    report.events_written += 1
+                elif outcome == "present":
+                    report.events_already_present += 1
+                else:
+                    report.conflicts.append(outcome)
 
-        report.dead_letters_written = _copy_dead_letters(source, backend)
-        report.findings_written = _copy_findings(source, backend)
+            report.dead_letters_written = _copy_dead_letters(source, backend, conn, source_id, report)
+            report.findings_written = _copy_findings(source, backend, conn, source_id, report)
 
         if verify:
             report.verified = _verify(source, backend, report)
+            if not backend.verify_chain().intact:
+                report.conflicts.append("destination hash chain is not intact")
+                report.verified = False
+    except psycopg.Error as exc:
+        raise BackendUnavailableError("SQLite import transaction failed") from exc
     finally:
         source.close()
     return report
 
 
 def _import_event(
-    backend, event_id, payload_hash, accepted_at, correlation_id, tenant, record
+    backend, conn, event_id, payload_hash, accepted_at, correlation_id, tenant, record
 ) -> str:
-    existing = backend._query(
-        f"SELECT payload_hash FROM {backend._events} WHERE event_id = %s", (event_id,)
-    )
+    existing = conn.execute(
+        f"SELECT payload_hash, accepted_at, record FROM {backend._events} WHERE event_id = %s",
+        (event_id,),
+    ).fetchone()
     if existing:
-        if existing[0][0] != payload_hash:
-            return (
-                f"{event_id}: destination holds a different payload "
-                f"(source {payload_hash[:12]}…, destination {existing[0][0][:12]}…) "
-                "— left untouched"
-            )
+        if existing[0] != payload_hash:
+            return f"{event_id}: destination holds a different payload — left untouched"
+        if not _same_instant(existing[1], accepted_at) or existing[2] != record:
+            return f"{event_id}: destination holds different historical evidence — left untouched"
         return "present"
 
-    backend._execute(
+    previous = backend._chain_head(conn)
+    conn.execute(
         f"INSERT INTO {backend._events} "
         "(event_id, payload_hash, accepted_at, observed_at, tenant, correlation_id, "
-        " source, action, record) "
-        "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) "
-        "ON CONFLICT (event_id) DO NOTHING",
+        " source, action, record, chain_prev, chain_hash) "
+        "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
         (
             event_id, payload_hash, accepted_at, record.get("observed_at"),
             tenant, correlation_id, record.get("source", ""), record.get("action", ""),
-            json.dumps(record, sort_keys=True),
+            json.dumps(record, sort_keys=True), previous,
+            chain_link(previous, payload_hash, event_id),
         ),
     )
     return "written"
 
 
-def _copy_dead_letters(source: sqlite3.Connection, backend) -> int:
-    try:
-        rows = source.execute(
-            "SELECT event_id, received_at, sender, reason, payload_hash, payload, "
-            "payload_withheld FROM dead_letters ORDER BY id"
-        ).fetchall()
-    except sqlite3.Error:
+def _fingerprint(value) -> str:
+    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
+
+
+def _receipt(backend, conn, source_id, kind, record_id):
+    return conn.execute(
+        f'SELECT content_hash, occurrences FROM "{backend.schema}".import_receipts '
+        "WHERE source_id = %s AND kind = %s AND record_id = %s",
+        (source_id, kind, record_id),
+    ).fetchone()
+
+
+def _save_receipt(backend, conn, source_id, kind, record_id, content_hash, occurrences=0):
+    conn.execute(
+        f'INSERT INTO "{backend.schema}".import_receipts '
+        "(source_id, kind, record_id, content_hash, occurrences) VALUES (%s,%s,%s,%s,%s) "
+        "ON CONFLICT (source_id, kind, record_id) DO UPDATE "
+        "SET content_hash = excluded.content_hash, occurrences = excluded.occurrences",
+        (source_id, kind, record_id, content_hash, occurrences),
+    )
+
+
+def _copy_dead_letters(source, backend, conn, source_id, report) -> int:
+    if not source.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='dead_letters'").fetchone():
         return 0
-    for event_id, received_at, sender, reason, payload_hash, payload, withheld in rows:
-        backend._execute(
+    rows = source.execute(
+        "SELECT id, event_id, received_at, sender, reason, payload_hash, payload, "
+        "payload_withheld FROM dead_letters ORDER BY id"
+    ).fetchall()
+    written = 0
+    for row_id, *data in rows:
+        record_id = str(row_id)
+        fingerprint = _fingerprint(data)
+        receipt = _receipt(backend, conn, source_id, "dead-letter", record_id)
+        if receipt:
+            if receipt[0] != fingerprint:
+                report.conflicts.append(f"dead-letter {record_id}: source changed after import")
+            continue
+        conn.execute(
             f'INSERT INTO "{backend.schema}".dead_letters '
             "(event_id, received_at, sender, reason, payload_hash, payload, payload_withheld) "
             "VALUES (%s, %s, %s, %s, %s, %s, %s)",
-            (event_id, received_at, sender, reason, payload_hash, payload, bool(withheld)),
+            (*data[:-1], bool(data[-1])),
         )
-    return len(rows)
+        _save_receipt(backend, conn, source_id, "dead-letter", record_id, fingerprint)
+        written += 1
+    return written
 
 
-def _copy_findings(source: sqlite3.Connection, backend) -> int:
-    try:
-        rows = source.execute(
-            "SELECT sender, source, action, field_path, outcome, persisted, "
-            "occurrences, first_seen, last_seen FROM secret_findings"
-        ).fetchall()
-    except sqlite3.Error:
+def _copy_findings(source, backend, conn, source_id, report) -> int:
+    if not source.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='secret_findings'").fetchone():
         return 0
+    rows = source.execute(
+        "SELECT sender, source, action, field_path, outcome, persisted, "
+        "occurrences, first_seen, last_seen FROM secret_findings"
+    ).fetchall()
+    written = 0
     for r in rows:
-        backend._execute(
+        record_id = _fingerprint(r[:5])
+        fingerprint = _fingerprint(r)
+        receipt = _receipt(backend, conn, source_id, "finding", record_id)
+        if receipt and receipt[0] == fingerprint:
+            continue
+        delta = r[6] - (receipt[1] if receipt else 0)
+        if delta < 0:
+            report.conflicts.append(f"finding {record_id}: source occurrence count decreased")
+            continue
+        conn.execute(
             f'INSERT INTO "{backend.schema}".secret_findings '
             "(sender, source, action, field_path, outcome, persisted, occurrences, "
             " first_seen, last_seen) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) "
             "ON CONFLICT (sender, source, action, field_path, outcome) DO UPDATE "
-            "SET occurrences = secret_findings.occurrences + excluded.occurrences",
-            (r[0], r[1], r[2], r[3], r[4], bool(r[5]), r[6], r[7], r[8]),
+            "SET occurrences = secret_findings.occurrences + excluded.occurrences, "
+            "persisted = secret_findings.persisted OR excluded.persisted, "
+            "first_seen = LEAST(secret_findings.first_seen, excluded.first_seen), "
+            "last_seen = GREATEST(secret_findings.last_seen, excluded.last_seen)",
+            (r[0], r[1], r[2], r[3], r[4], bool(r[5]), delta, r[7], r[8]),
         )
-    return len(rows)
+        _save_receipt(backend, conn, source_id, "finding", record_id, fingerprint, r[6])
+        written += 1
+    return written
 
 
 def _verify(source: sqlite3.Connection, backend, report: MigrationReport) -> bool:
@@ -200,4 +266,4 @@ def _same_instant(destination, source_iso: str) -> bool:
         return False
     if want.tzinfo is None or destination.tzinfo is None:
         return False
-    return abs(destination.timestamp() - want.timestamp()) < 1.0
+    return destination == want

audit_core/postgres_backend.py

@@ -45,6 +45,7 @@ from audit_core.integrity import (
     chain_link,
     evaluate_tamper_evidence,
     load_attestation,
+    order_chain_rows,
     verify_rows,
 )
 from audit_core.redaction import Finding
@@ -164,6 +165,21 @@ MIGRATIONS: list[tuple[str, str]] = [
         ALTER TABLE {schema}.events ADD COLUMN IF NOT EXISTS chain_prev text;
         """,
     ),
+    (
+        "0007-import-receipts",
+        """
+        CREATE INDEX IF NOT EXISTS events_chain_prev_idx ON {schema}.events (chain_prev);
+        CREATE INDEX IF NOT EXISTS events_chain_hash_idx ON {schema}.events (chain_hash);
+        CREATE TABLE IF NOT EXISTS {schema}.import_receipts (
+          source_id text NOT NULL,
+          kind text NOT NULL,
+          record_id text NOT NULL,
+          content_hash text NOT NULL,
+          occurrences bigint NOT NULL DEFAULT 0,
+          PRIMARY KEY (source_id, kind, record_id)
+        );
+        """,
+    ),
 ]
 
 # Rejection reasons whose payload must never be persisted. Storing the body of
@@ -380,11 +396,7 @@ class PostgresAuditBackend:
         try:
             with self.pool.connection() as conn, conn.transaction():
                 conn.execute("SELECT pg_advisory_xact_lock(%s)", (CHAIN_LOCK_KEY,))
-                head = conn.execute(
-                    f"SELECT chain_hash FROM {self._events} "
-                    "ORDER BY accepted_at DESC, event_id DESC LIMIT 1"
-                ).fetchone()
-                previous = head[0] if head and head[0] else GENESIS
+                previous = self._chain_head(conn)
                 link = chain_link(previous, payload_hash, event.event_id)
                 inserted = conn.execute(
                     f"""
@@ -445,7 +457,8 @@ class PostgresAuditBackend:
         return [{"accepted_at": _iso(at), **rec} for rec, at in rows]
 
     def event_counts(
-        self, source: str, since: str, until: str, tenant: str | None = None
+        self, source: str, since: str, until: str, tenant: str | None = None,
+        *, stream_instance_id: str | None = None
     ) -> list[dict]:
         """Per-class counts for one source over a bounded window.
 
@@ -462,6 +475,10 @@ class PostgresAuditBackend:
         if tenant is not None:
             sql += " AND tenant = %s"
             params += (tenant,)
+        if stream_instance_id is not None:
+            sql += " AND jsonb_typeof(record->'details'->'data'->'stream_instance_id') = 'string'"
+            sql += " AND record->'details'->'data'->>'stream_instance_id' = %s"
+            params += (stream_instance_id,)
         sql += " GROUP BY action ORDER BY action"
         return [{"class": action, "count": count} for action, count in self._query(sql, params)]
 
@@ -566,7 +583,7 @@ class PostgresAuditBackend:
             (),
         )
         return verify_rows(
-            [
+            order_chain_rows([
                 ChainRow(
                     event_id=r[0],
                     payload_hash=r[1],
@@ -575,7 +592,7 @@ class PostgresAuditBackend:
                     accepted_at=_iso(r[4]),
                 )
                 for r in rows
-            ],
+            ]),
             attestation=attestation,
         )
 
@@ -634,6 +651,23 @@ class PostgresAuditBackend:
         except Exception as exc:  # pool/libpq can raise outside psycopg.Error
             raise BackendUnavailableError(str(exc)) from exc
 
+    def _chain_head(self, conn) -> str:
+        """Find the unreferenced link while holding CHAIN_LOCK_KEY.
+
+        accepted_at is historical evidence, not insertion order: an imported
+        record may predate a record already held without rewriting its chain.
+        """
+        heads = conn.execute(
+            f"SELECT e.chain_hash FROM {self._events} e WHERE NOT EXISTS ("
+            f"SELECT 1 FROM {self._events} successor "
+            "WHERE successor.chain_prev = e.chain_hash) LIMIT 2"
+        ).fetchall()
+        if len(heads) == 1:
+            return heads[0][0]
+        if heads or conn.execute(f"SELECT EXISTS (SELECT 1 FROM {self._events})").fetchone()[0]:
+            raise BackendUnavailableError("audit chain has no unique head")
+        return GENESIS
+
     def _backfill_chain(self, conn) -> None:
         """Fill chain columns on rows accepted before migration 0006.
 

audit_core/sqlite_backend.py

@@ -220,7 +220,8 @@ class SQLiteAuditBackend:
         return [{"accepted_at": at, **json.loads(rec)} for rec, at in rows]
 
     def event_counts(
-        self, source: str, since: str, until: str, tenant: str | None = None
+        self, source: str, since: str, until: str, tenant: str | None = None,
+        *, stream_instance_id: str | None = None
     ) -> list[dict]:
         """Per-class counts for one source over a bounded window.
 
@@ -238,6 +239,10 @@ class SQLiteAuditBackend:
         if tenant is not None:
             sql += " AND tenant = ?"
             params += (tenant,)
+        if stream_instance_id is not None:
+            sql += " AND json_type(record, '$.details.data.stream_instance_id') = 'text'"
+            sql += " AND json_extract(record, '$.details.data.stream_instance_id') = ?"
+            params += (stream_instance_id,)
         sql += " GROUP BY action ORDER BY action"
         return [{"class": action, "count": count} for action, count in self._query(sql, params)]