66 lines
3.0 KiB
Python
66 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Compare one configured Claude model on the public synthetic 14-case suite.
|
|
|
|
Run server-side with the planner modules and scoped OAuth mount. This invokes
|
|
the same multi-pass backend without creating an HTTP job. Output is synthetic
|
|
evidence only; progress contains metadata. Never accepts a roster input file.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
sys.path.insert(0, os.environ.get("SUITE_PROBE_MODULE_DIR", "/opt/planner"))
|
|
import suite_backends
|
|
from suite_contract import (EXECUTION_REVISION, MODELS, PROMPT_REVISION, Problem,
|
|
encoded, validate_request, validate_result)
|
|
from suite_multipass import generate, preflight_workflow
|
|
from suite_synthetic import fixture, score
|
|
|
|
|
|
def main():
|
|
"""Run one fixed fixture and emit normalized results plus measured metadata."""
|
|
model = MODELS["claude"]["model"]
|
|
if model not in {"claude-opus-5-5", "claude-sonnet-5-5"}:
|
|
raise SystemExit("Configure one of the two approved comparison models")
|
|
if os.environ.get("SUITE_PROBE_CLAUDE_BIN"):
|
|
suite_backends.CLAUDE_BIN = os.environ["SUITE_PROBE_CLAUDE_BIN"]
|
|
request, expected = fixture(14)
|
|
request["routing"] = {"allow_external": True, "allowed_external_providers": ["claude"]}
|
|
request = validate_request(request, ["claude"])
|
|
selected = preflight_workflow(request)
|
|
started, last_report = time.monotonic(), 0.0
|
|
last_stage = None
|
|
|
|
def progress(value):
|
|
"""Emit stage and elapsed time without model-generated content."""
|
|
nonlocal last_report, last_stage
|
|
stage = value["current_pass"]
|
|
if stage != last_stage or time.monotonic() - last_report >= 20:
|
|
print(json.dumps({"model": model, "stage": stage,
|
|
"elapsed_seconds": value["job_elapsed_seconds"],
|
|
"completed_passes": value["completed_model_passes"]}),
|
|
file=sys.stderr, flush=True)
|
|
last_report, last_stage = time.monotonic(), stage
|
|
|
|
report = {"model": model, "execution_revision": EXECUTION_REVISION,
|
|
"prompt_revision": PROMPT_REVISION, "selection": selected,
|
|
"request_bytes": len(encoded(request)), "case_count": 14,
|
|
"automatic_retries": 0, "test_path": "server_side_multipass_backend"}
|
|
try:
|
|
result, metadata = generate(request, selected, threading.Event(), "192.168.22.8", progress)
|
|
validate_result(result, request)
|
|
report.update(status="completed", result=result, metadata=metadata,
|
|
quality=score(result, expected),
|
|
singletons=sum(len(g["members"]) == 1 for g in result["groups"]))
|
|
except Problem as exc:
|
|
report.update(status="failed", **exc.document())
|
|
report["wall_seconds"] = round(time.monotonic() - started, 3)
|
|
print(json.dumps(report, sort_keys=True), flush=True)
|
|
return 0 if report["status"] == "completed" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|