the-mindscape-of-bro-tsong/scripts/run_round04_blind_routing.py

import json
import sys
from collections import Counter
from pathlib import Path


SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from run_selector_demo import recommend  # noqa: E402


REPORT_NAME = "Round04_blind_routing_evaluation_report_2026-06-18.md"


def read_json(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))


def load_blind_inputs(root):
    return read_json(root / "selector" / "round04_blind_inputs.json")


def selected_model_ids(result):
    return [item["model_id"] for item in result.get("selected_models", [])]


def rejected_model_ids(result):
    return [item["model_id"] for item in result.get("rejected_models", [])]


def evaluate_blind_input(root, item):
    selector_result = recommend(root, item["input_text"])
    return {
        "input_id": item["input_id"],
        "category_for_owner_review": item["category_for_owner_review"],
        "control_case_type": item.get("control_case_type"),
        "input_text": item["input_text"],
        "why_included": item.get("why_included", ""),
        "selected_models": selected_model_ids(selector_result),
        "rejected_models": rejected_model_ids(selector_result),
        "no_call": selector_result.get("no_call"),
        "routing_notes": selector_result.get("routing_notes", ""),
        "selected_model_details": selector_result.get("selected_models", []),
        "rejected_model_details": selector_result.get("rejected_models", []),
    }


def format_model_details(details):
    if not details:
        return ["  - none"]
    lines = []
    for detail in details:
        lines.append(
            f"  - `{detail['model_id']}` score={detail['score']} decision={detail['decision']}"
        )
        if detail.get("reasons"):
            lines.append("    - reasons: " + "; ".join(detail["reasons"]))
        if detail.get("penalties"):
            lines.append("    - penalties: " + "; ".join(detail["penalties"]))
    return lines


def render_report(input_set, results):
    selected_counter = Counter()
    category_counter = Counter()
    no_call_count = 0
    for result in results:
        category_counter[result["category_for_owner_review"]] += 1
        if result["no_call"]:
            no_call_count += 1
        if not result["selected_models"]:
            selected_counter["none"] += 1
        for model_id in result["selected_models"]:
            selected_counter[model_id] += 1

    lines = [
        "# Round 04 Blind Routing Evaluation Report",
        "",
        "Date: 2026-06-18",
        "",
        f"Blind input set: `{input_set['blind_input_set_id']}`",
        "",
        "Expected routing labels: `not included before first blind run`",
        "",
        "This report records current selector routing only. It does not score correctness, does not generate final answers, does not modify selector rules, and does not upgrade model status.",
        "",
        "## Scope",
        "",
        "- Round 04 target: test routing stability on new inputs.",
        "- Runtime surface: current rule-based selector plus model-level QPI / IA gates.",
        "- QPI contract status: represented here by QPI selection and routing details; no full QPI answer generation is performed.",
        "- IA status: represented here by selector selection / rejection only; no Intellectual Archaeology answer generation is performed.",
        "",
        "## Summary",
        "",
        f"Inputs evaluated: {len(results)}",
        f"No-call results: {no_call_count}",
        "",
        "### Selected Model Distribution",
        "",
    ]

    for key, count in sorted(selected_counter.items()):
        lines.append(f"- `{key}`: {count}")

    lines.extend([
        "",
        "### Coverage Categories",
        "",
    ])
    for key, count in sorted(category_counter.items()):
        lines.append(f"- `{key}`: {count}")

    lines.extend([
        "",
        "## Case Results",
        "",
    ])

    for result in results:
        lines.extend([
            f"### {result['input_id']}",
            "",
            f"- category_for_owner_review: `{result['category_for_owner_review']}`",
        ])
        if result.get("control_case_type"):
            lines.append(f"- control_case_type: `{result['control_case_type']}`")
        lines.extend([
            f"- input_text: `{result['input_text']}`",
            f"- selected_models: `{result['selected_models']}`",
            f"- rejected_models: `{result['rejected_models']}`",
            f"- no_call: `{result['no_call']}`",
            f"- routing_notes: {result['routing_notes']}",
            "",
            "Selected model details:",
        ])
        lines.extend(format_model_details(result["selected_model_details"]))
        lines.extend(["", "Rejected model details:"])
        lines.extend(format_model_details(result["rejected_model_details"]))
        lines.append("")

    lines.extend([
        "## Review Reminder",
        "",
        "This is the first blind routing output. Failure labels, expected routing, and rule changes should be added only after owner / GPT / CCRA review of these observed results.",
        "",
    ])

    return "\n".join(lines)


def write_report(root, input_set, results):
    report_path = root / "reports" / REPORT_NAME
    report_path.write_text(render_report(input_set, results), encoding="utf-8")
    return report_path


def main():
    root = Path(__file__).resolve().parents[1]
    input_set = load_blind_inputs(root)
    results = [
        evaluate_blind_input(root, item)
        for item in input_set.get("inputs", [])
    ]
    report_path = write_report(root, input_set, results)
    print(f"round04 blind routing report written to {report_path}")
    print(f"inputs evaluated: {len(results)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())