Prepare railiance01 delivery: dynamic leases, migrate Job, operator runbook

VaultDynamicSecret pulls database/creds/* so a rotating lease is not frozen
into KV. Runtime sets AUDIT_CORE_AUTO_MIGRATE=0; schema is a Job with the
migration lease. Image base is digest-pinned. Namespace and NetworkPolicies
are on the cluster; Deployment waits for the attended OpenBao ESO token.

audit_core/cli.py

@@ -69,6 +69,29 @@ def build_parser() -> argparse.ArgumentParser:
     cleanup_parser.add_argument("--retention-days", type=int, default=None)
     cleanup_parser.set_defaults(func=cleanup)
 
+    schema_parser = sub.add_parser(
+        "migrate",
+        help="Apply pending PostgreSQL schema migrations and exit.",
+    )
+    schema_parser.add_argument(
+        "--to-postgres", dest="destination", default=None,
+        help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.",
+    )
+    schema_parser.add_argument("--schema", default="audit_core")
+    schema_parser.set_defaults(func=migrate_schema)
+
+    replay_parser = sub.add_parser(
+        "replay",
+        help="Re-submit a stored event through accept(); must reconcile as duplicate.",
+    )
+    replay_parser.add_argument("--event-id", required=True)
+    replay_parser.add_argument(
+        "--to-postgres", dest="destination", default=None,
+        help="DSN; defaults to AUDIT_CORE_DATABASE_URL or a mounted credential dir.",
+    )
+    replay_parser.add_argument("--schema", default="audit_core")
+    replay_parser.set_defaults(func=replay_event)
+
     migrate_parser = sub.add_parser(
         "migrate-store", help="Move audit records from a SQLite store into PostgreSQL."
     )
@@ -87,6 +110,54 @@ def build_parser() -> argparse.ArgumentParser:
     return parser
 
 
+def _postgres_backend(dsn: str | None, schema: str, *, migrate: bool):
+    import os
+
+    from audit_core.postgres_backend import PostgresAuditBackend
+
+    return PostgresAuditBackend(
+        dsn if dsn is not None else os.environ.get("AUDIT_CORE_DATABASE_URL") or "",
+        schema=schema,
+        credential_dir=os.environ.get("AUDIT_CORE_CREDENTIAL_DIR"),
+        migrate=migrate,
+    )
+
+
+def migrate_schema(args: argparse.Namespace) -> int:
+    """Apply pending schema migrations using the current connection.
+
+    Production runs this as a Job with the migration lease. The runtime
+    Deployment sets AUDIT_CORE_AUTO_MIGRATE=0 so the app role never needs
+    CREATE TABLE.
+    """
+    backend = _postgres_backend(args.destination, args.schema, migrate=False)
+    try:
+        applied = backend.migrate()
+    finally:
+        backend.close()
+    print(json.dumps({"ok": True, "schema": args.schema, "applied": applied}, sort_keys=True))
+    return 0
+
+
+def replay_event(args: argparse.Namespace) -> int:
+    """Reconcile a stored event. A first-acceptance result is a custody defect."""
+    backend = _postgres_backend(args.destination, args.schema, migrate=False)
+    try:
+        result = backend.replay(args.event_id)
+    except KeyError:
+        print(json.dumps({"ok": False, "error": "not_found", "event_id": args.event_id}))
+        return 1
+    finally:
+        backend.close()
+    print(json.dumps({
+        "ok": True,
+        "event_id": args.event_id,
+        "duplicate": result.duplicate,
+        "reference": result.reference,
+    }, sort_keys=True))
+    return 0 if result.duplicate else 2
+
+
 def migrate_store(args: argparse.Namespace) -> int:
     import os
 

audit_core/ingestion.py

@@ -499,6 +499,10 @@ def build_backend() -> IdempotentAuditBackend:
             statement_timeout_ms=int(
                 os.environ.get("AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS", "30000")
             ),
+            # Production mounts the runtime role, which cannot CREATE TABLE.
+            # Schema changes run as a Job with the migration lease
+            # (AUDIT-WP-0005-T02). Local and brokered use keep the default.
+            migrate=_env_flag("AUDIT_CORE_AUTO_MIGRATE", default=True),
         )
     path = os.environ.get("AUDIT_CORE_DATABASE_PATH", "/data/audit-core.db")
     log.warning(
@@ -508,6 +512,14 @@ def build_backend() -> IdempotentAuditBackend:
     return SQLiteAuditBackend(path)
 
 
+def _env_flag(name: str, *, default: bool) -> bool:
+    """Parse a boolean environment flag. Unset or empty keeps ``default``."""
+    raw = os.environ.get(name)
+    if raw is None or not raw.strip():
+        return default
+    return raw.strip().lower() not in {"0", "false", "no", "off"}
+
+
 def main() -> None:
     logging.basicConfig(
         level=os.environ.get("AUDIT_CORE_LOG_LEVEL", "INFO"),