"""Gate B2 measurement: raw-text Planner selection against the frozen 13A-02 oracle. Does not overwrite 13A-01, 13A-02, corpus, or gold. Does not change matchers. The primary denominator remains the 108 admitted prompts. Discovery-spec scoring is a secondary cohort and never replaces that denominator. """ from __future__ import annotations import argparse import hashlib import json import sys from collections import Counter from datetime import UTC, datetime from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.domain.projects.discovery_schemas import DiscoverySpec, ProductDefinition # noqa: E402 from scripts.measure_generation_reliability_13a import ( # noqa: E402 corpus, plan_observation, rate, score, ) PROTECTED_OUTPUTS = { ROOT / "docs/audits/fixtures/13a_run_01", ROOT / "docs/audits/fixtures/13a_run_02", } FROZEN_13A02 = ROOT / "docs/audits/fixtures/13a_run_02" FROZEN_RAW_CONTROL = { "raw_planner_exact_product": "76/108", "raw_planner_precision": "104/118", "raw_planner_recall": "104/139", "crm_exact_product": "3/10", "crm_recall": "8/23", "mixed_quote_exact": "10/10", } CONTACT_QUOTE_BLOCKS = frozenset({"forms.contact", "forms.quote_request"}) CORPUS_PATH = ROOT / "scripts/phase_13a_corpus.txt" 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 fraction(metric: dict) -> str | None: if metric.get("denominator") in (None, 0) and metric.get("rate") is None: return None return f"{metric['numerator']}/{metric['denominator']}" def compact_observation(observation: dict) -> dict: return { "predicted_product_blocks": observation["predicted_product_blocks"], "true_positive": observation["true_positive"], "false_positive": observation["false_positive"], "false_negative": observation["false_negative"], "exact_product_match": observation["exact_product_match"], "plan_satisfiable": observation["plan_satisfiable"], "empty_selected_plan": not observation["predicted_product_blocks"], } def raw_spec(prompt: str) -> DiscoverySpec: return DiscoverySpec(product=ProductDefinition(purpose=prompt)) def load_frozen_raw_controls(source: Path = FROZEN_13A02) -> dict[str, dict]: trials = {} for path in sorted((source / "trials").glob("13A-*.json")): row = json.loads(path.read_text(encoding="utf-8")) trials[row["id"]] = row if len(trials) != 120: raise ValueError(f"Expected 120 frozen 13A-02 trials, found {len(trials)}") return trials def load_discovery_specs(source: Path, repeat: str) -> dict[str, dict]: """Load finalized Discovery specs from a B1 receipt. Accepts either ``source/run_0N/trials`` or ``source/trials``. Missing specs stay unmeasured; this never invents a ready spec. """ candidates = [ source / repeat / "trials", source / "trials", ] trials_dir = next((path for path in candidates if path.is_dir()), None) if trials_dir is None: return {} specs = {} for path in sorted(trials_dir.glob("13A-*.json")): row = json.loads(path.read_text(encoding="utf-8")) result = (row.get("discovery") or {}).get("result") or {} turn = (row.get("discovery") or {}).get("raw_turn") or {} spec = result.get("spec") or turn.get("spec") if spec: specs[row["id"]] = spec return specs def observe_rows(rows: list[dict], *, spec_by_id: dict[str, dict] | None = None) -> list[dict]: observed = [] for row in rows: if spec_by_id is None: spec = raw_spec(row["prompt"]) else: payload = spec_by_id.get(row["id"]) if payload is None: observed.append( { **row, "observation": None, "unmeasured_reason": "no_finalized_discovery_spec", } ) continue spec = DiscoverySpec.model_validate(payload) observation = compact_observation(plan_observation(spec, row["expected_product_blocks"])) observed.append({**row, "observation": observation}) return observed def score_admitted(rows: list[dict]) -> dict: admitted = [ row["observation"] for row in rows if row["oracle_kind"] == "admitted" and row.get("observation") ] return score(admitted) def stratum_score(rows: list[dict], stratum: str) -> dict: admitted = [ row["observation"] for row in rows if row["oracle_kind"] == "admitted" and row["stratum"] == stratum and row.get("observation") ] return score(admitted) def count_false_positives(rows: list[dict], *, stratum: str, blocks: frozenset[str]) -> dict: admitted = [ row for row in rows if row["oracle_kind"] == "admitted" and row["stratum"] == stratum and row.get("observation") ] hits = [ row["id"] for row in admitted if set(row["observation"]["false_positive"]) & blocks ] return { "numerator": len(hits), "denominator": len(admitted), "rate": len(hits) / len(admitted) if admitted else None, "ids": hits, "unmeasured_reason": None if admitted else "no admitted rows in stratum", } def empty_selected_plan(rows: list[dict]) -> dict: admitted = [ row for row in rows if row["oracle_kind"] == "admitted" and row.get("observation") ] empty = [row["id"] for row in admitted if row["observation"]["empty_selected_plan"]] return rate(len(empty), len(admitted)) | {"ids": empty} def mixed_quote_exact(rows: list[dict]) -> dict: mixed = stratum_score(rows, "mixed")["exact_product_match_rate"] quote = stratum_score(rows, "quote")["exact_product_match_rate"] both = mixed["numerator"] == mixed["denominator"] == 10 and quote["numerator"] == 10 return { "mixed_exact": fraction(mixed), "quote_exact": fraction(quote), "mixed_quote_exact": "10/10" if both else f"{mixed['numerator'] + quote['numerator']}/20", "preserved": both, } def b2_columns(rows: list[dict]) -> dict: overall = score_admitted(rows) crm = stratum_score(rows, "crm") documents = stratum_score(rows, "documents") catalog = stratum_score(rows, "catalog") # GR-05: "support" stratum did not exist pre-B2; stratum_score() already # tolerates an absent/empty stratum (returns an unmeasured score), so this # is safe on any corpus that predates GR-05 too. support = stratum_score(rows, "support") return { "raw_planner_precision": fraction(overall["planner_precision"]), "raw_planner_recall": fraction(overall["planner_recall"]), "raw_planner_exact_product": fraction(overall["exact_product_match_rate"]), "crm_exact_product": fraction(crm["exact_product_match_rate"]), "crm_recall": fraction(crm["planner_recall"]), "documents_exact": fraction(documents["exact_product_match_rate"]), "documents_recall": fraction(documents["planner_recall"]), "catalog_exact": fraction(catalog["exact_product_match_rate"]), "catalog_recall": fraction(catalog["planner_recall"]), "support_ticket_exact": fraction(support["exact_product_match_rate"]), "support_ticket_recall": fraction(support["planner_recall"]), "booking_contact_false_positives": count_false_positives( rows, stratum="booking", blocks=frozenset({"forms.contact"}) ), "negation_false_positives": count_false_positives( rows, stratum="boundary", blocks=CONTACT_QUOTE_BLOCKS ), "empty_selected_plan": empty_selected_plan(rows), "mixed_quote_exact": mixed_quote_exact(rows), "raw_text_planner_control_only": overall, "raw_control_by_stratum": { group: score( [ row["observation"] for row in rows if row["oracle_kind"] == "admitted" and row["stratum"] == group and row.get("observation") ] ) for group in sorted( {row["stratum"] for row in rows if row["oracle_kind"] == "admitted"} ) }, } def predicted_disagreements(rows: list[dict], frozen: dict[str, dict]) -> list[dict]: disagreements = [] for row in rows: frozen_row = frozen.get(row["id"]) if frozen_row is None: # Rows admitted after the frozen 13A-02 120-row set (e.g. GR-05's # appended support-ticket witnesses) have nothing to compare # against -- that is not drift, just no historical control yet. continue control = frozen_row["raw_text_planner_control"] observed = row.get("observation") if observed is None: continue if observed["predicted_product_blocks"] != control["predicted_product_blocks"]: disagreements.append( { "id": row["id"], "stratum": row["stratum"], "frozen": control["predicted_product_blocks"], "observed": observed["predicted_product_blocks"], } ) return disagreements def compare_frozen_control(columns: dict) -> dict: observed = { "raw_planner_exact_product": columns["raw_planner_exact_product"], "raw_planner_precision": columns["raw_planner_precision"], "raw_planner_recall": columns["raw_planner_recall"], "crm_exact_product": columns["crm_exact_product"], "crm_recall": columns["crm_recall"], "mixed_quote_exact": columns["mixed_quote_exact"]["mixed_quote_exact"], } return { "required": FROZEN_RAW_CONTROL, "observed": observed, "reproduced": observed == FROZEN_RAW_CONTROL, "drifted": [key for key, value in FROZEN_RAW_CONTROL.items() if observed[key] != value], } def discovery_loss(raw_rows: list[dict], spec_rows: list[dict]) -> dict: """Gold held on raw control but lost after applying a Discovery spec.""" raw_by_id = {row["id"]: row for row in raw_rows if row.get("observation")} lost = [] compared = 0 for row in spec_rows: raw = raw_by_id.get(row["id"]) if raw is None or row.get("observation") is None: continue if row["oracle_kind"] != "admitted": continue compared += 1 raw_obs = raw["observation"] spec_obs = row["observation"] if raw_obs["exact_product_match"] and not spec_obs["exact_product_match"]: lost.append( { "id": row["id"], "stratum": row["stratum"], "gold": row["expected_product_blocks"], "raw_predicted": raw_obs["predicted_product_blocks"], "spec_predicted": spec_obs["predicted_product_blocks"], "spec_false_negative": spec_obs["false_negative"], "spec_false_positive": spec_obs["false_positive"], } ) return { "numerator": len(lost), "denominator": compared, "rate": len(lost) / compared if compared else None, "ids": [item["id"] for item in lost], "cells": lost, "unmeasured_reason": None if compared else "no overlapping Discovery specs", } def source_hashes() -> dict[str, str]: paths = ( "src/domain/app_blocks/planner.py", "src/domain/app_blocks/compiler.py", "scripts/phase_13a_corpus.txt", "scripts/measure_generation_reliability_13a.py", "scripts/measure_planner_selection_13b_b2.py", "docs/audits/fixtures/13a_run_02_freeze.json", ) return {path: hashlib.sha256((ROOT / path).read_bytes()).hexdigest() for path in paths} def measure( out: Path, *, kind: str = "before", discovery_source: Path | None = None, ) -> dict: out = out.resolve() protected = out in PROTECTED_OUTPUTS or any( out == path or path in out.parents for path in PROTECTED_OUTPUTS ) if protected: raise ValueError("Refuse writes into frozen 13A-01 or 13A-02 receipts") if (out / "scorecard.json").exists(): raise ValueError("Existing measurement must not be overwritten; select a new output folder") rows = corpus(CORPUS_PATH) frozen = load_frozen_raw_controls() raw_rows = observe_rows(rows) columns = b2_columns(raw_rows) comparison = compare_frozen_control(columns) disagreements = predicted_disagreements(raw_rows, frozen) discovery_columns = { "discovery_spec_planner_precision": None, "discovery_spec_planner_recall": None, "discovery_spec_planner_exact_product": None, "discovery_spec_planner_reproducibility": None, "discovery_to_planner_semantic_loss": None, "unmeasured_reason": ( "B2 before-score is raw-text control only; Discovery-spec cohort is GR-04." ), } discovery_repeats: dict[str, list[dict]] = {} if discovery_source is not None: source = Path(discovery_source) repeat_scores = {} for repeat in ("run_01", "run_02"): specs = load_discovery_specs(source, repeat) if not specs: continue observed = observe_rows(rows, spec_by_id=specs) discovery_repeats[repeat] = observed scored = score_admitted(observed) repeat_scores[repeat] = { "measured_admitted": sum( 1 for row in observed if row["oracle_kind"] == "admitted" and row.get("observation") ), "precision": fraction(scored["planner_precision"]), "recall": fraction(scored["planner_recall"]), "exact_product": fraction(scored["exact_product_match_rate"]), "score": scored, "loss_vs_raw": discovery_loss(raw_rows, observed), } pair_ids = [] if "run_01" in discovery_repeats and "run_02" in discovery_repeats: first = {row["id"]: row for row in discovery_repeats["run_01"]} second = {row["id"]: row for row in discovery_repeats["run_02"]} agreed = 0 compared = 0 for prompt_id, left in first.items(): right = second.get(prompt_id) if ( left.get("observation") is None or right is None or right.get("observation") is None ): continue compared += 1 if ( left["observation"]["predicted_product_blocks"] == right["observation"]["predicted_product_blocks"] ): agreed += 1 else: pair_ids.append(prompt_id) reproducibility = rate(agreed, compared) | {"disagreement_ids": pair_ids} else: reproducibility = rate( 0, 0, "Both B1 repeats are required to score discovery_spec_planner_reproducibility.", ) primary = repeat_scores.get("run_01") or next(iter(repeat_scores.values()), None) if primary: discovery_columns = { "discovery_spec_planner_precision": primary["precision"], "discovery_spec_planner_recall": primary["recall"], "discovery_spec_planner_exact_product": primary["exact_product"], "discovery_spec_planner_reproducibility": reproducibility, "discovery_to_planner_semantic_loss": primary["loss_vs_raw"], "repeats": repeat_scores, "unmeasured_reason": None, } else: discovery_columns["unmeasured_reason"] = ( f"No finalized Discovery specs found under {source}" ) metadata = { "phase": "13B", "gate": "B2", "measurement_kind": f"raw_planner_{kind}", "started_at": datetime.now(UTC).isoformat(), "corpus_size": len(rows), "capability_scoring_eligible": 108, "primary_denominator": "108 admitted prompts; raw ProductDefinition.purpose only", "frozen_13a02_control": FROZEN_RAW_CONTROL, "planner_changed": False, "catalog_changed": False, "corpus_changed": False, "gold_changed": False, "matcher_changed": False, "gold_entered_runtime": False, "source_hashes": source_hashes(), "frozen_13a02_source_hashes": json.loads( (ROOT / "docs/audits/fixtures/13a_run_02/metadata.json").read_text(encoding="utf-8") )["source_hashes"], "strata": dict(Counter(row["stratum"] for row in rows)), } scorecard = { "phase": "13B", "gate": "B2", "measurement_kind": metadata["measurement_kind"], "before_score_reproduced": comparison["reproduced"] if kind == "before" else None, "frozen_13a02_comparison": comparison, "predicted_set_disagreements_vs_13a02": { "numerator": len(disagreements), "denominator": 120, "ids": [item["id"] for item in disagreements], "cells": disagreements, }, **{key: columns[key] for key in ( "raw_planner_precision", "raw_planner_recall", "raw_planner_exact_product", "crm_exact_product", "crm_recall", "documents_exact", "documents_recall", "catalog_exact", "catalog_recall", "support_ticket_exact", "support_ticket_recall", "booking_contact_false_positives", "negation_false_positives", "empty_selected_plan", "mixed_quote_exact", )}, **discovery_columns, "raw_text_planner_control_only": columns["raw_text_planner_control_only"], "raw_control_by_stratum": columns["raw_control_by_stratum"], "do_not": [ "overwrite 13A-01 or 13A-02", "change corpus or gold", "begin matcher changes before before-score reproduces", "start 13C or 13E", "invent product surfaces", ], } write_json(out / "metadata.json", metadata) write_json( out / "trials.json", [ { "id": row["id"], "stratum": row["stratum"], "oracle_kind": row["oracle_kind"], "prompt": row["prompt"], "expected_product_blocks": row["expected_product_blocks"], "observation": row["observation"], "unmeasured_reason": row.get("unmeasured_reason"), } for row in raw_rows ], ) if discovery_repeats: write_json( out / "discovery_spec_trials.json", { repeat: [ { "id": row["id"], "stratum": row["stratum"], "oracle_kind": row["oracle_kind"], "expected_product_blocks": row["expected_product_blocks"], "observation": row["observation"], "unmeasured_reason": row.get("unmeasured_reason"), } for row in observed ] for repeat, observed in discovery_repeats.items() }, ) write_json(out / "scorecard.json", scorecard) return scorecard def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", required=True) parser.add_argument("--kind", choices=("before", "after", "checkpoint"), default="before") parser.add_argument("--discovery-source", default=None) args = parser.parse_args() scorecard = measure( Path(args.out), kind=args.kind, discovery_source=Path(args.discovery_source) if args.discovery_source else None, ) print(json.dumps({ "kind": args.kind, "reproduced": scorecard["frozen_13a02_comparison"]["reproduced"], "observed": scorecard["frozen_13a02_comparison"]["observed"], "disagreements": scorecard["predicted_set_disagreements_vs_13a02"]["numerator"], }, indent=2), flush=True) if args.kind == "before" and not scorecard["frozen_13a02_comparison"]["reproduced"]: raise SystemExit("BEFORE-SCORE DID NOT REPRODUCE FROZEN 13A-02 CONTROL. STOP MATCHER WORK.") if __name__ == "__main__": main()