董宏凯
1aeee86984
P0-1 migration: detect the same-named stage-3 execution_stage CHECK by sqltext (must contain rework + manager_review), SQLite rebuild / PG-MySQL drop+add, PG boolean default TRUE, validate_schema checks constraint content + seat_tasks status CHECK/unique/FKs; tests start from a real stage-3 schema. P0-2: a missing/invalid manager review (timeout/exception/empty/invalid JSON/missing per-seat decision) is never an acceptance - system rework under the cap, contract fallback (reassign/controlled_gap/ask_user) afterwards, source=system events. P0-3: unaccepted deliveries are isolated from the fact ledger, cross-review and final report (deferred merge, delivery_json envelope, accepted_task_results in checkpoint, accepted-only synthesis, no turn_results fallback). P0-4: seat-task init is atomic get-or-create over the full expected set; recovery fills missing tasks without overwriting frozen contracts; quality gate blocks missing required tasks. P1: zero config values honoured; duplicate detection persists SHA-256 conclusion digests in history (survives restarts); findings never satisfy evidence, evidence_refs must resolve to trusted ledger refs (unverified_evidence_ref); conversation SSE projection tests added and manager/finalizer messages are no longer mislabelled as seat.delivery.submitted.
395 lines
16 KiB
Python
395 lines
16 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,
|
|
execution_stage: str | None = None,
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {"content": content}
|
|
if usage:
|
|
payload["usage"] = usage
|
|
if metadata:
|
|
payload["metadata"] = metadata
|
|
# The conversation projection must be able to tell a real seat delivery
|
|
# apart from manager / review / finalizer messages — carry the stage.
|
|
if execution_stage:
|
|
payload["execution_stage"] = execution_stage
|
|
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})
|
|
|
|
|
|
# ── Stage 4 (R3-Q) quality gate events ──────────────────────────────────
|
|
|
|
|
|
def make_seat_delivery_submitted(
|
|
agent_id: str,
|
|
turn_index: int,
|
|
*,
|
|
task_key: str,
|
|
attempt: int,
|
|
deterministic_passed: bool,
|
|
failure_codes: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""A seat task submitted its delivery and passed/failed deterministic checks."""
|
|
return _make("seat.delivery.submitted", agent_id, turn_index, {
|
|
"task_key": task_key,
|
|
"attempt": attempt,
|
|
"deterministic_passed": deterministic_passed,
|
|
"failure_codes": list(failure_codes or []),
|
|
})
|
|
|
|
|
|
def make_manager_review_decision(
|
|
agent_id: str,
|
|
turn_index: int,
|
|
*,
|
|
task_key: str,
|
|
decision: str,
|
|
reason: str,
|
|
source: str,
|
|
attempt: int,
|
|
) -> dict[str, Any]:
|
|
"""Server-finalized manager review decision for one seat task."""
|
|
return _make("manager.review.decision", agent_id, turn_index, {
|
|
"task_key": task_key,
|
|
"decision": decision,
|
|
"reason": _truncate(reason, 500),
|
|
"source": source,
|
|
"attempt": attempt,
|
|
})
|
|
|
|
|
|
def make_seat_rework_assigned(
|
|
agent_id: str,
|
|
turn_index: int,
|
|
*,
|
|
task_key: str,
|
|
rework_count: int,
|
|
requirements: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""A seat task is sent back for rework with explicit revision requirements."""
|
|
return _make("seat.rework.assigned", agent_id, turn_index, {
|
|
"task_key": task_key,
|
|
"rework_count": rework_count,
|
|
"revision_requirements": [str(r) for r in (requirements or [])][:16],
|
|
})
|
|
|
|
|
|
def make_seat_reassigned(
|
|
agent_id: str,
|
|
turn_index: int,
|
|
*,
|
|
task_key: str,
|
|
from_agent_id: str,
|
|
reassign_count: int,
|
|
) -> dict[str, Any]:
|
|
"""A seat task is reassigned to a backup seat (agent_id = new owner)."""
|
|
return _make("seat.reassigned", agent_id, turn_index, {
|
|
"task_key": task_key,
|
|
"from_agent_id": from_agent_id,
|
|
"reassign_count": reassign_count,
|
|
})
|
|
|
|
|
|
def make_stage_quality_gate(
|
|
agent_id: str | None,
|
|
turn_index: int,
|
|
*,
|
|
decision: str,
|
|
total: int,
|
|
accepted: int,
|
|
controlled_gap: int,
|
|
blocked_required: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Stage quality gate outcome: passed / blocked / passed_with_gaps."""
|
|
return _make("stage.quality_gate", agent_id or "", turn_index, {
|
|
"decision": decision,
|
|
"total": total,
|
|
"accepted": accepted,
|
|
"controlled_gap": controlled_gap,
|
|
"blocked_required": list(blocked_required or []),
|
|
})
|
|
|
|
|
|
# ── 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
|