"""Read-only 13A measurement: live Discovery + Planner, separate raw-text control. No project creation, database writes, migrations, provider substitution or repair. The raw-text control is NOT the production Discovery pipeline. Unavailable later stages have null rates, never inferred successes. Run once with --live-discovery. """ # ruff: noqa: E402 -- direct CLI execution needs the repository path before src imports. from __future__ import annotations import argparse import asyncio import hashlib import json import shutil 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 src.config import get_settings from src.domain.app_blocks.compiler import compile_install_plan from src.domain.app_blocks.planner import product_intent_from_spec, select_capabilities from src.domain.app_blocks.registry import load_default_registry from src.domain.projects.discovery_catalog import build_discovery_payload from src.domain.projects.discovery_readiness import finalize_discovery_result from src.domain.projects.discovery_schemas import DiscoverySpec, ProductDefinition from src.integrations.project_intake_client import discover_project REGISTRY = load_default_registry() PRODUCT_BLOCKS = { "crm.customer", "crm.lead", "crm.pipeline", "forms.contact", "forms.quote_request", "case.management", "workflow.assignment", "task.management", "activity.timeline", "notifications.in_app", "files.attachments", "documents", "booking.appointment", "catalog.product", "support.ticket", "workflow.approval", } LATER_METRICS = ( "codegen_success_rate", "must_mount_conformance_rate", "unexpected_semantic_surface_rate", "structural_validation_rate", "actual_build_rate", "runtime_smoke_rate", "repair_required_rate", "repair_success_rate", "terminal_generation_failure_rate", ) def corpus(path: Path) -> list[dict]: rows = [] for line in path.read_text(encoding="utf-8").splitlines(): if not line or line.startswith("#"): continue group, gold, prompt = line.split("|", 2) index = len(rows) + 1 rows.append( { "id": f"13A-{index:03}", "stratum": group, "prompt": prompt, "expected_product_blocks": [] if gold == "-" else gold.split(","), "oracle_kind": "clarify" if group == "ambiguous" else ("admission_required" if index in (119, 120) else "admitted"), "provenance": "authored realistic request; not sampled production customer traffic", } ) # GR-05 appended a 10-row "support" stratum (121-130) and GR-06 appended # a 10-row "approval" stratum (131-140) after the frozen 13A-01/13A-02 # 120-row set. Appending, never inserting, preserves every existing # 13A-NNN id -- including the admission_required check above, which is # intentionally still hardcoded to (119, 120). assert len(rows) == 140 assert len({r["prompt"] for r in rows}) == 140 assert all(set(r["expected_product_blocks"]) <= PRODUCT_BLOCKS for r in rows) return rows def plan_observation(spec: DiscoverySpec, gold: list[str]) -> dict: selected = {b for _, b, _ in select_capabilities(product_intent_from_spec(spec))} predicted = selected & PRODUCT_BLOCKS expected = set(gold) plan = compile_install_plan(spec, registry=REGISTRY) return { "predicted_product_blocks": sorted(predicted), "true_positive": sorted(predicted & expected), "false_positive": sorted(predicted - expected), "false_negative": sorted(expected - predicted), "exact_product_match": predicted == expected, "plan_satisfiable": not plan.unsatisfiable, "plan": plan.model_dump(mode="json"), "required_ui_slots": sum( slot.type == "ui_slot" for block in plan.blocks for slot in REGISTRY.get(block.id).slots ), } 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 score(observations: list[dict]) -> dict: tp = sum(len(r["true_positive"]) for r in observations) fp = sum(len(r["false_positive"]) for r in observations) fn = sum(len(r["false_negative"]) for r in observations) negatives = sum( len(PRODUCT_BLOCKS) - len(r["true_positive"]) - len(r["false_negative"]) for r in observations ) return { "planner_precision": rate(tp, tp + fp), "planner_recall": rate(tp, tp + fn), "false_positive_capability_rate": rate(fp, negatives), "false_negative_capability_rate": rate(fn, tp + fn), "exact_product_match_rate": rate( sum(r["exact_product_match"] for r in observations), len(observations) ), "plan_satisfiability_rate": rate( sum(r["plan_satisfiable"] for r in observations), len(observations) ), } def write_json(path: Path, value): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") async def measure(args): settings = get_settings() rows = corpus(ROOT / "scripts/phase_13a_corpus.txt") output = Path(args.out).resolve() output.mkdir(parents=True, exist_ok=True) if (output / "scorecard.json").exists(): raise ValueError("Existing measurement must not be overwritten; select a new output folder") write_json(output / "corpus.json", rows) corpus_hash = hashlib.sha256((output / "corpus.json").read_bytes()).hexdigest() from sqlalchemy import text from src.core.database import async_session_factory, engine from src.domain.projects.discovery_service import available_templates engine.echo = False async with async_session_factory() as db: await db.execute(text("SET TRANSACTION READ ONLY")) project = ( ( await db.execute( text( "SELECT id, workspace_id, type, status, source_kind FROM projects " "WHERE id=:pid AND workspace_id=:wid" ), {"pid": UUID(args.project_id), "wid": UUID(args.workspace_id)}, ) ) .mappings() .first() ) if not project: raise ValueError("Project does not belong to the requested evaluation workspace") templates = await available_templates(db, UUID(args.workspace_id)) templates = [ SimpleNamespace( **{ key: getattr(item, key) for key in ("ref", "name", "description", "framework", "tags") } ) for item in templates ] revisions = list( (await db.execute(text("SELECT version_num FROM alembic_version"))).scalars() ) await db.rollback() await engine.dispose() metadata = { "phase": "13A", "started_at": datetime.now(UTC).isoformat(), "corpus_sha256": corpus_hash, "corpus_size": len(rows), "strata": dict(Counter(r["stratum"] for r in rows)), "discovery_model": settings.project_intake_model, "timeout_seconds": settings.project_intake_timeout_seconds, "templates": "Live workspace-eligible template catalog, read only", "template_count": len(templates), "database_alembic_revisions": revisions, "project_type": project["type"], "workspace_id": args.workspace_id, "project_id": args.project_id, "workspace_verified": True, "discovery_mode": "live production adapter + Core finalizer" if args.live_discovery else "not_run", "configured_codegen_provider": settings.ai_provider.value, "configured_codegen_backend": settings.codegen_execution_backend, "configured_codegen_executable_available": bool(shutil.which(settings.cursor_cli_path)), "configured_codegen_wrapper_available": bool( settings.cursor_agent_script and Path(settings.cursor_agent_script).is_file() ), "gold_policy": ( "Predeclared required product blocks; structural dependencies excluded; " "no gold supplied to model." ), "turn_policy": ( "One initial user turn; clarification retained, no invented answers or repair." ), "raw_control_policy": ( "Exact prompt in ProductDefinition.purpose only; " "no inferred actors/entities/auth or normalized matcher phrases." ), "source_hashes": { p: hashlib.sha256((ROOT / p).read_bytes()).hexdigest() for p in ( "src/domain/app_blocks/planner.py", "src/domain/app_blocks/compiler.py", "src/integrations/project_intake_client.py", "src/domain/projects/discovery_readiness.py", "scripts/measure_generation_reliability_13a.py", "scripts/phase_13a_corpus.txt", ) }, } write_json(output / "metadata.json", metadata) records = [] # These controls are never substituted for missing live model responses. for row in rows: raw = DiscoverySpec(product=ProductDefinition(purpose=row["prompt"])) observation = plan_observation(raw, row["expected_product_blocks"]) records.append( {**row, "raw_text_planner_control": observation, "discovery": {"status": "not_run"}} ) write_json(output / "raw_controls.json", records) semaphore = asyncio.Semaphore(4) start_lock = asyncio.Lock() last_start = [0.0] async def one(record): async with semaphore: async with start_lock: await asyncio.sleep(max(0, last_start[0] + 1.1 - time.monotonic())) last_start[0] = time.monotonic() state = {"messages": [{"role": "user", "content": record["prompt"]}]} payload = build_discovery_payload(project["type"], "generated", state, templates) started = time.monotonic() trace = {} try: turn = await asyncio.wait_for( discover_project(payload, settings, trace=trace), timeout=settings.project_intake_timeout_seconds, ) result = finalize_discovery_result(turn, payload) record["discovery"] = { "status": "success", "raw_turn": turn.model_dump(mode="json"), "result": result.model_dump(mode="json"), "usage": (trace.get("response") or {}).get("usage"), } except Exception as exc: record["discovery"] = { "status": "error", "error_type": type(exc).__name__, "http_status": getattr(exc, "status_code", None), } record["discovery"]["elapsed_seconds"] = round(time.monotonic() - started, 3) if record["discovery"]["status"] == "success": try: record["live_planner"] = await asyncio.to_thread( plan_observation, result.spec, record["expected_product_blocks"] ) except Exception as exc: record["planner_error"] = type(exc).__name__ record["later_stage_status"] = { "codegen": "not_attempted_pending_environment_parity", "build": "not_attempted_no_generated_project", "runtime": "not_attempted_no_generated_project", "repair": "not_attempted_baseline_only", } write_json(output / "trials" / f"{record['id']}.json", record) print( json.dumps( { "id": record["id"], "status": record["discovery"]["status"], "error": record["discovery"].get("error_type"), } ), flush=True, ) if args.live_discovery: await asyncio.gather(*(one(record) for record in records)) admitted = [r for r in records if r["oracle_kind"] == "admitted"] live = [r for r in admitted if "live_planner" in r] ready = [r for r in live if r["discovery"]["result"]["decision"] == "ready"] completed = [r for r in records if r["discovery"]["status"] == "success"] scorecard = { "status": "partial_baseline_downstream_unmeasured", "phase_13b_authorized": False, "corpus_size": len(records), "capability_scoring_eligible": len(admitted), "discovery_call_completion_rate": rate( len(completed), len(records) if args.live_discovery else 0 ), "first_turn_ready_rate": rate(len(ready), len(admitted)), "intent_understanding_rate": rate( 0, 0, "Requires independent semantic review; block agreement alone is not understanding.", ), "live_planner_on_returned_specs": score([r["live_planner"] for r in live]), "live_planner_on_ready_specs": score([r["live_planner"] for r in ready]), "raw_text_planner_control_only": score([r["raw_text_planner_control"] for r in admitted]), "errors": dict( Counter( r["discovery"].get("error_type") for r in records if r["discovery"]["status"] == "error" ) ), "decisions": dict(Counter(r["discovery"]["result"]["decision"] for r in completed)), "later_stages": { name: rate( 0, 0, "No production Codegen/build/runtime trial completed; " "prerequisite environment not established.", ) for name in LATER_METRICS }, "raw_control_by_stratum": { group: score([r["raw_text_planner_control"] for r in admitted if r["stratum"] == group]) for group in sorted({r["stratum"] for r in admitted}) }, } write_json(output / "results.json", records) write_json(output / "scorecard.json", scorecard) print(json.dumps(scorecard), flush=True) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", required=True) parser.add_argument("--workspace-id", required=True) parser.add_argument("--project-id", required=True) parser.add_argument("--live-discovery", action="store_true") asyncio.run(measure(parser.parse_args()))