董宏凯andCursor
f82363d81f
- Backend: action_coordinator, capability_snapshot, event_normalizer, tool_executor updates - Backend: approved_tool_executor with write guard, action leases, migration script - Frontend: maf-roundtable module (API client, event timeline, approval cards, fact flow panel) - Frontend: MafRoutes, MafSessionsPage, MafRunDetailPage - Tests: action_leases, migration_phase5, phase5_entry_e2e, real_write_e2e/http_e2e - Docs: update implementation plan Co-authored-by: Cursor <cursoragent@cursor.com>
297 lines
13 KiB
Python
297 lines
13 KiB
Python
"""Translate DeerFlow runtime SSE events into MAF roundtable event types.
|
|
|
|
Maps raw StreamBridge chunks (LangGraph Platform-aligned: ``metadata``,
|
|
``values``, ``updates``, ``messages``, ``error``, ``end``) into the durable
|
|
MAF event taxonomy persisted to ``maf_roundtable_events``.
|
|
|
|
Design constraints:
|
|
- Pure functions only (no I/O, no state).
|
|
- Never fabricates events: only emits what the runtime actually produced.
|
|
- Skill events come ONLY from structured tool/skill metadata, never from
|
|
guessing based on AI text content.
|
|
- All payloads are sanitized (credentials/tokens/cookies stripped).
|
|
- Supports source_event_id for dedup (messages + updates may overlap).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.maf_roundtable.sanitize import sanitize_payload
|
|
|
|
|
|
def normalize_stream_event(
|
|
sse_event: str,
|
|
data: Any,
|
|
*,
|
|
agent_id: str,
|
|
turn_index: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""Convert a single DeerFlow SSE frame into zero or more MAF event dicts.
|
|
|
|
Each returned dict has:
|
|
``{event_type, agent_id, turn_index, payload, source_event_id?}``.
|
|
"""
|
|
if sse_event == "metadata":
|
|
return [_make("run.started", agent_id, turn_index, {"run_id": _get(data, "run_id"), "thread_id": _get(data, "thread_id")})]
|
|
|
|
if sse_event == "messages":
|
|
return _normalize_messages(data, agent_id, turn_index)
|
|
|
|
if sse_event == "updates":
|
|
return _normalize_updates(data, agent_id, turn_index)
|
|
|
|
if sse_event == "error":
|
|
return [_make("run.failed", agent_id, turn_index, {"message": _get(data, "message", ""), "name": _get(data, "name", "")})]
|
|
|
|
# Structured skill lifecycle events (emitted by DeerFlow skill middleware,
|
|
# never guessed from AI text content).
|
|
if sse_event in ("invoke_skill", "skill_search_step"):
|
|
skill_name = _get(data, "skill_name", "") or ""
|
|
if sse_event == "invoke_skill":
|
|
return [_make("skill.started", agent_id, turn_index, {"skill_name": skill_name, "message": _get(data, "message", "")})]
|
|
return [_make("skill.selected", agent_id, turn_index, {"skill_name": skill_name, "message": _get(data, "message", "")})]
|
|
|
|
if sse_event == "skill_result":
|
|
return [_make("skill.completed", agent_id, turn_index, {"skill_name": _get(data, "skill_name", ""), "message": _get(data, "message", "")})]
|
|
|
|
if sse_event == "skill_error":
|
|
return [_make("skill.failed", agent_id, turn_index, {"skill_name": _get(data, "skill_name", ""), "error": _get(data, "reason", "") or _get(data, "message", "")})]
|
|
|
|
if sse_event == "artifact_created":
|
|
return [_make("artifact.created", agent_id, turn_index, {"artifact_type": _get(data, "artifact_type", ""), "artifact_id": _get(data, "artifact_id", ""), "title": _get(data, "title", "")})]
|
|
|
|
return []
|
|
|
|
|
|
def make_turn_started(agent_id: str, turn_index: int, payload: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
return _make("agent.turn_started", agent_id, turn_index, payload or {})
|
|
|
|
|
|
def make_turn_completed(
|
|
agent_id: str,
|
|
turn_index: int,
|
|
*,
|
|
content: str,
|
|
usage: dict[str, Any] | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {"content": content}
|
|
if usage:
|
|
payload["usage"] = usage
|
|
if metadata:
|
|
payload["metadata"] = metadata
|
|
return _make("agent.message.completed", agent_id, turn_index, payload)
|
|
|
|
|
|
def make_run_completed(agent_id: str | None, turn_index: int, payload: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
return _make("run.completed", agent_id or "", turn_index, payload or {})
|
|
|
|
|
|
def make_run_failed(agent_id: str | None, turn_index: int, *, error_code: str, message: str) -> dict[str, Any]:
|
|
return _make("run.failed", agent_id or "", turn_index, {"error_code": error_code, "message": message})
|
|
|
|
|
|
def make_run_cancelled(agent_id: str | None, turn_index: int) -> dict[str, Any]:
|
|
return _make("run.cancelled", agent_id or "", turn_index, {})
|
|
|
|
|
|
# ── skill lifecycle events ─────────────────────────────────────────────────
|
|
|
|
|
|
def make_skill_selected(agent_id: str, turn_index: int, *, skill_name: str, skill_id: str = "") -> dict[str, Any]:
|
|
return _make("skill.selected", agent_id, turn_index, {"skill_name": skill_name, "skill_id": skill_id})
|
|
|
|
|
|
def make_skill_started(agent_id: str, turn_index: int, *, skill_name: str, skill_id: str = "") -> dict[str, Any]:
|
|
return _make("skill.started", agent_id, turn_index, {"skill_name": skill_name, "skill_id": skill_id})
|
|
|
|
|
|
def make_skill_completed(agent_id: str, turn_index: int, *, skill_name: str, result_preview: str = "") -> dict[str, Any]:
|
|
return _make("skill.completed", agent_id, turn_index, {"skill_name": skill_name, "result_preview": _truncate(result_preview, 500)})
|
|
|
|
|
|
def make_skill_failed(agent_id: str, turn_index: int, *, skill_name: str, error: str = "") -> dict[str, Any]:
|
|
return _make("skill.failed", agent_id, turn_index, {"skill_name": skill_name, "error": _truncate(error, 500)})
|
|
|
|
|
|
# ── artifact events ────────────────────────────────────────────────────────
|
|
|
|
|
|
def make_artifact_created(agent_id: str, turn_index: int, *, artifact_type: str, artifact_id: str = "", title: str = "") -> dict[str, Any]:
|
|
return _make("artifact.created", agent_id, turn_index, {"artifact_type": artifact_type, "artifact_id": artifact_id, "title": title})
|
|
|
|
|
|
# ── approval lifecycle events ──────────────────────────────────────────────
|
|
|
|
|
|
def make_action_waiting_approval(agent_id: str, turn_index: int, *, action_id: str, approval_id: str, tool_name: str, summary: str = "") -> dict[str, Any]:
|
|
return _make("action.waiting_approval", agent_id, turn_index, {"action_id": action_id, "approval_id": approval_id, "tool_name": tool_name, "summary": _truncate(summary, 500)})
|
|
|
|
|
|
def make_action_approved(agent_id: str, turn_index: int, *, action_id: str, approval_id: str, decided_by: str = "") -> dict[str, Any]:
|
|
return _make("action.approved", agent_id, turn_index, {"action_id": action_id, "approval_id": approval_id, "decided_by": decided_by})
|
|
|
|
|
|
def make_action_rejected(agent_id: str, turn_index: int, *, action_id: str, approval_id: str, decided_by: str = "") -> dict[str, Any]:
|
|
return _make("action.rejected", agent_id, turn_index, {"action_id": action_id, "approval_id": approval_id, "decided_by": decided_by})
|
|
|
|
|
|
def make_action_unknown(agent_id: str, turn_index: int, *, action_id: str, reason_code: str) -> dict[str, Any]:
|
|
"""Record an inconclusive external outcome without leaking tool arguments."""
|
|
return _make("action.unknown", agent_id, turn_index, {"action_id": action_id, "reason_code": reason_code})
|
|
|
|
|
|
def make_action_waiting_target_lease(agent_id: str, turn_index: int, *, action_id: str, tool_name: str) -> dict[str, Any]:
|
|
"""A second approved write is serialized behind an active target lease."""
|
|
return _make("action.waiting_target_lease", agent_id, turn_index, {"action_id": action_id, "tool_name": tool_name})
|
|
|
|
|
|
# ── workflow checkpoint events ─────────────────────────────────────────────
|
|
|
|
|
|
def make_workflow_checkpointed(agent_id: str | None, turn_index: int, *, checkpoint_id: str, reason: str = "") -> dict[str, Any]:
|
|
return _make("workflow.checkpointed", agent_id or "", turn_index, {"checkpoint_id": checkpoint_id, "reason": reason})
|
|
|
|
|
|
# ── internal helpers ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make(event_type: str, agent_id: str, turn_index: int, payload: dict[str, Any], source_event_id: str | None = None) -> dict[str, Any]:
|
|
ev: dict[str, Any] = {
|
|
"event_type": event_type,
|
|
"agent_id": agent_id,
|
|
"turn_index": turn_index,
|
|
"payload": sanitize_payload(payload),
|
|
}
|
|
if source_event_id:
|
|
ev["source_event_id"] = source_event_id
|
|
return ev
|
|
|
|
|
|
def _get(data: Any, key: str, default: Any = None) -> Any:
|
|
if isinstance(data, dict):
|
|
return data.get(key, default)
|
|
return default
|
|
|
|
|
|
def _normalize_messages(data: Any, agent_id: str, turn_index: int) -> list[dict[str, Any]]:
|
|
"""Handle ``messages`` SSE event (streaming token deltas).
|
|
|
|
Format: ``[message_id, {type, content, ...}]`` (messages-tuple mode).
|
|
"""
|
|
events: list[dict[str, Any]] = []
|
|
if not isinstance(data, (list, tuple)) or len(data) < 2:
|
|
return events
|
|
msg_id = str(data[0]) if data[0] else None
|
|
msg = data[1] if len(data) > 1 else None
|
|
if not isinstance(msg, dict):
|
|
return events
|
|
|
|
msg_type = msg.get("type") or msg.get("role") or ""
|
|
content = msg.get("content")
|
|
|
|
if msg_type in ("ai", "assistant"):
|
|
tool_calls = msg.get("tool_calls")
|
|
if tool_calls:
|
|
for tc in tool_calls:
|
|
if not isinstance(tc, dict):
|
|
continue
|
|
tc_id = tc.get("id", "")
|
|
events.append(_make("tool.requested", agent_id, turn_index, {
|
|
"tool_name": tc.get("name", ""),
|
|
"tool_call_id": tc_id,
|
|
"args": _summarize_args(tc.get("args")),
|
|
}, source_event_id=f"{msg_id}:{tc_id}" if msg_id and tc_id else None))
|
|
if content:
|
|
text = _extract_text(content)
|
|
if text:
|
|
events.append(_make("agent.message.delta", agent_id, turn_index, {"text": text}, source_event_id=msg_id))
|
|
|
|
elif msg_type == "tool":
|
|
tc_id = msg.get("tool_call_id", "")
|
|
status = msg.get("status", "completed")
|
|
event_type = "tool.completed" if status != "error" else "tool.failed"
|
|
events.append(_make(event_type, agent_id, turn_index, {
|
|
"tool_name": msg.get("name", ""),
|
|
"tool_call_id": tc_id,
|
|
"content_preview": _truncate(_extract_text(content), 500),
|
|
"status": status,
|
|
}, source_event_id=f"tool:{tc_id}" if tc_id else None))
|
|
|
|
return events
|
|
|
|
|
|
def _normalize_updates(data: Any, agent_id: str, turn_index: int) -> list[dict[str, Any]]:
|
|
"""Handle ``updates`` SSE event (node completion with full output).
|
|
|
|
Only emits tool lifecycle events from structured data. Does NOT guess
|
|
skill execution from text content.
|
|
"""
|
|
events: list[dict[str, Any]] = []
|
|
if not isinstance(data, dict):
|
|
return events
|
|
|
|
for node_name, node_output in data.items():
|
|
if not isinstance(node_output, dict):
|
|
continue
|
|
messages = node_output.get("messages")
|
|
if not isinstance(messages, list):
|
|
continue
|
|
for msg in messages:
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
msg_type = msg.get("type") or msg.get("role") or ""
|
|
if msg_type in ("ai", "assistant"):
|
|
tool_calls = msg.get("tool_calls")
|
|
if tool_calls:
|
|
for tc in tool_calls:
|
|
if not isinstance(tc, dict):
|
|
continue
|
|
tc_id = tc.get("id", "")
|
|
events.append(_make("tool.started", agent_id, turn_index, {
|
|
"tool_name": tc.get("name", ""),
|
|
"tool_call_id": tc_id,
|
|
"node": node_name,
|
|
}, source_event_id=f"upd:{tc_id}" if tc_id else None))
|
|
elif msg_type == "tool":
|
|
tc_id = msg.get("tool_call_id", "")
|
|
status = msg.get("status", "completed")
|
|
event_type = "tool.completed" if status != "error" else "tool.failed"
|
|
events.append(_make(event_type, agent_id, turn_index, {
|
|
"tool_name": msg.get("name", ""),
|
|
"tool_call_id": tc_id,
|
|
"content_preview": _truncate(_extract_text(msg.get("content")), 500),
|
|
"status": status,
|
|
"node": node_name,
|
|
}, source_event_id=f"upd:tool:{tc_id}" if tc_id else None))
|
|
return events
|
|
|
|
|
|
def _extract_text(content: Any) -> str:
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts = []
|
|
for p in content:
|
|
if isinstance(p, dict) and p.get("type") == "text":
|
|
parts.append(p.get("text", ""))
|
|
elif isinstance(p, str):
|
|
parts.append(p)
|
|
return "\n".join(parts)
|
|
return ""
|
|
|
|
|
|
def _truncate(text: str, max_len: int) -> str:
|
|
if len(text) <= max_len:
|
|
return text
|
|
return text[:max_len] + "..."
|
|
|
|
|
|
def _summarize_args(args: Any) -> dict[str, Any] | None:
|
|
if not isinstance(args, dict):
|
|
return None
|
|
from app.maf_roundtable.sanitize import sanitize
|
|
result = sanitize(args)
|
|
return result if isinstance(result, dict) else None
|