import ast
import asyncio
import importlib.util
import io
import json
import os
import re
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock


SCRIPT = Path(__file__).parents[1] / "scripts" / "dispatch.py"
loaded = sys.modules.get("agent_task_dispatch")
if loaded is not None:
    dispatch = loaded
else:
    SPEC = importlib.util.spec_from_file_location("agent_task_dispatch", SCRIPT)
    assert SPEC and SPEC.loader
    dispatch = importlib.util.module_from_spec(SPEC)
    sys.modules[SPEC.name] = dispatch
    SPEC.loader.exec_module(dispatch)


def make_test_task(root: Path) -> dispatch.Task:
    plan = root / "PLAN-local-G05.md"
    review = root / "CODE_REVIEW-local-G05.md"
    plan.write_text("<!-- task=test plan=0 tag=TEST -->\n", encoding="utf-8")
    review.write_text("<!-- task=test plan=0 tag=TEST -->\n", encoding="utf-8")
    return dispatch.Task(
        name="test",
        directory=root,
        plan=plan,
        review=review,
        user_review=None,
        recovery=False,
        lane="local",
        grade=5,
    )


class ObservationOutputTest(unittest.TestCase):
    def test_banner_preserves_existing_format_and_nested_task_identity(self):
        buffer = io.StringIO()
        with mock.patch("sys.stdout", buffer):
            dispatch.banner("START", "group/subtask/task_name", ["line 1", "line 2"])
        output = buffer.getvalue()
        expected = (
            "------------------------------------------\n"
            "START: task_name\n"
            "------------------------------------------\n"
            "task=group/subtask/task_name\n"
            "line 1\n"
            "line 2\n"
        )
        self.assertEqual(output, expected)

        buffer_flat = io.StringIO()
        with mock.patch("sys.stdout", buffer_flat):
            dispatch.banner("START", "task_name")
        output_flat = buffer_flat.getvalue()
        expected_flat = (
            "------------------------------------------\n"
            "START: task_name\n"
            "------------------------------------------\n"
        )
        self.assertEqual(output_flat, expected_flat)

    def test_attempt_event_is_one_flushed_stdout_line(self):
        buffer = io.StringIO()
        with mock.patch("sys.stdout", buffer):
            dispatch.attempt_event("[test-prefix]", "event message detail")
        output = buffer.getvalue()
        self.assertEqual(output, "[test-prefix] event message detail\n")

    def test_dispatch_compatibility_aliases_point_to_observation_module(self):
        self.assertEqual(dispatch.SEP, dispatch.observation.SEP)
        self.assertIs(dispatch.banner, dispatch.observation.banner)
        self.assertIs(dispatch.attempt_event, dispatch.observation.attempt_event)

    def test_observation_module_identity_is_reused(self):
        module1 = dispatch.load_sibling_observation_module()
        module2 = dispatch.load_sibling_observation_module()
        self.assertIs(module1, module2)
        self.assertIs(module1, sys.modules["agent_task_dispatcher_observation"])

    def test_dispatch_has_no_direct_stdout_print_calls(self):
        source = SCRIPT.read_text(encoding="utf-8")
        tree = ast.parse(source, filename=str(SCRIPT))
        stdout_prints = []
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                func = node.func
                if isinstance(func, ast.Name) and func.id == "print":
                    is_stderr = False
                    for kw in node.keywords:
                        if kw.arg == "file":
                            val = kw.value
                            if (
                                isinstance(val, ast.Attribute)
                                and isinstance(val.value, ast.Name)
                                and val.value.id == "sys"
                                and val.attr == "stderr"
                            ):
                                is_stderr = True
                                break
                    if not is_stderr:
                        stdout_prints.append(node.lineno)
        self.assertEqual(
            stdout_prints,
            [],
            f"found direct stdout print() calls on lines: {stdout_prints}",
        )


class ObservationInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
    async def test_heartbeat_and_child_output_stay_in_logs_not_user_event_stream(self):
        with tempfile.TemporaryDirectory() as temporary:
            workspace = Path(temporary)
            (workspace / ".git").mkdir()
            task = make_test_task(workspace)
            store = dispatch.StateStore(workspace)
            session_id = "11111111-1111-1111-1111-111111111111"

            def command_for(
                spec,
                prompt,
                cwd,
                actual_session_id,
                attempt_dir,
                resume_session=None,
                resume_session_dir=None,
            ):
                self.assertEqual(actual_session_id, session_id)
                native = attempt_dir / "native-sessions" / f"session_{session_id}.jsonl"
                child = (
                    "from pathlib import Path\n"
                    "import sys,time\n"
                    "path = Path(sys.argv[1])\n"
                    "path.parent.mkdir(parents=True, exist_ok=True)\n"
                    "path.write_text("
                    "'{\"type\":\"session\",\"version\":3,\"id\":\"test\","
                    "\"timestamp\":\"2026-07-25T00:00:00.000Z\","
                    "\"cwd\":\"/tmp/test\"}\\n', encoding='utf-8')\n"
                    "time.sleep(0.05)\n"
                    "print('done', flush=True)\n"
                )
                return [sys.executable, "-c", child, str(native)]

            spec = dispatch.AgentSpec(
                "runtime-agent",
                "runtime-model",
                "runtime-target",
                native_resume=True,
                target_id="runtime-target",
                execution_class="local_model",
                runtime={
                    "command": ["runtime-command", "{prompt}"],
                    "session_path": "native-sessions/session_{session_id}.jsonl",
                    "native_session_monitor": True,
                    "output_format": "text",
                },
            )
            try:
                with (
                    mock.patch.object(dispatch, "build_command", side_effect=command_for),
                    mock.patch.object(dispatch.uuid, "uuid4", return_value=session_id),
                    mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
                    mock.patch("builtins.print") as print_mock,
                ):
                    rc, failure, locator = await dispatch.invoke(
                        workspace, store, task, "review", spec, "Reply briefly."
                    )
            finally:
                store.close()

            self.assertEqual(rc, 0)
            self.assertIsNone(failure)
            record = json.loads(locator.read_text(encoding="utf-8"))
            self.assertTrue(record["native_session_path"].endswith(f"{session_id}.jsonl"))
            self.assertIsInstance(record["native_session_mtime_ns"], int)
            heartbeat = Path(record["heartbeat_log"]).read_text(encoding="utf-8")
            self.assertIn("[heartbeat] 작업중...", heartbeat)
            self.assertIn("native_session=", heartbeat)
            self.assertIn("native_mtime_ns=", heartbeat)
            stream = Path(record["stream_log"]).read_text(encoding="utf-8")
            self.assertIn("[stdout] done", stream)
            self.assertNotIn("[heartbeat]", stream)
            normalized = Path(record["normalized_output_log"]).read_text(
                encoding="utf-8"
            )
            self.assertIn("done", normalized)
            visible_output = "\n".join(
                " ".join(str(value) for value in call.args)
                for call in print_mock.call_args_list
            )
            self.assertIn("locator=", visible_output)
            self.assertNotIn("작업중...", visible_output)
            self.assertNotIn("done", visible_output)


class SkillObservationContractTest(unittest.TestCase):
    def test_dispatcher_owns_observation_and_caller_wakes_only_for_attention(self):
        skill = (
            Path(__file__).parents[1] / "SKILL.md"
        ).read_text(encoding="utf-8")
        self.assertIn("The dispatcher owns deterministic scheduling, recovery", skill)
        self.assertIn("Launch the live dispatcher as one persistent foreground process", skill)
        self.assertIn("Do not count internal helper coroutines as agent slots", skill)
        self.assertIn("actual stream or native-session progress", skill)
        self.assertIn("PID/start-token/process-marker evidence", skill)
        self.assertIn("never queries quota before admission", skill)
        self.assertIn("confirmed quota/rate-limit error advances directly", skill)
        self.assertIn("Use the bundled default catalog", skill)


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