import argparse import json import sys from datetime import datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from run_selector_demo import load_selector_rules, recommend TARGETED_CASES = { "R04-BI-010", "R04-BI-011", "R04-BI-012", "R04-BI-013", "R04-BI-014", "R04-BI-015", "R04-BI-016", "R04-BI-017", "R04-BI-025", "R04-BI-026", "R04-BI-027", "R04-BI-028", "R04-BI-029", "R04-BI-030", "R04-BI-031", "R04-BI-032", "R04-BI-033", "R04-BI-038", } OUTPUT_DIR = Path("reports") / "round05_1_selector_patch_audit" BEFORE_SNAPSHOT = "before_selector_behavior.json" AFTER_SNAPSHOT = "after_selector_behavior.json" DIFF_REPORT_JSON = "round05_1_behavior_diff.json" DIFF_REPORT_MD = "round05_1_behavior_diff.md" DIFF_FIELDS = [ "selected_model_ids", "rejected_model_ids", "scores", "no_call", "reasons", "penalties", ] def read_json(path): return json.loads(Path(path).read_text(encoding="utf-8")) def write_json(path, data): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def route_from_selected(selected_model_ids, no_call): if no_call or not selected_model_ids: return "no_call" return "+".join(selected_model_ids) def model_field_map(selector_result, field_name): mapping = {} for item in selector_result.get("selected_models", []) + selector_result.get("rejected_models", []): mapping[item["model_id"]] = item.get(field_name, []) return mapping def model_score_map(selector_result): scores = {} for item in selector_result.get("selected_models", []) + selector_result.get("rejected_models", []): scores[item["model_id"]] = item.get("score", 0.0) return scores def build_behavior_record(source, case_id, input_text, selector_result, targeted=False, metadata=None): selected_model_ids = [item["model_id"] for item in selector_result.get("selected_models", [])] rejected_model_ids = [item["model_id"] for item in selector_result.get("rejected_models", [])] no_call = bool(selector_result.get("no_call")) return { "case_id": case_id, "source": source, "input": input_text, "targeted": targeted, "selected_model_ids": selected_model_ids, "rejected_model_ids": rejected_model_ids, "route": route_from_selected(selected_model_ids, no_call), "no_call": no_call, "scores": model_score_map(selector_result), "reasons": model_field_map(selector_result, "reasons"), "penalties": model_field_map(selector_result, "penalties"), "metadata": metadata or {}, } def load_round04_cases(root): data = read_json(root / "selector" / "round04_blind_inputs.json") cases = [] for item in data.get("inputs", []): case_id = item["input_id"] cases.append({ "source": "round04_blind_pool", "case_id": case_id, "input": item["input_text"], "metadata": { "category_for_owner_review": item.get("category_for_owner_review"), "why_included": item.get("why_included"), }, }) return cases def load_calibration_cases(root): data = read_json(root / "selector" / "selector_calibration_inputs.json") cases = [] for item in data.get("inputs", []): cases.append({ "source": "selector_calibration_inputs", "case_id": item["case_id"], "input": item["input"], "metadata": { "category": item.get("category"), "expected_selector_behavior": item.get("expected_selector_behavior"), "expected_notes": item.get("expected_notes"), }, }) return cases def load_regression_cases(root): data = read_json(root / "tests" / "regression_cases.json") cases = [] for item in data.get("regression_cases", []): cases.append({ "source": "aggregate_regression_cases", "case_id": item["case_id"], "input": item["input"], "metadata": { "model_id": item.get("model_id"), "case_type": item.get("case_type"), "expected_primary_model": item.get("expected_primary_model"), "negative_expected_models": item.get("negative_expected_models", []), }, }) return cases def collect_cases(root): cases = [] cases.extend(load_round04_cases(root)) cases.extend(load_calibration_cases(root)) cases.extend(load_regression_cases(root)) return cases def build_snapshot(root, snapshot_type): records = [] for case in collect_cases(root): result = recommend(root, case["input"]) records.append(build_behavior_record( source=case["source"], case_id=case["case_id"], input_text=case["input"], selector_result=result, targeted=case["case_id"] in TARGETED_CASES, metadata=case["metadata"], )) selector_rules = load_selector_rules(root) return { "snapshot_type": snapshot_type, "generated_at": datetime.now().isoformat(timespec="seconds"), "selector_version": selector_rules.get("selector_version"), "case_count": len(records), "targeted_cases": sorted(TARGETED_CASES), "records": records, } def changed_fields(before_record, after_record): return [ field for field in DIFF_FIELDS if before_record.get(field) != after_record.get(field) ] def diff_behavior_records(before_records, after_records, targeted_cases=None): targeted_cases = set(targeted_cases or TARGETED_CASES) before_by_id = {record["case_id"]: record for record in before_records} after_by_id = {record["case_id"]: record for record in after_records} common_ids = sorted(set(before_by_id) & set(after_by_id)) missing_before = sorted(set(after_by_id) - set(before_by_id)) missing_after = sorted(set(before_by_id) - set(after_by_id)) changed = [] for case_id in common_ids: before_record = before_by_id[case_id] after_record = after_by_id[case_id] fields = changed_fields(before_record, after_record) if not fields: continue is_targeted = case_id in targeted_cases item = { "case_id": case_id, "source": after_record.get("source"), "targeted": is_targeted, "expected_by_policy": is_targeted, "changed_fields": fields, "before": {field: before_record.get(field) for field in DIFF_FIELDS}, "after": {field: after_record.get(field) for field in DIFF_FIELDS}, "disposition_required": not is_targeted, } if not is_targeted: item["owner_disposition"] = "defer" item["ccra_disposition"] = "defer" item["follow_up"] = "Owner / CCRA disposition required before closeout." changed.append(item) targeted_changes = [item for item in changed if item["targeted"]] non_target_changes = [item for item in changed if not item["targeted"]] return { "summary": { "status": "ATTENTION" if non_target_changes else "PASS", "total_cases_compared": len(common_ids), "changed_cases": len(changed), "targeted_changes": len(targeted_changes), "non_target_changes": len(non_target_changes), "missing_before_cases": len(missing_before), "missing_after_cases": len(missing_after), }, "changed_cases": changed, "targeted_changes": targeted_changes, "non_target_changes": non_target_changes, "missing_before_cases": missing_before, "missing_after_cases": missing_after, } def load_snapshot(path): data = read_json(path) return data.get("records", []) def write_snapshot_report(path, snapshot): lines = [ f"# Round 05.1 Selector Behavior Snapshot: {snapshot['snapshot_type']}", "", f"Selector version: `{snapshot.get('selector_version')}`", f"Cases captured: {snapshot.get('case_count')}", "", "## Sources", "", ] source_counts = {} for record in snapshot.get("records", []): source_counts[record["source"]] = source_counts.get(record["source"], 0) + 1 for source, count in sorted(source_counts.items()): lines.append(f"- `{source}`: {count}") lines.extend(["", "## Targeted Cases", ""]) for case_id in snapshot.get("targeted_cases", []): lines.append(f"- `{case_id}`") path.write_text("\n".join(lines) + "\n", encoding="utf-8") def write_diff_markdown(path, diff): summary = diff["summary"] lines = [ "# Round 05.1 Selector Behavior Diff", "", f"Status: `{summary['status']}`", "", f"Total cases compared: {summary['total_cases_compared']}", f"Changed cases: {summary['changed_cases']}", f"Targeted changes: {summary['targeted_changes']}", f"Non-target changes: {summary['non_target_changes']}", f"Missing before cases: {summary['missing_before_cases']}", f"Missing after cases: {summary['missing_after_cases']}", "", "## Targeted Changes", "", ] for item in diff["targeted_changes"]: lines.append(f"- `{item['case_id']}` ({item['source']}): changed_fields={item['changed_fields']}") if not diff["targeted_changes"]: lines.append("- None") lines.extend(["", "## Non-Target Changes", ""]) for item in diff["non_target_changes"]: lines.append( f"- `{item['case_id']}` ({item['source']}): changed_fields={item['changed_fields']}; " f"owner_disposition=`{item['owner_disposition']}`; ccra_disposition=`{item['ccra_disposition']}`" ) if not diff["non_target_changes"]: lines.append("- None") lines.extend(["", "## Missing Cases", ""]) if diff["missing_before_cases"] or diff["missing_after_cases"]: lines.append(f"- missing_before_cases: `{diff['missing_before_cases']}`") lines.append(f"- missing_after_cases: `{diff['missing_after_cases']}`") else: lines.append("- None") path.write_text("\n".join(lines) + "\n", encoding="utf-8") def write_snapshot(root, output_dir, snapshot_type): output_dir.mkdir(parents=True, exist_ok=True) snapshot = build_snapshot(root, snapshot_type) json_name = BEFORE_SNAPSHOT if snapshot_type == "before" else AFTER_SNAPSHOT md_name = json_name.replace(".json", ".md") write_json(output_dir / json_name, snapshot) write_snapshot_report(output_dir / md_name, snapshot) return output_dir / json_name def write_diff(output_dir): before_path = output_dir / BEFORE_SNAPSHOT after_path = output_dir / AFTER_SNAPSHOT diff = diff_behavior_records( load_snapshot(before_path), load_snapshot(after_path), targeted_cases=TARGETED_CASES, ) write_json(output_dir / DIFF_REPORT_JSON, diff) write_diff_markdown(output_dir / DIFF_REPORT_MD, diff) return output_dir / DIFF_REPORT_JSON def main(argv=None): parser = argparse.ArgumentParser(description="Capture and diff Round 05.1 selector behavior.") parser.add_argument("mode", choices=["baseline", "after", "diff"]) parser.add_argument("--output-dir", default=str(OUTPUT_DIR)) args = parser.parse_args(argv) root = Path(__file__).resolve().parents[1] output_dir = root / args.output_dir if args.mode == "baseline": path = write_snapshot(root, output_dir, "before") print(f"before snapshot written to {path}") return 0 if args.mode == "after": path = write_snapshot(root, output_dir, "after") print(f"after snapshot written to {path}") return 0 path = write_diff(output_dir) print(f"behavior diff written to {path}") return 0 if __name__ == "__main__": sys.exit(main())