#!/usr/bin/env python3
"""Unit tests for hierarchical helpers (no network)."""

from __future__ import annotations

import json
import os
import sys
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
os.environ["CONFIG_DIR"] = str(ROOT / "config")
sys.path.insert(0, str(ROOT / "router"))

from hierarchical import (  # noqa: E402
    Subtask,
    classify_plan_reply,
    deterministic_checks,
    digest_text,
    find_pending_plan,
    format_plan_confirm,
    parse_plan_json,
    should_run_hierarchical,
    should_verify,
    topological_waves,
    worker_model_for,
)
from agent_hier import (  # noqa: E402
    _first_path_hint,
    context_fill_for_forward,
    format_context_fill,
    inject_plan_context,
    messages_have_tool_activity,
    request_has_tools,
    synthetic_first_tool_completion,
)


class HierarchicalHelpersTest(unittest.TestCase):
    def test_parse_plan_json(self) -> None:
        raw = """
        {
          "subtasks": [
            {"id": "1", "prompt": "do a", "worker_tier": "simple", "depends_on": []},
            {"id": "2", "prompt": "do b", "worker_tier": "medium_code", "depends_on": ["1"]},
            {"id": "3", "prompt": "do c", "worker_tier": "hard", "depends_on": ["2"]}
          ],
          "acceptance": ["works"]
        }
        """
        tasks, acc = parse_plan_json(raw, max_subtasks=5)
        self.assertEqual(len(tasks), 3)
        self.assertEqual(acc, ["works"])
        self.assertEqual(tasks[1].depends_on, ["1"])
        self.assertEqual(tasks[2].worker_tier, "hard")

    def test_parse_plan_caps_subtasks(self) -> None:
        items = [
            {
                "id": str(i),
                "prompt": f"p{i}",
                "worker_tier": "simple",
                "depends_on": [],
            }
            for i in range(1, 10)
        ]
        raw = json.dumps({"subtasks": items, "acceptance": []})
        tasks, _ = parse_plan_json(raw, max_subtasks=5)
        self.assertEqual(len(tasks), 5)

    def test_topological_waves(self) -> None:
        tasks = [
            Subtask("1", "a", "simple", []),
            Subtask("2", "b", "simple", ["1"]),
            Subtask("3", "c", "simple", []),
        ]
        waves = topological_waves(tasks)
        self.assertEqual({t.id for t in waves[0]}, {"1", "3"})
        self.assertEqual([t.id for t in waves[1]], ["2"])

    def test_should_run_hierarchical(self) -> None:
        self.assertTrue(
            should_run_hierarchical(
                tier_value="COMPLEX",
                header="auto",
                quality_mode="auto",
                enabled=True,
            )
        )
        self.assertFalse(
            should_run_hierarchical(
                tier_value="SIMPLE",
                header="auto",
                quality_mode="auto",
                enabled=True,
            )
        )
        self.assertTrue(
            should_run_hierarchical(
                tier_value="SIMPLE",
                header="force",
                quality_mode="auto",
                enabled=True,
            )
        )
        self.assertFalse(
            should_run_hierarchical(
                tier_value="COMPLEX",
                header="off",
                quality_mode="auto",
                enabled=True,
            )
        )
        self.assertFalse(
            should_run_hierarchical(
                tier_value="COMPLEX",
                header="auto",
                quality_mode="economy",
                enabled=True,
            )
        )

    def test_should_verify(self) -> None:
        self.assertFalse(
            should_verify(
                policy="on_fail_or_hard",
                checks_ok=True,
                hard_used=False,
                quality_mode="auto",
            )
        )
        self.assertTrue(
            should_verify(
                policy="on_fail_or_hard",
                checks_ok=False,
                hard_used=False,
                quality_mode="auto",
            )
        )
        self.assertTrue(
            should_verify(
                policy="on_fail_or_hard",
                checks_ok=True,
                hard_used=True,
                quality_mode="auto",
            )
        )
        self.assertTrue(
            should_verify(
                policy="never",
                checks_ok=False,
                hard_used=True,
                quality_mode="max",
            )
        )

    def test_deterministic_checks(self) -> None:
        ok, fail = deterministic_checks({"1": "hello"}, [], hard_used=False)
        self.assertTrue(ok)
        self.assertEqual(fail, [])
        ok2, fail2 = deterministic_checks({"1": ""}, [], hard_used=False)
        self.assertFalse(ok2)
        self.assertEqual(fail2, ["1"])
        ok3, _ = deterministic_checks({"1": "hello"}, [], hard_used=True)
        self.assertFalse(ok3)

    def test_digest_and_worker_map(self) -> None:
        self.assertIn("truncated", digest_text("x" * 1000, 50))
        # hard → DeepSeek (b-complex); quality=max → Max (c-complex)
        self.assertEqual(worker_model_for("hard"), "b-complex")
        self.assertEqual(
            worker_model_for("hard", quality_mode="max"), "c-complex"
        )
        self.assertEqual(worker_model_for("simple"), "a-simple")
        self.assertEqual(worker_model_for("medium_code"), "a-medium-code")

    def test_parse_plan_paths_and_goals(self) -> None:
        raw = json.dumps(
            {
                "subtasks": [
                    {
                        "id": "1",
                        "prompt": "fix router",
                        "worker_tier": "medium_code",
                        "depends_on": [],
                        "paths": [r"C:\Users\alexc\IdeaProjects\eventHub\EventHubAiRouter\router\router.py"],
                        "edit_goal": "bump timeout",
                        "constraints": ["keep SSE"],
                    }
                ],
                "acceptance": ["timeout higher"],
            }
        )
        tasks, acc = parse_plan_json(raw, max_subtasks=3)
        self.assertEqual(len(tasks), 1)
        self.assertTrue(tasks[0].paths[0].endswith("router.py"))
        self.assertEqual(tasks[0].edit_goal, "bump timeout")
        self.assertEqual(tasks[0].constraints, ["keep SSE"])
        self.assertEqual(acc, ["timeout higher"])
        payload = format_plan_confirm(tasks, acc, user_goal="bump")
        self.assertIn("bump timeout", payload)
        pending = find_pending_plan([{"role": "assistant", "content": payload}])
        self.assertIsNotNone(pending)
        self.assertEqual(pending["subtasks"][0]["edit_goal"], "bump timeout")

    def test_executor_never_synthetic_and_no_llama(self) -> None:
        from agent_hier import (
            executor_fallback_completion,
            pick_agent_executor_model,
            should_escalate_executor,
        )
        from rules_loader import load_orchestration

        hier = load_orchestration().get("hierarchical") or {}
        self.assertEqual(hier.get("planner_model"), "novita-planner")
        self.assertEqual(hier.get("agent_executor_model"), "a-medium-code")
        self.assertEqual(hier.get("executor_synthetic_mode"), "kickstart_only")
        self.assertEqual(pick_agent_executor_model({"agent_executor_model": "a-simple"}), "a-medium-code")
        self.assertEqual(pick_agent_executor_model(hier), "a-medium-code")
        stop = executor_fallback_completion(
            plan={"subtasks": [{"prompt": "edit x", "paths": ["a.py"]}]},
            tools=[
                {
                    "type": "function",
                    "function": {"name": "read_file", "parameters": {"type": "object"}},
                }
            ],
            model="a-medium-code",
            messages=[{"role": "user", "content": "go"}],
            cfg={"executor_synthetic_mode": "never"},
            reason="litellm_408",
        )
        choice = (stop.get("choices") or [{}])[0]
        self.assertEqual(choice.get("finish_reason"), "stop")
        msg = choice.get("message") or {}
        self.assertFalse(msg.get("tool_calls"))
        # escalate after edit failures
        fail_msgs = [
            {
                "role": "tool",
                "content": "Error: expected struct Edit",
                "tool_call_id": "1",
            },
            {
                "role": "tool",
                "content": "Error: expected struct Edit",
                "tool_call_id": "2",
            },
        ]
        self.assertTrue(
            should_escalate_executor(
                fail_msgs, {"agent_escalate_after_edit_failures": 2}
            )
        )
        self.assertTrue(hier.get("executor_escalate_after_read", False))
        self.assertEqual(int(hier.get("executor_tool_loop_timeout_sec") or 0), 45)
        yml = r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\ift\traefik\dynamic_conf.yml"
        read_ok = [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "r1",
                        "type": "function",
                        "function": {
                            "name": "read_file",
                            "arguments": json.dumps({"path": yml}),
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "tool_call_id": "r1",
                "content": "http:\n  routers:\n    api:\n      rule: Host(`x`)\n",
            },
        ]
        self.assertTrue(
            should_escalate_executor(
                read_ok, {"executor_escalate_after_read": True}
            )
        )
        from agent_hier import (
            force_edit_after_read_completion,
            sanitize_outbound_messages,
            should_force_edit_after_read,
        )

        self.assertTrue(
            should_force_edit_after_read(
                read_ok, {"executor_force_edit_after_read": True}
            )
        )
        # After an edit was already issued — never force another (edit loop bug)
        after_edit = read_ok + [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "e1",
                        "type": "function",
                        "function": {
                            "name": "edit_file",
                            "arguments": json.dumps(
                                {
                                    "path": yml,
                                    "edits": [
                                        {"old_text": "x", "new_text": "y"}
                                    ],
                                }
                            ),
                        },
                    }
                ],
            },
            {"role": "tool", "tool_call_id": "e1", "content": "ok"},
        ]
        self.assertFalse(
            should_force_edit_after_read(
                after_edit, {"executor_force_edit_after_read": True}
            )
        )
        from agent_hier import should_stop_after_edit, stop_after_edit_completion

        self.assertTrue(should_stop_after_edit(after_edit, {}))
        done = stop_after_edit_completion(model="b-complex", messages=after_edit)
        self.assertEqual(done["choices"][0]["finish_reason"], "stop")
        self.assertIn("Готово", done["choices"][0]["message"]["content"])
        forced = force_edit_after_read_completion(
            plan={"subtasks": [{"prompt": "traefik", "paths": [yml]}]},
            tools=[
                {
                    "type": "function",
                    "function": {"name": "read_file", "parameters": {"type": "object"}},
                },
                {
                    "type": "function",
                    "function": {"name": "edit_file", "parameters": {"type": "object"}},
                },
            ],
            model="b-complex",
            messages=read_ok,
        )
        self.assertEqual(
            forced["choices"][0]["message"]["tool_calls"][0]["function"]["name"],
            "edit_file",
        )
        dirty = [
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": "r1",
                        "type": "function",
                        "function": {"name": "read_file", "arguments": {"path": yml}},
                    }
                ],
                "reasoning_content": "secret",
            },
            {"role": "tool", "tool_call_id": "orphan", "content": "x"},
            {"role": "tool", "tool_call_id": "r1", "content": "ok"},
        ]
        clean = sanitize_outbound_messages(dirty)
        self.assertEqual(clean[0]["content"], "")
        self.assertNotIn("reasoning_content", clean[0])
        self.assertIsInstance(clean[0]["tool_calls"][0]["function"]["arguments"], str)
        self.assertEqual(len([m for m in clean if m.get("role") == "tool"]), 1)

        from agent_hier import (
            compress_messages_for_tool_loop,
            rewrite_redundant_reread_completion,
        )

        fat = "x" * 9000
        compressed = compress_messages_for_tool_loop(
            [{"role": "tool", "tool_call_id": "r1", "content": fat}],
            newest_tool_result_chars=3500,
        )
        self.assertLessEqual(len(compressed[0]["content"]), 3520)
        self.assertIn("TRUNCATED", compressed[0]["content"])
        tools = [
            {
                "type": "function",
                "function": {"name": "read_file", "parameters": {"type": "object"}},
            },
            {
                "type": "function",
                "function": {"name": "edit_file", "parameters": {"type": "object"}},
            },
        ]
        reread = {
            "choices": [
                {
                    "finish_reason": "tool_calls",
                    "message": {
                        "role": "assistant",
                        "tool_calls": [
                            {
                                "id": "r2",
                                "type": "function",
                                "function": {
                                    "name": "read_file",
                                    "arguments": json.dumps({"path": yml}),
                                },
                            }
                        ],
                    },
                }
            ]
        }
        rewritten = rewrite_redundant_reread_completion(
            reread,
            messages=read_ok,
            plan={
                "subtasks": [
                    {
                        "prompt": "add comment",
                        "paths": [yml],
                        "edit_goal": "marker",
                    }
                ]
            },
            tools=tools,
            model="b-complex",
            cfg={"executor_rewrite_reread": True},
        )
        name = rewritten["choices"][0]["message"]["tool_calls"][0]["function"]["name"]
        self.assertEqual(name, "edit_file")
        # Root docker-compose.yml does not exist in EventHubDevOps
        from agent_hier import _is_bogus_devops_path

        self.assertTrue(
            _is_bogus_devops_path(
                r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\docker-compose.yml"
            )
        )
        self.assertTrue(
            _is_bogus_devops_path(
                r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\ift\traefik\traefik.yml"
            )
        )
        self.assertFalse(
            _is_bogus_devops_path(
                r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\ift\docker-compose.core.yml"
            )
        )
        self.assertFalse(
            _is_bogus_devops_path(
                r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\ift\traefik\dynamic_conf.yml"
            )
        )
        bogus_read = {
            "choices": [
                {
                    "finish_reason": "tool_calls",
                    "message": {
                        "role": "assistant",
                        "tool_calls": [
                            {
                                "id": "r3",
                                "type": "function",
                                "function": {
                                    "name": "read_file",
                                    "arguments": json.dumps(
                                        {
                                            "path": r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\docker-compose.yml"
                                        }
                                    ),
                                },
                            }
                        ],
                    },
                }
            ]
        }
        fixed = rewrite_redundant_reread_completion(
            bogus_read,
            messages=read_ok,
            plan={
                "subtasks": [
                    {"prompt": "traefik EventHubDevOps", "paths": [yml]}
                ]
            },
            tools=tools,
            model="b-complex",
            cfg={"executor_rewrite_reread": True},
        )
        # Already have traefik body → must edit, not chase root compose
        self.assertEqual(
            fixed["choices"][0]["message"]["tool_calls"][0]["function"]["name"],
            "edit_file",
        )


    def test_plan_confirm_helpers(self) -> None:
        tasks = [Subtask("1", "fix auth", "simple", [])]
        text = format_plan_confirm(tasks, ["ok"], user_goal="fix login")
        self.assertIn("ожидает утверждения", text)
        self.assertIn("<!--hier-plan-v1:", text)
        pending = find_pending_plan(
            [{"role": "assistant", "content": text}]
        )
        self.assertIsNotNone(pending)
        self.assertEqual(pending["subtasks"][0]["prompt"], "fix auth")
        self.assertEqual(classify_plan_reply("ok"), "approve")
        self.assertEqual(classify_plan_reply("отмена"), "cancel")
        self.assertEqual(classify_plan_reply("добавь шаг про тесты"), "amend")

    def test_agent_tools_helpers(self) -> None:
        self.assertTrue(request_has_tools({"tools": [{"type": "function"}]}))
        self.assertFalse(request_has_tools({"tools": []}))
        self.assertTrue(
            messages_have_tool_activity(
                [{"role": "tool", "content": "done", "tool_call_id": "1"}]
            )
        )
        self.assertFalse(messages_have_tool_activity([{"role": "user", "content": "hi"}]))
        msgs = inject_plan_context(
            [{"role": "user", "content": "go"}],
            {"subtasks": [{"id": "1", "prompt": "edit x", "worker_tier": "simple"}]},
        )
        self.assertEqual(msgs[0]["role"], "system")
        self.assertIn("edit x", msgs[0]["content"])

    def test_devops_path_hint_absolute(self) -> None:
        plan = {
            "subtasks": [
                {
                    "id": "1",
                    "prompt": "Править EventHubDevOps Traefik ift dynamic_conf.yml",
                    "worker_tier": "medium_code",
                }
            ]
        }
        tools = [
            {
                "type": "function",
                "function": {"name": "read_file", "parameters": {"type": "object"}},
            }
        ]
        messages = [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "1",
                        "type": "function",
                        "function": {
                            "name": "list_directory",
                            "arguments": json.dumps(
                                {
                                    "path": r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps"
                                }
                            ),
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "tool_call_id": "1",
                "content": "ift\nstage\nREADME.md",
            },
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "2",
                        "type": "function",
                        "function": {
                            "name": "read_file",
                            "arguments": json.dumps(
                                {
                                    "path": r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\src"
                                }
                            ),
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "tool_call_id": "2",
                "content": r"Path not found: C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\src",
            },
        ]
        hint = _first_path_hint(plan, tools, messages)
        self.assertTrue(
            hint.lower().replace("/", "\\").endswith("ift\\traefik\\dynamic_conf.yml")
        )
        self.assertRegex(hint, r"^[A-Za-z]:\\")
        self.assertIn("EventHubDevOps", hint)
        self.assertNotIn("\\src", hint.lower().replace("/", "\\"))
        synth = synthetic_first_tool_completion(
            plan=plan, tools=tools, model="a-medium-code", messages=messages
        )
        args = json.loads(
            synth["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"]
        )
        self.assertEqual(args["path"], hint)
        minimal = inject_plan_context([], plan, minimal=True)
        self.assertIn(r"ift\traefik\dynamic_conf.yml", minimal[0]["content"])
        self.assertIn("NO src/", minimal[0]["content"])

    def test_synth_does_not_reread_same_yml(self) -> None:
        plan = {
            "subtasks": [
                {
                    "id": "1",
                    "prompt": "EventHubDevOps Traefik dynamic_conf.yml IFT links",
                    "worker_tier": "medium_code",
                }
            ]
        }
        yml = r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\ift\traefik\dynamic_conf.yml"
        tools = [
            {
                "type": "function",
                "function": {"name": "read_file", "parameters": {"type": "object"}},
            },
            {
                "type": "function",
                "function": {"name": "edit_file", "parameters": {"type": "object"}},
            },
        ]
        messages = [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "r1",
                        "type": "function",
                        "function": {
                            "name": "read_file",
                            "arguments": json.dumps({"path": yml}),
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "tool_call_id": "r1",
                "content": (
                    "tls:\n"
                    "  stores:\n"
                    "    default:\n"
                    "      defaultCertificate:\n"
                    "        certFile: /etc/traefik/certs/traefik.crt\n"
                    "        keyFile: /etc/traefik/certs/traefik.key\n"
                    "http:\n"
                    "  routers:\n"
                    "    api:\n"
                    "      rule: Host(`api.ift.eventhub.local`)\n"
                    "      service: api\n"
                ),
            },
        ]
        synth = synthetic_first_tool_completion(
            plan=plan, tools=tools, model="a-medium-code", messages=messages
        )
        tc = synth["choices"][0]["message"]["tool_calls"][0]
        self.assertEqual(tc["function"]["name"], "edit_file")
        args = json.loads(tc["function"]["arguments"])
        self.assertEqual(args["path"], yml)
        self.assertIn("edits", args)
        self.assertIsInstance(args["edits"], list)
        self.assertIn("old_text", args["edits"][0])
        self.assertIn("new_text", args["edits"][0])
        self.assertNotEqual(args["edits"][0].get("old_text"), "http:")
        self.assertGreaterEqual(args["edits"][0]["old_text"].count("\n"), 2)
        self.assertIn("calentiq-ift-services-links", args["edits"][0]["new_text"])
        # Failed edit → retry with another tool_call (must NOT stop the agent)
        messages_fail = messages + [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "e1",
                        "type": "function",
                        "function": {
                            "name": "edit_file",
                            "arguments": json.dumps(args),
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "tool_call_id": "e1",
                "content": 'invalid type: string "Add a middleware", expected struct Edit',
            },
        ]
        synth_fail = synthetic_first_tool_completion(
            plan=plan, tools=tools, model="a-medium-code", messages=messages_fail
        )
        # expected struct Edit → retry edit/tool, not stop
        self.assertEqual(synth_fail["choices"][0]["finish_reason"], "tool_calls")
        self.assertTrue(synth_fail["choices"][0]["message"].get("tool_calls"))
        # mismatch → must re-read (Zed: read the file again)
        messages_mismatch = messages + [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "e2",
                        "type": "function",
                        "function": {
                            "name": "edit_file",
                            "arguments": json.dumps(args),
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "tool_call_id": "e2",
                "content": (
                    "Could not find matching text for edit at index 0. "
                    "The old_text did not match any content in the file. "
                    "Please read the file again to get the current content."
                ),
            },
        ]
        synth_mm = synthetic_first_tool_completion(
            plan=plan, tools=tools, model="a-medium-code", messages=messages_mismatch
        )
        self.assertEqual(
            synth_mm["choices"][0]["message"]["tool_calls"][0]["function"]["name"],
            "read_file",
        )
        mm_args = json.loads(
            synth_mm["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"]
        )
        self.assertEqual(mm_args["path"], yml)
        # After re-read success → edit_file, NEVER another read_file (infinite loop)
        body2 = (
            "tls:\n"
            "  stores:\n"
            "    default:\n"
            "      defaultCertificate:\n"
            "        certFile: /etc/traefik/certs/traefik.crt\n"
            "        keyFile: /etc/traefik/certs/traefik.key\n"
            "http:\n"
            "  routers:\n"
            "    api:\n"
            "      rule: Host(`api.ift.eventhub.local`)\n"
            "      service: api\n"
        )
        messages_after_reread = messages_mismatch + [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "r2",
                        "type": "function",
                        "function": {
                            "name": "read_file",
                            "arguments": json.dumps({"path": yml}),
                        },
                    }
                ],
            },
            {"role": "tool", "tool_call_id": "r2", "content": body2 + "\n<<TRUNCATED>>"},
        ]
        synth_after = synthetic_first_tool_completion(
            plan=plan,
            tools=tools,
            model="a-medium-code",
            messages=messages_after_reread,
        )
        self.assertEqual(
            synth_after["choices"][0]["message"]["tool_calls"][0]["function"]["name"],
            "edit_file",
        )
        # With line-number prefixes stripped, edit still works
        from agent_hier import _strip_read_file_line_numbers

        numbered = "     1\ttls:\n     2\t  stores:\n     3\t    default:\n"
        stripped = _strip_read_file_line_numbers(numbered)
        self.assertEqual(stripped.rstrip("\n"), "tls:\n  stores:\n    default:")

    def test_context_fill_counter(self) -> None:
        line = format_context_fill(
            used_chars=3000, budget_chars=6000, model_window_tokens=32768, label="ctx"
        )
        self.assertIn("3000/6000", line)
        self.assertIn("(50%)", line)
        self.assertIn("[", line)
        forward = {
            "messages": [
                {"role": "system", "content": "x" * 100},
                {"role": "user", "content": "y" * 200},
            ],
            "tools": [{"type": "function", "function": {"name": "read_file"}}],
        }
        ctx_line, meta = context_fill_for_forward(
            forward, {"context_window_tokens": 32768}, budget_chars=6000
        )
        self.assertIn("ctx ", ctx_line)
        self.assertGreater(meta["ctx_used_chars"], 0)
        self.assertEqual(meta["ctx_budget_chars"], 6000)
        self.assertIn("ctx_pct", meta)

    def test_enforce_forward_budget_caps_overfill(self) -> None:
        from agent_hier import enforce_forward_budget, measure_forward_size

        forward = {
            "messages": [
                {"role": "system", "content": "plan " + ("p" * 500)},
                {
                    "role": "assistant",
                    "tool_calls": [
                        {
                            "id": "1",
                            "type": "function",
                            "function": {
                                "name": "read_file",
                                "arguments": json.dumps({"path": "a.yml"}),
                            },
                        }
                    ],
                },
                {"role": "tool", "tool_call_id": "1", "content": "DATA" * 2000},
                {
                    "role": "assistant",
                    "tool_calls": [
                        {
                            "id": "2",
                            "type": "function",
                            "function": {
                                "name": "read_file",
                                "arguments": json.dumps({"path": "b.yml"}),
                            },
                        }
                    ],
                },
                {"role": "tool", "tool_call_id": "2", "content": "MORE" * 2000},
            ],
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": f"tool_{i}",
                        "description": "d" * 40,
                        "parameters": {"type": "object", "properties": {"path": {"type": "string"}}},
                    },
                }
                for i in range(10)
            ],
        }
        before = measure_forward_size(forward)["total_chars"]
        self.assertGreater(before, 6000)
        capped = enforce_forward_budget(forward, budget_chars=6000)
        after = measure_forward_size(capped)["total_chars"]
        self.assertLessEqual(after, 6000)
        self.assertEqual(capped["messages"][0]["role"], "system")

    def test_slim_edit_file_schema_preserves_edit_struct(self) -> None:
        from agent_hier import slim_tools_for_executor

        tools = [
            {
                "type": "function",
                "function": {
                    "name": "edit_file",
                    "description": "edit a file " + ("x" * 200),
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "path": {"type": "string"},
                            "edits": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "old_text": {"type": "string"},
                                        "new_text": {"type": "string"},
                                    },
                                },
                            },
                        },
                    },
                },
            }
        ]
        slim = slim_tools_for_executor(tools, max_tools=5, max_desc_chars=80)
        params = slim[0]["function"]["parameters"]
        edits = params["properties"]["edits"]
        self.assertEqual(edits["type"], "array")
        self.assertEqual(edits["items"]["type"], "object")
        self.assertIn("old_text", edits["items"]["properties"])
        self.assertIn("new_text", edits["items"]["properties"])
        # must NOT be array of strings (that caused expected struct Edit)
        self.assertNotEqual(edits["items"].get("type"), "string")

    def test_synthetic_abort_limits(self) -> None:
        from agent_hier import synthetic_abort_reason, synthetic_first_tool_completion

        yml = r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\ift\traefik\dynamic_conf.yml"
        tools = [
            {
                "type": "function",
                "function": {"name": "read_file", "parameters": {"type": "object"}},
            },
            {
                "type": "function",
                "function": {"name": "edit_file", "parameters": {"type": "object"}},
            },
        ]
        # 3 prior gateway synthetics → hard stop
        msgs = []
        for i in range(3):
            msgs.append(
                {
                    "role": "assistant",
                    "tool_calls": [
                        {
                            "id": f"call_synth_{i:04d}abcd",
                            "type": "function",
                            "function": {
                                "name": "edit_file",
                                "arguments": json.dumps(
                                    {
                                        "path": yml,
                                        "edits": [
                                            {
                                                "old_text": "http:\n  routers:",
                                                "new_text": "http:\n  routers:\n    x:",
                                            }
                                        ],
                                    }
                                ),
                            },
                        }
                    ],
                }
            )
            msgs.append(
                {
                    "role": "tool",
                    "tool_call_id": f"call_synth_{i:04d}abcd",
                    "content": "Could not find matching text for edit at index 0.",
                }
            )
        self.assertIsNotNone(synthetic_abort_reason(msgs, max_synthetic=3))
        stop = synthetic_first_tool_completion(
            plan={"subtasks": [{"prompt": "traefik EventHubDevOps"}]},
            tools=tools,
            model="a-medium-code",
            messages=msgs,
            max_synthetic=3,
        )
        self.assertEqual(stop["choices"][0]["finish_reason"], "stop")
        self.assertIn("Стоп", stop["choices"][0]["message"]["content"])
        self.assertFalse(stop["choices"][0]["message"].get("tool_calls"))

    def test_kickstart_only_stops_mid_tool_loop(self) -> None:
        from agent_hier import executor_fallback_completion

        yml = r"C:\Users\alexc\IdeaProjects\eventHub\EventHubDevOps\ift\traefik\dynamic_conf.yml"
        tools = [
            {
                "type": "function",
                "function": {"name": "read_file", "parameters": {"type": "object"}},
            },
            {
                "type": "function",
                "function": {"name": "edit_file", "parameters": {"type": "object"}},
            },
        ]
        cfg = {
            "executor_synthetic_mode": "kickstart_only",
            "executor_max_synthetic_continues": 1,
        }
        plan = {"subtasks": [{"prompt": "traefik EventHubDevOps dynamic_conf"}]}
        # No tools yet → kickstart read_file
        kick = executor_fallback_completion(
            plan=plan,
            tools=tools,
            model="a-medium-code",
            messages=[],
            cfg=cfg,
            reason="litellm_408",
        )
        self.assertEqual(kick["choices"][0]["finish_reason"], "tool_calls")
        self.assertEqual(
            kick["choices"][0]["message"]["tool_calls"][0]["function"]["name"],
            "read_file",
        )
        # Already in tool loop → after successful read prefer edit_file
        mid = [
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "r1",
                        "type": "function",
                        "function": {
                            "name": "read_file",
                            "arguments": json.dumps({"path": yml}),
                        },
                    }
                ],
            },
            {
                "role": "tool",
                "tool_call_id": "r1",
                "content": (
                    "tls:\n"
                    "  stores:\n"
                    "    default:\n"
                    "      defaultCertificate:\n"
                    "        certFile: /etc/traefik/certs/traefik.crt\n"
                    "        keyFile: /etc/traefik/certs/traefik.key\n"
                    "http:\n"
                    "  routers:\n"
                    "    api:\n"
                    "      rule: Host(`api.ift.eventhub.local`)\n"
                ),
            },
        ]
        stop = executor_fallback_completion(
            plan=plan,
            tools=tools,
            model="a-medium-code",
            messages=mid,
            cfg=cfg,
            reason="litellm_408",
        )
        # After successful read, prefer edit_file over dead-end stop
        self.assertEqual(
            stop["choices"][0]["message"]["tool_calls"][0]["function"]["name"],
            "edit_file",
        )
        args = json.loads(
            stop["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"]
        )
        self.assertEqual(args["path"], yml)


if __name__ == "__main__":
    unittest.main()