runtime-behavior-probe

1"""Disposable Python probe scaffold.2 3Copy this file to a temporary location and adapt it for one narrow question.4Recommended usage from the repository root:5 6    uv run python /tmp/probe.py7 8If you want structured artifacts for repeat-heavy or benchmark probes:9 10    PROBE_OUTPUT_DIR=/tmp/probe-run uv run python /tmp/probe.py11"""12 13from __future__ import annotations14 15import json16import os17import platform18import shutil19import statistics20import subprocess21import sys22import time23import uuid24from collections import Counter, defaultdict25from importlib import metadata26from pathlib import Path27 28SCENARIO = "replace-me"29RUN_LABEL = "replace-me"30MODE = "single-shot"31APPROVED_ENV_VARS: list[str] = []32OUTPUT_DIR_ENV = "PROBE_OUTPUT_DIR"33 34RESULTS: list[dict[str, object]] = []35 36 37def _git_value(*args: str) -> str:38    result = subprocess.run(39        ["git", *args],40        check=False,41        capture_output=True,42        text=True,43    )44    if result.returncode != 0:45        return "unknown"46    return result.stdout.strip() or "unknown"47 48 49def _package_version(name: str) -> str | None:50    try:51        return metadata.version(name)52    except metadata.PackageNotFoundError:53        return None54 55 56def _output_dir() -> Path | None:57    value = os.getenv(OUTPUT_DIR_ENV)58    if not value:59        return None60    return Path(value)61 62 63def _write_json(path: Path, payload: object) -> None:64    path.parent.mkdir(parents=True, exist_ok=True)65    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")66 67 68def emit(kind: str, **payload: object) -> None:69    print(70        json.dumps(71            {72                "ts": round(time.time(), 3),73                "kind": kind,74                **payload,75            },76            sort_keys=True,77        )78    )79 80 81def runtime_context() -> dict[str, object]:82    approved = {name: ("set" if os.getenv(name) else "unset") for name in APPROVED_ENV_VARS}83    package_versions = {84        name: version85        for name in ("openai", "agents")86        if (version := _package_version(name)) is not None87    }88    return {89        "scenario": SCENARIO,90        "run_label": RUN_LABEL,91        "mode": MODE,92        "cwd": os.getcwd(),93        "script_path": str(Path(__file__).resolve()),94        "python_executable": sys.executable,95        "python_version": sys.version.split()[0],96        "platform": platform.platform(),97        "git_commit": _git_value("rev-parse", "HEAD"),98        "git_branch": _git_value("rev-parse", "--abbrev-ref", "HEAD"),99        "uv_path": shutil.which("uv"),100        "package_versions": package_versions,101        "approved_env_vars": approved,102        "output_dir": str(_output_dir()) if _output_dir() else None,103    }104 105 106def start_case(case_id: str, *, mode: str = MODE, note: str | None = None) -> None:107    emit("case_start", case_id=case_id, mode=mode, note=note)108 109 110def record_case_result(111    case_id: str,112    observation_summary: str,113    result_flag: str,114    *,115    mode: str = MODE,116    is_warmup: bool = False,117    total_latency_s: float | None = None,118    first_token_latency_s: float | None = None,119    metrics: dict[str, object] | None = None,120    error: str | None = None,121) -> None:122    payload: dict[str, object] = {123        "case_id": case_id,124        "mode": mode,125        "is_warmup": is_warmup,126        "observation_summary": observation_summary,127        "result_flag": result_flag,128        "metrics": metrics or {},129        "error": error,130    }131    if total_latency_s is not None:132        payload["total_latency_s"] = total_latency_s133    if first_token_latency_s is not None:134        payload["first_token_latency_s"] = first_token_latency_s135    RESULTS.append(payload)136    emit("case_result", **payload)137 138 139def summarize_results() -> dict[str, object]:140    by_case: defaultdict[str, list[dict[str, object]]] = defaultdict(list)141    for result in RESULTS:142        by_case[str(result["case_id"])].append(result)143 144    summary_cases: dict[str, object] = {}145    for case_id, items in by_case.items():146        measured = [item for item in items if not bool(item.get("is_warmup"))]147        latencies = [148            float(item["total_latency_s"])149            for item in measured150            if item.get("total_latency_s") is not None151        ]152        first_token_latencies = [153            float(item["first_token_latency_s"])154            for item in measured155            if item.get("first_token_latency_s") is not None156        ]157        result_flags = Counter(str(item["result_flag"]) for item in measured or items)158        observations = [str(item["observation_summary"]) for item in (measured or items)[:3]]159        summary_cases[case_id] = {160            "mode": str(items[-1]["mode"]),161            "runs": len(measured),162            "warmups": len(items) - len(measured),163            "result_flags": dict(result_flags),164            "median_total_latency_s": (statistics.median(latencies) if latencies else None),165            "mean_total_latency_s": statistics.mean(latencies) if latencies else None,166            "median_first_token_latency_s": (167                statistics.median(first_token_latencies) if first_token_latencies else None168            ),169            "observations": observations,170        }171 172    return {173        "scenario": SCENARIO,174        "run_label": RUN_LABEL,175        "mode": MODE,176        "result_count": len(RESULTS),177        "cases": summary_cases,178        "result_flags": dict(Counter(str(item["result_flag"]) for item in RESULTS)),179    }180 181 182def finalize(exit_code: int) -> None:183    metadata_payload = {184        "exit_code": exit_code,185        "runtime_context": runtime_context(),186    }187    summary_payload = summarize_results()188    emit("summary", metadata=metadata_payload, summary=summary_payload)189 190    output_dir = _output_dir()191    if not output_dir:192        return193 194    metadata_path = output_dir / "metadata.json"195    results_path = output_dir / "results.json"196    summary_path = output_dir / "summary.json"197    _write_json(metadata_path, metadata_payload)198    _write_json(results_path, RESULTS)199    _write_json(summary_path, summary_payload)200    emit(201        "artifact_paths",202        metadata_path=str(metadata_path),203        results_path=str(results_path),204        summary_path=str(summary_path),205    )206 207 208def main() -> int:209    case_id = os.getenv("PROBE_CASE_ID", f"case-{uuid.uuid4().hex[:8]}")210    emit("banner", context=runtime_context())211    start_case(case_id)212 213    # Replace this block with the narrow runtime question you want to test.214    observation = "replace-me"215    result_flag = "expected"216 217    record_case_result(218        case_id=case_id,219        observation_summary=observation,220        result_flag=result_flag,221    )222    finalize(exit_code=0)223    return 0224 225 226if __name__ == "__main__":227    raise SystemExit(main())228