run_precision_ablation.py

"""run_precision_ablation.py - ③④ 精度消融实验

测量 finding 验证器 (③) 和 杀手共识 (④) 对 review 精度的真实收益。
对 10 个 bug 金标准跑 4 个条件, 复用生产代码 QALoop._merge_findings /
_apply_verdicts 保证测的是真逻辑 (不是重写):

  baseline : 单 enhanced reviewer (复现 group A ~71%)
  ③ verify : reviewer → finding-verifier 反驳 → 丢弃 kill (抬 precision)
  ④ consensus: reviewer + skeptic 双视角, consensus_min=2 (共识才留)
  ③+④      : 先共识再反驳 (叠加)

前置: 与 run_group_a 相同 — ANTHROPIC_BASE_URL + ANTHROPIC_AUTH_TOKEN +
      ANTHROPIC_DEFAULT_SONNET_MODEL (见 experiment/REPRODUCE.md)。

运行:
    cd loop-engine/experiment
    python run_precision_ablation.py
"""
import sys, os, json
sys.path.insert(0, os.path.dirname(__file__))           # experiment/ 自身
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))  # loop-engine/
from bug_seeds import SEEDS
from llm_call import make_call_fn, extract_json_array, read_agent_body
from qa_loop import QALoop, Finding


# ---- 系统提示 (enhanced reviewer + skeptic + verifier) ----
REVIEWER_SYS = None  # 懒读: read_agent_body("../agents/code-reviewer.md")
SKEPTIC_SYS = """You are the SKEPTIC in a code review. Review ALL perspectives.
For EACH candidate issue, actively try to REFUTE it: can it actually trigger?
is it pre-existing? can it be reproduced? Report ONLY findings you CANNOT refute
and are >=80% sure are real. High-signal over volume. Return ONLY a JSON array."""
VERIFIER_SYS = None  # read_agent_body("../agents/finding-verifier.md")

CODE_USER = """审查以下Python代码找出bug(综合所有视角: 安全/正确性/边界/质量)。
返回JSON数组, 每项: {{"severity","file","category","issue","fix_hint"}}。无问题返回 []。

代码:
```python
{code}
```"""

SKEPTIC_USER = """你是杀手视角。审查以下代码, 对每个候选问题先尝试反驳。
只返回你**无法反驳且≥80%确信**的真问题。返回JSON数组, 每项:
{{"severity","file","category","issue","fix_hint"}}。无法反驳的真问题都没有则返回 []。

代码:
```python
{code}
```"""

VERIFY_USER = """你是独立的 finding 验证器。对下面每条 finding 主动反驳: 能否真触发?
是否既有问题? 默认 kill (不确定就杀)。返回JSON数组, 每项:
{{"file","issue","verdict":"keep|kill","confidence":0-1,"evidence"}}。

代码:
```python
{code}
```

findings:
{findings}"""

SYNONYMS = {
    "sql injection": ["sql injection", "sql注入", "参数化", "拼接查询", "注入"],
    "command injection": ["command injection", "命令注入", "os.system", "shell"],
    "xss": ["xss", "跨站", "转义", "escape", "innerHTML", "未转义"],
    "null": ["null", "none", "空值", "空指针", "空对象", "nullable", "未检查"],
    "off-by-one": ["off-by-one", "差一", "越界", "边界错误", "<=", "边界"],
    "discount": ["discount", "折扣", "方向", "加而非减"],
    "empty": ["empty", "空串", "空字符串", "空输入", "空参数"],
    "race": ["race", "竞态", "并发", "线程安全", "锁", "check-then"],
    "atomic": ["atomic", "原子", "事务", "一致性", "部分更新", "partial"],
    "divide": ["divide", "除零", "除以零", "division", "zero"],
}


def hit(finding, expected):
    blob = " ".join(str(finding.get(k, "")).lower() for k in ("issue", "category", "fix_hint"))
    keys = SYNONYMS.get(expected.lower(), [expected.lower()])
    return any(k.lower() in blob for k in keys)


def to_finding(d: dict) -> Finding:
    """dict → Finding (只取已知字段, 容错)"""
    return Finding(
        severity=d.get("severity", "Minor"),
        file=d.get("file", ""),
        category=d.get("category", ""),
        issue=d.get("issue", ""),
        impact=d.get("impact", ""),
        fix_hint=d.get("fix_hint", ""),
    )


def review_call(call, sys_prompt, user_prompt, code):
    text = call(user_prompt.format(code=code), sys_prompt)
    return [to_finding(d) for d in extract_json_array(text)]


# ---- 4 个条件 ----

def cond_baseline(call, code):
    return review_call(call, REVIEWER_SYS, CODE_USER, code)


def cond_verify(call, code):
    """③: reviewer → finding-verifier 反驳 → 丢弃 kill"""
    findings = review_call(call, REVIEWER_SYS, CODE_USER, code)
    if not findings:
        return []
    vtext = call(VERIFY_USER.format(code=code,
                                    findings=json.dumps([{"file": f.file, "issue": f.issue}
                                                         for f in findings], ensure_ascii=False)),
                 VERIFIER_SYS)
    verdicts = extract_json_array(vtext)
    return QALoop._apply_verdicts(findings, verdicts)


def cond_consensus(call, code):
    """④: reviewer + skeptic 双视角, consensus_min=2 (共识才留)"""
    reviewer_findings = review_call(call, REVIEWER_SYS, CODE_USER, code)
    skeptic_findings = review_call(call, SKEPTIC_SYS, SKEPTIC_USER, code)
    return QALoop._merge_findings([reviewer_findings, skeptic_findings], consensus_min=2)


def cond_both(call, code):
    """③+④: 先共识再反驳"""
    consensus = cond_consensus(call, code)
    if not consensus:
        return []
    vtext = call(VERIFY_USER.format(code=code,
                                    findings=json.dumps([{"file": f.file, "issue": f.issue}
                                                         for f in consensus], ensure_ascii=False)),
                 VERIFIER_SYS)
    verdicts = extract_json_array(vtext)
    return QALoop._apply_verdicts(consensus, verdicts)


CONDITIONS = [
    ("baseline (enhanced)", cond_baseline),
    ("③ verify", cond_verify),
    ("④ consensus", cond_consensus),
    ("③+④", cond_both),
]


def run_condition(call, fn):
    """跑 10 个种子, 返回 (hits, total_reported)"""
    hits, total = 0, 0
    for s in SEEDS:
        findings = fn(call, s.code)
        total += len(findings)
        if any(hit({"issue": f.issue, "category": f.category, "fix_hint": f.fix_hint},
                   s.expected_finding) for f in findings):
            hits += 1
    return hits, total


def main():
    global REVIEWER_SYS, VERIFIER_SYS
    REVIEWER_SYS = read_agent_body("../agents/code-reviewer.md")
    VERIFIER_SYS = read_agent_body("../agents/finding-verifier.md")

    print("=" * 64)
    print("③④ 精度消融实验 (10 bug 金标准)")
    print("=" * 64)
    call = make_call_fn("ANTHROPIC_DEFAULT_SONNET_MODEL")
    print(f"模型: {call.model_name}\n")

    n = len(SEEDS)
    print(f"{'条件':<24}{'命中':>6}{'报数':>6}{'FP':>6}{'precision':>12}{'recall':>8}")
    print("-" * 64)
    for label, fn in CONDITIONS:
        hits, total = run_condition(call, fn)
        fp = total - hits
        precision = hits / total if total else 0
        recall = hits / n
        print(f"{label:<24}{hits:>5}/{n}{total:>6}{fp:>6}{precision:>11.0%}{recall:>8.0%}")

    print("\n解读:")
    print("- baseline 应接近 group A 的 71% (校验脚本可信)。")
    print("- ③/④ 若 precision 上升 → 结构性补强有效 (绕开 LLM 自评过度自信)。")
    print("- recall 若从 100% 下降是预期 (用 recall 换 precision); 看是否划算。")
    print("- ③+④ 叠加边际递减, 不一定比单独 ③ 或 ④ 更好。")


if __name__ == "__main__":
    main()