from __future__ import annotations

import html
import importlib.util
import json
import re
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

from agentic_swmm.agent import mcp_cache, mcp_client
from agentic_swmm.agent.mcp_client import McpClientError
from agentic_swmm.agent.mcp_pool import ensure_session_pool
from agentic_swmm.agent.permissions import is_allowed_write_path, is_evidence_path
from agentic_swmm.agent.policy import capability_summary
from agentic_swmm.agent.types import ToolCall
from agentic_swmm.commands.plot import DEFAULT_NODE_ATTR, NODE_ATTRIBUTE_CHOICES, NODE_ATTRIBUTE_LABELS, _find_inp, _find_out, _read_manifest, rainfall_timeseries_options
from agentic_swmm.config import runtime_state_path
from agentic_swmm.providers.base import ProviderToolCall
from agentic_swmm.runtime.registry import discover_skills, load_mcp_registry
from agentic_swmm.utils.paths import repo_root, script_path
from agentic_swmm.utils.subprocess_runner import runtime_env


# CONCURRENCY-OWNER: PRD-GF-CORE
#
# ``supports_gap_fill`` and ``required_file_args`` are the two gap-fill
# (PRD-GF-CORE) hooks on every ToolSpec. Both default to fail-safe
# values so existing tools without explicit opt-in keep their pre-PRD
# behaviour. The handler-wrapping logic that actually intercepts
# ``gap_signal`` lives in ``agentic_swmm.agent.runtime_loop.invoke_tool_with_gap_fill``;
# this file only declares the dataclass fields and routes ``execute``
# through the wrapper.
@dataclass(frozen=True)
class ToolSpec:
    name: str
    description: str
    parameters: dict[str, Any]
    handler: Callable[[ToolCall, Path], dict[str, Any]]
    # PRD_runtime: ``is_read_only=True`` lets ``Profile.QUICK`` auto-approve
    # the tool without prompting. Default ``False`` is fail-safe — a new
    # tool only joins the auto-approve set when its author explicitly
    # marks it read-only.
    is_read_only: bool = False
    # CONCURRENCY-OWNER: PRD-GF-CORE
    # ``supports_gap_fill=True`` opts the tool into the gap-fill state
    # machine. The runtime then:
    #   * runs the pre-flight L1 scanner over ``required_file_args``
    #     before each invocation,
    #   * intercepts ``{"ok": false, "gap_signal": {...}}`` results
    #     and routes them through the proposer/UI/recorder.
    # Default ``False`` is fail-safe: legacy tools without the flag
    # raise on missing inputs exactly as they did pre-PRD.
    supports_gap_fill: bool = False
    # The tuple of argument names that point at files which must
    # exist before the tool runs. Used by the L1 pre-flight scanner.
    # Empty tuple disables pre-flight scanning (the tool then only
    # surfaces L3 gaps via in-band ``gap_signal``).
    required_file_args: tuple[str, ...] = ()

    def schema(self) -> dict[str, Any]:
        return {"type": "function", "name": self.name, "description": self.description, "parameters": self.parameters}


class AgentToolRegistry:
    def __init__(self) -> None:
        self._tools = _build_tools()

    @property
    def names(self) -> set[str]:
        return set(self._tools)

    def sorted_names(self) -> list[str]:
        return sorted(self._tools)

    def schemas(self) -> list[dict[str, Any]]:
        return [self._tools[name].schema() for name in sorted(self._tools)]

    def validate(self, call: ProviderToolCall) -> ToolCall:
        if call.name not in self._tools:
            raise ValueError(f"planner requested unsupported tool: {call.name}")
        return ToolCall(call.name, dict(call.arguments))

    def execute(self, call: ToolCall, session_dir: Path) -> dict[str, Any]:
        spec = self._tools.get(call.name)
        if spec is None:
            return {"tool": call.name, "args": call.args, "ok": False, "summary": f"unsupported tool: {call.name}"}
        # CONCURRENCY-OWNER: PRD-GF-CORE
        # Tools with ``supports_gap_fill=True`` are wrapped through the
        # detect/propose/record/retry state machine. Tools without the
        # opt-in flag bypass the wrapper entirely so existing behaviour
        # is untouched. The wrapper itself is a no-op when neither L1
        # pre-flight nor in-band ``gap_signal`` is in play, so the
        # branch below is cheap.
        if getattr(spec, "supports_gap_fill", False):
            from agentic_swmm.agent.runtime_loop import invoke_tool_with_gap_fill

            return invoke_tool_with_gap_fill(
                spec, call, session_dir, lambda c, sd: spec.handler(c, sd)
            )
        return spec.handler(call, session_dir)

    def is_read_only(self, name: str) -> bool:
        """Return whether ``name`` is a read-only tool.

        Unknown tools fall through to ``False`` — fail-safe.
        """
        spec = self._tools.get(name)
        if spec is None:
            return False
        return bool(spec.is_read_only)

    def describe(self, name: str) -> str | None:
        """Return the ``ToolSpec.description`` for ``name``, or ``None``.

        Used by the UX-3 tool spinner (issue #58) to show the first
        sentence of the description next to the running tool name.
        Unknown tools return ``None`` so callers can fall back to the
        bare tool name.
        """
        spec = self._tools.get(name)
        if spec is None:
            return None
        return spec.description

    def output_for_model(self, result: dict[str, Any]) -> dict[str, Any]:
        allowed_keys = {
            "tool",
            "args",
            "ok",
            "return_code",
            "summary",
            "stdout_tail",
            "stderr_tail",
            "path",
            "chars",
            "excerpt",
            "results",
            "servers",
            "tools",
            "mapped_tools",
            "capabilities",
            "recovery",
            "fallback_tools",
            "node_suggestions",
            # PRD-Z `request_expert_review` adds two fields the planner
            # needs to see: ``approved`` (Y/N answer) and ``decision_id``
            # (the ID of the human_decisions record).
            "approved",
            "decision_id",
            # PRD-Y: ``select_skill`` returns the skill's tool subset; the
            # planner needs to see both the tool list and the bound name.
            "skill_name",
            "source",
            # CONCURRENCY-OWNER: PRD-GF-L5
            # ``request_gap_judgement`` returns ``resume_mode=llm_replan``
            # plus ``gap_kind`` so the planner's replan-injection branch
            # can recognise the L5 result and pull the recorded decision
            # into the next LLM turn as a user_clarification message.
            "resume_mode",
            "gap_kind",
        }
        return {key: value for key, value in result.items() if key in allowed_keys}


def _object(properties: dict[str, Any], required: list[str] | None = None) -> dict[str, Any]:
    schema: dict[str, Any] = {"type": "object", "properties": properties, "additionalProperties": False}
    if required:
        schema["required"] = required
    return schema


# ---------------------------------------------------------------------------
# MCP-routed handler factory (PRD-Y "Handler rewrite — uniform pattern")
# ---------------------------------------------------------------------------
#
# Every deterministic-SWMM ToolSpec's handler is built from this factory so
# the audit trail of "agent → skill → MCP → Python script" is one
# transport. The factory itself is intentionally small: per-tool argument
# mapping (snake_case → camelCase, default-node detection, output-dir
# resolution) lives in dedicated ``_*_mcp_args`` builders below so we keep
# the handler body uniform.


def _make_mcp_routed_handler(
    server: str,
    tool: str,
    *,
    args_mapper: Callable[[ToolCall, Path], dict[str, Any] | dict[str, Any]] | None = None,
) -> Callable[[ToolCall, Path], dict[str, Any]]:
    """Build a ToolSpec handler that forwards the call through ``MCPPool``.

    ``args_mapper`` is an optional pre-call hook that may:
    * translate ToolSpec snake_case argument names into the MCP server's
      camelCase property names,
    * resolve relative paths / inject defaults (e.g. node auto-detect),
    * return a fail-soft result dict early when validation fails — that
      dict is returned verbatim so handlers behave the same way the
      old in-process subprocess handlers did when args were bad.

    The handler returns a flat ``{tool, args, ok, results, summary}``
    dict shaped like the historical subprocess handlers, so existing
    planner / reporting code does not need updating.
    """

    def handler(call: ToolCall, session_dir: Path) -> dict[str, Any]:
        if args_mapper is None:
            mcp_args: dict[str, Any] = dict(call.args)
        else:
            mapped = args_mapper(call, session_dir)
            if isinstance(mapped, dict) and mapped.get("ok") is False and "summary" in mapped:
                # ``_failure``-shaped early return — surface it unchanged.
                return mapped
            mcp_args = mapped if isinstance(mapped, dict) else {}
        pool = ensure_session_pool()
        if pool is None:
            return {
                "tool": call.name,
                "args": call.args,
                "ok": False,
                "summary": (
                    f"MCP transport unavailable for {server}.{tool}: "
                    "no MCP server registry configured. "
                    "Run: bash scripts/install_mcp_deps.sh (or aiswmm setup --install-mcp)."
                ),
            }
        try:
            result = pool.call_tool(server, tool, mcp_args)
        except McpClientError as exc:
            return {
                "tool": call.name,
                "args": call.args,
                "ok": False,
                "summary": f"MCP transport failed: {exc}",
            }
        return _wrap_mcp_result(call, server, tool, result)

    # Synthetic introspection attribute — the lock-in test
    # ``tests/test_handler_lockin_no_direct_subprocess.py`` reads this to
    # verify that every deterministic-SWMM ToolSpec handler is built via
    # this factory and not a legacy subprocess shim.
    handler._mcp_routing = {"server": server, "tool": tool}  # type: ignore[attr-defined]
    return handler


def _wrap_mcp_result(
    call: ToolCall,
    server: str,
    tool: str,
    result: dict[str, Any],
) -> dict[str, Any]:
    """Convert the raw MCP ``tools/call`` result into a ToolSpec response.

    The MCP server returns a JSON-RPC ``result`` object — usually with a
    ``content`` array of text blocks. We pass the body through under the
    ``results`` key, and synthesise an ``excerpt`` from the joined text
    blocks so existing reporting code that reads ``stdout_tail`` /
    ``excerpt`` still surfaces useful context to the user.
    """

    excerpt = ""
    content = result.get("content") if isinstance(result, dict) else None
    if isinstance(content, list):
        chunks = []
        for item in content:
            if isinstance(item, dict) and item.get("type") == "text":
                text = str(item.get("text") or "")
                if text:
                    chunks.append(text)
        excerpt = "\n".join(chunks)[:4000]
    summary = f"called {server}.{tool}"
    return {
        "tool": call.name,
        "args": call.args,
        "ok": True,
        "results": result,
        "excerpt": excerpt,
        "summary": summary,
    }


def _build_tools() -> dict[str, ToolSpec]:
    specs = [
        ToolSpec("audit_run", "Audit a run directory and write deterministic provenance/comparison/note artifacts.", _object({"run_dir": {"type": "string"}, "workflow_mode": {"type": "string"}, "objective": {"type": "string"}}, ["run_dir"]), _audit_run_tool),
        ToolSpec("apply_patch", "Apply a unified diff patch to repository files. Writes are repo-only and blocked for .git/.venv/secret paths.", _object({"patch": {"type": "string"}, "allow_evidence_edits": {"type": "boolean"}}, ["patch"]), _apply_patch_tool),
        ToolSpec("build_inp", "Assemble a SWMM INP from explicit CSV/JSON/text inputs using the swmm-builder skill.", _object({"subcatchments_csv": {"type": "string"}, "params_json": {"type": "string"}, "network_json": {"type": "string"}, "rainfall_json": {"type": "string"}, "raingage_json": {"type": "string"}, "timeseries_text": {"type": "string"}, "config_json": {"type": "string"}, "default_gage_id": {"type": "string"}, "out_inp": {"type": "string"}, "out_manifest": {"type": "string"}}, ["subcatchments_csv", "params_json", "network_json", "out_inp", "out_manifest"]), _build_inp_tool),
        ToolSpec("capabilities", "Describe what this runtime can and cannot access.", _object({}), _capabilities_tool, is_read_only=True),
        ToolSpec("demo_acceptance", "Run the prepared acceptance demo through the Agentic SWMM CLI.", _object({"run_id": {"type": "string"}, "keep_existing": {"type": "boolean"}}), _demo_acceptance_tool),
        ToolSpec("doctor", "Run the built-in Agentic SWMM runtime doctor.", _object({}), _doctor_tool),
        ToolSpec("format_rainfall", "Format rainfall CSV into SWMM TIMESERIES text and metadata JSON using the swmm-climate skill.", _object({"input_csv": {"type": "string"}, "out_json": {"type": "string"}, "out_timeseries": {"type": "string"}, "series_name": {"type": "string"}, "timestamp_column": {"type": "string"}, "value_column": {"type": "string"}, "value_units": {"type": "string"}, "unit_policy": {"type": "string", "enum": ["strict", "convert_to_mm_per_hr"]}, "timestamp_policy": {"type": "string", "enum": ["strict", "sort"]}}, ["input_csv", "out_json", "out_timeseries"]), _format_rainfall_tool),
        ToolSpec("git_diff", "Read the current repository diff or diff stat.", _object({"stat_only": {"type": "boolean"}, "path": {"type": "string"}}), _git_diff_tool, is_read_only=True),
        ToolSpec("inspect_plot_options", "Inspect a run directory or INP file and return selectable rainfall series, nodes, and node output attributes for plotting.", _object({"run_dir": {"type": "string"}, "inp_path": {"type": "string"}, "out_file": {"type": "string"}}, []), _inspect_plot_options_tool, is_read_only=True),
        ToolSpec("list_dir", "List a repository directory.", _object({"path": {"type": "string"}}), _list_dir_tool, is_read_only=True),
        ToolSpec("list_mcp_servers", "List configured local MCP servers.", _object({}), _list_mcp_servers_tool, is_read_only=True),
        ToolSpec("list_mcp_tools", "List tools exposed by one configured MCP server.", _object({"server": {"type": "string"}, "timeout_seconds": {"type": "integer"}, "refresh": {"type": "boolean"}, "cache_ttl_seconds": {"type": "integer"}}, ["server"]), _list_mcp_tools_tool, is_read_only=True),
        ToolSpec("call_mcp_tool", "Call a tool exposed by a configured local MCP server.", _object({"server": {"type": "string"}, "tool": {"type": "string"}, "arguments": {"type": "object"}}, ["server", "tool"]), _call_mcp_tool_tool),
        ToolSpec("list_skills", "List available repository skills.", _object({}), _list_skills_tool, is_read_only=True),
        ToolSpec("network_qa", "Validate a SWMM network JSON using the swmm-network QA script.", _object({"network_json": {"type": "string"}, "report_json": {"type": "string"}}, ["network_json"]), _network_qa_tool),
        ToolSpec("network_to_inp", "Export a SWMM network JSON to INP section text using the swmm-network script.", _object({"network_json": {"type": "string"}, "out_path": {"type": "string"}}, ["network_json", "out_path"]), _network_to_inp_tool),
        ToolSpec("plot_run", "Create a rainfall-runoff plot from a run directory using selected rainfall series, node, and node output attribute.", _object({"run_dir": {"type": "string"}, "node": {"type": "string"}, "node_attr": {"type": "string"}, "rain_ts": {"type": "string"}, "rain_kind": {"type": "string", "enum": ["intensity_mm_per_hr", "depth_mm_per_dt", "cumulative_depth_mm"]}, "out_png": {"type": "string"}}, ["run_dir"]), _plot_run_tool),
        ToolSpec("read_file", "Read a repository file and return a bounded excerpt.", _object({"path": {"type": "string"}}, ["path"]), _read_file_tool, is_read_only=True),
        ToolSpec("read_skill", "Read a skill contract from skills/<skill_name>/SKILL.md.", _object({"skill_name": {"type": "string"}}, ["skill_name"]), _read_skill_tool, is_read_only=True),
        ToolSpec(
            "recall_memory",
            (
                "Look up the lesson section for an exact failure_pattern name "
                "from memory/modeling-memory/lessons_learned.md.\n"
                "USE WHEN: you know the exact failure_pattern name (e.g. "
                "'peak_flow_parse_missing') and want a precise lookup.\n"
                "DO NOT USE WHEN: user is chatting, or the question is general "
                "(prefer recall_memory_search)."
            ),
            _object({"pattern": {"type": "string"}}, ["pattern"]),
            _recall_memory_tool,
            is_read_only=True,
        ),
        ToolSpec(
            "recall_memory_search",
            (
                "Retrieve the top-k most similar historical entries from the "
                "RAG corpus (memory/rag-memory/) for a natural-language query.\n"
                "USE WHEN: you have a natural-language question or do not know "
                "the failure_pattern name. Returns up to top-k entries with "
                "run_id, source_path, case_name, score, and matched_terms.\n"
                "DO NOT USE WHEN: you have an exact pattern name (prefer "
                "recall_memory) or the question is unrelated to past runs."
            ),
            _object(
                {"query": {"type": "string"}, "top_k": {"type": "integer"}},
                ["query"],
            ),
            _recall_memory_search_tool,
            is_read_only=True,
        ),
        ToolSpec(
            "recall_session_history",
            (
                "Search prior chat sessions in the SQLite session store for relevant past work.\n"
                "USE WHEN: user mentions '上次/昨天/上周/before/previously/continue', or you need "
                "to check whether a similar question / failure pattern has been encountered before.\n"
                "DO NOT USE WHEN: question has no temporal cue and current-session context is sufficient."
            ),
            _object(
                {
                    "query": {"type": "string"},
                    "case_name": {"type": "string"},
                    "limit": {"type": "integer"},
                },
                ["query"],
            ),
            _recall_session_history_tool,
            is_read_only=True,
        ),
        ToolSpec(
            "record_fact",
            (
                "Append a candidate project fact to the staging file for later user review.\n"
                "USE WHEN: user just expressed a durable preference, project convention, or "
                "confirmed fix recipe that future sessions should remember.\n"
                "DO NOT USE WHEN: ephemeral state, file path, secret, or anything you are not "
                "certain the user wants persisted."
            ),
            _object(
                {"text": {"type": "string"}, "source_session_id": {"type": "string"}},
                ["text"],
            ),
            _record_fact_tool,
            is_read_only=False,
        ),
        ToolSpec(
            "request_expert_review",
            (
                "Pause the agent and request expert review.\n"
                "USE WHEN: a QA threshold has been crossed and a "
                "hydrologically consequential decision must be human-approved "
                "before continuing. Pattern must match one of the documented "
                "HITL thresholds (see docs/hitl-thresholds.md).\n"
                "DO NOT USE WHEN: low-stakes confirmation or routine reasoning."
            ),
            _object(
                {
                    "run_dir": {"type": "string"},
                    "pattern": {"type": "string"},
                    "evidence_ref": {"type": "string"},
                    "message": {"type": "string"},
                },
                ["run_dir", "pattern", "evidence_ref", "message"],
            ),
            _request_expert_review_tool,
            # is_read_only=False — QUICK profile must NEVER auto-approve
            # the HITL pause (PRD-Z hard requirement).
            is_read_only=False,
        ),
        # CONCURRENCY-OWNER: PRD-GF-L5
        # L5 subjective judgement entry point. The LLM invokes this
        # tool explicitly when it identifies a hydrological choice that
        # has no single right answer (pour point, storm event, metric
        # weighting, …). The handler:
        #   1. asks the LLM to enumerate N candidates with each one's
        #      tradeoff cited (``llm_enumerator``),
        #   2. shows the per-gap pause UI (``ui_per_gap``),
        #   3. records an L5 ``GapDecision`` via the gap-fill recorder.
        # ``supports_gap_fill=False`` because L5 does *not* travel the
        # L1/L3 ``gap_signal`` interception path — the agent calls the
        # tool directly. ``is_read_only=False`` because judgement must
        # never be auto-approved by the QUICK profile.
        ToolSpec(
            "request_gap_judgement",
            (
                "Request a subjective hydrological judgement from the human "
                "expert with enumerated candidates.\n"
                "USE WHEN: a hydrological choice has no single right answer "
                "(pour point ambiguity, storm event selection from a "
                "calibration window, metric weighting, continuity tolerance "
                "deviation). The LLM enumerator will list candidates with "
                "each one's tradeoff cited; the modeller picks one with an "
                "optional free-form note; the planner re-plans on the next "
                "turn with the decision as a user_clarification message.\n"
                "DO NOT USE WHEN: a missing path or parameter value can be "
                "proposed deterministically (those flow through the L1/L3 "
                "gap-fill path automatically). For free-form pauses without "
                "structured candidates, use request_expert_review."
            ),
            _object(
                {
                    "gap_kind": {
                        "type": "string",
                        "enum": [
                            "pour_point",
                            "storm_event_selection",
                            "metric_weighting",
                            "continuity_tolerance",
                        ],
                    },
                    "context": {"type": "object"},
                    "evidence_ref": {"type": "string"},
                },
                ["gap_kind", "context", "evidence_ref"],
            ),
            _request_gap_judgement_tool,
            # is_read_only=False — judgement must never be auto-approved.
            is_read_only=False,
            # supports_gap_fill=False — L5 is a separate mechanism from
            # the L1/L3 ``gap_signal`` interception path. The agent
            # invokes this tool explicitly; the runtime does not wrap
            # it with the GF-CORE state machine.
            supports_gap_fill=False,
        ),
        ToolSpec("run_swmm_inp", "Run a repository or imported external .inp file through the constrained swmm-runner CLI wrapper.", _object({"inp_path": {"type": "string"}, "run_id": {"type": "string"}, "run_dir": {"type": "string"}, "node": {"type": "string"}}, ["inp_path"]), _run_swmm_inp_tool),
        ToolSpec("run_allowed_command", "Run an allowlisted local command such as pytest, python -m agentic_swmm.cli, node scripts/*.mjs, or swmm5.", _object({"command": {"type": "array", "items": {"type": "string"}}, "timeout_seconds": {"type": "integer"}}, ["command"]), _run_allowed_command_tool),
        ToolSpec("run_tests", "Run pytest on selected repository test paths.", _object({"paths": {"type": "array", "items": {"type": "string"}}, "timeout_seconds": {"type": "integer"}}), _run_tests_tool),
        ToolSpec("search_files", "Search text files in the repository.", _object({"query": {"type": "string"}, "glob": {"type": "string"}, "max_results": {"type": "integer"}}), _search_files_tool, is_read_only=True),
        ToolSpec(
            "select_skill",
            (
                "Commit to a workflow skill and receive its full tool list.\n"
                "USE WHEN: you are about to invoke a deterministic SWMM operation "
                "and have identified which skill provides it (e.g. swmm-builder, "
                "swmm-runner, swmm-plot). The response gives you the skill's "
                "tools (name + description + parameters); pick one and call it next.\n"
                "DO NOT USE WHEN: you only need agent-internal tools (memory recall, "
                "workflow mode selection, plot option inspection, file / dir / git "
                "inspection). Those are always available without selecting a skill."
            ),
            _object({"skill_name": {"type": "string"}}, ["skill_name"]),
            _select_skill_tool,
            is_read_only=True,
        ),
        ToolSpec("select_workflow_mode", "Select the top-level swmm-end-to-end operating mode and report required/missing inputs before running tools. OPTIONAL but recommended: pass `mode` with your identified workflow mode (one of calibration / uncertainty / prepared_inp_cli / full_modular_build / existing_run_plot / audit_only_or_comparison / prepared_demo) so the tool uses your classification directly instead of re-deriving intent via legacy keyword matching.", _object({"goal": {"type": "string"}, "mode": {"type": "string", "enum": ["calibration", "uncertainty", "prepared_inp_cli", "full_modular_build", "existing_run_plot", "audit_only_or_comparison", "prepared_demo"], "description": "OPTIONAL but recommended. The workflow mode you have identified from the user's goal. If provided and valid, the tool uses this directly. If absent or invalid, falls back to keyword matching (legacy compatibility)."}, "inp_path": {"type": "string"}, "run_dir": {"type": "string"}, "node": {"type": "string"}, "network_json": {"type": "string"}, "subcatchments_csv": {"type": "string"}, "rainfall_input": {"type": "string"}, "landuse_input": {"type": "string"}, "soil_input": {"type": "string"}, "observed_flow": {"type": "string"}, "fuzzy_config": {"type": "string"}, "baseline_run_dir": {"type": "string"}}, ["goal"]), _select_workflow_mode_tool, is_read_only=True),
        ToolSpec("summarize_memory", "Summarize audited runs into the modeling-memory directory.", _object({"runs_dir": {"type": "string"}, "out_dir": {"type": "string"}}, ["runs_dir"]), _summarize_memory_tool),
        ToolSpec(
            "retrieve_memory",
            "Retrieve relevant audited-run memory cards for a query using the swmm-rag-memory skill's hybrid keyword/embedding retriever. Returns source-cited matches that the planner can synthesize into a grounded answer.",
            _object(
                {
                    "query": {"type": "string"},
                    "top_k": {"type": "integer"},
                    "retriever": {"type": "string", "enum": ["keyword", "hybrid"]},
                    "project": {"type": "string"},
                },
                ["query"],
            ),
            _retrieve_memory_tool,
            is_read_only=True,
        ),
        ToolSpec("web_fetch_url", "Fetch and summarize a web page. Web evidence is not SWMM run evidence.", _object({"url": {"type": "string"}, "max_chars": {"type": "integer"}}), _web_fetch_url_tool, is_read_only=True),
        ToolSpec("web_search", "Run a lightweight web search and return cited result URLs. Web evidence is not SWMM run evidence.", _object({"query": {"type": "string"}, "allowed_domains": {"type": "array", "items": {"type": "string"}}, "max_results": {"type": "integer"}}), _web_search_tool, is_read_only=True),
    ]
    return {spec.name: spec for spec in specs}


def _doctor_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    return _run_cli_tool(call, session_dir, ["doctor"])


def _request_expert_review_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Thin shim around :func:`agentic_swmm.hitl.request_expert_review`.

    The real handler lives in the ``hitl`` package so the tool wiring
    and the pause/prompt/record logic can evolve independently. The
    shim is the integration seam the registry calls through.
    """
    from agentic_swmm.hitl.request_expert_review import request_expert_review

    return request_expert_review(call, session_dir)


# CONCURRENCY-OWNER: PRD-GF-L5
def _build_default_llm_provider() -> Any:
    """Construct the LLM provider used by the L5 enumerator.

    Late import so this file stays free of the provider import in
    environments where the optional LLM extras are not installed. The
    provider seam matches what the planner uses (``respond_with_tools``).

    PRD-09: construction is routed through ``make_provider`` so the L5
    enumerator honours ``provider.default`` (``openai`` or
    ``claude_sdk``) instead of hard-coding the OpenAI backend.
    """
    from agentic_swmm.config import load_config
    from agentic_swmm.providers.factory import make_provider

    config = load_config()
    provider_name = config.get("provider.default", "openai")
    model = config.get(f"{provider_name}.model")
    if not model and provider_name == "openai":
        model = "gpt-5.5-2026-04-23"
    return make_provider(provider_name, model=model)


def _is_tty_for_l5() -> bool:
    """Return whether stdin+stdout are both TTYs (the L5 prompt seam).

    Pulled out as a helper so tests can monkeypatch it without faking
    sys.stdin.isatty across an entire process.
    """
    import sys as _sys

    return _sys.stdin.isatty() and _sys.stdout.isatty()


def _request_gap_judgement_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Handler for the ``request_gap_judgement`` ToolSpec (PRD-GF-L5).

    Routes the call through the enumerator + per-gap UI + recorder.
    The flow:

    1. Validate the three required arguments (``gap_kind``,
       ``context``, ``evidence_ref``). Missing args return
       ``{ok: false}`` with a clear summary — the planner can retry
       with corrected args.
    2. Ask the LLM to enumerate N candidates with tradeoffs cited
       via :func:`enumerate_candidates`. The call is recorded under
       ``09_audit/llm_calls.jsonl`` with ``caller=gap_fill.enumerator``.
    3. Show the per-gap pause UI via :func:`prompt_judgement`. The
       user picks one candidate and optionally supplies a free-form
       note. Non-TTY contexts raise :class:`JudgementBlocked` —
       judgement is never automated.
    4. Build an L5 :class:`GapDecision` and persist it via
       :func:`record_gap_decisions`. The decision carries
       ``resume_mode="llm_replan"`` and
       ``enumerator_llm_call_id`` for round-trip lookup.
    5. Return ``{ok, decision_id, resume_mode, gap_kind, summary}``
       so the planner's replan-injection branch can pull the decision
       into the next LLM turn.
    """
    # Import the *modules* (not the symbols) so test-side
    # ``mock.patch("agentic_swmm.gap_fill.llm_enumerator.enumerate_candidates", ...)``
    # rebinds the attribute the handler actually reads. Importing the
    # names directly would late-bind once into this function's locals
    # and ignore the patch.
    from agentic_swmm.gap_fill import (
        llm_enumerator as _llm_enumerator,
        ui_per_gap as _ui_per_gap,
    )
    from agentic_swmm.gap_fill.protocol import (
        GapDecision,
        ProposerInfo,
        new_decision_id,
        new_gap_id,
    )
    from agentic_swmm.gap_fill.recorder import record_gap_decisions

    EnumeratorParseError = _llm_enumerator.EnumeratorParseError
    JudgementBlocked = _ui_per_gap.JudgementBlocked
    JudgementDeferred = _ui_per_gap.JudgementDeferred

    gap_kind = str(call.args.get("gap_kind") or "").strip()
    evidence_ref = str(call.args.get("evidence_ref") or "").strip()
    raw_context = call.args.get("context")
    if not gap_kind:
        return _failure(call, "gap_kind is required")
    if not evidence_ref:
        return _failure(call, "evidence_ref is required")
    if not isinstance(raw_context, dict):
        return _failure(call, "context is required and must be an object")
    context = dict(raw_context)

    # Build the default provider lazily — the tests stub
    # ``enumerate_candidates`` before this attribute is even read, so
    # the OpenAI SDK never needs to be available in CI.
    try:
        provider = _build_default_llm_provider()
    except Exception:  # pragma: no cover - defensive: tests stub before this matters
        provider = None
    try:
        candidates, enumerator_call_id = _llm_enumerator.enumerate_candidates(
            gap_kind=gap_kind,
            context=context,
            evidence_ref=evidence_ref,
            n_candidates=3,
            llm_provider=provider,
            run_dir=session_dir,
        )
    except EnumeratorParseError as exc:
        return _failure(call, f"enumerator parse error: {exc}")
    except Exception as exc:  # pragma: no cover - defensive
        return _failure(call, f"enumerator failed: {exc}")

    try:
        user_pick_id, user_note = _ui_per_gap.prompt_judgement(
            gap_kind=gap_kind,
            candidates=candidates,
            evidence_ref=evidence_ref,
            llm_call_id=enumerator_call_id,
            is_tty=_is_tty_for_l5(),
        )
    except JudgementDeferred as exc:
        return {
            "tool": call.name,
            "args": dict(call.args),
            "ok": False,
            "summary": f"judgement deferred: {exc}",
        }
    except JudgementBlocked as exc:
        # L5 always blocks in headless mode — surface a clean,
        # actionable failure so a CI run aborts with intent.
        return {
            "tool": call.name,
            "args": dict(call.args),
            "ok": False,
            "summary": (
                f"L5 judgement blocked: {exc}. L5 is human-only and cannot "
                "be auto-approved or registry-resolved."
            ),
        }

    from agentic_swmm.gap_fill.protocol import _now_utc_iso  # type: ignore

    decision_id = new_decision_id()
    gap_id = new_gap_id()
    decision = GapDecision(
        decision_id=decision_id,
        gap_id=gap_id,
        severity="L5",
        # ``field`` is the L1/L3 argument-name slot. For L5 the closest
        # analogue is the gap_kind, so we store it there too — gives the
        # provenance ledger something to grep on without inventing a new
        # field shape.
        field=gap_kind,
        # Synthetic proposer info: L5 has no machine-side proposer (the
        # enumerator presents options but never chooses). ``source="human"``
        # and ``confidence="HIGH"`` reflect "the human made the call".
        proposer=ProposerInfo(
            source="human",
            confidence="HIGH",
            llm_call_id=enumerator_call_id,
        ),
        proposed_value=None,
        final_value=user_pick_id,
        proposer_overridden=False,
        decided_by="human",
        decided_at=_now_utc_iso(),
        resume_mode="llm_replan",
        human_decisions_ref=None,
        gap_kind=gap_kind,
        candidates=tuple(candidates),
        user_pick=user_pick_id,
        user_note=user_note,
        enumerator_llm_call_id=enumerator_call_id,
    )

    try:
        # GF-CORE's recorder rebuilds the GapDecision when it
        # populates ``human_decisions_ref``, which drops the L5
        # extension fields. We capture the returned (enriched) record
        # so we can restore the L5 block in the ledger below. The
        # cross-link in ``experiment_provenance.json`` is still
        # correct — only the L5-specific keys in ``gap_decisions.json``
        # need re-stitching. This keeps GF-CORE's recorder untouched.
        enriched = record_gap_decisions(session_dir, [decision])
    except Exception as exc:  # pragma: no cover - defensive
        return _failure(call, f"recorder failed: {exc}")

    _restitch_l5_fields_in_ledger(
        session_dir=session_dir,
        decision_id=enriched[0].decision_id,
        human_decisions_ref=enriched[0].human_decisions_ref,
        l5_decision=decision,
    )

    return {
        "tool": call.name,
        "args": dict(call.args),
        "ok": True,
        "decision_id": decision_id,
        "resume_mode": "llm_replan",
        "gap_kind": gap_kind,
        "summary": (
            f"L5 judgement recorded for gap_kind={gap_kind!r}; "
            f"user_pick={user_pick_id!r}"
        ),
    }


# CONCURRENCY-OWNER: PRD-GF-L5
def _restitch_l5_fields_in_ledger(
    *,
    session_dir: Path,
    decision_id: str,
    human_decisions_ref: str | None,
    l5_decision: Any,
) -> None:
    """Re-write the L5-specific fields back into ``gap_decisions.json``.

    The GF-CORE recorder rebuilds the :class:`GapDecision` when it
    populates ``human_decisions_ref``, and the rebuild drops the L5
    extension fields (``gap_kind`` / ``candidates`` / ``user_pick`` /
    ``user_note`` / ``enumerator_llm_call_id``). We avoid touching
    GF-CORE's module by post-processing the ledger here: locate the
    record by ``decision_id`` and splice the L5 block back in. The
    write goes through tmp-file + ``os.replace`` for atomicity.
    """
    import json as _json
    import os as _os
    import tempfile as _tempfile

    ledger_path = Path(session_dir) / "09_audit" / "gap_decisions.json"
    if not ledger_path.is_file():
        return
    try:
        payload = _json.loads(ledger_path.read_text(encoding="utf-8"))
    except (OSError, _json.JSONDecodeError):
        return
    if not isinstance(payload, dict):
        return
    decisions_list = payload.get("decisions")
    if not isinstance(decisions_list, list):
        return

    l5_block = {
        "gap_kind": l5_decision.gap_kind,
        "candidates": [c.to_dict() for c in l5_decision.candidates],
        "user_pick": l5_decision.user_pick,
        "user_note": l5_decision.user_note,
        "enumerator_llm_call_id": l5_decision.enumerator_llm_call_id,
    }
    found = False
    for entry in decisions_list:
        if isinstance(entry, dict) and entry.get("decision_id") == decision_id:
            entry.update(l5_block)
            # Preserve the recorder's human_decisions_ref — the
            # ``enriched`` decision already carries it, but if the
            # caller passed an override use that.
            if human_decisions_ref is not None:
                entry["human_decisions_ref"] = human_decisions_ref
            found = True
            break
    if not found:
        return

    # Atomic write — same pattern as the GF-CORE recorder.
    fd, tmp_name = _tempfile.mkstemp(
        prefix=ledger_path.name + ".",
        suffix=".tmp",
        dir=str(ledger_path.parent),
    )
    try:
        with _os.fdopen(fd, "w", encoding="utf-8") as handle:
            _json.dump(payload, handle, indent=2, sort_keys=True)
            handle.flush()
            _os.fsync(handle.fileno())
        _os.replace(tmp_name, ledger_path)
    except Exception:
        try:
            _os.unlink(tmp_name)
        except OSError:
            pass
        raise


def _apply_patch_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    patch = str(call.args.get("patch") or "")
    if not patch.strip():
        return _failure(call, "patch is required")
    touched = _patch_paths(patch)
    if not touched:
        return _failure(call, "patch did not contain recognizable file paths")
    allow_evidence = bool(call.args.get("allow_evidence_edits"))
    for path in touched:
        full = _repo_path(path)
        if full is None:
            return _failure(call, f"patch path must be inside repository: {path}")
        if not is_allowed_write_path(full):
            return _failure(call, f"patch path is blocked by policy: {path}")
        if is_evidence_path(full) and not allow_evidence:
            return _failure(call, f"patch modifies evidence/generated memory path; set allow_evidence_edits only for explicit regenerate tasks: {path}")
    patch_path = session_dir / "tool_results" / "apply_patch.diff"
    patch_path.parent.mkdir(parents=True, exist_ok=True)
    patch_path.write_text(patch, encoding="utf-8")
    proc = subprocess.run(["git", "apply", str(patch_path)], cwd=repo_root(), capture_output=True, text=True)
    return {
        "tool": call.name,
        "args": {"path_count": len(touched), "allow_evidence_edits": allow_evidence},
        "ok": proc.returncode == 0,
        "return_code": proc.returncode,
        "path": str(patch_path),
        "stdout_tail": _tail(proc.stdout),
        "stderr_tail": _tail(proc.stderr),
        "summary": f"applied patch to {len(touched)} path(s)" if proc.returncode == 0 else "patch failed",
    }


# PRD #128: ``_demo_acceptance_tool`` moved to ``tool_handlers/demo.py``.
# Re-exported here so import paths stay stable.
from agentic_swmm.agent.tool_handlers.demo import (  # noqa: E402,F401
    _demo_acceptance_tool,
)


# ---------------------------------------------------------------------------
# Deterministic-SWMM handlers — MCP-routed (PRD-Y)
# ---------------------------------------------------------------------------
#
# Each handler below is the output of ``_make_mcp_routed_handler`` paired
# with a per-tool ``_*_args_mapper`` function. The mapper:
#   1. validates required snake_case arguments (paths exist / are in-repo
#      / have the right suffix) and returns a ``_failure`` dict on
#      problems so the planner sees the same fail-soft shape it always
#      did,
#   2. translates snake_case ToolSpec argument names into the camelCase
#      property names each MCP server expects (mirrors
#      ``mcp/<server>/server.js`` schemas).


def _audit_run_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``audit_run`` args to ``swmm-experiment-audit`` MCP schema."""

    run_dir = call.args.get("run_dir")
    if not isinstance(run_dir, str) or not run_dir.strip():
        return _failure(call, "missing required argument: run_dir")
    args: dict[str, Any] = {"runDir": run_dir}
    if call.args.get("workflow_mode"):
        args["workflowMode"] = str(call.args["workflow_mode"])
    if call.args.get("objective"):
        args["objective"] = str(call.args["objective"])
    return args


_audit_run_tool = _make_mcp_routed_handler(
    "swmm-experiment-audit", "audit_run", args_mapper=_audit_run_args
)


def _summarize_memory_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``summarize_memory`` args to ``swmm-modeling-memory`` MCP schema.

    The MCP server requires both ``runsDir`` and ``outDir``; if the caller
    omits ``out_dir`` we default to ``memory/modeling-memory`` (the same
    default the CLI used).
    """

    runs_dir = call.args.get("runs_dir")
    if not isinstance(runs_dir, str) or not runs_dir.strip():
        return _failure(call, "missing required argument: runs_dir")
    out_dir = call.args.get("out_dir") or "memory/modeling-memory"
    return {"runsDir": str(runs_dir), "outDir": str(out_dir)}


_summarize_memory_tool = _make_mcp_routed_handler(
    "swmm-modeling-memory", "summarize_memory", args_mapper=_summarize_memory_args
)


# -- swmm-rag-memory retrieval (Issue #124 Part A) ----------------------------
#
# The skill ships ``skills/swmm-rag-memory/scripts/retrieve_memory.py`` which
# already exposes a stable CLI; the agent invokes it via a subprocess so we
# don't pull the embedding library into the planner's hot path. Output is the
# script's JSON document (or markdown via ``--format markdown``); we return
# its trailing tail in ``stdout_tail`` so the planner sees citations.

_RAG_SKILL_DIR_RELATIVE = ("skills", "swmm-rag-memory", "scripts")


def _retrieve_memory_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Shell out to the swmm-rag-memory ``retrieve_memory.py`` script.

    Why a subprocess and not an in-process import? The retriever pulls in
    ``rag_memory_lib`` which lives under ``skills/swmm-rag-memory/scripts/``
    (not on the package import path) and optionally loads embedding vectors.
    Subprocess isolation keeps a corrupt index from poisoning the planner's
    Python state mid-turn.
    """
    query = call.args.get("query")
    if not isinstance(query, str) or not query.strip():
        return _failure(call, "missing required argument: query")
    script_path = repo_root().joinpath(*_RAG_SKILL_DIR_RELATIVE, "retrieve_memory.py")
    if not script_path.is_file():
        return _failure(call, f"retrieve_memory script not found at {script_path}")
    cli_args: list[str] = [str(script_path), "--query", query]
    top_k = call.args.get("top_k")
    if isinstance(top_k, int) and top_k > 0:
        cli_args.extend(["--top-k", str(top_k)])
    retriever = call.args.get("retriever")
    if retriever in ("keyword", "hybrid"):
        cli_args.extend(["--retriever", retriever])
    project = call.args.get("project")
    if isinstance(project, str) and project.strip():
        cli_args.extend(["--project", project])
    return _run_script_tool(call, session_dir, cli_args)


# -- Memory recall tools (PRD M1, M6, M7.1) -----------------------------------
#
# PRD #128: the four memory-family handlers (`_recall_memory_tool`,
# `_recall_memory_search_tool`, `_recall_session_history_tool`,
# `_record_fact_tool`) moved to ``tool_handlers/swmm_memory.py`` along
# with their token-budget / lessons-path helpers. Re-exported here so
# import paths stay stable for ``_build_tools`` and downstream code.

from agentic_swmm.agent.tool_handlers.swmm_memory import (  # noqa: E402,F401
    _recall_memory_search_tool,
    _recall_memory_tool,
    _recall_session_history_tool,
    _record_fact_tool,
)


def _read_file_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    path = _repo_path(str(call.args["path"]))
    if path is None:
        return _failure(call, "refusing to read outside repository")
    if not path.exists() or not path.is_file():
        return _failure(call, f"file not found: {path}")
    text = path.read_text(encoding="utf-8", errors="replace")
    return {"tool": call.name, "args": call.args, "ok": True, "path": str(path), "chars": len(text), "excerpt": text[:4000], "summary": f"read {path.relative_to(repo_root())}"}


def _list_skills_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    skills = [{"name": str(r.get("name")), "enabled": bool(r.get("enabled", True)), "path": str(r.get("path"))} for r in discover_skills()]
    return {"tool": call.name, "args": call.args, "ok": True, "skills": skills, "summary": f"{len(skills)} skills available", "excerpt": json.dumps(skills, indent=2)[:4000]}


def _read_skill_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    skill_name = str(call.args["skill_name"])
    if not re.fullmatch(r"[A-Za-z0-9_.-]+", skill_name):
        return _failure(call, "invalid skill name")
    path = _repo_path(f"skills/{skill_name}/SKILL.md")
    if path is None or not path.exists() or not path.is_file():
        return _failure(call, f"skill not found: {skill_name}")
    text = path.read_text(encoding="utf-8", errors="replace")
    return {"tool": call.name, "args": call.args, "ok": True, "path": str(path), "chars": len(text), "excerpt": text[:4000], "summary": f"read skill {skill_name}"}


def _run_swmm_inp_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``run_swmm_inp`` args to ``swmm-runner.swmm_run`` MCP schema.

    Path validation (in-repo + suffix) and default ``run_dir`` / ``node``
    selection mirror the historical in-process handler so behaviour is
    identical for the caller.
    """

    inp = _resolve_inp_for_run(call)
    if isinstance(inp, dict):
        return inp
    run_dir = _optional_repo_output_dir(call, "run_dir")
    if isinstance(run_dir, dict):
        return run_dir
    if run_dir is None:
        run_id = str(call.args.get("run_id") or f"{_safe_name(inp.stem)}-{int(time.time())}")
        run_dir = repo_root() / "runs" / "agent" / _safe_name(run_id)
    default_node = _node_suggestions(str(inp), limit=1)
    node = str(call.args.get("node") or (default_node[0] if default_node else "O1"))
    return {"inp": str(inp), "runDir": str(run_dir), "node": node}


_run_swmm_inp_tool = _make_mcp_routed_handler(
    "swmm-runner", "swmm_run", args_mapper=_run_swmm_inp_args
)


def _inspect_plot_options_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    run_dir: Path | None = None
    if call.args.get("run_dir"):
        resolved_run_dir = _required_repo_dir(call, "run_dir")
        if isinstance(resolved_run_dir, dict):
            return resolved_run_dir
        run_dir = resolved_run_dir

    inp: Path | None = None
    if call.args.get("inp_path"):
        inp = _resolve_existing_inp(str(call.args["inp_path"]))
    elif run_dir is not None:
        manifest = _read_manifest(run_dir)
        inp = _find_inp(run_dir, manifest)

    out_file: Path | None = None
    if call.args.get("out_file"):
        out_file = _repo_path(str(call.args["out_file"]))
        if out_file is None or not out_file.exists() or not out_file.is_file():
            return _failure(call, f"out_file must be an existing repository file: {call.args['out_file']}")
    elif run_dir is not None:
        manifest = _read_manifest(run_dir)
        out_file = _find_out(run_dir, manifest)

    rainfall_options = rainfall_timeseries_options(inp) if inp is not None else []
    node_options = _node_suggestions(str(inp), limit=100) if inp is not None else []
    node_attribute_options = _node_attribute_options(out_file, node_options)
    default_rain = next((option["name"] for option in rainfall_options if option.get("used_by_raingage")), None)
    if default_rain is None and rainfall_options:
        default_rain = rainfall_options[0]["name"]
    default_node = node_options[0] if node_options else None

    selections_needed: list[str] = []
    if len(rainfall_options) > 1:
        selections_needed.append("rain_ts")
    if len(node_options) > 1:
        selections_needed.append("node")
    if len(node_attribute_options) > 1:
        selections_needed.append("node_attr")
    user_prompt = ""
    if selections_needed:
        user_prompt = "Please choose " + ", ".join(selections_needed) + " before plotting."

    result = {
        "inp": str(inp) if inp is not None else None,
        "out_file": str(out_file) if out_file is not None else None,
        "rainfall_options": rainfall_options,
        "node_options": node_options,
        "node_attribute_options": node_attribute_options,
        "defaults": {"rain_ts": default_rain, "node": default_node, "node_attr": DEFAULT_NODE_ATTR},
        "selections_needed": selections_needed,
        "user_prompt": user_prompt,
    }
    return {"tool": call.name, "args": call.args, "ok": True, "results": result, "summary": f"rain={len(rainfall_options)} nodes={len(node_options)} attrs={len(node_attribute_options)}"}


def _select_skill_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Return the skill's tool subset (PRD-Y two-level planner surface).

    Lazy-imports ``SkillRouter`` because ``skill_router`` already imports
    from ``tool_registry`` — keeping the import inside the handler avoids
    a load-time cycle.
    """

    from agentic_swmm.agent.skill_router import AGENT_INTERNAL_SKILL, SkillRouter

    skill_name = call.args.get("skill_name")
    if not isinstance(skill_name, str) or not skill_name.strip():
        return _failure(call, "skill_name is required")
    skill_name = skill_name.strip()
    router = SkillRouter(AgentToolRegistry())
    try:
        bundle = router.tools_for(skill_name)
    except KeyError:
        known = ", ".join(router.list_skills())
        return _failure(call, f"unknown skill: {skill_name} (known: {known})")
    entries = [
        {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.parameters,
            "is_read_only": bool(tool.is_read_only),
        }
        for tool in bundle.tools
    ]
    summary = (
        f"selected skill {skill_name}: {len(entries)} tool(s) "
        f"({bundle.source})"
    )
    if skill_name == AGENT_INTERNAL_SKILL:
        summary += " — available without further select_skill calls"
    return {
        "tool": call.name,
        "args": call.args,
        "ok": True,
        "skill_name": skill_name,
        "source": bundle.source,
        "tools": entries,
        "summary": summary,
    }


# Valid workflow modes the LLM may pass via the ``mode`` argument.
# PRD-04: derived from the workflow-mode adapter registry — adding an
# adapter file is enough to make a new mode acceptable here. The
# ToolSpec input_schema enum above still has to be updated for the
# LLM-visible schema (that surface stays in this module per PRD-04
# scope). Computed once at import time because
# ``intent_disambiguator`` imports this name directly and uses it in
# module-level response-schema construction.
from agentic_swmm.agent.workflow_modes import all_modes as _all_modes  # noqa: E402

_VALID_MODE_ENUM = frozenset(_all_modes())


def _build_response_for_mode(
    call: ToolCall, mode: str, goal: str, provided: dict[str, str]
) -> dict[str, Any]:
    """Validate inputs for ``mode`` and build the tool's response payload.

    Shared between the explicit-mode short-circuit (when the LLM passed
    ``mode``) and the legacy keyword-fallback path. The ``provided`` dict
    is read-only here; auto-infer of ``run_dir`` from global state is the
    caller's responsibility and runs only in the keyword-fallback path.

    PRD-04. The per-mode ``required_inputs`` / ``recommended_next_tools``
    / ``evidence_boundary`` triples live on adapter classes in
    ``agentic_swmm.agent.workflow_modes``. The registry is the single
    source of truth; this function only adds rules that depend on the
    caller's state (the goal-text-driven baseline_run_dir addendum for
    comparison audits) or on the ``needs_user_inputs`` sentinel that
    the keyword fallback emits when no registered mode applies.
    """

    # Late import keeps the registry an optional dependency of this
    # module (avoiding an import cycle with the planner package).
    from agentic_swmm.agent.workflow_modes import get_mode_spec

    spec = get_mode_spec(mode)
    if spec is not None:
        required = list(spec.required_inputs)
        next_tools = list(spec.recommended_next_tools)
        boundary = spec.evidence_boundary
        # Goal-text-dependent rule: audit-comparison appends a second
        # run dir to required inputs. Lives at the call site because
        # it depends on the goal string, not on declarative spec state.
        if mode == "audit_only_or_comparison" and (
            "compare" in goal or "comparison" in goal or "比较" in goal
        ):
            required.append("baseline_run_dir")
    else:
        # ``needs_user_inputs`` is a sentinel the keyword fallback emits
        # when no registered mode applies. It is not a workflow the
        # agent dispatches; it is a prompt asking the user to provide
        # either a prepared INP or the full-build input set.
        required = ["inp_path or full modular build inputs"]
        next_tools = []
        boundary = "No SWMM execution should start until a prepared INP or complete build inputs are provided."

    missing = [item for item in required if item not in provided]
    if "inp_path or full modular build inputs" in required:
        missing = ["SWMM INP path, or network_json + subcatchments_csv + rainfall_input + landuse_input + soil_input"]
    if mode == "prepared_demo":
        missing = []

    node_suggestions = _node_suggestions(provided.get("inp_path"))
    plot_selection_options = _plot_selection_options_for_inp(provided.get("inp_path"))
    result = {
        "mode": mode,
        "top_level_contract": "skills/swmm-end-to-end/SKILL.md",
        "required_inputs": required,
        "provided_inputs": sorted(provided),
        "provided_values": provided,
        "missing_inputs": missing,
        "recommended_next_tools": [] if missing else next_tools,
        "stop_reason": "missing critical input" if missing else None,
        "evidence_boundary": boundary,
        "user_prompt": _workflow_user_prompt(mode, missing),
        "node_suggestions": node_suggestions,
        "plot_selection_options": plot_selection_options,
    }
    return {"tool": call.name, "args": call.args, "ok": True, "results": result, "summary": f"mode={mode} missing={len(missing)}"}


def compute_intent_signals(goal: str) -> dict[str, bool]:
    """Compute the keyword-derived ``wants_*`` flags for a goal.

    PRD #121 made ``agentic_swmm.agent.intent_classifier`` the single
    source of truth for keyword-driven intent extraction. This function
    is now a thin adapter that returns the legacy ``compute_intent_signals``
    dict shape so existing callers (``_select_workflow_mode_tool`` and
    the planner's auto-route disambiguator trigger from PRD #111) keep
    working byte-for-byte. New callers should use
    ``intent_classifier.classify_intent`` directly.
    """

    # Late import keeps the tool_registry → intent_classifier edge one
    # way; intent_classifier must stay dependency-free of tool_registry.
    from agentic_swmm.agent.intent_classifier import classify_intent

    return classify_intent(goal).as_dict()


def _select_workflow_mode_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    goal = str(call.args.get("goal") or "").lower()
    # ``mode`` is consumed below as a routing argument, not a workflow
    # input; exclude it from ``provided`` so it does not pollute the
    # echoed ``provided_inputs`` / ``provided_values`` fields.
    provided = {
        key: str(value).strip()
        for key, value in call.args.items()
        if key != "mode" and isinstance(value, str) and value.strip()
    }

    # PRD-INTENT-OVERMATCH (#95): the LLM may pass an explicit ``mode``
    # to bypass keyword-based intent re-derivation. If valid, use it
    # directly; do NOT auto-infer ``run_dir`` from global state when the
    # mode is explicit (auto-infer was part of the original bug — see
    # the PRD's Done Criteria #4).
    explicit_mode = call.args.get("mode")
    if isinstance(explicit_mode, str) and explicit_mode in _VALID_MODE_ENUM:
        return _build_response_for_mode(call, explicit_mode, goal, provided)

    # Legacy keyword-fallback path. NOTE: Chinese keywords are part of
    # the bilingual goal-routing contract. The `??` placeholders
    # previously here were a non-UTF-8 sync regression (P0-2 in #79).
    # Do not replace these with ASCII placeholders again — the
    # regression test in `tests/test_select_workflow_mode_bilingual.py`
    # will trip on any reintroduced `??` pair.
    #
    # PRD #111: signals are computed by ``compute_intent_signals`` so
    # the planner's auto-route disambiguator trigger and this keyword
    # fallback share one source of truth.
    signals = compute_intent_signals(goal)
    wants_calibration = signals["wants_calibration"]
    wants_uncertainty = signals["wants_uncertainty"]
    wants_audit = signals["wants_audit"]
    wants_plot = signals["wants_plot"]
    wants_demo = signals["wants_demo"]
    wants_run = signals["wants_run"]
    has_inp = bool(provided.get("inp_path"))
    has_run_dir = bool(provided.get("run_dir"))
    if (wants_plot or wants_audit) and not has_run_dir:
        active_run_dir = _active_run_dir_from_global_state()
        if active_run_dir:
            provided["run_dir"] = active_run_dir
            has_run_dir = True
    full_build_inputs = ["network_json", "subcatchments_csv", "rainfall_input", "landuse_input", "soil_input"]
    has_full_build = all(provided.get(key) for key in full_build_inputs)

    # PRD #111: compound run+demo+plot must pin to ``prepared_demo``
    # *before* the plot branch fires. Without this guard the plot
    # branch hijacks any goal that mentions ``plot`` once a run_dir is
    # auto-inferred from global state — which is exactly the bug from
    # ``runs/2026-05-16/120740_todcreek_run``.
    if wants_run and wants_demo and not has_inp:
        mode = "prepared_demo"
    elif wants_plot and has_run_dir:
        mode = "existing_run_plot"
    elif wants_calibration:
        mode = "calibration"
    elif wants_uncertainty:
        mode = "uncertainty"
    elif has_inp:
        mode = "prepared_inp_cli"
    elif wants_demo:
        mode = "prepared_demo"
    elif wants_audit and not has_inp:
        mode = "audit_only_or_comparison"
    elif has_full_build:
        mode = "full_modular_build"
    else:
        mode = "needs_user_inputs"

    return _build_response_for_mode(call, mode, goal, provided)


def _plot_run_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``plot_run`` args to ``swmm-plot.plot_rain_runoff_si`` MCP schema.

    The MCP server requires ``inp``/``out``/``outPng`` explicitly; the
    legacy CLI handler resolved the first two from the run-dir manifest
    internally. We do the same resolution here so the planner can keep
    passing just ``run_dir``.
    """

    run_dir = _required_repo_dir(call, "run_dir")
    if isinstance(run_dir, dict):
        return run_dir
    manifest = _read_manifest(run_dir)
    inp_path = _find_inp(run_dir, manifest)
    out_path = _find_out(run_dir, manifest)
    if inp_path is None or not inp_path.is_file():
        return _failure(call, f"could not resolve .inp from {run_dir}")
    if out_path is None or not out_path.is_file():
        return _failure(call, f"could not resolve .out from {run_dir}")
    if call.args.get("out_png"):
        out_png = _repo_output_path(str(call.args["out_png"]))
        if out_png is None or out_png.suffix.lower() != ".png":
            return _failure(call, "out_png must be a repository-relative .png path")
    else:
        # The MCP server requires outPng. Match the historical CLI default
        # (``07_plots/fig_<node>_<attr>.png`` under the run dir).
        node_for_default = re.sub(
            r"[^A-Za-z0-9_.-]+", "_", str(call.args.get("node") or "node")
        ).strip("_") or "node"
        attr_for_default = re.sub(
            r"[^A-Za-z0-9_.-]+", "_", str(call.args.get("node_attr") or "series")
        ).strip("_") or "series"
        out_png = run_dir / "07_plots" / f"fig_{node_for_default}_{attr_for_default}.png"
        out_png.parent.mkdir(parents=True, exist_ok=True)
    args: dict[str, Any] = {
        "inp": str(inp_path),
        "out": str(out_path),
        "outPng": str(out_png),
    }
    if call.args.get("node"):
        args["node"] = str(call.args["node"])
    if call.args.get("node_attr"):
        args["nodeAttr"] = str(call.args["node_attr"])
    if call.args.get("rain_ts"):
        args["rainTs"] = str(call.args["rain_ts"])
    if call.args.get("rain_kind"):
        args["rainKind"] = str(call.args["rain_kind"])
    return args


_plot_run_tool = _make_mcp_routed_handler(
    "swmm-plot", "plot_rain_runoff_si", args_mapper=_plot_run_args
)


def _network_qa_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``network_qa`` args to ``swmm-network.qa`` MCP schema.

    The MCP server's ``qa`` tool only accepts ``networkJsonPath`` — the
    optional ``report_json`` from the ToolSpec surface is ignored; we
    still validate it so the planner gets the same error message it
    used to. (The QA JSON ends up in the MCP server's stdout content.)
    """

    network_json = _required_repo_file(call, "network_json", suffix=".json")
    if isinstance(network_json, dict):
        return network_json
    if call.args.get("report_json"):
        report = _repo_output_path(str(call.args["report_json"]))
        if report is None or report.suffix.lower() != ".json":
            return _failure(call, "report_json must be a repository-relative .json path")
    return {"networkJsonPath": str(network_json)}


_network_qa_tool = _make_mcp_routed_handler(
    "swmm-network", "qa", args_mapper=_network_qa_args
)


def _network_to_inp_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``network_to_inp`` args to ``swmm-network.export_inp`` MCP schema.

    The MCP tool only accepts ``networkJsonPath`` (writes the .inp into a
    tmp directory) — ``out_path`` semantics from the legacy ToolSpec
    surface are preserved by post-processing the MCP response in
    ``_wrap_mcp_result``. The validation here keeps the planner-facing
    error parity.
    """

    network_json = _required_repo_file(call, "network_json", suffix=".json")
    if isinstance(network_json, dict):
        return network_json
    out_path = _repo_output_path(str(call.args["out_path"]))
    if out_path is None or out_path.suffix.lower() not in {".inp", ".txt"}:
        return _failure(call, "out_path must be a repository-relative .inp or .txt path")
    return {"networkJsonPath": str(network_json)}


_network_to_inp_tool = _make_mcp_routed_handler(
    "swmm-network", "export_inp", args_mapper=_network_to_inp_args
)


def _format_rainfall_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``format_rainfall`` args to ``swmm-climate.format_rainfall`` MCP."""

    input_csv = _required_repo_file(call, "input_csv", suffix=".csv")
    if isinstance(input_csv, dict):
        return input_csv
    out_json = _repo_output_path(str(call.args["out_json"]))
    out_timeseries = _repo_output_path(str(call.args["out_timeseries"]))
    if out_json is None or out_json.suffix.lower() != ".json":
        return _failure(call, "out_json must be a repository-relative .json path")
    if out_timeseries is None or out_timeseries.suffix.lower() not in {".txt", ".dat"}:
        return _failure(call, "out_timeseries must be a repository-relative .txt or .dat path")
    args: dict[str, Any] = {
        "inputCsvPath": str(input_csv),
        "outputJsonPath": str(out_json),
        "outputTimeseriesPath": str(out_timeseries),
    }
    snake_to_camel = {
        "series_name": "seriesName",
        "timestamp_column": "timestampColumn",
        "value_column": "valueColumn",
        "value_units": "valueUnits",
        "unit_policy": "unitPolicy",
        "timestamp_policy": "timestampPolicy",
    }
    for snake, camel in snake_to_camel.items():
        if call.args.get(snake):
            args[camel] = str(call.args[snake])
    return args


_format_rainfall_tool = _make_mcp_routed_handler(
    "swmm-climate", "format_rainfall", args_mapper=_format_rainfall_args
)


def _build_inp_args(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    """Map ``build_inp`` args to ``swmm-builder.build_inp`` MCP schema."""

    resolved: dict[str, Path] = {}
    for key, suffix in {"subcatchments_csv": ".csv", "params_json": ".json", "network_json": ".json"}.items():
        path = _required_repo_file(call, key, suffix=suffix)
        if isinstance(path, dict):
            return path
        resolved[key] = path
    out_inp = _repo_output_path(str(call.args["out_inp"]))
    out_manifest = _repo_output_path(str(call.args["out_manifest"]))
    if out_inp is None or out_inp.suffix.lower() != ".inp":
        return _failure(call, "out_inp must be a repository-relative .inp path")
    if out_manifest is None or out_manifest.suffix.lower() != ".json":
        return _failure(call, "out_manifest must be a repository-relative .json path")
    args: dict[str, Any] = {
        "subcatchmentsCsvPath": str(resolved["subcatchments_csv"]),
        "paramsJsonPath": str(resolved["params_json"]),
        "networkJsonPath": str(resolved["network_json"]),
        "outInpPath": str(out_inp),
        "outManifestPath": str(out_manifest),
    }
    optional_paths = {
        "rainfall_json": ("rainfallJsonPath", ".json"),
        "raingage_json": ("raingageJsonPath", ".json"),
        "timeseries_text": ("timeseriesTextPath", None),
        "config_json": ("configJsonPath", ".json"),
    }
    for snake, (camel, suffix) in optional_paths.items():
        if call.args.get(snake):
            path = _required_repo_file(call, snake, suffix=suffix)
            if isinstance(path, dict):
                return path
            args[camel] = str(path)
    if call.args.get("default_gage_id"):
        args["defaultGageId"] = str(call.args["default_gage_id"])
    return args


_build_inp_tool = _make_mcp_routed_handler(
    "swmm-builder", "build_inp", args_mapper=_build_inp_args
)


def _list_dir_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    path = _repo_path(str(call.args.get("path") or "."))
    if path is None or not path.exists() or not path.is_dir():
        return _failure(call, "directory must exist inside repository")
    entries = [{"name": item.name, "type": "dir" if item.is_dir() else "file", "path": str(item.relative_to(repo_root()))} for item in sorted(path.iterdir())[:200]]
    return {"tool": call.name, "args": call.args, "ok": True, "results": entries, "summary": f"{len(entries)} entries"}


def _search_files_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    query = str(call.args.get("query") or "").strip()
    if not query:
        return _failure(call, "query is required")
    pattern = _normalize_search_glob(str(call.args.get("glob") or "*"))
    max_results = int(call.args.get("max_results") or 50)
    results: list[dict[str, Any]] = []
    try:
        paths = repo_root().rglob(pattern)
    except ValueError as exc:
        return _failure(call, f"invalid glob pattern: {exc}")
    for path in paths:
        if len(results) >= max_results:
            break
        if not path.is_file() or any(part in {".git", ".venv", "__pycache__"} for part in path.parts):
            continue
        try:
            text = path.read_text(encoding="utf-8")
        except (UnicodeDecodeError, OSError):
            continue
        for lineno, line in enumerate(text.splitlines(), start=1):
            if query.lower() in line.lower():
                results.append({"path": str(path.relative_to(repo_root())), "line": lineno, "text": line.strip()[:300]})
                break
    return {"tool": call.name, "args": call.args, "ok": True, "glob": pattern, "results": results, "summary": f"{len(results)} match(es)"}


def _normalize_search_glob(pattern: str) -> str:
    cleaned = pattern.strip() or "*"
    # pathlib requires "**" to be a complete path component. LLM planners often
    # produce "**.inp" when they mean a recursive extension search.
    cleaned = re.sub(r"(?<!/)\*\*\.([A-Za-z0-9_*?[\]-]+)$", r"**/*.\1", cleaned)
    cleaned = re.sub(r"/\*\*\.([A-Za-z0-9_*?[\]-]+)$", r"/**/*.\1", cleaned)
    return cleaned


def _git_diff_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    command = ["git", "diff", "--stat" if call.args.get("stat_only", True) else "--"]
    if call.args.get("path"):
        command.extend(["--", str(call.args["path"])])
    proc = subprocess.run(command, cwd=repo_root(), capture_output=True, text=True)
    return {"tool": call.name, "args": call.args, "ok": proc.returncode == 0, "return_code": proc.returncode, "excerpt": proc.stdout[:8000], "stderr_tail": _tail(proc.stderr), "summary": "git diff read" if proc.returncode == 0 else "git diff failed"}


# PRD #128: ``_web_fetch_url_tool`` and ``_web_search_tool`` moved to
# ``tool_handlers/web.py``. Re-exported here so import paths stay stable.
from agentic_swmm.agent.tool_handlers.web import (  # noqa: E402,F401
    _web_fetch_url_tool,
    _web_search_tool,
)


def _list_mcp_servers_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    servers = load_mcp_registry()
    return {"tool": call.name, "args": call.args, "ok": True, "servers": servers, "summary": f"{len(servers)} configured MCP server(s)"}


def _list_mcp_tools_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    server = _mcp_server(str(call.args["server"]))
    if server is None:
        return _mcp_failure(call, f"MCP server not found: {call.args['server']}")
    timeout = int(call.args.get("timeout_seconds") or 5)
    refresh = bool(call.args.get("refresh"))
    ttl = int(call.args.get("cache_ttl_seconds") or mcp_cache.DEFAULT_TTL_SECONDS)
    if not refresh:
        cached = mcp_cache.read_cached_tools(server, ttl_seconds=ttl)
        if cached is not None:
            mapped = [_map_mcp_tool_schema(str(server["name"]), tool) for tool in cached if isinstance(tool, dict)]
            return {
                "tool": call.name,
                "args": call.args,
                "ok": True,
                "tools": cached,
                "mapped_tools": mapped,
                "cache": "hit",
                "summary": f"{len(cached)} cached MCP tool(s) on {server['name']}; {len(mapped)} schema(s) mapped for planner inspection",
            }
    try:
        tools = mcp_client.list_tools(str(server["command"]), [str(arg) for arg in server.get("args", [])], timeout=timeout)
    except Exception as exc:
        return _mcp_failure(call, f"MCP tools/list failed: {exc}")
    cache_path = mcp_cache.write_cached_tools(server, tools)
    mapped = [_map_mcp_tool_schema(str(server["name"]), tool) for tool in tools if isinstance(tool, dict)]
    return {
        "tool": call.name,
        "args": call.args,
        "ok": True,
        "tools": tools,
        "mapped_tools": mapped,
        "cache": "refresh" if refresh else "miss",
        "cache_path": str(cache_path),
        "summary": f"{len(tools)} MCP tool(s) on {server['name']}; {len(mapped)} schema(s) mapped for planner inspection; cached schema",
    }


def _call_mcp_tool_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    server = _mcp_server(str(call.args["server"]))
    if server is None:
        return _mcp_failure(call, f"MCP server not found: {call.args['server']}")
    arguments = call.args.get("arguments") if isinstance(call.args.get("arguments"), dict) else {}
    try:
        result = mcp_client.call_tool(str(server["command"]), [str(arg) for arg in server.get("args", [])], str(call.args["tool"]), arguments)
    except Exception as exc:
        return _mcp_failure(call, f"MCP tools/call failed: {exc}", server=str(server["name"]))
    return {"tool": call.name, "args": call.args, "ok": True, "results": result, "summary": f"called MCP tool {server['name']}.{call.args['tool']}"}


def _mcp_failure(call: ToolCall, summary: str, *, server: str | None = None) -> dict[str, Any]:
    result = _failure(call, summary)
    result["recovery"] = "Use list_mcp_servers/list_mcp_tools to refresh available MCP tools, then retry with corrected server/tool/arguments or fall back to the CLI wrapper."
    result["fallback_tools"] = _mcp_fallback_tools(server or str(call.args.get("server") or ""))
    return result


def _map_mcp_tool_schema(server_name: str, tool: dict[str, Any]) -> dict[str, Any]:
    name = str(tool.get("name") or "tool")
    description = str(tool.get("description") or f"MCP tool exposed by {server_name}.")
    schema = tool.get("inputSchema")
    if not isinstance(schema, dict):
        schema = tool.get("schema") if isinstance(tool.get("schema"), dict) else {}
    parameters = _normalize_json_schema(schema)
    return {
        "server": server_name,
        "mcp_tool": name,
        "planner_tool": "call_mcp_tool",
        "description": description,
        "arguments": {"server": server_name, "tool": name, "arguments_schema": parameters},
    }


def _normalize_json_schema(schema: dict[str, Any]) -> dict[str, Any]:
    if not schema:
        return {"type": "object", "properties": {}, "additionalProperties": True}
    normalized = dict(schema)
    normalized.setdefault("type", "object")
    normalized.setdefault("properties", {})
    normalized.setdefault("additionalProperties", True)
    if not isinstance(normalized.get("properties"), dict):
        normalized["properties"] = {}
    return normalized


def _mcp_fallback_tools(server_name: str) -> list[str]:
    mapping = {
        "swmm-builder": ["build_inp"],
        "swmm-climate": ["format_rainfall"],
        "swmm-network": ["network_qa", "network_to_inp"],
        "swmm-plot": ["plot_run"],
        "swmm-runner": ["run_swmm_inp"],
    }
    return mapping.get(server_name, ["list_mcp_servers", "list_mcp_tools"])


def _capabilities_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    names = sorted(_build_tools())
    return {"tool": call.name, "args": call.args, "ok": True, "capabilities": capability_summary(names), "summary": "runtime capabilities returned"}


def _workflow_user_prompt(mode: str, missing: list[str]) -> str:
    if not missing:
        return "Inputs are sufficient for the selected workflow mode. Continue with the recommended tools."
    if mode == "needs_user_inputs":
        return "Please provide a SWMM INP path, or the complete full-build input set: network_json, subcatchments_csv, rainfall_input, landuse_input, and soil_input."
    return "Please provide: " + ", ".join(missing)


def _active_run_dir_from_global_state() -> str | None:
    path = runtime_state_path()
    if not path.exists():
        return None
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return None
    if not isinstance(payload, dict):
        return None
    value = payload.get("active_run_dir")
    return str(value) if value else None


def _run_cli_tool(call: ToolCall, session_dir: Path, cli_args: list[str]) -> dict[str, Any]:
    return _run_process_tool(call, session_dir, [sys.executable, "-m", "agentic_swmm.cli", *cli_args], cwd=repo_root())


def _run_script_tool(call: ToolCall, session_dir: Path, cli_args: list[str]) -> dict[str, Any]:
    return _run_process_tool(call, session_dir, [sys.executable, *cli_args], cwd=repo_root())


def _run_process_tool(call: ToolCall, session_dir: Path, command: list[str], *, cwd: Path, timeout: int = 120) -> dict[str, Any]:
    started = datetime.now(timezone.utc)
    try:
        proc = subprocess.run(command, cwd=cwd, capture_output=True, text=True, env=runtime_env(), timeout=timeout)
        timed_out = False
    except subprocess.TimeoutExpired as exc:
        proc = subprocess.CompletedProcess(command, 124, stdout=exc.stdout or "", stderr=exc.stderr or f"command timed out after {timeout}s")
        timed_out = True
    finished = datetime.now(timezone.utc)
    stdout = _process_text(proc.stdout)
    stderr = _process_text(proc.stderr)
    safe_name = _safe_name(call.name)
    stdout_path = session_dir / "tool_results" / f"{safe_name}.stdout.txt"
    stderr_path = session_dir / "tool_results" / f"{safe_name}.stderr.txt"
    stdout_path.parent.mkdir(parents=True, exist_ok=True)
    stdout_path.write_text(stdout, encoding="utf-8")
    stderr_path.write_text(stderr, encoding="utf-8")
    return {"tool": call.name, "args": call.args, "command": command, "ok": proc.returncode == 0, "return_code": proc.returncode, "timed_out": timed_out, "started_at_utc": started.isoformat(timespec="seconds"), "finished_at_utc": finished.isoformat(timespec="seconds"), "stdout_file": str(stdout_path), "stderr_file": str(stderr_path), "stdout_tail": _tail(stdout), "stderr_tail": _tail(stderr), "summary": _summarize_cli_result(call.name, stdout, proc.returncode)}


def _run_tests_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    paths = call.args.get("paths")
    test_paths = [str(path) for path in paths] if isinstance(paths, list) and paths else ["tests"]
    for path in test_paths:
        resolved = _repo_path(path)
        if resolved is None:
            return _failure(call, f"test path must be inside repository: {path}")
    timeout = int(call.args.get("timeout_seconds") or 120)
    if importlib.util.find_spec("pytest") is None and len(test_paths) == 1 and test_paths[0].endswith(".py"):
        return _run_process_tool(call, session_dir, [sys.executable, test_paths[0]], cwd=repo_root(), timeout=timeout)
    return _run_process_tool(call, session_dir, [sys.executable, "-m", "pytest", *test_paths], cwd=repo_root(), timeout=timeout)


def _run_allowed_command_tool(call: ToolCall, session_dir: Path) -> dict[str, Any]:
    command = call.args.get("command")
    if not isinstance(command, list) or not all(isinstance(item, str) and item for item in command):
        return _failure(call, "command must be a non-empty string array")
    if not _command_allowed(command):
        return _failure(call, "command is not allowlisted")
    timeout = int(call.args.get("timeout_seconds") or 120)
    return _run_process_tool(call, session_dir, command, cwd=repo_root(), timeout=timeout)


def _summarize_cli_result(tool: str, stdout: str, return_code: int) -> str:
    if return_code != 0:
        return f"{tool} failed"
    parsed = _try_json(stdout)
    if isinstance(parsed, dict):
        if "run_dir" in parsed:
            return f"run_dir={parsed['run_dir']}"
        if "experiment_note" in parsed:
            return f"audit_note={parsed['experiment_note']}"
        if "ok" in parsed and "issue_count" in parsed:
            return f"ok={parsed['ok']} issue_count={parsed['issue_count']}"
        if "outputs" in parsed:
            return "outputs=" + json.dumps(parsed["outputs"], sort_keys=True)[:500]
    stripped = stdout.strip().splitlines()
    return stripped[-1] if stripped else "completed"


def _repo_path(value: str) -> Path | None:
    raw = Path(value).expanduser()
    candidate = raw.resolve() if raw.is_absolute() else (repo_root() / raw).resolve()
    try:
        candidate.relative_to(repo_root().resolve())
    except ValueError:
        return None
    return candidate


def _repo_output_path(value: str) -> Path | None:
    path = _repo_path(value)
    if path is not None:
        path.parent.mkdir(parents=True, exist_ok=True)
    return path


def _node_suggestions(inp_path: str | None, limit: int = 8) -> list[str]:
    if not inp_path:
        return []
    candidate = _resolve_existing_inp(inp_path)
    if candidate is None:
        return []
    sections: dict[str, list[str]] = {"[OUTFALLS]": [], "[JUNCTIONS]": []}
    section: str | None = None
    for line in candidate.read_text(encoding="utf-8", errors="ignore").splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith(";"):
            continue
        if stripped.startswith("[") and stripped.endswith("]"):
            section = stripped.upper()
            continue
        if section in {"[OUTFALLS]", "[JUNCTIONS]"}:
            name = stripped.split()[0]
            if name not in sections[section]:
                sections[section].append(name)
    suggestions = [*sections["[OUTFALLS]"], *sections["[JUNCTIONS]"]]
    deduped = list(dict.fromkeys(suggestions))
    return deduped[:limit]


def _plot_selection_options_for_inp(inp_path: str | None) -> dict[str, Any]:
    if not inp_path:
        return {"rainfall_options": [], "node_attribute_options": _default_node_attribute_options()}
    inp = _resolve_existing_inp(inp_path)
    if inp is None:
        return {"rainfall_options": [], "node_attribute_options": _default_node_attribute_options()}
    return {
        "rainfall_options": rainfall_timeseries_options(inp),
        "node_attribute_options": _default_node_attribute_options(),
    }


def _node_attribute_options(out_file: Path | None, node_options: list[str]) -> list[dict[str, Any]]:
    if out_file is None or not out_file.exists():
        return _default_node_attribute_options()
    try:
        from swmmtoolbox import catalog

        rows = catalog(str(out_file), "node")
    except Exception:
        return _default_node_attribute_options()
    attrs: list[str] = []
    for row in rows:
        if len(row) < 3 or row[0] != "node":
            continue
        node, attr = str(row[1]), str(row[2])
        if node_options and node not in node_options:
            continue
        if attr not in attrs:
            attrs.append(attr)
    if not attrs:
        return _default_node_attribute_options()
    preferred = [attr for attr in NODE_ATTRIBUTE_CHOICES if attr in attrs]
    remainder = [attr for attr in attrs if attr not in preferred]
    return [{"name": attr, "label": NODE_ATTRIBUTE_LABELS.get(attr, attr.replace("_", " "))} for attr in [*preferred, *remainder]]


def _default_node_attribute_options() -> list[dict[str, str]]:
    return [{"name": attr, "label": NODE_ATTRIBUTE_LABELS.get(attr, attr.replace("_", " "))} for attr in NODE_ATTRIBUTE_CHOICES]


def _resolve_existing_inp(value: str) -> Path | None:
    path = _repo_path(value)
    if path is not None and path.exists() and path.is_file() and path.suffix.lower() == ".inp":
        return path
    external = Path(value).expanduser()
    try:
        external = external.resolve()
    except OSError:
        return None
    if external.exists() and external.is_file() and external.suffix.lower() == ".inp":
        return external
    return _find_repo_inp(value)


def _process_text(value: Any) -> str:
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    return str(value or "")


def _patch_paths(patch: str) -> list[str]:
    paths: list[str] = []
    for line in patch.splitlines():
        if line.startswith("+++ b/") or line.startswith("--- a/"):
            path = line[6:].strip()
            if path != "/dev/null" and path not in paths:
                paths.append(path)
        elif line.startswith("diff --git "):
            parts = line.split()
            for part in parts[2:4]:
                if part.startswith(("a/", "b/")):
                    path = part[2:]
                    if path not in paths:
                        paths.append(path)
    return paths


def _command_allowed(command: list[str]) -> bool:
    exe = Path(command[0]).name.lower()
    if exe in {"pytest", "pytest.exe"}:
        return True
    if exe in {"python", "python.exe"} or command[0] == sys.executable:
        return len(command) >= 3 and command[1] == "-m" and command[2] in {"pytest", "agentic_swmm.cli"}
    if exe in {"node", "node.exe"}:
        return len(command) >= 2 and _repo_path(command[1]) is not None and Path(command[1]).suffix == ".mjs" and str(Path(command[1])).replace("\\", "/").startswith("scripts/")
    if exe in {"swmm5", "swmm5.exe", "swmm5.cmd"}:
        return True
    return False


def _required_repo_file(call: ToolCall, key: str, *, suffix: str | None = None) -> Path | dict[str, Any]:
    value = call.args.get(key)
    if not isinstance(value, str) or not value.strip():
        return _failure(call, f"missing required file argument: {key}")
    path = _repo_path(value)
    if path is None:
        return _failure(call, f"{key} must be inside repository")
    if suffix and path.suffix.lower() != suffix:
        return _failure(call, f"{key} must end with {suffix}")
    if not path.exists() or not path.is_file():
        return _failure(call, f"file not found: {path}")
    return path


def _resolve_inp_for_run(call: ToolCall) -> Path | dict[str, Any]:
    raw = str(call.args.get("inp_path", "")).strip()
    if not raw:
        return _failure(call, "missing required file argument: inp_path")
    repo_file = _required_repo_file(call, "inp_path", suffix=".inp")
    if not isinstance(repo_file, dict):
        return repo_file
    resolved = _find_repo_inp(raw)
    if resolved is not None:
        return resolved
    external = Path(raw).expanduser()
    try:
        external = external.resolve()
    except OSError:
        return _failure(call, f"inp_path could not be resolved: {raw}")
    if external.suffix.lower() != ".inp":
        return _failure(call, "inp_path must end with .inp")
    if not external.exists() or not external.is_file():
        return _failure(call, f"external INP file not found: {external}")
    return external


def _find_repo_inp(value: str) -> Path | None:
    if not value or Path(value).is_absolute() or "/" in value:
        return None
    root = repo_root() / "examples"
    if not root.exists():
        return None
    matches = sorted(path for path in root.rglob(value) if path.is_file() and path.suffix.lower() == ".inp")
    return matches[0] if matches else None


def _required_repo_dir(call: ToolCall, key: str) -> Path | dict[str, Any]:
    value = call.args.get(key)
    if not isinstance(value, str) or not value.strip():
        return _failure(call, f"missing required directory argument: {key}")
    path = _repo_path(value)
    if path is None:
        return _failure(call, f"{key} must be inside repository")
    if not path.exists() or not path.is_dir():
        return _failure(call, f"directory not found: {path}")
    return path


def _optional_repo_output_dir(call: ToolCall, key: str) -> Path | dict[str, Any] | None:
    value = call.args.get(key)
    if value is None:
        return None
    if not isinstance(value, str) or not value.strip():
        return _failure(call, f"{key} must be a non-empty string")
    path = _repo_path(value)
    if path is None:
        return _failure(call, f"{key} must be inside repository")
    path.mkdir(parents=True, exist_ok=True)
    return path


def _mcp_server(name: str) -> dict[str, Any] | None:
    for server in load_mcp_registry():
        if str(server.get("name")) == name:
            return server
    return None


def _strip_html(text: str) -> str:
    text = re.sub(r"(?is)<script.*?</script>|<style.*?</style>", " ", text)
    text = re.sub(r"(?s)<[^>]+>", " ", text)
    return re.sub(r"\s+", " ", html.unescape(text)).strip()


def _failure(call: ToolCall, summary: str) -> dict[str, Any]:
    return {"tool": call.name, "args": call.args, "ok": False, "summary": summary}


def _try_json(text: str) -> Any:
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return None


def _tail(text: str, max_chars: int = 2000) -> str:
    return text.strip()[-max_chars:]


def _safe_name(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9_.-]+", "-", value.strip())
    return cleaned.strip("-") or "agent"
