qihuaiflow/backend/python/app/core/engine_event.py

"""双 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,
    )