From af6a7e6139105671d69d8613f2be08e16b0c1bd7 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 28 Oct 2026 13:06:17 +0800 Subject: [PATCH 1/2] feat(benchmark): add token capacity admission Signed-off-by: huangruiteng --- .../capabilities/benchmark_toolkit/README.md | 87 +- .../benchmark_toolkit/__init__.py | 38 + .../benchmark_toolkit/catalog_entry.py | 124 ++ .../benchmark_toolkit/token_capacity.py | 1010 +++++++++++++++++ loopx/cli_commands/benchmark_dispatch.py | 22 +- .../cli_commands/benchmark_token_capacity.py | 213 ++++ .../test_benchmark_token_capacity.py | 502 ++++++++ 7 files changed, 1990 insertions(+), 6 deletions(-) create mode 100644 loopx/capabilities/benchmark_toolkit/token_capacity.py create mode 100644 loopx/cli_commands/benchmark_token_capacity.py create mode 100644 tests/capabilities/test_benchmark_token_capacity.py diff --git a/loopx/capabilities/benchmark_toolkit/README.md b/loopx/capabilities/benchmark_toolkit/README.md index 0e3ce875ab..5860e57a73 100644 --- a/loopx/capabilities/benchmark_toolkit/README.md +++ b/loopx/capabilities/benchmark_toolkit/README.md @@ -804,11 +804,12 @@ A countable experiment uses the toolkit in this order: 1. When domain guidance is a factor, qualify the four-arm contract and freeze its task-goal hashes before preregistration. 2. Read the project experiment board before selecting or launching another case. -3. Read or configure the concurrency envelope, then reconcile its reservations with - exact runner liveness. +3. Read or configure the concurrency envelope and, when provider throughput is + shared, the token-capacity envelope; reconcile both reservations with exact + runner liveness. 4. Declare a `run_permission_policy_v0` and preflight the runner boundary. -5. Atomically admit a case slot immediately before the independently authorized - runner launch. +5. Atomically reserve token capacity, pass its fresh provider-capacity receipt to + case-slot admission, and launch only after both admissions succeed. 6. Upsert the planned or running row, then launch one frozen case/arm; do not expose evaluator sources or official feedback. 7. Capture ATIF tool evidence and a runner-owned runtime attestation. @@ -821,7 +822,8 @@ A countable experiment uses the toolkit in this order: 11. Run the independent verifier only after the agent phase. 12. Reduce the official result through the benchmark-owned scoring path. 13. Upsert terminal score, countability, effort, treatment fidelity, and insight - status; release its reservation; then read the matched-comparison projection. + status; release its case and token reservations; then read the matched-comparison + projection. 14. Apply attempt-countability, treatment-fidelity, and matched-pair gates before any comparison claim. @@ -902,6 +904,81 @@ host default such as `/tmp`; if that binding cannot be proven, report `unresolve LoopX never records raw metrics, paths, provider logs, or the receipt in the envelope, and the receipt does not grant launch authority. +### Token capacity + +When a provider quota is expressed in tokens per minute, configure the goal-scoped +token envelope as a second admission dimension. The target is an integer fraction +of the hard quota. Cold-start reserve absorbs short request bursts, and evaluator +reserve prevents solver backfill from consuming capacity needed to close completed +runs: + +```bash +loopx benchmark token-capacity-configure \ + --goal-id \ + --quota-tokens-per-minute 10000000 \ + --target-utilization-basis-points 9000 \ + --cold-start-reserve-tokens-per-minute 500000 \ + --evaluator-reserve-tokens-per-minute 1000000 \ + --cooldown-seconds 90 \ + --execute \ + --format json +``` + +The runner converts provider usage into request-level +`benchmark_token_usage_sample_v0` records. `uncached_input_tokens`, +`cached_input_tokens`, and `output_tokens` are disjoint counters, so cached input is +never counted twice. A measured zero-request sample is an explicit coverage +heartbeat. A request whose usage cannot be obtained must be recorded as +`metering_state=unmetered`; any such request in the exact trailing 60-second window +makes new admission unresolved. Estimated but complete usage remains visible as +`complete_estimated` and is counted at its supplied value. + +```bash +loopx benchmark token-capacity-observe \ + --goal-id \ + --sample-json token-usage-sample.json \ + --execute \ + --format json + +loopx benchmark token-capacity-admit \ + --goal-id \ + --run-id \ + --workload-class \ + --tokens-per-minute \ + --execute \ + --format json +``` + +Admission uses the greater of rolling observed tokens and current reservations. A +solver is checked below the evaluator reserve; an evaluator may consume that +reserve. The successful decision contains a fresh +`benchmark_resource_headroom_receipt_v0` with a `provider_capacity=sufficient` +check. Pass that receipt to `concurrency-admit`. If case-slot admission fails, +release the token reservation before trying another run. + +Record an HTTP 429 or equivalent provider capacity rejection with +`token-capacity-reject`. LoopX keeps the longer of the policy cooldown and a bounded +provider `Retry-After`, and new token admissions remain closed until it expires. +The usage window, cooldown, and exact run reservations are persisted, so a +controller restart produces the same decision. Reservations are not process +liveness: after a typed terminal or runner-invalid transition, release the same run +from both envelopes. Reconcile orphaned reservations from authoritative runner +liveness; LoopX never discovers or terminates jobs itself. + +```bash +loopx benchmark token-capacity-reject \ + --goal-id \ + --retry-after-seconds 120 \ + --execute \ + --format json + +loopx benchmark token-capacity-release \ + --goal-id \ + --run-id \ + --execute \ + --format json +``` + Read back the gate with `concurrency-status`. To disable it, rerun `concurrency-configure` with the same capacity values and omit `--require-resource-headroom-receipt`; existing active reservations are preserved. diff --git a/loopx/capabilities/benchmark_toolkit/__init__.py b/loopx/capabilities/benchmark_toolkit/__init__.py index 4fcca4a0a4..e2b719d09b 100644 --- a/loopx/capabilities/benchmark_toolkit/__init__.py +++ b/loopx/capabilities/benchmark_toolkit/__init__.py @@ -228,6 +228,26 @@ BENCHMARK_TREATMENT_CONTINUATION_RECEIPT_SCHEMA_VERSION, build_benchmark_treatment_continuation_receipt, ) +from .token_capacity import ( + BENCHMARK_TOKEN_CAPACITY_DECISION_SCHEMA_VERSION, + BENCHMARK_TOKEN_CAPACITY_FILENAME, + BENCHMARK_TOKEN_CAPACITY_POLICY_SCHEMA_VERSION, + BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION, + BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION, + admit_benchmark_token_capacity, + build_benchmark_token_capacity_policy, + build_benchmark_token_capacity_status, + configure_benchmark_token_capacity, + default_benchmark_token_capacity_path, + normalize_benchmark_token_capacity_policy, + normalize_benchmark_token_capacity_state, + normalize_benchmark_token_usage_sample, + observe_benchmark_token_usage, + read_benchmark_token_capacity_state, + record_benchmark_token_capacity_rejection, + release_benchmark_token_capacity, + render_benchmark_token_capacity_markdown, +) __all__ = [ "ATIF_SCHEMA_VERSION", @@ -272,6 +292,11 @@ "BENCHMARK_STUDY_MANIFEST_SCHEMA_VERSION", "BENCHMARK_TREATMENT_CONTINUATION_OBSERVATION_SCHEMA_VERSION", "BENCHMARK_TREATMENT_CONTINUATION_RECEIPT_SCHEMA_VERSION", + "BENCHMARK_TOKEN_CAPACITY_DECISION_SCHEMA_VERSION", + "BENCHMARK_TOKEN_CAPACITY_FILENAME", + "BENCHMARK_TOKEN_CAPACITY_POLICY_SCHEMA_VERSION", + "BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION", + "BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION", "BENCHMARK_UPLOAD_ENVELOPE_SCHEMA_VERSION", "BENCHMARK_UPLOAD_READBACK_RECEIPT_SCHEMA_VERSION", "DEFAULT_RUN_PERMISSION_ALLOWED_ACTIONS", @@ -321,6 +346,7 @@ "RunnerOwnedProviderGateway", "StdioNativeGoalTransport", "admit_benchmark_case", + "admit_benchmark_token_capacity", "attach_native_goal", "benchmark_experiment_board_row_key", "build_benchmark_adaptive_concurrency_decision", @@ -341,6 +367,8 @@ "build_benchmark_runtime_observation", "build_benchmark_study_dashboard", "build_benchmark_treatment_continuation_receipt", + "build_benchmark_token_capacity_policy", + "build_benchmark_token_capacity_status", "build_benchmark_upload_envelope", "build_native_codex_isolation_envelope", "build_native_goal_public_trajectory_summary", @@ -357,9 +385,11 @@ "compact_native_goal_receipt", "compact_run_permission_policy_for_quota", "configure_benchmark_concurrency_envelope", + "configure_benchmark_token_capacity", "convert_traex_events_to_atif", "default_benchmark_concurrency_envelope_path", "default_benchmark_experiment_board_path", + "default_benchmark_token_capacity_path", "filter_public_benchmark_artifact_paths", "inspect_benchmark_source_revision_fence", "inspect_native_codex_profile", @@ -378,8 +408,12 @@ "normalize_benchmark_resource_headroom_receipt", "normalize_benchmark_rebalance_inventory", "normalize_benchmark_study_manifest", + "normalize_benchmark_token_capacity_policy", + "normalize_benchmark_token_capacity_state", + "normalize_benchmark_token_usage_sample", "normalize_benchmark_upload_envelope", "observe_native_goal_event", + "observe_benchmark_token_usage", "preview_benchmark_experiment_board_reconcile", "preview_benchmark_experiment_board_upsert", "probe_native_goal_process", @@ -387,11 +421,15 @@ "read_benchmark_experiment_board_rows", "read_benchmark_local_upload_records", "read_benchmark_upload_receipt", + "read_benchmark_token_capacity_state", "rebase_native_codex_loopx_workspace_state", "refresh_native_goal_status", "release_benchmark_case", + "record_benchmark_token_capacity_rejection", + "release_benchmark_token_capacity", "render_benchmark_concurrency_markdown", "render_benchmark_rebalance_plan_markdown", + "render_benchmark_token_capacity_markdown", "render_benchmark_experiment_board_markdown", "render_native_codex_goal_prompt", "run_native_goal_process", diff --git a/loopx/capabilities/benchmark_toolkit/catalog_entry.py b/loopx/capabilities/benchmark_toolkit/catalog_entry.py index 4381c441c9..73107bcf80 100644 --- a/loopx/capabilities/benchmark_toolkit/catalog_entry.py +++ b/loopx/capabilities/benchmark_toolkit/catalog_entry.py @@ -205,6 +205,68 @@ "remain runner-owned" ), }, + { + "command": ( + "loopx benchmark token-capacity-configure --goal-id " + "--quota-tokens-per-minute " + "--target-utilization-basis-points " + "--cold-start-reserve-tokens-per-minute " + "--evaluator-reserve-tokens-per-minute " + "--execute --format json" + ), + "purpose": ( + "Configure a durable rolling token-capacity envelope with explicit " + "burst and evaluator reserves." + ), + "write_boundary": ( + "project-local aggregate policy and reservations only; provider " + "routes, credentials, request text, and launch authority remain " + "runner-owned" + ), + }, + { + "command": ( + "loopx benchmark token-capacity-observe --goal-id " + "--sample-json --execute --format json" + ), + "purpose": ( + "Record disjoint cached, uncached, and output token counts or an " + "explicit unmetered request in the exact trailing minute." + ), + "write_boundary": ( + "typed aggregate usage only; no provider payload, prompt, response, " + "credential, endpoint, or raw log is accepted" + ), + }, + { + "command": ( + "loopx benchmark token-capacity-admit --goal-id " + "--run-id --workload-class " + "--tokens-per-minute --execute --format json" + ), + "purpose": ( + "Atomically reserve forecast throughput and emit a fresh provider-" + "capacity headroom receipt for case-slot admission." + ), + "write_boundary": ( + "compact run reservation only; callers own usage collection, " + "case-slot rollback, launch, liveness reconciliation, and release" + ), + }, + { + "command": ( + "loopx benchmark token-capacity-reject --goal-id " + "[--retry-after-seconds ] --execute --format json" + ), + "purpose": ( + "Persist a bounded provider-capacity cooldown after an HTTP 429 or " + "equivalent typed rejection." + ), + "write_boundary": ( + "timestamp and bounded cooldown only; no error body, provider " + "identity, route, or credential is persisted" + ), + }, { "command": ( "loopx benchmark concurrency-tune --goal-id " @@ -394,11 +456,14 @@ "read_experiment_board_before_launch_or_case_selection", "qualify_four_arm_contract_when_domain_guidance_is_a_factor", "configure_or_read_concurrency_envelope_before_launch", + "configure_or_read_token_capacity_when_provider_throughput_is_shared", "reconcile_admission_ledger_with_real_runner_liveness", "backfill_when_concurrency_status_reports_underfilled", "qualify_source_revision_before_each_new_run_admission", "qualify_resource_headroom_when_the_envelope_requires_a_receipt", + "record_complete_or_explicitly_unmetered_token_usage", "tune_target_occupancy_from_fresh_runner_feedback_when_opted_in", + "atomically_reserve_token_capacity_before_case_slot_admission", "atomically_admit_case_slot_before_runner_launch", "upsert_preregistered_or_running_row_when_a_run_starts", "classify_exact_runtime_observation_during_active_monitor_cycles", @@ -406,6 +471,7 @@ "adjudicate_restricted_access_suspicion_after_solver_and_score_terminal", "upsert_terminal_score_countability_effort_and_insight_status", "release_case_slot_after_terminal_or_runner_invalid_transition", + "release_token_capacity_after_terminal_or_runner_invalid_transition", "read_matched_comparisons_before_selecting_the_next_arm", ], "board_commands": { @@ -460,6 +526,42 @@ "and score transitions. When configured, admission requires a fresh " "public-safe resource-headroom receipt but never persists raw metrics." ), + "token_capacity_commands": { + "status": ( + "loopx benchmark token-capacity-status --goal-id " + "--format json" + ), + "configure": ( + "loopx benchmark token-capacity-configure --goal-id " + "--quota-tokens-per-minute " + "--target-utilization-basis-points " + "--cold-start-reserve-tokens-per-minute " + "--evaluator-reserve-tokens-per-minute " + "--execute --format json" + ), + "observe": ( + "loopx benchmark token-capacity-observe --goal-id " + "--sample-json --execute --format json" + ), + "admit": ( + "loopx benchmark token-capacity-admit --goal-id " + "--run-id --workload-class " + "--tokens-per-minute --execute --format json" + ), + "reject": ( + "loopx benchmark token-capacity-reject --goal-id " + "[--retry-after-seconds ] --execute --format json" + ), + "release": ( + "loopx benchmark token-capacity-release --goal-id " + "--run-id --execute --format json" + ), + }, + "token_capacity_boundary": ( + "The token envelope owns aggregate metering, forecast reservations, and " + "provider cooldown only. The runner owns provider adaptation, complete " + "coverage, job discovery, launch, rollback, liveness, and termination." + ), "selection_rule": ( "Keep diagnostic-only explore rows separate and make paired claims " "only from matched_pair_countable comparisons." @@ -987,6 +1089,26 @@ "module": "loopx.capabilities.benchmark_toolkit.concurrency_envelope", "doc": "loopx/capabilities/benchmark_toolkit/README.md", }, + { + "schema_version": "benchmark_token_capacity_policy_v0", + "module": "loopx.capabilities.benchmark_toolkit.token_capacity", + "doc": "loopx/capabilities/benchmark_toolkit/README.md", + }, + { + "schema_version": "benchmark_token_usage_sample_v0", + "module": "loopx.capabilities.benchmark_toolkit.token_capacity", + "doc": "loopx/capabilities/benchmark_toolkit/README.md", + }, + { + "schema_version": "benchmark_token_capacity_state_v0", + "module": "loopx.capabilities.benchmark_toolkit.token_capacity", + "doc": "loopx/capabilities/benchmark_toolkit/README.md", + }, + { + "schema_version": "benchmark_token_capacity_decision_v0", + "module": "loopx.capabilities.benchmark_toolkit.token_capacity", + "doc": "loopx/capabilities/benchmark_toolkit/README.md", + }, { "schema_version": "benchmark_adaptive_concurrency_policy_v0", "module": "loopx.capabilities.benchmark_toolkit.adaptive_concurrency", @@ -1077,6 +1199,7 @@ ( "python -m pytest tests/capabilities/test_benchmark_toolkit.py " "tests/capabilities/test_benchmark_concurrency_envelope.py " + "tests/capabilities/test_benchmark_token_capacity.py " "tests/capabilities/test_benchmark_experiment_board.py " "tests/capabilities/test_benchmark_source_revision_fence.py -q" ) @@ -1094,6 +1217,7 @@ "Integrity qualification establishes countability eligibility only; an independent official result and matched experiment contract are still required.", "Post-run analyst access never widens the solving agent's evidence boundary or grants feedback reuse in another scored run.", "Concurrency reservations are not runtime liveness: runners must reconcile exact processes before release or backfill.", + "Token reservations are not runtime liveness or provider attribution: runners must report complete metering coverage, roll back a token reservation after case-slot denial, and reconcile exact processes before release.", "The experiment board is a compact projection, not a score authority: benchmark-family runners and scoring adapters remain outside the active capability.", ], "next_real_step": ( diff --git a/loopx/capabilities/benchmark_toolkit/token_capacity.py b/loopx/capabilities/benchmark_toolkit/token_capacity.py new file mode 100644 index 0000000000..ec470c2adc --- /dev/null +++ b/loopx/capabilities/benchmark_toolkit/token_capacity.py @@ -0,0 +1,1010 @@ +from __future__ import annotations + +import json +from collections.abc import Mapping +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +from ...domain_state import default_domain_state_file_path +from ...file_lock import exclusive_file_lock +from ...registry import atomic_write_json +from .concurrency_envelope import ( + BENCHMARK_RESOURCE_HEADROOM_RECEIPT_SCHEMA_VERSION, + _now_iso, + _reject_unknown_fields, + _timestamp, +) +from .experiment_identity import experiment_token_text as _token + +BENCHMARK_TOKEN_CAPACITY_POLICY_SCHEMA_VERSION = "benchmark_token_capacity_policy_v0" +BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION = "benchmark_token_usage_sample_v0" +BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION = "benchmark_token_capacity_state_v0" +BENCHMARK_TOKEN_CAPACITY_DECISION_SCHEMA_VERSION = ( + "benchmark_token_capacity_decision_v0" +) +BENCHMARK_TOKEN_CAPACITY_FILENAME = "token-capacity.json" + +_WINDOW = timedelta(minutes=1) +_MAX_COOLDOWN_SECONDS = 3600 +_MAX_TOKEN_VALUE = 10**15 +_METERING_STATES = {"measured", "estimated", "unmetered"} +_WORKLOAD_CLASSES = {"solver", "evaluator"} +_POLICY_FIELDS = { + "schema_version", + "quota_tokens_per_minute", + "target_utilization_basis_points", + "cold_start_reserve_tokens_per_minute", + "evaluator_reserve_tokens_per_minute", + "cooldown_seconds", +} +_SAMPLE_FIELDS = { + "schema_version", + "sample_id", + "observed_at", + "metering_state", + "request_count", + "uncached_input_tokens", + "cached_input_tokens", + "output_tokens", +} +_RESERVATION_FIELDS = { + "run_id", + "workload_class", + "tokens_per_minute", + "reserved_at", +} +_COOLDOWN_FIELDS = {"last_rejected_at", "until"} +_STATE_FIELDS = { + "schema_version", + "configured_at", + "updated_at", + "policy", + "samples", + "reservations", + "cooldown", +} + + +def _bounded_token_int( + value: Any, + *, + field: str, + minimum: int = 0, + maximum: int = _MAX_TOKEN_VALUE, +) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < minimum: + qualifier = "positive" if minimum == 1 else "non-negative" + raise ValueError(f"{field} must be a {qualifier} integer") + if value > maximum: + raise ValueError(f"{field} must not exceed {maximum}") + return value + + +def build_benchmark_token_capacity_policy( + *, + quota_tokens_per_minute: int, + target_utilization_basis_points: int = 9000, + cold_start_reserve_tokens_per_minute: int = 0, + evaluator_reserve_tokens_per_minute: int = 0, + cooldown_seconds: int = 60, +) -> dict[str, Any]: + return normalize_benchmark_token_capacity_policy( + { + "schema_version": BENCHMARK_TOKEN_CAPACITY_POLICY_SCHEMA_VERSION, + "quota_tokens_per_minute": quota_tokens_per_minute, + "target_utilization_basis_points": target_utilization_basis_points, + "cold_start_reserve_tokens_per_minute": ( + cold_start_reserve_tokens_per_minute + ), + "evaluator_reserve_tokens_per_minute": ( + evaluator_reserve_tokens_per_minute + ), + "cooldown_seconds": cooldown_seconds, + } + ) + + +def normalize_benchmark_token_capacity_policy( + value: Mapping[str, Any], +) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise TypeError("benchmark token capacity policy must be an object") + _reject_unknown_fields(value, allowed=_POLICY_FIELDS, field="policy") + if value.get("schema_version") != BENCHMARK_TOKEN_CAPACITY_POLICY_SCHEMA_VERSION: + raise ValueError("benchmark token capacity policy schema mismatch") + quota = _bounded_token_int( + value.get("quota_tokens_per_minute"), + field="quota_tokens_per_minute", + minimum=1, + ) + utilization = _bounded_token_int( + value.get("target_utilization_basis_points"), + field="target_utilization_basis_points", + minimum=1, + maximum=10_000, + ) + cold_start_reserve = _bounded_token_int( + value.get("cold_start_reserve_tokens_per_minute"), + field="cold_start_reserve_tokens_per_minute", + ) + evaluator_reserve = _bounded_token_int( + value.get("evaluator_reserve_tokens_per_minute"), + field="evaluator_reserve_tokens_per_minute", + ) + target = quota * utilization // 10_000 + if cold_start_reserve + evaluator_reserve > target: + raise ValueError( + "cold-start and evaluator reserves cannot exceed target token capacity" + ) + return { + "schema_version": BENCHMARK_TOKEN_CAPACITY_POLICY_SCHEMA_VERSION, + "quota_tokens_per_minute": quota, + "target_utilization_basis_points": utilization, + "cold_start_reserve_tokens_per_minute": cold_start_reserve, + "evaluator_reserve_tokens_per_minute": evaluator_reserve, + "cooldown_seconds": _bounded_token_int( + value.get("cooldown_seconds"), + field="cooldown_seconds", + minimum=1, + maximum=_MAX_COOLDOWN_SECONDS, + ), + } + + +def normalize_benchmark_token_usage_sample( + value: Mapping[str, Any], +) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise TypeError("benchmark token usage sample must be an object") + _reject_unknown_fields(value, allowed=_SAMPLE_FIELDS, field="sample") + if value.get("schema_version") != BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION: + raise ValueError("benchmark token usage sample schema mismatch") + metering_state = _token(value.get("metering_state"), field="metering_state") + if metering_state not in _METERING_STATES: + raise ValueError("metering_state is unsupported") + sample = { + "schema_version": BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION, + "sample_id": _token(value.get("sample_id"), field="sample_id"), + "observed_at": _timestamp(value.get("observed_at"), field="observed_at"), + "metering_state": metering_state, + "request_count": _bounded_token_int( + value.get("request_count"), field="request_count" + ), + "uncached_input_tokens": _bounded_token_int( + value.get("uncached_input_tokens"), field="uncached_input_tokens" + ), + "cached_input_tokens": _bounded_token_int( + value.get("cached_input_tokens"), field="cached_input_tokens" + ), + "output_tokens": _bounded_token_int( + value.get("output_tokens"), field="output_tokens" + ), + } + token_total = sum( + sample[field] + for field in ( + "uncached_input_tokens", + "cached_input_tokens", + "output_tokens", + ) + ) + if metering_state == "unmetered": + if sample["request_count"] == 0: + raise ValueError("unmetered samples must identify at least one request") + if token_total != 0: + raise ValueError("unmetered samples cannot claim token counts") + elif sample["request_count"] == 0 and token_total != 0: + raise ValueError("token counts require at least one observed request") + return sample + + +def _normalize_reservation(value: Mapping[str, Any]) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise TypeError("benchmark token reservation must be an object") + _reject_unknown_fields(value, allowed=_RESERVATION_FIELDS, field="reservation") + workload_class = _token(value.get("workload_class"), field="workload_class") + if workload_class not in _WORKLOAD_CLASSES: + raise ValueError("workload_class is unsupported") + return { + "run_id": _token(value.get("run_id"), field="run_id"), + "workload_class": workload_class, + "tokens_per_minute": _bounded_token_int( + value.get("tokens_per_minute"), + field="tokens_per_minute", + minimum=1, + ), + "reserved_at": _timestamp(value.get("reserved_at"), field="reserved_at"), + } + + +def _normalize_cooldown(value: Mapping[str, Any] | None) -> dict[str, str] | None: + if value is None: + return None + if not isinstance(value, Mapping): + raise TypeError("cooldown must be an object or null") + _reject_unknown_fields(value, allowed=_COOLDOWN_FIELDS, field="cooldown") + last_rejected_at = _timestamp( + value.get("last_rejected_at"), field="last_rejected_at" + ) + until = _timestamp(value.get("until"), field="until") + if datetime.fromisoformat(until) <= datetime.fromisoformat(last_rejected_at): + raise ValueError("cooldown.until must be later than last_rejected_at") + return {"last_rejected_at": last_rejected_at, "until": until} + + +def normalize_benchmark_token_capacity_state( + value: Mapping[str, Any], +) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise TypeError("benchmark token capacity state must be an object") + _reject_unknown_fields(value, allowed=_STATE_FIELDS, field="state") + if value.get("schema_version") != BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION: + raise ValueError("benchmark token capacity state schema mismatch") + samples_value = value.get("samples") + reservations_value = value.get("reservations") + if not isinstance(samples_value, list): + raise TypeError("samples must be a list") + if not isinstance(reservations_value, list): + raise TypeError("reservations must be a list") + samples = [normalize_benchmark_token_usage_sample(item) for item in samples_value] + sample_ids = [item["sample_id"] for item in samples] + if len(sample_ids) != len(set(sample_ids)): + raise ValueError("samples cannot contain duplicate sample_id values") + reservations = [_normalize_reservation(item) for item in reservations_value] + run_ids = [item["run_id"] for item in reservations] + if len(run_ids) != len(set(run_ids)): + raise ValueError("reservations cannot contain duplicate run_id values") + configured_at = _timestamp(value.get("configured_at"), field="configured_at") + updated_at = _timestamp(value.get("updated_at"), field="updated_at") + if datetime.fromisoformat(updated_at) < datetime.fromisoformat(configured_at): + raise ValueError("updated_at cannot be earlier than configured_at") + return { + "schema_version": BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION, + "configured_at": configured_at, + "updated_at": updated_at, + "policy": normalize_benchmark_token_capacity_policy(value.get("policy")), + "samples": sorted( + samples, key=lambda item: (item["observed_at"], item["sample_id"]) + ), + "reservations": sorted(reservations, key=lambda item: item["run_id"]), + "cooldown": _normalize_cooldown(value.get("cooldown")), + } + + +def default_benchmark_token_capacity_path( + *, project: str | Path = ".", goal_id: str +) -> Path: + return default_domain_state_file_path( + project=project, + goal_id=goal_id, + domain_pack="benchmark_toolkit", + filename=BENCHMARK_TOKEN_CAPACITY_FILENAME, + ) + + +def read_benchmark_token_capacity_state(path: str | Path) -> dict[str, Any] | None: + target = Path(path).expanduser() + if not target.exists(): + return None + return normalize_benchmark_token_capacity_state( + json.loads(target.read_text(encoding="utf-8")) + ) + + +def _target_tokens(policy: Mapping[str, Any]) -> int: + return ( + policy["quota_tokens_per_minute"] + * policy["target_utilization_basis_points"] + // 10_000 + ) + + +def _window_samples( + samples: list[dict[str, Any]], *, observed_at: str +) -> tuple[list[dict[str, Any]], bool]: + now = datetime.fromisoformat(observed_at) + lower = now - _WINDOW + future = any(datetime.fromisoformat(item["observed_at"]) > now for item in samples) + return ( + [ + item + for item in samples + if lower < datetime.fromisoformat(item["observed_at"]) <= now + ], + future, + ) + + +def _usage_projection( + samples: list[dict[str, Any]], *, observed_at: str +) -> dict[str, Any]: + window, future = _window_samples(samples, observed_at=observed_at) + states = {item["metering_state"] for item in window} + if future: + coverage = "unresolved_future_sample" + reason_codes = ["token_usage_sample_from_future"] + elif not window: + coverage = "unobserved" + reason_codes = ["token_usage_unobserved"] + elif "unmetered" in states: + coverage = "partial_unmetered" + reason_codes = ["token_usage_partial_unmetered"] + elif "estimated" in states: + coverage = "complete_estimated" + reason_codes = [] + else: + coverage = "complete_measured" + reason_codes = [] + totals = { + field: sum(item[field] for item in window) + for field in ( + "request_count", + "uncached_input_tokens", + "cached_input_tokens", + "output_tokens", + ) + } + totals["tokens"] = ( + totals["uncached_input_tokens"] + + totals["cached_input_tokens"] + + totals["output_tokens"] + ) + return { + "window_seconds": int(_WINDOW.total_seconds()), + "window_started_at": (datetime.fromisoformat(observed_at) - _WINDOW) + .isoformat() + .replace("+00:00", "Z"), + "observed_at": observed_at, + "coverage": coverage, + "reason_codes": reason_codes, + "sample_count": len(window), + **totals, + } + + +def _reservation_totals(reservations: list[dict[str, Any]]) -> dict[str, int]: + solver = sum( + item["tokens_per_minute"] + for item in reservations + if item["workload_class"] == "solver" + ) + evaluator = sum( + item["tokens_per_minute"] + for item in reservations + if item["workload_class"] == "evaluator" + ) + return {"total": solver + evaluator, "solver": solver, "evaluator": evaluator} + + +def build_benchmark_token_capacity_status( + state: Mapping[str, Any] | None, + *, + observed_at: str | None = None, +) -> dict[str, Any]: + timestamp = _timestamp(observed_at or _now_iso(), field="observed_at") + if state is None: + return { + "ok": True, + "schema_version": BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION, + "configured": False, + "observed_at": timestamp, + "next_action": "configure_token_capacity", + "path_recorded": False, + } + normalized = normalize_benchmark_token_capacity_state(state) + policy = normalized["policy"] + usage = _usage_projection(normalized["samples"], observed_at=timestamp) + reservations = _reservation_totals(normalized["reservations"]) + target = _target_tokens(policy) + solver_limit = ( + target + - policy["cold_start_reserve_tokens_per_minute"] + - policy["evaluator_reserve_tokens_per_minute"] + ) + evaluator_limit = target - policy["cold_start_reserve_tokens_per_minute"] + cooldown = normalized["cooldown"] + cooldown_active = bool( + cooldown + and datetime.fromisoformat(timestamp) + < datetime.fromisoformat(cooldown["until"]) + ) + global_basis = max(usage["tokens"], reservations["total"]) + overcommitted = ( + global_basis > evaluator_limit or reservations["solver"] > solver_limit + ) + return { + "ok": True, + "schema_version": BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION, + "configured": True, + "configured_at": normalized["configured_at"], + "updated_at": normalized["updated_at"], + "observed_at": timestamp, + "policy": policy, + "limits": { + "target_tokens_per_minute": target, + "solver_tokens_per_minute": solver_limit, + "evaluator_tokens_per_minute": evaluator_limit, + }, + "usage": usage, + "reservations": { + **reservations, + "count": len(normalized["reservations"]), + "active": normalized["reservations"], + }, + "capacity_basis_tokens_per_minute": global_basis, + "overcommitted": overcommitted, + "admission_ready": ( + not usage["reason_codes"] and not cooldown_active and not overcommitted + ), + "cooldown": { + "active": cooldown_active, + "until": cooldown["until"] if cooldown_active and cooldown else None, + }, + "next_action": ( + "wait_for_metering" + if usage["reason_codes"] + else "wait_for_provider_cooldown" + if cooldown_active + else "release_or_reduce_token_load" + if overcommitted + else "evaluate_token_admission" + ), + "path_recorded": False, + } + + +def configure_benchmark_token_capacity( + path: str | Path, + policy: Mapping[str, Any], + *, + execute: bool = False, + agent_id: str | None = None, + observed_at: str | None = None, +) -> dict[str, Any]: + target = Path(path).expanduser() + timestamp = _timestamp(observed_at or _now_iso(), field="observed_at") + compact_policy = normalize_benchmark_token_capacity_policy(policy) + + def build( + current: Mapping[str, Any] | None, + ) -> tuple[dict[str, Any], dict[str, Any]]: + if current is None: + configured_at = timestamp + samples: list[dict[str, Any]] = [] + reservations: list[dict[str, Any]] = [] + cooldown = None + else: + normalized = normalize_benchmark_token_capacity_state(current) + if datetime.fromisoformat(timestamp) < datetime.fromisoformat( + normalized["updated_at"] + ): + raise ValueError("configuration time cannot move state backward") + configured_at = normalized["configured_at"] + samples = normalized["samples"] + reservations = normalized["reservations"] + cooldown = normalized["cooldown"] + next_state = normalize_benchmark_token_capacity_state( + { + "schema_version": BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION, + "configured_at": configured_at, + "updated_at": timestamp, + "policy": compact_policy, + "samples": samples, + "reservations": reservations, + "cooldown": cooldown, + } + ) + return ( + { + **build_benchmark_token_capacity_status( + next_state, observed_at=timestamp + ), + "dry_run": not execute, + "write_performed": False, + }, + next_state, + ) + + if not execute: + result, _ = build(read_benchmark_token_capacity_state(target)) + return result + with exclusive_file_lock( + target, agent_id=agent_id, operation="benchmark_token_capacity_configure" + ): + result, next_state = build(read_benchmark_token_capacity_state(target)) + atomic_write_json(target, next_state) + result["write_performed"] = True + return result + + +def observe_benchmark_token_usage( + path: str | Path, + sample: Mapping[str, Any], + *, + execute: bool = False, + agent_id: str | None = None, + recorded_at: str | None = None, +) -> dict[str, Any]: + target = Path(path).expanduser() + compact_sample = normalize_benchmark_token_usage_sample(sample) + timestamp = _timestamp(recorded_at or _now_iso(), field="recorded_at") + + def build( + current: Mapping[str, Any] | None, + ) -> tuple[dict[str, Any], dict[str, Any] | None]: + if current is None: + return ( + { + "ok": False, + "reason_codes": ["token_capacity_not_configured"], + "dry_run": not execute, + "write_performed": False, + }, + None, + ) + normalized = normalize_benchmark_token_capacity_state(current) + existing = next( + ( + item + for item in normalized["samples"] + if item["sample_id"] == compact_sample["sample_id"] + ), + None, + ) + if existing is not None and existing != compact_sample: + raise ValueError("sample_id is already recorded with different usage") + if existing is not None: + readback_at = ( + max( + datetime.fromisoformat(timestamp), + datetime.fromisoformat(normalized["updated_at"]), + ) + .isoformat() + .replace("+00:00", "Z") + ) + status = build_benchmark_token_capacity_status( + normalized, observed_at=readback_at + ) + return ( + { + **status, + "recorded": True, + "idempotent": True, + "dry_run": not execute, + "write_performed": False, + }, + None, + ) + latest = max( + datetime.fromisoformat(normalized["updated_at"]), + datetime.fromisoformat(timestamp), + ) + latest_at = latest.isoformat().replace("+00:00", "Z") + lower = latest - _WINDOW + samples = [ + item + for item in [*normalized["samples"], compact_sample] + if datetime.fromisoformat(item["observed_at"]) > lower + ] + next_state = normalize_benchmark_token_capacity_state( + { + **normalized, + "updated_at": latest_at, + "samples": samples, + } + ) + return ( + { + **build_benchmark_token_capacity_status( + next_state, observed_at=latest_at + ), + "recorded": True, + "sample": compact_sample, + "dry_run": not execute, + "write_performed": False, + }, + next_state, + ) + + if not execute: + result, _ = build(read_benchmark_token_capacity_state(target)) + return result + with exclusive_file_lock( + target, agent_id=agent_id, operation="benchmark_token_usage_observe" + ): + result, next_state = build(read_benchmark_token_capacity_state(target)) + if next_state is not None: + atomic_write_json(target, next_state) + result["write_performed"] = True + return result + + +def record_benchmark_token_capacity_rejection( + path: str | Path, + *, + retry_after_seconds: int | None = None, + execute: bool = False, + agent_id: str | None = None, + observed_at: str | None = None, +) -> dict[str, Any]: + target = Path(path).expanduser() + timestamp = _timestamp(observed_at or _now_iso(), field="observed_at") + if retry_after_seconds is not None: + retry_after_seconds = _bounded_token_int( + retry_after_seconds, + field="retry_after_seconds", + minimum=1, + maximum=_MAX_COOLDOWN_SECONDS, + ) + + def build( + current: Mapping[str, Any] | None, + ) -> tuple[dict[str, Any], dict[str, Any] | None]: + if current is None: + return ( + { + "ok": False, + "reason_codes": ["token_capacity_not_configured"], + "dry_run": not execute, + "write_performed": False, + }, + None, + ) + normalized = normalize_benchmark_token_capacity_state(current) + if datetime.fromisoformat(timestamp) < datetime.fromisoformat( + normalized["updated_at"] + ): + raise ValueError("rejection time cannot move state backward") + seconds = max( + normalized["policy"]["cooldown_seconds"], retry_after_seconds or 0 + ) + until = ( + (datetime.fromisoformat(timestamp) + timedelta(seconds=seconds)) + .isoformat() + .replace("+00:00", "Z") + ) + existing = normalized["cooldown"] + if existing and datetime.fromisoformat( + existing["until"] + ) > datetime.fromisoformat(until): + until = existing["until"] + next_state = normalize_benchmark_token_capacity_state( + { + **normalized, + "updated_at": timestamp, + "cooldown": {"last_rejected_at": timestamp, "until": until}, + } + ) + return ( + { + **build_benchmark_token_capacity_status( + next_state, observed_at=timestamp + ), + "recorded": True, + "dry_run": not execute, + "write_performed": False, + }, + next_state, + ) + + if not execute: + result, _ = build(read_benchmark_token_capacity_state(target)) + return result + with exclusive_file_lock( + target, agent_id=agent_id, operation="benchmark_token_capacity_reject" + ): + result, next_state = build(read_benchmark_token_capacity_state(target)) + if next_state is not None: + atomic_write_json(target, next_state) + result["write_performed"] = True + return result + + +def _headroom_receipt(*, observed_at: str, state: str) -> dict[str, Any]: + expires_at = ( + (datetime.fromisoformat(observed_at) + timedelta(minutes=1)) + .isoformat() + .replace("+00:00", "Z") + ) + return { + "schema_version": BENCHMARK_RESOURCE_HEADROOM_RECEIPT_SCHEMA_VERSION, + "observed_at": observed_at, + "expires_at": expires_at, + "checks": [{"kind": "provider_capacity", "state": state}], + } + + +def _token_admission( + state: Mapping[str, Any] | None, + *, + run_id: str, + workload_class: str, + tokens_per_minute: int, + admitted_at: str, +) -> tuple[dict[str, Any], dict[str, Any] | None]: + compact_run_id = _token(run_id, field="run_id") + compact_workload_class = _token(workload_class, field="workload_class") + if compact_workload_class not in _WORKLOAD_CLASSES: + raise ValueError("workload_class is unsupported") + compact_tpm = _bounded_token_int( + tokens_per_minute, field="tokens_per_minute", minimum=1 + ) + if state is None: + return ( + { + "ok": False, + "admitted": False, + "reason_codes": ["token_capacity_not_configured"], + "write_performed": False, + }, + None, + ) + normalized = normalize_benchmark_token_capacity_state(state) + if datetime.fromisoformat(admitted_at) < datetime.fromisoformat( + normalized["updated_at"] + ): + raise ValueError("admission time cannot move state backward") + existing = next( + ( + item + for item in normalized["reservations"] + if item["run_id"] == compact_run_id + ), + None, + ) + if existing is not None and ( + existing["workload_class"] != compact_workload_class + or existing["tokens_per_minute"] != compact_tpm + ): + raise ValueError("run_id is already reserved with different token capacity") + status = build_benchmark_token_capacity_status(normalized, observed_at=admitted_at) + reasons = list(status["usage"]["reason_codes"]) + if status["cooldown"]["active"]: + reasons.append("provider_capacity_cooldown") + reservations = normalized["reservations"] + current = _reservation_totals(reservations) + projected_reservations = reservations + if existing is None: + projected_reservations = [ + *reservations, + { + "run_id": compact_run_id, + "workload_class": compact_workload_class, + "tokens_per_minute": compact_tpm, + "reserved_at": admitted_at, + }, + ] + projected = _reservation_totals(projected_reservations) + global_limit = status["limits"]["evaluator_tokens_per_minute"] + added_tpm = 0 if existing is not None else compact_tpm + global_basis = max(status["usage"]["tokens"], current["total"]) + added_tpm + if compact_workload_class == "evaluator": + capacity_basis = global_basis + limit = global_limit + else: + solver_added_tpm = 0 if existing is not None else compact_tpm + capacity_basis = ( + max(status["usage"]["tokens"], current["solver"]) + solver_added_tpm + ) + limit = status["limits"]["solver_tokens_per_minute"] + if global_basis > global_limit or capacity_basis > limit: + reasons.append("token_capacity_exhausted") + reasons = sorted(set(reasons)) + headroom_state = ( + "unresolved" + if any( + reason + in { + "token_usage_unobserved", + "token_usage_partial_unmetered", + "token_usage_sample_from_future", + } + for reason in reasons + ) + else "insufficient" + if reasons + else "sufficient" + ) + decision = { + "ok": not reasons, + "schema_version": BENCHMARK_TOKEN_CAPACITY_DECISION_SCHEMA_VERSION, + "admitted": not reasons, + "idempotent": existing is not None, + "reason_codes": reasons, + "run": { + "run_id": compact_run_id, + "workload_class": compact_workload_class, + "tokens_per_minute": compact_tpm, + }, + "observed_at": admitted_at, + "coverage": status["usage"]["coverage"], + "rolling_usage_tokens_per_minute": status["usage"]["tokens"], + "projected_reserved_tokens_per_minute": projected["total"], + "projected_solver_tokens_per_minute": projected["solver"], + "projected_evaluator_tokens_per_minute": projected["evaluator"], + "capacity_basis_tokens_per_minute": capacity_basis, + "admission_limit_tokens_per_minute": limit, + "resource_headroom_receipt": _headroom_receipt( + observed_at=admitted_at, state=headroom_state + ), + "write_performed": False, + } + if reasons or existing is not None: + return decision, None + next_state = normalize_benchmark_token_capacity_state( + { + **normalized, + "updated_at": admitted_at, + "reservations": projected_reservations, + } + ) + return decision, next_state + + +def admit_benchmark_token_capacity( + path: str | Path, + *, + run_id: str, + workload_class: str, + tokens_per_minute: int, + execute: bool = False, + agent_id: str | None = None, + admitted_at: str | None = None, +) -> dict[str, Any]: + target = Path(path).expanduser() + timestamp = _timestamp(admitted_at or _now_iso(), field="admitted_at") + + def build( + current: Mapping[str, Any] | None, + ) -> tuple[dict[str, Any], dict[str, Any] | None]: + decision, next_state = _token_admission( + current, + run_id=run_id, + workload_class=workload_class, + tokens_per_minute=tokens_per_minute, + admitted_at=timestamp, + ) + decision["dry_run"] = not execute + return decision, next_state + + if not execute: + result, _ = build(read_benchmark_token_capacity_state(target)) + return result + with exclusive_file_lock( + target, agent_id=agent_id, operation="benchmark_token_capacity_admit" + ): + result, next_state = build(read_benchmark_token_capacity_state(target)) + if next_state is not None: + atomic_write_json(target, next_state) + result["write_performed"] = True + return result + + +def release_benchmark_token_capacity( + path: str | Path, + *, + run_id: str, + execute: bool = False, + agent_id: str | None = None, + released_at: str | None = None, +) -> dict[str, Any]: + target = Path(path).expanduser() + compact_run_id = _token(run_id, field="run_id") + timestamp = _timestamp(released_at or _now_iso(), field="released_at") + + def build( + current: Mapping[str, Any] | None, + ) -> tuple[dict[str, Any], dict[str, Any] | None]: + if current is None: + return ( + { + "ok": False, + "released": False, + "reason_codes": ["token_capacity_not_configured"], + "dry_run": not execute, + "write_performed": False, + }, + None, + ) + normalized = normalize_benchmark_token_capacity_state(current) + if datetime.fromisoformat(timestamp) < datetime.fromisoformat( + normalized["updated_at"] + ): + raise ValueError("release time cannot move state backward") + retained = [ + item + for item in normalized["reservations"] + if item["run_id"] != compact_run_id + ] + if len(retained) == len(normalized["reservations"]): + return ( + { + "ok": True, + "released": True, + "idempotent": True, + "reason_codes": [], + "dry_run": not execute, + "write_performed": False, + }, + None, + ) + next_state = normalize_benchmark_token_capacity_state( + { + **normalized, + "updated_at": timestamp, + "reservations": retained, + } + ) + return ( + { + "ok": True, + "released": True, + "idempotent": False, + "reason_codes": [], + "status": build_benchmark_token_capacity_status( + next_state, observed_at=timestamp + ), + "dry_run": not execute, + "write_performed": False, + }, + next_state, + ) + + if not execute: + result, _ = build(read_benchmark_token_capacity_state(target)) + return result + with exclusive_file_lock( + target, agent_id=agent_id, operation="benchmark_token_capacity_release" + ): + result, next_state = build(read_benchmark_token_capacity_state(target)) + if next_state is not None: + atomic_write_json(target, next_state) + result["write_performed"] = True + return result + + +def render_benchmark_token_capacity_markdown(payload: dict[str, Any]) -> str: + lines = ["# Benchmark token capacity", ""] + for field in ( + "configured", + "admitted", + "released", + "recorded", + "coverage", + "capacity_basis_tokens_per_minute", + "admission_limit_tokens_per_minute", + "next_action", + "write_performed", + ): + if field in payload: + lines.append(f"- {field}: {payload[field]}") + if payload.get("reason_codes"): + lines.append(f"- reason_codes: {', '.join(payload['reason_codes'])}") + usage = payload.get("usage") + if isinstance(usage, Mapping): + lines.append(f"- coverage: {usage.get('coverage')}") + lines.append(f"- rolling_tokens_per_minute: {usage.get('tokens')}") + return "\n".join(lines) + "\n" + + +__all__ = [ + "BENCHMARK_TOKEN_CAPACITY_DECISION_SCHEMA_VERSION", + "BENCHMARK_TOKEN_CAPACITY_FILENAME", + "BENCHMARK_TOKEN_CAPACITY_POLICY_SCHEMA_VERSION", + "BENCHMARK_TOKEN_CAPACITY_STATE_SCHEMA_VERSION", + "BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION", + "admit_benchmark_token_capacity", + "build_benchmark_token_capacity_policy", + "build_benchmark_token_capacity_status", + "configure_benchmark_token_capacity", + "default_benchmark_token_capacity_path", + "normalize_benchmark_token_capacity_policy", + "normalize_benchmark_token_capacity_state", + "normalize_benchmark_token_usage_sample", + "observe_benchmark_token_usage", + "read_benchmark_token_capacity_state", + "record_benchmark_token_capacity_rejection", + "release_benchmark_token_capacity", + "render_benchmark_token_capacity_markdown", +] diff --git a/loopx/cli_commands/benchmark_dispatch.py b/loopx/cli_commands/benchmark_dispatch.py index c802b18e57..52520a8779 100644 --- a/loopx/cli_commands/benchmark_dispatch.py +++ b/loopx/cli_commands/benchmark_dispatch.py @@ -35,6 +35,11 @@ handle_benchmark_study_command, register_benchmark_study_commands, ) +from .benchmark_token_capacity import ( + BENCHMARK_TOKEN_CAPACITY_COMMANDS, + handle_benchmark_token_capacity_command, + register_benchmark_token_capacity_commands, +) from .benchmark_trajectory import ( handle_benchmark_trajectory_command, @@ -46,7 +51,9 @@ PrintPayload = Callable[..., None] BENCHMARK_PROJECT_COMMANDS = ( - BENCHMARK_CONCURRENCY_COMMANDS | BENCHMARK_EXPERIMENT_BOARD_COMMANDS + BENCHMARK_CONCURRENCY_COMMANDS + | BENCHMARK_EXPERIMENT_BOARD_COMMANDS + | BENCHMARK_TOKEN_CAPACITY_COMMANDS ) @@ -97,6 +104,8 @@ def _benchmark_project_parser(args: argparse.Namespace) -> argparse.ArgumentPars return args.benchmark_rebalance_parser if args.benchmark_command in BENCHMARK_CONCURRENCY_COMMANDS: return args.benchmark_concurrency_parser + if args.benchmark_command in BENCHMARK_TOKEN_CAPACITY_COMMANDS: + return args.benchmark_token_capacity_parser return args.benchmark_experiment_board_parser @@ -115,6 +124,9 @@ def register_benchmark_command_group( register_benchmark_boundary_commands(benchmark_sub, add_subcommand_format) register_benchmark_continuation_commands(benchmark_sub, add_subcommand_format) register_benchmark_concurrency_commands(benchmark_sub, add_subcommand_format) + register_benchmark_token_capacity_commands( + benchmark_sub, add_subcommand_format + ) register_benchmark_rebalance_commands(benchmark_sub, add_subcommand_format) register_benchmark_experiment_board_commands( benchmark_sub, @@ -176,6 +188,14 @@ def handle_benchmark_command( print_payload=print_payload, output_format=output_format, ) + if handled is not None: + return handled + handled = handle_benchmark_token_capacity_command( + args, + project=project, + print_payload=print_payload, + output_format=output_format, + ) if handled is not None: return handled return handle_benchmark_experiment_board_command( diff --git a/loopx/cli_commands/benchmark_token_capacity.py b/loopx/cli_commands/benchmark_token_capacity.py new file mode 100644 index 0000000000..8a591e7b54 --- /dev/null +++ b/loopx/cli_commands/benchmark_token_capacity.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +import argparse +import json +import sys +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from ..capabilities.benchmark_toolkit import ( + admit_benchmark_token_capacity, + build_benchmark_token_capacity_policy, + build_benchmark_token_capacity_status, + configure_benchmark_token_capacity, + default_benchmark_token_capacity_path, + observe_benchmark_token_usage, + read_benchmark_token_capacity_state, + record_benchmark_token_capacity_rejection, + release_benchmark_token_capacity, + render_benchmark_token_capacity_markdown, +) + +PrintPayload = Callable[ + [dict[str, object], str, Callable[[dict[str, object]], str]], None +] +OutputFormat = Callable[[argparse.Namespace], str] + +BENCHMARK_TOKEN_CAPACITY_COMMANDS = { + "token-capacity-status", + "token-capacity-configure", + "token-capacity-observe", + "token-capacity-reject", + "token-capacity-admit", + "token-capacity-release", +} + + +def _add_location_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--goal-id", required=True) + parser.add_argument( + "--project", + help="Explicit project root. Defaults to the registry goal repository.", + ) + parser.set_defaults(benchmark_token_capacity_parser=parser) + + +def _add_execute_argument(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--execute", + action="store_true", + help="Write project-local domain state. Without this flag, preview only.", + ) + parser.add_argument( + "--agent-id", + help="Optional public-safe actor id recorded only in lock-holder diagnostics.", + ) + + +def register_benchmark_token_capacity_commands( + benchmark_subparsers: argparse._SubParsersAction, + add_subcommand_format: Callable[[argparse.ArgumentParser], None], +) -> None: + status = benchmark_subparsers.add_parser( + "token-capacity-status", + help="Read rolling token usage, reservations, and provider cooldown.", + ) + add_subcommand_format(status) + _add_location_arguments(status) + + configure = benchmark_subparsers.add_parser( + "token-capacity-configure", + help="Preview or configure a provider-neutral tokens-per-minute envelope.", + ) + add_subcommand_format(configure) + _add_location_arguments(configure) + configure.add_argument("--quota-tokens-per-minute", required=True, type=int) + configure.add_argument("--target-utilization-basis-points", type=int, default=9000) + configure.add_argument( + "--cold-start-reserve-tokens-per-minute", type=int, default=0 + ) + configure.add_argument("--evaluator-reserve-tokens-per-minute", type=int, default=0) + configure.add_argument("--cooldown-seconds", type=int, default=60) + _add_execute_argument(configure) + + observe = benchmark_subparsers.add_parser( + "token-capacity-observe", + help="Record one typed, request-level token usage sample.", + ) + add_subcommand_format(observe) + _add_location_arguments(observe) + observe.add_argument( + "--sample-json", + required=True, + help="benchmark_token_usage_sample_v0 JSON path, or - for stdin.", + ) + _add_execute_argument(observe) + + reject = benchmark_subparsers.add_parser( + "token-capacity-reject", + help="Record a provider capacity rejection and start a bounded cooldown.", + ) + add_subcommand_format(reject) + _add_location_arguments(reject) + reject.add_argument("--retry-after-seconds", type=int) + _add_execute_argument(reject) + + admit = benchmark_subparsers.add_parser( + "token-capacity-admit", + help="Atomically reserve forecast token capacity before case admission.", + ) + add_subcommand_format(admit) + _add_location_arguments(admit) + admit.add_argument("--run-id", required=True) + admit.add_argument( + "--workload-class", required=True, choices=("solver", "evaluator") + ) + admit.add_argument("--tokens-per-minute", required=True, type=int) + _add_execute_argument(admit) + + release = benchmark_subparsers.add_parser( + "token-capacity-release", + help="Release one exact run's token capacity reservation.", + ) + add_subcommand_format(release) + _add_location_arguments(release) + release.add_argument("--run-id", required=True) + _add_execute_argument(release) + + +def _read_json_argument(value: str) -> Any: + if value == "-": + return json.load(sys.stdin) + return json.loads(Path(value).expanduser().read_text(encoding="utf-8")) + + +def handle_benchmark_token_capacity_command( + args: argparse.Namespace, + *, + project: Path | None, + print_payload: PrintPayload, + output_format: OutputFormat, +) -> int | None: + if args.benchmark_command not in BENCHMARK_TOKEN_CAPACITY_COMMANDS: + return None + try: + if project is None: + raise ValueError("benchmark project route is required") + path = default_benchmark_token_capacity_path( + project=project, goal_id=args.goal_id + ) + if args.benchmark_command == "token-capacity-status": + payload: dict[str, Any] = build_benchmark_token_capacity_status( + read_benchmark_token_capacity_state(path) + ) + elif args.benchmark_command == "token-capacity-configure": + payload = configure_benchmark_token_capacity( + path, + build_benchmark_token_capacity_policy( + quota_tokens_per_minute=args.quota_tokens_per_minute, + target_utilization_basis_points=( + args.target_utilization_basis_points + ), + cold_start_reserve_tokens_per_minute=( + args.cold_start_reserve_tokens_per_minute + ), + evaluator_reserve_tokens_per_minute=( + args.evaluator_reserve_tokens_per_minute + ), + cooldown_seconds=args.cooldown_seconds, + ), + execute=args.execute, + agent_id=args.agent_id, + ) + elif args.benchmark_command == "token-capacity-observe": + payload = observe_benchmark_token_usage( + path, + _read_json_argument(args.sample_json), + execute=args.execute, + agent_id=args.agent_id, + ) + elif args.benchmark_command == "token-capacity-reject": + payload = record_benchmark_token_capacity_rejection( + path, + retry_after_seconds=args.retry_after_seconds, + execute=args.execute, + agent_id=args.agent_id, + ) + elif args.benchmark_command == "token-capacity-admit": + payload = admit_benchmark_token_capacity( + path, + run_id=args.run_id, + workload_class=args.workload_class, + tokens_per_minute=args.tokens_per_minute, + execute=args.execute, + agent_id=args.agent_id, + ) + else: + payload = release_benchmark_token_capacity( + path, + run_id=args.run_id, + execute=args.execute, + agent_id=args.agent_id, + ) + payload["goal_id"] = args.goal_id + payload["path_recorded"] = False + except (OSError, TypeError, ValueError): + args.benchmark_token_capacity_parser.error( + "invalid benchmark token capacity input" + ) + print_payload( + payload, output_format(args), render_benchmark_token_capacity_markdown + ) + return 0 if payload.get("ok") is not False else 2 diff --git a/tests/capabilities/test_benchmark_token_capacity.py b/tests/capabilities/test_benchmark_token_capacity.py new file mode 100644 index 0000000000..a92efbc7af --- /dev/null +++ b/tests/capabilities/test_benchmark_token_capacity.py @@ -0,0 +1,502 @@ +from __future__ import annotations + +import json +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from loopx.capabilities.benchmark_toolkit import ( + BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION, + admit_benchmark_token_capacity, + build_benchmark_token_capacity_policy, + build_benchmark_token_capacity_status, + configure_benchmark_token_capacity, + normalize_benchmark_resource_headroom_receipt, + observe_benchmark_token_usage, + read_benchmark_token_capacity_state, + record_benchmark_token_capacity_rejection, + release_benchmark_token_capacity, +) +from tests.source_cli import run_source_cli + + +def _sample( + sample_id: str, + observed_at: datetime, + *, + metering_state: str = "measured", + requests: int = 1, + uncached: int = 0, + cached: int = 0, + output: int = 0, +) -> dict[str, object]: + return { + "schema_version": BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION, + "sample_id": sample_id, + "observed_at": observed_at.isoformat(), + "metering_state": metering_state, + "request_count": requests, + "uncached_input_tokens": uncached, + "cached_input_tokens": cached, + "output_tokens": output, + } + + +def _configure(path: Path, *, now: datetime) -> None: + configured = configure_benchmark_token_capacity( + path, + build_benchmark_token_capacity_policy( + quota_tokens_per_minute=10_000_000, + target_utilization_basis_points=9000, + cold_start_reserve_tokens_per_minute=500_000, + evaluator_reserve_tokens_per_minute=1_000_000, + cooldown_seconds=90, + ), + execute=True, + observed_at=now.isoformat(), + ) + assert configured["write_performed"] is True + + +def test_exact_window_and_disjoint_cache_accounting(tmp_path: Path) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=2)) + + samples = [ + _sample( + "expired-boundary", + now - timedelta(seconds=60), + uncached=9_000_000, + ), + _sample( + "inside-window", + now - timedelta(seconds=59), + uncached=1_000_000, + cached=2_000_000, + output=500_000, + ), + _sample("coverage-heartbeat", now, requests=0), + ] + for sample in samples: + result = observe_benchmark_token_usage( + path, sample, execute=True, recorded_at=now.isoformat() + ) + assert result["recorded"] is True + + status = build_benchmark_token_capacity_status( + read_benchmark_token_capacity_state(path), observed_at=now.isoformat() + ) + assert status["usage"] == { + "window_seconds": 60, + "window_started_at": (now - timedelta(seconds=60)) + .isoformat() + .replace("+00:00", "Z"), + "observed_at": now.isoformat().replace("+00:00", "Z"), + "coverage": "complete_measured", + "reason_codes": [], + "sample_count": 2, + "request_count": 1, + "uncached_input_tokens": 1_000_000, + "cached_input_tokens": 2_000_000, + "output_tokens": 500_000, + "tokens": 3_500_000, + } + + +def test_partial_unmetered_traffic_fails_closed(tmp_path: Path) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=1)) + observe_benchmark_token_usage( + path, + _sample("measured", now, uncached=100), + execute=True, + recorded_at=now.isoformat(), + ) + observe_benchmark_token_usage( + path, + _sample("unknown", now, metering_state="unmetered"), + execute=True, + recorded_at=now.isoformat(), + ) + + decision = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=1_000_000, + execute=True, + admitted_at=now.isoformat(), + ) + assert decision["admitted"] is False + assert decision["reason_codes"] == ["token_usage_partial_unmetered"] + receipt = normalize_benchmark_resource_headroom_receipt( + decision["resource_headroom_receipt"] + ) + assert receipt["checks"] == [{"kind": "provider_capacity", "state": "unresolved"}] + assert read_benchmark_token_capacity_state(path)["reservations"] == [] + + +def test_future_usage_sample_is_retained_and_fails_closed(tmp_path: Path) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=1)) + observed = observe_benchmark_token_usage( + path, + _sample("clock-skew", now + timedelta(seconds=1), uncached=100), + execute=True, + recorded_at=now.isoformat(), + ) + decision = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=1_000_000, + execute=True, + admitted_at=now.isoformat(), + ) + + assert observed["usage"]["coverage"] == "unresolved_future_sample" + assert decision["reason_codes"] == ["token_usage_sample_from_future"] + + +def test_complete_estimated_usage_is_counted_and_visible(tmp_path: Path) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=1)) + observe_benchmark_token_usage( + path, + _sample( + "estimated", + now, + metering_state="estimated", + uncached=1_000_000, + cached=500_000, + output=500_000, + ), + execute=True, + recorded_at=now.isoformat(), + ) + decision = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=1_000_000, + execute=True, + admitted_at=now.isoformat(), + ) + + assert decision["admitted"] is True + assert decision["coverage"] == "complete_estimated" + assert decision["rolling_usage_tokens_per_minute"] == 2_000_000 + + +def test_429_cooldown_holds_then_recovers_without_losing_state( + tmp_path: Path, +) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=1)) + observe_benchmark_token_usage( + path, + _sample("healthy", now, requests=0), + execute=True, + recorded_at=now.isoformat(), + ) + rejected = record_benchmark_token_capacity_rejection( + path, + retry_after_seconds=120, + execute=True, + observed_at=now.isoformat(), + ) + assert rejected["cooldown"] == { + "active": True, + "until": (now + timedelta(seconds=120)).isoformat().replace("+00:00", "Z"), + } + + during_cooldown = now + timedelta(seconds=60) + observe_benchmark_token_usage( + path, + _sample("still-metered", during_cooldown, requests=0), + execute=True, + recorded_at=during_cooldown.isoformat(), + ) + held = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=1_000_000, + execute=True, + admitted_at=during_cooldown.isoformat(), + ) + assert held["reason_codes"] == ["provider_capacity_cooldown"] + + recovered_at = now + timedelta(seconds=121) + observe_benchmark_token_usage( + path, + _sample("recovered", recovered_at, requests=0), + execute=True, + recorded_at=recovered_at.isoformat(), + ) + admitted = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=1_000_000, + execute=True, + admitted_at=recovered_at.isoformat(), + ) + assert admitted["admitted"] is True + assert admitted["write_performed"] is True + + +def test_evaluator_reserve_cannot_be_consumed_by_solver(tmp_path: Path) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=1)) + observe_benchmark_token_usage( + path, + _sample("healthy", now, requests=0), + execute=True, + recorded_at=now.isoformat(), + ) + + solver = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=7_500_000, + execute=True, + admitted_at=now.isoformat(), + ) + blocked_solver = admit_benchmark_token_capacity( + path, + run_id="solver-b", + workload_class="solver", + tokens_per_minute=1, + execute=True, + admitted_at=now.isoformat(), + ) + evaluator = admit_benchmark_token_capacity( + path, + run_id="evaluator-a", + workload_class="evaluator", + tokens_per_minute=1_000_000, + execute=True, + admitted_at=now.isoformat(), + ) + + assert solver["admitted"] is True + assert solver["admission_limit_tokens_per_minute"] == 7_500_000 + assert blocked_solver["reason_codes"] == ["token_capacity_exhausted"] + assert evaluator["admitted"] is True + assert evaluator["admission_limit_tokens_per_minute"] == 8_500_000 + + +def test_restart_preserves_reservations_and_idempotent_release(tmp_path: Path) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=1)) + observe_benchmark_token_usage( + path, + _sample("healthy", now, requests=0), + execute=True, + recorded_at=now.isoformat(), + ) + first = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=2_000_000, + execute=True, + admitted_at=now.isoformat(), + ) + restored = read_benchmark_token_capacity_state(path) + repeated = admit_benchmark_token_capacity( + path, + run_id="solver-a", + workload_class="solver", + tokens_per_minute=2_000_000, + execute=True, + admitted_at=now.isoformat(), + ) + status = build_benchmark_token_capacity_status( + restored, observed_at=now.isoformat() + ) + + assert first["write_performed"] is True + assert repeated["idempotent"] is True + assert repeated["write_performed"] is False + assert status["reservations"]["total"] == 2_000_000 + assert status["capacity_basis_tokens_per_minute"] == 2_000_000 + + released = release_benchmark_token_capacity( + path, + run_id="solver-a", + execute=True, + released_at=(now + timedelta(seconds=1)).isoformat(), + ) + repeated_release = release_benchmark_token_capacity( + path, + run_id="solver-a", + execute=True, + released_at=(now + timedelta(seconds=2)).isoformat(), + ) + assert released["write_performed"] is True + assert repeated_release["idempotent"] is True + + +def test_parallel_admission_is_atomic_at_token_limit(tmp_path: Path) -> None: + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + path = tmp_path / "token-capacity.json" + _configure(path, now=now - timedelta(minutes=1)) + observe_benchmark_token_usage( + path, + _sample("healthy", now, requests=0), + execute=True, + recorded_at=now.isoformat(), + ) + + def reserve(index: int) -> dict[str, object]: + return admit_benchmark_token_capacity( + path, + run_id=f"solver-{index}", + workload_class="solver", + tokens_per_minute=3_000_000, + execute=True, + admitted_at=now.isoformat(), + ) + + with ThreadPoolExecutor(max_workers=4) as executor: + decisions = list(executor.map(reserve, range(4))) + + assert sum(decision["admitted"] is True for decision in decisions) == 2 + state = read_benchmark_token_capacity_state(path) + assert state is not None + assert len(state["reservations"]) == 2 + + +def test_cli_composes_token_receipt_with_existing_headroom_contract( + tmp_path: Path, +) -> None: + now = datetime.now(UTC) + configured = run_source_cli( + [ + "benchmark", + "token-capacity-configure", + "--goal-id", + "fixture-goal", + "--project", + str(tmp_path), + "--quota-tokens-per-minute", + "10000000", + "--target-utilization-basis-points", + "9000", + "--evaluator-reserve-tokens-per-minute", + "1000000", + "--execute", + "--format", + "json", + ], + check=True, + ) + assert json.loads(configured.stdout)["write_performed"] is True + + sample_path = tmp_path / "sample.json" + sample_path.write_text( + json.dumps(_sample("cli-coverage", now, requests=0)), encoding="utf-8" + ) + observed = run_source_cli( + [ + "benchmark", + "token-capacity-observe", + "--goal-id", + "fixture-goal", + "--project", + str(tmp_path), + "--sample-json", + str(sample_path), + "--execute", + "--format", + "json", + ], + check=True, + ) + assert json.loads(observed.stdout)["usage"]["coverage"] == "complete_measured" + + admitted = run_source_cli( + [ + "benchmark", + "token-capacity-admit", + "--goal-id", + "fixture-goal", + "--project", + str(tmp_path), + "--run-id", + "solver-cli", + "--workload-class", + "solver", + "--tokens-per-minute", + "1000000", + "--execute", + "--format", + "json", + ], + check=True, + ) + payload = json.loads(admitted.stdout) + assert payload["admitted"] is True + assert payload["resource_headroom_receipt"]["checks"] == [ + {"kind": "provider_capacity", "state": "sufficient"} + ] + assert str(tmp_path) not in admitted.stdout + + run_source_cli( + [ + "benchmark", + "concurrency-configure", + "--goal-id", + "fixture-goal", + "--project", + str(tmp_path), + "--max-active-cases", + "1", + "--max-baseline-cases", + "1", + "--max-test-cases", + "1", + "--require-resource-headroom-receipt", + "--execute", + "--format", + "json", + ], + check=True, + ) + receipt_path = tmp_path / "headroom.json" + receipt_path.write_text( + json.dumps(payload["resource_headroom_receipt"]), encoding="utf-8" + ) + case_admission = run_source_cli( + [ + "benchmark", + "concurrency-admit", + "--goal-id", + "fixture-goal", + "--project", + str(tmp_path), + "--run-id", + "solver-cli", + "--case-id", + "case-cli", + "--arm-role", + "treatment", + "--resource-headroom-json", + str(receipt_path), + "--execute", + "--format", + "json", + ], + check=True, + ) + assert json.loads(case_admission.stdout)["admitted"] is True From 5e5ab17e375d144c0d59ca67eaa440a8e0f4586b Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 28 Oct 2026 18:16:49 +0800 Subject: [PATCH 2/2] fix(benchmark): serialize token clocks and scope optional guidance Signed-off-by: huangruiteng --- .../capabilities/benchmark_toolkit/README.md | 19 +- .../benchmark_toolkit/catalog_entry.py | 9 +- .../benchmark_toolkit/token_capacity.py | 11 +- .../references/repair-patterns.md | 1 + .../test_benchmark_token_capacity.py | 164 ++++++++++++++++++ 5 files changed, 189 insertions(+), 15 deletions(-) diff --git a/loopx/capabilities/benchmark_toolkit/README.md b/loopx/capabilities/benchmark_toolkit/README.md index 5860e57a73..db403198af 100644 --- a/loopx/capabilities/benchmark_toolkit/README.md +++ b/loopx/capabilities/benchmark_toolkit/README.md @@ -805,11 +805,13 @@ A countable experiment uses the toolkit in this order: task-goal hashes before preregistration. 2. Read the project experiment board before selecting or launching another case. 3. Read or configure the concurrency envelope and, when provider throughput is - shared, the token-capacity envelope; reconcile both reservations with exact + shared, the token-capacity envelope; reconcile configured reservations with exact runner liveness. 4. Declare a `run_permission_policy_v0` and preflight the runner boundary. -5. Atomically reserve token capacity, pass its fresh provider-capacity receipt to - case-slot admission, and launch only after both admissions succeed. +5. Atomically admit a case slot immediately before the independently authorized + runner launch. When a token envelope is configured, first record token usage, + reserve token capacity, and pass its fresh provider-capacity receipt to case-slot + admission; launch only after both admissions succeed. 6. Upsert the planned or running row, then launch one frozen case/arm; do not expose evaluator sources or official feedback. 7. Capture ATIF tool evidence and a runner-owned runtime attestation. @@ -822,8 +824,8 @@ A countable experiment uses the toolkit in this order: 11. Run the independent verifier only after the agent phase. 12. Reduce the official result through the benchmark-owned scoring path. 13. Upsert terminal score, countability, effort, treatment fidelity, and insight - status; release its case and token reservations; then read the matched-comparison - projection. + status; release its case reservation and, when configured, its token reservation; + then read the matched-comparison projection. 14. Apply attempt-countability, treatment-fidelity, and matched-pair gates before any comparison claim. @@ -906,8 +908,11 @@ envelope, and the receipt does not grant launch authority. ### Token capacity -When a provider quota is expressed in tokens per minute, configure the goal-scoped -token envelope as a second admission dimension. The target is an integer fraction +Token capacity is optional. Campaigns without a configured token envelope retain +case-slot admission without token observation, reservation, cooldown reporting, or +release. For a provider quota expressed in tokens per minute, configure the +goal-scoped token envelope as a second admission dimension; the following token +steps apply to that configured campaign. The target is an integer fraction of the hard quota. Cold-start reserve absorbs short request bursts, and evaluator reserve prevents solver backfill from consuming capacity needed to close completed runs: diff --git a/loopx/capabilities/benchmark_toolkit/catalog_entry.py b/loopx/capabilities/benchmark_toolkit/catalog_entry.py index 73107bcf80..6ee203099d 100644 --- a/loopx/capabilities/benchmark_toolkit/catalog_entry.py +++ b/loopx/capabilities/benchmark_toolkit/catalog_entry.py @@ -461,9 +461,9 @@ "backfill_when_concurrency_status_reports_underfilled", "qualify_source_revision_before_each_new_run_admission", "qualify_resource_headroom_when_the_envelope_requires_a_receipt", - "record_complete_or_explicitly_unmetered_token_usage", + "record_complete_or_explicitly_unmetered_token_usage_when_token_capacity_is_configured", "tune_target_occupancy_from_fresh_runner_feedback_when_opted_in", - "atomically_reserve_token_capacity_before_case_slot_admission", + "atomically_reserve_token_capacity_before_case_slot_admission_when_token_capacity_is_configured", "atomically_admit_case_slot_before_runner_launch", "upsert_preregistered_or_running_row_when_a_run_starts", "classify_exact_runtime_observation_during_active_monitor_cycles", @@ -471,7 +471,7 @@ "adjudicate_restricted_access_suspicion_after_solver_and_score_terminal", "upsert_terminal_score_countability_effort_and_insight_status", "release_case_slot_after_terminal_or_runner_invalid_transition", - "release_token_capacity_after_terminal_or_runner_invalid_transition", + "release_token_capacity_after_terminal_or_runner_invalid_transition_when_token_capacity_is_configured", "read_matched_comparisons_before_selecting_the_next_arm", ], "board_commands": { @@ -558,6 +558,9 @@ ), }, "token_capacity_boundary": ( + "Only campaigns with a configured token envelope require token usage " + "observation, reservation, cooldown reporting, and release; other " + "campaigns retain case-slot admission without token configuration. " "The token envelope owns aggregate metering, forecast reservations, and " "provider cooldown only. The runner owns provider adaptation, complete " "coverage, job discovery, launch, rollback, liveness, and termination." diff --git a/loopx/capabilities/benchmark_toolkit/token_capacity.py b/loopx/capabilities/benchmark_toolkit/token_capacity.py index ec470c2adc..bcd1cddeb7 100644 --- a/loopx/capabilities/benchmark_toolkit/token_capacity.py +++ b/loopx/capabilities/benchmark_toolkit/token_capacity.py @@ -463,12 +463,13 @@ def configure_benchmark_token_capacity( observed_at: str | None = None, ) -> dict[str, Any]: target = Path(path).expanduser() - timestamp = _timestamp(observed_at or _now_iso(), field="observed_at") compact_policy = normalize_benchmark_token_capacity_policy(policy) def build( current: Mapping[str, Any] | None, ) -> tuple[dict[str, Any], dict[str, Any]]: + # Executed builds sample implicit time only after acquiring the state lock. + timestamp = _timestamp(observed_at or _now_iso(), field="observed_at") if current is None: configured_at = timestamp samples: list[dict[str, Any]] = [] @@ -528,11 +529,11 @@ def observe_benchmark_token_usage( ) -> dict[str, Any]: target = Path(path).expanduser() compact_sample = normalize_benchmark_token_usage_sample(sample) - timestamp = _timestamp(recorded_at or _now_iso(), field="recorded_at") def build( current: Mapping[str, Any] | None, ) -> tuple[dict[str, Any], dict[str, Any] | None]: + timestamp = _timestamp(recorded_at or _now_iso(), field="recorded_at") if current is None: return ( { @@ -629,7 +630,6 @@ def record_benchmark_token_capacity_rejection( observed_at: str | None = None, ) -> dict[str, Any]: target = Path(path).expanduser() - timestamp = _timestamp(observed_at or _now_iso(), field="observed_at") if retry_after_seconds is not None: retry_after_seconds = _bounded_token_int( retry_after_seconds, @@ -641,6 +641,7 @@ def record_benchmark_token_capacity_rejection( def build( current: Mapping[str, Any] | None, ) -> tuple[dict[str, Any], dict[str, Any] | None]: + timestamp = _timestamp(observed_at or _now_iso(), field="observed_at") if current is None: return ( { @@ -853,11 +854,11 @@ def admit_benchmark_token_capacity( admitted_at: str | None = None, ) -> dict[str, Any]: target = Path(path).expanduser() - timestamp = _timestamp(admitted_at or _now_iso(), field="admitted_at") def build( current: Mapping[str, Any] | None, ) -> tuple[dict[str, Any], dict[str, Any] | None]: + timestamp = _timestamp(admitted_at or _now_iso(), field="admitted_at") decision, next_state = _token_admission( current, run_id=run_id, @@ -891,11 +892,11 @@ def release_benchmark_token_capacity( ) -> dict[str, Any]: target = Path(path).expanduser() compact_run_id = _token(run_id, field="run_id") - timestamp = _timestamp(released_at or _now_iso(), field="released_at") def build( current: Mapping[str, Any] | None, ) -> tuple[dict[str, Any], dict[str, Any] | None]: + timestamp = _timestamp(released_at or _now_iso(), field="released_at") if current is None: return ( { diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index 478c28a13b..76147d599b 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -5,6 +5,7 @@ teaches a reusable control-plane lesson. | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair | | --- | --- | --- | --- | --- | +| `implicit_clock_precedes_state_lock` | Concurrent state mutations reject a healthy request as a backward timestamp even though a fresh retry succeeds. | The exact operation, persisted `updated_at`, and a real-clock reproduction where the first caller pauses before acquiring the lock and a later caller commits first. | Implicit time was sampled before lock acquisition, so request order and commit order diverged. | Sample implicit mutation time inside the existing locked state transition. Preserve explicit backward-time rejection and observation-specific out-of-order rules. Cover every adjacent writer with reversed lock acquisition, unchanged-state negative cases, and the real CLI/file-lock path; do not clamp explicit input, weaken the guard, or add retries. | | `stride_inline_outcome_basis_ordering` | A batched milestone dry-run succeeds, but execution returns `stride_review_basis_mismatch` after an outcome refinement has committed. | Exact Turn/obligation/context, the returned `goal_outcome_update` status and operation, current acceptance and run/spend receipts; compare due Stride with ordinary batching. | Sequential refinement changes the context that the decision references; a preview using pre-update facts hides that change. | Feed the typed acceptance owner's prospective contract to dry-run planning. For a changed due-review milestone, refine before a fresh same-Turn guard, read the new facts and author the complete decision; unchanged work needs no update. On partial failure retain applied/replayed receipts and original IDs, reassess fresh context, then settle once. Never assume rollback, blindly replace a context token, lower the gate or replay work. Cover real File/SQLite preview/refusal/recovery, ordinary batching and single debit. | | `native_code_cell_source_normalization` | An optional native context adapter always falls back to full after an output-budget pragma, despite complete retained input. | The same live model call, native cell source, child identity and complete model-visible capture bytes; compare the executable source separately from native framing. | Native removes the documented first-line execution pragma before recording the cell, while the adapter compares that body with the unprocessed model call. | Normalize only that first-line JSON-object pragma and retain exact remaining source, live call/cell ordering and complete-byte verification. Cover unchanged framing, altered bodies, malformed/extra headers and stale or partial input, plus real native resume and recovery. Preserve automatic same-capture full fallback; do not add an admission or owner question. | | `native_history_output_budget` | A complete CLI or code-cell result becomes a clipped or missing page in the next model input despite a larger direct-result budget. | Compare the CLI result, completed native cell, actual inference input and effective model history-output policy at the same native version. | Direct-result and retained-history limits are separate; an unknown loopback model may use a smaller byte-based fallback than the intended host's token-based policy. | Keep the original low-budget failures. Qualify a distinct matched workload under the intended declared history budget without changing live host limits or frozen criteria. Size batches within both budgets, retain complete-byte checks and same-capture recovery, and never count incomplete trials as savings or runtime output as retention proof. | diff --git a/tests/capabilities/test_benchmark_token_capacity.py b/tests/capabilities/test_benchmark_token_capacity.py index a92efbc7af..5134d810b7 100644 --- a/tests/capabilities/test_benchmark_token_capacity.py +++ b/tests/capabilities/test_benchmark_token_capacity.py @@ -2,8 +2,14 @@ import json from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager from datetime import UTC, datetime, timedelta from pathlib import Path +from threading import Event + +import pytest + +from loopx.capabilities.benchmark_toolkit import token_capacity from loopx.capabilities.benchmark_toolkit import ( BENCHMARK_TOKEN_USAGE_SAMPLE_SCHEMA_VERSION, @@ -377,6 +383,103 @@ def reserve(index: int) -> dict[str, object]: assert len(state["reservations"]) == 2 +def _mutate(path: Path, operation: str, run_id: str, **kwargs): + if operation == "configure": + return configure_benchmark_token_capacity( + path, + build_benchmark_token_capacity_policy(quota_tokens_per_minute=10_000_000), + **kwargs, + ) + if operation == "reject": + return record_benchmark_token_capacity_rejection(path, **kwargs) + if operation == "release": + return release_benchmark_token_capacity(path, run_id=run_id, **kwargs) + return admit_benchmark_token_capacity( + path, run_id=run_id, workload_class="solver", tokens_per_minute=100, **kwargs + ) + + +@pytest.mark.parametrize("operation", ("configure", "reject", "admit", "release")) +def test_implicit_time_follows_lock_order( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, operation: str +) -> None: + path = tmp_path / "token-capacity.json" + now = datetime.now(UTC) - timedelta(seconds=1) + _configure(path, now=now) + observe_benchmark_token_usage( + path, _sample("healthy", now, requests=0), execute=True + ) + if operation == "release": + for run_id in ("solver-a", "solver-b"): + _mutate(path, "admit", run_id, execute=True) + + waiting, resume = Event(), Event() + real_lock = token_capacity.exclusive_file_lock + + @contextmanager + def reverse_lock(*args, **kwargs): + if kwargs.get("agent_id") == "delayed": + waiting.set() + assert resume.wait(10), "later caller did not finish" + with real_lock(*args, **kwargs): + yield + + monkeypatch.setattr(token_capacity, "exclusive_file_lock", reverse_lock) + with ThreadPoolExecutor(max_workers=1) as executor: + delayed = executor.submit( + _mutate, path, operation, "solver-a", execute=True, agent_id="delayed" + ) + try: + assert waiting.wait(10), "first caller did not reach the lock" + ahead = _mutate(path, operation, "solver-b", execute=True) + assert ahead["write_performed"] is True + finally: + resume.set() + assert delayed.result(timeout=10)["write_performed"] is True + + state = read_benchmark_token_capacity_state(path) + assert state is not None + if operation == "admit": + assert {item["run_id"] for item in state["reservations"]} == { + "solver-a", + "solver-b", + } + elif operation == "release": + assert state["reservations"] == [] + elif operation == "reject": + assert ( + build_benchmark_token_capacity_status(state)["cooldown"]["active"] is True + ) + + +@pytest.mark.parametrize( + ("operation", "time_field"), + ( + ("configure", "observed_at"), + ("reject", "observed_at"), + ("admit", "admitted_at"), + ("release", "released_at"), + ), +) +@pytest.mark.parametrize("execute", (False, True)) +def test_explicit_backward_mutation_time_is_rejected( + tmp_path: Path, operation: str, time_field: str, execute: bool +) -> None: + path = tmp_path / "token-capacity.json" + now = datetime(2026, 10, 28, 10, 0, tzinfo=UTC) + _configure(path, now=now) + before = path.read_bytes() + with pytest.raises(ValueError, match="cannot move state backward"): + _mutate( + path, + operation, + "solver-a", + execute=execute, + **{time_field: (now - timedelta(microseconds=1)).isoformat()}, + ) + assert path.read_bytes() == before + + def test_cli_composes_token_receipt_with_existing_headroom_contract( tmp_path: Path, ) -> None: @@ -500,3 +603,64 @@ def test_cli_composes_token_receipt_with_existing_headroom_contract( check=True, ) assert json.loads(case_admission.stdout)["admitted"] is True + + +def test_cli_without_token_policy_retains_case_admission(tmp_path: Path) -> None: + catalog = json.loads( + run_source_cli( + ["capability", "show", "benchmark-toolkit", "--format", "json"] + ).stdout + )["capability"]["agent_usage"] + for prefix in ( + "record_complete_or_explicitly_unmetered_token_usage", + "atomically_reserve_token_capacity_before_case_slot_admission", + "release_token_capacity_after_terminal_or_runner_invalid_transition", + ): + (step,) = [ + step for step in catalog["required_sequence"] if step.startswith(prefix) + ] + assert step.endswith("_when_token_capacity_is_configured") + + location = ["--goal-id", "fixture-goal", "--project", str(tmp_path)] + execution = ["--execute", "--format", "json"] + status = json.loads( + run_source_cli( + ["benchmark", "token-capacity-status", *location, "--format", "json"] + ).stdout + ) + assert status["configured"] is False + run_source_cli( + [ + "benchmark", + "concurrency-configure", + *location, + "--max-active-cases", + "1", + "--max-baseline-cases", + "1", + "--max-test-cases", + "1", + *execution, + ] + ) + admitted = json.loads( + run_source_cli( + [ + "benchmark", + "concurrency-admit", + *location, + "--run-id", + "solver-cli", + "--case-id", + "case-cli", + "--arm-role", + "treatment", + *execution, + ] + ).stdout + ) + assert admitted["admitted"] is True + path = token_capacity.default_benchmark_token_capacity_path( + project=tmp_path, goal_id="fixture-goal" + ) + assert not path.exists()