LuminousRuoxi
94a26227c1
把对话页从固定状态机换成 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 + 真实字节回放」这一层。
511 lines
21 KiB
Python
511 lines
21 KiB
Python
"""对话底座（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
|