794 lines
31 KiB
Python
794 lines
31 KiB
Python
"""Conversation service for the MAF roundtable V2 (App layer).
|
|
|
|
Orchestrates the harness-layer store with business rules that belong to the
|
|
application layer: session blueprint construction, ownership enforcement,
|
|
public DTO projection, and fail-closed behaviour when configuration is missing.
|
|
|
|
Blueprint generation lives here (not in the harness) because it depends on
|
|
deployment config (``MafRoundtableConfig``) which is an app-layer concern.
|
|
The harness ``MafRoundtableStore`` never imports ``app.*``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from app.maf_roundtable.dispatch_contracts import (
|
|
DispatchConfigurationError,
|
|
MafDispatchCandidateSet,
|
|
MafDispatchCatalog,
|
|
)
|
|
from deerflow.persistence.maf_roundtable.conversations import (
|
|
conversation_public_to_dict,
|
|
)
|
|
from deerflow.persistence.maf_roundtable.messages import message_public_to_dict
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableMessageRow,
|
|
MafRoundtableRunRow,
|
|
)
|
|
from deerflow.persistence.maf_roundtable.store import MafRoundtableStore
|
|
|
|
if TYPE_CHECKING:
|
|
from app.maf_roundtable.conversation_event_hub import ConversationEventHub
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConversationServiceError(Exception):
|
|
def __init__(self, code: str, message: str, status_code: int = 400):
|
|
self.code = code
|
|
self.message = message
|
|
self.status_code = status_code
|
|
super().__init__(message)
|
|
|
|
|
|
class ConversationNotFoundError(ConversationServiceError):
|
|
def __init__(self) -> None:
|
|
super().__init__("not_found", "Conversation not found", 404)
|
|
|
|
|
|
class ConversationConflictError(ConversationServiceError):
|
|
def __init__(self) -> None:
|
|
super().__init__("conflict", "A run is already active for this conversation", 409)
|
|
|
|
|
|
class ConversationNoSessionConfigError(ConversationServiceError):
|
|
def __init__(self) -> None:
|
|
super().__init__(
|
|
"no_session_config",
|
|
"Cannot start a run: no trusted session configuration available",
|
|
503,
|
|
)
|
|
|
|
|
|
class ConversationDisabledError(ConversationServiceError):
|
|
def __init__(self) -> None:
|
|
super().__init__("disabled", "MAF roundtable feature is disabled", 503)
|
|
|
|
|
|
class ConversationDispatchError(ConversationServiceError):
|
|
def __init__(self, message: str = "MAF dispatch is not configured") -> None:
|
|
super().__init__("dispatch_unavailable", message, 503)
|
|
|
|
|
|
class ConversationRuntimeUnavailableError(ConversationServiceError):
|
|
def __init__(self) -> None:
|
|
super().__init__(
|
|
"runtime_unavailable",
|
|
"MAF workflow runtime is not available",
|
|
503,
|
|
)
|
|
|
|
|
|
class MafConversationService:
|
|
"""App-layer service for user-visible MAF conversations."""
|
|
|
|
def __init__(
|
|
self,
|
|
store: MafRoundtableStore,
|
|
*,
|
|
maf_config: Any = None,
|
|
event_hub: ConversationEventHub | None = None,
|
|
agent_store: Any | None = None,
|
|
app_config: Any | None = None,
|
|
dispatcher: Any | None = None,
|
|
enabled: bool | None = None,
|
|
) -> None:
|
|
self._store = store
|
|
self._maf_config = maf_config
|
|
self._hub = event_hub
|
|
self._agent_store = agent_store
|
|
self._app_config = app_config
|
|
self._dispatcher = dispatcher
|
|
self._enabled_override = enabled
|
|
self._watchers: dict[str, asyncio.Task[None]] = {}
|
|
|
|
@property
|
|
def _enabled(self) -> bool:
|
|
"""Mutating conversation APIs obey the same feature flag as MAF runs."""
|
|
if self._enabled_override is not None:
|
|
if not self._enabled_override:
|
|
return False
|
|
elif not bool(getattr(self._maf_config, "enabled", False)):
|
|
return False
|
|
return self.conversation_dispatch_ready
|
|
|
|
@property
|
|
def conversation_dispatch_ready(self) -> bool:
|
|
"""Expose only non-sensitive route readiness to the gateway/UI."""
|
|
if self._maf_config is None:
|
|
return False
|
|
return MafDispatchCatalog(self._maf_config).conversation_dispatch_ready
|
|
|
|
@property
|
|
def conversation_dispatch_status(self) -> str:
|
|
if self._maf_config is None:
|
|
return "configuration_unavailable"
|
|
return MafDispatchCatalog(self._maf_config).readiness()[1]
|
|
|
|
def _read_scope_user(self, *, user_id: str, is_admin: bool) -> str | None:
|
|
"""Keep administrator diagnostics read-only and config-gated."""
|
|
if is_admin and bool(getattr(self._maf_config, "admin_diagnostic_access", False)):
|
|
return None
|
|
return user_id
|
|
|
|
async def _build_dispatch_blueprint(
|
|
self,
|
|
*,
|
|
user_id: str,
|
|
title: str,
|
|
content: str,
|
|
task_context: dict[str, Any] | None,
|
|
) -> tuple[dict[str, Any] | None, MafDispatchCandidateSet]:
|
|
cfg = self._maf_config
|
|
if cfg is None:
|
|
raise ConversationDispatchError("MAF session configuration is unavailable")
|
|
|
|
try:
|
|
candidate_set = MafDispatchCatalog(cfg).build_candidate_set(
|
|
task_context=task_context,
|
|
)
|
|
except DispatchConfigurationError as exc:
|
|
raise ConversationDispatchError(str(exc)) from exc
|
|
|
|
snapshot_inputs = MafDispatchCatalog.candidate_participants(
|
|
candidate_set,
|
|
include_seats=True,
|
|
)
|
|
participants = MafDispatchCatalog.candidate_participants(
|
|
candidate_set,
|
|
include_seats=False,
|
|
)
|
|
capability_snapshot: dict[str, Any] | None = None
|
|
|
|
from app.maf_roundtable.capability_snapshot import (
|
|
AgentNotAccessibleError,
|
|
InvalidExcludedToolsError,
|
|
build_trusted_capability_snapshot,
|
|
seat_snapshot_for_agent,
|
|
)
|
|
|
|
if self._app_config is not None:
|
|
try:
|
|
capability_snapshot = await build_trusted_capability_snapshot(
|
|
user_id=user_id,
|
|
participants=snapshot_inputs,
|
|
app_config=self._app_config,
|
|
agent_store=self._agent_store,
|
|
session_model_name=getattr(cfg, "default_model_name", None),
|
|
session_reasoning_effort=getattr(cfg, "default_reasoning_effort", None),
|
|
)
|
|
except AgentNotAccessibleError as exc:
|
|
raise ConversationDispatchError(
|
|
f"configured MAF agent is not accessible: {exc.agent_id}"
|
|
) from exc
|
|
except InvalidExcludedToolsError as exc:
|
|
raise ConversationDispatchError(
|
|
f"configured MAF agent has invalid excluded tools: {exc.agent_id}"
|
|
) from exc
|
|
|
|
for participant in participants:
|
|
participant["capability_snapshot"] = seat_snapshot_for_agent(
|
|
capability_snapshot, str(participant["agent_id"])
|
|
)
|
|
|
|
payload = {
|
|
"maf_v2": {
|
|
"dispatch_contract_enabled": True,
|
|
"candidate_set": candidate_set.model_dump(mode="json"),
|
|
"candidate_seat_agent_ids": list(candidate_set.candidate_seat_agent_ids),
|
|
"manager_agent_id": candidate_set.manager_agent_id,
|
|
}
|
|
}
|
|
blueprint = {
|
|
"title": title,
|
|
"task_text": content,
|
|
"mode": candidate_set.mode,
|
|
"model_name": getattr(cfg, "default_model_name", None),
|
|
"reasoning_effort": getattr(cfg, "default_reasoning_effort", None),
|
|
"manager_agent_id": candidate_set.manager_agent_id,
|
|
"max_rounds": candidate_set.max_rounds,
|
|
"write_policy": getattr(cfg, "default_write_policy", "confirm_high_risk"),
|
|
"capability_snapshot": capability_snapshot,
|
|
"task_payload": payload,
|
|
"participants": participants,
|
|
"start_run": True,
|
|
}
|
|
return blueprint, candidate_set
|
|
|
|
async def create_conversation(
|
|
self, *, user_id: str, title: str = "", task_context: dict[str, Any] | None = None
|
|
) -> dict[str, Any]:
|
|
if not self._enabled:
|
|
raise ConversationDisabledError()
|
|
row_data = await self._store.conversations.create_conversation({
|
|
"owner_user_id": user_id,
|
|
"title": title,
|
|
"task_context": task_context,
|
|
})
|
|
conv_row = await self._get_conv_row(row_data["id"])
|
|
if conv_row is None:
|
|
raise ConversationNotFoundError()
|
|
return conversation_public_to_dict(conv_row)
|
|
|
|
async def list_conversations(
|
|
self, *, user_id: str, limit: int = 50, offset: int = 0
|
|
) -> tuple[list[dict[str, Any]], int]:
|
|
rows = await self._store.conversations.list_conversations(
|
|
user_id=user_id, limit=limit, offset=offset
|
|
)
|
|
total = await self._store.conversations.count_conversations(user_id=user_id)
|
|
public = []
|
|
for r in rows:
|
|
conv_row = await self._get_conv_row(r["id"])
|
|
if conv_row is not None:
|
|
public.append(conversation_public_to_dict(conv_row))
|
|
return public, total
|
|
|
|
async def get_conversation(self, *, conversation_id: str, user_id: str, is_admin: bool = False) -> dict[str, Any]:
|
|
effective_user = self._read_scope_user(user_id=user_id, is_admin=is_admin)
|
|
row = await self._store.conversations.get_conversation(conversation_id, user_id=effective_user)
|
|
if row is None:
|
|
raise ConversationNotFoundError()
|
|
conv_row = await self._get_conv_row(conversation_id)
|
|
if conv_row is None:
|
|
raise ConversationNotFoundError()
|
|
return conversation_public_to_dict(conv_row)
|
|
|
|
async def list_messages(
|
|
self, *, conversation_id: str, user_id: str, is_admin: bool = False,
|
|
before_seq: int | None = None, limit: int = 50,
|
|
) -> tuple[list[dict[str, Any]], int]:
|
|
effective_user = self._read_scope_user(user_id=user_id, is_admin=is_admin)
|
|
conv = await self._store.conversations.get_conversation(conversation_id, user_id=effective_user)
|
|
if conv is None:
|
|
raise ConversationNotFoundError()
|
|
|
|
rows = await self._store.messages.list_messages_desc(
|
|
conversation_id, limit=limit, before_seq=before_seq
|
|
)
|
|
public = []
|
|
for r in rows:
|
|
msg_row = await self._get_msg_row(r["id"])
|
|
if msg_row is not None:
|
|
public.append(message_public_to_dict(msg_row))
|
|
return public, int(conv.get("last_stream_seq", 0))
|
|
|
|
async def send_message(
|
|
self, *, conversation_id: str, user_id: str, content: str
|
|
) -> dict[str, Any]:
|
|
if not self._enabled:
|
|
raise ConversationDisabledError()
|
|
conv = await self._store.conversations.get_conversation(conversation_id, user_id=user_id)
|
|
if conv is None:
|
|
raise ConversationNotFoundError()
|
|
|
|
try:
|
|
blueprint, _candidate_set = await self._build_dispatch_blueprint(
|
|
user_id=user_id,
|
|
title=conv.get("title", ""),
|
|
content=content,
|
|
task_context=conv.get("task_context"),
|
|
)
|
|
except ConversationServiceError:
|
|
raise
|
|
|
|
if blueprint is None:
|
|
raise ConversationNoSessionConfigError()
|
|
|
|
# Do not create a durable running record if the MAF SDK is missing.
|
|
# Unit tests may inject a small fake dispatcher without this optional
|
|
# readiness method.
|
|
if self._dispatcher is not None:
|
|
check_sdk = getattr(self._dispatcher, "check_sdk_available", None)
|
|
if callable(check_sdk):
|
|
try:
|
|
check_sdk()
|
|
except Exception as exc: # noqa: BLE001 - safe public error
|
|
logger.warning("MAF runtime unavailable before dispatch: %s", type(exc).__name__)
|
|
raise ConversationRuntimeUnavailableError() from exc
|
|
|
|
result = await self._store.send_message_and_start_run(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=user_id,
|
|
content=content,
|
|
session_blueprint=blueprint,
|
|
)
|
|
|
|
if result is None:
|
|
raise ConversationNotFoundError()
|
|
if result.get("conflict"):
|
|
raise ConversationConflictError()
|
|
if result.get("no_session_config"):
|
|
raise ConversationNoSessionConfigError()
|
|
|
|
if self._hub is not None:
|
|
try:
|
|
await self._hub.publish(conversation_id)
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("hub publish failed conv=%s", conversation_id)
|
|
|
|
if result.get("run") is None:
|
|
raise ConversationDispatchError("dispatch run was not created")
|
|
await self._dispatch_run(
|
|
conversation_id=conversation_id,
|
|
user_id=user_id,
|
|
run=result["run"],
|
|
)
|
|
|
|
return {
|
|
"message_id": result["message"]["id"] if result.get("message") else "",
|
|
"conversation_id": conversation_id,
|
|
"seq": result.get("seq", 0),
|
|
}
|
|
|
|
async def _publish_conversation_event(
|
|
self,
|
|
*,
|
|
conversation_id: str,
|
|
owner_user_id: str,
|
|
event_type: str,
|
|
event_data: dict[str, Any],
|
|
source_event_id: str | None = None,
|
|
) -> bool:
|
|
"""Persist first, then notify; no internal run/thread ids are used.
|
|
|
|
P0-4: when ``source_event_id`` is provided, dedup is enforced — if an
|
|
event with the same ``(conversation_id, source_event_id)`` already
|
|
exists, the method returns ``False`` (no duplicate written). This
|
|
makes projection idempotent across retries and restarts.
|
|
|
|
Returns ``True`` if a new event was written, ``False`` if deduped.
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
async with self._store.session_factory() as db:
|
|
async with db.begin():
|
|
conv = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow)
|
|
.where(MafRoundtableConversationRow.id == conversation_id)
|
|
.with_for_update()
|
|
)
|
|
).scalars().first()
|
|
if conv is None or conv.owner_user_id != owner_user_id:
|
|
return False
|
|
|
|
# P0-4: dedup check — skip if already projected.
|
|
if source_event_id is not None:
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationEventRow,
|
|
)
|
|
|
|
existing = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationEventRow.id).where(
|
|
MafRoundtableConversationEventRow.conversation_id == conversation_id,
|
|
MafRoundtableConversationEventRow.source_event_id == source_event_id,
|
|
).limit(1)
|
|
)
|
|
).first()
|
|
if existing is not None:
|
|
return False
|
|
|
|
await self._store.append_conversation_item(
|
|
db,
|
|
conversation=conv,
|
|
event_type=event_type,
|
|
event_data=event_data,
|
|
run_id=None,
|
|
source_event_id=source_event_id,
|
|
)
|
|
if self._hub is not None:
|
|
try:
|
|
await self._hub.publish(conversation_id)
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("hub publish failed conv=%s", conversation_id)
|
|
return True
|
|
|
|
_EVENT_TYPE_MAP = {
|
|
"agent.turn_started": "agent.message.start",
|
|
"agent.message.completed": "agent.message.complete",
|
|
"action.waiting_approval": "approval.required",
|
|
"action.approved": "approval.decided",
|
|
"action.rejected": "approval.decided",
|
|
"run.started": "run.status",
|
|
"run.failed": "run.error",
|
|
"run.cancelled": "run.cancelled",
|
|
"manager.decision": "manager.dispatch",
|
|
"manager.dispatch": "manager.dispatch",
|
|
"clarification.required": "clarification.required",
|
|
"workflow.checkpointed": "run.status",
|
|
"action.waiting_target_lease": "tool.progress",
|
|
"action.unknown": "tool.failed",
|
|
}
|
|
|
|
_ALLOWED_TYPES = frozenset({
|
|
"agent.message.start", "agent.message.delta", "agent.message.complete",
|
|
"seat.delivery.submitted", "manager.dispatch", "manager.review.decision",
|
|
"tool.requested", "tool.started", "tool.progress", "tool.completed", "tool.failed",
|
|
"skill.selected", "skill.started", "skill.completed", "skill.failed",
|
|
"artifact.created", "approval.required", "approval.decided", "clarification.required",
|
|
"fact.confirmed", "review.completed", "run.status", "run.error",
|
|
"run.cancelled", "run.completed",
|
|
})
|
|
|
|
_SAFE_KEYS: dict[str, dict[str, bool]] = {
|
|
"agent.message.start": {"thread_id": False},
|
|
"agent.message.delta": {"text": True},
|
|
"agent.message.complete": {"content": True, "usage": True, "metadata": False},
|
|
"tool.requested": {"tool_name": True},
|
|
"tool.started": {"tool_name": True, "node": True},
|
|
"tool.progress": {"tool_name": True, "status": True},
|
|
"tool.completed": {"tool_name": True, "status": True, "content_preview": True},
|
|
"tool.failed": {"tool_name": True, "status": True, "error_code": True},
|
|
"skill.selected": {"skill_name": True, "message": True},
|
|
"skill.started": {"skill_name": True, "message": True},
|
|
"skill.completed": {"skill_name": True, "result_preview": True},
|
|
"skill.failed": {"skill_name": True, "error": True},
|
|
"artifact.created": {"artifact_type": True, "title": True},
|
|
"approval.required": {"tool_name": True, "summary": True},
|
|
"approval.decided": {"decided_by": True},
|
|
"run.status": {"checkpoint_id": True, "reason": True, "status": True},
|
|
"run.error": {"error_code": True, "message": True},
|
|
"run.cancelled": {},
|
|
"run.completed": {"status": True},
|
|
"manager.dispatch": {
|
|
"intent_type": True, "execution_mode": True, "convergence_path": True,
|
|
"selected_seat_agent_ids": True, "assignments": True, "next_speaker": True,
|
|
"reason": True, "source": True, "clarification_prompt": True,
|
|
},
|
|
"clarification.required": {"prompt": True, "reason": True},
|
|
"manager.review.decision": {"decision": True, "reason": True, "quality": True},
|
|
"fact.confirmed": {"fact_id": True, "summary": True},
|
|
"review.completed": {"summary": True, "decision": True},
|
|
"seat.delivery.submitted": {"content": True, "schema_status": True},
|
|
}
|
|
|
|
def _build_projection_items(
|
|
self,
|
|
event: dict[str, Any],
|
|
source_audit_event_id: str | None,
|
|
) -> list[dict[str, Any]]:
|
|
source_type = str(event.get("event_type") or "")
|
|
mapped_type = self._EVENT_TYPE_MAP.get(source_type, source_type)
|
|
if mapped_type not in self._ALLOWED_TYPES:
|
|
return []
|
|
|
|
payload = event.get("payload")
|
|
if not isinstance(payload, dict):
|
|
payload = {}
|
|
safe_keys = self._SAFE_KEYS.get(mapped_type, {})
|
|
data: dict[str, Any] = {
|
|
key: payload[key]
|
|
for key in safe_keys
|
|
if key in payload and safe_keys[key]
|
|
}
|
|
if event.get("agent_id"):
|
|
data["agent_id"] = event["agent_id"]
|
|
if event.get("turn_index") is not None:
|
|
data["turn_index"] = event["turn_index"]
|
|
|
|
source_evt_id = source_audit_event_id or event.get("id") or event.get("source_event_id")
|
|
|
|
items: list[dict[str, Any]] = [{
|
|
"event_type": mapped_type,
|
|
"event_data": data,
|
|
"source_event_id": source_evt_id,
|
|
}]
|
|
|
|
if source_type == "agent.message.completed":
|
|
delivery_data: dict[str, Any] = {
|
|
key: payload[key]
|
|
for key in ("content", "schema_status")
|
|
if key in payload
|
|
}
|
|
if event.get("agent_id"):
|
|
delivery_data["agent_id"] = event["agent_id"]
|
|
items.append({
|
|
"event_type": "seat.delivery.submitted",
|
|
"event_data": delivery_data,
|
|
"source_event_id": f"{source_evt_id}:delivery" if source_evt_id else None,
|
|
})
|
|
|
|
return items
|
|
|
|
async def _resolve_conversation_for_run(self, run_id: str) -> tuple[str, str] | None:
|
|
from sqlalchemy import select
|
|
async with self._store.session_factory() as db:
|
|
row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow)
|
|
.join(
|
|
MafRoundtableRunRow,
|
|
MafRoundtableConversationRow.internal_session_id
|
|
== MafRoundtableRunRow.session_id,
|
|
)
|
|
.where(MafRoundtableRunRow.id == run_id)
|
|
)
|
|
).scalars().first()
|
|
if row is None:
|
|
return None
|
|
return row.id, row.owner_user_id
|
|
|
|
async def bridge_workflow_event(
|
|
self,
|
|
*,
|
|
run_id: str,
|
|
event: dict[str, Any],
|
|
source_audit_event_id: str | None = None,
|
|
) -> bool:
|
|
"""Project a persisted per-run event onto the V2 conversation stream.
|
|
|
|
Returns ``True`` if at least one new conversation event was written,
|
|
``False`` if all items were deduped or unmapped.
|
|
|
|
All derived events from one audit event (e.g. ``agent.message.complete``
|
|
+ ``seat.delivery.submitted``) are written in a single transaction —
|
|
if any insert fails, the entire batch rolls back.
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
items = self._build_projection_items(event, source_audit_event_id)
|
|
if not items:
|
|
return False
|
|
|
|
resolved = await self._resolve_conversation_for_run(run_id)
|
|
if resolved is None:
|
|
return False
|
|
conversation_id, _owner_user_id = resolved
|
|
|
|
written = 0
|
|
async with self._store.session_factory() as db:
|
|
async with db.begin():
|
|
conv = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow)
|
|
.where(MafRoundtableConversationRow.id == conversation_id)
|
|
.with_for_update()
|
|
)
|
|
).scalars().first()
|
|
if conv is None:
|
|
return False
|
|
written = await self._store.append_conversation_events_idempotently(
|
|
db, conversation=conv, items=items,
|
|
)
|
|
if written > 0 and self._hub is not None:
|
|
try:
|
|
await self._hub.publish(conversation_id)
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("hub publish failed conv=%s", conversation_id)
|
|
return written > 0
|
|
|
|
async def reconcile_conversation_events(self, *, run_id: str) -> int:
|
|
"""Re-project audit events for a run into the conversation stream.
|
|
|
|
Scans ``maf_roundtable_events`` for the given run and re-bridges each
|
|
event. Because ``bridge_workflow_event`` is idempotent (dedup via
|
|
``source_event_id``), this is safe to call after a crash, a restart,
|
|
or a projection failure. Events already projected are silently skipped.
|
|
|
|
Returns the number of **new** conversation events written (deduped
|
|
events are NOT counted).
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableEventRow
|
|
|
|
written = 0
|
|
async with self._store.session_factory() as db:
|
|
rows = (
|
|
await db.execute(
|
|
select(MafRoundtableEventRow)
|
|
.where(MafRoundtableEventRow.run_id == run_id)
|
|
.order_by(MafRoundtableEventRow.seq.asc())
|
|
)
|
|
).scalars().all()
|
|
|
|
for row in rows:
|
|
event_dict: dict[str, Any] = {
|
|
"id": row.id,
|
|
"event_type": row.event_type,
|
|
"agent_id": row.agent_id,
|
|
"turn_index": row.turn_index,
|
|
"source_event_id": row.source_event_id,
|
|
"payload": row.payload_json,
|
|
}
|
|
try:
|
|
if await self.bridge_workflow_event(
|
|
run_id=run_id,
|
|
event=event_dict,
|
|
source_audit_event_id=row.id,
|
|
):
|
|
written += 1
|
|
except Exception: # noqa: BLE001
|
|
logger.warning(
|
|
"reconcile: failed to project audit event %s for run %s",
|
|
row.id, run_id, exc_info=True,
|
|
)
|
|
return written
|
|
|
|
async def reconcile_active_conversations(self) -> int:
|
|
"""Scan active/recent MAF runs and reconcile their conversation projections.
|
|
|
|
Called by the background reconcile task and the SSE connect trigger.
|
|
Returns total new conversation events written across all runs.
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableRunRow,
|
|
)
|
|
|
|
active_statuses = ("pending", "running", "waiting_approval", "waiting_input")
|
|
total = 0
|
|
async with self._store.session_factory() as db:
|
|
rows = (
|
|
await db.execute(
|
|
select(MafRoundtableRunRow.id)
|
|
.join(
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableConversationRow.active_run_id == MafRoundtableRunRow.id,
|
|
)
|
|
.where(MafRoundtableRunRow.status.in_(active_statuses))
|
|
)
|
|
).scalars().all()
|
|
|
|
for run_id in rows:
|
|
try:
|
|
total += await self.reconcile_conversation_events(run_id=str(run_id))
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("background reconcile: run %s failed", run_id, exc_info=True)
|
|
return total
|
|
|
|
async def _dispatch_run(self, *, conversation_id: str, user_id: str, run: dict[str, Any]) -> None:
|
|
if self._dispatcher is None:
|
|
# R1 unit callers can still exercise persistence without starting
|
|
# the optional SDK runtime. The production dependency wiring always
|
|
# supplies a dispatcher.
|
|
return
|
|
session = await self._store.sessions.get_session(run["session_id"], user_id=user_id)
|
|
if session is None:
|
|
raise ConversationDispatchError("internal MAF session could not be loaded")
|
|
try:
|
|
await self._dispatcher.dispatch(session, run, user_id=user_id)
|
|
except Exception as exc: # noqa: BLE001 - reconcile durable state
|
|
logger.exception("MAF dispatch failed run=%s", run["id"])
|
|
await self._mark_dispatch_failed(
|
|
conversation_id=conversation_id,
|
|
user_id=user_id,
|
|
run_id=run["id"],
|
|
error_code=type(exc).__name__[:64],
|
|
)
|
|
raise ConversationRuntimeUnavailableError() from exc
|
|
|
|
wait_for_run = getattr(self._dispatcher, "wait_for_run", None)
|
|
if callable(wait_for_run):
|
|
watcher = asyncio.create_task(
|
|
self._watch_run(
|
|
conversation_id=conversation_id,
|
|
user_id=user_id,
|
|
run_id=run["id"],
|
|
wait_for_run=wait_for_run,
|
|
)
|
|
)
|
|
self._watchers[run["id"]] = watcher
|
|
watcher.add_done_callback(lambda _task, rid=run["id"]: self._watchers.pop(rid, None))
|
|
|
|
async def _mark_dispatch_failed(
|
|
self, *, conversation_id: str, user_id: str, run_id: str, error_code: str
|
|
) -> None:
|
|
await self._store.runs.update_run(run_id, {
|
|
"status": "failed",
|
|
"finished_at": datetime.now(UTC),
|
|
"error_code": error_code,
|
|
"error_message_safe": "MAF workflow dispatch failed",
|
|
})
|
|
run = await self._store.runs.get_run(run_id)
|
|
if run is not None:
|
|
await self._store.sessions.update_session(
|
|
run["session_id"], user_id=user_id, fields={"status": "failed"}
|
|
)
|
|
await self._store.conversations.set_active_run(conversation_id, None, "failed")
|
|
await self._publish_conversation_event(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=user_id,
|
|
event_type="run.error",
|
|
event_data={"status": "failed", "error_code": error_code},
|
|
)
|
|
|
|
async def _watch_run(
|
|
self,
|
|
*,
|
|
conversation_id: str,
|
|
user_id: str,
|
|
run_id: str,
|
|
wait_for_run: Any,
|
|
) -> None:
|
|
try:
|
|
await wait_for_run(run_id)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("MAF conversation watcher failed run=%s", run_id)
|
|
return
|
|
|
|
try:
|
|
await self.reconcile_conversation_events(run_id=run_id)
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("watcher reconcile failed run=%s", run_id, exc_info=True)
|
|
|
|
run = await self._store.runs.get_run(run_id)
|
|
if run is None:
|
|
return
|
|
status_value = str(run.get("status") or "failed")
|
|
terminal = status_value in {"completed", "failed", "cancelled"}
|
|
await self._store.conversations.set_active_run(
|
|
conversation_id,
|
|
None if terminal else run_id,
|
|
status_value,
|
|
)
|
|
event_type = {
|
|
"completed": "run.completed",
|
|
"failed": "run.error",
|
|
"cancelled": "run.cancelled",
|
|
}.get(status_value, "run.status")
|
|
data: dict[str, Any] = {"status": status_value}
|
|
if run.get("error_code"):
|
|
data["error_code"] = run["error_code"]
|
|
await self._publish_conversation_event(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=user_id,
|
|
event_type=event_type,
|
|
event_data=data,
|
|
)
|
|
|
|
async def _get_conv_row(self, conversation_id: str) -> MafRoundtableConversationRow | None:
|
|
from sqlalchemy import select
|
|
async with self._store.session_factory() as session:
|
|
row = (
|
|
await session.execute(
|
|
select(MafRoundtableConversationRow).where(
|
|
MafRoundtableConversationRow.id == conversation_id
|
|
)
|
|
)
|
|
).scalars().first()
|
|
return row
|
|
|
|
async def _get_msg_row(self, message_id: str) -> MafRoundtableMessageRow | None:
|
|
from sqlalchemy import select
|
|
async with self._store.session_factory() as session:
|
|
row = (
|
|
await session.execute(
|
|
select(MafRoundtableMessageRow).where(
|
|
MafRoundtableMessageRow.id == message_id
|
|
)
|
|
)
|
|
).scalars().first()
|
|
return row
|