Prepare supervised Coulomb identity and export image release

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3

tools/coulomb-image-release.py

@@ -0,0 +1,185 @@
+"""Prepare/read-only-check a sealed release; live mode is attended founder work.
+
+This launcher does not mint an approval or bypass gated-executor. The existing
+gated-executor profile covers Kubernetes, not the root-owned backup helper refresh.
+Only the founder runs --execute in their own terminal after reviewing the packet.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import io
+import json
+import os
+import shlex
+import subprocess
+import sys
+import tarfile
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+PLAN = ROOT / "release/coulomb-identity-export-23ffe2c.json"
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+def prepare():
+    plan = json.loads(PLAN.read_text())
+    paths = [
+        ROOT / name
+        for name in (
+            "tools/coulomb-image-release.py",
+            "tools/coulomb_image_release_node.py",
+            "tools/coulomb_image_probe.py",
+            "release/identity-export-values.yaml",
+            "release/identity-export-snapshot.py",
+        )
+    ]
+    paths += sorted((ROOT / "charts/coulomb-social").rglob("*"))
+    paths += sorted((ROOT / "release/identity-export-evidence").glob("*.json"))
+    paths = [p for p in paths if p.is_file()]
+    if any(p.is_symlink() or not p.resolve().is_relative_to(ROOT) for p in paths):
+        raise RuntimeError("regular_reviewed_files_required")
+    content = {str(p.relative_to(ROOT)): p.read_bytes() for p in paths}
+    plan["files"] = {
+        name: hashlib.sha256(data).hexdigest() for name, data in content.items()
+    }
+    for name, expected in plan["rehearsal_evidence_sha256"].items():
+        if plan["files"].get("release/identity-export-evidence/" + name) != expected:
+            raise RuntimeError("reviewed_candidate_evidence_changed")
+    digest = "sha256:" + hashlib.sha256(canonical(plan)).hexdigest()
+    content["manifest.json"] = canonical(plan)
+    return plan, digest, content
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--preflight", action="store_true")
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--expect-digest")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    if args.preflight and args.execute:
+        raise SystemExit("choose read-only preflight or attended execution")
+    plan, digest, content = prepare()
+    if args.expect_digest and args.expect_digest != digest:
+        raise SystemExit("reviewed proposal changed; do not execute")
+    if not args.preflight and not args.execute:
+        print(
+            json.dumps(
+                {
+                    "status": "prepared",
+                    "proposal_digest": digest,
+                    "file_count": len(content),
+                    "candidate": plan["candidate_image"],
+                    "live_changes": False,
+                }
+            )
+        )
+        return
+    if args.execute:
+        if not args.expect_digest or not sys.stdin.isatty():
+            raise SystemExit(
+                "founder must review and run the exact digest in their own terminal"
+            )
+        print("Reviewed proposal: " + digest)
+        print(
+            "Coulomb-only image update; existing database schema retained, fresh paired backup, postflight and backup image-pin refresh."
+        )
+        if (
+            input("Type DEPLOY followed by the full digest to proceed: ").strip()
+            != "DEPLOY " + digest
+        ):
+            raise SystemExit("not approved; no live change")
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w:") as archive:
+        for name, data in content.items():
+            entry = tarfile.TarInfo(name)
+            entry.size, entry.mode = len(data), 0o600
+            archive.addfile(entry, io.BytesIO(data))
+    remote = "/tmp/csoc-reviewed-image-" + digest.removeprefix("sha256:")
+    receiver = r"""
+import hashlib,io,json,os,pathlib,sys,tarfile
+root=pathlib.Path(sys.argv[1]);expected=sys.argv[2]
+raw=sys.stdin.buffer.read(2*1024**2+1);assert len(raw)<=2*1024**2
+with tarfile.open(fileobj=io.BytesIO(raw),mode='r:') as archive:
+ entries=archive.getmembers();assert len({e.name for e in entries})==len(entries)
+ assert all(e.isfile() and not pathlib.Path(e.name).is_absolute() and '..' not in pathlib.Path(e.name).parts for e in entries)
+ data={e.name:archive.extractfile(e).read() for e in entries}
+ plan=json.loads(data['manifest.json'])
+ canonical=json.dumps(plan,sort_keys=True,separators=(',',':')).encode()
+ assert 'sha256:'+hashlib.sha256(canonical).hexdigest()==expected
+ assert set(data)==set(plan['files'])|{'manifest.json'}
+ for name,digest in plan['files'].items():assert hashlib.sha256(data[name]).hexdigest()==digest
+ root.mkdir(mode=0o700,exist_ok=True)
+ assert root.lstat().st_uid==os.getuid() and root.lstat().st_mode&0o077==0
+ for name,value in data.items():
+  target=root/name;target.parent.mkdir(parents=True,exist_ok=True,mode=0o700)
+  if target.exists():assert not target.is_symlink() and target.read_bytes()==value;continue
+  fd=os.open(target,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
+  with os.fdopen(fd,'wb') as stream:stream.write(value)
+"""
+    staged = subprocess.run(
+        [
+            "ssh",
+            "railiance01",
+            "python3 -c "
+            + shlex.quote(receiver)
+            + " "
+            + shlex.quote(remote)
+            + " "
+            + shlex.quote(digest),
+        ],
+        input=buffer.getvalue(),
+        capture_output=True,
+        timeout=60,
+        check=False,
+    )
+    if staged.returncode:
+        raise SystemExit("reviewed capsule staging refused; no live change")
+    node_hash = plan["files"]["tools/coulomb_image_release_node.py"]
+    guard = (
+        "import hashlib,pathlib,sys;p=pathlib.Path("
+        + repr(remote + "/tools/coulomb_image_release_node.py")
+        + ");"
+        "b=p.read_bytes();assert hashlib.sha256(b).hexdigest()=="
+        + repr(node_hash)
+        + ";"
+        "sys.argv=['reviewed-release','--capsule',"
+        + repr(remote)
+        + ",'--expect-digest',"
+        + repr(digest)
+        + ","
+        + repr("--execute" if args.execute else "--preflight")
+        + "];exec(compile(b,str(p),'exec'),{'__name__':'__main__'})"
+    )
+    result = subprocess.run(
+        ["ssh", "railiance01", "sudo -n python3 -c " + shlex.quote(guard)],
+        capture_output=True,
+        timeout=3600,
+        check=False,
+    )
+    try:
+        receipt = json.loads(result.stdout)
+    except (ValueError, TypeError):
+        receipt = {
+            "status": "remote_execution_unconfirmed",
+            "proposal_digest": digest,
+            "journal": "/var/lib/coulomb-social-backup/releases/"
+            + digest.removeprefix("sha256:")
+            + "/receipt.json",
+        }
+    if args.output:
+        fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(fd, "w") as stream:
+            json.dump(receipt, stream, indent=2)
+    print(json.dumps(receipt))
+    if result.returncode:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

tools/coulomb_image_probe.py

@@ -0,0 +1,134 @@
+"""Read-only data/schema and view probe; runs inside the existing app container."""
+
+import hashlib
+import json
+import sys
+import uuid
+from pathlib import Path
+
+import django
+
+django.setup()
+from coulomb_social.apps.members.models import Member, User
+from coulomb_social.apps.spaces.models import Space
+from django.conf import settings
+from django.db import connection
+from django.db.models import Count
+
+assert connection.vendor == "postgresql"
+assert connection.settings_dict["NAME"] in {
+    "coulomb_social_db",
+    "coulomb_social_restore",
+}
+tables = {}
+with connection.cursor() as cursor:
+    cursor.execute(
+        "SELECT tablename FROM pg_tables WHERE schemaname='public' "
+        "AND tablename!='django_migrations' ORDER BY tablename"
+    )
+    names = [row[0] for row in cursor.fetchall()]
+    for table in names:
+        cursor.execute("SELECT * FROM " + connection.ops.quote_name(table))
+        rows = sorted(
+            json.dumps(row, default=str, separators=(",", ":"))
+            for row in cursor.fetchall()
+        )
+        tables[table] = {
+            "rows": len(rows),
+            "sha256": hashlib.sha256("\n".join(rows).encode()).hexdigest(),
+        }
+    cursor.execute(
+        "SELECT conname FROM pg_constraint WHERE conrelid='spaces_space'::regclass"
+    )
+    constraints = {row[0] for row in cursor.fetchall()}
+    cursor.execute(
+        "SELECT COUNT(*) FROM django_migrations WHERE app='spaces' AND name='0003_space_slug_global_uniq'"
+    )
+    applied = cursor.fetchone()[0] == 1
+    cursor.execute("SELECT * FROM django_migrations")
+    migration_rows = sorted(json.dumps(row, default=str, separators=(",", ":"))
+                            for row in cursor.fetchall())
+root = Path(settings.CONTENT_ROOT)
+files = {
+    str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest()
+    for p in root.rglob("*")
+    if p.is_file()
+}
+result = {
+    "business_tables": len(tables),
+    "migration_rows_sha256": hashlib.sha256("\n".join(migration_rows).encode()).hexdigest(),
+    "business_rows": sum(t["rows"] for t in tables.values()),
+    "business_rows_sha256": hashlib.sha256(
+        json.dumps(tables, sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest(),
+    "content_files": len(files),
+    "content_files_sha256": hashlib.sha256(
+        json.dumps(files, sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest(),
+    "global_slug_constraint": "spaces_space_slug_global_uniq" in constraints,
+    "tenant_slug_constraint": "spaces_space_tenant_slug_uniq" in constraints,
+    "migration_0003_applied": applied,
+    "duplicate_slug_groups": Space.objects.values("slug")
+    .annotate(n=Count("pk"))
+    .filter(n__gt=1)
+    .count(),
+}
+if len(sys.argv) > 1 and sys.argv[1] == "views":
+    from coulomb_social.apps.spaces import views
+    from coulomb_social.apps.spaces.content import load_space_page
+    from django.test import RequestFactory
+    from django.urls import reverse
+
+    assert not settings.FLEX_AUTH_BASE_URL
+    factory = RequestFactory()
+    rendered_count, denied_count = 0, 0
+    for space in Space.objects.filter(is_active=True):
+        paths = sorted((root / "spaces" / space.slug / "pages").glob("*.md"))
+        if not paths:
+            continue
+        owner = space.created_by
+        if owner is None:
+            owner = (
+                space.memberships.filter(role="owner")
+                .select_related("member__user")
+                .first()
+                .member
+            )
+        token = "unsaved-release-check-" + uuid.uuid4().hex
+        outsider = Member(
+            user=User(username=token),
+            tenant_id=space.tenant_id,
+            subject=token,
+            issuer="https://release-check.invalid",
+            user_engine_user_id=token,
+        )
+        for path in paths:
+            page = path.stem
+            loaded = load_space_page(space, page)
+            assert loaded.error is None
+
+            def request(user, slug=space.slug, page_slug=page):
+                req = factory.get(
+                    reverse("spaces:detail", kwargs={"slug": slug}),
+                    {"page": page_slug},
+                )
+                req.user, req.session = user, {}
+                return req
+
+            response = views.space_detail(request(owner.user), slug=space.slug)
+            assert (
+                response.status_code == 200 and loaded.html.encode() in response.content
+            )
+            rendered_count += 1
+            denied = views.space_detail(request(outsider.user), slug=space.slug)
+            assert denied.status_code == 404
+            if loaded.html:
+                assert loaded.html.encode() not in denied.content
+            denied_count += 1
+    assert rendered_count > 0 and denied_count == rendered_count
+    result.update(
+        actual_owner_pages_rendered=rendered_count,
+        unsaved_outsider_page_denials=denied_count,
+        production_identity_or_grant_created=False,
+    )
+print(json.dumps(result))

tools/coulomb_image_release_node.py

@@ -0,0 +1,525 @@
+"""Attended founder-only node release; argv/data are sealed by the reviewed capsule.
+
+This is a supervisor command, not a new agent credential or approval boundary.
+No Secret object/output is fetched. Existing approved backup custody stays local.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import stat
+import subprocess
+import tempfile
+import time
+import urllib.request
+from datetime import UTC, datetime
+from pathlib import Path
+
+JOURNAL_ROOT = Path("/var/lib/coulomb-social-backup/releases")
+SNAPSHOT = Path("/opt/coulomb-social-backup/snapshot-content-and-db.py")
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+class Refused(RuntimeError):
+    pass
+
+
+class Release:
+    def __init__(self, capsule, expected, read_only=False):
+        self.root = capsule
+        self.plan = json.loads((capsule / "manifest.json").read_text())
+        frozen = {}
+        if "sha256:" + sha(canonical(self.plan)) != expected:
+            raise Refused("proposal_digest_differs")
+        for name, digest in self.plan["files"].items():
+            path = capsule / name
+            if (
+                not path.resolve().is_relative_to(capsule.resolve())
+                or Path(name).is_absolute()
+                or ".." in Path(name).parts
+                or path.is_symlink()
+                or not path.is_file()
+                or sha(path.read_bytes()) != digest
+            ):
+                raise Refused("capsule_file_differs")
+            frozen[name] = path.read_bytes()
+            if sha(frozen[name]) != digest:
+                raise Refused("capsule_changed_during_verification")
+        if os.geteuid() != 0:
+            raise Refused("attended_founder_root_execution_required")
+        self.env = dict(os.environ, KUBECONFIG="/etc/rancher/k3s/k3s.yaml")
+        self.read_only = read_only
+        self.receipt = {
+            "schema": "coulomb.guarded-image-release.v1",
+            "proposal_digest": expected,
+            "started_at": datetime.now(UTC).isoformat(),
+            "status": "preflight",
+            "production_data_restored": False,
+            "timers_activated": False,
+        }
+        self.journal = JOURNAL_ROOT / expected.removeprefix("sha256:")
+        if read_only:
+            return
+        if self.journal.exists():
+            raise Refused("prior_execution_requires_review")
+        self.journal.mkdir(parents=True, mode=0o700)
+        self.root = self.journal / "capsule"
+        self.root.mkdir(mode=0o700)
+        for name, data in frozen.items():
+            target = self.root / name
+            target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            with os.fdopen(fd, "wb") as stream:
+                stream.write(data)
+        self.receipt["pid"] = os.getpid()
+        self.save()
+
+    def save(self):
+        if self.read_only:
+            return
+        fd, name = tempfile.mkstemp(prefix=".receipt-", dir=self.journal)
+        with os.fdopen(fd, "w") as stream:
+            json.dump(self.receipt, stream, sort_keys=True, indent=2)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(name, self.journal / "receipt.json")
+        directory = os.open(self.journal, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+
+    def command(self, arguments, payload=None, env=None, timeout=180, allowed=(0,)):
+        # In particular, give the paired capture its existing app-resume window
+        # on timeout; subprocess.run would SIGKILL it before that cleanup.
+        with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr:
+            process = subprocess.Popen(
+                arguments,
+                env=env or self.env,
+                stdin=subprocess.PIPE if payload is not None else subprocess.DEVNULL,
+                stdout=stdout,
+                stderr=stderr,
+            )
+            try:
+                process.communicate(payload, timeout=timeout)
+            except subprocess.TimeoutExpired:
+                process.terminate()
+                try:
+                    process.wait(timeout=450)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait()
+                raise Refused("private_child_timed_out_after_cleanup_window") from None
+            stdout.seek(0)
+            body = stdout.read(8 * 1024**2 + 1)
+            if len(body) > 8 * 1024**2:
+                raise Refused("bounded_child_output_required")
+            result = subprocess.CompletedProcess(arguments, process.returncode, body)
+        if result.returncode not in allowed:
+            raise Refused("private_child_refused")
+        return result
+
+    def kdoc(self, *arguments):
+        return json.loads(self.command(["kubectl", *arguments, "-o", "json"]).stdout)
+
+    def history(self):
+        return json.loads(
+            self.command(
+                [
+                    "helm",
+                    "history",
+                    "coulomb-social",
+                    "-n",
+                    "coulomb-social",
+                    "-o",
+                    "json",
+                ]
+            ).stdout
+        )
+
+    def deployment(self):
+        return self.kdoc("get", "deployment", "coulomb-social", "-n", "coulomb-social")
+
+    def probe(self, mode="data"):
+        source = (self.root / "tools/coulomb_image_probe.py").read_text()
+        result = self.command(
+            [
+                "kubectl",
+                "exec",
+                "-n",
+                "coulomb-social",
+                "deploy/coulomb-social",
+                "--",
+                "python",
+                "-c",
+                source,
+                mode,
+            ]
+        )
+        return json.loads(result.stdout)
+
+    def same_data(self, actual):
+        for name in (
+            "business_tables",
+            "business_rows",
+            "business_rows_sha256",
+            "content_files",
+            "content_files_sha256",
+            "migration_rows_sha256",
+        ):
+            if actual[name] != self.plan["expected_data"][name]:
+                raise Refused("approved_recovered_dataset_drifted")
+        if actual["duplicate_slug_groups"] != 0 or actual["tenant_slug_constraint"] is not True:
+            raise Refused("storage_invariant_refused")
+        if (actual.get("global_slug_constraint") is not True
+                or actual.get("migration_0003_applied") is not True):
+            raise Refused("already_migrated_baseline_required")
+
+    def preflight(self):
+        self.verify_installed_helpers()
+        if (
+            self.kdoc("get", "namespace", "kube-system")["metadata"]["uid"]
+            != self.plan["cluster_uid"]
+        ):
+            raise Refused("wrong_cluster")
+        node = self.kdoc("get", "nodes")["items"]
+        if len(node) != 1 or not any(
+            a["address"] == "92.205.62.239" for a in node[0]["status"]["addresses"]
+        ):
+            raise Refused("wrong_node")
+        conditions = {v["type"]: v["status"] for v in node[0]["status"]["conditions"]}
+        if conditions.get("Ready") != "True" or any(
+            conditions.get(k) != "False"
+            for k in ("MemoryPressure", "DiskPressure", "PIDPressure")
+        ):
+            raise Refused("node_pressure")
+        allocated = self.command(
+            ["kubectl", "describe", "node", node[0]["metadata"]["name"]]
+        ).stdout.decode()
+        requests = {}
+        for resource in ("cpu", "memory"):
+            found = re.search(
+                r"(?m)^\s+" + resource + r"\s+\S+\s+\((\d+)%\)", allocated
+            )
+            if not found or int(found[1]) > 90:
+                raise Refused("node_request_headroom_refused")
+            requests[resource] = int(found[1])
+        self.receipt["node_request_percent"] = requests
+        history = self.history()
+        if (
+            history[-1]["revision"] != self.plan["baseline_revision"]
+            or history[-1]["status"] != "deployed"
+        ):
+            raise Refused("helm_baseline_changed")
+        manifest = self.command(
+            ["helm", "get", "manifest", "coulomb-social", "-n", "coulomb-social"]
+        ).stdout
+        if sha(manifest) != self.plan["baseline_helm_manifest_sha256"]:
+            raise Refused("helm_manifest_changed")
+        deployment = self.deployment()
+        if (
+            deployment["metadata"]["uid"] != self.plan["deployment_uid"]
+            or sha(canonical(deployment["spec"])) != self.plan["baseline_spec_sha256"]
+        ):
+            raise Refused("live_spec_changed")
+        if (
+            deployment["status"].get("readyReplicas") != 1
+            or deployment["spec"]["strategy"]["type"] != "Recreate"
+        ):
+            raise Refused("single_ready_writer_required")
+        # Helm's three-way merge preserves unmanaged pod metadata. Bind the
+        # complete desired live spec rather than treating kubectl apply's
+        # last-applied projection as Helm's result.
+        image_patch = [
+            {"op": "test", "path": "/spec", "value": deployment["spec"]},
+            {"op": "replace", "path": "/spec/template/spec/containers/0/image",
+             "value": self.plan["candidate_image"]},
+        ]
+        preview = self.kdoc(
+            "patch", "deployment", "coulomb-social", "-n", "coulomb-social",
+            "--type=json", "--dry-run=server", "-p", json.dumps(image_patch),
+        )
+        if sha(canonical(preview["spec"])) != self.plan["candidate_spec_sha256"]:
+            raise Refused("image_only_server_preview_differs")
+        pvc = self.kdoc("get", "pvc", "coulomb-social-content", "-n", "coulomb-social")
+        if (
+            pvc["metadata"]["uid"] != self.plan["pvc_uid"]
+            or pvc["status"]["phase"] != "Bound"
+        ):
+            raise Refused("pvc_changed")
+        apps = self.kdoc("get", "applications.argoproj.io", "-A")["items"]
+        if any(
+            a.get("spec", {}).get("destination", {}).get("namespace")
+            == "coulomb-social"
+            for a in apps
+        ):
+            raise Refused("gitops_adoption_requires_owner_path")
+        rendered = self.command(
+            [
+                "helm",
+                "template",
+                "coulomb-social",
+                str(self.root / "charts/coulomb-social"),
+                "-f",
+                str(self.root / "release/identity-export-values.yaml"),
+                "-n",
+                "coulomb-social",
+            ]
+        ).stdout
+        if sha(rendered) != self.plan["candidate_render_sha256"]:
+            raise Refused("candidate_render_changed")
+        self.command(["kubectl", "diff", "-f", "-"], payload=rendered, allowed=(0, 1))
+        self.command(
+            ["kubectl", "apply", "--dry-run=server", "-f", "-"], payload=rendered
+        )
+        self.command(
+            [
+                "helm",
+                "upgrade",
+                "coulomb-social",
+                str(self.root / "charts/coulomb-social"),
+                "-f",
+                str(self.root / "release/identity-export-values.yaml"),
+                "-n",
+                "coulomb-social",
+                "--dry-run=server",
+            ]
+        )
+        before = self.probe()
+        self.same_data(before)
+        self.receipt["preflight"] = before
+        self.save()
+
+    def verify_installed_helpers(self):
+        for path, expected in self.plan["installed_helper_hashes"].items():
+            metadata = Path(path).lstat()
+            if (
+                not stat.S_ISREG(metadata.st_mode)
+                or metadata.st_uid != 0
+                or metadata.st_mode & 0o022
+                or sha(Path(path).read_bytes()) != expected
+            ):
+                raise Refused("installed_backup_helper_changed")
+    def capture(self):
+        self.verify_installed_helpers()
+        environment = dict(self.env)
+        for name in ("primary.env", "recipient.env"):
+            path = Path("/etc/coulomb-social-backup") / name
+            metadata = path.lstat()
+            if (
+                not stat.S_ISREG(metadata.st_mode)
+                or metadata.st_uid != 0
+                or metadata.st_mode & 0o077
+            ):
+                raise Refused("private_own_backup_configuration_required")
+            environment.update(
+                dict(
+                    line.split("=", 1)
+                    for line in path.read_text().splitlines()
+                    if line and not line.startswith("#")
+                )
+            )
+        output = self.command(
+            ["python3", "/opt/coulomb-social-backup/paired-backup.py", "--execute"],
+            env=environment,
+            timeout=1200,
+        )
+        point = json.loads(output.stdout)
+        if (
+            point.get("status") != "offsite_fetched_verified"
+            or not point.get("remote_retrieval_verified")
+            or not point.get("version_id")
+            or point.get("version_id") == "null"
+            or point.get("image") != self.plan["old_image"]
+            or point.get("age_recipient") != self.plan["public_recipient"]
+            or not point.get("immutable_create")
+            or not point.get("destination", "").startswith(
+                "s3://railiance-platform-pg-backup/platform-pg/application-archives/coulomb-social/"
+            )
+        ):
+            raise Refused("fresh_paired_primary_readback_required")
+        self.receipt["fresh_paired_point"] = point
+        self.save()
+        self.same_data(self.probe())
+
+    def restore_pin(self, old):
+        path = SNAPSHOT
+        if (
+            sha(path.read_bytes())
+            == self.plan["files"]["release/identity-export-snapshot.py"]
+        ):
+            self.replace_pin(path, old)
+        elif sha(path.read_bytes()) != sha(old):
+            raise Refused("snapshot_pin_changed_during_rollback")
+
+    def replace_pin(self, path, value):
+        fd, name = tempfile.mkstemp(prefix=".reviewed-snapshot-", dir=path.parent)
+        try:
+            with os.fdopen(fd, "wb") as stream:
+                stream.write(value)
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(name, 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(name):
+                os.unlink(name)
+
+    def execute(self):
+        began = time.monotonic()
+        self.preflight()
+        self.capture()
+        old_pin = SNAPSHOT.read_bytes()
+        self.receipt["status"] = "upgrade_started"
+        self.save()
+        try:
+            self.command(
+                [
+                    "helm",
+                    "upgrade",
+                    "coulomb-social",
+                    str(self.root / "charts/coulomb-social"),
+                    "-f",
+                    str(self.root / "release/identity-export-values.yaml"),
+                    "-n",
+                    "coulomb-social",
+                    "--atomic",
+                    "--wait",
+                    "--timeout",
+                    "5m",
+                ],
+                timeout=660,
+            )
+            after = self.deployment()
+            if (
+                sha(canonical(after["spec"])) != self.plan["candidate_spec_sha256"]
+                or after["status"].get("readyReplicas") != 1
+            ):
+                raise Refused("candidate_spec_or_readiness_differs")
+            probe = self.probe("views")
+            self.same_data(probe)
+            for suffix in ("/healthz", "/", "/static/spaces/copy-content.js"):
+                with urllib.request.urlopen(
+                    "https://app.coulomb.social" + suffix, timeout=20
+                ) as response:
+                    body = response.read(1024**2)
+                    if response.status != 200:
+                        raise Refused("public_http_postflight_refused")
+                    if (
+                        suffix.endswith("copy-content.js")
+                        and sha(body) != self.plan["clipboard_static_sha256"]
+                    ):
+                        raise Refused("served_static_differs")
+            path = SNAPSHOT
+            if (
+                sha(path.read_bytes())
+                != self.plan["installed_helper_hashes"][str(path)]
+            ):
+                raise Refused("snapshot_pin_drifted")
+            self.replace_pin(
+                path, (self.root / "release/identity-export-snapshot.py").read_bytes()
+            )
+            self.command(
+                [
+                    "python3",
+                    str(path),
+                    "--destination",
+                    "/tmp/csoc-post-image-server-review",
+                ]
+            )
+            self.receipt.update(
+                status="image_candidate_deployed_verified",
+                postflight=probe,
+                helm_revision=self.history()[-1]["revision"],
+                snapshot_pin_refreshed=True,
+                public_health_landing_static_verified=True,
+                existing_global_constraint_preserved=True,
+            )
+        except Exception:
+            self.receipt["rollback_requested"] = True
+            try:
+                self.command(
+                    [
+                        "helm",
+                        "rollback",
+                        "coulomb-social",
+                        str(self.plan["baseline_revision"]),
+                        "-n",
+                        "coulomb-social",
+                        "--wait",
+                        "--timeout",
+                        "5m",
+                    ],
+                    timeout=360,
+                )
+                self.restore_pin(old_pin)
+                restored = self.deployment()
+                if (sha(canonical(restored["spec"])) != self.plan["baseline_spec_sha256"]
+                        or restored["status"].get("readyReplicas") != 1):
+                    raise Refused("rollback_spec_or_readiness_differs")
+                self.receipt["image_rollback_completed"] = True
+            except Exception:  # noqa: BLE001 -- retain explicit unconfirmed rollback outcome
+                self.receipt["image_rollback_completed"] = False
+            raise
+        self.receipt.update(
+            elapsed_seconds=round(time.monotonic() - began, 3),
+            completed_at=datetime.now(UTC).isoformat(),
+            private_values_printed=False,
+        )
+        self.save()
+        return self.receipt
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--capsule", type=Path, required=True)
+    parser.add_argument("--expect-digest", required=True)
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--preflight", action="store_true")
+    args = parser.parse_args()
+    if not args.execute and not args.preflight:
+        print(json.dumps({"status": "prepared", "live_changes": False}))
+        return
+    if args.execute and args.preflight:
+        raise SystemExit("choose read-only preflight or attended execution")
+    release = Release(args.capsule, args.expect_digest, read_only=args.preflight)
+    try:
+        if args.preflight:
+            release.preflight()
+            print(
+                json.dumps(
+                    dict(
+                        release.receipt,
+                        status="live_release_preflight_verified",
+                        live_changes=False,
+                    )
+                )
+            )
+            return
+        print(json.dumps(release.execute()))
+    except Exception as error:  # noqa: BLE001 -- private child diagnostics are suppressed
+        release.receipt.update(
+            status="release_refused_or_failed", failure_type=type(error).__name__
+        )
+        release.save()
+        print(json.dumps(release.receipt))
+        raise SystemExit(1) from None
+
+
+if __name__ == "__main__":
+    main()