"""
Central tool-policy registry and pre-invoke gate for Mugi.

Defines capability flags for every registered tool, a policy context that
encodes the current session's permissions, and a single gate function that
all tool invocation paths must call before executing a tool.

See docs/harness_threat_model.md for the normative threat model.
"""
from __future__ import annotations

import logging
import os
from dataclasses import dataclass, field
from enum import IntFlag, auto
from contextvars import ContextVar, Token
from typing import Any

_log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Stable prefix for blocked-tool result strings.  Tests, UI, and NDJSON
# logging assert on this so it must never change without a migration.
# ---------------------------------------------------------------------------
POLICY_BLOCKED_PREFIX = "[Policy blocked]"


# ---------------------------------------------------------------------------
# Capability flags — orthogonal; a tool may have multiple.
# ---------------------------------------------------------------------------
class Cap(IntFlag):
    READ_FS = auto()
    WRITE_FS = auto()
    NETWORK = auto()
    SHELL_LIKE = auto()
    SUBPROCESS_OTHER = auto()
    GIT_MUTATE = auto()
    CLIPBOARD_READ = auto()
    CLIPBOARD_WRITE = auto()
    CALENDAR_READ = auto()
    REMINDER_MUTATE = auto()
    EMAIL_READ = auto()
    EMAIL_COMPOSE = auto()
    MEMORY_WRITE = auto()
    SCHEDULE_MUTATE = auto()
    SCREEN_CAPTURE = auto()
    UX_SIDE_EFFECT = auto()
    PLUGIN_BUNDLED = auto()
    MCP_EXTERNAL = auto()
    PLUGIN_OR_WORKSPACE_MUTATE = auto()
    META = auto()  # find_tools, delegate_task — discovery or delegation
    READ_ONLY_DISCOVERY = auto()
    BROWSER_UI_AUTOMATION = auto()  # shared-browser / CDP driving primitives (Neko)


# Capabilities blocked under plan_mode by default.
_PLAN_MODE_BLOCKED = (
    Cap.WRITE_FS
    | Cap.SHELL_LIKE
    | Cap.GIT_MUTATE
    | Cap.CLIPBOARD_WRITE
    | Cap.REMINDER_MUTATE
    | Cap.EMAIL_COMPOSE
    | Cap.MEMORY_WRITE
    | Cap.SCHEDULE_MUTATE
    | Cap.PLUGIN_OR_WORKSPACE_MUTATE
    | Cap.UX_SIDE_EFFECT
)

_UNATTENDED_EXTERNAL_BLOCKED = (
    Cap.SHELL_LIKE
    | Cap.GIT_MUTATE
    | Cap.PLUGIN_OR_WORKSPACE_MUTATE
    | Cap.CLIPBOARD_WRITE
    | Cap.EMAIL_COMPOSE
    | Cap.BROWSER_UI_AUTOMATION
)

# Opt-in env var for unattended browser automation. When set to a truthy value
# on the backend process, the unattended-external gate exempts tools whose only
# blocking capability is ``BROWSER_UI_AUTOMATION``. Any additional blocked caps
# (SHELL_LIKE, GIT_MUTATE, etc.) still deny the call.
_UNATTENDED_BROWSER_ENV = "MUGI_ALLOW_UNATTENDED_BROWSER_AUTOMATION"
_UNATTENDED_WEB_ACT_ENV = "MUGI_ALLOW_UNATTENDED_WEB_ACT"
_ENV_TRUTHY = frozenset({"1", "true", "yes", "on"})

_WEB_ACT_TOOL_NAMES: frozenset[str] = frozenset({"run_web_script", "write_web_act_file"})


def _unattended_browser_opt_in() -> bool:
    return os.environ.get(_UNATTENDED_BROWSER_ENV, "").strip().lower() in _ENV_TRUTHY


def _unattended_web_act_opt_in() -> bool:
    return os.environ.get(_UNATTENDED_WEB_ACT_ENV, "").strip().lower() in _ENV_TRUTHY


# ---------------------------------------------------------------------------
# Core tool → capability registry
# ---------------------------------------------------------------------------
_TOOL_CAPABILITIES: dict[str, Cap] = {
    # -- file_ops --
    "read_file": Cap.READ_FS,
    "list_directory": Cap.READ_FS,
    "write_file": Cap.WRITE_FS,
    "edit_file": Cap.WRITE_FS,
    "delete_file": Cap.WRITE_FS,
    "copy_file": Cap.WRITE_FS,
    "move_file": Cap.WRITE_FS,
    "split_text_file": Cap.WRITE_FS,
    "merge_text_files": Cap.WRITE_FS,
    "save_to_file": Cap.WRITE_FS,
    "write_text_file": Cap.WRITE_FS,
    "append_to_text_file": Cap.WRITE_FS,
    "patch_user_profile": Cap.WRITE_FS,
    "patch_user_model": Cap.MEMORY_WRITE,
    "propose_user_model_update": Cap.MEMORY_WRITE,
    "list_pending_proposals": Cap.READ_FS,
    "read_file_chunk": Cap.READ_FS,
    "submit_ocr_chunk_corrections": Cap.WRITE_FS,
    "correct_ocr_text": Cap.WRITE_FS,
    "get_file_metadata": Cap.READ_FS,
    "verify_file_exists": Cap.READ_FS,
    "get_default_save_location": Cap.READ_FS,
    "suggest_save_filename": Cap.READ_FS,
    "commit_workspace_evolution": Cap.PLUGIN_OR_WORKSPACE_MUTATE | Cap.GIT_MUTATE,
    # -- pdf_tools --
    "extract_pdf_text": Cap.READ_FS,
    "split_pdf": Cap.WRITE_FS,
    "merge_pdfs": Cap.WRITE_FS,
    # -- web_tools --
    "search_web": Cap.NETWORK,
    "search_web_public": Cap.NETWORK,
    "read_webpage": Cap.NETWORK,
    "get_captured_browser_page": Cap.NETWORK,
    "request_browser_page": Cap.NETWORK | Cap.BROWSER_UI_AUTOMATION,
    "summarize_url": Cap.NETWORK,
    "screenshot_webpage": Cap.NETWORK | Cap.WRITE_FS,
    "download_file": Cap.NETWORK | Cap.WRITE_FS,
    "capture_network_traffic": Cap.NETWORK,
    "run_web_script": Cap.NETWORK | Cap.SUBPROCESS_OTHER | Cap.WRITE_FS,
    "write_web_act_file": Cap.NETWORK | Cap.WRITE_FS,
    # -- clipboard --
    "get_clipboard_text": Cap.CLIPBOARD_READ,
    "set_clipboard_text": Cap.CLIPBOARD_WRITE,
    "get_clipboard_image": Cap.CLIPBOARD_READ,
    "copy_image_to_clipboard": Cap.CLIPBOARD_WRITE,
    "copy_image_from_url_to_clipboard": Cap.CLIPBOARD_WRITE | Cap.NETWORK,
    "copy_current_clipboard_image_again": Cap.CLIPBOARD_WRITE,
    "view_image": Cap.READ_FS,
    # -- system --
    "shell_execute": Cap.SHELL_LIKE,
    "get_current_directory": Cap.READ_FS,
    "run_python_code": Cap.SHELL_LIKE,
    "show_display_window": Cap.UX_SIDE_EFFECT,
    "search_filesystem": Cap.READ_FS,
    "spotlight_search": Cap.READ_FS | Cap.SUBPROCESS_OTHER,
    "get_recent_files": Cap.READ_ONLY_DISCOVERY,
    # Read-only host diagnostics: curated no-sudo subprocesses (nettop, lpstat,
    # system_profiler, ...) — deliberately NOT SHELL_LIKE. Mutating fixes are
    # documented to go through shell_execute, which keeps its own gate.
    "inspect_system_activity": Cap.READ_ONLY_DISCOVERY | Cap.SUBPROCESS_OTHER,
    "run_host_diagnostics": Cap.READ_ONLY_DISCOVERY | Cap.SUBPROCESS_OTHER,
    "capture_screenshot": Cap.SCREEN_CAPTURE,
    "compress_files": Cap.WRITE_FS,
    "extract_archive": Cap.WRITE_FS,
    "get_current_time": Cap.READ_ONLY_DISCOVERY,
    # -- vision_ocr --
    "ocr_scanned_pdf": Cap.READ_FS | Cap.NETWORK,
    "ocr_scanned_pdf_to_file": Cap.READ_FS | Cap.WRITE_FS,
    "ocr_image_text": Cap.READ_FS,
    # -- tesseract --
    "ocr_complex_document": Cap.READ_FS | Cap.SUBPROCESS_OTHER,
    "ocr_complex_document_to_file": Cap.READ_FS | Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    # -- kraken --
    "ocr_with_kraken_api": Cap.READ_FS | Cap.SUBPROCESS_OTHER,
    "ocr_with_kraken_api_to_file": Cap.READ_FS | Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    # -- memory_tools --
    "append_to_memory": Cap.MEMORY_WRITE,
    "migrate_memory_to_shards": Cap.MEMORY_WRITE,
    "get_memory_size": Cap.READ_FS,
    "summarize_old_memory": Cap.MEMORY_WRITE,
    "archive_old_memory": Cap.MEMORY_WRITE,
    "write_session_state": Cap.MEMORY_WRITE,
    "read_session_state": Cap.READ_FS,
    "get_daily_memory_paths": Cap.READ_FS,
    "append_working_buffer": Cap.MEMORY_WRITE,
    "get_recent_feedback": Cap.READ_FS,
    "search_memory": Cap.READ_FS,
    "search_memory_and_knowledge_corpora": Cap.READ_FS,
    "refresh_memory": Cap.MEMORY_WRITE,
    # -- research --
    "deep_web_search": Cap.NETWORK,
    "read_and_analyze_page": Cap.NETWORK,
    "analyze_multiple_sources": Cap.NETWORK,
    "append_research_scratch": Cap.WRITE_FS,
    "save_research_findings": Cap.WRITE_FS,
    "suggest_follow_up_queries": Cap.READ_ONLY_DISCOVERY,
    "reflect_research_state": Cap.READ_ONLY_DISCOVERY,
    "plan_research_subquestions": Cap.READ_ONLY_DISCOVERY,
    "generate_final_research_report": Cap.WRITE_FS,
    # -- capture_tools --
    "capture_to_inbox": Cap.WRITE_FS,
    "list_captures": Cap.READ_FS,
    "process_capture_inbox": Cap.READ_FS | Cap.WRITE_FS,
    "list_pattern_connections": Cap.READ_FS,
    "list_pattern_action_cards": Cap.READ_FS,
    "list_capture_projects": Cap.READ_FS,
    "create_capture_project": Cap.WRITE_FS,
    "file_capture": Cap.READ_FS | Cap.WRITE_FS,
    "tag_capture": Cap.READ_FS | Cap.WRITE_FS,
    "process_capture_project": Cap.READ_FS | Cap.WRITE_FS,
    "list_project_captures": Cap.READ_FS,
    "list_project_notes": Cap.READ_FS,
    "search_vault_notes": Cap.READ_FS,
    "list_recent_vault_notes": Cap.READ_FS,
    "get_note": Cap.READ_FS,
    "create_vault_note": Cap.WRITE_FS,
    "update_vault_note": Cap.WRITE_FS,
    "append_to_vault_note": Cap.WRITE_FS,
    "summarize_capture_project": Cap.READ_FS,
    # -- vault_graph_tools --
    "vault_graph_neighbors": Cap.READ_FS,
    "vault_graph_linked_mentions": Cap.READ_FS,
    "vault_graph_list_suggestions": Cap.READ_FS,
    "vault_graph_refresh_suggestions": Cap.READ_FS | Cap.WRITE_FS,
    "vault_graph_confirm_edge": Cap.READ_FS | Cap.WRITE_FS,
    "vault_graph_reject_edge": Cap.READ_FS | Cap.WRITE_FS,
    "vault_graph_subgraph": Cap.READ_FS,
    "vault_graph_insights": Cap.READ_FS,
    # -- task_plan_tools --
    "update_task_plan": Cap.WRITE_FS,
    "clear_task_plan": Cap.WRITE_FS,
    "get_task_plan": Cap.READ_FS,
    # -- kanban_tools (planner / worker; encrypted kanban_storage — no shell/network) --
    "kanban_list_tasks": Cap.READ_FS,
    "kanban_show_task": Cap.READ_FS,
    "kanban_list_runs": Cap.READ_FS,
    "kanban_create_task": Cap.WRITE_FS,
    "kanban_create_batch": Cap.WRITE_FS,
    "convene_council": Cap.WRITE_FS,
    "kanban_cancel_task": Cap.WRITE_FS,
    "kanban_post_summary": Cap.MEMORY_WRITE,
    "kanban_post_progress": Cap.MEMORY_WRITE,
    # -- git_workspace --
    "git_project_status": Cap.READ_FS | Cap.SUBPROCESS_OTHER,
    "git_project_restore": Cap.GIT_MUTATE | Cap.SUBPROCESS_OTHER,
    "git_project_commit": Cap.GIT_MUTATE | Cap.SUBPROCESS_OTHER,
    "git_project_push": Cap.GIT_MUTATE | Cap.SUBPROCESS_OTHER | Cap.NETWORK,
    # -- calendar_tools --
    "get_upcoming_calendar_events": Cap.CALENDAR_READ,
    # -- plugin_creation --
    "create_plugin": Cap.PLUGIN_OR_WORKSPACE_MUTATE | Cap.WRITE_FS,
    "create_dashboard_plugin": Cap.PLUGIN_OR_WORKSPACE_MUTATE | Cap.WRITE_FS,
    "validate_plugin": Cap.READ_FS,
    "diagnose_plugin_sandbox_migration": Cap.READ_FS,
    # -- watchlist_tools --
    "add_watch_target": Cap.WRITE_FS,
    "list_watch_targets": Cap.READ_FS,
    "remove_watch_target": Cap.WRITE_FS,
    "compare_watch_baseline": Cap.READ_FS | Cap.NETWORK,
    # -- reminders_tools --
    "get_reminders": Cap.CALENDAR_READ,
    "add_reminder": Cap.REMINDER_MUTATE,
    "remind_me_in": Cap.REMINDER_MUTATE,
    "remind_at": Cap.REMINDER_MUTATE,
    # -- email_tools --
    "list_inbox": Cap.EMAIL_READ | Cap.NETWORK,
    "compose_email_draft": Cap.EMAIL_COMPOSE,
    # -- feed_tools --
    "add_feed": Cap.WRITE_FS,
    "list_feeds": Cap.READ_FS,
    "remove_feed": Cap.WRITE_FS,
    "check_feeds": Cap.NETWORK | Cap.READ_FS,
    # -- weather_tools --
    "get_weather": Cap.NETWORK,
    # -- subagent_tools --
    "delegate_task": Cap.META,
    "append_delegation_scratch": Cap.MEMORY_WRITE,
    "read_delegation_scratch": Cap.READ_FS,
    # -- mesh_tools --
    "contact_peer": Cap.NETWORK,
    "mesh_send_files": Cap.NETWORK,
    "list_mesh_peers": Cap.READ_ONLY_DISCOVERY,
    "mesh_conversation_status": Cap.READ_ONLY_DISCOVERY,
    # -- background_tools --
    "run_in_background": Cap.META,
    "check_background_task": Cap.READ_ONLY_DISCOVERY,
    "stop_background_task": Cap.META,
    "list_background_tasks": Cap.READ_ONLY_DISCOVERY,
    # -- skills_tools --
    "skills_list": Cap.READ_FS,
    "skill_view": Cap.READ_FS,
    "skill_manage": Cap.PLUGIN_OR_WORKSPACE_MUTATE | Cap.WRITE_FS,
    # -- drive_tools --
    "list_drives": Cap.READ_ONLY_DISCOVERY,
    "resolve_drive": Cap.META,
    "dismiss_drive": Cap.META,
    "create_drive": Cap.WRITE_FS,
    # -- knowledge_tools --
    "digest_corpus": Cap.READ_FS | Cap.WRITE_FS | Cap.NETWORK,
    "list_knowledge": Cap.READ_ONLY_DISCOVERY,
    "query_knowledge": Cap.READ_FS,
    "ask_studied_corpus": Cap.READ_FS,
    "read_knowledge_source": Cap.READ_FS,
    "explore_knowledge_connections": Cap.READ_FS,
    # -- knowledge_graph_tools --
    "knowledge_graph_subgraph": Cap.READ_FS,
    "knowledge_graph_neighbors": Cap.READ_FS,
    "knowledge_graph_list_suggestions": Cap.READ_FS,
    "knowledge_graph_confirm_edge": Cap.READ_FS | Cap.WRITE_FS,
    "knowledge_graph_reject_edge": Cap.READ_FS | Cap.WRITE_FS,
    "knowledge_graph_reindex_corpus": Cap.READ_FS | Cap.WRITE_FS,
    "knowledge_graph_gitnexus_hint": Cap.READ_FS,
    # -- session_search_tools --
    "search_sessions": Cap.READ_FS,
    # -- schedule_tools --
    "schedule_task": Cap.SCHEDULE_MUTATE,
    "list_scheduled_tasks": Cap.READ_ONLY_DISCOVERY,
    "cancel_scheduled_task": Cap.SCHEDULE_MUTATE,
    # -- timer_tools (Phase 2b) --
    "schedule_timer": Cap.SCHEDULE_MUTATE,
    "list_timers": Cap.READ_ONLY_DISCOVERY,
    "cancel_timer": Cap.SCHEDULE_MUTATE,
    # -- tool_index --
    "find_tools": Cap.READ_ONLY_DISCOVERY,
    # -- CLI: brew --
    "ensure_mugi_homebrew": Cap.SHELL_LIKE,
    "mugi_brew_install": Cap.SHELL_LIKE,
    # -- CLI: ffmpeg --
    "get_media_info": Cap.READ_FS | Cap.SUBPROCESS_OTHER,
    "convert_video": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "extract_audio": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "convert_audio": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "trim_media": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "concat_media": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "extract_frame": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "merge_audio_video": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "remove_audio": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "extract_subtitles": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "burn_subtitles": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "split_media": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "change_speed": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "rotate_video": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "images_to_video": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "extract_frames_interval": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "normalize_audio": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    # -- CLI: runner --
    "run_installed_cli": Cap.SHELL_LIKE,
    "install_cli_package": Cap.SHELL_LIKE,
    # -- bundled plugin: yt_dlp --
    "download_video_yt_dlp": Cap.NETWORK | Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    # -- hyperframes --
    "hyperframes_create_video": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "hyperframes_render": Cap.WRITE_FS | Cap.SUBPROCESS_OTHER,
    "hyperframes_lint": Cap.READ_FS | Cap.SUBPROCESS_OTHER,
    "hyperframes_doctor": Cap.SUBPROCESS_OTHER,
}

# Conservative default for unknown tools (MCP, plugins without manifest).
_UNKNOWN_TOOL_DEFAULT = Cap.NETWORK | Cap.WRITE_FS | Cap.SHELL_LIKE


def _plugin_tool_names_equivalent(a: str, b: str) -> bool:
    """True if *a* and *b* are the same tool name, allowing ``__`` vs ``_`` collapse."""
    if a == b:
        return True
    return a.replace("__", "_") == b.replace("__", "_")


def _capabilities_from_loaded_plugin_tool(tool_name: str) -> Cap | None:
    """If a loaded plugin exposes this tool name, use its manifest permissions.

    Covers static registration gaps and LLM/backend name collapse (``__`` vs ``_``).
    """
    if not tool_name:
        return None
    try:
        from agent_loop_tools import _get_plugin_manager
    except Exception:
        return None
    pm = _get_plugin_manager()
    if pm is None:
        return None
    try:
        for lp in pm.loaded_plugins:
            if lp.error:
                continue
            for t in lp.context._tools:
                reg = getattr(t, "name", None) or ""
                if reg and _plugin_tool_names_equivalent(tool_name, reg):
                    return caps_from_plugin_permissions(lp.manifest.permissions)
    except Exception:
        return None
    return None


def get_capabilities(tool_name: str) -> Cap:
    """Return capability flags for a tool.  Unknown tools get conservative defaults."""
    tool_name = str(tool_name or "").strip()
    if tool_name in _TOOL_CAPABILITIES:
        return _TOOL_CAPABILITIES[tool_name]
    lazy = _capabilities_from_loaded_plugin_tool(tool_name)
    if lazy is not None:
        return lazy
    collapsed = tool_name.replace("__", "_") if "__" in tool_name else None
    if collapsed and collapsed != tool_name:
        if collapsed in _TOOL_CAPABILITIES:
            return _TOOL_CAPABILITIES[collapsed]
        lazy_c = _capabilities_from_loaded_plugin_tool(collapsed)
        if lazy_c is not None:
            return lazy_c
    return _UNKNOWN_TOOL_DEFAULT


def register_tool(tool_name: str, caps: Cap) -> None:
    """Register or update capabilities for a tool (e.g. from plugin manifest merge)."""
    _TOOL_CAPABILITIES[tool_name] = caps


# Sentinel: tool name was not in _TOOL_CAPABILITIES before a plugin claimed it.
_PLUGIN_REG_ABSENT = object()

# slug -> {tool_name: previous Cap entry or _PLUGIN_REG_ABSENT}
_PLUGIN_REGISTRATIONS: dict[str, dict[str, Any]] = {}


def caps_from_plugin_permissions(permissions: dict | None) -> Cap:
    """Map plugin.toml [permissions] to policy caps (approved plugins only).

    Unknown keys are ignored. Filesystem must be one of none|read|read_write.
    ``browser_ui_automation = true`` marks the tool as a shared-browser /
    CDP driving primitive so the unattended-external gate can treat it as
    such (default-deny, opt-in via ``MUGI_ALLOW_UNATTENDED_BROWSER_AUTOMATION``).
    """
    perms = permissions or {}
    caps = Cap(0)
    fs = str(perms.get("filesystem", "none") or "none").strip()
    if fs == "read":
        caps |= Cap.READ_FS
    elif fs == "read_write":
        caps |= Cap.READ_FS | Cap.WRITE_FS
    if bool(perms.get("network", False)):
        caps |= Cap.NETWORK
    if bool(perms.get("shell", False)):
        caps |= Cap.SHELL_LIKE
    if bool(perms.get("browser_ui_automation", False)):
        caps |= Cap.BROWSER_UI_AUTOMATION
    return caps


def unregister_plugin_tool_capabilities(slug: str) -> None:
    """Remove policy entries previously registered for *slug*; restore prior caps."""
    snap = _PLUGIN_REGISTRATIONS.pop(slug, {})
    for tn, prev in snap.items():
        if prev is _PLUGIN_REG_ABSENT:
            _TOOL_CAPABILITIES.pop(tn, None)
        else:
            _TOOL_CAPABILITIES[tn] = prev


def register_plugin_tools_for_slug(
    slug: str, tool_names: list[str], permissions: dict | None,
    tool_permissions: dict[str, dict] | None = None,
) -> None:
    """Register capabilities for tools exposed by an approved loaded plugin.

    Call :func:`unregister_plugin_tool_capabilities` for the same *slug* before
    loading/reloading so stale tool names are cleared first.

    *tool_permissions* is an optional mapping from tool name → per-tool
    permission overrides (same keys as *permissions*).  Any key present in the
    per-tool dict overrides the plugin-level default; missing keys inherit from
    *permissions*.

    Local LLM backends sometimes collapse ``__`` to ``_`` in function names when
    emitting tool calls. Register both the canonical name and a single-underscore
    alias so policy matches what the model actually sends.
    """
    plugin_caps = caps_from_plugin_permissions(permissions)
    tp = tool_permissions or {}
    snap: dict[str, Any] = {}

    def _claim(name: str, caps: Cap) -> None:
        if not name or name in snap:
            return
        snap[name] = (
            _TOOL_CAPABILITIES[name] if name in _TOOL_CAPABILITIES else _PLUGIN_REG_ABSENT
        )
        _TOOL_CAPABILITIES[name] = caps

    for raw in tool_names:
        tn = str(raw or "").strip()
        if not tn:
            continue
        # Merge per-tool overrides into plugin-level permissions.
        if tn in tp:
            merged = dict(permissions or {})
            merged.update(tp[tn])
            tool_caps = caps_from_plugin_permissions(merged)
        else:
            tool_caps = plugin_caps
        _claim(tn, tool_caps)
        if "__" in tn:
            collapsed = tn.replace("__", "_")
            while "__" in collapsed:
                collapsed = collapsed.replace("__", "_")
            if collapsed != tn:
                _claim(collapsed, tool_caps)
    if snap:
        _PLUGIN_REGISTRATIONS[slug] = snap
    else:
        _PLUGIN_REGISTRATIONS.pop(slug, None)


# ---------------------------------------------------------------------------
# ToolPolicyContext — immutable snapshot of session permissions
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class ToolPolicyContext:
    """Immutable snapshot of the policy state for a single agent run.

    Built once per ``run_raw_agent`` invocation via ``context_from_prefs_and_flags``.
    Passed (by value) across thread boundaries — safe to share between parallel
    tool calls or subagent workers.
    """
    plan_mode: bool = False
    allow_shell_execute: bool = True
    background_automation: bool = False
    scheduled_job: bool = False
    unattended_external: bool = False
    messaging: bool = False
    mesh_sourced: bool = False
    hooks_triggered: bool = False
    subagent_mode: bool = False
    idle_tick: bool = False
    quick_panel: bool = False
    dashboard_surface: bool = False
    kanban_worker: bool = False
    kanban_web_act: bool = False
    # Per-task shell opt-in (kernel kanban worker, spec §4). Default deny;
    # only the dispatcher sets this true from operator-set task metadata.
    kanban_allow_shell: bool = False
    recent_user_text: str = ""


@dataclass(frozen=True)
class ToolExposurePolicy:
    """Shared schema/runtime visibility preferences read from app prefs."""

    allow_shell_execute: bool = True
    mcp_schemas_in_tiered_parent: bool = True


def get_tool_exposure_policy() -> ToolExposurePolicy:
    prefs = _load_prefs()
    return ToolExposurePolicy(
        allow_shell_execute=bool(prefs.get("allow_shell_execute", False)),
        mcp_schemas_in_tiered_parent=bool(prefs.get("mcp_schemas_in_tiered_parent", True)),
    )


def context_from_prefs_and_flags(
    *,
    plan_mode: bool | None = None,
    allow_shell_execute: bool | None = None,
    background_automation: bool = False,
    scheduled_job: bool = False,
    unattended_external: bool = False,
    messaging: bool = False,
    mesh_sourced: bool = False,
    hooks_triggered: bool = False,
    subagent_mode: bool = False,
    idle_tick: bool = False,
    quick_panel: bool = False,
    dashboard_surface: bool = False,
    kanban_worker: bool = False,
    kanban_web_act: bool = False,
    kanban_allow_shell: bool = False,
    recent_user_text: str = "",
) -> ToolPolicyContext:
    """Build a ``ToolPolicyContext`` from live prefs + caller-supplied flags.

    When ``plan_mode`` or ``allow_shell_execute`` are ``None`` (the default),
    the value is read from ``~/.mugi/app_prefs.json`` at call time.
    """
    prefs = _load_prefs()

    if plan_mode is None:
        plan_mode = prefs.get("plan_mode_enabled", False)
    if allow_shell_execute is None:
        allow_shell_execute = prefs.get("allow_shell_execute", True)

    return ToolPolicyContext(
        plan_mode=plan_mode,
        allow_shell_execute=allow_shell_execute,
        background_automation=background_automation,
        scheduled_job=scheduled_job,
        unattended_external=unattended_external,
        messaging=messaging,
        mesh_sourced=mesh_sourced,
        hooks_triggered=hooks_triggered,
        subagent_mode=subagent_mode,
        idle_tick=idle_tick,
        quick_panel=quick_panel,
        dashboard_surface=dashboard_surface,
        kanban_worker=kanban_worker,
        kanban_web_act=kanban_web_act,
        kanban_allow_shell=kanban_allow_shell,
        recent_user_text=(recent_user_text or "")[:4000],
    )


def is_non_foreground_personal_context(ctx: ToolPolicyContext | None = None) -> bool:
    """True when direct personal-model / USER.md patches must be refused.

    Matches the User Model system-prompt injection gate in ``agent_loop_run``:
    only real foreground interactive chat may mutate personal facts directly.
    """
    if ctx is None:
        ctx = get_current_policy_context()
    if ctx is None:
        return False
    return bool(
        ctx.subagent_mode
        or ctx.background_automation
        or ctx.scheduled_job
        or ctx.idle_tick
        or ctx.messaging
        or ctx.mesh_sourced
        or ctx.hooks_triggered
        or ctx.unattended_external
    )


def is_background_personal_proposal_context(ctx: ToolPolicyContext | None = None) -> bool:
    """True when even proposal creation should be discouraged (idle/scheduled/background)."""
    if ctx is None:
        ctx = get_current_policy_context()
    if ctx is None:
        return True
    return bool(ctx.background_automation or ctx.scheduled_job or ctx.idle_tick)


# Tools that must not be offered in the mesh *inbound* subagent tool schema: they
# are also blocked at invoke time when ``mesh_sourced`` (relay / fan-out). Keeping
# them out of the schema avoids wasted turns and policy-error prose in replies.
MESH_SOURCED_RELAY_BLOCKED_TOOL_NAMES: frozenset[str] = frozenset(
    {"contact_peer", "mesh_send_files", "delegate_task", "run_in_background"}
)

# Peer-initiated mesh turns: deny filesystem, network, shell, memory writes, etc.
_MESH_SOURCED_BLOCKED_CAPS: Cap = (
    Cap.READ_FS
    | Cap.WRITE_FS
    | Cap.NETWORK
    | Cap.SHELL_LIKE
    | Cap.SUBPROCESS_OTHER
    | Cap.GIT_MUTATE
    | Cap.MEMORY_WRITE
    | Cap.SCREEN_CAPTURE
    | Cap.EMAIL_READ
    | Cap.EMAIL_COMPOSE
    | Cap.CLIPBOARD_WRITE
    | Cap.SCHEDULE_MUTATE
    | Cap.PLUGIN_OR_WORKSPACE_MUTATE
    | Cap.MCP_EXTERNAL
    | Cap.BROWSER_UI_AUTOMATION
    | Cap.REMINDER_MUTATE
)


# ---------------------------------------------------------------------------
# The gate — single function all invoke paths must call
# ---------------------------------------------------------------------------
def policy_gate_tool_call(
    tool_name: str,
    args: dict[str, Any] | None = None,
    ctx: ToolPolicyContext | None = None,
) -> str | None:
    """Check whether *tool_name* is allowed under the current policy.

    Returns ``None`` if the call is allowed, or a human-readable blocked
    message (prefixed with ``POLICY_BLOCKED_PREFIX``) if denied.

    Gate evaluation order (short-circuits on first denial):
      1. plan_mode blocks tools with ``_PLAN_MODE_BLOCKED`` capabilities
      2. allow_shell_execute blocks ``SHELL_LIKE`` tools
      3. unattended_external blocks high-risk mutation/execution tools
      4. mesh_sourced blocks relay/fan-out tools to prevent worm behavior
    """
    if ctx is None:
        ctx = get_current_policy_context() or context_from_prefs_and_flags()

    tool_name = str(tool_name or "").strip()
    caps = get_capabilities(tool_name)

    # 1. Plan mode
    if ctx.plan_mode and (caps & _PLAN_MODE_BLOCKED):
        blocked_caps = caps & _PLAN_MODE_BLOCKED
        msg = (
            f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' is not available in "
            f"plan mode (requires: {_cap_names(blocked_caps)}). "
            f"Describe what you would do instead."
        )
        _emit_ndjson("policy_denied", tool_name, {"reason": "plan_mode", "caps": _cap_names(blocked_caps)})
        return msg

    # 2. Shell pref
    if not ctx.allow_shell_execute and (caps & Cap.SHELL_LIKE):
        # Unknown tools use _UNKNOWN_TOOL_DEFAULT (includes SHELL_LIKE). Same UX as a
        # plugin tool whose manifest sets shell=true, but the diagnosis differs.
        if caps == _UNKNOWN_TOOL_DEFAULT:
            extra = ""
            tn = tool_name.lower()
            if tn.startswith("neko_") or tn in ("open_neko_dashboard", "get_neko_room_link", "install_neko", "open_url_in_neko"):
                extra = (
                    " This often means ~/.mugi/plugins/neko/plugin.py is older than plugin.toml or the template: "
                    "copy templates/plugin_examples/neko/plugin.py (and plugin.toml) from the Mugi repo, then "
                    "Settings → Plugins → Reload (re-approve if prompted)."
                )
            msg = (
                f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' is not registered in the "
                "policy layer (unload/reload plugins, re-approve after edits, or check "
                "the tool name matches the plugin). While shell access is disabled, "
                "unrecognized tools are blocked."
                + extra
            )
        else:
            msg = (
                f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' requires shell "
                f"access, which is disabled in preferences."
            )
        _emit_ndjson("policy_denied", tool_name, {"reason": "shell_disabled"})
        return msg

    # 3. unattended external automation guard
    if ctx.unattended_external and (caps & _UNATTENDED_EXTERNAL_BLOCKED):
        blocked_caps = caps & _UNATTENDED_EXTERNAL_BLOCKED
        # Explicit operator opt-in for shared-browser automation. The env var
        # exempts only the BROWSER_UI_AUTOMATION bit; any other blocked cap
        # (SHELL_LIKE, GIT_MUTATE, …) still denies the call.
        if (blocked_caps & Cap.BROWSER_UI_AUTOMATION) and _unattended_browser_opt_in():
            blocked_caps &= ~Cap.BROWSER_UI_AUTOMATION
        if blocked_caps:
            msg = (
                f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' is blocked for unattended external runs "
                f"(requires: {_cap_names(blocked_caps)})."
            )
            if caps & Cap.BROWSER_UI_AUTOMATION and not _unattended_browser_opt_in():
                msg += (
                    f" Set {_UNATTENDED_BROWSER_ENV}=1 on the Mugi backend process to allow"
                    " shared-browser automation from webhooks/cron; see docs/harness_threat_model.md."
                )
            _emit_ndjson(
                "policy_denied",
                tool_name,
                {"reason": "unattended_external", "caps": _cap_names(blocked_caps)},
            )
            return msg

    if tool_name in _WEB_ACT_TOOL_NAMES and (
        ctx.unattended_external
        or ctx.background_automation
        or ctx.scheduled_job
        or ctx.idle_tick
    ):
        kanban_web_act_ok = False
        if ctx.kanban_web_act and ctx.kanban_worker:
            try:
                from tools.web_act import web_act_enabled

                kanban_web_act_ok = web_act_enabled()
            except Exception:
                kanban_web_act_ok = False
        if not kanban_web_act_ok and not _unattended_web_act_opt_in():
            msg = (
                f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' is blocked for unattended or "
                "background runs (web-act scripts execute arbitrary Python with network access)."
            )
            msg += (
                f" Set {_UNATTENDED_WEB_ACT_ENV}=1 on the Mugi backend process to allow; "
                "see docs/harness_threat_model.md."
            )
            _emit_ndjson("policy_denied", tool_name, {"reason": "unattended_web_act"})
            return msg

    # 4. mesh-sourced safety guard (prevent relay / recursive fan-out).
    if ctx.mesh_sourced and tool_name in MESH_SOURCED_RELAY_BLOCKED_TOOL_NAMES:
        msg = (
            f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' is blocked for mesh-sourced runs "
            "to prevent relay/fan-out behavior."
        )
        _emit_ndjson("policy_denied", tool_name, {"reason": "mesh_sourced_relay_block"})
        return msg

    # 4b. mesh-sourced: block high-risk capabilities on peer-initiated turns.
    if ctx.mesh_sourced:
        blocked_mesh = caps & _MESH_SOURCED_BLOCKED_CAPS
        if blocked_mesh:
            msg = (
                f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' is blocked for mesh-sourced runs "
                f"(requires: {_cap_names(blocked_mesh)})."
            )
            _emit_ndjson(
                "policy_denied",
                tool_name,
                {"reason": "mesh_sourced_capability_block", "caps": _cap_names(blocked_mesh)},
            )
            return msg

    # 5. idle_tick guard: prevent schedule_task during continuity ticks to avoid
    #    unbounded cron job creation from idle-initiated runs.
    if ctx.idle_tick and tool_name == "schedule_task":
        msg = (
            f"{POLICY_BLOCKED_PREFIX} Tool '{tool_name}' is not available during idle "
            "continuity runs. Ask the user to schedule tasks directly."
        )
        _emit_ndjson("policy_denied", tool_name, {"reason": "idle_tick_schedule_block"})
        return msg

    # 6. Background automation (heartbeat, drive cycles, consolidation): git restore of
    #    workspace personality files discards uncommitted user fixes — only interactive
    #    chat should run git_project_restore on these paths.
    if ctx.background_automation and tool_name == "git_project_restore":
        _hit = _git_restore_paths_hit_protected_workspace_files((args or {}).get("paths"))
        if _hit:
            msg = (
                f"{POLICY_BLOCKED_PREFIX} Tool 'git_project_restore' cannot target workspace "
                f"personality or identity files during background runs ({', '.join(_hit)}). "
                "Restore matches the last commit and throws away uncommitted edits. "
                "Ask the user to discard changes from chat if they intend that, or commit first."
            )
            _emit_ndjson(
                "policy_denied",
                tool_name,
                {"reason": "background_git_restore_identity_block", "paths": _hit},
            )
            return msg

    return None


# Workspace-root markdown files where git restore during automation is unsafe (align with
# tools/file_ops._EVOLUTION_COMMIT_WHITELIST + BOOTSTRAP.md).
_GIT_RESTORE_PROTECTED_BASE_NAMES: frozenset[str] = frozenset(
    {
        "PERSONA.MD",
        "IDENTITY.MD",
        "USER.MD",
        "SOUL.MD",
        "MEMORY.MD",
        "JOURNAL.MD",
        "HEARTBEAT.MD",
        "TOOLS.MD",
        "AGENTS.MD",
        "SESSION-STATE.MD",
        "CONSTITUTION.MD",
        "BOOTSTRAP.MD",
    }
)


def _git_restore_paths_hit_protected_workspace_files(paths: Any) -> list[str]:
    """Return sorted unique paths/basenames that must not be restored during automation."""
    if not isinstance(paths, list):
        return []
    hit: list[str] = []
    seen: set[str] = set()
    for p in paths:
        if not isinstance(p, str):
            continue
        s = p.strip()
        if not s:
            continue
        norm = s.replace("\\", "/")
        lower = norm.lower()
        if "memory/user_model/" in lower:
            label = "memory/user_model/"
            if label not in seen:
                seen.add(label)
                hit.append(label)
            continue
        base = os.path.basename(s.rstrip("/"))
        if not base:
            continue
        if base.upper() in _GIT_RESTORE_PROTECTED_BASE_NAMES and base not in seen:
            seen.add(base)
            hit.append(base)
    return sorted(hit)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _cap_names(caps: Cap) -> str:
    """Human-readable comma-separated capability names."""
    return ", ".join(c.name for c in Cap if c in caps)


# ---------------------------------------------------------------------------
# Optional NDJSON observability log
# ---------------------------------------------------------------------------
_NDJSON_LOG_PATH: str | None = None

_CURRENT_POLICY_CONTEXT: ContextVar[ToolPolicyContext | None] = ContextVar(
    "mugi_current_policy_context",
    default=None,
)


def set_current_policy_context(ctx: ToolPolicyContext | None) -> Token:
    """Set the current run policy context in a ContextVar."""
    return _CURRENT_POLICY_CONTEXT.set(ctx)


def reset_current_policy_context(token: Token) -> None:
    """Reset ContextVar to previous value."""
    _CURRENT_POLICY_CONTEXT.reset(token)


def get_current_policy_context() -> ToolPolicyContext | None:
    """Return ContextVar policy context (if set for this run/thread)."""
    return _CURRENT_POLICY_CONTEXT.get()


def _init_ndjson_log() -> str | None:
    """Return the NDJSON log path if enabled (``~/.mugi/logs/harness_policy.ndjson``)."""
    global _NDJSON_LOG_PATH
    if _NDJSON_LOG_PATH is not None:
        return _NDJSON_LOG_PATH if _NDJSON_LOG_PATH != "" else None
    try:
        prefs = _load_prefs()
        if not prefs.get("harness_policy_log_enabled", False):
            _NDJSON_LOG_PATH = ""
            return None
        logs_dir = os.path.expanduser("~/.mugi/logs")
        os.makedirs(logs_dir, exist_ok=True)
        _NDJSON_LOG_PATH = os.path.join(logs_dir, "harness_policy.ndjson")
        return _NDJSON_LOG_PATH
    except Exception:
        _NDJSON_LOG_PATH = ""
        return None


def _emit_ndjson(event_type: str, tool_name: str, extra: dict[str, Any] | None = None) -> None:
    """Append a single NDJSON line to the policy log (best-effort, no stdout)."""
    import json
    import time as _time

    path = _init_ndjson_log()
    if not path:
        return
    entry: dict[str, Any] = {
        "ts": _time.time(),
        "event": event_type,
        "tool": tool_name,
    }
    if extra:
        entry.update(extra)
    try:
        with open(path, "a") as f:
            f.write(json.dumps(entry, default=str) + "\n")
    except Exception:
        pass


def emit_policy_event(event_type: str, tool_name: str, extra: dict[str, Any] | None = None) -> None:
    """Best-effort NDJSON emission for policy-adjacent observability events."""
    _emit_ndjson(event_type, tool_name, extra)


def _load_prefs() -> dict[str, Any]:
    try:
        from utils.app_prefs_store import load_app_prefs_cached
        return load_app_prefs_cached()
    except Exception:
        return {}
