fix(data): 漏斗族七修(审计 P1-1/2/3+P2-14/15/16+P3-21/24)——水位随commit边界(崩溃窗口事件不蒸发)+main兜底emergency落盘rc=1; --limit循环顶budget_exhausted真截断(假开关); llm-cap改每夜新鲜总量+candidates_dropped(逐日×逐域截断溢出永久丢); rc优先级checkpoint>噪声失败; 否定守卫4字窗contains+被拦命中降级LLM候选(已终止的质押类confidence=1.0反向落库); 生产接线补--token-budget 3M; events_in先于调用入册+_dump_raw_fail死码删; llm_calls勘误lane未跑归零(10-04生产误报58实0); fusion spec §20.16台账 [nas]

2026-10-05 09:15:58 +08:00
parent 79c6b8b921
commit 14f30a00f8
6 changed files with 239 additions and 27 deletions
@@ -222,6 +222,147 @@ def test_llm_lane_budget_stop_leaves_rest_pending(root):
    assert ctx.tokens_used > 0 and ctx.budget_stopped


+class ScriptedClient:
+    """P2-14/P3-21 用: 脚本化应答 [("raise", exc) | ("ok", resp, usage)]."""
+
+    def __init__(self, script):
+        self.script = list(script)
+        self.calls = 0
+
+    async def chat_json(self, messages, *, temperature=0.2, max_tokens=2000,
+                        want_usage=False):
+        self.calls += 1
+        item = self.script.pop(0)
+        if item[0] == "raise":
+            raise item[1]
+        return item[1], dict(item[2])
+
+
+def _cand(i, title="t"):
+    return {"content_hash": f"h{i}", "src_domain": "news_meta",
+            "src_id": f"X{i}", "stock_code": "000001",
+            "event_date": _day(1), "title": title, "text_snippet": "s",
+            "selected_at": _day(1)}
+
+
+def test_limit_actually_caps_llm_calls(root):
+    """P1-2: --limit 全链零检查=假开关 → 循环顶 budget_exhausted 真截断."""
+    ctx = _mk_ctx(limit=2)
+    led, rec = _ledgers()
+    fake = FakeClient({"events": []})
+    n = asyncio.run(cf._llm_lane(ctx, fake, "m", [_cand(i) for i in range(4)],
+                                 led, rec))
+    assert n == 0 and len(fake.calls) == 2 and ctx.stage_units == 2
+    # limit 班不落 marker(冒烟语义): 全部可次夜重试
+    assert not any(cd.is_done("funnel", "llm", f"h{i}") for i in range(4))
+
+
+def test_llm_cap_nightly_total_and_dropped(root, monkeypatch, capsys):
+    """P1-3: cap 改每夜总量 —— 入队前统一截断, 溢出记 candidates_dropped
+    (旧逐日×逐域截断, 溢出永久静默丢不可观测)."""
+    import types
+    import sanguo_api.llm as llm_mod
+    for off in (2, 1):
+        d = _day(off)
+        _seed_day(root, "news_meta", d, _news_rows(d, [
+            (f"N{off}", "600519", "重大合同签署进展", None)]))
+        _seed_day(root, "ann_meta", d, [{
+            "announcement_id": f"B{off}", "sec_code": "000002",
+            "ann_time": d, "title": "重大合同签署进展", "content": None,
+            "short_title": None}])
+    monkeypatch.setattr(
+        llm_mod, "resolve_config",
+        lambda: types.SimpleNamespace(model="glm-test"))
+    fake = FakeClient({"events": []})
+    monkeypatch.setattr(llm_mod, "LLMClient", lambda cfg: fake)
+    rc = cf.main(["--llm-cap", "3"])
+    assert rc == 0
+    stats = monitor_checks.parse_stat_lines(
+        capsys.readouterr().out).get("funnel")
+    assert stats["candidates_fresh"] == 4      # 语义=发现的总量
+    assert stats["candidates_dropped"] == 1    # 截断丢弃可见(不入 pending)
+    assert stats["llm_calls"] == 3
+    assert stats["pending_after"] == 0         # 3 件全消费; 丢弃件不占 pending
+
+
+def test_watermark_not_advanced_when_commit_fails(root, monkeypatch):
+    """P1-1: 数据 commit 失败 → 水位不得前移(旧水位先行=崩溃窗口永久丢)."""
+    day = _day(1)
+    _seed_day(root, "news_meta", day, _news_rows(day, [
+        ("A1", "600519", "控股股东拟增持公司股份公告", None)]))
+
+    def _boom():
+        raise RuntimeError("disk full")
+    monkeypatch.setattr(cd, "flush_domains", _boom)
+    rc = cf.main(["--no-llm"])
+    assert rc == 1                              # crashed 兜底 fail loud
+    assert cf._load_watermark() == {}           # 水位未前移 → 次夜重走
+
+
+def test_uncaught_exception_saves_remainder_rc1(root, monkeypatch):
+    """P1-1 兜底: 未捕获异常不跳过残余回写(旧直接传播, pending 蒸发)."""
+    cf._save_pending([_cand(1)])
+
+    def _boom(ctx, domain, args, ledgers, recents, fresh, stats):
+        raise RuntimeError("boom")
+    monkeypatch.setattr(cf, "_walk_domain", _boom)
+    rc = cf.main(["--no-llm"])
+    assert rc == 1
+    assert [r["src_id"] for r in _pending(root)] == ["X1"]
+
+
+def test_checkpoint_rc3_overrides_noise_fail(root, monkeypatch, capsys):
+    """P2-14: 墙停/预算 checkpoint 优先于噪声 LLM 失败(旧 rc() failed→1)."""
+    import types
+    import sanguo_api.llm as llm_mod
+    day = _day(1)
+    _seed_day(root, "news_meta", day, _news_rows(day, [
+        (f"A{i}", "000001", "重大合同签署进展", None) for i in range(3)]))
+    monkeypatch.setattr(
+        llm_mod, "resolve_config",
+        lambda: types.SimpleNamespace(model="glm-test"))
+    fake = ScriptedClient([
+        ("raise", RuntimeError("transient")),                 # 噪声失败
+        ("ok", {"events": []}, {"prompt_tokens": 10, "completion_tokens": 5}),
+    ])
+    monkeypatch.setattr(llm_mod, "LLMClient", lambda cfg: fake)
+    rc = cf.main(["--token-budget", "15"])      # 第2件后触顶 → BudgetStop
+    assert rc == 3                              # 旧代码=1(噪声失败覆盖)
+    assert "stopped=budget" in capsys.readouterr().out
+
+
+def test_config_error_reports_zero_llm_calls(root, monkeypatch, capsys):
+    """10-05 勘误: lane 未跑时 llm_calls 不得残留 walk 段计数
+    (10-04 19:33 生产 config_error 夜误报 llm_calls=58 实 0)."""
+    import sanguo_api.llm as llm_mod
+    day = _day(1)
+    _seed_day(root, "news_meta", day, _news_rows(day, [
+        ("A1", "000001", "重大合同签署进展", None)]))
+
+    def _no_cfg():
+        raise ValueError("SANGUO_LLM_API_KEY 未设置")
+    monkeypatch.setattr(llm_mod, "resolve_config", _no_cfg)
+    rc = cf.main([])
+    assert rc == 1
+    stats = monitor_checks.parse_stat_lines(
+        capsys.readouterr().out).get("funnel")
+    assert stats["llm_calls"] == 0
+    assert stats["pending_after"] == 1
+
+
+def test_events_in_recorded_despite_call_failure(root):
+    """P3-21: 输入先于调用入册 —— 调用失败 events_in 行保留(可回放对拍)."""
+    ctx = _mk_ctx()
+    led, rec = _ledgers()
+    fake = ScriptedClient([("raise", RuntimeError("boom"))])
+    n = asyncio.run(cf._llm_lane(ctx, fake, "m", [_cand(1)], led, rec))
+    ctx.commit()
+    assert n == 0 and ctx.failed == 1
+    ins = _read_domain(root, "events_in").to_dict("records")
+    assert len(ins) == 1
+    assert not cd.is_done("funnel", "llm", "h1")   # 未 done → 次夜重试
+
+
 def test_llm_prompt_carries_row_text(root):
    ctx = _mk_ctx()
    led, rec = _ledgers()
@@ -46,6 +46,22 @@ def test_empty_and_none_safe():
    assert lx.is_llm_candidate("", None) is False


+def test_negation_window_catches_distant_negation():
+    # P2-15(10-05): 否定窗扩 4 字 contains —— 旧 2 字 endswith 漏「已终止
+    # 的质押」类间隔措辞, pledge_new/neg 以 confidence=1.0 直落库
+    assert lx.match_events("关于已终止的质押的公告") == []
+    # 被否定命中降级为 LLM 候选(不再静默丢, 有第二层纠错路径)
+    assert lx.is_llm_candidate("关于已终止的质押的公告") is True
+
+
+def test_shangwei_negation_family():
+    # 尚未/仍未/不排除 三族负向测试(P2-15 派单验收面)
+    assert lx.match_events("公司尚未扭亏的说明") == []
+    assert lx.is_llm_candidate("公司尚未扭亏的说明") is True
+    assert lx.match_events("股东仍未公布增持计划的说明") == []
+    assert lx.is_llm_candidate("不排除减持公司股份的可能") is True
+
+
 def test_event_types_registry_stable():
    assert set(lx.EVENT_TYPES) == {
        "holder_increase", "holder_decrease", "buyback", "pledge_new",