667 lines
27 KiB
Python
667 lines
27 KiB
Python
"""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
|