392 lines
15 KiB
Python
392 lines
15 KiB
Python
"""HTTP router for MAF roundtable Conversation API at ``/api/maf-roundtable/conversations``.
|
|
|
|
Phase 1 (R1) scope:
|
|
- POST /conversations — create blank user conversation
|
|
- GET /conversations — list current user's history
|
|
- GET /conversations/{id} — public conversation DTO
|
|
- GET /conversations/{id}/messages — paginated message history
|
|
- POST /conversations/{id}/messages — atomic send + run start
|
|
|
|
Phase 2 (R2) adds:
|
|
- GET /conversations/{id}/stream — SSE real-time event stream
|
|
|
|
User isolation: owner-only reads/writes. Admin diagnostic access is read-only.
|
|
Internal fields (owner_user_id, internal_session_id, active_run_id, thread ids)
|
|
are never returned.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, Request, status
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
from app.maf_roundtable.conversation_event_hub import ConversationEventHub
|
|
from app.maf_roundtable.conversation_schemas import (
|
|
ConversationListResponse,
|
|
ConversationResponse,
|
|
CreateConversationRequest,
|
|
MessageListResponse,
|
|
SendMessageRequest,
|
|
SendMessageResponse,
|
|
)
|
|
from app.maf_roundtable.conversation_service import (
|
|
ConversationConflictError,
|
|
ConversationDisabledError,
|
|
ConversationNoSessionConfigError,
|
|
ConversationNotFoundError,
|
|
ConversationServiceError,
|
|
MafConversationService,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/maf-roundtable", tags=["maf-roundtable-conversations"])
|
|
|
|
_SSE_HEADERS = {
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
}
|
|
|
|
|
|
def _get_conversation_service(request: Request) -> MafConversationService:
|
|
svc = getattr(request.app.state, "maf_conversation_service", None)
|
|
if svc is None:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="MAF conversation service not available")
|
|
return svc
|
|
|
|
|
|
def _get_event_hub(request: Request) -> ConversationEventHub | None:
|
|
return getattr(request.app.state, "maf_conversation_event_hub", None)
|
|
|
|
|
|
def _user_ctx(request: Request) -> tuple[str, bool]:
|
|
"""Match the main MAF router's authenticated-user convention.
|
|
|
|
``AuthMiddleware`` stamps the resolved account on ``request.state.user``;
|
|
it does not create ``user_id`` / ``is_admin`` attributes. Reading those
|
|
nonexistent attributes collapses every authenticated caller into the
|
|
fallback account and breaks owner isolation.
|
|
"""
|
|
user = getattr(request.state, "user", None)
|
|
if user is None:
|
|
return "default", False
|
|
return str(user.id), getattr(user, "system_role", None) == "admin"
|
|
|
|
|
|
# ── REST endpoints (Phase 1) ──────────────────────────────────────────────
|
|
|
|
|
|
def _check_stream_ready(request: Request) -> None:
|
|
stream_ready = getattr(request.app.state, "maf_conversation_stream_ready", True)
|
|
if not stream_ready:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=getattr(request.app.state, "maf_conversation_stream_status", "stream_not_ready"),
|
|
)
|
|
|
|
|
|
@router.post("/conversations", response_model=ConversationResponse, status_code=status.HTTP_201_CREATED)
|
|
async def create_conversation(body: CreateConversationRequest, request: Request):
|
|
_check_stream_ready(request)
|
|
svc = _get_conversation_service(request)
|
|
user_id, _ = _user_ctx(request)
|
|
try:
|
|
return await svc.create_conversation(
|
|
user_id=user_id,
|
|
title=body.title.strip(),
|
|
task_context=body.task_context,
|
|
)
|
|
except ConversationDisabledError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="MAF roundtable feature is disabled",
|
|
)
|
|
|
|
|
|
@router.get("/conversations", response_model=ConversationListResponse)
|
|
async def list_conversations(
|
|
request: Request,
|
|
limit: int = Query(default=50, ge=1, le=200),
|
|
offset: int = Query(default=0, ge=0),
|
|
):
|
|
svc = _get_conversation_service(request)
|
|
user_id, _ = _user_ctx(request)
|
|
conversations, total = await svc.list_conversations(user_id=user_id, limit=limit, offset=offset)
|
|
return {"conversations": conversations, "total": total}
|
|
|
|
|
|
@router.get("/conversations/{conversation_id}", response_model=ConversationResponse)
|
|
async def get_conversation(conversation_id: str, request: Request):
|
|
svc = _get_conversation_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
return await svc.get_conversation(conversation_id=conversation_id, user_id=user_id, is_admin=is_admin)
|
|
except ConversationNotFoundError:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")
|
|
|
|
|
|
@router.get("/conversations/{conversation_id}/messages", response_model=MessageListResponse)
|
|
async def list_messages(
|
|
conversation_id: str,
|
|
request: Request,
|
|
before_seq: int | None = Query(default=None, ge=1),
|
|
limit: int = Query(default=50, ge=1, le=200),
|
|
):
|
|
svc = _get_conversation_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
messages, last_stream_seq = await svc.list_messages(
|
|
conversation_id=conversation_id, user_id=user_id, is_admin=is_admin,
|
|
before_seq=before_seq, limit=limit,
|
|
)
|
|
return {"messages": messages, "last_stream_seq": last_stream_seq}
|
|
except ConversationNotFoundError:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")
|
|
|
|
|
|
@router.post("/conversations/{conversation_id}/messages", response_model=SendMessageResponse, status_code=status.HTTP_201_CREATED)
|
|
async def send_message(conversation_id: str, body: SendMessageRequest, request: Request):
|
|
_check_stream_ready(request)
|
|
svc = _get_conversation_service(request)
|
|
user_id, _ = _user_ctx(request)
|
|
try:
|
|
return await svc.send_message(conversation_id=conversation_id, user_id=user_id, content=body.content)
|
|
except ConversationNotFoundError:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")
|
|
except ConversationConflictError:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="A run is already active")
|
|
except ConversationNoSessionConfigError:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="No trusted session configuration")
|
|
except ConversationDisabledError:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="MAF roundtable feature is disabled")
|
|
except ConversationServiceError as exc:
|
|
raise HTTPException(status_code=exc.status_code, detail=exc.message)
|
|
|
|
|
|
# ── SSE endpoint (Phase 2 / R2) ───────────────────────────────────────────
|
|
|
|
|
|
# Event types that mark the end of a run's lifecycle. When the SSE stream
|
|
# encounters any of these in order, it sends the event and then closes the
|
|
# connection so the client doesn't sit idle receiving heartbeats.
|
|
_TERMINAL_EVENT_TYPES = frozenset({
|
|
"run.completed",
|
|
"run.error",
|
|
"run.cancelled",
|
|
})
|
|
|
|
|
|
def _format_sse_event(event: dict[str, Any]) -> str:
|
|
"""Format one conversation event as an SSE frame per §4.3 contract.
|
|
|
|
``event:`` field carries the event type; ``data:`` carries the full
|
|
envelope; ``id:`` carries the event_id for Last-Event-ID reconnection.
|
|
"""
|
|
event_type = event.get("type", "message")
|
|
event_id = event.get("event_id")
|
|
payload = json.dumps(event, default=str, ensure_ascii=False)
|
|
parts = [f"event: {event_type}", f"data: {payload}"]
|
|
if event_id:
|
|
parts.append(f"id: {event_id}")
|
|
parts.append("")
|
|
parts.append("")
|
|
return "\n".join(parts)
|
|
|
|
|
|
def _format_heartbeat() -> str:
|
|
"""SSE comment frame — keeps the connection alive without producing a
|
|
client-visible event."""
|
|
return ": heartbeat\n\n"
|
|
|
|
|
|
def _is_terminal(event: dict[str, Any]) -> bool:
|
|
return event.get("type") in _TERMINAL_EVENT_TYPES
|
|
|
|
|
|
def build_conversation_event_generator(
|
|
*,
|
|
service: MafConversationService,
|
|
hub: ConversationEventHub,
|
|
request: Request,
|
|
conversation_id: str,
|
|
after_seq: int = 0,
|
|
) -> Any:
|
|
"""Extracted async generator for SSE streaming — testable without HTTP.
|
|
|
|
The ordering is:
|
|
|
|
1. **Reconcile** any missed projections for the active run (P0-1 auto).
|
|
2. **Open subscription** (register before any DB query — P0-1 fix).
|
|
3. **Replay** from DB: ``list_events(after_seq)``.
|
|
4. **Stabilise**: re-query while the hub generation advanced during replay.
|
|
5. **Snapshot** marker.
|
|
6. **Real-time**: ``wait_for_change`` → query DB → yield events.
|
|
|
|
On terminal events the stream sends the event and closes (P0-2 fix).
|
|
Heartbeat every ``hub.heartbeat_interval`` seconds.
|
|
"""
|
|
store = service._store # noqa: SLF001
|
|
|
|
async def _gen():
|
|
try:
|
|
active_run_id = await _get_active_run_id(store, conversation_id)
|
|
if active_run_id:
|
|
try:
|
|
await service.reconcile_conversation_events(run_id=active_run_id)
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("SSE connect reconcile failed conv=%s", conversation_id, exc_info=True)
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("SSE connect reconcile lookup failed conv=%s", conversation_id, exc_info=True)
|
|
|
|
sub = hub.open_subscription(conversation_id)
|
|
last_seq = after_seq
|
|
try:
|
|
try:
|
|
replay_events = await store.conversation_events.list_events(
|
|
conversation_id, after_seq=last_seq,
|
|
)
|
|
for evt in replay_events:
|
|
if await request.is_disconnected():
|
|
return
|
|
yield _format_sse_event(evt)
|
|
last_seq = max(last_seq, int(evt.get("seq", last_seq)))
|
|
if _is_terminal(evt):
|
|
return
|
|
except Exception:
|
|
logger.exception("SSE replay error conv=%s", conversation_id)
|
|
yield _format_sse_event({
|
|
"seq": last_seq,
|
|
"event_id": f"replay_error_{last_seq}",
|
|
"conversation_id": conversation_id,
|
|
"type": "run.error",
|
|
"occurred_at": datetime.now(UTC).isoformat(),
|
|
"data": {"error": "replay_failed"},
|
|
})
|
|
return
|
|
|
|
while sub.has_pending:
|
|
sub.mark_seen()
|
|
late_events = await store.conversation_events.list_events(
|
|
conversation_id, after_seq=last_seq,
|
|
)
|
|
if not late_events:
|
|
break
|
|
for evt in late_events:
|
|
if await request.is_disconnected():
|
|
return
|
|
yield _format_sse_event(evt)
|
|
last_seq = max(last_seq, int(evt.get("seq", last_seq)))
|
|
if _is_terminal(evt):
|
|
return
|
|
|
|
snapshot = {
|
|
"seq": last_seq,
|
|
"event_id": f"snapshot_{conversation_id}_{last_seq}",
|
|
"conversation_id": conversation_id,
|
|
"type": "conversation.snapshot",
|
|
"occurred_at": datetime.now(UTC).isoformat(),
|
|
"data": {"last_stream_seq": last_seq},
|
|
}
|
|
if not await request.is_disconnected():
|
|
yield _format_sse_event(snapshot)
|
|
|
|
heartbeat_interval = hub.heartbeat_interval
|
|
while True:
|
|
if hub.is_closed:
|
|
return
|
|
if await request.is_disconnected():
|
|
return
|
|
|
|
changed = await sub.wait_for_change(timeout=heartbeat_interval)
|
|
if not changed:
|
|
if hub.is_closed:
|
|
return
|
|
yield _format_heartbeat()
|
|
continue
|
|
|
|
new_events = await store.conversation_events.list_events(
|
|
conversation_id, after_seq=last_seq,
|
|
)
|
|
for evt in new_events:
|
|
if await request.is_disconnected():
|
|
return
|
|
yield _format_sse_event(evt)
|
|
last_seq = max(last_seq, int(evt.get("seq", last_seq)))
|
|
if _is_terminal(evt):
|
|
return
|
|
except asyncio.CancelledError:
|
|
logger.debug("SSE stream cancelled conv=%s", conversation_id)
|
|
raise
|
|
except Exception:
|
|
logger.exception("SSE stream error conv=%s", conversation_id)
|
|
return
|
|
finally:
|
|
sub.close()
|
|
|
|
return _gen()
|
|
|
|
|
|
async def _get_active_run_id(store: Any, conversation_id: str) -> str | None:
|
|
from sqlalchemy import select
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow
|
|
async with store.session_factory() as db:
|
|
row = (
|
|
await db.execute(
|
|
select(MafRoundtableConversationRow.active_run_id).where(
|
|
MafRoundtableConversationRow.id == conversation_id,
|
|
)
|
|
)
|
|
).first()
|
|
return row[0] if row and row[0] else None
|
|
|
|
|
|
@router.get("/conversations/{conversation_id}/stream")
|
|
async def stream_conversation(
|
|
conversation_id: str,
|
|
request: Request,
|
|
after_seq: int = Query(default=0, ge=0),
|
|
):
|
|
"""SSE stream for one conversation (§4.2)."""
|
|
svc = _get_conversation_service(request)
|
|
hub = _get_event_hub(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
|
|
if hub is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="SSE event hub not configured",
|
|
)
|
|
|
|
stream_ready = getattr(request.app.state, "maf_conversation_stream_ready", True)
|
|
if not stream_ready:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=getattr(request.app.state, "maf_conversation_stream_status", "stream_not_ready"),
|
|
)
|
|
|
|
try:
|
|
await svc.get_conversation(
|
|
conversation_id=conversation_id, user_id=user_id, is_admin=is_admin,
|
|
)
|
|
except ConversationNotFoundError:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")
|
|
|
|
return StreamingResponse(
|
|
build_conversation_event_generator(
|
|
service=svc,
|
|
hub=hub,
|
|
request=request,
|
|
conversation_id=conversation_id,
|
|
after_seq=after_seq,
|
|
),
|
|
media_type="text/event-stream",
|
|
headers=_SSE_HEADERS,
|
|
)
|