EventHubAiRouter/router/agent_hier.py

"""Agent-native hierarchical: plan/verify on gateway, tools stay with Zed."""

from __future__ import annotations

import json
import logging
import re
import time
import uuid
from typing import Any

from hierarchical import (
    find_pending_plan,
    worker_model_for,
)

log = logging.getLogger("ai-router")

# Prefer coding tools; Zed sends huge schemas for diagnostics/browser/etc.
_TOOL_PRIORITY = (
    "read_file",
    "write_file",
    "edit_file",
    "create_file",
    "delete_file",
    "move_file",
    "find_path",
    "grep",
    "search",
    "list_directory",
    "terminal",
    "bash",
    "shell",
    "run_command",
    "execute",
    "open",
    "save",
    "replace",
    "apply_patch",
)


def request_has_tools(body: dict[str, Any]) -> bool:
    tools = body.get("tools")
    return isinstance(tools, list) and len(tools) > 0


def messages_have_tool_activity(messages: list[dict[str, Any]] | None) -> bool:
    for msg in messages or []:
        if not isinstance(msg, dict):
            continue
        role = str(msg.get("role") or "")
        if role == "tool":
            return True
        if msg.get("tool_calls"):
            return True
        if msg.get("function_call"):
            return True
    return False


def _msg_chars(msg: dict[str, Any]) -> int:
    content = msg.get("content")
    n = 0
    if isinstance(content, str):
        n += len(content)
    elif isinstance(content, list):
        for block in content:
            if isinstance(block, dict):
                n += len(str(block.get("text") or ""))
                n += len(str(block.get("content") or ""))
    elif content is not None:
        n += len(json.dumps(content, ensure_ascii=False, default=str))
    if msg.get("tool_calls"):
        n += len(json.dumps(msg.get("tool_calls"), ensure_ascii=False, default=str))
    if msg.get("name"):
        n += len(str(msg.get("name")))
    return n


def estimate_tokens_from_chars(n: int) -> int:
    return max(0, (n + 3) // 4)


def measure_forward_size(forward: dict[str, Any]) -> dict[str, int]:
    """Chars actually sent to LiteLLM (messages + slim tools)."""
    msgs = forward.get("messages") or []
    tools = forward.get("tools") or []
    msg_chars = sum(_msg_chars(m) for m in msgs if isinstance(m, dict))
    tools_chars = len(json.dumps(tools, ensure_ascii=False, default=str)) if tools else 0
    total = msg_chars + tools_chars
    return {
        "msg_chars": msg_chars,
        "tools_chars": tools_chars,
        "total_chars": total,
        "msgs": len(msgs) if isinstance(msgs, list) else 0,
        "tools": len(tools) if isinstance(tools, list) else 0,
        "est_tokens": estimate_tokens_from_chars(total),
    }


def format_context_fill(
    *,
    used_chars: int,
    budget_chars: int,
    model_window_tokens: int = 32768,
    label: str = "ctx",
    tools_chars: int = 0,
) -> str:
    """Human-readable context fill for Zed progress stream."""
    budget = max(1, int(budget_chars or 1))
    window = max(1, int(model_window_tokens or 1))
    pct_budget = min(999, int(round(100.0 * used_chars / budget)))
    tok = estimate_tokens_from_chars(used_chars)
    pct_model = min(999, int(round(100.0 * tok / window)))
    filled = min(10, max(0, min(pct_budget, 100) // 10))
    bar = "█" * filled + "░" * (10 - filled)
    over = " OVER" if used_chars > budget else ""
    tools_bit = f" tools={tools_chars}" if tools_chars else ""
    return (
        f"{label} {used_chars}/{budget} ({pct_budget}%{over}) [{bar}]"
        f"{tools_bit} ~{tok}tok · model≈{pct_model}%/{window}"
    )


def context_fill_for_forward(
    forward: dict[str, Any],
    cfg: dict[str, Any],
    *,
    budget_chars: int | None = None,
) -> tuple[str, dict[str, Any]]:
    """Return (progress line, meta fields) for an executor/plan forward body."""
    size = measure_forward_size(forward)
    budget = int(
        budget_chars
        if budget_chars is not None
        else cfg.get("executor_tool_loop_chars")
        or cfg.get("executor_input_chars")
        or 6000
    )
    window = int(cfg.get("context_window_tokens") or 32768)
    line = format_context_fill(
        used_chars=size["total_chars"],
        budget_chars=budget,
        model_window_tokens=window,
        label="ctx",
        tools_chars=size["tools_chars"],
    )
    meta = {
        "ctx_used_chars": size["total_chars"],
        "ctx_budget_chars": budget,
        "ctx_pct": min(999, int(round(100.0 * size["total_chars"] / max(1, budget)))),
        "ctx_est_tokens": size["est_tokens"],
        "ctx_tools_chars": size["tools_chars"],
        "ctx_msgs": size["msgs"],
        "ctx_window_tokens": window,
        "ctx_fill": line,
    }
    return line, meta


def trim_keeping_system(
    messages: list[dict[str, Any]],
    *,
    max_chars: int,
) -> list[dict[str, Any]]:
    """Trim from the oldest, but always keep a leading system message."""
    if not messages:
        return []
    if messages[0].get("role") == "system":
        sys = messages[0]
        sys_n = _msg_chars(sys)
        rest = trim_messages_for_executor(
            messages[1:], max_chars=max(0, max_chars - sys_n)
        )
        return [sys, *rest]
    return trim_messages_for_executor(messages, max_chars=max_chars)


def sanitize_outbound_messages(
    messages: list[dict[str, Any]] | None,
) -> list[dict[str, Any]]:
    """Novita DeepSeek returns 400 on content:null / orphan tools / bad args."""
    cleaned: list[dict[str, Any]] = []
    for msg in messages or []:
        if not isinstance(msg, dict):
            continue
        m = dict(msg)
        role = str(m.get("role") or "")
        for junk in (
            "reasoning",
            "reasoning_content",
            "reasoning_details",
            "provider_specific_fields",
        ):
            m.pop(junk, None)
        if role == "assistant":
            tcs = m.get("tool_calls")
            if isinstance(tcs, list) and tcs:
                if m.get("content") is None:
                    m["content"] = ""
                clean_tcs: list[dict[str, Any]] = []
                for tc in tcs:
                    if not isinstance(tc, dict):
                        continue
                    tc2 = dict(tc)
                    fn = tc2.get("function")
                    if isinstance(fn, dict):
                        fn2 = dict(fn)
                        args = fn2.get("arguments")
                        if args is None:
                            fn2["arguments"] = "{}"
                        elif isinstance(args, dict):
                            fn2["arguments"] = json.dumps(args, ensure_ascii=False)
                        elif not isinstance(args, str):
                            fn2["arguments"] = str(args)
                        tc2["function"] = fn2
                    if not tc2.get("type"):
                        tc2["type"] = "function"
                    clean_tcs.append(tc2)
                m["tool_calls"] = clean_tcs
            elif m.get("content") is None:
                m["content"] = ""
        elif role == "tool":
            if m.get("content") is None:
                m["content"] = ""
        else:
            if m.get("content") is None:
                m["content"] = ""
        cleaned.append(m)

    offered: set[str] = set()
    for m in cleaned:
        if m.get("role") != "assistant":
            continue
        for tc in m.get("tool_calls") or []:
            if isinstance(tc, dict):
                cid = str(tc.get("id") or "")
                if cid:
                    offered.add(cid)
    out: list[dict[str, Any]] = []
    for m in cleaned:
        if m.get("role") == "tool":
            cid = str(m.get("tool_call_id") or "")
            if not cid or cid not in offered:
                continue
        out.append(m)
    return out


def _count_edit_tool_calls(messages: list[dict[str, Any]] | None) -> int:
    n = 0
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            if "edit" in name or "write" in name:
                n += 1
    return n


def _last_tool_result_ok(messages: list[dict[str, Any]] | None) -> bool:
    for msg in reversed(messages or []):
        if not isinstance(msg, dict) or msg.get("role") != "tool":
            continue
        low = str(msg.get("content") or "").lower()
        if not low.strip():
            return False
        if any(
            x in low
            for x in (
                "not found",
                "could not find matching",
                "error",
                "failed",
                "expected struct",
            )
        ):
            return False
        return True
    return False


def should_stop_after_edit(
    messages: list[dict[str, Any]] | None,
    cfg: dict[str, Any] | None = None,
) -> bool:
    """After any edit_file was issued — do not force another edit loop."""
    cfg = cfg or {}
    if _count_edit_tool_calls(messages) < 1:
        return False
    # Prefer stop when last edit looks done (success or already failed enough)
    if _edit_failure_count(messages) >= int(
        cfg.get("executor_max_edit_failures", 4) or 4
    ):
        return True
    # Last assistant turn was edit → stop after tool result (success or fail once)
    for msg in reversed(messages or []):
        if not isinstance(msg, dict):
            continue
        if msg.get("role") == "tool":
            continue
        if msg.get("role") == "assistant" and msg.get("tool_calls"):
            names = [
                str(((tc.get("function") or {}).get("name")) or "").lower()
                for tc in (msg.get("tool_calls") or [])
                if isinstance(tc, dict)
            ]
            return any("edit" in n or "write" in n for n in names)
        if msg.get("role") in ("assistant", "user"):
            break
    return _count_edit_tool_calls(messages) >= 1


def should_force_edit_after_read(
    messages: list[dict[str, Any]] | None,
    cfg: dict[str, Any] | None = None,
    plan: dict[str, Any] | None = None,
) -> bool:
    """Only once: right after a successful read, before any edit was attempted.

    Bug before: returned True on ANY trailing tool result — including successful
    edit_file — which made Zed apply the same marker edit in an infinite loop.
    """
    cfg = cfg or {}
    if not cfg.get("executor_force_edit_after_read", True):
        return False
    # DevOps without runtime facts — do not blind-edit
    try:
        from runtime_probe import devops_blocks_blind_force_edit

        if devops_blocks_blind_force_edit(plan):
            return False
    except Exception:  # noqa: BLE001
        pass
    if not _successful_read_keys(messages):
        return False
    # Already edited (or tried) → never force again
    if _count_edit_tool_calls(messages) >= 1:
        return False
    if _edit_failure_count(messages) > 0:
        return False
    if _edit_mismatch_recently(messages):
        return False
    # Last assistant tool_calls must be read_* (skip tool results — do not
    # treat edit success as "force another edit")
    for msg in reversed(messages or []):
        if not isinstance(msg, dict):
            continue
        if msg.get("role") == "tool":
            continue
        if msg.get("role") == "assistant" and msg.get("tool_calls"):
            names = [
                str(((tc.get("function") or {}).get("name")) or "").lower()
                for tc in (msg.get("tool_calls") or [])
                if isinstance(tc, dict)
            ]
            if any("edit" in n or "write" in n for n in names):
                return False
            if any(
                any(k in n for k in ("read", "open", "cat", "get_file")) for n in names
            ):
                return True
            return False
        if msg.get("role") in ("user", "system"):
            break
    return False


def force_edit_after_read_completion(
    *,
    plan: dict[str, Any] | None,
    tools: list[Any],
    model: str,
    messages: list[dict[str, Any]] | None,
) -> dict[str, Any]:
    # Prefer deterministic Traefik port fix from live probe
    try:
        from runtime_probe import force_runtime_fix_edit

        targeted = force_runtime_fix_edit(plan=plan, tools=tools, model=model)
        if targeted:
            return targeted
    except Exception:  # noqa: BLE001
        pass
    path = _last_successful_read_path(messages) or _path_for_user_status(
        plan, tools, messages
    )
    if not path:
        return _stop_synth(model, "Нет пути для edit после read_file.")
    return _continue_edit_or_grep(
        model=model,
        tools=tools,
        path=path,
        plan=plan,
        messages=messages,
    )


def stop_after_edit_completion(
    *,
    model: str,
    messages: list[dict[str, Any]] | None,
) -> dict[str, Any]:
    path = _last_successful_read_path(messages) or _last_edit_path(messages)
    ok = _last_tool_result_ok(messages) and _edit_failure_count(messages) == 0
    if ok:
        text = "Готово: правка применена."
        if path:
            text += f" Файл: `{path}`."
        text += " Если нужно ещё — напиши одной фразой."
    else:
        fails = _edit_failure_count(messages)
        text = (
            f"Правка не применилась ({fails} ошибка). "
            "Нужен точный фрагмент из файла, либо правь вручную."
        )
        if path:
            text += f" Файл: `{path}`."
    return _stop_synth(model, text)


def enforce_forward_budget(
    forward: dict[str, Any],
    *,
    budget_chars: int,
) -> dict[str, Any]:
    """Hard-cap messages+tools so measure_forward_size ≤ budget (stops 160%+ spills)."""
    out = dict(forward)
    budget = max(500, int(budget_chars))
    for round_i in range(10):
        size = measure_forward_size(out)
        if size["total_chars"] <= budget:
            return out
        tools = list(out.get("tools") or [])
        msgs = list(out.get("messages") or [])
        # 1) Drop lowest-priority tools while tools take >35% of budget
        if tools and size["tools_chars"] > budget * 0.35:
            out["tools"] = tools[:-1]
            continue
        # 2) Aggressively truncate tool role contents
        max_tool = max(200, 900 - round_i * 80)
        new_msgs: list[dict[str, Any]] = []
        for m in msgs:
            if not isinstance(m, dict):
                continue
            mm = dict(m)
            if mm.get("role") == "tool" and isinstance(mm.get("content"), str):
                if len(mm["content"]) > max_tool:
                    mm["content"] = mm["content"][: max_tool - 1] + "…"
            new_msgs.append(mm)
        msgs = new_msgs
        # 3) Shrink message window (keep system)
        msg_budget = max(400, budget - len(json.dumps(out.get("tools") or [], ensure_ascii=False)))
        msgs = trim_keeping_system(msgs, max_chars=msg_budget)
        # 4) Drop oldest non-system pairs if still over
        if len(msgs) > 2 and measure_forward_size({"messages": msgs, "tools": out.get("tools")})[
            "total_chars"
        ] > budget:
            if msgs[0].get("role") == "system":
                msgs = [msgs[0], *msgs[2:]]
            else:
                msgs = msgs[1:]
        out["messages"] = msgs
    # Final hard cut on last tool payload if still over
    size = measure_forward_size(out)
    if size["total_chars"] > budget:
        msgs = list(out.get("messages") or [])
        for i in range(len(msgs) - 1, -1, -1):
            m = msgs[i]
            if isinstance(m, dict) and m.get("role") == "tool" and isinstance(m.get("content"), str):
                room = max(80, budget - (size["total_chars"] - len(m["content"])))
                m = dict(m)
                m["content"] = m["content"][:room] + "…"
                msgs[i] = m
                out["messages"] = msgs
                break
    return out


def trim_messages_for_executor(
    messages: list[dict[str, Any]],
    *,
    max_chars: int = 24000,
) -> list[dict[str, Any]]:
    """Keep newest messages under budget — huge Zed dumps make Novita hang past timeout."""
    if max_chars <= 0:
        return list(messages)
    kept: list[dict[str, Any]] = []
    total = 0
    for msg in reversed(messages or []):
        if not isinstance(msg, dict):
            continue
        size = _msg_chars(msg)
        if kept and total + size > max_chars:
            break
        kept.append(msg)
        total += size
    kept.reverse()
    return kept


def compress_messages_for_tool_loop(
    messages: list[dict[str, Any]],
    *,
    max_chars: int = 6000,
    max_tool_result_chars: int = 1200,
    newest_tool_result_chars: int = 3500,
    max_messages: int = 8,
) -> list[dict[str, Any]]:
    """Keep plan + newest tool turns; truncate huge tool payloads (Zed dumps)."""
    compressed: list[dict[str, Any]] = []
    tool_indices: list[int] = []
    for msg in messages or []:
        if not isinstance(msg, dict):
            continue
        m = dict(msg)
        if m.get("role") == "tool":
            tool_indices.append(len(compressed))
            content = m.get("content")
            if isinstance(content, str) and len(content) > max_tool_result_chars:
                m["content"] = (
                    content[: max_tool_result_chars - 14].rstrip() + "\n<<TRUNCATED>>"
                )
        elif m.get("role") == "assistant" and m.get("tool_calls"):
            # Novita DeepSeek 400 on content:null — use empty string
            m["content"] = ""
        compressed.append(m)
    # Keep the newest tool result fuller (edit_file needs exact text, not "…")
    if tool_indices:
        newest_i = tool_indices[-1]
        orig = None
        # recover original from messages by counting tool roles
        tool_n = 0
        for msg in messages or []:
            if isinstance(msg, dict) and msg.get("role") == "tool":
                tool_n += 1
                if tool_n == len(tool_indices):
                    orig = msg.get("content")
                    break
        if isinstance(orig, str) and orig.strip():
            # Cap newest read — large dumps make Novita coder hang for minutes
            keep_cap = max(800, int(newest_tool_result_chars or 3500))
            keep = min(len(orig), keep_cap)
            restored = orig[:keep]
            if len(orig) > keep:
                restored = restored.rstrip() + "\n<<TRUNCATED>>"
            compressed[newest_i] = dict(compressed[newest_i])
            compressed[newest_i]["content"] = restored
    # Prefer trailing window (tool loop is newest-first relevance)
    if len(compressed) > max_messages:
        # Keep leading system if present
        head: list[dict[str, Any]] = []
        rest = compressed
        if compressed and compressed[0].get("role") == "system":
            head = [compressed[0]]
            rest = compressed[1:]
        rest = rest[-(max_messages - len(head)) :]
        compressed = head + rest
    return trim_messages_for_executor(compressed, max_chars=max_chars)


def inject_plan_context(
    messages: list[dict[str, Any]],
    plan_payload: dict[str, Any] | None,
    *,
    max_chars: int = 24000,
    minimal: bool = False,
    tool_loop: bool = False,
    newest_tool_result_chars: int = 3500,
    path_resolve: bool = False,
) -> list[dict[str, Any]]:
    """Build executor messages.

    minimal=True (first turn after plan approve): drop Zed history/AGENTS
    dumps — tools+history makes Novita hang past timeout.
    tool_loop=True: compress tool results + keep only newest turns.
    path_resolve=True: cheap find/list only — do not invent read paths.
    """
    tasks = (plan_payload or {}).get("subtasks") or []
    plan_blob = json.dumps(plan_payload or {}, ensure_ascii=False)
    if path_resolve:
        qs = (plan_payload or {}).get("path_resolve_queries") or []
        q_line = ", ".join(str(q) for q in qs[:4]) or "dynamic_conf.yml"
        root = _DEFAULT_DEVOPS_ROOT
        sys_msg = {
            "role": "system",
            "content": "\n".join(
                [
                    "You locate files for an approved plan. Do NOT invent paths.",
                    "Call find_path or list_directory only — no read_file/edit yet.",
                    f"Search under: {root}",
                    f"Queries: {q_line}",
                    "EventHubDevOps Traefik file is ift\\traefik\\dynamic_conf.yml "
                    "(NOT traefik.yml). Compose: ift\\docker-compose.core.yml "
                    "(NO root docker-compose.yml).",
                ]
            ),
        }
        return [
            sys_msg,
            {
                "role": "user",
                "content": (
                    f"Find real files for: {q_line}. "
                    f"Start with find_path under {root}."
                ),
            },
        ]

    lines = [
        "You are a coding agent. Execute the approved plan via tools.",
        "Do not paste full files as markdown — use edit/write tools.",
        "Reply with a tool call only — no long prose.",
        "edit_file MUST be: {\"path\":\"...\",\"edits\":[{\"old_text\":\"...\",\"new_text\":\"...\"}]}. "
        "Never pass a string as edits (Zed: expected struct Edit).",
    ]
    if _looks_like_devops_plan(plan_blob):
        root = _DEFAULT_DEVOPS_ROOT
        yml = _join_under(root, _DEVOPS_KEY_RELS[0])
        lines.extend(
            [
                f"EventHubDevOps root (absolute): {root}",
                f"Target file: {yml}",
                "There is NO src/ under EventHubDevOps.",
                "There is NO root docker-compose.yml — use ift\\docker-compose.core.yml "
                "(or stage\\…).",
                "Traefik: ONLY ift\\traefik\\dynamic_conf.yml "
                "(NOT traefik.yml / traefik.toml / static.yml).",
                "Do NOT use relative paths (ift/... fails in Zed multi-root).",
                "Read each file at most once; then edit_file — never re-read the same path.",
                "Bad Gateway: fix Traefik backend URL/port to match the open container port "
                "(do NOT invent traefik.yml or random markers).",
            ]
        )
        facts = (plan_payload or {}).get("runtime_facts")
        if isinstance(facts, dict) and facts:
            lines.append("RUNTIME FACTS (from live probe — trust these over guesses):")
            lines.append(
                f"  host={facts.get('host')} http={facts.get('http_status')} "
                f"service={facts.get('service')} open_ports={facts.get('open_ports')}"
            )
            if facts.get("suggested_backend_url"):
                lines.append(
                    f"  REQUIRED FIX: set loadbalancer url to {facts.get('suggested_backend_url')} "
                    f"(replace http://{facts.get('service')}:80 if present)."
                )
            if facts.get("hint"):
                lines.append(f"  hint: {facts.get('hint')}")
    for i, item in enumerate(tasks):
        if not isinstance(item, dict):
            continue
        tid = item.get("id") or str(i + 1)
        prompt = item.get("prompt") or ""
        tier = item.get("worker_tier") or "simple"
        deps = item.get("depends_on") or item.get("deps") or []
        dep_s = f", deps={deps}" if deps else ""
        lines.append(f"{tid}. [{tier}] {prompt}{dep_s}")
        paths = item.get("paths") or []
        if isinstance(paths, list) and paths:
            lines.append("   paths: " + ", ".join(str(p) for p in paths[:6]))
        edit_goal = item.get("edit_goal") or item.get("goal") or ""
        if edit_goal:
            lines.append(f"   edit_goal: {edit_goal}")
        constraints = item.get("constraints") or []
        if isinstance(constraints, list) and constraints:
            lines.append(
                "   constraints: " + "; ".join(str(c) for c in constraints[:6])
            )
    goal = (plan_payload or {}).get("user_goal") or ""
    if goal:
        lines.append("User goal: " + str(goal)[:800])
    acc = (plan_payload or {}).get("acceptance") or []
    if acc:
        lines.append("Acceptance: " + "; ".join(str(x) for x in acc))
    sys_msg = {"role": "system", "content": "\n".join(lines)}

    if minimal:
        start = "Plan approved. Start with step 1. Call tools now."
        path_hints: list[str] = []
        for item in tasks:
            if isinstance(item, dict):
                for p in item.get("paths") or []:
                    if p and str(p) not in path_hints:
                        path_hints.append(str(p))
        if path_hints:
            start = (
                "Plan approved. Prefer absolute paths. Start with read_file on: "
                + ", ".join(path_hints[:3])
            )
        if _looks_like_devops_plan(plan_blob):
            start = (
                "Plan approved. Call read_file now with absolute path "
                f"{_join_under(_DEFAULT_DEVOPS_ROOT, _DEVOPS_KEY_RELS[0])} "
                "(no src/, no relative ift/...)."
            )
        return [
            sys_msg,
            {"role": "user", "content": start},
        ]

    if tool_loop:
        lines.append(
            "Tool loop: do NOT call read_file on paths already present in tool results. "
            "Next action must be edit/write/terminal."
        )
        sys_msg = {"role": "system", "content": "\n".join(lines)}
        # Leave room for sys_msg inside max_chars
        sys_n = _msg_chars(sys_msg)
        trimmed = compress_messages_for_tool_loop(
            messages,
            max_chars=max(400, max_chars - sys_n),
            max_tool_result_chars=1200,
            newest_tool_result_chars=newest_tool_result_chars,
            max_messages=6,
        )
        if trimmed and trimmed[0].get("role") == "system":
            out = [sys_msg, *trimmed[1:]]
        else:
            out = [sys_msg, *trimmed]
        return trim_keeping_system(out, max_chars=max_chars)

    trimmed = trim_messages_for_executor(messages, max_chars=max_chars)
    if not plan_payload:
        return trimmed
    return [sys_msg, *trimmed]


def _tool_name(tool: dict[str, Any]) -> str:
    fn = tool.get("function") if isinstance(tool.get("function"), dict) else {}
    return str(fn.get("name") or tool.get("name") or "")


def _tool_priority(name: str) -> int:
    low = name.lower()
    for i, key in enumerate(_TOOL_PRIORITY):
        if key in low:
            return i
    return 1000 + len(low)


def _skeleton_params(params: Any) -> dict[str, Any]:
    """Drop nested descriptions / oneOf / $ref — they bloat Novita to death.

    Preserve array-of-object shapes for Zed `edits: Edit[]` — flattening to
    `items: string` makes the model pass a prose string →
    `expected struct Edit`.
    """
    if not isinstance(params, dict):
        return {"type": "object", "properties": {}}
    props_in = params.get("properties")
    props_out: dict[str, Any] = {}
    if isinstance(props_in, dict):
        for key, spec in list(props_in.items())[:16]:
            t = "string"
            if isinstance(spec, dict):
                raw_t = spec.get("type")
                if isinstance(raw_t, str):
                    t = raw_t
                elif isinstance(raw_t, list) and raw_t:
                    t = str(raw_t[0])
            key_s = str(key)
            if t == "array":
                items = spec.get("items") if isinstance(spec, dict) else None
                # Zed edit_file.edits → [{old_text, new_text}]
                if key_s == "edits" or (
                    isinstance(items, dict)
                    and (
                        items.get("type") == "object"
                        or isinstance(items.get("properties"), dict)
                    )
                ):
                    item_props: dict[str, Any] = {
                        "old_text": {"type": "string"},
                        "new_text": {"type": "string"},
                    }
                    if isinstance(items, dict) and isinstance(
                        items.get("properties"), dict
                    ):
                        # Keep known Edit-like fields if present under other names
                        src = items["properties"]
                        if "old_string" in src and "old_text" not in src:
                            item_props = {
                                "old_string": {"type": "string"},
                                "new_string": {"type": "string"},
                            }
                        elif "old_text" in src or "new_text" in src:
                            item_props = {
                                "old_text": {"type": "string"},
                                "new_text": {"type": "string"},
                            }
                    props_out[key_s] = {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": item_props,
                            "required": list(item_props.keys()),
                        },
                    }
                else:
                    props_out[key_s] = {
                        "type": "array",
                        "items": {"type": "string"},
                    }
            elif t == "object":
                props_out[key_s] = {"type": "object"}
            elif t == "number" or t == "integer" or t == "boolean":
                props_out[key_s] = {"type": t}
            else:
                props_out[key_s] = {"type": "string"}
    required = params.get("required")
    req_out = (
        [str(r) for r in required if str(r) in props_out][:12]
        if isinstance(required, list)
        else []
    )
    out: dict[str, Any] = {"type": "object", "properties": props_out}
    if req_out:
        out["required"] = req_out
    return out


_ZED_EDIT_FILE_PARAMS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "path": {"type": "string"},
        "edits": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "old_text": {"type": "string"},
                    "new_text": {"type": "string"},
                },
                "required": ["old_text", "new_text"],
            },
        },
    },
    "required": ["path", "edits"],
}


def slim_tools_for_executor(
    tools: list[Any] | None,
    *,
    max_tools: int = 12,
    max_desc_chars: int = 80,
    full_schema_chars: int = 12000,
) -> list[Any]:
    """Keep coding tools; use full schemas under threshold, else skeletonize.

    Full Zed JSON Schema (~20KB+) hangs weak models — threshold avoids blind
    skeleton when payload is already small.
    """
    if not isinstance(tools, list):
        return []
    ranked = [t for t in tools if isinstance(t, dict) and _tool_name(t)]
    ranked.sort(key=lambda t: _tool_priority(_tool_name(t)))
    ranked = ranked[:max_tools]
    raw_chars = len(json.dumps(ranked, ensure_ascii=False, default=str))
    use_full = raw_chars <= max(500, int(full_schema_chars))
    out: list[Any] = []
    for tool in ranked:
        fn_in = tool.get("function") if isinstance(tool.get("function"), dict) else {}
        name = str(fn_in.get("name") or tool.get("name") or "tool")
        desc = fn_in.get("description")
        if not isinstance(desc, str):
            desc = name
        if len(desc) > max_desc_chars:
            desc = desc[: max_desc_chars - 1] + "…"
        low = name.lower()
        if use_full and "edit_file" not in low and low != "edit":
            # Pass through trimmed description only
            params = fn_in.get("parameters")
            if not isinstance(params, dict):
                params = {"type": "object", "properties": {}}
            out.append(
                {
                    "type": "function",
                    "function": {
                        "name": name,
                        "description": desc,
                        "parameters": params,
                    },
                }
            )
            continue
        if "edit_file" in low or low == "edit":
            params = dict(_ZED_EDIT_FILE_PARAMS)
            desc = (
                "Apply edits: {path, edits:[{old_text, new_text}]}. "
                "edits MUST be objects, never a string."
            )[:max_desc_chars]
        else:
            params = _skeleton_params(fn_in.get("parameters"))
        out.append(
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": desc,
                    "parameters": params,
                },
            }
        )
    return out


def pick_agent_executor_model(cfg: dict[str, Any]) -> str:
    explicit = cfg.get("agent_executor_model")
    if explicit:
        model = str(explicit)
        # Never hang tool-loop on Llama simple lanes
        if model in ("a-simple", "b-simple", "c-simple", "a-medium-ops", "b-medium-ops", "c-medium-ops"):
            return worker_model_for("medium_code")
        return model
    return worker_model_for("medium_code")


def pick_agent_escalate_model(cfg: dict[str, Any]) -> str:
    """Stronger model after repeated edit failures (DeepSeek / Max)."""
    explicit = cfg.get("agent_escalate_model")
    if explicit:
        return str(explicit)
    return worker_model_for("hard")


def should_escalate_executor(
    messages: list[dict[str, Any]] | None,
    cfg: dict[str, Any],
) -> bool:
    """Escalate mid tool_loop after edit failures or a successful read.

    Cheap coder routinely hangs 2–3 minutes on fat read_file payloads; once the
    file is in context, jump to DeepSeek for the edit decision.
    """
    max_ef = int(cfg.get("agent_escalate_after_edit_failures", 2) or 2)
    if _edit_failure_count(messages) >= max(1, max_ef):
        return True
    if cfg.get("executor_escalate_after_read", True) and _successful_read_keys(
        messages
    ):
        return True
    return False


def prepare_agent_executor_forward(
    body: dict[str, Any],
    messages: list[dict[str, Any]],
    plan: dict[str, Any] | None,
    cfg: dict[str, Any],
    *,
    executor: str | None = None,
    stream: bool | None = None,
    minimal: bool | None = None,
    path_resolve: bool = False,
) -> dict[str, Any]:
    """Build LiteLLM body for Zed tool executor.

    Critical: num_retries=0 and fallbacks=[] — otherwise LiteLLM stacks
    retries×fallback chain past gateway httpx ReadTimeout.
    Never fall back to Llama on tools (except dedicated path_resolve).
    """
    from path_resolve import discovery_tools_only, pick_path_resolve_model

    model = executor or pick_agent_executor_model(cfg)
    if path_resolve:
        model = pick_path_resolve_model(cfg)
    elif should_escalate_executor(messages, cfg):
        model = pick_agent_escalate_model(cfg)
    plan = sanitize_plan_paths(plan)
    max_ctx = int(cfg.get("executor_input_chars", 12000))
    timeout = float(
        cfg.get("executor_timeout_sec") or cfg.get("call_timeout_sec") or 90
    )
    retries = int(cfg.get("executor_num_retries", 0))
    in_tool_loop = messages_have_tool_activity(messages)
    use_minimal = (
        bool(cfg.get("executor_minimal_context", True))
        if minimal is None
        else bool(minimal)
    )
    if path_resolve:
        use_minimal = True
        timeout = float(cfg.get("path_resolve_timeout_sec") or 45)
    elif in_tool_loop:
        use_minimal = False
        max_ctx = int(cfg.get("executor_tool_loop_chars", 6000))
        timeout = float(cfg.get("executor_tool_loop_timeout_sec") or timeout or 100)

    tools = slim_tools_for_executor(
        body.get("tools") if isinstance(body.get("tools"), list) else [],
        max_tools=int(cfg.get("executor_max_tools", 12)),
        max_desc_chars=int(cfg.get("executor_tool_desc_chars", 80)),
        full_schema_chars=int(cfg.get("executor_tools_full_chars", 12000)),
    )
    if path_resolve:
        tools = discovery_tools_only(
            tools, max_tools=int(cfg.get("path_resolve_max_tools", 3) or 3)
        )
    tools_chars = len(json.dumps(tools, ensure_ascii=False, default=str)) if tools else 0
    # Tools count toward the same budget — reserve space so msgs+tools ≤ max_ctx
    while tools and tools_chars > max(800, max_ctx // 3):
        tools = tools[:-1]
        tools_chars = len(json.dumps(tools, ensure_ascii=False, default=str))
    msg_budget = max(600, max_ctx - tools_chars)

    newest_tool = int(cfg.get("executor_newest_tool_chars", 3500) or 3500)
    msgs = inject_plan_context(
        messages,
        plan,
        max_chars=msg_budget,
        minimal=use_minimal,
        tool_loop=in_tool_loop and not path_resolve,
        newest_tool_result_chars=newest_tool,
        path_resolve=path_resolve,
    )
    msgs = sanitize_outbound_messages(msgs)
    forward: dict[str, Any] = {
        "model": model,
        "messages": msgs,
        "tools": tools,
        # Force a tool call — prose-only replies waste the whole timeout budget
        "tool_choice": "required",
        "max_tokens": int(cfg.get("executor_max_tokens") or 768),
        "temperature": 0.1,
        "stream": bool(stream) if stream is not None else False,
        "num_retries": retries,
        "fallbacks": [],
        "timeout": timeout,
    }
    return enforce_forward_budget(forward, budget_chars=max_ctx)


# Zed multi-root: relative paths fail ("not in the project"). Prefer absolute.
_DEFAULT_DEVOPS_ROOT = r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps"
_DEVOPS_KEY_RELS = (
    r"ift\traefik\dynamic_conf.yml",
    r"ift\traefik\dynamic_conf.loadtest.yml",
    r"ift\docker-compose.core.yml",
    r"README.md",
)
_ABS_WIN_RE = re.compile(r"[A-Za-z]:\\[^\"'\n\r]+")


def _norm_path_key(path: str) -> str:
    return path.replace("/", "\\").rstrip("\\").lower()


def _join_under(root: str, rel: str) -> str:
    return root.rstrip("\\/") + "\\" + rel.replace("/", "\\").lstrip("\\/")


def _looks_like_devops_plan(text: str) -> bool:
    low = (text or "").lower()
    return any(
        k in low
        for k in (
            "traefik",
            "балансир",
            "eventhubdevops",
            "ift.calentiq",
            "ai-router.ift",
        )
    )


def _paths_mentioned(text: str) -> list[str]:
    found = re.findall(
        r"([A-Za-z0-9_./\\-]+\.(?:ya?ml|toml|md|erl|ts|tsx|js|go|py|json|sh))",
        text or "",
    )
    return found


def _extract_abs_win_paths(text: str) -> list[str]:
    out: list[str] = []
    for m in _ABS_WIN_RE.finditer(text or ""):
        p = m.group(0).rstrip("\\/.,;:)")
        # json.dumps of Windows paths → doubled backslashes
        while "\\\\" in p:
            p = p.replace("\\\\", "\\")
        if p:
            out.append(p)
    return out


def _tool_arg_paths(messages: list[dict[str, Any]] | None) -> list[str]:
    """Paths from tool_call arguments (prefer parsed JSON over regex on dumps)."""
    out: list[str] = []
    for msg in messages or []:
        if not isinstance(msg, dict) or not msg.get("tool_calls"):
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                parsed = None
            if isinstance(parsed, dict):
                for key in ("path", "file", "target", "directory", "dir"):
                    val = parsed.get(key)
                    if isinstance(val, str) and val.strip():
                        out.append(val.strip())
            else:
                out.extend(_extract_abs_win_paths(str(args)))
    return out


def _devops_root_from_messages(messages: list[dict[str, Any]] | None) -> str | None:
    for p in _tool_arg_paths(messages):
        norm = p.replace("/", "\\")
        m = re.search(r"(?i)^(.+?\\EventHubDevOps)(?:\\|$)", norm)
        if m:
            raw = m.group(1)
            idx = raw.lower().rfind("eventhubdevops")
            return raw[:idx] + "EventHubDevOps"
    for msg in messages or []:
        if not isinstance(msg, dict):
            continue
        blob = json.dumps(msg, ensure_ascii=False, default=str)
        for p in _extract_abs_win_paths(blob):
            m = re.search(r"(?i)^(.+?\\EventHubDevOps)(?:\\|$)", p)
            if m:
                raw = m.group(1)
                idx = raw.lower().rfind("eventhubdevops")
                return raw[:idx] + "EventHubDevOps"
    return None


def _failed_path_keys(messages: list[dict[str, Any]] | None) -> set[str]:
    failed: set[str] = set()
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "tool":
            continue
        content = str(msg.get("content") or "")
        low = content.lower()
        if not (
            "not found" in low
            or "not in the project" in low
            or "path not found" in low
        ):
            continue
        for p in _extract_abs_win_paths(content):
            failed.add(_norm_path_key(p))
        # Zed often returns JSON: {"Text":"C:\\...\\file.yml not found"}
        for m in re.finditer(
            r'([A-Za-z]:\\[^"\n\r]+?)\s+not found', content, re.I
        ):
            failed.add(_norm_path_key(m.group(1).strip()))
        for p in _paths_mentioned(content):
            failed.add(_norm_path_key(p))
        m = re.search(r"Path\s+(.+?)\s+is not", content, re.I)
        if m:
            failed.add(_norm_path_key(m.group(1).strip()))
        m2 = re.search(r"Path not found:\s*(.+)", content, re.I)
        if m2:
            failed.add(_norm_path_key(m2.group(1).strip()))
    return failed


def _seen_path_keys(messages: list[dict[str, Any]] | None) -> set[str]:
    seen: set[str] = set()
    for msg in messages or []:
        if not isinstance(msg, dict):
            continue
        if msg.get("role") == "assistant" and msg.get("tool_calls"):
            for tc in msg.get("tool_calls") or []:
                if not isinstance(tc, dict):
                    continue
                args = (tc.get("function") or {}).get("arguments") or ""
                try:
                    parsed = json.loads(args) if isinstance(args, str) else args
                except Exception:  # noqa: BLE001
                    parsed = None
                if isinstance(parsed, dict):
                    for key in ("path", "file", "target", "directory", "dir"):
                        val = parsed.get(key)
                        if isinstance(val, str) and val.strip():
                            seen.add(_norm_path_key(val))
                for p in _extract_abs_win_paths(str(args)):
                    seen.add(_norm_path_key(p))
                for p in _paths_mentioned(str(args)):
                    seen.add(_norm_path_key(p))
        blob = json.dumps(msg, ensure_ascii=False, default=str)
        for p in _paths_mentioned(blob):
            if "\\" in p or "/" in p:
                seen.add(_norm_path_key(p))
    return seen


def _is_bogus_devops_path(path: str) -> bool:
    key = _norm_path_key(path)
    if key.endswith("\\src") or key.endswith("/src"):
        return True
    # bare relative under multi-root Zed — reject for DevOps synthetics
    if not re.match(r"^[a-z]:\\", key) and (
        key.startswith("ift\\")
        or key.startswith("ift/")
        or key.startswith("eventhubdevops\\")
        or key.startswith("eventhubdevops/")
    ):
        return True
    # EventHubDevOps has no root compose — only ift|stage/docker-compose.*.yml
    base = key.rsplit("\\", 1)[-1].rsplit("/", 1)[-1]
    if base in ("docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml"):
        if "\\ift\\" not in key and "/ift/" not in key and "\\stage\\" not in key and "/stage/" not in key:
            return True
    # Classic Traefik static names — this repo only has dynamic_conf*.yml under */traefik/
    if "\\traefik\\" in key or "/traefik/" in key:
        allowed_traefik = {
            "dynamic_conf.yml",
            "dynamic_conf.yaml",
            "dynamic_conf.loadtest.yml",
            "dynamic_conf.loadtest.yaml",
            ".gitkeep",
        }
        if base not in allowed_traefik and not base.endswith(".crt") and not base.endswith(".key"):
            return True
    # Other frequently invented static Traefik filenames anywhere under DevOps
    if base in (
        "traefik.yml",
        "traefik.yaml",
        "traefik.toml",
        "traefik-static.yml",
        "static.yml",
        "static_conf.yml",
    ):
        return True
    return False


def sanitize_plan_paths(plan: dict[str, Any] | None) -> dict[str, Any] | None:
    """Drop invented DevOps paths from planner JSON before kickstart/hints."""
    if not isinstance(plan, dict):
        return plan
    out = dict(plan)
    tasks = out.get("subtasks")
    if not isinstance(tasks, list):
        return out
    fixed_tasks: list[Any] = []
    for item in tasks:
        if not isinstance(item, dict):
            fixed_tasks.append(item)
            continue
        it = dict(item)
        paths = it.get("paths")
        if isinstance(paths, list):
            clean = [
                p
                for p in paths
                if isinstance(p, str) and p.strip() and not _is_bogus_devops_path(p)
            ]
            if not clean and _looks_like_devops_plan(json.dumps(plan, ensure_ascii=False)):
                clean = [
                    _join_under(_DEFAULT_DEVOPS_ROOT, _DEVOPS_KEY_RELS[0])
                ]
            it["paths"] = clean
        fixed_tasks.append(it)
    out["subtasks"] = fixed_tasks
    return out


def _first_path_hint(
    plan: dict[str, Any] | None,
    tools: list[Any],
    messages: list[dict[str, Any]] | None = None,
) -> str | None:
    plan_text = json.dumps(plan or {}, ensure_ascii=False)
    blob_all = plan_text + "\n" + json.dumps(messages or [], ensure_ascii=False, default=str)
    devops = _looks_like_devops_plan(blob_all)
    root = _devops_root_from_messages(messages)
    if devops and not root:
        root = _DEFAULT_DEVOPS_ROOT

    candidates: list[str] = []
    if root:
        for rel in _DEVOPS_KEY_RELS:
            candidates.append(_join_under(root, rel))
        for p in _paths_mentioned(plan_text):
            if re.match(r"^[A-Za-z]:\\", p) or p.startswith("/"):
                candidates.append(p)
            elif not _is_bogus_devops_path(p):
                if "src" not in p.replace("\\", "/").split("/"):
                    candidates.append(_join_under(root, p))

    seen = _seen_path_keys(messages)
    failed = _failed_path_keys(messages)
    read_counts = _read_path_counts(messages)
    success = _successful_read_keys(messages)

    def _ok(path: str) -> bool:
        key = _norm_path_key(path)
        if key in seen or key in failed or key in success:
            return False
        if read_counts.get(key, 0) >= 1:
            return False
        if devops and _is_bogus_devops_path(path):
            return False
        return True

    for path in candidates:
        if _ok(path):
            return path
    if root:
        fallback = _join_under(root, "README.md")
        if _ok(fallback):
            return fallback
        # Do NOT re-return dynamic_conf.yml — that caused infinite read loops
        return None
    names = [_tool_name(t) for t in tools if isinstance(t, dict)]
    if any("read" in n.lower() for n in names):
        return "README.md"
    return "AGENTS.md"


def _read_path_counts(messages: list[dict[str, Any]] | None) -> dict[str, int]:
    counts: dict[str, int] = {}
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            if not any(k in name for k in ("read", "open", "cat", "get_file")):
                continue
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                parsed = None
            path = None
            if isinstance(parsed, dict):
                for key in ("path", "file", "target"):
                    val = parsed.get(key)
                    if isinstance(val, str) and val.strip():
                        path = val.strip()
                        break
            if path:
                k = _norm_path_key(path)
                counts[k] = counts.get(k, 0) + 1
    return counts


def _successful_read_keys(messages: list[dict[str, Any]] | None) -> set[str]:
    """Paths whose read tool result looks like file content (not an error)."""
    call_path: dict[str, str] = {}
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            if not any(k in name for k in ("read", "open", "cat", "get_file")):
                continue
            cid = str(tc.get("id") or "")
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                parsed = None
            if isinstance(parsed, dict):
                for key in ("path", "file", "target"):
                    val = parsed.get(key)
                    if isinstance(val, str) and val.strip() and cid:
                        call_path[cid] = val.strip()
                        break
    ok: set[str] = set()
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "tool":
            continue
        cid = str(msg.get("tool_call_id") or "")
        path = call_path.get(cid)
        if not path:
            continue
        content = str(msg.get("content") or "")
        low = content.lower()
        if (
            "not found" in low
            or "not in the project" in low
            or "path not found" in low
            or "error" in low[:80]
        ):
            continue
        if len(content.strip()) < 8:
            continue
        ok.add(_norm_path_key(path))
    return ok


def _pick_read_tool(tools: list[Any]) -> str:
    for t in tools:
        if not isinstance(t, dict):
            continue
        name = _tool_name(t)
        low = name.lower()
        if "read" in low and "file" in low:
            return name
        if low in ("read", "open", "cat"):
            return name
    for t in tools:
        if isinstance(t, dict) and _tool_name(t):
            return _tool_name(t)
    return "read_file"


def _pick_edit_tool(tools: list[Any]) -> str | None:
    prefer = (
        "edit_file",
        "apply_patch",
        "search_replace",
        "replace",
        "write_file",
        "create_file",
        "edit",
        "write",
    )
    names = {_tool_name(t): t for t in tools if isinstance(t, dict) and _tool_name(t)}
    low_map = {n.lower(): n for n in names}
    for key in prefer:
        for low, orig in low_map.items():
            if key == low or key in low:
                return orig
    return None


def _synthetic_tool_completion(
    *,
    model: str,
    tool_name: str,
    arguments: dict[str, Any],
) -> dict[str, Any]:
    call_id = f"call_synth_{uuid.uuid4().hex[:10]}"
    return {
        "id": f"chatcmpl-synth-{uuid.uuid4().hex[:12]}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model,
        "choices": [
            {
                "index": 0,
                "finish_reason": "tool_calls",
                "message": {
                    "role": "assistant",
                    "content": "",
                    "tool_calls": [
                        {
                            "id": call_id,
                            "type": "function",
                            "function": {
                                "name": tool_name,
                                "arguments": json.dumps(
                                    arguments, ensure_ascii=False
                                ),
                            },
                        }
                    ],
                },
            }
        ],
    }


def _strip_read_file_line_numbers(content: str) -> str:
    """Zed read_file prefixes 'NNNNNN\\t' — must strip before edit old_text."""
    out: list[str] = []
    for line in content.replace("\r\n", "\n").split("\n"):
        # right-aligned 6-digit field + tab (Zed agent read_file)
        if len(line) >= 7 and line[6] == "\t" and line[:6].strip().isdigit():
            out.append(line[7:])
        else:
            m = re.match(r"^\s*\d+\t(.*)$", line)
            out.append(m.group(1) if m else line)
    return "\n".join(out)


_TRUNC_MARKERS = ("\n<<TRUNCATED>>", "<<TRUNCATED>>", "\n…", "…", "\n...", "...")


def _strip_tool_truncation(content: str) -> str:
    """Remove gateway/compress truncation suffixes; keep usable file head."""
    c = content.replace("\r\n", "\n")
    changed = True
    while changed:
        changed = False
        for mark in _TRUNC_MARKERS:
            if c.endswith(mark):
                c = c[: -len(mark)].rstrip()
                changed = True
                break
    return c


def _is_edit_mismatch_text(content: str) -> bool:
    low = content.lower()
    return any(
        x in low
        for x in (
            "did not match",
            "could not find matching",
            "old_text did not match",
            "read the file again",
            "matching text for edit",
            "no match for old_text",
        )
    )


def _looks_like_file_body(content: str) -> bool:
    """True if tool content is file text (not an edit error / empty)."""
    c = _strip_tool_truncation(content).strip()
    if len(c) < 20:
        return False
    low = c.lower()
    if _is_edit_mismatch_text(c):
        return False
    if (
        "not found" in low
        or "not in the project" in low
        or low.startswith("error")
        or "invalid type" in low
        or "expected struct" in low
        or "failed to edit" in low
    ):
        return False
    return True


def _tool_result_for_path(
    messages: list[dict[str, Any]] | None, path: str
) -> str | None:
    """Return successful read_file content for path, if present in history."""
    want = _norm_path_key(path)
    call_path: dict[str, str] = {}
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            if not any(k in name for k in ("read", "open", "cat", "get_file")):
                continue
            cid = str(tc.get("id") or "")
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                parsed = None
            if isinstance(parsed, dict):
                p = parsed.get("path") or parsed.get("file") or parsed.get("target")
                if isinstance(p, str) and cid:
                    call_path[cid] = p
    # Prefer newest matching read (allow truncated head — strip marker)
    found: str | None = None
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "tool":
            continue
        cid = str(msg.get("tool_call_id") or "")
        p = call_path.get(cid)
        if not p or _norm_path_key(p) != want:
            continue
        content = str(msg.get("content") or "")
        if not _looks_like_file_body(content):
            continue
        body = _strip_tool_truncation(content)
        found = _strip_read_file_line_numbers(body)
    return found


def _edit_mismatch_recently(messages: list[dict[str, Any]] | None) -> bool:
    """True only if latest edit failed with mismatch AND no successful re-read after it.

    Scanning any of the last N tools caused infinite read_file after a good re-read.
    """
    # Map tool_call_id → path for reads
    read_call_path: dict[str, str] = {}
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            cid = str(tc.get("id") or "")
            if not cid:
                continue
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                parsed = None
            path = None
            if isinstance(parsed, dict):
                p = parsed.get("path") or parsed.get("file") or parsed.get("target")
                if isinstance(p, str) and p.strip():
                    path = p.strip()
            if path and any(k in name for k in ("read", "open", "cat", "get_file")):
                read_call_path[cid] = path

    edit_path = _last_edit_path(messages)
    want = _norm_path_key(edit_path) if edit_path else None
    for msg in reversed(messages or []):
        if not isinstance(msg, dict) or msg.get("role") != "tool":
            continue
        content = str(msg.get("content") or "")
        cid = str(msg.get("tool_call_id") or "")
        # Successful read of the edited path after mismatch → do NOT re-read again
        if want and cid in read_call_path:
            if _norm_path_key(read_call_path[cid]) == want and _looks_like_file_body(
                content
            ):
                return False
        if _is_edit_mismatch_text(content):
            return True
    return False


def _last_edit_path(messages: list[dict[str, Any]] | None) -> str | None:
    for msg in reversed(messages or []):
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in reversed(msg.get("tool_calls") or []):
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            if "edit" not in name:
                continue
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                continue
            if isinstance(parsed, dict):
                p = parsed.get("path")
                if isinstance(p, str) and p.strip():
                    return p.strip()
    return None


def _edit_failure_count(messages: list[dict[str, Any]] | None) -> int:
    n = 0
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "tool":
            continue
        low = str(msg.get("content") or "").lower()
        if any(
            x in low
            for x in (
                "failed to edit",
                "edit failed",
                "not unique",
                "ошибка редактир",
                "could not find",
                "multiple matches",
                "expected struct edit",
                "invalid type",
                "fuzzy match",
                "no match",
                "did not match",
                "matching text for edit",
                "eof while parsing",
                "invalid json",
                "error parsing input json",
                "parent directory doesn't exist",
            )
        ):
            n += 1
    return n


def _tried_old_texts(messages: list[dict[str, Any]] | None) -> set[str]:
    tried: set[str] = set()
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            if "edit" not in name:
                continue
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                continue
            if not isinstance(parsed, dict):
                continue
            for edit in parsed.get("edits") or []:
                if isinstance(edit, dict):
                    ot = edit.get("old_text") or edit.get("old_string")
                    if isinstance(ot, str) and ot.strip():
                        tried.add(ot)
            ot2 = parsed.get("old_text") or parsed.get("old_string")
            if isinstance(ot2, str) and ot2.strip():
                tried.add(ot2)
    return tried


def _unique_snippet_edit(
    content: str,
    *,
    exclude: set[str] | None = None,
) -> dict[str, str] | None:
    """Build a unique old→new replace from file body (avoid short ambiguous strings)."""
    marker = "# calentiq-ift-services-links: pending"
    if marker in content:
        return None
    exclude = exclude or set()
    lines = content.replace("\r\n", "\n").split("\n")
    # Prefer longer windows first after a failed edit (more unique)
    for width in (8, 7, 6, 5, 4, 3):
        limit = min(120, max(0, len(lines) - width + 1))
        for i in range(limit):
            chunk_lines = lines[i : i + width]
            if any(not ln.strip() for ln in chunk_lines):
                continue
            if any("…" in ln or ln.strip() == "..." for ln in chunk_lines):
                continue
            old = "\n".join(chunk_lines)
            if len(old) < 40:
                continue
            if old in exclude:
                continue
            if content.count(old) != 1:
                continue
            new = old + "\n" + marker
            return {"old_text": old, "new_text": new}
    return None


def _edit_args_for_path(
    path: str,
    plan: dict[str, Any] | None,
    messages: list[dict[str, Any]] | None = None,
) -> dict[str, Any] | None:
    """Zed edit_file shape: {path, edits:[{old_text, new_text}]} — not flat old_string."""
    del plan  # reserved for future plan-aware snippets
    body = _tool_result_for_path(messages, path)
    if not body:
        return None
    snippet = _unique_snippet_edit(body, exclude=_tried_old_texts(messages))
    if not snippet:
        return None
    return {
        "path": path,
        "edits": [
            {
                "old_text": snippet["old_text"],
                "new_text": snippet["new_text"],
            }
        ],
    }


def count_synthetic_tool_calls(messages: list[dict[str, Any]] | None) -> int:
    """How many gateway-injected tool_calls already ran (call_synth_* ids)."""
    n = 0
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            cid = str(tc.get("id") or "")
            if cid.startswith("call_synth_"):
                n += 1
    return n


def _assistant_tool_fingerprint(msg: dict[str, Any]) -> str | None:
    tcs = msg.get("tool_calls") or []
    if not isinstance(tcs, list) or not tcs:
        return None
    parts: list[str] = []
    for tc in tcs:
        if not isinstance(tc, dict):
            continue
        fn = tc.get("function") or {}
        name = str(fn.get("name") or "")
        args = str(fn.get("arguments") or "")[:240]
        parts.append(f"{name}:{args}")
    return "|".join(parts) if parts else None


def _sticky_tool_loop(messages: list[dict[str, Any]] | None, *, repeat: int = 3) -> bool:
    """True if the last `repeat` assistant tool turns are identical (Zed spin)."""
    fps: list[str] = []
    for msg in reversed(messages or []):
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        if not msg.get("tool_calls"):
            continue
        fp = _assistant_tool_fingerprint(msg)
        if not fp:
            continue
        fps.append(fp)
        if len(fps) >= repeat:
            break
    return len(fps) >= repeat and len(set(fps)) == 1


def synthetic_abort_reason(
    messages: list[dict[str, Any]] | None,
    *,
    max_synthetic: int = 3,
    max_edit_failures: int = 4,
) -> str | None:
    """If non-None, gateway must stop (finish=stop) — no more synthetic tool_calls."""
    synth_n = count_synthetic_tool_calls(messages)
    if max_synthetic > 0 and synth_n >= max_synthetic:
        return (
            f"Стоп: слишком много synthetic continue ({synth_n}≥{max_synthetic}). "
            "Модель Novita/LiteLLM отвечает 408 или пустым tool_calls — цикл прерван. "
            "Сократи задачу / попробуй снова; правку файла лучше сделать точечно вручную."
        )
    fails = _edit_failure_count(messages)
    if max_edit_failures > 0 and fails >= max_edit_failures:
        return (
            f"Стоп: edit_file падал {fails} раз(а). "
            "Нужен точный old_text из актуального read_file (без усечения) "
            "в формате {\"path\":\"...\",\"edits\":[{\"old_text\":\"...\",\"new_text\":\"...\"}]}."
        )
    if _sticky_tool_loop(messages, repeat=3):
        return (
            "Стоп: одинаковые tool_calls повторяются — агент зациклился. "
            "Прервано, чтобы не крутить litellm 408 → synthetic бесконечно."
        )
    return None


def _stop_synth(model: str, text: str) -> dict[str, Any]:
    return {
        "id": f"chatcmpl-synth-{uuid.uuid4().hex[:12]}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model,
        "choices": [
            {
                "index": 0,
                "finish_reason": "stop",
                "message": {"role": "assistant", "content": text},
            }
        ],
    }


def _continue_edit_or_grep(
    *,
    model: str,
    tools: list[Any],
    path: str,
    plan: dict[str, Any] | None,
    messages: list[dict[str, Any]] | None,
) -> dict[str, Any]:
    """Never finish_reason=stop after edit fail — that kills the Zed agent loop."""
    read_counts = _read_path_counts(messages)
    path_key = _norm_path_key(path)
    already_read = (
        read_counts.get(path_key, 0) >= 1
        or path_key in _successful_read_keys(messages)
        or bool(_tool_result_for_path(messages, path))
    )
    # Mismatch only if it is newer than any successful re-read of that path
    if _edit_mismatch_recently(messages):
        rpath = _last_edit_path(messages) or path
        return _synthetic_tool_completion(
            model=model,
            tool_name=_pick_read_tool(tools),
            arguments={"path": rpath},
        )
    edit_name = _pick_edit_tool(tools)
    if edit_name:
        args = _edit_args_for_path(path, plan, messages)
        if args:
            return _synthetic_tool_completion(
                model=model, tool_name=edit_name, arguments=args
            )
        # No body yet → one read; never re-read the same path in a loop
        if not already_read:
            return _synthetic_tool_completion(
                model=model,
                tool_name=_pick_read_tool(tools),
                arguments={"path": path},
            )
    # Fallback: grep for routers section to re-anchor without full re-read
    grep_name = None
    for t in tools:
        if not isinstance(t, dict):
            continue
        n = _tool_name(t).lower()
        if "grep" in n or n in ("search", "find"):
            grep_name = _tool_name(t)
            break
    if grep_name:
        return _synthetic_tool_completion(
            model=model,
            tool_name=grep_name,
            arguments={
                "pattern": "routers:|services:|middlewares:",
                "path": path.rsplit("\\", 1)[0] if "\\" in path else path,
            },
        )
    # Last resort: still emit edit_file with a minimal unique line from body
    body = _tool_result_for_path(messages, path) or ""
    lines = [ln for ln in body.split("\n") if ln.strip()]
    if edit_name and len(lines) >= 1:
        # pick longest unique single line
        for ln in sorted(lines, key=len, reverse=True):
            if len(ln) >= 24 and body.count(ln) == 1 and ln not in _tried_old_texts(messages):
                marker = "# calentiq-ift-services-links: pending"
                return _synthetic_tool_completion(
                    model=model,
                    tool_name=edit_name,
                    arguments={
                        "path": path,
                        "edits": [
                            {
                                "old_text": ln,
                                "new_text": ln + "\n" + marker,
                            }
                        ],
                    },
                )
    # Absolute last: ask model via a no-op-looking but valid tool — list_directory parent
    list_name = None
    for t in tools:
        if not isinstance(t, dict):
            continue
        n = _tool_name(t).lower()
        if "list" in n and "dir" in n:
            list_name = _tool_name(t)
            break
    if list_name:
        parent = path.rsplit("\\", 1)[0] if "\\" in path else path.rsplit("/", 1)[0]
        return _synthetic_tool_completion(
            model=model,
            tool_name=list_name,
            arguments={"path": parent},
        )
    # Only if no tools at all
    return _stop_synth(
        model,
        "edit_file нужен в формате "
        '{"path":"...","edits":[{"old_text":"...","new_text":"..."}]}. '
        f"Файл уже в контексте: {path}",
    )


def _last_successful_read_path(
    messages: list[dict[str, Any]] | None,
) -> str | None:
    """Absolute path of the newest successful read_file (for status / edit)."""
    success = _successful_read_keys(messages)
    if not success:
        return None
    last: str | None = None
    for msg in messages or []:
        if not isinstance(msg, dict) or msg.get("role") != "assistant":
            continue
        for tc in msg.get("tool_calls") or []:
            if not isinstance(tc, dict):
                continue
            name = str((tc.get("function") or {}).get("name") or "").lower()
            if not any(k in name for k in ("read", "open", "cat", "get_file")):
                continue
            args = (tc.get("function") or {}).get("arguments") or ""
            try:
                parsed = json.loads(args) if isinstance(args, str) else args
            except Exception:  # noqa: BLE001
                parsed = None
            if not isinstance(parsed, dict):
                continue
            for key in ("path", "file", "target"):
                val = parsed.get(key)
                if isinstance(val, str) and _norm_path_key(val) in success:
                    last = val.strip()
    return last


def _path_for_user_status(
    plan: dict[str, Any] | None,
    tools: list[Any],
    messages: list[dict[str, Any]] | None,
) -> str | None:
    """Prefer already-read / plan path — never the next unread KEY file (loadtest)."""
    got = _last_successful_read_path(messages)
    if got:
        return got
    for item in (plan or {}).get("subtasks") or []:
        if not isinstance(item, dict):
            continue
        for p in item.get("paths") or []:
            if isinstance(p, str) and p.strip() and not _is_bogus_devops_path(p):
                return p.strip()
    return _first_path_hint(plan, tools, messages)


def timeout_stop_message(reason: str, *, path: str | None = None) -> str:
    base = (
        "Executor (Novita/LiteLLM) не вернул tool_calls вовремя "
        f"({reason}). Synthetic mid-loop отключён — иначе Zed крутит ctx/408. "
        "Повтори короче или правь файл вручную."
    )
    if path:
        return base + f" Цель: `{path}`."
    return base


def midloop_stop_message(reason: str, *, path: str | None = None) -> str:
    """User-facing stop after tools already ran — never 'synthetic 1≥1' noise."""
    base = (
        f"Novita/LiteLLM timeout в tool_loop ({reason}). "
        "Повторный synthetic edit не делается (это и давало бесконечный ctx). "
        "Уточни шаг одной фразой / «ok» ещё раз, либо правь файл вручную."
    )
    if path:
        return base + f" Файл: `{path}`."
    return base


def executor_fallback_completion(
    *,
    plan: dict[str, Any] | None,
    tools: list[Any],
    model: str,
    messages: list[dict[str, Any]] | None,
    cfg: dict[str, Any] | None,
    reason: str = "timeout",
) -> dict[str, Any]:
    """Decide gateway fallback when LiteLLM 408 / empty tool_calls.

    Modes (hierarchical.executor_synthetic_mode):
      never: always stop with a clear message — no fake tool_calls.
      kickstart_only (default): at most one synthetic *read_file* before any
        tool activity; mid tool_loop → finish=stop (no fake edits).
      always: legacy synthetic edit/read with hard abort limits.
    """
    cfg = cfg or {}
    mode = str(cfg.get("executor_synthetic_mode") or "kickstart_only").lower()
    max_s = int(cfg.get("executor_max_synthetic_continues", 1) or 1)
    max_ef = int(cfg.get("executor_max_edit_failures", 4) or 4)
    in_loop = messages_have_tool_activity(messages)
    synth_n = count_synthetic_tool_calls(messages)

    if mode == "never":
        path = _path_for_user_status(plan, tools, messages)
        return _stop_synth(model, timeout_stop_message(reason, path=path))
    if mode != "always":
        # kickstart_only: never invent edit_file after tools already ran
        if in_loop or synth_n >= max(1, max_s):
            path = _path_for_user_status(plan, tools, messages)
            fails = _edit_failure_count(messages)
            if fails >= max_ef:
                text = (
                    f"Стоп: edit_file/write_file падал {fails} раз(а). "
                    "Нужен точный old_text из read_file "
                    '{"path":"...","edits":[{"old_text":"...","new_text":"..."}]}.'
                )
            elif _sticky_tool_loop(messages, repeat=3):
                text = (
                    "Стоп: одинаковые tool_calls повторяются — цикл прерван."
                )
            elif path and _last_successful_read_path(messages):
                # Prefer one more real edit attempt over a dead-end stop after 408
                cont = _continue_edit_or_grep(
                    model=model,
                    tools=tools,
                    path=path,
                    plan=plan,
                    messages=messages,
                )
                if completion_has_tool_calls(cont):
                    tc0 = (cont.get("choices") or [{}])[0].get("message", {}).get(
                        "tool_calls"
                    ) or []
                    name = str(
                        ((tc0[0].get("function") or {}).get("name")) if tc0 else ""
                    ).lower()
                    if "edit" in name or "write" in name or "grep" in name:
                        return cont
                text = midloop_stop_message(reason, path=path)
            else:
                # Do NOT surface synthetic_abort_reason "1≥1" — kickstart counts as 1
                text = midloop_stop_message(reason, path=path)
            return _stop_synth(model, text)
        # First kick only: read primary file — model continues after Zed returns body
        plan_text = json.dumps(plan or {}, ensure_ascii=False)
        devops = _looks_like_devops_plan(
            plan_text
            + "\n"
            + json.dumps(messages or [], ensure_ascii=False, default=str)
        )
        root = _devops_root_from_messages(messages)
        if devops and not root:
            root = _DEFAULT_DEVOPS_ROOT
        path = _first_path_hint(plan, tools, messages)
        if not path and root:
            path = _join_under(root, _DEVOPS_KEY_RELS[0])
        if not path:
            return _stop_synth(model, timeout_stop_message(reason))
        return _synthetic_tool_completion(
            model=model,
            tool_name=_pick_read_tool(tools),
            arguments={"path": path},
        )

    # Legacy always-mode
    return synthetic_first_tool_completion(
        plan=plan,
        tools=tools,
        model=model,
        messages=messages,
        max_synthetic=max_s,
        max_edit_failures=max_ef,
    )


def synthetic_first_tool_completion(
    *,
    plan: dict[str, Any] | None,
    tools: list[Any],
    model: str,
    messages: list[dict[str, Any]] | None = None,
    max_synthetic: int = 3,
    max_edit_failures: int = 4,
) -> dict[str, Any]:
    """Unblock Zed on timeout / empty tool_calls — keep the tool loop alive.

    Hard-stops after too many gateway synthetics / edit failures / sticky repeats
    so litellm 408 → synthetic cannot spin forever (ctx fill spam in Zed).
    """
    abort = synthetic_abort_reason(
        messages,
        max_synthetic=max_synthetic,
        max_edit_failures=max_edit_failures,
    )
    if abort:
        return _stop_synth(model, abort)

    plan_text = json.dumps(plan or {}, ensure_ascii=False)
    devops = _looks_like_devops_plan(
        plan_text + "\n" + json.dumps(messages or [], ensure_ascii=False, default=str)
    )
    root = _devops_root_from_messages(messages)
    if devops and not root:
        root = _DEFAULT_DEVOPS_ROOT
    primary = _join_under(root, _DEVOPS_KEY_RELS[0]) if root else None

    read_counts = _read_path_counts(messages)
    success = _successful_read_keys(messages)
    edit_name = _pick_edit_tool(tools)

    def _should_edit(path: str) -> bool:
        key = _norm_path_key(path)
        return key in success or read_counts.get(key, 0) >= 1

    target = primary
    path_hint = _first_path_hint(plan, tools, messages)
    if path_hint and _should_edit(path_hint):
        target = path_hint
    elif primary and _should_edit(primary):
        target = primary

    # After failed edit(s): retry with a different unique old_text — never stop the agent
    if target and _edit_failure_count(messages) > 0 and edit_name:
        return _continue_edit_or_grep(
            model=model,
            tools=tools,
            path=target,
            plan=plan,
            messages=messages,
        )

    if primary and _should_edit(primary) and edit_name:
        args = _edit_args_for_path(primary, plan, messages)
        if args:
            return _synthetic_tool_completion(
                model=model, tool_name=edit_name, arguments=args
            )
        return _continue_edit_or_grep(
            model=model,
            tools=tools,
            path=primary,
            plan=plan,
            messages=messages,
        )

    path = path_hint
    if path and not _should_edit(path):
        return _synthetic_tool_completion(
            model=model,
            tool_name=_pick_read_tool(tools),
            arguments={"path": path},
        )
    if path and _should_edit(path) and edit_name:
        args = _edit_args_for_path(path, plan, messages)
        if args:
            return _synthetic_tool_completion(
                model=model, tool_name=edit_name, arguments=args
            )
        return _continue_edit_or_grep(
            model=model, tools=tools, path=path, plan=plan, messages=messages
        )
    if primary and edit_name:
        args = _edit_args_for_path(primary, plan, messages)
        if args:
            return _synthetic_tool_completion(
                model=model, tool_name=edit_name, arguments=args
            )
        return _continue_edit_or_grep(
            model=model,
            tools=tools,
            path=primary,
            plan=plan,
            messages=messages,
        )
    if primary:
        return _continue_edit_or_grep(
            model=model,
            tools=tools,
            path=primary,
            plan=plan,
            messages=messages,
        )
    return _synthetic_tool_completion(
        model=model,
        tool_name=_pick_read_tool(tools),
        arguments={"path": "README.md"},
    )


def plan_payload_from_meta(meta: dict[str, Any]) -> dict[str, Any] | None:
    payload = meta.get("plan_payload")
    if isinstance(payload, dict) and payload.get("subtasks"):
        return sanitize_plan_paths(payload)
    return None


def merge_agent_meta(
    base: dict[str, Any],
    *,
    executor: str,
    plan: dict[str, Any] | None,
    phase: str,
) -> dict[str, Any]:
    out = dict(base)
    out["mode"] = "hierarchical_agent"
    out["agent_phase"] = phase
    out["executor_model"] = executor
    out["selected_model"] = executor
    if plan:
        out["plan_payload"] = plan
        out["subtask_count"] = len(plan.get("subtasks") or [])
    return out


def meta_header(meta: dict[str, Any] | None) -> str:
    """Serialize router meta for HTTP headers (latin-1 only).

    Cyrillic plan_payload in X-Router-Meta → UnicodeEncodeError → 500 in Zed.
    """
    skip = {"plan_payload", "plan", "content", "messages", "body"}
    safe: dict[str, Any] = {}
    for key, val in (meta or {}).items():
        if key in skip:
            if key == "plan_payload" and isinstance(val, dict):
                safe["subtask_count"] = len(val.get("subtasks") or [])
            continue
        if isinstance(val, (str, int, float, bool)) or val is None:
            safe[key] = val
        elif isinstance(val, (list, dict)):
            try:
                encoded = json.dumps(val, ensure_ascii=True, default=str)
            except Exception:  # noqa: BLE001
                continue
            if len(encoded) <= 600:
                safe[key] = json.loads(encoded)
        else:
            safe[key] = str(val)[:80]
    return json.dumps(safe, ensure_ascii=True, default=str)


def completion_has_tool_calls(data: dict[str, Any] | None) -> bool:
    if not isinstance(data, dict):
        return False
    choice = (data.get("choices") or [{}])[0]
    msg = choice.get("message") or {}
    tcs = msg.get("tool_calls")
    return isinstance(tcs, list) and len(tcs) > 0


def _tool_calls_from_completion(data: dict[str, Any] | None) -> list[dict[str, Any]]:
    if not isinstance(data, dict):
        return []
    choice = (data.get("choices") or [{}])[0]
    msg = choice.get("message") or {}
    tcs = msg.get("tool_calls") if isinstance(msg, dict) else None
    return [tc for tc in tcs if isinstance(tc, dict)] if isinstance(tcs, list) else []


def _read_paths_from_tool_calls(tcs: list[dict[str, Any]]) -> list[str]:
    paths: list[str] = []
    for tc in tcs:
        name = str((tc.get("function") or {}).get("name") or "").lower()
        if not any(k in name for k in ("read", "open", "cat", "get_file")):
            return []  # mixed with non-read → leave alone
        args = (tc.get("function") or {}).get("arguments") or ""
        try:
            parsed = json.loads(args) if isinstance(args, str) else args
        except Exception:  # noqa: BLE001
            parsed = None
        path = None
        if isinstance(parsed, dict):
            for key in ("path", "file", "target"):
                val = parsed.get(key)
                if isinstance(val, str) and val.strip():
                    path = val.strip()
                    break
        if path:
            paths.append(path)
    return paths


def rewrite_redundant_reread_completion(
    data: dict[str, Any],
    *,
    messages: list[dict[str, Any]] | None,
    plan: dict[str, Any] | None,
    tools: list[Any],
    model: str,
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """If model only re-reads already-loaded / bogus paths, force a better tool.

    After a 408→DeepSeek escalate we still saw read_file on the same yml —
    that looks like 'долго читает файл' and burns another round-trip.
    Also blocks invented EventHubDevOps\\docker-compose.yml (does not exist).
    """
    cfg = cfg or {}
    if not cfg.get("executor_rewrite_reread", True):
        return data
    tcs = _tool_calls_from_completion(data)
    if not tcs:
        return data
    paths = _read_paths_from_tool_calls(tcs)
    if not paths:
        return data
    success = _successful_read_keys(messages)
    read_counts = _read_path_counts(messages)
    failed = _failed_path_keys(messages)
    already = all(
        _norm_path_key(p) in success or read_counts.get(_norm_path_key(p), 0) >= 1
        for p in paths
    )
    bogus_or_failed = any(
        _is_bogus_devops_path(p) or _norm_path_key(p) in failed for p in paths
    )
    if not already and not bogus_or_failed:
        return data

    if bogus_or_failed and not already:
        # Prefer editing a file already in context over inventing another read
        if success:
            # recover a real path string matching a successful key
            path_for_edit = None
            for msg in messages or []:
                if not isinstance(msg, dict) or msg.get("role") != "assistant":
                    continue
                for tc in msg.get("tool_calls") or []:
                    if not isinstance(tc, dict):
                        continue
                    name = str((tc.get("function") or {}).get("name") or "").lower()
                    if not any(k in name for k in ("read", "open", "cat", "get_file")):
                        continue
                    args = (tc.get("function") or {}).get("arguments") or ""
                    try:
                        parsed = json.loads(args) if isinstance(args, str) else args
                    except Exception:  # noqa: BLE001
                        parsed = None
                    if not isinstance(parsed, dict):
                        continue
                    for key in ("path", "file", "target"):
                        val = parsed.get(key)
                        if isinstance(val, str) and _norm_path_key(val) in success:
                            path_for_edit = val.strip()
            if path_for_edit:
                log.info(
                    "rewrite bogus read → edit already-read path=%s", path_for_edit
                )
                return _continue_edit_or_grep(
                    model=model,
                    tools=tools,
                    path=path_for_edit,
                    plan=plan,
                    messages=messages,
                )
        # Invented / previously missing path → read a real DevOps file instead
        good = _first_path_hint(plan, tools, messages)
        if not good:
            return data
        log.info("rewrite bogus/failed read_file → %s", good)
        return _synthetic_tool_completion(
            model=model,
            tool_name=_pick_read_tool(tools),
            arguments={"path": good},
        )

    # Prefer edit on the path that was re-requested
    path = paths[0]
    rewritten = _continue_edit_or_grep(
        model=model,
        tools=tools,
        path=path,
        plan=plan,
        messages=messages,
    )
    # If continue still wants read (mismatch case), keep original re-read
    r_tcs = _tool_calls_from_completion(rewritten)
    if r_tcs:
        r_name = str((r_tcs[0].get("function") or {}).get("name") or "").lower()
        if any(k in r_name for k in ("read", "open", "cat", "get_file")):
            return data
    log.info(
        "rewrite redundant read_file → %s path=%s",
        (
            ((r_tcs[0].get("function") or {}).get("name"))
            if r_tcs
            else "stop"
        ),
        path,
    )
    return rewritten


def completion_preview(data: dict[str, Any] | None, *, n: int = 160) -> str:
    """Short preview for logs when model returns prose instead of tools."""
    if not isinstance(data, dict):
        return ""
    choice = (data.get("choices") or [{}])[0]
    msg = choice.get("message") or {}
    fr = choice.get("finish_reason")
    content = msg.get("content") if isinstance(msg, dict) else None
    rc = msg.get("reasoning_content") if isinstance(msg, dict) else None
    text = content if isinstance(content, str) and content.strip() else ""
    if not text and isinstance(rc, str):
        text = rc
    return f"finish={fr} content={str(text)[:n]!r}"


def force_kickstart_read(
    *,
    plan: dict[str, Any] | None,
    tools: list[Any],
    model: str,
    messages: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Unblock Zed: one read_file on plan path (first turn only)."""
    path = _first_path_hint(plan, tools, messages)
    if not path:
        root = _devops_root_from_messages(messages) or _DEFAULT_DEVOPS_ROOT
        path = _join_under(root, _DEVOPS_KEY_RELS[0])
    return _synthetic_tool_completion(
        model=model,
        tool_name=_pick_read_tool(tools),
        arguments={"path": path},
    )


def completion_to_sse_chunks(
    data: dict[str, Any],
    *,
    cid: str,
    model: str,
    drop_content_if_tools: bool = True,
) -> list[bytes]:
    """Turn a non-stream chat.completion into OpenAI SSE chunks for Zed.

    Emit tool_calls in a single delta (full arguments) — fragmented argument
    streams confuse some Zed builds and look like a stuck/one-shot agent.
    """
    choice = (data.get("choices") or [{}])[0]
    msg = choice.get("message") or {}
    finish = choice.get("finish_reason") or "stop"
    chunks: list[bytes] = []

    def pack(delta: dict[str, Any], fr: str | None = None) -> bytes:
        payload = {
            "id": cid,
            "object": "chat.completion.chunk",
            "created": int(data.get("created") or 0) or int(time.time()),
            "model": model,
            "choices": [{"index": 0, "delta": delta, "finish_reason": fr}],
        }
        return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode("utf-8")

    tool_calls = msg.get("tool_calls")
    has_tools = isinstance(tool_calls, list) and bool(tool_calls)

    chunks.append(pack({"role": "assistant"}))

    content = msg.get("content")
    if (
        isinstance(content, str)
        and content.strip()
        and not (drop_content_if_tools and has_tools)
    ):
        text = content
        step = 800
        for i in range(0, len(text), step):
            chunks.append(pack({"content": text[i : i + step]}))

    if has_tools:
        finish = "tool_calls"
        streamed = []
        for i, tc in enumerate(tool_calls):
            if not isinstance(tc, dict):
                continue
            fn = tc.get("function") or {}
            args = fn.get("arguments") or ""
            if not isinstance(args, str):
                args = json.dumps(args, ensure_ascii=False)
            streamed.append(
                {
                    "index": i,
                    "id": tc.get("id") or f"call_{i}",
                    "type": tc.get("type") or "function",
                    "function": {
                        "name": fn.get("name") or "",
                        "arguments": args,
                    },
                }
            )
        chunks.append(pack({"tool_calls": streamed}, finish))
    else:
        chunks.append(pack({}, finish))
    chunks.append(b"data: [DONE]\n\n")
    return chunks