董宏凯andCursor
f82363d81f
- Backend: action_coordinator, capability_snapshot, event_normalizer, tool_executor updates - Backend: approved_tool_executor with write guard, action leases, migration script - Frontend: maf-roundtable module (API client, event timeline, approval cards, fact flow panel) - Frontend: MafRoutes, MafSessionsPage, MafRunDetailPage - Tests: action_leases, migration_phase5, phase5_entry_e2e, real_write_e2e/http_e2e - Docs: update implementation plan Co-authored-by: Cursor <cursoragent@cursor.com>
634 lines
28 KiB
Python
634 lines
28 KiB
Python
"""Service layer for the MAF roundtable experiment.
|
|
|
|
Enforces the cross-cutting policies the individual repositories don't:
|
|
|
|
- **User isolation (MAF-01-07)**: every read / write is scoped to the caller's
|
|
``user_id``. Read and write scopes are deliberately **separate**:
|
|
``admin_diagnostic_access`` only widens the *read* scope (cross-user
|
|
diagnostics) — it never grants cross-user *writes*. Every mutating operation
|
|
(update / delete session, start / cancel run, append event, decide approval)
|
|
is owner-checked via ``_write_scope_user`` regardless of the diagnostic flag.
|
|
- **Run gating + atomic launch**: a session may have at most one active run
|
|
(``starting`` / ``running`` / ``waiting_*``). The launch is a **single
|
|
database transaction** (``store.claim_and_create_pending_run``) that
|
|
conditionally updates the session to ``running`` and inserts a ``pending``
|
|
run together — not a multi-step claim/create/set_status dance, and not a
|
|
Python lock. Concurrent callers cannot both win; a mid-transaction failure
|
|
rolls everything back so the session never sticks in a half-claimed state.
|
|
- **Feature flag**: when ``maf_roundtable.enabled`` is off, write paths raise
|
|
``MafDisabledError`` (the router maps it to 503); reads stay open so admins
|
|
can review history.
|
|
- **Config-driven defaults**: ``mode`` / ``write_policy`` / ``max_rounds``
|
|
default to the deployed ``MafRoundtableConfig`` values when the create
|
|
request omits them; an explicit request value always wins.
|
|
|
|
The service does **not** load the MAF SDK — that is Phase 2+ (the
|
|
``DeerFlowAgentAdapter``). Phase 1 ships the persistence + API skeleton; until
|
|
a real executor exists, a launched run is parked in ``pending`` (clearly a
|
|
reserved / not-yet-executing state) rather than pretending to be ``running``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from app.maf_roundtable.sanitize import sanitize, sanitize_payload
|
|
from deerflow.config.maf_roundtable_config import MafRoundtableConfig
|
|
from deerflow.persistence.maf_roundtable.store import MafRoundtableStore
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MafDisabledError(RuntimeError):
|
|
"""Raised when a write is attempted while the feature flag is off."""
|
|
|
|
|
|
class MafNotFoundError(KeyError):
|
|
"""Raised when a session / run / action / approval is missing or out of scope."""
|
|
|
|
def __init__(self, what: str, ident: str) -> None:
|
|
super().__init__(f"{what} not found or out of scope: {ident}")
|
|
self.what = what
|
|
self.ident = ident
|
|
|
|
|
|
class MafConflictError(RuntimeError):
|
|
"""Raised when a session is in a state that rejects the requested edit."""
|
|
|
|
|
|
class MafForbiddenError(PermissionError):
|
|
"""Raised when a caller is authenticated but not the owner of the target
|
|
resource (e.g. an admin trying to *write* another user's session while
|
|
only holding cross-user *read* via the diagnostic flag)."""
|
|
|
|
def __init__(self, what: str, ident: str) -> None:
|
|
super().__init__(f"not the owner of {what}: {ident}")
|
|
self.what = what
|
|
self.ident = ident
|
|
|
|
|
|
class MafConfigurationError(RuntimeError):
|
|
"""MAF is enabled but a required safety prerequisite is missing.
|
|
|
|
Raised by :func:`validate_maf_prerequisites` on the startup path so the
|
|
feature can never come up in a half-safe state ("MAF on, approval guard not
|
|
mounted"). The gateway lifespan catches this and forces the feature OFF
|
|
(clearly logged) rather than aborting the whole (experimental, opt-in)
|
|
service.
|
|
"""
|
|
|
|
|
|
def get_maf_write_guard_readiness(*, app_config: Any | None = None) -> dict[str, Any]:
|
|
"""Return non-sensitive WriteGuard readiness information for MAF health.
|
|
|
|
A missing secret intentionally does not disable read-only MAF work:
|
|
WriteGuard fails closed before any external write can execute. ``ready``
|
|
is therefore the signal that the complete approval + encryption path is
|
|
usable, while ``enabled`` describes whether the middleware is mounted.
|
|
"""
|
|
import os
|
|
|
|
from deerflow.config.write_guard_config import get_write_guard_config
|
|
from deerflow.reflection import resolve_variable
|
|
|
|
wg = getattr(app_config, "write_guard", None) if app_config is not None else None
|
|
if wg is None:
|
|
wg = get_write_guard_config()
|
|
|
|
provider = getattr(wg, "provider", None)
|
|
provider_configured = bool(
|
|
provider
|
|
and getattr(provider, "policy_use", "")
|
|
and getattr(provider, "interceptor_use", "")
|
|
)
|
|
bridge_resolvable = False
|
|
if provider_configured:
|
|
try:
|
|
resolve_variable(provider.policy_use)
|
|
resolve_variable(provider.interceptor_use)
|
|
bridge_resolvable = True
|
|
except Exception: # noqa: BLE001 - health must remain safe and non-fatal
|
|
bridge_resolvable = False
|
|
|
|
secret_configured = bool((os.getenv("MAF_ACTION_SECRET_KEY") or "").strip())
|
|
if not getattr(wg, "enabled", False):
|
|
status = "disabled"
|
|
elif not provider_configured or not bridge_resolvable:
|
|
status = "provider_unavailable"
|
|
elif not secret_configured:
|
|
status = "missing_secret"
|
|
else:
|
|
status = "ready"
|
|
|
|
return {
|
|
"enabled": bool(getattr(wg, "enabled", False)),
|
|
"ready": status == "ready",
|
|
"secret_configured": secret_configured,
|
|
"status": status,
|
|
}
|
|
|
|
|
|
def validate_maf_prerequisites() -> None:
|
|
"""Fail loudly if MAF is enabled without its full safety stack in place.
|
|
|
|
Checks, in order:
|
|
1. ``write_guard.enabled`` is true (the approval guard is mounted);
|
|
2. ``write_guard.provider`` is complete (``policy_use`` + ``interceptor_use``);
|
|
3. the bridge factories actually resolve (the app-layer guard is importable);
|
|
4. secret readiness is reported separately; a missing secret must not
|
|
disable read-only MAF work, while writes fail closed before execution.
|
|
|
|
Raises ``MafConfigurationError`` on the first unmet condition. Callers on
|
|
the startup path treat that as "force the feature off"; direct unit-test
|
|
construction of :class:`MafRoundtableService` does NOT call this.
|
|
"""
|
|
from deerflow.config.write_guard_config import get_write_guard_config
|
|
from deerflow.reflection import resolve_variable
|
|
|
|
wg = get_write_guard_config()
|
|
if not wg.enabled:
|
|
raise MafConfigurationError(
|
|
"maf_roundtable.enabled=true requires write_guard.enabled=true "
|
|
"(the approval guard must be mounted)"
|
|
)
|
|
provider = wg.provider
|
|
if provider is None or not provider.policy_use or not provider.interceptor_use:
|
|
raise MafConfigurationError(
|
|
"maf_roundtable.enabled=true requires a complete write_guard.provider "
|
|
"(policy_use + interceptor_use)"
|
|
)
|
|
try:
|
|
resolve_variable(provider.policy_use)
|
|
resolve_variable(provider.interceptor_use)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise MafConfigurationError(
|
|
f"write_guard bridge factories are not resolvable: {exc}"
|
|
) from exc
|
|
|
|
|
|
# Session statuses that may be edited (config changes).
|
|
_EDITABLE_STATUSES = {"draft", "completed", "failed", "cancelled"}
|
|
# Session statuses that mean a run is in flight (no config edits). ``starting``
|
|
# is the transient CAS-claim state between "claimed" and "run row created".
|
|
# ``cancel_failed`` blocks edits because the cancel cascade is incomplete and
|
|
# the may still have active sub-runs.
|
|
_RUNNING_STATUSES = {"starting", "running", "waiting_approval", "waiting_input", "cancelling", "cancel_failed"}
|
|
|
|
|
|
class MafRoundtableService:
|
|
def __init__(
|
|
self,
|
|
store: MafRoundtableStore,
|
|
*,
|
|
enabled: bool,
|
|
admin_diagnostic_access: bool,
|
|
config: MafRoundtableConfig | None = None,
|
|
agent_store: Any | None = None,
|
|
app_config: Any | None = None,
|
|
) -> None:
|
|
self._store = store
|
|
self._admin_diagnostic_access = admin_diagnostic_access
|
|
self._agent_store = agent_store
|
|
self._app_config = app_config
|
|
# Defaults source for create_session; falls back to the pydantic
|
|
# defaults when no config is injected (keeps unit tests lightweight).
|
|
self._config = config or MafRoundtableConfig()
|
|
|
|
# A missing MAF_ACTION_SECRET_KEY does not disable read-only MAF
|
|
# operation. The action coordinator raises before creating or
|
|
# executing an external-write action, so writes fail closed until the
|
|
# deployment injects the secret through the environment.
|
|
self._enabled = enabled
|
|
|
|
@property
|
|
def store(self) -> MafRoundtableStore:
|
|
return self._store
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return self._enabled
|
|
|
|
@property
|
|
def config(self) -> MafRoundtableConfig:
|
|
return self._config
|
|
|
|
def _require_enabled(self) -> None:
|
|
if not self._enabled:
|
|
raise MafDisabledError("MAF roundtable feature is disabled")
|
|
|
|
# ── permission scopes ──────────────────────────────────────────────────
|
|
|
|
def _read_scope_user(self, user_id: str | None, is_admin: bool) -> str | None:
|
|
"""User id to scope **read** queries by.
|
|
|
|
- Admin with diagnostic access: ``None`` (cross-user read for diagnostics).
|
|
- Anyone else: their own ``user_id`` (enforced).
|
|
"""
|
|
if is_admin and self._admin_diagnostic_access:
|
|
return None
|
|
return user_id
|
|
|
|
def _write_scope_user(self, user_id: str | None, is_admin: bool) -> str | None:
|
|
"""User id to scope **write** operations by.
|
|
|
|
The diagnostic flag is read-only — it never widens the write scope.
|
|
Every mutating operation is owner-checked against the caller's own
|
|
``user_id`` (admin or not). A future explicit operator/admin force-stop
|
|
would be a separate, audited code path, not a reuse of this flag.
|
|
"""
|
|
return user_id
|
|
|
|
# ── sessions ──────────────────────────────────────────────────────────
|
|
|
|
async def create_session(
|
|
self, *, user_id: str, body: dict[str, Any], is_admin: bool = False
|
|
) -> dict[str, Any]:
|
|
self._require_enabled()
|
|
cfg = self._config
|
|
from app.maf_roundtable.capability_snapshot import (
|
|
AgentNotAccessibleError,
|
|
InvalidExcludedToolsError,
|
|
build_trusted_capability_snapshot,
|
|
seat_snapshot_for_agent,
|
|
)
|
|
from deerflow.config import get_app_config
|
|
|
|
app_config = self._app_config or get_app_config()
|
|
raw_participants = list(body.get("participants") or [])
|
|
# Strip any client-supplied snapshot / tool metadata — never trusted.
|
|
participants: list[dict[str, Any]] = []
|
|
for p in raw_participants:
|
|
participants.append(
|
|
{
|
|
"agent_id": p.get("agent_id"),
|
|
"display_name": p.get("display_name", ""),
|
|
"role": p.get("role", "seat"),
|
|
"sort_order": p.get("sort_order", 0),
|
|
"thread_id": p.get("thread_id"),
|
|
"excluded_tools": p.get("excluded_tools") or [],
|
|
}
|
|
)
|
|
|
|
try:
|
|
trusted_snapshot = await build_trusted_capability_snapshot(
|
|
user_id=user_id,
|
|
participants=participants,
|
|
app_config=app_config,
|
|
agent_store=self._agent_store,
|
|
session_model_name=body.get("model_name"),
|
|
session_reasoning_effort=body.get("reasoning_effort"),
|
|
session_excluded_tools=body.get("excluded_tools"),
|
|
)
|
|
except AgentNotAccessibleError as exc:
|
|
raise MafForbiddenError("agent", exc.agent_id) from exc
|
|
except InvalidExcludedToolsError as exc:
|
|
raise MafConflictError(
|
|
f"excluded_tools not allowed for agent {exc.agent_id}: {exc.tools}"
|
|
) from exc
|
|
|
|
for p in participants:
|
|
aid = p.get("agent_id")
|
|
if aid:
|
|
p["capability_snapshot"] = seat_snapshot_for_agent(trusted_snapshot, str(aid))
|
|
|
|
session_data: dict[str, Any] = {
|
|
"user_id": user_id,
|
|
"title": body.get("title", ""),
|
|
"task_text": body["task_text"],
|
|
"task_payload": body.get("task_payload"),
|
|
"mode": body.get("mode") or cfg.default_mode,
|
|
"status": "draft",
|
|
"model_name": body.get("model_name"),
|
|
"reasoning_effort": body.get("reasoning_effort"),
|
|
"manager_agent_id": body.get("manager_agent_id"),
|
|
"max_rounds": body.get("max_rounds") if body.get("max_rounds") is not None else cfg.default_max_rounds,
|
|
"write_policy": body.get("write_policy") or cfg.default_write_policy,
|
|
"capability_snapshot": trusted_snapshot,
|
|
}
|
|
session = await self._store.sessions.create_session(session_data)
|
|
if participants:
|
|
await self._store.participants.replace_participants(session["id"], participants)
|
|
return await self.get_session(session_id=session["id"], user_id=user_id, is_admin=is_admin)
|
|
|
|
async def get_session(self, *, session_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
|
|
scope = self._read_scope_user(user_id, is_admin)
|
|
session = await self._store.sessions.get_session(session_id, user_id=scope)
|
|
if session is None:
|
|
raise MafNotFoundError("session", session_id)
|
|
participants = await self._store.participants.list_participants(session_id)
|
|
session["participants"] = participants
|
|
return session
|
|
|
|
async def list_sessions(
|
|
self,
|
|
*,
|
|
user_id: str | None,
|
|
is_admin: bool = False,
|
|
status: str | None = None,
|
|
mode: str | None = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
) -> tuple[list[dict[str, Any]], int]:
|
|
scope = self._read_scope_user(user_id, is_admin)
|
|
rows = await self._store.sessions.list_sessions(
|
|
user_id=scope, status=status, mode=mode, limit=limit, offset=offset
|
|
)
|
|
total = await self._store.sessions.count_sessions(user_id=scope, status=status, mode=mode)
|
|
# Attach participants to each row (cheap — Phase 1 lists are bounded).
|
|
for row in rows:
|
|
row["participants"] = await self._store.participants.list_participants(row["id"])
|
|
return rows, total
|
|
|
|
async def update_session(
|
|
self, *, session_id: str, user_id: str | None, fields: dict[str, Any], is_admin: bool = False
|
|
) -> dict[str, Any]:
|
|
self._require_enabled()
|
|
# Writes are owner-scoped even for admins (diagnostic access is read-only).
|
|
scope = self._write_scope_user(user_id, is_admin)
|
|
existing = await self._store.sessions.get_session(session_id, user_id=scope)
|
|
if existing is None:
|
|
# Distinguish "missing" from "owned by someone else" so an admin
|
|
# (who can read cross-user) gets a clear 403 rather than a 404.
|
|
if is_admin and await self._store.sessions.get_session(session_id, user_id=None) is not None:
|
|
raise MafForbiddenError("session", session_id)
|
|
raise MafNotFoundError("session", session_id)
|
|
if existing["status"] in _RUNNING_STATUSES:
|
|
raise MafConflictError(
|
|
f"session {session_id} is running ({existing['status']}); config edits not allowed"
|
|
)
|
|
# capability_snapshot is frozen — never accept it on update.
|
|
fields.pop("capability_snapshot", None)
|
|
updated = await self._store.sessions.update_session(session_id, user_id=scope, fields=fields)
|
|
if updated is None:
|
|
raise MafNotFoundError("session", session_id)
|
|
if "participants" in fields and fields["participants"] is not None:
|
|
await self._store.participants.replace_participants(session_id, fields["participants"])
|
|
return await self.get_session(session_id=session_id, user_id=user_id, is_admin=is_admin)
|
|
|
|
async def delete_session(self, *, session_id: str, user_id: str | None, is_admin: bool = False) -> None:
|
|
self._require_enabled()
|
|
scope = self._write_scope_user(user_id, is_admin)
|
|
# Pre-check existence cross-user so an admin gets 403, not 404.
|
|
if await self._store.sessions.get_session(session_id, user_id=scope) is None:
|
|
if is_admin and await self._store.sessions.get_session(session_id, user_id=None) is not None:
|
|
raise MafForbiddenError("session", session_id)
|
|
raise MafNotFoundError("session", session_id)
|
|
ok = await self._store.sessions.delete_session(session_id, user_id=scope)
|
|
if not ok:
|
|
raise MafNotFoundError("session", session_id)
|
|
|
|
# ── runs ────────────────────────────────────────────────────────────────
|
|
|
|
async def create_run(
|
|
self, *, session_id: str, user_id: str | None, workflow_version: str | None = None, is_admin: bool = False
|
|
) -> dict[str, Any]:
|
|
self._require_enabled()
|
|
# Owner-only write: an admin cannot start another user's run. The
|
|
# owner_user_id also rides in the atomic SQL (not just this pre-check).
|
|
if not user_id:
|
|
raise MafNotFoundError("session", session_id)
|
|
|
|
run = await self._store.claim_and_create_pending_run(
|
|
session_id=session_id,
|
|
owner_user_id=user_id,
|
|
workflow_version=workflow_version,
|
|
)
|
|
if run is not None:
|
|
return run
|
|
|
|
# Classify the failed claim: missing / wrong-owner / already active.
|
|
any_session = await self._store.sessions.get_session(session_id, user_id=None)
|
|
if any_session is None:
|
|
raise MafNotFoundError("session", session_id)
|
|
if any_session["user_id"] != user_id:
|
|
if is_admin:
|
|
raise MafForbiddenError("session", session_id)
|
|
raise MafNotFoundError("session", session_id)
|
|
raise MafConflictError(
|
|
f"session {session_id} already has an active run; concurrent launch rejected"
|
|
)
|
|
|
|
async def get_run(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
|
|
run = await self._store.runs.get_run(run_id)
|
|
if run is None:
|
|
raise MafNotFoundError("run", run_id)
|
|
# Enforce session owner scope (read scope — admins may diagnose).
|
|
scope = self._read_scope_user(user_id, is_admin)
|
|
session = await self._store.sessions.get_session(run["session_id"], user_id=scope)
|
|
if session is None:
|
|
raise MafNotFoundError("run", run_id)
|
|
return run
|
|
|
|
async def list_runs(self, *, session_id: str, user_id: str | None, is_admin: bool = False) -> list[dict[str, Any]]:
|
|
scope = self._read_scope_user(user_id, is_admin)
|
|
session = await self._store.sessions.get_session(session_id, user_id=scope)
|
|
if session is None:
|
|
raise MafNotFoundError("session", session_id)
|
|
return await self._store.runs.list_runs(session_id)
|
|
|
|
async def cancel_run(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
|
|
self._require_enabled()
|
|
if not user_id:
|
|
raise MafNotFoundError("run", run_id)
|
|
|
|
result = await self._store.cancel_run_atomically(run_id=run_id, owner_user_id=user_id)
|
|
if result is not None:
|
|
return result
|
|
|
|
# Classify the miss: missing vs wrong-owner.
|
|
run = await self._store.runs.get_run(run_id)
|
|
if run is None:
|
|
raise MafNotFoundError("run", run_id)
|
|
if is_admin:
|
|
raise MafForbiddenError("run", run_id)
|
|
raise MafNotFoundError("run", run_id)
|
|
|
|
async def start_cancel(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
|
|
"""Phase 1 of two-phase cancel: atomically set run+session to ``cancelling``.
|
|
|
|
The caller (router) then calls ``dispatcher.cancel(run_id)`` to
|
|
cascade to DeerFlow, then calls ``cancel_run`` to finalize to
|
|
``cancelled``.
|
|
"""
|
|
self._require_enabled()
|
|
if not user_id:
|
|
raise MafNotFoundError("run", run_id)
|
|
|
|
result = await self._store.start_cancel_atomically(run_id=run_id, owner_user_id=user_id)
|
|
if result is not None:
|
|
return result
|
|
|
|
# Classify the miss.
|
|
run = await self._store.runs.get_run(run_id)
|
|
if run is None:
|
|
raise MafNotFoundError("run", run_id)
|
|
if is_admin:
|
|
raise MafForbiddenError("run", run_id)
|
|
raise MafNotFoundError("run", run_id)
|
|
|
|
async def finalize_cancel(self, *, run_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
|
|
"""Phase 2 of two-phase cancel: finalize ``cancelling`` → ``cancelled``."""
|
|
return await self.cancel_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
|
|
async def mark_cancel_failed(
|
|
self,
|
|
*,
|
|
run_id: str,
|
|
user_id: str | None,
|
|
is_admin: bool = False,
|
|
reason: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Mark a cancelling run as ``cancel_failed`` (non-terminal).
|
|
|
|
Used when the dispatcher's cancel cascade could not confirm all
|
|
DeerFlow sub-runs were cancelled. The run AND session both enter
|
|
``cancel_failed`` atomically so their statuses never diverge.
|
|
The user can retry cancel later — ``cancel_run_atomically``
|
|
accepts ``cancel_failed`` as a valid input state.
|
|
A failed cancel must NEVER masquerade as ``cancelled``.
|
|
"""
|
|
self._require_enabled()
|
|
if not user_id:
|
|
raise MafNotFoundError("run", run_id)
|
|
|
|
scope = self._write_scope_user(user_id, is_admin)
|
|
run = await self._store.runs.get_run(run_id)
|
|
if run is None:
|
|
raise MafNotFoundError("run", run_id)
|
|
# Owner (or admin) check.
|
|
session = await self._store.sessions.get_session(run["session_id"], user_id=scope)
|
|
if session is None:
|
|
if is_admin and await self._store.sessions.get_session(run["session_id"], user_id=None) is not None:
|
|
raise MafForbiddenError("run", run_id)
|
|
raise MafNotFoundError("run", run_id)
|
|
|
|
safe_reason = str(sanitize(reason))[:500] if reason else "one or more sub-runs could not be cancelled"
|
|
updated = await self._store.mark_cancel_failed_atomically(
|
|
run_id=run_id,
|
|
owner_user_id=user_id,
|
|
error_code="CancelCascadeIncomplete",
|
|
error_message=safe_reason,
|
|
)
|
|
if updated is None:
|
|
raise MafNotFoundError("run", run_id)
|
|
return updated
|
|
|
|
# ── events ─────────────────────────────────────────────────────────────
|
|
|
|
async def append_event(
|
|
self,
|
|
*,
|
|
run_id: str,
|
|
user_id: str | None,
|
|
event_type: str,
|
|
agent_id: str | None = None,
|
|
turn_index: int | None = None,
|
|
payload: dict[str, Any] | None = None,
|
|
source_event_id: str | None = None,
|
|
is_admin: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Append an event to a run's stream (internal / executor only).
|
|
|
|
There is **no** public HTTP write endpoint for events — Phase-1 audit
|
|
integrity requires that only the backend Workflow executor (Phase 2+)
|
|
call this method. Frontend联调 mocks belong in tests / scripts, not in
|
|
a production-reachable route.
|
|
"""
|
|
self._require_enabled()
|
|
# Owner-only write: an admin cannot forge events into another user's run.
|
|
run = await self._store.runs.get_run(run_id)
|
|
if run is None:
|
|
raise MafNotFoundError("run", run_id)
|
|
scope = self._write_scope_user(user_id, is_admin)
|
|
session = await self._store.sessions.get_session(run["session_id"], user_id=scope)
|
|
if session is None:
|
|
if is_admin and await self._store.sessions.get_session(run["session_id"], user_id=None) is not None:
|
|
raise MafForbiddenError("run", run_id)
|
|
raise MafNotFoundError("run", run_id)
|
|
return await self._store.events.append_event(
|
|
session_id=run["session_id"],
|
|
run_id=run_id,
|
|
event_type=event_type,
|
|
agent_id=agent_id,
|
|
turn_index=turn_index,
|
|
payload=sanitize_payload(payload),
|
|
source_event_id=source_event_id,
|
|
)
|
|
|
|
async def list_events(
|
|
self, *, run_id: str, user_id: str | None, after_seq: int = 0, limit: int = 1000, is_admin: bool = False
|
|
) -> tuple[list[dict[str, Any]], int]:
|
|
# get_run enforces owner read-scope (raises MafNotFoundError if out of scope).
|
|
await self.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
events = await self._store.events.list_events(run_id, after_seq=after_seq, limit=limit)
|
|
latest = await self._store.events.latest_seq(run_id)
|
|
return events, latest
|
|
|
|
# ── actions ────────────────────────────────────────────────────────────
|
|
|
|
async def list_actions(
|
|
self, *, run_id: str, user_id: str | None, is_admin: bool = False
|
|
) -> list[dict[str, Any]]:
|
|
from deerflow.persistence.maf_roundtable.actions import action_to_public_dict
|
|
|
|
run = await self.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
return [
|
|
action_to_public_dict(action)
|
|
for action in await self._store.actions.list_actions(run_id=run["id"])
|
|
]
|
|
|
|
async def get_action(self, *, action_id: str, user_id: str | None, is_admin: bool = False) -> dict[str, Any]:
|
|
from deerflow.persistence.maf_roundtable.actions import action_to_public_dict
|
|
|
|
action = await self._store.actions.get_action(action_id)
|
|
if action is None:
|
|
raise MafNotFoundError("action", action_id)
|
|
scope = self._read_scope_user(user_id, is_admin)
|
|
session = await self._store.sessions.get_session(action["session_id"], user_id=scope)
|
|
if session is None:
|
|
raise MafNotFoundError("action", action_id)
|
|
return action_to_public_dict(action)
|
|
|
|
# ── approvals ──────────────────────────────────────────────────────────
|
|
|
|
async def list_approvals(
|
|
self, *, run_id: str, user_id: str | None, is_admin: bool = False
|
|
) -> list[dict[str, Any]]:
|
|
run = await self.get_run(run_id=run_id, user_id=user_id, is_admin=is_admin)
|
|
# Reads are observational. Expiry is performed by the dedicated
|
|
# scanner; decision/resume paths retain the fail-closed pre-CAS check.
|
|
return await self._store.approvals.list_approvals(run_id=run["id"])
|
|
|
|
async def decide_approval(
|
|
self,
|
|
*,
|
|
approval_id: str,
|
|
decision: str,
|
|
user_id: str | None,
|
|
is_admin: bool = False,
|
|
) -> dict[str, Any]:
|
|
self._require_enabled()
|
|
approval = await self._store.approvals.get_approval(approval_id)
|
|
if approval is None:
|
|
raise MafNotFoundError("approval", approval_id)
|
|
# Owner-only write: an admin cannot decide another user's approval.
|
|
scope = self._write_scope_user(user_id, is_admin)
|
|
session = await self._store.sessions.get_session(approval["session_id"], user_id=scope)
|
|
if session is None:
|
|
if is_admin and await self._store.sessions.get_session(approval["session_id"], user_id=None) is not None:
|
|
raise MafForbiddenError("approval", approval_id)
|
|
raise MafNotFoundError("approval", approval_id)
|
|
# The timeout transition is server-owned and happens before the
|
|
# decision CAS, so an expired approval can never be revived by a late
|
|
# browser request.
|
|
if approval.get("run_id"):
|
|
await self._store.expire_due_approvals(run_id=approval["run_id"])
|
|
try:
|
|
updated = await self._store.approvals.decide(approval_id, decision=decision, decided_by=user_id)
|
|
except ValueError as exc:
|
|
raise MafConflictError(str(exc)) from exc
|
|
if updated is None:
|
|
raise MafNotFoundError("approval", approval_id)
|
|
return updated
|