docs+tests: #873 事件契约与双 Claw Adapter 文档及哨兵测试

- 新增 tests/test_engine_event.py (31 用例: EngineEvent 线格式/to_wire/from_wire,
  SseAdapter 裁剪+帧格式+frames, BufferAdapter 累积/媒体/诊断/截断, Diagnosis,
  test_dual_claw_wiring_source_guard 校验 chat.py/im_llm.py 接线)
- 新增 docs/dual-claw-event-contract.md (#873 契约文档 7 节)
- 全量 pytest 64 用例全绿 (含 test_dual_claw/test_node_registry 无回归)
- 仅文档/测试,非运行时代码,不 push
This commit is contained in:
QihuAI Design committed 2026-08-04 17:28:57 +08:00
1 parent 33d0f9c32e
commit c1f8b164bb
2 files changed
+536

No files matched your search

+396
View File
@@ -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='<img src="https://x/y.png" alt="图">',
)
))
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 类回归)"
+140
View File
@@ -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`:
- 媒体(以 `<img` / `<video` 开头)→ 返回结构化 `{"__media__": True, "media_type": "image"|"video", "url": "..."}`(IM 内联发图/发视频依赖,禁止退化成纯文本);
- 空 body + error → `_render_failure()`;空 body + generating → 「内容正在生成」;否则正文 + 诊断尾注 + 生成提示;
- 超 `_MAX_ASSEMBLED_CHARS=20000` 截断并标注「内容过长已截断」。
---
## 6. Web / IM 双 Claw 接线方式
| Claw | 入口 | Adapter | 线格式 |
|------|------|---------|--------|
| Web | `chat.py` `/messages/stream` | `SseAdapter`(委托,线格式逐字节一致) | SSE `data:` 逐帧 |
| IM Bot | `im_llm.py`(灰阶门控) | `BufferAdapter` | 整条 `str` / `__media__` dict |
- `im_llm.py` 经 `BufferAdapter` 复用 `EngineCore`(`_call_engine_core`),媒体走快路径;
- **IM_USE_ENGINE_CORE 灰阶门(默认 OFF)**:生产零改动,开启后 IM 走统一引擎,单引擎红利即时生效;
- 两个 Adapter 必须分别被 `chat.py` / `im_llm.py` 引用(接线哨兵 `test_engine_event.py::test_dual_claw_wiring_source_guard` 锁死)。
---
## 7. 测试与可观测
- `tests/test_engine_event.py`:EngineEvent 工厂 / `to_wire` 兼容 / `from_wire` / SseAdapter 裁剪与 SSE 格式 / BufferAdapter 累积与媒体 / 双 Claw 接线契约(确定性、无 DB、无网络);
- `tests/test_dual_claw.py`(R-DUAL-*):Web/IM 能力一致、命名同源、`__media__` 保留、OTel `tenant_id`+`claw_type` 标签;
- 指标(#876):`aiflow_span_count` / `aiflow_span_duration_ms` / `aiflow_llm_route_total`,维度含 `claw_type ∈ {web, im_wechat}`、`tenant_id`,Grafana 看板 `qihuaiflow-dual-claw-876`。