351 lines
12 KiB
Python
351 lines
12 KiB
Python
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())
|