董宏凯andCursor
99642fca6c
Backend:
- Conversation API at /api/maf-roundtable/conversations: create/list/detail,
paginated messages, atomic send+run start, owner-only isolation,
admin read-only diagnostic access
- SSE real-time event stream (GET /conversations/{id}/stream) backed by
ConversationEventHub + conversation_events store; multi-worker deployment doc
- Deterministic server-owned stage-3 dispatch contracts (dispatch_contracts.py):
routing resolved from trusted MAF deployment config instead of manager-chosen
agent ids; config.yaml keeps stage-3 routing deployment-owned
(default_manager_agent_id / seat ids / business_routes / max_seats_per_run)
- Conversation persistence (conversations/messages stores) + v2 migration script
- Tests: conversations, conversation stream, migration, phase3 conversation dispatch
Frontend:
- Vitest test infra for MAF (vitest.maf.config.ts + test:maf script) with
component tests for ApprovalCard/RunHeader/SessionsPage and use-maf-events
- Session form lib, flow-graph dialog / run-detail / sessions page updates,
routes and nav tweaks
- Docs: R0 legacy interaction baseline mapping, updated implementation plan
Co-authored-by: Cursor <cursoragent@cursor.com>
671 lines
27 KiB
Python
671 lines
27 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,
|
|
MafDispatchCatalog,
|
|
MafDispatchPlan,
|
|
MafIntentDecision,
|
|
resolve_intent,
|
|
)
|
|
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:
|
|
return self._enabled_override
|
|
return bool(getattr(self._maf_config, "enabled", False))
|
|
|
|
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, MafIntentDecision, MafDispatchPlan | None]:
|
|
cfg = self._maf_config
|
|
if cfg is None:
|
|
raise ConversationDispatchError("MAF session configuration is unavailable")
|
|
|
|
intent = resolve_intent(content, task_context)
|
|
plan: MafDispatchPlan | None = None
|
|
participants: list[dict[str, Any]] = []
|
|
capability_snapshot: dict[str, Any] | None = None
|
|
|
|
if intent.intent_type != "clarification":
|
|
try:
|
|
plan = MafDispatchCatalog(cfg).build_plan(
|
|
intent=intent,
|
|
task_context=task_context,
|
|
)
|
|
except DispatchConfigurationError as exc:
|
|
raise ConversationDispatchError(str(exc)) from exc
|
|
|
|
from app.maf_roundtable.capability_snapshot import (
|
|
AgentNotAccessibleError,
|
|
InvalidExcludedToolsError,
|
|
build_trusted_capability_snapshot,
|
|
seat_snapshot_for_agent,
|
|
)
|
|
|
|
participants = MafDispatchCatalog.participants(plan)
|
|
if self._app_config is not None:
|
|
try:
|
|
capability_snapshot = await build_trusted_capability_snapshot(
|
|
user_id=user_id,
|
|
participants=participants,
|
|
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": {
|
|
"intent_type": intent.intent_type,
|
|
"reason_code": intent.reason_code,
|
|
"business_code": (task_context or {}).get("business_code"),
|
|
"dispatch_plan": plan.model_dump(mode="json") if plan is not None else None,
|
|
}
|
|
}
|
|
blueprint = {
|
|
"title": title,
|
|
"task_text": content,
|
|
"mode": getattr(plan, "mode", None) or getattr(cfg, "default_mode", "structured"),
|
|
"model_name": getattr(cfg, "default_model_name", None),
|
|
"reasoning_effort": getattr(cfg, "default_reasoning_effort", None),
|
|
"manager_agent_id": (
|
|
plan.manager_agent_id if plan is not None
|
|
else getattr(cfg, "default_manager_agent_id", None)
|
|
),
|
|
"max_rounds": getattr(plan, "max_rounds", None) or getattr(cfg, "default_max_rounds", 8),
|
|
"write_policy": getattr(cfg, "default_write_policy", "confirm_high_risk"),
|
|
"capability_snapshot": capability_snapshot,
|
|
"task_payload": payload,
|
|
"participants": participants,
|
|
"start_run": intent.intent_type != "clarification",
|
|
}
|
|
return blueprint, intent, plan
|
|
|
|
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, intent, plan = 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 intent.intent_type != "clarification" and 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()
|
|
|
|
await self._publish_conversation_event(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=user_id,
|
|
event_type="manager.intent",
|
|
event_data={
|
|
"intent_type": intent.intent_type,
|
|
"confidence": intent.confidence,
|
|
"reason_code": intent.reason_code,
|
|
},
|
|
)
|
|
|
|
if intent.intent_type == "clarification":
|
|
await self._publish_conversation_event(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=user_id,
|
|
event_type="clarification.required",
|
|
event_data={
|
|
"prompt": intent.clarification_prompt,
|
|
"reason_code": intent.reason_code,
|
|
},
|
|
)
|
|
else:
|
|
if plan is None or result.get("run") is None:
|
|
raise ConversationDispatchError("dispatch plan did not produce a run")
|
|
await self._publish_conversation_event(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=user_id,
|
|
event_type="manager.dispatch",
|
|
event_data={
|
|
"intent_type": plan.intent_type,
|
|
"mode": plan.mode,
|
|
"business_code": plan.business_code,
|
|
"manager_agent_id": plan.manager_agent_id,
|
|
"seat_agent_ids": list(plan.seat_agent_ids),
|
|
"seat_count": len(plan.seat_agent_ids),
|
|
},
|
|
)
|
|
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
|
|
|
|
async def bridge_workflow_event(self, *, run_id: str, event: dict[str, Any]) -> None:
|
|
"""Project a persisted per-run event onto the V2 conversation stream.
|
|
|
|
The per-run event ledger remains the audit source of truth. This
|
|
projection is deliberately lossy: only user-visible lifecycle fields
|
|
are copied, while tool arguments, tool-call ids, thread ids, run ids,
|
|
and raw exception bodies are discarded before the SSE event is saved.
|
|
"""
|
|
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
|
|
conversation_id = row.id
|
|
owner_user_id = row.owner_user_id
|
|
|
|
source_type = str(event.get("event_type") or "")
|
|
mapped_type = {
|
|
"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",
|
|
"workflow.checkpointed": "run.status",
|
|
"action.waiting_target_lease": "tool.progress",
|
|
"action.unknown": "tool.failed",
|
|
}.get(source_type, source_type)
|
|
|
|
allowed_types = {
|
|
"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",
|
|
"fact.confirmed", "review.completed", "run.status", "run.error",
|
|
"run.cancelled", "run.completed",
|
|
}
|
|
if mapped_type not in allowed_types:
|
|
return
|
|
|
|
payload = event.get("payload")
|
|
if not isinstance(payload, dict):
|
|
payload = {}
|
|
safe_keys = {
|
|
"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": {"next_speaker": 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},
|
|
}.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"]
|
|
|
|
# P0-4: use the source audit event's stable id for dedup.
|
|
source_evt_id = event.get("id") or event.get("source_event_id")
|
|
|
|
await self._publish_conversation_event(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=owner_user_id,
|
|
event_type=mapped_type,
|
|
event_data=data,
|
|
source_event_id=source_evt_id,
|
|
)
|
|
if source_type == "agent.message.completed":
|
|
# A completed structured seat turn is also a delivery milestone;
|
|
# this gives the timeline a stable gate separate from raw text.
|
|
delivery_data = {
|
|
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"]
|
|
await self._publish_conversation_event(
|
|
conversation_id=conversation_id,
|
|
owner_user_id=owner_user_id,
|
|
event_type="seat.delivery.submitted",
|
|
event_data=delivery_data,
|
|
source_event_id=f"{source_evt_id}:delivery" if source_evt_id else None,
|
|
)
|
|
|
|
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
|
|
|
|
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
|