"""DeerFlow Agent Adapter for the MAF roundtable experiment (Phase 2). Drives each MAF seat's actual execution through the existing DeerFlow Agent Runtime, preserving tools / skills / MCP / sandbox / identity / middleware semantics. The adapter does NOT wrap individual LangChain tools — it invokes a full agent run per seat turn via ``DeerFlowRuntimeClient``. Architecture: - One DeerFlow thread per seat (bound at session start, reused across turns). - In-process ``start_run`` via DeerFlowRuntimeClient (no fake Request). - User context injected via ExecutionPrincipal (server-generated, never from body). - StreamBridge subscription translates runtime SSE → MAF events in real time. - Cancellation propagates from MAF run → DeerFlow run via RunManager.cancel. - All seat runs persisted to ``maf_roundtable_seat_runs`` (survives restart). - Critical audit events (tool.*, skill.*, approval.*, artifact.*) MUST persist; audit failure stops external writes; seat turn failures are soft-failed into ``seat_outcomes`` / ``partial_failure`` (structured mode). Boundary: lives in ``app.maf_roundtable`` (may import ``deerflow.*`` and ``app.gateway.*``); never imported by ``deerflow.*``. """ from __future__ import annotations import asyncio import logging import os import time from dataclasses import dataclass, field from typing import Any from app.maf_roundtable.context_builder import build_seat_input from app.maf_roundtable.event_normalizer import ( make_run_cancelled, make_run_failed, make_turn_completed, make_turn_started, normalize_stream_event, ) from app.maf_roundtable.runtime_client import DeerFlowRuntimeClient, ExecutionPrincipal from app.maf_roundtable.sanitize import sanitize from deerflow.runtime.stream_bridge.base import END_SENTINEL, HEARTBEAT_SENTINEL logger = logging.getLogger(__name__) _IDLE_TIMEOUT_S = int(os.getenv("MAF_SEAT_IDLE_TIMEOUT_SECONDS", "1200") or "1200") _IDLE_CHECK_INTERVAL_S = 10.0 _MAX_CONTEXT_CHARS = int(os.getenv("MAF_MAX_CONTEXT_CHARS", "12000") or "12000") _CRITICAL_EVENT_TYPES = frozenset({ "tool.requested", "tool.started", "tool.completed", "tool.failed", "skill.selected", "skill.started", "skill.completed", "skill.failed", "artifact.created", "action.waiting_approval", "action.approved", "action.rejected", "run.completed", "run.failed", "run.cancelled", }) class AuditEventError(RuntimeError): """Raised when a critical audit event fails to persist.""" @dataclass class SeatResponse: """Standardized response from a seat turn (MAF-02-08).""" agent_id: str turn_index: int content: str tool_calls: list[dict[str, Any]] = field(default_factory=list) usage: dict[str, Any] | None = None metadata: dict[str, Any] | None = None error: str | None = None latency_ms: int | None = None class DeerFlowAgentAdapter: """Drives MAF seat execution through the DeerFlow Agent Runtime. Usage:: client = DeerFlowRuntimeClient(app) adapter = DeerFlowAgentAdapter(client, service=svc, user_id="u1") await adapter.ensure_threads(session_id, participants) response = await adapter.run_seat_turn(...) """ def __init__( self, runtime_client: DeerFlowRuntimeClient, *, service: Any, user_id: str, ) -> None: self._client = runtime_client self._service = service self._user_id = user_id self._seen_source_ids: set[str] = set() # ── MAF-02-01: thread lifecycle ───────────────────────────────────────── async def ensure_threads( self, session_id: str, run_id: str, participants: list[dict[str, Any]] ) -> dict[str, str]: """Create and bind a DeerFlow thread for each participant that lacks one. Thread creation is wrapped in set_current_user so thread_meta.user_id matches the MAF session owner. Returns ``{agent_id: thread_id}``. """ principal = ExecutionPrincipal(user_id=self._user_id, session_id=session_id, run_id=run_id) thread_map: dict[str, str] = {} for p in participants: agent_id = p.get("agent_id", "") existing = p.get("thread_id") if existing: thread_map[agent_id] = existing continue thread_id = await self._client.create_thread(principal) await self._service.store.participants.bind_thread( session_id, agent_id, thread_id=thread_id ) thread_map[agent_id] = thread_id return thread_map # ── MAF-02-02/03/04/05: seat turn execution ───────────────────────────── async def run_seat_turn( self, *, session_id: str, run_id: str, agent_id: str, turn_index: int, message: str, thread_id: str, execution_stage: str = "expert", model_name: str | None = None, reasoning_effort: str | None = None, thinking_enabled: bool = True, subagent_enabled: bool = False, excluded_tools: list[str] | None = None, stage_instruction: str = "", prior_turns: list[dict[str, Any]] | None = None, fact_ledger: dict[str, Any] | None = None, role_description: str = "", task_text: str = "", memory_recall_disabled: bool | None = None, memory_injection_enabled: bool | None = None, capability_snapshot: dict[str, Any] | None = None, write_policy: str = "confirm_high_risk", approved_tool: str | None = None, wait_timeout_seconds: float | None = None, ) -> SeatResponse: """Execute one seat turn through the DeerFlow Agent Runtime.""" t0 = time.perf_counter() seat_message = build_seat_input( task_text=task_text or message, stage_instruction=stage_instruction, role_description=role_description, prior_turns=prior_turns or [], fact_ledger=fact_ledger, max_context_chars=_MAX_CONTEXT_CHARS, ) if not task_text: seat_message = message await self._emit_critical(run_id, make_turn_started(agent_id, turn_index, {"thread_id": thread_id})) principal = ExecutionPrincipal(user_id=self._user_id, session_id=session_id, run_id=run_id) # Phase-3 P0-1: atomic "occupy or read". Only the caller that wins the # insert (created=True) may start a DeerFlow run. An existing row is # handled by status so we NEVER silently start a second DeerFlow # execution for the same four-tuple (run, agent, turn, stage). seat_run, created = await self._service.store.seat_runs.get_or_create_seat_run( maf_run_id=run_id, agent_id=agent_id, turn_index=turn_index, execution_stage=execution_stage, deerflow_thread_id=thread_id, ) seat_run_id = seat_run["id"] prior_status = seat_run.get("status") if not created: if prior_status == "completed": return self._seat_response_from_snapshot( seat_run=seat_run, agent_id=agent_id, turn_index=turn_index, ) if prior_status in ("pending", "running", "cancelling"): effective_timeout = wait_timeout_seconds if wait_timeout_seconds is not None else 600.0 seat_run = await self._service.store.seat_runs.wait_for_terminal_seat_run( maf_run_id=run_id, agent_id=agent_id, turn_index=turn_index, execution_stage=execution_stage, timeout_seconds=effective_timeout, ) if seat_run["status"] == "completed": return self._seat_response_from_snapshot( seat_run=seat_run, agent_id=agent_id, turn_index=turn_index, ) return SeatResponse( agent_id=agent_id, turn_index=turn_index, content="", error=f"stage '{execution_stage}' ended as {seat_run['status']}", metadata={"seat_run_id": seat_run["id"], "terminal": seat_run["status"]}, latency_ms=int((time.perf_counter() - t0) * 1000), ) return SeatResponse( agent_id=agent_id, turn_index=turn_index, content="", error=f"stage '{execution_stage}' previously terminal ({prior_status}); " f"manual recovery required", metadata={"terminal": prior_status, "seat_run_id": seat_run_id}, latency_ms=int((time.perf_counter() - t0) * 1000), ) try: response = await self._execute_run( principal=principal, seat_run_id=seat_run_id, run_id=run_id, agent_id=agent_id, turn_index=turn_index, thread_id=thread_id, message=seat_message, model_name=model_name, reasoning_effort=reasoning_effort, thinking_enabled=thinking_enabled, subagent_enabled=subagent_enabled, excluded_tools=excluded_tools or [], memory_recall_disabled=memory_recall_disabled, memory_injection_enabled=memory_injection_enabled, capability_snapshot=capability_snapshot, write_policy=write_policy, approved_tool=approved_tool, ) except asyncio.CancelledError: await self._service.store.seat_runs.mark_cancelled(seat_run_id) await self._emit_critical(run_id, make_run_cancelled(agent_id, turn_index)) raise except Exception as exc: latency = int((time.perf_counter() - t0) * 1000) # Sanitize before persisting / emitting — the raw exception body may # embed params or credentials; only the redacted text is stored. safe_error = str(sanitize(str(exc)))[:500] await self._service.store.seat_runs.mark_failed(seat_run_id, error=safe_error) await self._emit_critical(run_id, make_run_failed(agent_id, turn_index, error_code=type(exc).__name__, message=safe_error)) return SeatResponse( agent_id=agent_id, turn_index=turn_index, content="", error=safe_error, latency_ms=latency, ) raw_snapshot = { "content": response.content, "tool_calls": response.tool_calls, "usage": response.usage, "metadata": response.metadata, } snapshot = sanitize(raw_snapshot) if not isinstance(snapshot, dict): raise RuntimeError("seat result snapshot sanitization produced invalid payload") await self._service.store.seat_runs.mark_completed(seat_run_id, result_snapshot=snapshot) response.latency_ms = int((time.perf_counter() - t0) * 1000) await self._emit_critical(run_id, make_turn_completed( agent_id, turn_index, content=response.content, usage=response.usage, metadata=response.metadata, )) return response async def _execute_run( self, *, principal: ExecutionPrincipal, seat_run_id: str, run_id: str, agent_id: str, turn_index: int, thread_id: str, message: str, model_name: str | None, reasoning_effort: str | None, thinking_enabled: bool, subagent_enabled: bool, excluded_tools: list[str], memory_recall_disabled: bool | None, memory_injection_enabled: bool | None, capability_snapshot: dict[str, Any] | None = None, write_policy: str = "confirm_high_risk", approved_tool: str | None = None, ) -> SeatResponse: record = await self._client.start_agent_run( principal, thread_id=thread_id, agent_id=agent_id, message=message, model_name=model_name, reasoning_effort=reasoning_effort, thinking_enabled=thinking_enabled, subagent_enabled=subagent_enabled, excluded_tools=excluded_tools, memory_recall_disabled=memory_recall_disabled, memory_injection_enabled=memory_injection_enabled, turn_index=turn_index, capability_snapshot=capability_snapshot, write_policy=write_policy, approved_tool=approved_tool, ) deerflow_run_id = getattr(record, "run_id", None) if deerflow_run_id: await self._service.store.seat_runs.mark_running(seat_run_id, deerflow_run_id=deerflow_run_id) await self._await_with_event_translation( record, deerflow_run_id, run_id=run_id, agent_id=agent_id, turn_index=turn_index ) messages = await self._client.read_thread_messages(principal, thread_id) content = self._latest_ai_text(messages) tool_calls = self._tool_calls_since_last_human(messages) usage = self._extract_usage(messages) return SeatResponse( agent_id=agent_id, turn_index=turn_index, content=content, tool_calls=tool_calls, usage=usage, metadata={"thread_id": thread_id, "deerflow_run_id": deerflow_run_id}, ) # ── MAF-02-07: real-time event translation ────────────────────────────── async def _await_with_event_translation( self, record: Any, deerflow_run_id: str | None, *, run_id: str, agent_id: str, turn_index: int, ) -> None: """Await the DeerFlow run while translating StreamBridge events to MAF events. P1.5 fix: ``if task in done: await task`` so exceptions propagate. """ task = getattr(record, "task", None) if task is None: return bridge = getattr(self._client.app.state, "stream_bridge", None) if bridge is None or deerflow_run_id is None: await task return loop = asyncio.get_running_loop() last_activity = loop.time() stuck = asyncio.Event() async def _watch() -> None: nonlocal last_activity async for ev in bridge.subscribe(deerflow_run_id, heartbeat_interval=_IDLE_CHECK_INTERVAL_S): if ev is END_SENTINEL: return now = loop.time() if ev is HEARTBEAT_SENTINEL: if now - last_activity >= _IDLE_TIMEOUT_S: stuck.set() return else: last_activity = now maf_events = normalize_stream_event( ev.event, ev.data, agent_id=agent_id, turn_index=turn_index ) for me in maf_events: await self._emit(run_id, me) watch = asyncio.create_task(_watch()) try: done, _ = await asyncio.wait({task, watch}, return_when=asyncio.FIRST_COMPLETED) if task in done: await task return if not stuck.is_set(): await task return idle_elapsed = loop.time() - last_activity logger.error( "MAF seat '%s' stalled: no output for %.0fs on run %s", agent_id, idle_elapsed, deerflow_run_id, ) await self._client.cancel_run(deerflow_run_id) try: await asyncio.wait_for(asyncio.shield(task), timeout=10.0) except (TimeoutError, asyncio.CancelledError, Exception): logger.warning("DeerFlow task did not finish after cancel for run %s", deerflow_run_id) raise RuntimeError(f"MAF seat '{agent_id}' stalled (no output for {_IDLE_TIMEOUT_S}s)") finally: if not watch.done(): watch.cancel() try: await watch except BaseException: pass # ── MAF-02-09: cancellation ───────────────────────────────────────────── async def cancel(self, maf_run_id: str) -> dict[str, Any]: """Cancel all active DeerFlow runs for a MAF run (two-phase cascade). Phase 1: mark all active seat runs as ``cancelling``. Phase 2: actually cancel each DeerFlow run, then mark the seat run as ``cancelled`` (success) or ``cancel_failed`` (failure). Returns a structured result so the caller (dispatcher → router) can decide whether to finalize the parent run as ``cancelled`` or leave it in a non-terminal ``cancelling`` / ``cancel_failed`` state. A failed cancel must NEVER masquerade as cancelled. Result shape:: { "all_cancelled": bool, "cancelled_run_ids": [...], # seat_run_ids that succeeded "failed_run_ids": [...], # seat_run_ids where cancel failed "still_active_run_ids": [...], # seat_run_ids with no deerflow id } """ seat_infos = await self._service.store.seat_runs.cancel_all_active(maf_run_id) cancelled_run_ids: list[str] = [] failed_run_ids: list[str] = [] still_active_run_ids: list[str] = [] for info in seat_infos: seat_run_id = info["seat_run_id"] deerflow_run_id = info.get("deerflow_run_id") if not deerflow_run_id: # No DeerFlow run to cancel — record as still-active/unknown. await self._service.store.seat_runs.mark_cancel_unknown( seat_run_id, error="no deerflow_run_id recorded" ) still_active_run_ids.append(seat_run_id) continue if await self._client.cancel_run(deerflow_run_id): await self._service.store.seat_runs.mark_cancelled(seat_run_id) cancelled_run_ids.append(seat_run_id) else: await self._service.store.seat_runs.mark_cancel_failed( seat_run_id, error="DeerFlow cancel_run returned False" ) failed_run_ids.append(seat_run_id) all_cancelled = not failed_run_ids and not still_active_run_ids and bool(seat_infos) # If there were no active seat runs at all, treat as fully cancelled # (nothing left to cancel). if not seat_infos: all_cancelled = True return { "all_cancelled": all_cancelled, "cancelled_run_ids": cancelled_run_ids, "failed_run_ids": failed_run_ids, "still_active_run_ids": still_active_run_ids, } # ── P1.6: audit event persistence ─────────────────────────────────────── async def _emit_critical(self, run_id: str, event: dict[str, Any]) -> None: """Emit a critical audit event. Failure raises AuditEventError. The persisted audit event row's ``id`` is passed to the conversation bridge as ``source_audit_event_id`` so projection dedup is always keyed on the real audit row — not on an ephemeral input field that may be absent for critical events. """ persisted: dict[str, Any] | None = None try: persisted = await self._service.append_event( run_id=run_id, user_id=self._user_id, event_type=event["event_type"], agent_id=event.get("agent_id"), turn_index=event.get("turn_index"), payload=event.get("payload"), ) except Exception as exc: logger.error("CRITICAL audit event %s failed to persist for run %s", event.get("event_type"), run_id, exc_info=True) raise AuditEventError( f"Critical audit event '{event.get('event_type')}' failed to persist: {exc}" ) from exc await self._bridge_conversation_event( run_id, event, source_audit_event_id=persisted["id"] if persisted else None, ) async def _emit(self, run_id: str, event: dict[str, Any]) -> None: """Emit a non-critical event (agent.message.delta). Failure degrades gracefully. Dedup: events with a source_event_id are deduped at the DB level via the (run_id, source_event_id) unique constraint — durable across restarts. The in-memory _seen_source_ids set is a fast-path cache only. """ source_id = event.get("source_event_id") if source_id: if source_id in self._seen_source_ids: return self._seen_source_ids.add(source_id) event_type = event.get("event_type", "") if event_type in _CRITICAL_EVENT_TYPES: await self._emit_critical(run_id, event) return persisted: dict[str, Any] | None = None try: persisted = await self._service.append_event( run_id=run_id, user_id=self._user_id, event_type=event_type, agent_id=event.get("agent_id"), turn_index=event.get("turn_index"), payload=event.get("payload"), source_event_id=source_id, ) except Exception: logger.warning("non-critical event %s failed for run %s (degraded)", event_type, run_id, exc_info=True) return await self._bridge_conversation_event( run_id, event, source_audit_event_id=persisted["id"] if persisted else None, ) async def _bridge_conversation_event( self, run_id: str, event: dict[str, Any], *, source_audit_event_id: str | None = None, ) -> None: """Best-effort projection into the V2 conversation event stream. The per-run audit write above remains authoritative. A projection failure must not cause a tool to execute twice or turn a completed seat run into a false runtime failure. ``source_audit_event_id`` is the real persisted audit row id — it is always preferred over ``event["id"]`` / ``event["source_event_id"]`` for dedup (P0-4 fix). When the bridge retries after a restart, the same audit id ensures the conversation event is written exactly once. P0-3: the entire ``app.state`` resolution is best-effort. If any link is missing (no app, no state, no conversation service), the projection is silently skipped — audit, recovery, cancel, lease, and tool execution safety are never affected. """ try: app = getattr(self._client, "app", None) state = getattr(app, "state", None) conversation_service = getattr(state, "maf_conversation_service", None) except Exception: # noqa: BLE001 return if conversation_service is None: return bridge = getattr(conversation_service, "bridge_workflow_event", None) if not callable(bridge): return try: await bridge( run_id=run_id, event=event, source_audit_event_id=source_audit_event_id, ) except Exception: # noqa: BLE001 - audit already persisted logger.warning( "conversation event projection failed for run %s (%s)", run_id, event.get("event_type"), exc_info=True, ) # ── internal helpers ──────────────────────────────────────────────────── def _seat_response_from_snapshot( self, *, seat_run: dict[str, Any], agent_id: str, turn_index: int, ) -> SeatResponse: snapshot = seat_run.get("result_snapshot") if not snapshot or not isinstance(snapshot, dict): return SeatResponse( agent_id=agent_id, turn_index=turn_index, content="", error="completed seat run has no replayable snapshot; manual recovery required", metadata={"seat_run_id": seat_run.get("id"), "terminal": seat_run.get("status")}, latency_ms=0, ) return SeatResponse( agent_id=agent_id, turn_index=turn_index, content=str(snapshot.get("content", "")), tool_calls=list(snapshot.get("tool_calls") or []), usage=snapshot.get("usage"), metadata=dict(snapshot.get("metadata") or {}), error=None, latency_ms=0, ) def _latest_ai_text(self, messages: list[dict[str, Any]]) -> str: for m in reversed(messages): if not isinstance(m, dict): continue if (m.get("type") or m.get("role")) not in {"ai", "assistant"}: continue text = self._message_text(m) if text.strip(): return text return "" def _message_text(self, message: dict[str, Any]) -> str: content = message.get("content") if isinstance(content, str): return content if isinstance(content, list): parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"] return "\n".join(parts) return "" def _tool_calls_since_last_human(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]: last_human = -1 for i, m in enumerate(messages): if isinstance(m, dict) and (m.get("type") or m.get("role")) in {"human", "user"}: last_human = i out: list[dict[str, Any]] = [] for m in messages[last_human + 1:]: if not isinstance(m, dict): continue if (m.get("type") or m.get("role")) not in {"ai", "assistant"}: continue for tc in m.get("tool_calls") or []: if not isinstance(tc, dict): continue name = tc.get("name") or "" if not name: continue out.append({"name": name, "args": tc.get("args") or {}, "id": tc.get("id") or ""}) return out def _extract_usage(self, messages: list[dict[str, Any]]) -> dict[str, Any] | None: for m in reversed(messages): if not isinstance(m, dict): continue usage = m.get("usage_metadata") or m.get("usage") if isinstance(usage, dict) and usage: return usage return None