scripts/d2_space_proof.py

#!/usr/bin/env python3
"""D2 — In-Space synthetic identity/lifecycle + recovery proof (opt-in, fail-safe, one-shot).

Runs INSIDE the deployed HF Space (via the entrypoint) on the ACTUAL Space runtime, BEFORE
uvicorn starts. It exercises the merged D2 identity code end-to-end on the real filesystem and
SQLite runtime, using a real in-process ES256 WebAuthn channel (the "equivalent cryptographically
verified WebAuthn test channel" the D2 runbook permits), and uploads REDACTED evidence to the
private D2 proof bucket so it can be verified out-of-band.

Strictly synthetic — NO participant data, NO real names/emails/phones/biometrics. It never
enables WAL, never modifies Payment Core / PDP / ASAP, and is NEVER fatal (a failure is caught
and the app still starts).

Stages (AMANPAY_D2_SPACE_PROOF_STAGE):
  * ``full``      — verify runtime facts (§5), bootstrap tenant+operator+customer, run the
                    lifecycle (§6-§9,§12), create an encrypted synthetic snapshot, emit evidence.
  * ``verify``    — after a factory rebuild: prove restore-on-start recovered the synthetic DB
                    (§10), emit post-recovery evidence, mark the generation restore_tested.
  * ``negatives`` — isolated negative-recovery checks (§11) on throwaway copies.

Evidence NEVER contains invitation codes, session/CSRF tokens, credential public-key bytes,
challenge values, private keys, or PII — only states, counts, hashes and generation ids.
"""
from __future__ import annotations

import contextlib
import hashlib
import json
import os
import secrets
import sqlite3
import tempfile
import time

# --- deployed D2 identity code (all under amanpay/, shipped in the image) ---
from amanpay.identity.config import D2Config, store_key
from amanpay.identity.operator import OperatorService
from amanpay.identity.service import IdentityService
from amanpay.identity.storage import D2Storage
from amanpay.simulation_storage import util

PROOF_VERSION = "1"
TENANT_SLUG = "d2-space-proof"
BUCKET = os.getenv("AMANPAY_D2_BUCKET") or os.getenv("AMANPAY_BUCKET") or "MHamdan/amanpay-d2-proof"
RP_ID = os.getenv("AMANPAY_D2_RP_ID", "mhamdan-amanpay.hf.space")
ORIGIN = os.getenv("AMANPAY_D2_ORIGIN", "https://mhamdan-amanpay.hf.space")


# ======================================================================= authenticator
class _SoftAuthenticator:
    """Minimal in-process ES256 (ECDSA P-256) WebAuthn authenticator — real signatures.

    Verification is performed by the deployed py_webauthn via the D2 WebAuthn service; nothing
    is weakened. ``backup`` toggles BE/BS so synced vs single-device counter handling is exercised.
    """

    def __init__(self, backup: bool = True):
        import cbor2  # noqa: F401 (availability check)
        self.backup = backup
        self._creds: dict = {}

    # -- helpers --
    @staticmethod
    def _b64u(data: bytes) -> str:
        import base64
        return base64.urlsafe_b64encode(data).decode().rstrip("=")

    def _flags(self, *, at: bool) -> int:
        f = 0x01 | 0x04                       # UP | UV
        if self.backup:
            f |= 0x08 | 0x10                  # BE | BS
        if at:
            f |= 0x40                         # AT
        return f

    def _auth_data(self, rp_id, *, sign_count, cred_id, priv, include_attested):
        import cbor2
        out = hashlib.sha256(rp_id.encode()).digest() + bytes([self._flags(at=include_attested)])
        out += sign_count.to_bytes(4, "big")
        if include_attested:
            n = priv.public_key().public_numbers()
            cose = cbor2.dumps({1: 2, 3: -7, -1: 1, -2: n.x.to_bytes(32, "big"),
                                -3: n.y.to_bytes(32, "big")})
            out += b"\x00" * 16 + len(cred_id).to_bytes(2, "big") + cred_id + cose
        return out

    def create(self, options_json: str, origin: str | None = None) -> dict:
        import cbor2
        from cryptography.hazmat.primitives.asymmetric import ec
        opts = json.loads(options_json)
        opts = opts.get("publicKey", opts)
        cred_id = secrets.token_bytes(20)
        priv = ec.generate_private_key(ec.SECP256R1())
        self._creds[self._b64u(cred_id)] = {"priv": priv, "sc": 0, "uh": opts["user"]["id"]}
        cdj = json.dumps({"type": "webauthn.create", "challenge": opts["challenge"],
                          "origin": origin or ORIGIN, "crossOrigin": False},
                         separators=(",", ":")).encode()
        ad = self._auth_data(opts["rp"]["id"], sign_count=0, cred_id=cred_id, priv=priv,
                             include_attested=True)
        att = cbor2.dumps({"fmt": "none", "attStmt": {}, "authData": ad})
        return {"id": self._b64u(cred_id), "rawId": self._b64u(cred_id), "type": "public-key",
                "response": {"clientDataJSON": self._b64u(cdj), "attestationObject": self._b64u(att),
                             "transports": ["internal", "hybrid"]}, "clientExtensionResults": {}}

    def get(self, options_json: str, credential_id: str | None = None,
            origin: str | None = None) -> dict:
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.asymmetric import ec
        opts = json.loads(options_json)
        opts = opts.get("publicKey", opts)
        cid = credential_id or next(iter(self._creds))
        st = self._creds[cid]
        if not self.backup:
            st["sc"] += 1
        cdj = json.dumps({"type": "webauthn.get", "challenge": opts["challenge"],
                          "origin": origin or ORIGIN, "crossOrigin": False},
                         separators=(",", ":")).encode()
        ad = self._auth_data(opts["rpId"], sign_count=st["sc"], cred_id=None, priv=None,
                             include_attested=False)
        sig = st["priv"].sign(ad + hashlib.sha256(cdj).digest(), ec.ECDSA(hashes.SHA256()))
        return {"id": cid, "rawId": cid, "type": "public-key",
                "response": {"clientDataJSON": self._b64u(cdj), "authenticatorData": self._b64u(ad),
                             "signature": self._b64u(sig), "userHandle": st["uh"]},
                "clientExtensionResults": {}}


# ============================================================================ infra
def _deployed_commit() -> str:
    for p in ("/app/build_info.json", "build_info.json"):
        with contextlib.suppress(Exception):
            return str(json.load(open(p)).get("commit") or "unknown")
    return os.getenv("AMANPAY_COMMIT", "unknown")


def _bucket():
    from amanpay.simulation_storage.hf_bucket_store import HFBucketObjectStore
    return HFBucketObjectStore(BUCKET, os.environ["HF_TOKEN"])


def _put_evidence(stage: str, result: dict) -> str | None:
    """Upload a REDACTED stage-result artifact to the private bucket. Returns its key or None."""
    try:
        result = _redact(result)
        result.update(proof_version=PROOF_VERSION, deployed_commit=_deployed_commit(),
                      stage=stage, created_at_iso=util.utc_stamp())
        key = f"d2-space-proof-evidence/{_deployed_commit()}/{stage}/{util.utc_stamp()}-result.json"
        _bucket().put(key, json.dumps(result, sort_keys=True, default=str).encode())
        return key
    except Exception as exc:                                # noqa: BLE001 — best effort
        print(f"[d2-space-proof] evidence upload failed (non-fatal): {type(exc).__name__}")
        return None


_FORBIDDEN = ("code", "token", "csrf", "secret", "public_key", "credential_id", "challenge",
              "pepper", "store_key", "password", "email", "phone")


def _redact(obj):
    """Defense-in-depth: drop any key that could carry a secret / credential / PII."""
    if isinstance(obj, dict):
        return {k: _redact(v) for k, v in obj.items()
                if not any(f in k.lower() for f in _FORBIDDEN)}
    if isinstance(obj, list):
        return [_redact(v) for v in obj]
    return obj


def _runtime_facts(storage: D2Storage) -> dict:
    rr = storage.db.runtime_report()
    db_path = storage.config.db_path
    return {
        "execution_env": "hf_space" if os.getenv("SPACE_ID") or os.path.isdir("/app") else "other",
        "space_id": os.getenv("SPACE_ID", ""),
        "db_path_under_tmp_amanpay_d2": db_path.startswith("/tmp/amanpay-d2"),
        "fs_category": rr.get("fs_category"), "fstype": rr.get("fstype"),
        "sqlite_version": rr.get("sqlite_version"), "sqlite_source_id": rr.get("sqlite_source_id"),
        "journal_mode": rr.get("journal_mode"), "synchronous": rr.get("synchronous"),
        "wal_status": rr.get("wal_status"),
        # foreign_keys is enforced ON by the store's connections (D1 pragma discipline); the
        # per-connection default OFF is why the audit connection sets it explicitly above.
        "foreign_keys_enforced": bool(storage.db.config.foreign_keys),
        "busy_timeout_ms": storage.config.busy_timeout_ms,
        "schema_version": storage.db.schema_version(),
        "no_wal_file": not os.path.exists(storage.config.wal_path),
        "no_shm_file": not os.path.exists(storage.config.shm_path),
    }


def _integrity(db_path: str) -> dict:
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("PRAGMA foreign_keys=ON")            # per-connection; make the check meaningful
        ic = conn.execute("PRAGMA integrity_check").fetchone()[0]
        fk = conn.execute("PRAGMA foreign_key_check").fetchall()
        return {"integrity_check": str(ic).lower(), "foreign_key_violations": len(fk)}
    finally:
        conn.close()


def _counts(storage: D2Storage) -> dict:
    from amanpay.identity.storage.schema import CORE_TABLES
    out = {}
    for t in CORE_TABLES:
        with contextlib.suppress(Exception):
            out[t] = storage.db.read(f"SELECT COUNT(*) FROM {t}")[0][0]
    out["last_event_seq"] = storage.db.read(
        "SELECT COALESCE(MAX(seq),0) FROM lifecycle_events")[0][0]
    return out


# ======================================================================= stage: full
def _stage_full() -> dict:
    cfg = D2Config()
    if not util.is_confirmed_local(cfg.db_path) and not cfg.dev_allow_unknown_fs:
        # /tmp on the Space is overlay (local); this only guards a misconfiguration.
        pass
    key = store_key()
    if not key:
        raise RuntimeError("AMANPAY_D2_STORE_KEY missing — refusing unencrypted synthetic proof")

    storage = D2Storage(config=cfg, object_store=_bucket(), source_commit=_deployed_commit(),
                        store_key=key)
    # Idempotent synthetic run: always start from a CLEAN local DB (this is throwaway proof data).
    # Opening with recover=True would restore prior synthetic accounts from the bucket and then a
    # fresh bootstrap would collide (handle_taken), so wipe the local DB and open without recovery.
    for p in (cfg.db_path, cfg.wal_path, cfg.shm_path):
        with contextlib.suppress(OSError):
            os.remove(p)
    ev: dict = {"outcome": "", "init_recovery": storage.open(recover=False)}
    try:
        facts = _runtime_facts(storage)
        facts.update(_integrity(cfg.db_path))
        ev["runtime_facts"] = facts
        if facts["journal_mode"] != "delete":
            raise RuntimeError(f"journal_mode={facts['journal_mode']} (must be delete)")

        ident = IdentityService.from_storage(storage)
        op = OperatorService(ident)
        repo = ident.repo
        now = time.time()

        tenant = repo.get_tenant_by_slug(TENANT_SLUG) or repo.create_tenant(
            slug=TENANT_SLUG, name_en="AmanPay Synthetic Event",
            name_ar="فعالية أمان باي التجريبية", now=now)

        # --- §6 operator bootstrap (in-process via the same service the CLI uses) ---
        op_auth = _SoftAuthenticator(backup=True)
        _inv, op_code = ident.invites.create(tenant_id=tenant.id, role="operator", now=now)
        oopts = ident.begin_enrollment(code=op_code, public_handle="d2proofoperator",
                                       display_alias="Synthetic Operator",
                                       preferred_language="en", consent_accepted=True, now=now)
        ores = ident.complete_enrollment(credential=op_auth.create(oopts), now=now + 1)
        operator, osess = ores.user, ores.issued.session
        ev["operator"] = {"status": operator.status, "role": operator.role,
                          "passkeys": repo.count_active_credentials(operator.id)}

        # --- §7 customer enrollment (operator issues the invitation) ---
        pairs = op.create_invitations(operator, role="customer", count=1, now=now + 2)
        cust_auth = _SoftAuthenticator(backup=True)
        # validate the invitation is atomic/one-time by re-reading state
        copts = ident.begin_enrollment(code=pairs[0][1], public_handle="d2proofuser",
                                       display_alias="Synthetic Participant",
                                       preferred_language="ar", consent_accepted=True, now=now + 3)
        cres = ident.complete_enrollment(credential=cust_auth.create(copts), now=now + 4)
        customer = cres.user
        ev["customer_enrolled"] = {"status": customer.status,
                                   "passkeys": repo.count_active_credentials(customer.id),
                                   "no_active_without_passkey":
                                       repo.count_active_credentials(customer.id) >= 1}
        # invitation cannot be reused
        try:
            ident.begin_enrollment(code=pairs[0][1], public_handle="reuse",
                                   display_alias="x", preferred_language="en",
                                   consent_accepted=True, now=now + 5)
            ev["invitation_reuse_blocked"] = False
        except Exception:                                  # noqa: BLE001
            ev["invitation_reuse_blocked"] = True

        # --- §8 lifecycle ---
        ev["lifecycle"] = _run_lifecycle(ident, op, operator, osess, cust_auth, customer, now)

        # --- §9 encrypted snapshot (synthetic-only, protected) ---
        gen = storage.snapshots.create_snapshot()
        storage.snapshots.update_manifest_flags(
            gen.generation_id, synthetic_only=True, deletion_protected=True,
            retained_for="d2_synthetic_acceptance")
        # verify by re-download + hash
        with tempfile.TemporaryDirectory() as td:
            dest = os.path.join(td, "s.db")
            storage.snapshots.download_snapshot(
                type(gen)(gen.generation_id, gen.prefix,
                          storage.snapshots.manifests.read_manifest(gen.prefix)), dest)
            redl_sha = util.sha256_file(dest)
        ev["snapshot"] = {"generation_id": gen.generation_id,
                          "encryption": gen.manifest.get("encryption"),
                          "key_version": gen.manifest.get("key_version"),
                          "schema_version": gen.manifest.get("schema_version"),
                          "sqlite_version": gen.manifest.get("sqlite_version"),
                          "upload_complete": gen.manifest.get("upload_complete"),
                          "synthetic_only": True, "deletion_protected": True,
                          "snapshot_sha_matches_after_redownload":
                              redl_sha == gen.manifest.get("snapshot_sha256"),
                          "row_counts": gen.manifest.get("row_counts")}
        ev["counts_after_full"] = _counts(storage)
        ev["outcome"] = "success"
        return ev
    finally:
        storage.close()


def _run_lifecycle(ident, op, operator, osess, cust_auth, customer, now) -> dict:
    r: dict = {}
    # login (discoverable) after logout
    lg = _login(ident, cust_auth, now + 10)
    r["login_discoverable"] = lg.user.id == customer.id
    # wrong-origin rejection: a genuine assertion whose clientData origin is NOT the RP origin.
    o = ident.begin_login(now=now + 11)
    wrong = cust_auth.get(o, origin="https://evil.example.com")
    r["wrong_origin_rejected"] = _expect_fail(
        lambda: ident.complete_login(credential=wrong, now=now + 12))
    # expired-ceremony rejection (ceremony/challenge TTL is 300s). NB: the result key avoids the
    # substring "challenge" so the evidence redactor does not elide it.
    o2 = ident.begin_login(now=now + 13)
    exp = cust_auth.get(o2)
    r["stale_ceremony_rejected"] = _expect_fail(
        lambda: ident.complete_login(credential=exp, now=now + 13 + 100_000))
    # replay rejection (same assertion twice)
    o3 = ident.begin_login(now=now + 14)
    a = cust_auth.get(o3)
    ident.complete_login(credential=a, now=now + 15)
    r["replay_rejected"] = _expect_fail(
        lambda: ident.complete_login(credential=a, now=now + 16))
    # multi-passkey: add a second (needs recent auth → fresh login session)
    lg2 = _login(ident, cust_auth, now + 20)
    sess2 = lg2.issued.session
    auth2 = _SoftAuthenticator(backup=False)
    addopts = ident.begin_add_passkey(lg2.user, sess2, now=now + 21)
    added = ident.complete_add_passkey(lg2.user, sess2, credential=auth2.create(addopts),
                                       nickname="Synthetic laptop", now=now + 22)
    r["passkeys_after_add"] = ident.repo.count_active_credentials(customer.id)
    # revoke the added (second) credential, authenticate with the remaining
    ident.revoke_passkey(lg2.user, sess2, added.id, now=now + 23)
    r["passkeys_after_revoke"] = ident.repo.count_active_credentials(customer.id)
    # now exactly one credential remains → the final one is protected from revocation
    last = ident.list_passkeys(lg2.user)[0]
    r["final_passkey_protected"] = _expect_fail(
        lambda: ident.revoke_passkey(lg2.user, sess2, last.id, now=now + 24))
    r["login_after_revoke"] = _login(ident, cust_auth, now + 25).user.id == customer.id
    # sessions + CSRF
    # CSRF (anti-forgery) rejection. NB: result keys avoid the substring "csrf" so the evidence
    # redactor does not elide them.
    s = ident.repo.get_session_by_token_hash(_hash(lg2.issued.token))
    r["antiforgery_missing_rejected"] = not ident.sessions.validate_csrf(s, None)
    r["antiforgery_mismatch_rejected"] = not ident.sessions.validate_csrf(s, "wrong")
    r["active_sessions_before_revoke_others"] = len(ident.repo.list_sessions(customer.id))
    ident.revoke_other_sessions(lg2.user, sess2, now=now + 26)
    r["active_sessions_after_revoke_others"] = len(ident.repo.list_sessions(customer.id))
    # pause (self) → sessions revoked, login rejected
    lg3 = _login(ident, cust_auth, now + 30)
    ident.pause_account(lg3.user, lg3.issued.session, now=now + 31)
    r["paused_state"] = ident.repo.get_user(customer.id).status
    r["paused_sessions_revoked"] = len(ident.repo.list_sessions(customer.id, active_only=True)) == 0
    r["paused_login_rejected"] = _expect_fail(lambda: _login(ident, cust_auth, now + 32))
    # operator reactivation (recent operator step-up from enrollment)
    op.reactivate_account(operator, osess, customer.id, now=now + 33)
    r["reactivated_state"] = ident.repo.get_user(customer.id).status
    r["reactivate_requires_signin"] = len(
        ident.repo.list_sessions(customer.id, active_only=True)) == 0
    # recovery (operator-issued) → new passkey; operator never mints a session
    _iid, rec_code = op.issue_recovery_invitation(operator, osess, customer.id, now=now + 34)
    rec_auth = _SoftAuthenticator(backup=True)
    ropts = ident.begin_recovery(code=rec_code, now=now + 35)
    rres = ident.complete_recovery(credential=rec_auth.create(ropts), now=now + 36)
    r["recovery_state"] = rres.user.status
    r["recovery_old_credential_revoked"] = _expect_fail(lambda: _login(ident, cust_auth, now + 37))
    r["recovery_new_credential_ok"] = _login(ident, rec_auth, now + 38).user.id == customer.id
    r["recovery_replay_blocked"] = _expect_fail(
        lambda: ident.begin_recovery(code=rec_code, now=now + 39))
    # §12 deletion (requires fresh passkey assertion)
    lgd = _login(ident, rec_auth, now + 40)
    dopts = ident.begin_delete(lgd.user, lgd.issued.session, now=now + 41)
    receipt = ident.complete_delete(lgd.user, lgd.issued.session,
                                    credential=rec_auth.get(dopts), now=now + 42)
    u = ident.repo.get_user(customer.id)
    r["deletion"] = {"deleted": receipt.get("deleted"), "status": u.status,
                     "profile_cleared": u.display_alias == "",
                     "credentials_revoked": ident.repo.count_active_credentials(customer.id) == 0,
                     "tombstone": bool(ident.repo.get_tombstone(customer.id)),
                     "generation_monotonic": u.profile_generation > 1,
                     "retention_disclosure_present": bool(receipt.get("note"))}
    r["deleted_login_rejected"] = _expect_fail(lambda: _login(ident, rec_auth, now + 43))
    return r


# --------------------------------------------------------------------- lifecycle helpers
def _login(ident, authr, now):
    o = ident.begin_login(now=now)
    return ident.complete_login(credential=authr.get(o), now=now + 0.5)


def _expect_fail(fn) -> bool:
    try:
        fn()
        return False
    except Exception:                                      # noqa: BLE001
        return True


def _hash(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


# ===================================================================== stage: verify
def _stage_verify() -> dict:
    """After a factory rebuild: prove restore-on-start recovered the synthetic DB."""
    cfg = D2Config()  # init_mode is False unless AMANPAY_D2_INITIALIZE=1 (should be off now)
    storage = D2Storage(config=cfg, object_store=_bucket(), source_commit=_deployed_commit(),
                        store_key=store_key())
    local_existed = os.path.exists(cfg.db_path) and os.path.getsize(cfg.db_path) > 0
    ev = {"local_db_existed_before_open": local_existed, "recovery_event": storage.open()}
    try:
        facts = _runtime_facts(storage)
        facts.update(_integrity(cfg.db_path))
        ev["runtime_facts"] = facts
        ev["counts_after_recovery"] = _counts(storage)
        rows = storage.db.read(
            "SELECT outcome, source, generation_id FROM recovery_events ORDER BY id DESC LIMIT 3")
        ev["recovery_events"] = [{"outcome": r[0], "source": r[1], "generation_id": r[2]}
                                 for r in rows]
        # tenant/operator/customer(+tombstone) survived
        t = storage.db.read("SELECT COUNT(*) FROM tenants WHERE slug=?", (TENANT_SLUG,))[0][0]
        ev["synthetic_tenant_restored"] = t == 1
        ev["tombstone_restored"] = storage.db.read(
            "SELECT COUNT(*) FROM deletion_tombstones")[0][0] >= 1
        ev["no_wal_no_shm"] = facts["no_wal_file"] and facts["no_shm_file"]
        # mark newest generation restore_tested (recovery succeeded from a bucket generation)
        gen_id = next((r[2] for r in rows if r[0] == "restored" and r[2]), "")
        if gen_id:
            with contextlib.suppress(Exception):
                storage.snapshots.update_manifest_flags(gen_id, restore_tested=True)
                ev["restore_tested_marked"] = gen_id
        ev["outcome"] = "success" if ev["recovery_event"].get("outcome") in (
            "restored", "opened_local") else "unexpected"
        return ev
    finally:
        storage.close()


# ================================================================== stage: negatives
def _stage_negatives() -> dict:
    """Isolated negative-recovery checks on throwaway copies (never touch the retained gen)."""
    from amanpay.identity.storage.integrity import D2IntegrityVerifier
    from amanpay.identity.storage.recovery import RecoveryFailed
    v = D2IntegrityVerifier()
    ev: dict = {}
    with tempfile.TemporaryDirectory() as td:
        # unsupported future schema
        p = os.path.join(td, "future.db")
        c = sqlite3.connect(p)
        c.execute("CREATE TABLE storage_metadata(key TEXT PRIMARY KEY, value TEXT)")
        c.execute("INSERT INTO storage_metadata VALUES('schema_version','999')")
        c.execute("INSERT INTO storage_metadata VALUES('kind','amanpay-d2-accounts')")
        c.commit(); c.close()
        ev["unsupported_future_schema_rejected"] = not v.verify_database(p).ok
        # corrupt file
        pc = os.path.join(td, "corrupt.db")
        open(pc, "wb").write(b"not a sqlite database")
        ev["corrupt_snapshot_rejected"] = not v.verify_database(pc).ok
    # wrong-key / missing-manifest / fail-safe are covered by the merged unit tests
    # (tests/test_d2_persistence.py); record that linkage honestly.
    ev["wrong_key_and_missing_manifest"] = "covered_by_tests/test_d2_persistence.py"
    ev["fail_safe_no_empty_db"] = "recovery raises RecoveryFailed without init_mode"
    ev["outcome"] = "success"
    return ev


# ============================================================================ main
def main() -> int:
    stage = os.getenv("AMANPAY_D2_SPACE_PROOF_STAGE", "full").strip().lower()
    started = time.time()
    print(f"[d2-space-proof] stage={stage} commit={_deployed_commit()}")
    try:
        if stage == "full":
            result = _stage_full()
        elif stage == "verify":
            result = _stage_verify()
        elif stage == "negatives":
            result = _stage_negatives()
        else:
            result = {"outcome": "invalid_stage", "stage": stage}
        result["duration_ms"] = int((time.time() - started) * 1000)
        key = _put_evidence(stage, result)
        print(f"[d2-space-proof] stage={stage} outcome={result.get('outcome')} evidence={key}")
    except Exception as exc:                                # noqa: BLE001 — never fatal
        detail = {"outcome": "error", "error_class": type(exc).__name__,
                  "error": str(exc)[:200], "duration_ms": int((time.time() - started) * 1000)}
        with contextlib.suppress(Exception):
            _put_evidence(stage, detail)
        print(f"[d2-space-proof] stage={stage} ERROR {type(exc).__name__}: {str(exc)[:160]}")
    return 0                                                # ALWAYS 0 — the app must start


if __name__ == "__main__":
    raise SystemExit(main())