scripts/d1_space_diagnostic.py

#!/usr/bin/env python3
"""D1 HF Space runtime diagnostic — opt-in, one-shot, dormant probe.

Runs ONLY when ``AMANPAY_D1_RUNTIME_DIAGNOSTIC=1`` (default disabled). It collects redacted
runtime facts about the ACTUAL deployed Space and:
  * enables NO D1 storage, initializes NO database, restores/creates NO snapshot, enables NO
    WAL, exposes NO endpoint, requires NO Dev Mode, prints NO credentials;
  * writes a redacted JSON to a local path (``/tmp/amanpay-d1-space-runtime.json``);
  * optionally uploads the redacted JSON to the private proof bucket ONLY when
    ``AMANPAY_D1_RUNTIME_DIAGNOSTIC_UPLOAD=1`` (upload failure is non-fatal);
  * is one-shot per (deployed commit, diagnostic version) so ordinary restarts don't spam.

The report always asserts ``storage_activated=false`` and ``database_created=false``.
"""

from __future__ import annotations

import json
import os
import shutil

from amanpay.simulation_storage import util
from amanpay.simulation_storage.config import default_local_db_path
from amanpay.simulation_storage.sqlite_provider import classify_runtime, select_provider

DIAG_VERSION = "1"
DEFAULT_ARTIFACT = "/tmp/amanpay-d1-space-runtime.json"


def _artifact_path(env: dict) -> str:
    return env.get("AMANPAY_D1_DIAG_OUT", DEFAULT_ARTIFACT)


def _bucket(env: dict) -> str:
    return env.get("AMANPAY_BUCKET", "MHamdan/amanpay-d1-proof")

# credential env vars we report by PRESENCE only (never value)
_CRED_ENV = ("HF_TOKEN", "AMANPAY_STORE_KEY", "AMANPAY_D1_PROOF_KEY",
             "AMANPAY_S3_KEY_ID", "AMANPAY_S3_SECRET")


def _deployed_commit(env: dict) -> str:
    # Explicit override wins (testing / forced re-runs); otherwise the deployed build_info.json
    # is authoritative in the container; GITHUB_SHA is a CI fallback.
    if env.get("AMANPAY_COMMIT"):
        return str(env["AMANPAY_COMMIT"])
    for path in ("/app/build_info.json", "build_info.json"):
        try:
            with open(path) as fh:
                c = json.load(fh).get("commit")
                if c:
                    return str(c)
        except Exception:  # noqa: BLE001
            pass
    return env.get("GITHUB_SHA") or "unknown"


def _sqlite_compile_options() -> list[str]:
    import sqlite3
    con = sqlite3.connect(":memory:")   # in-memory only — creates NO database file
    try:
        return [r[0] for r in con.execute("PRAGMA compile_options").fetchall()]
    finally:
        con.close()


def _writable(parent: str) -> bool:
    try:
        os.makedirs(parent, exist_ok=True)
        probe = os.path.join(parent, ".d1_diag_write_probe")
        with open(probe, "w") as fh:
            fh.write("ok")
        os.remove(probe)
        return True
    except OSError:
        return False


def _disk(parent: str) -> dict:
    try:
        du = shutil.disk_usage(parent)
        return {"free_gb": round(du.free / 2**30, 2), "total_gb": round(du.total / 2**30, 2)}
    except OSError:
        return {"free_gb": None, "total_gb": None}


def build_report(env: dict | None = None, *, db_path: str | None = None) -> dict:
    """Pure, side-effect-free* report builder. (*only an in-memory SQLite connection.)"""
    env = env if env is not None else os.environ
    provider = select_provider()
    status, reason = classify_runtime(provider.version, provider.source_id)
    db_path = db_path or default_local_db_path()
    parent = os.path.dirname(os.path.abspath(db_path)) or "."
    real = os.path.realpath(db_path)
    fs_category, fstype = util.classify_filesystem(parent)
    wal_approved = (status == "APPROVED_WAL_RUNTIME")
    fs_local = (fs_category == "local")
    wal_eligible = wal_approved and fs_local
    if not wal_approved:
        wal_reason = f"runtime not WAL-approved [{status}]: {reason}"
    elif not fs_local:
        wal_reason = f"filesystem {fstype!r} classified {fs_category!r} (not confirmed local)"
    else:
        wal_reason = "approved runtime + confirmed-local filesystem"

    import sys
    return {
        "diagnostic_version": DIAG_VERSION,
        "diagnostic_timestamp": util.utc_stamp(),
        "deployed_commit": _deployed_commit(env),
        "python_version": sys.version.split()[0],
        "sqlite_version": provider.version,
        "sqlite_source_id": provider.source_id,
        "sqlite_compile_options": _sqlite_compile_options(),
        "approved_runtime_classification": status,
        "approved_runtime_reason": reason,
        "intended_active_db_path": db_path,
        "resolved_real_path": real,
        "filesystem_type": fstype,
        "filesystem_classification": fs_category,   # local | network | unknown
        "writable": _writable(parent),
        "disk": _disk(parent),
        "process_pid": os.getpid(),
        "worker_count": env.get("WEB_CONCURRENCY") or env.get("UVICORN_WORKERS") or "1 (single uvicorn process)",
        "hardware_env": env.get("SPACE_HARDWARE") or env.get("HF_SPACE_HARDWARE") or None,
        "persistent_volume_detected": os.path.isdir("/data"),
        "bucket_credential_presence": {k: bool(env.get(k)) for k in _CRED_ENV},
        "AMANPAY_D1_STORAGE_ENABLED": env.get("AMANPAY_D1_STORAGE_ENABLED", "unset"),
        "AMANPAY_D1_PROOF_MODE": env.get("AMANPAY_D1_PROOF_MODE", "unset"),
        "selected_default_journal_mode": "wal" if wal_eligible else "delete (rollback)",
        "wal_eligibility": wal_eligible,
        "wal_eligibility_reason": wal_reason,
        # hard invariants — the diagnostic NEVER activates storage or creates a DB
        "storage_activated": False,
        "database_created": False,
    }


def _marker_path(commit: str) -> str:
    return f"/tmp/.amanpay-d1-diag.{commit}.{DIAG_VERSION}.done"


def _maybe_upload(report: dict, env: dict) -> str:
    """Upload the redacted report to the private bucket. Non-fatal; returns a status string."""
    if env.get("AMANPAY_D1_RUNTIME_DIAGNOSTIC_UPLOAD") != "1":
        return "upload_disabled"
    token = env.get("HF_TOKEN")
    if not token:
        return "upload_skipped_no_token"
    try:
        from amanpay.simulation_storage.hf_bucket_store import HFBucketObjectStore
        key = f"runtime-diagnostics/{report['deployed_commit']}/{report['diagnostic_timestamp']}.json"
        HFBucketObjectStore(_bucket(env), token).put(key, json.dumps(report, sort_keys=True).encode())
        return f"uploaded:{key}"
    except Exception as exc:  # noqa: BLE001 — upload failure must not affect the caller
        return f"upload_failed:{type(exc).__name__}"


def _run_diagnostic(env: dict) -> int:
    """Build the report, write the local artifact, then (only on success) the one-shot marker."""
    report = build_report(env)
    marker = _marker_path(report["deployed_commit"])
    if os.path.exists(marker) and env.get("AMANPAY_D1_RUNTIME_DIAGNOSTIC_FORCE") != "1":
        print(f"[d1-diagnostic] already ran for commit {report['deployed_commit']} "
              f"(marker present); skipping. Set AMANPAY_D1_RUNTIME_DIAGNOSTIC_FORCE=1 to re-run.")
        return 0

    report["upload"] = _maybe_upload(report, env)   # never raises (returns a status string)
    artifact = _artifact_path(env)
    artifact_written = False
    try:
        os.makedirs(os.path.dirname(os.path.abspath(artifact)), exist_ok=True)
        with open(artifact, "w") as fh:
            json.dump(report, fh, indent=2, sort_keys=True)
        artifact_written = True
        print(f"[d1-diagnostic] wrote {artifact} (upload_status={report['upload']})")
    except OSError as exc:
        # No local artifact → do NOT create the marker; a later restart retries naturally.
        print(f"[d1-diagnostic] local write failed ({type(exc).__name__}); NOT creating one-shot "
              f"marker; a later restart will retry. storage_activated=false database_created=false")

    # One-shot marker ONLY after a successful local artifact write (regardless of upload outcome).
    if artifact_written:
        try:
            with open(marker, "w") as fh:
                fh.write(report["diagnostic_timestamp"])
        except OSError:
            pass   # marker is an optimization; failing to write it is harmless
        print(json.dumps({k: report.get(k) for k in (
            "deployed_commit", "python_version", "sqlite_version", "sqlite_source_id",
            "approved_runtime_classification", "filesystem_classification", "writable",
            "selected_default_journal_mode", "wal_eligibility", "storage_activated",
            "database_created", "upload")}, indent=2))
    return 0


def main() -> int:
    env = os.environ
    if env.get("AMANPAY_D1_RUNTIME_DIAGNOSTIC") != "1":
        print("[d1-diagnostic] disabled (set AMANPAY_D1_RUNTIME_DIAGNOSTIC=1 to run).")
        return 0
    # ANY diagnostic failure (filesystem/SQLite inspection, artifact write, unexpected) must NEVER
    # propagate, activate storage/WAL, create a DB, or block application startup. Log a short
    # REDACTED result (exception TYPE only — never the message/env values) and continue.
    try:
        return _run_diagnostic(env)
    except Exception as exc:  # noqa: BLE001 — fail-safe by design
        print(f"[d1-diagnostic] failed ({type(exc).__name__}); continuing to app startup. "
              f"storage_activated=false database_created=false")
        return 0


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