Prepare supervised Coulomb storage migration release and guarded node execution

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b

docs/coulomb-storage-release-2026-10-29.md

@@ -0,0 +1,85 @@
+# Storage/content hardening release — exact supervised proposal
+
+Prepared, not deployed. Decision 05c9b00f-e4e4-4593-bdeb-c63fe6a4fe15 is pending
+founder disposition. Earlier backup/custody/schedule approvals remain in force.
+This packet does not reopen them or approve Bubble/apex cutover.
+
+Proposal digest:
+`sha256:3389c506d86f1613c6567a2bba5c29ae22b812f19b011ebe454bb08af39a4132`.
+It binds 12 capsule members including all node/launcher/probe code, exact chart and
+values, dataset/schema expectations and the candidate snapshot-helper source.
+Any changed reviewed byte produces another digest and refuses execution.
+
+## Exact result
+
+Target: `ADMINISTER @ realm:kubernetes/railiance01`, namespace/release
+coulomb-social, current Helm revision 17. Candidate product source 17d0eb6 at
+`sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd`.
+The candidate contains global slug constraint 0003, tested content revision/recovery
+hardening and strict original-date import guards. Renderer policy is unchanged.
+
+The attended command rechecks cluster/node, exact live spec/PVC/Helm baseline,
+headroom, no Argo adoption, data equality and server/client dry-runs. It then runs
+the already admitted fresh paired backup and immutable primary readback, checks
+the dataset again, and performs Helm --atomic/--wait with Recreate. The three
+resource manifests change only the app image; the PVC, configuration, resources
+and secret references are preserved. Startup applies the atomic SQL constraint;
+it does not rename/merge/delete business rows or change grants.
+
+Read-only postflight checks schema, all business/content aggregate hashes, actual
+owner pages and unsaved-outsider denials, plus public health/landing/static bytes.
+Only then refresh the root-owned snapshot helper's image pin to the new digest
+and repeat its server dry-run. No private key/recovery pair is delivered to the
+worker; no email, caller, issuer, DNS, apex, timer or production-data restore.
+
+## Evidence and limits
+
+- Packaged real recovered dataset: forward/refusal/reverse/reapply preserve all
+  12 business tables/91 rows and82 files; candidate renders 39 pages with owner/viewer/
+  ungranted checks. Old image also reads the migrated data and denies outsiders.
+- Live data hashes match the independently protected recovered point. Current
+  image 9a646d8e, one Ready replica, revision 17, Recreate, Bound PVC, no pressure.
+  Requests:68% CPU / 76% memory. Helm lint passes; exact sealed live preflight passes.
+- Product make check: 246 passed, including mandatory browser journeys. Ten
+  synthetic executor cases cover tampering, data drift, replay refusal, staging
+  immutability, unattended-launch refusal, rollback and capture cleanup timeout.
+
+The fresh capture normally pauses/resumes in about 54 seconds; previous app
+upgrades took about 20 seconds. Helm has a five-minute wait plus rollback handling.
+Failure preserves post-capture writes and the fresh point. Automatic rollback
+restores the prior image/revision, not database contents; the added uniqueness
+constraint may remain. Its separate SQL inverse and old-image read compatibility
+are rehearsed. A failed/uncertain attempt requires journal inspection, not replay.
+
+No new node/PVC/service capacity is committed. One existing-size encrypted
+generation/readback and normal API/storage usage are added. No EUR autopilot
+budget or risk grant is claimed; execution remains supervised.
+
+## Founder execution
+
+Run in your own terminal from the prepared deployment checkout:
+
+```sh
+python3 /tmp/csoc-storage-release-owner-20261029/tools/coulomb-storage-release.py \
+  --execute \
+  --expect-digest sha256:3389c506d86f1613c6567a2bba5c29ae22b812f19b011ebe454bb08af39a4132 \
+  --output /tmp/csoc-storage-release-result-20261029.json
+```
+
+Review the digest and type DEPLOY followed by that full digest when prompted.
+The command uses the founder's existing SSH/sudo path and returns metadata only.
+No Scaleway login or credential pasted into chat is required. Reply done afterward;
+the agent must inspect the node receipt and postflight before recording acceptance.
+The durable node journal is
+`/var/lib/coulomb-social-backup/releases/<proposal-hex>/receipt.json`.
+
+This is an attended supervisor command, not a new credential/approval service.
+The active gated-executor profile is Kubernetes-only and cannot perform the
+root-owned node-helper refresh. Its configuration is left unchanged.
+
+The exact-action requirement comes from the-custodian's
+docs/agent-environment-orientation.md §4: “Every live change needs the founder's
+go-ahead”, and docs/agent-autonomy-decision.md: supervised privileged actions use
+an exact proposal; “The agent cannot approve itself.” This image/schema action
+has no current exact founder approval. Preparation and all observed checks above
+are complete; the app has not been changed by this packet.

helm/coulomb-social-values.yaml

@@ -3,8 +3,8 @@
 
 image:
   repository: forgejo.coulomb.social/coulomb/coulomb-social
-  tag: "dd83c90"
-  digest: "sha256:9a646d8e01a61ba5316d3c2eb0273064e17f2d0300ad07af766ccedda65ecb18"
+  tag: "17d0eb6"
+  digest: "sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd"
 
 # Public host: app.coulomb.social (parallel to Bubble on apex coulomb.social).
 # Apex cutover is deferred until data + self-registration are ready.

release/coulomb-storage-17d0eb6.json

@@ -0,0 +1,70 @@
+{
+  "schema": "coulomb.reviewed-storage-release.v1",
+  "proposal_id": "CSOC-STORAGE-20261029",
+  "scope": "ADMINISTER @ realm:kubernetes/railiance01; supervised founder execution only",
+  "cluster_uid": "a553c742-0115-43d4-99a4-a5ca56fe0786",
+  "baseline_revision": 17,
+  "baseline_helm_manifest_sha256": "da80414e13e7c5fd254803a241f21ade223b4b84e68ed908535e36e854b523a2",
+  "candidate_render_sha256": "865b6a22b3c50c65a7417bb80de609b7e0290f54a99edebff497b2259bc067a2",
+  "deployment_uid": "2dbe661a-d232-4477-bbd1-833813077fbe",
+  "pvc_uid": "9888abeb-9d52-4c70-a23d-b5494780d9c3",
+  "baseline_spec_sha256": "3185c30b966bc4a65d90e32dc8e0027bce957fc1fb66d10661e7e405ac215456",
+  "candidate_spec_sha256": "013aca79ec910b52ed2c2df7a18050c436751005320d1dec616590962da30b8e",
+  "old_image": "forgejo.coulomb.social/coulomb/coulomb-social@sha256:9a646d8e01a61ba5316d3c2eb0273064e17f2d0300ad07af766ccedda65ecb18",
+  "candidate_image": "forgejo.coulomb.social/coulomb/coulomb-social@sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd",
+  "product_source": "17d0eb67f397e3d1f5b9b63880df5054aa263c5a",
+  "expected_data": {
+    "business_tables": 12,
+    "business_rows": 91,
+    "business_rows_sha256": "1f404b7468011f8e006ed784628ade14bd03b53617282063f6d300df36f5090f",
+    "content_files": 82,
+    "content_files_sha256": "e3113909deccdab8135332c9e9fcacbed88f3216d9a1178f6b63da943aca3deb",
+    "global_slug_constraint": false,
+    "tenant_slug_constraint": true,
+    "migration_0003_applied": false,
+    "duplicate_slug_groups": 0
+  },
+  "public_recipient": "age1qm30jk8q0cezey652dk2dhcr3d9vs528v6lg8mavl6d4q3e3vgnqmrmyy3",
+  "installed_helper_hashes": {
+    "/opt/coulomb-social-backup/paired-backup.py": "43e119acb6a4933b4e144c349a091e074e1a91ab61dbe15a65c8264c1d47661a",
+    "/opt/coulomb-social-backup/snapshot-content-and-db.py": "f98604294fdba5f1664529bd932592a641a22bb73ec4583bbabba8a9591cf8e3",
+    "/opt/railiance-platform/scripts/primary_application_archive.py": "a3f0a61e869a6012edb4b5288be14e3b0c4a9a1273b090320ab146d12e5aa94d"
+  },
+  "clipboard_static_sha256": "347eabe0c8d93bf439df02b5a08a7ea71004600e6ff4e8a42cecbee52a9f0413",
+  "protected_recovered_point": {
+    "captured_at": "2026-10-29T21:22:46Z",
+    "object_key": "platform-pg/application-archives/coulomb-social/2026/10/29/20261029T212246Z-fd128164c7bb22a3a6c42acf.tar.age",
+    "version_id": "1793309020705518",
+    "ciphertext_sha256": "acbf92eda5074761bb6d00170e6e459e787359afcd377f2f18324ba17496510a",
+    "protected_kit": {
+      "automatic_expiry": false,
+      "bucket": "railiance-coulomb-social-recovery",
+      "bytes": 277585920,
+      "object_key": "kits/db05c8f3b5587dc1fc09a5352c317dd3357527dbe72366ab7ca1041db578c834/recovery-kit.tar",
+      "project_id": "06d107bc-9d84-4672-b2a0-95701cc0c07f",
+      "schema": "platform.coulomb-protected-recovery-kit.v1",
+      "sha256": "db05c8f3b5587dc1fc09a5352c317dd3357527dbe72366ab7ca1041db578c834",
+      "version_id": "1793309971213507"
+    },
+    "protected_recovery_proof": {
+      "project_id": "06d107bc-9d84-4672-b2a0-95701cc0c07f",
+      "bucket": "railiance-coulomb-social-recovery",
+      "object_key": "kits/receipts/a77477cac53086b83ad2cb512973ae0d1dfecba028a29a92e06a1bcfe5287f29/recovery-proof.json",
+      "version_id": "1793310476676486",
+      "sha256": "a77477cac53086b83ad2cb512973ae0d1dfecba028a29a92e06a1bcfe5287f29",
+      "bytes": 4237,
+      "independent_readback_verified": true,
+      "automatic_expiry": false
+    }
+  },
+  "fresh_backup_before_helm": true,
+  "requires_dataset_equality_before_and_after": true,
+  "node_request_percent_limit": 90,
+  "helm_atomic_timeout_seconds": 300,
+  "rollback": "Restore Helm revision17/image; preserve all post-capture data. Added uniqueness may remain. SQL inverse tested separately; no automatic production data restore.",
+  "new_resources_committed": 0,
+  "autopilot_or_eur_budget_claimed": false,
+  "approval_status": "awaiting_founder_execution",
+  "supervision_decision_id": "05c9b00f-e4e4-4593-bdeb-c63fe6a4fe15",
+  "current_preflight_is_read_only": true
+}

release/snapshot-content-and-db.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""Node-side quiesced coulomb-social snapshot. Binary output stays private.
+
+Run on railiance01 after reviewed live approval. Does not restore production.
+Default is server dry-run; --execute briefly pauses the app and captures a
+paired pg_dump/content archive. A separate isolated restore must validate it.
+"""
+from __future__ import annotations
+
+import argparse
+import fcntl
+import hashlib
+import json
+import os
+import signal
+import stat
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+NS = 'coulomb-social'
+IMAGE = ('forgejo.coulomb.social/coulomb/coulomb-social@sha256:'
+         '512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd')
+CLUSTER_UID = 'a553c742-0115-43d4-99a4-a5ca56fe0786'
+LOCK_PATH = Path('/var/lib/coulomb-social-backup/snapshot.lock')
+
+
+class Refused(RuntimeError):
+    pass
+
+
+def run(args: list[str], *, payload: bytes | None = None, timeout: int = 180) -> bytes:
+    p = subprocess.run(args, input=payload, capture_output=True, timeout=timeout)
+    if p.returncode:
+        raise Refused('command_failed')
+    return p.stdout
+
+
+def k(*args: str, payload: bytes | None = None) -> bytes:
+    return run(['kubectl', *args], payload=payload)
+
+
+def document(*args: str) -> dict:
+    return json.loads(k(*args, '-o', 'json'))
+
+
+def private_write(path: Path, value: bytes) -> None:
+    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(fd, 'wb') as f:
+        f.write(value)
+
+
+def snapshot() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--execute', action='store_true')
+    parser.add_argument('--destination', type=Path, required=True)
+    args = parser.parse_args()
+    os.environ['KUBECONFIG'] = '/etc/rancher/k3s/k3s.yaml'
+    started = time.monotonic()
+    if document('get', 'namespace', 'kube-system')['metadata']['uid'] != CLUSTER_UID:
+        raise Refused('wrong_cluster')
+    dep = document('get', 'deployment', 'coulomb-social', '-n', NS)
+    spec = dep['spec']['template']['spec']
+    if dep['spec']['replicas'] != 1 or spec['containers'][0]['image'] != IMAGE:
+        raise Refused('unexpected_release')
+    if document('get', 'pvc', 'coulomb-social-content', '-n', NS)['status']['phase'] != 'Bound':
+        raise Refused('pvc_not_bound')
+    cluster = document('get', 'cluster', 'apps-pg', '-n', 'databases')
+    primary = cluster['status']['currentPrimary']
+    name = 'csoc-paired-snapshot-' + time.strftime('%Y%m%d%H%M%S', time.gmtime())
+    pod = {
+        'apiVersion': 'v1', 'kind': 'Pod',
+        'metadata': {'name': name, 'namespace': NS, 'labels': {'csoc-recovery': name}},
+        'spec': {
+            'restartPolicy': 'Never', 'automountServiceAccountToken': False,
+            'securityContext': {'runAsNonRoot': True, 'runAsUser': 10001,
+                                'runAsGroup': 10001, 'fsGroup': 10001,
+                                'fsGroupChangePolicy': 'OnRootMismatch'},
+            'containers': [{'name': 'snapshot', 'image': IMAGE,
+                'command': ['python', '-c', 'import time;time.sleep(1800)'],
+                'resources': {'requests': {'cpu': '10m', 'memory': '64Mi'},
+                              'limits': {'cpu': '100m', 'memory': '256Mi'}},
+                'securityContext': {'allowPrivilegeEscalation': False,
+                                    'readOnlyRootFilesystem': True,
+                                    'capabilities': {'drop': ['ALL']}},
+                'volumeMounts': [{'name': 'content', 'mountPath': '/snapshot', 'readOnly': True}]}],
+            'volumes': [{'name': 'content', 'persistentVolumeClaim':
+                         {'claimName': 'coulomb-social-content', 'readOnly': True}}],
+        },
+    }
+    body = json.dumps(pod).encode()
+    k('create', '--dry-run=server', '-f', '-', payload=body)
+    if not args.execute:
+        print(json.dumps({'status': 'reviewed', 'server_dry_run': True,
+                          'image_pinned': True, 'pvc_read_only': True,
+                          'db_scope': 'coulomb_social_db', 'replicas_restored_to': 1}))
+        return 0
+    dest = args.destination
+    if not dest.is_absolute() or dest.exists():
+        raise Refused('destination_must_be_new_absolute_directory')
+    dest.mkdir(mode=0o700)
+    paused = False
+    created = False
+    receipt = {'status': 'incomplete', 'app': 'coulomb-social',
+               'database': 'coulomb_social_db', 'image': IMAGE,
+               'started_at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}
+    try:
+        paused = True
+        k('scale', 'deployment/coulomb-social', '-n', NS, '--replicas=0')
+        k('wait', '--for=delete', 'pod', '-n', NS,
+          '-l', 'app.kubernetes.io/instance=coulomb-social', '--timeout=120s')
+        created = True
+        k('create', '-f', '-', payload=body)
+        k('wait', '--for=condition=Ready', 'pod/' + name, '-n', NS, '--timeout=120s')
+        dump = k('exec', '-n', 'databases', primary, '-c', 'postgres', '--',
+                 'pg_dump', '-U', 'postgres', '-d', 'coulomb_social_db', '-Fc')
+        if not dump.startswith(b'PGDMP'):
+            raise Refused('invalid_pg_dump')
+        private_write(dest / 'database.dump', dump)
+        archive_code = ('import io,sys,tarfile; b=io.BytesIO(); '
+                        't=tarfile.open(fileobj=b,mode="w:gz"); '
+                        't.add("/snapshot",arcname="content"); '
+                        't.close();sys.stdout.buffer.write(b.getvalue())')
+        content = k('exec', '-n', NS, name, '--', 'python', '-c', archive_code)
+        private_write(dest / 'content.tar.gz', content)
+        receipt.update(status='captured', writers_quiesced=True,
+                       database_sha256=hashlib.sha256(dump).hexdigest(),
+                       content_sha256=hashlib.sha256(content).hexdigest(),
+                       database_bytes=len(dump), content_bytes=len(content))
+    finally:
+        try:
+            if created:
+                k('delete', 'pod', name, '-n', NS, '--wait=true', '--timeout=120s')
+        finally:
+            if paused:
+                k('scale', 'deployment/coulomb-social', '-n', NS, '--replicas=1')
+                k('rollout', 'status', 'deployment/coulomb-social', '-n', NS, '--timeout=180s')
+                receipt['app_resumed'] = True
+        receipt['elapsed_seconds'] = round(time.monotonic() - started, 2)
+        private_write(dest / 'receipt.json', json.dumps(receipt, indent=2).encode())
+    print(json.dumps(receipt))
+    return 0
+
+
+def main() -> int:
+    if '--execute' not in sys.argv:
+        return snapshot()
+    LOCK_PATH.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    parent = LOCK_PATH.parent.lstat()
+    if (not stat.S_ISDIR(parent.st_mode) or parent.st_uid != os.geteuid()
+            or parent.st_mode & 0o077):
+        raise Refused('lock_directory_not_private')
+    fd = os.open(LOCK_PATH, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(fd, 'w') as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            raise Refused('snapshot_already_running') from None
+        def terminate(signum, frame):
+            raise Refused('snapshot_interrupted')
+        previous = signal.signal(signal.SIGTERM, terminate)
+        try:
+            return snapshot()
+        finally:
+            signal.signal(signal.SIGTERM, previous)
+
+
+if __name__ == '__main__':
+    try:
+        sys.exit(main())
+    except Exception as exc:
+        # Never expose child output, kubectl objects, credentials or raw exception messages.
+        print(json.dumps({'status': 'failed', 'failure_type': type(exc).__name__}))
+        sys.exit(1)

tests/test_coulomb_storage_release.py

@@ -0,0 +1,142 @@
+"""Synthetic release safety cases; no cluster, credentials or real content."""
+import copy
+import importlib.util
+import json
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+SOURCE = Path(__file__).resolve().parents[1] / "tools/coulomb_storage_release_node.py"
+spec = importlib.util.spec_from_file_location("storage_release", SOURCE)
+release = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(release)
+
+launcher_spec = importlib.util.spec_from_file_location("storage_launcher", SOURCE.with_name("coulomb-storage-release.py"))
+launcher = importlib.util.module_from_spec(launcher_spec)
+launcher_spec.loader.exec_module(launcher)
+
+
+def capsule(tmp_path):
+    root = tmp_path / "capsule"
+    root.mkdir()
+    (root / "probe.py").write_bytes(b"synthetic read-only probe")
+    plan = {"files": {"probe.py": release.sha((root / "probe.py").read_bytes())}}
+    (root / "manifest.json").write_text(json.dumps(plan))
+    return root, "sha256:" + release.sha(release.canonical(plan))
+
+
+def test_capsule_tampering_refuses_before_privileged_work(tmp_path, monkeypatch):
+    root, digest = capsule(tmp_path)
+    monkeypatch.setattr(release.os, "geteuid", lambda: 0)
+    monkeypatch.setattr(release, "JOURNAL_ROOT", tmp_path / "journal")
+    (root / "probe.py").write_bytes(b"changed")
+    with pytest.raises(release.Refused, match="capsule_file_differs"):
+        release.Release(root, digest)
+    assert not (tmp_path / "journal").exists()
+
+
+def test_preflight_never_creates_an_execution_journal(tmp_path, monkeypatch):
+    root, digest = capsule(tmp_path)
+    monkeypatch.setattr(release.os, "geteuid", lambda: 0)
+    monkeypatch.setattr(release, "JOURNAL_ROOT", tmp_path / "journal")
+    instance = release.Release(root, digest, read_only=True)
+    instance.save()
+    assert not (tmp_path / "journal").exists()
+
+
+def test_prior_attempt_is_not_blindly_replayed(tmp_path, monkeypatch):
+    root, digest = capsule(tmp_path)
+    journal = tmp_path / "journal"
+    (journal / digest.removeprefix("sha256:")).mkdir(parents=True)
+    monkeypatch.setattr(release.os, "geteuid", lambda: 0)
+    monkeypatch.setattr(release, "JOURNAL_ROOT", journal)
+    with pytest.raises(release.Refused, match="prior_execution_requires_review"):
+        release.Release(root, digest)
+
+
+def test_mutable_staging_is_frozen_before_live_work(tmp_path, monkeypatch):
+    root, digest = capsule(tmp_path)
+    monkeypatch.setattr(release.os, "geteuid", lambda: 0)
+    monkeypatch.setattr(release, "JOURNAL_ROOT", tmp_path / "journal")
+    instance = release.Release(root, digest)
+    (root / "probe.py").write_bytes(b"changed after verification")
+    assert (instance.root / "probe.py").read_bytes() == b"synthetic read-only probe"
+    assert (instance.journal / "receipt.json").is_file()
+
+
+@pytest.mark.parametrize("field", ["business_rows_sha256", "content_files_sha256"])
+def test_dataset_drift_is_not_accepted(field):
+    instance = object.__new__(release.Release)
+    baseline = {"business_tables": 2, "business_rows": 3, "business_rows_sha256": "synthetic-a",
+                "content_files": 4, "content_files_sha256": "synthetic-b"}
+    instance.plan = {"expected_data": baseline}
+    changed = dict(baseline, duplicate_slug_groups=0, tenant_slug_constraint=True)
+    changed[field] = "different"
+    with pytest.raises(release.Refused, match="dataset_drifted"):
+        instance.same_data(changed)
+
+
+def test_failed_preflight_never_captures_or_upgrades():
+    instance = object.__new__(release.Release)
+    instance.preflight = Mock(side_effect=release.Refused("guard stopped"))
+    instance.capture = Mock()
+    instance.command = Mock()
+    with pytest.raises(release.Refused):
+        instance.execute()
+    instance.capture.assert_not_called()
+    instance.command.assert_not_called()
+
+
+def test_postflight_failure_rolls_back_image_without_restoring_data(tmp_path, monkeypatch):
+    pin = tmp_path / "snapshot.py"
+    pin.write_bytes(b"old pin")
+    monkeypatch.setattr(release, "SNAPSHOT", pin)
+    instance = object.__new__(release.Release)
+    instance.receipt = {}
+    instance.root = tmp_path
+    spec = {"synthetic": "candidate"}
+    baseline = {"business_tables": 2, "business_rows": 3, "business_rows_sha256": "synthetic-a",
+                "content_files": 4, "content_files_sha256": "synthetic-b"}
+    instance.plan = {"candidate_spec_sha256": release.sha(release.canonical(spec)), "baseline_revision": 17,
+                     "expected_data": baseline}
+    instance.preflight = Mock()
+    instance.capture = Mock()
+    instance.save = Mock()
+    instance.deployment = lambda: {"spec": copy.deepcopy(spec), "status": {"readyReplicas": 1}}
+    instance.probe = lambda mode: dict(baseline, business_rows_sha256="changed")
+    instance.command = Mock(return_value=SimpleNamespace(stdout=b"", returncode=0))
+    instance.restore_pin = Mock()
+    with pytest.raises(release.Refused, match="dataset_drifted"):
+        instance.execute()
+    commands = [call.args[0] for call in instance.command.call_args_list]
+    assert commands[0][:2] == ["helm", "upgrade"] and "--atomic" in commands[0]
+    assert commands[1][:4] == ["helm", "rollback", "coulomb-social", "17"]
+    assert not any("pg_restore" in command for command in commands)
+    instance.restore_pin.assert_called_once_with(b"old pin")
+    assert instance.receipt["image_rollback_completed"] is True
+
+
+def test_live_launcher_rejects_unattended_execution_before_ssh(monkeypatch):
+    _, digest, _ = launcher.prepare()
+    monkeypatch.setattr(launcher.sys, "argv", ["release", "--execute", "--expect-digest", digest])
+    monkeypatch.setattr(launcher.sys, "stdin", SimpleNamespace(isatty=lambda: False))
+    command = Mock()
+    monkeypatch.setattr(launcher.subprocess, "run", command)
+    with pytest.raises(SystemExit, match="own terminal"):
+        launcher.main()
+    command.assert_not_called()
+
+
+def test_timeout_gives_the_existing_capture_resume_window(monkeypatch):
+    instance = object.__new__(release.Release)
+    instance.env = {}
+    process = Mock()
+    process.communicate.side_effect = release.subprocess.TimeoutExpired(["synthetic"], 1)
+    monkeypatch.setattr(release.subprocess, "Popen", Mock(return_value=process))
+    with pytest.raises(release.Refused, match="cleanup_window"):
+        instance.command(["synthetic"], timeout=1)
+    process.terminate.assert_called_once()
+    process.wait.assert_called_once_with(timeout=450)
+    process.kill.assert_not_called()

tools/coulomb-storage-release.py

@@ -0,0 +1,181 @@
+"""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-storage-17d0eb6.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-storage-release.py",
+            "tools/coulomb_storage_release_node.py",
+            "tools/coulomb_storage_probe.py",
+            "helm/coulomb-social-values.yaml",
+            "release/snapshot-content-and-db.py",
+        )
+    ]
+    paths += sorted((ROOT / "charts/coulomb-social").rglob("*"))
+    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()
+    }
+    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/migration, 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-storage-" + 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_storage_release_node.py"]
+    guard = (
+        "import hashlib,pathlib,sys;p=pathlib.Path("
+        + repr(remote + "/tools/coulomb_storage_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_storage_probe.py

@@ -0,0 +1,130 @@
+"""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
+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),
+    "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_storage_release_node.py

@@ -0,0 +1,505 @@
+"""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-storage-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_storage_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",
+        ):
+            if actual[name] != self.plan["expected_data"][name]:
+                raise Refused("approved_recovered_dataset_drifted")
+        if actual["duplicate_slug_groups"] != 0 or not actual["tenant_slug_constraint"]:
+            raise Refused("storage_invariant_refused")
+
+    def preflight(self):
+        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")
+        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 / "helm/coulomb-social-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 / "helm/coulomb-social-values.yaml"),
+                "-n",
+                "coulomb-social",
+                "--dry-run=server",
+            ]
+        )
+        before = self.probe()
+        self.same_data(before)
+        if before["global_slug_constraint"] or before["migration_0003_applied"]:
+            raise Refused("migration_baseline_changed")
+        self.receipt["preflight"] = before
+        self.save()
+
+    def capture(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")
+        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/snapshot-content-and-db.py"]
+        ):
+            self.replace_pin(path, old)
+
+    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 / "helm/coulomb-social-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)
+            if (
+                not probe["global_slug_constraint"]
+                or not probe["migration_0003_applied"]
+            ):
+                raise Refused("production_migration_not_verified")
+            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/snapshot-content-and-db.py").read_bytes()
+            )
+            self.command(
+                [
+                    "python3",
+                    str(path),
+                    "--destination",
+                    "/tmp/csoc-post-storage-server-review",
+                ]
+            )
+            self.receipt.update(
+                status="storage_candidate_deployed_verified",
+                postflight=probe,
+                helm_revision=self.history()[-1]["revision"],
+                snapshot_pin_refreshed=True,
+                public_health_landing_static_verified=True,
+                global_constraint_applied=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)
+                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()