feat: 对话层 Agent 化（对话底座 + 自研 skill）

把对话页从固定状态机换成 skill 驱动的 Agent：顾问说什么都原样发给 agent，由 SKILL.md + 工具决定该聊还是该调工具。工作台（workbenchView）未改动。后端 - llm.py：新增 stream_chat()（流式 + tools），并带 reasoning 端点自动降级（gemini-3.7-flash 拒绝 effort:none，首次 400 后锁定重试） - agent.py：最小对话底座内核——skill 注入 + 工具循环 + SSE 事件 - tools.py：8 个工具包住既有能力（analyze/diagnose/scaffold/reference/ translate/recheck/record_note/confirm），延迟 import main 复用已验证的 endpoint 处理器与测试接缝，prompts.py/schemas.py/evidence.py 未改一行 - main.py：POST /api/chat/stream（SSE） skill - skills/hvr-rewrite/SKILL.md：分诊规则、首轮盘点、AI 味清单对齐 blader/humanizer 的 A–E 分类（含强度校准与反误报）、诚实性硬约束前端 - 流式正文 + 工具卡片 + 选项按钮；工具 payload 直接喂既有渲染函数（renderUnderstanding/renderDiagnosis/renderRecheck），没有第二套 UI - 删除随 agent 化失效的 runDiagnose / applyUnderstandingCorrection 死代码验证（真实模型，非 mock） - 65 单测全绿（test_agent.py 30 + test_demo.py 35） - 逐轮实跑：盘点 → 确认 → 诊断 → 复检，SSE 事件序列与 payload kind 均符合契约 - 前端 readSSE 用真实响应字节按 7/64/全量三种切块回放，事件序列一致 - 实测修掉两个只在真跑时暴露的问题：模型调完分析直接调确认工具导致正文为空（SKILL.md 补「正文先行」硬规则）；伪标题 `**N. 标题**` 让列表判定失败、短横线漏成字面字符（前端改逐行分组渲染）未验证：浏览器人工走查（无浏览器自动化环境），仅到「真实 HTTP + 真实字节回放」这一层。
2026-09-11 10:25:12 +08:00
parent 493c7f6416
commit 94a26227c1
7 changed files with 2170 additions and 73 deletions
@@ -0,0 +1,510 @@
+"""对话底座（agent.py / tools.py）的 hermetic 测试。
+
+不碰真实模型：stream_chat 用脚本化 stub，工具层转调的 endpoint 处理器
+通过 monkeypatch main.get_client 打桩（与 test_demo.py 同一个接缝）。
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+from fastapi.testclient import TestClient
+
+import main as main_mod
+from agent import MAX_TURNS, build_system_prompt, get_session, load_skill, run_agent_stream
+from llm import LLMError, LlmClient
+from main import app
+from tools import ToolContext, execute, openai_tools
+
+PARAGRAPHS = [
+    "Growing up, I always thought success was a straight line.",
+    "The failure taught me that resilience is the language of growth.",
+]
+
+
+# ---------------------------------------------------------------- stubs
+class StreamStub(LlmClient):
+    """脚本化 stream_chat。每个脚本项 = 一轮 LLM 的事件列表。"""
+
+    def __init__(self, script):
+        super().__init__(api_key="sk-test", base_url="http://stub", model="stub")
+        self.script = list(script)
+        self.seen: list[list[dict]] = []
+
+    def stream_chat(self, messages, tools=None, **kwargs):
+        self.seen.append([dict(m) for m in messages])
+        item = self.script.pop(0)
+        if isinstance(item, Exception):
+            raise item
+        yield from item
+
+
+class CompleterStub:
+    """工具层打桩：按序返回待定 JSON。"""
+
+    def __init__(self, payloads):
+        self.payloads = list(payloads)
+        self.calls = 0
+
+    def complete_json(self, system, user, validate=None):
+        self.calls += 1
+        data = self.payloads.pop(0)
+        if validate is not None:
+            problem = validate(data)
+            if problem:
+                raise LLMError(problem)
+        return data
+
+
+def tool_turn(name, args=None, call_id="c1"):
+    return [
+        {
+            "type": "tool_calls",
+            "tool_calls": [
+                {"id": call_id, "name": name, "arguments": json.dumps(args or {})}
+            ],
+        },
+        {"type": "done", "finish_reason": "tool_calls"},
+    ]
+
+
+def text_turn(text):
+    return [{"type": "text", "text": text}, {"type": "done", "finish_reason": "stop"}]
+
+
+def names(events):
+    return [e["event"] for e in events]
+
+
+def first(events, name):
+    return next(e for e in events if e["event"] == name)
+
+
+def analyze_payload(paragraphs=PARAGRAPHS):
+    return {
+        "essay_summary": "写一次失败后的转变",
+        "paragraph_alignment": {},
+        "prompt_alignment": {"prompt_intent": "讲一次失败", "current_alignment": "基本回应"},
+        "paragraphs": [
+            {
+                "id": f"p{i + 1}",
+                "original_text": p,
+                "natural_meaning_zh": f"第{i + 1}段中文",
+                "semantic_anchor": f"锚点{i + 1}",
+                "optional_content_opportunity": "",
+            }
+            for i, p in enumerate(paragraphs)
+        ],
+    }
+
+
+def new_session(paragraphs=PARAGRAPHS, **kw):
+    return get_session(None, {"paragraphs": list(paragraphs), **kw})
+
+
+class FakeResp:
+    def __init__(self, data):
+        self._data = data
+
+    def model_dump(self):
+        return self._data
+
+
+class CaptureHandler:
+    """替换 endpoint 处理器，抓住工具实际发出的请求对象。"""
+
+    def __init__(self, response):
+        self.response = response
+        self.request = None
+
+    def __call__(self, req):
+        self.request = req
+        return self.response
+
+
+# ---------------------------------------------------------------- 参数透传
+def test_diagnose_forwards_advisor_constraints(monkeypatch):
+    """顾问确认的锚点与分段要求必须到达诊断——工具化改造最容易在这里丢参数。"""
+    cap = CaptureHandler(FakeResp({"patterns": [], "paragraph_briefs": []}))
+    monkeypatch.setattr(main_mod, "diagnose", cap)
+    ctx = ToolContext(
+        paragraphs=PARAGRAPHS,
+        constraints=["全篇更口语"],
+        confirmed_anchors=["锚点A", "锚点B"],
+        paragraph_constraints={"p1": ["保留 secret code"]},
+    )
+    execute("diagnose_essay", {}, ctx)
+
+    assert cap.request.confirmed_anchors == ["锚点A", "锚点B"]
+    assert cap.request.paragraph_constraints == {"p1": ["保留 secret code"]}
+    assert cap.request.global_constraints == ["全篇更口语"]
+
+
+def test_diagnose_falls_back_to_analysis_anchors(monkeypatch):
+    """顾问没单独确认时，退回理解阶段的锚点——保持改造前 requestDiagnosis 的行为。"""
+    cap = CaptureHandler(FakeResp({"patterns": [], "paragraph_briefs": []}))
+    monkeypatch.setattr(main_mod, "diagnose", cap)
+    ctx = ToolContext(paragraphs=PARAGRAPHS)
+    ctx.analysis = analyze_payload()
+    execute("diagnose_essay", {}, ctx)
+    assert cap.request.confirmed_anchors == ["锚点1", "锚点2"]
+
+
+def test_record_note_reaches_diagnosis(monkeypatch):
+    """顾问口头提的要求必须落成结构化约束，否则下一轮诊断收不到（PRD §10.1）。"""
+    ctx = ToolContext(paragraphs=PARAGRAPHS)
+    execute("record_advisor_note", {"note": "重点不是接受不确定", "paragraph_id": "p2"}, ctx)
+    execute("record_advisor_note", {"note": "全篇更口语"}, ctx)
+
+    assert ctx.paragraph_constraints == {"p2": ["重点不是接受不确定"]}
+    assert ctx.constraints == ["全篇更口语"]
+
+    cap = CaptureHandler(FakeResp({"patterns": [], "paragraph_briefs": []}))
+    monkeypatch.setattr(main_mod, "diagnose", cap)
+    execute("diagnose_essay", {}, ctx)
+    assert cap.request.paragraph_constraints == {"p2": ["重点不是接受不确定"]}
+    assert cap.request.global_constraints == ["全篇更口语"]
+
+
+def test_record_note_deduplicates_and_emits_payload():
+    ctx = ToolContext(paragraphs=PARAGRAPHS)
+    execute("record_advisor_note", {"note": "保留 secret code", "paragraph_id": "p1"}, ctx)
+    result = execute("record_advisor_note", {"note": "保留 secret code", "paragraph_id": "p1"}, ctx)
+    assert ctx.paragraph_constraints["p1"] == ["保留 secret code"], "重复记录不该堆叠"
+    assert result.payload["kind"] == "constraint"  # 前端靠它同步本地副本
+    assert execute("record_advisor_note", {"note": "  "}, ctx).ok is False
+
+
+def test_record_note_rejects_hallucinated_paragraph():
+    """段号越界要报错，不能静默降级成全篇要求——那会污染整篇的约束。"""
+    ctx = ToolContext(paragraphs=PARAGRAPHS)
+    result = execute("record_advisor_note", {"note": "保留比喻", "paragraph_id": "p9"}, ctx)
+    assert result.ok is False
+    assert ctx.paragraph_constraints == {}
+    assert ctx.constraints == [], "越界段号不该被当成全篇要求收下"
+
+
+def test_recheck_forwards_diagnosis_and_constraints(monkeypatch):
+    """不带 diagnosis，「原 Pattern 是否缓解」这一维就没有对照物。"""
+    cap = CaptureHandler(FakeResp({"status": "pass", "global_checks": {}, "revision_targets": []}))
+    monkeypatch.setattr(main_mod, "recheck", cap)
+    ctx = ToolContext(paragraphs=PARAGRAPHS, rewrite_paragraphs=["改1", "改2"])
+    ctx.diagnosis = {"overall_diagnosis": "总述", "patterns": [], "paragraph_briefs": []}
+    ctx.paragraph_constraints = {"p2": ["保留比喻"]}
+    execute("recheck_rewrite", {}, ctx)
+
+    assert cap.request.diagnosis is not None
+    assert cap.request.diagnosis.overall_diagnosis == "总述"
+    assert cap.request.paragraph_constraints == {"p2": ["保留比喻"]}
+
+
+# ---------------------------------------------------------------- skill 注入
+def test_load_skill_strips_frontmatter():
+    text = load_skill("hvr-rewrite")
+    assert not text.startswith("---"), "frontmatter 必须剥掉——它是给注册链路读的元数据，对模型是噪声"
+    assert "name: hvr-rewrite" not in text
+    assert "AI 味检测清单" in text
+    assert "不是 X，而是 Y" in text
+    assert "这些不要动手" in text
+
+
+def test_system_prompt_carries_skill_and_essay_context():
+    ctx = ToolContext(prompt="Write about a failure.", word_limit=650, paragraphs=PARAGRAPHS)
+    prompt = build_system_prompt(ctx)
+    assert "AI 味检测清单" in prompt  # skill 全文进 system prompt（照搬 engine 的固定注入）
+    assert "<essay_context>" in prompt and "</essay_context>" in prompt
+    assert "<p1>Growing up" in prompt
+    assert "Write about a failure." in prompt
+    assert "已完成步骤：无" in prompt
+
+
+def test_system_prompt_marks_completed_steps():
+    ctx = ToolContext(paragraphs=PARAGRAPHS)
+    ctx.analysis = {"paragraphs": []}
+    assert "已完成理解" in build_system_prompt(ctx)
+
+
+# ---------------------------------------------------------------- 工具循环
+def test_tool_loop_runs_tool_then_answers(monkeypatch):
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()]))
+    stub = StreamStub([tool_turn("analyze_essay"), text_turn("这篇的问题在结尾。")])
+    sess = new_session()
+
+    events = list(run_agent_stream(sess, "帮我看看这篇", stub))
+
+    assert names(events)[:2] == ["meta", "session_id"]
+    call = first(events, "tool_call")
+    assert call["data"]["tool"] == "analyze_essay"
+    result = first(events, "tool_result")
+    assert result["data"]["ok"] is True
+    assert result["data"]["payload"]["kind"] == "analysis"  # 前端靠 payload 渲染理解卡
+    assert first(events, "token")["data"]["text"] == "这篇的问题在结尾。"
+    assert events[-1] == {"event": "done", "data": {"finish_reason": "stop"}}
+    assert len(stub.seen) == 2, "工具轮 + 回答轮 = 两次 LLM 调用"
+
+
+def test_text_and_tool_call_in_one_turn(monkeypatch):
+    """同一轮里既有正文又有工具调用：正文要流出去，工具也要执行。
+
+    SKILL.md 的首轮盘点要求「先写盘点正文，再在同一次回复里调 let_user_confirm」——
+    这条路走不通的话，顾问屏幕上只剩几张工具卡片，一个字都看不到。"""
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()]))
+    mixed = [
+        {"type": "text", "text": "这篇在写什么："},
+        {
+            "type": "tool_calls",
+            "tool_calls": [{"id": "c1", "name": "analyze_essay", "arguments": "{}"}],
+        },
+        {"type": "done", "finish_reason": "tool_calls"},
+    ]
+    stub = StreamStub([mixed, text_turn("盘点如上。")])
+    events = list(run_agent_stream(new_session(), "请开始首轮盘点。", stub))
+
+    streamed = "".join(e["data"]["text"] for e in events if e["event"] == "token")
+    assert streamed == "这篇在写什么：盘点如上。"
+    assert [e["data"]["tool"] for e in events if e["event"] == "tool_call"] == ["analyze_essay"]
+    # 带工具调用的那轮，正文也必须落进历史，否则下一轮模型不记得自己说过什么
+    assert stub.seen[1][-2]["content"] == "这篇在写什么："
+
+
+def test_tool_result_in_history_is_summary_not_full_payload(monkeypatch):
+    """诊断 JSON 有 ~2.5k tokens，原样回灌会让每轮成本滚雪球——进历史的必须是摘要。"""
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()]))
+    stub = StreamStub([tool_turn("analyze_essay"), text_turn("好。")])
+    list(run_agent_stream(new_session(), "看看", stub))
+
+    second_turn = stub.seen[1]
+    tool_msg = next(m for m in second_turn if m.get("role") == "tool")
+    assert "语义锚点" in tool_msg["content"]  # 摘要内容
+    assert tool_msg["content"] not in json.dumps(analyze_payload())  # 不是原始 payload
+    assert len(tool_msg["content"]) < 800
+
+
+def test_analysis_is_stashed_for_later_tools(monkeypatch):
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()]))
+    stub = StreamStub([tool_turn("analyze_essay"), text_turn("好。")])
+    sess = new_session()
+    list(run_agent_stream(sess, "看看", stub))
+    assert sess["ctx"].analysis is not None  # 后续工具与前端上下文复用
+
+
+def test_parallel_tool_calls_all_execute(monkeypatch):
+    monkeypatch.setattr(
+        main_mod, "get_client", lambda: CompleterStub([analyze_payload()])
+    )
+    stub = StreamStub(
+        [
+            [
+                {
+                    "type": "tool_calls",
+                    "tool_calls": [
+                        {"id": "c1", "name": "analyze_essay", "arguments": "{}"},
+                        {"id": "c2", "name": "translate_to_chinese", "arguments": '{"paragraph_id": "p1"}'},
+                    ],
+                },
+                {"type": "done", "finish_reason": "tool_calls"},
+            ],
+            text_turn("好了。"),
+        ]
+    )
+    events = list(run_agent_stream(new_session(), "看看", stub))
+    calls = [e["data"]["tool"] for e in events if e["event"] == "tool_call"]
+    assert calls == ["analyze_essay", "translate_to_chinese"]
+    assert len([e for e in events if e["event"] == "tool_result"]) == 2
+
+
+# ---------------------------------------------------------------- 交互型工具
+def test_let_user_confirm_pauses_the_loop():
+    stub = StreamStub(
+        [tool_turn("let_user_confirm", {"question": "先改哪一段？", "choices": ["p1", "p2"]})]
+    )
+    events = list(run_agent_stream(new_session(), "开始", stub))
+
+    clarify = first(events, "tool_request_clarify")
+    assert clarify["data"]["question"] == "先改哪一段？"
+    assert clarify["data"]["choices"] == ["p1", "p2"]
+    assert events[-1] == {"event": "done", "data": {"finish_reason": "await_user"}}
+    assert len(stub.seen) == 1, "暂停后不应再调 LLM——等顾问点选"
+
+
+def test_confirm_requires_question_and_choices():
+    result = execute("let_user_confirm", {"question": "?", "choices": []}, ToolContext())
+    assert result.ok is False
+
+
+# ---------------------------------------------------------------- 失败路径
+def test_tool_failure_does_not_kill_the_turn():
+    """工具失败必须让 LLM 看见并如实转述（SKILL.md 诚实性硬约束第 4 条）。"""
+    stub = StreamStub([tool_turn("recheck_rewrite"), text_turn("还没有改写稿，先改。")])
+    events = list(run_agent_stream(new_session(), "复检一下", stub))
+
+    result = first(events, "tool_result")
+    assert result["data"]["ok"] is False
+    assert "改写稿" in result["data"]["result_summary"]
+    assert first(events, "token")["data"]["text"] == "还没有改写稿，先改。"
+
+
+def test_unknown_tool_is_reported_not_raised():
+    stub = StreamStub([tool_turn("no_such_tool"), text_turn("我换个办法。")])
+    events = list(run_agent_stream(new_session(), "?", stub))
+    assert first(events, "tool_result")["data"]["ok"] is False
+
+
+def test_malformed_tool_arguments_do_not_crash():
+    stub = StreamStub(
+        [
+            [
+                {
+                    "type": "tool_calls",
+                    "tool_calls": [{"id": "c1", "name": "analyze_essay", "arguments": "{not json"}],
+                },
+                {"type": "done", "finish_reason": "tool_calls"},
+            ],
+            text_turn("好。"),
+        ]
+    )
+    events = list(run_agent_stream(new_session(), "?", stub))
+    assert first(events, "tool_call")["data"]["args"] == {}
+
+
+def test_no_essay_fails_closed_without_calling_llm():
+    stub = StreamStub([])
+    events = list(run_agent_stream(new_session(paragraphs=[]), "看看", stub))
+    assert first(events, "error")["data"]["code"] == "no_essay"
+    assert stub.seen == [], "没有原文就不该付 LLM 调用的钱"
+
+
+def test_llm_error_becomes_error_frame():
+    stub = StreamStub([LLMError("调用大模型失败（HTTP 402）")])
+    events = list(run_agent_stream(new_session(), "看看", stub))
+    err = first(events, "error")
+    assert err["data"]["code"] == "llm_error"
+    assert "402" in err["data"]["msg"]
+
+
+def test_max_turns_guard(monkeypatch):
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()] * MAX_TURNS))
+    stub = StreamStub([tool_turn("analyze_essay", call_id=f"c{i}") for i in range(MAX_TURNS)])
+    events = list(run_agent_stream(new_session(), "一直调工具", stub))
+    assert first(events, "error")["data"]["code"] == "max_turns"
+
+
+# ---------------------------------------------------------------- 会话
+def test_session_reuses_context_and_history(monkeypatch):
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()]))
+    stub = StreamStub([tool_turn("analyze_essay"), text_turn("好。")])
+    sess = new_session()
+    list(run_agent_stream(sess, "看看", stub))
+
+    again = get_session(sess["sid"], {"paragraphs": PARAGRAPHS})
+    assert again is sess
+    assert again["ctx"].analysis is not None, "续接会话不能丢已完成的分析"
+    assert any(m.get("role") == "user" for m in again["messages"])
+
+
+def test_context_refresh_keeps_completed_work(monkeypatch):
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()]))
+    sess = new_session()
+    list(run_agent_stream(sess, "看看", StreamStub([tool_turn("analyze_essay"), text_turn("好。")])))
+
+    # 顾问在工作台改完了稿，前端重发上下文
+    get_session(sess["sid"], {"paragraphs": PARAGRAPHS, "rewrite_paragraphs": ["改后1", "改后2"]})
+    assert sess["ctx"].rewrite_paragraphs == ["改后1", "改后2"]
+    assert sess["ctx"].analysis is not None
+
+
+def test_unknown_session_id_starts_fresh_instead_of_erroring():
+    sess = get_session("deadbeefdeadbeef", {"paragraphs": PARAGRAPHS})
+    assert sess["sid"] != "deadbeefdeadbeef"
+
+
+def test_new_session_hydrates_analysis_and_diagnosis():
+    """刷新/进程重启后前端新开会话，第一轮必须把已有的分析/诊断带过来。
+
+    丢了它们不会报错——复检会静默降级（少 Pattern 对照、锚点为空），
+    属于「不报错的错」，所以这里钉死。"""
+    analysis = analyze_payload()
+    diagnosis = {"overall_diagnosis": "像 AI", "patterns": [], "paragraph_briefs": []}
+    sess = get_session(None, {"paragraphs": PARAGRAPHS, "analysis": analysis, "diagnosis": diagnosis})
+    assert sess["ctx"].analysis == analysis
+    assert sess["ctx"].diagnosis == diagnosis
+
+
+def test_refresh_does_not_wipe_hydrated_results():
+    """有了 session_id 之后前端不再重发分析/诊断，重发也不该把它们抹掉。"""
+    sess = get_session(None, {"paragraphs": PARAGRAPHS, "analysis": analyze_payload()})
+    get_session(sess["sid"], {"paragraphs": PARAGRAPHS, "rewrite_paragraphs": ["改后1", "改后2"]})
+    assert sess["ctx"].analysis is not None
+    assert sess["ctx"].rewrite_paragraphs == ["改后1", "改后2"]
+
+
+# ---------------------------------------------------------------- 工具声明
+def test_tool_declarations_are_openai_shaped():
+    specs = openai_tools()
+    assert {s["function"]["name"] for s in specs} == {
+        "analyze_essay",
+        "diagnose_essay",
+        "get_writing_scaffold",
+        "get_reference_snippet",
+        "translate_to_chinese",
+        "recheck_rewrite",
+        "record_advisor_note",
+        "let_user_confirm",
+    }
+    for spec in specs:
+        assert spec["type"] == "function"
+        fn = spec["function"]
+        assert fn["description"].strip()
+        assert fn["parameters"]["type"] == "object"
+
+
+def test_paragraph_id_is_normalized():
+    """模型给 'P2'/'2'/'p2' 都该认；越界或给不出就返回空，不猜。"""
+    ctx = ToolContext(paragraphs=PARAGRAPHS)
+    assert ctx.paragraph("p2") == PARAGRAPHS[1]
+    assert ctx.paragraph("P2") == PARAGRAPHS[1]
+    assert ctx.paragraph("2") == PARAGRAPHS[1]
+    assert ctx.paragraph("9") == ""
+    assert ctx.paragraph("") == ""
+    assert ctx.paragraph("third") == ""
+
+
+def test_scaffold_tool_rejects_unknown_paragraph():
+    result = execute("get_writing_scaffold", {"paragraph_id": "p9"}, ToolContext(paragraphs=PARAGRAPHS))
+    assert result.ok is False
+    assert "找不到段落" in result.error
+
+
+# ---------------------------------------------------------------- HTTP 端点
+def test_chat_stream_endpoint_emits_wire_format(monkeypatch):
+    monkeypatch.setattr(main_mod, "get_client", lambda: CompleterStub([analyze_payload()]))
+    monkeypatch.setattr(
+        main_mod.agent,
+        "run_agent_stream",
+        lambda sess, msg, client: iter(
+            [
+                {"event": "meta", "data": {"trace_id": "t1"}},
+                {"event": "token", "data": {"text": "你好"}},
+                {"event": "done", "data": {"finish_reason": "stop"}},
+            ]
+        ),
+    )
+    with TestClient(app) as c:
+        r = c.post("/api/chat/stream", json={"message": "hi", "paragraphs": PARAGRAPHS})
+    assert r.status_code == 200
+    assert r.headers["content-type"].startswith("text/event-stream")
+    assert 'event: token\ndata: {"text": "你好"}' in r.text
+    assert r.text.rstrip().endswith("data: {\"finish_reason\": \"stop\"}")
+
+
+def test_chat_stream_reports_client_error_as_frame(monkeypatch):
+    def boom():
+        raise LLMError("缺少 OpenRouter Key")
+
+    monkeypatch.setattr(main_mod, "get_client", boom)
+    with TestClient(app) as c:
+        r = c.post("/api/chat/stream", json={"message": "hi", "paragraphs": PARAGRAPHS})
+    assert r.status_code == 200, "流已开，状态码改不了——失败靠 error 帧告诉前端"
+    assert "event: error" in r.text
+    assert "缺少 OpenRouter Key" in r.text