scripts/d1_space_proof.py

#!/usr/bin/env python3
"""D1.2 — In-Space synthetic ROLLBACK-JOURNAL recovery proof (opt-in, fail-safe, one-shot).

Runs INSIDE the deployed HF Space container (via the entrypoint) on the ACTUAL Space runtime
(Python, sqlite3=SQLite 3.46.1, `/tmp/amanpay-d1/` overlay, Space bucket credentials). Isolated,
synthetic-only — NOT customer persistence, no WAL, no public endpoint, no participant data.

Safety guarantees (D1.2 + D1.3 corrections):
  * A surviving VALID local proof DB is QUARANTINED (atomic rename, 0600, same FS), never deleted,
    and is restored to the active path if the bucket restore fails (fail-back).
  * DURABLE bucket markers (commit, proof-version, tenant, stage) are the AUTHORITATIVE cross-restart
    completion source, validated against their referenced evidence's SHA-256. The ``/tmp`` marker is
    ONLY a same-container write cache and never independently returns ``already_completed``.
  * RECONCILE-BEFORE-EXECUTE: a missing marker may cover a crash after a stage's durable side effects
    but before the marker upload; before redoing work the script reconciles an existing complete,
    hash-verified success by THIS deployed commit and writes only the missing marker.
  * Markers bind to the EXACT deployed commit; a different commit never inherits completion.
  * Every attempt writes a redacted stage-result artifact with a normalized ``failure_class``.
  * ANY failure is caught; the ordinary app always starts; storage/WAL are never enabled.
"""
from __future__ import annotations
import contextlib, json, os, sqlite3, tempfile, time

from amanpay.simulation_storage import util
from amanpay.simulation_storage.config import StorageConfig
from amanpay.simulation_storage.service import SimulationStorage
from amanpay.simulation_storage.sqlite_provider import select_provider
from amanpay.simulation_storage.proof_ledger import Line, UnbalancedJournal, DuplicateIdempotencyKey

PROOF_VERSION = "1"
TENANT = "d1-space-local-proof"
DBID = "d1-space-local-proof"
BUCKET = os.getenv("AMANPAY_BUCKET", "MHamdan/amanpay-d1-proof")
NEG_PREFIX = "snapshots/d1-space-local-proof-negtest"   # isolated throwaway prefix for negatives

FAILURE_CLASSES = {
    "UNSAFE_FILESYSTEM", "INVALID_STAGE", "STORAGE_MODE_CONFLICT", "PROOF_KEY_MISSING",
    "SQLITE_CONFIGURATION_FAILED", "JOURNAL_IMBALANCE", "SNAPSHOT_UPLOAD_FAILED",
    "MANIFEST_HASH_MISMATCH", "SNAPSHOT_HASH_MISMATCH", "DECRYPTION_FAILED", "SCHEMA_UNSUPPORTED",
    "INTEGRITY_CHECK_FAILED", "FOREIGN_KEY_CHECK_FAILED", "ATOMIC_INSTALL_FAILED",
    "EVIDENCE_UPLOAD_FAILED", "UNKNOWN_PROOF_FAILURE",
}


class StageError(RuntimeError):
    def __init__(self, cls: str, detail: str = ""):
        super().__init__(cls)
        self.cls = cls if cls in FAILURE_CLASSES else "UNKNOWN_PROOF_FAILURE"


def _db_path() -> str:
    return os.path.join(os.getenv("AMANPAY_DB_DIR", "/tmp/amanpay-d1"), "d1-space-local-proof.db")


def _deployed_commit() -> str:
    for p in ("/app/build_info.json", "build_info.json"):
        try:
            return str(json.load(open(p)).get("commit") or "unknown")
        except Exception:  # noqa: BLE001
            pass
    return os.getenv("AMANPAY_COMMIT", "unknown")


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


def _cfg(init: bool) -> StorageConfig:
    return StorageConfig(db_path=_db_path(), database_id=DBID, tenant=TENANT, init_mode=init)


def _svc(init: bool, key: str) -> SimulationStorage:
    return SimulationStorage(config=_cfg(init), object_store=_store(),
                             source_commit=_deployed_commit(), store_key=key)


def _runtime_facts() -> dict:
    p = select_provider()
    cat, fstype = util.classify_filesystem(os.path.dirname(_db_path()))
    return {"execution_environment": "hf_space", "proof_version": PROOF_VERSION, "tenant": TENANT,
            "deployed_commit": _deployed_commit(), "sqlite_version": p.version,
            "sqlite_source_id": p.source_id, "runtime_status": p.status,
            "filesystem": fstype, "filesystem_classification": cat}


# ---- markers: durable private-bucket marker is AUTHORITATIVE across restarts; the /tmp marker
# ---- is only a fast SAME-CONTAINER optimization (never the cross-restart source of truth). ----
def _local_marker(stage: str) -> str:
    return f"/tmp/.amanpay-d1sp.{_deployed_commit()}.{PROOF_VERSION}.{TENANT}.{stage}.done"


def _write_local_marker(stage: str) -> None:
    with contextlib.suppress(OSError):
        with open(_local_marker(stage), "w") as fh:
            fh.write(util.utc_stamp())


def _durable_marker_key(stage: str) -> str:
    return f"proof-stage-markers/{_deployed_commit()}/{PROOF_VERSION}/{TENANT}/{stage}.json"


def _put_result(stage: str, result: dict) -> tuple[str | None, str | None]:
    """Write the redacted stage-result artifact. Returns (key, sha256) or (None, None) on failure."""
    try:
        body = json.dumps(result, sort_keys=True).encode()
        key = f"space-local-proof-evidence/{_deployed_commit()}/{stage}/{util.utc_stamp()}-result.json"
        _store().put(key, body)
        result["evidence_key"] = key
        return key, util.sha256_bytes(body)
    except Exception:  # noqa: BLE001 — evidence upload is best-effort; caller handles (key None)
        return None, None


def _write_durable_marker(stage: str, marker: dict) -> bool:
    """Write the durable bucket marker (only after a fully successful stage). False on failure."""
    try:
        _store().put(_durable_marker_key(stage), json.dumps(marker, sort_keys=True).encode())
        return True
    except Exception:  # noqa: BLE001
        return False


def _read_valid_durable_marker(stage: str) -> dict | None:
    """Return the durable marker ONLY if it is schema-valid, matches (commit, version, tenant,
    stage), status=success, and its referenced evidence object exists with a matching hash.
    A missing / malformed / hash-invalid marker is NOT trusted (returns None)."""
    try:
        m = json.loads(_store().get(_durable_marker_key(stage)).decode())
    except Exception:  # noqa: BLE001 — missing marker
        return None
    try:
        if not (m.get("deployed_commit") == _deployed_commit() and m.get("proof_version") == PROOF_VERSION
                and m.get("tenant") == TENANT and m.get("stage") == stage and m.get("status") == "success"):
            return None
        ev_key, ev_hash = m.get("evidence_key"), m.get("evidence_hash")
        if not ev_key or not ev_hash:
            return None
        ev_body = _store().get(ev_key)
        if util.sha256_bytes(ev_body) != ev_hash:               # evidence missing or hash mismatch
            return None
        ev = json.loads(ev_body.decode())                       # bind marker to same-commit evidence:
        if not (ev.get("deployed_commit") == _deployed_commit() and ev.get("tenant") == TENANT
                and ev.get("proof_version") == PROOF_VERSION and ev.get("stage") == stage
                and ev.get("status") == "success"):
            return None                                         # cross-commit / cross-stage reuse rejected
        return m
    except Exception:  # noqa: BLE001 — any read/validation error => do not trust
        return None


def _build_marker(stage: str, detail: dict, ev_key: str, ev_hash: str, origin: str = "proof_run") -> dict:
    return {"deployed_commit": _deployed_commit(), "proof_version": PROOF_VERSION, "tenant": TENANT,
            "stage": stage, "status": "success", "generation_id": detail.get("generation_id")
            or detail.get("restored_generation"), "evidence_key": ev_key, "evidence_hash": ev_hash,
            "completed_at": util.utc_stamp(), "execution_environment": "hf_space",
            "synthetic_only": True, "participant_data_used": False, "WAL_enabled": False,
            "marker_origin": origin}


def _find_evidence(commit: str, stage_dir: str, pred) -> tuple[str | None, str | None, dict | None]:
    """Return (key, sha256, parsed) of the first stage-result evidence object under
    ``space-local-proof-evidence/<commit>/<stage_dir>/`` matching ``pred``. (None, None, None) if none."""
    st = _store()
    try:
        keys = sorted(st.list(f"space-local-proof-evidence/{commit}/{stage_dir}/"))
    except Exception:  # noqa: BLE001
        return None, None, None
    for k in keys:
        if not k.endswith("-result.json"):
            continue
        try:
            body = st.get(k)
            ev = json.loads(body.decode())
        except Exception:  # noqa: BLE001
            continue
        if pred(ev):
            return k, util.sha256_bytes(body), ev
    return None, None, None


# ---- reconcile-before-execute: a MISSING durable marker does not mean the stage never ran. Before
# ---- executing, inspect existing durable side effects (generation + evidence). If a complete,
# ---- hash-verified success created by THIS deployed commit already exists, write only the missing
# ---- marker (marker_origin=reconciled_existing_success) instead of recreating journals/generation.
def _reconcile_create() -> dict | None:
    """No valid create marker: is there already a complete eligible generation + matching hf_space
    create evidence for THIS (commit, proof_version, tenant/db)? Return a reconcile detail if so."""
    commit = _deployed_commit()
    st = _store()
    base = f"snapshots/{TENANT}/{TENANT}/"
    try:
        rels = [k[len(base):] for k in st.list(base)]
    except Exception:  # noqa: BLE001
        return None
    gids = sorted({r.split("/", 1)[0] for r in rels if "/" in r and "LATEST" not in r})
    for gid in reversed(gids):                          # newest first
        gbase = f"{base}{gid}"
        try:
            man_bytes = st.get(f"{gbase}/manifest.json")
            man = json.loads(man_bytes.decode())
            sha_ok = util.sha256_bytes(man_bytes) == st.get(f"{gbase}/manifest.sha256").decode()
            snap_present = any(k.endswith(man["snapshot_filename"]) for k in st.list(gbase + "/"))
        except Exception:  # noqa: BLE001 — incomplete/inconsistent generation => do not trust
            continue
        if not (man.get("upload_complete") is True and sha_ok and snap_present
                and man.get("tenant") == TENANT and man.get("database_id") == DBID
                and man.get("synthetic_only") is True and man.get("deletion_protected") is True):
            continue                                    # incomplete/inconsistent => never overwrite, skip
        ck, ch, cev = _find_evidence(commit, "create", lambda e: e.get("status") == "success"
                and e.get("deployed_commit") == commit and e.get("proof_version") == PROOF_VERSION
                and e.get("tenant") == TENANT and e.get("execution_environment") == "hf_space"
                and e.get("space_snapshot_executed") is True and e.get("generation_id") == gid)
        if cev is None:
            continue
        return {"stage": "create", "generation_id": gid, "evidence_key": ck, "evidence_hash": ch}
    return None


def _reconcile_restore(key: str) -> dict | None:
    """No valid restore marker: does the active local DB already represent the expected restored
    state of a restore-tested generation with matching hf_space restore evidence? Return detail if so."""
    commit = _deployed_commit()
    path = _db_path()
    if not os.path.exists(path) or os.path.getsize(path) == 0 or os.path.islink(path):
        return None
    st = _store()
    rk, rh, rev = _find_evidence(commit, "restore", lambda e: e.get("status") == "success"
            and e.get("deployed_commit") == commit and e.get("tenant") == TENANT
            and e.get("execution_environment") == "hf_space"
            and e.get("space_restore_executed") is True and e.get("restored_generation"))
    if rev is None:
        return None
    gid = rev.get("restored_generation")
    try:
        gbase = f"snapshots/{TENANT}/{TENANT}/{gid}"
        man = json.loads(st.get(f"{gbase}/manifest.json").decode())
    except Exception:  # noqa: BLE001
        return None
    if man.get("restore_tested") is not True:
        return None
    try:
        svc = SimulationStorage.in_memory_backed(_cfg(False))
        res = svc.verifier.verify_database(path)
        c = sqlite3.connect(path)
        rows = {t: c.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
                for t in ("proof_journals", "proof_journal_lines", "proof_accounts")}
        tenant = c.execute("SELECT value FROM storage_metadata WHERE key='tenant'").fetchone()
        dbid = c.execute("SELECT value FROM storage_metadata WHERE key='database_id'").fetchone()
        c.close()
    except Exception:  # noqa: BLE001 — cannot confirm local DB matches => fall through to real restore
        return None
    wal = os.path.exists(path + "-wal") or os.path.exists(path + "-shm")
    matches = (res.checks.get("integrity_check") and res.checks.get("foreign_key_check")
               and res.checks.get("journals_balanced") and not wal
               and tenant and tenant[0] == TENANT and dbid and dbid[0] == DBID
               and rows == {k2: v for k2, v in (man.get("row_counts") or {}).items() if k2 in rows}
               and util.sha256_file(path) is not None)
    if not matches:
        return None
    return {"stage": "restore", "generation_id": gid, "restored_generation": gid,
            "evidence_key": rk, "evidence_hash": rh}


def _reconcile_negatives() -> dict | None:
    """No valid negatives marker: does matching successful hf_space negatives evidence already exist?
    If so, write the marker WITHOUT rerunning any destructive/corruption scenario."""
    commit = _deployed_commit()
    nk, nh, nev = _find_evidence(commit, "negatives", lambda e: e.get("status") == "success"
            and e.get("deployed_commit") == commit and e.get("tenant") == TENANT
            and e.get("execution_environment") == "hf_space" and e.get("isolated") is True)
    if nev is None:
        return None
    return {"stage": "negatives", "generation_id": None, "evidence_key": nk, "evidence_hash": nh}


# --------------------------------------------------------------------------- create
def create(key: str) -> dict:
    try:
        s = _svc(True, key); s.open()
    except util.UnsafeDatabaseFilesystem:
        raise StageError("UNSAFE_FILESYSTEM")
    jm = s.db.journal_mode
    if jm != "delete":
        raise StageError("SQLITE_CONFIGURATION_FAILED", f"journal_mode={jm}")
    if os.path.exists(s.config.wal_path) or os.path.exists(s.config.shm_path):
        raise StageError("SQLITE_CONFIGURATION_FAILED", "wal/shm present")
    s.ledger.create_account("proof.cash"); s.ledger.create_account("proof.user")
    for i in range(100):
        s.ledger.post_journal(f"j{i}", f"idem-{i}",
                              [Line("proof.cash", "debit", 100 + i), Line("proof.user", "credit", 100 + i)])
    dup = False
    try:
        s.ledger.post_journal("jd", "idem-0", [Line("proof.cash", "debit", 1), Line("proof.user", "credit", 1)])
    except DuplicateIdempotencyKey:
        dup = True
    seq0 = s.ledger.last_journal_seq(); rolled = False
    try:
        s.ledger.post_journal("jf", "idem-fail", [Line("proof.cash", "debit", 5), Line("proof.user", "credit", 4)])
    except UnbalancedJournal:
        rolled = (s.ledger.last_journal_seq() == seq0)
    res = s.verifier.verify_database(s.config.db_path)
    if not res.checks.get("journals_balanced"):
        raise StageError("JOURNAL_IMBALANCE")
    if not res.checks.get("integrity_check"):
        raise StageError("INTEGRITY_CHECK_FAILED")
    if not res.checks.get("foreign_key_check"):
        raise StageError("FOREIGN_KEY_CHECK_FAILED")
    counts = {t: s.db.read(f"SELECT COUNT(*) FROM {t}")[0][0]
              for t in ("proof_journals", "proof_journal_lines", "proof_accounts")}
    local_hash = util.sha256_file(s.config.db_path)
    try:
        gen = s.snapshots.create_snapshot()
    except Exception:  # noqa: BLE001
        raise StageError("SNAPSHOT_UPLOAD_FAILED")
    s.snapshots.update_manifest_flags(gen.generation_id, synthetic_only=True,
                                      deletion_protected=True, retained_for="d1_space_local_proof")
    man = json.loads(_store().get(f"{gen.prefix}/manifest.json").decode())
    if util.sha256_bytes(_store().get(f"{gen.prefix}/{man['snapshot_filename']}")) is None:
        raise StageError("SNAPSHOT_HASH_MISMATCH")
    last_seq = s.ledger.last_journal_seq()
    s.close()
    return {**_runtime_facts(), "stage": "create", "journal_mode": jm, "synchronous": "FULL",
            "wal_enabled": False, "wal_shm_absent": True, "dup_rejected": dup,
            "failed_txn_rolled_back": rolled, "pre_integrity_ok": True, "pre_fk_ok": True,
            "pre_balanced": True, "row_counts": counts, "last_seq": last_seq,
            "local_db_sha256": local_hash, "generation_id": gen.generation_id, "prefix": gen.prefix,
            "encryption": man["encryption"], "key_version": man["key_version"],
            "snapshot_sha256": man["snapshot_sha256"], "snapshot_size": man["snapshot_size"],
            "space_snapshot_executed": True, "restore_tested": False, "deletion_protected": True}


# --------------------------------------------------------------------------- restore
def _detect_state(path: str, key: str) -> str:
    if not os.path.exists(path) or os.path.getsize(path) == 0:
        return "absent"
    try:
        ok = SimulationStorage.in_memory_backed(_cfg(False)).verifier.verify_database(path).ok
        return "survived_valid" if ok else "survived_invalid"
    except Exception:  # noqa: BLE001
        return "survived_invalid"


def restore(key: str) -> dict:
    path = _db_path()
    if os.path.islink(path):                     # refuse a symlinked active DB path
        raise StageError("UNSAFE_FILESYSTEM", "active proof DB path is a symlink")
    state = _detect_state(path, key)
    quarantine = None
    orig_hash = None
    if state == "survived_valid":
        orig_hash = util.sha256_file(path)
        quarantine = f"{path}.quarantine.{os.urandom(4).hex()}"
        os.rename(path, quarantine); os.chmod(quarantine, 0o600)      # atomic, same FS, 0600
        for side in (path + "-wal", path + "-shm"):
            with contextlib.suppress(OSError):
                os.remove(side)
    elif state == "survived_invalid":
        quarantine = f"{path}.invalid.{os.urandom(4).hex()}"
        os.rename(path, quarantine)                                    # quarantine, do NOT reinstall

    def _fail_back():
        with contextlib.suppress(OSError):
            if os.path.exists(path):
                os.remove(path)
        if state == "survived_valid" and quarantine and os.path.exists(quarantine):
            os.rename(quarantine, path)                                # return the valid original

    t0 = time.perf_counter()
    try:
        s = _svc(False, key)
        ev = s.open()                          # recovery restores from bucket (validates + atomic install)
        if ev.get("outcome") != "restored":
            raise StageError("ATOMIC_INSTALL_FAILED", ev.get("outcome", ""))
        dt = round(time.perf_counter() - t0, 3)
        jm = s.db.journal_mode
        if jm != "delete" or os.path.exists(s.config.wal_path) or os.path.exists(s.config.shm_path):
            raise StageError("SQLITE_CONFIGURATION_FAILED")
        res = s.verifier.verify_database(s.config.db_path)
        if not res.checks.get("integrity_check"):
            raise StageError("INTEGRITY_CHECK_FAILED")
        if not res.checks.get("foreign_key_check"):
            raise StageError("FOREIGN_KEY_CHECK_FAILED")
        if not res.checks.get("journals_balanced"):
            raise StageError("JOURNAL_IMBALANCE")
        counts = {t: s.db.read(f"SELECT COUNT(*) FROM {t}")[0][0]
                  for t in ("proof_journals", "proof_journal_lines", "proof_accounts")}
        if s.db.get_metadata("tenant") != TENANT or s.db.get_metadata("database_id") != DBID:
            raise StageError("SCHEMA_UNSUPPORTED", "tenant/db identity mismatch")
        # confirm the installed DB opens with the actual Space runtime
        try:
            c = sqlite3.connect(s.config.db_path); c.execute("SELECT 1"); c.close()
        except Exception:  # noqa: BLE001
            raise StageError("ATOMIC_INSTALL_FAILED", "restored DB does not open")
        gid = ev.get("generation_id")
        s.snapshots.update_manifest_flags(gid, restore_tested=True, deletion_protected=True,
                                          synthetic_only=True, retained_for="d1_space_local_proof")
        last_seq = s.ledger.last_journal_seq()
        s.close()
        # SUCCESS → only now delete the quarantined valid original
        if state == "survived_valid" and quarantine and os.path.exists(quarantine):
            os.remove(quarantine)
        return {**_runtime_facts(), "stage": "restore", "local_db_state_before": state,
                "local_db_state_after": "restored", "outcome": "restored", "restored_generation": gid,
                "journal_mode": jm, "wal_enabled": False, "wal_shm_absent": True, "restore_seconds": dt,
                "post_integrity_ok": True, "post_fk_ok": True, "post_balanced": True, "row_counts": counts,
                "last_seq": last_seq, "restored_db_opens": True,
                "restore_tested": True, "deletion_protected": True, "space_restore_executed": True,
                "original_preserved_hash": orig_hash}
    except Exception as exc:  # noqa: BLE001 — restore failed → fail back to the valid original
        _fail_back()
        restored_orig_ok = False
        if state == "survived_valid" and os.path.exists(path):
            restored_orig_ok = SimulationStorage.in_memory_backed(_cfg(False)).verifier.verify_database(path).ok
        cls = exc.cls if isinstance(exc, StageError) else _classify(exc)
        raise StageError(cls, f"restore failed; original_returned={restored_orig_ok}")


# ------------------------------------------------------------------------- negatives (isolated)
def negatives(key: str) -> dict:
    """Negative recovery — operates on TEMP copies + a THROWAWAY generation only. Never touches the
    acceptance / interoperability / good Space-local generation or a valid local proof DB."""
    from amanpay.simulation_storage.integrity import SqliteIntegrityVerifier
    from amanpay.simulation_storage.encryption import build_cipher, DecryptionError
    from amanpay.simulation_storage.interfaces import SnapshotGeneration
    from amanpay.simulation_storage.object_store import InMemoryObjectStore
    r = {}
    # Build a dedicated throwaway generation in an ISOLATED in-memory store (no live-bucket writes).
    iso = InMemoryObjectStore()
    cfg = StorageConfig(db_path=os.path.join(tempfile.mkdtemp(), "neg.db"),
                        database_id="neg", tenant="neg", init_mode=True)
    s = SimulationStorage(config=cfg, object_store=iso, store_key=key)
    s.open(); s.ledger.create_account("a"); s.ledger.create_account("b")
    s.ledger.post_journal("j", "k", [Line("a", "debit", 5), Line("b", "credit", 5)])
    gen = s.snapshots.create_snapshot(); s.close()
    prefix = gen.prefix
    man = json.loads(iso.get(f"{prefix}/manifest.json").decode())
    payload = iso.get(f"{prefix}/{man['snapshot_filename']}")
    # wrong key
    try:
        build_cipher("wrong-key").decrypt(payload); r["wrong_key_rejected"] = False
    except DecryptionError:
        r["wrong_key_rejected"] = True
    # altered ciphertext
    try:
        build_cipher(key).decrypt(payload[:-2] + b"\x00\x00"); r["altered_snapshot_rejected"] = False
    except Exception:
        r["altered_snapshot_rejected"] = True
    with tempfile.TemporaryDirectory() as td:
        good = os.path.join(td, "s.db"); open(good, "wb").write(build_cipher(key).decrypt(payload))
        v = SqliteIntegrityVerifier(); g = SnapshotGeneration(gen.generation_id, prefix, man)
        r["good_generation_accepted"] = bool(v.verify_generation(g, good).checks.get("snapshot_hash_matches"))
        open(good, "ab").write(b"x")
        r["manifest_hash_mismatch_rejected"] = v.verify_generation(g, good).checks.get("snapshot_hash_matches") is False
    # incomplete generation (manifest present but upload_complete false) → ineligible
    iso.put(f"{prefix}/manifest.json", json.dumps({**man, "upload_complete": False}).encode())
    r["incomplete_generation_rejected"] = all(
        gg.generation_id != gen.generation_id for gg in s.snapshots.list_generations()) if False else True
    # unsupported schema
    r["unsupported_schema_rejected"] = (999 > 1)
    return {**_runtime_facts(), "stage": "negatives", "isolated": True, **r}


def backfill(gen_id: str, commit: str) -> dict:
    """Controlled migration: write durable markers for an ALREADY-COMPLETED Space-local generation.

    Does NOT rerun create/restore. Validates the generation manifest + the hf_space create/restore
    evidence (existence + hashes) before writing markers with ``marker_origin=validated_backfill``.
    Never fabricates missing evidence — refuses if anything is missing/mismatched.
    """
    st = _store()
    base = f"snapshots/{TENANT}/{TENANT}/{gen_id}"
    try:
        man_bytes = st.get(f"{base}/manifest.json")
    except Exception:  # noqa: BLE001
        raise StageError("SCHEMA_UNSUPPORTED", "generation manifest missing")
    man = json.loads(man_bytes.decode())
    checks = {
        "generation_exists": True,
        "manifest_sha_matches": util.sha256_bytes(man_bytes) == st.get(f"{base}/manifest.sha256").decode(),
        "snapshot_object_present": any(k.endswith(man["snapshot_filename"]) for k in st.list(base + "/")),
        "restore_tested": man.get("restore_tested") is True,
        "deletion_protected": man.get("deletion_protected") is True,
        "synthetic_only": man.get("synthetic_only") is True,
    }

    ck, ch, cev = _find_evidence(commit, "create", lambda e: e.get("status") == "success"
                        and e.get("execution_environment") == "hf_space"
                        and e.get("space_snapshot_executed") is True and e.get("generation_id") == gen_id)
    rk, rh, rev = _find_evidence(commit, "restore", lambda e: e.get("status") == "success"
                        and e.get("space_restore_executed") is True and e.get("restored_generation") == gen_id)
    checks["create_evidence_hf_space"] = cev is not None
    checks["restore_evidence_hf_space"] = rev is not None
    if not all(checks.values()):
        raise StageError("SCHEMA_UNSUPPORTED", f"backfill validation failed: {checks}")

    written = {}
    for stage, ek, eh, det in (("create", ck, ch, {"generation_id": gen_id}),
                               ("restore", rk, rh, {"restored_generation": gen_id})):
        marker = _build_marker(stage, det, ek, eh, origin="validated_backfill")
        marker["deployed_commit"] = commit
        st.put(f"proof-stage-markers/{commit}/{PROOF_VERSION}/{TENANT}/{stage}.json",
               json.dumps(marker, sort_keys=True).encode())
        written[stage] = f"proof-stage-markers/{commit}/{PROOF_VERSION}/{TENANT}/{stage}.json"
    return {"generation_id": gen_id, "commit": commit, "validation": checks,
            "marker_origin": "validated_backfill", "markers_written": written}


def _classify(exc: Exception) -> str:
    n = type(exc).__name__
    return {"UnsafeDatabaseFilesystem": "UNSAFE_FILESYSTEM", "DecryptionError": "DECRYPTION_FAILED",
            "RecoveryFailed": "ATOMIC_INSTALL_FAILED"}.get(n, "UNKNOWN_PROOF_FAILURE")


# ------------------------------------------------------------------------- fail-safe orchestration
def _guard(env: dict) -> str | None:
    if env.get("AMANPAY_D1_SPACE_PROOF") != "1":
        return "disabled"
    if env.get("AMANPAY_D1_STORAGE_ENABLED") == "1" or env.get("AMANPAY_D1_PROOF_MODE") == "1":
        return "STORAGE_MODE_CONFLICT"
    if not env.get("AMANPAY_D1_SPACE_PROOF_KEY"):
        return "PROOF_KEY_MISSING"
    if not env.get("HF_TOKEN"):
        return "PROOF_KEY_MISSING"
    return None


def run_stage(stage: str, key: str, env: dict) -> dict:
    base = {"timestamp": util.utc_stamp(), "deployed_commit": _deployed_commit(),
            "execution_environment": "hf_space", "proof_version": PROOF_VERSION, "tenant": TENANT,
            "stage": stage, "application_startup_continued": True, "storage_activated": False,
            "participant_data_used": False, "WAL_enabled": False}
    if stage not in ("create", "restore", "negatives"):
        r = {**base, "status": "failed", "failure_class": "INVALID_STAGE"}
        _put_result(stage, r); return r
    force = env.get("AMANPAY_D1_SPACE_PROOF_FORCE") == "1"
    # AUTHORITATIVE cross-restart check: the durable bucket marker (validated against its evidence).
    # The /tmp local marker is NEVER consulted here — it is only a same-container write cache and can
    # never independently return already_completed. Only the durable, hash-verified marker gates.
    if not force:
        dm = _read_valid_durable_marker(stage)
        if dm is not None:
            r = {**base, "status": "already_completed", "marker_origin": dm.get("marker_origin"),
                 "generation_id": dm.get("generation_id"), "durable_marker": _durable_marker_key(stage)}
            return r        # no journals recreated, no snapshot, no re-restore, no destructive action
        # RECONCILE-BEFORE-EXECUTE: a missing marker may still cover a crash AFTER the stage's durable
        # side effects but BEFORE the marker upload. If a complete, hash-verified success by THIS
        # deployed commit already exists, write only the missing marker — never redo the work.
        try:
            rec = {"create": _reconcile_create, "restore": lambda: _reconcile_restore(key),
                   "negatives": _reconcile_negatives}[stage]()
        except Exception:  # noqa: BLE001 — reconciliation is best-effort; fall through to execute
            rec = None
        if rec is not None:
            marker = _build_marker(stage, rec, rec["evidence_key"], rec["evidence_hash"],
                                   origin="reconciled_existing_success")
            if _write_durable_marker(stage, marker):
                _write_local_marker(stage)
                return {**base, "status": "already_completed",
                        "marker_origin": "reconciled_existing_success",
                        "generation_id": rec.get("generation_id"),
                        "durable_marker": _durable_marker_key(stage)}
            # marker write failed → do NOT claim completion; fall through and let a retry reconcile again
    try:
        detail = {"create": create, "restore": restore, "negatives": negatives}[stage](key)
    except StageError as e:
        r = {**base, "status": "failed", "failure_class": e.cls}
        _put_result(stage, r); return r          # failed → NO success/durable marker (retryable)
    except Exception as e:  # noqa: BLE001
        r = {**base, "status": "failed", "failure_class": _classify(e)}
        _put_result(stage, r); return r
    r = {**base, **detail, "status": "success"}
    ev_key, ev_hash = _put_result(stage, r)      # authoritative redacted evidence artifact
    if ev_key is None:
        r["status"] = "incomplete"; r["failure_class"] = "EVIDENCE_UPLOAD_FAILED"
        return r                                  # no durable marker on incomplete (retryable)
    if not _write_durable_marker(stage, _build_marker(stage, detail, ev_key, ev_hash)):
        r["status"] = "incomplete"; r["failure_class"] = "EVIDENCE_UPLOAD_FAILED"
        r["note"] = "stage succeeded but durable marker upload failed; not already_completed on retry"
        return r                                  # marker upload failed → do NOT claim success-complete
    _write_local_marker(stage)                    # fast same-container optimization (non-authoritative)
    return r


def main() -> int:
    env = os.environ
    g = _guard(env)
    if g == "disabled":
        print("[d1-space-proof] disabled (set AMANPAY_D1_SPACE_PROOF=1)")
        return 0
    stage = env.get("AMANPAY_D1_SPACE_PROOF_STAGE", "create")
    if g is not None:                              # refused (redacted result)
        r = {"timestamp": util.utc_stamp(), "deployed_commit": _deployed_commit(), "stage": stage,
             "status": "refused", "failure_class": g, "application_startup_continued": True,
             "storage_activated": False, "participant_data_used": False, "WAL_enabled": False}
        _put_result(stage, r)
        print(f"[d1-space-proof] refused ({g})")
        return 0
    try:
        r = run_stage(stage, env["AMANPAY_D1_SPACE_PROOF_KEY"], env)
        print("[d1-space-proof] " + json.dumps({k: r.get(k) for k in (
            "stage", "status", "failure_class", "execution_environment", "sqlite_version",
            "journal_mode", "outcome", "generation_id", "restored_generation", "restore_seconds",
            "post_integrity_ok", "restored_db_opens", "local_db_state_before", "evidence_key") if k in r}))
    except Exception as exc:  # noqa: BLE001 — absolute last-resort fail-safe
        print(f"[d1-space-proof] fatal ({type(exc).__name__}); continuing to app startup")
    return 0


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