"""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