import argparse
import asyncio
import importlib.util
import json
import os
import stat
import subprocess
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest import mock


SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "dispatch.py"
SPEC = importlib.util.spec_from_file_location("agent_task_dispatch_test", SCRIPT)
dispatch = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
sys.modules[SPEC.name] = dispatch
SPEC.loader.exec_module(dispatch)


def catalog_value(command: str = "/bin/true") -> dict:
    targets = {
        "primary": {
            "agent": "runner-primary",
            "model": "model-primary",
            "execution_class": "local_model",
            "selfcheck_required": True,
            "runtime": {
                "command": [command, "--workspace", "{workspace}", "--model", "{model}", "{prompt}"],
                "resume_command": [
                    command,
                    "--resume",
                    "{resume_session}",
                    "--session-dir",
                    "{resume_session_dir}",
                    "{prompt}",
                ],
                "environment": {"TARGET_ID": "{target_id}"},
                "output_format": "jsonl",
                "native_session_monitor": True,
                "session_path": "sessions/{session_id}.jsonl",
            },
        },
        "alternate": {
            "agent": "runner-alternate",
            "model": "model-alternate",
            "execution_class": "cloud_model",
            "runtime": {"command": [command, "{prompt}"]},
        },
    }
    routes = {"worker": {}, "review": {}}
    for stage in routes:
        for lane in ("local", "cloud"):
            for grade in range(1, 11):
                routes[stage][f"{lane}-G{grade:02d}"] = {
                    "candidates": ["primary", "alternate"],
                    "rule_id": f"{stage}-{lane}-{grade:02d}",
                    "reason_codes": ["injected-route"],
                }
    return {"schema_version": "1.0", "targets": targets, "routes": routes}


def write_catalog(root: Path, value: dict | None = None) -> Path:
    path = root / "execution-catalog.json"
    path.write_text(json.dumps(value or catalog_value()), encoding="utf-8")
    return path


def write_plan(root: Path, *, task_name: str = "group/01_task") -> Path:
    directory = root / "agent-task" / task_name
    directory.mkdir(parents=True, exist_ok=True)
    path = directory / "PLAN-cloud-G05.md"
    path.write_text(
        f"<!-- task={task_name} plan=0 tag=API -->\n\n"
        "# Plan\n\n## Modified Files Summary\n\n"
        "| File | Action |\n|---|---|\n| `src/item.txt` | modify |\n",
        encoding="utf-8",
    )
    return path


def task_from_plan(root: Path, plan: Path) -> dispatch.Task:
    directory = plan.parent
    return dispatch.Task(
        name="group/01_task",
        directory=directory,
        plan=plan,
        review=None,
        user_review=None,
        recovery=False,
        index=1,
        write_set={str((root / "src/item.txt").resolve())},
        write_set_known=True,
        plan_hash=dispatch.sha256_file(plan),
    )


class RuntimeCatalogDispatcherTests(unittest.TestCase):
    def setUp(self):
        self.previous_catalog = dispatch.EXECUTION_CATALOG_PATH

    def tearDown(self):
        dispatch.EXECUTION_CATALOG_PATH = self.previous_catalog

    def test_dispatcher_default_timestamps_use_kst(self):
        timestamp = datetime.fromisoformat(dispatch.now_iso())

        self.assertEqual(dispatch.DISPATCHER_TIMEZONE_NAME, "Asia/Seoul")
        self.assertEqual(timestamp.utcoffset(), timedelta(hours=9))
        self.assertRegex(
            dispatch.work_log_now_kst(),
            r"^\d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} KST$",
        )

    def test_agent_spec_is_loaded_from_persisted_catalog_evidence(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            catalog = write_catalog(root)
            plan = write_plan(root)
            dispatch.EXECUTION_CATALOG_PATH = catalog
            selector = dispatch._selector_module()
            decision = selector.select_execution_target(plan, catalog_path=catalog)
            spec = dispatch.agent_spec_from_decision(decision)
        self.assertEqual(spec.target_id, "primary")
        self.assertEqual(spec.cli, "runner-primary")
        self.assertEqual(spec.model, "model-primary")
        self.assertTrue(spec.native_resume)
        self.assertEqual(spec.runtime["command"][0], "/bin/true")

    def test_agent_spec_rejects_catalog_change_after_selection(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            catalog = write_catalog(root)
            plan = write_plan(root)
            selector = dispatch._selector_module()
            decision = selector.select_execution_target(plan, catalog_path=catalog)
            changed = catalog_value()
            changed["targets"]["primary"]["model"] = "changed"
            catalog.write_text(json.dumps(changed), encoding="utf-8")
            with self.assertRaisesRegex(dispatch.ExecutionDecisionError, "변경"):
                dispatch.agent_spec_from_decision(decision)

    def test_command_is_expanded_only_from_runtime_template(self):
        spec = dispatch.AgentSpec(
            "opaque-agent",
            "opaque-model",
            "opaque-agent/opaque-model",
            target_id="opaque-id",
            reasoning_effort="high",
            runtime={
                "command": ["runner", "{workspace}", "{model}", "{reasoning_effort}", "{session_id}", "{attempt_dir}", "{prompt}"],
                "resume_command": [
                    "runner",
                    "resume",
                    "{reasoning_effort}",
                    "{resume_session}",
                    "{resume_session_dir}",
                    "{prompt}",
                ],
            },
        )
        command = dispatch.build_command(
            spec,
            "do work",
            Path("/workspace"),
            "session-1",
            Path("/attempt"),
        )
        resumed = dispatch.build_command(
            spec,
            "continue",
            Path("/workspace"),
            "session-1",
            Path("/attempt"),
            resume_session=Path("/attempt/session.jsonl"),
            resume_session_dir=Path("/attempt"),
        )
        self.assertEqual(command, ["runner", "/workspace", "opaque-model", "high", "session-1", "/attempt", "do work"])
        self.assertEqual(
            resumed,
            [
                "runner",
                "resume",
                "high",
                "/attempt/session.jsonl",
                "/attempt",
                "continue",
            ],
        )

    def test_json_only_reasoning_effort_change_controls_command(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            value = catalog_value()
            primary = value["targets"]["primary"]
            primary["reasoning_effort"] = "low"
            primary["runtime"]["command"].insert(-1, "{reasoning_effort}")
            primary["runtime"]["resume_command"].insert(
                -1, "{reasoning_effort}"
            )
            catalog = write_catalog(root, value)
            plan = write_plan(root)
            selector = dispatch._selector_module()

            low_decision = selector.select_execution_target(
                plan, catalog_path=catalog
            )
            low_spec = dispatch.agent_spec_from_decision(low_decision)
            low_command = dispatch.build_command(
                low_spec,
                "work",
                root,
                "session-1",
                root / "attempt",
            )

            primary["reasoning_effort"] = "high"
            catalog.write_text(json.dumps(value), encoding="utf-8")
            high_decision = selector.select_execution_target(
                plan, catalog_path=catalog
            )
            high_spec = dispatch.agent_spec_from_decision(high_decision)
            high_command = dispatch.build_command(
                high_spec,
                "work",
                root,
                "session-2",
                root / "attempt",
            )

        self.assertIn("low", low_command)
        self.assertNotIn("high", low_command)
        self.assertIn("high", high_command)
        self.assertNotIn("low", high_command)

    def test_preflight_checks_executable_and_optional_probe(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            value = catalog_value("/bin/true")
            value["targets"]["primary"]["runtime"]["preflight_command"] = ["/bin/true", "--check"]
            catalog = write_catalog(root, value)
            dispatch.preflight_execution_catalog(catalog)

    def test_preflight_rejects_missing_command(self):
        with TemporaryDirectory() as tmp:
            catalog = write_catalog(Path(tmp), catalog_value("definitely-missing-command"))
            with self.assertRaisesRegex(dispatch.ExecutionDecisionError, "command not found"):
                dispatch.preflight_execution_catalog(catalog)

    def test_persisted_decision_failover_uses_next_runtime_target(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            catalog = write_catalog(root)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            dispatch.EXECUTION_CATALOG_PATH = catalog
            with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
                store = dispatch.StateStore(root)
                try:
                    initial, first_spec = dispatch.persisted_execution_decision(store, task, stage="worker")
                    failed, second_spec = dispatch.persisted_execution_decision(
                        store,
                        task,
                        stage="worker",
                        transition="failover",
                        failure_class="provider-quota",
                    )
                finally:
                    store.close()
        self.assertEqual(initial["selected"]["target_id"], "primary")
        self.assertEqual(first_spec.model, "model-primary")
        self.assertEqual(failed["selected"]["target_id"], "alternate")
        self.assertEqual(second_spec.model, "model-alternate")
        self.assertNotIn("quota", failed)

    def test_generic_failure_budget_is_persistent_per_target(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            catalog = write_catalog(root)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            dispatch.EXECUTION_CATALOG_PATH = catalog
            with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
                store = dispatch.StateStore(root)
                try:
                    first, _ = dispatch.persisted_execution_decision(
                        store, task, stage="worker"
                    )
                    first_budget = dispatch.GenericFailureBudget.from_decision(
                        store, task, first
                    )
                    self.assertEqual(first_budget.record_failure(), 1)
                    self.assertEqual(first_budget.record_failure(), 2)
                    self.assertEqual(first_budget.record_failure(), 3)
                    second, _ = dispatch.persisted_execution_decision(
                        store,
                        task,
                        stage="worker",
                        transition="failover",
                        failure_class="generic-error",
                    )
                    second_budget = dispatch.GenericFailureBudget.from_decision(
                        store, task, second
                    )
                    self.assertEqual(second_budget.count(), 0)
                    self.assertEqual(first_budget.count(), 3)
                finally:
                    store.close()

    def test_catalog_revision_acceptance_resets_only_stale_worker_state(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            catalog = write_catalog(root)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            dispatch.EXECUTION_CATALOG_PATH = catalog
            with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
                store = dispatch.StateStore(root)
                try:
                    decision, _ = dispatch.persisted_execution_decision(
                        store, task, stage="worker"
                    )
                    state = store.task_state(task)
                    state.update(
                        active_stage="worker",
                        active_locator="/missing/locator.json",
                        recovery_failures={"worker": 2},
                        stage_failure_budgets={
                            f'{decision["work_unit_id"]}|worker': {"count": 2}
                        },
                        generic_failure_budgets={
                            f'{decision["work_unit_id"]}|worker|primary': {"count": 2}
                        },
                    )
                    store.save()
                    accepted = store.accept_catalog_revision(
                        [task],
                        catalog_revision="replacement-revision",
                        catalog_source="/runtime/replacement.json",
                        persist=True,
                    )
                    state = store.task_state(task)
                finally:
                    store.close()
        self.assertEqual(accepted, [task.name])
        self.assertNotIn("worker", state["execution_decisions"])
        self.assertNotIn("worker", state["recovery_failures"])
        self.assertEqual(state["stage_failure_budgets"], {})
        self.assertEqual(state["generic_failure_budgets"], {})
        self.assertIsNone(state["active_stage"])
        self.assertEqual(
            state["catalog_revision_acceptance_history"][-1]["previous_selected"],
            decision["selected"],
        )

    def test_retry_blocked_resets_to_fresh_attempt_without_prior_context(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            catalog = write_catalog(root)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            dispatch.EXECUTION_CATALOG_PATH = catalog
            with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
                store = dispatch.StateStore(root)
                try:
                    decision, _ = dispatch.persisted_execution_decision(store, task, stage="worker")
                    state = store.task_state(task)
                    state.update(
                        blocked="runtime failure",
                        active_stage=None,
                        active_locator=None,
                        blocker_evidence={
                            "role": "worker",
                            "failure_class": "provider-quota",
                            "locator": "/tmp/locator.json",
                            "selected": decision["selected"],
                            "work_unit_id": decision["work_unit_id"],
                        },
                        recovery_failures={"worker": 3},
                        stage_failure_budgets={"unit|worker": {"count": 3}},
                        generic_failure_budgets={"unit|worker|target": {"count": 2}},
                        route_transition_history=[{"transition": "resume"}],
                        retry_failover_pending=True,
                        retry_failover_context={"locator": "/tmp/locator.json"},
                    )
                    counter_key = f"{task.name}|{task.plan_hash}|worker"
                    store.data.setdefault("attempt_counters", {})[counter_key] = 5
                    store.save()
                    store.reset_for_fresh_restart("group")
                    state = store.task_state(task)
                    counters = dict(store.data["attempt_counters"])
                finally:
                    store.close()
        self.assertFalse(state["retry_failover_pending"])
        self.assertIsNone(state["retry_failover_context"])
        self.assertIsNone(state["blocker_evidence"])
        self.assertIsNone(state["blocked"])
        self.assertEqual(state["execution_decisions"], {})
        self.assertEqual(state["route_transition_history"], [])
        self.assertEqual(state["recovery_failures"], {})
        self.assertEqual(state["stage_failure_budgets"], {})
        self.assertEqual(state["generic_failure_budgets"], {})
        self.assertNotIn(counter_key, counters)
        self.assertNotIn("quota_snapshot", state)
        self.assertNotIn("retry_quota_refresh_pending", state)

    def test_runtime_error_classifier_keeps_provider_quota(self):
        failure, evidence = dispatch.classify_failure_with_evidence(
            "HTTP 429 resource exhausted: quota reached"
        )
        self.assertEqual(failure, "provider-quota")
        self.assertIsNotNone(evidence)

    def test_auxiliary_log_diagnostics_reads_only_current_attempt_append(self):
        with TemporaryDirectory() as tmp:
            path = Path(tmp) / "provider.log"
            path.write_text(
                "old error: Usage limit reached for 5 hour\n",
                encoding="utf-8",
            )
            offset = path.stat().st_size
            path.write_text(
                path.read_text(encoding="utf-8")
                + "stream error: Usage limit reached for 5 hour\n"
                + "Aborting non-transient provider quota retry\n",
                encoding="utf-8",
            )

            diagnostics = dispatch.auxiliary_log_diagnostics(path, offset)

        self.assertEqual(len(diagnostics), 1)
        self.assertIn("Usage limit reached", diagnostics[0])

    def test_output_validation_capability_rejection_is_provider_terminal(self):
        failure, evidence = dispatch.classify_failure_with_evidence(
            "no provider supports the required output validation capability"
        )
        self.assertEqual(failure, "provider-connection")
        self.assertIsNotNone(evidence)

    def test_generic_json_terminal_diagnostic_has_no_agent_branch(self):
        diagnostic = dispatch.terminal_diagnostic(
            "opaque-agent",
            "stdout",
            json.dumps({"type": "turn.failed", "error": {"code": 429}}),
        )
        self.assertIn("429", diagnostic or "")
        self.assertIsNone(
            dispatch.terminal_diagnostic(
                "opaque-agent",
                "stdout",
                json.dumps({"type": "message", "text": "quota design notes"}),
            )
        )

    def test_agent_end_terminal_diagnostic_respects_retry_and_stop_reason(self):
        retrying = {
            "type": "agent_end",
            "willRetry": True,
            "messages": [
                {
                    "role": "assistant",
                    "stopReason": "error",
                    "errorMessage": "provider_tunnel_error",
                }
            ],
        }
        failed = {
            **retrying,
            "willRetry": False,
        }
        succeeded = {
            "type": "agent_end",
            "willRetry": False,
            "messages": [{"role": "assistant", "stopReason": "stop"}],
        }

        self.assertIsNone(
            dispatch.terminal_diagnostic(
                "opaque-agent", "stdout", json.dumps(retrying)
            )
        )
        self.assertIn(
            "provider_tunnel_error",
            dispatch.terminal_diagnostic(
                "opaque-agent", "stdout", json.dumps(failed)
            )
            or "",
        )
        self.assertEqual(dispatch.json_agent_terminal_outcome(succeeded), "succeeded")
        self.assertIsNone(
            dispatch.terminal_diagnostic(
                "opaque-agent", "stdout", json.dumps(succeeded)
            )
        )

    def test_agent_end_diagnostic_ignores_historical_context_words(self):
        failed = {
            "type": "agent_end",
            "willRetry": False,
            "messages": [
                {
                    "role": "toolResult",
                    "content": [
                        {
                            "type": "text",
                            "text": "context window token limit max_tokens model unavailable",
                        }
                    ],
                },
                {
                    "role": "assistant",
                    "stopReason": "error",
                    "errorMessage": "502: provider_tunnel_error: recovery_failed",
                },
            ],
        }

        diagnostic = dispatch.terminal_diagnostic(
            "pi", "stdout", json.dumps(failed)
        )
        failure, evidence = dispatch.classify_failure_with_evidence(diagnostic or "")

        self.assertEqual(failure, "provider-connection")
        self.assertIn("provider_tunnel_error", evidence or "")
        self.assertNotIn("max_tokens", diagnostic or "")

    def _invoke_fake_json_event(
        self,
        event: dict | list[dict],
        *,
        stderr_line: str | None = None,
        terminal_success: str = "agent_end",
    ) -> tuple[int, str | None, dict, str]:
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            runner = root / "fake_json_runner.py"
            events = event if isinstance(event, list) else [event]
            runner.write_text(
                "import json\n"
                "import sys\n"
                f"events = {events!r}\n"
                "for event in events:\n"
                "    print(json.dumps(event))\n"
                f"print({stderr_line!r}, file=sys.stderr) if {stderr_line!r} else None\n",
                encoding="utf-8",
            )
            runtime = {
                "command": [sys.executable, str(runner)],
                "output_format": "jsonl",
                "terminal_success": terminal_success,
            }
            agent = dispatch.AgentSpec(
                "fake-json-runner",
                "fake-model",
                "fake-json-runner/fake-model",
                target_id="fake-json-target",
                runtime=runtime,
            )
            with mock.patch.dict(
                os.environ,
                {"XDG_STATE_HOME": str(root / "state")},
            ):
                store = dispatch.StateStore(root)
                try:
                    return_code, failure, locator = asyncio.run(
                        dispatch.invoke(
                            root,
                            store,
                            task,
                            "worker",
                            agent,
                            "fake prompt",
                        )
                    )
                    record = json.loads(locator.read_text(encoding="utf-8"))
                    self.assertEqual(record["cli"], "fake-json-runner")
                    work_log = (
                        root / "agent-task" / "group" / "WORK_LOG.md"
                    ).read_text(encoding="utf-8")
                finally:
                    store.close()
        return return_code, failure, record, work_log

    def test_zero_exit_agent_error_is_not_recorded_as_success(self):
        event = {
            "type": "agent_end",
            "willRetry": False,
            "messages": [
                {
                    "role": "assistant",
                    "stopReason": "error",
                    "errorMessage": "provider_tunnel_error: recovery_failed",
                }
            ],
        }

        return_code, failure, record, work_log = self._invoke_fake_json_event(event)

        self.assertEqual(return_code, 0)
        self.assertEqual(failure, "provider-connection")
        self.assertEqual(record["status"], "failed")
        self.assertNotIn("succeeded:0", work_log)

    def test_stdout_terminal_error_survives_later_stderr_diagnostic(self):
        event = {
            "type": "agent_end",
            "willRetry": False,
            "messages": [
                {
                    "role": "assistant",
                    "stopReason": "error",
                    "errorMessage": "opaque terminal failure",
                }
            ],
        }

        return_code, failure, record, work_log = self._invoke_fake_json_event(
            event,
            stderr_line="cleanup warning",
        )

        self.assertEqual(return_code, 0)
        self.assertEqual(failure, "generic-error")
        self.assertEqual(record["failure_evidence_source"], "fake-json-runner:stdout")
        self.assertNotIn("succeeded:0", work_log)

    def test_required_agent_end_success_event_fails_closed_when_missing(self):
        event = {"type": "message", "text": "partial output only"}

        return_code, failure, record, work_log = self._invoke_fake_json_event(event)

        self.assertEqual(return_code, 0)
        self.assertEqual(failure, "generic-error")
        self.assertEqual(record["failure_source"], "dispatcher-terminal-contract")
        self.assertNotIn("succeeded:0", work_log)

    def test_required_agent_end_success_event_accepts_final_stop(self):
        event = {
            "type": "agent_end",
            "willRetry": False,
            "messages": [{"role": "assistant", "stopReason": "stop"}],
        }

        return_code, failure, record, work_log = self._invoke_fake_json_event(event)

        self.assertEqual(return_code, 0)
        self.assertIsNone(failure)
        self.assertEqual(record["status"], "succeeded")
        self.assertIn("succeeded:0", work_log)

    def test_turn_completed_overrides_recovered_transport_diagnostics(self):
        events = [
            {
                "type": "error",
                "message": "Reconnecting after websocket certificate failure",
            },
            {
                "type": "item.completed",
                "item": {
                    "type": "error",
                    "message": "Falling back from WebSockets to HTTPS transport",
                },
            },
            {"type": "turn.completed", "usage": {"input_tokens": 1}},
        ]

        return_code, failure, record, work_log = self._invoke_fake_json_event(
            events,
            stderr_line="ERROR websocket UnknownIssuer before HTTPS fallback",
            terminal_success="turn_completed",
        )

        self.assertEqual(return_code, 0)
        self.assertIsNone(failure)
        self.assertEqual(record["status"], "succeeded")
        self.assertIn("succeeded:0", work_log)

    def test_turn_failed_remains_terminal_failure(self):
        return_code, failure, record, work_log = self._invoke_fake_json_event(
            [
                {"type": "error", "message": "request failed"},
                {"type": "turn.failed", "error": {"message": "terminal"}},
            ],
            terminal_success="turn_completed",
        )

        self.assertEqual(return_code, 0)
        self.assertEqual(failure, "generic-error")
        self.assertEqual(record["status"], "failed")
        self.assertNotIn("succeeded:0", work_log)

    def test_required_turn_completed_event_fails_closed_when_missing(self):
        return_code, failure, record, work_log = self._invoke_fake_json_event(
            {"type": "message", "text": "partial output only"},
            terminal_success="turn_completed",
        )

        self.assertEqual(return_code, 0)
        self.assertEqual(failure, "generic-error")
        self.assertEqual(record["failure_source"], "dispatcher-terminal-contract")
        self.assertNotIn("succeeded:0", work_log)

    def test_zero_exit_worker_without_implementation_progress_is_generic_error(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
            review.write_text(
                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
                encoding="utf-8",
            )
            task.review = review
            runner = root / "fake_json_runner.py"
            runner.write_text(
                "import json\n"
                "print(json.dumps({'type': 'agent_end', 'willRetry': False, "
                "'messages': [{'role': 'assistant', 'stopReason': 'stop'}]}))\n",
                encoding="utf-8",
            )
            agent = dispatch.AgentSpec(
                "fake-json-runner",
                "fake-model",
                "fake-json-runner/fake-model",
                target_id="fake-json-target",
                completion_selfcheck_on_incomplete=True,
                runtime={
                    "command": [sys.executable, str(runner)],
                    "output_format": "jsonl",
                    "terminal_success": "agent_end",
                },
            )
            with mock.patch.dict(
                os.environ,
                {"XDG_STATE_HOME": str(root / "state")},
            ):
                store = dispatch.StateStore(root)
                try:
                    return_code, failure, locator = asyncio.run(
                        dispatch.invoke(
                            root, store, task, "worker", agent, "fake prompt"
                        )
                    )
                    record = json.loads(locator.read_text(encoding="utf-8"))
                    work_log = (
                        root / "agent-task" / "group" / "WORK_LOG.md"
                    ).read_text(encoding="utf-8")
                finally:
                    store.close()

        self.assertEqual(return_code, 0)
        self.assertEqual(failure, "generic-error")
        self.assertEqual(
            record["failure_source"],
            "dispatcher-worker-completion-contract",
        )
        self.assertEqual(record["status"], "failed")
        self.assertNotIn("succeeded:0", work_log)

    def test_changed_worker_with_incomplete_evidence_can_advance_to_completion_selfcheck(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
            review.write_text(
                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
                encoding="utf-8",
            )
            task.review = review
            claimed = root / "src" / "item.txt"
            claimed.parent.mkdir(parents=True)
            runner = root / "completion_selfcheck_runner.py"
            runner.write_text(
                "import json\n"
                f"open({str(claimed)!r}, 'w', encoding='utf-8').write('changed')\n"
                "print(json.dumps({'type': 'agent_end', 'willRetry': False, "
                "'messages': [{'role': 'assistant', 'stopReason': 'stop'}]}))\n",
                encoding="utf-8",
            )
            agent = dispatch.AgentSpec(
                "fake-json-runner",
                "fake-model",
                "fake-json-runner/fake-model",
                target_id="fake-json-target",
                completion_selfcheck_on_incomplete=True,
                runtime={
                    "command": [sys.executable, str(runner)],
                    "output_format": "jsonl",
                    "terminal_success": "agent_end",
                },
            )
            with mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}):
                store = dispatch.StateStore(root)
                try:
                    return_code, failure, locator = asyncio.run(
                        dispatch.invoke(root, store, task, "worker", agent, "fake prompt")
                    )
                    record = json.loads(locator.read_text(encoding="utf-8"))
                finally:
                    store.close()

        self.assertEqual(return_code, 0)
        self.assertIsNone(failure)
        self.assertEqual(record["status"], "succeeded")
        self.assertTrue(record["completion_selfcheck_on_incomplete"])

    def test_worker_done_with_incomplete_evidence_returns_to_worker_stage(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
            review.write_text(
                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
                encoding="utf-8",
            )
            task.review = review

            stage = dispatch.task_stage(task, {"worker_done": True})

        self.assertEqual(stage, "worker")

    def test_worker_done_with_conditional_selfcheck_routes_incomplete_evidence_to_selfcheck(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            value = catalog_value()
            value["targets"]["primary"][
                "completion_selfcheck_on_incomplete"
            ] = True
            catalog = write_catalog(root, value)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
            review.write_text(
                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
                encoding="utf-8",
            )
            task.review = review
            selector = dispatch._selector_module()
            decision = selector.select_execution_target(plan, catalog_path=catalog)

            stage = dispatch.task_stage(
                task,
                {
                    "worker_done": True,
                    "selfcheck_done": False,
                    "completion_selfcheck_pending": True,
                    "completing_decision": decision,
                },
            )

        self.assertEqual(stage, "selfcheck")

    def test_incomplete_conditional_selfcheck_without_resume_fails_over_to_next_worker(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            value = catalog_value()
            value["targets"]["primary"]["runtime"].pop(
                "native_session_monitor"
            )
            value["targets"]["primary"]["runtime"].pop("resume_command")
            value["targets"]["primary"][
                "completion_selfcheck_on_incomplete"
            ] = True
            catalog = write_catalog(root, value)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
            review.write_text(
                "## Implementation Checklist\n\n- [ ] Implement the task.\n",
                encoding="utf-8",
            )
            task.review = review
            dispatch.EXECUTION_CATALOG_PATH = catalog
            selector = dispatch._selector_module()
            decision = selector.select_execution_target(
                plan, catalog_path=catalog
            )
            with mock.patch.dict(
                os.environ, {"XDG_STATE_HOME": str(root / "state")}
            ):
                store = dispatch.StateStore(root)
                try:
                    store.update_task(
                        task,
                        worker_done=True,
                        selfcheck_done=False,
                        completion_selfcheck_pending=True,
                        completing_decision=decision,
                        execution_decisions={"worker": decision},
                    )
                    with mock.patch.object(
                        dispatch,
                        "run_escalating",
                        new=mock.AsyncMock(
                            return_value=(
                                True,
                                root / "selfcheck-locator.json",
                            )
                        ),
                    ):
                        asyncio.run(dispatch.run_selfcheck(root, store, task))
                    state = store.task_state(task)
                finally:
                    store.close()

        self.assertFalse(state["worker_done"])
        self.assertFalse(state["completion_selfcheck_pending"])
        self.assertIsNone(state["blocked"])
        self.assertEqual(
            state["execution_decisions"]["worker"]["selected"]["target_id"],
            "alternate",
        )

    def test_silent_native_session_is_terminated_and_classified_as_stall(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            runner = root / "silent_runner.py"
            runner.write_text(
                "import time\n"
                "time.sleep(30)\n",
                encoding="utf-8",
            )
            agent = dispatch.AgentSpec(
                "silent-runner",
                "silent-model",
                "silent-runner/silent-model",
                native_resume=True,
                target_id="silent-target",
                runtime={
                    "command": [sys.executable, str(runner)],
                    "native_session_monitor": True,
                    "session_path": "sessions/{session_id}.jsonl",
                },
            )
            with (
                mock.patch.dict(
                    os.environ,
                    {"XDG_STATE_HOME": str(root / "state")},
                ),
                mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
                mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
            ):
                store = dispatch.StateStore(root)
                try:
                    return_code, failure, locator = asyncio.run(
                        dispatch.invoke(
                            root,
                            store,
                            task,
                            "worker",
                            agent,
                            "fake prompt",
                        )
                    )
                    record = json.loads(locator.read_text(encoding="utf-8"))
                finally:
                    store.close()

        self.assertLess(return_code, 0)
        self.assertEqual(failure, "session-stall")
        self.assertEqual(record["status"], "failed")
        self.assertEqual(record["failure_source"], "dispatcher-stall-timeout")
        self.assertEqual(record["termination_initiator"], "dispatcher")
        self.assertGreaterEqual(record["session_stall_seconds"], 0.05)
        self.assertIn("native_silence_inspection", record)

    def test_non_native_active_command_execution_is_not_classified_as_stall(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            runner = root / "command_runner.py"
            runner.write_text(
                "import json, time\n"
                "item = {'id': 'command-1', 'type': 'command_execution'}\n"
                "print(json.dumps({'type': 'item.started', 'item': item}), flush=True)\n"
                "time.sleep(0.15)\n"
                "print(json.dumps({'type': 'item.completed', 'item': item}), flush=True)\n",
                encoding="utf-8",
            )
            agent = dispatch.AgentSpec(
                "command-runner",
                "command-model",
                "command-runner/command-model",
                target_id="command-target",
                runtime={
                    "command": [sys.executable, str(runner)],
                    "output_format": "jsonl",
                },
            )
            with (
                mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}),
                mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
                mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
            ):
                store = dispatch.StateStore(root)
                try:
                    return_code, failure, locator = asyncio.run(
                        dispatch.invoke(root, store, task, "worker", agent, "fake prompt")
                    )
                    record = json.loads(locator.read_text(encoding="utf-8"))
                finally:
                    store.close()

        self.assertEqual(return_code, 0)
        self.assertIsNone(failure)
        self.assertEqual(record["status"], "succeeded")
        self.assertEqual(record["active_command_execution_ids"], [])
        self.assertNotIn("session_stall_seconds", record)

    def test_completed_non_native_command_returns_to_stall_detection(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            runner = root / "completed_command_runner.py"
            runner.write_text(
                "import json, time\n"
                "item = {'id': 'command-1', 'type': 'command_execution'}\n"
                "print(json.dumps({'type': 'item.started', 'item': item}), flush=True)\n"
                "print(json.dumps({'type': 'item.completed', 'item': item}), flush=True)\n"
                "time.sleep(30)\n",
                encoding="utf-8",
            )
            agent = dispatch.AgentSpec(
                "completed-command-runner",
                "command-model",
                "completed-command-runner/command-model",
                target_id="command-target",
                runtime={
                    "command": [sys.executable, str(runner)],
                    "output_format": "jsonl",
                },
            )
            with (
                mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}),
                mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
                mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
            ):
                store = dispatch.StateStore(root)
                try:
                    return_code, failure, locator = asyncio.run(
                        dispatch.invoke(root, store, task, "worker", agent, "fake prompt")
                    )
                    record = json.loads(locator.read_text(encoding="utf-8"))
                finally:
                    store.close()

        self.assertLess(return_code, 0)
        self.assertEqual(failure, "session-stall")
        self.assertEqual(record["active_command_execution_ids"], [])
        self.assertIn("stream_silence_inspection", record)

    def test_session_stall_resume_reuses_runtime_stream_session_id(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            runner = root / "stall_resume_runner.py"
            capture = root / "resume-argv.json"
            runner.write_text(
                "import json, sys, time\n"
                "args = sys.argv[1:]\n"
                "if args and args[0] == 'resume':\n"
                "    capture = args[1]\n"
                "    open(capture, 'w', encoding='utf-8').write(json.dumps(args[2:]))\n"
                "    print(json.dumps({'type': 'message', 'sessionID': args[2], 'text': 'continued'}), flush=True)\n"
                "else:\n"
                "    print(json.dumps({'type': 'step_start', 'sessionID': 'ses-runtime-1'}), flush=True)\n"
                "    time.sleep(30)\n",
                encoding="utf-8",
            )
            agent = dispatch.AgentSpec(
                "opencode",
                "glm-5.2",
                "opencode/glm-5.2",
                target_id="opencode-test",
                runtime={
                    "command": [sys.executable, str(runner)],
                    "resume_command": [
                        sys.executable,
                        str(runner),
                        "resume",
                        str(capture),
                        "{resume_session}",
                        "{prompt}",
                    ],
                    "output_format": "jsonl",
                    "session_stall_resume": True,
                },
            )
            with (
                mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}),
                mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
                mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
            ):
                store = dispatch.StateStore(root)
                try:
                    first_rc, first_failure, first_locator = asyncio.run(
                        dispatch.invoke(root, store, task, "worker", agent, "start")
                    )
                    first_record = json.loads(first_locator.read_text(encoding="utf-8"))
                    self.assertEqual(
                        dispatch.stall_resume_session_id(agent, first_locator),
                        "ses-runtime-1",
                    )
                    second_rc, second_failure, second_locator = asyncio.run(
                        dispatch.invoke(
                            root,
                            store,
                            task,
                            "worker",
                            agent,
                            dispatch.continuation_prompt(
                                task,
                                "worker",
                                first_locator,
                                resume_same_native_session=True,
                            ),
                            resume_locator=first_locator,
                        )
                    )
                    second_record = json.loads(second_locator.read_text(encoding="utf-8"))
                    resumed_args = json.loads(capture.read_text(encoding="utf-8"))
                finally:
                    store.close()

        self.assertLess(first_rc, 0)
        self.assertEqual(first_failure, "session-stall")
        self.assertEqual(first_record["session_id"], "ses-runtime-1")
        self.assertEqual(first_record["session_id_source"], "runtime-stream")
        self.assertEqual(second_rc, 0)
        self.assertIsNone(second_failure)
        self.assertEqual(second_record["session_id"], "ses-runtime-1")
        self.assertEqual(second_record["resume_mode"], "runtime-session-id")
        self.assertEqual(resumed_args[0], "ses-runtime-1")
        self.assertIn("Continue this session", resumed_args[1])
        second_record.update(status="failed", failure_class="session-stall")
        self.assertIsNone(
            dispatch.stall_resume_session_id(agent, second_locator, second_record)
        )

    def test_catalog_source_is_in_runtime_audit_evidence(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            catalog = write_catalog(root)
            plan = write_plan(root)
            selector = dispatch._selector_module()
            decision = selector.select_execution_target(plan, catalog_path=catalog)
            evidence = dispatch.selector_runtime_evidence(decision)
        self.assertEqual(evidence["catalog"]["source"], str(catalog.resolve()))
        self.assertNotIn("quota", evidence)


class GenericDispatcherContractTests(unittest.TestCase):
    def test_official_review_owns_reconstructable_verification_evidence(self):
        common_skills = Path(__file__).parents[2]
        review_skill = (common_skills / "code-review" / "SKILL.md").read_text(
            encoding="utf-8"
        )
        review_template = (
            common_skills / "plan" / "templates" / "review-stub-template.md"
        ).read_text(encoding="utf-8")

        self.assertIn(
            "Verification, defect analysis, and fix selection are review-agent-owned at verdict time.",
            review_skill,
        )
        self.assertIn(
            "Missing or incomplete implementation-recorded verification output alone is not `Required`.",
            review_skill,
        )
        self.assertIn(
            "do not delegate diagnosis or solution selection to the plan skill or implementing agent",
            review_skill,
        )
        self.assertIn(
            "Run applicable required verification and record fresh command/output",
            review_template,
        )
        plan_skill = (common_skills / "plan" / "SKILL.md").read_text(encoding="utf-8")
        self.assertIn(
            "treat the reviewer's closed finding packet as the decision authority",
            plan_skill,
        )
        self.assertIn("contains no worker-owned diagnosis", plan_skill)

    def _implementation_review_errors(self, text: str) -> list[str]:
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            task = task_from_plan(root, plan)
            review = plan.parent / "CODE_REVIEW-cloud-G05.md"
            review.write_text(text, encoding="utf-8")
            task.review = review
            return dispatch.implementation_review_errors(task)

    def test_explicit_blocker_evidence_can_reach_official_review(self):
        errors = self._implementation_review_errors(
            "## Overview\n\n"
            "status=BLOCKED — external runtime input is missing\n\n"
            "## Implementation Checklist\n\n"
            "- [x] Captured available evidence.\n"
            "- [ ] Run the external verification after access is ready.\n\n"
            "## Verification Results\n\n"
            "Resume condition: Register the missing route and provide the named runtime input.\n"
        )

        self.assertEqual(errors, [])

    def test_blocker_placeholder_does_not_bypass_checklist_gate(self):
        errors = self._implementation_review_errors(
            "## Overview\n\n"
            "status=BLOCKED\n\n"
            "## Implementation Checklist\n\n"
            "- [x] Captured available evidence.\n"
            "- [ ] Run the external verification.\n\n"
            "## Verification Results\n\n"
            "Resume condition: <none or exact condition>\n"
        )

        self.assertEqual(errors, ["구현 체크리스트 미완료"])

    def test_verdict_parser_accepts_inline_code_value(self):
        text = "## Code Review Result\n\n- Overall Verdict: `FAIL`\n"

        self.assertEqual(dispatch.verdict_from_text(text), "FAIL")

    def test_invalid_reasoning_effort_is_target_unavailable(self):
        output = (
            '{"type":"error","error":{"data":{"message":'
            '"reasoning_effort must be one of none, low, medium, or high"}}}'
        )

        failure, evidence = dispatch.classify_failure_with_evidence(output)

        self.assertEqual(failure, "model-unavailable")
        self.assertIn("reasoning_effort", evidence or "")

    def test_selfcheck_work_log_uses_worker_plan_artifact(self):
        with TemporaryDirectory() as tmp:
            directory = Path(tmp) / "agent-task" / "group" / "01_task"
            directory.mkdir(parents=True)
            plan = directory / "PLAN-local-G05.md"
            review = directory / "CODE_REVIEW-cloud-G05.md"
            plan.touch()
            review.touch()
            task = dispatch.Task(
                name="group/01_task",
                directory=directory,
                plan=plan,
                review=review,
                user_review=None,
                recovery=False,
            )

            worker = dispatch.work_log_task_name(task, "worker")
            selfcheck = dispatch.work_log_task_name(task, "selfcheck")
            official_review = dispatch.work_log_task_name(task, "review")

        self.assertTrue(worker.endswith("/PLAN-local-G05.md"))
        self.assertEqual(selfcheck, worker)
        self.assertTrue(official_review.endswith("/CODE_REVIEW-cloud-G05.md"))

    def test_parallel_limit_contract(self):
        self.assertEqual(dispatch.validated_max_parallel(0), 0)
        self.assertEqual(dispatch.validated_max_parallel(3), 3)
        with self.assertRaises(ValueError):
            dispatch.validated_max_parallel(-1)

    def test_modified_files_summary_is_canonicalized(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            write_set, diagnostics = dispatch.inspect_write_set(plan, root)
        self.assertEqual(diagnostics, [])
        self.assertEqual(write_set, {str((root / "src/item.txt").resolve())})

    def test_outside_workspace_claim_is_rejected(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            plan.write_text(
                plan.read_text(encoding="utf-8").replace("`src/item.txt`", "`../outside.txt`"),
                encoding="utf-8",
            )
            _, diagnostics = dispatch.inspect_write_set(plan, root)
        self.assertTrue(any("outside" in item.lower() or "workspace" in item.lower() for item in diagnostics))

    def test_selector_evidence_uses_agent_model_fields(self):
        decision = {
            "work_unit_id": "group/01::plan-0::tag-API",
            "selected": {"target_id": "a", "agent": "runner", "model": "model"},
            "candidates": [
                {"candidate_rank": 1, "target_id": "a", "agent": "runner", "model": "model"}
            ],
            "decision": {"rule_id": "rule", "policy_priority": 1, "reason_codes": []},
            "transition": {"trigger": "initial"},
        }
        lines = dispatch.selector_evidence_lines(decision)
        self.assertIn("candidates=#1:runner/model", lines)
        self.assertFalse(any("quota" in line for line in lines))

    def test_validate_plan_mode_does_not_require_catalog(self):
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            plan = write_plan(root)
            completed = subprocess.run(
                [sys.executable, str(SCRIPT), "--workspace", str(root), "--validate-plan", str(plan)],
                capture_output=True,
                text=True,
                env={key: value for key, value in os.environ.items() if key != "AGENT_TASK_EXECUTION_CATALOG"},
                check=False,
            )
        self.assertEqual(completed.returncode, 0, completed.stderr)

    def test_dry_run_uses_bundled_catalog_by_default(self):
        with TemporaryDirectory() as tmp:
            (Path(tmp) / "agent-task").mkdir()
            bin_dir = Path(tmp) / "bin"
            bin_dir.mkdir()
            for executable in ("agy", "claude", "codex", "opencode", "pi"):
                path = bin_dir / executable
                path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
                path.chmod(stat.S_IRWXU)
            env = {
                key: value
                for key, value in os.environ.items()
                if key != "AGENT_TASK_EXECUTION_CATALOG"
            }
            env["XDG_STATE_HOME"] = str(Path(tmp) / "state")
            env["PATH"] = f"{bin_dir}{os.pathsep}{env.get('PATH', '')}"
            completed = subprocess.run(
                [sys.executable, str(SCRIPT), "--workspace", tmp, "--dry-run"],
                capture_output=True,
                text=True,
                env=env,
                check=False,
            )
        self.assertEqual(completed.returncode, 0, completed.stderr)
        self.assertNotIn("missing_execution_catalog", completed.stderr)


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