"""Server-trusted tool side-effect / risk metadata for MAF WriteGuard. Harness-only — never imports ``app.*``. Used when building frozen capability snapshots and when classifying tools at runtime. Client-supplied metadata is never trusted; only deployment config (``ToolConfig`` extras) and this registry apply. Tools without explicit metadata default to ``write`` / ``R3`` (requires approval under ``confirm_high_risk``). """ from __future__ import annotations from collections.abc import Mapping from typing import Any _DEFAULT_SIDE_EFFECT = "write" _DEFAULT_RISK_LEVEL = "R3" # Built-in tools whose side effects are known server-side (not name-prefix guessing # for arbitrary tools — only this curated allow-list). _BUILTIN_METADATA: dict[str, tuple[str, str]] = { "memory": ("read", "R0"), "ask_clarification": ("read", "R0"), "view_image": ("read", "R0"), "present_file": ("read", "R0"), "web_search": ("read", "R1"), "web_fetch": ("read", "R1"), "read_file": ("read", "R0"), "grep": ("read", "R0"), "glob": ("read", "R0"), "ls": ("read", "R0"), "read_peer_delivery": ("read", "R0"), "invoke_skill": ("skill", "R1"), "skill_search": ("skill", "R1"), "task": ("read", "R2"), "bash": ("write", "R3"), "write_file": ("write", "R3"), "edit_file": ("write", "R3"), } def resolve_tool_metadata( tool_name: str, *, config_extra: dict[str, Any] | None = None, maf_metadata: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Return trusted metadata for one tool. Precedence is intentionally server-only and explicit: 1. MAF deployment metadata keyed by the exact runtime tool name; 2. existing server ``ToolConfig`` extras; 3. the curated built-in allow-list; 4. conservative ``write`` / ``R3`` fallback. ``maf_metadata`` is passed by the application while building a frozen capability snapshot. It is never accepted from a session request or a model message. """ side = _DEFAULT_SIDE_EFFECT risk = _DEFAULT_RISK_LEVEL target_key_fields: list[str] = [] target_key_namespace = "default" registered = maf_metadata.get(tool_name) if maf_metadata else None if registered is not None: candidate_side = registered.get("side_effect") if isinstance(registered, Mapping) else getattr(registered, "side_effect", None) candidate_risk = registered.get("risk_level") if isinstance(registered, Mapping) else getattr(registered, "risk_level", None) if candidate_side in {"read", "skill", "write"}: side = str(candidate_side) if candidate_risk in {"R0", "R1", "R2", "R3", "R4"}: risk = str(candidate_risk) candidate_fields = registered.get("target_key_fields") if isinstance(registered, Mapping) else getattr(registered, "target_key_fields", None) candidate_namespace = registered.get("target_key_namespace") if isinstance(registered, Mapping) else getattr(registered, "target_key_namespace", None) if isinstance(candidate_fields, list) and all(isinstance(item, str) for item in candidate_fields): target_key_fields = list(candidate_fields) if isinstance(candidate_namespace, str) and candidate_namespace: target_key_namespace = candidate_namespace elif config_extra: candidate_side = config_extra.get("side_effect") candidate_risk = config_extra.get("risk_level") if candidate_side in {"read", "skill", "write"}: side = str(candidate_side) if candidate_risk in {"R0", "R1", "R2", "R3", "R4"}: risk = str(candidate_risk) candidate_fields = config_extra.get("target_key_fields") candidate_namespace = config_extra.get("target_key_namespace") if isinstance(candidate_fields, list) and all(isinstance(item, str) for item in candidate_fields): target_key_fields = list(candidate_fields) if isinstance(candidate_namespace, str) and candidate_namespace: target_key_namespace = candidate_namespace elif tool_name in _BUILTIN_METADATA: side, risk = _BUILTIN_METADATA[tool_name] return { "name": tool_name, "tool_name": tool_name, "side_effect": side, "risk_level": risk, "target_key_fields": target_key_fields, "target_key_namespace": target_key_namespace, }