scripts/collect_paid_api_launch_evidence.py

#!/usr/bin/env python3
"""Collect sanitized Kaiju Coder 7 paid API launch evidence.

This script helps fill release/paid-api-launch-evidence.json without storing
API keys, secret values, full prompts, or model responses. It is preview-only by
default; pass --write to update the evidence file.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import statistics
import sys
import time
import urllib.error
import urllib.request
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_OUT = ROOT / "release/paid-api-launch-evidence.json"
MODEL_ID = "kaiju-coder-7"
DEFAULT_ROUTE = "/v1/chat/completions"
DEFAULT_USER_AGENT = "KaijuCoder7LaunchEvidence/1.0"
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")),
    ("bearer_token", re.compile(r"\bBearer\s+[A-Za-z0-9._~+/-]{24,}={0,2}\b", re.IGNORECASE)),
    ("private_key_block", re.compile(r"-----BEGIN (?:RSA |OPENSSH |EC |DSA )?PRIVATE KEY-----")),
]


def utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def load_existing(path: Path) -> dict[str, Any]:
    if not path.is_file():
        return {}
    return json.loads(path.read_text(encoding="utf-8"))


def secret_findings(text: str) -> list[str]:
    return sorted({label for label, pattern in SECRET_PATTERNS if pattern.search(text)})


def assert_sanitized(payload: dict[str, Any]) -> None:
    rendered = json.dumps(payload, sort_keys=True)
    findings = secret_findings(rendered)
    if findings:
        raise SystemExit("Refusing to write secret-looking evidence: " + ", ".join(findings))


def api_url(base_url: str, path: str) -> str:
    return base_url.rstrip("/") + path


def request_json(url: str, payload: dict[str, Any], api_key: str, request_id: str, timeout: int) -> tuple[int, str, float]:
    body = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=body,
        method="POST",
        headers={
            "authorization": f"Bearer {api_key}",
            "content-type": "application/json",
            "user-agent": DEFAULT_USER_AGENT,
            "x-request-id": request_id,
        },
    )
    start = time.perf_counter()
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            content_type = response.headers.get("content-type", "")
            response.read()
            return response.status, content_type, (time.perf_counter() - start) * 1000
    except urllib.error.HTTPError as exc:
        exc.read()
        return exc.code, exc.headers.get("content-type", ""), (time.perf_counter() - start) * 1000


def probe_health(base_url: str, timeout: int) -> tuple[int, float] | None:
    start = time.perf_counter()
    try:
        request = urllib.request.Request(
            api_url(base_url, "/health"),
            headers={"user-agent": DEFAULT_USER_AGENT},
        )
        with urllib.request.urlopen(request, timeout=timeout) as response:
            response.read()
            return response.status, (time.perf_counter() - start) * 1000
    except Exception:
        return None


def percentile_95(values: list[float]) -> float:
    if len(values) == 1:
        return values[0]
    try:
        return statistics.quantiles(values, n=20, method="inclusive")[18]
    except statistics.StatisticsError:
        return max(values)


def run_staging_samples(args: argparse.Namespace) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
    if args.skip_live_request:
        return None, None
    if not args.api_base_url:
        raise SystemExit("--api-base-url is required unless --skip-live-request is set")
    api_key = os.environ.get(args.api_key_env)
    if not api_key:
        raise SystemExit(f"{args.api_key_env} is not set; refusing to read API keys from arguments")

    latencies: list[float] = []
    first_request_id = ""
    first_status = 0
    first_streamed = False
    url = api_url(args.api_base_url, DEFAULT_ROUTE)
    sample_count = max(args.live_samples, 1)
    for index in range(sample_count):
        request_id = f"kaiju-paid-staging-{uuid.uuid4()}"
        payload = {
            "model": MODEL_ID,
            "stream": True,
            "max_tokens": 48,
            "messages": [
                {
                    "role": "user",
                    "content": "Return a short Kaiju Coder 7 paid API staging smoke response.",
                }
            ],
        }
        status, content_type, latency_ms = request_json(url, payload, api_key, request_id, args.timeout)
        if index == 0:
            first_request_id = request_id
            first_status = status
            first_streamed = "event-stream" in content_type.lower()
        if status == 200:
            latencies.append(latency_ms)

    request_evidence = {
        "status": "pass" if first_status == 200 and first_streamed else "pending",
        "checked_at": utc_now(),
        "route": DEFAULT_ROUTE,
        "model": MODEL_ID,
        "http_status": first_status,
        "streamed": first_streamed,
        "request_id": first_request_id,
    }
    latency_evidence = {
        "status": "pass" if len(latencies) >= 5 else "pending",
        "checked_at": utc_now(),
        "route": DEFAULT_ROUTE,
        "sample_count": len(latencies),
        "p95_ms": round(percentile_95(latencies), 2) if latencies else 0,
        "max_acceptable_ms": args.max_acceptable_ms,
    }
    return request_evidence, latency_evidence


def add_optional_manual_evidence(evidence: dict[str, Any], args: argparse.Namespace) -> None:
    checked_at = args.checked_at or utc_now()
    if args.public_route_ok:
        health = probe_health(args.api_base_url, args.timeout) if args.api_base_url else None
        exposure_mode = args.public_route_mode
        evidence["public_route_mode"] = {
            "status": "pass",
            "checked_at": checked_at,
            "exposure_mode": exposure_mode,
            "route": args.api_base_url,
            "result": f"{exposure_mode} route resolves to the intended Kaiju Worker"
            + (f"; /health={health[0]} in {health[1]:.0f}ms" if health else ""),
        }
    if args.wrangler_secret_name:
        evidence["wrangler_secrets_verified"] = {
            "status": "pass",
            "checked_at": checked_at,
            "command": "wrangler secret list",
            "observed_names": sorted(set(args.wrangler_secret_name)),
        }
    if args.d1_migration_result:
        evidence["d1_migration_applied"] = {
            "status": "pass",
            "checked_at": checked_at,
            "command": args.d1_migration_command,
            "migration": "0001_paid_api.sql",
            "result": args.d1_migration_result,
        }
    if args.stripe_checkout_topup_pass:
        evidence["stripe_checkout_topup_staging"] = {
            "status": "pass",
            "checked_at": checked_at,
            "mode": args.stripe_mode,
            "webhook_event": "checkout.session.completed",
            "credited_api_key_id": args.credited_api_key_id,
            "idempotency_checked": args.stripe_idempotency_checked,
        }
    if args.rollback_result:
        evidence["rollback_exercised"] = {
            "status": "pass",
            "checked_at": checked_at,
            "command": args.rollback_command,
            "result": args.rollback_result,
        }
    if args.staging_request_id:
        evidence["worker_to_gojira_staging_request"] = {
            "status": "pass",
            "checked_at": checked_at,
            "route": DEFAULT_ROUTE,
            "model": MODEL_ID,
            "http_status": args.staging_http_status,
            "streamed": args.staging_streamed,
            "request_id": args.staging_request_id,
        }
    if args.paid_route_p95_ms is not None:
        evidence["paid_route_latency"] = {
            "status": "pass",
            "checked_at": checked_at,
            "route": DEFAULT_ROUTE,
            "sample_count": args.paid_route_sample_count,
            "p95_ms": args.paid_route_p95_ms,
            "max_acceptable_ms": args.max_acceptable_ms,
        }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
    parser.add_argument("--write", action="store_true", help="Write the evidence file. Default is preview only.")
    parser.add_argument("--merge-existing", action="store_true", help="Merge with existing evidence at --out.")
    parser.add_argument("--checked-at", help="Override checked_at timestamp for manual evidence.")
    parser.add_argument("--api-base-url", default="", help="Public paid API base URL, for example https://api.example.com.")
    parser.add_argument("--api-key-env", default="KAIJU_PAID_API_KEY", help="Environment variable containing the staging API key.")
    parser.add_argument("--timeout", type=int, default=120)
    parser.add_argument("--skip-live-request", action="store_true", help="Do not call the paid API.")
    parser.add_argument("--live-samples", type=int, default=5)
    parser.add_argument("--max-acceptable-ms", type=float, default=120_000)
    parser.add_argument("--public-route-ok", action="store_true", help="Record public custom-domain route evidence.")
    parser.add_argument(
        "--public-route-mode",
        choices=["workers_dev", "custom_domain"],
        default="workers_dev",
        help="Public route type used for the launch evidence.",
    )
    parser.add_argument("--wrangler-secret-name", action="append", default=[], help="Observed Wrangler secret name. Repeatable.")
    parser.add_argument("--d1-migration-result", choices=["success", "already_applied"])
    parser.add_argument(
        "--d1-migration-command",
        default="wrangler d1 migrations apply KAIJU_BILLING_DB --remote",
    )
    parser.add_argument("--stripe-checkout-topup-pass", action="store_true")
    parser.add_argument("--stripe-mode", default="test")
    parser.add_argument("--credited-api-key-id", default="key_staging_001")
    parser.add_argument("--stripe-idempotency-checked", action="store_true")
    parser.add_argument("--rollback-result", choices=["success"])
    parser.add_argument("--rollback-command", default="wrangler rollback")
    parser.add_argument("--staging-request-id", help="Sanitized request id from a separate staging request.")
    parser.add_argument("--staging-http-status", type=int, default=200)
    parser.add_argument("--staging-streamed", action="store_true")
    parser.add_argument("--paid-route-p95-ms", type=float)
    parser.add_argument("--paid-route-sample-count", type=int, default=5)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    evidence = load_existing(args.out) if args.merge_existing else {}
    add_optional_manual_evidence(evidence, args)
    request_evidence, latency_evidence = run_staging_samples(args)
    if request_evidence:
        evidence["worker_to_gojira_staging_request"] = request_evidence
    if latency_evidence:
        evidence["paid_route_latency"] = latency_evidence

    assert_sanitized(evidence)
    rendered = json.dumps(evidence, indent=2, sort_keys=True) + "\n"
    if args.write:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(rendered, encoding="utf-8")
        print(f"Wrote sanitized paid API evidence to {args.out}")
    else:
        print(rendered, end="")
        print("Preview only. Pass --write to update the evidence file.", file=sys.stderr)
    return 0


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