Prepare coherent encrypted offsite backup operations and retry recovery

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a11538-05ea-7b63-b3c7-a2df5f38898d

deploy/backup/coulomb-social-backup-freshness.service

@@ -0,0 +1,11 @@
+[Unit]
+Description=Check coulomb-social paired backup age and local encrypted hash
+
+[Service]
+Type=oneshot
+User=root
+ExecStart=/usr/bin/python3 /opt/coulomb-social-backup/paired-backup.py --check-freshness
+NoNewPrivileges=true
+ProtectSystem=strict
+ProtectHome=true
+

deploy/backup/coulomb-social-backup-freshness.timer

@@ -0,0 +1,10 @@
+[Unit]
+Description=Hourly freshness check for coulomb-social paired backups
+
+[Timer]
+OnCalendar=hourly
+Persistent=true
+Unit=coulomb-social-backup-freshness.service
+
+[Install]
+WantedBy=timers.target

deploy/backup/coulomb-social-paired-backup.service

@@ -0,0 +1,23 @@
+[Unit]
+Description=Coherent encrypted coulomb-social database and content backup
+After=network-online.target k3s.service
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+User=root
+UMask=0077
+StateDirectory=coulomb-social-backup
+StateDirectoryMode=0700
+# Owner-delivered credential references only; values never enter this repo.
+EnvironmentFile=/etc/coulomb-social-backup/offsite.env
+ExecStart=/usr/bin/python3 /opt/coulomb-social-backup/paired-backup.py --execute
+TimeoutStartSec=30min
+TimeoutStopSec=5min
+KillMode=mixed
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ReadWritePaths=/var/lib/coulomb-social-backup
+ProtectHome=true
+

deploy/backup/coulomb-social-paired-backup.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=Daily paired coulomb-social recovery point (target RPO 24h)
+
+[Timer]
+OnCalendar=*-*-* 03:15:00 UTC
+Persistent=true
+RandomizedDelaySec=5min
+Unit=coulomb-social-paired-backup.service
+
+[Install]
+WantedBy=timers.target

docs/release/paired-backup-operations-2026-10-07.md

@@ -0,0 +1,99 @@
+# Paired backup operations candidate — 2026-10-07
+
+CSOC-WP-0008-T02 now has an executable encrypted offsite publication wrapper and
+schedule. Platform admission is proposed in RPF-WP-0053. No timer was installed
+or enabled, no credential copied and no production archive uploaded this session.
+
+Product source owns coherent capture and app checks; railiance-platform owns
+execution admission, offsite custody, escrow and monitoring. Reused Option A
+library: railiance-platform commit 22651d851f071b92ab316f82eca8aef949f5aba1,
+lib/railiance-backup-common.sh. Routing: warden route show
+railiance-backup-offsite-lane. Its workload identity remains unresolved in the
+catalog and must be admitted before unattended operation.
+
+## Proposed contract
+
+| Item | Candidate contract |
+| --- | --- |
+| Target | railiance01; kube-system UID a553c742-0115-43d4-99a4-a5ca56fe0786 |
+| Data | coulomb_social_db on apps-pg and entire coulomb-social-content PVC |
+| Coherence | Exact app digest, one replica, scale to zero, capture both stores, resume before encryption/upload |
+| Schedule | Daily 03:15 UTC with up to five minutes jitter; persistent timer |
+| Targets | Proposed RPO 24h; provisional attended RTO 60min; acceptance/full measurement open |
+| Offsite | Independent coulomb-social prefix; directory, overwrite and retrieval await admission |
+| Retention | Fourteen rotating daily slots and four Sunday weekly slots; one encrypted pair per slot |
+| Health | Hourly check fails at age >26h or local encrypted hash mismatch; evidence reception unproven |
+| Escrow | Worker gets public age recipient; private identity remains with recovery custody |
+
+Slots bound artifact count using the file-drop lane without listing/deletion
+credentials. Normal rotation covers fourteen daily and four weekly points.
+Missed runs leave older points in their slots; strict age expiry needs a separate
+provider policy. Unrelated prefixes/manual snapshots are outside this policy.
+The node service candidate uses root/k3s admin guarded to this app/database;
+platform may choose its bounded activity-core executor instead. This is pending
+owner/founder admission, not an unattended grant. No live control was relaxed.
+
+## Implementation and validation
+
+scripts/paired-backup.py validates hashes, complete capture and app resume, wraps
+database.dump/content.tar.gz/receipt.json into one private tar, encrypts using
+age and uploads only ciphertext through the platform helper. last-success.json
+advances only after all required uploads succeed. remote_retrieval_verified=false
+explicitly distinguishes PUT acceptance from independent readback/recovery.
+
+Publication and direct capture both take nonblocking locks. Snapshot termination
+attempts cleanup/resume; the parent permits a cleanup window before killing a
+stuck child. Host loss/SIGKILL still needs a custodian replica check. Failure
+retains one private pending directory and blocks another unattended capture.
+Reviewed --execute --resume-pending republishes that valid pair without another
+pause. Success clears only its own staging directory and keeps bounded local
+encrypted slots; older manual private pairs remain intact.
+
+Validation: make check passed Ruff and all 140 tests, including mandatory Chromium
+journeys. Eighteen focused backup/restore cases cover corruption, overlap,
+termination/resume, partial daily/weekly upload, freshness, slot rotation across
+year boundaries and retry without recapture. Real age encrypt/decrypt roundtrip
+with an ephemeral synthetic identity matched both stores/receipt hashes.
+systemd-analyze verify passed all four units. Node server dry-run passed on the
+current production digest/PVC without pause/capture/upload/helper pod creation.
+Earlier production-pair isolated restore matched 13 tables, 104 rows, sequences
+and 75 files (private-visuals-2026-10-07.md). Its 4.14 seconds measures restoration
+only, not full RTO. No production data or escrow identity entered regression tests.
+
+## Activation and recovery
+
+RPF-WP-0053-T01/T02 must admit execution identity, prefix/overwrite contract and
+bounded custody. Node candidate expects platform-delivered
+RAILIANCE_BACKUP_NC_TOKEN and RAILIANCE_BACKUP_NC_WEBDAV_URL in root-owned
+mode0600 /etc/coulomb-social-backup/offsite.env. Never copy another workload's
+Secret or emit that file. Do not deliver AGE_PRIVATE_KEY to the worker.
+
+After exact founder activation, install pinned scripts in
+/opt/coulomb-social-backup/{paired-backup.py,snapshot-content-and-db.py}, reviewed
+library at /opt/railiance-platform/lib/railiance-backup-common.sh, age, kubectl
+and four deploy/backup units through the admitted platform path. Use root0700
+/var/lib/coulomb-social-backup. Record installed hashes, custody-delivery reference
+and admission decision. App releases must update/install the helper's expected
+digest; mismatch refuses capture.
+
+First run without --execute for a server dry-run. Perform one approved service
+run, inspect redacted receipt/readiness and independently retrieve ciphertext.
+Compare its SHA-256 with the evidence receipt; custody decrypts in an attended
+private workspace. Unpack only the three expected regular files, rejecting
+links/duplicates/other names; verify paired hashes; run
+scripts/restore-paired-snapshot.py. It restores only an internal Docker database
+and safe content tree, compares all COPY rows/sequences and files, and publishes
+no ports. Rehearse app migrations/rendering/owner and denied ungranted reads.
+Record retrieval/decryption/restore/app-ready time against the proposed RTO.
+Production restoration is a separate approved incident action. Preserve failed
+drill evidence/source ciphertext; never replace production after a failed drill.
+
+Enable timers only under the admitted schedule. Connect failed/freshness units to
+the established evidence owner and prove reception of a synthetic stale signal.
+Abort by disabling the timer and stopping the service with its five-minute
+cleanup window; confirm one Ready app replica and no helper pod before release
+work. Preserve pending pairs/last-success receipts. Image rollback does not
+require production database restore by default.
+
+T02 stays progress until custody, offsite readback/restore, monitored operation,
+recovery ownership and the service contract are accepted.

scripts/paired-backup.py

@@ -0,0 +1,231 @@
+#!/usr/bin/env python3
+"""Node-side scheduled paired backup using the platform age/WebDAV lane.
+
+Default: preflight only. Live execution requires approval of the schedule,
+quiescence and offsite consumer. No escrow key is needed to make a backup.
+"""
+from __future__ import annotations
+
+import argparse
+import fcntl
+import hashlib
+import json
+import os
+import shutil
+import signal
+import stat
+import subprocess
+import sys
+import tarfile
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+
+
+class Refused(RuntimeError):
+    pass
+
+
+def digest(path: Path) -> str:
+    with path.open('rb') as stream:
+        return hashlib.file_digest(stream, 'sha256').hexdigest()
+
+
+def validate_pair(directory: Path) -> dict:
+    receipt = json.loads((directory / 'receipt.json').read_text())
+    if (receipt.get('status') != 'captured' or not receipt.get('writers_quiesced')
+            or not receipt.get('app_resumed') or receipt.get('database') != 'coulomb_social_db'):
+        raise Refused('pair_incomplete')
+    for filename, prefix in [('database.dump', 'database'), ('content.tar.gz', 'content')]:
+        path = directory / filename
+        if path.is_symlink() or not path.is_file():
+            raise Refused('pair_invalid')
+        if digest(path) != receipt[prefix + '_sha256']:
+            raise Refused('pair_hash_mismatch')
+        if path.stat().st_size != receipt[prefix + '_bytes']:
+            raise Refused('pair_size_mismatch')
+    return receipt
+
+
+def retention_slots(now: datetime) -> list[str]:
+    # Fixed slots work with the platform file-drop lane: no listing/deletion
+    # credential is introduced. Daily slots hold fourteen calendar days;
+    # Sunday weekly slots hold four completed weeks. Uploads replace a slot.
+    names = [f'csoc-pair-daily-{now.date().toordinal() % 14:02d}.tar.age']
+    if now.isoweekday() == 7:
+        week = now.date().toordinal() // 7
+        names.append(f'csoc-pair-weekly-{week % 4:02d}.tar.age')
+    return names
+
+
+def command(argv: list[str], *, env=None, timeout=900) -> None:
+    # File-backed output avoids pipe deadlocks and keeps binary/private child
+    # output out of diagnostics. Termination gives the snapshot its resume window.
+    with tempfile.TemporaryFile() as output:
+        process = subprocess.Popen(argv, env=env, stdout=output, stderr=output)
+        try:
+            result = process.wait(timeout=timeout)
+        except BaseException:
+            process.terminate()
+            try:
+                process.wait(timeout=400)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+            raise
+        if result:
+            raise Refused('backup_command_failed')
+
+
+def upload(encrypted: Path, name: str, common: Path) -> None:
+    script = ('source "$1"; railiance_backup_require_openbao_lane; '
+              'RAILIANCE_BACKUP_NC_PREFIX=coulomb-social; '
+              'railiance_backup_nc_upload "$2" "$3"')
+    command(['bash', '-euo', 'pipefail', '-c', script, 'paired-backup',
+             str(common), str(encrypted), name])
+
+
+def write_receipt(path: Path, receipt: dict) -> None:
+    fd, name = tempfile.mkstemp(prefix='receipt-', dir=path.parent)
+    temporary = Path(name)
+    with os.fdopen(fd, 'w') as stream:
+        json.dump(receipt, stream, indent=2)
+        stream.flush()
+        os.fsync(stream.fileno())
+    temporary.replace(path)
+
+
+def archive_pair(directory: Path, archive: Path) -> None:
+    with archive.open('xb') as stream:
+        os.chmod(archive, 0o600)
+        with tarfile.open(fileobj=stream, mode='w') as bundle:
+            for name in ['database.dump', 'content.tar.gz', 'receipt.json']:
+                bundle.add(directory / name, arcname=name, recursive=False)
+
+
+def run_backup(state: Path, common: Path, recipient: str, now: datetime, *, resume=False) -> dict:
+    # One owner-only state directory, one process, and one recovery point.
+    state.mkdir(mode=0o700, parents=True, exist_ok=True)
+    info = state.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
+        raise Refused('state_directory_not_private')
+    fd = os.open(state / 'backup.lock', 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('backup_already_running') from None
+        pending = list(state.glob('pending-*'))
+        if resume:
+            if len(pending) != 1 or pending[0].is_symlink():
+                raise Refused('one_pending_pair_required')
+            stage = pending[0]
+        else:
+            if pending:
+                raise Refused('pending_backup_requires_review')
+            stage = Path(tempfile.mkdtemp(prefix='pending-', dir=state))
+        pair = stage / 'pair'
+        helper = Path(__file__).with_name('snapshot-content-and-db.py')
+        if not resume:
+            command([sys.executable, str(helper), '--execute', '--destination', str(pair)])
+        captured = validate_pair(pair)
+        publication = Path(tempfile.mkdtemp(prefix='publication-', dir=stage))
+        archive = publication / 'pair.tar'
+        archive_pair(pair, archive)
+        encrypted = publication / 'pair.tar.age'
+        command(['age', '-r', recipient, '-o', str(encrypted), str(archive)])
+        if not encrypted.is_file() or encrypted.stat().st_size == 0:
+            raise Refused('encryption_missing')
+        os.chmod(encrypted, 0o600)
+        captured_at = datetime.fromisoformat(captured['started_at'].replace('Z', '+00:00'))
+        names = retention_slots(captured_at)
+        for name in names:
+            upload(encrypted, name, common)
+        receipt = {
+            'status': 'offsite_provider_accepted',
+            'captured_at': captured['started_at'],
+            'completed_at': datetime.now(UTC).isoformat(),
+            'image': captured['image'], 'slots': names,
+            'encrypted_sha256': digest(encrypted),
+            'encrypted_bytes': encrypted.stat().st_size,
+            'database_sha256': captured['database_sha256'],
+            'content_sha256': captured['content_sha256'],
+            'pause_and_capture_seconds': captured['elapsed_seconds'],
+            'remote_retrieval_verified': False,
+        }
+        # Keep matching local encrypted slots; replace only after all uploads
+        # succeeded. Failed attempts keep their private pair for investigation.
+        for name in names:
+            target = state / name
+            temporary = state / (name + '.new')
+            shutil.copyfile(encrypted, temporary)
+            os.chmod(temporary, 0o600)
+            temporary.replace(target)
+        write_receipt(state / 'last-success.json', receipt)
+        shutil.rmtree(stage)
+        return receipt
+
+
+def check_freshness(state: Path, now: datetime, hours: int = 26) -> dict:
+    receipt = json.loads((state / 'last-success.json').read_text())
+    captured = datetime.fromisoformat(receipt['captured_at'].replace('Z', '+00:00'))
+    age = (now - captured).total_seconds()
+    if age < 0 or age > hours * 3600:
+        raise Refused('backup_stale')
+    if receipt.get('status') != 'offsite_provider_accepted':
+        raise Refused('backup_not_uploaded')
+    for name in receipt['slots']:
+        if name not in {f'csoc-pair-{kind}-{number:02d}.tar.age'
+                        for kind, count in [('daily', 14), ('weekly', 4)]
+                        for number in range(count)}:
+            raise Refused('invalid_slot')
+        if digest(state / name) != receipt['encrypted_sha256']:
+            raise Refused('local_encrypted_hash_mismatch')
+    return {'status': 'fresh', 'age_hours': round(age / 3600, 2),
+            'remote_retrieval_verified': receipt['remote_retrieval_verified']}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--execute', action='store_true')
+    parser.add_argument('--resume-pending', action='store_true',
+                        help='Publish the one reviewed pending pair without pausing the app again')
+    parser.add_argument('--check-freshness', action='store_true')
+    parser.add_argument('--state-dir', type=Path, default=Path('/var/lib/coulomb-social-backup'))
+    parser.add_argument('--platform-common', type=Path,
+                        default=Path('/opt/railiance-platform/lib/railiance-backup-common.sh'))
+    parser.add_argument('--age-recipient', default=
+                        'age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz')
+    args = parser.parse_args()
+    if args.resume_pending and not args.execute:
+        raise Refused('resume_requires_execute')
+    now = datetime.now(UTC)
+    if args.check_freshness:
+        print(json.dumps(check_freshness(args.state_dir, now)))
+    elif args.execute:
+        if not args.state_dir.is_absolute() or not args.platform_common.is_file():
+            raise Refused('invalid_backup_configuration')
+        # Fail before pausing the app if encryption or credential custody is absent.
+        command(['age', '--version'])
+        command(['bash', '-euo', 'pipefail', '-c',
+                 'source "$1"; railiance_backup_require_openbao_lane',
+                 'paired-backup', str(args.platform_common)])
+        print(json.dumps(run_backup(args.state_dir, args.platform_common, args.age_recipient,
+                                    now, resume=args.resume_pending)))
+    else:
+        helper = Path(__file__).with_name('snapshot-content-and-db.py')
+        command([sys.executable, str(helper), '--destination', '/tmp/csoc-backup-review'])
+        print(json.dumps({'status': 'reviewed', 'live_changes': False,
+                          'offsite_auth_checked': False, 'slots': retention_slots(now)}))
+    return 0
+
+
+if __name__ == '__main__':
+    try:
+        def terminate(signum, frame):
+            raise Refused('backup_interrupted')
+        signal.signal(signal.SIGTERM, terminate)
+        sys.exit(main())
+    except Exception as exc:
+        print(json.dumps({'status': 'failed', 'failure_type': type(exc).__name__}))
+        sys.exit(1)

scripts/snapshot-content-and-db.py

@@ -8,9 +8,12 @@ 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
@@ -20,6 +23,7 @@ NS = 'coulomb-social'
 IMAGE = ('forgejo.coulomb.social/coulomb/coulomb-social@sha256:'
          '91673ba7625f905c7a4d0c3d52b45ccc7e7746f387d1e84aab331d68994c5df9')
 CLUSTER_UID = 'a553c742-0115-43d4-99a4-a5ca56fe0786'
+LOCK_PATH = Path('/var/lib/coulomb-social-backup/snapshot.lock')
 
 
 class Refused(RuntimeError):
@@ -47,7 +51,7 @@ def private_write(path: Path, value: bytes) -> None:
         f.write(value)
 
 
-def main() -> int:
+def snapshot() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--execute', action='store_true')
     parser.add_argument('--destination', type=Path, required=True)
@@ -139,6 +143,29 @@ def main() -> int:
     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())

tests/test_paired_backup.py

@@ -0,0 +1,161 @@
+"""Paired backup publication must never declare partial or stale recovery fresh."""
+import fcntl
+import hashlib
+import importlib.util
+import json
+import tarfile
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def backup():
+    path = Path(__file__).resolve().parents[1] / 'scripts/paired-backup.py'
+    spec = importlib.util.spec_from_file_location('csoc_paired_backup', path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+NOW = datetime(2026, 10, 11, 3, 15, tzinfo=UTC)  # Sunday: daily + weekly
+
+
+def pair(directory):
+    directory.mkdir(mode=0o700)
+    receipt = {
+        'status': 'captured', 'writers_quiesced': True, 'app_resumed': True,
+        'database': 'coulomb_social_db', 'image': 'synthetic@sha256:fixture',
+        'started_at': NOW.isoformat(), 'elapsed_seconds': 45,
+    }
+    for name, key, data in [('database.dump', 'database', b'PGDMPsynthetic'),
+                            ('content.tar.gz', 'content', b'synthetic-content')]:
+        (directory / name).write_bytes(data)
+        (directory / name).chmod(0o600)
+        receipt[key + '_bytes'] = len(data)
+        receipt[key + '_sha256'] = hashlib.sha256(data).hexdigest()
+    (directory / 'receipt.json').write_text(json.dumps(receipt))
+    (directory / 'receipt.json').chmod(0o600)
+    return receipt
+
+
+def setup(backup, monkeypatch, tmp_path, failure=None):
+    state = tmp_path / 'state'
+    calls, uploads = [], []
+
+    def command(argv, **kwargs):
+        calls.append(argv)
+        if '--execute' in argv:
+            if failure == 'capture':
+                raise backup.Refused('capture_failed')
+            pair(Path(argv[-1]))
+        else:
+            assert argv[:3] == ['age', '-r', 'age1-public-test-key']
+            with tarfile.open(argv[-1]) as bundle:
+                assert set(bundle.getnames()) == {'database.dump', 'content.tar.gz', 'receipt.json'}
+                assert bundle.extractfile('database.dump').read() == b'PGDMPsynthetic'
+            if failure == 'encrypt':
+                raise backup.Refused('encryption_failed')
+            Path(argv[4]).write_bytes(b'age-encryption.org/v1\nsynthetic-ciphertext')
+
+    def upload(encrypted, name, common):
+        assert encrypted.stat().st_mode & 0o777 == 0o600
+        assert encrypted.read_bytes().startswith(b'age-encryption.org/')
+        uploads.append(name)
+        if failure == 'upload' or (failure == 'weekly' and 'weekly' in name):
+            raise backup.Refused('upload_failed')
+
+    monkeypatch.setattr(backup, 'command', command)
+    monkeypatch.setattr(backup, 'upload', upload)
+    return state, calls, uploads
+
+
+def test_success_is_one_encrypted_pair_in_bounded_slots(backup, monkeypatch, tmp_path):
+    state, calls, uploads = setup(backup, monkeypatch, tmp_path)
+    receipt = backup.run_backup(state, tmp_path / 'common.sh', 'age1-public-test-key', NOW)
+    assert receipt['status'] == 'offsite_provider_accepted'
+    assert not receipt['remote_retrieval_verified']
+    assert uploads == receipt['slots'] and len(uploads) == 2
+    assert len(calls) == 2
+    assert not list(state.glob('pending-*'))
+    assert not list(state.rglob('database.dump'))
+    assert not list(state.rglob('pair.tar'))
+    assert state.stat().st_mode & 0o777 == 0o700
+    assert (state / 'last-success.json').stat().st_mode & 0o777 == 0o600
+    assert backup.check_freshness(state, NOW + timedelta(hours=25))['status'] == 'fresh'
+    with pytest.raises(backup.Refused, match='backup_stale'):
+        backup.check_freshness(state, NOW + timedelta(hours=27))
+    (state / uploads[0]).write_bytes(b'corrupt')
+    with pytest.raises(backup.Refused, match='local_encrypted_hash_mismatch'):
+        backup.check_freshness(state, NOW)
+
+
+@pytest.mark.parametrize('failure', ['capture', 'encrypt', 'upload', 'weekly'])
+def test_partial_failures_never_advance_success_and_preserve_private_evidence(
+        backup, monkeypatch, tmp_path, failure):
+    state, _, _ = setup(backup, monkeypatch, tmp_path, failure)
+    with pytest.raises(backup.Refused):
+        backup.run_backup(state, tmp_path / 'common.sh', 'age1-public-test-key', NOW)
+    assert not (state / 'last-success.json').exists()
+    pending = list(state.glob('pending-*'))
+    assert len(pending) == 1 and pending[0].stat().st_mode & 0o777 == 0o700
+    # An unattended failure cannot grow private plaintext copies indefinitely.
+    with pytest.raises(backup.Refused, match='pending_backup_requires_review'):
+        backup.run_backup(state, tmp_path / 'common.sh', 'age1-public-test-key', NOW)
+    assert len(list(state.glob('pending-*'))) == 1
+
+
+def test_live_lock_and_nonprivate_state_refuse_before_capture(backup, monkeypatch, tmp_path):
+    state, calls, _ = setup(backup, monkeypatch, tmp_path)
+    state.mkdir(mode=0o700)
+    with (state / 'backup.lock').open('w') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        with pytest.raises(backup.Refused, match='backup_already_running'):
+            backup.run_backup(state, tmp_path / 'common.sh', 'age1-public-test-key', NOW)
+    state.chmod(0o755)
+    with pytest.raises(backup.Refused, match='state_directory_not_private'):
+        backup.run_backup(state, tmp_path / 'common.sh', 'age1-public-test-key', NOW)
+    assert calls == []
+
+
+def test_pair_hashes_and_resume_receipt_required(backup, tmp_path):
+    directory = tmp_path / 'pair'
+    receipt = pair(directory)
+    assert backup.validate_pair(directory) == receipt
+    (directory / 'database.dump').write_bytes(b'corrupt')
+    with pytest.raises(backup.Refused, match='pair_hash_mismatch'):
+        backup.validate_pair(directory)
+    receipt['app_resumed'] = False
+    (directory / 'receipt.json').write_text(json.dumps(receipt))
+    with pytest.raises(backup.Refused, match='pair_incomplete'):
+        backup.validate_pair(directory)
+
+
+def test_retention_covers_fourteen_days_four_weeks_and_year_boundary(backup):
+    names = set()
+    for day in range(365):
+        slots = backup.retention_slots(datetime(2026, 12, 20, tzinfo=UTC)
+                                       + timedelta(days=day))
+        names.update(slots)
+        assert len(slots) in {1, 2}
+    assert len(names) == 18
+    assert len({backup.retention_slots(NOW + timedelta(days=day))[0]
+                for day in range(14)}) == 14
+    assert backup.retention_slots(NOW)[0] == backup.retention_slots(NOW + timedelta(days=14))[0]
+    assert len({backup.retention_slots(NOW + timedelta(weeks=week))[1]
+                for week in range(4)}) == 4
+
+
+def test_reviewed_pending_pair_can_publish_without_recapture(backup, monkeypatch, tmp_path):
+    state, _, _ = setup(backup, monkeypatch, tmp_path, 'upload')
+    with pytest.raises(backup.Refused):
+        backup.run_backup(state, tmp_path / 'common.sh', 'age1-public-test-key', NOW)
+    assert list(state.glob('pending-*/pair/database.dump'))
+    _, calls, uploads = setup(backup, monkeypatch, tmp_path)
+    receipt = backup.run_backup(state, tmp_path / 'common.sh', 'age1-public-test-key',
+                                NOW + timedelta(hours=1), resume=True)
+    assert len(calls) == 1 and calls[0][0] == 'age'
+    assert receipt['captured_at'] == NOW.isoformat()
+    assert len(uploads) == 2
+    assert not list(state.glob('pending-*'))

tests/test_snapshot_recovery.py

@@ -1,6 +1,8 @@
 """Recovery must resume the app even when snapshot capture or cleanup fails."""
+import fcntl
 import importlib.util
 import json
+import signal
 import sys
 from pathlib import Path
 
@@ -45,6 +47,7 @@ def setup_fake(helper, monkeypatch, tmp_path, failure=None, execute=True):
     monkeypatch.setattr(sys, 'argv', argv)
     monkeypatch.setattr(helper, 'document', document)
     monkeypatch.setattr(helper, 'k', k)
+    monkeypatch.setattr(helper, 'LOCK_PATH', tmp_path / 'snapshot.lock')
     return calls, dest
 
 
@@ -77,3 +80,27 @@ def test_success_artifacts_private_and_paired(helper, monkeypatch, tmp_path):
     assert receipt['status'] == 'captured'
     assert receipt['database'] == 'coulomb_social_db'
     assert not any('pg_restore' in c for c in calls)
+
+
+def test_second_snapshot_refuses_without_pausing(helper, monkeypatch, tmp_path):
+    calls, _ = setup_fake(helper, monkeypatch, tmp_path)
+    with helper.LOCK_PATH.open('w') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        with pytest.raises(helper.Refused, match='snapshot_already_running'):
+            helper.main()
+    assert calls == []
+
+
+def test_termination_during_capture_resumes_app(helper, monkeypatch, tmp_path):
+    calls, _ = setup_fake(helper, monkeypatch, tmp_path)
+    original = helper.k
+    def terminate_during_dump(*args, **kwargs):
+        if 'pg_dump' in args:
+            handler = signal.getsignal(signal.SIGTERM)
+            handler(signal.SIGTERM, None)
+        return original(*args, **kwargs)
+    monkeypatch.setattr(helper, 'k', terminate_during_dump)
+    with pytest.raises(helper.Refused, match='snapshot_interrupted'):
+        helper.main()
+    assert any('--replicas=1' in call for call in calls)
+    assert any(call[0] == 'rollout' for call in calls)

workplans/CSOC-WP-0008-production-transfer-acceptance.md

@@ -98,6 +98,16 @@ counts, markdown/asset hashes, rendering and denied cross-member reads. Record
 RPO/RTO, retention, custody and a rollback drill. No acceptance from DB-only
 snapshots; no private archive in git or State Hub.
 
+2026-10-07 operations increment: scripts/paired-backup.py and four deploy/backup
+units package both stores into one age-encrypted pair using the platform WebDAV
+lane, bounded daily/weekly slots, freshness/hash checks, capture locks and a
+reviewed failed-pair retry. Snapshot termination attempts app resume. All 140
+tests pass through make check, including 18 focused recovery cases. Synthetic
+real age roundtrip, unit verification and node server dry-run pass. See
+docs/release/paired-backup-operations-2026-10-07.md and proposed RPF-WP-0053.
+No timer enabled, credential copied or production offsite write performed.
+T02 remains progress for admission, retrieval/recovery and operating acceptance.
+
 ## T03 — Rehearse and reconcile the actual transfer cohort
 
 ```task

workplans/CSOC-WP-0010-platform-email-facilities.md

@@ -68,6 +68,18 @@ state_hub_task_id: "ca41be06-3611-5647-b469-4f89f56f77f9"
 
 Extend existing email-connect with explicit caller identities, approved templates, per-caller limits, idempotent retries, suppression/bounce handling and redacted delivery states. Replace forgejo sender with platform identity only through reviewed custody and rollout. Reuse native SMTP for authentication messages, HTTP notification API for supported application templates; do not widen current user-engine-only ingress globally.
 
+2026-10-07 retry correction implemented in email-connect under EMAIL-WP-0006.
+Durable tenant/client/registration reservations prevent duplicate challenge mail
+across retries, changed tracing IDs, restart and concurrent connections. Unknown
+outcomes block automatic resend; definitive retryable refusal cancels the failed
+challenge before retry. Existing challenge links remain usable, with legacy
+retries conservative. All 39 transactional tests and all 101 repository tests
+passed with the synthetic GreenMail SMTP/IMAP harness; the real verification
+journey confirms one message after restart/retry. Static ESO bootstrap retired
+in source per platform handoff, without a live custody change. Candidate release
+is reviewed in email-connect/docs/verification-retry-review-2026-10-07.md.
+T04 stays progress for live activation and caller/sender/limits admission.
+
 ## Establish reply and mailbox operations
 
 ```task