478 lines
21 KiB
Python
478 lines
21 KiB
Python
"""HTTP router for the MAF roundtable experiment at ``/api/maf-roundtable``.
|
|
|
|
Phase 1 ships:
|
|
|
|
- ``GET /health`` — feature-flag probe (open read; the frontend reads
|
|
``enabled`` to decide whether to show the entry).
|
|
- Session CRUD: ``POST /sessions``, ``GET /sessions``, ``GET /sessions/{id}``,
|
|
``PUT /sessions/{id}``, ``DELETE /sessions/{id}``.
|
|
- Run lifecycle: ``POST /sessions/{id}/runs``, ``GET /runs/{id}``,
|
|
``GET /sessions/{id}/runs``, ``POST /runs/{id}/cancel``,
|
|
``POST /runs/{id}/resume``, ``POST /runs/{id}/recover``.
|
|
- Event stream: ``GET /runs/{id}/events?after_seq=N`` (断线补拉) — **read-only**;
|
|
there is no public ``POST /runs/{id}/events``. Real workflow events are
|
|
written only by the backend executor via ``service.append_event``.
|
|
- Actions: ``GET /runs/{id}/actions``, ``GET /actions/{id}``.
|
|
- Approvals: ``GET /runs/{id}/approvals``, ``POST /approvals/{id}/decision``.
|
|
|
|
User isolation (MAF-01-07): every read / write is scoped to the caller's
|
|
``user_id``; admins may *read* cross-user when ``admin_diagnostic_access`` is
|
|
on, but **writes** remain owner-only (admin write of another user's resource
|
|
→ 403). A non-admin user cannot read or mutate another user's session / run /
|
|
action / approval — the store returns ``None`` for out-of-scope rows and the
|
|
router surfaces that as ``404`` (not ``403``, so the existence of another
|
|
user's session is not leaked).
|
|
|
|
Feature flag (MAF-01-04): when ``maf_roundtable.enabled`` is off, write paths
|
|
return ``503 Service Disabled``; reads stay open so admins can review history.
|
|
The MAF SDK is **not** imported by this router — Phase 2+ imports it lazily
|
|
inside ``app.maf_roundtable.service`` only when a run is actually started.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, Request, status
|
|
|
|
from app.maf_roundtable import (
|
|
ActionListResponse,
|
|
ActionResponse,
|
|
ApprovalDecisionRequest,
|
|
ApprovalListResponse,
|
|
ApprovalResponse,
|
|
CreateRunRequest,
|
|
CreateSessionRequest,
|
|
EventListResponse,
|
|
EventResponse,
|
|
HealthResponse,
|
|
MafConflictError,
|
|
MafDisabledError,
|
|
MafForbiddenError,
|
|
MafNotFoundError,
|
|
RunListResponse,
|
|
RunResponse,
|
|
SessionListResponse,
|
|
SessionResponse,
|
|
UpdateSessionRequest,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/maf-roundtable", tags=["maf-roundtable"])
|
|
|
|
|
|
# ── helpers ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _get_service(request: Request):
|
|
"""Resolve the MAF service from app.state. Returns None when the feature
|
|
is fully off (no store wired) — the router then 503s on writes.
|
|
"""
|
|
return getattr(request.app.state, "maf_roundtable_service", None)
|
|
|
|
|
|
def _user_ctx(request: Request) -> tuple[str | None, bool]:
|
|
"""Return ``(user_id, is_admin)`` for the current request.
|
|
|
|
When auth is disabled (no user stamped on request.state), the caller is
|
|
treated as the ``"default"`` user (same convention as
|
|
``get_effective_user_id``) and non-admin — so a no-auth deployment stays
|
|
owner-scoped to ``"default"`` rather than accidentally bypassing the
|
|
owner filter.
|
|
"""
|
|
user = getattr(request.state, "user", None)
|
|
if user is None:
|
|
return "default", False
|
|
return str(user.id), getattr(user, "system_role", None) == "admin"
|
|
|
|
|
|
def _map_service_error(exc: Exception) -> HTTPException:
|
|
if isinstance(exc, MafDisabledError):
|
|
return HTTPException(status_code=503, detail="MAF roundtable feature is disabled")
|
|
if isinstance(exc, MafForbiddenError):
|
|
return HTTPException(status_code=403, detail=f"not the owner of {exc.what}")
|
|
if isinstance(exc, MafNotFoundError):
|
|
return HTTPException(status_code=404, detail=f"{exc.what} not found")
|
|
if isinstance(exc, MafConflictError):
|
|
return HTTPException(status_code=409, detail=str(exc))
|
|
return HTTPException(status_code=500, detail=str(exc))
|
|
|
|
|
|
def _require_service(request: Request):
|
|
svc = _get_service(request)
|
|
if svc is None:
|
|
raise HTTPException(status_code=503, detail="MAF roundtable store not available")
|
|
return svc
|
|
|
|
|
|
# ── health ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("/health", response_model=HealthResponse)
|
|
async def get_health(request: Request) -> HealthResponse:
|
|
"""Feature-flag probe. Open read — the frontend reads ``enabled`` to
|
|
decide whether to show the MAF entry. Does not require the store.
|
|
"""
|
|
from app.maf_roundtable.service import get_maf_write_guard_readiness
|
|
from deerflow.config import get_app_config
|
|
|
|
app_config = get_app_config()
|
|
cfg = app_config.maf_roundtable
|
|
svc = _get_service(request)
|
|
store_available = svc is not None and svc.store is not None
|
|
from app.maf_roundtable.dispatch_contracts import MafDispatchCatalog
|
|
|
|
dispatch_ready, dispatch_status = MafDispatchCatalog(cfg).readiness()
|
|
readiness = get_maf_write_guard_readiness(app_config=app_config)
|
|
stream_ready = bool(getattr(request.app.state, "maf_conversation_stream_ready", False))
|
|
stream_status = str(getattr(request.app.state, "maf_conversation_stream_status", "unknown"))
|
|
conv_backend = str(getattr(request.app.state, "maf_conversation_event_backend", "memory"))
|
|
return HealthResponse(
|
|
enabled=bool(cfg.enabled and dispatch_ready and svc is not None and svc.enabled),
|
|
conversation_dispatch_ready=bool(cfg.enabled and dispatch_ready and svc is not None and svc.enabled),
|
|
conversation_dispatch_status=dispatch_status,
|
|
conversation_stream_ready=stream_ready,
|
|
conversation_stream_status=stream_status,
|
|
conversation_event_backend=conv_backend,
|
|
store_available=store_available,
|
|
default_mode=cfg.default_mode,
|
|
default_write_policy=cfg.default_write_policy,
|
|
default_max_rounds=cfg.default_max_rounds,
|
|
admin_diagnostic_access=cfg.admin_diagnostic_access,
|
|
write_guard_enabled=readiness["enabled"],
|
|
write_guard_ready=readiness["ready"],
|
|
action_secret_configured=readiness["secret_configured"],
|
|
write_guard_status=readiness["status"],
|
|
sse_hub_type="in_memory",
|
|
sse_multi_worker=False,
|
|
sse_warning="Single-worker only. Multi-worker requires Redis Pub/Sub.",
|
|
)
|
|
|
|
|
|
# ── sessions ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.post("/sessions", response_model=SessionResponse)
|
|
async def create_session(request: Request, body: CreateSessionRequest) -> SessionResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
session = await svc.create_session(user_id=user_id, body=body.model_dump(), is_admin=is_admin)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return SessionResponse(**session)
|
|
|
|
|
|
@router.get("/sessions", response_model=SessionListResponse)
|
|
async def list_sessions(
|
|
request: Request,
|
|
status: str | None = Query(default=None),
|
|
mode: str | None = Query(default=None),
|
|
limit: int = Query(default=100, ge=1, le=500),
|
|
offset: int = Query(default=0, ge=0),
|
|
) -> SessionListResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
rows, total = await svc.list_sessions(
|
|
user_id=user_id, is_admin=is_admin, status=status, mode=mode, limit=limit, offset=offset
|
|
)
|
|
return SessionListResponse(sessions=[SessionResponse(**r) for r in rows], total=total)
|
|
|
|
|
|
@router.get("/sessions/{session_id}", response_model=SessionResponse)
|
|
async def get_session(request: Request, session_id: str) -> SessionResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
session = await svc.get_session(session_id=session_id, user_id=user_id, is_admin=is_admin)
|
|
except MafNotFoundError as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return SessionResponse(**session)
|
|
|
|
|
|
@router.put("/sessions/{session_id}", response_model=SessionResponse)
|
|
async def update_session(request: Request, session_id: str, body: UpdateSessionRequest) -> SessionResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
session = await svc.update_session(
|
|
session_id=session_id, user_id=user_id, fields=body.model_dump(exclude_unset=True), is_admin=is_admin
|
|
)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return SessionResponse(**session)
|
|
|
|
|
|
@router.delete("/sessions/{session_id}")
|
|
async def delete_session(request: Request, session_id: str) -> dict[str, Any]:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
await svc.delete_session(session_id=session_id, user_id=user_id, is_admin=is_admin)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return {"deleted": True, "id": session_id}
|
|
|
|
|
|
# ── runs ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.post("/sessions/{session_id}/runs", response_model=RunResponse)
|
|
async def create_run(request: Request, session_id: str, body: CreateRunRequest) -> RunResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
|
|
# Check MAF SDK availability BEFORE creating the run — so the user gets a
|
|
# clean 503 rather than a run that will immediately fail.
|
|
from app.maf_roundtable.dispatcher import MafSdkNotAvailableError, get_dispatcher
|
|
|
|
dispatcher = get_dispatcher(request.app)
|
|
try:
|
|
dispatcher.check_sdk_available()
|
|
except MafSdkNotAvailableError as exc:
|
|
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
|
|
|
try:
|
|
run = await svc.create_run(
|
|
session_id=session_id, user_id=user_id, workflow_version=body.workflow_version, is_admin=is_admin
|
|
)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
|
|
# Launch the workflow in the background. The run is already persisted as
|
|
# "pending"; the dispatcher transitions it to "running" then completes/fails it.
|
|
# Auto-dispatch can be disabled (e.g. in tests) via app.state.maf_auto_dispatch=False.
|
|
if getattr(request.app.state, "maf_auto_dispatch", True):
|
|
session = await svc.get_session(session_id=session_id, user_id=user_id, is_admin=is_admin)
|
|
await dispatcher.dispatch(session, run, user_id=user_id or "default")
|
|
|
|
return RunResponse(**run)
|
|
|
|
|
|
@router.get("/sessions/{session_id}/runs", response_model=RunListResponse)
|
|
async def list_runs(request: Request, session_id: str) -> RunListResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
runs = await svc.list_runs(session_id=session_id, user_id=user_id, is_admin=is_admin)
|
|
except MafNotFoundError as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return RunListResponse(runs=[RunResponse(**r) for r in runs])
|
|
|
|
|
|
@router.get("/runs/{run_id}", response_model=RunResponse)
|
|
async def get_run(request: Request, run_id: str) -> RunResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
run = await svc.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
except MafNotFoundError as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return RunResponse(**run)
|
|
|
|
|
|
@router.post("/runs/{run_id}/cancel", response_model=RunResponse)
|
|
async def cancel_run(request: Request, run_id: str) -> RunResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
|
|
# Phase 1: atomically set run+session to cancelling (not yet cancelled).
|
|
try:
|
|
await svc.start_cancel(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
|
|
# Phase 2: cascade cancel to the background task + DeerFlow sub-runs.
|
|
from app.maf_roundtable.dispatcher import get_dispatcher
|
|
|
|
dispatcher = get_dispatcher(request.app)
|
|
result = await dispatcher.cancel(run_id)
|
|
|
|
# Phase 3: finalize based on the structured cancel result.
|
|
# - all_cancelled → finalize to ``cancelled`` (terminal).
|
|
# - some sub-runs failed/still active → mark ``cancel_failed`` (non-terminal)
|
|
# so the failure is visible and retryable. A failed cancel must NEVER
|
|
# masquerade as cancelled.
|
|
try:
|
|
if result.get("all_cancelled"):
|
|
final_run = await svc.finalize_cancel(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
else:
|
|
failed = result.get("failed_run_ids") or []
|
|
still_active = result.get("still_active_run_ids") or []
|
|
detail = []
|
|
if failed:
|
|
detail.append(f"{len(failed)} sub-run(s) failed to cancel")
|
|
if still_active:
|
|
detail.append(f"{len(still_active)} sub-run(s) had no DeerFlow run id")
|
|
reason = "; ".join(detail) or "cancel cascade incomplete"
|
|
final_run = await svc.mark_cancel_failed(
|
|
run_id=run_id, user_id=user_id, is_admin=is_admin, reason=reason
|
|
)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
|
|
return RunResponse(**final_run)
|
|
|
|
|
|
@router.post("/runs/{run_id}/resume", response_model=RunResponse)
|
|
async def resume_run(request: Request, run_id: str) -> RunResponse:
|
|
"""Resume after approvals are decided (owner-only).
|
|
|
|
If the run is not ``waiting_approval`` or still has pending approvals,
|
|
returns 409.
|
|
"""
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
run = await svc.get_run(run_id=run_id, user_id=user_id, is_admin=False)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
|
|
if run.get("status") != "waiting_approval":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"run {run_id} is not waiting_approval (status={run.get('status')})",
|
|
)
|
|
|
|
from app.maf_roundtable.dispatcher import get_dispatcher
|
|
|
|
dispatcher = get_dispatcher(request.app)
|
|
try:
|
|
updated = await dispatcher.resume(run_id, user_id=user_id or "default")
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"resume failed: {type(exc).__name__}",
|
|
) from exc
|
|
if updated is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="run not found")
|
|
return RunResponse(**updated)
|
|
|
|
|
|
@router.post("/runs/{run_id}/recover", response_model=RunResponse)
|
|
async def recover_run(request: Request, run_id: str) -> RunResponse:
|
|
"""Graph-level recover after process restart (owner-only).
|
|
|
|
Re-enters the workflow from checkpoint; completed_nodes are skipped.
|
|
Does not silently scan all runs at startup.
|
|
"""
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
run = await svc.get_run(run_id=run_id, user_id=user_id, is_admin=False)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
|
|
if run.get("status") not in ("running", "waiting_approval"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"run {run_id} cannot be recovered (status={run.get('status')})",
|
|
)
|
|
|
|
from app.maf_roundtable.dispatcher import get_dispatcher
|
|
|
|
dispatcher = get_dispatcher(request.app)
|
|
try:
|
|
updated = await dispatcher.recover(run_id, user_id=user_id or "default")
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"recover failed: {type(exc).__name__}",
|
|
) from exc
|
|
if updated is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="run not found")
|
|
return RunResponse(**updated)
|
|
|
|
|
|
# ── events ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("/runs/{run_id}/events", response_model=EventListResponse)
|
|
async def list_events(
|
|
request: Request,
|
|
run_id: str,
|
|
after_seq: int = Query(default=0, ge=0),
|
|
limit: int = Query(default=1000, ge=1, le=10000),
|
|
) -> EventListResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
events, latest = await svc.list_events(
|
|
run_id=run_id, user_id=user_id, after_seq=after_seq, limit=limit, is_admin=is_admin
|
|
)
|
|
except MafNotFoundError as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return EventListResponse(events=[EventResponse(**e) for e in events], latest_seq=latest)
|
|
|
|
|
|
# ── actions ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("/runs/{run_id}/actions", response_model=ActionListResponse)
|
|
async def list_actions(request: Request, run_id: str) -> ActionListResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
actions = await svc.list_actions(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
except MafNotFoundError as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return ActionListResponse(actions=[ActionResponse(**a) for a in actions])
|
|
|
|
|
|
@router.get("/actions/{action_id}", response_model=ActionResponse)
|
|
async def get_action(request: Request, action_id: str) -> ActionResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
action = await svc.get_action(action_id=action_id, user_id=user_id, is_admin=is_admin)
|
|
except MafNotFoundError as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return ActionResponse(**action)
|
|
|
|
|
|
# ── approvals ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("/runs/{run_id}/approvals", response_model=ApprovalListResponse)
|
|
async def list_approvals(request: Request, run_id: str) -> ApprovalListResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
approvals = await svc.list_approvals(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
except MafNotFoundError as exc:
|
|
raise _map_service_error(exc) from exc
|
|
return ApprovalListResponse(approvals=[ApprovalResponse(**a) for a in approvals])
|
|
|
|
|
|
@router.post("/approvals/{approval_id}/decision", response_model=ApprovalResponse)
|
|
async def decide_approval(request: Request, approval_id: str, body: ApprovalDecisionRequest) -> ApprovalResponse:
|
|
svc = _require_service(request)
|
|
user_id, is_admin = _user_ctx(request)
|
|
try:
|
|
approval = await svc.decide_approval(
|
|
approval_id=approval_id, decision=body.decision, user_id=user_id, is_admin=is_admin
|
|
)
|
|
except (MafDisabledError, MafForbiddenError, MafNotFoundError, MafConflictError) as exc:
|
|
raise _map_service_error(exc) from exc
|
|
|
|
# P1: after a successful approval decision, resume the workflow from its
|
|
# checkpoint. Only the owner can resume (decide_approval already enforced
|
|
# owner-only write). The resume loads the checkpoint, restores
|
|
# RoundtableState.from_dict(), and executes the approved action (at-most-once).
|
|
run_id = approval.get("run_id")
|
|
if run_id and body.decision in ("approved", "rejected"):
|
|
from app.maf_roundtable.dispatcher import get_dispatcher
|
|
|
|
dispatcher = get_dispatcher(request.app)
|
|
try:
|
|
await dispatcher.resume(run_id, user_id=user_id or "default")
|
|
except Exception:
|
|
# Resume failure should not fail the approval response itself —
|
|
# the approval is already persisted. Log for observability.
|
|
import logging
|
|
|
|
logging.getLogger(__name__).exception(
|
|
"MAF resume failed for run %s after approval %s", run_id, approval_id
|
|
)
|
|
|
|
return ApprovalResponse(**approval)
|