"""Tool-argument normalizer used before tool invocation.

This module has two explicit field maps:
- ``TOOL_TOKEN_FIELDS`` for opaque model-facing identifiers (uuid, hex ids,
  bg handles, stream ids, attachment refs).
- ``TOOL_PATH_FIELDS`` for path-bearing arguments and attachment/path repair.

Ordering is intentional: aliases/coercions -> token normalization -> path
normalization. We only normalize known schema keys; there is no global string
walk over free-text fields (query/topic/body/prompts).

Top-level ``url`` string arguments are stripped of all whitespace for every tool
(models often emit ``https://host. tld/...``); tools may still defensively strip.

Note for debugging/support: traces and chat history can contain model-original
argument strings, while tool execution uses this module's normalized dict.
"""
from __future__ import annotations

import json
import logging
import os
import re
from typing import Any

import attachment_registry as _ar
from utils.llm_token_normalize import (
    normalize_attachment_ref_token,
    normalize_bg_handle_token,
    normalize_hex_id_token,
    normalize_kt_task_id_token,
    normalize_stream_id_token,
    normalize_uuid_token,
)
from utils.path_safety import normalize_path_spaces

_log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Per-tool field map: which named arguments carry file paths.
#   key   = tool name
#   value = dict mapping arg-name -> {"intent": "read"|"write"|"rw",
#                                      "list": True|False,
#                                      "json_key": True if inside input_json}
#
# "list" means the arg is a list of paths (e.g. file_paths, input_paths).
# "json_key" means the path field lives inside a JSON-string arg named
# ``input_json``; the normalizer will parse, mutate, and re-serialize.
# ---------------------------------------------------------------------------

TOOL_PATH_FIELDS: dict[str, dict[str, dict[str, Any]]] = {
    # --- file_ops ---
    "read_file":                     {"filepath": {"intent": "read"}},
    "list_directory":                {"dir_path": {"intent": "read"}},
    "delete_file":                   {"path": {"intent": "write", "recovery_resolve_allow_write": False}},
    "copy_file":                     {"source_path": {"intent": "read"}, "destination_path": {"intent": "write"}},
    "move_file":                     {"source_path": {"intent": "read"}, "destination_path": {"intent": "write"}},
    "split_text_file":               {"file_path": {"intent": "read", "json_key": True},
                                      "output_dir": {"intent": "write", "json_key": True}},
    "merge_text_files":              {"file_paths": {"intent": "read", "json_key": True, "list": True},
                                      "output_path": {"intent": "write", "json_key": True}},
    "save_to_file":                  {"file_path": {"intent": "write"}},
    "write_text_file":               {"file_path": {"intent": "write"}},
    "append_to_text_file":           {"file_path": {"intent": "write"}},
    "read_file_chunk":               {"file_path": {"intent": "read"}},
    "submit_ocr_chunk_corrections":  {"file_path": {"intent": "read"},
                                      "output_file": {"intent": "write"}},
    "get_file_metadata":             {"path": {"intent": "read"}},
    "verify_file_exists":            {"file_path": {"intent": "read"}},
    "suggest_save_filename":         {"source_file_path": {"intent": "read"}},
    # --- pdf_tools ---
    "extract_pdf_text":              {"file_path": {"intent": "read"}},
    "split_pdf":                     {"file_path": {"intent": "read", "json_key": True},
                                      "output_dir": {"intent": "write", "json_key": True}},
    "merge_pdfs":                    {"file_paths": {"intent": "read", "json_key": True, "list": True},
                                      "output_path": {"intent": "write", "json_key": True}},
    # --- vision_ocr ---
    "ocr_scanned_pdf":               {"file_path": {"intent": "read"}},
    "ocr_image_text":                {"file_path": {"intent": "read"}},
    "ocr_scanned_pdf_to_file":       {"pdf_path": {"intent": "read"},
                                      "output_file_path": {"intent": "write"}},
    # --- tesseract ---
    "ocr_complex_document":          {"file_path": {"intent": "read"}},
    "ocr_complex_document_to_file":  {"pdf_path": {"intent": "read"},
                                      "output_file_path": {"intent": "write"}},
    # --- kraken ---
    "ocr_with_kraken_api":           {"file_path": {"intent": "read"}},
    "ocr_with_kraken_api_to_file":   {"pdf_path": {"intent": "read"},
                                      "output_file_path": {"intent": "write"}},
    # --- web_tools ---
    "screenshot_webpage":            {"output_path": {"intent": "write"}},
    "download_file":                 {"file_path": {"intent": "write", "json_key": True}},
    # --- research ---
    "generate_final_research_report": {"scratch_path": {"intent": "read"}},
    # --- system ---
    "show_display_window":           {"filename": {"intent": "read"}},
    "search_filesystem":             {"root_path": {"intent": "read"}},
    "capture_screenshot":            {"output_path": {"intent": "write"}},
    "compress_files":                {"files": {"intent": "read", "list": True},
                                      "output_zip": {"intent": "write"}},
    "extract_archive":               {"archive_path": {"intent": "read"},
                                      "output_dir": {"intent": "write"}},
    # --- clipboard ---
    "get_clipboard_image":           {"save_path": {"intent": "write"}},
    "copy_image_to_clipboard":       {"image_path": {"intent": "read"}},
    # --- cli/ffmpeg ---
    "get_media_info":                {"filepath": {"intent": "read"}},
    "convert_video":                 {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "extract_audio":                 {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "convert_audio":                 {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "trim_media":                    {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "concat_media":                  {"input_paths": {"intent": "read", "list": True},
                                      "output_path": {"intent": "write"}},
    "extract_frame":                 {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "merge_audio_video":             {"video_path": {"intent": "read"},
                                      "audio_path": {"intent": "read"},
                                      "output_path": {"intent": "write"}},
    "remove_audio":                  {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "extract_subtitles":             {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "burn_subtitles":                {"input_path": {"intent": "read"},
                                      "subtitle_path": {"intent": "read"},
                                      "output_path": {"intent": "write"}},
    "split_media":                   {"input_path": {"intent": "read"}, "output_dir": {"intent": "write"}},
    "change_speed":                  {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "rotate_video":                  {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    "images_to_video":               {"input_paths": {"intent": "read", "list": True},
                                      "output_path": {"intent": "write"}},
    "extract_frames_interval":       {"input_path": {"intent": "read"}, "output_dir": {"intent": "write"}},
    "normalize_audio":               {"input_path": {"intent": "read"}, "output_path": {"intent": "write"}},
    # --- skills ---
    # skill_name is a slug (e.g. docker-setup), not a workspace path — listing it here
    # caused path-arg recovery to inject directories like ~/.mugi/workspace into skill_name.
    "skill_view":                    {"file": {"intent": "read"}},
    "skill_manage":                  {"file": {"intent": "rw"}},
    # --- mesh ---
    "mesh_send_files":              {"paths": {"intent": "read", "list": True}},
}

# Per-tool field map for opaque non-path tokens that may be corrupted by LLM whitespace.
# key   = tool name
# value = dict mapping arg-name -> {"kind": "uuid"|"hex_id"|"bg_handle"|"stream_id"|"attachment_ref",
#                                   "lengths": (optional hex lengths),
#                                   "list": True|False,
#                                   "json_key": True if field lives inside input_json}
TOOL_TOKEN_FIELDS: dict[str, dict[str, dict[str, Any]]] = {
    # --- background / realtime handles ---
    "check_background_task": {"handle_id": {"kind": "bg_handle"}},
    "stop_background_task": {"handle_id": {"kind": "bg_handle"}},
    # --- schedule ---
    "cancel_scheduled_task": {"job_id": {"kind": "hex_id", "lengths": (12,)}},
    # --- watchlist / feeds ---
    "remove_watch_target": {"watch_id": {"kind": "hex_id", "lengths": (8,)}},
    "compare_watch_baseline": {"watch_id": {"kind": "hex_id", "lengths": (8,)}},
    "remove_feed": {"feed_id": {"kind": "hex_id", "lengths": (8,)}},
    # --- vault / captures ---
    "get_note": {"note_id": {"kind": "uuid"}},
    "update_vault_note": {"note_id": {"kind": "uuid"}},
    "append_to_vault_note": {"note_id": {"kind": "uuid"}},
    "vault_graph_neighbors": {"note_id": {"kind": "uuid"}},
    "vault_graph_linked_mentions": {"note_id": {"kind": "uuid"}},
    "vault_graph_subgraph": {"note_id": {"kind": "uuid"}},
    "vault_graph_insights": {"note_id": {"kind": "uuid"}},
    "vault_graph_confirm_edge": {"edge_id": {"kind": "uuid"}},
    "vault_graph_reject_edge": {"edge_id": {"kind": "uuid"}},
    # Delegation scratch / job ids are 12-hex (see subagent_runner); models often insert spaces.
    "read_delegation_scratch": {"delegation_job_id": {"kind": "hex_id", "lengths": (8, 12)}},
    "append_delegation_scratch": {"delegation_job_id": {"kind": "hex_id", "lengths": (8, 12)}},
    "delegate_task": {"delegation_job_id": {"kind": "hex_id", "lengths": (8, 12)}},
    # Kanban task ids are generated as kt_<14 hex>; models sometimes split with spaces.
    "kanban_show_task": {"task_id": {"kind": "kt_task"}},
    "kanban_list_runs": {"task_id": {"kind": "kt_task"}},
    "kanban_cancel_task": {"task_id": {"kind": "kt_task"}},
    "kanban_post_summary": {"task_id": {"kind": "kt_task"}},
    "kanban_post_progress": {"task_id": {"kind": "kt_task"}},
}

# Common LLM arg-name drift aliases -> canonical tool argument names.
TOOL_ARG_ALIASES: dict[str, dict[str, str]] = {
    # read_file_chunk uses file_path; models often reuse that name for read_file.
    "read_file": {
        "file_path": "filepath",
        "path": "filepath",
        "file": "filepath",
        "filename": "filepath",
        "target": "filepath",
    },
    "read_file_chunk": {"filepath": "file_path"},
    "create_plugin": {"code": "plugin_py_content"},
    "ocr_scanned_pdf_to_file": {"output_path": "output_file_path"},
    "ocr_complex_document_to_file": {"output_path": "output_file_path"},
    "ocr_with_kraken_api_to_file": {"output_path": "output_file_path"},
    "search_filesystem": {
        "path": "root_path",
        "root": "root_path",
        "rootPath": "root_path",
        "rootDir": "root_path",
    },
    "skill_manage": {"name": "skill_name", "skill": "skill_name",
                      "skill_content": "content", "skill_py_content": "content",
                      "skill_md_content": "content"},
    "schedule_task": {"task": "prompt", "time": "schedule", "when": "schedule"},
    "update_vault_note": {"text": "body"},
    # Models often use `text` for append-style tools; file tools take `content`.
    "write_text_file": {"text": "content"},
    "save_to_file": {"text": "content"},
    "append_to_text_file": {"text": "content"},
    # Models often emit URL-ish keys other than the schema's `url`.
    "read_webpage": {
        "page_url": "url",
        "link": "url",
        "website": "url",
        "source_url": "url",
        "href": "url",
        "page": "url",
        "website_url": "url",
        "target_url": "url",
        "uri": "url",
    },
    "read_and_analyze_page": {
        "page_url": "url",
        "link": "url",
        "website": "url",
        "source_url": "url",
        "href": "url",
        "page": "url",
        "website_url": "url",
        "target_url": "url",
        "uri": "url",
    },
    "append_research_scratch": {
        "subject": "topic",
        "title": "topic",
        "heading": "topic",
        "body": "content",
        "text": "content",
        "notes": "content",
        "findings": "content",
    },
    # Same drift as file/research scratch tools: main prose field must be `content`.
    "append_delegation_scratch": {
        "text": "content",
        "body": "content",
        "notes": "content",
    },
    # Long-term memory entry is `entry`; models often use generic prose keys.
    "append_to_memory": {
        "content": "entry",
        "text": "entry",
        "body": "entry",
        "note": "entry",
        "notes": "entry",
        "topic": "shard",
        "category": "shard",
        "shard_name": "shard",
    },
    "search_sessions": {
        "thread_id": "target_thread_id",
        "thread": "target_thread_id",
        "conversation_id": "target_thread_id",
    },
    "set_clipboard_text": {"content": "text"},
    "write_session_state": {
        "content": "value",
        "text": "value",
        "body": "value",
    },
    "save_research_findings": {
        "content": "findings",
        "text": "findings",
        "body": "findings",
        "report": "findings",
    },
    "generate_final_research_report": {
        "content": "findings",
        "text": "findings",
        "body": "findings",
    },
    "run_python_code": {
        "script": "code",
        "python": "code",
        "python_code": "code",
        "snippet": "code",
        "source": "code",
    },
    "submit_ocr_chunk_corrections": {
        "content": "corrected_content",
        "text": "corrected_content",
        "corrected_text": "corrected_content",
    },
    "correct_ocr_text": {"content": "text"},
    "commit_workspace_evolution": {
        "content": "message",
        "text": "message",
        "body": "message",
    },
    "compose_email_draft": {"content": "body", "text": "body"},
    "remind_me_in": {"content": "message", "text": "message"},
    "remind_at": {"content": "message", "text": "message"},
    # Watchlist tool signatures are easy for models to confuse with other URL tools.
    # add_watch_target takes `target_url` (not `url`); trace d844b076-* hit validation
    # guard twice because the model emitted `url`.
    "add_watch_target": {
        "url": "target_url",
        "page_url": "target_url",
        "website": "target_url",
        "website_url": "target_url",
        "link": "target_url",
        "href": "target_url",
        "uri": "target_url",
        "source_url": "target_url",
    },
    # compare_watch_baseline takes `watch_id` + `current_content`; models often emit
    # `watch_target_id` (confusing the param with the watch "target") or shorten
    # `current_content` to `content`.
    "compare_watch_baseline": {
        "watch_target_id": "watch_id",
        "target_id": "watch_id",
        "content": "current_content",
        "page_content": "current_content",
    },
}

# Models often use `filepath` (read_file) or `pdf_path` for PDF/OCR tools that take `file_path`.
_OCR_PDF_SINGLE_FILE_PATH_ALIASES: dict[str, str] = {
    "filepath": "file_path",
    "path": "file_path",
    "pdf_path": "file_path",
    "source_path": "file_path",
    "source_file": "file_path",
}
for _tn in (
    "extract_pdf_text",
    "ocr_scanned_pdf",
    "ocr_image_text",
    "ocr_complex_document",
    "ocr_with_kraken_api",
):
    TOOL_ARG_ALIASES[_tn] = dict(_OCR_PDF_SINGLE_FILE_PATH_ALIASES)

_SINGLE_INPUT_FILE_PATH_READ_TOOLS: frozenset[str] = frozenset(
    {
        "extract_pdf_text",
        "ocr_scanned_pdf",
        "ocr_image_text",
        "ocr_complex_document",
        "ocr_with_kraken_api",
    }
)


def _merge_longest_file_path_when_duplicates(tool_name: str, args: dict) -> None:
    """If both `file_path` and drift keys (e.g. `filepath`) are present, keep the longest string.

    Hermes/XML often supplies a truncated `file_path` plus a fuller `filepath` for the same PDF.
    Must run **before** :func:`_apply_aliases`, which would otherwise drop the alias when canonical exists.
    """
    if tool_name not in _SINGLE_INPUT_FILE_PATH_READ_TOOLS:
        return
    if not isinstance(args, dict):
        return
    canonical = "file_path"
    drift = ("filepath", "path", "pdf_path", "source_path", "source_file")
    candidates: list[tuple[str, str]] = []
    for k in (canonical,) + drift:
        if k not in args:
            continue
        v = args.get(k)
        if isinstance(v, str):
            s = v.strip()
            if s:
                candidates.append((k, s))
    if len(candidates) < 2:
        return
    _, best_v = max(candidates, key=lambda kv: len(kv[1]))
    args[canonical] = best_v
    for k, _s in candidates:
        if k != canonical:
            args.pop(k, None)


# ---------------------------------------------------------------------------
# Core normalizer
# ---------------------------------------------------------------------------

def _resolve_single(value: str) -> str:
    """Resolve a single string value: strip, attachment ID, then malformed-path repair."""
    if not value or not isinstance(value, str):
        return value
    value = value.strip()
    resolved = _ar.resolve_if_id(value)
    if resolved != value:
        return resolved
    repaired = normalize_path_spaces(value)
    if repaired != value:
        _ar.bump_counter("malformed_path_repair")
        return repaired
    return value


def _resolve_value(value: Any, field_meta: dict) -> Any:
    """Resolve a field value (scalar string or list of strings)."""
    is_list = field_meta.get("list", False)
    if is_list:
        if isinstance(value, list):
            return [_resolve_single(v) if isinstance(v, str) else v for v in value]
        if isinstance(value, str):
            return _resolve_single(value)
        return value
    if isinstance(value, str):
        return _resolve_single(value)
    return value


_PATH_MARKUP_ARTIFACT_LITERALS = {
    "/tool_call",
    "tool_call",
    "/tool_response",
    "tool_response",
    "/function",
    "function",
    "/parameter",
    "parameter",
    "/invoke",
    "invoke",
    "/think",
    "think",
}
_PATH_CONFUSABLE_SEPARATORS = {
    "\u2215",  # division slash
    "\u2044",  # fraction slash
    "\uff0f",  # fullwidth slash
    "\uff3c",  # fullwidth backslash
    "\uff1a",  # fullwidth colon
}


def path_value_looks_like_tool_markup_artifact(value: str) -> bool:
    """Return True when *value* looks like tool-call markup, not a real path.

    This is a correctness/anti-glitch guard for malformed local-model tool calls.
    It is not a replacement for path-safety boundaries like safe_resolve_path.
    """
    if not isinstance(value, str):
        return False
    trimmed = value.strip()
    if not trimmed:
        return False
    lowered = trimmed.lower()
    if lowered in _PATH_MARKUP_ARTIFACT_LITERALS:
        return True
    if any(tok in trimmed for tok in _PATH_CONFUSABLE_SEPARATORS):
        return True
    if any(ch in trimmed for ch in ("\n", "\r", "\x00")):
        return True
    if re.search(r"</?\s*(tool_call|tool_response|function|parameter|invoke)\b", lowered):
        return True
    if any(tok in lowered for tok in ("<|tool_call", "<tool_call|>", "<|channel>", "<channel|>")):
        return True
    if "//tool_call" in lowered or "//tool_response" in lowered or "//function" in lowered or "//parameter" in lowered or "//invoke" in lowered:
        return True
    normalized = lowered.replace("\\", "/")
    basename = normalized.rsplit("/", 1)[-1]
    if basename in {"tool_call", "tool_response", "function", "parameter", "invoke", "think"} and "." not in basename:
        return True
    return False


def tool_call_has_path_markup_artifact(tool_name: str, args_after_normalize: dict[str, Any]) -> bool:
    """Return True if any mapped path field looks like tool markup artifacts.

    Only fields registered in TOOL_PATH_FIELDS are checked. Non-string path values
    are ignored so regular tool validation can handle those type errors.
    """
    if not isinstance(args_after_normalize, dict):
        return False
    field_map = TOOL_PATH_FIELDS.get(tool_name)
    if not field_map:
        return False

    parsed_input_json: dict[str, Any] | None = None
    if any(meta.get("json_key") for meta in field_map.values()):
        raw_json = args_after_normalize.get("input_json")
        if isinstance(raw_json, str):
            try:
                loaded = json.loads(raw_json)
            except (json.JSONDecodeError, TypeError):
                loaded = None
        elif isinstance(raw_json, dict):
            loaded = raw_json
        else:
            loaded = None
        if isinstance(loaded, dict):
            parsed_input_json = loaded

    for field_name, meta in field_map.items():
        container = parsed_input_json if meta.get("json_key") else args_after_normalize
        if not isinstance(container, dict) or field_name not in container:
            continue
        value = container[field_name]
        if meta.get("list"):
            if isinstance(value, list):
                if any(path_value_looks_like_tool_markup_artifact(v) for v in value if isinstance(v, str)):
                    return True
            elif isinstance(value, str) and path_value_looks_like_tool_markup_artifact(value):
                return True
            continue
        if isinstance(value, str) and path_value_looks_like_tool_markup_artifact(value):
            return True
    return False


def iter_scalar_top_level_path_fields(tool_name: str) -> list[tuple[str, dict[str, Any]]]:
    """Path arguments that live on the tool dict (not inside ``input_json``), excluding list fields."""
    fm = TOOL_PATH_FIELDS.get(tool_name) or {}
    out: list[tuple[str, dict[str, Any]]] = []
    for field_name, meta in fm.items():
        if meta.get("json_key") or meta.get("list"):
            continue
        out.append((field_name, meta))
    return out


def path_fields_with_markup_artifacts(tool_name: str, args: dict[str, Any]) -> list[str]:
    """Names of scalar path fields whose values look like XML/tool markup (e.g. ``/function``)."""
    if not isinstance(args, dict):
        return []
    work = dict(args)
    try:
        normalize_tool_args(tool_name, work)
    except Exception:
        pass
    bad: list[str] = []
    for field_name, meta in iter_scalar_top_level_path_fields(tool_name):
        val = work.get(field_name)
        if isinstance(val, str) and path_value_looks_like_tool_markup_artifact(val):
            bad.append(field_name)
    return bad


def recovery_allow_write_for_path_field(tool_name: str, field_name: str) -> bool:
    """Whether ``safe_resolve_path`` should allow write-side resolution when recovering this path."""
    fm = TOOL_PATH_FIELDS.get(tool_name) or {}
    meta = fm.get(field_name) or {}
    if "recovery_resolve_allow_write" in meta:
        return bool(meta["recovery_resolve_allow_write"])
    return (meta.get("intent") or "read") in ("write", "rw")


def _normalize_token_scalar(value: Any, field_meta: dict) -> Any:
    if not isinstance(value, str):
        return value
    kind = (field_meta.get("kind") or "").strip()
    if kind == "uuid":
        return normalize_uuid_token(value)
    if kind == "hex_id":
        lengths = field_meta.get("lengths")
        if isinstance(lengths, tuple) and lengths:
            return normalize_hex_id_token(value, allowed_lengths=lengths)
        return normalize_hex_id_token(value)
    if kind == "bg_handle":
        return normalize_bg_handle_token(value)
    if kind == "stream_id":
        return normalize_stream_id_token(value)
    if kind == "attachment_ref":
        return normalize_attachment_ref_token(value)
    if kind == "kt_task":
        return normalize_kt_task_id_token(value)
    if kind:
        raise ValueError(f"Unknown token kind '{kind}' in TOOL_TOKEN_FIELDS")
    return value


def _normalize_token_value(value: Any, field_meta: dict) -> Any:
    if field_meta.get("list"):
        if isinstance(value, list):
            return [_normalize_token_scalar(v, field_meta) for v in value]
        return _normalize_token_scalar(value, field_meta)
    return _normalize_token_scalar(value, field_meta)


def _apply_token_normalization(tool_name: str, args: dict) -> dict:
    token_map = _token_field_map_for_tool(tool_name)
    if not token_map:
        return args

    has_json_keys = any(m.get("json_key") for m in token_map.values())
    if has_json_keys:
        raw_json = args.get("input_json")
        parsed = None
        if isinstance(raw_json, str):
            try:
                parsed = json.loads(raw_json)
            except (json.JSONDecodeError, TypeError):
                parsed = None
        elif isinstance(raw_json, dict):
            parsed = raw_json
        if isinstance(parsed, dict):
            changed = False
            for field_name, meta in token_map.items():
                if not meta.get("json_key"):
                    continue
                if field_name not in parsed:
                    continue
                old = parsed[field_name]
                new = _normalize_token_value(old, meta)
                if new is not old and new != old:
                    parsed[field_name] = new
                    changed = True
            if changed or isinstance(raw_json, dict):
                args["input_json"] = json.dumps(parsed)

    for field_name, meta in token_map.items():
        if meta.get("json_key"):
            continue
        if field_name not in args:
            continue
        old = args[field_name]
        new = _normalize_token_value(old, meta)
        if new is not old and new != old:
            args[field_name] = new
    return args


def _token_field_map_for_tool(tool_name: str) -> dict[str, dict[str, Any]]:
    static_map = TOOL_TOKEN_FIELDS.get(tool_name)
    dynamic_map = _load_mcp_token_fields(tool_name)
    if static_map and dynamic_map:
        merged = dict(dynamic_map)
        merged.update(static_map)
        return merged
    if static_map:
        return static_map
    if dynamic_map:
        return dynamic_map
    return {}


def _load_mcp_token_fields(tool_name: str) -> dict[str, dict[str, Any]]:
    if not tool_name.startswith("mcp__"):
        return {}
    parts = tool_name.split("__", 2)
    if len(parts) != 3:
        return {}
    server_id, inner_tool_name = parts[1], parts[2]
    try:
        from mcp_bridge.config import load_mcp_config

        cfg = load_mcp_config()
        server_cfg = ((cfg.get("mcpServers") or {}).get(server_id) or {})
        raw_manifest = server_cfg.get("token_fields") or {}
        if not isinstance(raw_manifest, dict):
            return {}
        raw_fields = raw_manifest.get(inner_tool_name) or raw_manifest.get("*") or {}
        if not isinstance(raw_fields, dict):
            return {}
        out: dict[str, dict[str, Any]] = {}
        for field_name, raw in raw_fields.items():
            if not isinstance(field_name, str) or not field_name.strip():
                continue
            if isinstance(raw, str):
                kind = raw.strip()
                meta = {"kind": kind}
            elif isinstance(raw, dict):
                kind = str(raw.get("kind") or "").strip()
                meta = {"kind": kind}
                if isinstance(raw.get("lengths"), list):
                    try:
                        meta["lengths"] = tuple(int(x) for x in raw.get("lengths", []))
                    except Exception:
                        pass
                if bool(raw.get("list")):
                    meta["list"] = True
                if bool(raw.get("json_key")):
                    meta["json_key"] = True
            else:
                continue
            if meta.get("kind") in {"uuid", "hex_id", "bg_handle", "stream_id", "attachment_ref"}:
                out[field_name] = meta
        return out
    except Exception:
        return {}


def _apply_aliases(tool_name: str, args: dict) -> dict:
    aliases = TOOL_ARG_ALIASES.get(tool_name) or {}
    if not aliases:
        return args
    for alias, canonical in aliases.items():
        if alias in args and canonical not in args:
            args[canonical] = args.pop(alias)
        elif alias in args:
            # Canonical already provided; drop alias to avoid schema conflicts.
            args.pop(alias, None)
    return args


def _normalize_path_literals_in_code(code: str) -> str:
    """Best-effort normalize malformed file-path literals in Python code strings."""
    if not isinstance(code, str) or not code or not any(ch.isspace() for ch in code):
        return code

    def _fix_literal(match: re.Match) -> str:
        quote = match.group(1)
        inner = match.group(2) if match.lastindex and match.lastindex >= 2 else match.group(1)
        if not inner:
            return match.group(0)
        lowered = inner.lower()
        if lowered.startswith("http://") or lowered.startswith("https://"):
            return match.group(0)
        if ("/" not in inner and "~" not in inner) or not any(ch.isspace() for ch in inner):
            return match.group(0)
        fixed = normalize_path_spaces(inner)
        if fixed == inner:
            return match.group(0)
        return f"{quote}{fixed}{quote}"

    # Normalize both single- and double-quoted short literals.
    code = re.sub(r"(')([^'\n]{1,800})(')", _fix_literal, code)
    code = re.sub(r'(")([^"\n]{1,800})(")', _fix_literal, code)
    return code


def _parse_bool_string(value: str) -> bool | None:
    if not isinstance(value, str):
        return None
    v = value.strip().lower()
    if v in {"true", "1", "yes", "on"}:
        return True
    if v in {"false", "0", "no", "off"}:
        return False
    return None


def _coerce_int_string(value: Any) -> Any:
    if isinstance(value, str):
        v = value.strip()
        if re.fullmatch(r"-?\d+", v):
            try:
                return int(v)
            except Exception:
                return value
    return value


def _normalize_url_whitespace(value: Any) -> Any:
    """Collapse all whitespace in URL strings (local/Hermes models often break hosts)."""
    if not isinstance(value, str):
        return value
    s = value.strip()
    if not s:
        return ""
    return re.sub(r"\s+", "", s)


def _coerce_tool_specific_args(tool_name: str, args: dict) -> dict:
    # delegate_task: some providers emit `tasks` as a JSON array; tolerate tuple from tests.
    if tool_name == "delegate_task":
        t = args.get("tasks")
        if isinstance(t, tuple):
            args["tasks"] = list(t)
        # JSON schema marks `task_description` required, but batch mode ignores it when `tasks`
        # parses to ≥1 valid subtask. Fill a non-empty placeholder so preflight/validation matches
        # runtime behavior (see tools/subagent_tools._parse_delegate_batch_tasks).
        td = args.get("task_description")
        _td_empty = td is None or (isinstance(td, str) and not td.strip())
        if _td_empty:
            from tools.subagent_tools import _parse_delegate_batch_tasks

            _items, _err = _parse_delegate_batch_tasks(args.get("tasks"))
            if _items and not _err:
                args["task_description"] = (
                    "(batch delegation; top-level `task_description` ignored when `tasks` is valid)"
                )

    if tool_name == "mesh_send_files":
        p = args.get("paths")
        if isinstance(p, str):
            s = p.strip()
            parsed: list[str] | None = None
            if s.startswith("["):
                try:
                    loaded = json.loads(s)
                except Exception:
                    loaded = None
                if isinstance(loaded, list):
                    parsed = [str(x).strip() for x in loaded if str(x).strip()]
            args["paths"] = parsed if parsed is not None else ([s] if s else [])
        elif isinstance(p, tuple):
            args["paths"] = [str(x).strip() for x in p if str(x).strip()]
        elif p is None:
            args["paths"] = []

    # create_plugin / create_dashboard_plugin: tolerate string bool drift.
    if tool_name in {"create_plugin", "create_dashboard_plugin"}:
        overwrite = args.get("overwrite")
        parsed = _parse_bool_string(overwrite) if isinstance(overwrite, str) else None
        if parsed is not None:
            args["overwrite"] = parsed

    # read_file_chunk often arrives with numeric args as strings from XML payloads.
    if tool_name == "read_file_chunk":
        for field in ("chunk_index", "num_chunks", "chunk_size"):
            if field in args:
                args[field] = _coerce_int_string(args[field])

    if tool_name == "read_and_analyze_page":
        focus_areas = args.get("focus_areas")
        if isinstance(focus_areas, str):
            parsed = None
            try:
                parsed = json.loads(focus_areas)
            except Exception:
                parsed = None
            if isinstance(parsed, list):
                args["focus_areas"] = [str(x).strip() for x in parsed if str(x).strip()]
            elif focus_areas.strip():
                args["focus_areas"] = [focus_areas.strip()]

    if tool_name in {"deep_web_search", "plan_research_subquestions"}:
        for field in ("num_results", "iteration_hint", "max_questions"):
            if field in args:
                args[field] = _coerce_int_string(args[field])

    if tool_name == "analyze_multiple_sources":
        ao = args.get("analysis_outputs")
        if not isinstance(ao, list) or not any(
            isinstance(x, str) and x.strip() for x in ao
        ):
            alt = args.get("sources")
            if isinstance(alt, list) and any(str(x).strip() for x in alt):
                args["analysis_outputs"] = [str(x).strip() for x in alt if str(x).strip()]
        ao2 = args.get("analysis_outputs")
        if isinstance(ao2, str):
            parsed = None
            try:
                parsed = json.loads(ao2)
            except Exception:
                parsed = None
            if isinstance(parsed, list):
                args["analysis_outputs"] = [str(x).strip() for x in parsed if str(x).strip()]
            elif ao2.strip():
                args["analysis_outputs"] = [ao2.strip()]
        oq = args.get("original_query")
        if not isinstance(oq, str) or not oq.strip():
            for key in ("query", "topic", "research_question"):
                v = args.get(key)
                if isinstance(v, str) and v.strip():
                    args["original_query"] = v.strip()
                    break

    if tool_name == "generate_final_research_report":
        src = args.get("sources")
        if isinstance(src, list):
            args["sources"] = [_normalize_url_whitespace(u) if isinstance(u, str) else u for u in src]

    if tool_name == "search_sessions":
        tid = args.get("target_thread_id")
        if isinstance(tid, str) and tid.strip():
            from tools.session_search_tools import _split_mugi_thread_prefix

            _, stripped = _split_mugi_thread_prefix(tid)
            if stripped:
                args["target_thread_id"] = stripped
        q = args.get("query")
        if isinstance(q, str) and q.strip():
            from tools.session_search_tools import _split_mugi_thread_prefix

            cleaned, auto_tid = _split_mugi_thread_prefix(q)
            if cleaned != q:
                args["query"] = cleaned
            if auto_tid and not args.get("target_thread_id"):
                args["target_thread_id"] = auto_tid
            if not args.get("target_thread_id"):
                from utils.recall_signals import extract_direct_thread_id

                q_for_extract = args.get("query") if isinstance(args.get("query"), str) else q
                extracted = extract_direct_thread_id(q_for_extract if isinstance(q_for_extract, str) else "")
                if extracted:
                    args["target_thread_id"] = extracted

    if "url" in args:
        args["url"] = _normalize_url_whitespace(args["url"])
    return args


def normalize_tool_args(tool_name: str, args: dict) -> dict:
    """Resolve attachment IDs and repair malformed paths in *args* for *tool_name*.

    Mutates and returns *args* (same dict).  Unknown tools or tools not in
    the field map pass through unchanged.

    Respects the attachment_mode setting from config:
      - ``dual``         – resolve IDs + repair paths (default)
      - ``id_preferred`` – same as dual (prompt already emphasises IDs)
      - ``id_strict``    – resolve IDs; reject unresolved ``att_*``/``attachment://`` refs
    """
    _merge_longest_file_path_when_duplicates(tool_name, args)
    args = _apply_aliases(tool_name, args)
    args = _coerce_tool_specific_args(tool_name, args)
    args = _apply_token_normalization(tool_name, args)

    # run_python_code often receives malformed path literals in generated snippets.
    if tool_name == "run_python_code" and isinstance(args.get("code"), str):
        args["code"] = _normalize_path_literals_in_code(args["code"])

    field_map = TOOL_PATH_FIELDS.get(tool_name)
    if not field_map:
        return args

    try:
        from core.config import get_attachment_mode
        mode = get_attachment_mode()
    except Exception:
        mode = "dual"

    has_json_keys = any(m.get("json_key") for m in field_map.values())

    if has_json_keys:
        raw_json = args.get("input_json")
        parsed = None
        if isinstance(raw_json, str):
            try:
                parsed = json.loads(raw_json)
            except (json.JSONDecodeError, TypeError):
                parsed = None
        elif isinstance(raw_json, dict):
            # Agent loop parses LLM JSON into a dict; tools like split_pdf expect input_json: str.
            # Resolve paths in the dict, then re-serialize so the tool receives a string.
            parsed = raw_json
        if isinstance(parsed, dict):
            changed = False
            for field_name, meta in field_map.items():
                if not meta.get("json_key"):
                    continue
                if field_name in parsed:
                    old = parsed[field_name]
                    new = _resolve_value(old, meta)
                    if new is not old and new != old:
                        parsed[field_name] = new
                        changed = True
                    if mode == "id_strict" and meta.get("intent") == "read":
                        _reject_unresolved(parsed.get(field_name), field_name, tool_name)
            # Re-serialize so tools that declare input_json: str get a string (avoids Pydantic validation errors).
            if changed or isinstance(raw_json, dict):
                args["input_json"] = json.dumps(parsed)

    for field_name, meta in field_map.items():
        if meta.get("json_key"):
            continue
        if field_name in args:
            old = args[field_name]
            new = _resolve_value(old, meta)
            if new is not old and new != old:
                args[field_name] = new
            if mode == "id_strict" and meta.get("intent") == "read":
                _reject_unresolved(args.get(field_name), field_name, tool_name)

    return args


def _reject_unresolved(value: Any, field_name: str, tool_name: str) -> None:
    """In strict mode, raise if value still looks like an unresolved attachment ref."""
    if not isinstance(value, str):
        return
    v = value.strip()
    if v.startswith(_ar.ATTACHMENT_ID_PREFIX) or (v.startswith("att_") and v[4:].isdigit()):
        raise ValueError(
            f"Unresolved attachment reference '{v}' for {tool_name}.{field_name}. "
            "The attachment may have expired or was never registered."
        )


def proactive_tool_allowed(
    tool_name: str,
    autonomy_mode: str,
    allowlist: list[str] | None = None,
    aggressive_allowlist: list[str] | None = None,
) -> bool:
    """Central proactive policy gate shared by autonomy callers."""
    mode = str(autonomy_mode or "bounded").strip().lower()
    if mode == "off":
        return False
    if mode == "aggressive":
        if aggressive_allowlist:
            agg = {str(x).strip() for x in aggressive_allowlist if str(x).strip()}
            if agg:
                return str(tool_name or "").strip() in agg
        return True
    allowed = {str(x).strip() for x in (allowlist or []) if str(x).strip()}
    return str(tool_name or "").strip() in allowed
