"""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, 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)

    @contextmanager
    def helper_evidence_files(self, path):
        """Open only the closed helper's known evidence; preserve crash recovery."""
        with ExitStack() as stack:
            def opened(name, flags, parent=None):
                fd = os.open(name, flags | os.O_NOFOLLOW, dir_fd=parent)
                stack.callback(os.close, fd)
                return fd

            directory = opened("/", os.O_RDONLY | os.O_DIRECTORY)
            for component in path.parts[1:]:
                directory = opened(component, os.O_RDONLY | os.O_DIRECTORY, directory)
            info = os.fstat(directory)
            if (info.st_uid not in {os.geteuid(), 10001}
                    or info.st_gid not in {os.getegid(), 10001}
                    or stat.S_IMODE(info.st_mode) != 0o700):
                raise Refused("private_helper_evidence_required")
            names = sorted(os.listdir(directory))
            if set(names) - {"execute.lock", "journal.json", "result.json"}:
                raise Refused("unexpected_helper_evidence")
            files, evidence = {}, {}
            for name in names:
                fd = opened(name, os.O_RDONLY | os.O_NONBLOCK, directory)
                info = os.fstat(fd)
                if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1
                        or info.st_uid not in {os.geteuid(), 10001}
                        or info.st_gid not in {os.getegid(), 10001}
                        or stat.S_IMODE(info.st_mode) != 0o600):
                    raise Refused("private_helper_evidence_required")
                if name == "execute.lock":
                    fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                digest = hashlib.sha256()
                while chunk := os.read(fd, 1024 * 1024):
                    digest.update(chunk)
                after = os.fstat(fd)
                if any(getattr(info, key) != getattr(after, key) for key in (
                        "st_dev", "st_ino", "st_mode", "st_uid", "st_gid", "st_nlink",
                        "st_size", "st_mtime_ns", "st_ctime_ns")):
                    raise Refused("helper_evidence_changed")
                files[name] = fd
                evidence[name] = {"sha256": digest.hexdigest(), "bytes": info.st_size,
                                  "device": info.st_dev, "inode": info.st_ino,
                                  "mtime_ns": info.st_mtime_ns}
            yield directory, files, evidence

    def helper_evidence(self, path):
        with self.helper_evidence_files(path) as (_, _, evidence):
            return evidence

    def seal_helper_evidence(self, path, expected):
        # Mixed helper/root custody is allowed only to resume this recorded step.
        with self.helper_evidence_files(path) as (directory, files, evidence):
            if evidence != expected:
                raise Refused("helper_evidence_changed")
            for fd in [*files.values(), directory]:
                os.fchown(fd, os.geteuid(), os.getegid())
                os.fsync(fd)
                info = os.fstat(fd)
                if (info.st_uid, info.st_gid) != (os.geteuid(), os.getegid()):
                    raise Refused("helper_evidence_custody_unconfirmed")
        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)
        return self.complete()

    def complete(self, **changes):
        """Seal evidence only after resume, when helper recovery is no longer needed."""
        self.deployment(1)
        path = self.directory / "private"
        for pod in self.node.doc("get", "pods", "-A")["items"]:
            if pod["metadata"]["name"] == self.helper:
                raise Refused("helper_still_present_at_closeout")
            for volume in pod["spec"].get("volumes", []):
                if "hostPath" in volume:
                    mount = Path(volume["hostPath"]["path"])
                    if mount == path or mount in path.parents or path in mount.parents:
                        raise Refused("helper_evidence_still_mounted")
        evidence = self.state.get("helper_evidence")
        if evidence is None:
            evidence = self.node.helper_evidence(path)
        self.record("custody_closeout_started", application_resumed=True,
                    helper_evidence=evidence, **changes)
        self.node.seal_helper_evidence(path, evidence)
        self.record("completed", helper_evidence_root_custody_verified=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")
            if self.state["phase"] == "completed":
                raise Refused("completed_attempt_must_not_be_replayed")
            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.
                phase = (self.state["failed_phase"] if self.state["phase"] == "requires_review"
                         else self.state["phase"])
                if (phase not in {
                        "resume_started", "custody_closeout_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"])
                return self.complete(resumed_reply_reconciled=True,
                                     post_resume_writes_preserved=True)
            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)