"""Path recovery, JSON lenience, and tool-argument hygiene for tool calls.

Extracted from agent_loop.py — do not import agent_loop from here.
"""
from __future__ import annotations

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

from tools.conversation_trace import emit_guardrail_event, is_trace_enabled
from utils.path_safety import normalize_path_spaces, safe_resolve_path
from utils.transcript_roles import is_real_user_message
from utils.llm_token_normalize import normalize_kt_task_id_token

_log = logging.getLogger(__name__)

_RECOVERY_PREFER_EXISTING_DIR: frozenset[str] = frozenset({"list_directory", "search_filesystem"})
_RECOVERY_CONFUSABLE_PATH_TOKENS = (
    "\u2215",  # division slash
    "\u2044",  # fraction slash
    "\uff0f",  # fullwidth slash
    "\uff3c",  # fullwidth backslash
    "\uff1a",  # fullwidth colon
)
_RECOVERY_SCHEME_RE = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE)
_RECOVERY_PATH_TOKEN_RE = re.compile(
    r"(?<![A-Za-z0-9_])((?:~|\.{1,2})?/[^<>'\"`\s]+|(?:\./|\.\./)[^<>'\"`\s]+)"
)
_RECOVERY_PATH_KV_RE_CACHE: re.Pattern[str] | None = None


def _recovery_path_kv_re() -> re.Pattern[str]:
    """Key=value path hints in prose — pattern covers all scalar path args from ``TOOL_PATH_FIELDS``."""
    global _RECOVERY_PATH_KV_RE_CACHE
    if _RECOVERY_PATH_KV_RE_CACHE is None:
        from tool_arg_normalizer import TOOL_PATH_FIELDS

        names = sorted(
            {
                k
                for fields in TOOL_PATH_FIELDS.values()
                for k, m in fields.items()
                if not m.get("list")
            }
        )
        inner = "|".join(re.escape(n) for n in names)
        _RECOVERY_PATH_KV_RE_CACHE = re.compile(
            rf"(?:{inner})\s*[:=]\s*[\"']([^\"'`\n]{{1,800}})[\"'`]",
            re.IGNORECASE,
        )
    return _RECOVERY_PATH_KV_RE_CACHE
def _extract_recovery_path_candidates(text: str) -> list[str]:
    """Extract unique path-like candidates from assistant content in one pass."""
    if not isinstance(text, str) or not text:
        return []
    out: list[str] = []
    seen: set[str] = set()

    for m in _recovery_path_kv_re().finditer(text):
        cand = (m.group(1) or "").strip()
        if cand and cand not in seen:
            seen.add(cand)
            out.append(cand)

    for m in _RECOVERY_PATH_TOKEN_RE.finditer(text):
        cand = (m.group(1) or "").strip()
        if cand and cand not in seen:
            seen.add(cand)
            out.append(cand)
    return out


def _validate_recovered_path_candidate(candidate: str) -> tuple[str | None, str]:
    """Return (normalized_candidate, reject_reason)."""
    if not isinstance(candidate, str):
        return None, "candidate_not_string"
    raw = candidate.strip()
    if not raw:
        return None, "candidate_empty"
    if len(raw) > 512:
        return None, "candidate_too_long"
    if any(tok in raw for tok in _RECOVERY_CONFUSABLE_PATH_TOKENS):
        return None, "candidate_confusable_separator"
    if any(ch in raw for ch in ("\n", "\r", "\t", "\x00")):
        return None, "candidate_control_char"
    if _RECOVERY_SCHEME_RE.match(raw):
        return None, "candidate_non_filesystem_scheme"
    if any(tok in raw for tok in ("*", "?", "[", "]", "{", "}")):
        return None, "candidate_glob_like"

    normalized = unicodedata.normalize("NFKC", raw)
    normalized = normalize_path_spaces(normalized.strip())
    if not normalized:
        return None, "candidate_normalized_empty"
    try:
        from tool_arg_normalizer import path_value_looks_like_tool_markup_artifact

        if path_value_looks_like_tool_markup_artifact(normalized):
            return None, "candidate_markup_artifact"
    except Exception:
        _log.debug("Recovery candidate markup check failed", exc_info=True)
    return normalized, ""


def _score_path_candidate_for_recovery(
    tool_name: str,
    cand: str,
    resolved: str,
    workspace_dir: str,
) -> int:
    """Higher is better when choosing among multiple filesystem path candidates."""
    score = 0
    low = cand.lower().strip()
    base = os.path.basename(low.strip("/"))
    if base in {"think", "parameter", "tool_call", "tool_response", "function"}:
        score -= 40
    is_file = os.path.isfile(resolved)
    is_dir = os.path.isdir(resolved)
    if tool_name in _RECOVERY_PREFER_EXISTING_DIR:
        if is_dir:
            score += 100
        elif is_file:
            score += 25
    else:
        if is_file:
            score += 100
        elif is_dir:
            score += 15
    ws_abs = ""
    try:
        ws_abs = os.path.abspath(os.path.expanduser(workspace_dir or ""))
    except Exception:
        ws_abs = ""
    if ws_abs:
        try:
            if os.path.commonpath([ws_abs, resolved]) == ws_abs:
                score += 20
        except Exception:
            pass
    if "/.mugi/workspace/" in low or low.startswith("~/.mugi/workspace/"):
        score += 20
    if base in {"bootstrap.md", "identity.md", "soul.md", "user.md", "memory.md", "agents.md"}:
        score += 15
    return score


def _repair_artifact_path_values_in_params(
    tool_name: str,
    params: dict[str, str],
    block: str,
    recovery_context: str | None = None,
) -> dict[str, str]:
    """Replace markup-like path fragments from XML parsing with paths found in the block.

    Local models often put the real path in CoT *outside* ``<tool_call>...</tool_call>`` while
    the XML parameter value is a junk fragment like ``/parameter``. When ``recovery_context``
    is the full assistant message for the turn, we scan it too.
    """
    from tool_arg_normalizer import (
        TOOL_PATH_FIELDS,
        iter_scalar_top_level_path_fields,
        normalize_tool_args,
        path_value_looks_like_tool_markup_artifact,
    )

    if not tool_name or not TOOL_PATH_FIELDS.get(tool_name) or not params:
        return params
    out = dict(params)
    tmp: dict[str, Any] = dict(out)
    try:
        normalize_tool_args(tool_name, tmp)
    except Exception:
        tmp = dict(out)
    fields_needing_repair: list[str] = []
    for field_name, _meta in iter_scalar_top_level_path_fields(tool_name):
        val = tmp.get(field_name)
        if isinstance(val, str) and path_value_looks_like_tool_markup_artifact(val):
            fields_needing_repair.append(field_name)
    if not fields_needing_repair:
        return out
    search_text = "\n".join(
        x for x in (block or "", (recovery_context or "").strip()) if x
    )
    candidates = _extract_recovery_path_candidates(search_text)
    valid: list[str] = []
    seen: set[str] = set()
    for c in candidates:
        n, _reason = _validate_recovered_path_candidate(c)
        if not n:
            continue
        try:
            if path_value_looks_like_tool_markup_artifact(n):
                continue
        except Exception:
            pass
        if n in seen:
            continue
        seen.add(n)
        valid.append(n)
    ci = 0
    for field_name in fields_needing_repair:
        if ci >= len(valid):
            break
        out[field_name] = valid[ci]
        ci += 1
    return out


def _repair_artifact_path_values_in_tool_args_dict(
    tool_name: str,
    args: dict[str, Any],
    block: str,
    recovery_context: str | None = None,
) -> dict[str, Any]:
    """Like ``_repair_artifact_path_values_in_params`` but preserves non-string argument values."""
    if not isinstance(args, dict):
        return args
    str_map = {k: v for k, v in args.items() if isinstance(v, str)}
    if not str_map:
        return args
    repaired_str = _repair_artifact_path_values_in_params(
        tool_name, str_map, block, recovery_context=recovery_context
    )
    out = dict(args)
    out.update(repaired_str)
    return out
def _escape_control_chars_in_json_strings(s: str) -> str:
    """Escape literal control characters (newlines, tabs, etc.) inside JSON string values.

    Local models sometimes emit unescaped newlines inside JSON strings, especially
    when generating multi-line content like YAML frontmatter.  Standard json.loads
    rejects these per RFC 8259.  This walks the string character-by-character,
    tracking whether we're inside a JSON string, and replaces literal control chars
    with their JSON escape sequences.
    """
    result: list[str] = []
    in_string = False
    esc = False
    for ch in s:
        if in_string:
            if esc:
                esc = False
                result.append(ch)
            elif ch == "\\":
                esc = True
                result.append(ch)
            elif ch == '"':
                in_string = False
                result.append(ch)
            elif ch == "\n":
                result.append("\\n")
            elif ch == "\r":
                result.append("\\r")
            elif ch == "\t":
                result.append("\\t")
            else:
                result.append(ch)
        else:
            if ch == '"':
                in_string = True
            result.append(ch)
    return "".join(result)


def _json_loads_lenient(s: str) -> Any:
    """json.loads with fallback that escapes literal control chars in string values."""
    try:
        return json.loads(s)
    except (json.JSONDecodeError, ValueError):
        return json.loads(_escape_control_chars_in_json_strings(s))


def _extract_balanced_json_value(text: str) -> str | None:
    """Return the first balanced JSON object/array prefix from text, or None."""
    if not isinstance(text, str):
        return None
    s = text.lstrip()
    if not s or s[0] not in "{[":
        return None
    opener = s[0]
    closer = "}" if opener == "{" else "]"
    depth = 0
    in_string = False
    escape = False
    for i, ch in enumerate(s):
        if in_string:
            if escape:
                escape = False
            elif ch == "\\":
                escape = True
            elif ch == '"':
                in_string = False
            continue
        if ch == '"':
            in_string = True
            continue
        if ch == opener:
            depth += 1
        elif ch == closer:
            depth -= 1
            if depth == 0:
                return s[: i + 1]
    return None


def _quote_json_value(m: re.Match) -> str:
    """Re.sub callback: quote an unquoted JSON value after a colon."""
    raw = m.group(1).strip()
    if raw.lower() in ("true", "false", "null"):
        return ": " + raw.lower()
    try:
        float(raw)
        return ": " + raw
    except ValueError:
        return ': "' + raw.replace('"', '\\"') + '"'


def _salvage_json_tool_call(block: str) -> dict | None:
    """Best-effort parse for malformed JSON tool-call payloads."""
    if not isinstance(block, str):
        return None
    name_match = re.search(r'"(?:name|function)"\s*:\s*"([^"]+)"', block)
    if not name_match:
        return None
    name = (name_match.group(1) or "").strip()
    if not name:
        return None
    args_str = "{}"
    args_match = re.search(r'"(?:arguments|parameters)"\s*:\s*', block)
    if args_match:
        tail = block[args_match.end() :]
        balanced = _extract_balanced_json_value(tail)
        if balanced:
            args_str = balanced
    if name in {"read_file", "read_file_chunk", "list_directory"}:
        try:
            parsed_args = _json_loads_lenient(args_str)
        except Exception:
            parsed_args = {}
        if isinstance(parsed_args, dict):
            try:
                from tool_arg_normalizer import normalize_tool_args, tool_call_has_path_markup_artifact

                normalized_args = normalize_tool_args(name, dict(parsed_args))
                if tool_call_has_path_markup_artifact(name, normalized_args):
                    return None
            except Exception:
                _log.debug("Salvage path-artifact check failed", exc_info=True)
    return {"name": name, "arguments": args_str}

# URL and UUID whitespace hygiene. Local models (especially under MLX /
# llama.cpp chat templates) sometimes emit strings with a space injected after
# every `.` — e.g. "https://gorilla. cs. berkeley. edu/" or a Vault note id
# like "4d 3255c8-99f9-4368-93f9-ff5d 2467f4ee". Those values fail at the
# transport layer and the agent just re-emits the same broken call, burning
# the entire round budget (see trace 33b07417-*). We repair them here before
# any other preflight so the cleaned call is validated and executed.
_URL_SCHEME_RE = re.compile(r"^https?://", re.IGNORECASE)
_URL_WHITESPACE_RE = re.compile(r"\s+")
# UUID shape with optional internal whitespace between hex/dash runs.
_UUID_LOOSE_RE = re.compile(
    r"^[0-9a-fA-F\s]{8,}-[0-9a-fA-F\s-]{20,}$"
)
_STRICT_UUID_RE = re.compile(
    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
)


def _collapse_spaced_uuids_in_jsonish_text(s: str) -> str:
    """Best-effort: collapse internal whitespace in spans that are corrupted UUIDs.

    Gemma / channel-style tool blocks often inject spaces inside UUIDs so ``json.loads``
    fails and salvage returns ``{}``.  This runs on the raw ``{...}`` substring *before*
    JSON parsing.
    """
    if not isinstance(s, str):
        return s
    if not any(ch.isspace() for ch in s):
        return s

    def repl(m: re.Match[str]) -> str:
        span = m.group(1)
        collapsed = _URL_WHITESPACE_RE.sub("", span)
        cl = collapsed.lower()
        if _STRICT_UUID_RE.match(cl):
            return collapsed.lower()
        return m.group(0)

    return re.sub(
        r"(?<![0-9a-fA-F])([0-9a-fA-F](?:[0-9a-fA-F\s-]){15,47})(?![0-9a-fA-F])",
        repl,
        s,
        flags=re.IGNORECASE,
    )


def _collapse_spaced_urls_in_jsonish_quoted_strings(s: str) -> str:
    """Strip internal whitespace from http(s) URLs inside double-quoted JSON-ish fragments."""
    if not isinstance(s, str) or "http" not in s.lower():
        return s

    def repl(m: re.Match[str]) -> str:
        inner = m.group(1)
        if _URL_SCHEME_RE.match(inner) and _URL_WHITESPACE_RE.search(inner):
            return '"' + _URL_WHITESPACE_RE.sub("", inner) + '"'
        return m.group(0)

    return re.sub(r'"((?:https?://)[^"]*)"', repl, s, flags=re.IGNORECASE)


def _preclean_gemma_tool_call_brace_args(raw_args: str) -> str:
    """Prepare Gemma ``call:tool{...}`` JSON-ish text before ``json.loads``."""
    s = raw_args or ""
    if not s.strip():
        return s
    s = _collapse_spaced_uuids_in_jsonish_text(s)
    s = _collapse_spaced_urls_in_jsonish_quoted_strings(s)
    return s


def _collapse_http_url_at(s: str, start: int) -> tuple[str, int] | None:
    """Collapse tokenizer spaces inside an http(s) URL starting at *start*; return (collapsed, end_idx)."""
    raw = s[start:]
    m = re.match(r"https?://", raw, re.IGNORECASE)
    if not m:
        return None
    scheme_end = start + m.end()
    i = scheme_end
    host_parts: list[str] = []
    while i < len(s):
        ch = s[i]
        if ch in "/?#":
            break
        if ch.isspace():
            while i < len(s) and s[i].isspace():
                i += 1
            continue
        if ch.isalnum() or ch in ".-":
            host_parts.append(ch)
            i += 1
            continue
        break
    host = "".join(host_parts)
    parts: list[str] = [s[start:scheme_end], host]
    if i < len(s) and s[i] == "/":
        j = i
        while j < len(s) and not s[j].isspace():
            j += 1
        parts.append(s[i:j])
        i = j
    elif i < len(s) and s[i] in "?#":
        anchor = s[i]
        i += 1
        frag: list[str] = [anchor]
        while i < len(s) and not s[i].isspace():
            frag.append(s[i])
            i += 1
        parts.append("".join(frag))
    collapsed = "".join(parts)
    return collapsed, i


def _repair_embedded_http_urls(s: str) -> tuple[str, bool]:
    """Repair spaced http(s) URLs when the whole string is not only a URL (e.g. shell commands)."""
    if not isinstance(s, str) or not s:
        return s, False
    if not _URL_WHITESPACE_RE.search(s):
        return s, False
    if not re.search(r"https?://", s, re.IGNORECASE):
        return s, False
    matches = list(re.finditer(r"https?://", s, re.IGNORECASE))
    if not matches:
        return s, False
    out: list[str] = []
    last = 0
    changed = False
    for m in matches:
        start = m.start()
        if start < last:
            continue
        collapsed_pair = _collapse_http_url_at(s, start)
        if collapsed_pair is None:
            continue
        ctext, end = collapsed_pair
        if ctext != s[start:end]:
            changed = True
        out.append(s[last:start])
        out.append(ctext)
        last = end
    out.append(s[last:])
    return "".join(out), changed


def _repair_arg_whitespace(value: Any) -> tuple[Any, bool]:
    """Return (value, changed). Repairs tokenizer-induced whitespace in URLs / UUIDs.

    - For strings starting with ``http://`` or ``https://`` that contain any
      internal whitespace, strip all whitespace.  URLs never legitimately
      contain unescaped whitespace, so this is safe.
    - For strings that *look* UUID-shaped but have stray whitespace, strip
      whitespace iff the result is a canonical UUID.
    - Walks lists / dicts recursively; leaves other scalars alone.
    """
    if isinstance(value, str):
        s = value
        if _URL_SCHEME_RE.match(s) and _URL_WHITESPACE_RE.search(s):
            return _URL_WHITESPACE_RE.sub("", s), True
        new_s, emb = _repair_embedded_http_urls(s)
        if emb:
            return new_s, True
        # Kanban task ids (kt_ + 14 hex) often get split by tokenizer spaces.
        normalized_kt = normalize_kt_task_id_token(s)
        if normalized_kt != s:
            return normalized_kt, True
        if _UUID_LOOSE_RE.match(s) and _URL_WHITESPACE_RE.search(s):
            stripped = _URL_WHITESPACE_RE.sub("", s).lower()
            if _STRICT_UUID_RE.match(stripped):
                return stripped, True
        return s, False
    if isinstance(value, list):
        out: list[Any] = []
        changed = False
        for item in value:
            new_item, item_changed = _repair_arg_whitespace(item)
            out.append(new_item)
            changed = changed or item_changed
        return out, changed
    if isinstance(value, dict):
        out_d: dict[Any, Any] = {}
        changed = False
        for k, v in value.items():
            new_v, v_changed = _repair_arg_whitespace(v)
            out_d[k] = new_v
            changed = changed or v_changed
        return out_d, changed
    return value, False


def _finalize_xml_parsed_arguments_string(args_str: str) -> str:
    """Apply URL/UUID whitespace repair to JSON ``arguments`` produced by XML salvage."""
    if not isinstance(args_str, str):
        return str(args_str)
    raw = args_str.strip()
    if not raw:
        return raw
    try:
        parsed = _json_loads_lenient(raw)
    except Exception:
        return args_str
    if not isinstance(parsed, (dict, list)):
        return args_str
    repaired, _ = _repair_arg_whitespace(parsed)
    try:
        return json.dumps(repaired, separators=(",", ":"))
    except Exception:
        return args_str


def _preflight_repair_arg_whitespace(
    tool_calls_list: list[dict[str, Any]],
    on_log: Any | None = None,
    thread_id: str | None = None,
    round_: int | None = None,
) -> list[dict[str, Any]]:
    """Rewrite each tool call's ``arguments`` JSON with whitespace-repaired values.

    Logs a single ``arg_whitespace_hygiene`` guardrail event per repaired call so
    operators can audit without spamming; no message is appended to the chat —
    the fix is silent to the model by design (retrying in one round beats
    another round of corrective prose).
    """
    if not isinstance(tool_calls_list, list) or not tool_calls_list:
        return tool_calls_list
    repaired_count = 0
    for tc in tool_calls_list:
        if not isinstance(tc, dict):
            continue
        raw = tc.get("arguments")
        if not isinstance(raw, str) or not raw.strip():
            continue
        try:
            parsed = _json_loads_lenient(raw)
        except Exception:
            continue
        if not isinstance(parsed, (dict, list)):
            continue
        new_parsed, changed = _repair_arg_whitespace(parsed)
        if not changed:
            continue
        try:
            tc["arguments"] = json.dumps(new_parsed, separators=(",", ":"))
        except Exception:
            _log.debug("arg hygiene: failed to re-serialize repaired args", exc_info=True)
            continue
        repaired_count += 1
        name = tc.get("name") or "?"
        if on_log:
            on_log(
                "AGENT",
                f"Arg hygiene: stripped stray whitespace in URL/UUID args for `{name}`.",
            )
        if is_trace_enabled() and thread_id and round_ is not None:
            try:
                emit_guardrail_event(
                    thread_id,
                    round_,
                    "arg_whitespace_hygiene",
                    "repaired",
                    details={"tool": name},
                )
            except Exception:
                _log.debug("arg hygiene: failed to emit guardrail event", exc_info=True)
    if repaired_count and on_log:
        _log.debug("arg hygiene repaired %d tool call(s)", repaired_count)
    return tool_calls_list


def _preflight_filter_path_artifact_tool_calls(
    tool_calls_list: list[dict[str, Any]],
    tools_by_name: dict[str, Any],
    full_tools_by_name: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[tuple[str, dict[str, Any], dict[str, Any]]]]:
    """Filter tool calls whose normalized path fields look like markup artifacts.

    Poisoned entries include (tool_name, parsed_args, original_tc) so callers can
    reconstruct the call when recovery succeeds.
    """
    filtered: list[dict[str, Any]] = []
    poisoned: list[tuple[str, dict[str, Any], dict[str, Any]]] = []
    for tc in tool_calls_list:
        name = tc.get("name", "")
        tool_obj = tools_by_name.get(name) or full_tools_by_name.get(name)
        if tool_obj is None:
            filtered.append(tc)
            continue
        try:
            parsed_args = _json_loads_lenient(tc.get("arguments", "{}"))
        except Exception:
            parsed_args = {}
        if not isinstance(parsed_args, dict):
            parsed_args = {}
        check_args = dict(parsed_args)
        try:
            from tool_arg_normalizer import normalize_tool_args, tool_call_has_path_markup_artifact

            normalize_tool_args(name, check_args)
            if tool_call_has_path_markup_artifact(name, check_args):
                poisoned.append((name, dict(parsed_args), dict(tc)))
                continue
        except Exception:
            _log.debug("Path-artifact preflight normalize failed", exc_info=True)
        filtered.append(tc)
    return filtered, poisoned
_QUERY_LIKE_FIELDS: frozenset[str] = frozenset({
    "query", "topic", "search_query", "question", "prompt",
})

# Scratch tools carry structured topic/content — never run search-query salvage on them.
_TOOLS_SKIP_QUERY_ARG_RECOVERY: frozenset[str] = frozenset({
    "append_research_scratch",
})

_SEARCH_QUERY_TOOLS: frozenset[str] = frozenset({
    "deep_web_search",
    "search_web",
    "search_web_public",
})

_WRITE_PATH_TOOLS: frozenset[str] = frozenset({
    "write_text_file",
    "save_to_file",
    "append_to_text_file",
})

_CONTENT_EXCLUDE_KEYS: frozenset[str] = frozenset({
    "content",
    "body",
    "html",
    "text",
    "data",
})

_HYPERFRAMES_INDEX_PATH_RE = re.compile(
    r"(/[^\s()\"']+/hyperframes/projects/[^/\s()\"']+/index\.html)",
    re.IGNORECASE,
)
_HF_INDEX_BYTES_RE = re.compile(
    r"index\.html:\s*\d+\s+bytes\s+\(([^)]+)\)",
    re.IGNORECASE,
)


def _compact_query_from_worker_brief(text: str) -> str:
    """Derive a short search/topic string from kanban/subagent Context/Task briefs."""
    raw = (text or "").strip()
    if not raw:
        return ""
    task_m = re.search(r"Task:\s*(.+?)(?:\n|$)", raw, re.IGNORECASE)
    if task_m:
        q = task_m.group(1).strip()
        if q:
            return q[:500]
    ctx_m = re.search(r"Context:\s*(.+?)(?:\nTask:|\Z)", raw, re.IGNORECASE | re.DOTALL)
    if ctx_m:
        q = " ".join(ctx_m.group(1).split())
        if q:
            return q[:500]
    return raw[:500]


def _extract_worker_task_brief(messages: list[dict[str, Any]]) -> str:
    """Return compact task text from the first Context/Task worker message (trim-safe)."""
    for msg in messages or []:
        if not is_real_user_message(msg):
            continue
        content = msg.get("content")
        candidate = ""
        if isinstance(content, str) and content.strip():
            candidate = content.strip()
        elif isinstance(content, list):
            for part in content:
                if isinstance(part, dict) and part.get("type") == "text":
                    t = part.get("text", "")
                    if isinstance(t, str) and t.strip():
                        candidate = t.strip()
                        break
        if not candidate or "Context:" not in candidate or "Task:" not in candidate:
            continue
        return _compact_query_from_worker_brief(candidate)
    return ""


def _is_unusable_recovered_query(text: str) -> bool:
    """Reject guard-retry blobs and other non-search strings from query recovery."""
    from utils.path_normalize import is_internal_guard_user_text

    if not isinstance(text, str) or not text.strip():
        return True
    if is_internal_guard_user_text(text):
        return True
    lower = text.lower()
    if "missing required arguments" in lower:
        return True
    if lower.startswith("your previous tool call"):
        return True
    if "for `append_research_scratch`" in lower:
        return True
    if "kanban completion repair" in lower:
        return True
    if "every tool call in the last round failed" in lower:
        return True
    if "before writing prose, repair" in lower:
        return True
    if len(text) > 400 and any(
        marker in lower
        for marker in (
            "retry with",
            "executable tool call",
            "after normalization",
            "return exactly one executable",
        )
    ):
        return True
    return False


def _recover_query_from_prior_tool_calls(
    messages: list[dict[str, Any]],
    tool_name: str,
    field: str,
) -> str:
    """Reuse the most recent non-empty query-like arg from earlier tool calls."""
    want = str(field or "query").strip()
    if not want:
        return ""
    for msg in reversed(messages or []):
        tcs = msg.get("tool_calls")
        if not isinstance(tcs, list):
            continue
        for tc in tcs:
            if not isinstance(tc, dict):
                continue
            name = str(tc.get("name") or "").strip()
            if not name and isinstance(tc.get("function"), dict):
                name = str(tc["function"].get("name") or "").strip()
            if name != tool_name:
                continue
            raw = tc.get("arguments")
            if raw is None and isinstance(tc.get("function"), dict):
                raw = tc["function"].get("arguments")
            parsed: dict[str, Any] | None = None
            if isinstance(raw, dict):
                parsed = raw
            elif isinstance(raw, str) and raw.strip():
                try:
                    p = json.loads(raw)
                    if isinstance(p, dict):
                        parsed = p
                except Exception:
                    parsed = None
            if not isinstance(parsed, dict):
                continue
            val = parsed.get(want)
            if isinstance(val, str) and val.strip():
                return val.strip()
    return ""


def prefill_kanban_xml_search_tool_calls(
    tool_calls: list[dict[str, Any]],
    *,
    messages: list[dict[str, Any]] | None = None,
    assistant_content: str = "",
) -> list[dict[str, Any]]:
    """Fill empty query/topic args on XML-salvaged calls before invalid-required handling."""
    if not tool_calls:
        return tool_calls
    out: list[dict[str, Any]] = []
    for tc in tool_calls:
        if not isinstance(tc, dict):
            out.append(tc)
            continue
        name = str(tc.get("name") or "").strip()
        if name not in _SEARCH_QUERY_TOOLS and name != "plan_research_subquestions":
            out.append(tc)
            continue
        try:
            parsed = _parse_tool_call_arguments(tc.get("arguments"))
        except Exception:
            parsed = {}
        if not isinstance(parsed, dict):
            parsed = {}
        missing = [
            f
            for f in ("query", "topic")
            if f in _QUERY_LIKE_FIELDS
            and (f not in parsed or not str(parsed.get(f) or "").strip())
        ]
        if not missing:
            out.append(tc)
            continue
        recovered, _reason = _attempt_recover_query_args(
            name,
            missing,
            parsed,
            list(messages or []),
            assistant_content=assistant_content,
        )
        if not recovered:
            out.append(tc)
            continue
        merged = dict(parsed)
        merged.update(recovered)
        row = dict(tc)
        row["arguments"] = json.dumps(merged, ensure_ascii=False)
        out.append(row)
    return out


def _attempt_recover_query_args(
    tool_name: str,
    missing_fields: list[str],
    raw_args: dict[str, Any],
    messages: list[dict[str, Any]],
    assistant_content: str = "",
) -> tuple[dict[str, Any] | None, str]:
    """Try to fill missing query-like string fields from context.

    Recovery sources, in priority order:
    1. Most recent real user message (if it looks like a search topic).
    2. Quoted phrases in the assistant's own content for this turn
       (the model often writes ``"I'll search for 'X'"`` then emits empty args).

    Returns (recovered_args_dict, reason_string) — same contract as
    ``_attempt_recover_read_tool_args`` so callers can chain.
    """
    if not missing_fields:
        return None, "no_missing_fields"

    if tool_name in _TOOLS_SKIP_QUERY_ARG_RECOVERY:
        return None, "tool_skips_query_recovery"

    query_missing = [f for f in missing_fields if f in _QUERY_LIKE_FIELDS]
    if not query_missing:
        return None, "no_query_fields_missing"

    # Only auto-fill when the tool call was essentially empty — the model chose
    # the right tool but failed to serialise any arguments.
    non_empty_provided = {
        k for k, v in (raw_args or {}).items()
        if v is not None and (not isinstance(v, str) or v.strip())
    }
    if non_empty_provided:
        return None, "non_empty_args_present"

    # Extract the most recent real user message text (skip internal guard injections).
    from utils.path_normalize import is_internal_guard_user_text

    user_text = ""
    _user_turns_scanned = 0
    for msg in reversed(messages or []):
        if not is_real_user_message(msg):
            continue
        _user_turns_scanned += 1
        if _user_turns_scanned > 12:
            break
        content = msg.get("content")
        candidate = ""
        if isinstance(content, str) and content.strip():
            candidate = content.strip()
        elif isinstance(content, list):
            for part in content:
                if isinstance(part, dict) and part.get("type") == "text":
                    t = part.get("text", "")
                    if isinstance(t, str) and t.strip():
                        candidate = t.strip()
                        break
        if not candidate:
            continue
        if is_internal_guard_user_text(candidate):
            continue
        user_text = candidate
        break

    # ── Source 1: user message ────────────────────────────────────────────
    user_text_usable = bool(user_text)
    if user_text_usable:
        if "missing required arguments" in user_text.lower():
            user_text_usable = False
        else:
            _MIN_QUERY_WORDS = 3
            _words = user_text.split()
            if len(_words) < _MIN_QUERY_WORDS:
                user_text_usable = False
            else:
                _lower = user_text.lower()
                _CONVERSATIONAL_STARTS = (
                    "you ", "please ", "try ", "can you ", "could you ",
                    "do it", "yes", "no", "ok", "sure", "thanks", "thank you",
                    "go ahead", "start", "stop", "wait", "never mind",
                )
                if any(_lower.startswith(p) for p in _CONVERSATIONAL_STARTS):
                    user_text_usable = False

    if user_text_usable:
        MAX_INJECT_LEN = 2000
        inject = user_text[:MAX_INJECT_LEN] if len(user_text) > MAX_INJECT_LEN else user_text
        if (
            tool_name in _SEARCH_QUERY_TOOLS
            and query_missing[0] == "query"
            and "Context:" in inject
            and "Task:" in inject
        ):
            inject = _compact_query_from_worker_brief(inject) or inject
        if _is_unusable_recovered_query(inject):
            user_text_usable = False
        else:
            field = query_missing[0]
            return {field: inject}, "query_recovered_from_user_message"

    # ── Source 2: quoted phrase in the assistant's own content ─────────────
    # Models often narrate their intent — "I'll search for 'iPlug2 GitHub'"
    # — then emit the tool call with empty arguments.  Prefer this over the
    # kanban task title, which query-recovery used to replay identically.
    if assistant_content:
        if _is_unusable_recovered_query(assistant_content):
            return None, "assistant_content_unusable"
        _QUOTE_PATTERNS = [
            re.compile(r'"(.+?)"', re.DOTALL),
            re.compile(r'[\u201c](.+?)[\u201d]', re.DOTALL),
            # Single quotes: require whitespace/start before opening quote
            # to avoid matching contractions like "I'll".
            re.compile(r"(?:^|(?<=\s))'(.+?)'(?=\s|[.,;:!?)\]]|$)", re.DOTALL),
            re.compile(r'[\u2018](.+?)[\u2019]', re.DOTALL),
        ]
        candidates: list[str] = []
        for pat in _QUOTE_PATTERNS:
            candidates.extend(pat.findall(assistant_content))
        best = ""
        for c in candidates:
            c = c.strip()
            if is_internal_guard_user_text(c):
                continue
            if 3 <= len(c) <= 200 and len(c.split()) >= 2:
                if len(c) > len(best):
                    best = c
        if best:
            if _is_unusable_recovered_query(best):
                return None, "assistant_quote_unusable"
            field = query_missing[0]
            return {field: best}, "query_recovered_from_assistant_content"

    extracted_from_message = _extract_worker_task_brief(messages)
    sticky_brief = ""
    try:
        from utils.kanban_dispatcher import current_kanban_worker_search_brief

        sticky_brief = (current_kanban_worker_search_brief() or "").strip()
    except Exception:
        sticky_brief = ""

    # Kanban workers pin task text before summarization trims Context/Task. MLX often
    # emits XML tool calls with `{}` and zero assistant prose — use the pin when the
    # message extract is gone (do not treat that as "replay task title" loop bait).
    if not extracted_from_message and sticky_brief:
        compact = _compact_query_from_worker_brief(sticky_brief) or sticky_brief
        if not _is_unusable_recovered_query(compact):
            field = query_missing[0]
            return {field: compact}, "query_recovered_from_kanban_search_brief"

    worker_brief = extracted_from_message
    if worker_brief and not _is_unusable_recovered_query(worker_brief):
        reuse_task_title = (
            tool_name in _SEARCH_QUERY_TOOLS
            and sticky_brief
            and extracted_from_message == sticky_brief
        )
        if not reuse_task_title:
            field = query_missing[0]
            query_val = worker_brief
            if tool_name in _SEARCH_QUERY_TOOLS and query_missing[0] == "query":
                query_val = _compact_query_from_worker_brief(worker_brief) or worker_brief
            return {field: query_val}, "query_recovered_from_worker_brief"

    if tool_name in _SEARCH_QUERY_TOOLS:
        prior = _recover_query_from_prior_tool_calls(messages, tool_name, query_missing[0])
        if prior and not _is_unusable_recovered_query(prior):
            field = query_missing[0]
            return {field: prior}, "query_recovered_from_prior_tool_call"

    if not user_text:
        return None, "no_user_text"
    return None, "user_text_not_usable_for_query"


def _parse_tool_call_arguments(raw: Any) -> dict[str, Any]:
    """Normalize tool-call argument payloads from assistant history rows."""
    if isinstance(raw, dict):
        if isinstance(raw.get("function"), dict):
            fn = raw["function"]
            inner = fn.get("arguments")
            if inner is not None:
                raw = inner
            else:
                raw = raw.get("arguments", raw)
        elif raw.get("arguments") is not None:
            raw = raw.get("arguments")
        else:
            return dict(raw)
    if isinstance(raw, str):
        try:
            parsed = json.loads(raw)
            return parsed if isinstance(parsed, dict) else {}
        except Exception:
            return {}
    return {}


def _args_for_path_candidate_extraction(raw_args: dict[str, Any]) -> dict[str, Any]:
    """Drop large/HTML ``content`` blobs so path salvage does not false-positive."""
    if not isinstance(raw_args, dict):
        return {}
    out: dict[str, Any] = {}
    for key, val in raw_args.items():
        if key in _CONTENT_EXCLUDE_KEYS and isinstance(val, str):
            sample = val[:240].lower()
            if len(val) > 400 or "<html" in sample or "<!doctype" in sample:
                continue
        out[key] = val
    return out


def _collect_recent_path_hints_from_messages(
    messages: list[dict[str, Any]],
) -> list[str]:
    """Recent filesystem paths implied by prior tool calls (newest first)."""
    hints: list[str] = []
    seen: set[str] = set()

    def _add(path: str) -> None:
        p = (path or "").strip()
        if not p or p in seen:
            return
        seen.add(p)
        hints.append(p)

    for msg in reversed(messages or []):
        role = msg.get("role")
        if role == "assistant":
            for tc in msg.get("tool_calls") or []:
                if not isinstance(tc, dict):
                    continue
                fn = tc.get("function") if isinstance(tc.get("function"), dict) else {}
                tool_name = str(tc.get("name") or fn.get("name") or "")
                args = _parse_tool_call_arguments(tc)
                if tool_name == "read_file":
                    for key in ("filepath", "file_path", "path"):
                        if args.get(key):
                            _add(str(args[key]))
                elif tool_name in _WRITE_PATH_TOOLS:
                    for key in ("file_path", "filepath", "path"):
                        if args.get(key):
                            _add(str(args[key]))
                elif tool_name == "hyperframes_create_video":
                    project = str(args.get("project_name") or "").strip()
                    if project:
                        try:
                            from utils.hyperframes_manager import HYPERFRAMES_PROJECTS_DIR

                            _add(os.path.join(HYPERFRAMES_PROJECTS_DIR, project, "index.html"))
                        except Exception:
                            _log.debug("hyperframes path hint failed", exc_info=True)
        elif role == "tool":
            content = str(msg.get("content") or "")
            for match in _HF_INDEX_BYTES_RE.finditer(content):
                _add(match.group(1))
            for match in _HYPERFRAMES_INDEX_PATH_RE.finditer(content):
                _add(match.group(1))
    return hints


def _pick_write_path_from_message_context(
    messages: list[dict[str, Any]],
    *,
    workspace_dir: str,
    raw_args: dict[str, Any],
    field_name: str,
    tool_name: str,
) -> tuple[str | None, str]:
    """Infer ``file_path`` for write tools from recent read/create tool context."""
    from tool_arg_normalizer import recovery_allow_write_for_path_field

    content = str((raw_args or {}).get("content") or "")
    prefer_index = bool(content) and (
        len(content) > 400
        or "<html" in content.lower()
        or "<!doctype" in content.lower()
    )
    hints = _collect_recent_path_hints_from_messages(messages)
    if not hints:
        return None, "no_context_paths"

    ordered = hints
    if prefer_index:
        index_hints = [h for h in hints if h.rstrip("/").lower().endswith("index.html")]
        ordered = index_hints + [h for h in hints if h not in index_hints]

    allow_write = recovery_allow_write_for_path_field(tool_name, field_name)
    for cand in ordered:
        normalized, reject = _validate_recovered_path_candidate(cand)
        if reject or not normalized:
            continue
        resolved, err = safe_resolve_path(
            workspace_dir,
            normalized,
            allow_write=allow_write,
            create_parents=False,
        )
        if err or not resolved:
            continue
        return normalized, "path_recovered_from_tool_context"
    return None, "no_valid_context_path"


def _attempt_recover_read_tool_args(
    tool_name: str,
    missing_fields: list[str],
    raw_args: dict[str, Any],
    assistant_content: str,
    workspace_dir: str,
    *,
    artifact_field: str | None = None,
    artifact_fields: list[str] | None = None,
    candidate_skip: int = 0,
    messages: list[dict[str, Any]] | None = None,
) -> tuple[dict[str, Any] | None, str]:
    """Recover filesystem path arguments when values are missing or XML salvage produced markup-like paths."""
    from tool_arg_normalizer import (
        iter_scalar_top_level_path_fields,
        path_value_looks_like_tool_markup_artifact,
        recovery_allow_write_for_path_field,
    )

    scalar_specs = iter_scalar_top_level_path_fields(tool_name)
    if not scalar_specs:
        return None, "tool_not_eligible"
    scalar_names = {f for f, _ in scalar_specs}

    if artifact_fields is not None:
        fields_to_recover = [f for f in artifact_fields if f in scalar_names]
    elif artifact_field is not None:
        if artifact_field not in scalar_names:
            return None, "artifact_field_not_recoverable"
        fields_to_recover = [artifact_field]
    else:
        mf = set(missing_fields or [])
        fields_to_recover = [f for f, _ in scalar_specs if f in mf]
        if not fields_to_recover:
            return None, "missing_field_not_recoverable"

    if not fields_to_recover:
        return None, "nothing_to_recover"

    if (
        tool_name == "read_file_chunk"
        and "chunk_index" in set(missing_fields or [])
        and artifact_field is None
        and artifact_fields is None
    ):
        return None, "chunk_index_missing_not_recoverable"

    def _finalize_recovered(recovered_paths: dict[str, Any]) -> tuple[dict[str, Any] | None, str]:
        merged = dict(raw_args or {})
        merged.update(recovered_paths)
        for fname, val in recovered_paths.items():
            if not isinstance(val, str):
                return None, "candidate_not_string"
            try:
                if path_value_looks_like_tool_markup_artifact(val):
                    return None, "candidate_markup_artifact"
            except Exception:
                _log.debug("Recovery path-artifact check failed", exc_info=True)
                return None, "candidate_markup_artifact"
            aw = recovery_allow_write_for_path_field(tool_name, fname)
            resolved, err = safe_resolve_path(
                workspace_dir,
                val,
                allow_write=aw,
                create_parents=False,
            )
            if err:
                return None, "path_safety_reject"
            if not resolved:
                return None, "path_resolution_failed"
        return merged, ""

    if messages and tool_name in _WRITE_PATH_TOOLS and len(fields_to_recover) == 1:
        _ctx_field = fields_to_recover[0]
        _ctx_path, _ctx_reason = _pick_write_path_from_message_context(
            messages,
            workspace_dir=workspace_dir,
            raw_args=raw_args if isinstance(raw_args, dict) else {},
            field_name=_ctx_field,
            tool_name=tool_name,
        )
        if _ctx_path:
            fin, _fin_reason = _finalize_recovered({_ctx_field: _ctx_path})
            if fin is not None:
                return fin, _ctx_reason

    candidate_text = assistant_content if isinstance(assistant_content, str) else ""
    slim_args = _args_for_path_candidate_extraction(raw_args if isinstance(raw_args, dict) else {})
    if slim_args:
        try:
            candidate_text = f"{candidate_text}\n{json.dumps(slim_args, ensure_ascii=False)}"
        except Exception:
            pass

    candidates = _extract_recovery_path_candidates(candidate_text)
    if not candidates:
        return None, "candidate_not_found"
    max_cands = 8 if len(fields_to_recover) > 1 else 4
    if len(candidates) > max_cands:
        if messages and tool_name in _WRITE_PATH_TOOLS and len(fields_to_recover) == 1:
            _ctx_field = fields_to_recover[0]
            _ctx_path, _ctx_reason = _pick_write_path_from_message_context(
                messages,
                workspace_dir=workspace_dir,
                raw_args=raw_args if isinstance(raw_args, dict) else {},
                field_name=_ctx_field,
                tool_name=tool_name,
            )
            if _ctx_path:
                fin, _fin_reason = _finalize_recovered({_ctx_field: _ctx_path})
                if fin is not None:
                    return fin, _ctx_reason
        return None, "candidate_count_limit_exceeded"

    valid_candidates: list[str] = []
    for cand in candidates:
        normalized, _reason = _validate_recovered_path_candidate(cand)
        if normalized:
            valid_candidates.append(normalized)
    if not valid_candidates:
        return None, "no_single_candidate"

    seen: set[str] = set()
    deduped: list[str] = []
    for c in valid_candidates:
        if c in seen:
            continue
        seen.add(c)
        deduped.append(c)
    valid_candidates = deduped
    if candidate_skip > 0:
        if candidate_skip >= len(valid_candidates):
            return None, "candidate_skip_exhausted"
        valid_candidates = valid_candidates[candidate_skip:]

    if len(fields_to_recover) == 1:
        field_name = fields_to_recover[0]
        allow_write_scoring = recovery_allow_write_for_path_field(tool_name, field_name)
        if len(valid_candidates) > 1:
            scored: list[tuple[int, str]] = []
            for cand in valid_candidates:
                resolved, err = safe_resolve_path(
                    workspace_dir,
                    cand,
                    allow_write=allow_write_scoring,
                    create_parents=False,
                )
                if err or not resolved:
                    scored.append((-1000, cand))
                    continue
                sc = _score_path_candidate_for_recovery(tool_name, cand, resolved, workspace_dir)
                scored.append((sc, cand))
            scored.sort(key=lambda item: item[0], reverse=True)
            if len(scored) > 1 and scored[0][0] == scored[1][0]:
                return None, "ambiguous_candidates"
            if scored[0][0] <= 0:
                return None, "ambiguous_candidates"
            recovered_path = scored[0][1]
        else:
            recovered_path = valid_candidates[0]

        return _finalize_recovered({field_name: recovered_path})

    rec: dict[str, Any] = {}
    seq_ok = len(valid_candidates) >= len(fields_to_recover)
    if seq_ok:
        for i, fname in enumerate(fields_to_recover):
            cand = valid_candidates[i]
            aw = recovery_allow_write_for_path_field(tool_name, fname)
            resolved, err = safe_resolve_path(workspace_dir, cand, allow_write=aw, create_parents=False)
            if err or not resolved:
                seq_ok = False
                break
            rec[fname] = cand
        if seq_ok:
            return _finalize_recovered(rec)

    used_cands: set[str] = set()
    rec_g: dict[str, Any] = {}
    for fname in fields_to_recover:
        aw = recovery_allow_write_for_path_field(tool_name, fname)
        best_cand: str | None = None
        best_score = -10**9
        for cand in valid_candidates:
            if cand in used_cands:
                continue
            resolved, err = safe_resolve_path(workspace_dir, cand, allow_write=aw, create_parents=False)
            if err or not resolved:
                continue
            sc = _score_path_candidate_for_recovery(tool_name, cand, resolved, workspace_dir)
            if sc > best_score:
                best_score = sc
                best_cand = cand
        if best_cand is None:
            return None, "field_recovery_failed"
        used_cands.add(best_cand)
        rec_g[fname] = best_cand
    return _finalize_recovered(rec_g)


def _try_recover_all_poisoned_path_calls(
    poisoned: list[tuple[str, dict[str, Any], dict[str, Any]]],
    assistant_content: str,
    workspace_dir: str,
    round_: int,
) -> tuple[list[dict[str, Any]] | None, str]:
    """Rebuild tool_calls when every entry failed path-artifact preflight (multi-call safe)."""
    if not poisoned:
        return None, "empty_poisoned"
    from tool_arg_normalizer import path_fields_with_markup_artifacts

    out: list[dict[str, Any]] = []
    skip = 0
    for name, pargs, otc in poisoned:
        fields = path_fields_with_markup_artifacts(name, pargs)
        if not fields:
            return None, "no_artifact_fields"
        rec, reason = _attempt_recover_read_tool_args(
            name,
            [],
            pargs,
            assistant_content,
            workspace_dir,
            artifact_fields=fields,
            candidate_skip=skip,
        )
        if rec is None:
            return None, reason or "recover_failed"
        skip += len(fields)
        out.append(
            {
                "id": otc.get("id", f"xmltc_recov_{round_}_{len(out)}"),
                "name": name,
                "arguments": json.dumps(rec),
            }
        )
    return out, ""

