"""Final B1 production_admission_capacity_gated 120-prompt measurement. Uses the frozen 13A-02 oracle prompts. Does not score or invoke Planner. Does not overwrite historical 13A or B1 fixtures. Does not change timeout, admission policy, capacity policy, gold, or Phase 12 contracts. """ from __future__ import annotations import argparse import asyncio import hashlib import inspect import itertools import json import subprocess import sys import time from collections import Counter from datetime import UTC, datetime from pathlib import Path from types import SimpleNamespace from uuid import UUID ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from sqlalchemy import text # noqa: E402 from src.config import get_settings # noqa: E402 from src.core.provider_capacity import watchdog_seconds # noqa: E402 from src.domain.projects.discovery_b1 import ( # noqa: E402 capacity_summary, capacity_trace_fields, classify_reproducibility_pair, score_b1, ) from src.domain.projects.discovery_catalog import build_discovery_payload # noqa: E402 from src.domain.projects.discovery_readiness import ( # noqa: E402 CONTENT_ARCHETYPES, MARKETPLACE_ARCHETYPES, finalize_discovery_result, ) from src.domain.projects.discovery_schemas import DiscoverySpec, Entity # noqa: E402 from src.integrations.project_intake_client import DISCOVERY_PROMPT, discover_project # noqa: E402 from src.workers.tasks.project_discovery import reserve_discovery_call # noqa: E402 EVALUATION_PROJECT_ID = "1f6556ff-4a16-4143-9b12-e9d9249d1d3d" EVALUATION_WORKSPACE_ID = "5e890ff3-5c7b-43f5-a4ad-75c47f90bdba" ORACLE_PATH = ROOT / "docs/audits/fixtures/13b_b1_live_oracle_prompts.json" ORACLE_PROMPTS_SHA256 = "f4bf800dcf399898189f9781d7bddff42763d75660ca204f62eca7b0da6349f0" TIMEOUT_TYPES = {"TimeoutError", "APITimeoutError"} MEASUREMENT_KIND = "production_admission_capacity_gated" OPTION_A_WATCHLIST = { "13A-025", "13A-042", "13A-051", "13A-060", "13A-063", "13A-075", "13A-085", "13A-103", "13A-117", } REQUIRED_BOUNDARY_CONTROLS = ("13A-042", "13A-060", "13A-085", "13A-117") BOOKING_ARCHETYPES = {"booking"} EMPTY_INPUT_PROMPTS = {"hello", "hi", "hey"} PROTECTED_OUTPUTS = { ROOT / "docs/audits/fixtures/13a_run_01", ROOT / "docs/audits/fixtures/13a_run_02", ROOT / "docs/audits/fixtures/13b_b1_completion/extraction_before", ROOT / "docs/audits/fixtures/13b_b1_admission_gated_01", ROOT / "docs/audits/fixtures/13b_b1_live_01", ROOT / "docs/audits/fixtures/13b_b1_live_02", ROOT / "docs/audits/fixtures/13b_b1_equal_load_01", ROOT / "docs/audits/fixtures/13b_b1_equal_load_02_with_rate_limit_fix", ROOT / "docs/audits/fixtures/13b_b1_equal_load_03_retry_only", ROOT / "docs/audits/fixtures/13b_b1_semantic_stability_live_01", ROOT / "docs/audits/fixtures/13b_b1_extraction_prod_host_01", ROOT / "docs/audits/fixtures/13b_b1_admission_prod_host_01.json", ROOT / "docs/audits/fixtures/13b_b1_admission_prod_host_01.operator_summary.json", ROOT / "docs/audits/fixtures/13b_b1_admission_gated_prod_host_02", ROOT / "docs/audits/fixtures/13b_b1_provider_limit_prod_host_03", ROOT / "docs/audits/fixtures/13b_b1_provider_limit_prod_host_04", } def write_json(path: Path, value) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def load_oracle(): payload = json.loads(ORACLE_PATH.read_text(encoding="utf-8")) prompts = payload["prompts"] if len(prompts) != 120: raise ValueError(f"Expected 120 frozen oracle prompts, found {len(prompts)}") digest = hashlib.sha256( json.dumps(prompts, sort_keys=True, ensure_ascii=False).encode() ).hexdigest() if digest != ORACLE_PROMPTS_SHA256: raise ValueError("Oracle prompt identity does not match the frozen 13A-02 set") if payload.get("planner_scored_by_this_harness") is not False: raise ValueError("This harness must not score Planner") return prompts def _refuse_protected(output: Path) -> None: resolved = output.resolve() for path in PROTECTED_OUTPUTS: target = path.resolve() if resolved == target or target in resolved.parents: raise ValueError("Refusing to write into a protected historical fixture") def _percentile(ordered: list[float], percent: int) -> float: if len(ordered) == 1: return ordered[0] rank = (percent / 100) * (len(ordered) - 1) low = int(rank) high = min(low + 1, len(ordered) - 1) return ordered[low] + (ordered[high] - ordered[low]) * (rank - low) def latency_distribution(elapsed: list[float]) -> dict: if not elapsed: return {"p50": None, "p95": None, "n": 0, "unmeasured_reason": "No elapsed samples."} ordered = sorted(elapsed) return { "p50": round(_percentile(ordered, 50), 3), "p95": round(_percentile(ordered, 95), 3), "n": len(ordered), "unmeasured_reason": None, } def rate(numerator: int, denominator: int, reason: str | None = None) -> dict: return { "numerator": numerator, "denominator": denominator, "rate": numerator / denominator if denominator else None, "unmeasured_reason": reason if not denominator else None, } def _names(items): return sorted( (item.get("name") or "").strip().casefold() for item in items or [] if isinstance(item, dict) ) def layer_a_projection(row): raw = (row.get("discovery") or {}).get("raw_turn") if raw is None: return None spec = raw.get("spec") or {} return { "entities": _names(spec.get("entities")), "workflow_count": len(spec.get("workflows") or []), "archetypes": sorted(spec.get("archetypes") or []), "actor_names": _names(spec.get("actors")), "requested_capabilities": sorted(raw.get("requested_capabilities") or []), } def _decision_key(row): discovery = row.get("discovery") or {} if discovery.get("status") != "success": return (discovery.get("status"), discovery.get("error_type")) result = discovery.get("result") or {} question = (result.get("question") or {}).get("decision_id") return (result.get("decision"), question) def _git_identity() -> dict: def _run(*args: str) -> str: try: return subprocess.check_output(args, cwd=ROOT, text=True).strip() except Exception: return "unmeasured" return { "git_head": _run("git", "rev-parse", "HEAD"), "git_head_short": _run("git", "rev-parse", "--short", "HEAD"), "git_origin_main": _run("git", "rev-parse", "origin/main"), "git_status_short": _run("git", "status", "--porcelain"), } def _seconds(ms: float | None) -> float | None: if ms is None: return None return round(float(ms) / 1000, 3) def _readiness_family(projection: dict | None) -> str | None: if not projection: return None archetypes = {item.strip().lower() for item in projection.get("archetypes") or []} if archetypes & {item.lower() for item in MARKETPLACE_ARCHETYPES}: return "marketplace" if archetypes & BOOKING_ARCHETYPES or any("booking" in item for item in archetypes): return "booking" if archetypes and archetypes <= {item.lower() for item in CONTENT_ARCHETYPES}: return "content" return "other" def _error_bucket(row: dict) -> str | None: discovery = row.get("discovery") or {} if discovery.get("status") != "error": return None error_type = discovery.get("error_type") if error_type in TIMEOUT_TYPES: return "provider_timeout" if error_type == "RateLimitError": return "final_429" if error_type == "ProviderCapacityUnavailableError": return "capacity_unavailable" if error_type == "AdmissionDenied": return "cost_abuse_denial" if error_type == "AdmissionCheckFailed": return "admission_infrastructure_error" return "other_execution_error" def run_execution_receipt(rows: list[dict]) -> dict: buckets = Counter(_error_bucket(row) for row in rows if _error_bucket(row)) successes = sum( 1 for row in rows if (row.get("discovery") or {}).get("status") == "success" ) initial_429 = sum( 1 for row in rows if (row.get("discovery") or {}).get("provider_429_initial") or (row.get("discovery") or {}).get("error_type") == "RateLimitError" ) capacity_deadline = sum( 1 for row in rows if ((row.get("discovery") or {}).get("provider_capacity") or {}).get( "capacity_deadline_exceeded" ) or (row.get("discovery") or {}).get("error_type") == "ProviderCapacityUnavailableError" ) return { "successes": successes, "provider_timeout": buckets["provider_timeout"], "initial_429": initial_429, "final_429": buckets["final_429"], "capacity_deadline": capacity_deadline, "capacity_unavailable": buckets["capacity_unavailable"], "cost_abuse_denial": buckets["cost_abuse_denial"], "admission_infrastructure_error": buckets["admission_infrastructure_error"], "other_execution_error": buckets["other_execution_error"], } def _provider_receipts(rows: list[dict]) -> dict: timeouts = 0 other_errors = 0 for row in rows: bucket = _error_bucket(row) if bucket == "provider_timeout": timeouts += 1 elif bucket in {"other_execution_error", "final_429"}: if bucket == "other_execution_error": other_errors += 1 elif bucket == "final_429": other_errors += 0 other_provider = 0 for row in rows: discovery = row.get("discovery") or {} if discovery.get("status") != "error": continue error_type = discovery.get("error_type") if error_type in TIMEOUT_TYPES: continue if error_type in { "RateLimitError", "ProviderCapacityUnavailableError", "AdmissionDenied", "AdmissionCheckFailed", }: continue other_provider += 1 return { "provider_timeout_count": timeouts, "other_provider_error_count": other_provider, } def _option_a_inspections(runs: list[list[dict]]) -> dict: indexed = [{row["id"]: row for row in run} for run in runs] family_flips = [] capability_disagreements = [] empty_vs_populated = [] bookings_over_extraction = [] negative_invention = [] for trial_id in sorted(indexed[0]): left = indexed[0][trial_id] right = indexed[1][trial_id] pa, pb = layer_a_projection(left), layer_a_projection(right) if pa is None or pb is None: continue family_a, family_b = _readiness_family(pa), _readiness_family(pb) watched = {family_a, family_b} & {"content", "booking", "marketplace"} if len(watched) > 1: family_flips.append( { "id": trial_id, "run_1": family_a, "run_2": family_b, "archetypes": [pa["archetypes"], pb["archetypes"]], } ) if pa["requested_capabilities"] != pb["requested_capabilities"]: capability_disagreements.append( { "id": trial_id, "run_1": pa["requested_capabilities"], "run_2": pb["requested_capabilities"], } ) left_empty = not pa["entities"] and pa["workflow_count"] == 0 right_empty = not pb["entities"] and pb["workflow_count"] == 0 if left_empty != right_empty: empty_vs_populated.append(trial_id) if left["prompt"] == "Help with bookings." and ( pa["entities"] or pa["workflow_count"] or pb["entities"] or pb["workflow_count"] ): bookings_over_extraction.append(trial_id) if left["prompt"].strip().casefold() in EMPTY_INPUT_PROMPTS and ( pa["entities"] or pa["workflow_count"] or pb["entities"] or pb["workflow_count"] ): negative_invention.append(trial_id) return { "negative_control_invention": { "count": len(negative_invention), "ids": negative_invention, "hello_negative_control_in_corpus": False, }, "help_with_bookings_over_extraction": { "count": len(bookings_over_extraction), "ids": bookings_over_extraction, }, "requested_capability_disagreement": { "count": len(capability_disagreements), "pairs": capability_disagreements, }, "empty_vs_populated_oscillation": { "count": len(empty_vs_populated), "ids": empty_vs_populated, }, "content_booking_marketplace_family_flips": { "count": len(family_flips), "pairs": family_flips, }, "boundary_controls": { trial_id: [layer_a_projection(run[trial_id]) for run in indexed if trial_id in run] for trial_id in REQUIRED_BOUNDARY_CONTROLS }, } def enrich_capacity_fields(trace: dict, admission: dict, started: float) -> dict: fields = capacity_trace_fields(trace) capacity = dict(fields.get("provider_capacity") or {}) demand = capacity.get("estimated_request_demand") or fields.get("estimated_request_tokens") reserved_after = capacity.get("local_reserved_after") if reserved_after is None: reserved_after = capacity.get("reserved_capacity") if ( capacity.get("local_reserved_before") is None and reserved_after is not None and demand is not None and capacity.get("capacity_granted") ): capacity["local_reserved_before"] = reserved_after - demand capacity["local_reserved_after"] = reserved_after if "capacity_immediate_grant" not in capacity: capacity["capacity_immediate_grant"] = bool( capacity.get("capacity_granted") and not capacity.get("capacity_queued") ) capacity["observed_provider_limit"] = capacity.get("observed_limit") capacity["observed_provider_remaining"] = capacity.get("observed_remaining") capacity["provider_reset"] = capacity.get("provider_reset_at_ms") fields["provider_capacity"] = capacity fields["cost_abuse_admission_ms"] = int(round(float(admission.get("check_seconds") or 0) * 1000)) fields["capacity_wait_ms"] = capacity.get("capacity_wait_ms") fields["end_to_end_ms"] = int(round((time.monotonic() - started) * 1000)) return fields def scorecard(runs: list[list[dict]]) -> dict: first = runs[0] all_rows = [row for run in runs for row in run] b1 = score_b1(first, timeout_policy_changed=False, repeats=runs) indexed = [{row["id"]: row for row in run} for run in runs] layer_a = { "pair_count": 0, "execution_error_pairs": 0, "entity_projection_disagreement": 0, "workflow_projection_disagreement": 0, "archetype_disagreement": 0, "actor_projection_disagreement": 0, "capability_projection_disagreement": 0, "boundary_help_with_bookings_entity_or_workflow": 0, "empty_input_invention": 0, "hello_negative_control_in_corpus": False, "watchlist": {}, "semantic_variance_taxonomy": {}, } taxonomy = Counter() for left, right in itertools.combinations(runs, 2): for a, b in zip(left, right, strict=True): layer_a["pair_count"] += 1 taxonomy[classify_reproducibility_pair(a, b)] += 1 pa, pb = layer_a_projection(a), layer_a_projection(b) if pa is None or pb is None: layer_a["execution_error_pairs"] += 1 continue layer_a["entity_projection_disagreement"] += pa["entities"] != pb["entities"] layer_a["workflow_projection_disagreement"] += ( pa["workflow_count"] != pb["workflow_count"] ) layer_a["archetype_disagreement"] += pa["archetypes"] != pb["archetypes"] layer_a["actor_projection_disagreement"] += pa["actor_names"] != pb["actor_names"] layer_a["capability_projection_disagreement"] += ( pa["requested_capabilities"] != pb["requested_capabilities"] ) if a["prompt"] == "Help with bookings." and ( pa["entities"] or pa["workflow_count"] or pb["entities"] or pb["workflow_count"] ): layer_a["boundary_help_with_bookings_entity_or_workflow"] += 1 if a["prompt"].strip().casefold() in EMPTY_INPUT_PROMPTS: layer_a["hello_negative_control_in_corpus"] = True if ( pa["entities"] or pa["workflow_count"] or pb["entities"] or pb["workflow_count"] ): layer_a["empty_input_invention"] += 1 layer_a["semantic_variance_taxonomy"] = dict(taxonomy) for trial_id in sorted(OPTION_A_WATCHLIST): layer_a["watchlist"][trial_id] = [ layer_a_projection(run[trial_id]) for run in indexed if trial_id in run ] capacity = capacity_summary(all_rows) provider = _provider_receipts(all_rows) wait = capacity.get("capacity_wait_ms_p50_p95") or {} return { "measurement_kind": MEASUREMENT_KIND, "phase": "13B", "gate": "B1", "corpus_size": len(first), "b1_operational_policy": "OPTION_A_AUTHORIZED", "raw_planner_control_preserved": True, "planner_scored": False, "frozen_planner_denominator_13a02": { "raw_planner_exact_product": "76/108", "raw_planner_precision": "104/118", "raw_planner_recall": "104/139", "CRM_exact": "3/10", "CRM_recall": "8/23", "mixed_quote_exact": "10/10", "mixed_quote_recall": "10/10", }, "discovery_completion_rate": b1["discovery_completion_rate"], "discovery_latency_p50_p95": b1["discovery_latency_p50_p95"], "discovery_provider_latency_p50_p95": b1["discovery_provider_latency_p50_p95"], "discovery_timeout_rate": b1["discovery_timeout_rate"], "discovery_error_types": b1["discovery_error_types"], "first_turn_ready_rate": b1["first_turn_ready_rate"], "unnecessary_clarification_rate": b1["unnecessary_clarification_rate"], "invented_assumption_rate": b1["invented_assumption_rate"], "same_prompt_reproducibility": b1["same_prompt_reproducibility"], "semantic_reproducibility": b1["semantic_reproducibility"], "timeout_policy_changed": False, "admission": b1["admission"], "run_execution": { f"run_{index:02}": run_execution_receipt(run) for index, run in enumerate(runs, start=1) }, "completed_in_both_runs": b1["same_prompt_reproducibility"].get("completed_in_all_runs"), "execution_error_pairs": layer_a["execution_error_pairs"], "provider_capacity": capacity, "provider_receipts": { "provider_successes": sum( 1 for row in all_rows if (row.get("discovery") or {}).get("status") == "success" ), "initial_provider_429_count": capacity["provider_429_initial"], "retry_attempt_count": capacity["retry_attempt_count"], "retry_success_count": capacity["provider_429_retry_success"], "retry_exhausted_count": capacity["retry_exhausted_count"], "final_provider_429_count": capacity["provider_429_final"], **provider, }, "capacity_wait_seconds": { "p50": _seconds(wait.get("p50")), "p95": _seconds(wait.get("p95")), "max": _seconds(capacity.get("capacity_wait_ms_max")), }, "layer_a_raw_turn": layer_a, "option_a_inspections": _option_a_inspections(runs), "acceptance_policy": ( "PHASE_13B_B1_EXIT_REVIEW.md B1 Operational Acceptance Decision OPTION_A" ), } def preflight(): settings = get_settings() prompts = load_oracle() if "EXTRACTION DISCIPLINE" not in DISCOVERY_PROMPT: raise ValueError("Production measurement requires the authorized extraction contract") if "INTENT FIDELITY" not in DISCOVERY_PROMPT: raise ValueError("Production measurement requires the intent-fidelity contract") entity_name = Entity.model_fields["name"].description or "" workflows = DiscoverySpec.model_fields["workflows"].description or "" if "grammatical number" not in entity_name: raise ValueError("Entity.name extraction description is not loaded") if "including when decision is clarify" not in workflows: raise ValueError("Workflow extraction description is not loaded") if settings.project_intake_timeout_seconds != 8: raise ValueError("Live B1 requires the frozen 8s timeout") if settings.project_intake_model != "gpt-4o-mini": raise ValueError("Model differs from the frozen baseline") if not settings.openai_api_key: raise ValueError("Live B1 requires the configured provider credential") if not settings.redis_url: raise ValueError("Live B1 requires configured Redis") if settings.project_intake_calls_per_minute != 60: raise ValueError("Expected unchanged 60/min admission policy") if getattr(settings, "provider_capacity_enabled", True) is not True: raise ValueError("Capacity manager must not be bypassed") if float(getattr(settings, "provider_capacity_safety_fraction", 0)) != 0.90: raise ValueError("Capacity safety fraction must remain 0.90") if float(getattr(settings, "provider_capacity_wait_seconds", 0)) != 60: raise ValueError("Capacity queue deadline must remain 60s") source = inspect.getsource(reserve_discovery_call) if "project-intake:budget:" not in source: raise ValueError("Admission implementation is not the production reserve_discovery_call") from src.integrations import project_intake_client as intake_client structured = inspect.getsource(intake_client._project_structured_call) if "manager.acquire" not in structured: raise ValueError("Discover path does not acquire provider capacity") if "max_retries=0" not in structured: raise ValueError("Provider SDK retries must remain 0") return { "preflight": "passed", "measurement_kind": MEASUREMENT_KIND, "b1_operational_policy": "OPTION_A_AUTHORIZED", "working_directory": str(ROOT), "oracle_count": len(prompts), "oracle_prompts_sha256": ORACLE_PROMPTS_SHA256, "repeats_default": 2, "expected_provider_calls": len(prompts) * 2, "concurrency": 4, "start_interval_seconds": 1.1, "timeout_seconds": 8, "model": settings.project_intake_model, "admission_policy_enabled": True, "configured_calls_per_minute": 60, "provider_capacity_enabled": True, "provider_capacity_safety_fraction": 0.90, "provider_capacity_wait_seconds": 60, "planner_scored": False, "provider_calls": 0, "required_environment_names": ["REDIS_URL", "DATABASE_URL", "OPENAI_API_KEY"], "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), **_git_identity(), } def verify(output: Path): scorecard = json.loads((output / "scorecard.json").read_text(encoding="utf-8")) metadata = json.loads((output / "metadata.json").read_text(encoding="utf-8")) oracle = json.loads((output / "oracle_identity.json").read_text(encoding="utf-8")) contract = json.loads((output / "request_contract.json").read_text(encoding="utf-8")) assert scorecard["measurement_kind"] == MEASUREMENT_KIND assert metadata["measurement_kind"] == MEASUREMENT_KIND assert contract["measurement_kind"] == MEASUREMENT_KIND assert metadata["environment"] == "production" assert metadata["admission_policy_enabled"] is True assert metadata["planner_scored"] is False assert contract["planner_scored"] is False assert oracle["oracle_prompts_sha256"] == ORACLE_PROMPTS_SHA256 assert scorecard["raw_planner_control_preserved"] is True assert scorecard["planner_scored"] is False prompts = load_oracle() for repeat in range(1, metadata["repeat_runs"] + 1): for case in prompts: path = output / f"run_{repeat:02}" / "trials" / f"{case['id']}.json" assert path.is_file(), f"missing {path}" return { "verification": "passed", "environment": metadata["environment"], "measurement_kind": scorecard["measurement_kind"], "repeat_runs": metadata["repeat_runs"], "oracle_count": len(prompts), "admission": scorecard["admission"], "discovery_completion_rate": scorecard["discovery_completion_rate"], "layer_a_raw_turn": scorecard["layer_a_raw_turn"], } async def live_context(): from src.core.database import async_session_factory, engine from src.domain.projects.discovery_service import available_templates engine.echo = False try: async with async_session_factory() as db: await db.execute(text("SET TRANSACTION READ ONLY")) project = ( ( await db.execute( text( "SELECT type, source_kind FROM projects " "WHERE id=:pid AND workspace_id=:wid" ), { "pid": UUID(EVALUATION_PROJECT_ID), "wid": UUID(EVALUATION_WORKSPACE_ID), }, ) ) .mappings() .first() ) if not project: raise ValueError("Project does not belong to the evaluation workspace") templates = [ SimpleNamespace( **{ key: getattr(item, key) for key in ("ref", "name", "description", "framework", "tags") } ) for item in await available_templates(db, UUID(EVALUATION_WORKSPACE_ID)) ] revisions = list( (await db.execute(text("SELECT version_num FROM alembic_version"))).scalars() ) await db.rollback() return dict(project), templates, revisions finally: await engine.dispose() async def measure(output: Path, repeats: int, environment: str, start_interval: float): _refuse_protected(output) if output.exists(): raise ValueError("Select a new experiment directory") if repeats < 2: raise ValueError("Live B1 requires at least two repeats") if start_interval < 1.1: raise ValueError("Live B1 must not exceed the baseline request start rate") settings = get_settings() preflight() project, templates, revisions = await live_context() cases = load_oracle() wall_started = time.monotonic() write_json( output / "oracle_identity.json", { "corpus": "13A-02", "oracle_count": len(cases), "oracle_prompts_sha256": ORACLE_PROMPTS_SHA256, "planner_scored": False, "expected_product_blocks_included_for_identity_only": True, }, ) metadata = { "phase": "13B", "gate": "B1", "measurement_kind": MEASUREMENT_KIND, "b1_operational_policy": "OPTION_A_AUTHORIZED", "environment": environment, "admission_policy_enabled": True, "provider_capacity_enabled": True, "provider_capacity_safety_fraction": settings.provider_capacity_safety_fraction, "provider_capacity_wait_seconds": settings.provider_capacity_wait_seconds, "configured_calls_per_minute": settings.project_intake_calls_per_minute, "repeat_runs": repeats, "concurrency": 4, "start_interval_seconds": start_interval, "timeout_seconds": 8.0, "timeout_policy_changed": False, "admission_policy_changed": False, "capacity_policy_changed_during_run": False, "retry_policy_changed_during_run": False, "planner_changed": False, "planner_scored": False, "corpus_changed": False, "gold_changed": False, "core_entity_canonicalization": False, "expected_provider_calls": len(cases) * repeats, "discovery_model": settings.project_intake_model, "project_id": EVALUATION_PROJECT_ID, "workspace_id": EVALUATION_WORKSPACE_ID, "database_alembic_revisions": revisions, "template_count": len(templates), "catalog_revision_matches_13a02": revisions == ["145"], "started_at": datetime.now(UTC).isoformat(), "status": "running", **_git_identity(), "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "do_not": [ "overwrite historical 13A or B1 fixtures", "begin Gate B2", "treat timeout increase as success", "change Planner or Phase 12 contracts", "broaden Option A after seeing results", ], } write_json(output / "metadata.json", metadata) write_json( output / "request_contract.json", { "measurement_kind": MEASUREMENT_KIND, "b1_operational_policy": "OPTION_A_AUTHORIZED", "corpus": "13A-02", "oracle_prompts_sha256": ORACLE_PROMPTS_SHA256, "repeats": repeats, "concurrency": 4, "start_interval_seconds": start_interval, "timeout_seconds": 8, "admission_policy_enabled": True, "provider_capacity_enabled": True, "planner_scored": False, "finalize": "src.domain.projects.discovery_readiness.finalize_discovery_result", "prompt_contains_extraction_discipline": "EXTRACTION DISCIPLINE" in DISCOVERY_PROMPT, }, ) semaphore, lock = asyncio.Semaphore(4), asyncio.Lock() last_start = 0.0 async def one(case, repeat): nonlocal last_start async with semaphore: async with lock: await asyncio.sleep(max(0, last_start + start_interval - time.monotonic())) last_start = time.monotonic() row = { "id": case["id"], "prompt": case["prompt"], "oracle_kind": case["oracle_kind"], "stratum": case["stratum"], } payload = build_discovery_payload( project["type"], project["source_kind"], {"messages": [{"role": "user", "content": case["prompt"]}]}, templates, ) started = time.monotonic() admission_started = time.monotonic() try: await reserve_discovery_call(settings) admission = { "attempted": True, "granted": True, "check_seconds": round(time.monotonic() - admission_started, 4), } except ValueError as exc: admission = { "attempted": True, "granted": False, "denial_kind": "policy", "reason": str(exc), "check_seconds": round(time.monotonic() - admission_started, 4), } except Exception as exc: admission = { "attempted": True, "granted": False, "denial_kind": "infrastructure", "reason": f"{type(exc).__name__}: {exc}", "check_seconds": round(time.monotonic() - admission_started, 4), } if not admission["granted"]: row["discovery"] = { "status": "error", "error_type": ( "AdmissionDenied" if admission.get("denial_kind") == "policy" else "AdmissionCheckFailed" ), "admission": admission, "elapsed_seconds": round(time.monotonic() - started, 3), } else: provider_started = time.monotonic() trace = {} try: turn = await asyncio.wait_for( discover_project(payload, settings, trace=trace), timeout=watchdog_seconds(settings, 8), ) result = finalize_discovery_result(turn, payload) extras = enrich_capacity_fields(trace, admission, started) row["discovery"] = { "status": "success", "raw_turn": turn.model_dump(mode="json"), "result": result.model_dump(mode="json"), "admission": admission, "provider_elapsed_seconds": round( (extras.get("provider_execution_ms") or 0) / 1000, 3 ) if extras.get("provider_execution_ms") is not None else round(time.monotonic() - provider_started, 3), "elapsed_seconds": round(time.monotonic() - started, 3), **extras, } except Exception as exc: extras = enrich_capacity_fields(trace, admission, started) row["discovery"] = { "status": "error", "error_type": type(exc).__name__, "admission": admission, "provider_elapsed_seconds": round( (extras.get("provider_execution_ms") or 0) / 1000, 3 ) if extras.get("provider_execution_ms") is not None else round(time.monotonic() - provider_started, 3), "elapsed_seconds": round(time.monotonic() - started, 3), **extras, } write_json(output / f"run_{repeat:02}" / "trials" / f"{case['id']}.json", row) print( json.dumps( {"run": repeat, "id": case["id"], "status": row["discovery"]["status"]} ), flush=True, ) return row runs = [] for repeat in range(1, repeats + 1): runs.append(await asyncio.gather(*(one(case, repeat) for case in cases))) scored = scorecard(runs) metadata.update( status="complete", finished_at=datetime.now(UTC).isoformat(), wall_clock_seconds=round(time.monotonic() - wall_started, 3), ) write_json(output / "metadata.json", metadata) write_json(output / "scorecard.json", scored) print(json.dumps({ "measurement_kind": scored["measurement_kind"], "run_execution": scored["run_execution"], "completed_in_both_runs": scored["completed_in_both_runs"], "execution_error_pairs": scored["execution_error_pairs"], "provider_capacity": scored["provider_capacity"], "provider_receipts": scored["provider_receipts"], "first_turn_ready_rate": scored["first_turn_ready_rate"], "same_prompt_reproducibility": scored["same_prompt_reproducibility"], "semantic_reproducibility": scored["semantic_reproducibility"], "layer_a_raw_turn": scored["layer_a_raw_turn"], "option_a_inspections": scored["option_a_inspections"], }, indent=2)) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", type=Path) parser.add_argument("--repeats", type=int, default=2) parser.add_argument("--start-interval", type=float, default=1.1) parser.add_argument("--environment", choices=("local", "production"), default="local") parser.add_argument("--preflight", action="store_true") parser.add_argument("--verify", type=Path) args = parser.parse_args() try: if args.preflight: print(json.dumps(preflight(), indent=2)) elif args.verify: print(json.dumps(verify(args.verify), indent=2)) elif args.out: asyncio.run(measure(args.out, args.repeats, args.environment, args.start_interval)) else: parser.error("Supply --preflight, --verify RECEIPT_DIR, or --out NEW_DIRECTORY") except Exception as exc: print( json.dumps({"diagnostic_failed": type(exc).__name__, "detail": str(exc)}), file=sys.stderr, ) raise SystemExit(1) from None