RPF-WP-0015-T03: post-reboot synthetic E2 mode in the reviewed sender lane

One contained login: project one synthetic sender, event accepted, replay
must be duplicate (exactly once), cleanup; no lease is touched.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 680842@bnt-lap001
Assistant-Session: 54eb9e8e-7d6b-4f0c-b7a6-a6db62075883

scripts/audit-core-database-lease-recovery.py

@@ -24,6 +24,9 @@ from typing import Any, Callable
 
 ROOT = Path(__file__).resolve().parents[1]
 TASK_ID = "RAILIANCE-WP-0024-T02"
+# The post-reboot synthetic E2 (T03 runbook step 8) reuses this receipt shape and lane.
+E2_TASK_ID = "RAILIANCE-WP-0024-T03"
+E2_CONFIRM = f"{E2_TASK_ID}:synthetic-e2:attended"
 PROCEDURE = "audit-core-database-lease-recovery"
 LEASE_PREFIX = "database/creds/audit-core-runtime"
 EXTERNAL_SECRET = "audit-core-database"
@@ -70,14 +73,16 @@ def parse_time(value: str) -> datetime:
     return parsed.astimezone(UTC)
 
 
-def load_approval(path: Path, now: datetime, *, require_open_window: bool) -> dict[str, Any]:
+def load_approval(
+    path: Path, now: datetime, *, require_open_window: bool, task_id: str = TASK_ID
+) -> dict[str, Any]:
     try:
         document = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, json.JSONDecodeError) as exc:
         raise ProcedureError("approval receipt is unavailable or invalid JSON") from exc
     if not isinstance(document, dict):
         raise ProcedureError("approval receipt must be an object")
-    if document.get("procedure") != PROCEDURE or document.get("task_id") != TASK_ID:
+    if document.get("procedure") != PROCEDURE or document.get("task_id") != task_id:
         raise ProcedureError("approval receipt has the wrong procedure or task")
     if document.get("status") != "approved" or not document.get("approval_id"):
         raise ProcedureError("approval receipt is not approved")
@@ -380,6 +385,50 @@ def force_refresh(remote: Any, baseline: dict[str, Any]) -> dict[str, Any]:
     return latest
 
 
+def synthetic_e2(args: argparse.Namespace) -> dict[str, Any]:
+    """Post-reboot E2: one synthetic event accepted exactly once, no lease touched.
+
+    baseline must be accepted; resending the same fixture must return duplicate
+    (exactly-once); the pod must not change while the event is written.
+    """
+    approval = load_approval(args.approval, datetime.now(UTC), require_open_window=True, task_id=E2_TASK_ID)
+    if args.confirm != E2_CONFIRM:
+        raise ProcedureError(f"synthetic E2 requires --confirm {E2_CONFIRM}")
+    remote = PREFLIGHT.Remote(args.remote)
+    driver = LoadDriver(
+        args.load_driver,
+        approval["synthetic_load"]["contract_id"],
+        approval["synthetic_load"]["driver_revision"],
+    )
+    before_pod = pod_state(remote)
+    if PREFLIGHT.endpoint_status(remote, "/readyz") != 200:
+        raise ProcedureError("audit-core is not ready for the synthetic E2")
+    accepted = driver.run("baseline")
+    if accepted.get("status") != "ready":
+        raise ProcedureError("synthetic E2 event was not accepted")
+    replay = driver.run("expect-recovered")
+    if replay.get("status") != "duplicate" or replay.get("fixture_id") != accepted.get("fixture_id"):
+        raise ProcedureError("synthetic E2 replay was not recognised as a duplicate")
+    load_cleanup = driver.run("cleanup")
+    after_pod = pod_state(remote)
+    return {
+        "procedure": PROCEDURE,
+        "mode": "post-reboot-synthetic-e2",
+        "task_id": E2_TASK_ID,
+        "approval_id": approval["approval_id"],
+        "completed_at": datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
+        "synthetic_event_accepted": True,
+        "replay_status": replay["status"],
+        "accepted_exactly_once": True,
+        "same_pod_uid": after_pod["uid"] == before_pod["uid"],
+        "restart_count_unchanged": after_pod["restart_count"] == before_pod["restart_count"],
+        "load_cleanup_status": load_cleanup.get("status"),
+        "synthetic_load_contract": approval["synthetic_load"]["contract_id"],
+        "synthetic_load_driver_revision": approval["synthetic_load"]["driver_revision"],
+        "secret_values_observed": False,
+    }
+
+
 def exercise(args: argparse.Namespace) -> dict[str, Any]:
     approval = load_approval(args.approval, datetime.now(UTC), require_open_window=True)
     if args.confirm != CONFIRM:
@@ -515,7 +564,7 @@ def emit_result(result: dict, result_file: "Path | None") -> None:
 
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser()
-    parser.add_argument("command", choices=["validate-approval", "exercise"])
+    parser.add_argument("command", choices=["validate-approval", "exercise", "synthetic-e2"])
     parser.add_argument("--approval", type=Path, required=True)
     parser.add_argument("--remote", default="railiance01")
     parser.add_argument(
@@ -540,7 +589,7 @@ def main(argv: list[str] | None = None) -> int:
                 raise ProcedureError("live exercise requires --contained (scripts/openbao-attended-exec.py)")
             if args.load_driver is None:
                 raise ProcedureError("live exercise requires --load-driver")
-            result = exercise(args)
+            result = synthetic_e2(args) if args.command == "synthetic-e2" else exercise(args)
     except (OSError, IndexError, ProcedureError, PREFLIGHT.PreflightError) as exc:
         if args.result_file is None:
             print(f"database lease recovery failed: {exc}", file=sys.stderr)