scripts/check_paid_api_readiness.py

#!/usr/bin/env python3
"""Check Kaiju Coder 7 paid API readiness without reading secrets.

The scaffold mode should pass for the local Worker implementation. The launch
mode is intentionally stricter and should fail until real Cloudflare bindings,
Stripe webhook evidence, staging requests, and rollback proof are attached.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
WORKER = ROOT / "gateway/cloudflare-worker"
WRANGLER = WORKER / "wrangler.jsonc"
SOURCE = WORKER / "src/index.js"
TESTS = WORKER / "test/index.test.js"
MIGRATION = WORKER / "migrations/0001_paid_api.sql"
PACKAGE = WORKER / "package.json"
PAID_DOC = ROOT / "release/PAID_API_READINESS.md"
RESOURCE_SCRIPT = ROOT / "scripts/prepare_paid_api_cloudflare_resources.sh"
DEFAULT_EVIDENCE = ROOT / "release/paid-api-launch-evidence.json"
EVIDENCE_EXAMPLE = ROOT / "release/paid-api-launch-evidence.example.json"
CF_BINDINGS_EXAMPLE = ROOT / "release/cloudflare-bindings.example.json"
SECRET_PATTERNS = [
    ("openai_api_key", re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_-]{20,}\b")),
    ("anthropic_api_key", re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b")),
    ("stripe_secret_key", re.compile(r"\b[rs]k_(?:live|test)_[A-Za-z0-9]{16,}\b")),
    ("stripe_webhook_secret", re.compile(r"\bwhsec_[A-Za-z0-9]{16,}\b")),
    ("huggingface_token", re.compile(r"\bhf_[A-Za-z0-9]{20,}\b")),
    ("github_token", re.compile(r"\b(?:ghp_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{22,})\b")),
    ("google_api_key", re.compile(r"\bAIza[0-9A-Za-z_-]{20,}\b")),
    ("private_key_block", re.compile(r"-----BEGIN (?:RSA |OPENSSH |EC |DSA )?PRIVATE KEY-----")),
    ("bearer_token", re.compile(r"\bBearer\s+[A-Za-z0-9._~+/-]{24,}={0,2}\b", re.IGNORECASE)),
]


@dataclass
class Check:
    name: str
    status: str
    detail: str


def file_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def strip_jsonc(text: str) -> str:
    text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
    lines = []
    for line in text.splitlines():
        in_string = False
        escaped = False
        output = []
        index = 0
        while index < len(line):
            char = line[index]
            nxt = line[index + 1] if index + 1 < len(line) else ""
            if char == "\\" and in_string:
                escaped = not escaped
                output.append(char)
            elif char == '"' and not escaped:
                in_string = not in_string
                output.append(char)
            elif char == "/" and nxt == "/" and not in_string:
                break
            else:
                escaped = False
                output.append(char)
            index += 1
        lines.append("".join(output))
    stripped = "\n".join(lines)
    return re.sub(r",\s*([}\]])", r"\1", stripped)


def load_wrangler(path: Path = WRANGLER) -> dict[str, Any]:
    return json.loads(strip_jsonc(file_text(path)))


def has_real_binding(bindings: list[dict[str, Any]], binding_name: str, id_field: str) -> bool:
    for binding in bindings:
        if binding.get("binding") != binding_name:
            continue
        value = str(binding.get(id_field, "")).strip()
        return bool(value) and not value.startswith("replace_with_")
    return False


def load_launch_evidence(path: Path) -> tuple[dict[str, Any], Check | None]:
    if not path.is_file():
        return {}, None
    text = file_text(path)
    findings = [label for label, pattern in SECRET_PATTERNS if pattern.search(text)]
    if findings:
        return {}, Check(
            "paid API launch evidence file",
            "fail",
            f"{path} appears to contain secret-looking values: {', '.join(sorted(set(findings)))}",
        )
    try:
        evidence = json.loads(text)
    except json.JSONDecodeError as exc:
        return {}, Check("paid API launch evidence file", "fail", f"{path} is invalid JSON: {exc}")
    if not isinstance(evidence, dict):
        return {}, Check("paid API launch evidence file", "fail", f"{path} must contain a JSON object")
    return evidence, Check("paid API launch evidence file", "pass", f"loaded sanitized launch evidence from {path}")


def evidence_item(
    checks: list[Check],
    evidence: dict[str, Any],
    path: Path,
    key: str,
    name: str,
    required_fields: list[str],
    validator: Any | None = None,
) -> None:
    item = evidence.get(key)
    if not item:
        checks.append(Check(name, "manual", f"attach sanitized evidence in {path} key `{key}`"))
        return
    if not isinstance(item, dict):
        checks.append(Check(name, "fail", f"{path} key `{key}` must be an object"))
        return
    if item.get("status") != "pass":
        checks.append(Check(name, "manual", f"{path} key `{key}` status is not pass"))
        return
    missing = [field for field in required_fields if item.get(field) in (None, "", [])]
    if missing:
        checks.append(Check(name, "manual", f"{path} key `{key}` missing fields: {', '.join(missing)}"))
        return
    if validator:
        validation = validator(item)
        if validation:
            checks.append(Check(name, validation[0], validation[1]))
            return
    checks.append(Check(name, "pass", f"{path} key `{key}` has required sanitized evidence"))


def validate_public_route_mode(item: dict[str, Any]) -> tuple[str, str] | None:
    if item.get("exposure_mode") != "custom_domain":
        return ("manual", "public route evidence must use exposure_mode=custom_domain before paid launch")
    route = str(item.get("route", ""))
    if not route.startswith("https://"):
        return ("manual", "public route evidence route must be an https URL")
    return None


def validate_secrets_verified(item: dict[str, Any]) -> tuple[str, str] | None:
    required = {"KAIJU_ORIGIN_URL", "KAIJU_ORIGIN_SECRET", "KAIJU_STRIPE_WEBHOOK_SECRET"}
    observed = set(item.get("observed_names") or [])
    missing = sorted(required - observed)
    if missing:
        return ("manual", "secret-name evidence missing: " + ", ".join(missing))
    return None


def validate_d1_migration(item: dict[str, Any]) -> tuple[str, str] | None:
    if item.get("migration") != "0001_paid_api.sql":
        return ("manual", "D1 migration evidence must name 0001_paid_api.sql")
    if item.get("result") not in {"success", "already_applied"}:
        return ("manual", "D1 migration result must be success or already_applied")
    return None


def validate_stripe_staging(item: dict[str, Any]) -> tuple[str, str] | None:
    if item.get("webhook_event") != "checkout.session.completed":
        return ("manual", "Stripe evidence must include checkout.session.completed")
    if item.get("idempotency_checked") is not True:
        return ("manual", "Stripe evidence must confirm duplicate webhook idempotency")
    return None


def validate_staging_request(item: dict[str, Any]) -> tuple[str, str] | None:
    if item.get("model") != "kaiju-coder-7":
        return ("fail", "staging request evidence must use model=kaiju-coder-7")
    if int(item.get("http_status") or 0) != 200:
        return ("manual", "staging request evidence must show HTTP 200")
    if item.get("streamed") is not True:
        return ("manual", "staging request evidence must confirm streaming")
    return None


def validate_rollback(item: dict[str, Any]) -> tuple[str, str] | None:
    if item.get("result") != "success":
        return ("manual", "rollback evidence must be a successful exercised rollback or route switch")
    return None


def validate_latency(item: dict[str, Any]) -> tuple[str, str] | None:
    p95_ms = float(item.get("p95_ms") or 0)
    sample_count = int(item.get("sample_count") or 0)
    max_acceptable_ms = float(item.get("max_acceptable_ms") or 0)
    if sample_count < 5:
        return ("manual", "latency evidence needs at least 5 staging samples")
    if max_acceptable_ms <= 0:
        return ("manual", "latency evidence must set max_acceptable_ms")
    if p95_ms <= 0 or p95_ms > max_acceptable_ms:
        return ("manual", f"p95_ms={p95_ms:g} exceeds max_acceptable_ms={max_acceptable_ms:g}")
    return None


def add_marker_check(checks: list[Check], name: str, text: str, markers: list[str], path: Path) -> None:
    missing = [marker for marker in markers if marker not in text]
    if missing:
        checks.append(Check(name, "fail", f"{path} missing markers: {', '.join(missing)}"))
    else:
        checks.append(Check(name, "pass", f"{path} contains required markers"))


def scaffold_checks(wrangler_path: Path = WRANGLER) -> list[Check]:
    checks: list[Check] = []
    source = file_text(SOURCE)
    tests = file_text(TESTS)
    migration = file_text(MIGRATION)
    package = json.loads(file_text(PACKAGE))
    paid_doc = file_text(PAID_DOC)
    resource_script = file_text(RESOURCE_SCRIPT)
    wrangler = load_wrangler(wrangler_path)

    add_marker_check(
        checks,
        "model id enforcement",
        source,
        [
            'const DEFAULT_MODEL_ID = "kaiju-coder-7"',
            "Unsupported model. Use",
            "payload.model = modelId",
        ],
        SOURCE,
    )
    add_marker_check(
        checks,
        "streaming and thinking controls",
        source,
        ["payload.stream = true", "enable_thinking: false", "thinking: false", "streamHeaders"],
        SOURCE,
    )
    add_marker_check(
        checks,
        "billing and debit/refund controls",
        source,
        ["KAIJU_BILLING_DB", "reserveCredit", "refundCredit", "markUsageDebited"],
        SOURCE,
    )
    add_marker_check(
        checks,
        "rate limit controls",
        source,
        ["KAIJU_RATE_LIMIT_KV", "rateLimit", "Rate limit exceeded"],
        SOURCE,
    )
    add_marker_check(
        checks,
        "secret-like prompt rejection",
        source,
        ["SECRET_PATTERNS", "secret_like_content", "Remove them before using Kaiju Coder 7"],
        SOURCE,
    )
    add_marker_check(
        checks,
        "stripe top-up webhook",
        source,
        ["verifyStripeSignature", "checkout.session.completed", "stripe_topup_credited"],
        SOURCE,
    )
    add_marker_check(
        checks,
        "artifact route controls",
        source,
        [
            "KAIJU_ARTIFACT_BUCKET",
            "uploadArtifact",
            "downloadArtifact",
            "artifact_stored",
            "Artifact appears to contain secrets or credentials",
        ],
        SOURCE,
    )
    add_marker_check(
        checks,
        "paid API tests",
        tests,
        [
            "rejects inactive paid API keys",
            "rejects paid API requests with insufficient credits before origin fetch",
            "rate limits authenticated paid API keys before debit",
            "credits paid API balance from signed Stripe checkout webhook",
            "rejects secret-looking prompt content before debit",
        ],
        TESTS,
    )
    add_marker_check(
        checks,
        "artifact route tests",
        tests,
        [
            "stores origin-uploaded artifacts in the account/request R2 namespace",
            "serves customer artifacts only through the authenticated account namespace",
            "rejects unsafe artifact paths before R2 storage",
            "rejects secret-looking artifact content before R2 storage",
        ],
        TESTS,
    )
    add_marker_check(
        checks,
        "D1 schema",
        migration,
        ["kaiju_api_keys", "kaiju_credit_ledger", "kaiju_usage_events"],
        MIGRATION,
    )
    add_marker_check(
        checks,
        "paid readiness docs",
        paid_doc,
        [
            "Do not sell the hosted API",
            "Harnessed customer-readiness pack",
            "Raw OpenCode multi-file pack remains a blocker",
        ],
        PAID_DOC,
    )

    if (
        package.get("scripts", {}).get("check")
        == "node --check src/index.js && node --check scripts/create-api-key.mjs && npm test && npm run check:deploy"
        and package.get("scripts", {}).get("check:deploy") == "npx wrangler deploy --dry-run"
    ):
        checks.append(Check("gateway check command", "pass", "npm run check covers syntax, Worker tests, and Wrangler dry-run deploy"))
    else:
        checks.append(Check("gateway check command", "fail", "package.json check/check:deploy scripts changed or missing"))

    if package.get("scripts", {}).get("prepare:cloudflare") == "bash ../../scripts/prepare_paid_api_cloudflare_resources.sh":
        checks.append(Check("Cloudflare resource prep command", "pass", "npm run prepare:cloudflare is wired"))
    else:
        checks.append(Check("Cloudflare resource prep command", "fail", "package.json prepare:cloudflare script is missing"))

    add_marker_check(
        checks,
        "Cloudflare resource prep script",
        resource_script,
        [
            "KAIJU_CF_RESOURCE_APPLY",
            "wrangler d1 create",
            "wrangler kv namespace create",
            "wrangler r2 bucket create",
            "wrangler d1 migrations apply",
            "wrangler rollback",
            "preflight:launch",
        ],
        RESOURCE_SCRIPT,
    )
    if EVIDENCE_EXAMPLE.is_file():
        checks.append(Check("paid API launch evidence template", "pass", f"{EVIDENCE_EXAMPLE} exists"))
    else:
        checks.append(Check("paid API launch evidence template", "fail", f"missing {EVIDENCE_EXAMPLE}"))

    if CF_BINDINGS_EXAMPLE.is_file():
        checks.append(Check("Cloudflare bindings template", "pass", f"{CF_BINDINGS_EXAMPLE} exists"))
    else:
        checks.append(Check("Cloudflare bindings template", "fail", f"missing {CF_BINDINGS_EXAMPLE}"))

    if wrangler.get("name") == "kaiju-api-gateway" and wrangler.get("main") == "src/index.js":
        checks.append(Check("wrangler scaffold config", "pass", "Worker name and entrypoint are present"))
    else:
        checks.append(Check("wrangler scaffold config", "fail", "wrangler name or entrypoint is missing"))

    return checks


def launch_checks(evidence_path: Path, wrangler_path: Path = WRANGLER) -> list[Check]:
    checks = scaffold_checks(wrangler_path)
    wrangler = load_wrangler(wrangler_path)
    evidence, evidence_check = load_launch_evidence(evidence_path)
    if evidence_check and evidence_check.status == "fail":
        checks.append(evidence_check)

    if has_real_binding(wrangler.get("d1_databases", []), "KAIJU_BILLING_DB", "database_id"):
        checks.append(Check("live D1 binding", "pass", "KAIJU_BILLING_DB has a non-placeholder database_id"))
    else:
        checks.append(Check("live D1 binding", "fail", "KAIJU_BILLING_DB is missing or still placeholder/commented"))

    if has_real_binding(wrangler.get("kv_namespaces", []), "KAIJU_RATE_LIMIT_KV", "id"):
        checks.append(Check("live KV binding", "pass", "KAIJU_RATE_LIMIT_KV has a non-placeholder id"))
    else:
        checks.append(Check("live KV binding", "fail", "KAIJU_RATE_LIMIT_KV is missing or still placeholder/commented"))

    if has_real_binding(wrangler.get("r2_buckets", []), "KAIJU_ARTIFACT_BUCKET", "bucket_name"):
        checks.append(Check("artifact R2 binding", "pass", "KAIJU_ARTIFACT_BUCKET is configured"))
    else:
        checks.append(Check("artifact R2 binding", "fail", "KAIJU_ARTIFACT_BUCKET is missing; artifact routes cannot launch"))

    if wrangler.get("workers_dev") is False:
        checks.append(Check("public route mode", "pass", "workers_dev is disabled for custom-domain launch"))
    else:
        evidence_item(
            checks,
            evidence,
            evidence_path,
            "public_route_mode",
            "public route mode",
            ["checked_at", "exposure_mode", "route", "result"],
            validate_public_route_mode,
        )

    evidence_item(
        checks,
        evidence,
        evidence_path,
        "wrangler_secrets_verified",
        "wrangler secret list confirms KAIJU_ORIGIN_URL, KAIJU_ORIGIN_SECRET, and KAIJU_STRIPE_WEBHOOK_SECRET",
        ["checked_at", "command", "observed_names"],
        validate_secrets_verified,
    )
    evidence_item(
        checks,
        evidence,
        evidence_path,
        "d1_migration_applied",
        "D1 migration 0001_paid_api.sql applied to the live billing database",
        ["checked_at", "command", "migration", "result"],
        validate_d1_migration,
    )
    evidence_item(
        checks,
        evidence,
        evidence_path,
        "stripe_checkout_topup_staging",
        "Stripe Checkout top-up products and webhook endpoint tested with metadata.kaiju_api_key_id",
        ["checked_at", "mode", "webhook_event", "credited_api_key_id", "idempotency_checked"],
        validate_stripe_staging,
    )
    evidence_item(
        checks,
        evidence,
        evidence_path,
        "worker_to_gojira_staging_request",
        "staging request passed through Worker to Gojira-B origin with model=kaiju-coder-7",
        ["checked_at", "route", "model", "http_status", "streamed", "request_id"],
        validate_staging_request,
    )
    evidence_item(
        checks,
        evidence,
        evidence_path,
        "rollback_exercised",
        "rollback command or route switch was exercised and recorded",
        ["checked_at", "command", "result"],
        validate_rollback,
    )
    evidence_item(
        checks,
        evidence,
        evidence_path,
        "paid_route_latency",
        "p95 latency for paid routes is recorded after staging traffic",
        ["checked_at", "route", "sample_count", "p95_ms", "max_acceptable_ms"],
        validate_latency,
    )

    return checks


def summarize(checks: list[Check], mode: str) -> dict[str, Any]:
    hard_fail = any(check.status == "fail" for check in checks)
    manual = any(check.status == "manual" for check in checks)
    ready = not hard_fail and (mode == "scaffold" or not manual)
    return {
        "mode": mode,
        "ready": ready,
        "summary": {
            "pass": sum(1 for check in checks if check.status == "pass"),
            "fail": sum(1 for check in checks if check.status == "fail"),
            "manual": sum(1 for check in checks if check.status == "manual"),
        },
        "checks": [asdict(check) for check in checks],
    }


def print_text(result: dict[str, Any]) -> None:
    print(f"Kaiju Coder 7 paid API readiness: mode={result['mode']} ready={result['ready']}")
    print(
        "Summary: "
        f"{result['summary']['pass']} pass, "
        f"{result['summary']['fail']} fail, "
        f"{result['summary']['manual']} manual"
    )
    for check in result["checks"]:
        print(f"[{check['status']}] {check['name']} - {check['detail']}")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--mode", choices=["scaffold", "launch"], default="scaffold")
    parser.add_argument("--evidence-file", type=Path, default=DEFAULT_EVIDENCE)
    parser.add_argument("--wrangler-config", type=Path, default=WRANGLER)
    parser.add_argument("--json", action="store_true", help="Print machine-readable JSON.")
    args = parser.parse_args()

    checks = scaffold_checks(args.wrangler_config) if args.mode == "scaffold" else launch_checks(args.evidence_file, args.wrangler_config)
    result = summarize(checks, args.mode)
    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print_text(result)
    return 0 if result["ready"] else 1


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