Establish pre-production record disposition and add the migration tool

AUDIT-WP-0005-T04.

Disposition: there are no pre-production records. No audit-core SQLite store
on this host, no mock-file-backend output, and no audit-core pod, deployment
or PVC on railiance01 - the only audit-* PVC there is OpenBao's own audit
device. Consistent with the history: WP-0003-T03 was cancelled before the
receiver was ever deployed, so every SQLite store that has existed was a test
fixture. Nothing is being discarded because nothing was ever accepted outside
tests.

The tool is built anyway because the SQLite path stays reachable - the
entrypoint falls back to it when AUDIT_CORE_DATABASE_URL is unset. If that
fallback is ever used in anger the records are audit records, and writing the
migration afterwards under pressure is the wrong time.

audit_core.migrate_store and `python -m audit_core migrate-store` transfer
events, dead letters and secret-finding counters. Records keep their original
event_id, payload_hash and accepted_at, which is why this bypasses accept():
that stamps acceptance with the current time, and a migration that rewrote
acceptance times would destroy the evidence it exists to preserve.

Idempotent, and verification reads back from the destination rather than
trusting the write path. A destination record with a differing payload hash is
reported as a conflict and left untouched - silently overwriting a stored audit
record is the same class of failure as losing it. Conflicts and failed
verification exit non-zero; a partial migration is not a success.

Tests 71 -> 77.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

WORK-RECORDS.md

@@ -28,7 +28,7 @@
 | task | AUDIT-WP-0004-T05 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md |
 | task | AUDIT-WP-0004-T06 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md |
 | task | AUDIT-WP-0004-T07 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md |
-| task | AUDIT-WP-0005-T01 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
+| task | AUDIT-WP-0005-T01 | done | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
 | task | AUDIT-WP-0005-T02 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
 | task | AUDIT-WP-0005-T03 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |
 | task | AUDIT-WP-0005-T04 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md |

audit_core/cli.py

@@ -69,9 +69,53 @@ def build_parser() -> argparse.ArgumentParser:
     cleanup_parser.add_argument("--retention-days", type=int, default=None)
     cleanup_parser.set_defaults(func=cleanup)
 
+    migrate_parser = sub.add_parser(
+        "migrate-store", help="Move audit records from a SQLite store into PostgreSQL."
+    )
+    migrate_parser.add_argument("--from-sqlite", required=True, dest="source")
+    migrate_parser.add_argument(
+        "--to-postgres", dest="destination", default=None,
+        help="DSN; defaults to AUDIT_CORE_DATABASE_URL",
+    )
+    migrate_parser.add_argument("--schema", default="audit_core")
+    migrate_parser.add_argument(
+        "--dry-run", action="store_true",
+        help="Report what would move without writing anything.",
+    )
+    migrate_parser.set_defaults(func=migrate_store)
+
     return parser
 
 
+def migrate_store(args: argparse.Namespace) -> int:
+    import os
+
+    from audit_core.migrate_store import migrate
+
+    if args.dry_run:
+        import sqlite3
+
+        with sqlite3.connect(f"file:{args.source}?mode=ro", uri=True) as source:
+            count = source.execute("SELECT count(*) FROM events").fetchone()[0]
+        print(json.dumps({"dry_run": True, "events_in_source": count}, sort_keys=True))
+        return 0
+
+    from audit_core.postgres_backend import PostgresAuditBackend
+
+    dsn = args.destination or os.environ.get("AUDIT_CORE_DATABASE_URL")
+    if not dsn:
+        raise SystemExit("--to-postgres or AUDIT_CORE_DATABASE_URL is required")
+
+    backend = PostgresAuditBackend(dsn, schema=args.schema)
+    try:
+        report = migrate(args.source, backend)
+    finally:
+        backend.close()
+    print(report.summary())
+    # A conflict or a failed verification is not a partial success.
+    return 0 if report.ok else 1
+
+
 def main() -> int:
     parser = build_parser()
     args = parser.parse_args()

audit_core/migrate_store.py

@@ -0,0 +1,203 @@
+"""Move audit records from a SQLite store into PostgreSQL (AUDIT-WP-0005-T04).
+
+As of 2026-08-10 there are no pre-production records to move: audit-core was
+never deployed, so the only SQLite stores that ever existed were test
+fixtures. See the workplan for that finding.
+
+The tool exists anyway because the SQLite path is still reachable — the
+entrypoint falls back to it when ``AUDIT_CORE_DATABASE_URL`` is unset. If that
+fallback is ever used in anger, the records it accumulates are audit records
+and cannot simply be dropped.
+
+Fidelity is the whole point. Records are transferred with their original
+``event_id``, ``payload_hash``, and ``accepted_at`` intact; a migration that
+rewrote acceptance times would destroy the evidence it was meant to preserve.
+That is why this bypasses ``accept()``, which stamps ``accepted_at`` with the
+current time.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from dataclasses import dataclass, field
+
+from audit_core.interface import BackendUnavailableError
+
+
+@dataclass
+class MigrationReport:
+    """What happened, in enough detail to be evidence."""
+
+    events_read: int = 0
+    events_written: int = 0
+    events_already_present: int = 0
+    dead_letters_written: int = 0
+    findings_written: int = 0
+    conflicts: list[str] = field(default_factory=list)
+    verified: bool = False
+
+    @property
+    def ok(self) -> bool:
+        return not self.conflicts and self.verified
+
+    def summary(self) -> str:
+        lines = [
+            f"events read:            {self.events_read}",
+            f"events written:         {self.events_written}",
+            f"already present:        {self.events_already_present}",
+            f"dead letters written:   {self.dead_letters_written}",
+            f"secret findings written:{self.findings_written}",
+            f"verified:               {self.verified}",
+        ]
+        if self.conflicts:
+            lines.append(f"CONFLICTS ({len(self.conflicts)}):")
+            lines += [f"  - {c}" for c in self.conflicts]
+        return "\n".join(lines)
+
+
+def migrate(sqlite_path: str, backend, *, verify: bool = True) -> 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
+    and never overwritten — the destination record wins, and a human decides.
+    """
+    report = MigrationReport()
+    source = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True)
+    try:
+        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)
+
+        report.dead_letters_written = _copy_dead_letters(source, backend)
+        report.findings_written = _copy_findings(source, backend)
+
+        if verify:
+            report.verified = _verify(source, backend, report)
+    finally:
+        source.close()
+    return report
+
+
+def _import_event(
+    backend, 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,)
+    )
+    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"
+            )
+        return "present"
+
+    backend._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",
+        (
+            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),
+        ),
+    )
+    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:
+        return 0
+    for event_id, received_at, sender, reason, payload_hash, payload, withheld in rows:
+        backend._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)),
+        )
+    return len(rows)
+
+
+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:
+        return 0
+    for r in rows:
+        backend._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]),
+        )
+    return len(rows)
+
+
+def _verify(source: sqlite3.Connection, backend, report: MigrationReport) -> bool:
+    """Confirm every source event is present with its hash and time intact.
+
+    Verification reads back rather than trusting the write path: a migration
+    that reports success without checking is the same as no migration at all.
+    """
+    rows = source.execute("SELECT event_id, payload_hash, accepted_at FROM events").fetchall()
+    for event_id, payload_hash, accepted_at in rows:
+        found = backend._query(
+            f"SELECT payload_hash, accepted_at FROM {backend._events} WHERE event_id = %s",
+            (event_id,),
+        )
+        if not found:
+            report.conflicts.append(f"{event_id}: missing from destination after migration")
+            return False
+        if found[0][0] != payload_hash:
+            report.conflicts.append(f"{event_id}: payload hash differs after migration")
+            return False
+        if not _same_instant(found[0][1], accepted_at):
+            report.conflicts.append(
+                f"{event_id}: accepted_at not preserved "
+                f"(source {accepted_at}, destination {found[0][1]})"
+            )
+            return False
+    return True
+
+
+def _same_instant(destination, source_iso: str) -> bool:
+    from datetime import datetime
+
+    try:
+        want = datetime.fromisoformat(str(source_iso).replace("Z", "+00:00"))
+    except ValueError:
+        return False
+    if not hasattr(destination, "timestamp"):
+        return False
+    if want.tzinfo is None or destination.tzinfo is None:
+        return False
+    return abs(destination.timestamp() - want.timestamp()) < 1.0

tests/test_migrate_store.py

@@ -0,0 +1,170 @@
+"""Migration fidelity tests (AUDIT-WP-0005-T04).
+
+The controlling property is that migrating audit records does not alter them.
+An event that arrives in the destination with a rewritten acceptance time has
+been damaged, not moved.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import uuid
+
+import pytest
+
+from audit_core.ingestion import IngestionApplication
+from audit_core.migrate_store import migrate
+from audit_core.sqlite_backend import SQLiteAuditBackend
+
+from tests.test_backend_conformance import HAVE_PG, PG_URL, make_event
+
+pytestmark = pytest.mark.skipif(not HAVE_PG, reason="needs PostgreSQL")
+
+
+@pytest.fixture
+def destination():
+    from audit_core.postgres_backend import PostgresAuditBackend
+
+    schema = f"mig_{uuid.uuid4().hex[:12]}"
+    backend = PostgresAuditBackend(PG_URL, schema=schema)
+    try:
+        yield backend
+    finally:
+        with backend.pool.connection() as conn:
+            conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
+        backend.close()
+
+
+@pytest.fixture
+def populated_sqlite(tmp_path):
+    """A SQLite store holding events, a dead letter, and a finding."""
+    path = str(tmp_path / "legacy.db")
+    source = SQLiteAuditBackend(path)
+    for i in range(3):
+        event = make_event(f"legacy-{i}")
+        source.accept(event, hashlib.sha256(f"h{i}".encode()).hexdigest())
+    source.record_rejection(
+        event_id="bad-1", reason="source_not_allowed", payload_hash="hh",
+        sender="user-engine", payload='{"source":"nope"}',
+    )
+    from audit_core.redaction import Finding
+
+    source.count_secret_findings(
+        sender="user-engine", source="user-engine", action="membership.added",
+        outcome="redacted", findings=[Finding("data.auth_token", True)],
+    )
+    return path, source
+
+
+def test_migration_preserves_identity_hash_and_time(populated_sqlite, destination):
+    path, source = populated_sqlite
+    original = {
+        row[0]: (row[1], row[2])
+        for row in source.db.execute(
+            "SELECT event_id, payload_hash, accepted_at FROM events"
+        ).fetchall()
+    }
+
+    report = migrate(path, destination)
+
+    assert report.events_read == 3
+    assert report.events_written == 3
+    assert report.conflicts == []
+    assert report.verified is True
+    assert report.ok is True
+
+    for event_id, (payload_hash, accepted_at) in original.items():
+        rows = destination._query(
+            f"SELECT payload_hash, accepted_at FROM {destination._events} "
+            "WHERE event_id = %s", (event_id,),
+        )
+        assert rows, f"{event_id} missing"
+        assert rows[0][0] == payload_hash
+        # Acceptance time is evidence; rewriting it would destroy what the
+        # migration exists to preserve.
+        assert rows[0][1].isoformat().startswith(accepted_at[:16])
+
+
+def test_migration_carries_dead_letters_and_findings(populated_sqlite, destination):
+    path, _ = populated_sqlite
+    report = migrate(path, destination)
+
+    assert report.dead_letters_written == 1
+    assert report.findings_written == 1
+    assert destination.dead_letters()[0]["reason"] == "source_not_allowed"
+    assert destination.secret_findings()[0]["field_path"] == "data.auth_token"
+
+
+def test_migration_is_idempotent(populated_sqlite, destination):
+    path, _ = populated_sqlite
+    first = migrate(path, destination)
+    second = migrate(path, destination)
+
+    assert first.events_written == 3
+    assert second.events_written == 0
+    assert second.events_already_present == 3
+    assert second.ok is True
+
+    count = destination._query(
+        f"SELECT count(*) FROM {destination._events}", ()
+    )[0][0]
+    assert count == 3
+
+
+def test_a_diverging_destination_record_is_a_conflict_not_an_overwrite(
+    populated_sqlite, destination
+):
+    """The destination wins and a human decides.
+
+    Silently overwriting a stored audit record with a differing one would be
+    the same class of failure as losing it.
+    """
+    path, _ = populated_sqlite
+    migrate(path, destination)
+
+    # Simulate divergence: same id, different payload hash in the source.
+    source = SQLiteAuditBackend(path)
+    source.db.execute("BEGIN IMMEDIATE")
+    source.db.execute(
+        "UPDATE events SET payload_hash = 'diverged' WHERE event_id = 'legacy-1'"
+    )
+    source.db.execute("COMMIT")
+
+    report = migrate(path, destination)
+    assert report.conflicts, "divergence must be reported"
+    assert "legacy-1" in report.conflicts[0]
+    assert report.ok is False
+
+    survived = destination._query(
+        f"SELECT payload_hash FROM {destination._events} WHERE event_id = 'legacy-1'", ()
+    )[0][0]
+    assert survived != "diverged", "destination record must not be overwritten"
+
+
+def test_migrated_events_remain_readable_and_append_only(populated_sqlite, destination):
+    path, _ = populated_sqlite
+    migrate(path, destination)
+
+    record = destination.get("legacy-0")
+    assert record["tenant"] == "tenant:friendly:binky"
+    assert destination.retention_policy.immutable is True
+
+    from audit_core.interface import BackendUnavailableError
+
+    with pytest.raises(BackendUnavailableError):
+        destination._execute(
+            f"DELETE FROM {destination._events} WHERE event_id = %s", ("legacy-0",)
+        )
+
+
+def test_empty_source_migrates_cleanly(tmp_path, destination):
+    """Today's actual case: nothing to move."""
+    path = str(tmp_path / "empty.db")
+    SQLiteAuditBackend(path)
+
+    report = migrate(path, destination)
+    assert report.events_read == 0
+    assert report.events_written == 0
+    assert report.ok is True

workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md

@@ -166,7 +166,7 @@ previous version.
 
 ```task
 id: AUDIT-WP-0005-T04
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "9010fb4a-a1b8-4ef7-b143-e33ca7cc0619"
 ```
@@ -182,6 +182,42 @@ events is the exact failure this service is meant to make impossible.
 Done when the disposition of every pre-production record is either migrated
 and verified, or explicitly and justifiably discarded.
 
+**Disposition, established 2026-08-10: there are no pre-production records.**
+Checked and found empty: no `/data/audit-core.db` or any other audit-core
+SQLite store on this host, no mock-file-backend output under `/tmp/audit-core`,
+and no audit-core pod, deployment, or PVC on railiance01. (The only `audit-*`
+PVC there is `audit-openbao-0`, which is OpenBao's own audit device and
+unrelated to this service.)
+
+That is consistent with the history rather than surprising: WP-0003-T03 was
+cancelled before the receiver was ever deployed, so the only SQLite stores that
+have ever existed were test fixtures with no custody value. Nothing is being
+discarded, because nothing was ever accepted outside tests.
+
+The migration tool was built regardless, because the SQLite path remains
+reachable: the entrypoint falls back to it when `AUDIT_CORE_DATABASE_URL` is
+unset. If that fallback is ever used in anger, the records it accumulates are
+audit records and cannot simply be dropped — and building the tool afterwards,
+under pressure, is the wrong time.
+
+`audit_core.migrate_store` plus `python -m audit_core migrate-store` transfer
+events, dead letters, and secret-finding counters. Fidelity is the point:
+records keep their original `event_id`, `payload_hash`, and `accepted_at`,
+which is why the tool bypasses `accept()` — that stamps acceptance with the
+current time, and a migration that rewrote acceptance times would destroy the
+evidence it exists to preserve.
+
+Migration is idempotent, and verification reads back from the destination
+rather than trusting the write path. Where the destination already holds an
+event with a *different* payload hash, the tool reports a conflict and leaves
+the destination untouched: silently overwriting a stored audit record with a
+differing one is the same class of failure as losing it. A conflict or failed
+verification exits non-zero — a partial migration is not a success.
+
+Six tests cover fidelity, dead letters and counters, idempotency, conflict
+handling, post-migration readability under the append-only trigger, and the
+empty-source case that is today's actual situation.
+
 ## T05 - Run the live failure matrix
 
 ```task