deerflow-code/offline-backend-20260512/backend/app/maf_roundtable/capability_snapshot.py

"""Build server-trusted, frozen MAF capability snapshots.

Clients must not supply tool metadata or full snapshots — only participant
``agent_id`` and optional per-seat ``excluded_tools`` (subtractive only).
"""

from __future__ import annotations

import logging
from typing import Any

from deerflow.config.agents_config import load_agent_config
from deerflow.config.app_config import AppConfig
from deerflow.tools.tool_metadata import resolve_tool_metadata
from deerflow.tools.tools import get_available_tools

logger = logging.getLogger(__name__)


class CapabilitySnapshotError(Exception):
    """Invalid participant / tool selection when building a trusted snapshot."""


class AgentNotAccessibleError(CapabilitySnapshotError):
    def __init__(self, agent_id: str) -> None:
        super().__init__(f"agent not accessible: {agent_id}")
        self.agent_id = agent_id


class InvalidExcludedToolsError(CapabilitySnapshotError):
    def __init__(self, agent_id: str, tools: list[str]) -> None:
        super().__init__(f"invalid excluded_tools for {agent_id}: {tools}")
        self.agent_id = agent_id
        self.tools = tools


async def _assert_user_may_use_agent(
    *,
    agent_store: Any | None,
    user_id: str,
    agent_id: str,
) -> None:
    """Ensure the user may run this agent (visible in catalog)."""
    if agent_store is not None:
        visible = await agent_store.get_visible(agent_id, user_id)
        if visible is None:
            raise AgentNotAccessibleError(agent_id)
        return
    cfg = load_agent_config(agent_id)
    if cfg is None:
        raise AgentNotAccessibleError(agent_id)


def _config_extra_for_tool(app_config: AppConfig, tool_name: str) -> dict[str, Any] | None:
    for tc in app_config.tools:
        if tc.name == tool_name:
            extra = dict(getattr(tc, "model_extra", None) or {})
            return extra or None
    return None


def _seat_tools_for_agent(
    *,
    app_config: AppConfig,
    agent_id: str,
    session_model_name: str | None,
    session_reasoning_effort: str | None,
    requested_excluded: list[str],
) -> tuple[list[str], list[str], list[dict[str, Any]]]:
    """Resolve allowed names, effective excluded, and tool metadata for one seat."""
    agent_config = load_agent_config(agent_id)
    tool_groups = getattr(agent_config, "tool_groups", None) if agent_config else None
    model_name = (
        getattr(agent_config, "model", None) if agent_config else None
    ) or session_model_name

    available = get_available_tools(
        groups=tool_groups,
        model_name=model_name,
        excluded_tools=[],
        app_config=app_config,
    )
    allowed_set = {t.name for t in available}
    excluded_set = set(requested_excluded or [])
    unknown_excluded = excluded_set - allowed_set
    if unknown_excluded:
        raise InvalidExcludedToolsError(agent_id, sorted(unknown_excluded))
    effective_excluded = sorted(excluded_set & allowed_set)
    effective_allowed = sorted(allowed_set - set(effective_excluded))

    tools_meta: list[dict[str, Any]] = []
    maf_tool_metadata = getattr(
        getattr(app_config, "maf_roundtable", None), "tool_metadata", None
    ) or {}
    for name in effective_allowed:
        extra = _config_extra_for_tool(app_config, name)
        tools_meta.append(
            resolve_tool_metadata(
                name,
                config_extra=extra,
                maf_metadata=maf_tool_metadata,
            )
        )

    return effective_allowed, effective_excluded, tools_meta


async def build_trusted_capability_snapshot(
    *,
    user_id: str,
    participants: list[dict[str, Any]],
    app_config: AppConfig,
    agent_store: Any | None = None,
    session_model_name: str | None = None,
    session_reasoning_effort: str | None = None,
    session_excluded_tools: list[str] | None = None,
) -> dict[str, Any]:
    """Build the frozen snapshot stored on the session and used by WriteGuard."""
    agents_out: list[dict[str, Any]] = []
    flat_tools: dict[str, dict[str, Any]] = {}

    for p in participants:
        agent_id = str(p.get("agent_id") or "")
        if not agent_id:
            continue
        await _assert_user_may_use_agent(
            agent_store=agent_store, user_id=user_id, agent_id=agent_id
        )
        agent_config = load_agent_config(agent_id)
        seat_excluded = list(p.get("excluded_tools") or [])
        if session_excluded_tools:
            seat_excluded = sorted(set(seat_excluded) | set(session_excluded_tools))

        allowed, excluded, tools_meta = _seat_tools_for_agent(
            app_config=app_config,
            agent_id=agent_id,
            session_model_name=session_model_name,
            session_reasoning_effort=session_reasoning_effort,
            requested_excluded=seat_excluded,
        )
        for meta in tools_meta:
            flat_tools[meta["tool_name"]] = meta

        agents_out.append(
            {
                "agent_id": agent_id,
                "role": p.get("role", "seat"),
                "model_name": getattr(agent_config, "model", None) if agent_config else session_model_name,
                "reasoning_effort": session_reasoning_effort,
                "thinking_enabled": True,
                "subagent_enabled": bool(getattr(agent_config, "subagent_enabled", False))
                if agent_config
                else False,
                "tool_groups": list(getattr(agent_config, "tool_groups", None) or [])
                if agent_config
                else None,
                "allowed_tool_names": allowed,
                "excluded_tools": excluded,
                "tools": tools_meta,
                "memory_recall_disabled": None,
                "memory_injection_enabled": None,
            }
        )

    return {
        "version": 1,
        "agents": agents_out,
        "tools": list(flat_tools.values()),
    }


def seat_snapshot_for_agent(full_snapshot: dict[str, Any], agent_id: str) -> dict[str, Any]:
    """Per-participant slice persisted on ``maf_roundtable_participants``."""
    for a in full_snapshot.get("agents") or []:
        if isinstance(a, dict) and a.get("agent_id") == agent_id:
            return dict(a)
    return {}