diff --git a/examples/train-a-router/README.md b/examples/train-a-router/README.md index 1382051a6..99791079c 100644 --- a/examples/train-a-router/README.md +++ b/examples/train-a-router/README.md @@ -14,7 +14,7 @@ You will: 6. **Compare cost estimates:** predict each request’s answer cost, then choose a policy on the full curve. 7. **Route a new question:** call only the selected answer model. -[One-page instructions](QUICKSTART.md) · [Labels and new answer collection](LABELS.md) · [Serve the encoder yourself](self-host/README.md) +[One-page instructions](QUICKSTART.md) · [Labels and new answer collection](LABELS.md) · [Serve the encoder yourself](self-host/README.md) · [Luna low and latency](effort/README.md) ## 1. Understand the data and baseline [Sol 80.8% accuracy; Luna 55.6%] @@ -265,7 +265,21 @@ The Cloud encoder measurement includes network time and saving the raw response The opening table combines these router estimates with the selected answer costs. The explorer retains the original answer-only curves and can add per-request Cloud encoder fees; it does not insert a native hosting estimate into the frozen evaluation. Its unpriced native and CPU costs remain unknown, not zero. -The cost calculations are: +### Answer latency and reasoning effort + +The answer models also contribute latency. All results above use **medium reasoning for both Sol and Luna**. On the 975 final questions, Luna medium was cheaper but slower than Sol. We then collected Luna low on those same questions: + +| Answer model | Accuracy | Answer cost / 10k | Mean answer time | 95th-percentile answer time | +|---|---:|---:|---:|---:| +| Sol, medium | 80.8% | $70.77 | 9.87 s | 27.79 s | +| Luna, medium | 55.6% | $12.23 | 22.01 s | 72.03 s | +| Luna, low | 44.3% | $5.52 | 9.04 s | 19.40 s | + +Luna low cost **54.9% less**, with an **11.3-point accuracy loss** versus medium. For the existing embedding classifier + learned-cost policy at the 80% cap, replacing Luna medium with low changed accuracy **79.9% → 77.2%**, total cost **$58.97 → $58.05 / 10k**, and mean answer time **11.81 → 9.94 seconds**. The router’s decisions and encoder fee stayed fixed. + +These are elapsed API-request times, excluding router inference; time to first token was not measured. Low and medium were collected on different days, so provider load also affects their latency difference. Adding separately measured average router time can estimate the combined mean; adding their 95th percentiles is invalid. The [Luna-low comparison](effort/README.md) includes LoRA, both illustrated embedding spending levels, uncertainty intervals, saved responses and CPU replay commands. + +### Cost calculation - **Sol and Luna:** sum each model's token cost only for questions sent to that model, then multiply by `10,000 / 975`. Input, cached-input and cache-write prices follow the saved usage; output tokens already include reasoning tokens, so reasoning is charged once. Multiplying a routing share by the model's overall mean cost would be wrong: routers select different questions, with different answer lengths. - **Qwen embeddings:** the recorded Cloud rate is **$0.06 per million input tokens**, rounded up per request to a whole **$0.00001 credit**: `ceil(3 × input_tokens / 500)` credits. The 975 test questions total 1,235,875 encoder tokens and **7,910 credits after rounding**: `$0.07910 × 10,000 / 975 = $0.811282`. Both neighbors and classifiers need **one embedding for every incoming request**, whichever answer model is selected. The rate and rounding rule reproduce all 91 saved singleton charges and 245 successful batch charges. The table prices separate live requests, even when the exercise reuses cached vectors; it does not spread the original batch-collection bill across future requests. diff --git a/examples/train-a-router/effort/README.md b/examples/train-a-router/effort/README.md new file mode 100644 index 000000000..4cdc3d775 --- /dev/null +++ b/examples/train-a-router/effort/README.md @@ -0,0 +1,105 @@ +# Lower Luna’s reasoning effort + +**Luna low cost 54.9% less than Luna medium, with an 11.3-point accuracy loss.** Its mean answer time was 9.0 seconds versus 22.0 seconds. We compared all **975 BBEH test questions**, reusing the existing medium recordings; the runs happened on different days, so provider load also affects the latency comparison. + +Sol stays at medium. The existing Laya, LoRA and embedding routers keep their original decisions, so this measures what happens when you change Luna’s reasoning setting without retraining the router. + +## Results + +Costs below are estimates per 10,000 requests. Accuracy loss and savings are relative to always using Sol. Mean and 95th-percentile times cover the selected answer request, before adding any router inference. + +| Answer model | Accuracy | Accuracy loss (points) | Answer cost | Savings | Mean time | 95th-percentile time | +|---|---:|---:|---:|---:|---:|---:| +| Sol, medium | 80.8% | 0.0 | $70.77 | 0.0% | 9.87 s | 27.79 s | +| Luna, medium | 55.6% | 25.2 | $12.23 | 82.7% | 22.01 s | 72.03 s | +| Luna, low | 44.3% | 36.5 | $5.52 | 92.2% | 9.04 s | 19.40 s | + +Luna low answered **432 questions correctly**, versus **542** at medium. The paired 95% interval for the accuracy loss is **8.2 to 14.4 points**. Average reasoning tokens fell from **2,126 to 783**. All 975 low responses completed with valid answer formats; there were no retries or truncated low answers. + +Changing Luna also changes the answers delivered by the routers: + +| Frozen policy | Luna effort | Accuracy | Accuracy loss (points) | Total cost | Savings | Mean answer time | 95th-percentile answer time | +|---|---|---:|---:|---:|---:|---:|---:| +| Laya + LoRA, 80% cap | medium | 78.3% | 2.6 | $57.85 | 18.2% | 12.45 s | 32.62 s | +| Laya + LoRA, 80% cap | low | 75.3% | 5.5 | $56.80 | 19.7% | 10.28 s | 27.52 s | +| LoRA + learned cost, 80% cap | medium | 79.0% | 1.8 | $58.68 | 17.1% | 12.08 s | 32.36 s | +| LoRA + learned cost, 80% cap | low | 77.1% | 3.7 | $57.79 | 18.3% | 10.25 s | 27.61 s | +| Embedding classifier + learned cost, 80% cap | medium | 79.9% | 0.9 | $58.97 | 16.7% | 11.81 s | 33.99 s | +| Embedding classifier + learned cost, 80% cap | low | 77.2% | 3.6 | $58.05 | 18.0% | 9.94 s | 27.13 s | +| Embedding classifier + learned cost, 60% cap | medium | 74.8% | 6.1 | $42.70 | 39.7% | 14.32 s | 44.65 s | +| Embedding classifier + learned cost, 60% cap | low | 68.8% | 12.0 | $40.19 | 43.2% | 9.49 s | 23.73 s | + +The caps refer to the original calibration answer-cost budgets. These are the same policies illustrated in the main example. Their decisions and Sol shares stay fixed; we do not choose new thresholds from these test answers. All nine comparisons, including unchanged Laya, neighbors and text controls, are in the [full report](https://huggingface.co/datasets/superlinked/train-a-router-bbeh/blob/main/effort/2026-10-26/luna-low/report.md). + +Total cost adds the original **$1.05 / 10k busy-H100 estimate for LoRA**, or **$0.81 / 10k SIE Cloud encoder fee** for embedding routers. Application hosting and idle capacity are extra. The router cost is unchanged between medium and low. [The main example](../README.md) explains the cost breakdown and serving measurements. + +For the embedding policy at the 80% cap, low saves another **$0.92 / 10k requests** and reduces observed mean answer time by **1.88 seconds**, while losing another **2.7 accuracy points**. Sol still supplies 79.2% of answers, so cutting Luna’s bill has a small effect on the total. Adding the separately measured encoder/classifier mean gives an estimated **12.02 → 10.14 seconds** for router plus answer; this is not a measured end-to-end service time. Percentiles cannot be added this way. + +[Download the responses and analysis](https://huggingface.co/datasets/superlinked/train-a-router-bbeh/tree/main/effort/2026-10-26/luna-low), or replay them below. + +## Replay the results on CPU + +No API key or GPU is needed. From a clone of this repository: + +```bash +# Enter the example and install the locked analysis dependencies. +cd examples/train-a-router +uv sync --frozen --group collection --group outcomes +``` + +```bash +# Download the original questions, reference answers and frozen routing policies. +uv run --frozen python router.py fetch --output data --group core +``` + +```bash +# Download and verify the Luna-low results, then unpack their saved API responses. +uv run --frozen python router.py fetch --release effort/release.json --output luna-low-evidence +tar -xzf luna-low-evidence/evidence.tar.gz -C luna-low-evidence +``` + +```bash +# Grade all 975 answers and compare the original policies with Luna low substituted. +uv run --frozen --group collection --group outcomes python -m effort.analyze \ + --data data --collection luna-low-evidence/collection --output luna-low-report +``` + +Open `luna-low-report/report.md`. `report.json` also contains results by task, paired uncertainty intervals, token counts, missing-value coverage and all nine policy comparisons. `graded-rows.json` contains the question-level measurements. Use a new report directory for each replay; the analyzer preserves previous reports. + +## Collect fresh Luna-low answers on CPU + +Use the same downloaded `data` directory. This makes paid OpenAI calls with **up to 16 requests in flight**. It calls only `gpt-6-luna`, with `reasoning.effort=low`; it reuses the existing Luna-medium and Sol-medium recordings for comparison. + +```bash +# Configure the OpenAI account that will pay for fresh Luna-low answers. +export OPENAI_API_KEY='your-openai-api-key' +``` + +```bash +# Collect Luna-low answers to all 975 questions into a fresh directory. +uv run --frozen --group collection python -m effort.collect_low \ + --inputs data/source/question-only/final-inputs.json \ + --manifest data/manifest.json --source-plan data/source/source-plan.json \ + --output my-luna-low --budget-usd 20.00 --deadline-seconds 5400 --execute +``` + +The budget is a ceiling, not a target. Our complete run used **$0.54 in estimated API usage**, plus worker hosting, and collection took **13.5 minutes on a Modal CPU worker**. Omitting `--execute` only saves the request plan. Repeating the exact command resumes that plan; completed answers and requests with uncertain execution are not sent again. Invalid or truncated answers are kept. Only explicit transient failures can be retried. + +```bash +# Grade your new answers using the same references and routing decisions. +uv run --frozen --group collection --group outcomes python -m effort.analyze \ + --data data --collection my-luna-low --output my-luna-low-report +``` + +## How to read the comparison + +- **Accuracy:** correct answers divided by all 975 questions, including invalid or truncated responses. Grading uses the original reference answers and normalization rules. +- **Cost:** input and output tokens at the study’s standard API prices, including cached-input accounting. Reasoning tokens are already part of output tokens and are charged once. Router costs are identified separately. +- **Answer time:** elapsed time for the selected API attempt, including client/network work. It excludes router inference, collection queue time and retry backoff. The 95th percentile is the time within which 95% of these answers finished; time to first token was not measured. +- **Uncertainty:** paired 95% intervals resample related question families together within each task. They describe this question mix, not a guarantee for new kinds of work. + +Low and medium answers were collected on different days. Their latency difference includes changes in provider load and serving conditions as well as reasoning effort. This is a comparison of saved request times, not a simultaneous load test. + +The routers learned from medium-effort answers and costs. Their route choices stay fixed here; no policy was selected using the new low-effort test answers. Training a router specifically for Luna low would require new training and calibration answers, followed by a fresh test. The main example’s medium-based results remain unchanged. + +[Return to the full example](../README.md) diff --git a/examples/train-a-router/effort/analyze.py b/examples/train-a-router/effort/analyze.py new file mode 100644 index 000000000..e0298fd74 --- /dev/null +++ b/examples/train-a-router/effort/analyze.py @@ -0,0 +1,588 @@ +"""Post-publication Luna-low comparison and unchanged medium-trained route replay. + +From the example root: uv sync --frozen --group collection --group outcomes +Then: uv run --frozen python -m effort.analyze --data DATA --collection RUN --output REPORT +This command has no network, model-fitting, threshold-selection or inference path. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +from collections import Counter +from decimal import Decimal +from pathlib import Path + +import numpy as np + +import collect +import evidence +import grade_v2 +import normalize +from effort import collect_low +from evaluation import frontier, utility_router + +ROOT = Path(__file__).resolve().parents[1] +N = 975 +RELEASE_REVISION = "86dd7494a4c2396bc9ffa890f994b9722610b357" +MANIFEST_SHA256 = "4fd8a49d760384ad745c1bd49b3ee61cb58fd8d98c71cef6552ff6dbfda66b55" +OFFICIAL_SHA256 = "4b4f06e5babb015de2ba639bae995a5526182ffb5b5890af54dfcf038580eb34" +METHODS = ( + "base", "lora", "knn", "embedding", "lora__textcost", "embedding__textcost", + "text_length__textcost", "text_topic__textcost", +) +ARMS = ("luna_low", "luna_medium", "sol_medium") +PROTOCOL = { + "study": "Post-publication comparison on all 975 existing final questions; not a new blind test.", + "change": "New gpt-6-luna low reasoning; saved Luna-medium and Sol-medium answers are reused.", + "grading": "Unchanged source-only normalize.py followed by grade_v2.decimal_match and pinned BBEH scorer.", + "selection": "No fitting or threshold selection. Exact saved Fraction rules and input scores are replayed.", + "policies": "Original README eight 80%-calibration-answer-cost-cap rules plus embedding__textcost at 60%.", + "transfer": "Routers and expected-cost estimates were trained for medium effort, not low effort.", + "denominator": "All eligible IDs, including invalid, truncated and unavailable answers; none dropped.", + "bootstrap": "2000 paired whole-family draws within each task, seed 89; fixed observed task weights.", + "uncertainty": "Pointwise intervals conditional on these tasks; no simultaneous or unseen-task guarantee.", + "latency": "Selected physical attempt wall time, not a concurrently measured routed service or TTFT.", + "confounding": "Low and medium were collected on different days; latency differences also reflect service/load changes.", + "cost": "Token-accounted standard list-price estimate, not invoice; reasoning is included in output tokens.", + "failures": "Unavailable answers count operationally incorrect; unknown usage/cost stays unknown, never zero-imputed.", + "router_cost": "Saved singleton encoder fees unchanged; native and small CPU hosting remain unpriced here.", +} + + +def require(value, message): + if not value: + raise ValueError(message) + + +def sha(path): + return evidence.sha(path) + + +def read(path): + return json.loads(Path(path).read_bytes()) + + +def write(path, value): + Path(path).write_text(json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False, allow_nan=False) + "\n", + encoding="utf-8") + + +def bound(root, name, expected): + path = evidence.artifact_path(root, name) + require(path.is_file() and sha(path) == expected, f"Changed or missing artifact: {name}") + return path + + +def checked_data(directory): + """Verify just the needed immutable public files; no weight or feature loading.""" + root = Path(directory) + manifest = evidence.checked_manifest((root / "manifest.json").read_bytes(), MANIFEST_SHA256) + entries = {item["path"]: item for item in manifest["files"]} + names = ( + "source/source-plan.json", "source/meta.json", "source/gold.json", + "source/question-only/final-inputs.json", "source/provenance/bbeh/evaluate.py", "pair.json", + "grades/grades-final.json", "reference/frozen-frontiers.json", "reference/final-report.json", + "metadata/presentation-policies.json", + ) + hashes = {} + for name in names: + entry = entries[name] + path = bound(root, name, entry["sha256"]) + require(path.stat().st_size == entry["size"], f"Published size changed: {name}") + hashes[name] = entry["sha256"] + source, pair = read(root / names[0]), read(root / "pair.json") + inputs = read(root / "source/question-only/final-inputs.json") + meta, gold = read(root / "source/meta.json"), read(root / "source/gold.json") + by_meta, by_gold = ({row["id"]: row for row in records} for records in (meta, gold)) + require(len(by_meta) == len(meta) and len(by_gold) == len(gold), "Duplicate source/reference ID") + for name in ("meta.json", "gold.json", "question-only/final-inputs.json"): + require(source["files_sha256"][name] == hashes["source/" + name], "Source file binding changed") + ids = [row["id"] for row in inputs] + require(len(ids) == N and len(set(ids)) == N, "Use the complete ordered 975-question population") + rows = [by_meta[identity] for identity in ids] + require({r["id"] for r in meta if r["retained"] and r["role"] == "final"} == set(ids), + "Final source population changed") + for item, row in zip(inputs, rows, strict=True): + require(row["retained"] is True and row["role"] == "final" + and row["request_sha256"] == hashlib.sha256(item["request"].encode()).hexdigest(), + "Inactive, cross-role or changed question") + collect.body_for(item, "luna") # Validate the original exact three-field input contract. + require(isinstance(by_gold[item["id"]]["target"], str), "Non-string reference") + grades = read(root / "grades/grades-final.json") + frozen, report = (read(root / f"reference/{name}.json") for name in ("frozen-frontiers", "final-report")) + choices = read(root / "metadata/presentation-policies.json") + source_sha, pair_sha = hashes["source/source-plan.json"], hashes["pair.json"] + require(pair["models"] == {"cheap": "gpt-6-luna", "strong": "gpt-6.1-sol"} + and pair["settings"] == collect.OPENAI_SETTINGS and pair["system_prompt"] == collect.SYSTEM + and pair["prices_usd_per_million"] == collect.PRICES, "Original answer recipe changed") + require(grades["format"] == "bbeh-paired-grades-v1" and grades["role"] == "final" + and grades["n"] == N and [r["id"] for r in grades["rows"]] == ids + and grades["models"] == pair["models"] and grades["source_plan_sha256"] == source_sha + and grades["pair_sha256"] == pair_sha and grades["grader_sha256"] == sha(grade_v2.__file__) + and grades["normalizer_sha256"] == sha(normalize.__file__) + and grades["official_grader_sha256"] == OFFICIAL_SHA256 + and grades["normalization_version"] == normalize.VERSION, "Original grading lineage changed") + require(frozen["format"] == "bbeh-frozen-frontiers-v1" and frozen["ids"] == ids + and frozen["frozen_before_final_outcomes"] is True and frozen["final_outcomes_accessed"] is False + and frozen["source_plan_sha256"] == source_sha and frozen["pair_sha256"] == pair_sha + and frozen["plan_sha256"] == entries["reference/embedding/plan.json"]["sha256"] + and frozen["implementation_sha256"]["utility"] == sha(utility_router.__file__), + "Original routing lineage changed") + require(report["frozen_sha256"] == hashes["reference/frozen-frontiers.json"] + and report["grades_sha256"] == hashes["grades/grades-final.json"] + and report["n"] == N, "Original report binding changed") + require(choices["format"] == "bbeh-presentation-policies-v1" and choices["final_answers_collected"] is False + and choices["frozen_sha256"] == report["frozen_sha256"] + and choices["calibration_sha256"] == frozen["calibration_sha256"], "Presentation rules changed") + for row, old, saved in zip(rows, grades["rows"], frozen["source_rows"], strict=True): + require(all(old[key] == saved[key] == row[key] for key in ("id", "task", "family", "role")), + "Source/family identity changed") + official = grade_v2.load_official(root / "source/provenance/bbeh/evaluate.py", OFFICIAL_SHA256) + return {"inputs": inputs, "rows": rows, "gold": by_gold, "grades": grades, "frozen": frozen, + "report": report, "choices": choices, "hashes": hashes, "official": official, "pair": pair} + + +def checked_plan(root, original): + plan = read(root / "plan.json") + inputs, hashes = original["inputs"], original["hashes"] + require(plan["format"] == "bbeh-luna-low-acquisition-v1" and plan["protocol"] == collect_low.PROTOCOL + and plan["role"] == "final" and plan["n"] == len(inputs) + and plan["ordered_ids"] == [r["id"] for r in inputs] + and plan["inputs_sha256"] == hashes["source/question-only/final-inputs.json"] + and plan["inputs_canonical_sha256"] == collect_low.digest(inputs) + and plan["manifest_sha256"] == MANIFEST_SHA256 + and plan["source_plan_sha256"] == hashes["source/source-plan.json"] + and plan["pair_sha256"] == hashes["pair.json"], "Low acquisition population/lineage changed") + release = read(ROOT / "release.json") + require(plan["release"] == release and plan["release_sha256"] == sha(ROOT / "release.json") + and release["manifest_sha256"] == MANIFEST_SHA256 + and f"/{RELEASE_REVISION}/" in release["manifest_url"], "Public release changed") + require(plan["prices_usd_per_million"] == collect.PRICES["gpt-6-luna"] + and plan["only_request_change"] == {"reasoning.effort": {"from": "medium", "to": "low"}} + and plan["no_quality_selection"] is True and plan["new_routing_fit"] is False, + "Low-effort comparison specification changed") + sources = ("effort/collect_low.py", "collect.py", "openai_population.py", "reader_io.py", "evidence.py") + require(set(plan["source_files_sha256"]) == set(sources), "Unknown collector source closure") + for name in sources: + require(plan["source_files_sha256"][name] == sha(ROOT / name), f"Collector source changed: {name}") + require(read(root / "inputs.json") == inputs, "Collected question bytes changed") + require(plan["request_body_sha256"] == {r["id"]: collect_low.digest(collect_low.body_for(r)) for r in inputs} + and plan["original_medium_body_sha256"] == { + r["id"]: collect_low.digest(collect.body_for(r, "luna")) for r in inputs}, "Request body changed") + return plan + + +def checked_attempts(root, plan, inputs): + """Reconcile the complete physical ledger, including attempts without usage.""" + root = Path(root) + summary, selection = read(root / "collection.json"), read(root / "selection.json") + plan_sha, ids = sha(root / "plan.json"), [r["id"] for r in inputs] + require(summary["format"] == "bbeh-luna-low-collection-v1" + and summary["plan_sha256"] == plan_sha and summary["selection_sha256"] == sha(root / "selection.json"), + "Collection/selection binding changed") + require(selection["format"] == "bbeh-luna-low-selection-v1" and selection["plan_sha256"] == plan_sha + and selection["ordered_ids"] == ids and [r["id"] for r in selection["entries"]] == ids + and selection["role"] == "final" and selection["arm"] == "luna" + and selection["model"] == "gpt-6-luna" and selection["reasoning_effort"] == "low" + and selection["first_terminal_response_only"] is True and selection["no_quality_retries"] is True, + "Selection contract changed") + chosen, ledger, declared = {}, [], set() + totals = [Decimal(0), Decimal(0), Decimal(0)] + for item, state in zip(inputs, selection["entries"], strict=True): + body, first, previous = collect_low.body_for(item), None, None + reserve = collect.reservation(body, "luna") + require(len(state["attempts"]) <= plan["protocol"]["max_attempts"], "Physical attempt cap exceeded") + for number, attempt in enumerate(state["attempts"], 1): + require(first is None and (previous is None or previous == "retry"), + "Terminal or uncertain response was retried") + prefix = f"attempts/{hashlib.sha256(item['id'].encode()).hexdigest()}/{number:03d}" + require(attempt["attempt"] == number and attempt["intent_path"] == prefix + "-intent.json", + "Attempt ID/order changed") + intent = read(bound(root, attempt["intent_path"], attempt["intent_sha256"])) + declared.add(attempt["intent_path"]) + identity = {"id": item["id"], "arm": "luna", "model": "gpt-6-luna", "reasoning_effort": "low", + "body_sha256": collect_low.digest(body)} + require(all(intent.get(k) == v for k, v in { + **identity, "attempt": number, "plan_sha256": plan_sha, "reserved_usd": str(reserve), + }.items()), "Intent/source binding changed") + if attempt["record_path"] is None: + require(attempt["record_sha256"] is None and not (root / (prefix + "-response.json")).exists(), + "Undeclared response") + response = {**identity, "status": "interrupted_unknown", "execution_unknown": True, + "terminal_response": False, "created_unix": intent["created_unix"], + "elapsed_seconds": None} + else: + require(attempt["record_path"] == prefix + "-response.json", "Response ID/path changed") + response = read(bound(root, attempt["record_path"], attempt["record_sha256"])) + declared.add(attempt["record_path"]) + require(all(response.get(k) == v for k, v in { + **identity, "attempt": number, "intent_sha256": attempt["intent_sha256"], + }.items()) and response.get("binding_valid", True) is True, "Response identity changed") + elapsed = response.get("elapsed_seconds") + require(elapsed is None or type(elapsed) in (int, float) and math.isfinite(elapsed) and elapsed >= 0, + "Invalid physical elapsed time") + amount, budget, unknown = collect_low.accounting(response, reserve, body) + previous = collect_low.decision(response, number) + if budget > reserve: + previous = "fatal" + require(attempt["action"] == previous and attempt["status"] == response["status"], + "Physical attempt decision changed") + expected = (str(amount) if not unknown else None, str(budget) if not unknown else None, str(unknown)) + require(tuple(attempt[k] for k in ("known_list_usd", "known_budget_usd", "unknown_reservation_usd")) + == expected, "Physical accounting changed") + totals = [a + b for a, b in zip(totals, (amount, budget, unknown), strict=True)] + ledger.append({"id": item["id"], **attempt, "elapsed_seconds": elapsed, + "http_status": response.get("http_status"), "error_type": response.get("error_type"), + "error_code": response.get("error_code"), "request_not_sent": response.get("request_not_sent", False)}) + if response.get("terminal_response"): + require(response.get("http_status") == 200 and previous == "terminal", + "Provider/model/effort/tier mismatch: cannot compare this response") + first = (number, response, attempt) + require(state["selected_attempt"] == (first[0] if first else None), "Not the first terminal response") + if first: + expected_state = "complete" if first[1]["status"] == "complete" else "terminal" + require(state["status"] == expected_state, "Selected receipt status changed") + chosen[item["id"]] = {"response": first[1] if first else None, + "attempt": first[2] if first else None, "state": state, + "unknown_bound_usd": str(sum((Decimal(a["unknown_reservation_usd"]) + for a in state["attempts"]), Decimal(0)))} + actual = {p.relative_to(root).as_posix() for p in (root / "attempts").rglob("*") if p.is_file()} + require(actual == declared, "Missing or undeclared physical attempt files") + states = selection["entries"] + require(summary["expected_responses"] == len(ids) and summary["physical_attempts"] == len(ledger) + and summary["selected_responses"] == sum(x["selected_attempt"] is not None for x in states) + and summary["complete_receipts"] == sum(x["status"] == "complete" for x in states) + and summary["status_counts"] == dict(Counter(x["status"] for x in states)), "Coverage totals changed") + for key, value in zip(("known_list_usd", "known_budget_usd", "unknown_reservation_usd"), totals, strict=True): + require(Decimal(summary[key]) == value, "Acquisition accounting totals changed") + require(Decimal(summary["cost_upper_bound_usd"]) == totals[1] + totals[2] + and totals[1] + totals[2] <= Decimal(plan["budget_usd"]), "Acquisition budget changed") + require(summary["status"] == ("complete" if summary["complete_receipts"] == len(ids) + and summary["stop"] is None else "incomplete"), "Completion status changed") + return chosen, ledger, summary + + +def grade_low(item, meta, target, selected, official): + response = selected["response"] or {"status": selected["state"]["status"]} + text = "".join(p.get("text") or "" for p in response.get("messages", []) if p["type"] == "output_text") + answer = collect.parse_answer(text, item["response_contract"]) if response.get("provider_status") == "completed" else None + if response.get("status") == "complete": + require(response.get("answer") == answer and response.get("valid_answer") == (answer is not None), + "Saved final-answer parse changed") + valid = response.get("status") == "complete" and answer is not None + canonical, rule = normalize.canonicalize(answer, item["request"], meta["task"]) + cost = response.get("cost") + amount = cost["usd_estimate"] if cost else None + if amount is not None: + upper = amount + elif selected["state"]["attempts"]: + upper = selected["unknown_bound_usd"] + else: + upper = None # A never-dispatched answer has no observed deployment-cost bound. + return { + "correct": int(valid and canonical is not None and grade_v2.decimal_match(official, canonical, target)), + "official_raw_correct": int(valid and official.evaluate_correctness(answer, target)), + "literal_correct": int(valid and answer.strip() == target.strip()), + "normalized_answer": canonical, "normalization_rule": rule, "valid": valid, + "status": response["status"], "provider_status": response.get("provider_status"), + "truncated": response.get("provider_status") == "incomplete", "usage": response.get("usage"), + "cost_usd": amount, "cost_upper_bound_usd": upper, "elapsed_seconds": response.get("elapsed_seconds"), + "record_sha256": selected["attempt"]["record_sha256"] if selected["attempt"] else None, + "available_answer": selected["response"] is not None and response.get("status") == "complete", + } + + +def original_arm(row, prefix, item): + values = {key.removeprefix(prefix + "_"): value for key, value in row.items() if key.startswith(prefix + "_")} + for name in ("correct", "official_raw_correct", "literal_correct"): + require(type(values[name]) is int and values[name] in (0, 1), "Nonbinary original correctness") + body = collect.body_for(item, "luna" if prefix == "cheap" else "sol") + cost = collect.openai_cost(values["usage"], body) + require(Decimal(str(values["cost_usd"])) == Decimal(cost["usd_estimate"]), "Original token cost mismatch") + return {**values, "cost_usd": cost["usd_estimate"], "cost_upper_bound_usd": cost["usd_estimate"], + "available_answer": values["status"] == "complete"} + + +def numbers(values): + known = [float(v) for v in values if v is not None] + require(all(math.isfinite(v) and v >= 0 for v in known), "Invalid nonnegative measurement") + return {"observed_n": len(known), "missing_n": len(values) - len(known), + "mean": float(np.mean(known)) if known else None, + "p50": float(np.quantile(known, .5)) if known else None, + "p95": float(np.quantile(known, .95)) if known else None, + "known_sum": float(sum(known))} + + +def usage_values(arms, name): + paths = { + "input_tokens": ("input_tokens",), "output_tokens": ("output_tokens",), + "reasoning_tokens": ("output_tokens_details", "reasoning_tokens"), + "cached_input_tokens": ("input_tokens_details", "cached_tokens"), + "cache_write_tokens": ("input_tokens_details", "cache_write_tokens"), + } + result = [] + for arm in arms: + usage = arm.get("usage") + value = usage + for key in paths[name]: + value = value.get(key) if isinstance(value, dict) else None + require(value is None or type(value) is int and value >= 0, "Invalid token measurement") + result.append(value) + return result + + +def summarize(arms, bootstrap=None): + n = len(arms) + correct = np.array([r["correct"] for r in arms], dtype=float) + costs = [r["cost_usd"] for r in arms] + known = sum((Decimal(str(x)) for x in costs if x is not None), Decimal(0)) + bounded = all(r.get("cost_upper_bound_usd") is not None for r in arms) + upper = sum((Decimal(str(r["cost_upper_bound_usd"])) for r in arms), Decimal(0)) if bounded else None + full = all(x is not None for x in costs) + result = { + "n": n, "correct": int(correct.sum()), "accuracy_percent": 100 * float(correct.mean()), + "official_raw_accuracy_percent": 100 * sum(r["official_raw_correct"] for r in arms) / n, + "literal_accuracy_percent": 100 * sum(r["literal_correct"] for r in arms) / n, + "unavailable_n": sum(not r["available_answer"] for r in arms), + "invalid_n": sum(not r["valid"] for r in arms), "invalid_percent": 100 * sum(not r["valid"] for r in arms) / n, + "truncated_n": sum(r["truncated"] for r in arms), + "truncated_percent": 100 * sum(r["truncated"] for r in arms) / n, + "status_counts": dict(Counter(r["status"] for r in arms)), + "provider_status_counts": dict(Counter(str(r["provider_status"]) for r in arms)), + "cost_status": "exact_selected_usage" if full else "incomplete_selected_usage", + "answer_usd_per_10000": float(known * 10000 / n) if full else None, + "known_selected_cost_usd": str(known), "cost_missing_n": sum(x is None for x in costs), + "selected_cost_bounds_usd": {"lower": str(known), "upper": str(upper) if upper is not None else None}, + "selected_elapsed_seconds": numbers([r.get("elapsed_seconds") for r in arms]), + "tokens": {name: numbers(usage_values(arms, name)) for name in ( + "input_tokens", "output_tokens", "reasoning_tokens", "cached_input_tokens", "cache_write_tokens")}, + } + if bootstrap is not None: + result["ci95"] = {"accuracy_percent": frontier.interval(100 * bootstrap.means(correct[:, None])[:, 0])} + for key, values, scale in ( + ("answer_usd_per_10000", costs, 10000), + ("selected_elapsed_mean_seconds", [r.get("elapsed_seconds") for r in arms], 1), + ("output_tokens_mean", usage_values(arms, "output_tokens"), 1), + ("reasoning_tokens_mean", usage_values(arms, "reasoning_tokens"), 1), + ): + result["ci95"][key] = mean_interval(bootstrap, values, scale) + return result + + +def mean_interval(bootstrap, values, scale=1): + if any(v is None for v in values): + return None + array = np.asarray(values, dtype=float) + return frontier.interval(scale * bootstrap.means(array[:, None])[:, 0]) + + +def compare(first, second, bootstrap): + """Paired differences use the same complete population and bootstrap draw.""" + y0, y1 = (np.array([r["correct"] for r in rows], dtype=float) for rows in (first, second)) + result = { + "n": len(first), "both_correct": int(np.sum((y0 == 1) & (y1 == 1))), + "both_wrong": int(np.sum((y0 == 0) & (y1 == 0))), + "first_only_correct": int(np.sum((y0 == 1) & (y1 == 0))), + "second_only_correct": int(np.sum((y0 == 0) & (y1 == 1))), + "accuracy_difference_pp": 100 * float(np.mean(y0 - y1)), + "accuracy_difference_pp_ci95": mean_interval(bootstrap, y0 - y1, 100), + "direction": "first minus second; negative cost/time difference means first is lower/faster", + } + for name, field, scale in (("answer_usd_per_10000", "cost_usd", 10000), + ("selected_elapsed_mean_seconds", "elapsed_seconds", 1)): + a, b = ([r.get(field) for r in rows] for rows in (first, second)) + if any(v is None for v in a + b): + result[name + "_difference"] = result[name + "_difference_ci95"] = None + result[name + "_reduction_percent"] = result[name + "_reduction_percent_ci95"] = None + else: + matrix = np.column_stack((np.asarray(a, dtype=float), np.asarray(b, dtype=float))) + means = bootstrap.means(matrix) + result[name + "_difference"] = scale * float(np.mean(matrix[:, 0] - matrix[:, 1])) + result[name + "_difference_ci95"] = frontier.interval(scale * (means[:, 0] - means[:, 1])) + result[name + "_reduction_percent"] = 100 * (1 - float(matrix[:, 0].mean() / matrix[:, 1].mean())) \ + if matrix[:, 1].mean() else None + with np.errstate(divide="ignore", invalid="ignore"): + result[name + "_reduction_percent_ci95"] = frontier.interval(100 * (1 - means[:, 0] / means[:, 1])) + return result + + +def route_metrics(arms, routes, fees): + value = summarize(arms) + value["strong_share"] = float(np.mean(routes)) + value["router_usd_per_10000"] = None if fees is None else 10000 * float(np.mean(fees)) + value["total_usd_per_10000"] = None if fees is None or value["answer_usd_per_10000"] is None else ( + value["answer_usd_per_10000"] + value["router_usd_per_10000"]) + return value + + +def replay_policies(original, rows, bootstrap): + frozen, choices, published = original["frozen"], original["choices"], original["report"] + definitions = [(name, "0.8") for name in METHODS] + [("embedding__textcost", "0.6")] + result = [] + for method, cap in definitions: + selected, saved = choices["choices"][method][cap], frozen["methods"][method] + candidate = next(r for r in saved["candidates"] if r["candidate_id"] == selected["candidate_id"]) + require(candidate["policy"] == selected["policy"] and candidate["calibration"] == selected["calibration"] + and candidate["candidate_id"] in saved["calibration_frontier_candidate_ids"], + "Presentation candidate differs from saved calibration rule") + scores, rule = saved["scores"], candidate["policy"] + require(scores["metadata_assisted"] is False, "Metadata-only policy is not a README deployable rule") + routes = utility_router.replay(rule, scores["probabilities"], scores["expected_usd"]) + require(len(routes) == len(rows), "Route population changed") + original_arms = [row["sol_medium" if use_sol else "luna_medium"] for row, use_sol in zip(rows, routes, strict=True)] + changed_arms = [row["sol_medium" if use_sol else "luna_low"] for row, use_sol in zip(rows, routes, strict=True)] + fees = frozen["encoder_fees_usd"] if scores["router_kind"] == "embedding" else None + if rule["mode"].startswith("always_"): + fees = [0.] * len(rows) + old, new = (route_metrics(a, routes, fees) for a in (original_arms, changed_arms)) + for value, arm_rows in ((old, original_arms), (new, changed_arms)): + correct = np.array([r["correct"] for r in arm_rows], dtype=float) + sol_correct = np.array([r["sol_medium"]["correct"] for r in rows], dtype=float) + value["quality_gap_vs_sol_pp"] = 100 * float(np.mean(sol_correct - correct)) + value["accuracy_percent_ci95"] = mean_interval(bootstrap, correct, 100) + value["quality_gap_vs_sol_pp_ci95"] = mean_interval(bootstrap, sol_correct - correct, 100) + prior = next(r for r in published["methods"][method]["points"] if r["candidate_id"] == candidate["candidate_id"]) + require(prior["rule"] == rule, "Published policy rule differs") + errors = {} + for key, current in (("answer_accuracy_percent", old["accuracy_percent"]), + ("answer_usd_per_10000", old["answer_usd_per_10000"]), + ("strong_share", old["strong_share"]), + ("router_usd_per_10000", old["router_usd_per_10000"]), + ("total_usd_per_10000", old["total_usd_per_10000"])): + expected = prior[key] + require((current is None) == (expected is None), "Original priced/unpriced scope changed") + errors[key] = None if current is None else abs(current - expected) + require(current is None or math.isclose(current, expected, rel_tol=0, abs_tol=1e-10), + f"Original replay parity failed: {method}/{key}") + comparison = compare(changed_arms, original_arms, bootstrap) + result.append({"method": method, "calibration_answer_cap_fraction": float(cap), + "candidate_id": candidate["candidate_id"], "rule": rule, + "routes_sha256": collect_low.digest(routes.astype(int).tolist()), + "strong_n": int(routes.sum()), "cheap_n": int((~routes).sum()), + "original_medium": old, "substituted_low": new, + "low_minus_medium": comparison, "original_replay_absolute_errors": errors, + "router_kind": scores["router_kind"], "same_routes_and_router_overhead": True, + "latency_excludes_router_and_retries": True}) + return result + + +def markdown(report): + def fmt(value, digits=3): + return "unknown" if value is None else f"{value:.{digits}f}" + + lines = ["# Luna reasoning-effort comparison", "", PROTOCOL["study"], "", PROTOCOL["confounding"], "", + (f"Collection status: **{report['new_acquisition']['status']}**; " + f"{report['new_acquisition']['selected_responses']}/{report['n']} selected responses; " + f"{report['arms']['luna_low']['unavailable_n']} unavailable answers counted operationally incorrect."), "", + "Accuracy uses all questions. Costs are selected-answer list-price estimates; times are selected-attempt wall time.", + "", "| Answer model | Accuracy % | USD / 10k | Mean s | p50 s | p95 s | Invalid / truncated |", + "|---|---:|---:|---:|---:|---:|---:|"] + for name, value in report["arms"].items(): + timing = value["selected_elapsed_seconds"] + lines.append(f"| {name} | {fmt(value['accuracy_percent'])} | {fmt(value['answer_usd_per_10000'])} | " + f"{fmt(timing['mean'])} | {fmt(timing['p50'])} | {fmt(timing['p95'])} | " + f"{value['invalid_n']} / {value['truncated_n']} |") + delta = report["comparisons"]["luna_low_vs_luna_medium"] + ci = delta["accuracy_difference_pp_ci95"] + lines += ["", (f"Luna low minus medium accuracy: {fmt(delta['accuracy_difference_pp'])} pp " + f"(paired 95% interval {fmt(ci['lower'])} to {fmt(ci['upper'])})."), + "", "## Fixed-route transfer", "", PROTOCOL["transfer"], "", + ("The original calibration rules and all route masks stay unchanged. Native/CPU hosting is unpriced; " + "embedding fees stay unchanged. No operating point is selected on these new answers."), "", + "| Method / calibration cap | Medium → low accuracy % | Medium → low answer USD / 10k | Medium → low mean s | Sol share % |", + "|---|---:|---:|---:|---:|"] + for point in report["policies"]: + a, b = point["original_medium"], point["substituted_low"] + lines.append(f"| {point['method']} / {point['calibration_answer_cap_fraction']:.0%} | " + f"{fmt(a['accuracy_percent'])} → {fmt(b['accuracy_percent'])} | " + f"{fmt(a['answer_usd_per_10000'])} → {fmt(b['answer_usd_per_10000'])} | " + f"{fmt(a['selected_elapsed_seconds']['mean'])} → {fmt(b['selected_elapsed_seconds']['mean'])} | " + f"{100 * a['strong_share']:.2f} |") + ledger = report["new_acquisition"] + lines += ["", (f"Acquisition: {ledger['physical_attempts']} physical attempts; known usage-priced subtotal " + f"${ledger['known_list_usd']}; additional unknown-attempt reservation bound " + f"${ledger['unknown_reservation_usd']}. This is separate from selected-answer deployment estimates."), + "", PROTOCOL["uncertainty"], "", ("Medians and p95s are descriptive. Missing values and their coverage " + "are explicit in report.json; observed-only timing summaries do not remove rows from accuracy."), + "", "No new model fitting, threshold calibration, or fresh blind-test claim is made.", ""] + return "\n".join(lines) + + +def analyze(data, collection, output, *, draws=2000): + require(type(draws) is int and draws > 0, "Invalid bootstrap draw count") + output = Path(output) + require(not output.exists(), "Preserve previous reports; use a new output directory") + original = checked_data(data) + root = Path(collection) + plan = checked_plan(root, original) + selected, ledger, acquisition = checked_attempts(root, plan, original["inputs"]) + rows = [] + for item, meta, old in zip(original["inputs"], original["rows"], original["grades"]["rows"], strict=True): + row = {key: meta[key] for key in ("id", "task", "family", "role")} + row["luna_low"] = grade_low(item, meta, original["gold"][item["id"]]["target"], + selected[item["id"]], original["official"]) + row["luna_medium"], row["sol_medium"] = original_arm(old, "cheap", item), original_arm(old, "strong", item) + rows.append(row) + bootstrap = frontier.Bootstrap(rows, draws) + arms = {name: [row[name] for row in rows] for name in ARMS} + by_task = {} + for task in sorted({r["task"] for r in rows}): + part = [r for r in rows if r["task"] == task] + task_bootstrap = frontier.Bootstrap(part, draws) + by_task[task] = {"n": len(part), "families": len({r["family"] for r in part}), + "arms": {name: summarize([r[name] for r in part]) for name in ARMS}, + "low_minus_medium": compare([r["luna_low"] for r in part], + [r["luna_medium"] for r in part], task_bootstrap)} + policies = replay_policies(original, rows, bootstrap) + report = { + "format": "bbeh-luna-effort-comparison-v1", "protocol": PROTOCOL, "bootstrap_draws": draws, + "n": len(rows), "families": len({r["family"] for r in rows}), "tasks": len(by_task), + "arms": {name: summarize(values, bootstrap) for name, values in arms.items()}, + "comparisons": {f"{first}_vs_{second}": compare(arms[first], arms[second], bootstrap) + for first, second in (("luna_low", "luna_medium"), ("luna_low", "sol_medium"), + ("luna_medium", "sol_medium"))}, + "by_task": by_task, "policies": policies, "new_acquisition": acquisition, + "physical_attempt_errors": dict(Counter(f"{r['http_status']}:{r['error_type']}:{r['error_code']}" + for r in ledger if r["action"] != "terminal")), + "physical_attempt_elapsed_seconds": numbers([r["elapsed_seconds"] for r in ledger]), + "original_acquisition_ledger": original["grades"]["acquisition_ledger"], + "original_replay_verified": True, "new_fit": False, "new_policy_selection": False, + "provenance": {"release_revision": RELEASE_REVISION, "manifest_sha256": MANIFEST_SHA256, + "original_files_sha256": original["hashes"], "collection_sha256": sha(root / "collection.json"), + "plan_sha256": sha(root / "plan.json"), "selection_sha256": sha(root / "selection.json"), + "analyzer_sha256": sha(__file__), "collector_source_sha256": plan["source_files_sha256"], + "normalizer_sha256": sha(normalize.__file__), "grade_v2_sha256": sha(grade_v2.__file__), + "official_sha256": OFFICIAL_SHA256, "utility_sha256": sha(utility_router.__file__), + "bootstrap_sha256": sha(frontier.__file__), "numpy_version": np.__version__, + "original_acquisition_protocol_sha256": original["grades"]["acquisition_protocol_sha256"], + "low_attempt_created_unix_range": [min((r["response"]["created_unix"] for r in selected.values() + if r["response"]), default=None), + max((r["response"]["created_unix"] for r in selected.values() + if r["response"]), default=None)]}, + } + output.mkdir(parents=True) + write(output / "graded-rows.json", {"format": "bbeh-luna-effort-rows-v1", "rows": rows, + "physical_attempts": ledger, "provenance": report["provenance"]}) + report["graded_rows_sha256"] = sha(output / "graded-rows.json") + write(output / "report.json", report) + (output / "report.md").write_text(markdown(report), encoding="utf-8") + return report + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data", required=True, type=Path) + parser.add_argument("--collection", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + args = parser.parse_args() + result = analyze(args.data, args.collection, args.output) + print(json.dumps({"n": result["n"], "original_replay_verified": result["original_replay_verified"], + "collection_status": result["new_acquisition"]["status"], + "report_sha256": sha(args.output / "report.json")})) + + +if __name__ == "__main__": + main() diff --git a/examples/train-a-router/effort/collect_low.py b/examples/train-a-router/effort/collect_low.py new file mode 100644 index 000000000..fcc2ede26 --- /dev/null +++ b/examples/train-a-router/effort/collect_low.py @@ -0,0 +1,446 @@ +"""Collect Luna-low on the published final questions without repeating uncertain calls. + +Run from the example root with ``python -m effort.collect_low --help``. +Only --execute makes requests; an unchanged output directory is resumable. +""" + +import argparse +from collections import Counter, deque +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait +from decimal import Decimal +import hashlib +import json +import os +from pathlib import Path +import time + +import httpx + +import collect +import evidence +import openai_population as original +from reader_io import exclusive_run, read, require, sha, sync_files + +ROOT = Path(collect.__file__).resolve().parent +MODEL = "gpt-6-luna" +ENDPOINT = "https://api.openai.com/v1/responses" +RELEASE_REVISION = "86dd7494a4c2396bc9ffa890f994b9722610b357" +MANIFEST_SHA256 = "4fd8a49d760384ad745c1bd49b3ee61cb58fd8d98c71cef6552ff6dbfda66b55" +INPUT_MEMBER = "source/question-only/final-inputs.json" +SOURCE_MEMBER = "source/source-plan.json" +PROTOCOL = { + "model": MODEL, "reasoning_effort": "low", "max_output_tokens": 16384, + "service_tier": "default", "store": False, "endpoint": ENDPOINT, + "workers": 16, "max_attempts": 3, "request_timeout_seconds": 600, + "connect_timeout_seconds": 30, "backoff_seconds": [5, 15], "retry_after_cap_seconds": 60, + "first_http_200_only": True, "quality_retries": False, "uncertain_execution_retries": False, + "reference_answers_on_input": False, "original_medium_or_sol_requests": False, + "timing": "Per-attempt wall time including client creation, HTTP and parsing; excluding checkpointing/backoff", +} +PRE_EXECUTION_ERRORS = {"ConnectTimeout", "ConnectError", "PoolTimeout"} + + +def source_hashes(): + return {"effort/collect_low.py": sha(__file__), "collect.py": sha(collect.__file__), + "openai_population.py": sha(original.__file__), "reader_io.py": sha(ROOT / "reader_io.py"), + "evidence.py": sha(evidence.__file__)} + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() + + +def body_for(item): + body = collect.body_for(item, "luna") + require(body["model"] == MODEL and body["reasoning"] == {"effort": "medium"} + and body["max_output_tokens"] == 16384 and body["service_tier"] == "default", + "The original answer-model protocol changed") + return {**body, "reasoning": {"effort": "low"}} + + +def prepare(inputs_path, manifest_path, source_plan_path, *, release_path=None, pair_path=None, + budget_usd="20.00", deadline_seconds=5400): + """Verify only public question inputs and metadata, never references or prior answers.""" + release_path, pair_path = Path(release_path or ROOT / "release.json"), Path(pair_path or ROOT / "pair.json") + release, pair, source = read(release_path), read(pair_path), read(source_plan_path) + require(release["format"] == "train-a-router-release-v1" + and release["manifest_sha256"] == MANIFEST_SHA256 + and release["manifest_url"] == f"https://huggingface.co/datasets/superlinked/train-a-router-bbeh/resolve/{RELEASE_REVISION}/manifest.json", + "Use the original immutable public BBEH release") + manifest = evidence.checked_manifest(Path(manifest_path).read_bytes(), MANIFEST_SHA256) + entries = {row["path"]: row for row in manifest["files"]} + for member, path in ((INPUT_MEMBER, inputs_path), (SOURCE_MEMBER, source_plan_path)): + require(member in entries and sha(path) == entries[member]["sha256"] + and Path(path).stat().st_size == entries[member]["size"], "Published input/source binding changed") + require(sha(inputs_path) == source["files_sha256"]["question-only/final-inputs.json"], "Final inputs changed") + require(pair["models"] == {"cheap": MODEL, "strong": "gpt-6.1-sol"} + and pair["settings"] == collect.OPENAI_SETTINGS and pair["system_prompt"] == collect.SYSTEM + and pair["prices_usd_per_million"] == collect.PRICES, "Original pair recipe changed") + items = read(inputs_path) + require(isinstance(items, list) and len(items) == 975, "Use all 975 published final questions") + ids = [item["id"] for item in items] + require(all(isinstance(value, str) and value for value in ids) and len(set(ids)) == len(ids), "Invalid/duplicate ID") + bodies = {item["id"]: digest(body_for(item)) for item in items} + cap = Decimal(str(budget_usd)) + require(cap.is_finite() and 0 < cap <= 20, "Budget must be positive and no more than USD20") + require(type(deadline_seconds) is int and 600 < deadline_seconds <= 5400, "Deadline must be601..5400seconds") + plan = { + "format": "bbeh-luna-low-acquisition-v1", "role": "final", "n": len(items), + "protocol": PROTOCOL, "ordered_ids": ids, "inputs_sha256": sha(inputs_path), + "inputs_canonical_sha256": digest(items), "request_body_sha256": bodies, + "original_medium_body_sha256": {item["id"]: digest(collect.body_for(item, "luna")) for item in items}, + "release_sha256": sha(release_path), "release": release, "manifest_sha256": sha(manifest_path), + "source_plan_sha256": sha(source_plan_path), "pair_sha256": sha(pair_path), + "prices_usd_per_million": collect.PRICES[MODEL], "budget_usd": str(cap), + "deadline_seconds": deadline_seconds, "deadline_scope": "Wall clock from first execute, including resumes", + "only_request_change": {"reasoning.effort": {"from": "medium", "to": "low"}}, + "source_files_sha256": source_hashes(), + "no_quality_selection": True, "new_routing_fit": False, + } + return items, plan + + +def acquire(item, client_factory=None): + """One physical request, readable response fields only; no automatic HTTP retry.""" + body = body_for(item) + row = {"id": item["id"], "arm": "luna", "model": MODEL, "reasoning_effort": "low", + "body_sha256": digest(body), "created_unix": time.time(), "terminal_response": False, + "execution_unknown": True, "request_not_sent": False} + started = time.monotonic() + try: + factory = client_factory or (lambda: httpx.Client(timeout=httpx.Timeout(600, connect=30), + follow_redirects=False, transport=httpx.HTTPTransport(retries=0), trust_env=False)) + with factory() as client: + response = client.post(ENDPOINT, json=body, headers={ + "Authorization": "Bearer " + os.environ["OPENAI_API_KEY"], "Content-Type": "application/json"}) + row.update(http_status=response.status_code, + retry_after_seconds=original.retry_after(response.headers.get("retry-after"))) + try: + payload = response.json() + except ValueError: + payload = None + if response.status_code != 200: + error = payload.get("error") if isinstance(payload, dict) else None + error = error if isinstance(error, dict) else {} + row.update(status="provider_error", execution_unknown=False, + error_code=original.safe_code(error.get("code")), error_type=original.safe_code(error.get("type"))) + return row + row["terminal_response"] = True + require(isinstance(payload, dict), "Invalid provider JSON object") + reasoning = payload.get("reasoning") + row.update(provider_status=payload.get("status"), resolved_model=payload.get("model"), + service_tier=payload.get("service_tier"), + resolved_reasoning_effort=reasoning.get("effort") if isinstance(reasoning, dict) else None) + messages = [] + for output in payload.get("output", []): + if output.get("type") == "message": + for part in output.get("content", []): + if part.get("type") in {"output_text", "refusal"}: + key = "text" if part["type"] == "output_text" else "refusal" + require(isinstance(part.get(key), str), "Invalid readable output content") + messages.append({"type": part["type"], key: part[key]}) + row["messages"] = messages + usage = payload.get("usage") + require(isinstance(usage, dict) and original.numeric_usage(usage), "Usage must be a readable numeric object") + row["usage"] = usage + row["cost"] = collect.openai_cost(usage, body) + row["execution_unknown"] = False + details = payload.get("incomplete_details") + row["incomplete_reason"] = original.safe_code(details.get("reason")) if isinstance(details, dict) else None + text = "".join(part["text"] for part in messages if part["type"] == "output_text") + row["answer"] = collect.parse_answer(text, item["response_contract"]) if row["provider_status"] == "completed" else None + row.update(status="complete", valid_answer=row["answer"] is not None) + if Decimal(row["cost"]["budget_usd"]) > collect.reservation(body, "luna"): + row["status"] = "charge_exceeds_reservation" + except Exception as error: + row.update(status="receipt_incomplete" if row["terminal_response"] else "failed_or_unknown", + error_type=type(error).__name__) + if type(error).__name__ in PRE_EXECUTION_ERRORS: + row.update(request_not_sent=True, execution_unknown=False) + finally: + row["elapsed_seconds"] = time.monotonic() - started + return row + + +def decision(row, attempt): + if (row.get("http_status") in {401, 403} or row.get("error_code") in original.FATAL_CODES + or row.get("error_type") in original.FATAL_CODES or row.get("binding_valid") is False + or row.get("status") == "charge_exceeds_reservation"): + return "fatal" + if row.get("terminal_response"): + if (row.get("resolved_model") != MODEL or row.get("service_tier") != "default" + or row.get("resolved_reasoning_effort") not in (None, "low")): + return "fatal" + return "terminal" + if row.get("execution_unknown", True): + return "uncertain" + transient = (row.get("http_status") in original.TRANSIENT_HTTP + or row.get("request_not_sent") is True and row.get("error_type") in PRE_EXECUTION_ERRORS) + if not transient: + return "incomplete" + if attempt >= PROTOCOL["max_attempts"]: + return "exhausted" + if (row.get("retry_after_seconds") or 0) > PROTOCOL["retry_after_cap_seconds"]: + return "deferred" + return "retry" + + +def accounting(row, reserve, body): + """Retain a bound for every unpriced physical attempt; never call missing usage free.""" + if row.get("cost") is not None: + expected = collect.openai_cost(row["usage"], body) + require(row["cost"] == expected, "Stored usage/cost mismatch") + return Decimal(expected["usd_estimate"]), Decimal(expected["budget_usd"]), Decimal(0) + if row.get("request_not_sent") is True: + require(row.get("error_type") in PRE_EXECUTION_ERRORS and not row.get("terminal_response") + and not row.get("http_status"), "Cannot classify a dispatched request as free") + return Decimal(0), Decimal(0), Decimal(0) + return Decimal(0), Decimal(0), reserve + + +def run(items, output, plan, commit=lambda: None, *, call=acquire, emit=True): + """Rolling16 requests; callback persists Volume writes before dispatch and after receipts.""" + with exclusive_run(output): + return _run(items, Path(output).resolve(), plan, commit, call, emit) + + +def _run(items, root, plan, commit, call, emit): + require(plan["format"] == "bbeh-luna-low-acquisition-v1" and plan["protocol"] == PROTOCOL, + "Changed low-effort protocol") + require(plan["ordered_ids"] == [row["id"] for row in items] and plan["n"] == len(items) + and len({row["id"] for row in items}) == len(items) + and plan["inputs_canonical_sha256"] == digest(items), "Changed fixed population") + require(plan["source_files_sha256"] == source_hashes(), "Collector source changed") + cap = Decimal(plan["budget_usd"]) + require(cap.is_finite() and 0 < cap <= 20 and 600 < plan["deadline_seconds"] <= 5400, "Invalid budget/deadline") + bodies = {row["id"]: body_for(row) for row in items} + require(plan["request_body_sha256"] == {key: digest(body) for key, body in bodies.items()}, "Request bodies changed") + dirty = set() + + def write(path, value, *, immutable=False): + path = root / path + require(not path.is_symlink() and path.resolve().is_relative_to(root), "Output path escapes collection") + if immutable and path.exists(): + require(path.read_bytes() == original.encoded(value), "Cannot change prior immutable evidence") + return + original.dump(path, value) + dirty.add(path) + + def checkpoint(): + if dirty: + sync_files(dirty, root) + commit() + dirty.clear() + + if not (root / "plan.json").exists(): + require(not any(path.name != ".collector.lock" for path in root.iterdir()), "Use an empty directory or exact prior plan") + write("plan.json", plan, immutable=True) + write("inputs.json", items, immutable=True) + plan_hash = sha(root / "plan.json") + start_path = root / "execution.json" + if start_path.exists(): + execution = read(start_path) + require(execution["plan_sha256"] == plan_hash, "Execution plan changed") + else: + execution = {"plan_sha256": plan_hash, "started_unix": time.time(), "deadline_seconds": plan["deadline_seconds"]} + write("execution.json", execution, immutable=True) + require(execution["deadline_seconds"] == plan["deadline_seconds"], "Execution deadline changed") + checkpoint() + if (root / "selection.json").exists(): + previous = read(root / "selection.json") + require(previous["plan_sha256"] == plan_hash and previous["ordered_ids"] == plan["ordered_ids"], "Prior selection changed") + for state in previous["entries"]: + for entry in state["attempts"]: + for field in ("intent", "record"): + if entry[field + "_path"] is not None: + path = root / entry[field + "_path"] + require(path.resolve().is_relative_to(root) and not path.is_symlink() + and sha(path) == entry[field + "_sha256"], "Prior attempt bytes changed") + states, item_map, queue = {}, {row["id"]: row for row in items}, deque() + known_list, known_budget, unknown = Decimal(0), Decimal(0), Decimal(0) + expected_files, stop = set(), None + + def record(state, intent_path, response_path, row, number, reserve): + nonlocal known_list, known_budget, unknown, stop + body = bodies[state["id"]] + matched = all(row.get(k) == value for k, value in { + "id": state["id"], "arm": "luna", "model": MODEL, "reasoning_effort": "low", "body_sha256": digest(body)}.items()) + require(row.get("binding_valid", matched) == matched, "Receipt binding flag changed") + amount, budget, bound = accounting(row, reserve, body) + known_list += amount + known_budget += budget + unknown += bound + action = decision({**row, "binding_valid": matched}, number) + if budget > reserve: + action = "fatal" + entry = {"attempt": number, "intent_path": intent_path.relative_to(root).as_posix(), + "intent_sha256": sha(intent_path), "record_path": response_path.relative_to(root).as_posix() if response_path.exists() else None, + "record_sha256": sha(response_path) if response_path.exists() else None, + "status": row["status"], "action": action, "known_list_usd": str(amount) if not bound else None, + "known_budget_usd": str(budget) if not bound else None, "unknown_reservation_usd": str(bound)} + state["attempts"].append(entry) + if row.get("terminal_response"): + state["selected_attempt"] = number + state["status"] = "complete" if action == "terminal" and row["status"] == "complete" else action + if action == "fatal": + stop = "provider_auth_quota_or_identity_error" + elif action == "deferred": + stop = "retry_after_exceeds_supported_cap" + if action == "retry": + delay = max(PROTOCOL["backoff_seconds"][number - 1], row.get("retry_after_seconds") or 0) + state["not_before_unix"] = row["created_unix"] + row["elapsed_seconds"] + delay + return action + + for item in items: + identity = item["id"] + state = {"id": identity, "status": "pending", "selected_attempt": None, "attempts": [], "not_before_unix": 0} + states[identity] = state + directory = root / "attempts" / hashlib.sha256(identity.encode()).hexdigest() + intents = sorted(directory.glob("*-intent.json")) + require(len(intents) <= PROTOCOL["max_attempts"], "Physical attempt cap exceeded") + for number, path in enumerate(intents, 1): + require(path.name == f"{number:03d}-intent.json" and not path.is_symlink(), "Invalid attempt sequence") + require(not state["attempts"] or state["attempts"][-1]["action"] == "retry", "Unauthorized repeat after terminal/uncertain attempt") + intent = read(path) + reserve = collect.reservation(bodies[identity], "luna") + require(all(intent.get(k) == v for k, v in { + "id": identity, "arm": "luna", "attempt": number, "model": MODEL, "reasoning_effort": "low", + "body_sha256": digest(bodies[identity]), "plan_sha256": plan_hash, "reserved_usd": str(reserve)}.items()), "Prior intent changed") + response = path.with_name(f"{number:03d}-response.json") + expected_files.add(path.relative_to(root).as_posix()) + if response.exists(): + require(not response.is_symlink(), "Response must be a regular file") + expected_files.add(response.relative_to(root).as_posix()) + row = read(response) + require(row.get("intent_sha256") == sha(path) and row.get("attempt") == number, "Response intent changed") + else: + row = {"id": identity, "arm": "luna", "model": MODEL, "reasoning_effort": "low", + "body_sha256": digest(bodies[identity]), "status": "interrupted_unknown", "execution_unknown": True, + "terminal_response": False, "created_unix": intent["created_unix"], "elapsed_seconds": None} + action = record(state, path, response, row, number, reserve) + if not intents or action == "retry": + queue.append(identity) + actual_files = {path.relative_to(root).as_posix() for path in (root / "attempts").rglob("*") if path.is_file()} + require(actual_files == expected_files, "Unaccounted attempt/response files") + require(known_budget + unknown <= cap, "Preserved cost exposure exceeds budget") + pending, completed_since_progress = {}, 0 + pause_until = 0 + dispatch_deadline = execution["started_unix"] + plan["deadline_seconds"] - PROTOCOL["request_timeout_seconds"] + with ThreadPoolExecutor(max_workers=PROTOCOL["workers"]) as pool: + while queue or pending: + if time.time() >= dispatch_deadline: + stop = stop or "dispatch_deadline" + candidates = [] + reserved = sum((entry[3] for entry in pending.values()), Decimal(0)) + for _ in range(len(queue)): + if stop or time.time() < pause_until or len(pending) + len(candidates) >= PROTOCOL["workers"]: + break + identity = queue.popleft() + state = states[identity] + reserve = collect.reservation(bodies[identity], "luna") + if time.time() < state["not_before_unix"] or known_budget + unknown + reserved + reserve > cap: + queue.append(identity) + continue + number = len(state["attempts"]) + 1 + directory = Path("attempts") / hashlib.sha256(identity.encode()).hexdigest() + path, response = directory / f"{number:03d}-intent.json", directory / f"{number:03d}-response.json" + require(number <= PROTOCOL["max_attempts"] and state["selected_attempt"] is None + and not (root / path).exists() and not (root / response).exists(), "Never overwrite/repeat a physical request") + write(path, {"id": identity, "arm": "luna", "model": MODEL, "reasoning_effort": "low", + "attempt": number, "body_sha256": digest(bodies[identity]), "plan_sha256": plan_hash, + "reserved_usd": str(reserve), "created_unix": time.time()}, immutable=True) + candidates.append((identity, path, response, reserve, number)) + reserved += reserve + if candidates: + checkpoint() # Bounded group of durable intents, before any request in the group. + for identity, path, response, reserve, number in candidates: + pending[pool.submit(call, item_map[identity])] = (identity, path, response, reserve, number) + if not pending: + if stop or not queue: + break + if not any(known_budget + unknown + collect.reservation(bodies[key], "luna") <= cap for key in queue): + stop = "budget_bound" + break + time.sleep(0.1) + continue + done, _ = wait(pending, timeout=0.5, return_when=FIRST_COMPLETED) + for future in done: + identity, path, response, reserve, number = pending.pop(future) + try: + row = future.result() + except Exception as error: + row = {"id": identity, "arm": "luna", "model": MODEL, "reasoning_effort": "low", + "body_sha256": digest(bodies[identity]), "status": "failed_or_unknown", "execution_unknown": True, + "terminal_response": False, "error_type": type(error).__name__, "created_unix": time.time(), "elapsed_seconds": None} + row.update(intent_sha256=sha(root / path), attempt=number) + write(response, row, immutable=True) + action = record(states[identity], root / path, root / response, row, number, reserve) + if action == "retry": + queue.append(identity) + if row.get("http_status") in {429, 503}: + pause_until = max(pause_until, states[identity]["not_before_unix"]) + completed_since_progress += 1 + if done: + checkpoint() + if emit and completed_since_progress >= 32: + completed_since_progress = 0 + progress = {"selected": sum(state["selected_attempt"] is not None for state in states.values()), + "expected": len(items), "known_list_usd": str(known_list), "unknown_reservation_usd": str(unknown), + "inflight": len(pending), "stop": stop} + write("progress.json", progress) + checkpoint() + print(json.dumps(progress), flush=True) + for state in states.values(): + if state["status"] in {"pending", "retry"}: + state["status"] = "incomplete_" + (stop or "unavailable") + selection = {"format": "bbeh-luna-low-selection-v1", "role": "final", "arm": "luna", "model": MODEL, + "reasoning_effort": "low", "plan_sha256": plan_hash, "ordered_ids": plan["ordered_ids"], + "entries": list(states.values()), "first_terminal_response_only": True, "no_quality_retries": True} + write("selection.json", selection) + complete = sum(state["status"] == "complete" for state in states.values()) + summary = {"format": "bbeh-luna-low-collection-v1", "status": "complete" if complete == len(items) and stop is None else "incomplete", + "expected_responses": len(items), "complete_receipts": complete, + "selected_responses": sum(state["selected_attempt"] is not None for state in states.values()), + "physical_attempts": sum(len(state["attempts"]) for state in states.values()), + "status_counts": dict(Counter(state["status"] for state in states.values())), + "known_list_usd": str(known_list), "known_budget_usd": str(known_budget), "unknown_reservation_usd": str(unknown), + "cost_upper_bound_usd": str(known_budget + unknown), "budget_usd": plan["budget_usd"], + "stop": stop, "plan_sha256": plan_hash, "selection_sha256": sha(root / "selection.json"), + "elapsed_seconds": time.time() - execution["started_unix"], "correctness_inspected": False} + write("collection.json", summary) + checkpoint() + return summary + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--inputs", type=Path, required=True) + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--source-plan", type=Path, required=True) + parser.add_argument("--release", type=Path) + parser.add_argument("--pair", type=Path) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--budget-usd", default="20.00") + parser.add_argument("--deadline-seconds", type=int, default=5400) + parser.add_argument("--execute", action="store_true") + args = parser.parse_args() + items, plan = prepare(args.inputs, args.manifest, args.source_plan, release_path=args.release, + pair_path=args.pair, budget_usd=args.budget_usd, deadline_seconds=args.deadline_seconds) + if args.execute: + require(bool(os.environ.get("OPENAI_API_KEY")), "Set OPENAI_API_KEY before --execute") + result = run(items, args.output, plan) + else: + with exclusive_run(args.output): + path = args.output / "plan.json" + if path.exists(): + require(path.read_bytes() == original.encoded(plan), "Cannot replace an existing plan") + else: + require(not any(p.name != ".collector.lock" for p in args.output.iterdir()), "Use an empty plan directory") + original.dump(path, plan) + sync_files([path], args.output) + result = {"prepared_only": True, "n": len(items), "plan_sha256": sha(path), "paid_calls": 0} + print(json.dumps(result, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/examples/train-a-router/effort/release.json b/examples/train-a-router/effort/release.json new file mode 100644 index 000000000..efb9cf370 --- /dev/null +++ b/examples/train-a-router/effort/release.json @@ -0,0 +1,7 @@ +{ + "dataset": "superlinked/train-a-router-bbeh", + "format": "train-a-router-release-v1", + "manifest_sha256": "e406935743a147e9c5e4044bac64fed8d03a7ae529010b3f32bf502e1584d51d", + "manifest_url": "https://huggingface.co/datasets/superlinked/train-a-router-bbeh/resolve/ff3d9b76029620095fb63e7f04df10f38ce8a2b4/effort/2026-10-26/luna-low/manifest.json", + "revision": "ff3d9b76029620095fb63e7f04df10f38ce8a2b4" +} diff --git a/examples/train-a-router/test_effort_analysis.py b/examples/train-a-router/test_effort_analysis.py new file mode 100644 index 000000000..7e1c0552e --- /dev/null +++ b/examples/train-a-router/test_effort_analysis.py @@ -0,0 +1,239 @@ +"""Small offline checks for effort grading, accounting and exact fixed-route replay.""" + +import hashlib +import tempfile +import unittest +from copy import deepcopy +from decimal import Decimal +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import numpy as np + +import collect +from effort import analyze, collect_low +from evaluation import frontier, utility_router + + +def item(identity="x", request="Return exactly yes."): + return {"id": identity, "request": request, "response_contract": {"type": "exact_string", "max_length": 2048}} + + +def response(question, answer="yes", *, truncated=False): + body = collect_low.body_for(question) + usage = {"input_tokens": 100, "output_tokens": 25, + "input_tokens_details": {"cached_tokens": 10, "cache_write_tokens": 20}, + "output_tokens_details": {"reasoning_tokens": 20}, "total_tokens": 125} + return {"id": question["id"], "arm": "luna", "model": "gpt-6-luna", "reasoning_effort": "low", + "body_sha256": collect_low.digest(body), "status": "complete", "http_status": 200, + "created_unix": 1000., "terminal_response": True, "execution_unknown": False, + "resolved_model": "gpt-6-luna", "service_tier": "default", "resolved_reasoning_effort": "low", + "provider_status": "incomplete" if truncated else "completed", "elapsed_seconds": 1.25, + "usage": usage, "cost": collect.openai_cost(usage, body), + "messages": [{"type": "output_text", "text": analyze.json.dumps({"answer": answer})}], + "answer": None if truncated else answer, "valid_answer": not truncated} + + +def collection(root, questions, attempts): + """Produce explicit immutable physical receipts, never run a transport/scheduler.""" + plan = {"protocol": collect_low.PROTOCOL, "budget_usd": "20.00"} + analyze.write(root / "plan.json", plan) + plan_sha, entries = analyze.sha(root / "plan.json"), [] + amounts = [Decimal(0), Decimal(0), Decimal(0)] + for question, records in zip(questions, attempts, strict=True): + body = collect_low.body_for(question) + reserve = collect.reservation(body, "luna") + state = {"id": question["id"], "status": "uncertain", "selected_attempt": None, "attempts": []} + for number, saved in enumerate(records, 1): + prefix = f"attempts/{hashlib.sha256(question['id'].encode()).hexdigest()}/{number:03d}" + (root / prefix).parent.mkdir(parents=True, exist_ok=True) + intent = {"id": question["id"], "arm": "luna", "model": "gpt-6-luna", "reasoning_effort": "low", + "body_sha256": collect_low.digest(body), "plan_sha256": plan_sha, "attempt": number, + "reserved_usd": str(reserve), "created_unix": 1000.} + analyze.write(root / (prefix + "-intent.json"), intent) + intent_sha = analyze.sha(root / (prefix + "-intent.json")) + row = deepcopy(saved) if saved is not None else { + "status": "interrupted_unknown", "execution_unknown": True, "terminal_response": False} + row.update(intent_sha256=intent_sha, attempt=number) + if saved is not None: + analyze.write(root / (prefix + "-response.json"), row) + amount, budget, unknown = collect_low.accounting(row, reserve, body) + amounts = [a + b for a, b in zip(amounts, (amount, budget, unknown), strict=True)] + action = collect_low.decision(row, number) + receipt = {"attempt": number, "intent_path": prefix + "-intent.json", "intent_sha256": intent_sha, + "record_path": prefix + "-response.json" if saved is not None else None, + "record_sha256": analyze.sha(root / (prefix + "-response.json")) if saved is not None else None, + "status": row["status"], "action": action, + "known_list_usd": None if unknown else str(amount), + "known_budget_usd": None if unknown else str(budget), "unknown_reservation_usd": str(unknown)} + state["attempts"].append(receipt) + state["status"] = "complete" if action == "terminal" and row["status"] == "complete" else action + if row.get("terminal_response"): + state["selected_attempt"] = number + entries.append(state) + selection = {"format": "bbeh-luna-low-selection-v1", "plan_sha256": plan_sha, "role": "final", "arm": "luna", + "model": "gpt-6-luna", "reasoning_effort": "low", "ordered_ids": [q["id"] for q in questions], + "entries": entries, "first_terminal_response_only": True, "no_quality_retries": True} + analyze.write(root / "selection.json", selection) + complete = sum(s["status"] == "complete" for s in entries) + summary = {"format": "bbeh-luna-low-collection-v1", "plan_sha256": plan_sha, + "selection_sha256": analyze.sha(root / "selection.json"), "expected_responses": len(entries), + "physical_attempts": sum(len(s["attempts"]) for s in entries), + "selected_responses": sum(s["selected_attempt"] is not None for s in entries), + "complete_receipts": complete, "status_counts": dict(analyze.Counter(s["status"] for s in entries)), + "known_list_usd": str(amounts[0]), "known_budget_usd": str(amounts[1]), + "unknown_reservation_usd": str(amounts[2]), "cost_upper_bound_usd": str(amounts[1] + amounts[2]), + "stop": None, "status": "complete" if complete == len(entries) else "incomplete"} + analyze.write(root / "collection.json", summary) + return plan + + +OFFICIAL = SimpleNamespace(preprocess_sample=str.strip, preprocess_reference=str.strip, + evaluate_correctness=lambda a, b: a.strip() == b.strip()) + + +def metric(correct, cost="0.01", elapsed=1.): + return {"correct": correct, "official_raw_correct": correct, "literal_correct": correct, + "valid": True, "truncated": False, "status": "complete", "provider_status": "completed", + "available_answer": True, "cost_usd": cost, "cost_upper_bound_usd": cost, + "elapsed_seconds": elapsed, "usage": {"input_tokens": 10, "output_tokens": 6, + "input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 4}}} + + +class EffortAnalysisTests(unittest.TestCase): + def test_valid_truncated_and_missing_are_retained_with_cost_bounds(self): + questions = [item("a"), item("b"), item("c")] + with tempfile.TemporaryDirectory() as name: + root = Path(name) + plan = collection(root, questions, [[response(questions[0])], [response(questions[1], truncated=True)], [None]]) + chosen, ledger, summary = analyze.checked_attempts(root, plan, questions) + graded = [analyze.grade_low(q, {"task": "test"}, "yes", chosen[q["id"]], OFFICIAL) for q in questions] + self.assertEqual([r["correct"] for r in graded], [1, 0, 0]) + self.assertEqual(len(ledger), 3) + self.assertEqual(summary["complete_receipts"], 2) # Includes provider-truncated terminal receipt. + self.assertTrue(graded[1]["truncated"]) + result = analyze.summarize(graded) + self.assertEqual(result["n"], 3) + self.assertIsNone(result["answer_usd_per_10000"]) + self.assertEqual(result["cost_missing_n"], 1) + self.assertGreater(Decimal(result["selected_cost_bounds_usd"]["upper"]), + Decimal(result["known_selected_cost_usd"])) + self.assertEqual(result["selected_elapsed_seconds"]["missing_n"], 1) + + def test_paid_retry_ledger_is_separate_from_selected_answer(self): + question = item() + failed = {**response(question), "http_status": 503, "terminal_response": False, + "status": "provider_error", "execution_unknown": False, "cost": None, "usage": None} + with tempfile.TemporaryDirectory() as name: + root = Path(name) + plan = collection(root, [question], [[failed, response(question)]]) + chosen, ledger, summary = analyze.checked_attempts(root, plan, [question]) + graded = analyze.grade_low(question, {"task": "test"}, "yes", chosen["x"], OFFICIAL) + self.assertEqual(len(ledger), 2) + self.assertEqual(graded["cost_usd"], response(question)["cost"]["usd_estimate"]) + self.assertGreater(Decimal(summary["unknown_reservation_usd"]), 0) + self.assertEqual(summary["status"], "complete") + + def test_hash_tampering_and_terminal_retry_fail_closed(self): + question = item() + with tempfile.TemporaryDirectory() as name: + root = Path(name) + plan = collection(root, [question], [[response(question)]]) + path = next((root / "attempts").rglob("*-response.json")) + path.write_text("{}", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "Changed or missing artifact"): + analyze.checked_attempts(root, plan, [question]) + with tempfile.TemporaryDirectory() as name: + root = Path(name) + plan = collection(root, [question], [[response(question), response(question)]]) + with self.assertRaisesRegex(ValueError, "retried"): + analyze.checked_attempts(root, plan, [question]) + + def test_normalization_and_decimal_do_not_use_binary_float_equivalence(self): + q = item(request="Pick one.\n(A) red\n(B) blue") + selected = {"response": response(q, "(B) blue"), "attempt": {"record_sha256": "x"}, + "state": {"attempts": [1]}, "unknown_bound_usd": "0"} + graded = analyze.grade_low(q, {"task": "shuffled_objects"}, "(B)", selected, OFFICIAL) + self.assertEqual(graded["correct"], 1) + self.assertEqual(graded["literal_correct"], 0) + self.assertFalse(analyze.grade_v2.decimal_match(OFFICIAL, "9007199254740992", "9007199254740993")) + + def test_stratified_whole_family_bootstrap_preserves_task_weights(self): + rows = [{"task": "a", "family": "family-a"}] * 3 + [{"task": "b", "family": "family-b"}] + bootstrap = frontier.Bootstrap(rows, 20) + values = np.array([[0.], [0.], [0.], [1.]]) + np.testing.assert_array_equal(bootstrap.means(values), np.full((20, 1), .25)) + same = [metric(x) for x in [0, 1, 1, 0]] + comparison = analyze.compare(same, same, bootstrap) + self.assertEqual(comparison["accuracy_difference_pp_ci95"]["lower"], 0) + self.assertEqual(comparison["answer_usd_per_10000_difference_ci95"]["upper"], 0) + with self.assertRaisesRegex(ValueError, "spans tasks"): + frontier.Bootstrap([{"task": "a", "family": "x"}, {"task": "b", "family": "x"}], 2) + + def test_exact_fraction_ties_signed_cost_and_bypasses(self): + p, c = [[.25, .5], [.5, .25], [.5, .5]], [[0, .25], [.25, 0], [0, 0]] + rule = {"format": utility_router.FORMAT, "mode": "utility", "lambda": {"numerator": "1", "denominator": "1"}} + self.assertEqual(utility_router.replay(rule, p, c).tolist(), [True, True, True]) + rule["lambda"] = {"numerator": "1000000000000000001", "denominator": "1000000000000000000"} + self.assertEqual(utility_router.replay(rule, p, c).tolist(), [False, True, True]) + for mode, expected in (("cost_limit", [False, True, True]), ("always_cheap", [False] * 3), + ("always_strong", [True] * 3)): + self.assertEqual(utility_router.replay({"format": utility_router.FORMAT, "mode": mode}, p, c).tolist(), expected) + + def test_token_accounting_does_not_add_reasoning_twice(self): + q = item() + r = response(q) + # 70 fresh input,10 cached,20 written,25 output including20 reasoning. + expected = (Decimal(70) * Decimal('.10') + Decimal(10) * Decimal('.01') + + Decimal(20) * Decimal('.125') + Decimal(25) * Decimal('.50')) / 1_000_000 + self.assertEqual(Decimal(r["cost"]["usd_estimate"]), expected) + old = {"cheap_" + k: v for k, v in metric(1).items()} + old.update(cheap_usage=r["usage"], cheap_cost_usd=float(expected)) + self.assertEqual(analyze.original_arm(old, "cheap", q)["cost_usd"], str(expected)) + + def test_replay_keeps_masks_and_checks_published_values(self): + rows = [{"id": str(i), "task": "a", "family": str(i), + "luna_medium": metric(i % 2), "luna_low": metric(1, "0.005", .5), + "sol_medium": metric(1, "0.05", 2.)} for i in range(4)] + p = [[.7, .8], [.5, .8], [.3, .8], [.1, .8]] + c = [[.01, .05]] * 4 + rule = {"format": utility_router.FORMAT, "mode": "utility", "lambda": {"numerator": "10", "denominator": "1"}} + routes = utility_router.replay(rule, p, c) + candidate = {"candidate_id": 3, "policy": rule, "calibration": {"fixed": True}} + methods, choices, report = {}, {}, {} + for name in analyze.METHODS: + embedding = name in ("embedding", "embedding__textcost", "knn") + methods[name] = {"scores": {"probabilities": p, "expected_usd": c, "metadata_assisted": False, + "router_kind": "embedding" if embedding else "native"}, + "candidates": [candidate], "calibration_frontier_candidate_ids": [3]} + choices[name] = {"0.8": candidate, "0.6": candidate} + old = [r["sol_medium" if mask else "luna_medium"] for r, mask in zip(rows, routes, strict=True)] + metrics = analyze.route_metrics(old, routes, [.001] * 4 if embedding else None) + report[name] = {"points": [{"candidate_id": 3, "rule": rule, + "answer_accuracy_percent": metrics["accuracy_percent"], + **{k: metrics[k] for k in ("strong_share", "answer_usd_per_10000", + "router_usd_per_10000", "total_usd_per_10000")}}]} + original = {"frozen": {"methods": methods, "encoder_fees_usd": [.001] * 4}, + "choices": {"choices": choices}, "report": {"methods": report}} + results = analyze.replay_policies(original, rows, frontier.Bootstrap(rows, 20)) + self.assertEqual(len(results), 9) + for point in results: + self.assertEqual(point["original_medium"]["strong_share"], point["substituted_low"]["strong_share"]) + self.assertEqual(point["original_medium"]["router_usd_per_10000"], point["substituted_low"]["router_usd_per_10000"]) + original["report"]["methods"]["base"]["points"][0]["strong_share"] += .01 + with self.assertRaisesRegex(ValueError, "replay parity"): + analyze.replay_policies(original, rows, frontier.Bootstrap(rows, 20)) + + def test_pinned_manifest_is_checked_before_any_grades(self): + with tempfile.TemporaryDirectory() as name: + root = Path(name) + analyze.write(root / "manifest.json", {"files": []}) + with patch.object(analyze, "read", side_effect=AssertionError("Must reject before grade reads")), \ + self.assertRaisesRegex(ValueError, "manifest does not match"): + analyze.checked_data(root) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/train-a-router/test_effort_collection.py b/examples/train-a-router/test_effort_collection.py new file mode 100644 index 000000000..acd67df1c --- /dev/null +++ b/examples/train-a-router/test_effort_collection.py @@ -0,0 +1,250 @@ +"""No-provider tests for the prospective Luna-low collection lifecycle.""" + +import copy +from decimal import Decimal +import hashlib +import json +import os +from pathlib import Path +import tempfile +import threading +import time +import unittest +from unittest.mock import patch + +import httpx + +import collect +from effort import collect_low as low + + +def item(number=0): + return {"id": f"bbeh:synthetic:{number:03d}", "request": "Which symbol follows A? Return B.", + "response_contract": {"type": "exact_string", "max_length": 2048}} + + +def plan(items): + return {"format": "bbeh-luna-low-acquisition-v1", "protocol": low.PROTOCOL, + "n": len(items), "ordered_ids": [row["id"] for row in items], + "inputs_canonical_sha256": low.digest(items), + "request_body_sha256": {row["id"]: low.digest(low.body_for(row)) for row in items}, + "source_files_sha256": low.source_hashes(), "budget_usd": "20.00", "deadline_seconds": 5400} + + +def success(row, *, answer="B"): + body = low.body_for(row) + usage = {"input_tokens": 100, "output_tokens": 10, + "input_tokens_details": {"cached_tokens": 20, "cache_write_tokens": 5}, + "output_tokens_details": {"reasoning_tokens": 4}, "total_tokens": 110} + return {"id": row["id"], "arm": "luna", "model": low.MODEL, "reasoning_effort": "low", + "body_sha256": low.digest(body), "created_unix": time.time(), "elapsed_seconds": 0.01, + "terminal_response": True, "execution_unknown": False, "request_not_sent": False, + "http_status": 200, "resolved_model": low.MODEL, "service_tier": "default", + "resolved_reasoning_effort": "low", "provider_status": "completed", "status": "complete", + "answer": answer, "valid_answer": answer is not None, + "messages": [{"type": "output_text", "text": json.dumps({"answer": answer})}], + "usage": usage, "cost": collect.openai_cost(usage, body)} + + +def transient(row, status=503): + return {"id": row["id"], "arm": "luna", "model": low.MODEL, "reasoning_effort": "low", + "body_sha256": low.digest(low.body_for(row)), "created_unix": time.time() - 100, + "elapsed_seconds": 0.01, "terminal_response": False, "execution_unknown": False, + "request_not_sent": False, "http_status": status, "retry_after_seconds": 0, + "status": "provider_error", "error_code": "overloaded", "error_type": "server_error"} + + +class CollectionTests(unittest.TestCase): + def test_request_changes_only_luna_effort_and_never_original_adapter(self): + settings = copy.deepcopy(collect.OPENAI_SETTINGS) + row = item() + medium, actual = collect.body_for(row, "luna"), low.body_for(row) + self.assertEqual(actual, {**medium, "reasoning": {"effort": "low"}}) + self.assertEqual(actual["model"], "gpt-6-luna") + self.assertEqual(actual["max_output_tokens"], 16384) + self.assertEqual(collect.OPENAI_SETTINGS, settings) + with self.assertRaises(ValueError): + low.body_for({**row, "reference": "B"}) + + def test_actual_http_adapter_preserves_readable_usage_and_hides_provider_extras(self): + requests = [] + + def handler(request): + requests.append(json.loads(request.content)) + return httpx.Response(200, json={"model": low.MODEL, "service_tier": "default", "status": "completed", + "reasoning": {"effort": "low"}, "id": "provider-response-id", + "output": [{"type": "reasoning", "encrypted_content": "synthetic-opaque"}, + {"type": "message", "content": [{"type": "output_text", "text": '{"answer":"B"}'}]}], + "usage": success(item())["usage"]}) + + with patch.dict(os.environ, {"OPENAI_API_KEY": "synthetic-key"}): + row = low.acquire(item(), lambda: httpx.Client(transport=httpx.MockTransport(handler))) + self.assertEqual(requests, [low.body_for(item())]) + self.assertEqual(row["answer"], "B") + self.assertEqual(row["cost"], collect.openai_cost(row["usage"], requests[0])) + self.assertGreaterEqual(row["elapsed_seconds"], 0) + for forbidden in ("synthetic-key", "provider-response-id", "synthetic-opaque", "Authorization"): + self.assertNotIn(forbidden, json.dumps(row)) + + def test_successful_resume_has_no_new_attempt_or_response_changes(self): + with tempfile.TemporaryDirectory() as temporary: + rows, root = [item(0), item(1)], Path(temporary) + checks = [] + result = low.run(rows, root, plan(rows), commit=lambda: checks.append(True), call=success, emit=False) + self.assertEqual(result["status"], "complete") + before = {str(path): path.read_bytes() for path in root.glob("attempts/*/*")} + with patch.object(low, "acquire", side_effect=AssertionError("Must not call")): + result = low.run(rows, root, plan(rows), call=lambda row: self.fail("Duplicate call"), emit=False) + self.assertEqual(result["physical_attempts"], 2) + self.assertEqual(before, {str(path): path.read_bytes() for path in root.glob("attempts/*/*")}) + self.assertTrue(checks) + + def test_no_retry_for_invalid_truncated_or_incomplete_terminal_receipt(self): + for variant in ("invalid", "truncated", "missing-usage"): + with self.subTest(variant=variant), tempfile.TemporaryDirectory() as temporary: + calls = [] + + def call(row): + calls.append(row["id"]) + value = success(row, answer=None) + if variant == "truncated": + value.update(provider_status="incomplete", incomplete_reason="max_output_tokens") + if variant == "missing-usage": + value.pop("cost") + value.pop("usage") + value.update(status="receipt_incomplete", execution_unknown=True) + return value + + low.run([item()], temporary, plan([item()]), call=call, emit=False) + low.run([item()], temporary, plan([item()]), call=call, emit=False) + entry = low.read(Path(temporary) / "selection.json")["entries"][0] + self.assertEqual(calls, [item()["id"]]) + self.assertEqual(entry["selected_attempt"], 1) + + def test_transient_retries_are_bounded_and_charge_every_unknown_attempt(self): + for succeed in (True, False): + with self.subTest(succeed=succeed), tempfile.TemporaryDirectory() as temporary: + calls = [] + + def call(row): + calls.append(row["id"]) + return success(row) if succeed and len(calls) == 3 else transient(row) + + result = low.run([item()], temporary, plan([item()]), call=call, emit=False) + self.assertEqual(len(calls), 3) + reserve = collect.reservation(low.body_for(item()), "luna") + self.assertEqual(Decimal(result["unknown_reservation_usd"]), reserve * (2 if succeed else 3)) + self.assertEqual(result["status"], "complete" if succeed else "incomplete") + low.run([item()], temporary, plan([item()]), call=call, emit=False) + self.assertEqual(len(calls), 3) + + def test_uncertain_response_and_missing_receipt_are_not_repeated(self): + with tempfile.TemporaryDirectory() as temporary: + row, root = item(), Path(temporary) + request_plan = plan([row]) + low.original.dump(root / "plan.json", request_plan) + path = root / "attempts" / hashlib.sha256(row["id"].encode()).hexdigest() / "001-intent.json" + reserve = collect.reservation(low.body_for(row), "luna") + low.original.dump(path, {"id": row["id"], "arm": "luna", "model": low.MODEL, "reasoning_effort": "low", + "attempt": 1, "body_sha256": low.digest(low.body_for(row)), "plan_sha256": low.sha(root / "plan.json"), + "reserved_usd": str(reserve), "created_unix": time.time() - 10}) + result = low.run([row], root, request_plan, call=lambda value: self.fail("Unknown repeated"), emit=False) + self.assertEqual(result["status_counts"], {"uncertain": 1}) + self.assertEqual(Decimal(result["unknown_reservation_usd"]), reserve) + self.assertIsNone(low.read(root / "selection.json")["entries"][0]["attempts"][0]["record_path"]) + self.assertFalse(path.with_name("001-response.json").exists()) + with tempfile.TemporaryDirectory() as temporary: + calls = [] + + def timeout(row): + calls.append(row["id"]) + return {**transient(row), "http_status": None, "error_type": "ReadTimeout", + "status": "failed_or_unknown", "execution_unknown": True} + + low.run([item()], temporary, plan([item()]), call=timeout, emit=False) + low.run([item()], temporary, plan([item()]), call=timeout, emit=False) + self.assertEqual(len(calls), 1) + + def test_identity_and_fatal_quota_stop_and_preserve_first_terminal(self): + for variant in ("model", "tier", "effort", "quota"): + with self.subTest(variant=variant), tempfile.TemporaryDirectory() as temporary: + calls = [] + + def call(row): + calls.append(row["id"]) + value = success(row) + if variant == "model": + value["resolved_model"] = "gpt-6.1-sol" + elif variant == "tier": + value["service_tier"] = "flex" + elif variant == "effort": + value["resolved_reasoning_effort"] = "medium" + else: + value = {**transient(row, 429), "error_code": "credit_balance_exhausted"} + return value + + result = low.run([item()], temporary, plan([item()]), call=call, emit=False) + self.assertEqual(result["stop"], "provider_auth_quota_or_identity_error") + self.assertEqual(result["status"], "incomplete") + entry = low.read(Path(temporary) / "selection.json")["entries"][0] + self.assertEqual(entry["selected_attempt"], None if variant == "quota" else 1) + low.run([item()], temporary, plan([item()]), call=call, emit=False) + self.assertEqual(len(calls), 1) + + def test_expired_deadline_and_insufficient_budget_dispatch_nothing(self): + for variant in ("deadline", "budget"): + with self.subTest(variant=variant), tempfile.TemporaryDirectory() as temporary: + root, request_plan = Path(temporary), plan([item()]) + if variant == "budget": + request_plan["budget_usd"] = "0.00001" + low.original.dump(root / "plan.json", request_plan) + if variant == "deadline": + low.original.dump(root / "execution.json", {"plan_sha256": low.sha(root / "plan.json"), + "started_unix": time.time() - 6000, "deadline_seconds": 5400}) + result = low.run([item()], root, request_plan, call=lambda row: self.fail("Unauthorized call"), emit=False) + self.assertEqual(result["physical_attempts"], 0) + self.assertEqual(result["stop"], "dispatch_deadline" if variant == "deadline" else "budget_bound") + + def test_all_declared_source_bindings_and_undeclared_responses_are_checked(self): + for name in low.source_hashes(): + with self.subTest(name=name), tempfile.TemporaryDirectory() as temporary: + request_plan = plan([item()]) + request_plan["source_files_sha256"][name] = "0" * 64 + with self.assertRaisesRegex(ValueError, "source changed"): + low.run([item()], temporary, request_plan, call=lambda row: self.fail("Called"), emit=False) + with tempfile.TemporaryDirectory() as temporary: + root, request_plan = Path(temporary), plan([item()]) + low.original.dump(root / "plan.json", request_plan) + low.original.dump(root / "attempts/unknown/001-response.json", {"undeclared": True}) + with self.assertRaisesRegex(ValueError, "Unaccounted"): + low.run([item()], root, request_plan, call=lambda row: self.fail("Called"), emit=False) + + def test_rolling_concurrency_never_exceeds_sixteen_and_intents_precede_calls(self): + with tempfile.TemporaryDirectory() as temporary: + root, active, maximum, lock = Path(temporary), 0, 0, threading.Lock() + committed = set() + + def commit(): + committed.update(path.relative_to(root).as_posix() for path in root.glob("attempts/*/*-intent.json")) + + def call(row): + nonlocal active, maximum + name = "attempts/" + hashlib.sha256(row["id"].encode()).hexdigest() + "/001-intent.json" + self.assertIn(name, committed) + with lock: + active += 1 + maximum = max(maximum, active) + time.sleep(0.01) + with lock: + active -= 1 + return success(row) + + items = [item(number) for number in range(33)] + result = low.run(items, root, plan(items), commit=commit, call=call, emit=False) + self.assertEqual(result["complete_receipts"], 33) + self.assertGreater(maximum, 1) + self.assertLessEqual(maximum, 16) + + +if __name__ == "__main__": + unittest.main()