build_calibration_error_queue.py

"""Build a review queue from scanner threshold-calibration errors.

The output is not training data. It is a human-review staging file for deciding
which calibration errors should become curated clean/poison examples, weak-label
quarantine rows, or scanner-policy test cases.
"""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


DEFAULT_CALIBRATION = Path(
    "research/_results/activation_scanner_calibration/"
    "20260603T102832Z-qwen-pooled-l13-15-threshold-calibration.json"
)
DEFAULT_OUTPUT = Path(
    "research/datasets/calibration_error_review_queue_qwen_pooled_policy_v3_warn030_2026-06-03.json"
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Convert calibration top errors into a data-curriculum review queue."
    )
    parser.add_argument(
        "--calibration",
        type=Path,
        default=DEFAULT_CALIBRATION,
        help="Calibration JSON produced by research.calibrate_scanner_thresholds.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=DEFAULT_OUTPUT,
        help="Review-queue JSON output path.",
    )
    parser.add_argument(
        "--pretty",
        action="store_true",
        help="Pretty-print the generated queue summary.",
    )
    return parser.parse_args()


def utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def queue_action(error_type: str, row: dict[str, Any]) -> str:
    style = row.get("style", "")
    source = row.get("source", "")
    static_findings = row.get("static_findings") or []

    if error_type in {"false_warn_or_block", "false_blocks"}:
        if static_findings:
            return "review_static_context_false_positive"
        if style in {"bipia_clean_email", "masb_content_safe", "skill_inject_clean"}:
            return "promote_benign_lookalike_candidate"
        return "review_clean_false_positive"

    if error_type == "missed_poison":
        if style.startswith("bipia_text_attack"):
            return "review_import_quality_or_reconstruct_hidden_carrier"
        if style == "masb_metadata_malicious" or "metadata" in source:
            return "quarantine_weak_metadata_label_until_content_available"
        return "promote_hard_positive_candidate"

    if error_type == "warned_but_not_blocked_poison":
        return "promote_near_block_positive_candidate"

    return "review"


def queue_reason(error_type: str, row: dict[str, Any]) -> str:
    style = row.get("style", "")
    activation = float(row.get("activation_score", 0.0))
    static_score = float(row.get("static_score", 0.0))
    risk = float(row.get("risk_score", 0.0))

    if error_type in {"false_warn_or_block", "false_blocks"}:
        if static_score > activation:
            return "Clean row is flagged mostly by static rules; check negation/context before block."
        return "Clean row has high activation risk; use as benign lookalike if label is correct."

    if error_type == "missed_poison":
        if style.startswith("bipia_text_attack"):
            return "Poison label may depend on hidden benchmark context not present in scanned text."
        return "Poison row is allowed at current warn threshold; review for stronger training signal."

    if error_type == "warned_but_not_blocked_poison":
        return f"Poison row is close to block threshold but only reaches risk {risk:.3f}."

    return "Review calibration error."


def priority(error_type: str, row: dict[str, Any]) -> str:
    style = row.get("style", "")
    risk = float(row.get("risk_score", 0.0))

    if error_type == "false_blocks" and style in {"bipia_clean_email", "masb_content_safe"}:
        return "p0"
    if error_type == "missed_poison" and risk < 0.1:
        return "p0"
    if error_type == "warned_but_not_blocked_poison" and risk >= 0.8:
        return "p1"
    if error_type == "false_warn_or_block":
        return "p1"
    return "p2"


def build_queue(calibration: dict[str, Any]) -> dict[str, Any]:
    entries: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    top_errors = calibration.get("top_errors", {})

    for dataset_name, error_groups in top_errors.items():
        for error_type, rows in error_groups.items():
            for row in rows:
                key = (dataset_name, row.get("id", ""))
                if key in seen:
                    continue
                seen.add(key)
                entries.append(
                    {
                        "queue_id": f"{dataset_name}:{error_type}:{row.get('id', 'unknown')}",
                        "dataset": dataset_name,
                        "error_type": error_type,
                        "priority": priority(error_type, row),
                        "recommended_action": queue_action(error_type, row),
                        "reason": queue_reason(error_type, row),
                        "review_decision": "needs_review",
                        "artifact_id": calibration.get("artifact", {}).get("artifact_id"),
                        "warn_threshold": calibration.get("config", {}).get("warn_threshold"),
                        "block_threshold": calibration.get("config", {}).get("block_threshold"),
                        "row": {
                            "id": row.get("id"),
                            "label": row.get("label"),
                            "style": row.get("style"),
                            "family": row.get("family"),
                            "source": row.get("source"),
                            "activation_score": row.get("activation_score"),
                            "static_score": row.get("static_score"),
                            "risk_score": row.get("risk_score"),
                            "policy_decision": row.get("policy_decision"),
                            "policy_reasons": row.get("policy_reasons", []),
                            "static_findings": row.get("static_findings", []),
                            "text_preview": row.get("text_preview"),
                        },
                    }
                )

    entries.sort(
        key=lambda item: (
            {"p0": 0, "p1": 1, "p2": 2}.get(item["priority"], 9),
            item["dataset"],
            item["error_type"],
            item["row"].get("style") or "",
            item["row"].get("id") or "",
        )
    )

    by_priority: dict[str, int] = {}
    by_action: dict[str, int] = {}
    by_error_type: dict[str, int] = {}
    for entry in entries:
        by_priority[entry["priority"]] = by_priority.get(entry["priority"], 0) + 1
        by_action[entry["recommended_action"]] = (
            by_action.get(entry["recommended_action"], 0) + 1
        )
        by_error_type[entry["error_type"]] = by_error_type.get(entry["error_type"], 0) + 1

    return {
        "created_at": utc_now(),
        "source_calibration": calibration.get("created_at"),
        "artifact": calibration.get("artifact"),
        "purpose": (
            "Human-review queue for data-curriculum and decision-policy updates. "
            "Rows are not automatically promoted into training."
        ),
        "summary": {
            "total_entries": len(entries),
            "by_priority": dict(sorted(by_priority.items())),
            "by_error_type": dict(sorted(by_error_type.items())),
            "by_recommended_action": dict(sorted(by_action.items())),
        },
        "review_guidance": [
            "Promote clean false positives only if the row is truly benign for tool/MCP poisoning.",
            "Quarantine rows where the unsafe instruction is absent from the scanned text.",
            "Use near-block poison rows to improve the block policy only after adding benign twins.",
            "Keep metadata-only malicious rows out of gold training unless package content is available.",
        ],
        "entries": entries,
    }


def main() -> None:
    args = parse_args()
    calibration = json.loads(args.calibration.read_text())
    queue = build_queue(calibration)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(queue, indent=2, ensure_ascii=False) + "\n")

    if args.pretty:
        print(json.dumps(queue["summary"], indent=2))
        print(f"Wrote {args.output}")


if __name__ == "__main__":
    main()