"""Tool-round guardrails: failure classification, retry prompts, loop detection."""
from __future__ import annotations

import json
import logging
import os
import re
from dataclasses import dataclass, field
from typing import Any

from agent_loop_tool_parse import (
    _TOOL_CALL_BLOCK_RE,
    _TOOL_CALL_TAG_RE,
    _extract_unclosed_tool_call_block,
    _parse_xml_tool_call_block,
    _parse_xml_tool_calls,
)
from agent_loop_tool_repair import _json_loads_lenient, _salvage_json_tool_call
from agent_tool_format import (
    NATIVE_TOOL_CALL_FORMAT_BULLETS,
    GUARD_TOOL_CALL_MINIMAL,
    GUARD_TOOL_CALL_SUFFIX,
)
from drive_engine import BACKGROUND_HEARTBEAT_THREAD_ID, HEARTBEAT_THREAD_PREFIX
from tools.conversation_trace import emit_guardrail_event, is_trace_enabled
from utils.transcript_roles import (
    extract_inner_tool_response_text,
    is_tool_result_carrier_message,
    is_real_user_message,
)

_log = logging.getLogger(__name__)

_NOOP_PROMISE_RE = re.compile(
    r"\b(let me|i(?:'|’)ll|i will|retry|do this correctly|get the full text|i did(?:n'?t| not) actually)\b",
    re.IGNORECASE,
)
# Follow-up intent without requiring a trailing ":" (local models often end with a period).
_NOOP_RETRY_PHRASE_RE = re.compile(
    r"\b(?:"
    r"one more time|"
    r"(?:try|retry)(?:ing)?\s+again|"
    r"try again|"
    r"another (?:try|attempt)|"
    r"attempt(?:ing)? again|"
    r"give(?: it)? another (?:try|go|shot)"
    r")\b",
    re.IGNORECASE,
)
_NOOP_ACTION_START_RE = re.compile(r"\b(let me|i(?:'|’)ll|i will)\b", re.IGNORECASE)
_NOOP_ACTION_VERB_RE = re.compile(
    r"\b(check|open|read|run|search|research(?:ing)?|fetch|inspect|analy[sz]e|call|use|look|pull|grab|try|retry|execute|start|create|build)\b",
    re.IGNORECASE,
)
_NOOP_NON_ACTION_PHRASE_RE = re.compile(
    r"\b(i(?:'|’)ll\s+take|i(?:'|’)ll\s+explain|i(?:'|’)ll\s+be\b)\b",
    re.IGNORECASE,
)
_NOOP_PLACEHOLDER_SUFFIX_RE = re.compile(r"(?::\s*$|\.\.\.\s*$)")
# Local models often end action promises with a period, not ":" or "...".
_NOOP_TRAILING_ACTION_PERIOD_RE = re.compile(
    r"(?i)(?:let me|i(?:'|’)ll|i will)\b[^\n]{0,300}\b"
    r"(?:check|open|read|run|search|research(?:ing)?|fetch|inspect|analy[sz]e|call|use|look|pull|grab|execute|start|create|build)\b"
    r"[^\n]{0,160}\.\s*$"
)
_NOOP_LET_ME_DO_IT_RE = re.compile(r"(?i)\blet me\b.*\b(?:just\s+)?do it\b")
_NOOP_PROPER_TAIL_RE = re.compile(r"\bproperly\s*\.?\s*$", re.IGNORECASE)
# Local/chat models sometimes emit `search_sessions(...)` as prose with no tool_calls.
_TEXT_ONLY_SEARCH_SESSIONS_SHIM_RE = re.compile(
    r"(?:^\s*|\n\s*)search_sessions\s*[\(\[]",
    re.IGNORECASE,
)
_TOOL_ERROR_HINT_RE = re.compile(
    r"(\[policy blocked\]|policy blocked|tool error:|validationerror|filenotfounderror|no such file|missing required positional argument|"
    r"timed out|timeout|invoke_tool|runtime request|\-32001|no such target|stale session)",
    re.IGNORECASE,
)
_TOOL_FAILURE_PREFIXES = (
    "tool error:",
    "batch rejected:",
    "error:",
    "unknown tool:",
    "path does not exist:",
    "path is a directory, not a file:",
    "path is not a directory:",
    "path exists but is not a file:",
    "file not found:",
    "file not found or not a file:",
    "file not readable:",
    "permission denied:",
    "operation blocked:",
    "blocked:",
    "refused:",
    "read failed:",
    "write failed:",
    "append failed:",
)


def _structured_tool_result_reports_errors(result_text: str) -> bool:
    """True when a JSON tool payload reports operational failure (e.g. lint ok:false)."""
    text = (result_text or "").strip()
    if not text.startswith("{"):
        return False
    try:
        data = json.loads(text)
    except Exception:
        return False
    if not isinstance(data, dict):
        return False
    if data.get("ok") is False:
        return True
    try:
        if int(data.get("errorCount") or 0) > 0:
            return True
    except (TypeError, ValueError):
        pass
    return False


def _tool_result_indicates_action_blocked(result_text: str) -> bool:
    """True when a tool result reports a blocked action (render/lint gate, etc.)."""
    lowered = (result_text or "").strip().lower()
    if not lowered:
        return False
    if "render blocked" in lowered or "(render blocked)" in lowered:
        return True
    if lowered.startswith("offline validation errors"):
        return True
    if _structured_tool_result_reports_errors(result_text):
        return True
    return False


def _tool_result_indicates_failure(result_text: str) -> bool:
    """Return True when a tool result clearly indicates failure."""
    text = (result_text or "").strip()
    if not text:
        return False
    if _tool_result_indicates_action_blocked(text):
        return True
    lowered = text.lower()
    if lowered.startswith("[search_sessions_error]"):
        return True
    return any(lowered.startswith(prefix) for prefix in _TOOL_FAILURE_PREFIXES)


def _tool_failure_kind(result_text: str, *, policy_blocked: bool = False) -> str | None:
    """Return a stable typed failure kind for tool outcomes."""
    if policy_blocked:
        return "policy_blocked"
    text = (result_text or "").strip()
    if not text:
        return None
    lower = text.lower()
    if lower.startswith("unknown tool:"):
        return "unknown_tool"
    if lower.startswith("tool error:"):
        return "tool_runtime_error"
    if lower.startswith("error:"):
        return "validation_error"
    if lower.startswith("operation blocked:") or lower.startswith("blocked:") or lower.startswith("refused:"):
        return "policy_blocked"
    if _tool_result_indicates_failure(text):
        return "tool_failure"
    return None


ROUND_OUTCOME_OK = "ok"
ROUND_OUTCOME_UNKNOWN = "unknown_tool"
ROUND_OUTCOME_POLICY = "policy_blocked"
_KANBAN_HANDOFF_MIN_RAW_CHARS = 80


def _classify_tool_result(result_text: str) -> str:
    """Classify a single tool result for the consecutive-failure loop guard.

    Returns one of ``ROUND_OUTCOME_OK`` / ``ROUND_OUTCOME_UNKNOWN`` /
    ``ROUND_OUTCOME_POLICY``.  Classification is stable regardless of the tier
    the model used to resolve the tool (``tools_by_name`` vs
    ``full_tools_by_name``) — by the time this runs, ``_invoke_tool`` has
    already produced either an "Unknown tool:" string (genuinely unregistered),
    a ``[Policy blocked]`` string, or a real result, so the guard does not
    falsely abort when ``find_tools`` is still in the middle of resolving.
    """
    # Imported lazily to avoid circulars during module load.
    from tool_policy import POLICY_BLOCKED_PREFIX as _POLICY_BLOCKED_PREFIX

    text = result_text if isinstance(result_text, str) else str(result_text or "")
    stripped = text.lstrip()
    if stripped.startswith("Unknown tool:"):
        return ROUND_OUTCOME_UNKNOWN
    if stripped.startswith(_POLICY_BLOCKED_PREFIX):
        return ROUND_OUTCOME_POLICY
    return ROUND_OUTCOME_OK


def _update_consecutive_failure_streaks(
    round_outcomes: list[tuple[str, str]],
    prev_unknown_streak: int,
    prev_policy_streak: int,
    prev_policy_names: tuple[str, ...],
) -> tuple[int, int, tuple[str, ...]]:
    """Apply streak update rules for a single finished round.

    Pure helper so the streak arithmetic is unit-testable without running the
    whole agent loop.  Returns updated ``(unknown_streak, policy_streak,
    last_policy_names)``.  Rules:

    - A round only counts when **every** call shares the same failure class.
    - Any OK outcome in the round resets both streaks.
    - Policy streak only increments when the same set of blocked tool names
      repeats — a preference change, not a model retry, fixes the problem.
    """
    if not round_outcomes:
        return prev_unknown_streak, prev_policy_streak, prev_policy_names

    kinds = {kind for _, kind in round_outcomes}
    all_unknown = kinds == {ROUND_OUTCOME_UNKNOWN}
    all_policy = kinds == {ROUND_OUTCOME_POLICY}
    had_ok = ROUND_OUTCOME_OK in kinds

    unknown_streak = prev_unknown_streak + 1 if all_unknown else (0 if had_ok else prev_unknown_streak)

    if all_policy:
        blocked_names = tuple(
            sorted({name for name, kind in round_outcomes if kind == ROUND_OUTCOME_POLICY and name})
        )
        if blocked_names and blocked_names == prev_policy_names:
            policy_streak = prev_policy_streak + 1
        else:
            policy_streak = 1
        last_policy_names = blocked_names
    elif had_ok:
        policy_streak = 0
        last_policy_names = ()
    else:
        policy_streak = prev_policy_streak
        last_policy_names = prev_policy_names

    return unknown_streak, policy_streak, last_policy_names


def _update_consecutive_hard_tool_error_streak(
    prev: int,
    *,
    tool_success_flags: list[bool],
    round_outcomes: list[tuple[str, str]],
    eligible: bool,
) -> int:
    """Increment when every tool in the round failed (``success`` false) with real tool errors.

    Unknown-tool and policy-only rounds are excluded — those have dedicated loop
    guards. Empty ``tool_success_flags`` leaves the streak unchanged.
    """
    if not eligible or not tool_success_flags:
        return prev
    if any(tool_success_flags):
        return 0
    if not round_outcomes:
        return 0
    kinds = {kind for _, kind in round_outcomes}
    if kinds == {ROUND_OUTCOME_UNKNOWN} or kinds == {ROUND_OUTCOME_POLICY}:
        return 0
    return prev + 1


def _visible_text_has_kanban_result(text: str) -> bool:
    """True when *text* contains a parseable trailing ``kanban-result`` fence."""
    try:
        from utils.kanban_context import parse_kanban_result_detailed

        return parse_kanban_result_detailed(text or "").parsed is not None
    except Exception:
        return False


def kanban_synthesize_missing_contract_enabled() -> bool:
    """Whether the harness may synthesize a trailing kanban-result fence at worker exit.

    Fails **closed** on a config-read error: synthesis is a backup, so a broken/unreadable
    config must not silently enable it (which would mask the reliability signal). When config is
    readable, the default stays **on** (``raw is None`` -> True) so the fallback is fully
    preserved in all normal operation — only a genuine read failure declines to synthesize.
    """
    try:
        from core.config import get_kanban_config

        raw = get_kanban_config().get("synthesize_missing_contract")
        if raw is None:
            return True
        if isinstance(raw, str):
            return raw.strip().lower() in {"1", "true", "yes", "on"}
        return bool(raw)
    except Exception:
        _log.warning(
            "kanban synthesize-contract config unreadable; failing closed (no synthesis this exit)",
            exc_info=True,
        )
        return False


@dataclass
class KanbanRunLedger:
    """Per-run progress snapshot for kanban worker completion synthesis."""

    task_kind: str = ""
    deliverable_type: str = ""
    tools_executed: set[str] = field(default_factory=set)
    recent_paths: list[str] = field(default_factory=list)
    append_scratch_count: int = 0
    scratch_path: str = ""
    report_generated: bool = False
    save_research_called: bool = False
    wrote_text: bool = False
    render_tools_succeeded: set[str] = field(default_factory=set)
    last_scratch_count_at_checkpoint: int = 0
    loop_break_reason: str = ""
    handoff_retries_exhausted: bool = False
    # Count of in-context turns the harness injected specifically to elicit/repair the
    # completion handoff (kanban completion-repair prompt + tool-budget close turn) BEFORE any
    # synthesis fallback fired. Phase 2c observability: lets salvage telemetry distinguish a
    # handoff we synthesized *after* giving the model a chance to fix it (repair_attempts > 0)
    # from one we synthesized with zero in-context repair (repair_attempts == 0, the worse case).
    handoff_repair_attempts: int = 0
    inband_summary: str = ""
    inband_metadata: dict[str, Any] = field(default_factory=dict)
    prose_excerpt: str = ""
    last_phase: str = ""
    research_gather_chars: int = 0
    research_gather_tool_count: int = 0

    def note_prose(self, text: str, *, max_chars: int = 400) -> None:
        clean = " ".join(str(text or "").split())
        if clean:
            self.prose_excerpt = clean[:max_chars]


def _sanitize_kanban_fence_text(text: str) -> str:
    """Strip triple-backticks so nested fences cannot truncate the kanban-result block."""
    return str(text or "").replace("```", "''").strip()


def _format_kanban_result_message(parsed: dict[str, Any]) -> str:
    summary = _sanitize_kanban_fence_text((parsed or {}).get("summary") or "")
    metadata = (parsed or {}).get("metadata")
    if not isinstance(metadata, dict):
        metadata = {}
    md_out: dict[str, Any] = {}
    for key, val in metadata.items():
        if isinstance(val, str):
            md_out[key] = _sanitize_kanban_fence_text(val)
        elif isinstance(val, list):
            md_out[key] = [
                _sanitize_kanban_fence_text(x) if isinstance(x, str) else x for x in val
            ]
        else:
            md_out[key] = val
    body = json.dumps(
        {"summary": summary, "metadata": md_out},
        ensure_ascii=False,
    )
    return f"{summary}\n\n```kanban-result\n{body}\n```"


def _parsed_with_ledger_tools(
    parsed: dict[str, Any],
    ledger: KanbanRunLedger,
) -> dict[str, Any]:
    """Attach run ledger tool names so acceptance ``require_tools_any`` can pass.

    Also stamps Phase 2c ``repair_attempts`` provenance: every synthesized/salvaged handoff
    routes through here, so recording the number of in-context repair turns we injected before
    synthesizing makes "synthesized after repair vs synthesized with zero repair" measurable in
    persisted run metadata (aggregated by ``kanban_storage.salvage_stats``).
    """
    from utils.kanban_context import merge_tools_used_metadata

    md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
    out = dict(parsed)
    merged = merge_tools_used_metadata(md, ledger.tools_executed)
    try:
        attempts = int(ledger.handoff_repair_attempts or 0)
    except (TypeError, ValueError):
        attempts = 0
    if attempts > 0 and "repair_attempts" not in merged:
        merged["repair_attempts"] = attempts
    out["metadata"] = merged
    return out


def _parsed_handoff_is_salvage(parsed: dict[str, Any]) -> bool:
    md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
    status = str(md.get("status") or "").strip().lower()
    if status in ("partial", "blocked"):
        return True
    blockers = md.get("blockers")
    if isinstance(blockers, list) and blockers:
        return True
    if isinstance(blockers, str) and str(blockers).strip():
        return True
    return False


# Blockers that describe a missing/garbled trailing fence (a *formatting* glitch the run
# ledger can legitimately disprove), NOT a substantive statement about the work being
# incomplete. Only these may auto-upgrade a partial/blocked fence to success.
_CONTRACT_FORMAT_BLOCKERS = frozenset(
    {
        "contract_missing",
        "contract_missing_prose_only",
        "contract_missing_after_research",
        "empty_visible_handoff",
    }
)


def _salvage_is_contract_format_only(parsed: dict[str, Any]) -> bool:
    """True only when a salvage fence's blockers are purely about a missing/garbled
    contract (a formatting glitch), not the model's honest self-assessment that the work
    itself is incomplete. A bare partial/blocked with no blockers is treated as a genuine
    self-assessment and is NOT format-only — so it is preserved, never silently upgraded."""
    md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
    blockers = md.get("blockers")
    items: list[str] = []
    if isinstance(blockers, list):
        items = [str(b).strip().lower() for b in blockers if str(b).strip()]
    elif isinstance(blockers, str) and blockers.strip():
        items = [blockers.strip().lower()]
    if not items:
        return False
    return all(b in _CONTRACT_FORMAT_BLOCKERS for b in items)


def _strip_misleading_success_status(parsed: dict[str, Any]) -> dict[str, Any]:
    """Successful completions use empty metadata.status, not completed/done."""
    md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
    status = str(md.get("status") or "").strip().lower()
    if status not in ("completed", "done", "success"):
        return parsed
    md = dict(md)
    md.pop("status", None)
    out = dict(parsed)
    out["metadata"] = md
    return out


def _handoff_should_resynthesize_from_ledger(
    parsed: dict[str, Any],
    ledger: KanbanRunLedger,
    *,
    task_metadata: dict[str, Any] | None = None,
) -> bool:
    """True when run progress supersedes a stale salvage fence or in-band summary.

    A model-authored partial/blocked fence is the model's honest self-assessment and is
    preserved — run-ledger progress only auto-upgrades it to success when the blockers are
    purely a missing/garbled-contract *formatting* glitch (``_salvage_is_contract_format_only``).
    This keeps salvage a backup that never silently overrides a substantive "not finished"
    signal from the model.
    """
    if _parsed_handoff_is_salvage(parsed) and not _salvage_is_contract_format_only(parsed):
        return False
    if ledger.inband_summary.strip():
        return True
    if not _ledger_deliverable_completed(ledger, task_metadata=task_metadata):
        return False
    return _parsed_handoff_is_salvage(parsed)


def _ledger_deliverable_completed(
    ledger: KanbanRunLedger,
    *,
    task_metadata: dict[str, Any] | None = None,
) -> bool:
    if ledger.report_generated or ledger.save_research_called:
        return True
    try:
        from utils.kanban_context import (
            kanban_task_expects_lint,
            kanban_task_expects_render,
            kanban_task_is_web_act_deliverable,
        )

        if kanban_task_is_web_act_deliverable(task_metadata):
            return "run_web_script" in ledger.render_tools_succeeded
        expects_render = kanban_task_expects_render(task_metadata)
        expects_lint = kanban_task_expects_lint(task_metadata)
    except Exception:
        expects_render = False
        expects_lint = False
    if expects_render:
        return bool(ledger.render_tools_succeeded)
    write_hits = {
        n
        for n in ledger.tools_executed
        if n in _WRITE_TEXT_TOOLS or str(n).endswith("_render")
    }
    wrote = bool(write_hits) and ledger.wrote_text
    if expects_lint and wrote:
        return any(str(n).endswith("_lint") for n in ledger.tools_executed)
    if ledger.render_tools_succeeded:
        return True
    return wrote


def _ledger_inband_completion_handoff(
    ledger: KanbanRunLedger,
    *,
    task_metadata: dict[str, Any] | None = None,
    prose: str = "",
) -> str | None:
    """Return a formatted completion fence from the worker's in-band
    ``kanban_post_summary`` when it passes verification, else ``None``.

    Shared by the missing-fence finalizer and the loop hard-stop so a worker that
    *finished its work and then stalled* hands off its completed deliverable
    rather than a synthesized "no progress" blocked salvage (incident
    kt_6cc4cd8e862947: research worker re-posted update_task_plan to the round cap
    after already posting a usable summary).
    """
    from utils.kanban_context import (
        normalize_salvage_handoff,
        verify_kanban_result,
    )

    if not str(getattr(ledger, "inband_summary", "") or "").strip():
        return None
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    # Provenance: a deliberate tool submission is a FIRST-CLASS completion
    # (handoff_source="tool_submit") under native_result_handoff; otherwise it is
    # the salvage substitution ("inband_summary"). salvage_stats counts
    # tool_submit as a clean completion (not salvage).
    _inband_source = "inband_summary"
    try:
        from utils.kanban_context import _native_result_handoff_enabled

        if _native_result_handoff_enabled():
            _inband_source = "tool_submit"
    except Exception:
        _log.debug("native_result_handoff flag read failed in finalize", exc_info=True)
    parsed: dict[str, Any] = _parsed_with_ledger_tools(
        {
            "summary": ledger.inband_summary.strip(),
            "metadata": {
                **dict(ledger.inband_metadata or {}),
                "handoff_source": _inband_source,
            },
        },
        ledger,
    )
    ok, _reason, _detail = verify_kanban_result(
        parsed,
        task_metadata=tm,
        worker_transcript=prose,
    )
    if not ok:
        return None
    parsed = normalize_salvage_handoff(
        parsed,
        worker_transcript=prose,
        task_metadata=tm,
    )
    return _format_kanban_result_message(parsed)


def synthesize_kanban_contract_from_ledger(
    ledger: KanbanRunLedger,
    *,
    task_metadata: dict[str, Any] | None = None,
    visible_prose: str = "",
) -> str:
    """Build a verifier-aligned fenced kanban-result from run progress."""
    from utils.kanban_context import (
        normalize_salvage_handoff,
        verify_kanban_result,
    )

    tm = task_metadata if isinstance(task_metadata, dict) else {}
    task_kind = str(ledger.task_kind or tm.get("task_kind") or "").strip().lower()
    prose = " ".join(str(visible_prose or ledger.prose_excerpt or "").split())
    artifact_paths = [p for p in ledger.recent_paths if p][:4]

    _inband_completion = _ledger_inband_completion_handoff(
        ledger, task_metadata=tm, prose=prose
    )
    if _inband_completion is not None:
        return _inband_completion

    if _ledger_deliverable_completed(ledger, task_metadata=tm) and len(prose) >= 12:
        summary = prose[:2000] if prose else "Kanban worker completed deliverable steps."
        # Provenance: success-shaped handoff synthesized by the harness from run-ledger
        # progress (the model omitted its trailing fence). Stamped so this never looks like
        # native contract compliance in storage/API — salvage must be observable.
        metadata: dict[str, Any] = {"handoff_source": "ledger_synthesized"}
        if artifact_paths:
            metadata["artifact_paths"] = artifact_paths
        if ledger.scratch_path:
            metadata.setdefault("artifact_paths", artifact_paths)
            if ledger.scratch_path not in (metadata.get("artifact_paths") or []):
                paths = list(metadata.get("artifact_paths") or [])
                paths.insert(0, ledger.scratch_path)
                metadata["artifact_paths"] = paths[:4]
        try:
            from utils.kanban_context import kanban_task_is_web_act_deliverable

            if kanban_task_is_web_act_deliverable(tm):
                run_dir = next(
                    (
                        p
                        for p in (metadata.get("artifact_paths") or artifact_paths)
                        if "/web_runs/kanban_" in str(p).replace("\\", "/")
                    ),
                    "",
                )
                if run_dir:
                    import os

                    if run_dir.endswith((".json", ".png", ".py")):
                        run_dir = os.path.dirname(run_dir)
                    metadata["web_act_run_dir"] = run_dir
                    try:
                        from tools.web_act import validate_web_act_result

                        validated = validate_web_act_result(run_dir, fresh_run_expected=True)
                        if validated.get("fresh_run_ok"):
                            metadata["fresh_run_ok"] = True
                        if validated.get("answer"):
                            metadata["answer"] = validated["answer"]
                        if validated.get("paths"):
                            metadata["artifact_paths"] = list(validated["paths"])[:8]
                    except Exception:
                        pass
        except Exception:
            pass
        parsed = _parsed_with_ledger_tools(
            {"summary": summary, "metadata": metadata},
            ledger,
        )
        ok, reason, detail = verify_kanban_result(
            parsed,
            task_metadata=tm,
            worker_transcript=prose,
        )
        if ok:
            parsed = normalize_salvage_handoff(
                parsed,
                worker_transcript=prose,
                task_metadata=tm,
            )
            return _format_kanban_result_message(parsed)

    blockers: list[str] = []
    failure_reason = "kanban_contract_synthesized"
    status = "blocked"
    if not prose.strip() and not ledger.tools_executed:
        blockers = ["empty_visible_handoff"]
        summary = "Kanban worker ended without visible handoff or tool progress."
        next_action = "Retry the task with an explicit tool step or trailing kanban-result fence."
    elif task_kind == "research" and ledger.append_scratch_count < 2:
        blockers = ["contract_missing_after_research"]
        status = "partial"
        failure_reason = "research_scratch_incomplete"
        excerpt = prose[:240] if prose else "Research gathered but not saved to scratch."
        summary = (
            f"Research run ended without a kanban-result fence. {excerpt}"
        ).strip()
        next_action = (
            "Call append_research_scratch at least twice with synthesis from prior searches, "
            "then generate_final_research_report or save_research_findings, "
            "then end with a trailing kanban-result fenced JSON block."
        )
    elif prose.strip():
        blockers = ["contract_missing_prose_only"]
        status = "partial" if task_kind == "research" else "blocked"
        failure_reason = "contract_missing_prose_only"
        summary = prose[:2000]
        next_action = (
            "Return the same substance in summary form inside a trailing "
            "kanban-result fenced JSON block (summary + metadata)."
        )
        if task_kind == "research":
            next_action = (
                "Summarize findings via save_research_findings or generate_final_research_report, "
                "then end with a trailing kanban-result fenced JSON block."
            )
    else:
        blockers = ["contract_missing"]
        summary = "Kanban worker ended without a parseable kanban-result handoff."
        next_action = "End with a trailing kanban-result fenced JSON block describing outcomes and next steps."

    metadata = {
        "status": status,
        "blockers": blockers,
        "next_action": next_action,
        "failure_reason": failure_reason,
        "handoff_source": "ledger_salvage",
    }
    if artifact_paths:
        metadata["artifact_paths"] = artifact_paths
    if ledger.scratch_path:
        metadata.setdefault("artifact_paths", [])
        if ledger.scratch_path not in metadata["artifact_paths"]:
            metadata["artifact_paths"] = [ledger.scratch_path, *metadata["artifact_paths"]][:4]

    parsed = _parsed_with_ledger_tools(
        {"summary": summary[:2000], "metadata": metadata},
        ledger,
    )
    parsed = normalize_salvage_handoff(
        parsed,
        worker_transcript=prose,
        task_metadata=tm,
    )
    ok, verifier_reason, verifier_detail = verify_kanban_result(
        parsed,
        task_metadata=tm,
        worker_transcript=prose,
    )
    if not ok:
        md = dict(parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {})
        md["status"] = "blocked"
        md["blockers"] = list(md.get("blockers") or []) or ["verifier_failed_on_synthesis"]
        md["failure_reason"] = verifier_reason or "verifier_failed_on_synthesis"
        if not str(md.get("next_action") or "").strip():
            md["next_action"] = (
                f"Fix handoff format ({verifier_detail or 'verifier failed'}) and retry with "
                "a trailing kanban-result fenced JSON block."
            )
        parsed = normalize_salvage_handoff(
            {"summary": str(parsed.get("summary") or summary)[:2000], "metadata": md},
            worker_transcript=prose,
            task_metadata=tm,
        )
    return _format_kanban_result_message(parsed)


def _format_existing_kanban_fence(
    source: str,
    ledger: KanbanRunLedger,
    *,
    task_metadata: dict[str, Any] | None = None,
) -> str | None:
    """Parse and format an existing fence, or None when synthesis should run."""
    from utils.kanban_context import normalize_salvage_handoff, parse_kanban_result_detailed

    if not (source or "").strip() or not _visible_text_has_kanban_result(source):
        return None
    try:
        detail = parse_kanban_result_detailed(source)
    except Exception:
        return None
    parsed = detail.parsed if isinstance(detail.parsed, dict) else None
    if parsed is None:
        return None
    if _handoff_should_resynthesize_from_ledger(
        parsed,
        ledger,
        task_metadata=task_metadata,
    ):
        return None
    parsed = _strip_misleading_success_status(
        _parsed_with_ledger_tools(parsed, ledger),
    )
    normalized = normalize_salvage_handoff(
        parsed,
        worker_transcript=source,
        task_metadata=task_metadata,
    )
    return _format_kanban_result_message(normalized)


COUNCIL_RESEARCHER_SEARCH_BUDGET = 4
COUNCIL_RESEARCHER_MIN_GATHER_TOOLS = 2
COUNCIL_RESEARCHER_MAX_CLOSE_RETRIES = 2
COUNCIL_RESEARCHER_MAX_GATHER_CONTINUATIONS = 1
COUNCIL_RESEARCHER_GATHER_TOOLS = frozenset({
    "read_file",
    "read_file_chunk",
    "read_webpage",
    "deep_web_search",
    "query_knowledge",
    "ask_studied_corpus",
})


def is_council_researcher_gather_tool(tool_name: str) -> bool:
    return str(tool_name or "").strip() in COUNCIL_RESEARCHER_GATHER_TOOLS


def council_researcher_gather_sufficient(
    gather_tool_count: int,
    *,
    search_budget_exhausted: bool = False,
    has_preseed: bool = False,
) -> bool:
    """True when Athena has done enough gather work to enter CLOSE (not merely started)."""
    if has_preseed or search_budget_exhausted:
        return True
    return int(gather_tool_count) >= int(COUNCIL_RESEARCHER_MIN_GATHER_TOOLS)


def is_council_researcher_task(task_metadata: dict[str, Any] | None) -> bool:
    try:
        from utils.council_orchestrator import is_council_task_metadata

        tm = task_metadata if isinstance(task_metadata, dict) else {}
        return (
            is_council_task_metadata(tm)
            and str(tm.get("council_seat") or "").strip().lower() == "researcher"
        )
    except Exception:
        return False


def council_tool_result_is_soft_guidance(result_text: str) -> bool:
    """Dedup / loop-guidance tool output — not a hard failure for council workers."""
    tl = str(result_text or "").strip().lower()
    if not tl:
        return False
    if tl.startswith("[loop warning]"):
        return True
    if tl.startswith("[council guidance]"):
        return True
    return any(
        marker in tl
        for marker in (
            "already searched",
            "very similar query",
            "previous search result",
            "do not call search_web again",
            "do not call deep_web_search again",
            "do not call search_web or deep_web_search again",
            "pre-seeded product source",
        )
    )


def council_researcher_search_budget_exceeded(search_count: int) -> bool:
    return int(search_count) >= int(COUNCIL_RESEARCHER_SEARCH_BUDGET)


def maybe_repair_council_researcher_handoff(
    handoff_text: str,
    *,
    task_metadata: dict[str, Any] | None,
    gather_excerpt: str,
    sig_labels: str = "",
    worker_prose: str = "",
) -> tuple[str, bool]:
    """Replace false blocked / missing-claims Researcher handoffs when gather text exists."""
    if not is_council_researcher_task(task_metadata):
        return handoff_text, False
    excerpt = " ".join(str(gather_excerpt or "").split())
    if len(excerpt) < 200:
        return handoff_text, False
    try:
        from utils.council_orchestrator import verify_council_seat_extended
        from utils.kanban_context import parse_kanban_result_detailed

        parsed = parse_kanban_result_detailed(handoff_text).parsed
        if parsed is None:
            repaired = synthesize_council_researcher_completion_handoff(
                excerpt,
                sig_labels=sig_labels or "council_finalize_salvage",
                worker_prose=worker_prose,
            )
            return repaired, True
        md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
        status = str(md.get("status") or "").strip().lower()
        claims = md.get("claims")
        has_claims = isinstance(claims, list) and len(claims) > 0
        ok, _, _ = verify_council_seat_extended(
            parsed,
            task={"metadata": task_metadata or {}},
            task_metadata=task_metadata or {},
        )
        if ok and has_claims and status not in ("blocked",):
            return handoff_text, False
        if status == "blocked" or not has_claims or not ok:
            repaired = synthesize_council_researcher_completion_handoff(
                excerpt,
                sig_labels=sig_labels or "council_finalize_salvage",
                worker_prose=worker_prose,
            )
            return repaired, True
    except Exception:
        _log.debug("council researcher handoff repair failed", exc_info=True)
    return handoff_text, False


def council_researcher_gather_evidence_block(
    gather_excerpt: str,
    *,
    max_chars: int = 2400,
) -> str:
    """Inline evidence recap for CLOSE nudges and retries (prevents 'no research done' hallucinations)."""
    raw = str(gather_excerpt or "").strip()
    if len(raw) < 120:
        return ""
    clean = _council_researcher_usable_gather_excerpt(raw)
    if len(clean) < 120:
        return ""
    return (
        "\n\nGathered evidence (you already executed search/read tools — use this; "
        "do NOT claim no research was done):\n"
        f"{clean[:max_chars]}"
    )


def format_council_researcher_gather_continuation_prompt(gather_tool_count: int) -> str:
    """Injected when Athena stops in GATHER with fewer than the contract minimum of sources.

    One search's snippets plus parametric memory produced stale recommendations (two-year-old
    model names for a 'best local LLM today' frame) — force one real follow-up gather.
    """
    n = max(int(gather_tool_count), 0)
    return (
        f"Council Researcher GATHER incomplete: you have used only {n} gather call(s); "
        "the contract requires at least two distinct sources before CLOSE.\n"
        "Continue gathering NOW — do not write your final claims yet:\n"
        "- Call read_webpage on the most promising URL from your search results. For "
        "time-sensitive frames (best/latest/current X), pick the most recently published "
        "source.\n"
        "- Or run ONE refined deep_web_search with different terms.\n"
        "Ground every model/product/version name in the gathered sources — your prior "
        "knowledge may be years out of date; snippets alone are not enough.\n"
        "After that gather call, CLOSE with synthesized claims and the trailing "
        "```kanban-result``` fence (council_seat: researcher, claims[])."
    )


def format_council_researcher_max_rounds_handoff_prompt(gather_excerpt: str) -> str:
    excerpt_block = council_researcher_gather_evidence_block(gather_excerpt)
    return (
        "Council Researcher FINAL handoff. Do NOT call any more tools.\n"
        "You already executed search/read tools above. Synthesize claims from that evidence "
        "for the Captain's frame."
        f"{excerpt_block}\n\n"
        "End with a trailing ```kanban-result``` JSON fence. Required metadata:\n"
        "- council_seat: researcher\n"
        "- claims: [{id, text, evidence[], confidence}] — at least one claim\n"
        "- uncertainties: [] or list gaps\n"
        "Do NOT set status to blocked. Do NOT say no research was conducted."
    )


def format_council_researcher_close_retry_prompt(
    gather_excerpt: str,
    *,
    reason: str = "",
) -> str:
    """Guard-retry prompt when CLOSE synthesis failed but gather evidence exists."""
    evidence = council_researcher_gather_evidence_block(gather_excerpt)
    reason_bits: list[str] = []
    r = str(reason or "").strip().lower()
    if r in ("close_tool_refusal", "close_invalid_args"):
        reason_bits.append(
            "You attempted to call tools during CLOSE — tools are disabled for this seat now."
        )
    elif r in ("invalid_handoff_needs_synthesis", "missing_fence_with_gather"):
        reason_bits.append(
            "Your prior response did not produce valid researcher claims "
            "(empty claims, blocked status, or falsely claimed no research was done)."
        )
    prefix = " ".join(reason_bits)
    if prefix:
        prefix = prefix + " "
    return (
        f"Council Researcher CLOSE retry: {prefix}"
        "Do NOT call any tools. Write plain-language claim sentences grounded in the evidence below. "
        "Put source URLs only in each claim's evidence[] array — never embed Title:/URL:/Snippet: in text."
        f"{evidence}\n\n"
        "End with a trailing ```kanban-result``` JSON fence. Required metadata:\n"
        "- council_seat: researcher\n"
        "- claims: [{id, text, evidence[], confidence}] — at least one claim\n"
        "- uncertainties: [] or list gaps\n"
        "Do NOT set status to blocked."
    )


def maybe_normalize_council_verifier_handoff(
    handoff_text: str,
    *,
    task_metadata: dict[str, Any] | None,
) -> tuple[str, bool]:
    """Drop a failure-ish metadata.status from a structurally complete Verifier verdict.

    Verifier models describe an all-claims-rejected verdict as ``status: "failed"``,
    but that is a *successful* verification run; the generic contract verifier reads
    failure statuses as worker failure and the seat gives up after the retry budget
    (kcn_c32156cfe1b340e6). Only strips when a structured verdict is present.
    """
    try:
        from utils.council_orchestrator import is_council_task_metadata
        from utils.kanban_context import (
            _COUNCIL_VERIFIER_VERDICT_FAILURE_STATUSES,
            parse_kanban_result_detailed,
        )

        tm = task_metadata if isinstance(task_metadata, dict) else {}
        if not is_council_task_metadata(tm):
            return handoff_text, False
        if str(tm.get("council_seat") or "").strip().lower() != "verifier":
            return handoff_text, False
        parsed = parse_kanban_result_detailed(handoff_text).parsed
        if not isinstance(parsed, dict):
            return handoff_text, False
        md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
        status = str(md.get("status") or "").strip().lower()
        if status not in _COUNCIL_VERIFIER_VERDICT_FAILURE_STATUSES:
            return handoff_text, False
        has_verdict = isinstance(md.get("verified_claims"), list) or (
            isinstance(md.get("rejections"), list) and md.get("rejections")
        )
        if not has_verdict:
            return handoff_text, False
        md = dict(md)
        md.pop("status", None)
        normalized = dict(parsed)
        normalized["metadata"] = md
        return _format_kanban_result_message(normalized), True
    except Exception:
        _log.debug("council verifier handoff normalize failed", exc_info=True)
    return handoff_text, False


def council_researcher_bad_handoff_retry_reason(
    handoff_text: str,
    *,
    task_metadata: dict[str, Any] | None,
    gather_excerpt: str,
) -> tuple[str, str] | None:
    """Return (guard_type, reason) when CLOSE should retry synthesis instead of salvage."""
    if not is_council_researcher_task(task_metadata):
        return None
    excerpt = " ".join(str(gather_excerpt or "").split())
    if len(excerpt) < 200:
        return None
    try:
        from utils.council_orchestrator import verify_council_seat_extended
        from utils.kanban_context import parse_kanban_result_detailed

        parsed = parse_kanban_result_detailed(handoff_text).parsed
        if parsed is None:
            return ("council_researcher_close_handoff", "missing_fence_with_gather")
        md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
        status = str(md.get("status") or "").strip().lower()
        claims = md.get("claims")
        has_claims = isinstance(claims, list) and len(claims) > 0
        ok, _, _ = verify_council_seat_extended(
            parsed,
            task={"metadata": task_metadata or {}},
            task_metadata=task_metadata or {},
        )
        if ok and has_claims and status not in ("blocked",):
            return None
        return ("council_researcher_close_handoff", "invalid_handoff_needs_synthesis")
    except Exception:
        _log.debug("council researcher bad handoff retry check failed", exc_info=True)
    return None


def finalize_kanban_worker_output(
    visible: str,
    raw: str,
    ledger: KanbanRunLedger,
    *,
    task_metadata: dict[str, Any] | None = None,
    gather_excerpt: str = "",
) -> tuple[str, bool]:
    """Ensure dispatcher-visible text includes a parseable kanban-result fence.

    Returns ``(final_text, synthesized)``.
    """
    visible_clean = (visible or "").strip()
    raw_clean = (raw or "").strip()
    prose = visible_clean or raw_clean

    for source in (visible_clean, raw_clean):
        formatted = _format_existing_kanban_fence(
            source,
            ledger,
            task_metadata=task_metadata,
        )
        if formatted is not None:
            normalized, was_normalized = maybe_normalize_council_verifier_handoff(
                formatted,
                task_metadata=task_metadata,
            )
            if was_normalized:
                return normalized, True
            repaired, was_repaired = maybe_repair_council_researcher_handoff(
                formatted,
                task_metadata=task_metadata,
                gather_excerpt=gather_excerpt,
                worker_prose=prose,
            )
            return repaired, was_repaired

    if is_council_researcher_task(task_metadata):
        excerpt = " ".join(str(gather_excerpt or "").split())
        if len(excerpt) >= 200:
            salvaged = synthesize_council_researcher_completion_handoff(
                excerpt,
                sig_labels="council_finalize_missing_fence",
                worker_prose=prose,
            )
            return salvaged, True

    if not kanban_synthesize_missing_contract_enabled():
        return visible, False

    try:
        from utils.council_orchestrator import is_council_task_metadata

        if is_council_task_metadata(task_metadata):
            return visible, False
    except Exception:
        _log.debug("council synthesize gate check failed", exc_info=True)

    synthesized = synthesize_kanban_contract_from_ledger(
        ledger,
        task_metadata=task_metadata,
        visible_prose=prose,
    )
    return synthesized, True


def _format_kanban_close_contract_prompt(*, prefer_tool: bool = False) -> str:
    """Close turn for a kanban worker that stopped without a completion handoff.

    Phase 2a mandatory close turn: distinct framing from the max-rounds prompt (the worker did
    NOT exhaust its budget — it simply ended without the required handoff). Native-only by contract;
    never instructs XML tool formats.

    When ``prefer_tool`` (Grain-reframe Stage B, ``[kanban].native_result_handoff`` on), ask the
    worker to CALL the native ``kanban_post_summary`` tool — tools remain available this round —
    with the trailing fence kept as the backup. Otherwise ask for a tool-less trailing fence (the
    pre-Stage-B path).
    """
    if prefer_tool:
        return (
            "You ended your turn without a completion handoff. Finish now by calling the "
            "kanban_post_summary tool with your final result:\n"
            "  kanban_post_summary(summary=\"<dense factual handoff>\", metadata={...}, "
            'status="success")\n'
            "This native tool call is the primary handoff. If you cannot finish, call it with "
            'status="partial" or status="blocked" plus metadata.blockers and metadata.next_action.\n'
            "As a backup you may instead end your message with a trailing fenced JSON block "
            "(```kanban-result``` ... ```) of the same shape. "
            "Put all user-visible content outside <think> tags."
        )
    return (
        "You ended your turn without a completion handoff. Do NOT call any more tools.\n"
        "Respond now with a concise visible summary of what you accomplished "
        "(even if partial or blocked) and end your message with a trailing fenced "
        "JSON block in this exact format:\n"
        "```kanban-result\n"
        '{"summary": "<dense factual handoff>", '
        '"metadata": {"status": "partial|done|blocked", ...}}\n'
        "```\n"
        "Put all user-visible content outside <think> tags. "
        "If work is incomplete, say so in summary and set metadata.status to "
        '"partial" with metadata.blockers when applicable.'
    )


def _format_kanban_max_rounds_handoff_prompt() -> str:
    """Tool-less final round for kanban workers that hit the tool-call budget."""
    return (
        "You have reached your tool-call budget for this kanban worker run. "
        "Do NOT call any more tools.\n"
        "Respond now with a concise visible summary of what you accomplished "
        "(even if partial or blocked) and end your message with a trailing fenced "
        "JSON block in this exact format:\n"
        "```kanban-result\n"
        '{"summary": "<dense factual handoff>", '
        '"metadata": {"status": "partial|done|blocked", ...}}\n'
        "```\n"
        "Put all user-visible content outside <think> tags. "
        "If work is incomplete, say so in summary and set metadata.status to "
        '"partial" with metadata.blockers when applicable.'
    )


def _kanban_handoff_retry_reason(
    *,
    raw_before_finalize: str,
    finalized_text: str,
    min_raw_chars: int = _KANBAN_HANDOFF_MIN_RAW_CHARS,
) -> tuple[str, str] | None:
    """Return ``(guard_type, reason)`` when a kanban final-answer repair should retry.

    Trigger A: finalize removed all visible content even though the model emitted
    substantial raw text. Trigger B: visible text exists but does not contain a
    parseable kanban-result fence (``contract_missing``). Trigger C: no visible
    text at all (common when the run stops on a tool round at max_rounds).
    """
    raw_clean = (raw_before_finalize or "").strip()
    final_clean = (finalized_text or "").strip()
    if not final_clean and len(raw_clean) >= max(1, int(min_raw_chars)):
        return ("kanban_handoff_empty_visible", "finalized_empty_after_sanitize")
    if not final_clean:
        return ("kanban_handoff_missing_result_fence", "empty_visible_handoff")
    try:
        from utils.kanban_context import parse_kanban_result_detailed

        parsed = parse_kanban_result_detailed(final_clean)
        raw_parsed = parse_kanban_result_detailed(raw_clean) if raw_clean else parsed
    except Exception:
        return None
    if parsed.parsed is None:
        if raw_parsed.parsed is not None:
            return ("kanban_handoff_fence_in_raw_only", "fence_stripped_from_visible")
        reason = str(parsed.reason or "missing_result_fence").strip() or "missing_result_fence"
        return ("kanban_handoff_missing_result_fence", reason)
    return None


def _kanban_unverified_write_handoff_reason(
    *,
    tools_executed: set[str],
    visible_tool_names: set[str],
    wrote_text: bool,
    task_metadata: dict[str, Any] | None = None,
) -> str | None:
    """Return a reason when a kanban worker wrote files but skipped available lint tools."""
    if not wrote_text:
        return None
    try:
        from utils.kanban_context import kanban_task_expects_lint

        if not kanban_task_expects_lint(task_metadata):
            return None
    except Exception:
        return None
    lint_tools = {n for n in visible_tool_names if str(n).endswith("_lint")}
    if not lint_tools:
        return None
    if tools_executed & lint_tools:
        return None
    return "unverified_write_before_handoff"


def _format_kanban_unverified_write_handoff_prompt(*, lint_tool_names: list[str]) -> str:
    tools_display = ", ".join(f"`{n}`" for n in lint_tool_names[:4])
    extra = f" (and {len(lint_tool_names) - 4} more)" if len(lint_tool_names) > 4 else ""
    return (
        "Kanban verification step: you wrote or updated file content this run but have not "
        f"run a lint/verify tool yet ({tools_display}{extra}). "
        "Run the appropriate lint tool on the artifact you changed and fix any reported errors "
        "before ending with your visible summary and ```kanban-result``` fence. "
        "If lint cannot pass, end with kanban-result metadata describing the blocking errors."
    )


def _format_render_blocked_recovery_prompt(*, tool_name: str, result_preview: str) -> str:
    preview = " ".join(str(result_preview or "").split())[:480]
    return (
        f"`{tool_name}` reported a blocked action and did not complete:\n"
        f"{preview or '(no preview)'}\n\n"
        "Fix the underlying artifact (for example vendor local scripts instead of remote "
        "`<script src>`, or lint errors in the target file), re-run lint if available, "
        "then retry the render tool. Do not end with prose-only status or a success "
        "kanban-result until the render tool returns success."
    )


def _kanban_render_incomplete_handoff_reason(
    *,
    tools_executed: set[str],
    visible_tool_names: set[str],
    render_tools_succeeded: set[str],
    render_blocked_seen: bool,
    task_metadata: dict[str, Any] | None = None,
) -> str | None:
    """Return a reason when a kanban worker exits without a successful render tool."""
    try:
        from utils.kanban_context import kanban_task_expects_render

        if not kanban_task_expects_render(task_metadata):
            return None
    except Exception:
        return None
    render_tools = {n for n in visible_tool_names if str(n).endswith("_render")}
    if not render_tools:
        return None
    attempted = render_tools & tools_executed
    if not attempted:
        return None
    if render_tools_succeeded & attempted:
        return None
    if render_blocked_seen or attempted:
        return "render_incomplete_before_handoff"
    return None


def _format_kanban_render_incomplete_handoff_prompt(*, render_tool_names: list[str]) -> str:
    tools_display = ", ".join(f"`{n}`" for n in render_tool_names[:4])
    return (
        "Kanban render verification: you invoked a render tool "
        f"({tools_display}) but none succeeded in this run. "
        "Fix the blocking validation/lint issues, execute a successful render tool call, "
        "then end with a ```kanban-result``` fence that includes the output path in metadata. "
        "If render cannot pass, use metadata.status=\"blocked\" with blockers describing the error."
    )


def _format_truncated_read_file_followup(
    *,
    path_bucket: str,
    kanban_video_render: bool = False,
    preview_chars: int = 5000,
) -> str:
    from utils.path_normalize import truncate_path_display

    path_display = truncate_path_display(path_bucket) if path_bucket else "this file"
    if kanban_video_render:
        return (
            f"Kanban video render: `read_file` on `{path_display}` is a preview "
            f"(up to {int(preview_chars):,} chars). The file on disk is not truncated or corrupt. "
            "Do not call read_file on this path again. Patch the HTML with write_text_file "
            "(data-composition-id, data-width, data-height, GSAP timeline registry), "
            "then run hyperframes_lint once and hyperframes_render once."
        )
    return (
        f"Note: `read_file` on `{path_display}` returned only the first {int(preview_chars):,} characters. "
        "Use `read_file_chunk` (chunk_index 0, 1, 2, ...) to read and patch the rest; "
        "do not call `read_file` on the same path again."
    )


def _format_hyperframes_lint_failure_followup(result_preview: str) -> str:
    preview = str(result_preview or "").strip()[:1200]
    return (
        "Kanban: hyperframes_lint reported errors. The composition file on disk is fine to edit. "
        "Use write_text_file to add data-composition-id, data-width, data-height on `<div id=\"root\">`, "
        "register GSAP timelines (`window.__timelines`), then run hyperframes_lint again, then "
        "hyperframes_render once. End with kanban-result metadata.output or artifact_paths "
        "(not only output_path). Do not loop read_file.\n"
        f"Lint output (excerpt): {preview}"
    )


def _is_recoverable_all_failed_round(
    *,
    tool_success_flags: list[bool],
    round_outcomes: list[tuple[str, str]],
) -> bool:
    """True when every tool call failed with runtime/validation-like outcomes."""
    if not tool_success_flags:
        return False
    if any(tool_success_flags):
        return False
    if not round_outcomes:
        return False
    kinds = {kind for _, kind in round_outcomes}
    if kinds == {ROUND_OUTCOME_UNKNOWN} or kinds == {ROUND_OUTCOME_POLICY}:
        return False
    return True


def _format_all_tools_failed_recovery_message(
    failed_details: list[tuple[str, str, str]],
    *,
    max_chars: int = 1800,
) -> str:
    """Return a compact recovery prompt from the latest failed tool round."""
    if not failed_details:
        return (
            "The last tool round failed and appears recoverable. Retry once with corrected tool arguments, "
            "a different tool, or a concrete blocker from tool output."
        )
    lines = [
        "Every tool call in the last round failed.",
        "Before writing prose, repair the next step using this evidence:",
    ]
    budget = max(400, int(max_chars))
    for tool_name, args_summary, preview in failed_details:
        label = str(tool_name or "tool")
        arg_text = (args_summary or "").strip()
        p = " ".join(str(preview or "").split())
        seg = f"- {label}({arg_text or '-'}) -> {p or 'no error preview captured'}"
        if sum(len(x) + 1 for x in lines) + len(seg) > budget:
            lines.append("- (additional failures omitted for brevity)")
            break
        lines.append(seg)
    lines.append("Suggested next step: fix args/tool choice and execute at least one tool call now.")
    return "\n".join(lines)


_MULTI_STEP_KEYWORDS = (
    "refactor", "migrate", "migration", "investigate", "debug", "audit",
    "implement", "overhaul", "rewrite", "port ", "build out", "integrate",
    "set up", "scaffold", "plan ", "plan:", "plan out",
)
_MULTI_STEP_ENUM_RE = re.compile(r"(^|\n)\s*(?:[-*]|\d+[.)])\s+\S", re.MULTILINE)
_MULTI_STEP_ORDER_RE = re.compile(
    r"\b(first|then|after that|afterwards|finally|lastly|next,)\b", re.IGNORECASE
)


def _multi_step_user_message(text: str) -> bool:
    """Heuristic: does *text* look like a multi-step request that warrants a plan?

    False positives are harmless (one extra nudge line); false negatives just
    mean no nudge. Thresholds are conservative.
    """
    if not isinstance(text, str):
        return False
    s = text.strip()
    if len(s) < 40:
        return False
    lower = s.lower()
    hits = 0
    if _MULTI_STEP_ENUM_RE.search(s):
        hits += 2  # explicit list beats everything
    if _MULTI_STEP_ORDER_RE.search(s):
        hits += 1
    if any(kw in lower for kw in _MULTI_STEP_KEYWORDS):
        hits += 1
    # Multiple questions in one turn often imply multi-part work.
    if s.count("?") >= 2:
        hits += 1
    return hits >= 2


def _is_heartbeat_background_thread(thread_id: str) -> bool:
    tid = (thread_id or "").strip()
    return tid == BACKGROUND_HEARTBEAT_THREAD_ID or tid.startswith(HEARTBEAT_THREAD_PREFIX)


def _heartbeat_exploratory_miss_ok(tool_name: str, result_text: str) -> bool:
    """True when a missing path is normal for background heartbeats (optional reads / probes)."""
    t = (result_text or "").strip().lower()
    if tool_name in ("read_file", "read_file_chunk", "list_directory"):
        return t.startswith("path does not exist:")
    # Heartbeat probes often ask location/time/weather questions without enough context.
    if tool_name in ("get_weather", "web_search", "search_web", "search_web_public"):
        return (
            "missing location" in t
            or "provide a location" in t
            or "no results found" in t
            or "no search results returned" in t
        )
    return False
def _has_recent_tool_error(messages: list[dict], lookback: int = 8) -> bool:
    if not isinstance(messages, list) or not messages:
        return False
    for m in reversed(messages[-lookback:]):
        if not isinstance(m, dict):
            continue
        content = m.get("content")
        text = ""
        if (m.get("role") or "").strip() == "tool" and isinstance(content, str):
            text = content
        elif is_tool_result_carrier_message(m):
            text = extract_inner_tool_response_text(content)
        if text and (
            _TOOL_ERROR_HINT_RE.search(text)
            or _tool_result_indicates_failure(text)
        ):
            return True
    return False


_MAX_COMPLETION_CRITIC_SNIPPET_CHARS = 800


def _last_user_text_goal(messages: list[dict]) -> str:
    for m in reversed(messages or []):
        if not isinstance(m, dict):
            continue
        if (m.get("role") or "").strip() != "user":
            continue
        raw = m.get("content")
        if not isinstance(raw, str):
            continue
        t = raw.strip()
        if t:
            return t
    return ""


def _recent_tool_digest_for_completion_critic(
    messages: list[dict],
    *,
    max_items: int = 6,
    max_total: int = 2400,
) -> str:
    chunks: list[str] = []
    total = 0
    for m in reversed(messages or []):
        if len(chunks) >= max_items:
            break
        if not isinstance(m, dict):
            continue
        role = (m.get("role") or "").strip()
        text = ""
        if role == "tool" and isinstance(m.get("content"), str):
            text = m["content"]
        elif is_tool_result_carrier_message(m):
            text = extract_inner_tool_response_text(m.get("content"))
        text = (text or "").strip()
        if not text:
            continue
        cap = _MAX_COMPLETION_CRITIC_SNIPPET_CHARS
        piece = text if len(text) <= cap else text[:cap] + "…"
        if total + len(piece) + 2 > max_total:
            break
        chunks.append(piece)
        total += len(piece) + 2
    chunks.reverse()
    return "\n---\n".join(chunks)


def _recent_policy_blocked_shell_execute(messages: list[dict], lookback: int = 8) -> bool:
    from tool_policy import POLICY_BLOCKED_PREFIX

    pref = POLICY_BLOCKED_PREFIX.lower()
    needle = "shell_execute"
    for m in reversed((messages or [])[-lookback:]):
        if not isinstance(m, dict):
            continue
        content = m.get("content")
        text = ""
        if (m.get("role") or "").strip() == "tool" and isinstance(content, str):
            text = content
        elif is_tool_result_carrier_message(m):
            text = extract_inner_tool_response_text(content)
        lowered = (text or "").lower()
        if pref in lowered and needle in lowered:
            return True
    return False


def _noop_tail_for_action_scan(text: str) -> str:
    """Last paragraph of the reply — models bury action promises after apologies."""
    parts = [p.strip() for p in re.split(r"\n\s*\n+", (text or "").strip()) if p.strip()]
    return parts[-1] if parts else (text or "").strip()


def _noop_action_promise_chunk(chunk: str, *, has_retry_phrase: bool) -> bool:
    """True when a text chunk is a short action placeholder that owes a tool call."""
    piece = (chunk or "").strip()
    if not piece:
        return False
    if not _NOOP_ACTION_START_RE.search(piece):
        return False
    if _NOOP_NON_ACTION_PHRASE_RE.search(piece):
        return False
    if _NOOP_LET_ME_DO_IT_RE.search(piece):
        return True
    if not (_NOOP_ACTION_VERB_RE.search(piece) or has_retry_phrase):
        return False
    if has_retry_phrase:
        return True
    if _NOOP_PROPER_TAIL_RE.search(piece):
        return True
    if _NOOP_PLACEHOLDER_SUFFIX_RE.search(piece):
        return True
    if _NOOP_TRAILING_ACTION_PERIOD_RE.search(piece):
        return True
    return False


def _is_noop_action_promise(content: str, messages: list[dict]) -> bool:
    """Detect "I'll do X" replies that should be a tool call instead."""
    if not isinstance(content, str):
        return False
    text = content.strip()
    if not text:
        return False
    recent_err = _has_recent_tool_error(messages)
    has_promise_cue = bool(_NOOP_PROMISE_RE.search(text))
    has_retry_phrase = bool(_NOOP_RETRY_PHRASE_RE.search(text))
    if not has_promise_cue and not has_retry_phrase:
        return False

    # Strong signal when we just saw a tool/runtime failure (including timeouts / CDP noise).
    if recent_err:
        # After a failed tool, models often emit a long explanation then "try one more time";
        # allow a much larger cap than the no-error path.
        if len(text) > 2200:
            return False
        if has_promise_cue:
            return True
        # Retry phrasing alone can appear in neutral prose ("one more time the API…");
        # require an explicit action lead-in.
        if has_retry_phrase and (
            _NOOP_ACTION_START_RE.search(text)
            or re.search(r"^\s*(?:trying|retrying)\b", text, re.IGNORECASE)
        ):
            return True
        return False

    # Long apologies often end with a short "let me …" line in the final paragraph.
    # Scan the tail even when the full message exceeds the short-reply cap.
    tail = _noop_tail_for_action_scan(text)
    tail_retry = bool(_NOOP_RETRY_PHRASE_RE.search(tail)) or has_retry_phrase
    if (tail != text or len(text) > 260) and _noop_action_promise_chunk(
        tail, has_retry_phrase=tail_retry
    ):
        return True

    # Short single-chunk replies (e.g. "Let me check both files:").
    if len(text) > 260:
        return False
    return _noop_action_promise_chunk(text, has_retry_phrase=has_retry_phrase)


def _is_text_only_search_sessions_shim(content: str) -> bool:
    """True when the model typed a search_sessions(...) line as plain text (no tool_calls)."""
    if not isinstance(content, str):
        return False
    text = content.strip()
    if not text or len(text) > 360:
        return False
    if _TEXT_ONLY_SEARCH_SESSIONS_SHIM_RE.search(text):
        return True
    if len(text) <= 240 and re.search(r"\bsearch_sessions\s*\(", text, re.IGNORECASE):
        return True
    return False


def _last_user_message_text(messages: list[dict]) -> str:
    for msg in reversed(messages or []):
        if str(msg.get("role") or "") == "user":
            return str(msg.get("content") or "")
    return ""


def _is_fabricated_tool_output(content: str, messages: list[dict]) -> bool:
    """Detect assistant echoing expected tool stdout without calling the tool."""
    if not isinstance(content, str):
        return False
    text = content.strip()
    if not text or len(text) > 160:
        return False
    user = _last_user_message_text(messages)
    if not user:
        return False
    if re.search(r"\bshell_execute\b", user, re.IGNORECASE):
        echo_m = re.search(
            r"""echo\s+['"]?([^'"\s;]+)['"]?""",
            user,
            re.IGNORECASE,
        )
        if echo_m and text == echo_m.group(1):
            return True
    return False


# --- Zero-tool substantive reply (announce-without-execute expansion) ---

_MANDATABLE_TOOL_NAMES = frozenset({
    "read_file", "write_file", "edit_file", "list_directory", "shell_execute",
    "search_web", "read_webpage", "search_memory", "append_to_memory",
    "kanban_create_batch", "kanban_list_tasks", "kanban_create_task",
    "kanban_show_task", "kanban_list_runs", "find_tools", "search_sessions",
    "query_knowledge", "ask_studied_corpus", "list_knowledge_corpora",
    "patch_user_model", "propose_user_model_update", "list_pending_proposals",
    "search_vault_notes", "create_vault_note", "get_note", "append_to_vault_note",
    "capture_to_inbox", "list_recent_vault_notes", "search_filesystem",
    "update_task_plan", "hyperframes_doctor", "get_clipboard_text",
    "get_upcoming_calendar_events",
})

_USER_ACT_NOW_RE = re.compile(
    r"(?i)\b(?:reliability test|tools?\s+only|do\s+not\s+guess|tool\s+call\s+required|"
    r"must\s+call|you\s+must\s+call|call\s+the\s+tool|no\s+plan\s+—\s+call)\b",
)

_USER_MUST_REFRESH_TOOL_RE = re.compile(
    r"(?i)\b(?:you just created|previous turn|prior (?:create|turn)|from your (?:prior|previous)|"
    r"on (?:that|the) (?:id|note id|task id)|id from your prior|that note id|"
    r"already created(?: successfully)? in the previous)\b",
)

_PRIOR_TURN_STATUS_REPLY_RE = re.compile(
    r"(?i)\b(?:already created|was already created|from the previous turn|"
    r"the id is|id is [`']?kt_|created successfully in the previous)\b",
)

_FABRICATED_NO_TASK_REPLY_RE = re.compile(
    r"(?i)\b(?:no task (?:was )?created|no task exists|isn't available in my context|"
    r"provide the task id|failed creation|prior .+ failed|couldn't find.*task id|"
    r"don't have the id from)\b",
)

_TASK_PLAN_FABRICATED_REPLY_RE = re.compile(
    r"(?i)\b(?:conducts?|task\s+plan|plan\s+steps?|steps?\s+(?:to|for)\b|outline\s+steps?)",
)

_STATUS_ONLY_REPLY_RE = re.compile(
    r"^(?:PROPOSED|PATCH_OK|YES|NO|FOUND|NOT_FOUND|EMPTY|todo|CHAIN_OK|PLAN_OK|"
    r"CAPTURED|ANSWERED|NO_CORPUS|KB_HIT|KB_MISS|MEMORY_HIT|MEMORY_MISS|"
    r"COUNT=\d+|LINES=\d+|SESSIONS=\d+|READY_COUNT=\d+|RUNS=\d+|"
    r"COUNT|GOOD|READ_OK)\.?$",
    re.IGNORECASE,
)

_KANBAN_STATUS_FIELD_REPLY_RE = re.compile(
    r"(?i)^status\s*[=:]\s*[a-z_]+(?:\.|$)",
)

_KANBAN_FABRICATED_TASK_JSON_RE = re.compile(
    r'\{[^{}]{0,400}["\']status["\']\s*:\s*["\'](?:todo|ready|queued|in_progress|done)',
    re.IGNORECASE,
)

_FABRICATED_SEARCH_RESULT_RE = re.compile(
    r"(?i)\b(?:here(?:'|’)s what i found|titles?\s+only|bullet titles|http://|https://)\b",
)

_DEFERRAL_REPLY_RE = re.compile(
    r"(?i)^(?:i (?:can'?t|cannot|don'?t|won'?t)|unable to|would you like|"
    r"do you want me to|please confirm|i need (?:you to|your|a)|"
    r"that file does not exist|is not present|isn'?t (?:in|available))",
)


def _has_tool_activity_since_last_user(messages: list[dict]) -> bool:
    """True when assistant issued tool_calls or tool results after the latest user turn."""
    return bool(_tools_called_since_last_user(messages))


def _tools_called_since_last_user(messages: list[dict]) -> set[str]:
    """Tool names executed since the latest user message (inclusive of that turn)."""
    names: set[str] = set()
    collecting = False
    for msg in messages or []:
        role = str(msg.get("role") or "")
        if role == "user":
            names.clear()
            collecting = True
            continue
        if not collecting:
            continue
        if role == "assistant":
            for tc in msg.get("tool_calls") or []:
                if not isinstance(tc, dict):
                    continue
                fn = tc.get("function") or {}
                nm = str(fn.get("name") or tc.get("name") or "").strip()
                if nm:
                    names.add(nm)
    return names


def _user_forbidden_tool_names(user: str) -> set[str]:
    """Tools the user explicitly forbade this turn (do NOT use X)."""
    if not user:
        return set()
    forbidden: set[str] = set()
    for name in _MANDATABLE_TOOL_NAMES:
        if re.search(
            rf"(?i)\b(?:do\s+not|don'?t)\s+(?:use\s+)?[`']?{re.escape(name)}[`']?",
            user,
        ):
            forbidden.add(name)
            continue
        # "do NOT patch_user_model or append_to_memory …"
        if re.search(
            rf"(?i)\b(?:do\s+not|don'?t)\b[^.\n]{{0,120}}\bor\s+[`']?{re.escape(name)}[`']?\b",
            user,
        ):
            forbidden.add(name)
    if re.search(r"(?i)\bkanban_create_task\b", user) and not re.search(
        r"(?i)\bkanban_create_batch\b", user
    ):
        forbidden.add("kanban_create_batch")
    return forbidden


def _missing_mandated_tools(messages: list[dict]) -> set[str]:
    """Mandated tool names not yet called since the latest user turn."""
    user = _last_user_message_text(messages)
    mandated = _mandated_tool_names_from_user(user)
    if not mandated:
        return set()
    return mandated - _tools_called_since_last_user(messages)


_FALSE_KERNEL_TOOL_DENIAL_RE = re.compile(
    r"(?i)(?:don'?t|do not)\s+have(?:\s+a)?\s+`?(\w+)`?\s+tool|"
    r"`?(\w+)`?\s+(?:is\s+)?not\s+(?:a\s+)?(?:tool|available)",
)


def _kernel_always_on_tool_names() -> frozenset[str]:
    try:
        from core.tools import KERNEL_TOOL_NAMES

        return frozenset(KERNEL_TOOL_NAMES)
    except Exception:
        return frozenset({"read_file", "list_directory", "find_tools"})


def _is_false_kernel_tool_denial(content: str, messages: list[dict]) -> bool:
    """Model claims a kernel tool is unavailable when the user required it."""
    missing = _missing_mandated_tools(messages)
    kernel_missing = missing & _kernel_always_on_tool_names()
    if not kernel_missing:
        return False
    text = (content or "").strip()
    if not text:
        return False
    for name in kernel_missing:
        if re.search(rf"(?i)\b{re.escape(name)}\b.*(?:not available|no tool|don'?t have)", text):
            return True
        if re.search(rf"(?i)(?:don'?t|do not) have.*\b{re.escape(name)}\b", text):
            return True
    return bool(_FALSE_KERNEL_TOOL_DENIAL_RE.search(text))


def _mandated_tool_names_from_user(user: str) -> set[str]:
    """Tool names the latest user message explicitly requires this turn."""
    if not user:
        return set()
    forbidden = _user_forbidden_tool_names(user)
    lower = user.lower()
    mandated: set[str] = set()
    tools_only = bool(re.search(r"(?i)\btools?\s+only\b", user))
    for name in _MANDATABLE_TOOL_NAMES:
        if name in forbidden:
            continue
        if name not in lower:
            continue
        if re.search(
            rf"(?i)\b(?:call|use|must\s+call|you\s+must\s+call|then\s+call)\s+[`']?{re.escape(name)}[`']?",
            user,
        ):
            mandated.add(name)
            continue
        if re.search(
            rf"(?i)\b{re.escape(name)}\b\s+(?:once|first|only|max_results|limit|query|question|path|command|on|with|title)",
            user,
        ):
            mandated.add(name)
            continue
        if re.search(rf"(?i)\b{re.escape(name)}\b\s+\S", user):
            mandated.add(name)
            continue
        if re.search(rf"(?i)^\s*{re.escape(name)}\b", user.strip()):
            mandated.add(name)
            continue
        if re.search(rf"(?i)\b(?:must|required)\b.*\b{re.escape(name)}\b", user):
            mandated.add(name)
            continue
        if tools_only:
            mandated.add(name)
    return mandated


def _is_deferral_reply(text: str) -> bool:
    stripped = (text or "").strip()
    if not stripped:
        return True
    if stripped.endswith("?") and len(stripped) < 320:
        return True
    return bool(_DEFERRAL_REPLY_RE.search(stripped))


def _is_forbidden_tool_substitution(messages: list[dict]) -> bool:
    """User forbade a tool but the model called it this turn."""
    user = _last_user_message_text(messages)
    forbidden = _user_forbidden_tool_names(user)
    if not forbidden:
        return False
    return bool(_tools_called_since_last_user(messages) & forbidden)


def _is_mandated_tool_mismatch_reply(content: str, messages: list[dict]) -> bool:
    """Required tools missing after wrong-tool or status-token replies this turn."""
    if _is_forbidden_tool_substitution(messages):
        return True
    if _is_false_kernel_tool_denial(content, messages):
        return True
    missing = _missing_mandated_tools(messages)
    if not missing:
        return False
    called = _tools_called_since_last_user(messages)
    user = _last_user_message_text(messages)
    text = (content or "").strip()
    if not text:
        return False
    if _is_deferral_reply(text) and not (called & _user_forbidden_tool_names(user)):
        return False
    if _is_noop_action_promise(text, messages):
        return False

    if not called:
        return False

    if _STATUS_ONLY_REPLY_RE.match(text):
        return True
    if _KANBAN_STATUS_FIELD_REPLY_RE.match(text):
        return True

    user_mandated = _mandated_tool_names_from_user(user)
    if missing and called and user_mandated and not (called & user_mandated):
        return True

    if missing & {"kanban_list_runs"} and "kanban_list_runs" not in called:
        if re.search(r"(?i)\bRUNS=\d+\b", text) or (
            len(text) > 80
            and re.search(r"(?i)\b(?:completed|lifecycle|operational)\b", text)
        ):
            return True

    if missing & {"kanban_show_task"} and "kanban_show_task" not in called:
        if _KANBAN_STATUS_FIELD_REPLY_RE.match(text) or _KANBAN_FABRICATED_TASK_JSON_RE.search(
            text
        ):
            return True

    read_missing = missing & {"read_file", "list_directory"}
    if read_missing and len(text) < 240:
        return True

    um_missing = missing & {"patch_user_model", "propose_user_model_update"}
    if um_missing:
        return True

    return False


def _is_zero_tool_substantive_reply(content: str, messages: list[dict]) -> bool:
    """Detect answers/status tokens that imply tools ran when none did this turn."""
    if not isinstance(content, str):
        return False
    if _has_tool_activity_since_last_user(messages):
        return False
    user = _last_user_message_text(messages)
    if not user:
        return False
    mandated = _mandated_tool_names_from_user(user)
    act_now = bool(_USER_ACT_NOW_RE.search(user))
    if not mandated and not act_now:
        return False
    text = content.strip()
    if not text or _is_deferral_reply(text):
        return False
    if _is_noop_action_promise(text, messages):
        return False

    if _STATUS_ONLY_REPLY_RE.match(text):
        return True
    if _KANBAN_STATUS_FIELD_REPLY_RE.match(text):
        return True

    kanban_show_mandated = mandated & {"kanban_show_task"}
    if kanban_show_mandated and re.search(r"(?i)\b(?:status|queued|ready|in_progress|done)\b", text):
        if len(text) < 120:
            return True

    kanban_list_mandated = mandated & {"kanban_list_tasks"}
    if kanban_list_mandated and re.match(r"(?i)^(?:COUNT|READY)=\d*\.?$", text):
        return True
    if kanban_list_mandated and re.match(r"^\d+\.?$", text):
        if re.search(r"(?i)\bCOUNT\b", user):
            return True

    kanban_runs_mandated = mandated & {"kanban_list_runs"}
    if kanban_runs_mandated and re.match(r"(?i)^RUNS=\d+\.?$", text):
        return True
    if kanban_runs_mandated and re.search(
        r"(?i)\b(?:completed successfully|full lifecycle|board state|kanban system)\b",
        text,
    ):
        return True

    if kanban_show_mandated and _KANBAN_FABRICATED_TASK_JSON_RE.search(text):
        return True

    kanban_create_mandated = mandated & _KANBAN_CREATE_TOOL_NAMES
    if kanban_create_mandated and len(text) <= 4:
        return True

    if _USER_MUST_REFRESH_TOOL_RE.search(user) and mandated:
        if _PRIOR_TURN_STATUS_REPLY_RE.search(text):
            return True
        if _FABRICATED_NO_TASK_REPLY_RE.search(text):
            return True
        if re.match(r"(?i)^RUNS=0\.?$", text) and "kanban_list_runs" in mandated:
            return True
        if _KANBAN_TASK_ID_RE.search(text) and mandated & {
            "kanban_show_task",
            "kanban_list_runs",
            "get_note",
            "append_to_vault_note",
        }:
            return True

    task_plan_mandated = mandated & {"find_tools", "update_task_plan"}
    if task_plan_mandated and "update_task_plan" in mandated:
        if _TASK_PLAN_FABRICATED_REPLY_RE.search(text) and len(text) < 200:
            return True
        if text.lower().startswith("conduct"):
            return True

    search_mandated = mandated & {"search_web", "read_webpage"}
    if search_mandated and (
        _FABRICATED_SEARCH_RESULT_RE.search(text) or re.search(r"https?://", text)
    ):
        return True

    corpus_mandated = mandated & {"query_knowledge", "ask_studied_corpus"}
    if corpus_mandated and len(text) < 120:
        return True

    um_mandated = mandated & {"patch_user_model", "propose_user_model_update"}
    if um_mandated and re.match(r"^(?:PROPOSED|PATCH_OK)\.?$", text, re.IGNORECASE):
        return True

    read_mandated = mandated & {"read_file", "list_directory"}
    if read_mandated:
        if text.startswith("# ") or "**Name:**" in text or text.startswith("- **"):
            return True
        if act_now or re.search(r"(?i)\breply\s+(?:YES|first line|line count|LINES=)", user):
            if len(text) < 240:
                return True

    vault_mandated = mandated & {"search_vault_notes", "get_note", "create_vault_note"}
    if vault_mandated and len(text) < 200 and not text.endswith("?"):
        return True

    if act_now and len(text) > 20:
        return True

    if re.match(r"^[👻✨\s]+.{20,}", text):
        return True

    return False


_KANBAN_TASK_ID_RE = re.compile(r"\bkt_[a-f0-9]{8,}\b", re.IGNORECASE)
_KANBAN_SHORT_TASK_ID_RE = re.compile(r"\bkt_[a-f0-9]{1,7}\b", re.IGNORECASE)
_FABRICATED_TOOL_CALL_JSON_RE = re.compile(
    r'\{\s*"(?:name|tool_name)"\s*:\s*"(?:kanban_|read_|search_|list_|find_)',
    re.IGNORECASE,
)
_KANBAN_CREATE_TOOL_NAMES = frozenset({"kanban_create_task", "kanban_create_batch"})
_USER_EXPECTS_KANBAN_CREATE_RE = re.compile(
    r"(?:"
    r"\bkanban_create_(?:task|batch)\b|"
    r"\bcreate(?:d)?\s+(?:a\s+)?(?:new\s+)?(?:kanban\s+)?task\b|"
    r"\b(?:new|created)\s+task\s+id\b|"
    r"\breply\s+with\s+(?:the\s+)?(?:new\s+)?task\s+id\b|"
    r"\bkt_\.\.\.|"
    r"\bkt_…"
    r")",
    re.IGNORECASE,
)


def _tool_message_name_and_body(msg: dict) -> tuple[str, str]:
    """Parse kernel/SSE tool rows that may be JSON carriers."""
    content = str(msg.get("content") or "")
    name = ""
    try:
        obj = json.loads(content)
    except (json.JSONDecodeError, TypeError):
        return name, content
    if isinstance(obj, dict):
        name = str(obj.get("tool_name") or "").strip()
        preview = obj.get("result_preview")
        if preview is not None:
            content = str(preview)
    return name, content


def _kanban_ids_from_create_tool_results(
    messages: list[dict],
    *,
    lookback: int = 32,
) -> set[str]:
    ids: set[str] = set()
    for msg in list(messages or [])[-max(1, lookback) :]:
        if str(msg.get("role") or "") != "tool":
            continue
        name, body = _tool_message_name_and_body(msg)
        if name and name not in _KANBAN_CREATE_TOOL_NAMES:
            continue
        if name in _KANBAN_CREATE_TOOL_NAMES or re.search(
            r"Created(?: \d+)? kanban task", body, re.IGNORECASE
        ):
            ids.update(_KANBAN_TASK_ID_RE.findall(body))
    return {x.lower() for x in ids}


def _is_placeholder_kanban_task_id(task_id: str) -> bool:
    m = re.match(r"kt_([a-f0-9]+)$", (task_id or "").strip(), re.IGNORECASE)
    if not m:
        return False
    hexpart = m.group(1).lower()
    if len(hexpart) < 8:
        return True
    return hexpart.replace("0", "") == ""


def _user_expects_kanban_create(messages: list[dict]) -> bool:
    user = _last_user_message_text(messages)
    if not user:
        return False
    return bool(_USER_EXPECTS_KANBAN_CREATE_RE.search(user))


def _is_fabricated_tool_call_json(content: str, messages: list[dict]) -> bool:
    """Detect assistant emitting tool-call-shaped JSON in content instead of native tools."""
    if not isinstance(content, str):
        return False
    text = content.strip()
    if not text or not _FABRICATED_TOOL_CALL_JSON_RE.search(text):
        return False
    if _has_tool_activity_since_last_user(messages):
        return False
    user = _last_user_message_text(messages)
    if not user:
        return False
    mandated = _mandated_tool_names_from_user(user)
    if not mandated:
        return False
    for name in mandated:
        if name in text:
            return True
    return bool(_FABRICATED_TOOL_CALL_JSON_RE.search(text))


def _is_fabricated_kanban_task_id(content: str, messages: list[dict]) -> bool:
    """Detect assistant inventing kt_* ids without a create_task/create_batch tool result."""
    if not isinstance(content, str):
        return False
    text = content.strip()
    if not text or not messages:
        return False
    if not _user_expects_kanban_create(messages):
        return False
    if _KANBAN_SHORT_TASK_ID_RE.search(text):
        return True
    claimed = _KANBAN_TASK_ID_RE.findall(text)
    if not claimed:
        return False
    created = _kanban_ids_from_create_tool_results(messages)
    for cid in claimed:
        if _is_placeholder_kanban_task_id(cid):
            return True
        if cid.lower() not in created:
            return True
    return False


def _required_tool_fields(tool_obj: Any) -> list[str]:
    """Return required argument names for a tool object."""
    try:
        schema = getattr(tool_obj, "args_schema", None)
        if schema is None:
            return []
        if hasattr(schema, "model_json_schema"):
            js = schema.model_json_schema()
        elif hasattr(schema, "schema"):
            js = schema.schema()
        else:
            return []
        required = js.get("required") or []
        return [r for r in required if isinstance(r, str)]
    except Exception:
        return []


def _missing_required_tool_args(tool_obj: Any, args: dict[str, Any]) -> list[str]:
    """Return missing required args for *tool_obj* given parsed *args* dict."""
    missing: list[str] = []
    required = _required_tool_fields(tool_obj)
    for field in required:
        val = args.get(field)
        if val is None:
            missing.append(field)
        elif isinstance(val, str) and not val.strip():
            missing.append(field)
    return missing


def _preflight_drop_tool_calls_missing_required(
    tool_calls_list: list[dict[str, Any]],
    tools_by_name: dict[str, Any],
    full_tools_by_name: dict[str, Any],
    on_log: Any | None = None,
    thread_id: str | None = None,
    round_: int | None = None,
) -> list[dict[str, Any]]:
    """Drop parallel tool calls that fail required-arg validation when at least one
    other call in the same batch is valid.

    Local / Hermes-style models often emit multiple XML tool calls where some have
    empty `{}` arguments. Previously the whole batch was rejected, burning retries
    even when other calls were executable.
    """
    if not isinstance(tool_calls_list, list) or len(tool_calls_list) < 2:
        return tool_calls_list

    from tool_arg_normalizer import normalize_tool_args

    classified: list[tuple[dict[str, Any], bool]] = []
    for tc in tool_calls_list:
        if not isinstance(tc, dict):
            classified.append((tc, True))
            continue
        name = tc.get("name") or ""
        tool_obj = tools_by_name.get(name) or full_tools_by_name.get(name)
        if tool_obj is None:
            classified.append((tc, True))
            continue
        try:
            parsed = _json_loads_lenient(tc.get("arguments") or "{}")
        except Exception:
            parsed = {}
        if not isinstance(parsed, dict):
            parsed = {}
        for_check = dict(parsed)
        try:
            normalize_tool_args(name, for_check)
        except Exception:
            _log.debug("preflight normalize_tool_args failed for %s", name, exc_info=True)
        missing = _missing_required_tool_args(tool_obj, for_check)
        classified.append((tc, not missing))

    kept = [tc for tc, ok in classified if ok]
    dropped = [tc for tc, ok in classified if not ok]
    if not dropped or not kept:
        return tool_calls_list
    if on_log:
        on_log(
            "AGENT",
            f"Dropped {len(dropped)} parallel tool call(s) missing required args; "
            f"executing {len(kept)} valid call(s).",
        )
    if is_trace_enabled() and thread_id and round_:
        try:
            _dropped_names = [
                (tc.get("name") if isinstance(tc, dict) else "") or ""
                for tc in dropped
            ]
            emit_guardrail_event(
                thread_id,
                int(round_),
                "preflight_parallel_drop",
                "missing_required_args",
                details={"dropped_count": len(dropped), "kept_count": len(kept), "dropped_tools": _dropped_names[:20]},
            )
        except Exception:
            _log.debug("Failed to emit preflight parallel-drop trace event", exc_info=True)
    return kept


def _is_read_chunk_only_turn(tool_calls_list: list[dict]) -> bool:
    """True when every planned tool call in this turn is read_file_chunk."""
    if not isinstance(tool_calls_list, list) or not tool_calls_list:
        return False
    for tc in tool_calls_list:
        if not isinstance(tc, dict):
            return False
        if (tc.get("name") or "") != "read_file_chunk":
            return False
    return True


def _read_chunk_turn_signatures(tool_calls_list: list[dict]) -> set[tuple[str, str, str]] | None:
    """Per-call (file, chunk, span) signatures for a read_file_chunk-only turn.

    Returns ``None`` when the turn is NOT read_file_chunk-only. Otherwise returns
    a set of ``(file_path, chunk_index, num_chunks)`` signatures so the caller
    can distinguish *progress* (reading a chunk not seen before) from a genuine
    *stall* (re-reading the same chunks). Sequentially reading a 60-chunk file
    is the expected behavior for research/review/synthesis tasks and must not
    be treated as a loop (incident kt_865237ad896842: a kanban design-review
    worker was hard-stopped mid-read by the chunk-loop guard and the run was
    salvaged as blocked).

    Unparseable arguments yield a sentinel signature that always counts as a
    repeat, so malformed call loops still trip the guard.
    """
    if not _is_read_chunk_only_turn(tool_calls_list):
        return None
    sigs: set[tuple[str, str, str]] = set()
    for tc in tool_calls_list:
        args = tc.get("arguments")
        if isinstance(args, str):
            try:
                args = json.loads(args)
            except (ValueError, TypeError):
                args = None
        if not isinstance(args, dict):
            args = tc.get("args") if isinstance(tc.get("args"), dict) else None
        if not isinstance(args, dict):
            sigs.add(("", "", ""))
            continue
        path = str(args.get("file_path") or args.get("filepath") or "").strip().lower()
        chunk = str(args.get("chunk_index", "")).strip()
        span = str(args.get("num_chunks", "") or "1").strip()
        sigs.add((path, chunk, span))
    return sigs


def _should_append_invalid_required_retry(messages: list[dict], expected_content: str) -> bool:
    """Avoid stacking identical synthetic retry nags back-to-back."""
    if not isinstance(messages, list) or not messages:
        return True
    last = messages[-1]
    if not isinstance(last, dict):
        return True
    if last.get("role") != "user":
        return True
    return (last.get("content") or "") != expected_content


def _format_invalid_required_retry_prompt(
    tool_name: str,
    missing: list[str],
    provided_keys: list[str],
    normalized_keys: list[str],
    raw_args: dict[str, Any] | None = None,
) -> str:
    """Build actionable retry guidance for missing required tool arguments."""
    provided_hint = ", ".join(provided_keys) if provided_keys else "(none)"
    normalized_hint = ", ".join(normalized_keys) if normalized_keys else "(none)"
    missing_set = {m.strip() for m in (missing or []) if isinstance(m, str)}
    extra_hints: list[str] = []
    if tool_name == "skill_manage":
        action = ""
        if isinstance(raw_args, dict):
            value = raw_args.get("action")
            if isinstance(value, str):
                action = value.strip().lower()
        if action == "list":
            extra_hints.append("For listing skills, call `skills_list` instead of `skill_manage(action='list')`.")
    if tool_name == "append_to_memory" and "entry" in missing_set:
        extra_hints.append(
            "For `append_to_memory`, include the full note in JSON `arguments.entry`; do not leave it only in prose or fake tags."
        )
    if tool_name == "append_working_buffer" and ({"human_summary", "agent_summary"} & missing_set):
        extra_hints.append(
            "For `append_working_buffer`, provide both `human_summary` and `agent_summary` as strings."
        )
    if tool_name == "write_session_state" and ({"key", "value"} & missing_set):
        extra_hints.append(
            "For `write_session_state`, include both `key` and `value`."
        )
    if tool_name == "read_file" and "filepath" in missing_set:
        extra_hints.append(
            "For `read_file`, pass a real filesystem path in `filepath`."
        )
    if tool_name == "read_file_chunk" and ({"file_path", "chunk_index"} & missing_set):
        extra_hints.append(
            "For `read_file_chunk`, use `file_path` (not `filepath`) and include `chunk_index` (0-based)."
        )
    if tool_name == "write_text_file" and ({"file_path", "content"} & missing_set):
        extra_hints.append(
            "For `write_text_file`, include BOTH `file_path` and `content` in arguments JSON."
        )
    if tool_name in {"write_text_file", "save_to_file", "append_to_text_file"} and "content" in missing_set:
        if "text" in provided_hint or "text" in normalized_hint:
            extra_hints.append(
                f"For `{tool_name}`, the file body belongs in JSON key `content`, not `text`. "
                "(`append_to_vault_note` uses `text`; file tools such as `write_text_file` use `content`.)"
            )
    if tool_name == "append_delegation_scratch" and "content" in missing_set:
        if "text" in provided_hint or "text" in normalized_hint:
            extra_hints.append(
                f"For `{tool_name}`, put the markdown to append in JSON key `content`, not `text`."
            )
    if tool_name == "append_to_memory" and "entry" in missing_set:
        if "content" in provided_hint or "text" in provided_hint:
            extra_hints.append(
                "For `append_to_memory`, the note text belongs in JSON key `entry` (not `content` or `text`)."
            )
    if tool_name == "write_session_state" and "value" in missing_set:
        if "content" in provided_hint or "text" in provided_hint:
            extra_hints.append(
                "For `write_session_state`, the persisted detail belongs in JSON key `value` (with `key` for the label)."
            )
    if tool_name == "set_clipboard_text" and "text" in missing_set:
        if "content" in provided_hint:
            extra_hints.append(
                "For `set_clipboard_text`, the string to copy belongs in JSON key `text`, not `content`."
            )
    if tool_name in {"save_research_findings", "generate_final_research_report"} and "findings" in missing_set:
        if "content" in provided_hint or "text" in provided_hint:
            extra_hints.append(
                f"For `{tool_name}`, the findings prose belongs in JSON key `findings`, not `content`/`text`."
            )
    if tool_name == "submit_ocr_chunk_corrections" and "corrected_content" in missing_set:
        if "content" in provided_hint or "text" in provided_hint:
            extra_hints.append(
                "For `submit_ocr_chunk_corrections`, the replacement chunk belongs in JSON key `corrected_content`."
            )
    if tool_name == "correct_ocr_text" and "text" in missing_set:
        if "content" in provided_hint:
            extra_hints.append(
                "For `correct_ocr_text`, the OCR text to correct belongs in JSON key `text`, not `content`."
            )
    if tool_name == "compose_email_draft" and "body" in missing_set:
        if "content" in provided_hint or "text" in provided_hint:
            extra_hints.append(
                "For `compose_email_draft`, the message body belongs in JSON key `body` (along with `to` and optional `subject`)."
            )
    if tool_name == "delegate_task" and "task_description" in missing_set:
        extra_hints.append(
            "For `delegate_task`, provide a non-empty string `task_description` for a single delegation, "
            "OR pass `tasks` as a JSON array of objects where **each** object has its own `task_description`. "
            "If you use batch `tasks`, do not leave `task_description` empty unless every batch item includes one."
        )
    if tool_name in {
        "extract_pdf_text",
        "ocr_scanned_pdf",
        "ocr_image_text",
        "ocr_complex_document",
        "ocr_with_kraken_api",
    } and "file_path" in missing_set:
        if "filepath" in provided_hint or "filepath" in normalized_hint:
            extra_hints.append(
                f"For `{tool_name}`, use JSON key `file_path` (not `filepath`) for the PDF or image path. "
                "Include the full filename, including spaces (e.g. `.../Ch 3 - 5 SCAN.pdf`)."
            )
    if tool_name == "append_research_scratch" and missing_set:
        extra_hints.append(
            "For `append_research_scratch`, use one tool call whose JSON arguments include "
            "**both** `topic` and `content` (full paragraph text). Do not send topic-only or content-only calls."
        )
    if tool_name in {"deep_web_search", "search_web", "search_web_public"} and "query" in missing_set:
        extra_hints.append(
            f"For `{tool_name}`, you MUST include `query` (a non-empty search string) in the JSON arguments."
        )
    if tool_name == "plan_research_subquestions" and "topic" in missing_set:
        extra_hints.append(
            "For `plan_research_subquestions`, you MUST include `topic` (a non-empty string) in the JSON arguments."
        )
    if tool_name == "read_and_analyze_page" and "url" in missing_set:
        extra_hints.append(
            "For `read_and_analyze_page`, you MUST include `url` (a full URL string) in the JSON arguments."
        )
    if tool_name == "run_python_code" and "code" in missing_set:
        if "script" in provided_hint or "snippet" in provided_hint or "source" in provided_hint:
            extra_hints.append(
                "For `run_python_code`, use JSON key `code` for the Python source (not `script`/`snippet`/`source`). "
            )
        extra_hints.append(
            "For `run_python_code`, you MUST include `code` (a Python code string) in the JSON arguments. "
            "Place the entire script inside the `code` value."
        )
    extra_hint = f" {' '.join(extra_hints)}" if extra_hints else ""
    return (
        f"Your previous tool call was invalid: `{tool_name}` is missing required arguments: {', '.join(missing)}. "
        f"Provided argument keys: {provided_hint}. "
        f"After normalization, keys were: {normalized_hint}."
        f"{extra_hint} "
        + GUARD_TOOL_CALL_SUFFIX
    )


_HEADING_RE = re.compile(r"^#+\s+(.+)", re.MULTILINE)


def _derive_topic_from_content(content: str) -> str:
    """Best-effort topic derivation from research content text."""
    if not content:
        return ""
    m = _HEADING_RE.search(content)
    if m:
        heading = m.group(1).strip().rstrip("#").strip()
        if len(heading) > 3:
            return heading[:120]
    for line in content.split("\n"):
        line = line.strip()
        if line and not line.startswith("<") and len(line) > 5:
            return line[:120]
    return ""


def _auto_fill_scratch_args(
    parsed_args: dict[str, Any],
    accumulated: dict[str, Any],
) -> tuple[dict[str, Any], bool]:
    """Fill missing ``topic`` or ``content`` for ``append_research_scratch``.

    Uses topic derivation from content and/or previously accumulated partial args
    from earlier retry rounds.  Returns (args, mutated) where *mutated* is True if
    the args dict was modified.
    """
    mutated = False
    for key in ("topic", "content"):
        if not parsed_args.get(key) and accumulated.get(key):
            parsed_args[key] = accumulated[key]
            mutated = True

    topic_val = str(parsed_args.get("topic") or "").strip()
    content_val = str(parsed_args.get("content") or "").strip()
    if topic_val and not content_val and len(topic_val) > 120:
        if ". " in topic_val or "\n" in topic_val or topic_val.count(" ") > 15:
            parsed_args["content"] = topic_val
            derived = _derive_topic_from_content(topic_val)
            if derived:
                parsed_args["topic"] = derived
            mutated = True

    if parsed_args.get("content") and not parsed_args.get("topic"):
        topic = _derive_topic_from_content(str(parsed_args["content"]))
        if topic:
            parsed_args["topic"] = topic
            mutated = True

    return parsed_args, mutated


def _recover_scratch_content_from_messages(
    messages: list[dict[str, Any]],
    *,
    topic: str = "",
    max_chars: int = 6000,
) -> str:
    """Synthesize ``content`` for append_research_scratch from recent research tool output."""
    from utils.path_normalize import is_internal_guard_user_text

    chunks: list[str] = []
    chars = 0
    for msg in reversed(messages or []):
        role = msg.get("role")
        content = msg.get("content")
        if role == "tool" and isinstance(content, str) and content.strip():
            chunk = content.strip()
        elif role == "assistant" and isinstance(content, str):
            continue
        elif role == "user" and isinstance(content, str):
            if is_internal_guard_user_text(content):
                continue
            continue
        else:
            continue
        if len(chunk) < 80:
            continue
        if chars + len(chunk) > max_chars:
            chunk = chunk[: max(0, max_chars - chars)]
        if not chunk:
            break
        chunks.append(chunk)
        chars += len(chunk)
        if chars >= max_chars:
            break
    if not chunks:
        return ""
    body = "\n\n---\n\n".join(reversed(chunks))
    if topic:
        return f"Research synthesis for {topic}:\n\n{body}"[:max_chars]
    return body[:max_chars]


def _infer_scratch_topic_from_context(messages: list[dict[str, Any]]) -> str:
    """Topic string for ``append_research_scratch`` from worker brief or sticky kanban claim."""
    from utils.transcript_roles import is_real_user_message

    for msg in messages or []:
        if not is_real_user_message(msg):
            continue
        content = msg.get("content")
        candidate = ""
        if isinstance(content, str) and content.strip():
            candidate = content.strip()
        elif isinstance(content, list):
            for part in content:
                if isinstance(part, dict) and part.get("type") == "text":
                    t = part.get("text", "")
                    if isinstance(t, str) and t.strip():
                        candidate = t.strip()
                        break
        if not candidate or "Task:" not in candidate:
            continue
        task_m = re.search(r"Task:\s*(.+?)(?:\n|$)", candidate, re.IGNORECASE)
        if task_m:
            topic = task_m.group(1).strip()
            if topic:
                return topic[:120]
    try:
        from utils.kanban_dispatcher import current_kanban_worker_search_brief

        sticky = (current_kanban_worker_search_brief() or "").strip()
        if sticky:
            derived = _derive_topic_from_content(sticky)
            return (derived or sticky)[:120]
    except Exception:
        pass
    return ""


def _format_path_artifact_retry_prompt(
    tool_name: str,
    *,
    bad_path_fields: list[str] | None = None,
) -> str:
    """Build retry guidance when tool-call path args contain markup artifacts."""
    field_note = ""
    if bad_path_fields:
        field_note = f" Suspected bad path field(s): {', '.join(bad_path_fields)}."
    return (
        f"Your previous tool call for `{tool_name}` used path-like values that look like tool markup artifacts "
        "(for example `tool_call` / `tool_response`) instead of real filesystem paths."
        f"{field_note} Retry with real filesystem paths only.\n" + GUARD_TOOL_CALL_SUFFIX
    )


def _cap_tool_diag_text(s: str, max_len: int = 200) -> str:
    s = s.replace("\n", " ").replace("\r", " ").strip()
    if len(s) <= max_len:
        return s
    return s[: max_len - 1] + "…"


def _diagnose_single_tool_call_block(
    block: str,
    idx: int,
    recovery_context: str,
) -> str:
    """One-line reason a single payload block did not parse (for retry prompts)."""
    block = (block or "").strip()
    if not block:
        return "empty payload between tool_call tags"
    if _parse_xml_tool_call_block(block, idx, recovery_context=recovery_context) is not None:
        return "block parsed in isolation (unexpected; report as bug if this recurs)"

    s = block.lstrip()
    if s.startswith("{"):
        try:
            obj = _json_loads_lenient(block)
        except Exception as e:
            if _salvage_json_tool_call(block) is not None:
                return (
                    f"JSON error: {_cap_tool_diag_text(str(e), 160)} "
                    "(partial salvage not enough for a full tool call)"
                )
            return f"invalid JSON: {_cap_tool_diag_text(str(e), 180)}"
        if not isinstance(obj, dict):
            return "top-level JSON is not an object with name/arguments"
        name = str(obj.get("name") or obj.get("function") or "").strip()
        if not name:
            return "JSON object has no non-empty `name` or `function`"
        return "JSON has a name but did not form an executable call (check arguments and schema keys)"

    if re.search(r"<function=", block) or re.search(r"<invoke\s+name=", block) or re.search(
        r"<arg_key>", block, re.IGNORECASE
    ):
        return "XML tool_call layout detected but name/parameters did not parse cleanly"
    if re.search(r"call:\s*\w+", block, re.IGNORECASE):
        return "Gemma-style call: line did not parse (check trailing JSON braces/escapes)"
    return f"unrecognized layout (not JSON, function=, invoke, arg_key, or call:). Snippet: {_cap_tool_diag_text(block, 120)}"


def _diagnose_tool_call_parse_failure(response_buffer: str) -> str:
    """Short operator/model-facing reason XML/native fallback found markup but no executable calls."""
    if not isinstance(response_buffer, str) or not response_buffer.strip():
        return "empty assistant content where tool_call markup was expected"
    if not _TOOL_CALL_TAG_RE.search(response_buffer):
        return "no tool_call open tag in buffer (inconsistent state)"

    raw_matches = _TOOL_CALL_BLOCK_RE.findall(response_buffer)
    blocks = [g1 or g2 for g1, g2 in raw_matches] if raw_matches else []

    if not blocks:
        uncl = _extract_unclosed_tool_call_block(response_buffer)
        if uncl is not None:
            if not (uncl or "").strip():
                return "unclosed <tool_call> with empty payload (output likely truncated before closing tag)"
            if _parse_xml_tool_call_block(uncl, 0, recovery_context=response_buffer) is None:
                return "unclosed block: " + _diagnose_single_tool_call_block(
                    uncl, 0, response_buffer
                )
            return "unclosed block should have parsed (internal inconsistency)"
        return "tool_call open tag(s) but no well-formed <tool_call>…</tool_call> block matched"

    reasons: list[str] = []
    for i, b in enumerate(blocks[:3]):
        reasons.append(f"block {i + 1}: {_diagnose_single_tool_call_block(b, i, response_buffer)}")
    if len(blocks) > 3:
        reasons.append(f"+{len(blocks) - 3} more block(s) not shown")
    return "; ".join(reasons)


def _format_malformed_tool_call_retry_prompt(*, diagnostic: str | None = None) -> str:
    """Build one-shot retry guidance when a <tool_call> payload was malformed."""
    detail = (diagnostic or "").strip()
    if not detail:
        return (
            "Your previous response attempted a tool call but the payload was malformed and could not be executed. "
            "Retry immediately with one native structured tool call.\n"
            + NATIVE_TOOL_CALL_FORMAT_BULLETS
            + "For read tools, use exact keys (`read_file.filepath`, `read_file_chunk.file_path`, "
            + "`read_file_chunk.chunk_index`, `list_directory.dir_path`). "
            + "For writes, use real paths (`write_text_file.file_path`, `save_to_file.file_path`). "
            + "Include all required fields and minimize surrounding text."
        )
    return (
        f"Your previous tool call could not be executed ({detail}).\n\n"
        + GUARD_TOOL_CALL_MINIMAL
    )


def _format_noop_promise_retry_prompt() -> str:
    """Build retry guidance when assistant promised action but emitted no tool call."""
    return (
        "Your previous reply promised a follow-up step (including phrases like \"try again\" or "
        "\"one more time\") but did not execute any tool call. "
        "Issue the required tool call(s) now in this turn — do not end with intent-only prose.\n"
        + GUARD_TOOL_CALL_SUFFIX
    )


def _format_shell_policy_recovery_prompt() -> str:
    return (
        "Your recent attempt used `shell_execute`, but the host blocked it under policy or plan mode. "
        "Do not promise shell or terminal actions. Prefer allowed read tools (`read_webpage`, `read_file`, "
        "`find_tools`) or ask the user to run commands locally.\n"
        + GUARD_TOOL_CALL_SUFFIX
    )


def _format_text_only_search_sessions_shim_retry_prompt() -> str:
    return (
        "Your last message looked like a `search_sessions(...)` tool call written as plain text, "
        "but no tool was executed. Issue a real `search_sessions` tool call with JSON arguments "
        "(or answer without claiming you searched).\n"
        + GUARD_TOOL_CALL_SUFFIX
    )


def _format_fabricated_tool_output_retry_prompt() -> str:
    return (
        "Your last reply repeated the expected tool output as plain text without running the tool. "
        "Call the requested tool (e.g. `shell_execute`) now and base your answer on the real result.\n"
        + GUARD_TOOL_CALL_SUFFIX
    )


def _format_fabricated_kanban_task_id_retry_prompt() -> str:
    return (
        "Your last reply included a kanban task id (kt_…) without a successful "
        "`kanban_create_task` or `kanban_create_batch` tool result in this turn. "
        "Do not invent or reuse placeholder ids. Call the create tool now, then reply with "
        "the id from the tool output.\n"
        + GUARD_TOOL_CALL_SUFFIX
    )


def _format_mandated_tool_retry_prompt(messages: list[dict]) -> str:
    """Retry when required tools were skipped, forbidden tools were used, or kernel tools denied."""
    user = _last_user_message_text(messages)
    missing = sorted(_missing_mandated_tools(messages))
    forbidden = sorted(_user_forbidden_tool_names(user))
    called = sorted(_tools_called_since_last_user(messages))
    kernel_missing = [n for n in missing if n in _kernel_always_on_tool_names()]

    parts: list[str] = []
    if forbidden and called:
        bad = sorted(set(called) & set(forbidden))
        if bad:
            parts.append(
                f"You called forbidden tool(s): {', '.join(f'`{n}`' for n in bad)}. "
                f"Do not substitute — use the tool the user asked for."
            )
    if missing:
        parts.append(
            "Still required this turn: "
            + ", ".join(f"`{n}`" for n in missing[:6])
            + "."
        )
    if kernel_missing:
        parts.append(
            "Those kernel tools are always available in your schema "
            f"({', '.join(f'`{n}`' for n in kernel_missing[:4])}) — do not claim they are missing."
        )
    lead = " ".join(parts) if parts else (
        "Your last reply did not satisfy the user's tool requirements."
    )
    return (
        lead
        + " Do not invent status tokens (YES/PROPOSED/FOUND). "
        "Call the required tool(s) now, then reply using the real tool output.\n"
        + GUARD_TOOL_CALL_SUFFIX
    )


def _format_zero_tool_substantive_retry_prompt(messages: list[dict]) -> str:
    """Build retry guidance when the model answered without calling required tools."""
    return _format_mandated_tool_retry_prompt(messages)


def _format_task_plan_continuation_prompt(stored_plan: dict | None = None) -> str:
    """Neutral nudge when the agent tries to finalize with an incomplete plan.

    When *stored_plan* is provided and still has open steps, a compact
    snapshot of just those steps is appended so the model sees **which**
    work remains — important salience after summarization/trim may have
    evicted the system-prompt plan block.

    Mirrors the non-authoritative framing in ``format_plan_for_injection`` —
    plan step text is model- and user-derived, so this reminder avoids
    treating those strings as policy. Callers append this as a ``user`` role
    message; it is visible in chat history.
    """
    generic = (
        "The recorded task plan still has steps that are pending or in_progress. "
        "Keep working: call the tools needed for the next step, then update the plan with "
        "`update_task_plan` (set `status: done`, `skipped`, or `blocked`). "
        "If the task is complete or no longer applicable, call `clear_task_plan(confirm=True)` "
        "or mark the remaining steps terminal, then provide the final answer."
    )
    snapshot = ""
    if stored_plan is not None:
        try:
            from utils.task_plan import format_open_steps_for_continuation
            snapshot = format_open_steps_for_continuation(stored_plan) or ""
        except Exception:
            _log.debug(
                "task_plan continuation snapshot: formatter failed; using generic prompt",
                exc_info=True,
            )
            snapshot = ""
    if not snapshot:
        return generic
    return f"{generic}{snapshot}"


def _format_coordinator_board_open_handoff_prompt(stored_plan: dict | None = None) -> str:
    """Nudge when coordinator tries to finalize while Kanban workers still own open plan steps."""
    generic = (
        "Kanban coordinator mode is active and workers still own the open task-plan steps. "
        "Do NOT call delegate_task, file writes, shell, or read_delegation_scratch "
        "in this thread — Kanban workers execute those steps.\n"
        "It is OK for the recorded task plan to stay in_progress while workers run. "
        "Give the operator a one-paragraph status of what is in_progress on the board, "
        "then stop. You will resume automatically when Kanban workers deliver results.\n"
        "Do not keep polling kanban_show_task on the same task id."
    )
    snapshot = ""
    if stored_plan is not None:
        try:
            from utils.task_plan import format_open_steps_for_continuation

            snapshot = format_open_steps_for_continuation(stored_plan) or ""
        except Exception:
            _log.debug("coordinator handoff plan snapshot failed", exc_info=True)
    if not snapshot:
        return generic
    return f"{generic}{snapshot}"


def _task_plan_enforcement_enabled() -> bool:
    """Honor the operator kill-switch for task-plan continuation enforcement.

    Set ``MUGI_TASK_PLAN_ENFORCE_COMPLETION=0`` (or ``false``/``no``/``off``)
    to disable without a code change. Default is enabled.
    """
    val = os.environ.get("MUGI_TASK_PLAN_ENFORCE_COMPLETION", "").strip().lower()
    if not val:
        return True
    return val not in {"0", "false", "no", "off"}


def _completion_critic_env_enabled() -> bool:
    """Kill-switch for the optional utility completion critic (default on)."""
    val = os.environ.get("MUGI_COMPLETION_CRITIC", "").strip().lower()
    if not val:
        return True
    return val not in {"0", "false", "no", "off"}


# ---------------------------------------------------------------------------
# Tool-call loop detector
#
# Symptom (trace d844b076-*): the agent repeated `find_tools` and
# `analyze_multiple_sources` with slightly-different text while the task plan
# stayed stuck. Hashing argument *values* would miss that because each call
# varied the query/sources text; hashing argument *keys* catches it because
# the shape of the call (which tool + which arg names) was identical.
#
# We treat the detector as a *heads-up*, not a retry: we emit a single
# neutral nudge per offending signature per run so the model can escalate to
# a different tool, change inputs substantially, or close the related plan
# step via ``update_task_plan`` and move on. It does not consume the shared
# ``_consume_guard_retry_budget`` because it does not force a retry turn —
# the model responds in the next round as usual.
# ---------------------------------------------------------------------------

_TOOL_CALL_LOOP_DEFAULT_THRESHOLD = 4
_TOOL_CALL_LOOP_RECOVERY_THRESHOLD = _TOOL_CALL_LOOP_DEFAULT_THRESHOLD + 2
_TOOL_CALL_LOOP_HARD_STOP_THRESHOLD = _TOOL_CALL_LOOP_RECOVERY_THRESHOLD + 2

# Tools where same-signature repetition is expected/benign and should NOT
# count toward the loop detector. These are stateful appends, polling, or
# chunked streams where every call advances the world by design.
_TOOL_CALL_LOOP_EXEMPT_TOOLS: frozenset[str] = frozenset({
    "update_task_plan",
    "clear_task_plan",
    "read_file_chunk",
    "append_research_scratch",
    "append_delegation_scratch",
    "append_to_vault_note",
    "append_to_memory",
    "append_working_buffer",
    "append_to_text_file",
    "write_session_state",
    "check_background_task",
    "stop_background_task",
})

# Read/probe tools where repeated calls to different paths should not share one loop signature.
_PATH_BUCKET_LOOP_TOOLS: frozenset[str] = frozenset({
    "read_file",
    "list_directory",
})

# Write tools where the same arg shape with different payloads is normal
# iterative development (write → run → debug → rewrite). Bucket by target path
# plus a digest of the payload so only true repeats — writing the *identical*
# content again — count toward the loop detector. Motivating trace: 6be8ae2b,
# where seven materially different web-act scripts shared one signature and
# the run was hard-stopped mid-debugging.
_CONTENT_BUCKET_LOOP_TOOLS: frozenset[str] = frozenset({
    "write_web_act_file",
    "write_text_file",
    "save_to_file",
    "save_research_findings",
    "skill_manage",
})

# Run/execute tools bucketed by script path: re-running a *different* script is
# progress; re-running the same script repeatedly still counts toward a loop.
_RUN_PATH_BUCKET_LOOP_TOOLS: frozenset[str] = frozenset({
    "run_web_script",
})

# Paginated reads where the *same* arg shape walks distinct slices of a file.
# Bucket by path AND the slice index so paging through chunks 0,1,2,… reads as
# progress, while re-reading the SAME chunk over and over still trips the loop.
_PAGINATED_READ_LOOP_TOOLS: frozenset[str] = frozenset({
    "read_file_chunk",
})
_PAGINATED_READ_PATH_KEYS: tuple[str, ...] = ("file_path", "filepath", "path")
_PAGINATED_READ_INDEX_KEYS: tuple[str, ...] = ("chunk_index", "page", "offset", "index")

# Plan-update tools where the SAME arg shape carries different payloads as the
# worker marks steps done. Bucket by a content hash of the plan so genuine
# progression (changed steps) reads as progress, while re-posting an identical
# plan over and over is a no-progress loop (incident kt_6cc4cd8e862947: a worker
# finished its research, then re-called update_task_plan to the round cap instead
# of handing off). Unlike read/probe tools the arg keys never vary, so a
# key-only signature could never tell thrash from progress — the content hash is
# what makes this tool loop-detectable at all.
_PLAN_UPDATE_LOOP_TOOLS: frozenset[str] = frozenset({
    "update_task_plan",
})
_PLAN_UPDATE_CONTENT_KEYS: tuple[str, ...] = ("steps", "title")

_CONTENT_BUCKET_CONTENT_KEYS: tuple[str, ...] = ("content", "text", "findings")
_CONTENT_BUCKET_PATH_KEYS: tuple[str, ...] = (
    "relative_path",
    "script_path",
    "file_path",
    "filepath",
    "path",
    "skill_name",
)

_TRUNCATED_READ_FILE_MARKER = "⚠️ TRUNCATED:"
_WRITE_TEXT_TOOLS: frozenset[str] = frozenset({
    "write_text_file",
    "save_to_file",
    "append_to_text_file",
    "save_research_findings",
})

_KANBAN_PROGRESS_MUTATION_TOOLS: frozenset[str] = _WRITE_TEXT_TOOLS | frozenset({
    "write_web_act_file",
})

# Filesystem probe tools only — repeated calls without a write trigger the
# kanban observe-loop hard stop. Web/search tools keep the generic nudge/recovery
# path but must not salvage-block research workers that legitimately search.
_OBSERVE_LOOP_TOOLS: frozenset[str] = frozenset({
    "read_file",
    "list_directory",
    "read_file_chunk",
}) | _PATH_BUCKET_LOOP_TOOLS

_DEFAULT_KANBAN_OBSERVE_LOOP_HARD_STOP_THRESHOLD = 6

# Chunked reads and identical plan re-posts should count toward observe-loop
# detection; other exempt tools stay benign. (Plan re-posts only trip when the
# plan content is unchanged — see _PLAN_UPDATE_LOOP_TOOLS signature bucketing.)
_KANBAN_OBSERVE_LOOP_EXEMPT: frozenset[str] = _TOOL_CALL_LOOP_EXEMPT_TOOLS - frozenset({
    "read_file_chunk",
    "update_task_plan",
})

LoopSignature = tuple[str, tuple[str, ...], str]
TrippedSignature = tuple[str, tuple[str, ...], str, int]


def _parse_tool_call_args(tc: dict[str, Any]) -> tuple[str, dict[str, Any] | None]:
    name = str(tc.get("name") or "").strip()
    if not name:
        return ("", None)
    raw = tc.get("arguments")
    parsed: dict[str, Any] | None = None
    if isinstance(raw, str) and raw:
        try:
            p = json.loads(raw)
            if isinstance(p, dict):
                parsed = p
        except Exception:
            parsed = None
    elif isinstance(raw, dict):
        parsed = raw
    if parsed is None:
        return (name, None)
    try:
        from tool_arg_normalizer import normalize_tool_args as _norm

        parsed = _norm(name, dict(parsed))
    except Exception:
        pass
    if not isinstance(parsed, dict):
        return (name, None)
    return (name, parsed)


def _primary_path_arg(tool_name: str, parsed: dict[str, Any]) -> str:
    try:
        from tool_arg_normalizer import iter_scalar_top_level_path_fields

        for field_name, meta in iter_scalar_top_level_path_fields(tool_name):
            if meta.get("intent") != "read":
                continue
            val = parsed.get(field_name)
            if isinstance(val, str) and val.strip():
                return val.strip()
    except Exception:
        pass
    for key in ("filepath", "dir_path", "file_path", "path", "project_name"):
        val = parsed.get(key)
        if isinstance(val, str) and val.strip():
            return val.strip()
    return ""


def _tool_call_loop_signature(
    tc: dict[str, Any],
    *,
    workspace_dir: str = "",
) -> LoopSignature:
    """Return ``(tool_name, sorted arg keys, path_bucket)`` for loop detection."""
    name, parsed = _parse_tool_call_args(tc)
    if not name:
        return ("", (), "")
    if parsed is None:
        return (name, (), "")
    keys = tuple(sorted(k for k in parsed.keys() if isinstance(k, str)))
    path_bucket = ""
    if (
        name in _PATH_BUCKET_LOOP_TOOLS
        or name.endswith("_lint")
        or name.endswith("_render")
    ):
        from utils.path_normalize import normalize_path_bucket

        raw_path = _primary_path_arg(name, parsed)
        if raw_path:
            path_bucket = normalize_path_bucket(raw_path, workspace=workspace_dir)
            if not path_bucket:
                import hashlib

                digest = hashlib.sha256(raw_path.encode("utf-8", errors="replace")).hexdigest()[:12]
                path_bucket = f"raw:{digest}"
    elif name in _PAGINATED_READ_LOOP_TOOLS:
        raw_path = ""
        for key in _PAGINATED_READ_PATH_KEYS:
            val = parsed.get(key)
            if isinstance(val, str) and val.strip():
                raw_path = val.strip()
                break
        bucket = ""
        if raw_path:
            from utils.path_normalize import normalize_path_bucket

            bucket = normalize_path_bucket(raw_path, workspace=workspace_dir) or raw_path
        slice_id = ""
        for key in _PAGINATED_READ_INDEX_KEYS:
            if key in parsed and parsed.get(key) is not None:
                slice_id = str(parsed.get(key)).strip()
                break
        path_bucket = f"{bucket}#slice={slice_id}"
    elif name in _PLAN_UPDATE_LOOP_TOOLS:
        import hashlib

        parts: list[str] = []
        for key in _PLAN_UPDATE_CONTENT_KEYS:
            val = parsed.get(key)
            if val is None:
                continue
            if not isinstance(val, str):
                try:
                    val = json.dumps(val, sort_keys=True, separators=(",", ":"))
                except (TypeError, ValueError):
                    val = str(val)
            if val.strip():
                digest = hashlib.sha256(
                    val.strip().encode("utf-8", errors="replace")
                ).hexdigest()[:12]
                parts.append(f"{key}:{digest}")
        path_bucket = "|".join(parts)
    elif name in _CONTENT_BUCKET_LOOP_TOOLS or name in _RUN_PATH_BUCKET_LOOP_TOOLS:
        parts: list[str] = []
        for key in _CONTENT_BUCKET_PATH_KEYS:
            val = parsed.get(key)
            if isinstance(val, str) and val.strip():
                parts.append(val.strip())
                break
        if name in _CONTENT_BUCKET_LOOP_TOOLS:
            for key in _CONTENT_BUCKET_CONTENT_KEYS:
                val = parsed.get(key)
                if isinstance(val, str) and val:
                    import hashlib

                    digest = hashlib.sha256(val.encode("utf-8", errors="replace")).hexdigest()[:12]
                    parts.append(f"sha:{digest}")
                    break
        path_bucket = "|".join(parts)
    return (name, keys, path_bucket)


def _coarse_tool_call_signature(tc: dict[str, Any]) -> tuple[str, tuple[str, ...]]:
    """Deprecated alias: returns ``(tool_name, arg_keys)`` without path bucket."""
    sig = _tool_call_loop_signature(tc, workspace_dir="")
    return (sig[0], sig[1])


def _update_tool_call_loop_counts(
    tool_calls_list: list[dict[str, Any]],
    counts: dict[LoopSignature, int],
    already_nudged: set[LoopSignature],
    *,
    workspace_dir: str = "",
    threshold: int = _TOOL_CALL_LOOP_DEFAULT_THRESHOLD,
    exempt: frozenset[str] = _TOOL_CALL_LOOP_EXEMPT_TOOLS,
) -> list[TrippedSignature]:
    """Increment per-signature counts and return newly-tripped signatures.

    Mutates both ``counts`` and ``already_nudged``. Each signature can only
    appear in the returned list once per run (subsequent crossings are
    silent) so the model does not get spammed by the same nudge.
    """
    tripped: list[TrippedSignature] = []
    if not isinstance(tool_calls_list, list):
        return tripped
    for tc in tool_calls_list:
        if not isinstance(tc, dict):
            continue
        sig = _tool_call_loop_signature(tc, workspace_dir=workspace_dir)
        name, keys, path_bucket = sig
        if not name or name in exempt:
            continue
        counts[sig] = counts.get(sig, 0) + 1
        if counts[sig] >= threshold and sig not in already_nudged:
            tripped.append((name, keys, path_bucket, counts[sig]))
            already_nudged.add(sig)
    return tripped


def _select_tool_call_loop_recovery_trips(
    tool_calls_list: list[dict[str, Any]],
    counts: dict[LoopSignature, int],
    already_recovery_nudged: set[LoopSignature],
    *,
    workspace_dir: str = "",
    threshold: int = _TOOL_CALL_LOOP_RECOVERY_THRESHOLD,
    exempt: frozenset[str] = _TOOL_CALL_LOOP_EXEMPT_TOOLS,
) -> list[TrippedSignature]:
    """Return signatures crossing the stronger recovery threshold.

    This helper does not mutate ``counts`` (the base detector already did).
    It only marks signatures in ``already_recovery_nudged`` so we inject at
    most one recovery nudge per signature per run.
    """
    tripped: list[TrippedSignature] = []
    if not isinstance(tool_calls_list, list):
        return tripped
    seen_round: set[LoopSignature] = set()
    for tc in tool_calls_list:
        if not isinstance(tc, dict):
            continue
        sig = _tool_call_loop_signature(tc, workspace_dir=workspace_dir)
        if sig in seen_round:
            continue
        seen_round.add(sig)
        name, keys, path_bucket = sig
        if not name or name in exempt:
            continue
        count = int(counts.get(sig, 0))
        if count >= threshold and sig not in already_recovery_nudged:
            tripped.append((name, keys, path_bucket, count))
            already_recovery_nudged.add(sig)
    return tripped


def kanban_observe_loop_hard_stop_threshold() -> int:
    """Configured observe-only hard-stop threshold for kanban workers."""
    try:
        from core.config import get_kanban_config

        raw = get_kanban_config().get("observe_loop_hard_stop_threshold")
        if raw is None:
            return _DEFAULT_KANBAN_OBSERVE_LOOP_HARD_STOP_THRESHOLD
        return max(3, int(raw))
    except Exception:
        return _DEFAULT_KANBAN_OBSERVE_LOOP_HARD_STOP_THRESHOLD


def _is_observe_loop_tool(tool_name: str) -> bool:
    name = str(tool_name or "").strip()
    if not name:
        return False
    if name in _OBSERVE_LOOP_TOOLS:
        return True
    # Identical plan re-posts are a no-progress loop (the signature bucket only
    # collides when the plan content is unchanged), so route them through the
    # same kanban observe hard-stop + salvage path as repeated reads.
    if name in _PLAN_UPDATE_LOOP_TOOLS:
        return True
    if name.endswith("_lint") or name.endswith("_render"):
        return True
    return False


def _is_verify_loop_tool(tool_name: str) -> bool:
    name = str(tool_name or "").strip()
    return bool(name) and (name.endswith("_lint") or name.endswith("_render"))


def _select_kanban_observe_loop_hard_stop_trips(
    tool_calls_list: list[dict[str, Any]],
    counts: dict[LoopSignature, int],
    already_hard_stopped: set[LoopSignature],
    *,
    workspace_dir: str = "",
    threshold: int | None = None,
    exempt: frozenset[str] = _KANBAN_OBSERVE_LOOP_EXEMPT,
) -> list[TrippedSignature]:
    """Return observe/probe signatures that exceed the kanban hard-stop threshold."""
    effective = threshold if threshold is not None else kanban_observe_loop_hard_stop_threshold()
    tripped = _select_tool_call_loop_hard_stop_trips(
        tool_calls_list,
        counts,
        already_hard_stopped,
        workspace_dir=workspace_dir,
        threshold=effective,
        exempt=exempt,
    )
    return [
        row for row in tripped
        if _is_observe_loop_tool(row[0]) and not _is_verify_loop_tool(row[0])
    ]


def _select_kanban_verify_loop_hard_stop_trips(
    tool_calls_list: list[dict[str, Any]],
    counts: dict[LoopSignature, int],
    already_hard_stopped: set[LoopSignature],
    *,
    workspace_dir: str = "",
    threshold: int | None = None,
    exempt: frozenset[str] = _TOOL_CALL_LOOP_EXEMPT_TOOLS,
) -> list[TrippedSignature]:
    """Return lint/render verify signatures that exceed the kanban hard-stop threshold."""
    effective = threshold if threshold is not None else kanban_observe_loop_hard_stop_threshold()
    tripped = _select_tool_call_loop_hard_stop_trips(
        tool_calls_list,
        counts,
        already_hard_stopped,
        workspace_dir=workspace_dir,
        threshold=effective,
        exempt=exempt,
    )
    return [
        row for row in tripped
        if _is_verify_loop_tool(row[0])
    ]


def reset_tool_call_loop_state(
    counts: dict[LoopSignature, int],
    nudged: set[LoopSignature],
    recovery_nudged: set[LoopSignature],
    hard_stopped: set[LoopSignature],
) -> int:
    """Decay loop-guard state after context compression/trim; returns affected count.

    Once messages are summarized or trimmed, the model can no longer see the
    calls (or nudges) the counters refer to — a repeat after that point is
    amnesia, not defiance. Escalating stale counters to a hard stop killed
    trace 6be8ae2b mid-task. But fully *clearing* the counts blinded every
    guard whenever compression fired more often than a loop could
    re-accumulate: worker kt_aa761234c9d642 re-read the same file 13 times
    without a single nudge because frequent trims zeroed the counters first.

    Decaying to 2/3 fixes both incidents: one-off repeats wash out to zero
    within a couple of compressions, while a persistent loop converges to a
    steady state of ~3x the per-compression repeat rate and still crosses the
    nudge/hard-stop thresholds. Nudge/recovery/hard-stop markers are cleared
    outright so post-trim guidance re-fires where the model can see it.
    """
    affected = len(counts)
    for sig in list(counts.keys()):
        decayed = (counts[sig] * 2) // 3
        if decayed <= 0:
            del counts[sig]
        else:
            counts[sig] = decayed
    nudged.clear()
    recovery_nudged.clear()
    hard_stopped.clear()
    return affected


def _select_tool_call_loop_hard_stop_trips(
    tool_calls_list: list[dict[str, Any]],
    counts: dict[LoopSignature, int],
    already_hard_stopped: set[LoopSignature],
    *,
    workspace_dir: str = "",
    threshold: int = _TOOL_CALL_LOOP_HARD_STOP_THRESHOLD,
    exempt: frozenset[str] = _TOOL_CALL_LOOP_EXEMPT_TOOLS,
) -> list[TrippedSignature]:
    """Return signatures that exceed the parent-run hard-stop threshold."""
    tripped: list[TrippedSignature] = []
    if not isinstance(tool_calls_list, list):
        return tripped
    seen_round: set[LoopSignature] = set()
    for tc in tool_calls_list:
        if not isinstance(tc, dict):
            continue
        sig = _tool_call_loop_signature(tc, workspace_dir=workspace_dir)
        if sig in seen_round:
            continue
        seen_round.add(sig)
        name, keys, path_bucket = sig
        if not name or name in exempt:
            continue
        count = int(counts.get(sig, 0))
        if count >= threshold and sig not in already_hard_stopped:
            tripped.append((name, keys, path_bucket, count))
            already_hard_stopped.add(sig)
    return tripped


def _format_tripped_signature_line(name: str, keys: tuple[str, ...], path_bucket: str, count: int) -> str:
    from utils.path_normalize import truncate_path_display

    keys_display = ", ".join(keys) if keys else "(no args)"
    if path_bucket:
        path_display = truncate_path_display(path_bucket)
        return f"- `{name}` (args: {keys_display}; path: `{path_display}`) — called {count} times"
    return f"- `{name}` (args: {keys_display}) \u2014 called {count} times"


def _append_anchor_block(msg: str, *, anchor_summary: str) -> str:
    if not anchor_summary:
        return msg
    return (
        msg
        + "\n\nKnown paths from this run (recorded, not authoritative):\n"
        + anchor_summary
        + "\nReuse these exact paths when retrying."
    )


def _format_tool_call_loop_nudge(
    tripped: list[TrippedSignature],
    *,
    anchor_summary: str = "",
    is_coordinator_parent: bool = False,
    truncated_read_paths: set[str] | None = None,
) -> str:
    """Neutral, non-authoritative nudge describing the repeated signatures.

    Framed like the task-plan continuation prompt so the prompt-injection
    boundary stays consistent: the assistant sees a ``user`` role message
    that records an observation and suggests next actions, not a directive.
    """
    if not tripped:
        return ""
    lines: list[str] = []
    for name, keys, path_bucket, count in tripped:
        lines.append(_format_tripped_signature_line(name, keys, path_bucket, count))
    msg = (
        "Heads up (recorded, not authoritative): one or more tools have been "
        "called repeatedly with the same shape of arguments without visibly "
        "advancing the task plan:\n"
        + "\n".join(lines)
        + "\nTry a different approach: use `find_tools` to discover a better "
        "tool, change the inputs substantially, or mark the related plan "
        "step `blocked` or `done` via `update_task_plan` and move on. "
        "This note is recorded for salience only; it is not a new instruction."
    )
    if is_coordinator_parent:
        msg = (
            "Heads up (recorded, not authoritative): Kanban coordinator mode — "
            "you have been polling the board repeatedly without new worker results:\n"
            + "\n".join(lines)
            + "\nStop re-polling the same task. Give the operator a brief status and end "
            "your turn; this thread will resume when Kanban workers deliver results."
        )
    _trunc = truncated_read_paths or set()
    if _trunc and any(n == "read_file" and pb in _trunc for n, _, pb, _ in tripped):
        msg += (
            "\nFor truncated files, switch to `read_file_chunk` instead of repeating `read_file` "
            "on the same path."
        )
    return _append_anchor_block(msg, anchor_summary=anchor_summary)


def _format_tool_call_loop_recovery(
    tripped: list[TrippedSignature],
    *,
    is_kanban_worker: bool = False,
    is_coordinator_parent: bool = False,
    observed_tool_names: set[str] | None = None,
    anchor_summary: str = "",
    truncated_read_paths: set[str] | None = None,
) -> str:
    """Stronger guidance for repeated loop signatures.

    This remains non-authoritative and advisory, but gives concrete next
    actions after we have already nudged once.
    """
    if not tripped:
        return ""
    lines: list[str] = []
    for name, keys, path_bucket, count in tripped:
        lines.append(_format_tripped_signature_line(name, keys, path_bucket, count))
    msg = (
        "Recovery nudge (recorded, not authoritative): repeated tool-call "
        "shapes are still looping without clear progress:\n"
        + "\n".join(lines)
        + "\nTry a materially different step now: use `find_tools` (if needed), "
        "change inputs substantially, or mark the current plan step `blocked`/`done` "
        "with `update_task_plan` before moving to the next step."
    )
    if is_coordinator_parent:
        msg = (
            "Recovery nudge (recorded, not authoritative): Kanban coordinator mode — "
            "repeated board polling without new worker output:\n"
            + "\n".join(lines)
            + "\nDo not call delegate_task or implementation tools. Briefly summarize board "
            "state for the operator and stop; you will resume on Kanban delivery."
        )
    if is_kanban_worker:
        msg += (
            "\nBecause this is a kanban worker, avoid free-form retries that repeat the "
            "same shape. If still blocked, end with a fenced `kanban-result` JSON block "
            "describing what failed and what exact next action is required."
        )
    _trunc = truncated_read_paths or set()
    if _trunc and any(n == "read_file" and pb in _trunc for n, _, pb, _ in tripped):
        msg += (
            "\nThis file was truncated by `read_file`. Use `read_file_chunk` to inspect and "
            "patch the relevant section, then re-run lint/render if needed."
        )
    if is_kanban_worker:
        msg += (
            "\nIf repeated read/probe/verify calls are not unblocking progress, switch to a "
            "concrete mutation (write/patch the artifact) or end with a fenced `kanban-result` "
            "block using metadata.status=\"blocked\" and a specific next_action."
        )
    return _append_anchor_block(msg, anchor_summary=anchor_summary)


def _format_kanban_observe_loop_salvage_prompt(
    tripped: list[TrippedSignature],
    *,
    anchor_summary: str = "",
    recent_paths: str = "",
    verify_mode: bool = False,
) -> str:
    """User-facing nudge when a kanban worker loops (no parseable fence — use synthesize)."""
    sig_labels = ", ".join(
        f"{n}({','.join(ks) if ks else '-'})x{c}"
        for (n, ks, _pb, c) in tripped
    ) if tripped else ("verify tools" if verify_mode else "observe-only tools")
    paths_hint = recent_paths.strip() or "paths you inspected this run"
    loop_kind = "verify" if verify_mode else "read/probe"
    msg = (
        "Kanban loop circuit break (recorded, not authoritative): "
        f"repeated {loop_kind} tool calls without making progress ({sig_labels}).\n\n"
        "Stop retrying the same pattern. End this run with a trailing ```kanban-result``` block "
        "using metadata.status=\"blocked\", blockers, next_action, and artifact_paths. "
        f"Focus the next_action on a concrete mutation or fix using {paths_hint}."
    )
    return _append_anchor_block(msg, anchor_summary=anchor_summary)


def _web_act_blockers_from_worker_transcript(worker_transcript: str) -> list[str]:
    """Collect web-act blocker slugs from recent ``run_web_script`` tool output."""
    try:
        from utils.web_act_diagnostics import (
            infer_web_act_blockers,
            merge_web_act_blockers,
            parse_blockers_from_tool_message,
        )
    except Exception:
        return []
    text = str(worker_transcript or "")
    found: list[str] = []
    for line in text.splitlines():
        if line.strip().lower().startswith("blockers:"):
            found = merge_web_act_blockers(found, parse_blockers_from_tool_message(line))
    if found:
        return found
    for chunk in reversed([c.strip() for c in text.split("\n\n") if c.strip()]):
        if "exit_code:" not in chunk:
            continue
        fresh_run = False
        exit_code: int | None = None
        validation_detail = ""
        for line in chunk.splitlines():
            ls = line.strip()
            low = ls.lower()
            if low.startswith("fresh_run:"):
                fresh_run = ls.split(":", 1)[1].strip().lower() in ("true", "1", "yes")
            elif low.startswith("exit_code:"):
                try:
                    exit_code = int(ls.split(":", 1)[1].strip())
                except (TypeError, ValueError):
                    exit_code = None
            elif low.startswith("validation_detail:"):
                validation_detail = ls.split(":", 1)[1].strip()
        inferred = infer_web_act_blockers(
            tool_message=chunk,
            fresh_run=fresh_run,
            exit_code=exit_code,
            validation_detail=validation_detail,
        )
        if inferred:
            return inferred
    return found


def synthesize_kanban_loop_salvage_handoff(
    tripped: list[TrippedSignature],
    *,
    recent_paths: str = "",
    verify_mode: bool = False,
    task_metadata: dict[str, Any] | None = None,
    worker_transcript: str = "",
) -> str:
    """Deterministic assistant output with a parseable blocked salvage handoff."""
    from utils.kanban_context import observe_loop_salvage_next_action

    sig_labels = ", ".join(
        f"{n}({','.join(ks) if ks else '-'})x{c}"
        for (n, ks, _pb, c) in tripped
    ) if tripped else ("verify tools" if verify_mode else "observe-only tools")
    artifact_paths: list[str] = [
        p.strip() for p in recent_paths.split(",") if p.strip()
    ][:4]
    if not artifact_paths:
        for _n, _ks, path_bucket, _c in tripped:
            pb = str(path_bucket or "").strip()
            if pb and pb not in artifact_paths:
                artifact_paths.append(pb)
            if len(artifact_paths) >= 4:
                break
    primary_path = artifact_paths[0] if artifact_paths else "the primary artifact for this task"
    blocker = "verify_loop_no_progress" if verify_mode else "observe_loop_no_progress"
    loop_kind = "verify" if verify_mode else "observe/probe"
    summary = (
        f"Circuit break: repeated {loop_kind} tool calls without progress ({sig_labels})."
    )
    next_action = observe_loop_salvage_next_action(
        task_metadata=task_metadata,
        primary_path=primary_path,
        verify_mode=verify_mode,
    )
    metadata: dict[str, Any] = {
        "status": "blocked",
        "blockers": [blocker],
        "next_action": next_action,
        "artifact_paths": artifact_paths,
        "failure_reason": (
            "kanban_verify_loop_hard_stop" if verify_mode else "kanban_observe_loop_hard_stop"
        ),
    }
    try:
        from utils.kanban_context import (
            kanban_task_is_video_render_deliverable,
            kanban_task_is_web_act_deliverable,
        )

        if kanban_task_is_video_render_deliverable(task_metadata):
            metadata["tools_required_next"] = [
                "write_text_file",
                "hyperframes_lint",
                "hyperframes_render",
            ]
        elif kanban_task_is_web_act_deliverable(task_metadata):
            metadata["tools_required_next"] = [
                "write_web_act_file",
                "run_web_script",
            ]
            wa_blockers = _web_act_blockers_from_worker_transcript(worker_transcript)
            if wa_blockers:
                try:
                    from utils.web_act_diagnostics import (
                        merge_web_act_blockers,
                        web_act_blocker_next_action_hint,
                    )

                    metadata["blockers"] = merge_web_act_blockers(
                        wa_blockers,
                        metadata.get("blockers"),
                    )
                    metadata["next_action"] = web_act_blocker_next_action_hint(
                        metadata["blockers"],
                        task_metadata=task_metadata,
                    )
                except Exception:
                    metadata["blockers"] = wa_blockers
    except Exception:
        pass
    try:
        from utils.council_orchestrator import is_council_task_metadata

        if is_council_task_metadata(task_metadata):
            seat = str((task_metadata or {}).get("council_seat") or "researcher").strip().lower()
            metadata["council_seat"] = seat
            metadata.setdefault(
                "failure_reason",
                "kanban_verify_loop_hard_stop" if verify_mode else "kanban_observe_loop_hard_stop",
            )
    except Exception:
        pass
    payload = {"summary": summary, "metadata": metadata}
    return _format_kanban_result_message(payload)


def synthesize_kanban_loop_hard_stop_handoff(
    tripped: list[TrippedSignature],
    *,
    ledger: "KanbanRunLedger | None" = None,
    recent_paths: str = "",
    verify_mode: bool = False,
    task_metadata: dict[str, Any] | None = None,
    worker_transcript: str = "",
    visible_prose: str = "",
) -> tuple[str, bool]:
    """Choose the handoff for a kanban worker loop hard-stop.

    Prefers a verifiable in-band completion (the worker finished and then stalled
    on a no-progress loop such as repeated ``update_task_plan``) over the blocked
    "circuit break" salvage, so completed work is delivered instead of discarded.
    Returns ``(handoff_text, used_inband_completion)``.

    Only the observe path (``verify_mode=False``) considers the in-band summary;
    verify-mode loops mean the worker has a file it cannot get past lint/render, so
    the existing blocked salvage stays authoritative there.
    """
    if ledger is not None and not verify_mode:
        prose = " ".join(
            str(visible_prose or getattr(ledger, "prose_excerpt", "") or "").split()
        )
        completion = _ledger_inband_completion_handoff(
            ledger, task_metadata=task_metadata, prose=prose
        )
        if completion is not None:
            return completion, True
    return (
        synthesize_kanban_loop_salvage_handoff(
            tripped,
            recent_paths=recent_paths,
            verify_mode=verify_mode,
            task_metadata=task_metadata,
            worker_transcript=worker_transcript,
        ),
        False,
    )


_RESEARCH_GATHER_LOOP_TOOLS: frozenset[str] = frozenset({
    "deep_web_search",
    "search_web",
    "search_web_public",
    "read_and_analyze_page",
    "analyze_multiple_sources",
})

# Only search-style tools trigger gather hard-stop (reads are progress, not loops).
_RESEARCH_GATHER_HARD_STOP_TOOLS: frozenset[str] = frozenset({
    "deep_web_search",
    "search_web",
    "search_web_public",
})


def kanban_research_gather_loop_hard_stop_threshold() -> int:
    try:
        from core.config import get_kanban_config

        raw = get_kanban_config().get("research_gather_loop_hard_stop_threshold")
        if raw is None:
            # Slightly above observe-loop hard-stop so one read + several searches can run.
            return kanban_observe_loop_hard_stop_threshold() + 2
        return max(6, int(raw))
    except Exception:
        return kanban_observe_loop_hard_stop_threshold()


def _is_research_gather_loop_tool(tool_name: str) -> bool:
    return str(tool_name or "").strip() in _RESEARCH_GATHER_LOOP_TOOLS


def _select_kanban_research_gather_loop_hard_stop_trips(
    tool_calls_list: list[dict[str, Any]],
    counts: dict[LoopSignature, int],
    already_hard_stopped: set[LoopSignature],
    *,
    ledger: KanbanRunLedger | None = None,
    workspace_dir: str = "",
    threshold: int | None = None,
    exempt: frozenset[str] = _TOOL_CALL_LOOP_EXEMPT_TOOLS,
) -> list[TrippedSignature]:
    """Gather-loop trips only when search/read signatures repeat without scratch progress."""
    effective = threshold if threshold is not None else kanban_research_gather_loop_hard_stop_threshold()
    tripped = _select_tool_call_loop_hard_stop_trips(
        tool_calls_list,
        counts,
        already_hard_stopped,
        workspace_dir=workspace_dir,
        threshold=effective,
        exempt=exempt,
    )
    if ledger is None:
        return [row for row in tripped if _is_research_gather_loop_tool(row[0])]
    scratch_stalled = (
        ledger.append_scratch_count <= ledger.last_scratch_count_at_checkpoint
    )
    if not scratch_stalled:
        return []
    return [row for row in tripped if str(row[0] or "").strip() in _RESEARCH_GATHER_HARD_STOP_TOOLS]


def kanban_gather_hard_stop_should_defer(
    ledger: KanbanRunLedger,
    *,
    gather_excerpt: str = "",
    append_scratch_count: int = 0,
    already_deferred: bool = False,
    min_excerpt_chars: int = 200,
    min_gather_chars: int = 500,
) -> bool:
    """Defer gather hard-stop once when the worker already has usable research text."""
    if int(append_scratch_count) > 0 or already_deferred:
        return False
    excerpt = " ".join(str(gather_excerpt or "").split())
    if len(excerpt) >= min_excerpt_chars:
        return True
    return int(ledger.research_gather_chars) >= min_gather_chars


def _council_gather_is_meta_noise(text: str) -> bool:
    """True when tool output is truncation/path noise, not source evidence."""
    lower = str(text or "").lower()
    if not lower.strip():
        return True
    meta_markers = (
        "truncated: this file has",
        "only the first 5000 are shown",
        "path does not exist:",
        "parent directory exists. entries:",
    )
    return any(marker in lower for marker in meta_markers)


def _council_researcher_usable_gather_excerpt(gather_excerpt: str) -> str:
    """Drop search-dedup boilerplate and other non-evidence tool noise from salvage input."""
    raw = str(gather_excerpt or "")
    if not raw.strip():
        return ""
    junk_markers = (
        "you already searched for a very similar query",
        "do not call search_web again",
        "do not call deep_web_search again",
        "do not call search_web or deep_web_search again",
        "previous search result — use this to answer",
        "[previous search result",
        "truncated: this file has",
        "only the first 5000 are shown",
        "do not save this truncated content",
        "use read_file_chunk",
        "path does not exist:",
        "parent directory exists. entries:",
    )
    kept: list[str] = []
    for para in re.split(r"\n{2,}", raw):
        chunk = " ".join(para.split())
        if not chunk:
            continue
        lower = chunk.lower()
        if any(marker in lower for marker in junk_markers):
            continue
        kept.append(chunk)
    return " ".join(kept)


def _extract_search_snippet_claims(excerpt: str, *, max_claims: int = 4) -> list[dict[str, Any]]:
    """Parse search_web Title/URL/Snippet blocks or deep_web_search markdown/JSON into claims."""
    import json
    import re

    claims: list[dict[str, Any]] = []

    def _append_claim(*, title: str, url: str, snippet: str) -> None:
        if len(claims) >= max_claims:
            return
        title = " ".join((title or "").split())
        url = (url or "").strip()
        snippet = " ".join((snippet or "").split())
        if not url and not title:
            return
        if snippet:
            text = snippet[:420]
        else:
            text = title[:420]
        text = re.sub(r"(?i)\b(title|url|snippet)\s*:", "", text).strip()
        if not text:
            return
        claims.append(
            {
                "id": f"c{len(claims) + 1}",
                "text": text[:500],
                "evidence": [url] if url else ["gathered search/read tool results"],
                "confidence": "medium" if snippet else "low",
            }
        )

    for block in re.split(r"(?=Title:\s)", excerpt, flags=re.IGNORECASE):
        chunk = block.strip()
        if not chunk:
            continue
        title_m = re.search(r"Title:\s*(.+?)(?:\n|$)", chunk, re.IGNORECASE)
        url_m = re.search(r"URL:\s*(\S+)", chunk, re.IGNORECASE)
        snippet_m = re.search(r"Snippet:\s*(.+?)(?=\n\nTitle:|\Z)", chunk, re.IGNORECASE | re.DOTALL)
        if not title_m or not url_m:
            continue
        _append_claim(
            title=title_m.group(1),
            url=url_m.group(1),
            snippet=snippet_m.group(1) if snippet_m else "",
        )
    if claims:
        return claims

    json_m = re.search(r"```json\s*(\[[\s\S]*?\])\s*```", excerpt)
    if json_m:
        try:
            rows = json.loads(json_m.group(1))
        except json.JSONDecodeError:
            rows = None
        if isinstance(rows, list):
            for row in rows:
                if not isinstance(row, dict):
                    continue
                _append_claim(
                    title=str(row.get("title") or ""),
                    url=str(row.get("url") or ""),
                    snippet=str(row.get("snippet") or ""),
                )
                if len(claims) >= max_claims:
                    break
    if claims:
        return claims

    for block in re.split(r"\n(?=\d+\.\s+\*\*)", excerpt):
        chunk = block.strip()
        if not chunk:
            continue
        title_m = re.search(r"^\d+\.\s+\*\*(.+?)\*\*", chunk, re.MULTILINE)
        url_m = re.search(r"-\s*URL:\s*(\S+)", chunk, re.IGNORECASE)
        snippet_m = re.search(
            r"-\s*Snippet:\s*(.+?)(?=\n\d+\.\s+\*\*|\n```|\Z)",
            chunk,
            re.DOTALL | re.IGNORECASE,
        )
        if not title_m or not url_m:
            continue
        _append_claim(
            title=title_m.group(1),
            url=url_m.group(1),
            snippet=snippet_m.group(1) if snippet_m else "",
        )
    return claims


def _extract_prose_claims(
    prose: str,
    urls: list[str],
    *,
    max_claims: int = 4,
) -> list[dict[str, Any]]:
    """Build claims from the worker's own synthesis text (preferred over raw snippets).

    The Researcher model usually writes a solid prose synthesis even when it fails the
    fence contract; discarding it for Title/URL/Snippet extraction produced claims the
    Verifier correctly rejected as snippet dumps (kcn_c32156cfe1b340e6).
    """
    import re

    raw = str(prose or "")
    if not raw.strip():
        return []
    # Drop fenced blocks (kanban-result or code) and private planning.
    raw = re.sub(r"```.*?```", " ", raw, flags=re.DOTALL)
    raw = re.sub(r"<think>.*?</think>", " ", raw, flags=re.DOTALL | re.IGNORECASE)
    # Strip markdown decoration but keep sentence text.
    raw = re.sub(r"[*_#`]+", "", raw)

    candidates: list[str] = []
    for line in raw.splitlines():
        chunk = " ".join(line.split()).strip("-• \t")
        if not chunk:
            continue
        # Sentence-split long paragraphs; keep bullets whole.
        parts = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9])", chunk) if len(chunk) > 320 else [chunk]
        candidates.extend(p.strip() for p in parts if p.strip())

    claims: list[dict[str, Any]] = []
    seen: set[str] = set()
    for cand in candidates:
        if not (40 <= len(cand) <= 500):
            continue
        lower = cand.lower()
        if "title:" in lower or "snippet:" in lower or lower.startswith(("url:", "http")):
            continue
        if "kanban" in lower or "handoff" in lower or "researcher seat" in lower:
            continue
        key = lower[:80]
        if key in seen:
            continue
        seen.add(key)
        claims.append(
            {
                "id": f"c{len(claims) + 1}",
                "text": cand[:500],
                "evidence": urls[:3] if urls else ["gathered search/read tool results"],
                "confidence": "medium",
            }
        )
        if len(claims) >= max_claims:
            break
    return claims


def _extract_repo_path_claims(excerpt: str, *, max_claims: int = 4) -> list[dict[str, Any]]:
    """Turn pre-seeded ### path headers + body into claims when salvage runs on seeded context."""
    import re

    claims: list[dict[str, Any]] = []
    for match in re.finditer(r"###\s+(\S+)\n([\s\S]*?)(?=\n###\s+|\Z)", excerpt):
        rel_path = match.group(1).strip()
        body = " ".join(match.group(2).split())
        if len(body) < 40:
            continue
        claims.append(
            {
                "id": f"c{len(claims) + 1}",
                "text": body[:500],
                "evidence": [rel_path],
                "confidence": "high",
            }
        )
        if len(claims) >= max_claims:
            break
    return claims


def _search_claim_source_for_salvage(raw_gather: str, excerpt: str) -> str:
    """Prefer unflattened gather text when deep_web_search/search_web structure is present."""
    import re

    raw = str(raw_gather or "")
    if not raw.strip():
        return str(excerpt or "")
    if (
        "## search results" in raw.lower()
        or "```json" in raw.lower()
        or re.search(r"(?im)^title:\s", raw)
        or re.search(r"\n\d+\.\s+\*\*", raw)
    ):
        return raw
    return str(excerpt or raw)


def synthesize_council_researcher_completion_handoff(
    gather_excerpt: str,
    *,
    sig_labels: str = "",
    worker_prose: str = "",
) -> str:
    """Turn gathered tool text into a valid Researcher kanban-result (completion, not blocked).

    When ``worker_prose`` (the model's own visible synthesis) is available, claims are
    extracted from it first; raw search-snippet extraction is the fallback only.
    """
    import re

    raw_gather = str(gather_excerpt or "")
    excerpt = _council_researcher_usable_gather_excerpt(raw_gather)
    if len(excerpt) < 80:
        excerpt = _council_researcher_usable_gather_excerpt(
            " ".join(raw_gather.split())
        )
    if _council_gather_is_meta_noise(excerpt):
        excerpt = ""
    # Pull evidence URLs from BOTH the filtered excerpt and the raw gather:
    # the junk filter can drop URL-bearing paragraphs, and claims salvaged
    # without real URLs are correctly rejected by the Verifier as
    # unsubstantiated (incident kcn_a5076cb276464735).
    urls = list(
        dict.fromkeys(
            re.findall(r"https?://[^\s\)\]>\"']+", excerpt)
            + re.findall(r"https?://[^\s\)\]>\"']+", raw_gather)
        )
    )[:8]
    uncertainties: list[str] = []
    claims: list[dict[str, Any]] = []

    prose_claims = _extract_prose_claims(worker_prose, urls)
    search_source = _search_claim_source_for_salvage(raw_gather, excerpt)

    # Dedup-only second searches can zero the filtered excerpt while Title/URL
    # blocks remain in raw tool text (``9ce6ae73`` thin salvage).
    if len(excerpt) < 80 and len(raw_gather) >= 200:
        claims = _extract_repo_path_claims(raw_gather)
        if not claims:
            claims = prose_claims or _extract_search_snippet_claims(search_source)
        if claims:
            excerpt = raw_gather

    if len(excerpt) >= 80:
        claims = _extract_repo_path_claims(excerpt)
        if not claims:
            claims = prose_claims or _extract_search_snippet_claims(search_source)
        if not claims:
            chunk_a = re.sub(r"(?i)\b(title|url|snippet)\s*:", "", excerpt[:700]).strip()
            chunk_b = (
                re.sub(r"(?i)\b(title|url|snippet)\s*:", "", excerpt[700:1400]).strip()
                if len(excerpt) > 700
                else ""
            )
            if chunk_a:
                claims.append(
                    {
                        "id": "c1",
                        "text": chunk_a,
                        "evidence": urls[:3] if urls else ["gathered search/read tool results"],
                        "confidence": "medium",
                    }
                )
            if chunk_b:
                claims.append(
                    {
                        "id": "c2",
                        "text": chunk_b,
                        "evidence": urls[3:6] if len(urls) > 3 else (urls[:1] or ["gathered search/read tool results"]),
                        "confidence": "low",
                    }
                )
    elif prose_claims:
        claims = prose_claims
        uncertainties.append("Claims synthesized from worker prose; tool evidence was thin.")
    else:
        uncertainties.append("Thin tool output; claims are provisional pending operator review.")
        claims.append(
            {
                "id": "c1",
                "text": excerpt or "Researcher gathered limited evidence before completion salvage.",
                "evidence": urls[:2] if urls else [],
                "confidence": "low",
            }
        )

    summary = (
        "Researcher seat compiled source-grounded claims from gathered tool results "
        "for the Captain's frame."
    )
    if sig_labels:
        summary += f" (Auto-completed after tool-loop guard: {sig_labels})"

    metadata: dict[str, Any] = {
        "council_seat": "researcher",
        "claims": claims,
        "uncertainties": uncertainties,
        "completion_salvage": True,
        "handoff_source": "council_researcher_salvage",
    }
    if sig_labels:
        metadata["salvage_sig"] = sig_labels[:240]
    return _format_kanban_result_message({"summary": summary[:2000], "metadata": metadata})


def synthesize_kanban_research_gather_salvage_handoff(
    tripped: list[TrippedSignature],
    *,
    recent_paths: str = "",
    append_scratch_count: int = 0,
    min_scratch_appends: int = 2,
    gather_excerpt: str = "",
) -> str:
    sig_labels = ", ".join(
        f"{n}({','.join(ks) if ks else '-'})x{c}"
        for (n, ks, _pb, c) in tripped
    ) if tripped else "research gather tools"
    remaining = max(0, min_scratch_appends - int(append_scratch_count))
    excerpt = " ".join(str(gather_excerpt or "").split())
    if len(excerpt) >= 48:
        summary = (
            f"Research gathered but append_research_scratch was not called before search loop break "
            f"({sig_labels}). Findings: {excerpt[:1800]}"
        ).strip()
        status = "partial"
        blockers = ["research_gather_loop_no_progress"]
        failure_reason = "kanban_research_gather_loop_hard_stop"
    else:
        summary = (
            f"Circuit break: repeated research gather without scratch progress ({sig_labels})."
        )
        status = "blocked"
        blockers = ["research_gather_loop_no_progress"]
        failure_reason = "kanban_research_gather_loop_hard_stop"
    next_action = (
        "Stop calling deep_web_search and read_and_analyze_page. "
        f"Call append_research_scratch with a dense synthesis of findings "
        f"({remaining} more append(s) needed before generate_final_research_report), "
        "then end with a trailing kanban-result fenced JSON block."
    )
    metadata: dict[str, Any] = {
        "status": status,
        "blockers": blockers,
        "next_action": next_action,
        "failure_reason": failure_reason,
        "tools_required_next": ["append_research_scratch"],
    }
    if recent_paths:
        metadata["artifact_paths"] = [p.strip() for p in recent_paths.split(",") if p.strip()][:4]
    payload = {"summary": summary[:2000], "metadata": metadata}
    return _format_kanban_result_message(payload)


# Research/plan tools whose repeated use without a deliverable counts as a non-progress round.
RESEARCH_GATHER_OR_PLAN_TOOLS: frozenset[str] = frozenset({
    "plan_research_subquestions",
    "deep_web_search",
    "search_web",
    "search_web_public",
    "read_and_analyze_page",
    "analyze_multiple_sources",
})

# Tools that fetch *real external evidence* (web/page/corpus). Distinct from planning:
# planning output (generated sub-questions) is NOT findings and must not satisfy the
# force-close evidence gate — otherwise a plan-only loop force-closes with nothing real
# to write and the model can only emit an honest "no findings" block.
RESEARCH_REAL_GATHER_TOOLS: frozenset[str] = frozenset({
    "deep_web_search",
    "search_web",
    "search_web_public",
    "read_and_analyze_page",
    "analyze_multiple_sources",
    "read_webpage",
    "summarize_url",
    "query_knowledge",
    "ask_studied_corpus",
})

# Planning/meta tools: they count toward non-progress, and are stripped during a forced
# SEARCH turn so the model cannot keep planning instead of actually gathering evidence.
RESEARCH_PLAN_ONLY_TOOLS: frozenset[str] = frozenset({
    "plan_research_subquestions",
    "reflect_research_state",
    "suggest_follow_up_queries",
})


def research_round_made_deliverable_progress(
    *,
    appended_scratch: bool,
    generated_report: bool,
) -> bool:
    """A research round 'progressed' only when it moved toward the deliverable.

    Gathering (search/plan/read) is not progress on its own — the worker must either
    accumulate synthesis (append_research_scratch) or produce the report. Used by the
    agent loop to count consecutive non-progress rounds before forcing a close.
    """
    return bool(appended_scratch or generated_report)


def should_force_research_close(
    *,
    nonprogress_rounds: int,
    threshold: int,
    force_close_attempts: int,
    max_attempts: int,
    report_generated: bool,
    gather_excerpt_chars: int,
    min_evidence_chars: int = 200,
    did_real_gather: bool = True,
) -> bool:
    """True when a (non-council) research worker should be forced to CLOSE from evidence.

    Doctrine-pure: this only *forces the model* to write its findings (or emit an honest
    blocked handoff) — it never synthesizes on the model's behalf. We require:
      * no report produced yet,
      * we have not already exhausted the force-close attempt cap (so an honest give-up
        stays reachable),
      * the worker has looped at least ``threshold`` consecutive non-progress rounds,
      * the worker actually ran a real search/read/corpus call (``did_real_gather``) — a
        plan-only loop has no findings to write, so closing it would only yield a hollow
        "no data" block; that case is handled by :func:`should_force_research_search`
        instead, and
      * there is usable gathered evidence to write up. With no evidence we deliberately do
        NOT force a close — normal max-rounds/budget handling applies, preserving an honest
        blocked outcome rather than pushing a premature give-up.

    ``did_real_gather`` defaults to True so existing callers/tests that predate the
    plan-only-loop split keep their original behavior.
    """
    if report_generated:
        return False
    if int(force_close_attempts) >= int(max_attempts):
        return False
    if int(nonprogress_rounds) < int(threshold):
        return False
    if not bool(did_real_gather):
        return False
    return int(gather_excerpt_chars) >= int(min_evidence_chars)


def should_force_research_search(
    *,
    nonprogress_rounds: int,
    threshold: int,
    search_attempts: int,
    max_attempts: int,
    report_generated: bool,
    did_real_gather: bool,
) -> bool:
    """True when a research worker is stuck PLANNING without ever gathering real evidence.

    The complement of :func:`should_force_research_close`. When the worker has looped on
    planning/meta tools to the non-progress threshold but has performed no real
    search/read/corpus call, forcing a CLOSE would only produce an honest "no findings"
    block (there is nothing to write). Instead we force a SEARCH turn: strip the planning
    tools, keep the gather tools, and tell the model to search now. This converts a
    plan-only stall into a real gather, after which the normal close path can deliver.

    Doctrine-pure: we only force the model to *gather*; we never fabricate findings, and
    the attempt cap keeps an honest blocked outcome reachable if the model still cannot
    answer after it has searched.
    """
    if report_generated:
        return False
    if bool(did_real_gather):
        return False
    if int(search_attempts) >= int(max_attempts):
        return False
    return int(nonprogress_rounds) >= int(threshold)


def format_research_force_search_prompt() -> str:
    """Hard SEARCH nudge for a research worker stuck planning without gathering.

    Used on the forced SEARCH turn (planning tools stripped, gather tools kept). The model
    must run a real search instead of generating more sub-questions. The harness never
    writes findings for it — it only forces it to gather real evidence first.
    """
    return (
        "STOP planning — you have repeatedly called plan_research_subquestions (or other "
        "planning/reflection tools) without ever running a real search, so you have no "
        "findings to report. Do NOT call plan_research_subquestions, reflect_research_state, "
        "or suggest_follow_up_queries this turn. Call deep_web_search NOW with a concrete, "
        "specific query drawn from your plan, then read the most relevant results. You must "
        "gather real evidence before you attempt to write findings or emit a blocked handoff."
    )


def format_research_force_close_prompt(*, gather_evidence_excerpt: str = "") -> str:
    """Hard CLOSE nudge for a looping research worker — forces the model to deliver.

    Mirrors the proven council CLOSE mechanism (tool-less round + explicit instruction)
    but for general research workers. The model must produce its findings as prose plus a
    trailing kanban-result fence, or emit a deliberate blocked handoff with a concrete
    reason. The harness does not write the answer for it.
    """
    evidence = ""
    excerpt = " ".join(str(gather_evidence_excerpt or "").split())
    if excerpt:
        evidence = f"\n\nEvidence you have already gathered (use this — do not search again):\n{excerpt[:1800]}"
    return (
        "STOP researching — you have already gathered enough to answer, and you have been "
        "repeating gather steps without producing a result. Do NOT call "
        "plan_research_subquestions, deep_web_search, search_web, read_and_analyze_page, or "
        "any other tool this turn. Write your complete findings NOW as visible prose, then "
        "end with a trailing ```kanban-result``` JSON fence whose summary captures the "
        "findings (put source URLs in the summary or metadata). If you genuinely cannot "
        "answer from what you have gathered, emit a kanban-result with "
        "metadata.status=\"blocked\" and a specific, concrete reason — not a generic "
        "\"ran out of budget\" or \"need to initialize research\"." + evidence
    )


def synthesize_kanban_invalid_args_salvage_handoff(
    tool_name: str,
    missing_fields: list[str],
    *,
    task_metadata: dict[str, Any] | None = None,
) -> str:
    """Deterministic blocked handoff when a kanban worker cannot form valid tool args."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    task_kind = str(tm.get("task_kind") or "").strip().lower()
    missing = ", ".join(str(f) for f in (missing_fields or []) if str(f).strip())
    name = str(tool_name or "tool").strip() or "tool"
    if task_kind == "research" and name in {
        "deep_web_search",
        "search_web",
        "search_web_public",
        "plan_research_subquestions",
    }:
        summary = (
            f"Research worker could not execute `{name}` with required arguments ({missing or 'query'})."
        )
        next_action = (
            f"Call `{name}` with a non-empty JSON `{missing or 'query'}` string, then "
            "`save_research_findings` with a dense summary, then end with a trailing "
            "kanban-result fenced JSON block."
        )
        blocker = "missing_research_tool_args"
    else:
        summary = f"Kanban worker could not execute `{name}` — required args missing ({missing or 'unknown'})."
        next_action = (
            f"Retry `{name}` with all required JSON arguments ({missing or 'see tool schema'}), "
            "then finish with a trailing kanban-result fenced JSON block."
        )
        blocker = "missing_tool_args"
    payload = {
        "summary": summary,
        "metadata": {
            "status": "blocked",
            "blockers": [blocker],
            "next_action": next_action,
            "failure_reason": "kanban_invalid_required_args",
            "tools_required_next": [name],
        },
    }
    return _format_kanban_result_message(payload)


def _format_tool_call_loop_hard_stop_message(
    tripped: list[TrippedSignature],
    *,
    anchor_summary: str = "",
    recent_paths: str = "",
    directory_snapshot: str = "",
) -> str:
    """User-visible hard-stop message when parent run exceeds loop threshold."""
    if not tripped:
        return (
            "I stopped because repeated tool-call signatures kept looping after recovery guidance. "
            "Please confirm the next concrete step or provide corrected tool parameters."
        )

    from utils.path_normalize import truncate_path_display

    has_path_bucket = any(path_bucket for _, _, path_bucket, _ in tripped)
    sig_labels = ", ".join(
        f"{n}({','.join(ks) if ks else '-'})x{c}"
        for (n, ks, _pb, c) in tripped
    )

    if not has_path_bucket:
        return (
            "I stopped because the same tool-call signature kept repeating after recovery guidance "
            f"({sig_labels}). Please confirm the next concrete step or provide corrected tool parameters."
        )

    lines: list[str] = []
    for name, keys, path_bucket, count in tripped:
        if path_bucket:
            disp = truncate_path_display(path_bucket)
            lines.append(f"- `{name}` → `{disp}` × {count}")
        else:
            keys_display = ", ".join(keys) if keys else "(no args)"
            lines.append(f"- `{name}` (args: {keys_display}) × {count}")

    msg = (
        "I stopped after repeated tool calls to the same path(s):\n"
        + "\n".join(lines)
    )
    if anchor_summary:
        msg += "\n\nKnown paths from this run:\n" + anchor_summary
    if recent_paths:
        msg += f"\n\nRecent paths tried: {recent_paths}"
    if directory_snapshot:
        msg += f"\n\nDirectory listing:\n{directory_snapshot}"
    msg += (
        "\n\nSuggested next step: use `list_directory` on the parent folder or managed project root, "
        "then read an existing path from tool results above — do not invent paths from skill names."
    )
    return msg


def _format_coordinator_board_open_message(snapshot: str = "") -> str:
    """Honest stop when coordinator parent tries to finish while board is open."""
    base = (
        "Kanban workers are still executing this goal. I'm pausing here instead of "
        "burning rounds polling the board — I'll resume when workers deliver results."
    )
    if snapshot and snapshot.strip():
        return base + "\n\nCurrent board:\n" + snapshot.strip()
    return base


def _format_council_active_message(snapshot: str = "") -> str:
    """Honest stop when Captain tries to answer while council seats are running."""
    base = (
        "Council seats are still deliberating (Researcher → Verifier → Contrarian). "
        "I'm pausing here — I'll synthesize a final answer when the contrarian review completes."
    )
    if snapshot and snapshot.strip():
        return base + "\n\nCouncil status:\n" + snapshot.strip()
    return base


def _format_council_awaiting_synthesis_message(snapshot: str = "") -> str:
    """Honest stop when seats finished but synthesis delivery has not reached chat yet."""
    base = (
        "Council seats have finished but the synthesized answer is not ready yet. "
        "I'm pausing here — a dedicated synthesis turn will follow shortly."
    )
    if snapshot and snapshot.strip():
        return base + "\n\nCouncil status:\n" + snapshot.strip()
    return base


_FUTURE_ACTION_PROMISE_RE = re.compile(
    r"\b(?:i(?:\s+am|\s+will|'m|'ll)|going to|about to)\b.*\b(?:start|create|run|execute|try|attempt|do)\b",
    re.IGNORECASE | re.DOTALL,
)
_FUTURE_ACTION_COMPLETION_DISQUALIFIER_RE = re.compile(
    r"\b(?:done|completed?|finished|resolved|fixed|saved|scheduled)\b",
    re.IGNORECASE,
)


def _looks_like_future_action_promise(text: str) -> bool:
    """Broad catch for prose promises that defer action to a future step."""
    if not isinstance(text, str):
        return False
    s = text.strip()
    if not s:
        return False
    if len(s) > 1800:
        return False
    if _FUTURE_ACTION_COMPLETION_DISQUALIFIER_RE.search(s):
        return False
    return bool(_FUTURE_ACTION_PROMISE_RE.search(s))


