"""双 Claw 引擎事件契约(#873 · P0)。 why: 平台有且只有两个 Claw 出口 —— Web Claw(SSE 逐帧) 与 IM Bot Claw(整条推送)。 在此之前,引擎事件是「散落在 chat_service 里的裸 dict」: * 构造点分散,字段全靠记忆,漏一个键前端就静默不渲染; * 没有 trace_id / tenant_id / claw 标签,可观测(#876)无处落脚; * 失败只有一句人话字符串,前端靠 result.includes('失败') 猜错误 —— 极脆。 本模块把事件收敛为「唯一构造入口」:所有事件必须经 make_* 工厂产出, 再由 to_wire() 统一序列化。新增字段一律「非空才输出」, 保证既有前端(ChatPage.tsx)零改动。 线格式契约(authoritative —— 前端已消费,禁止破坏): 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 消费): 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} 附加(新增,前端可选消费): "diagnosis": {...}, "phase": "start|ok|fail", "node": {"id","type","label"} # #874 节点身份 """ from __future__ import annotations import time from dataclasses import dataclass, field from typing import Any, Dict, Optional __all__ = [ "SCHEMA_VERSION", "DELTA", "REASONING", "STEP", "DONE", "ERROR", "EVENT_TYPES", "STEP_THINKING", "STEP_SKILL", "STEP_MCP", "STEP_AGENT", "STEP_KB", "STEP_MEDIA", "PHASE_START", "PHASE_OK", "PHASE_FAIL", "CLAW_WEB", "CLAW_IM", "SEV_INFO", "SEV_WARN", "SEV_ERROR", "Diagnosis", "EngineEvent", "make_delta", "make_reasoning", "make_step", "make_done", "make_error", "from_wire", ] SCHEMA_VERSION = 1 # ---- 事件类型(线格式 type 值,前端已硬编码,禁止改名)---- DELTA = "delta" REASONING = "reasoning" STEP = "step" DONE = "done" ERROR = "error" EVENT_TYPES = frozenset({DELTA, REASONING, STEP, DONE, ERROR}) # ---- step 子类(前端按此选图标 / 渲染分支)---- STEP_THINKING = "thinking" STEP_SKILL = "skill" STEP_MCP = "mcp" STEP_AGENT = "agent" # 新增:外部 Agent 调用 STEP_KB = "kb" # 新增:知识库检索 STEP_MEDIA = "media" # 新增:媒体生成 # ---- 执行阶段 ---- PHASE_START = "start" PHASE_OK = "ok" PHASE_FAIL = "fail" # ---- Claw 通道标签(可观测 #876 的核心维度)---- CLAW_WEB = "web" CLAW_IM = "im" # ---- 诊断严重级 ---- SEV_INFO = "info" SEV_WARN = "warn" SEV_ERROR = "error" _SEV_ICON = {SEV_INFO: "ℹ️", SEV_WARN: "⚠️", SEV_ERROR: "❌"} @dataclass class Diagnosis: """结构化诊断。 取代「把失败原因塞进 result 字符串、前端再 includes('失败') 猜」的旧做法: * Web Claw 可据此渲染诊断卡片(图标/配色/重试按钮); * IM Bot Claw 无卡片 UI,用 to_text() 降级为纯文本行。 """ code: str = "" # 机器可读,如 "MCP_TIMEOUT" / "SKILL_NOT_FOUND" severity: str = SEV_INFO # info | warn | error summary: str = "" # 一句话结论(面向用户) detail: str = "" # 细节(可含技术信息,前端折叠展示) hint: str = "" # 可执行建议 retryable: bool = False # 是否值得重试 def to_dict(self) -> Dict[str, Any]: return { "code": self.code, "severity": self.severity, "summary": self.summary, "detail": self.detail, "hint": self.hint, "retryable": self.retryable, } def to_text(self) -> str: """IM Bot Claw 的纯文本降级渲染。""" icon = _SEV_ICON.get(self.severity, "•") lines = [f"{icon} {self.summary or self.code or '执行提示'}"] if self.detail: lines.append(self.detail) if self.hint: lines.append(f"建议:{self.hint}") return "\n".join(lines) @classmethod def from_dict(cls, d: Any) -> "Optional[Diagnosis]": if isinstance(d, Diagnosis): return d if not isinstance(d, dict): return None return cls( code=str(d.get("code") or ""), severity=str(d.get("severity") or SEV_INFO), summary=str(d.get("summary") or ""), detail=str(d.get("detail") or ""), hint=str(d.get("hint") or ""), retryable=bool(d.get("retryable")), ) @dataclass class EngineEvent: """引擎事件的唯一表示。禁止在业务代码里手搓等价 dict —— 用 make_* 工厂。""" type: str content: str = "" seq: int = 0 ts: float = field(default_factory=time.time) trace_id: str = "" tenant_id: str = "" claw: str = "" diagnosis: Optional[Diagnosis] = None # extra 承载各事件类型的专属键(message_id / step / process_steps ...), # 序列化时平铺到顶层,以保持与既有线格式逐字节兼容。 extra: Dict[str, Any] = field(default_factory=dict) # -- 谓词 --------------------------------------------------------- def is_terminal(self) -> bool: return self.type in (DONE, ERROR) def is_text_chunk(self) -> bool: return self.type in (DELTA, REASONING) # -- 序列化 ------------------------------------------------------- def to_wire(self) -> Dict[str, Any]: """序列化为通道线格式。 兼容铁律:既有键原样输出;新增键(seq/ts/claw/trace_id/tenant_id/diagnosis) 一律「非空才输出」,既不撑大帧体积,也不触发前端未知分支。 """ wire: Dict[str, Any] = {"type": self.type} if self.type in (DELTA, REASONING): wire["content"] = self.content elif self.content: wire["content"] = self.content for key, val in self.extra.items(): wire[key] = val if self.seq: wire["seq"] = self.seq if self.claw: wire["claw"] = self.claw if self.trace_id: wire["trace_id"] = self.trace_id if self.tenant_id: wire["tenant_id"] = self.tenant_id if self.diagnosis is not None: wire["diagnosis"] = self.diagnosis.to_dict() return wire def to_text(self) -> str: """无结构 UI 时(IM Bot)的降级文本;返回空串表示「本事件无需展示」。""" if self.type in (DELTA, DONE): return self.content if self.type == ERROR: if self.diagnosis is not None: return self.diagnosis.to_text() return f"❌ {self.extra.get('message') or self.content or '执行失败'}" if self.type == STEP and self.diagnosis is not None: return self.diagnosis.to_text() return "" # ===================================================================== # 工厂 —— 唯一构造入口 # ===================================================================== def _ctx(kw: Dict[str, Any]) -> Dict[str, Any]: """抽出通用上下文字段,避免每个工厂重复列参数。""" return { "seq": int(kw.pop("seq", 0) or 0), "trace_id": str(kw.pop("trace_id", "") or ""), "tenant_id": str(kw.pop("tenant_id", "") or ""), "claw": str(kw.pop("claw", "") or ""), } def make_delta(content: str, **kw) -> EngineEvent: """正文增量。""" return EngineEvent(type=DELTA, content=content or "", **_ctx(kw)) def make_reasoning(content: str, **kw) -> EngineEvent: """思维链增量。""" return EngineEvent(type=REASONING, content=content or "", **_ctx(kw)) def make_step( kind: str, *, name: str = "", text: str = "", server: str = "", args: Optional[Dict[str, Any]] = None, result: str = "", phase: str = "", index: Optional[int] = None, diagnosis: Any = None, node: Optional[Dict[str, Any]] = None, **kw, ) -> EngineEvent: """过程步骤(思考 / 技能 / MCP / Agent / KB / 媒体)。 index 用于「原地更新同一步骤」:先发 phase=start 占位, 执行完再用同一 index 发 phase=ok|fail 覆盖,前端即得到实时进度。 node(#874)承载图节点身份 {id, type, label}。设计书原文写的是 `EngineEvent(kind="node")`,但前端只认既有 step 类型,发未知 type 会静默丢帧; 故 kind 仍取节点映射后的已知类型,节点身份改由本字段附加 —— 非空才输出, 老前端忽略,新前端可据此渲染 DAG 节点进度。 """ step: Dict[str, Any] = {"type": kind} if kind == STEP_THINKING: step["text"] = text or name or "" else: if name: step["name"] = name if server: step["server"] = server if text: step["text"] = text step["args"] = args or {} step["result"] = result or "" if phase: step["phase"] = phase if isinstance(node, dict) and node: step["node"] = node diag = Diagnosis.from_dict(diagnosis) if diagnosis is not None else None if diag is not None: step["diagnosis"] = diag.to_dict() # result 为空时前端会判定为「运行中」,故失败必须给出可读 result 兜底 if not step["result"] and diag.severity == SEV_ERROR: step["result"] = diag.to_text() extra: Dict[str, Any] = {"step": step} if isinstance(index, int) and index >= 0: extra["index"] = index return EngineEvent(type=STEP, diagnosis=diag, extra=extra, **_ctx(kw)) def make_done( *, message_id: Any = None, conversation_id: Any = None, content: str = "", process_steps: Optional[list] = None, used_skills: Optional[list] = None, used_mcp: Optional[list] = None, generating: bool = False, **kw, ) -> EngineEvent: """终帧。字段集必须与既有 done 帧完全一致。""" extra = { "message_id": message_id, "conversation_id": conversation_id, "content": content or "", "process_steps": process_steps or [], "used_skills": used_skills or [], "used_mcp": used_mcp or [], "generating": bool(generating), } return EngineEvent(type=DONE, content=content or "", extra=extra, **_ctx(kw)) def make_error(message: str, *, diagnosis: Any = None, **kw) -> EngineEvent: """错误终帧。""" diag = Diagnosis.from_dict(diagnosis) if diagnosis is not None else None return EngineEvent( type=ERROR, diagnosis=diag, extra={"message": message or ""}, **_ctx(kw), ) def from_wire(payload: Any) -> Optional[EngineEvent]: """把遗留裸 dict 提升为 EngineEvent(迁移期双向兼容的关键)。 chat_service.py 里仍有直接 await on_event({...}) 的老代码, Adapter 收到后一律经此函数归一化,之后全链路只面对 EngineEvent。 """ if isinstance(payload, EngineEvent): return payload if not isinstance(payload, dict): return None d = dict(payload) etype = str(d.pop("type", "") or "") if not etype: return None content = d.pop("content", "") if not isinstance(content, str): # done 帧的 content 理论上是 str;异常类型保留原值到 extra,避免丢数据 d["content"] = content content = "" diag = Diagnosis.from_dict(d.pop("diagnosis", None)) seq = d.pop("seq", 0) trace_id = d.pop("trace_id", "") tenant_id = d.pop("tenant_id", "") claw = d.pop("claw", "") d.pop("ts", None) # done 帧的 content 需回填到 extra,才能在 to_wire() 时原样吐回 if etype == DONE: d["content"] = content return EngineEvent( type=etype, content=content, seq=int(seq or 0), trace_id=str(trace_id or ""), tenant_id=str(tenant_id or ""), claw=str(claw or ""), diagnosis=diag, extra=d, )