#!/usr/bin/env python3 """Measure real Codex/Claude CLI completion without storing keys or model output. Run only after configuring your chosen client and route. Every attempt may incur metered usage. This is a basic marker test, not a coding-quality or TTFT test. Python 3.9+; standard library only. """ import argparse from datetime import datetime, timezone import json import math from pathlib import Path import shutil import statistics import subprocess import sys import tempfile import time MARKER = "superfastcode-benchmark-ok" PROMPT = "Do not use tools or inspect any files. Reply with exactly: " + MARKER def utc_now(): return datetime.now(timezone.utc).isoformat() def quantile_nearest_rank(values, fraction): ordered = sorted(values) return ordered[max(0, math.ceil(len(ordered) * fraction) - 1)] if ordered else None def run_attempt(client_path, client, model, timeout): # Separate working directory and final-message file for each attempt. # Start outside project repositories; user-level client settings remain active. with tempfile.TemporaryDirectory(prefix="gateway-marker-") as directory: final_path = Path(directory) / "final-message.txt" if client == "codex": command = [client_path, "exec", "--model", model, "--skip-git-repo-check", "--sandbox", "read-only", "--output-last-message", str(final_path), PROMPT] else: # Keep the positional prompt before variadic tool flags, which would # otherwise consume it as another tool name rather than a prompt. command = [client_path, "--print", PROMPT, "--model", model, "--output-format", "json", "--tools", "", "--disallowedTools", "mcp__*"] started_at = utc_now() started = time.perf_counter() exit_code = None error_kind = None success = False try: completed = subprocess.run(command, cwd=directory, capture_output=True, text=True, timeout=timeout, check=False) exit_code = completed.returncode if exit_code != 0: error_kind = "process_exit" elif client == "codex": final_text = final_path.read_text(encoding="utf-8") if final_path.exists() else "" success = final_text.strip() == MARKER error_kind = None if success else "marker_missing" else: try: payload = json.loads(completed.stdout) if not isinstance(payload, dict) or not isinstance(payload.get("result"), str): error_kind = "invalid_output" else: success = payload.get("is_error") is not True and payload["result"].strip() == MARKER error_kind = None if success else "marker_missing" except (ValueError, TypeError): error_kind = "invalid_output" except subprocess.TimeoutExpired: error_kind = "timeout" except (OSError, UnicodeError): error_kind = "local_io_error" return {"started_at": started_at, "finished_at": utc_now(), "elapsed_ms": round((time.perf_counter() - started) * 1000, 3), "exit_code": exit_code, "success": success, "error_kind": error_kind} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--client", required=True, choices=("codex", "claude")) parser.add_argument("--model", required=True, help="Exact model available to your configured route") parser.add_argument("--route-label", required=True, help="Non-secret label, not a key or URL") parser.add_argument("--runs", type=int, default=5) parser.add_argument("--timeout", type=int, default=120, help="Per-attempt seconds (1–600)") parser.add_argument("--interval", type=float, default=2, help="Seconds between attempts (1–60)") parser.add_argument("--output", required=True, type=Path, help="New JSONL file; existing files are never overwritten") args = parser.parse_args() if not 1 <= args.runs <= 100: parser.error("--runs must be between 1 and 100") if not 1 <= args.timeout <= 600 or not 1 <= args.interval <= 60: parser.error("--timeout must be 1–600 and --interval must be 1–60") if len(args.route_label) > 80 or any(not (c.isalnum() or c in "-_.") for c in args.route_label): parser.error("--route-label accepts up to 80 letters, numbers, hyphens, dots or underscores") client_path = shutil.which(args.client) if client_path is None: parser.error("Client is not installed or is absent from PATH") try: version_result = subprocess.run([client_path, "--version"], capture_output=True, text=True, timeout=15, check=False) if version_result.returncode != 0: parser.error("Client version command failed; check your local installation") version = version_result.stdout.strip()[:200] except (OSError, subprocess.TimeoutExpired): parser.error("Could not read the local client version") records = [] try: output_file = args.output.open("x", encoding="utf-8") except OSError: parser.error("Cannot create result file; use a new path in an existing writable directory") try: with output_file: for index in range(args.runs): record = {"attempt": index + 1, "client": args.client, "client_version": version, "model_requested": args.model, "model_observed": None, "route_label": args.route_label, "metric": "cli_end_to_end_ms", **run_attempt(client_path, args.client, args.model, args.timeout)} records.append(record) output_file.write(json.dumps(record, ensure_ascii=True) + "\n") output_file.flush() outcome = "completed" if record["success"] else record["error_kind"] print(f"{index + 1}/{args.runs}: {outcome}, {record['elapsed_ms']:.1f} ms", flush=True) if index + 1 < args.runs: time.sleep(args.interval) except KeyboardInterrupt: print("Interrupted; already written attempts remain in the result file.", file=sys.stderr) return 130 durations = [row["elapsed_ms"] for row in records if row["success"]] failures = {} for row in records: if not row["success"]: failures[row["error_kind"]] = failures.get(row["error_kind"], 0) + 1 summary = {"attempts": len(records), "marker_successes": len(durations), "success_rate": len(durations) / len(records), "failures": failures, "successful_median_ms": statistics.median(durations) if durations else None, "successful_p95_ms": quantile_nearest_rank(durations, 0.95), "ttft_ms": None, "actual_cost": None, "note": "CLI elapsed includes startup and retries. Marker success is not code quality. No TTFT or cost is inferred."} print(json.dumps(summary, indent=2)) return 0 if len(durations) == len(records) else 1 if __name__ == "__main__": sys.exit(main())