Prepare attended source-date supervisor and recovered-pair rehearsal

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3

coulomb_social/apps/members/client_export.py

@@ -88,6 +88,12 @@ def schema_inventory():
             "django_admin_log": "Export only events attributable to selected client objects.",
             "process_logs": "Operator retention applies; outside application data export.",
             "backup_copies": "Independent retention/expiry requires operator disposition.",
+            "date_correction_journals": (
+                "Operator-private date-operations journals retain original index bytes, plan and "
+                "backup metadata outside CONTENT_ROOT. Include retained client bytes in operator "
+                "offboarding handoff and explicit expiry; the app export cannot read "
+                "that root store."
+            ),
         },
         "operational_schema": [
             {"model": model._meta.label_lower, "table": model._meta.db_table,

docs/release/client-data-export-review-2026-10-30.md

@@ -20,6 +20,12 @@ than copying or silently discarding it. Independent backups and process logs sta
 under operator retention. Exporting data does not delete any of these stores or
 revoke sessions.
 
+The operational inventory also names the prepared supervisor's private
+`date-operations` journals outside CONTENT_ROOT. They can retain original private
+index bytes, plan and backup metadata. The app cannot read that root store;
+operator offboarding must account for any retained client bytes and their explicit
+expiry separately. The supervisor rehearsal creates no production journal.
+
 The inventory derives field names, database columns, types, nullability and
 relations from the loaded Django models. Every application field has an explicit
 classification. A new model or unclassified field refuses export until reviewed.

docs/release/source-date-journal-review-2026-10-30.md

@@ -110,10 +110,10 @@ The candidate image uses explicit source overlays; no image is released here.
 The focused planner/journal/operation suite passes 50 cases; `make check` passes
 364 cases including protected browsers, with zero skips.
 
-Remaining wrapper requirements: bind exact reviewed source/plan/target and fresh
-backup/readback hashes; verify the external writer fence and full live SQL/content
-baseline; invoke this operation in the pinned helper; supervise the privileged
-action; refuse outstanding journal state;
-retain the journal on durable private storage; perform full unrelated-row/file
-postflight and resolve every pending state before resuming writers. Rehearse that
-wrapper against the independently recovered real pair before productive use.
+The external supervisor source and recovered-pair wrapper rehearsal are now
+described in source-date-supervisor-review-2026-10-30.md. They add exact capsule
+binding, native capture/source bookends, pinned helper/fence checks, private durable
+journals and explicit recovery/resume handling. The rehearsal's control plane and
+provider remain simulated. Actual storage release, native observation/admission,
+fresh verified backup, external fence and productive correction/postflight remain
+required. Source tests and isolated rehearsal do not establish those live gates.

docs/release/source-date-supervisor-rehearsal-2026-10-30.json

@@ -0,0 +1,58 @@
+{
+  "schema": "coulomb.supervised-date-isolated-rehearsal.v1",
+  "status": "controller_and_real_helper_rehearsed",
+  "scenarios": [
+    {
+      "scenario": "normal",
+      "status": "settled",
+      "actual_helper_wire_and_data_operation": true,
+      "capture_calls": 1,
+      "apply_calls": 1,
+      "all_tables_compared": 13,
+      "all_files_compared": 82
+    },
+    {
+      "scenario": "lost_apply_reply",
+      "status": "settled",
+      "actual_helper_wire_and_data_operation": true,
+      "capture_calls": 1,
+      "apply_calls": 1,
+      "all_tables_compared": 13,
+      "all_files_compared": 82
+    }
+  ],
+  "kubernetes_and_root_operations_simulated": true,
+  "capture_and_provider_simulated": true,
+  "production_access": false,
+  "native_observer_not_executed": true,
+  "production_writes": false,
+  "live_fence_or_native_backup_proved": false,
+  "source_overlay_used": true,
+  "candidate_image_id": "sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd",
+  "retained_recovery_source_unchanged": true,
+  "disposable_sql_clones_removed": true,
+  "sources_sha256": {
+    "scripts/source_date_supervisor.py": "a212bac3307611d27791eb857323b6344ab138a31a03120f908bda0ef1bf30c7",
+    "scripts/source_date_wire.py": "044480950df86194a91136284de734801a463ebbdbbfa2cb1b0b6ffd1e91bc82",
+    "scripts/prepare-source-date-capsule.py": "1d1b788f7e1abeb076832bf57779245cb13c16d581048fbb5b7b39957ff5a8e8",
+    "scripts/observe-source-date-target.py": "0e55360f092e2975cb547413dbc5b414088f3a6e56882005c422013bb13796a4",
+    "scripts/rehearse-source-date-supervisor.py": "51eff1511b7c961a0761723abcbd2076a648f564c7d4f06b35fa38fac0763d62",
+    "tests/test_source_date_supervisor.py": "ebe06e24e91b0e7c9b275442fc77f79cfcb810eb00a51d8b5b7622afaf7966a7"
+  },
+  "modules_sha256": {
+    "date_backfill.py": "84ce5d3369389e1af9ab999aad3804456fd2f66e654e290496be6930cd4885e4",
+    "date_backfill_journal.py": "47af997c02b3910058e5dc8c593dd8135a68be94e909c672a946f54601eb621e",
+    "date_backfill_operation.py": "50f962f6c42fd47dbbbdea3f06d38ccfef063e784bdb6361b8a2325cbe825e18"
+  },
+  "private_plan_sha256": "44369d5abbf92d3e6d52e909cdc1e1c3a90b8817be7027c3a43846fbfc381ef8",
+  "restore_receipt_sha256": "5f745cdae0a34c4a98f0e4429ceea49475b4e5cd9d2f8f09dea6be000e200258",
+  "acceptance_gate": {
+    "command": "make check",
+    "passed": 383,
+    "failures": 0,
+    "errors": 0,
+    "skipped": 0,
+    "required_browser_journeys_included": true,
+    "junit_sha256": "89b349cec61070d1c51faf20056c9e55ff4f9c274d30298fe3fac4f9d8e008dd"
+  }
+}

docs/release/source-date-supervisor-review-2026-10-30.md

@@ -0,0 +1,127 @@
+# Attended creation-date supervisor
+
+Prepared source, not a productive correction or an admitted production capsule.
+The separate storage release proposal `3389c506` is unchanged and still requires
+its founder-terminal execution. Observe and seal the date target only after that
+release: migration 0003 changes the complete-store baseline.
+
+## Bound operation
+
+`scripts/observe-source-date-target.py` performs read-only native observation on
+the content node. It requires the expected cluster/deployment/PVC UIDs and exact
+image digest, verifies the local machine against the Kubernetes node, reads only
+the installed helper hashes and public recipient, and probes SQL/content metadata.
+It checks the stable Helm release, sole Ready application writer, global slug
+constraint and a server-dry-run helper. It publishes a new private capsule under
+`/tmp`; it does not capture, scale, create a live helper or change dates.
+
+`scripts/prepare-source-date-capsule.py` also supports offline preparation from
+private plan, target and baseline files. Neither preparation path grants approval.
+Each attempt has a new random identifier. Its digest binds the manifest, exact
+target, complete-store baseline, plan, supervisor, helper protocol and all three
+date modules. Changing any reviewed byte requires another digest.
+
+`scripts/source_date_supervisor.py` defaults to read-only preflight. Execution
+requires node root, a terminal and the founder's exact `APPLY-DATE <digest>`
+confirmation. It rechecks the deployment, actual running image, physical node,
+PVC, Helm release, absent autoscaling/GitOps adoption, installed backup hashes and
+inactive backup timers. It invokes the existing admitted paired capture worker,
+checks immutable primary readback and local ciphertext metadata/hash, and compares
+the complete SQL/content baseline before and after capture.
+
+The supervisor then holds backup/snapshot locks, scales the pinned deployment to
+zero with resource-version/current-scale preconditions, waits for application
+pods to disappear and refuses unknown PVC users. A single pinned helper mounts
+the existing content PVC and an operator-private journal directory on that same
+node. It uses application Secret references without retrieving their values,
+no service-account token, a read-only root filesystem, dropped capabilities and
+bounded memory/CPU/tmp storage. Server admission must preserve the reviewed
+process, environment, mounts and security fields; the actual helper UID/spec are
+checked again before use.
+
+The helper rechecks the complete baseline, takes the existing content lock and
+PostgreSQL table locks, and invokes the reviewed durable date operation. Only the
+selected SQL creation date and index date insertion may change. All tables and
+content hashes must match the derived result before the exact helper is deleted
+with a UID precondition and the application is resumed. This is an attended
+sole-writer maintenance operation; native observation and execution must establish
+the real writer boundary. Synthetic control-plane fixtures do not establish it.
+
+## Durable state and recovery
+
+The root receipt and frozen capsule live under
+`/var/lib/coulomb-social-backup/date-operations/<proposal-hex>/`.
+The helper's `private/` directory is owned by application UID 10001 with mode 0700;
+the root ancestor is private and the application does not mount this store.
+The journal contains original private index bytes, the plan and original backup
+binding. Include it in operator client-offboarding handoff and explicit expiry.
+The app's client export cannot read that root store. No expiry is performed here.
+
+Uncertain outcomes retain the stopped application, helper and durable state.
+Never replay `--execute` or replace the original backup with a new capture.
+Explicit founder-attended `--recover` uses the frozen capsule and original digest:
+
+| Observed state | Recovery |
+| --- | --- |
+| Original helper operation still holds its kernel lock | Refuse; keep writers fenced. |
+| Pending journal | Reconcile against committed SQL using the original backup binding. |
+| Completed, rolled-back or absent journal | Require its exact expected complete-store fingerprint. |
+| Lost helper-create/delete reply | Inspect the pinned name, UID and spec; recreate only the same helper when absent. |
+| Original helper has terminated | Delete the matching terminal UID and recreate only for inspection/reconciliation. |
+| Replaced helper, unknown writer or unrelated data drift | Refuse for operator review. |
+| Lost scale-up reply, exact application already running | Settle rollout/receipt without restoring historical data or overwriting later writes. |
+
+Remote helper work is time-bounded. An interrupted process releases SQL/kernel
+locks; its durable journal determines reconciliation. Normal pod deletion is
+followed by disappearance checks. No forced deletion or generic data restore is
+used. A receipt phase of `completed` means the attempt is settled; correction is
+proved only by `correction_status: completed` and native postflight. An absent or
+rolled-back correction does not finish WP-0014-T04.
+
+## Operator sequence after the storage release
+
+Run the observer from the reviewed source checkout on the content node, supplying
+the private accepted-cohort plan and a fresh output directory:
+
+```sh
+sudo python3 scripts/observe-source-date-target.py \
+  --plan /private/date-backfill-plan.json \
+  --output /tmp/csoc-date-capsule-NEW \
+  --image forgejo.coulomb.social/coulomb/coulomb-social@sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd \
+  --cluster-uid a553c742-0115-43d4-99a4-a5ca56fe0786 \
+  --deployment-uid 2dbe661a-d232-4477-bbd1-833813077fbe \
+  --pvc-uid 9888abeb-9d52-4c70-a23d-b5494780d9c3
+```
+
+Review that newly observed capsule and exact proposal before founder execution:
+
+```sh
+sudo python3 /tmp/csoc-date-capsule-NEW/supervisor.py \
+  --capsule /tmp/csoc-date-capsule-NEW \
+  --expect-digest sha256:REVIEWED_DATE_PROPOSAL --execute
+```
+
+These placeholders are deliberately not an executable admitted proposal. For an
+uncertain attempt, run its frozen root `capsule/supervisor.py` with that capsule,
+the same digest and `--recover`, then inspect the durable receipt and native
+postflight. Do not type the founder confirmation on the founder's behalf.
+The deployment-owner rule is recorded in the storage release review: “The agent
+cannot approve itself.” No timer, mail, grant, production restore, bulk import,
+apex change or Bubble retirement is authorized by this preparation.
+
+## Evidence
+
+`tests/test_source_date_supervisor.py` covers 19 focused lifecycle/refusal cases.
+`scripts/rehearse-source-date-supervisor.py` uses the real prepared capsule,
+supervisor, helper protocol and operation against a disposable clone of the
+independently recovered PostgreSQL/content pair. It exercises normal completion
+and a lost reply after the actual SQL/file commit, then explicit reconciliation
+without another capture or apply. All 13 tables and 82 files are compared; the
+retained original is unchanged and the clone is removed.
+
+The companion `source-date-supervisor-rehearsal-2026-10-30.json` binds reviewed
+source hashes and records the boundaries: Kubernetes/root fencing and provider
+capture are simulated, and the native observer was not executed. No new live
+backup, production capsule, runtime admission or productive correction is claimed.
+The full `make check` gate passes 383 cases, including mandatory browser journeys,
+with zero failures, errors or skips; Ruff and whitespace checks pass.

scripts/observe-source-date-target.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+"""Read-only native observation and private capsule preparation, after storage release.
+
+Run on the content node. No backup, scale, pod creation, schema/date update,
+credential delivery or timer activation is performed. Server dry-run is used.
+"""
+
+import argparse
+import base64
+import json
+import os
+import runpy
+import tempfile
+from pathlib import Path
+
+from source_date_supervisor import (
+    CLAIM,
+    NS,
+    Capsule,
+    Node,
+    Refused,
+    Supervisor,
+    canonical,
+    sha,
+    sync_directory,
+)
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def observe(plan, output, *, image, cluster_uid, deployment_uid, pvc_uid, node=None):
+    if os.geteuid() != 0:
+        raise Refused("native_node_root_read_context_required")
+    output = output.absolute()
+    if (output.exists() or any(p.is_symlink() for p in [output, *output.parents])
+            or output.resolve().is_relative_to(ROOT)
+            or not output.resolve().is_relative_to(Path("/tmp"))):
+        raise Refused("fresh_private_tmp_output_outside_checkout_required")
+    node = node or Node()
+    dep = node.doc("get", "deployment", NS, "-n", NS)
+    spec = dep["spec"]
+    if (dep["metadata"]["uid"] != deployment_uid or spec["replicas"] != 1
+            or len(spec["template"]["spec"]["containers"]) != 1
+            or spec["template"]["spec"]["containers"][0]["image"] != image):
+        raise Refused("storage_release_or_target_prerequisite_missing")
+    if node.doc("get", "namespace", "kube-system")["metadata"]["uid"] != cluster_uid:
+        raise Refused("cluster_changed")
+    pvc = node.doc("get", "pvc", CLAIM, "-n", NS)
+    if pvc["metadata"]["uid"] != pvc_uid or pvc["status"]["phase"] != "Bound":
+        raise Refused("pvc_changed")
+    pods = [p for p in node.doc("get", "pods", "-n", NS)["items"]
+            if p.get("status", {}).get("phase") not in {"Succeeded", "Failed"}
+            and any(v.get("persistentVolumeClaim", {}).get("claimName") == CLAIM
+                    for v in p["spec"].get("volumes", []))]
+    if len(pods) != 1:
+        raise Refused("single_application_pod_required")
+    host = node.doc("get", "node", pods[0]["spec"]["nodeName"])
+    node.verify_local_node(host)
+    pins = node.installation_pins()
+    history = json.loads(node.command(["helm", "history", NS, "-n", NS, "-o", "json"]))
+    if not history or history[-1]["status"] != "deployed":
+        raise Refused("stable_helm_release_required")
+    target = {"cluster_uid": cluster_uid, "deployment_uid": deployment_uid, "pvc_uid": pvc_uid,
+              "node_name": host["metadata"]["name"], "node_uid": host["metadata"]["uid"],
+              "image": image, "deployment_spec_sha256": sha(canonical(spec)),
+              "helm_revision": history[-1]["revision"],
+              "helm_manifest_sha256": sha(node.command(["helm", "get", "manifest", NS, "-n", NS])),
+              **pins}
+    modules = {name: (ROOT / "coulomb_social/apps/spaces" / name).read_bytes() for name in
+               ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py")}
+    payload = {"mode": "probe", "database_name": "coulomb_social_db",
+               "modules": {n: base64.b64encode(v).decode() for n, v in modules.items()},
+               "module_hashes": {n: sha(v) for n, v in modules.items()}}
+    baseline = json.loads(node.command(
+        ["kubectl", "exec", "-i", "-n", NS, pods[0]["metadata"]["name"], "--", "env",
+         "PYTHONDONTWRITEBYTECODE=1", "python", "-c",
+         (ROOT / "scripts/source_date_wire.py").read_text()], canonical(payload)))["fingerprint"]
+    prepare = runpy.run_path(str(ROOT / "scripts/prepare-source-date-capsule.py"))["prepare"]
+    with tempfile.TemporaryDirectory(prefix="csoc-native-date-observation-") as temporary:
+        work = Path(temporary)
+        (work / "baseline.json").write_bytes(canonical(baseline))
+        (work / "target.json").write_bytes(canonical(target))
+        prepared = prepare(plan, work / "baseline.json", work / "target.json", work / "capsule")
+        capsule = Capsule(work / "capsule", prepared["proposal_digest"])
+        check = Supervisor(capsule, node)
+        _, proof = check.preflight()
+        output.mkdir(mode=0o700)
+        # Publish exactly the already checked capsule, retaining its attempt id.
+        for name in [*capsule.content, "manifest.json"]:
+            data = (work / "capsule" / name).read_bytes()
+            fd = os.open(output / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            with os.fdopen(fd, "wb") as stream:
+                stream.write(data)
+                stream.flush()
+                os.fsync(stream.fileno())
+        sync_directory(output)
+        sync_directory(output.parent)
+    return {**prepared, "status": "native_read_only_capsule_prepared_not_admitted",
+            "baseline_rechecked": proof["fingerprint"] == baseline,
+            "helper_server_dry_run_sha256": check.preview_helper_spec,
+            "capture_performed": False, "live_changes": False}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--plan", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--image", required=True)
+    parser.add_argument("--cluster-uid", required=True)
+    parser.add_argument("--deployment-uid", required=True)
+    parser.add_argument("--pvc-uid", required=True)
+    args = parser.parse_args()
+    print(json.dumps(observe(args.plan, args.output, image=args.image, cluster_uid=args.cluster_uid,
+                             deployment_uid=args.deployment_uid, pvc_uid=args.pvc_uid)))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        print(json.dumps({"status": "observation_refused", "failure_type": type(exc).__name__}))
+        raise SystemExit(1)

scripts/prepare-source-date-capsule.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Seal existing private date plan, current target pins and complete-store baseline.
+
+Offline preparation only. It creates no approval and never contacts the cluster.
+Acquire the target/baseline AFTER the separately supervised storage release.
+"""
+
+import argparse
+import json
+import os
+import uuid
+from pathlib import Path
+
+from source_date_supervisor import Capsule, canonical, sha, sync_directory
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def prepare(plan, baseline, target, output):
+    output = output.absolute()
+    if (output.exists() or any(p.is_symlink() for p in [output, *output.parents])
+            or output.resolve().is_relative_to(ROOT)):
+        raise ValueError("Fresh private destination outside checkout required")
+    sources = {
+        "supervisor.py": ROOT / "scripts/source_date_supervisor.py",
+        "wire.py": ROOT / "scripts/source_date_wire.py",
+        **{name: ROOT / "coulomb_social/apps/spaces" / name for name in
+           ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py")},
+        "plan.json": plan,
+    }
+    if any(p.is_symlink() or not p.is_file() or p.resolve().is_relative_to(output)
+           for p in [*sources.values(), baseline, target]):
+        raise ValueError("Regular existing inputs required")
+    content = {name: path.read_bytes() for name, path in sources.items()}
+    manifest = {"schema": "coulomb.supervised-date-capsule.v1",
+                "attempt_id": uuid.uuid4().hex,
+                "target": json.loads(target.read_bytes()),
+                "baseline": json.loads(baseline.read_bytes()),
+                "files": {name: sha(data) for name, data in content.items()},
+                "requires_founder_attended_execution": True,
+                "approval_status": "prepared_not_admitted"}
+    digest = "sha256:" + sha(canonical(manifest))
+    # A private temporary validation directory avoids publishing an invalid or
+    # partial capsule under the requested final name.
+    import tempfile
+
+    with tempfile.TemporaryDirectory(prefix="csoc-date-capsule-") as temporary:
+        stage = Path(temporary)
+        for name, data in {**content, "manifest.json": canonical(manifest)}.items():
+            (stage / name).write_bytes(data)
+        Capsule(stage, digest)
+    output.mkdir(mode=0o700)
+    for name, data in {**content, "manifest.json": canonical(manifest)}.items():
+        fd = os.open(output / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(fd, "wb") as stream:
+            stream.write(data)
+            stream.flush()
+            os.fsync(stream.fileno())
+    sync_directory(output)
+    sync_directory(output.parent)
+    return {"status": "prepared_not_admitted", "proposal_digest": digest,
+            "file_count": len(content) + 1, "production_access": False, "live_changes": False}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--plan", type=Path, required=True)
+    parser.add_argument("--baseline", type=Path, required=True)
+    parser.add_argument("--target", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    print(json.dumps(prepare(args.plan, args.baseline, args.target, args.output)))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        raise SystemExit("Date capsule preparation refused: " + type(exc).__name__)

scripts/rehearse-source-date-supervisor.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+"""Rehearse controller + exact helper wire on clones of retained recovered data.
+
+Kubernetes, root ownership and fresh provider capture are explicit simulations.
+The packaged Django/PostgreSQL/content operations and helper protocol are real.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import json
+import os
+import re
+import runpy
+import subprocess
+import tempfile
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--restore-receipt", type=Path, required=True)
+    parser.add_argument("--plan", type=Path, required=True)
+    parser.add_argument("--candidate-image", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    if args.output.exists() or args.output.is_symlink():
+        raise ValueError("Fresh output required")
+    previous = runpy.run_path(str(ROOT / "scripts/rehearse-source-date-journal.py"))
+    fixture = runpy.run_path(str(ROOT / "tests/test_source_date_supervisor.py"))
+    prepare = runpy.run_path(str(ROOT / "scripts/prepare-source-date-capsule.py"))["prepare"]
+    control, FakeNode = fixture["supervisor"], fixture["FakeNode"]
+    run, hashes = previous["PREVIOUS"]["run"], previous["PREVIOUS"]["hashes"]
+    raw_receipt, plan_bytes = args.restore_receipt.read_bytes(), args.plan.read_bytes()
+    state = json.loads(raw_receipt)
+    if (state["status"] != "restored_verified"
+            or not re.fullmatch(r"csoc-recovery-\d{14}", state["database_container"])
+            or state["network"] != state["database_container"] + "-net"):
+        raise ValueError("Retained isolated restore required")
+    net = json.loads(run(["docker", "network", "inspect", state["network"]]))[0]
+    db = json.loads(run(["docker", "inspect", state["database_container"]]))[0]
+    if (not net["Internal"] or set(net["Containers"]) != {db["Id"]}
+            or not db["State"]["Running"] or db["Image"] != state["postgres_image_id"]
+            or set(db["NetworkSettings"]["Networks"]) != {state["network"]}
+            or "recovery-db" not in db["NetworkSettings"]["Networks"][state["network"]]["Aliases"]
+            or any(db["NetworkSettings"]["Ports"].values())):
+        raise ValueError("Original recovery isolation changed")
+    image = run(["docker", "image", "inspect", args.candidate_image,
+                 "--format", "{{.Id}}"]).decode().strip()
+    if not re.fullmatch(r"sha256:[a-f0-9]{64}", image):
+        raise ValueError("Retained image required")
+    pinned_image = "forgejo.coulomb.social/coulomb/coulomb-social@" + image
+    root = Path(state["restored_directory"]) / "content"
+    if any(p.is_symlink() for p in [root, *root.parents]) or not root.resolve().is_relative_to(
+            Path("/tmp")):
+        raise ValueError("Temporary recovery directory required")
+    originals = hashes(root)
+    wire = (ROOT / "scripts/source_date_wire.py").read_bytes()
+    modules = {name: (ROOT / "coulomb_social/apps/spaces" / name).read_bytes() for name in
+               ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py")}
+    payload_modules = {"modules": {n: base64.b64encode(v).decode() for n, v in modules.items()},
+                       "module_hashes": {n: sha(v) for n, v in modules.items()}}
+    outcomes = []
+    for scenario in ("normal", "lost_apply_reply"):
+        clone = "csoc_date_" + uuid.uuid4().hex
+        with tempfile.TemporaryDirectory(prefix="csoc-date-supervisor-rehearsal-") as temporary:
+            import shutil
+
+            work = Path(temporary)
+            shutil.copytree(root, work / "content")
+            (work / "plan.json").write_bytes(plan_bytes)
+            (work / "probe").mkdir(mode=0o700)
+            for name, data in modules.items():
+                (work / name).write_bytes(data)
+
+            def application(database, code, payload=None, operation=None):
+                name = "csoc-date-supervisor-" + uuid.uuid4().hex
+                command = ["docker", "run", "--rm", "-i", "--name", name,
+                           "--network", state["network"], "--read-only",
+                           "--tmpfs", "/tmp:size=33554432", "--cap-drop=ALL",
+                           "--security-opt=no-new-privileges", "--user",
+                           f"{os.getuid()}:{os.getgid()}",
+                           "--entrypoint", "/app/.venv/bin/python",
+                           "-e", "DJANGO_SETTINGS_MODULE=coulomb_social.settings.base",
+                           "-e", f"DATABASE_URL=postgres://postgres@recovery-db/{database}",
+                           "-e", "CONTENT_ROOT=/app/var/content", "-e", "FORGEJO_BASE_URL=",
+                           "-e", "PYTHONDONTWRITEBYTECODE=1",
+                           "-v", str(work / "content") + ":/app/var/content",
+                           "-v", str(work / "content") + ":/rehearsal-content:ro",
+                           "-v", str(work / "plan.json") + ":/review/plan.json:ro",
+                           "-v", str(operation or work / "probe") + ":/operation"]
+                for member in modules:
+                    command += ["-v", str(work / member)
+                                + ":/app/coulomb_social/apps/spaces/" + member + ":ro"]
+                command += [image, "-c", code]
+                try:
+                    result = subprocess.run(command, input=payload, capture_output=True,
+                                            timeout=210)
+                finally:
+                    subprocess.run(["docker", "rm", "-f", name], capture_output=True,
+                                   timeout=30, check=False)
+                if result.returncode:
+                    fd, diagnostic = tempfile.mkstemp(prefix="csoc-supervisor-private-diagnostics-")
+                    with os.fdopen(fd, "wb") as stream:
+                        stream.write(result.stdout + result.stderr)
+                    print(json.dumps({"private_diagnostics": diagnostic}))
+                    raise ValueError("Private helper refused")
+                rows = [json.loads(line) for line in result.stdout.splitlines()
+                        if line.startswith(b"{")]
+                if len(rows) != 1:
+                    raise ValueError("Single helper result required")
+                return rows[0]
+
+            before = application("coulomb_social_restore", previous["SNAPSHOT"])
+            run(["docker", "exec", state["database_container"], "createdb", "-U", "postgres",
+                 "-T", "coulomb_social_restore", clone])
+            try:
+                migration = ("import django; django.setup(); "
+                             "from django.db import connection; "
+                             "from django.db.migrations.executor import MigrationExecutor; "
+                             "MigrationExecutor(connection).migrate(["
+                             "('spaces','0003_space_slug_global_uniq')]);\n")
+                baseline = application(clone, migration + wire.decode(), control.canonical({
+                    **payload_modules, "mode": "probe", "database_name": clone}))["fingerprint"]
+
+                class NativeHelper(FakeNode):
+                    def command(self, arguments, payload=None, **options):
+                        if arguments[:2] != ["kubectl", "exec"]:
+                            return super().command(arguments, payload, **options)
+                        message = json.loads(payload)
+                        # Translate the synthetic Kubernetes transport to the
+                        # actual isolated clone; live Capsule.payload always pins
+                        # coulomb_social_db. No production endpoint is accepted.
+                        message["database_name"] = clone
+                        private = self.supervisor.directory / "private"
+                        result = application(clone, arguments[-1], control.canonical(message),
+                                             private if private.exists() else None)
+                        if message["mode"] == "apply":
+                            self.apply_count += 1
+                            if scenario == "lost_apply_reply":
+                                self.fail = "apply_reply"
+                                self.trip("apply_reply")
+                        return control.canonical(result)
+
+                    def capture(self, target):
+                        self.capture_count += 1
+                        now = datetime.now(UTC).isoformat()
+                        key = ("platform-pg/application-archives/coulomb-social/"
+                               "synthetic-control.tar.age")
+                        return {"schema": "platform.application-primary-archive.v1",
+                                "application": "coulomb-social",
+                                "status": "offsite_fetched_verified",
+                                "captured_at": now, "completed_at": now, "image": pinned_image,
+                                "age_recipient": "age1synthetic", "immutable_create": True,
+                                "remote_retrieval_verified": True,
+                                "version_id": "synthetic-version",
+                                "object_key": key, "destination": "s3://railiance-platform-pg-backup/"
+                                + key, "encrypted_sha256": "a" * 64, "ciphertext_sha256": "a" * 64,
+                                "database_sha256": "b" * 64, "content_sha256": "c" * 64,
+                                "encrypted_bytes": 1, "ciphertext_bytes": 1,
+                                "capture_and_provider_simulated": True}
+
+                node = NativeHelper()
+                node.dep["spec"]["template"]["spec"]["containers"][0]["image"] = pinned_image
+                target = {"cluster_uid": "cluster-id", "deployment_uid": "deployment-id",
+                          "node_uid": "node-id", "node_name": "synthetic-node", "pvc_uid": "pvc-id",
+                          "image": pinned_image, "deployment_spec_sha256": sha(control.canonical(
+                              node.dep["spec"])), "helm_revision": 18,
+                          "helm_manifest_sha256": sha(b"synthetic helm manifest"),
+                          "public_recipient": "age1synthetic", "installed_helpers": {
+                              n: "f" * 64 for n in control.HELPERS}}
+                (work / "baseline.json").write_bytes(control.canonical(baseline))
+                (work / "target.json").write_bytes(control.canonical(target))
+                prepared = prepare(work / "plan.json", work / "baseline.json", work / "target.json",
+                                   work / "capsule")
+                capsule = control.Capsule(work / "capsule", prepared["proposal_digest"])
+                operation = control.Supervisor(capsule, node, state_root=work / "state")
+                node.supervisor = operation
+                try:
+                    result = operation.execute()
+                except control.Refused:
+                    if scenario != "lost_apply_reply":
+                        raise
+                    assert node.dep["spec"]["replicas"] == 0
+                    result = operation.recover()
+                assert result["phase"] == "completed" and result["correction_status"] == "completed"
+                assert node.capture_count == node.apply_count == 1
+                assert node.dep["spec"]["replicas"] == 1 and not node.helper
+                after = application("coulomb_social_restore", previous["SNAPSHOT"])
+                if before["tables_sha256"] != after["tables_sha256"] or hashes(root) != originals:
+                    raise ValueError("Original recovery source changed")
+                outcomes.append({"scenario": scenario, "status": "settled",
+                                 "actual_helper_wire_and_data_operation": True,
+                                 "capture_calls": 1, "apply_calls": 1,
+                                 "all_tables_compared": baseline["tables"],
+                                 "all_files_compared": baseline["files"]})
+            finally:
+                run(["docker", "exec", state["database_container"], "dropdb", "-U", "postgres",
+                     "--force", clone])
+    proof = {"schema": "coulomb.supervised-date-isolated-rehearsal.v1",
+             "status": "controller_and_real_helper_rehearsed", "scenarios": outcomes,
+             "kubernetes_and_root_operations_simulated": True,
+             "capture_and_provider_simulated": True, "production_access": False,
+             "native_observer_not_executed": True,
+             "production_writes": False, "live_fence_or_native_backup_proved": False,
+             "source_overlay_used": True, "candidate_image_id": image,
+             "retained_recovery_source_unchanged": True, "disposable_sql_clones_removed": True,
+             "sources_sha256": {p: sha((ROOT / p).read_bytes()) for p in (
+                 "scripts/source_date_supervisor.py", "scripts/source_date_wire.py",
+                 "scripts/prepare-source-date-capsule.py",
+                 "scripts/observe-source-date-target.py",
+                 "scripts/rehearse-source-date-supervisor.py",
+                 "tests/test_source_date_supervisor.py")},
+             "modules_sha256": {n: sha(v) for n, v in modules.items()},
+             "private_plan_sha256": sha(plan_bytes), "restore_receipt_sha256": sha(raw_receipt)}
+    fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(fd, "w") as stream:
+        json.dump(proof, stream, indent=2)
+        stream.write("\n")
+    print(json.dumps(proof))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (ValueError, OSError, KeyError, subprocess.SubprocessError):
+        raise SystemExit("Supervisor rehearsal refused; inspect private diagnostics locally.")

scripts/source_date_supervisor.py

@@ -0,0 +1,662 @@
+"""Node supervisor for an exact founder-reviewed date-correction capsule.
+
+Default is read-only. This tool is not an approval service: only the founder may
+run execute/recover in their terminal under the deployment owner's supervision.
+Uncertain attempts retain their helper, journals and stopped application.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import copy
+import fcntl
+import hashlib
+import json
+import os
+import re
+import stat
+import subprocess
+import sys
+import tempfile
+from contextlib import ExitStack, contextmanager
+from datetime import UTC, datetime
+from pathlib import Path
+
+NS = "coulomb-social"
+CLAIM = "coulomb-social-content"
+STATE_ROOT = Path("/var/lib/coulomb-social-backup/date-operations")
+MEMBERS = {"supervisor.py", "wire.py", "date_backfill.py", "date_backfill_journal.py",
+           "date_backfill_operation.py", "plan.json"}
+HELPERS = {"/opt/coulomb-social-backup/paired-backup.py",
+           "/opt/coulomb-social-backup/snapshot-content-and-db.py",
+           "/opt/railiance-platform/scripts/primary_application_archive.py"}
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode()
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+class Refused(RuntimeError):
+    pass
+
+
+def private_file(path):
+    info = path.lstat()
+    if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077
+            or info.st_nlink != 1 or any(p.is_symlink() for p in path.parents)):
+        raise Refused("private_root_file_required")
+    return path.read_bytes()
+
+
+def sync_directory(path):
+    descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def save(path, value):
+    fd, temporary = tempfile.mkstemp(prefix=".state-", dir=path.parent)
+    try:
+        with os.fdopen(fd, "wb") as stream:
+            stream.write(canonical(value))
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+class Capsule:
+    def __init__(self, path, expected):
+        self.manifest = json.loads((path / "manifest.json").read_bytes())
+        if "sha256:" + sha(canonical(self.manifest)) != expected:
+            raise Refused("proposal_digest_changed")
+        self.digest = expected
+        if (self.manifest.get("schema") != "coulomb.supervised-date-capsule.v1"
+                or set(self.manifest["files"]) != MEMBERS
+                or self.manifest.get("requires_founder_attended_execution") is not True
+                or not re.fullmatch(r"[a-f0-9]{32}", self.manifest.get("attempt_id", ""))):
+            raise Refused("exact_capsule_members_required")
+        self.content = {}
+        for name, digest in self.manifest["files"].items():
+            member = path / name
+            if not member.is_file() or any(p.is_symlink() for p in [member, *member.parents]):
+                raise Refused("capsule_links_refused")
+            data = member.read_bytes()
+            if sha(data) != digest:
+                raise Refused("capsule_member_changed")
+            self.content[name] = data
+        self.target = self.manifest["target"]
+        fields = {"cluster_uid", "deployment_uid", "pvc_uid", "node_name", "node_uid", "image",
+                  "deployment_spec_sha256", "helm_revision", "helm_manifest_sha256",
+                  "public_recipient", "installed_helpers"}
+        if (set(self.target) != fields or set(self.target["installed_helpers"]) != HELPERS
+                or not re.fullmatch(r"[a-z0-9.-]+", self.target["node_name"])
+                or not re.fullmatch(r"forgejo\.coulomb\.social/coulomb/"
+                                    r"coulomb-social@sha256:[a-f0-9]{64}", self.target["image"])):
+            raise Refused("exact_application_target_required")
+        hashes = [self.target[n] for n in ("deployment_spec_sha256", "helm_manifest_sha256")]
+        hashes += list(self.target["installed_helpers"].values())
+        if (any(not re.fullmatch(r"[a-f0-9]{64}", h) for h in hashes)
+                or type(self.target["helm_revision"]) is not int or self.target["helm_revision"] < 1
+                or not self.target["public_recipient"].startswith("age1")):
+            raise Refused("exact_target_pins_required")
+        self.plan = json.loads(self.content["plan.json"])
+        self.baseline = self.manifest["baseline"]
+        if (self.plan.get("schema") != "coulomb.source-date-backfill-plan.v1"
+                or self.plan.get("blockers") or not self.plan.get("prepared_change_set_valid")
+                or set(self.baseline) != {"tables_sha256", "files_sha256",
+                                         "tables", "rows", "files"}
+                or any(not re.fullmatch(r"[a-f0-9]{64}", self.baseline[n])
+                       for n in ("tables_sha256", "files_sha256"))
+                or any(type(self.baseline[n]) is not int or self.baseline[n] < 1
+                       for n in ("tables", "rows", "files"))):
+            raise Refused("exact_reviewed_date_plan_and_baseline_required")
+
+    def payload(self, mode, backup=None):
+        names = ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py")
+        return {"mode": mode, "plan": self.plan, "baseline": self.baseline,
+                "database_name": "coulomb_social_db",
+                "image": self.target["image"], "recipient": self.target["public_recipient"],
+                "backup": backup,
+                "modules": {n: base64.b64encode(self.content[n]).decode() for n in names},
+                "module_hashes": {n: self.manifest["files"][n] for n in names}}
+
+
+class Node:
+    def __init__(self):
+        self.env = {**os.environ, "KUBECONFIG": "/etc/rancher/k3s/k3s.yaml"}
+
+    def command(self, arguments, payload=None, *, env=None, timeout=210):
+        with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr:
+            process = subprocess.Popen(arguments,
+                                       stdin=subprocess.PIPE if payload else subprocess.DEVNULL,
+                                       stdout=stdout, stderr=stderr, env=env or self.env)
+            try:
+                process.communicate(payload, timeout=timeout)
+            except BaseException:
+                process.terminate()
+                try:
+                    process.wait(timeout=400)  # permit admitted backup's resume cleanup
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait()
+                raise Refused("child_outcome_requires_review") from None
+            stdout.seek(0)
+            data = stdout.read(8 * 1024 * 1024 + 1)
+            if process.returncode or len(data) > 8 * 1024 * 1024:
+                raise Refused("private_child_refused")
+            return data
+
+    def doc(self, *arguments):
+        return json.loads(self.command(["kubectl", *arguments, "-o", "json"]))
+
+    def verify_local_node(self, node):
+        machine = Path("/etc/machine-id").read_text().strip()
+        if not machine or machine != node["status"]["nodeInfo"]["machineID"]:
+            raise Refused("run_on_the_pinned_content_node")
+
+    def installation_pins(self):
+        helpers = {}
+        for name in HELPERS:
+            path = Path(name)
+            info = path.lstat()
+            if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022
+                    or info.st_nlink != 1 or any(p.is_symlink() for p in path.parents)):
+                raise Refused("installed_backup_helper_changed")
+            helpers[name] = sha(path.read_bytes())
+        fields = dict(line.split("=", 1) for line in private_file(
+            Path("/etc/coulomb-social-backup/recipient.env")).decode().splitlines()
+            if line and not line.startswith("#"))
+        if set(fields) != {"COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT"}:
+            raise Refused("public_recipient_environment_required")
+        return {"installed_helpers": helpers,
+                "public_recipient": fields["COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT"]}
+
+    def verify_backup_installation(self, target):
+        expected = {name: target[name] for name in ("installed_helpers", "public_recipient")}
+        if self.installation_pins() != expected:
+            raise Refused("installed_backup_configuration_changed")
+        for name in ("paired-backup", "backup-freshness"):
+            result = subprocess.run(["systemctl", "is-active", f"coulomb-social-{name}.timer"],
+                                    capture_output=True, timeout=15)
+            if result.returncode not in {3, 4}:
+                raise Refused("timer_state_requires_owner_review")
+
+    def capture(self, target):
+        env = dict(self.env)
+        allowed = {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY",
+                   "COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT"}
+        for name in ("primary.env", "recipient.env"):
+            raw = private_file(Path("/etc/coulomb-social-backup") / name)
+            for line in raw.decode().splitlines():
+                if line and not line.startswith("#"):
+                    key, value = line.split("=", 1)
+                    if key not in allowed:
+                        raise Refused("unexpected_backup_environment_field")
+                    env[key] = value
+        if env.get("COULOMB_SOCIAL_BACKUP_AGE_RECIPIENT") != target["public_recipient"]:
+            raise Refused("recipient_binding_changed")
+        point = json.loads(self.command(
+            ["python3", "/opt/coulomb-social-backup/paired-backup.py", "--execute"],
+            env=env, timeout=1200))
+        # Validate before scaling or executing any date code. The installed
+        # admitted helper owns the actual immutable publication/readback.
+        if (point.get("status") != "offsite_fetched_verified"
+                or point.get("remote_retrieval_verified") is not True
+                or point.get("immutable_create") is not True
+                or point.get("image") != target["image"]
+                or point.get("age_recipient") != target["public_recipient"]
+                or not point.get("version_id") or point["version_id"] == "null"):
+            raise Refused("fresh_native_capture_required")
+        name = point.get("local_archive", "")
+        if not re.fullmatch(r"\d{8}T\d{6}Z-[a-f0-9]{24}\.tar\.age", name):
+            raise Refused("own_local_ciphertext_required")
+        ciphertext = Path("/var/lib/coulomb-social-backup") / name
+        info = ciphertext.lstat()
+        if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077
+                or info.st_nlink != 1 or info.st_size != point["ciphertext_bytes"]):
+            raise Refused("private_complete_ciphertext_required")
+        with ciphertext.open("rb") as stream:
+            digest = hashlib.file_digest(stream, "sha256").hexdigest()
+        captured = datetime.fromisoformat(point["captured_at"].replace("Z", "+00:00"))
+        completed = datetime.fromisoformat(point["completed_at"].replace("Z", "+00:00"))
+        now = datetime.now(UTC)
+        prefix = "platform-pg/application-archives/coulomb-social/"
+        if (point.get("application") != NS or digest != point["ciphertext_sha256"]
+                or digest != point["encrypted_sha256"] or captured.tzinfo is None
+                or completed.tzinfo is None or not captured <= completed <= now
+                or (now - captured).total_seconds() > 600
+                or not point["object_key"].startswith(prefix)
+                or not point["object_key"].endswith("/" + name)
+                or point["destination"] != "s3://railiance-platform-pg-backup/"
+                + point["object_key"]):
+            raise Refused("fresh_native_ciphertext_binding_changed")
+        return point
+
+    def prepare_root(self, root):
+        root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        info = root.lstat()
+        if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077
+                or any(p.is_symlink() for p in root.parents)):
+            raise Refused("private_supervisor_directory_required")
+        sync_directory(root)
+        sync_directory(root.parent)
+
+    @contextmanager
+    def operation_lock(self, root):
+        self.prepare_root(root)
+        fd = os.open(root / "supervisor.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+        with os.fdopen(fd, "w"):
+            info = os.fstat(fd)
+            if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid()
+                    or info.st_mode & 0o077 or info.st_nlink != 1):
+                raise Refused("private_supervisor_lock_required")
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            yield
+
+    def lock_backups(self, stack):
+        for name in ("backup.lock", "snapshot.lock"):
+            path = Path("/var/lib/coulomb-social-backup") / name
+            private_file(path)
+            stream = stack.enter_context(path.open("r+"))
+            fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+    def helper_directory(self, path):
+        if not path.exists():
+            path.mkdir(mode=0o700)
+            os.chown(path, 10001, 10001)
+        info = path.lstat()
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != 10001 or info.st_mode & 0o077:
+            raise Refused("private_helper_journal_directory_required")
+        sync_directory(path)
+        sync_directory(path.parent)
+
+    def read_state(self, path):
+        return json.loads(private_file(path))
+
+
+class Supervisor:
+    def __init__(self, capsule, node, *, state_root=STATE_ROOT):
+        self.capsule, self.node = capsule, node
+        self.target, self.state_root = capsule.target, state_root
+        self.directory = state_root / capsule.digest.removeprefix("sha256:")
+        self.helper = "csoc-date-" + capsule.digest.removeprefix("sha256:")[:20]
+        self.state = None
+
+    def record(self, phase, **changes):
+        self.state.update(changes, phase=phase, updated_at=datetime.now(UTC).isoformat())
+        save(self.directory / "receipt.json", self.state)
+
+    def deployment(self, replicas):
+        dep = self.node.doc("get", "deployment", NS, "-n", NS)
+        spec = copy.deepcopy(dep["spec"])
+        if dep["metadata"]["uid"] != self.target["deployment_uid"] or spec["replicas"] != replicas:
+            raise Refused("deployment_identity_or_scale_changed")
+        spec["replicas"] = 1
+        if sha(canonical(spec)) != self.target["deployment_spec_sha256"]:
+            raise Refused("deployment_spec_changed")
+        if (len(spec["template"]["spec"]["containers"]) != 1
+                or spec["template"]["spec"]["containers"][0]["image"] != self.target["image"]
+                or spec["strategy"]["type"] != "Recreate"):
+            raise Refused("single_pinned_recreate_writer_required")
+        claims = [(v["name"], v["persistentVolumeClaim"]["claimName"])
+                  for v in spec["template"]["spec"].get("volumes", [])
+                  if "persistentVolumeClaim" in v]
+        if claims != [("content", CLAIM)]:
+            raise Refused("exact_content_claim_required")
+        return dep
+
+    def writers(self):
+        return [pod for pod in self.node.doc("get", "pods", "-n", NS)["items"]
+                if pod.get("status", {}).get("phase") not in {"Succeeded", "Failed"}
+                and any(v.get("persistentVolumeClaim", {}).get("claimName") == CLAIM
+                        for v in pod["spec"].get("volumes", []))]
+
+    def identity(self):
+        if self.node.doc("get", "namespace", "kube-system")["metadata"]["uid"] != self.target[
+                "cluster_uid"]:
+            raise Refused("cluster_changed")
+        node = self.node.doc("get", "node", self.target["node_name"])
+        self.node.verify_local_node(node)
+        if (node["metadata"]["uid"] != self.target["node_uid"] or node["spec"].get("unschedulable")
+                or not any(c["type"] == "Ready" and c["status"] == "True"
+                           for c in node["status"]["conditions"])):
+            raise Refused("node_changed_or_not_ready")
+        pvc = self.node.doc("get", "pvc", CLAIM, "-n", NS)
+        if pvc["metadata"]["uid"] != self.target["pvc_uid"] or pvc["status"]["phase"] != "Bound":
+            raise Refused("pvc_changed")
+        if self.node.doc("get", "hpa", "-n", NS)["items"]:
+            raise Refused("autoscaling_requires_owner_path")
+        if any(a.get("spec", {}).get("destination", {}).get("namespace") == NS for a in
+               self.node.doc("get", "applications.argoproj.io", "-A")["items"]):
+            raise Refused("gitops_requires_owner_path")
+        history = json.loads(self.node.command(["helm", "history", NS, "-n", NS, "-o", "json"]))
+        if (not history or history[-1]["status"] != "deployed"
+                or history[-1]["revision"] != self.target["helm_revision"]
+                or sha(self.node.command(["helm", "get", "manifest", NS, "-n", NS]))
+                != self.target["helm_manifest_sha256"]):
+            raise Refused("helm_release_changed")
+
+    def wire(self, pod, mode, backup=None):
+        return json.loads(self.node.command(
+            ["kubectl", "exec", "-i", "-n", NS, pod, "--", "env",
+             "PYTHONDONTWRITEBYTECODE=1", "python", "-c", self.capsule.content["wire.py"].decode()],
+            canonical(self.capsule.payload(mode, backup))))
+
+    def preflight(self):
+        self.identity()
+        dep = self.deployment(1)
+        pods = self.writers()
+        if (len(pods) != 1 or pods[0]["spec"].get("nodeName") != self.target["node_name"]
+                or pods[0].get("metadata", {}).get("deletionTimestamp")
+                or not any(c["type"] == "Ready" and c["status"] == "True"
+                           for c in pods[0]["status"].get("conditions", []))):
+            raise Refused("single_ready_application_writer_required")
+        containers = pods[0]["spec"].get("containers", [])
+        statuses = pods[0]["status"].get("containerStatuses", [])
+        if (len(containers) != 1 or containers[0]["image"] != self.target["image"]
+                or len(statuses) != 1 or not statuses[0].get("imageID", "").endswith(
+                    self.target["image"].split("@", 1)[1])):
+            raise Refused("actual_running_image_changed")
+        owner = pods[0]["metadata"]["ownerReferences"]
+        if len(owner) != 1 or owner[0]["kind"] != "ReplicaSet":
+            raise Refused("unexpected_application_writer")
+        rs = self.node.doc("get", "replicaset", owner[0]["name"], "-n", NS)
+        if not any(r["uid"] == self.target["deployment_uid"]
+                   for r in rs["metadata"]["ownerReferences"]):
+            raise Refused("unexpected_application_controller")
+        self.node.verify_backup_installation(self.target)
+        result = self.wire(pods[0]["metadata"]["name"], "prepare")
+        if (result["fingerprint"] != self.capsule.baseline
+                or not result["storage_constraint_verified"]):
+            raise Refused("reviewed_dataset_or_storage_prerequisite_changed")
+        requested = self.helper_manifest(dep)
+        dry = json.loads(self.node.command(["kubectl", "create", "--dry-run=server",
+                                           "-f", "-", "-o", "json"], canonical(requested)))
+        self.validate_helper(dry["spec"], requested["spec"])
+        self.preview_helper_spec = sha(canonical(dry["spec"]))
+        return dep, result
+
+    def validate_helper(self, actual, requested):
+        if (len(actual.get("containers", [])) != 1 or actual.get("initContainers")
+                or actual.get("hostNetwork") or actual.get("hostPID") or actual.get("hostIPC")
+                or actual.get("automountServiceAccountToken") is not False
+                or actual.get("nodeName") != self.target["node_name"]
+                or actual.get("securityContext") != requested["securityContext"]
+                or actual.get("volumes") != requested["volumes"]):
+            raise Refused("unreviewed_helper_pod_admission")
+        container = actual["containers"][0]
+        for key in ("image", "command", "env", "envFrom", "securityContext", "resources",
+                    "volumeMounts"):
+            if container.get(key) != requested["containers"][0].get(key):
+                raise Refused("unreviewed_helper_container_admission")
+        if container.get("args") or container.get("lifecycle"):
+            raise Refused("unreviewed_helper_process_admission")
+
+    def create_helper(self, dep):
+        self.node.helper_directory(self.directory / "private")
+        helper = self.helper_manifest(dep)
+        dry = json.loads(self.node.command(["kubectl", "create", "--dry-run=server",
+                                           "-f", "-", "-o", "json"], canonical(helper)))
+        self.validate_helper(dry["spec"], helper["spec"])
+        digest = sha(canonical(dry["spec"]))
+        if self.state.get("helper_spec_sha256", digest) != digest:
+            raise Refused("helper_admission_changed")
+        self.record("helper_create_started", helper_spec_sha256=digest, helper_uid=None)
+        created = json.loads(self.node.command(
+            ["kubectl", "create", "-f", "-", "-o", "json"], canonical(helper)))
+        self.record("helper_created", helper_uid=created["metadata"]["uid"])
+        self.node.command(["kubectl", "wait", "--for=condition=Ready", f"pod/{self.helper}",
+                           "-n", NS, "--timeout=120s"])
+        self.assert_fenced(helper=True)
+
+    def helper_manifest(self, dep):
+        app = dep["spec"]["template"]["spec"]
+        container = app["containers"][0]
+        mounts = [v for v in container.get("volumeMounts", [])
+                  if v["mountPath"] == "/app/var/content"]
+        if len(mounts) != 1 or mounts[0]["name"] != "content":
+            raise Refused("expected_content_mount_required")
+        return {"apiVersion": "v1", "kind": "Pod", "metadata": {
+            "name": self.helper, "namespace": NS,
+            "annotations": {"csoc-date-proposal": self.capsule.digest}}, "spec": {
+                "nodeName": self.target["node_name"], "restartPolicy": "Never",
+                "automountServiceAccountToken": False,
+                "securityContext": {"runAsNonRoot": True, "runAsUser": 10001,
+                                    "runAsGroup": 10001, "fsGroup": 10001,
+                                    "seccompProfile": {"type": "RuntimeDefault"}},
+                "imagePullSecrets": copy.deepcopy(app.get("imagePullSecrets", [])),
+                "containers": [{"name": "operation", "image": self.target["image"],
+                    "imagePullPolicy": "IfNotPresent",
+                    "command": ["python", "-c", "import time; time.sleep(1800)"],
+                    "envFrom": copy.deepcopy(container.get("envFrom", [])),
+                    "env": copy.deepcopy(container.get("env", [])),
+                    "securityContext": {"allowPrivilegeEscalation": False,
+                                        "readOnlyRootFilesystem": True,
+                                        "capabilities": {"drop": ["ALL"]}},
+                    "resources": {"requests": {"cpu": "10m", "memory": "64Mi"},
+                                  "limits": {"cpu": "100m", "memory": "256Mi"}},
+                    "volumeMounts": [{"name": "content", "mountPath": "/app/var/content"},
+                                     {"name": "operation", "mountPath": "/operation"},
+                                     {"name": "temporary", "mountPath": "/tmp"}]}],
+                "volumes": [{"name": "content", "persistentVolumeClaim": {"claimName": CLAIM}},
+                            {"name": "operation", "hostPath": {
+                                "path": str(self.directory / "private"), "type": "Directory"}},
+                            {"name": "temporary", "emptyDir": {
+                                "medium": "Memory", "sizeLimit": "32Mi"}}]}}
+
+    def assert_fenced(self, helper=False):
+        self.identity()
+        self.deployment(0)
+        writers = self.writers()
+        if helper:
+            if (len(writers) != 1 or writers[0]["metadata"]["name"] != self.helper
+                    or writers[0]["metadata"]["uid"] != self.state["helper_uid"]
+                    or sha(canonical(writers[0]["spec"])) != self.state["helper_spec_sha256"]):
+                raise Refused("helper_identity_or_writer_fence_changed")
+        elif writers:
+            raise Refused("application_writers_not_gone")
+
+    def finish(self, expected):
+        self.assert_fenced(helper=True)
+        if self.wire(self.helper, "probe")["fingerprint"] != expected:
+            raise Refused("full_dataset_postflight_changed")
+        self.record("helper_delete_started", expected_final=expected)
+        options = {"apiVersion": "v1", "kind": "DeleteOptions", "gracePeriodSeconds": 30,
+                   "preconditions": {"uid": self.state["helper_uid"]}}
+        self.node.command(["kubectl", "delete", "--raw",
+                           f"/api/v1/namespaces/{NS}/pods/{self.helper}", "-f", "-"],
+                          canonical(options))
+        self.node.command(["kubectl", "wait", "--for=delete", f"pod/{self.helper}", "-n", NS,
+                           "--timeout=120s"])
+        self.assert_fenced()
+        dep = self.deployment(0)
+        self.record("resume_started")
+        self.node.command(["kubectl", "scale", f"deployment/{NS}", "-n", NS,
+                           "--replicas=1", "--current-replicas=0",
+                           "--resource-version=" + dep["metadata"]["resourceVersion"]])
+        self.node.command(["kubectl", "rollout", "status", f"deployment/{NS}", "-n", NS,
+                           "--timeout=180s"])
+        self.deployment(1)
+        self.record("completed", application_resumed=True)
+        return self.state
+
+    def execute(self):
+        with self.node.operation_lock(self.state_root):
+            return self._execute()
+
+    def _execute(self):
+        if self.directory.exists():
+            raise Refused("prior_attempt_requires_recovery_not_replay")
+        dep, before = self.preflight()
+        self.directory.mkdir(mode=0o700, parents=True)
+        sync_directory(self.state_root)
+        self.state = {"schema": "coulomb.supervised-date-attempt.v1",
+                      "proposal_digest": self.capsule.digest, "application_resumed": False,
+                      "pid": os.getpid(), "expected_after": before["expected_after"],
+                      "helper_spec_sha256": self.preview_helper_spec}
+        frozen = self.directory / "capsule"
+        frozen.mkdir(mode=0o700)
+        for name, data in {**self.capsule.content,
+                           "manifest.json": canonical(self.capsule.manifest)}.items():
+            fd = os.open(frozen / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            with os.fdopen(fd, "wb") as stream:
+                stream.write(data)
+                stream.flush()
+                os.fsync(stream.fileno())
+        sync_directory(frozen)
+        self.record("capture_started")
+        try:
+            point = self.node.capture(self.target)
+            self.record("captured", backup=point)
+            _, after = self.preflight()
+            if after != before or self.preview_helper_spec != self.state["helper_spec_sha256"]:
+                raise Refused("dataset_changed_across_capture")
+            with ExitStack() as locks:
+                self.node.lock_backups(locks)
+                dep = self.deployment(1)
+                self.record("fencing")
+                self.node.command(["kubectl", "scale", f"deployment/{NS}", "-n", NS,
+                                   "--replicas=0", "--current-replicas=1",
+                                   "--resource-version=" + dep["metadata"]["resourceVersion"]])
+                self.node.command(["kubectl", "wait", "--for=delete", "pod", "-n", NS,
+                                   "-l", "app.kubernetes.io/instance=" + NS, "--timeout=120s"])
+                self.assert_fenced()
+                self.create_helper(dep)
+                if self.wire(self.helper, "prepare") != before:
+                    raise Refused("dataset_changed_before_correction")
+                self.record("apply_started")
+                result = self.wire(self.helper, "apply", point)
+                if (result.get("status") != "completed" or not result.get("whole_store_postflight")
+                        or result.get("fingerprint") != before["expected_after"]):
+                    raise Refused("date_result_unconfirmed")
+                self.record("applied", operation=result, correction_status="completed")
+                return self.finish(before["expected_after"])
+        except BaseException as exc:
+            self.record("requires_review", failed_phase=self.state["phase"],
+                        failure_reason=str(exc) if isinstance(exc, Refused) else type(exc).__name__)
+            raise
+
+    def recover(self):
+        """Explicit attended reconciliation; never captures or replays apply."""
+        with self.node.operation_lock(self.state_root), ExitStack() as locks:
+            self.node.lock_backups(locks)
+            self.state = self.node.read_state(self.directory / "receipt.json")
+            if (self.state.get("schema") != "coulomb.supervised-date-attempt.v1"
+                    or self.state["proposal_digest"] != self.capsule.digest):
+                raise Refused("original_attempt_required")
+            self.identity()
+            dep = self.node.doc("get", "deployment", NS, "-n", NS)
+            if dep["spec"]["replicas"] == 1:
+                # A lost response after scale-up must never undo later writes.
+                if (self.state.get("failed_phase", self.state["phase"]) != "resume_started"
+                        or "expected_final" not in self.state):
+                    raise Refused("running_application_requires_owner_review")
+                self.deployment(1)
+                self.node.command(["kubectl", "rollout", "status", f"deployment/{NS}",
+                                   "-n", NS, "--timeout=180s"])
+                self.record("completed", application_resumed=True,
+                            resumed_reply_reconciled=True, post_resume_writes_preserved=True)
+                return self.state
+            self.deployment(0)
+            writers = self.writers()
+            if not writers:
+                # Lost deletion reply or interruption before helper creation:
+                # recreate only the same pinned helper against retained state.
+                terminal = [p for p in self.node.doc("get", "pods", "-n", NS)["items"]
+                            if p["metadata"]["name"] == self.helper]
+                if terminal:
+                    old = terminal[0]
+                    if (len(terminal) != 1 or old.get("status", {}).get("phase") not in {
+                            "Succeeded", "Failed"}
+                            or sha(canonical(old["spec"])) != self.state.get("helper_spec_sha256")
+                            or old["metadata"].get("annotations", {}).get("csoc-date-proposal")
+                            != self.capsule.digest
+                            or (self.state.get("helper_uid") and self.state["helper_uid"]
+                                != old["metadata"]["uid"])):
+                        raise Refused("terminal_helper_identity_requires_review")
+                    self.record("terminal_helper_delete_started", helper_uid=old["metadata"]["uid"])
+                    self.node.command(["kubectl", "delete", "--raw",
+                                       f"/api/v1/namespaces/{NS}/pods/{self.helper}", "-f", "-"],
+                                      canonical({"apiVersion": "v1", "kind": "DeleteOptions",
+                                                 "gracePeriodSeconds": 30,
+                                                 "preconditions": {
+                                                     "uid": self.state["helper_uid"]}}))
+                    self.node.command(["kubectl", "wait", "--for=delete", f"pod/{self.helper}",
+                                       "-n", NS, "--timeout=120s"])
+                self.create_helper(dep)
+                writers = self.writers()
+            if (len(writers) != 1 or writers[0]["metadata"]["name"] != self.helper
+                    or sha(canonical(writers[0]["spec"])) != self.state.get("helper_spec_sha256")):
+                raise Refused("original_helper_requires_owner_review")
+            if (self.state.get("helper_uid")
+                    and self.state["helper_uid"] != writers[0]["metadata"]["uid"]):
+                raise Refused("helper_replaced_requires_owner_review")
+            self.state["helper_uid"] = writers[0]["metadata"]["uid"]
+            self.assert_fenced(helper=True)
+            observed = self.wire(self.helper, "inspect", self.state.get("backup"))
+            status = observed.get("status")
+            if status == "operation_busy":
+                raise Refused("original_operation_still_running")
+            if status == "pending":
+                self.record("reconcile_started")
+                observed = self.wire(self.helper, "reconcile", self.state["backup"])
+                status = observed.get("status")
+            expected = (self.state["expected_after"] if status == "completed"
+                        else self.capsule.baseline)
+            if status not in {"completed", "rolled_back", "absent"} or observed.get(
+                    "fingerprint") != expected:
+                raise Refused("reconciled_dataset_requires_owner_review")
+            self.record("reconciled", correction_status=status)
+            return self.finish(expected)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--capsule", type=Path, required=True)
+    parser.add_argument("--expect-digest", required=True)
+    modes = parser.add_mutually_exclusive_group()
+    modes.add_argument("--execute", action="store_true")
+    modes.add_argument("--recover", action="store_true")
+    args = parser.parse_args()
+    capsule = Capsule(args.capsule, args.expect_digest)
+    if sha(Path(__file__).read_bytes()) != capsule.manifest["files"]["supervisor.py"]:
+        raise Refused("run_the_reviewed_supervisor_source")
+    if os.geteuid() != 0:
+        raise Refused("node_root_context_required")
+    supervisor = Supervisor(capsule, Node())
+    if args.execute or args.recover:
+        if not sys.stdin.isatty():
+            raise Refused("founder_attended_terminal_required")
+        action = "RECOVER-DATE" if args.recover else "APPLY-DATE"
+        if input("Type " + action + " " + capsule.digest + " to continue: ").strip() != (
+                action + " " + capsule.digest):
+            raise Refused("not_confirmed")
+        print(json.dumps(supervisor.recover() if args.recover else supervisor.execute()))
+    else:
+        _, proof = supervisor.preflight()
+        print(json.dumps({"status": "read_only_preflight", "proposal_digest": capsule.digest,
+                          "before": proof, "live_changes": False}))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        print(json.dumps({"status": "refused", "failure_type": type(exc).__name__,
+                          "reason": str(exc) if isinstance(exc, Refused) else type(exc).__name__}))
+        raise SystemExit(1)

scripts/source_date_wire.py

@@ -0,0 +1,139 @@
+"""Pinned helper protocol. Stdin/private journals hold data; stdout is metadata only."""
+
+import base64
+import copy
+import fcntl
+import hashlib
+import json
+import os
+import signal
+import stat
+import sys
+import tempfile
+import types
+from pathlib import Path
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode()
+
+
+def run(payload):
+    import django
+
+    django.setup()
+    for filename in ("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py"):
+        source = base64.b64decode(payload["modules"][filename], validate=True)
+        if hashlib.sha256(source).hexdigest() != payload["module_hashes"][filename]:
+            raise ValueError("Reviewed module changed")
+        name = "coulomb_social.apps.spaces." + filename[:-3]
+        module = types.ModuleType(name)
+        module.__package__ = "coulomb_social.apps.spaces"
+        module.__file__ = "<reviewed-" + filename + ">"
+        sys.modules[name] = module
+        exec(compile(source, module.__file__, "exec"), module.__dict__)
+
+    from django.conf import settings
+    from django.db import connection
+
+    from coulomb_social.apps.spaces.date_backfill import prepare_index_date_patch
+    from coulomb_social.apps.spaces.date_backfill_operation import (
+        _set_date,
+        _snapshot,
+        execute_date_operation,
+        fingerprint,
+    )
+
+    if (connection.vendor != "postgresql" or settings.CONTENT_ROOT != "/app/var/content"
+            or connection.settings_dict["NAME"] != payload["database_name"]):
+        raise ValueError("Pinned application database/content context required")
+    with connection.cursor() as cursor:
+        cursor.execute("SET lock_timeout='5s'")
+        cursor.execute("SET statement_timeout='60s'")
+        cursor.execute("SELECT conname FROM pg_constraint "
+                       "WHERE conrelid='public.spaces_space'::regclass")
+        constraints = {r[0] for r in cursor.fetchall()}
+        cursor.execute("SELECT count(*) FROM django_migrations WHERE app='spaces' "
+                       "AND name='0003_space_slug_global_uniq'")
+        migrated = cursor.fetchone()[0] == 1
+    if "spaces_space_slug_global_uniq" not in constraints or not migrated:
+        raise ValueError("Storage release prerequisite missing")
+    root = Path(settings.CONTENT_ROOT)
+    mode = payload["mode"]
+    if mode in {"probe", "prepare"}:
+        snapshot = _snapshot(root)
+        result = {"fingerprint": fingerprint(snapshot), "storage_constraint_verified": True}
+        if mode == "prepare":
+            if result["fingerprint"] != payload["baseline"]:
+                raise ValueError("Reviewed dataset differs")
+            selected = payload["plan"]["selection"]
+            name = f"spaces/{selected['space_slug']}/pages/index.md"
+            patched = prepare_index_date_patch(payload["plan"], (root / name).read_bytes())
+            expected = copy.deepcopy(snapshot)
+            _set_date(expected, payload["plan"], payload["plan"]["source_created_at"])
+            expected["files"][name] = hashlib.sha256(patched).hexdigest()
+            result["expected_after"] = fingerprint(expected)
+        return result
+    if mode not in {"apply", "inspect", "reconcile"}:
+        raise ValueError("Unsupported operation")
+    journal = Path("/operation/journal.json")
+    lock_fd = os.open("/operation/execute.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(lock_fd, "w"):
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            return {"status": "operation_busy", "writers_must_remain_fenced": True}
+        if mode == "inspect":
+            status = "absent"
+            if journal.exists() or journal.is_symlink():
+                fd = os.open(journal, os.O_RDONLY | os.O_NOFOLLOW)
+                with os.fdopen(fd, "rb") as stream:
+                    info = os.fstat(stream.fileno())
+                    raw = stream.read(4 * 1024 * 1024 + 1)
+                if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid()
+                        or info.st_mode & 0o077 or info.st_nlink != 1
+                        or len(raw) > 4 * 1024 * 1024):
+                    raise ValueError("Private bound journal required")
+                record = json.loads(raw)
+                original = base64.b64decode(record["original"], validate=True)
+                patched = prepare_index_date_patch(payload["plan"], original)
+                if (record["schema"] != "coulomb.source-date-journal.v1"
+                        or record["plan"] != payload["plan"]
+                        or record["backup_receipt_sha256"] != hashlib.sha256(
+                            canonical(payload["backup"])).hexdigest()
+                        or record["patched_sha256"] != hashlib.sha256(patched).hexdigest()
+                        or record["state"] not in {"pending", "completed", "rolled_back"}):
+                    raise ValueError("Original operation journal binding changed")
+                status = record["state"]
+            return {"status": status, "fingerprint": fingerprint(_snapshot(root))}
+        proof_path = Path("/operation/result.json")
+        result = execute_date_operation(
+            root, payload["plan"], journal, payload["baseline"], payload["backup"],
+            image=payload["image"], recipient=payload["recipient"], reconcile=mode == "reconcile")
+        result["fingerprint"] = fingerprint(_snapshot(root))
+        fd, temporary = tempfile.mkstemp(dir="/operation", prefix=".result-")
+        with os.fdopen(fd, "wb") as stream:
+            stream.write(canonical(result))
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, proof_path)
+        directory = os.open("/operation", os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+        return result
+
+
+if __name__ == "__main__":
+    # A lost kubectl client must not leave an unbounded helper operation behind.
+    # SIGALRM kills this process/SQL connection; the durable journal reconciles it.
+    signal.alarm(150)
+    try:
+        raw = sys.stdin.buffer.read(4 * 1024 * 1024 + 1)
+        if len(raw) > 4 * 1024 * 1024:
+            raise ValueError("Bounded input required")
+        print(json.dumps(run(json.loads(raw))))
+    except Exception as exc:
+        print(json.dumps({"status": "refused", "failure_type": type(exc).__name__}))
+        raise SystemExit(1)

tests/test_source_date_supervisor.py

@@ -0,0 +1,367 @@
+"""Supervisor lifecycle against a synthetic Kubernetes/backup adapter, no network."""
+
+import copy
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location("source_date_supervisor",
+                                             ROOT / "scripts/source_date_supervisor.py")
+supervisor = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(supervisor)
+IMAGE = "forgejo.coulomb.social/coulomb/coulomb-social@sha256:" + "a" * 64
+BASE = {"tables_sha256": "b" * 64, "files_sha256": "c" * 64,
+        "tables": 13, "rows": 112, "files": 82}
+AFTER = {**BASE, "tables_sha256": "d" * 64, "files_sha256": "e" * 64}
+
+
+class FakeNode(supervisor.Node):
+    """Only control-plane/capture effects are simulated; production paths never open."""
+
+    def __init__(self):
+        self.dep = {"metadata": {"uid": "deployment-id", "resourceVersion": "17"}, "spec": {
+            "replicas": 1, "strategy": {"type": "Recreate"}, "template": {"spec": {
+                "containers": [{"name": "app", "image": IMAGE,
+                                "env": [{"name": "CONTENT_ROOT", "value": "/app/var/content"}],
+                                "envFrom": [{"secretRef": {"name": "synthetic-app-env"}}],
+                                "volumeMounts": [{"name": "content",
+                                                  "mountPath": "/app/var/content"}]}],
+                "volumes": [{"name": "content", "persistentVolumeClaim": {
+                    "claimName": supervisor.CLAIM}}]}}}}
+        self.fp, self.helper, self.journal = BASE, None, "absent"
+        self.calls, self.fail, self.capture_count, self.apply_count = [], None, 0, 0
+        self.next_uid = 1
+        self.busy = False
+        self.sidecar = False
+        self.extra_writer = False
+        self.running_image = None
+
+    def trip(self, name):
+        if self.fail == name:
+            self.fail = None
+            raise supervisor.Refused("synthetic lost response: " + name)
+
+    def doc(self, *args):
+        kind = args[1]
+        if kind == "namespace":
+            return {"metadata": {"uid": "cluster-id"}}
+        if kind == "node":
+            return {"metadata": {"uid": "node-id", "name": "synthetic-node"},
+                    "spec": {}, "status": {
+                "conditions": [{"type": "Ready", "status": "True"}]}}
+        if kind == "pvc":
+            return {"metadata": {"uid": "pvc-id"}, "status": {"phase": "Bound"}}
+        if kind in {"hpa", "applications.argoproj.io"}:
+            return {"items": []}
+        if kind == "deployment":
+            return copy.deepcopy(self.dep)
+        if kind == "replicaset":
+            return {"metadata": {"ownerReferences": [{"uid": "deployment-id"}]}}
+        if kind == "pods":
+            pods = []
+            if self.dep["spec"]["replicas"]:
+                pods.append({"metadata": {"name": "synthetic-app", "ownerReferences": [
+                    {"kind": "ReplicaSet", "name": "synthetic-rs"}]}, "spec": {
+                        "nodeName": "synthetic-node", "containers": [{"image": self.running_image
+                            or self.dep["spec"]["template"]["spec"]["containers"][0]["image"]}],
+                        "volumes": [{"persistentVolumeClaim": {
+                            "claimName": supervisor.CLAIM}}]}, "status": {
+                                "phase": "Running", "conditions": [
+                                    {"type": "Ready", "status": "True"}],
+                                "containerStatuses": [{"imageID": self.dep["spec"]["template"][
+                                    "spec"]["containers"][0]["image"]}]}})
+            if self.helper:
+                pods.append(copy.deepcopy(self.helper))
+            if self.extra_writer:
+                pods.append({"metadata": {"name": "unreviewed-writer"}, "spec": {"volumes": [
+                    {"persistentVolumeClaim": {"claimName": supervisor.CLAIM}}]},
+                    "status": {"phase": "Running"}})
+            return {"items": pods}
+        raise AssertionError(args)
+
+    def command(self, args, payload=None, **kwargs):
+        self.calls.append(args[:4])
+        if args[:2] == ["helm", "history"]:
+            return supervisor.canonical([{"revision": 18, "status": "deployed"}])
+        if args[:3] == ["helm", "get", "manifest"]:
+            return b"synthetic helm manifest"
+        if args[:2] == ["kubectl", "exec"]:
+            message = json.loads(payload)
+            mode = message["mode"]
+            if mode == "prepare":
+                return supervisor.canonical({"fingerprint": self.fp,
+                                             "storage_constraint_verified": True,
+                                             "expected_after": AFTER})
+            if mode == "probe":
+                return supervisor.canonical({"fingerprint": self.fp})
+            if mode == "inspect":
+                status = "operation_busy" if self.busy else self.journal
+                return supervisor.canonical({"status": status,
+                                             "fingerprint": self.fp})
+            if mode == "reconcile":
+                assert self.journal == "pending"
+                self.journal, self.fp = "rolled_back", BASE
+                return supervisor.canonical({"status": self.journal, "fingerprint": self.fp})
+            assert mode == "apply"
+            self.apply_count += 1
+            if self.fail == "apply_pending":
+                self.journal = "pending"
+                self.trip("apply_pending")
+            self.journal, self.fp = "completed", AFTER
+            self.trip("apply_reply")
+            return supervisor.canonical({"status": "completed", "whole_store_postflight": True,
+                                         "fingerprint": self.fp})
+        if args[:2] == ["kubectl", "scale"]:
+            current = next(a for a in args if a.startswith("--current-replicas=")).split("=")[1]
+            assert int(current) == self.dep["spec"]["replicas"]
+            wanted = next(a for a in args if a.startswith("--replicas=")).split("=")[1]
+            self.dep["spec"]["replicas"] = int(wanted)
+            return b""
+        if args[:2] == ["kubectl", "create"]:
+            pod = json.loads(payload)
+            if self.sidecar:
+                pod["spec"]["containers"].append({"name": "injected"})
+            if "--dry-run=server" not in args:
+                assert self.dep["spec"]["replicas"] == 0 and not self.helper
+                pod["metadata"]["uid"] = "helper-" + str(self.next_uid)
+                self.next_uid += 1
+                pod["status"] = {"phase": "Running"}
+                self.helper = copy.deepcopy(pod)
+                self.trip("create_reply")
+            return supervisor.canonical(pod)
+        if args[:3] == ["kubectl", "delete", "--raw"]:
+            assert json.loads(payload)["preconditions"]["uid"] == self.helper["metadata"]["uid"]
+            self.helper = None
+            self.trip("delete_reply")
+            return b""
+        if args[:3] == ["kubectl", "rollout", "status"]:
+            self.trip("resume_reply")
+            return b""
+        assert args[:2] == ["kubectl", "wait"], args
+        return b""
+
+    def verify_backup_installation(self, target):
+        self.trip("helper_hash")
+
+    def verify_local_node(self, node):
+        pass  # synthetic cluster/root adapter, never a native machine-id claim
+
+    def installation_pins(self):
+        return {"installed_helpers": {name: "f" * 64 for name in supervisor.HELPERS},
+                "public_recipient": "age1synthetic"}
+
+    def capture(self, target):
+        self.capture_count += 1
+        self.trip("capture")
+        return {"status": "synthetic-capture"}
+
+    def lock_backups(self, stack):
+        self.trip("backup_lock")
+
+    def helper_directory(self, path):
+        path.mkdir(mode=0o700, exist_ok=True)
+
+    def read_state(self, path):
+        return json.loads(path.read_bytes())
+
+
+@pytest.fixture
+def prepared(tmp_path):
+    node = FakeNode()
+    target = {"cluster_uid": "cluster-id", "deployment_uid": "deployment-id", "node_uid": "node-id",
+              "node_name": "synthetic-node", "pvc_uid": "pvc-id", "image": IMAGE,
+              "deployment_spec_sha256": supervisor.sha(supervisor.canonical(node.dep["spec"])),
+              "helm_revision": 18,
+              "helm_manifest_sha256": supervisor.sha(b"synthetic helm manifest"),
+              "public_recipient": "age1synthetic", "installed_helpers": {
+                  name: "f" * 64 for name in supervisor.HELPERS}}
+    capsule = tmp_path / "capsule"
+    capsule.mkdir(mode=0o700)
+    for name in supervisor.MEMBERS:
+        (capsule / name).write_text("synthetic source")
+    (capsule / "plan.json").write_text(json.dumps({
+        "schema": "coulomb.source-date-backfill-plan.v1", "blockers": [],
+        "prepared_change_set_valid": True}))
+    manifest = {"schema": "coulomb.supervised-date-capsule.v1", "baseline": BASE, "target": target,
+                "attempt_id": "1" * 32, "requires_founder_attended_execution": True,
+                "files": {name: supervisor.sha((capsule / name).read_bytes())
+                          for name in supervisor.MEMBERS}}
+    (capsule / "manifest.json").write_bytes(supervisor.canonical(manifest))
+    digest = "sha256:" + supervisor.sha(supervisor.canonical(manifest))
+    frozen = supervisor.Capsule(capsule, digest)
+    root = tmp_path / "state"
+    return supervisor.Supervisor(frozen, node, state_root=root), node, capsule, digest
+
+
+def test_success_captures_fences_applies_checks_and_resumes_once(prepared):
+    operation, node, _, _ = prepared
+    assert not operation.state_root.exists()
+    operation.preflight()
+    assert not operation.state_root.exists() and node.capture_count == 0
+    result = operation.execute()
+    assert result["phase"] == "completed" and result["correction_status"] == "completed"
+    assert node.capture_count == node.apply_count == 1
+    assert node.dep["spec"]["replicas"] == 1 and not node.helper
+    with pytest.raises(supervisor.Refused, match="prior_attempt"):
+        operation.execute()
+    assert node.apply_count == 1
+
+
+@pytest.mark.parametrize("failure", ["apply_reply", "apply_pending",
+                                      "create_reply", "delete_reply"])
+def test_uncertainty_preserves_fence_and_recovers_without_new_capture_or_apply(prepared, failure):
+    operation, node, _, _ = prepared
+    node.fail = failure
+    with pytest.raises(supervisor.Refused):
+        operation.execute()
+    assert node.dep["spec"]["replicas"] == 0
+    applied = node.apply_count
+    result = operation.recover()
+    assert result["phase"] == "completed" and node.dep["spec"]["replicas"] == 1
+    assert node.capture_count == 1 and node.apply_count == applied
+    assert not node.helper
+    if failure == "apply_pending":
+        assert result["correction_status"] == "rolled_back"
+
+
+def test_live_original_operation_blocks_recovery_and_never_restarts(prepared):
+    operation, node, _, _ = prepared
+    node.fail = "apply_pending"
+    with pytest.raises(supervisor.Refused):
+        operation.execute()
+    node.busy = True
+    with pytest.raises(supervisor.Refused, match="still_running"):
+        operation.recover()
+    assert node.capture_count == node.apply_count == 1 and node.dep["spec"]["replicas"] == 0
+
+
+def test_lost_resume_reply_preserves_new_productive_writes(prepared):
+    operation, node, _, _ = prepared
+    node.fail = "resume_reply"
+    with pytest.raises(supervisor.Refused):
+        operation.execute()
+    assert node.dep["spec"]["replicas"] == 1
+    node.fp = {**AFTER, "rows": 113}  # legitimate write after the application resumed
+    result = operation.recover()
+    assert result["post_resume_writes_preserved"] and node.fp["rows"] == 113
+    assert node.capture_count == node.apply_count == 1
+
+
+@pytest.mark.parametrize("problem", ["pod_injection", "foreign_writer",
+                                      "spec_drift", "helper_hash", "pod_image"])
+def test_preflight_refusals_never_capture_or_scale(prepared, problem):
+    operation, node, _, _ = prepared
+    if problem == "pod_injection":
+        node.sidecar = True
+    elif problem == "foreign_writer":
+        node.extra_writer = True
+    elif problem == "spec_drift":
+        node.dep["spec"]["template"]["spec"]["containers"][0]["image"] += "0"
+    elif problem == "pod_image":
+        node.running_image = IMAGE + "0"
+    else:
+        node.fail = "helper_hash"
+    with pytest.raises(supervisor.Refused):
+        operation.execute()
+    assert node.capture_count == node.apply_count == 0 and node.dep["spec"]["replicas"] == 1
+
+
+def test_changed_capsule_refuses_before_any_backend_access(prepared):
+    _, _, path, digest = prepared
+    (path / "wire.py").write_text("changed")
+    with pytest.raises(supervisor.Refused, match="member_changed"):
+        supervisor.Capsule(path, digest)
+
+
+def test_unrelated_data_drift_after_uncertain_result_never_resumes(prepared):
+    operation, node, _, _ = prepared
+    node.fail = "apply_reply"
+    with pytest.raises(supervisor.Refused):
+        operation.execute()
+    node.fp = {**AFTER, "rows": 999}
+    with pytest.raises(supervisor.Refused, match="dataset_requires_owner_review"):
+        operation.recover()
+    assert node.dep["spec"]["replicas"] == 0 and node.helper
+    assert node.capture_count == node.apply_count == 1
+
+
+def test_replaced_helper_is_never_deleted_or_used_for_recovery(prepared):
+    operation, node, _, _ = prepared
+    node.fail = "apply_reply"
+    with pytest.raises(supervisor.Refused):
+        operation.execute()
+    node.helper["metadata"]["uid"] = "foreign-helper"
+    with pytest.raises(supervisor.Refused, match="helper_replaced"):
+        operation.recover()
+    assert node.dep["spec"]["replicas"] == 0
+    assert node.helper["metadata"]["uid"] == "foreign-helper"
+
+
+def test_terminal_original_helper_is_uid_deleted_and_recreated_only_for_reconciliation(prepared):
+    operation, node, _, _ = prepared
+    node.fail = "apply_reply"
+    with pytest.raises(supervisor.Refused):
+        operation.execute()
+    node.helper["status"]["phase"] = "Succeeded"
+    result = operation.recover()
+    assert result["correction_status"] == "completed" and not node.helper
+    assert node.next_uid == 3 and node.capture_count == node.apply_count == 1
+
+
+def test_capsule_preparation_is_private_fresh_and_has_a_new_attempt_id(prepared, tmp_path):
+    import runpy
+    import sys
+
+    operation, _, source, _ = prepared
+    sys.path.insert(0, str(ROOT / "scripts"))
+    try:
+        prepare = runpy.run_path(str(ROOT / "scripts/prepare-source-date-capsule.py"))["prepare"]
+    finally:
+        sys.path.pop(0)
+    baseline, target = tmp_path / "baseline.json", tmp_path / "target.json"
+    baseline.write_bytes(supervisor.canonical(BASE))
+    target.write_bytes(supervisor.canonical(operation.target))
+    proofs = []
+    for name in ("first", "second"):
+        output = tmp_path / name
+        proof = prepare(source / "plan.json", baseline, target, output)
+        capsule = supervisor.Capsule(output, proof["proposal_digest"])
+        assert capsule.manifest["requires_founder_attended_execution"]
+        assert output.stat().st_mode & 0o777 == 0o700
+        assert all(p.stat().st_mode & 0o777 == 0o600 for p in output.iterdir())
+        proofs.append(proof["proposal_digest"])
+        with pytest.raises(ValueError, match="Fresh private"):
+            prepare(source / "plan.json", baseline, target, output)
+    assert proofs[0] != proofs[1]
+
+
+@pytest.mark.parametrize("changed_image", [False, True])
+def test_observer_prepares_without_capturing_or_refuses_missing_release(
+        prepared, tmp_path, monkeypatch, changed_image):
+    import runpy
+    import sys
+
+    operation, node, source, _ = prepared
+    sys.path.insert(0, str(ROOT / "scripts"))
+    try:
+        observer = runpy.run_path(str(ROOT / "scripts/observe-source-date-target.py"))["observe"]
+    finally:
+        sys.path.pop(0)
+    monkeypatch.setattr(supervisor.os, "geteuid", lambda: 0)  # simulated root context only
+    output = tmp_path / "observed-capsule"
+    if changed_image:
+        node.dep["spec"]["template"]["spec"]["containers"][0]["image"] += "0"
+        with pytest.raises(RuntimeError, match="prerequisite_missing"):
+            observer(source / "plan.json", output, image=IMAGE, cluster_uid="cluster-id",
+                     deployment_uid="deployment-id", pvc_uid="pvc-id", node=node)
+        assert not output.exists()
+    else:
+        proof = observer(source / "plan.json", output, image=IMAGE, cluster_uid="cluster-id",
+                         deployment_uid="deployment-id", pvc_uid="pvc-id", node=node)
+        capsule = supervisor.Capsule(output, proof["proposal_digest"])
+        assert capsule.baseline == BASE and capsule.target == operation.target
+        assert proof["baseline_rechecked"] and not proof["live_changes"]
+    assert node.capture_count == node.apply_count == 0 and node.dep["spec"]["replicas"] == 1

workplans/CSOC-WP-0012-durable-recovery-service-readiness.md

@@ -262,6 +262,13 @@ unchanged. Temporary private archive removed. Proof:
 docs/release/client-data-export-rehearsal-2026-10-30.json. Full make check340passes,
 zero skips. No deployment, production export, teardown or expiry was performed.
 
+2026-10-30 operational inventory: the prepared date supervisor introduces an
+operator-private root date-operation store containing original client index bytes,
+plan and backup binding. Client export inventory now names this store explicitly;
+the app cannot read/export it. Include any future retained journals in operator
+offboarding handoff and explicit expiry under the open service contract work.
+No production date journal or expiry was created by the isolated rehearsal.
+
 ## Completion
 
 Finish when every task above has its stated evidence and acceptance, or hand off any newly discovered residual to a live work record first. Cross-owner dependencies remain their existing canonical records.

workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md

@@ -210,6 +210,24 @@ productive correction is implied.50 focused and364 full gate cases pass, zero sk
 The exact supervised node wrapper, live backup/source bookends, pinned helper/PVC
 admission, durable journal placement and operator postflight/resume remain open.
 
+2026-10-30 supervisor follow-through: added exact private capsule preparation,
+read-only native target observer and founder-terminal node supervisor. The wrapper
+binds the physical node/PVC/deployment/Helm/image, source and full dataset, checks
+fresh admitted capture/readback bookends, fences application/PVC writers, and runs
+the operation through a pinned helper with a durable private host journal. Explicit
+recovery handles original-operation busy state, pending/terminal journal outcomes,
+lost create/delete/apply/resume replies and terminal helper replacement without
+replaying apply or overwriting post-resume writes.19 focused cases pass. Review:
+docs/release/source-date-supervisor-review-2026-10-30.md. Native observation and an
+admitted date capsule must follow the still-pending storage release; none exists
+for production yet. Source/rehearsal completion does not finish this task.
+The real helper protocol and controller now pass normal completion and a lost
+post-commit reply on disposable recovered-pair clones. Both cases compare all13
+tables/82files and recover without a second apply/capture; originals are unchanged
+and clones removed. Kubernetes/root/provider actions are explicitly simulated.
+Proof: docs/release/source-date-supervisor-rehearsal-2026-10-30.json.
+`make check`:383 passed, zero skips, including required browser journeys.
+
 ## Completion
 
 Finish when every task above has its stated evidence and acceptance, or hand off any newly discovered residual to a live work record first. Cross-owner dependencies remain their existing canonical records.