diff --git a/backend/python/tests/test_engine_event.py b/backend/python/tests/test_engine_event.py new file mode 100644 index 0000000..1d2a526 --- /dev/null +++ b/backend/python/tests/test_engine_event.py @@ -0,0 +1,396 @@ +"""#873 双 Claw 事件契约 · 单元哨兵(EngineEvent / ClawAdapter 层)。 + +为什么存在这个文件(#873 残留的「哨兵补」): + R-DUAL 哨兵(test_dual_claw.py) 只验证了「Web/IM 双端能力一致 + OTel 标签」, + 但从未直接断言 EngineEvent 工厂与两个 Adapter 的**线格式契约**。 + 这两份契约是单引擎 / 双 Claw 的物理基础 —— 一旦被破坏,前端静默不渲染, + 且排查极难。本文件把它们锁死为确定性单测(无 DB / 无网络)。 + +验证范围: + 1. EngineEvent 工厂:make_delta / reasoning / step / done / error 的线格式 + 2. to_wire() 与既有 ChatPage.tsx 逐键兼容(新增键非空才输出) + 3. from_wire() 把遗留裸 dict 提升为 EngineEvent(迁移期双向兼容) + 4. SseAdapter._thin() 默认剥离热帧元数据、终帧保留 trace_id + 5. SseAdapter.frame() SSE 线格式 + frames() 生成器直到哨兵 + 6. BufferAdapter 累积 / 媒体结构化返回 / 诊断尾注 / 截断 + 7. 双 Claw 接线契约:SseAdapter.claw=="web" / BufferAdapter.claw=="im" + + chat.py / im_llm.py 必须引用对应 Adapter(消灭第二份实现) +""" +import asyncio +import json +import os +import sys + +import pytest + +_HERE = os.path.dirname(os.path.abspath(__file__)) +APP_DIR = os.path.abspath(os.path.join(_HERE, "..", "app")) +sys.path.insert(0, os.path.join(_HERE, "..")) + +from app.core.engine_event import ( # noqa: E402 + CLAW_IM, + CLAW_WEB, + DELTA, + DONE, + ERROR, + REASONING, + STEP, + SEV_ERROR, + SEV_INFO, + Diagnosis, + EngineEvent, + make_delta, + make_done, + make_error, + make_reasoning, + make_step, + from_wire, +) +from app.core.claw_adapters import ( # noqa: E402 + SseAdapter, + BufferAdapter, +) + + +# ===================================================================== +# 1. EngineEvent 工厂线格式 +# ===================================================================== +def test_make_delta_wire(): + ev = make_delta("你好") + assert ev.type == DELTA + assert ev.to_wire() == {"type": "delta", "content": "你好"} + + +def test_make_reasoning_wire(): + ev = make_reasoning("让我想想") + assert ev.to_wire() == {"type": "reasoning", "content": "让我想想"} + + +def test_make_step_thinking(): + ev = make_step("thinking", text="分析中") + w = ev.to_wire() + assert w["type"] == STEP + assert w["step"] == {"type": "thinking", "text": "分析中", "args": {}, "result": ""} + + +def test_make_step_skill_mcp(): + ev = make_step("skill", name="天气查询", server="", args={"city": "北京"}, result="晴") + w = ev.to_wire() + step = w["step"] + assert step["type"] == "skill" + assert step["name"] == "天气查询" + assert step["args"] == {"city": "北京"} + assert step["result"] == "晴" + + ev2 = make_step("mcp", server="srv", name="tool_a", args={"x": 1}) + step2 = ev2.to_wire()["step"] + assert step2["server"] == "srv" + assert step2["name"] == "tool_a" + + +def test_make_step_index_and_node(): + ev = make_step("skill", name="s", index=3, node={"id": "n1", "type": "llm", "label": "L"}) + w = ev.to_wire() + assert w["index"] == 3 + assert w["step"]["node"] == {"id": "n1", "type": "llm", "label": "L"} + + +def test_make_step_error_diagnosis_fallback_result(): + ev = make_step( + "mcp", + server="srv", + name="t", + phase="fail", + diagnosis={"code": "MCP_TIMEOUT", "severity": "error", "summary": "超时", "detail": "d", "hint": "重试"}, + ) + w = ev.to_wire() + # 失败且 result 为空时,必须给可读 result 兜底,否则前端判定「运行中」 + assert w["step"]["result"] == "❌ 超时" + assert w["step"]["phase"] == "fail" + assert w["step"]["diagnosis"]["code"] == "MCP_TIMEOUT" + + +def test_make_done_full_fields(): + ev = make_done( + message_id="m1", + conversation_id="c1", + content="最终回复", + process_steps=[{"type": "thinking", "text": "x"}], + used_skills=["s"], + used_mcp=["m"], + generating=False, + ) + w = ev.to_wire() + assert w["type"] == DONE + assert w["message_id"] == "m1" + assert w["conversation_id"] == "c1" + assert w["content"] == "最终回复" + assert w["process_steps"] == [{"type": "thinking", "text": "x"}] + assert w["used_skills"] == ["s"] + assert w["used_mcp"] == ["m"] + assert w["generating"] is False + + +def test_make_error_wire(): + ev = make_error("崩了", diagnosis={"code": "BOOM", "severity": "error", "summary": "内部错误"}) + w = ev.to_wire() + assert w["type"] == ERROR + assert w["message"] == "崩了" + assert w["diagnosis"]["code"] == "BOOM" + + +def test_ctx_strips_common_keys(): + # 通用上下文键(seq/trace_id/tenant_id/claw) 必须从 kw 抽出,不污染 extra + ev = make_delta("x", seq=5, trace_id="t1", tenant_id="ten", claw="web") + assert ev.seq == 5 + assert ev.trace_id == "t1" + assert ev.tenant_id == "ten" + assert ev.claw == "web" + assert ev.to_wire() == {"type": "delta", "content": "x"} # 默认不输出元数据 + + +# ===================================================================== +# 2. to_wire 与既有前端逐键兼容(新增键非空才输出) +# ===================================================================== +def test_to_wire_omits_empty_meta_by_default(): + ev = make_delta("hi") + w = ev.to_wire() + # 默认不输出 seq/claw/trace_id/tenant_id/diagnosis + assert "seq" not in w + assert "claw" not in w + assert "trace_id" not in w + assert "tenant_id" not in w + assert "diagnosis" not in w + + +def test_to_wire_emits_meta_when_present(): + ev = make_done(message_id="m", conversation_id="c", content="ok", trace_id="tr") + w = ev.to_wire() + # done 终帧本身带 content/message_id 等,trace_id 因非空而输出 + assert w["trace_id"] == "tr" + assert w["message_id"] == "m" + + +# ===================================================================== +# 3. from_wire 把遗留裸 dict 提升为 EngineEvent +# ===================================================================== +def test_from_wire_delta_dict(): + ev = from_wire({"type": "delta", "content": "hi"}) + assert isinstance(ev, EngineEvent) + assert ev.type == DELTA + assert ev.content == "hi" + + +def test_from_wire_passthrough_engineevent(): + orig = make_delta("x") + assert from_wire(orig) is orig + + +def test_from_wire_rejects_malformed(): + assert from_wire(None) is None + assert from_wire("not a dict") is None + assert from_wire({"content": "no type"}) is None + + +def test_from_wire_done_roundtrip_content(): + d = {"type": "done", "message_id": "m", "conversation_id": "c", + "content": "fin", "process_steps": [], "used_skills": [], "used_mcp": [], "generating": False} + ev = from_wire(d) + w = ev.to_wire() + # content 必须原样吐回,前端才能渲染终帧 + assert w["content"] == "fin" + assert w["message_id"] == "m" + + +# ===================================================================== +# 4. SseAdapter 热帧裁剪 + SSE 线格式 +# ===================================================================== +def test_sse_adapter_thins_meta_by_default(): + sse = SseAdapter() + ev = make_delta("热帧", seq=1, trace_id="t", tenant_id="ten", claw="web") + wire = sse._thin(ev, ev.to_wire()) + # 默认剥离 seq/claw/tenant_id;trace_id 在热帧也剥离 + assert "seq" not in wire + assert "claw" not in wire + assert "tenant_id" not in wire + assert "trace_id" not in wire + assert wire == {"type": "delta", "content": "热帧"} + + +def test_sse_adapter_keeps_trace_on_terminal(): + sse = SseAdapter() + ev = make_done(message_id="m", conversation_id="c", content="ok", trace_id="TRACE-1") + wire = sse._thin(ev, ev.to_wire()) + # 终帧保留 trace_id(排障唯一抓手),其余元数据仍剥离 + assert wire["trace_id"] == "TRACE-1" + assert "seq" not in wire + assert "tenant_id" not in wire + + +def test_sse_adapter_wire_meta_true_keeps_all(): + sse = SseAdapter(wire_meta=True) + ev = make_delta("x", seq=2, trace_id="t", tenant_id="ten", claw="web") + wire = sse._thin(ev, ev.to_wire()) + assert wire["seq"] == 2 + assert wire["claw"] == "web" + assert wire["trace_id"] == "t" + assert wire["tenant_id"] == "ten" + + +def test_sse_frame_format(): + frame = SseAdapter.frame({"type": "delta", "content": "hi"}) + assert frame == 'data: {"type": "delta", "content": "hi"}\n\n' + + +def test_sse_frames_generator_until_sentinel(): + async def _run(): + sse = SseAdapter() + await sse.emit(make_delta("一")) + await sse.emit(make_done(message_id="m", conversation_id="c", content="终", trace_id="T")) + await sse.close() + frames = [f async for f in sse.frames()] + return frames + + frames = asyncio.run(_run()) + assert len(frames) == 2 + assert frames[0] == 'data: {"type": "delta", "content": "一"}\n\n' + assert frames[1].startswith('data: {"type": "done"') + assert 'trace_id": "T"' in frames[1] + assert not frames[1].endswith('"}\n\n') or '"}\n\n' in frames[1] + + +def test_sse_emit_counts_and_swallow_exceptions(): + async def _run(): + sse = SseAdapter() + + async def boom(ev): + raise RuntimeError("adapter must not crash engine") + sse._handle = boom # type: ignore + await sse.emit(make_delta("a")) + await sse.emit(make_delta("b")) + return dict(sse.counts) + + counts = asyncio.run(_run()) + # 异常被吞掉,但计数仍记录(Adapter 永不反噬引擎) + assert counts.get("delta") == 2 + + +# ===================================================================== +# 5. BufferAdapter 累积 / 媒体 / 诊断 +# ===================================================================== +def test_buffer_accumulates_delta_and_done_priority(): + buf = BufferAdapter() + asyncio.run(buf.emit(make_delta("第一句"))) + asyncio.run(buf.emit(make_delta("第二句"))) + # 还没 done 时,text 是累积的 delta + assert buf.text == "第一句第二句" + asyncio.run(buf.emit(make_done(message_id="m", conversation_id="c", content="终稿"))) + # done 的 content 优先于累积 delta + assert buf.text == "终稿" + out = buf.assemble() + assert out == "终稿" + + +def test_buffer_drops_reasoning_by_default_keeps_when_flagged(): + buf = BufferAdapter() + asyncio.run(buf.emit(make_reasoning("偷偷想"))) + asyncio.run(buf.emit(make_delta("公开内容"))) + assert "偷偷想" not in buf.text + + buf2 = BufferAdapter(keep_reasoning=True) + asyncio.run(buf2.emit(make_reasoning("可见思考"))) + asyncio.run(buf2.emit(make_delta("正文"))) + assert "可见思考" in buf2.text + + +def test_buffer_extracts_media_struct(): + buf = BufferAdapter() + asyncio.run(buf.emit( + make_done( + message_id="m", conversation_id="c", + content='图', + ) + )) + out = buf.assemble() + # 媒体必须返回结构化 dict(IM 内联发图依赖),禁止退化成纯文本 + assert isinstance(out, dict) + assert out.get("__media__") is True + assert out.get("media_type") == "image" + assert out.get("url") == "https://x/y.png" + + +def test_buffer_diagnosis_tail_note(): + buf = BufferAdapter() + asyncio.run(buf.emit(make_delta("正文"))) + asyncio.run(buf.emit( + make_step("mcp", server="s", name="t", phase="fail", + diagnosis={"code": "E1", "severity": "error", "summary": "工具挂了"}) + )) + asyncio.run(buf.emit(make_done(message_id="m", conversation_id="c", content="正文"))) + out = buf.assemble() + assert "工具挂了" in out # 失败步骤诊断降级成尾注 + + +def test_buffer_error_render(): + buf = BufferAdapter() + asyncio.run(buf.emit(make_error("生成中断", diagnosis={"code": "X", "severity": "error", "summary": "失败"}))) + out = buf.assemble() + assert "失败" in out + + +def test_buffer_truncates_oversized(): + buf = BufferAdapter() + big = "标题\n" + ("长内容" * 6000) + asyncio.run(buf.emit(make_delta(big))) + asyncio.run(buf.emit(make_done(message_id="m", conversation_id="c", content=big))) + out = buf.assemble() + assert len(out) <= 20000 + 40 # 截断上限 + 尾注余量 + assert "已截断" in out + + +# ===================================================================== +# 6. Diagnosis 结构 +# ===================================================================== +def test_diagnosis_to_text_im_degradation(): + d = Diagnosis(code="C", severity=SEV_ERROR, summary="坏了", detail="细节", hint="重试") + txt = d.to_text() + assert "坏了" in txt + assert "细节" in txt + assert "重试" in txt + + +def test_diagnosis_roundtrip(): + d = Diagnosis(code="C", severity=SEV_INFO, summary="s", detail="d", hint="h", retryable=True) + dd = Diagnosis.from_dict(d.to_dict()) + assert dd.code == "C" and dd.retryable is True + + +# ===================================================================== +# 7. 双 Claw 接线契约(消灭第二份实现) +# ===================================================================== +def test_dual_claw_channel_labels(): + assert SseAdapter().claw == CLAW_WEB == "web" + assert BufferAdapter().claw == CLAW_IM == "im" + + +def test_dual_claw_wiring_source_guard(): + """chat.py 必须引用 SseAdapter;im_llm.py 必须引用 BufferAdapter。 + + 这是 #873 单引擎原则的硬约束:能力只在 EngineCore 实现一次, + 两个 Claw 的差异只能收敛在 claw_adapters.py 的两个 Adapter 里, + 禁止在 chat/im_llm 再写一份事件序列化逻辑。 + """ + api_dir = os.path.join(APP_DIR, "api") + chat_src = "" + im_src = "" + for name in ("chat.py", "chat_service.py"): + p = os.path.join(api_dir, name) + if os.path.exists(p): + chat_src = open(p, "r", encoding="utf-8").read() + break + im_path = os.path.join(api_dir, "im_llm.py") + if os.path.exists(im_path): + im_src = open(im_path, "r", encoding="utf-8").read() + + assert "SseAdapter" in chat_src, "chat.py 未引用 SseAdapter(Web Claw 事件序列化可能回退到内联实现)" + assert "BufferAdapter" in im_src, "im_llm.py 未引用 BufferAdapter(IM Claw 可能又出现第二份实现,#866 类回归)" diff --git a/docs/dual-claw-event-contract.md b/docs/dual-claw-event-contract.md new file mode 100644 index 0000000..6ac94ee --- /dev/null +++ b/docs/dual-claw-event-contract.md @@ -0,0 +1,140 @@ +# 双 Claw 引擎事件契约(#873 · P0) + +> 任务归属:加速带 #873(事件契约结构化:diagnosis + IM Bot 缓冲) +> 关联:#874 NodeRegistry(节点身份)、#876 OTel(双 Claw 标签)、#877 落库集成 +> 真相源:`app/core/engine_event.py` + `app/core/claw_adapters.py`(容器 `qihu-aiflow`,镜像 `with-deps=df7adefc7333`) +> 单测哨兵:`tests/test_engine_event.py`(EngineEvent/Adapter 层) + `tests/test_dual_claw.py`(R-DUAL 双端一致性) + +--- + +## 1. Why:为什么需要收敛事件契约 + +此前引擎事件是「散落在 `chat_service` 里的裸 dict」: + +- 构造点分散,字段全靠记忆,漏一个键前端就**静默不渲染**; +- 没有 `trace_id` / `tenant_id` / `claw` 标签,可观测(#876)无处落脚; +- 失败只有一句人话字符串,前端靠 `result.includes('失败')` 猜错误 —— 极脆。 + +#873 把事件收敛为「**唯一构造入口**」:所有事件必须经 `make_*` 工厂产出,再由 `to_wire()` 统一序列化。新增字段一律「**非空才输出**」,保证既有前端(`ChatPage.tsx`)零改动。 + +--- + +## 2. 单引擎原则(最高约束) + +``` +能力只在 EngineCore(`_send_message_core`) 实现一次; +Adapter 只做「引擎事件 → 通道」的搬运,不含任何业务判断。 + + ┌──────────────┐ on_event(EngineEvent) ┌──────────────┐ + │ EngineCore │ ────────────────────────► │ ClawAdapter │ + └──────────────┘ └──────┬───────┘ + │ + ┌───────────────────────────────┴────────────────────────┐ + ▼ ▼ + SseAdapter (Web Claw) BufferAdapter (IM Bot Claw) + 逐帧 `data: {...}\n\n` 缓冲装配 → 整条 str | 媒体 dict +``` + +**历史教训(#866)**:IM 侧曾出现「第二份实现」,一次全量部署就把它冲掉、能力静默回退。因此两个 Claw 的差异**必须、且只能**收敛在本文件的 `SseAdapter` / `BufferAdapter` 里。任何把引擎事件序列化逻辑写回 `chat.py` / `im_llm.py` 的提交,都会被 `test_engine_event.py` 与 `test_dual_claw.py` 的接线哨兵捕获。 + +--- + +## 3. 线格式契约(authoritative —— 前端已消费,禁止破坏) + +| 事件 | 线格式(dict) | +|------|----------------| +| `delta` | `{"type":"delta","content": str}` | +| `reasoning` | `{"type":"reasoning","content": str}` | +| `step` | `{"type":"step","step": {...}, "index"?: int}` | +| `done` | `{"type":"done","message_id","conversation_id","content","process_steps","used_skills","used_mcp","generating"}` | +| `error` | `{"type":"error","message": str}` | + +`step` 对象契约(`AiProcessBlock` 消费): + +| kind | 结构 | +|------|------| +| `thinking` | `{"type":"thinking","text": str}` | +| `skill` | `{"type":"skill","name": str,"args": dict,"result": str}` | +| `mcp` | `{"type":"mcp","server": str,"name": str,"args": dict,"result": str}` | +| `agent`/`kb`/`media` | 同构,按 `type` 选渲染分支 | + +**附加字段(新增,前端可选消费,非空才输出)**: +- `diagnosis`: `{code, severity, summary, detail, hint, retryable}` —— 结构化诊断,取代「把原因塞进 result 字符串」; +- `phase`: `"start" | "ok" | "fail"` —— 步骤执行阶段; +- `node`: `{"id","type","label"}` —— #874 图节点身份; +- `seq` / `trace_id` / `tenant_id` / `claw` —— 可观测维度(#876)。 + +--- + +## 4. EngineEvent API(`engine_event.py`) + +### 工厂(唯一构造入口) +- `make_delta(content, *, seq=0, trace_id="", tenant_id="", claw="")` → 正文增量 +- `make_reasoning(content, **)` → 思维链增量 +- `make_step(kind, *, name="", text="", server="", args=None, result="", phase="", index=None, diagnosis=None, node=None, **)` → 过程步骤 + - `index`:同一步骤「先 `phase=start` 占位,执行完同 `index` 发 `phase=ok|fail` 覆盖」→ 前端实时进度; + - `node`(#874):承载图节点身份,非空才输出,老前端忽略; + - `diagnosis` 为 `error` 且 `result` 为空时,自动用 `diagnosis.to_text()` 兜底 `result`(否则前端判定「运行中」)。 +- `make_done(*, message_id, conversation_id, content="", process_steps=None, used_skills=None, used_mcp=None, generating=False, **)` → 终帧(字段集与遗留 done 帧**完全一致**) +- `make_error(message, *, diagnosis=None, **)` → 错误终帧 + +### 序列化 +- `EngineEvent.to_wire() -> dict`:兼容铁律 —— 既有键原样输出;新增键一律「非空才输出」(不撑大帧体积、不触发未知分支)。 +- `EngineEvent.to_text() -> str`:无结构 UI(IM Bot)降级文本;返回空串表示「本事件无需展示」。 + +### 诊断 +- `Diagnosis(code, severity, summary, detail, hint, retryable)`: + - `to_dict()` / `from_dict(d)` 双向; + - `to_text()`:IM 纯文本降级(`ℹ️/⚠️/❌` 图标 + summary + detail + 建议)。 + +### 迁移兼容 +- `from_wire(payload)`:把遗留裸 dict 提升为 `EngineEvent`(Adapter 收到老代码 `on_event({...})` 后一律归一化)。`type` 缺失 / 非 dict → 返回 `None`(Adapter 丢弃并告警)。 + +--- + +## 5. ClawAdapter API(`claw_adapters.py`) + +基类 `BaseClawAdapter`: +- 实例可直接当 `on_event` 回调传给 EngineCore(`async def __call__(event)` → `emit`); +- `emit(event)`:先 `from_wire` 归一化 → 计数 → `_handle`(异常被吞掉,Adapter 永不反噬引擎); +- `_normalize(event)`:补全 `seq` / `trace_id` / `tenant_id` / `claw`(Adapter 自身默认值兜底)。 + +### SseAdapter(Web Claw,逐帧) +- `claw = "web"`;构造 `SseAdapter(queue=None, *, wire_meta=False)`; +- `_thin(ev, wire)`:**默认剥离** `seq/trace_id/tenant_id/claw`(实测 delta 单帧 46B→117B 膨胀 2.5x,2000 帧多传 ~140KB 噪音,前端不消费);**终帧保留 `trace_id`**(排障唯一抓手); +- `frame(wire)` → `"data: " + json.dumps(wire, ensure_ascii=False) + "\n\n"`; +- `frames()` 异步生成器:从 queue 取帧,遇 `__end__` 哨兵结束; +- `close()`:幂等,投送哨兵帧。 + +### BufferAdapter(IM Bot Claw,缓冲装配) +- `claw = "im"`;构造 `BufferAdapter(*, keep_reasoning=False, keep_diagnosis=True)`; +- 累积规则: + - `delta` 累积正文; + - `reasoning` 默认丢弃(思维链不该刷屏微信),`keep_reasoning=True` 打开; + - `step` 的 `error` 级 `diagnosis` 降级为文本尾注(用户至少知道哪步炸了); + - `done.content` 优先于累积 delta(引擎已做最终裁剪/落库); +- `assemble() -> str | dict`: + - 媒体(以 `