#!/usr/bin/env python3
"""Fail if model-facing code reintroduces XML tool-call coaching or XML conversation history.

Native OpenAI tool_calls are mandatory for model steering. XML/channel salvage may
still run server-side in agent_loop_tool_parse / agent_loop_run, but prompts and
history must stay native-first.

Usage:
    python scripts/check_native_tool_call_invariant.py

Exit codes:
    0 — OK
    1 — invariant violated
    2 — script error
"""
from __future__ import annotations

import re
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]

# Files whose string literals are shown to the model (or define that vocabulary).
MODEL_PROMPT_FILES = (
    REPO_ROOT / "agent_tool_format.py",
    REPO_ROOT / "utils/agent_loop_tool_context.py",
    REPO_ROOT / "agent_loop_guardrails.py",
    REPO_ROOT / "subagent_routing.py",
)

# Forbidden substrings in model prompt files (coaching XML as a format).
FORBIDDEN_PROMPT_SUBSTRINGS = (
    "XML fallback only",
    "XML fallback when",
    "Reply with exactly one valid <tool_call",
    "Reply with exactly one valid `<tool_call",
    "emit exactly one <tool_call>",
    "emit exactly one `<tool_call",
    "wrap tools in legacy XML prose",
    "Accepted payload styles:",
    "Could not parse your tool_call markup",
)

# agent_loop_run.py patterns that must not return (XML-history entrainment).
RUN_FORBIDDEN_PATTERNS = (
    re.compile(
        r"_use_openai_tool_message_format\s*=\s*\(\s*not\s+_used_xml_tool_calls\s*\)",
        re.MULTILINE,
    ),
    re.compile(
        r"if\s+not\s+_used_xml_tool_calls\s*:\s*\n\s*assistant_msg\[\"tool_calls\"\]",
        re.MULTILINE,
    ),
    re.compile(
        r"if\s+_used_xml_tool_calls\s*:\s*\n\s*messages\.append\(\{\"role\":\s*\"user\"",
        re.MULTILINE,
    ),
)

# agent_loop_run.py patterns that must be present (native path landed).
RUN_REQUIRED_SNIPPETS = (
    "_harvest_stream_choice_native_tool_calls",
    "_use_openai_tool_message_format",
    "_normalize_tool_calls_for_mlx_native_history",
    "native_history_from_xml_salvage",
)

RUN_FILE = REPO_ROOT / "agent_loop_run.py"
TOOL_CONTEXT_FILE = REPO_ROOT / "utils/agent_loop_tool_context.py"

# Kernel model client (core/model.py) — the native path must stay landed:
# stream harvest, one-shot non-stream recovery, and content salvage emitting
# native-shaped ToolCalls. Content salvage now lives in core/_salvage.py (kept
# model.py under the kernel line budget); both files are server-side only and
# must never contain model-facing XML coaching (checked via FORBIDDEN_PROMPT_SUBSTRINGS).
CORE_MODEL_FILE = REPO_ROOT / "core/model.py"
CORE_MODEL_REQUIRED_SNIPPETS = (
    "harvest_message_tool_calls",
    "_try_non_stream_recovery",
    "resolve_content_salvage",
    'recovered_via="content_salvage"',
    'recovered_via = "non_stream"',
)

# Content salvage (core/_salvage.py) — built-in Hermes XML salvage plus the
# native-id normalization; the hook dispatcher must keep emitting call_* ids.
CORE_SALVAGE_FILE = REPO_ROOT / "core/_salvage.py"
CORE_SALVAGE_REQUIRED_SNIPPETS = (
    "salvage_xml_tool_calls",
    "normalize_call_ids",
    "resolve_content_salvage",
)


def _read(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except OSError as exc:
        print(f"ERROR: cannot read {path}: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc


def check_prompt_files() -> list[str]:
    errors: list[str] = []
    for path in MODEL_PROMPT_FILES:
        if not path.is_file():
            errors.append(f"missing expected file: {path.relative_to(REPO_ROOT)}")
            continue
        text = _read(path)
        for needle in FORBIDDEN_PROMPT_SUBSTRINGS:
            if needle in text:
                errors.append(f"{path.relative_to(REPO_ROOT)}: forbidden prompt substring: {needle!r}")
        if path == TOOL_CONTEXT_FILE:
            if "base = TOOL_REMINDER_XML" in text:
                errors.append(
                    f"{path.relative_to(REPO_ROOT)}: build_tool_format_reminder must not assign TOOL_REMINDER_XML"
                )
    return errors


def check_agent_loop_run() -> list[str]:
    errors: list[str] = []
    if not RUN_FILE.is_file():
        return [f"missing {RUN_FILE.relative_to(REPO_ROOT)}"]
    text = _read(RUN_FILE)
    for pat in RUN_FORBIDDEN_PATTERNS:
        if pat.search(text):
            errors.append(
                f"{RUN_FILE.relative_to(REPO_ROOT)}: forbidden pattern (XML history entrainment): {pat.pattern[:60]}…"
            )
    for snippet in RUN_REQUIRED_SNIPPETS:
        if snippet not in text:
            errors.append(
                f"{RUN_FILE.relative_to(REPO_ROOT)}: required native-tool snippet missing: {snippet!r}"
            )
    return errors


def check_core_model() -> list[str]:
    errors: list[str] = []
    if not CORE_MODEL_FILE.is_file():
        return [f"missing {CORE_MODEL_FILE.relative_to(REPO_ROOT)}"]
    text = _read(CORE_MODEL_FILE)
    for snippet in CORE_MODEL_REQUIRED_SNIPPETS:
        if snippet not in text:
            errors.append(
                f"{CORE_MODEL_FILE.relative_to(REPO_ROOT)}: required native-tool snippet missing: {snippet!r}"
            )
    for needle in FORBIDDEN_PROMPT_SUBSTRINGS:
        if needle in text:
            errors.append(
                f"{CORE_MODEL_FILE.relative_to(REPO_ROOT)}: forbidden prompt substring: {needle!r}"
            )
    return errors


def check_core_salvage() -> list[str]:
    errors: list[str] = []
    if not CORE_SALVAGE_FILE.is_file():
        return [f"missing {CORE_SALVAGE_FILE.relative_to(REPO_ROOT)}"]
    text = _read(CORE_SALVAGE_FILE)
    for snippet in CORE_SALVAGE_REQUIRED_SNIPPETS:
        if snippet not in text:
            errors.append(
                f"{CORE_SALVAGE_FILE.relative_to(REPO_ROOT)}: required native-tool snippet missing: {snippet!r}"
            )
    for needle in FORBIDDEN_PROMPT_SUBSTRINGS:
        if needle in text:
            errors.append(
                f"{CORE_SALVAGE_FILE.relative_to(REPO_ROOT)}: forbidden prompt substring: {needle!r}"
            )
    return errors


def main() -> int:
    errors = (
        check_prompt_files()
        + check_agent_loop_run()
        + check_core_model()
        + check_core_salvage()
    )
    if errors:
        print("Native tool-call invariant FAILED:\n", file=sys.stderr)
        for err in errors:
            print(f"  - {err}", file=sys.stderr)
        print(
            "\nSee .cursor/rules/native-tool-calls-invariant.mdc\n",
            file=sys.stderr,
        )
        return 1
    print("Native tool-call invariant OK")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
