import asyncio from datetime import datetime, timezone, timedelta import importlib.util import json import os import re import signal import subprocess import sys import tempfile import time import unittest import uuid from pathlib import Path from types import SimpleNamespace from unittest import mock SCRIPT = Path(__file__).parents[1] / "scripts" / "dispatch.py" 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 pi_session_jsonl(events, version=dispatch.PI_SESSION_SCHEMA_VERSION): values = [ { "type": "session", "version": version, "id": "test-session", "timestamp": "2026-07-25T00:00:00.000Z", "cwd": "/tmp/test", } ] parent_id = None for index, event in enumerate(events): value = dict(event) value.setdefault("id", f"entry-{index}") value.setdefault("parentId", parent_id) values.append(value) parent_id = value["id"] return "".join(json.dumps(value) + "\n" for value in values) def write_legacy_quota_attempts( runs: Path, task: dispatch.Task, *, cli: str = "claude", model: str = "claude-opus-4-8", reasoning_effort: str | None = "xhigh", dispatcher_sha256: str = "older-dispatcher", ) -> list[Path]: locators = [] event = json.dumps( { "type": "rate_limit_event", "rate_limit_info": {"status": "rejected"}, } ) for attempt_number in range(dispatch.RECOVERY_FAILURE_LIMIT): attempt = runs / f"legacy-attempt-{attempt_number}" attempt.mkdir(parents=True) locator = attempt / "locator.json" locator.write_text( json.dumps( { "task": task.name, "plan_number": dispatch.plan_number(task), "role": "worker", "attempt": attempt_number, "status": "failed", "failure_class": "generic-error", "cli": cli, "model": model, "reasoning_effort": reasoning_effort, "dispatcher_source_sha256": dispatcher_sha256, } ), encoding="utf-8", ) if cli == "agy": (attempt / "stream.log").write_text( "[stdout] AGY request failed\n", encoding="utf-8", ) (attempt / "agy-cli.log").write_text( ( "rpc failed: code = ResourceExhausted " "status=RESOURCE_EXHAUSTED HTTP 429 quota exceeded\n" ), encoding="utf-8", ) else: (attempt / "stream.log").write_text( f"[stdout] {event}\n", encoding="utf-8", ) locators.append(locator) return locators class CommandConstructionTest(unittest.TestCase): def test_agy_print_receives_prompt_before_timeout_option(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) prompt = "Implement the active plan." command = dispatch.build_command( dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)"), prompt, workspace, "test-session", workspace / "attempt", ) self.assertEqual(command[:5], ["agy", "--print", prompt, "--print-timeout", "8h"]) self.assertEqual( command[-2:], ["--log-file", str(workspace / "attempt" / "agy-cli.log")] ) class TaskStageTest(unittest.TestCase): def make_task(self, root: Path, review_text: str = ""): plan = root / "PLAN-local-G05.md" review = root / "CODE_REVIEW-local-G05.md" plan.write_text("\n", encoding="utf-8") review.write_text("\n" + review_text, encoding="utf-8") return dispatch.Task( name="test", directory=root, plan=plan, review=review, user_review=None, recovery=False, lane="local", grade=5, ) @staticmethod def blocking_user_review_text(): return ( "# User Review Required - test\n\n" "## 상태\n\nUSER_REVIEW\n\n" "## 사유\n\n" "- 유형: milestone-lock\n" "- 연결 대상: agent-roadmap/phase/p/milestones/m.md\n\n" "## 차단 근거\n\n" "- 차단 판단 근거: API ownership decision blocks implementation.\n\n" "## 연결 결정 필요\n\n" "- [ ] API ownership 선택\n\n" "## 재개 조건\n\n" "- Milestone 결정 반영 후 재개\n" ) def test_default_or_arbitrary_status_text_does_not_start_review(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task( root, "## 사용자 리뷰 요청\n- 상태: 없음\n" "## unrelated\n- 상태: 확인 필요\n", ) self.assertEqual(dispatch.task_stage(task, {}), "worker") def test_verdict_text_outside_official_section_does_not_start_review(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task( root, "## 검증 결과\n" "명령 출력 예시: 종합 판정: PASS\n", ) self.assertEqual(dispatch.task_stage(task, {}), "worker") def test_exact_official_verdict_section_starts_review_recovery(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task( root, "## 코드리뷰 결과\n" "- **종합 판정**: WARN\n", ) self.assertEqual(dispatch.task_stage(task, {}), "review") def test_only_explicit_user_review_file_stops_the_loop(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root, "- 상태: 없음\n") task.user_review = root / "USER_REVIEW.md" task.user_review.write_text( self.blocking_user_review_text(), encoding="utf-8" ) task.plan = None task.review = None task.recovery = True self.assertEqual(dispatch.task_stage(task, {}), "user-review") def test_user_review_without_blocking_contract_is_state_blocked(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) task.user_review = root / "USER_REVIEW.md" task.user_review.write_text( "## 상태\n\nUSER_REVIEW\n", encoding="utf-8" ) task.plan = None task.review = None task.recovery = True self.assertEqual(dispatch.task_stage(task, {}), "blocked") def test_user_review_with_active_pair_is_state_blocked(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) task.user_review = root / "USER_REVIEW.md" task.user_review.write_text( self.blocking_user_review_text(), encoding="utf-8" ) self.assertEqual(dispatch.task_stage(task, {}), "blocked") def test_user_review_only_directory_without_logs_is_readable(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) directory = workspace / "agent-task" / "group" / "01_gate" directory.mkdir(parents=True) user_review = directory / "USER_REVIEW.md" user_review.write_text( self.blocking_user_review_text(), encoding="utf-8" ) task = dispatch.read_task_directory(workspace, directory) self.assertIsNotNone(task) self.assertEqual(dispatch.task_stage(task, {}), "user-review") def test_user_review_none_values_do_not_form_a_valid_stop(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) task.user_review = root / "USER_REVIEW.md" task.user_review.write_text( "## 상태\n\nUSER_REVIEW\n\n" "## 사유\n\n" "- 유형: milestone-lock\n" "- 연결 대상: 없음\n\n" "## 차단 근거\n\n" "- 차단 판단 근거: 없음\n\n" "## 연결 결정 필요\n\n" "- [ ] 없음\n\n" "## 재개 조건\n\n" "- 없음\n", encoding="utf-8", ) task.plan = None task.review = None task.recovery = True self.assertEqual(dispatch.task_stage(task, {}), "blocked") def test_completed_implementation_checklist_does_not_bypass_worker(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task( root, "- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 채운다.\n", ) self.assertEqual(dispatch.task_stage(task, {}), "worker") def test_pi_worker_success_requires_selfcheck_before_review(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) local_decision = { "work_unit_id": "test::plan-0::tag-TEST", "stage": "worker", "selected": { "adapter": "pi", "target": "iop/ornith:35b", "execution_class": "local_model", "selfcheck_required": True, }, } cloud_decision = { "work_unit_id": "test::plan-0::tag-TEST", "stage": "worker", "selected": { "adapter": "claude", "target": "claude-opus-4-8", "execution_class": "cloud_model", "selfcheck_required": False, }, } self.assertEqual( dispatch.task_stage( task, {"worker_done": True, "selfcheck_done": False, "completing_decision": local_decision}, ), "selfcheck", ) self.assertEqual( dispatch.task_stage( task, {"worker_done": True, "selfcheck_done": True, "completing_decision": local_decision}, ), "review", ) self.assertEqual( dispatch.task_stage( task, {"worker_done": True, "selfcheck_done": False, "completing_decision": cloud_decision}, ), "review", ) def test_local_route_grade_boundaries(self): with tempfile.TemporaryDirectory() as temporary: task = self.make_task(Path(temporary)) expected = { 5: ("pi", "ornith:35b", True), 6: ("pi", "ornith:35b", True), 9: ("claude", "claude-opus-4-8", False), 10: ("claude", "claude-opus-4-8", False), } for grade, (cli, model, local_pi) in expected.items(): with self.subTest(grade=grade): assert task.plan is not None graded_plan = task.plan.with_name(f"PLAN-local-G{grade:02d}.md") task.plan.rename(graded_plan) task.plan = graded_plan task.grade = grade decision = dispatch.select_execution_decision(task, stage="worker") spec = dispatch.agent_spec_from_decision(decision) self.assertEqual(spec.cli, cli) self.assertEqual(spec.model, model) self.assertEqual(spec.local_pi, local_pi) def test_local_g07_g08_route_uses_explicit_kst_boundaries(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) nighttime = datetime(2026, 7, 26, 1, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: task = self.make_task(Path(temporary)) for grade in (7, 8): assert task.plan is not None graded_plan = task.plan.with_name(f"PLAN-local-G{grade:02d}.md") task.plan.rename(graded_plan) task.plan = graded_plan task.grade = grade day_dec = dispatch.select_execution_decision(task, stage="worker", evaluated_at=daytime) self.assertEqual(day_dec["selected"]["adapter"], "agy") self.assertEqual(day_dec["selected"]["target"], "Gemini 3.6 Flash (Medium)") night_dec = dispatch.select_execution_decision(task, stage="worker", evaluated_at=nighttime) self.assertEqual(night_dec["selected"]["adapter"], "pi") self.assertEqual(night_dec["selected"]["target"], "iop/laguna-s:2.1") def test_selfcheck_requires_nonempty_checklist_values(self): with tempfile.TemporaryDirectory() as temporary: task = self.make_task( Path(temporary), "## 구현 항목별 완료 여부\n\n" "| 항목 | 완료 여부 |\n|---|---|\n| TEST-1 | [ ] |\n\n" "## 구현 체크리스트\n\n- [ ] TEST-1\n", ) self.assertEqual( dispatch.implementation_review_errors(task), ["구현 체크리스트 미완료"], ) def test_selfcheck_accepts_any_nonempty_checklist_values(self): with tempfile.TemporaryDirectory() as temporary: task = self.make_task( Path(temporary), "## 구현 항목별 완료 여부\n\n" "| 항목 | 완료 여부 |\n|---|---|\n| TEST-1 | [ ] |\n\n" "## 구현 체크리스트\n\n" "- [x] TEST-1\n" "- [v] TEST-2\n" "- [✅] TEST-3\n", ) self.assertEqual(dispatch.implementation_review_errors(task), []) class CompletingTargetSelfcheckTest(unittest.IsolatedAsyncioTestCase): """Verify selfcheck is determined by the completing decision's execution_class. - Worker success persists the actual completing decision with execution_class. - selfcheck schedules exactly once when execution_class=local_model. - local selfcheck reuses the completing target without re-evaluating selector. - Gemini→Laguna, Laguna→Gemini, cloud completions follow the policy. - Restart does not duplicate selfcheck execution. - Provider-deny guard prevents actual provider calls during tests. """ _WORK_UNIT_ID = "completing_target_test::plan-0::tag-TEST" _CLOUD_CASES = ( ("agy", "Gemini 3.6 Flash (Medium)"), ("claude", "claude-opus-4-8"), ("codex", "gpt-5.6-sol"), ) @classmethod def make_cloud_decision( cls, adapter: str, target: str, execution_class: object = "cloud_model", ) -> dict: return { "work_unit_id": cls._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": adapter, "target": target, "execution_class": execution_class, "selfcheck_required": False, }, } def setUp(self) -> None: super().setUp() self._provider_deny = mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=RuntimeError( "provider invoke must not be called in selfcheck tests" )), ) self._build_command_deny = mock.patch.object( dispatch, "build_command", side_effect=RuntimeError( "build_command must not be called in selfcheck tests" ), ) self._provider_deny.start() self._build_command_deny.start() def tearDown(self) -> None: self._provider_deny.stop() self._build_command_deny.stop() super().tearDown() def make_task(self, workspace: Path, lane: str = "local", grade: int = 8) -> dispatch.Task: directory = workspace / "agent-task" / "completing_target_test" directory.mkdir(parents=True, exist_ok=True) header = f"\n" (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") tasks = dispatch.scan_tasks(workspace, None) return tasks[0] def make_locator(self, workspace: Path, cli: str, model: str) -> Path: attempt = workspace / f"attempt-{cli}" attempt.mkdir(parents=True, exist_ok=True) locator = attempt / "locator.json" locator.write_text( json.dumps({"cli": cli, "model": model}), encoding="utf-8", ) return locator async def test_worker_persists_actual_completing_decision_and_execution_class(self): """Worker success records the completing decision, not the initial one. Simulates a Gemini→Laguna failover where the worker actually completed on Laguna. The persisted completing decision must reflect the actual target (Laguna), not the initial target (Gemini). """ with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: # Initial decision: cloud (Gemini) initial_decision = { "schema_version": "1.0", "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "execution_class": "cloud_model", "selfcheck_required": False, }, "decision": { "rule_id": "test-rule", "policy_priority": 0, "reason_codes": [], "evaluated_at": "2026-07-26T14:00:00+09:00", "timezone": "Asia/Seoul", "time_window": {}, }, "transition": {"trigger": "initial"}, "stage": "worker", "lane": "local", "grade": 8, "candidates": [], "quota": {}, } # Simulate failover: worker completed on Laguna laguna_decision = { "schema_version": "1.0", "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, "decision": { "rule_id": "test-rule", "policy_priority": 0, "reason_codes": [], "evaluated_at": "2026-07-26T14:00:00+09:00", "timezone": "Asia/Seoul", "time_window": {}, }, "transition": {"trigger": "failover"}, "stage": "worker", "lane": "local", "grade": 8, "candidates": [], "quota": {}, } loc_laguna = self.make_locator(workspace, "pi", "laguna-s:2.1") # Persist the completing decision to execution_decisions["worker"] # so _mark_worker_done can find it as the authoritative source. store.task_state(task) # initialize state store.data["tasks"][task.name]["execution_decisions"]["worker"] = laguna_decision store.save() def mock_persisted_execution_decision(store_obj, task_obj, *, stage, **kwargs): if stage == "worker": return laguna_decision, dispatch.AgentSpec( "pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True ) return initial_decision, dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)" ) with ( mock.patch.object( dispatch, "persisted_execution_decision", side_effect=mock_persisted_execution_decision, ), mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, loc_laguna)), ), ): await dispatch.run_worker( workspace, store, task ) state = store.task_state(task) self.assertTrue(state["worker_done"]) self.assertEqual(state["execution_class"], "local_model") self.assertFalse(state["selfcheck_done"]) completing = state["completing_decision"] self.assertEqual(completing["selected"]["adapter"], "pi") self.assertEqual(completing["selected"]["target"], "iop/laguna-s:2.1") self.assertEqual(completing["selected"]["execution_class"], "local_model") self.assertTrue(completing["selected"]["selfcheck_required"]) # Verify stage is selfcheck, not review self.assertEqual(dispatch.task_stage(task, state), "selfcheck") finally: store.close() async def test_local_completing_decision_triggers_selfcheck(self): """execution_class=local_model schedules exactly one selfcheck.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: local_decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } loc_laguna = self.make_locator(workspace, "pi", "laguna-s:2.1") # Manually set worker_done with local completing decision store.update_task( task, worker_done=True, worker_cli="pi", worker_model="laguna-s:2.1", completing_decision=local_decision, execution_class="local_model", selfcheck_done=False, blocked=None, ) state = store.task_state(task) self.assertEqual(dispatch.task_stage(task, state), "selfcheck") self.assertTrue( dispatch.completing_decision_requires_selfcheck(state) ) # Run selfcheck once with ( mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, loc_laguna)), ), mock.patch.object( dispatch, "implementation_review_errors", return_value=[], ), ): await dispatch.run_selfcheck( workspace, store, task ) state2 = store.task_state(task) self.assertTrue(state2["selfcheck_done"]) self.assertEqual(dispatch.task_stage(task, state2), "review") finally: store.close() async def test_cloud_completing_decision_skips_selfcheck(self): """execution_class=cloud_model skips selfcheck entirely.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: cloud_decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "claude", "target": "claude-opus-4-8", "execution_class": "cloud_model", "selfcheck_required": False, }, } store.update_task( task, worker_done=True, worker_cli="claude", worker_model="claude-opus-4-8", completing_decision=cloud_decision, execution_class="cloud_model", selfcheck_done=True, blocked=None, ) state = store.task_state(task) self.assertFalse( dispatch.completing_decision_requires_selfcheck(state) ) self.assertEqual(dispatch.task_stage(task, state), "review") finally: store.close() async def test_selfcheck_reuses_completing_target_no_selector_call(self): """Selfcheck uses the completing decision's target, not re-evaluating selector.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: local_decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } loc_laguna = self.make_locator(workspace, "pi", "laguna-s:2.1") store.update_task( task, worker_done=True, worker_cli="pi", worker_model="laguna-s:2.1", completing_decision=local_decision, execution_class="local_model", selfcheck_done=False, blocked=None, ) selector_calls = [] with ( mock.patch.object( dispatch, "persisted_execution_decision", side_effect=lambda *a, **kw: selector_calls.append(1) or ( {}, dispatch.AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True) ), ), mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, loc_laguna)), ), mock.patch.object( dispatch, "implementation_review_errors", return_value=[], ), ): await dispatch.run_selfcheck( workspace, store, task ) # persisted_execution_decision must NOT be called during selfcheck self.assertEqual(len(selector_calls), 0) state = store.task_state(task) self.assertTrue(state["selfcheck_done"]) finally: store.close() async def test_restart_does_not_duplicate_selfcheck(self): """After restart, already-completed selfcheck is not re-executed.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: local_decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } store.update_task( task, worker_done=True, worker_cli="pi", worker_model="laguna-s:2.1", completing_decision=local_decision, execution_class="local_model", selfcheck_done=True, blocked=None, ) state = store.task_state(task) self.assertEqual(dispatch.task_stage(task, state), "review") # selfcheck is required for local_model but already completed self.assertTrue( dispatch.completing_decision_requires_selfcheck(state) ) self.assertTrue(state["selfcheck_done"]) finally: store.close() async def test_identity_mismatch_fails_closed(self): """Missing or malformed completing decision blocks selfcheck.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: # No completing_decision set store.update_task( task, worker_done=True, worker_cli="pi", worker_model="laguna-s:2.1", selfcheck_done=False, blocked=None, ) with mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock() ) as run_escalating: await dispatch.run_selfcheck( workspace, store, task ) self.assertEqual(run_escalating.await_count, 0) state = store.task_state(task) self.assertIn("completing decision이 없어", state["blocked"]) finally: store.close() async def test_identity_mismatch_decision_blocked_at_scheduler(self): """worker_done=True with missing completing decision blocks at scheduler entry.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: # worker_done=True but no completing_decision store.update_task( task, worker_done=True, worker_cli="pi", worker_model="laguna-s:2.1", selfcheck_done=False, blocked=None, ) state = store.task_state(task) # Should be blocked, not review self.assertEqual(dispatch.task_stage(task, state), "blocked") self.assertFalse( dispatch.completing_decision_requires_selfcheck(state) ) finally: store.close() async def test_identity_mismatch_malformed_decision_blocked(self): """worker_done=True with malformed completing decision blocks at scheduler.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: # completing_decision with invalid schema store.update_task( task, worker_done=True, worker_cli="pi", worker_model="laguna-s:2.1", completing_decision={"selected": None}, selfcheck_done=False, blocked=None, ) state = store.task_state(task) self.assertEqual(dispatch.task_stage(task, state), "blocked") finally: store.close() async def test_canonical_model_command_generation(self): """Selfcheck spec.model is normalized (iop/ prefix stripped) for command generation.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: local_decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } spec = dispatch._spec_from_completing_decision(local_decision) # Model should be normalized (iop/ prefix stripped) self.assertEqual(spec.model, "laguna-s:2.1") # Display should preserve canonical identity self.assertEqual(spec.display, "pi/iop/laguna-s:2.1") # local_pi should be True self.assertTrue(spec.local_pi) # adapter should be pi self.assertEqual(spec.cli, "pi") # Temporarily disable the build_command deny guard for this test self._build_command_deny.stop() try: # Verify build_command generates correct command command = dispatch.build_command( spec, "test prompt", workspace, "test-session", workspace / "attempt", ) self.assertIn("--provider", command) self.assertIn("iop", command) self.assertIn("--model", command) # Model should be laguna-s:2.1 (not iop/laguna-s:2.1) model_idx = command.index("--model") self.assertEqual(command[model_idx + 1], "laguna-s:2.1") finally: self._build_command_deny.start() finally: store.close() async def test_selector_probe_not_invoked_during_selfcheck(self): """Selfcheck does not call selector or quota probe.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: local_decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } loc_laguna = self.make_locator(workspace, "pi", "laguna-s:2.1") store.update_task( task, worker_done=True, worker_cli="pi", worker_model="laguna-s:2.1", completing_decision=local_decision, execution_class="local_model", selfcheck_done=False, blocked=None, ) selector_select_calls = [] quota_probe_calls = [] def mock_select(*args, **kwargs): selector_select_calls.append(1) raise RuntimeError("selector must not be called") def mock_quota_probe(*args, **kwargs): quota_probe_calls.append(1) raise RuntimeError("quota probe must not be called") # Patch the selector module directly selector_module = dispatch._selector_module() with ( mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, loc_laguna)), ), mock.patch.object( dispatch, "implementation_review_errors", return_value=[], ), mock.patch.object( selector_module.policy, "select_policy", side_effect=mock_select, ), mock.patch.object( selector_module, "probe_candidate_quota", side_effect=mock_quota_probe, ), ): await dispatch.run_selfcheck( workspace, store, task ) self.assertEqual(len(selector_select_calls), 0) self.assertEqual(len(quota_probe_calls), 0) state = store.task_state(task) self.assertTrue(state["selfcheck_done"]) finally: store.close() def test_completing_decision_requires_selfcheck_matrix(self): """Matrix: local_model→True, cloud_model→False, missing→False.""" local_state = { "completing_decision": { "selected": { "execution_class": "local_model", }, }, } cloud_state = { "completing_decision": { "selected": { "execution_class": "cloud_model", }, }, } missing_state = {"worker_done": True} empty_selected_state = { "completing_decision": {"selected": {}}, } self.assertTrue( dispatch.completing_decision_requires_selfcheck(local_state) ) self.assertFalse( dispatch.completing_decision_requires_selfcheck(cloud_state) ) self.assertFalse( dispatch.completing_decision_requires_selfcheck(missing_state) ) self.assertFalse( dispatch.completing_decision_requires_selfcheck(empty_selected_state) ) def test_completing_decision_validation_matrix(self): """_completing_decision_is_valid enforces stage, work_unit_id, and schema contract.""" workspace = Path(tempfile.mkdtemp()) (workspace / ".git").mkdir() task = self.make_task(workspace) try: valid_pi = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } # Canonical valid cloud decisions for every adapter via class factory valid_cloud_cases = { adapter: self.make_cloud_decision(adapter, target) for adapter, target in self._CLOUD_CASES } wrong_stage = { "work_unit_id": self._WORK_UNIT_ID, "stage": "review", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } wrong_work_unit = { "work_unit_id": "wrong-task::plan-1::tag-WRONG", "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } pi_with_selfcheck_false = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": False, }, } cloud_with_selfcheck_true = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "claude", "target": "claude-opus-4-8", "execution_class": "cloud_model", "selfcheck_required": True, }, } missing = {} no_selected = {"selected": None} invalid_execution_class = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "invalid", "selfcheck_required": True, }, } self.assertTrue( dispatch._completing_decision_is_valid(task, {"completing_decision": valid_pi}) ) # Every canonical cloud adapter must be accepted as valid for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter, target=target): self.assertTrue( dispatch._completing_decision_is_valid( task, {"completing_decision": valid_cloud_cases[adapter]} ) ) self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": wrong_stage}) ) self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": wrong_work_unit}) ) self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": pi_with_selfcheck_false}) ) self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": cloud_with_selfcheck_true}) ) self.assertFalse( dispatch._completing_decision_is_valid(task, missing) ) self.assertFalse( dispatch._completing_decision_is_valid(task, no_selected) ) self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": invalid_execution_class}) ) # cloud adapter + local_model + selfcheck_required=False must be rejected for every adapter for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter, target=target): invalid = self.make_cloud_decision(adapter, target, "local_model") self.assertFalse( dispatch._completing_decision_is_valid( task, {"completing_decision": invalid} ) ) # non-string selected fields must be rejected (adapter, target, execution_class) non_string_adapter = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": 123, "target": "claude-opus-4-8", "execution_class": "cloud_model", "selfcheck_required": False, }, } self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": non_string_adapter}) ) non_string_target = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "claude", "target": None, "execution_class": "cloud_model", "selfcheck_required": False, }, } self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": non_string_target}) ) non_string_execution_class = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "claude", "target": "claude-opus-4-8", "execution_class": None, "selfcheck_required": False, }, } self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": non_string_execution_class}) ) # empty string selected fields must be rejected empty_adapter = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "", "target": "claude-opus-4-8", "execution_class": "cloud_model", "selfcheck_required": False, }, } self.assertFalse( dispatch._completing_decision_is_valid(task, {"completing_decision": empty_adapter}) ) # direct validator must receive full decision shape, not selected sub-dict for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter): invalid_full = self.make_cloud_decision(adapter, target, "local_model") with self.assertRaises(dispatch.ExecutionDecisionError): dispatch._spec_from_completing_decision(invalid_full) finally: import shutil shutil.rmtree(workspace, ignore_errors=True) def test_spec_from_completing_decision_normalizes_pi_target(self): """_spec_from_completing_decision strips iop/ prefix from model.""" decision = { "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } spec = dispatch._spec_from_completing_decision(decision) self.assertEqual(spec.model, "laguna-s:2.1") self.assertEqual(spec.display, "pi/iop/laguna-s:2.1") self.assertTrue(spec.local_pi) def test_spec_from_completing_decision_rejects_invalid_pi_target(self): """_spec_from_completing_decision rejects Pi target without iop/ prefix.""" decision = { "selected": { "adapter": "pi", "target": "laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } with self.assertRaises(dispatch.ExecutionDecisionError): dispatch._spec_from_completing_decision(decision) def test_spec_from_completing_decision_rejects_non_pi_cloud(self): """_spec_from_completing_decision rejects cloud target with selfcheck_required=True.""" decision = { "selected": { "adapter": "claude", "target": "claude-opus-4-8", "execution_class": "cloud_model", "selfcheck_required": True, }, } with self.assertRaises(dispatch.ExecutionDecisionError): dispatch._spec_from_completing_decision(decision) def test_mark_worker_done_validates_pi_identity(self): """_mark_worker_done rejects Pi decision with mismatched worker model.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } store.task_state(task) # initialize state store.data["tasks"][task.name]["execution_decisions"]["worker"] = decision store.save() # Worker model doesn't match target with self.assertRaises(dispatch.ExecutionDecisionError): dispatch._mark_worker_done( store, task, initial_decision=decision, worker_cli="pi", worker_model="ornith:35b", ) state = store.task_state(task) self.assertFalse(state["worker_done"]) finally: store.close() def test_mark_worker_done_validates_cloud_identity(self): """_mark_worker_done rejects cloud decision with mismatched worker CLI.""" with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "claude", "target": "claude-opus-4-8", "execution_class": "cloud_model", "selfcheck_required": False, }, } store.task_state(task) # initialize state store.data["tasks"][task.name]["execution_decisions"]["worker"] = decision store.save() # Worker CLI doesn't match adapter with self.assertRaises(dispatch.ExecutionDecisionError): dispatch._mark_worker_done( store, task, initial_decision=decision, worker_cli="codex", worker_model="gpt-5.6-sol", ) state = store.task_state(task) self.assertFalse(state["worker_done"]) finally: store.close() def test_mark_worker_done_does_not_fallback_from_malformed_persisted_worker_decision( self, ): """Malformed persisted worker decision blocks without falling back to initial. When execution_decisions["worker"] exists but is malformed (e.g. missing selected block), _mark_worker_done must raise rather than silently reverting to the initial_decision parameter. """ with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: malformed_decision = {"selected": None} valid_initial = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } store.task_state(task) # initialize state store.data["tasks"][task.name]["execution_decisions"]["worker"] = malformed_decision store.save() with self.assertRaises(dispatch.ExecutionDecisionError): dispatch._mark_worker_done( store, task, initial_decision=valid_initial, worker_cli="pi", worker_model="laguna-s:2.1", ) state = store.task_state(task) self.assertFalse(state["worker_done"]) self.assertIsNone(state.get("completing_decision")) finally: store.close() def test_mark_worker_done_rejects_cloud_decision_with_local_execution_class( self, ): """_mark_worker_done rejects cloud adapter + local_model + False for every adapter. Regression: cloud adapter with local_model execution_class and selfcheck_required=False must not be accepted as a valid completing decision. This prevents worker_done from being recorded with a wrong execution_class that would route restart to selfcheck. """ for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: bad_decision = self.make_cloud_decision(adapter, target, "local_model") store.task_state(task) store.data["tasks"][task.name]["execution_decisions"]["worker"] = bad_decision store.save() with self.assertRaises(dispatch.ExecutionDecisionError): dispatch._mark_worker_done( store, task, initial_decision=bad_decision, worker_cli=adapter, worker_model=target, ) state = store.task_state(task) self.assertFalse(state["worker_done"]) finally: store.close() def test_restart_with_malformed_completed_state_blocks_not_selfcheck(self): """Restart with malformed completing decision blocks, does not enter selfcheck. Regression: when persisted completing_decision has non-string selected fields or invalid adapter/class/selfcheck combination, task_stage() must return 'blocked' rather than 'selfcheck'. """ for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: # worker_done=True with a completing decision that has non-string execution_class bad_decision = self.make_cloud_decision(adapter, target, 42) store.update_task( task, worker_done=True, worker_cli=adapter, worker_model=target, completing_decision=bad_decision, execution_class="local_model", selfcheck_done=False, blocked=None, ) state = store.task_state(task) self.assertTrue(state["worker_done"]) # task_stage must not return selfcheck for invalid completing decision stage = dispatch.task_stage(task, state) self.assertNotEqual(stage, "selfcheck") # should be blocked because _completing_decision_is_valid returns False self.assertEqual(stage, "blocked") finally: store.close() def test_restart_with_cloud_local_mismatch_blocks(self): """Restart with cloud adapter + local_model + False blocks for every adapter, not selfcheck. Regression: cloud adapter with local_model execution_class must not route restart to selfcheck. """ for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: bad_decision = self.make_cloud_decision(adapter, target, "local_model") store.update_task( task, worker_done=True, worker_cli=adapter, worker_model=target, completing_decision=bad_decision, execution_class="local_model", selfcheck_done=False, blocked=None, ) state = store.task_state(task) stage = dispatch.task_stage(task, state) self.assertNotEqual(stage, "selfcheck") self.assertEqual(stage, "blocked") finally: store.close() def test_mark_worker_done_validates_cloud_decision_commit_matrix(self): """Valid cloud decision commits worker_done=True, selfcheck_done=True, stage=review. Regression: every cloud adapter (agy/claude/codex) with a valid completing decision must record worker completion and advance to review without selfcheck. This covers the valid half of the adapter × validity × consumption path matrix. """ for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter, target=target): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: valid_decision = self.make_cloud_decision(adapter, target) store.task_state(task) store.data["tasks"][task.name]["execution_decisions"]["worker"] = valid_decision store.save() dispatch._mark_worker_done( store, task, initial_decision=valid_decision, worker_cli=adapter, worker_model=target, ) state = store.task_state(task) self.assertTrue( state["worker_done"], f"{adapter}: worker_done must be True after valid commit", ) self.assertTrue( state["selfcheck_done"], f"{adapter}: selfcheck_done must be True for cloud_model", ) self.assertEqual( state["execution_class"], "cloud_model", f"{adapter}: execution_class must remain cloud_model", ) self.assertEqual( dispatch.task_stage(task, state), "review", f"{adapter}: task_stage must advance to review after valid cloud completion", ) # completing_decision must be persisted with validated shape persisted = state["completing_decision"] self.assertEqual( persisted["selected"]["adapter"], adapter ) self.assertEqual( persisted["selected"]["execution_class"], "cloud_model" ) finally: store.close() def test_restart_valid_cloud_advances_to_review(self): """Restart with valid cloud completing decision goes to review, not selfcheck. Regression: a task that already has worker_done=True with a valid cloud completing decision must resume to review on restart. """ for adapter, target in self._CLOUD_CASES: with self.subTest(adapter=adapter, target=target): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: valid_decision = self.make_cloud_decision(adapter, target) store.update_task( task, worker_done=True, worker_cli=adapter, worker_model=target, completing_decision=valid_decision, execution_class="cloud_model", selfcheck_done=True, blocked=None, ) state = store.task_state(task) self.assertTrue(state["worker_done"]) self.assertTrue(state["selfcheck_done"]) stage = dispatch.task_stage(task, state) self.assertNotEqual(stage, "selfcheck") self.assertEqual( stage, "review", f"{adapter}: restart with valid cloud must go to review", ) finally: store.close() async def test_run_worker_completion_mismatch_blocks_task_without_raise( self, ): """run_worker converts completion validation failure to task-local blocker. When the persisted worker decision's runtime identity does not match the actual worker that completed, run_worker must catch the error, keep worker_done=False, record a task-local blocked reason, and return normally—never propagating ExecutionDecisionError to the scheduler. """ with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: # Persisted decision says Pi/Laguna, but worker actually completed # as cloud/codex — identity mismatch. laguna_decision = { "work_unit_id": self._WORK_UNIT_ID, "stage": "worker", "selected": { "adapter": "pi", "target": "iop/laguna-s:2.1", "execution_class": "local_model", "selfcheck_required": True, }, } loc_codex = self.make_locator(workspace, "codex", "gpt-5.6-sol") store.task_state(task) # initialize state store.data["tasks"][task.name]["execution_decisions"]["worker"] = laguna_decision store.save() def mock_persisted(*a, **kw): return laguna_decision, dispatch.AgentSpec( "pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True ) # run_worker returns normally; does not raise. with mock.patch.object( dispatch, "persisted_execution_decision", side_effect=mock_persisted, ), mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, loc_codex)), ): await dispatch.run_worker( workspace, store, task ) state = store.task_state(task) self.assertFalse(state["worker_done"]) self.assertIsNotNone(state.get("blocked")) self.assertIn("worker completion validation failed", state["blocked"]) # Provider-deny guard must not have been bypassed self.assertIsNone(state.get("active_locator")) finally: store.close() class LegacyWorkLogContractHelpers: @staticmethod def completed_replacements(): return { "### 목표와 범위\n\n- 미작성": ( "### 목표와 범위\n\n- PLAN 범위 구현 및 검증" ), "### 체크포인트\n\n- 기록 없음": ( "### 체크포인트\n\n" "- 2026-07-24T00:01:00Z | 구현 | 완료 | 핵심 경로 수정 | " "evidence=`src/test.go` | next=검증" ), "### 예상 밖 이슈\n\n- 기록 없음": ( "### 예상 밖 이슈\n\n" "- 2026-07-24T00:02:00Z | correctness | 계획 밖 race 가능성 | " "impact=동시성 오류 | action=수정 및 테스트 | disposition=해결" ), "### 검증\n\n- 기록 없음": ( "### 검증\n\n- `go test ./...` - PASS" ), "- 상태: 미작성": "- 상태: 완료", "- 요약: 미작성": "- 요약: 구현 및 검증 완료", "- 완료 항목: 미작성": "- 완료 항목: 계획 체크리스트 전체", "- 변경 파일: 미작성": "- 변경 파일: `src/test.go`", "- 검증: 미작성": "- 검증: `go test ./...` PASS", "- 미해결/후속: 미작성": "- 미해결/후속: 없음", "- 예상 밖 이슈 요약: 미작성": ( "- 예상 밖 이슈 요약: race 가능성 수정 완료" ), "- CODE_REVIEW 동기화: 미작성": "- CODE_REVIEW 동기화: 완료", } def make_completed_log(self, root: Path, task=None): task = task or TaskStageTest().make_task(root) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) execution_id = "test__p0__worker__a00" path = dispatch.append_work_log_attempt( task, execution_id, "worker", spec, root / "locator.json", "2026-07-24T00:00:00+00:00", ) text = path.read_text(encoding="utf-8") for before, after in self.completed_replacements().items(): self.assertIn(before, text) text = text.replace(before, after, 1) path.write_text(text, encoding="utf-8") return task, path, execution_id def test_template_and_attempt_require_checkpoints_and_final_report(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = TaskStageTest().make_task(root) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) execution_id = "test__p0__worker__a00" path = dispatch.append_work_log_attempt( task, execution_id, "worker", spec, root / "locator.json", "2026-07-24T00:00:00+00:00", ) rendered = path.read_text(encoding="utf-8") self.assertNotIn(dispatch.WORK_LOG_TEMPLATE_START, rendered) self.assertIn(f"## 실행 `{execution_id}`", rendered) status, errors = dispatch.work_log_attempt_result(path, execution_id) self.assertIsNone(status) self.assertIn("체크포인트 미작성", errors) self.assertIn("최종 리포트 상태 미작성", errors) def test_completed_attempt_preserves_unexpected_issue_and_runtime_result(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, path, execution_id = self.make_completed_log(root) status, errors = dispatch.work_log_attempt_result(path, execution_id) self.assertEqual(status, "완료") self.assertEqual(errors, []) dispatch.append_work_log_runtime_result( path, execution_id, exit_code=0, failure_class=None, locator=root / "locator.json", ) text = path.read_text(encoding="utf-8") self.assertIn("계획 밖 race 가능성", text) self.assertIn(f"### 런타임 종료 기록 `{execution_id}`", text) self.assertIn("- failure_class: `none`", text) def test_checkpoint_cannot_be_replaced_with_none(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, path, execution_id = self.make_completed_log(root) text = path.read_text(encoding="utf-8") checkpoint = self.completed_replacements()[ "### 체크포인트\n\n- 기록 없음" ] path.write_text( text.replace(checkpoint, "### 체크포인트\n\n- 없음", 1), encoding="utf-8", ) status, errors = dispatch.work_log_attempt_result(path, execution_id) self.assertEqual(status, "완료") self.assertIn("체크포인트 형식 불일치", errors) def test_unexpected_issue_accepts_explicit_none(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, path, execution_id = self.make_completed_log(root) text = path.read_text(encoding="utf-8") unexpected = self.completed_replacements()[ "### 예상 밖 이슈\n\n- 기록 없음" ] path.write_text( text.replace(unexpected, "### 예상 밖 이슈\n\n- 없음", 1), encoding="utf-8", ) status, errors = dispatch.work_log_attempt_result(path, execution_id) self.assertEqual(status, "완료") self.assertEqual(errors, []) def test_code_review_sync_must_be_exactly_complete(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, path, execution_id = self.make_completed_log(root) text = path.read_text(encoding="utf-8") path.write_text( text.replace( "- CODE_REVIEW 동기화: 완료", "- CODE_REVIEW 동기화: 실패", 1, ), encoding="utf-8", ) status, errors = dispatch.work_log_attempt_result(path, execution_id) self.assertEqual(status, "완료") self.assertIn( "최종 리포트 CODE_REVIEW 동기화는 완료여야 한다", errors, ) def test_completed_review_checklist_does_not_depend_on_worker_log(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = TaskStageTest().make_task( root, "- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 " "실제 구현 내용과 검증 출력으로 채운다.\n" "- [x] `WORK_LOG.md` 현재 실행 블록의 체크포인트, 예상 밖 이슈, " "검증, 최종 리포트를 모두 채운다. 이 항목이 완료되기 전에는 " "구현이 완료된 것이 아니다.\n", ) task.plan.write_text( task.plan.read_text(encoding="utf-8") + "## 작업 로그 계약\n", encoding="utf-8", ) self.assertEqual(dispatch.task_stage(task, {}), "review") def test_prompt_requires_checkpoints_unexpected_issues_and_final_report(self): path = Path("/tmp/task/WORK_LOG.md") prompt = dispatch.work_log_prompt(path, "task__p0__worker__a00") self.assertIn("checkpoint after each meaningful phase", prompt) self.assertIn("unexpected issues", prompt) self.assertIn("최종 리포트", prompt) self.assertIn("CODE_REVIEW", prompt) def test_state_loss_skips_execution_id_already_present_in_work_log(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task, _, _ = self.make_completed_log(root) store = mock.Mock() store.next_attempt.side_effect = [0, 1] attempt, execution_id = dispatch.next_execution_identity( store, task, "worker", ) self.assertEqual(attempt, 1) self.assertEqual(execution_id, "test__p0__worker__a01") self.assertEqual(store.next_attempt.call_count, 2) class WorkLogInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): def test_work_log_timestamp_uses_compact_kst_format(self): fixed_kst = datetime(2026, 7, 26, 7, 40, 15, tzinfo=dispatch.KST) with mock.patch.object(dispatch, "datetime") as datetime_mock: datetime_mock.now.return_value = fixed_kst self.assertEqual(dispatch.work_log_now_kst(), "26-07-26 07:40:15") datetime_mock.now.assert_called_once_with(dispatch.KST) self.assertRegex(dispatch.now_iso(), r"\+00:00$") async def test_invoke_writes_dispatcher_owned_milestone_timeline(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) command = [ sys.executable, "-c", ( "import os;" f"assert os.environ.get('{dispatch.AGENT_PROCESS_MARKER_ENV}');" "print('work complete', flush=True)" ), ] spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) try: with mock.patch.object( dispatch, "build_command", return_value=command, ) as build_command: rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 0) self.assertIsNone(failure) record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual( record["work_log"], str((workspace / dispatch.WORK_LOG_NAME).resolve()), ) self.assertTrue( record["agent_process_marker"].startswith( "test__p0__worker__a00__" ) ) self.assertEqual(record["status"], "succeeded") prompt = build_command.call_args.args[1] self.assertEqual(prompt, "Read the plan.") self.assertNotIn("checkpoint after each meaningful phase", prompt) log = (workspace / dispatch.WORK_LOG_NAME).read_text(encoding="utf-8") self.assertIn("Dispatcher-owned execution timeline", log) self.assertRegex(log, r"\| \d+ \| \d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| START \|") self.assertRegex(log, r"\| \d+ \| \d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| FINISH \|") self.assertIn("| START | test | worker | 0 | pi | running |", log) self.assertIn("| FINISH | test | worker | 0 | pi | succeeded:0 |", log) async def test_invoke_does_not_require_model_written_work_log(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) command = [sys.executable, "-c", "print('done without report')"] spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) try: with mock.patch.object( dispatch, "build_command", return_value=command, ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 0) self.assertIsNone(failure) async def test_locator_refresh_failure_does_not_abort_live_model(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) real_write_json = dispatch.write_json failed_once = False def flaky_write_json(path, value): nonlocal failed_once if ( path.name == "locator.json" and value.get("agent_pid") is not None and not failed_once ): failed_once = True raise OSError("transient locator write failure") return real_write_json(path, value) spec = dispatch.AgentSpec( "pi", "ornith:35b", "pi", local_pi=True ) try: with ( mock.patch.object( dispatch, "build_command", return_value=[ sys.executable, "-c", "print('completed', flush=True)", ], ), mock.patch.object( dispatch, "write_json", side_effect=flaky_write_json ), ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Work." ) finally: store.close() self.assertTrue(failed_once) self.assertEqual(rc, 0) self.assertIsNone(failure) record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["status"], "succeeded") self.assertIn("transient locator write failure", record["locator_write_error"]) record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["status"], "succeeded") self.assertNotIn("work_log_contract_errors", record) async def test_existing_milestone_log_is_preserved_and_extended(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) (workspace / dispatch.WORK_LOG_NAME).write_text( "# malformed\n", encoding="utf-8", ) store = dispatch.StateStore(workspace) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) command = [sys.executable, "-c", "print('done')"] try: with mock.patch.object( dispatch, "build_command", return_value=command ) as build_command: rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 0) self.assertIsNone(failure) build_command.assert_called_once() record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["status"], "succeeded") log = (workspace / dispatch.WORK_LOG_NAME).read_text(encoding="utf-8") self.assertIn("# malformed", log) self.assertIn("## Dispatcher Timeline", log) async def test_runtime_log_write_failure_finishes_locator_as_blocked_failure(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) command = [sys.executable, "-c", "print('done')"] spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) try: with ( mock.patch.object( dispatch, "build_command", return_value=command, ), mock.patch.object( dispatch, "append_milestone_event", side_effect=[ workspace / dispatch.WORK_LOG_NAME, OSError("disk full"), ], ), ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 0) self.assertEqual(failure, "work-log-runtime-write") record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["status"], "failed") self.assertEqual(record["failure_class"], "work-log-runtime-write") self.assertEqual(record["work_log_runtime_error"], "disk full") async def test_cancelled_invoke_finishes_locator_and_runtime_record(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) command = [ sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(60)", ] spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) invocation = None try: with mock.patch.object( dispatch, "build_command", return_value=command, ): invocation = asyncio.create_task( dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) ) locator = None for _ in range(100): candidates = list(store.runs.glob("*/locator.json")) if candidates: candidate = candidates[0] record = json.loads( candidate.read_text(encoding="utf-8") ) output = Path(record["output_log"]) if ( output.is_file() and "ready" in output.read_text(encoding="utf-8") ): locator = candidate break await asyncio.sleep(0.01) self.assertIsNotNone(locator) invocation.cancel() with self.assertRaises(asyncio.CancelledError): await invocation finally: if invocation is not None and not invocation.done(): invocation.cancel() await asyncio.gather(invocation, return_exceptions=True) store.close() assert locator is not None record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["status"], "failed") self.assertEqual(record["exit_code"], "cancelled") self.assertEqual(record["failure_class"], "cancelled") log = (workspace / dispatch.WORK_LOG_NAME).read_text(encoding="utf-8") self.assertIn("| FINISH | test | worker | 0 | pi | failed:cancelled |", log) async def test_heartbeat_updates_output_and_locator_with_pi_native_session(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) session_id = "11111111-1111-1111-1111-111111111111" def command_for( spec, prompt, cwd, actual_session_id, attempt_dir, pi_resume_session=None, ): self.assertEqual(actual_session_id, session_id) native = attempt_dir / "pi-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("pi", "ornith:35b", "pi", local_pi=True) 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), ): 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) async def test_pi_silent_awaiting_model_is_inspected_without_termination(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) session_id = "22222222-2222-2222-2222-222222222222" def command_for( spec, prompt, cwd, actual_session_id, attempt_dir, pi_resume_session=None, ): native = attempt_dir / "pi-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" "{\"type\":\"message\",\"id\":\"assistant-1\"," "\"parentId\":null,\"message\":{\"role\":\"assistant\"," "\"content\":[{\"type\":\"toolCall\",\"id\":\"call-1\"," "\"name\":\"read\"}]}}\\n" "{\"type\":\"message\",\"id\":\"result-1\"," "\"parentId\":\"assistant-1\"," "\"message\":{\"role\":\"toolResult\"," "\"toolCallId\":\"call-1\",\"content\":[]}}\\n', " "encoding='utf-8')\n" "time.sleep(0.08)\n" ) return [sys.executable, "-c", child, str(native)] spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi", local_pi=True) 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.object( dispatch, "PI_MODEL_RESPONSE_STALL_SECONDS", 0.03 ), ): 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.assertEqual(record["status"], "succeeded") self.assertEqual(record["pi_session_phase"], "awaiting-model") self.assertEqual( record["pi_session_phase_reason"], "all-tool-results-recorded", ) self.assertEqual(record["pi_expected_tool_call_ids"], ["call-1"]) self.assertEqual(record["pi_completed_tool_call_ids"], ["call-1"]) self.assertEqual(record["pi_pending_tool_call_ids"], []) inspection = record["pi_silence_inspection"] self.assertGreaterEqual(inspection["silence_seconds"], 0.03) self.assertIn("stream_tail", inspection) heartbeat = Path(record["heartbeat_log"]).read_text(encoding="utf-8") self.assertIn("[silence-inspection]", heartbeat) async def test_pi_silent_starting_state_is_inspected_without_termination(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) session_id = "24242424-2424-2424-2424-242424242424" def command_for( spec, prompt, cwd, actual_session_id, attempt_dir, pi_resume_session=None, ): native = attempt_dir / "pi-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.08)\n" ) return [sys.executable, "-c", child, str(native)] spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi", local_pi=True) 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.object( dispatch, "PI_MODEL_RESPONSE_STALL_SECONDS", 0.03 ), ): 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.assertEqual(record["status"], "succeeded") self.assertEqual(record["pi_session_phase"], "starting") self.assertIsNone(record["pi_stall_timeout_seconds"]) self.assertIn("pi_silence_inspection", record) heartbeat = Path(record["heartbeat_log"]).read_text(encoding="utf-8") self.assertNotIn("[session-stall]", heartbeat) async def test_pi_json_stream_progress_prevents_native_only_stall(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) session_id = "23232323-2323-2323-2323-232323232323" def command_for( spec, prompt, cwd, actual_session_id, attempt_dir, pi_resume_session=None, ): native = attempt_dir / "pi-sessions" / f"session_{session_id}.jsonl" child = ( "from pathlib import Path\n" "import json,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" "for _ in range(8):\n" " print(json.dumps({'type': 'message_update'}), flush=True)\n" " time.sleep(0.015)\n" ) return [sys.executable, "-c", child, str(native)] spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi", local_pi=True) 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), ): 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.assertEqual(record["status"], "succeeded") self.assertEqual(record["pi_activity_state"], "streaming") stream = Path(record["stream_log"]).read_text(encoding="utf-8") self.assertIn('[stdout] {"type": "message_update"}', stream) async def test_pi_incomplete_tool_batch_has_no_automatic_timeout(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) session_id = "33333333-3333-3333-3333-333333333333" def command_for( spec, prompt, cwd, actual_session_id, attempt_dir, pi_resume_session=None, ): native = attempt_dir / "pi-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" "{\"type\":\"message\",\"id\":\"assistant-1\"," "\"parentId\":null,\"message\":{\"role\":\"assistant\"," "\"content\":[{\"type\":\"toolCall\",\"id\":\"call-a\"," "\"name\":\"read\"},{\"type\":\"toolCall\",\"id\":\"call-b\"," "\"name\":\"bash\"}]}}\\n" "{\"type\":\"message\",\"id\":\"result-a\"," "\"parentId\":\"assistant-1\"," "\"message\":{\"role\":\"toolResult\"," "\"toolCallId\":\"call-a\",\"content\":[]}}\\n', " "encoding='utf-8')\n" "time.sleep(0.08)\n" "with path.open('a', encoding='utf-8') as stream:\n" " stream.write(" "'{\"type\":\"message\",\"id\":\"result-b\"," "\"parentId\":\"result-a\"," "\"message\":{\"role\":\"toolResult\"," "\"toolCallId\":\"call-b\",\"content\":[]}}\\n')\n" "print('done', flush=True)\n" ) return [sys.executable, "-c", child, str(native)] spec = dispatch.AgentSpec( "pi", "laguna-s:2.1", "pi", local_pi=True ) 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.object( dispatch, "PI_MODEL_RESPONSE_STALL_SECONDS", 0.02 ), ): 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.assertIsNone(record["pi_stall_timeout_seconds"]) heartbeat = Path(record["heartbeat_log"]).read_text(encoding="utf-8") self.assertNotIn("[session-stall]", heartbeat) async def test_resume_heartbeat_preserves_prior_native_session_path(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) native = workspace / "prior-session.jsonl" native.write_text(pi_session_jsonl([]), encoding="utf-8") prior_locator = workspace / "prior-locator.json" prior_locator.write_text( json.dumps( { "session_id": "resume-session", "native_session_path": str(native), } ), encoding="utf-8", ) def command_for( spec, prompt, cwd, actual_session_id, attempt_dir, pi_resume_session=None, ): self.assertEqual(pi_resume_session, native) return [ sys.executable, "-c", "import time; time.sleep(0.05); print('done', flush=True)", ] spec = dispatch.AgentSpec( "pi", "laguna-s:2.1", "pi", local_pi=True ) try: with ( mock.patch.object( dispatch, "build_command", side_effect=command_for ), mock.patch.object( dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01 ), ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "review", spec, "Continue.", resume_locator=prior_locator, ) finally: store.close() self.assertEqual(rc, 0) self.assertIsNone(failure) record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["native_session_path"], str(native)) self.assertEqual( record["resumed_from_locator"], str(prior_locator) ) async def test_provider_stderr_requires_and_preserves_exact_evidence(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) provider_line = ( "provider_tunnel_error: dial tcp 192.0.2.1:8001: " "connect: connection refused" ) command = [ sys.executable, "-c", "import sys; sys.stderr.write(sys.argv[1] + '\\n'); " "raise SystemExit(1)", provider_line, ] spec = dispatch.AgentSpec( "pi", "ornith:35b", "pi", local_pi=True ) try: with mock.patch.object( dispatch, "build_command", return_value=command, ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan." ) finally: store.close() self.assertEqual(rc, 1) self.assertEqual(failure, "provider-connection") record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual( record["failure_source"], "provider-terminal-diagnostic" ) self.assertTrue(record["provider_transport_failure_confirmed"]) self.assertEqual(record["failure_evidence_source"], "pi:stderr") self.assertEqual(record["failure_evidence_excerpt"], provider_line) self.assertEqual(record["dispatcher_pid"], dispatch.os.getpid()) self.assertEqual( record["dispatcher_source_path"], str(dispatch.DISPATCHER_SOURCE_PATH) ) self.assertEqual( record["dispatcher_source_sha256"], dispatch.DISPATCHER_SOURCE_SHA256, ) self.assertEqual( record["dispatcher_source_current_sha256"], dispatch.DISPATCHER_SOURCE_SHA256, ) self.assertTrue(record["dispatcher_source_matches_loaded"]) self.assertEqual( dispatch.DISPATCHER_SOURCE_SHA256, dispatch.sha256_file(SCRIPT), ) report = dispatch.failure_report_lines(failure, locator) self.assertIn("provider_transport_failure_confirmed=true", report) self.assertIn(f"dispatcher_pid={dispatch.os.getpid()}", report) self.assertIn( f"dispatcher_source_sha256={dispatch.DISPATCHER_SOURCE_SHA256}", report, ) self.assertIn( "dispatcher_source_matches_loaded=true", report, ) self.assertIn(f"provider_evidence={provider_line}", report) async def test_claude_session_limit_stderr_records_provider_quota(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) diagnostic = ( "You've hit your session limit · resets 9pm (Asia/Seoul)" ) command = [ sys.executable, "-c", "import sys; sys.stderr.write(sys.argv[1] + '\\n'); " "raise SystemExit(1)", diagnostic, ] spec = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) try: with mock.patch.object( dispatch, "build_command", return_value=command, ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 1) self.assertEqual(failure, "provider-quota") record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["failure_source"], "cli-terminal-diagnostic") self.assertEqual(record["failure_evidence_source"], "claude:stderr") self.assertEqual(record["failure_evidence_excerpt"], diagnostic) self.assertEqual(record["reasoning_effort"], "xhigh") self.assertFalse(record["provider_transport_failure_confirmed"]) async def test_claude_structured_rate_limit_stdout_records_provider_quota(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) rate_limit_event = json.dumps( { "type": "rate_limit_event", "rate_limit_info": { "status": "rejected", "rateLimitType": "five_hour", "overageStatus": "rejected", }, }, ensure_ascii=False, ) result_event = json.dumps( { "type": "result", "subtype": "success", "is_error": True, "terminal_reason": "api_error", "api_error_status": 429, "result": ( "You've hit your session limit · " "resets 9pm (Asia/Seoul)" ), }, ensure_ascii=False, ) command = [ sys.executable, "-c", "import sys; print(sys.argv[1]); print(sys.argv[2]); " "raise SystemExit(1)", rate_limit_event, result_event, ] spec = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) try: with mock.patch.object( dispatch, "build_command", return_value=command, ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 1) self.assertEqual(failure, "provider-quota") record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["failure_source"], "cli-terminal-diagnostic") self.assertEqual(record["failure_evidence_source"], "claude:stdout") self.assertIn('"api_error_status": 429', record["failure_evidence_excerpt"]) self.assertFalse(record["provider_transport_failure_confirmed"]) async def test_agy_cli_log_quota_records_provider_quota(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) diagnostic = ( "rpc failed: code=ResourceExhausted " "status=RESOURCE_EXHAUSTED HTTP 429 quota exceeded" ) spec = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)", ) def build_agy_command( _spec, _prompt, _workspace, _session_id, attempt_dir, **_kwargs, ): return [ sys.executable, "-c", ( "from pathlib import Path; " "Path(__import__('sys').argv[1]).write_text(" "__import__('sys').argv[2] + '\\n', encoding='utf-8'); " "raise SystemExit(1)" ), str(attempt_dir / "agy-cli.log"), diagnostic, ] try: with ( mock.patch.object( dispatch, "build_command", side_effect=build_agy_command, ), mock.patch.object( dispatch, "agy_conversations", return_value={}, ), ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 1) self.assertEqual(failure, "provider-quota") record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["failure_source"], "cli-terminal-diagnostic") self.assertEqual( record["failure_evidence_source"], "agy:cli-log", ) self.assertEqual(record["failure_evidence_excerpt"], diagnostic) async def test_agy_cli_log_quota_with_zero_exit_records_provider_quota(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) diagnostic = ( "agent executor error: model unreachable: " "RESOURCE_EXHAUSTED (code 429): Individual quota reached" ) spec = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)", ) def build_agy_command( _spec, _prompt, _workspace, _session_id, attempt_dir, **_kwargs, ): return [ sys.executable, "-c", ( "from pathlib import Path; " "Path(__import__('sys').argv[1]).write_text(" "__import__('sys').argv[2] + '\\n', encoding='utf-8')" ), str(attempt_dir / "agy-cli.log"), diagnostic, ] try: with ( mock.patch.object( dispatch, "build_command", side_effect=build_agy_command, ), mock.patch.object( dispatch, "agy_conversations", return_value={}, ), ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan.", ) finally: store.close() self.assertEqual(rc, 0) self.assertEqual(failure, "provider-quota") record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["status"], "failed") self.assertEqual(record["exit_code"], 0) self.assertEqual(record["failure_source"], "cli-terminal-diagnostic") self.assertEqual( record["failure_evidence_source"], "agy:cli-log", ) self.assertEqual(record["failure_evidence_excerpt"], diagnostic) self.assertFalse(record["provider_transport_failure_confirmed"]) async def test_exit_143_is_process_termination_not_provider_failure(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = TaskStageTest().make_task(workspace) store = dispatch.StateStore(workspace) provider_line = ( "provider_tunnel_error: dial tcp 192.0.2.1:8001: " "connect: connection refused" ) spec = dispatch.AgentSpec( "pi", "ornith:35b", "pi", local_pi=True ) try: with mock.patch.object( dispatch, "build_command", return_value=[ sys.executable, "-c", "import sys; sys.stderr.write(sys.argv[1] + '\\n'); " "raise SystemExit(143)", provider_line, ], ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "worker", spec, "Read the plan." ) finally: store.close() self.assertEqual(rc, 143) self.assertEqual(failure, "process-terminated") record = json.loads(locator.read_text(encoding="utf-8")) self.assertEqual(record["failure_source"], "process-termination") self.assertFalse(record["provider_transport_failure_confirmed"]) self.assertEqual(record["termination_signal"], "SIGTERM") self.assertTrue(record["termination_signal_inferred"]) self.assertEqual(record["termination_initiator"], "unknown") class ReviewControlTest(unittest.TestCase): def test_classifies_claude_session_limit_as_provider_quota(self): diagnostic = ( "You've hit your session limit · resets 9pm (Asia/Seoul)" ) self.assertEqual( dispatch.classify_failure_with_evidence(diagnostic), ("provider-quota", diagnostic), ) def test_claude_assistant_text_is_not_a_terminal_diagnostic(self): assistant_event = json.dumps( { "type": "assistant", "message": { "role": "assistant", "content": [ { "type": "text", "text": "You've hit your session limit", } ], }, } ) self.assertIsNone( dispatch.terminal_diagnostic("claude", "stdout", assistant_event) ) def test_claude_rejected_rate_limit_event_is_terminal_diagnostic(self): event = json.dumps( { "type": "rate_limit_event", "rate_limit_info": {"status": "rejected"}, } ) diagnostic = dispatch.terminal_diagnostic("claude", "stdout", event) self.assertIsNotNone(diagnostic) self.assertEqual( dispatch.classify_failure_with_evidence(diagnostic or ""), ("provider-quota", diagnostic), ) def test_agy_structured_resource_exhausted_is_terminal_diagnostic(self): event = json.dumps( { "type": "error", "error": { "code": 429, "status": "RESOURCE_EXHAUSTED", "message": "Quota exceeded", }, } ) diagnostic = dispatch.terminal_diagnostic("agy", "stdout", event) self.assertIsNotNone(diagnostic) self.assertEqual( dispatch.classify_failure_with_evidence(diagnostic or ""), ("provider-quota", diagnostic), ) def test_agy_assistant_quota_text_is_not_a_terminal_diagnostic(self): event = json.dumps( { "type": "assistant", "status": "rejected", "error": {"code": 429}, "content": "The quota exceeded message is handled in the code.", } ) self.assertIsNone( dispatch.terminal_diagnostic("agy", "stdout", event) ) def test_agy_log_diagnostic_requires_strong_quota_evidence(self): with tempfile.TemporaryDirectory() as temporary: log = Path(temporary) / "agy-cli.log" log.write_text( "quota configuration loaded\n" "ERROR quota configuration refresh failed\n" "status=RESOURCE_EXHAUSTED HTTP 429 quota exceeded\n", encoding="utf-8", ) self.assertEqual( dispatch.agy_log_diagnostics(log), ["status=RESOURCE_EXHAUSTED HTTP 429 quota exceeded"], ) def test_claude_promotion_targets_terra_high(self): claude = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) promoted = dispatch.promoted_spec(claude, recovery_count=0) self.assertEqual( promoted, dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ), ) assert promoted is not None command = dispatch.build_command( promoted, "Read the plan.", Path("/workspace"), "session-id", Path("/attempt"), ) self.assertIn("gpt-5.6-terra", command) self.assertIn('model_reasoning_effort="high"', command) self.assertEqual( dispatch.effective_reasoning_effort(promoted), "high", ) self.assertIs( dispatch.promoted_spec(promoted, recovery_count=0), promoted, ) def test_regular_codex_and_claude_routes_keep_xhigh_effort(self): codex = dispatch.AgentSpec( "codex", "gpt-5.6-sol", "codex/gpt-5.6-sol xhigh", ) claude = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) self.assertEqual(dispatch.effective_reasoning_effort(codex), "xhigh") self.assertEqual(dispatch.effective_reasoning_effort(claude), "xhigh") def test_classifies_provider_tunnel_connection_refusal(self): provider_line = ( "provider_tunnel_error: dial tcp 192.0.2.1:8001: " "connect: connection refused" ) self.assertEqual( dispatch.classify_failure(provider_line), "provider-connection", ) self.assertEqual( dispatch.classify_failure_with_evidence( f"unrelated warning\n{provider_line}" ), ("provider-connection", provider_line), ) def test_generic_tool_stderr_is_not_provider_transport_evidence(self): weak_lines = [ "pytest setup failed: connection refused while opening fixture", "dial tcp 127.0.0.1:9999: connect: connection refused", "curl error: failure when receiving data from the peer", ] for line in weak_lines: with self.subTest(line=line): self.assertEqual( dispatch.classify_failure_with_evidence(line), ("generic-error", None), ) def test_provider_stream_requires_strong_backend_or_sse_context(self): line = ( "Backend for model crashed before streaming started: " "SSE stream before DONE" ) self.assertEqual( dispatch.classify_failure_with_evidence(line), ("provider-stream-disconnect", line), ) def test_pi_stdout_provider_words_are_not_terminal_diagnostics(self): line = "provider_tunnel_error: connection refused" self.assertIsNone(dispatch.terminal_diagnostic("pi", "stdout", line)) def test_dispatcher_source_provenance_detects_hot_edit(self): changed_sha256 = "f" * 64 self.assertNotEqual(changed_sha256, dispatch.DISPATCHER_SOURCE_SHA256) with mock.patch.object( dispatch, "sha256_file", return_value=changed_sha256, ): provenance = dispatch.dispatcher_source_provenance() self.assertEqual( provenance["dispatcher_source_sha256"], dispatch.DISPATCHER_SOURCE_SHA256, ) self.assertEqual( provenance["dispatcher_source_current_sha256"], changed_sha256 ) self.assertFalse(provenance["dispatcher_source_matches_loaded"]) def test_pi_phase_reads_large_last_jsonl_event(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" path.write_text( pi_session_jsonl( [ { "type": "message", "message": { "role": "assistant", "content": [ { "type": "toolCall", "id": "large-result", "name": "read", } ], }, }, { "type": "message", "message": { "role": "toolResult", "toolCallId": "large-result", "content": [ {"type": "text", "text": "x" * 20000} ], }, }, ] ), encoding="utf-8", ) self.assertEqual( dispatch.pi_native_session_phase(str(path)), "awaiting-model", ) def test_pi_phase_keeps_incomplete_sequential_batch_tool_running(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" events = [ { "type": "message", "message": { "role": "assistant", "content": [ {"type": "toolCall", "id": "call-a", "name": "read"}, {"type": "toolCall", "id": "call-b", "name": "bash"}, ], }, }, { "type": "message", "message": { "role": "toolResult", "toolCallId": "call-a", "content": [], }, }, ] path.write_text( pi_session_jsonl(events), encoding="utf-8", ) state = dispatch.pi_native_session_state(str(path)) self.assertEqual(state.phase, "tool-running") self.assertEqual(state.expected_tool_call_ids, ("call-a", "call-b")) self.assertEqual(state.completed_tool_call_ids, ("call-a",)) self.assertEqual(state.pending_tool_call_ids, ("call-b",)) def test_pi_phase_waits_for_model_only_after_entire_batch_completes(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" events = [ { "type": "message", "message": { "role": "assistant", "content": [ {"type": "toolCall", "id": "call-a", "name": "read"}, {"type": "toolCall", "id": "call-b", "name": "bash"}, ], }, }, { "type": "message", "message": { "role": "toolResult", "toolCallId": "call-a", "content": [], }, }, { "type": "message", "message": { "role": "toolResult", "toolCallId": "call-b", "content": [], }, }, ] path.write_text( pi_session_jsonl(events), encoding="utf-8", ) state = dispatch.pi_native_session_state(str(path)) self.assertEqual(state.phase, "awaiting-model") self.assertEqual(state.completed_tool_call_ids, ("call-a", "call-b")) self.assertEqual(state.pending_tool_call_ids, ()) def test_pi_phase_marks_unknown_schema_without_assuming_tool_completion(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" path.write_text( pi_session_jsonl( [ { "type": "message", "message": { "role": "toolResult", "content": [], }, }, ] ), encoding="utf-8", ) state = dispatch.pi_native_session_state(str(path)) self.assertEqual(state.phase, "unknown") self.assertEqual(state.reason, "tool-result-id-missing") def test_pi_phase_does_not_treat_unknown_assistant_content_as_final(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" path.write_text( pi_session_jsonl( [ { "type": "message", "message": { "role": "assistant", "content": [ { "type": "futureToolCall", "id": "unknown-call", } ], }, }, ] ), encoding="utf-8", ) state = dispatch.pi_native_session_state(str(path)) self.assertEqual(state.phase, "unknown") self.assertEqual(state.reason, "unsupported-assistant-content") def test_pi_phase_marks_future_session_version_unknown(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" path.write_text( pi_session_jsonl( [ { "type": "message", "message": { "role": "user", "content": [], }, }, ], version=dispatch.PI_SESSION_SCHEMA_VERSION + 1, ), encoding="utf-8", ) state = dispatch.pi_native_session_state(str(path)) self.assertEqual(state.phase, "unknown") self.assertEqual( state.reason, f"unsupported-session-version:" f"{dispatch.PI_SESSION_SCHEMA_VERSION + 1}", ) def test_pi_phase_marks_corrupt_session_entries_unknown(self): header = pi_session_jsonl([]).encode() cases = { "missing-parent": header + json.dumps( { "type": "message", "id": "message-without-parent", "message": { "role": "user", "content": [], }, } ).encode() + b"\n", "invalid-utf8": header + b'{"type":"message","id":"bad-\\xff"}\n', } for name, content in cases.items(): with self.subTest(name=name), tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" path.write_bytes(content) state = dispatch.pi_native_session_state(str(path)) self.assertEqual(state.phase, "unknown") def test_pi_phase_follows_only_the_active_session_branch(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "session.jsonl" path.write_text( pi_session_jsonl( [ { "type": "message", "id": "root-user", "parentId": None, "message": { "role": "user", "content": [], }, }, { "type": "message", "id": "abandoned-assistant", "parentId": "root-user", "message": { "role": "assistant", "content": [{"type": "text", "text": "done"}], }, }, { "type": "custom", "id": "active-branch-marker", "parentId": "root-user", }, ] ), encoding="utf-8", ) state = dispatch.pi_native_session_state(str(path)) self.assertEqual(state.phase, "awaiting-model") self.assertEqual(state.reason, "user-message") def test_detects_codex_collaboration_wait(self): line = ( '{"type":"item.started","item":{"type":"collab_tool_call",' '"tool":"wait"}}' ) self.assertEqual(dispatch.codex_collaboration_tool(line), "wait") def test_ignores_completed_or_non_json_events(self): self.assertIsNone( dispatch.codex_collaboration_tool( '{"type":"item.completed","item":{"type":"collab_tool_call",' '"tool":"wait"}}' ) ) self.assertIsNone(dispatch.codex_collaboration_tool("not json")) def test_prompts_keep_local_work_and_official_review_roles_separate(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = TaskStageTest().make_task(root) pi = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) selfcheck = dispatch.base_prompt(task, "selfcheck", pi) review = dispatch.base_prompt( task, "review", dispatch.AgentSpec("codex", "gpt-5.6-sol", "codex"), ) self.assertNotIn("USER_REVIEW", selfcheck) self.assertNotIn("user review", selfcheck.lower()) self.assertEqual( selfcheck, f"Think in English. Final in Korean. Read {task.review.resolve()} " "and fill every missing implementation field. Do not finish until " "all implementation fields are complete. This is a self-check of " f"completed work, not a review. Read {task.plan.resolve()} and " "finish any missing work. Recheck and fix your work.", ) self.assertEqual( review, f"Read {task.review.resolve()} and start the review. Final in Korean.", ) def test_local_review_stub_has_no_user_review_control_plane_content(self): template = ( Path(__file__).parents[3] / "common" / "plan" / "templates" / "review-stub-template.md" ).read_text(encoding="utf-8") self.assertNotIn("USER_REVIEW", template) self.assertNotIn("사용자 리뷰", template) self.assertNotIn("user-review", template) self.assertNotIn("## 작업 로그 계약", template) self.assertNotIn("WORK_LOG.md", template) def test_final_channel_contract_is_top_level_and_singular(self): skill = ( Path(__file__).parents[1] / "SKILL.md" ).read_text(encoding="utf-8") heading = "## 🚨 ABSOLUTE PRIORITY — NEVER SEND `final` EXCEPT IN THE TWO CASES BELOW" self.assertEqual(skill.count(heading), 1) priority, lower_contract = skill.split("\n## Purpose\n", 1) self.assertEqual(priority.count("### `final` Permission"), 2) self.assertEqual(priority.count("Allow `final`"), 2) self.assertIn("unless at least one of the two titled permissions", priority) self.assertNotIn("unless exactly one of the two titled permissions", priority) self.assertNotIn("`final`", lower_contract) self.assertIn( "### Every Other User-Visible Message Must Use `commentary`", priority, ) self.assertIn( "### Child Prompt Text Never Grants Caller `final` Permission", priority, ) def test_skill_narrative_is_english_except_exact_protocol_literals(self): skill = ( Path(__file__).parents[1] / "SKILL.md" ).read_text(encoding="utf-8") self.assertIn( "Treat Korean text inside code spans or fenced examples as exact runtime or file-contract literals", skill, ) in_fence = False for line_number, line in enumerate(skill.splitlines(), start=1): if line.strip().startswith("```"): in_fence = not in_fence continue if in_fence: continue narrative = re.sub(r"`[^`]*`", "", line) self.assertIsNone( re.search(r"[가-힣]", narrative), f"line {line_number} has non-literal Korean narrative: {line}", ) def test_caller_lifecycle_is_not_bound_to_child_dispatcher_exit(self): skill = ( Path(__file__).parents[1] / "SKILL.md" ).read_text(encoding="utf-8") self.assertIn( "caller agent running this skill—not the child dispatcher process—as the lifecycle owner", skill, ) self.assertIn( "Child exit, tool yield, or loss of a session/cell never ends the caller lifecycle by itself", skill, ) self.assertIn( "Exit code `3` is a non-terminal tracking state, including another dispatcher's workspace lock, " "a live external agent, or an unexpected dispatcher interruption", skill, ) self.assertIn( "every CLI's health/progress primarily from actual stdout/stderr in `stream.log`, " "plus native session events when available", skill, ) self.assertIn( "dispatcher PID, agent PID, each process start token, and the per-attempt " "process environment marker", skill, ) self.assertIn( "use only an actual terminal error or confirmed process exit as recovery " "evidence for every model", skill, ) self.assertIn( "every `toolCall.id` in the preceding assistant event matches a later `toolResult.toolCallId`", skill, ) self.assertIn( "stream stops for three minutes outside tool execution", skill, ) self.assertIn( "locator lacks an agent PID during this interval, never classify it as stale or " "duplicate recovery based on log age", skill, ) self.assertIn( "original exception is a persistent-state error, do not convert it to exit `2` if any agent was running", skill, ) self.assertIn( "do not return successful exit `0` while any attempt directory remains", skill, ) self.assertIn( "share a budget of 10 consecutive automatic recovery failures for the same task stage", skill, ) self.assertIn( "On the 10th failure, block that task and do not auto-resume after cooldown", skill, ) self.assertIn( "legacy locator `session-stall` as a record of an earlier dispatcher timeout policy, not as provider failure", skill, ) self.assertIn( "Never classify exit code `143` as provider failure without actual provider terminal evidence", skill, ) self.assertIn( "Do not generalize one `pi -p` fresh/isolated session attempt to a Pi TUI or system-wide provider outage", skill, ) self.assertIn("provider_transport_failure_confirmed", skill) self.assertIn( "Do not infer provider failure from `connection refused`, `dial tcp`, or `curl` peer failure in ordinary tool/test stderr", skill, ) self.assertIn( "A running Python dispatcher does not hot-reload source edits", skill, ) self.assertIn("dispatcher_source_sha256", skill) self.assertIn("`dispatcher_source_matches_loaded=false`", skill) self.assertIn( "KST-night `local-G07`–`local-G08` Laguna locator `context-limit`/`session-stall`", skill, ) self.assertIn( "fresh session and `세션응답복구재시도` only for other legacy Pi `session-stall` recovery", skill, ) def test_work_log_archive_ownership_is_consistent_across_skills(self): skills_root = Path(__file__).parents[3] dispatcher_skill = ( Path(__file__).parents[1] / "SKILL.md" ).read_text(encoding="utf-8") review_skill = ( skills_root / "common" / "code-review" / "SKILL.md" ).read_text(encoding="utf-8") plan_skill = ( skills_root / "common" / "plan" / "SKILL.md" ).read_text(encoding="utf-8") self.assertIn( "append the final `FINISH` and move the generated `WORK_LOG.md`", dispatcher_skill, ) self.assertIn("work_log_N.log", dispatcher_skill) self.assertIn( "append `FINISH` with `reconciled:verified-complete-archive`", dispatcher_skill, ) self.assertIn( "pidless stream/native evidence remains live", dispatcher_skill, ) self.assertIn( "task-group `agent-task/{task_group}/WORK_LOG.md`는 " "이동·복사·삭제·이름 변경하지 않는다", review_skill, ) self.assertIn( "review process 종료 뒤 dispatcher가 마지막 `FINISH`를 append", review_skill, ) self.assertIn( "dispatcher가 마지막 `FINISH` 기록과 group 완료 확인 뒤 " "`work_log_N.log`로 archive한다", plan_skill, ) class ProcessTerminationTest(unittest.IsolatedAsyncioTestCase): async def test_terminates_the_exact_process_group(self): class Process: pid = 12345 returncode = None async def wait(self): self.returncode = -signal.SIGTERM return self.returncode process = Process() with mock.patch.object( dispatch.os, "killpg", side_effect=[None, ProcessLookupError], ) as killpg: await dispatch.terminate_process_group(process) self.assertEqual(killpg.call_args_list[0], mock.call(12345, signal.SIGTERM)) self.assertEqual(killpg.call_args_list[1], mock.call(12345, 0)) async def test_kills_sigterm_ignoring_descendant_and_closes_pipe(self): child_script = ( "import signal,time;" "signal.signal(signal.SIGTERM,signal.SIG_IGN);" "time.sleep(60)" ) parent_script = ( "import signal,subprocess,sys,time;" "signal.signal(signal.SIGTERM,signal.SIG_IGN);" f"subprocess.Popen([sys.executable,'-c',{child_script!r}]);" "print('ready',flush=True);" "time.sleep(60)" ) process = await asyncio.create_subprocess_exec( sys.executable, "-c", parent_script, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, start_new_session=True, ) try: assert process.stdout is not None self.assertEqual( await asyncio.wait_for(process.stdout.readline(), timeout=1), b"ready\n", ) await dispatch.terminate_process_group(process, grace_seconds=0.05) self.assertEqual(process.returncode, -signal.SIGKILL) self.assertEqual( await asyncio.wait_for(process.stdout.read(), timeout=1), b"", ) finally: if process.returncode is None: await dispatch.terminate_process_group(process, grace_seconds=0.05) class ReviewRetryTest(unittest.IsolatedAsyncioTestCase): def make_task(self, root: Path): return TaskStageTest().make_task(root) async def test_claude_provider_quota_promotes_to_terra_high(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) claude = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) terra = dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ) locators = [root / "locator-0.json", root / "locator-1.json"] results = [ (1, "provider-quota", locators[0]), (0, None, locators[1]), ] with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=results), ) as invoke: success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", claude, ) self.assertTrue(success) self.assertEqual(locator, locators[1]) self.assertEqual(invoke.await_count, 2) self.assertEqual(invoke.await_args_list[0].args[4], claude) self.assertEqual(invoke.await_args_list[1].args[4], terra) async def test_agy_and_claude_quota_chain_promotes_to_terra(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) agy = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)", ) claude = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) terra = dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ) locators = [ root / "locator-0.json", root / "locator-1.json", root / "locator-2.json", ] with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock( side_effect=[ (1, "provider-quota", locators[0]), (1, "provider-quota", locators[1]), (0, None, locators[2]), ] ), ) as invoke: success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", agy, ) self.assertTrue(success) self.assertEqual(locator, locators[2]) self.assertEqual( [call.args[4] for call in invoke.await_args_list], [agy, claude, terra], ) async def test_process_termination_retries_same_claude_target(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) claude = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) locators = [root / "locator-0.json", root / "locator-1.json"] with ( mock.patch.object( dispatch, "invoke", new=mock.AsyncMock( side_effect=[ (1, "process-terminated", locators[0]), (0, None, locators[1]), ] ), ) as invoke, mock.patch.object( dispatch.asyncio, "sleep", new=mock.AsyncMock(), ), ): success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", claude, ) self.assertTrue(success) self.assertEqual(locator, locators[1]) self.assertEqual( [call.args[4] for call in invoke.await_args_list], [claude, claude], ) async def test_legacy_generic_quota_blocker_resumes_directly_on_terra(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = self.make_task(root) store = dispatch.StateStore(root) locator = write_legacy_quota_attempts( store.runs, task, )[-1] state = store.task_state(task) state.update( blocked=( "worker recovery failure limit exhausted: 10/10 " f"locator={locator}" ), recovery_failures={"worker": 10}, ) recovery = dispatch.legacy_promotion_recovery( store.runs, task, state, ) self.assertIsNotNone(recovery) initial_route = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)", ) terra = dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ) completed_locator = root / "completed-locator.json" try: with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock( return_value=(0, None, completed_locator) ), ) as invoke: success, actual_locator = await dispatch.run_escalating( root, store, task, "worker", initial_route, initial_resume_locator=locator, ) recovered_state = store.task_state(task) finally: store.close() self.assertTrue(success) self.assertEqual(actual_locator, completed_locator) self.assertEqual(invoke.await_count, 1) self.assertEqual(invoke.await_args.args[4], terra) self.assertIsNone(recovered_state["blocked"]) self.assertEqual( recovered_state["recovery_failures"], {}, ) self.assertEqual( recovered_state["legacy_terminal_reclassification"][ "failed_cli" ], "claude", ) async def test_legacy_agy_quota_log_resumes_on_claude_then_terra(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = self.make_task(root) store = dispatch.StateStore(root) locator = write_legacy_quota_attempts( store.runs, task, cli="agy", model="Gemini 3.6 Flash (High)", reasoning_effort=None, )[-1] state = store.task_state(task) state.update( blocked=( "worker recovery failure limit exhausted: 10/10 " f"locator={locator}" ), recovery_failures={"worker": 10}, ) recovery = dispatch.legacy_promotion_recovery( store.runs, task, state, ) self.assertIsNotNone(recovery) assert recovery is not None self.assertEqual(recovery.failed_cli, "agy") self.assertEqual(recovery.evidence_source, "agy:cli-log") initial_route = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)", ) claude = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) terra = dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ) locators = [ root / "claude-locator.json", root / "terra-locator.json", ] try: with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock( side_effect=[ (1, "provider-quota", locators[0]), (0, None, locators[1]), ] ), ) as invoke: success, actual_locator = await dispatch.run_escalating( root, store, task, "worker", initial_route, initial_resume_locator=locator, ) finally: store.close() self.assertTrue(success) self.assertEqual(actual_locator, locators[1]) self.assertEqual( [call.args[4] for call in invoke.await_args_list], [claude, terra], ) async def test_worker_persists_the_actual_promoted_target(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = self.make_task(root) store = dispatch.StateStore(root) initial_route = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)", ) attempt = store.runs / "completed-attempt" attempt.mkdir() locator = attempt / "locator.json" locator.write_text( json.dumps( { "cli": "codex", "model": "gpt-5.6-terra", "reasoning_effort": "high", } ), encoding="utf-8", ) completing_decision = { "work_unit_id": "test::plan-0::tag-TEST", "stage": "worker", "selected": { "adapter": "codex", "target": "gpt-5.6-terra", "execution_class": "cloud_model", "selfcheck_required": False, }, } try: # Persist completing decision to execution_decisions["worker"] store.task_state(task) store.data["tasks"][task.name]["execution_decisions"]["worker"] = completing_decision store.save() with ( mock.patch.object( dispatch, "persisted_execution_decision", return_value=(completing_decision, initial_route), ), mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, locator)), ), ): await dispatch.run_worker( root, store, task, ) state = store.task_state(task) finally: store.close() self.assertTrue(state["worker_done"]) self.assertEqual(state["worker_cli"], "codex") self.assertEqual(state["worker_model"], "gpt-5.6-terra") self.assertTrue(state["selfcheck_done"]) self.assertEqual( state["execution_class"], "cloud_model" ) self.assertEqual( state["completing_decision"]["selected"]["adapter"], "codex" ) async def test_retries_two_control_violations_then_succeeds(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) spec = dispatch.AgentSpec("codex", "gpt-5.6-sol", "codex") locators = [root / f"locator-{index}.json" for index in range(3)] results = [ (1, "review-control-violation", locators[0]), (1, "review-control-violation", locators[1]), (0, None, locators[2]), ] with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=results) ) as invoke: success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "review", spec ) self.assertTrue(success) self.assertEqual(locator, locators[2]) self.assertEqual(invoke.await_count, 3) async def test_review_control_retries_do_not_create_a_blocker(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) spec = dispatch.AgentSpec("codex", "gpt-5.6-sol", "codex") locators = [root / f"locator-{index}.json" for index in range(4)] results = [ (1, "review-control-violation", locators[0]), (1, "review-control-violation", locators[1]), (1, "review-control-violation", locators[2]), (0, None, locators[3]), ] with ( mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=results) ) as invoke, mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "review", spec ) self.assertTrue(success) self.assertEqual(locator, locators[3]) self.assertEqual(invoke.await_count, 4) async def test_does_not_retry_obsolete_model_work_log_failure(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) locators = [root / "locator-0.json", root / "locator-1.json"] results = [ (0, "work-log-incomplete", locators[0]), (0, None, locators[1]), ] with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=results) ) as invoke: success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", spec ) self.assertFalse(success) self.assertEqual(locator, locators[0]) self.assertEqual(invoke.await_count, 1) async def test_retries_pi_session_stall_twice_with_same_model(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi", local_pi=True) locators = [root / f"locator-{index}.json" for index in range(3)] results = [ (1, "session-stall", locators[0]), (1, "session-stall", locators[1]), (0, None, locators[2]), ] with ( mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=results) ) as invoke, mock.patch.object( dispatch.asyncio, "sleep", new=mock.AsyncMock() ), ): success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", spec ) self.assertTrue(success) self.assertEqual(locator, locators[2]) self.assertEqual(invoke.await_count, 3) self.assertTrue(all(call.args[4] == spec for call in invoke.await_args_list)) self.assertEqual( invoke.await_args_list[1].args[-1], "Think in English. Final in Korean. Continue this session and " "complete the current task.", ) self.assertEqual( invoke.await_args_list[1].kwargs["resume_locator"], locators[0], ) self.assertEqual( invoke.await_args_list[2].kwargs["resume_locator"], locators[1], ) def test_failed_laguna_locator_is_recovered_after_dispatcher_restart(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) native = root / "session.jsonl" native.write_text("{}\n", encoding="utf-8") locator = root / "locator.json" locator.write_text( json.dumps( { "cli": "pi", "model": "laguna-s:2.1", "status": "failed", "failure_class": "session-stall", "native_session_path": str(native), } ), encoding="utf-8", ) self.assertEqual( dispatch.laguna_resume_locator( {"active_locator": str(locator)} ), locator, ) async def test_retries_pi_connection_and_generic_failures(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) locators = [root / f"locator-{index}.json" for index in range(3)] results = [ (1, "provider-connection", locators[0]), (1, "generic-error", locators[1]), (0, None, locators[2]), ] with ( mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=results) ) as invoke, mock.patch.object( dispatch.asyncio, "sleep", new=mock.AsyncMock() ) as sleep, ): success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", spec ) self.assertTrue(success) self.assertEqual(locator, locators[2]) self.assertEqual(invoke.await_count, 3) self.assertEqual(sleep.await_count, 2) async def test_pi_tenth_failure_blocks_without_cooldown(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) locators = [ root / f"locator-{index}.json" for index in range(dispatch.RECOVERY_FAILURE_LIMIT) ] results = [ (1, "provider-stream-disconnect", locator) for locator in locators ] sleep_observations = [] async def observe_sleep(delay): sleep_observations.append(delay) with ( mock.patch.object( dispatch, "invoke", new=mock.AsyncMock(side_effect=results) ) as invoke, mock.patch.object( dispatch.asyncio, "sleep", new=mock.AsyncMock(side_effect=observe_sleep), ), ): success, locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", spec, ) self.assertFalse(success) self.assertEqual(locator, locators[-1]) self.assertEqual(invoke.await_count, dispatch.RECOVERY_FAILURE_LIMIT) self.assertEqual( len(sleep_observations), dispatch.RECOVERY_FAILURE_LIMIT - 1 ) async def test_recovery_failure_limit_survives_dispatcher_restart(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = self.make_task(root) store = dispatch.StateStore(root) locator = root / "locator.json" store.update_task( task, recovery_failures={"worker": dispatch.RECOVERY_FAILURE_LIMIT - 1} ) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) try: with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock( return_value=(1, "generic-error", locator) ), ) as invoke: success, actual_locator = await dispatch.run_escalating( root, store, task, "worker", spec ) self.assertFalse(success) self.assertEqual(actual_locator, locator) self.assertEqual(invoke.await_count, 1) self.assertIn( "recovery failure limit exhausted", store.task_state(task)["blocked"], ) finally: store.close() async def test_already_exhausted_recovery_budget_does_not_invoke_model(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = self.make_task(root) store = dispatch.StateStore(root) locator = root / "locator.json" store.update_task( task, recovery_failures={"worker": dispatch.RECOVERY_FAILURE_LIMIT} ) spec = dispatch.AgentSpec("pi", "ornith:35b", "pi", local_pi=True) try: with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock() ) as invoke: success, actual_locator = await dispatch.run_escalating( root, store, task, "worker", spec, initial_resume_locator=locator, ) self.assertFalse(success) self.assertEqual(actual_locator, locator) self.assertEqual(invoke.await_count, 0) finally: store.close() async def test_does_not_promote_work_log_infrastructure_failure(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) task = self.make_task(root) spec = dispatch.AgentSpec("codex", "gpt-5.6-sol", "codex") locator = root / "locator.json" with mock.patch.object( dispatch, "invoke", new=mock.AsyncMock( return_value=(1, "work-log-runtime-write", locator) ), ) as invoke: success, actual_locator = await dispatch.run_escalating( root, mock.Mock(), task, "worker", spec, ) self.assertFalse(success) self.assertEqual(actual_locator, locator) self.assertEqual(invoke.await_count, 1) class RepetitionLimitTest(unittest.IsolatedAsyncioTestCase): async def test_exhausted_selfcheck_budget_does_not_invoke_model(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = TaskStageTest().make_task(root) store = dispatch.StateStore(root) completing_decision = { "work_unit_id": "test::plan-0::tag-TEST", "stage": "worker", "selected": { "adapter": "pi", "target": "iop/ornith:35b", "execution_class": "local_model", "selfcheck_required": True, }, } store.update_task( task, selfcheck_incomplete=dispatch.SELF_CHECK_INCOMPLETE_LIMIT, completing_decision=completing_decision, ) try: with mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock() ) as run_escalating: await dispatch.run_selfcheck( root, store, task, ) self.assertEqual(run_escalating.await_count, 0) self.assertIn( "limit already exhausted", store.task_state(task)["blocked"] ) finally: store.close() async def test_exhausted_review_budget_does_not_invoke_model(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = TaskStageTest().make_task(root) store = dispatch.StateStore(root) store.update_task( task, review_no_progress=dispatch.REVIEW_NO_PROGRESS_LIMIT ) try: with mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock() ) as run_escalating: result = await dispatch.run_review(root, store, task) self.assertIsNone(result) self.assertEqual(run_escalating.await_count, 0) self.assertIn( "limit already exhausted", store.task_state(task)["blocked"] ) finally: store.close() async def test_selfcheck_incomplete_tenth_pass_blocks_task(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = TaskStageTest().make_task(root) store = dispatch.StateStore(root) completing_decision = { "work_unit_id": "test::plan-0::tag-TEST", "stage": "worker", "selected": { "adapter": "pi", "target": "iop/ornith:35b", "execution_class": "local_model", "selfcheck_required": True, }, } store.update_task( task, selfcheck_incomplete=dispatch.SELF_CHECK_INCOMPLETE_LIMIT - 1, completing_decision=completing_decision, ) locator = root / "locator.json" try: with ( mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, locator)), ) as run_escalating, mock.patch.object( dispatch, "implementation_review_errors", return_value=["검증 결과 미완성"], ), ): await dispatch.run_selfcheck( root, store, task, ) self.assertEqual(run_escalating.await_count, 1) self.assertIn( "selfcheck implementation fields remain incomplete", store.task_state(task)["blocked"], ) finally: store.close() async def test_review_tenth_no_progress_pass_blocks_task(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / ".git").mkdir() task = TaskStageTest().make_task(root) store = dispatch.StateStore(root) store.update_task( task, review_no_progress=dispatch.REVIEW_NO_PROGRESS_LIMIT - 1, ) locator = root / "locator.json" try: with ( mock.patch.object( dispatch, "run_escalating", new=mock.AsyncMock(return_value=(True, locator)), ), mock.patch.object( dispatch, "task_signature", return_value="unchanged" ), mock.patch.object( dispatch, "review_fingerprints", return_value=set() ), ): result = await dispatch.run_review(root, store, task) self.assertIsNone(result) self.assertIn( "review made no progress", store.task_state(task)["blocked"] ) finally: store.close() class BlockerDrainTest(unittest.IsolatedAsyncioTestCase): async def test_user_review_only_holds_its_dependency_closure(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() group = workspace / "agent-task" / "m-test" gate_dir = group / "01_gate" dependent_dir = group / "02+01_dependent" independent_dir = group / "03_independent" for directory in (gate_dir, dependent_dir, independent_dir): directory.mkdir(parents=True) user_review = gate_dir / "USER_REVIEW.md" user_review.write_text( TaskStageTest.blocking_user_review_text(), encoding="utf-8" ) gate = dispatch.Task( name="m-test/01_gate", directory=gate_dir, plan=None, review=None, user_review=user_review, recovery=True, index=1, ) def runnable_task(name, directory, index, deps=()): plan = directory / "PLAN-local-G05.md" review = directory / "CODE_REVIEW-local-G05.md" plan.write_text( f"\n", encoding="utf-8", ) review.write_text("", encoding="utf-8") return dispatch.Task( name=name, directory=directory, plan=plan, review=review, user_review=None, recovery=False, index=index, deps=deps, lane="local", grade=5, plan_hash=f"{name}-hash", ) dependent = runnable_task( "m-test/02+01_dependent", dependent_dir, 2, ("01",) ) independent = runnable_task( "m-test/03_independent", independent_dir, 3 ) completed_archive = workspace / "completed-independent" completed_archive.mkdir() (completed_archive / "complete.log").write_text( "complete\n", encoding="utf-8" ) args = SimpleNamespace( task_group="m-test", retry_blocked=False, dry_run=False ) store = dispatch.StateStore(workspace) try: with ( mock.patch.object( dispatch, "scan_tasks", side_effect=[ [gate, dependent, independent], [gate, dependent], ], ), mock.patch.object( dispatch, "run_worker", new=mock.AsyncMock(return_value=str(completed_archive)), ) as run_worker, ): result = await dispatch.dispatch_with_store( args, workspace, store ) self.assertEqual(result, 2) self.assertEqual(run_worker.await_count, 1) self.assertEqual( run_worker.await_args.args[2].name, independent.name ) orchestration = store.data["orchestrations"]["m-test"]["tasks"] self.assertEqual(orchestration[gate.name]["status"], "blocked") self.assertEqual( orchestration[dependent.name]["status"], "waiting" ) self.assertEqual( orchestration[independent.name]["status"], "complete" ) self.assertEqual( store.data["orchestrations"]["m-test"]["status"], "blocked", ) self.assertEqual( orchestration[independent.name]["archive"], str(completed_archive.resolve()), ) finally: store.close() async def test_runtime_blocker_still_drains_independent_task(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() group = workspace / "agent-task" / "m-test" gate_dir = group / "01_gate" independent_dir = group / "02_independent" gate_dir.mkdir(parents=True) independent_dir.mkdir(parents=True) gate = TaskStageTest().make_task(gate_dir) gate.name = "m-test/01_gate" gate.index = 1 gate.plan_hash = "gate-hash" independent = TaskStageTest().make_task(independent_dir) independent.name = "m-test/02_independent" independent.index = 2 independent.plan_hash = "independent-hash" completed_archive = workspace / "completed-independent" completed_archive.mkdir() (completed_archive / "complete.log").write_text( "complete\n", encoding="utf-8" ) completed_tasks: list[str] = [] async def fake_worker(workspace_path, state_store, task, *args, **kwargs): if task.name == gate.name: state_store.update_task( task, blocked="worker recovery failure limit exhausted: 10/10", ) return None completed_tasks.append(task.name) return str(completed_archive) args = SimpleNamespace( task_group="m-test", retry_blocked=False, dry_run=False ) store = dispatch.StateStore(workspace) try: with ( mock.patch.object( dispatch, "scan_tasks", side_effect=[[gate, independent], [gate]], ), mock.patch.object(dispatch, "run_worker", new=fake_worker), ): result = await dispatch.dispatch_with_store( args, workspace, store ) self.assertEqual(result, 2) self.assertEqual(completed_tasks, [independent.name]) group_state = store.data["orchestrations"]["m-test"] orchestration = group_state["tasks"] self.assertEqual(group_state["status"], "blocked") self.assertEqual(orchestration[gate.name]["status"], "blocked") self.assertEqual( orchestration[independent.name]["status"], "complete" ) store.prepare_orchestration("m-test", [gate], workspace) group_state = store.data["orchestrations"]["m-test"] self.assertEqual(group_state["status"], "running") self.assertEqual( group_state["tasks"][gate.name]["status"], "active" ) self.assertNotIn( "reason", group_state["tasks"][gate.name] ) finally: store.close() async def test_unexpected_agent_exception_drains_sibling_then_returns_three(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() group = workspace / "agent-task" / "m-test" failed_dir = group / "01_failed" sibling_dir = group / "02_sibling" failed_dir.mkdir(parents=True) sibling_dir.mkdir(parents=True) failed = TaskStageTest().make_task(failed_dir) failed.name = "m-test/01_failed" failed.index = 1 failed.plan_hash = "failed-hash" sibling = TaskStageTest().make_task(sibling_dir) sibling.name = "m-test/02_sibling" sibling.index = 2 sibling.plan_hash = "sibling-hash" completed_archive = workspace / "completed-sibling" completed_archive.mkdir() (completed_archive / "complete.log").write_text( "complete\n", encoding="utf-8" ) events: list[str] = [] async def fake_worker(workspace_path, state_store, task, *args, **kwargs): if task.name == failed.name: raise RuntimeError("unexpected control failure") events.append("sibling-started") await asyncio.sleep(0.02) events.append("sibling-finished") return str(completed_archive) args = SimpleNamespace( task_group="m-test", retry_blocked=False, dry_run=False ) store = dispatch.StateStore(workspace) try: with ( mock.patch.object( dispatch, "scan_tasks", side_effect=[[failed, sibling], [failed]], ), mock.patch.object(dispatch, "run_worker", new=fake_worker), ): result = await dispatch.dispatch_with_store( args, workspace, store ) self.assertEqual(result, 3) self.assertEqual(events, ["sibling-started", "sibling-finished"]) group_state = store.data["orchestrations"]["m-test"] self.assertEqual(group_state["status"], "running") self.assertNotEqual( group_state["tasks"][failed.name]["status"], "blocked", ) self.assertEqual( group_state["tasks"][sibling.name]["status"], "complete", ) finally: store.close() async def test_review_preflight_failure_still_drains_independent_worker(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() review_dir = workspace / "agent-task" / "m-test" / "01_review" worker_dir = workspace / "agent-task" / "m-test" / "02_worker" review_dir.mkdir(parents=True) worker_dir.mkdir(parents=True) review = TaskStageTest().make_task(review_dir) review.name = "m-test/01_review" review.index = 1 review.plan_hash = "review-hash" worker = TaskStageTest().make_task(worker_dir) worker.name = "m-test/02_worker" worker.index = 2 worker.plan_hash = "worker-hash" completed_archive = workspace / "completed-worker" completed_archive.mkdir() (completed_archive / "complete.log").write_text( "complete\n", encoding="utf-8" ) args = SimpleNamespace( task_group="m-test", retry_blocked=False, dry_run=False ) store = dispatch.StateStore(workspace) store.update_task( review, worker_done=True, selfcheck_done=True, completing_decision={ "work_unit_id": "test::plan-0::tag-TEST", "stage": "worker", "selected": { "adapter": "codex", "target": "gpt-5.6-sol", "execution_class": "cloud_model", "selfcheck_required": False, }, }, execution_class="cloud_model", ) try: with ( mock.patch.object( dispatch, "scan_tasks", side_effect=[[review, worker], [review]], ), mock.patch.object( dispatch, "ensure_review_shared_state", side_effect=RuntimeError("shared helper unavailable"), ), mock.patch.object( dispatch, "run_worker", new=mock.AsyncMock(return_value=str(completed_archive)), ) as run_worker, mock.patch.object( dispatch, "run_review", new=mock.AsyncMock() ) as run_review, ): result = await dispatch.dispatch_with_store( args, workspace, store ) self.assertEqual(result, 2) self.assertEqual(run_worker.await_count, 1) self.assertEqual(run_review.await_count, 0) orchestration = store.data["orchestrations"]["m-test"]["tasks"] self.assertEqual(orchestration[review.name]["status"], "blocked") self.assertEqual(orchestration[worker.name]["status"], "complete") finally: store.close() async def test_invalidated_complete_archive_cannot_end_with_success(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task_dir = workspace / "agent-task" / "m-test" / "01_task" task_dir.mkdir(parents=True) task = TaskStageTest().make_task(task_dir) task.name = "m-test/01_task" task.index = 1 task.plan_hash = "task-hash" archive = workspace / "completed-task" archive.mkdir() complete_log = archive / "complete.log" complete_log.write_text("complete\n", encoding="utf-8") scan_count = 0 def scan_side_effect(*args, **kwargs): nonlocal scan_count scan_count += 1 if scan_count == 1: return [task] complete_log.unlink() return [] args = SimpleNamespace( task_group="m-test", retry_blocked=False, dry_run=False ) store = dispatch.StateStore(workspace) try: with ( mock.patch.object( dispatch, "scan_tasks", side_effect=scan_side_effect ), mock.patch.object( dispatch, "run_worker", new=mock.AsyncMock(return_value=str(archive)), ), ): result = await dispatch.dispatch_with_store( args, workspace, store ) self.assertEqual(result, 2) self.assertEqual( store.data["orchestrations"]["m-test"]["status"], "blocked", ) finally: store.close() async def test_external_active_task_returns_non_terminal_exit_three(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() directory = workspace / "agent-task" / "m-test" / "01_active" directory.mkdir(parents=True) task = TaskStageTest().make_task(directory) task.name = "m-test/01_active" task.index = 1 task.plan_hash = "active-hash" locator = workspace / "locator.json" args = SimpleNamespace( task_group="m-test", retry_blocked=False, dry_run=False ) store = dispatch.StateStore(workspace) store.mark_active(task, "worker", locator) try: with ( mock.patch.object( dispatch, "scan_tasks", return_value=[task] ), mock.patch.object( dispatch, "external_active_is_live", return_value=(True, "pid=123"), ), mock.patch.object( dispatch, "run_worker", new=mock.AsyncMock() ) as run_worker, ): result = await dispatch.dispatch_with_store( args, workspace, store ) self.assertEqual(result, 3) self.assertEqual(run_worker.await_count, 0) finally: store.close() class ReviewSchedulingTest(unittest.TestCase): def test_all_ready_reviews_are_selected_without_numeric_cap(self): reviews = [ (mock.sentinel.review_a, "review"), (mock.sentinel.review_b, "review"), (mock.sentinel.review_c, "review"), ] worker = (mock.sentinel.worker, "worker") selected, deferred, reason = dispatch.select_dispatch_candidates( [*reviews, worker] ) self.assertEqual(selected, [*reviews, worker]) self.assertEqual(deferred, []) self.assertEqual(reason, "") def test_new_reviews_join_already_running_review_phase(self): reviews = [ (mock.sentinel.review_a, "review"), (mock.sentinel.review_b, "review"), ] worker = (mock.sentinel.worker, "selfcheck") selected, deferred, reason = dispatch.select_dispatch_candidates( [*reviews, worker] ) self.assertEqual(selected, [*reviews, worker]) self.assertEqual(deferred, []) self.assertEqual(reason, "") def test_only_declared_same_group_live_predecessors_delay_task(self): task = dispatch.Task( name="group/03+01,02_join", directory=Path("/tmp/group/03+01,02_join"), plan=None, review=None, user_review=None, recovery=False, deps=("01", "02"), ) live = dispatch.live_predecessors( task, { "group/01_core", "other/02_unrelated", "group/04_parallel", }, ) self.assertEqual(live, ["01"]) def test_task_without_declared_dependency_ignores_live_siblings(self): task = dispatch.Task( name="group/04_parallel", directory=Path("/tmp/group/04_parallel"), plan=None, review=None, user_review=None, recovery=False, ) self.assertEqual( dispatch.live_predecessors( task, {"group/01_core", "group/02_other"}, ), [], ) class WriteSetTest(unittest.TestCase): def test_normalizes_relative_and_absolute_aliases(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) source = workspace / "src" / "shared.go" source.parent.mkdir() source.write_text("package src\n", encoding="utf-8") relative_plan = workspace / "relative.md" absolute_plan = workspace / "absolute.md" relative_plan.write_text( "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" "| `./src/shared.go` | TEST-1 |\n", encoding="utf-8", ) absolute_plan.write_text( "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" f"| `{source}` | TEST-1 |\n", encoding="utf-8", ) relative, relative_known = dispatch.extract_write_set( relative_plan, workspace ) absolute, absolute_known = dispatch.extract_write_set( absolute_plan, workspace ) self.assertTrue(relative_known) self.assertTrue(absolute_known) self.assertEqual(relative, absolute) self.assertEqual(relative, {str(source.resolve())}) def test_recovery_restores_write_set_from_matching_archived_plan(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) task = workspace / "agent-task" / "recovery" task.mkdir(parents=True) header = "\n" (task / "plan_local_G05_2.log").write_text( header + "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" "| `src/recovery.go` | TEST-1 |\n", encoding="utf-8", ) (task / "code_review_local_G05_2.log").write_text( header + "## 코드리뷰 결과\n- 종합 판정: WARN\n", encoding="utf-8", ) (task / "plan_cloud_G09_3.log").write_text( "\n" "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" "| `src/unrelated.go` | OTHER-1 |\n", encoding="utf-8", ) [scanned] = dispatch.scan_tasks(workspace, None) self.assertTrue(scanned.write_set_known) self.assertEqual( scanned.write_set, {str((workspace / "src" / "recovery.go").resolve())}, ) self.assertEqual(scanned.errors, []) def test_recovery_without_matching_plan_does_not_block_dispatch(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) task = workspace / "agent-task" / "recovery" task.mkdir(parents=True) (task / "code_review_local_G05_2.log").write_text( "\n" "## 코드리뷰 결과\n" "- 종합 판정: WARN\n", encoding="utf-8", ) [scanned] = dispatch.scan_tasks(workspace, None) self.assertFalse(scanned.write_set_known) self.assertEqual(scanned.errors, []) def test_rejects_broad_or_outside_workspace_write_sets(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / "src").mkdir() plan = workspace / "unsafe.md" plan.write_text( "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" "| `src/` | TEST-1 |\n" "| `../outside.go` | TEST-2 |\n" "| `src/*.go` | TEST-3 |\n", encoding="utf-8", ) write_set, known = dispatch.extract_write_set(plan, workspace) self.assertFalse(known) self.assertEqual(write_set, set()) def test_review_progress_signature_ignores_dispatcher_work_log(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) task = TaskStageTest().make_task(workspace) before = dispatch.task_signature(workspace, task) (workspace / dispatch.WORK_LOG_NAME).write_text( "| FINISH | test | review |\n", encoding="utf-8" ) after_work_log = dispatch.task_signature(workspace, task) self.assertEqual(after_work_log, before) assert task.review is not None task.review.write_text( "## 코드리뷰 결과\n- 종합 판정: WARN\n", encoding="utf-8", ) after_review = dispatch.task_signature(workspace, task) self.assertNotEqual(after_review, before) class WorkLogArchiveTest(unittest.TestCase): def complete_archive( self, workspace: Path, task_name: str, month: str = "07", suffix: str = "", ) -> Path: archive = ( workspace / "agent-task" / "archive" / "2026" / month / f"{task_name}{suffix}" ) archive.mkdir(parents=True) (archive / "complete.log").write_text( f"complete {task_name}\n", encoding="utf-8", ) return archive def test_archives_split_group_log_with_next_cross_month_number(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) active_group = workspace / "agent-task" / "group" active_group.mkdir(parents=True) source = active_group / dispatch.WORK_LOG_NAME source.write_text("final timeline\n", encoding="utf-8") (active_group / "01_done").mkdir() old_group = ( workspace / "agent-task" / "archive" / "2026" / "06" / "group" ) old_group.mkdir(parents=True) (old_group / "work_log_0.log").write_text( "old timeline\n", encoding="utf-8", ) archive = self.complete_archive( workspace, "group/01_done", ) archived, errors = dispatch.archive_completed_group_work_logs( workspace, {"group/01_done"}, {"group/01_done": str(archive)}, set(), ) destination = archive.parent / "work_log_1.log" self.assertEqual(errors, {}) self.assertEqual( archived, {"group": str(destination.resolve())}, ) self.assertEqual( destination.read_text(encoding="utf-8"), "final timeline\n", ) self.assertFalse(source.exists()) self.assertFalse(active_group.exists()) def test_does_not_archive_until_every_group_task_is_complete_and_idle(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) active_group = workspace / "agent-task" / "group" active_group.mkdir(parents=True) source = active_group / dispatch.WORK_LOG_NAME source.write_text("in progress\n", encoding="utf-8") archive = self.complete_archive( workspace, "group/01_done", ) archived, errors = dispatch.archive_completed_group_work_logs( workspace, {"group/01_done", "group/02_open"}, {"group/01_done": str(archive)}, {"group/02_open"}, ) self.assertEqual(archived, {}) self.assertEqual(errors, {}) self.assertEqual( source.read_text(encoding="utf-8"), "in progress\n", ) self.assertFalse((archive.parent / "work_log_0.log").exists()) def test_archives_single_task_log_inside_its_suffixed_archive(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) active_group = workspace / "agent-task" / "single" active_group.mkdir(parents=True) source = active_group / dispatch.WORK_LOG_NAME source.write_text("single timeline\n", encoding="utf-8") archive = self.complete_archive( workspace, "single", suffix="_1", ) archived, errors = dispatch.archive_completed_group_work_logs( workspace, {"single"}, {"single": str(archive)}, set(), ) destination = archive / "work_log_0.log" self.assertEqual(errors, {}) self.assertEqual( archived, {"single": str(destination.resolve())}, ) self.assertEqual( destination.read_text(encoding="utf-8"), "single timeline\n", ) self.assertFalse(active_group.exists()) def test_closes_unmatched_start_before_archiving_completed_group(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) source = ( workspace / "agent-task" / "group" / dispatch.WORK_LOG_NAME ) locator = workspace / "runs" / "attempt" / "locator.json" dispatch.append_work_log_event( source, task_name="group/01_done", event="START", execution_id="group__01_done__p0__review__a00", role="review", attempt=0, model="codex/gpt-5.6-sol xhigh", result="running", locator=locator, ) archive = self.complete_archive( workspace, "group/01_done", ) archived, errors = dispatch.archive_completed_group_work_logs( workspace, {"group/01_done"}, {"group/01_done": str(archive)}, set(), ) destination = archive.parent / "work_log_0.log" text = destination.read_text(encoding="utf-8") self.assertEqual(errors, {}) self.assertEqual( archived, {"group": str(destination.resolve())}, ) self.assertEqual(text.count("| START |"), 1) self.assertEqual(text.count("| FINISH |"), 1) self.assertIn( "reconciled:verified-complete-archive", text, ) self.assertEqual( dispatch.unfinished_work_log_attempts(destination), [], ) def test_normalizes_legacy_work_log_already_moved_by_review(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) archive = self.complete_archive(workspace, "single") legacy = archive / dispatch.WORK_LOG_NAME legacy.write_text("legacy timeline\n", encoding="utf-8") archived, errors = dispatch.archive_completed_group_work_logs( workspace, {"single"}, {"single": str(archive)}, set(), ) destination = archive / "work_log_0.log" self.assertEqual(errors, {}) self.assertEqual( archived, {"single": str(destination.resolve())}, ) self.assertFalse(legacy.exists()) self.assertEqual( destination.read_text(encoding="utf-8"), "legacy timeline\n", ) def test_archive_failure_preserves_source_and_reports_retryable_error(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) active_group = workspace / "agent-task" / "group" active_group.mkdir(parents=True) source = active_group / dispatch.WORK_LOG_NAME source.write_text("keep me\n", encoding="utf-8") archive = self.complete_archive( workspace, "group/01_done", ) with mock.patch.object( Path, "replace", side_effect=OSError("disk full"), ): archived, errors = dispatch.archive_completed_group_work_logs( workspace, {"group/01_done"}, {"group/01_done": str(archive)}, set(), ) self.assertEqual(archived, {}) self.assertIn("group", errors) self.assertIn("disk full", errors["group"]) self.assertEqual( source.read_text(encoding="utf-8"), "keep me\n", ) def test_existing_destination_is_never_overwritten(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) active_group = workspace / "agent-task" / "group" active_group.mkdir(parents=True) source = active_group / dispatch.WORK_LOG_NAME source.write_text("new timeline\n", encoding="utf-8") archive = self.complete_archive( workspace, "group/01_done", ) destination = archive.parent / "work_log_0.log" destination.write_text("existing timeline\n", encoding="utf-8") with mock.patch.object( dispatch, "next_work_log_archive_number", return_value=0, ): archived, errors = dispatch.archive_completed_group_work_logs( workspace, {"group/01_done"}, {"group/01_done": str(archive)}, set(), ) self.assertEqual(archived, {}) self.assertIn("이미 존재한다", errors["group"]) self.assertEqual( destination.read_text(encoding="utf-8"), "existing timeline\n", ) self.assertEqual( source.read_text(encoding="utf-8"), "new timeline\n", ) def test_process_marker_recovers_liveness_when_pid_write_was_lost(self): with tempfile.TemporaryDirectory() as temporary: locator = Path(temporary) / "locator.json" locator.write_text( json.dumps( { "status": "running", "agent_process_marker": "attempt-marker", } ), encoding="utf-8", ) with mock.patch.object( dispatch, "marked_agent_process_pids", return_value=[123, 456], ): live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertTrue(live) self.assertIn("123,456", detail) with mock.patch.object( dispatch, "marked_agent_process_pids", return_value=[], ): live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertFalse(live) self.assertIn("absent from the process table", detail) def test_process_marker_finds_spawned_agent_process(self): marker = f"test-marker-{uuid.uuid4()}" environment = { **os.environ, dispatch.AGENT_PROCESS_MARKER_ENV: marker, } process = subprocess.Popen( [ sys.executable, "-c", "import time; time.sleep(5)", ], env=environment, ) try: found: list[int] = [] for _ in range(50): found = dispatch.marked_agent_process_pids(marker) if process.pid in found: break time.sleep(0.01) self.assertIn(process.pid, found) finally: process.terminate() process.wait(timeout=5) def test_orchestration_keeps_pidless_stream_evidence_active(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() stream = workspace / "stream.log" stream.write_text("reasoning\n", encoding="utf-8") locator = workspace / "locator.json" locator.write_text( json.dumps( { "status": "running", "cli": "codex", "stream_log": str(stream), } ), encoding="utf-8", ) store = dispatch.StateStore(workspace) try: store.data["orchestrations"] = { "group": { "status": "running", "tasks": { "group/01_done": { "status": "active", "archive": None, } }, } } store.data["tasks"] = { "group/01_done": { "active_locator": str(locator), } } live = dispatch.orchestration_live_agent_processes( store, "group", ) self.assertIn("group/01_done", live) self.assertIn( "time-based duplicate recovery is disabled", live["group/01_done"], ) finally: store.close() def test_restart_waits_for_live_writer_then_reconciles_and_archives(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task_directory = workspace / "agent-task" / "group" / "01_done" task_directory.mkdir(parents=True) plan = task_directory / "PLAN-local-G05.md" review = task_directory / "CODE_REVIEW-local-G05.md" plan.write_text( "\n", encoding="utf-8", ) review.write_text( "\n", encoding="utf-8", ) task = dispatch.Task( name="group/01_done", directory=task_directory, plan=plan, review=review, user_review=None, recovery=False, plan_hash=dispatch.sha256_file(plan), lane="local", grade=5, ) source = dispatch.append_milestone_event( task, event="START", execution_id="group__01_done__p0__review__a00", role="review", attempt=0, model="codex/gpt-5.6-sol xhigh", result="running", locator=workspace / "runs" / "attempt" / "locator.json", ) store = dispatch.StateStore(workspace) try: store.prepare_orchestration("group", [task], workspace) archive = ( workspace / "agent-task" / "archive" / "2026" / "07" / "group" / "01_done" ) archive.parent.mkdir(parents=True) (task_directory / "complete.log").write_text( "complete\n", encoding="utf-8", ) task_directory.rename(archive) destination = archive.parent / "work_log_0.log" wait_observations: list[bool] = [] async def observe_wait(seconds): self.assertEqual( seconds, dispatch.STREAM_HEARTBEAT_SECONDS, ) wait_observations.append( source.is_file() and not destination.exists() ) with ( mock.patch.object( dispatch, "orchestration_live_agent_processes", side_effect=[ {"group/01_done": "agent_pid=123 alive"}, {}, ], ), mock.patch.object( dispatch.asyncio, "sleep", new=observe_wait, ), ): result = asyncio.run( dispatch.dispatch_with_store( SimpleNamespace( task_group="group", retry_blocked=False, dry_run=False, ), workspace, store, ) ) self.assertEqual(result, 0) self.assertEqual(wait_observations, [True]) self.assertFalse(source.exists()) self.assertTrue(destination.is_file()) text = destination.read_text(encoding="utf-8") self.assertIn("| FINISH |", text) self.assertIn( "reconciled:verified-complete-archive", text, ) self.assertEqual( store.data["orchestrations"]["group"]["status"], "complete", ) finally: store.close() def test_dispatcher_returns_three_when_completed_log_archive_needs_retry(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() (workspace / "agent-task").mkdir() archive = self.complete_archive( workspace, "group/01_done", ) store = dispatch.StateStore(workspace) try: store.mark_orchestration_task_complete( "group", "group/01_done", archive, ) args = SimpleNamespace( task_group="group", retry_blocked=False, dry_run=False, ) with ( mock.patch.object(dispatch, "scan_tasks", return_value=[]), mock.patch.object( dispatch, "archive_completed_group_work_logs", return_value=({}, {"group": "disk full"}), ), ): result = asyncio.run( dispatch.dispatch_with_store( args, workspace, store, ) ) self.assertEqual(result, 3) self.assertEqual( store.data["orchestrations"]["group"]["status"], "running", ) finally: store.close() class OrchestrationPersistenceTest(unittest.TestCase): def make_task(self, workspace: Path, name: str = "task"): directory = workspace / "agent-task" / name directory.mkdir(parents=True) plan = directory / "PLAN-local-G05.md" review = directory / "CODE_REVIEW-local-G05.md" plan.write_text( f"\n" "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" "| `src/task.go` | TEST-1 |\n", encoding="utf-8", ) review.write_text( f"\n", encoding="utf-8", ) return dispatch.scan_tasks(workspace, None)[0] def test_complete_archive_removes_only_its_task_attempt_logs(self): with tempfile.TemporaryDirectory() as temporary: runs = Path(temporary) / "runs" completed = runs / "completed-attempt" other = runs / "other-attempt" for attempt, task_name in ((completed, "group/01_done"), (other, "group/02_open")): attempt.mkdir(parents=True) (attempt / "locator.json").write_text( json.dumps({"task": task_name}), encoding="utf-8" ) for name in ("stream.log", "heartbeat.log", "session.jsonl"): (attempt / name).write_text("evidence\n", encoding="utf-8") removed = dispatch.cleanup_completed_task_attempt_logs( runs, "group/01_done" ) self.assertEqual(removed, 1) self.assertFalse(completed.exists()) self.assertTrue(other.is_dir()) self.assertTrue((other / "stream.log").is_file()) def test_mark_complete_removes_task_attempt_logs(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() store = dispatch.StateStore(workspace) try: attempt = store.runs / "attempt" attempt.mkdir() (attempt / "locator.json").write_text( json.dumps({"task": "group/01_done"}), encoding="utf-8" ) (attempt / "stream.log").write_text("stream\n", encoding="utf-8") archive = workspace / "archive" archive.mkdir() (archive / "complete.log").write_text("complete\n", encoding="utf-8") store.mark_orchestration_task_complete( "group", "group/01_done", archive ) self.assertFalse(attempt.exists()) finally: store.close() def test_reconcile_keeps_attempt_logs_until_active_writer_exits(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() archive = workspace / "archive" archive.mkdir() (archive / "complete.log").write_text( "complete\n", encoding="utf-8", ) store = dispatch.StateStore(workspace) try: store.mark_orchestration_task_complete( "group", "group/01_done", archive, ) attempt = store.runs / "live-attempt" attempt.mkdir() (attempt / "locator.json").write_text( json.dumps({"task": "group/01_done"}), encoding="utf-8", ) (attempt / "stream.log").write_text( "still running\n", encoding="utf-8", ) completed, errors = store.reconcile_orchestration( "group", workspace, {"group/01_done"}, ) self.assertEqual(errors, {}) self.assertIn("group/01_done", completed) self.assertTrue(attempt.is_dir()) store.reconcile_orchestration("group", workspace, set()) self.assertFalse(attempt.exists()) finally: store.close() def test_attempt_log_cleanup_failure_does_not_revoke_completion(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() store = dispatch.StateStore(workspace) try: attempt = store.runs / "attempt" attempt.mkdir() (attempt / "locator.json").write_text( json.dumps({"task": "group/01_done"}), encoding="utf-8" ) archive = workspace / "archive" archive.mkdir() (archive / "complete.log").write_text( "complete\n", encoding="utf-8" ) with mock.patch.object( dispatch.shutil, "rmtree", side_effect=OSError("transient cleanup failure"), ): store.mark_orchestration_task_complete( "group", "group/01_done", archive ) with mock.patch.object( dispatch, "scan_tasks", return_value=[], ): result = asyncio.run( dispatch.dispatch_with_store( SimpleNamespace( task_group="group", retry_blocked=False, dry_run=False, ), workspace, store, ) ) record = store.data["orchestrations"]["group"]["tasks"][ "group/01_done" ] self.assertEqual(result, 3) self.assertEqual(record["status"], "complete") self.assertTrue(attempt.is_dir()) self.assertEqual( dispatch.cleanup_completed_task_attempt_logs( store.runs, "group/01_done" ), 1, ) self.assertFalse(attempt.exists()) with mock.patch.object(dispatch, "scan_tasks", return_value=[]): result = asyncio.run( dispatch.dispatch_with_store( SimpleNamespace( task_group="group", retry_blocked=False, dry_run=False, ), workspace, store, ) ) self.assertEqual(result, 0) finally: store.close() def test_attempt_log_cleanup_pending_precedes_other_terminal_blocker(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() pending_dir = workspace / "agent-task" / "group" / "02_pending" pending_dir.mkdir(parents=True) pending = TaskStageTest().make_task(pending_dir) pending.name = "group/02_pending" pending.index = 2 pending.plan_hash = "pending-hash" store = dispatch.StateStore(workspace) try: store.prepare_orchestration("group", [pending], workspace) attempt = store.runs / "attempt" attempt.mkdir() (attempt / "locator.json").write_text( json.dumps({"task": "group/01_done"}), encoding="utf-8" ) archive = workspace / "archive" archive.mkdir() (archive / "complete.log").write_text( "complete\n", encoding="utf-8" ) with ( mock.patch.object( dispatch.shutil, "rmtree", side_effect=OSError("transient cleanup failure"), ), mock.patch.object(dispatch, "scan_tasks", return_value=[]), ): store.mark_orchestration_task_complete( "group", "group/01_done", archive ) result = asyncio.run( dispatch.dispatch_with_store( SimpleNamespace( task_group="group", retry_blocked=False, dry_run=False, ), workspace, store, ) ) self.assertEqual(result, 3) self.assertEqual( store.data["orchestrations"]["group"]["status"], "running", ) self.assertTrue(attempt.is_dir()) finally: store.close() def test_external_liveness_ignores_heartbeat_mtime(self): with tempfile.TemporaryDirectory() as temporary: attempt = Path(temporary) stream = attempt / "stream.log" heartbeat = attempt / "heartbeat.log" locator = attempt / "locator.json" stream.write_text("old model output\n", encoding="utf-8") heartbeat.write_text("fresh heartbeat\n", encoding="utf-8") stale_at = time.time() - dispatch.CODEX_STREAM_STALL_SECONDS - 1 os.utime(stream, (stale_at, stale_at)) locator.write_text( json.dumps( { "status": "running", "cli": "agy", "stream_log": str(stream), "heartbeat_log": str(heartbeat), } ), encoding="utf-8", ) live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertTrue(live) self.assertIn("stream inactive=", detail) self.assertIn("time-based duplicate recovery is disabled", detail) record = json.loads(locator.read_text(encoding="utf-8")) record["agent_pid"] = 999_999_999 locator.write_text(json.dumps(record), encoding="utf-8") live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertFalse(live) self.assertIn("recorded agent process identity", detail) record.pop("agent_pid") locator.write_text(json.dumps(record), encoding="utf-8") stream.touch() live, _ = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertTrue(live) def test_external_liveness_keeps_silent_live_agent_process(self): with tempfile.TemporaryDirectory() as temporary: attempt = Path(temporary) stream = attempt / "stream.log" locator = attempt / "locator.json" stream.write_text("old model output\n", encoding="utf-8") stale_at = time.time() - (60 * 60) os.utime(stream, (stale_at, stale_at)) locator.write_text( json.dumps( { "status": "running", "cli": "pi", "agent_pid": os.getpid(), "stream_log": str(stream), } ), encoding="utf-8", ) live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertTrue(live) self.assertIn("agent_pid=", detail) def test_external_liveness_never_times_out_pidless_exact_tool_execution(self): with tempfile.TemporaryDirectory() as temporary: attempt = Path(temporary) stream = attempt / "stream.log" native = attempt / "session.jsonl" locator = attempt / "locator.json" stream.write_text("old model output\n", encoding="utf-8") native.write_text( pi_session_jsonl( [ { "type": "message", "message": { "role": "assistant", "content": [ { "type": "toolCall", "id": "slow-tool", "name": "bash", } ], }, } ] ), encoding="utf-8", ) stale_at = time.time() - (24 * 60 * 60) os.utime(stream, (stale_at, stale_at)) os.utime(native, (stale_at, stale_at)) locator.write_text( json.dumps( { "status": "running", "cli": "pi", "stream_log": str(stream), "native_session_path": str(native), } ), encoding="utf-8", ) live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertTrue(live) self.assertIn("time-based duplicate recovery is disabled", detail) record = json.loads(locator.read_text(encoding="utf-8")) record["agent_pid"] = 999_999_999 locator.write_text(json.dumps(record), encoding="utf-8") live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertFalse(live) self.assertIn("recorded agent process identity", detail) def test_external_liveness_rejects_reused_pid_identity(self): with tempfile.TemporaryDirectory() as temporary: attempt = Path(temporary) stream = attempt / "stream.log" locator = attempt / "locator.json" stream.write_text("old model output\n", encoding="utf-8") stale_at = time.time() - dispatch.CODEX_STREAM_STALL_SECONDS - 1 os.utime(stream, (stale_at, stale_at)) locator.write_text( json.dumps( { "status": "running", "cli": "agy", "agent_pid": os.getpid(), "agent_process_start_token": "not-the-current-process", "stream_log": str(stream), } ), encoding="utf-8", ) live, detail = dispatch.external_active_is_live( {"active_locator": str(locator)} ) self.assertFalse(live) self.assertIn("recorded agent process identity", detail) def test_retry_blocked_only_clears_selected_task_group(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() self.make_task(workspace, "alpha/01_task") self.make_task(workspace, "beta/01_task") tasks = { task.name: task for task in dispatch.scan_tasks(workspace, None) } alpha = tasks["alpha/01_task"] beta = tasks["beta/01_task"] store = dispatch.StateStore(workspace) try: for task in (alpha, beta): store.update_task( task, blocked="recovery failure limit exhausted: 10/10", review_no_progress=10, selfcheck_incomplete=10, recovery_failures={"worker": 10}, ) store.clear_blocked("alpha") alpha_state = store.task_state(alpha) self.assertIsNone(alpha_state["blocked"]) self.assertEqual(alpha_state["review_no_progress"], 0) self.assertEqual(alpha_state["selfcheck_incomplete"], 0) self.assertEqual(alpha_state["recovery_failures"], {}) beta_state = store.task_state(beta) self.assertIsNotNone(beta_state["blocked"]) self.assertEqual(beta_state["review_no_progress"], 10) self.assertEqual(beta_state["selfcheck_incomplete"], 10) self.assertEqual(beta_state["recovery_failures"], {"worker": 10}) finally: store.close() def test_legacy_promotion_recovery_requires_older_source_and_typed_event(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() self.make_task(workspace, "alpha/01_task") task = dispatch.scan_tasks(workspace, None)[0] runs = workspace / ".git" / "agent-task-dispatcher" / "runs" locators = write_legacy_quota_attempts(runs, task) locator = locators[-1] state = { "blocked": ( "worker recovery failure limit exhausted: 10/10 " f"locator={locator}" ), "recovery_failures": {"worker": 10}, } recovery = dispatch.legacy_promotion_recovery( runs, task, state, ) self.assertIsNotNone(recovery) assert recovery is not None self.assertEqual(recovery.failure_class, "provider-quota") self.assertEqual(recovery.evidence_source, "claude:stdout") record = json.loads(locator.read_text(encoding="utf-8")) record["dispatcher_source_sha256"] = ( dispatch.DISPATCHER_SOURCE_SHA256 ) locator.write_text(json.dumps(record), encoding="utf-8") self.assertIsNone( dispatch.legacy_promotion_recovery(runs, task, state) ) def test_legacy_promotion_recovery_rejects_mixed_attempt_history(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() self.make_task(workspace, "alpha/01_task") task = dispatch.scan_tasks(workspace, None)[0] runs = workspace / ".git" / "agent-task-dispatcher" / "runs" locators = write_legacy_quota_attempts(runs, task) mixed_stream = locators[4].parent / "stream.log" mixed_stream.write_text( "[stdout] " + json.dumps( { "type": "assistant", "message": { "content": "You've hit your session limit" }, } ) + "\n", encoding="utf-8", ) state = { "blocked": ( "worker recovery failure limit exhausted: 10/10 " f"locator={locators[-1]}" ), "recovery_failures": {"worker": 10}, } self.assertIsNone( dispatch.legacy_promotion_recovery(runs, task, state) ) def test_persisted_legacy_promotion_survives_dispatcher_restart(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() self.make_task(workspace, "alpha/01_task") task = dispatch.scan_tasks(workspace, None)[0] runs = workspace / ".git" / "agent-task-dispatcher" / "runs" locator = write_legacy_quota_attempts(runs, task)[-1] blocked_state = { "blocked": ( "worker recovery failure limit exhausted: 10/10 " f"locator={locator}" ), "recovery_failures": {"worker": 10}, } recovery = dispatch.legacy_promotion_recovery( runs, task, blocked_state, ) assert recovery is not None restarted_state = { "blocked": None, "recovery_failures": {"worker": 1}, "legacy_terminal_reclassification": { "role": recovery.role, "failure_class": recovery.failure_class, "evidence_source": recovery.evidence_source, "prior_dispatcher_sha256": recovery.prior_dispatcher_sha256, "current_dispatcher_sha256": dispatch.DISPATCHER_SOURCE_SHA256, "locator": str(recovery.locator), "failed_cli": recovery.failed_cli, "failed_model": recovery.failed_model, "failed_reasoning_effort": recovery.failed_reasoning_effort, }, } restored = ( dispatch.pending_persisted_legacy_promotion_recovery( task, restarted_state, ) ) self.assertIsNotNone(restored) assert restored is not None self.assertEqual(restored.failed_cli, "claude") self.assertEqual( dispatch.promoted_spec( dispatch.failed_spec_from_recovery(restored), 0, ).model, "gpt-5.6-terra", ) def test_persisted_legacy_agy_promotion_survives_dispatcher_restart(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() self.make_task(workspace, "alpha/01_task") task = dispatch.scan_tasks(workspace, None)[0] runs = workspace / ".git" / "agent-task-dispatcher" / "runs" locator = write_legacy_quota_attempts( runs, task, cli="agy", model="Gemini 3.6 Flash (High)", reasoning_effort=None, )[-1] blocked_state = { "blocked": ( "worker recovery failure limit exhausted: 10/10 " f"locator={locator}" ), "recovery_failures": {"worker": 10}, } recovery = dispatch.legacy_promotion_recovery( runs, task, blocked_state, ) assert recovery is not None restarted_state = { "blocked": None, "recovery_failures": {"worker": 1}, "legacy_terminal_reclassification": { "role": recovery.role, "failure_class": recovery.failure_class, "evidence_source": recovery.evidence_source, "prior_dispatcher_sha256": recovery.prior_dispatcher_sha256, "current_dispatcher_sha256": dispatch.DISPATCHER_SOURCE_SHA256, "locator": str(recovery.locator), "failed_cli": recovery.failed_cli, "failed_model": recovery.failed_model, "failed_reasoning_effort": recovery.failed_reasoning_effort, }, } restored = ( dispatch.pending_persisted_legacy_promotion_recovery( task, restarted_state, ) ) self.assertIsNotNone(restored) assert restored is not None self.assertEqual(restored.failed_cli, "agy") self.assertEqual(restored.evidence_source, "agy:cli-log") self.assertEqual( dispatch.promoted_spec( dispatch.failed_spec_from_recovery(restored), 0, ).cli, "claude", ) def test_corrupt_persistent_state_fails_closed_and_releases_lock(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) state_root = workspace / ".git" / "agent-task-dispatcher" state_root.mkdir(parents=True) state_path = state_root / "state.json" state_path.write_text("{broken", encoding="utf-8") with self.assertRaisesRegex( RuntimeError, "dispatcher state를 읽을 수 없다" ): dispatch.StateStore(workspace) state_path.write_text("{}\n", encoding="utf-8") reopened = dispatch.StateStore(workspace) reopened.close() def test_live_workspace_lock_is_non_terminal_exit_three(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() (workspace / "agent-task").mkdir() owner = dispatch.StateStore(workspace) args = SimpleNamespace( workspace=str(workspace), task_group=None, dry_run=False, retry_blocked=False, ) try: with mock.patch.object(dispatch, "parse_args", return_value=args): result = dispatch.main() self.assertEqual(result, 3) finally: owner.close() def test_unexpected_dispatcher_exception_is_non_terminal_exit_three(self): args = SimpleNamespace( workspace=".", task_group=None, dry_run=False, retry_blocked=False, ) with ( mock.patch.object(dispatch, "parse_args", return_value=args), mock.patch.object( dispatch, "dispatch", new=mock.AsyncMock(side_effect=RuntimeError("transient failure")), ), ): result = dispatch.main() self.assertEqual(result, 3) def test_scheduler_exception_waits_for_running_agent_tasks(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() (workspace / "agent-task").mkdir() args = SimpleNamespace( workspace=str(workspace), task_group=None, dry_run=False, retry_blocked=False, ) events: list[str] = [] async def failing_scheduler(args, workspace, store): async def running_agent(): events.append("started") await asyncio.sleep(0.02) events.append("finished") asyncio.create_task(running_agent()) await asyncio.sleep(0) raise dispatch.DispatcherTerminalStateError("scheduler failed") with mock.patch.object( dispatch, "dispatch_with_store", new=failing_scheduler, ): with self.assertRaisesRegex( dispatch.DispatcherInterruptedWithActiveWork, "scheduler failed", ): asyncio.run(dispatch.dispatch(args)) self.assertEqual(events, ["started", "finished"]) def test_dry_run_retry_blocked_does_not_clear_persistent_limits(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) store.update_task( task, blocked="recovery failure limit exhausted: 10/10", review_no_progress=10, selfcheck_incomplete=10, recovery_failures={"review": 10}, ) store.close() args = SimpleNamespace( workspace=str(workspace), task_group=None, dry_run=True, retry_blocked=True, ) result = asyncio.run(dispatch.dispatch(args)) self.assertEqual(result, 2) reopened = dispatch.StateStore(workspace) try: state = reopened.task_state(task) self.assertEqual( state["blocked"], "recovery failure limit exhausted: 10/10", ) self.assertEqual(state["review_no_progress"], 10) self.assertEqual(state["selfcheck_incomplete"], 10) self.assertEqual( state["recovery_failures"], {"review": 10} ) finally: reopened.close() def test_child_restart_detects_task_that_disappeared_without_complete_log(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) first = dispatch.StateStore(workspace) first.prepare_orchestration("__all__", [task], workspace) first.close() task.plan.unlink() task.review.unlink() task.directory.rmdir() restarted = dispatch.StateStore(workspace) restarted.prepare_orchestration("__all__", [], workspace) completed, errors = restarted.reconcile_orchestration( "__all__", workspace, set() ) self.assertEqual(completed, {}) self.assertIn(task.name, errors) self.assertIn("새 complete.log archive 모두에서 사라졌다", errors[task.name]) restarted.close() def test_child_restart_recovers_new_complete_archive_not_baseline_archive(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() old_archive = workspace / "agent-task" / "archive" / "2026" / "06" / "task" old_archive.mkdir(parents=True) (old_archive / "complete.log").write_text("old\n", encoding="utf-8") task = self.make_task(workspace) first = dispatch.StateStore(workspace) first.prepare_orchestration("__all__", [task], workspace) first.close() new_archive = workspace / "agent-task" / "archive" / "2026" / "07" / "task_1" new_archive.parent.mkdir(parents=True) (task.directory / "complete.log").write_text("new\n", encoding="utf-8") task.directory.rename(new_archive) restarted = dispatch.StateStore(workspace) restarted.prepare_orchestration("__all__", [], workspace) completed, errors = restarted.reconcile_orchestration( "__all__", workspace, set() ) self.assertEqual(errors, {}) self.assertEqual(completed, {"task": str(new_archive.resolve())}) restarted.close() def test_preexisting_incomplete_archive_cannot_become_false_new_completion(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() incomplete_archive = ( workspace / "agent-task" / "archive" / "2026" / "06" / "task" ) incomplete_archive.mkdir(parents=True) task = self.make_task(workspace) first = dispatch.StateStore(workspace) first.prepare_orchestration("__all__", [task], workspace) first.close() task.plan.unlink() task.review.unlink() task.directory.rmdir() (incomplete_archive / "complete.log").write_text( "late unrelated completion\n", encoding="utf-8", ) restarted = dispatch.StateStore(workspace) restarted.prepare_orchestration("__all__", [], workspace) completed, errors = restarted.reconcile_orchestration( "__all__", workspace, set() ) self.assertEqual(completed, {}) self.assertIn(task.name, errors) restarted.close() def test_dry_run_does_not_create_persistent_orchestration_state(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() self.make_task(workspace) args = SimpleNamespace( workspace=str(workspace), task_group=None, dry_run=True, retry_blocked=False, ) result = asyncio.run(dispatch.dispatch(args)) self.assertEqual(result, 0) state_path = workspace / ".git" / "agent-task-dispatcher" / "state.json" self.assertFalse(state_path.exists()) def test_explicit_unobserved_task_group_cannot_report_success(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() (workspace / "agent-task").mkdir() args = SimpleNamespace( workspace=str(workspace), task_group="missing-group", dry_run=False, retry_blocked=False, ) result = asyncio.run(dispatch.dispatch(args)) self.assertEqual(result, 2) state = json.loads( ( workspace / ".git" / "agent-task-dispatcher" / "state.json" ).read_text(encoding="utf-8") ) self.assertEqual( state["orchestrations"]["missing-group"]["status"], "blocked", ) class RouteDecisionPersistenceTest(unittest.TestCase): def make_task( self, workspace: Path, name: str = "route/01_unit", *, lane: str = "local", grade: int = 5, ): directory = workspace / "agent-task" / name directory.mkdir(parents=True) header = f"\n" (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text( header + "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" + "| src/route.py | ROUTE-1 |\n", encoding="utf-8", ) (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text( header, encoding="utf-8", ) return next(task for task in dispatch.scan_tasks(workspace, None) if task.name == name) def test_reopen_body_edit_and_generation_reset_preserve_or_reset_pin(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) first = dispatch.StateStore(workspace) try: initial, worker_spec = dispatch.persisted_execution_decision( first, task, stage="worker" ) review, review_spec = dispatch.persisted_execution_decision( first, task, stage="review" ) self.assertEqual(initial["transition"]["trigger"], "initial") self.assertEqual(worker_spec.cli, "pi") self.assertEqual(review_spec.cli, "codex") self.assertEqual(review["transition"]["trigger"], "initial") self.assertEqual( [entry["stage"] for entry in first.task_state(task)["route_transition_history"]], ["worker", "review"], ) finally: first.close() reopened = dispatch.StateStore(workspace) try: reopened_task = dispatch.scan_tasks(workspace, None)[0] resumed, resumed_spec = dispatch.persisted_execution_decision( reopened, reopened_task, stage="worker" ) self.assertEqual(resumed["transition"]["trigger"], "resume") self.assertEqual(resumed_spec.display, "pi/iop/ornith:35b") assert reopened_task.plan is not None reopened_task.plan.write_text( reopened_task.plan.read_text(encoding="utf-8") + "\n본문만 변경\n", encoding="utf-8", ) body_edited = dispatch.scan_tasks(workspace, None)[0] self.assertEqual(body_edited.plan_hash, reopened_task.plan_hash) body_resume, _ = dispatch.persisted_execution_decision( reopened, body_edited, stage="worker" ) self.assertEqual(body_resume["transition"]["trigger"], "resume") assert body_edited.plan is not None and body_edited.review is not None for path in (body_edited.plan, body_edited.review): path.write_text( path.read_text(encoding="utf-8").replace("plan=0", "plan=1"), encoding="utf-8", ) next_generation = dispatch.scan_tasks(workspace, None)[0] reset, _ = dispatch.persisted_execution_decision( reopened, next_generation, stage="worker" ) self.assertEqual(reset["transition"]["trigger"], "initial") reset_history = reopened.task_state(next_generation)[ "route_transition_history" ] self.assertEqual(len(reset_history), 1) self.assertEqual(reset_history[0]["stage"], "worker") self.assertEqual(reset_history[0]["transition"], "initial") self.assertEqual( reset_history[0]["work_unit_id"], reset["work_unit_id"] ) self.assertEqual(reset_history[0]["selected"], reset["selected"]) self.assertEqual(reset_history[0]["decision"], reset["decision"]) self.assertEqual(reset_history[0]["quota"], reset["quota"]) self.assertNotIn("rule_id", reset_history[0]) self.assertNotIn("priority", reset_history[0]) self.assertNotIn("quota_snapshot", reset_history[0]) finally: reopened.close() def test_resume_keeps_pin_across_kst_boundary(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) task = self.make_task(workspace) initial = dispatch.select_execution_decision( task, stage="worker", evaluated_at=datetime(2026, 7, 25, 6, 59, tzinfo=dispatch.KST), ) resumed = dispatch.select_execution_decision( task, stage="worker", prior_decision=initial, evaluated_at=datetime(2026, 7, 25, 7, 0, tzinfo=dispatch.KST), ) self.assertEqual(resumed["transition"]["trigger"], "resume") self.assertEqual(resumed["selected"], initial["selected"]) self.assertTrue(resumed["decision"]["pinned"]) def test_rejects_tampered_canonical_target_and_selector_load_error(self): with tempfile.TemporaryDirectory() as temporary: task = self.make_task(Path(temporary)) decision = dispatch.select_execution_decision(task, stage="worker") tampered = json.loads(json.dumps(decision)) tampered["selected"] = { "adapter": "agy", "target": "untrusted-target", "execution_class": "local_model", "selfcheck_required": False, } with self.assertRaises(dispatch.ExecutionDecisionError): dispatch.agent_spec_from_decision(tampered) for failure in (OSError("load failed"), SyntaxError("broken selector"), RuntimeError("loader crashed")): with self.subTest(failure=type(failure).__name__): with mock.patch.object(dispatch, "_selector_module", side_effect=failure): with self.assertRaises(dispatch.ExecutionDecisionError): dispatch.select_execution_decision(task, stage="worker") def test_malformed_or_exhausted_state_blocks_only_its_task(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() exhausted = self.make_task( workspace, "route/01_exhausted", lane="cloud", grade=7 ) healthy = self.make_task(workspace, "route/02_healthy") store = dispatch.StateStore(workspace) try: store.update_task( exhausted, quota_snapshot={ "source": "test", "targets": [{ "adapter": "claude", "target": "claude-opus-4-8", "status": "exhausted", }], }, ) asyncio.run(dispatch.run_worker(workspace, store, exhausted)) self.assertIn("no_eligible_target", store.task_state(exhausted)["blocked"]) _, healthy_spec = dispatch.persisted_execution_decision( store, healthy, stage="worker" ) self.assertEqual(healthy_spec.cli, "pi") store.update_task( healthy, execution_decisions={"worker": {"malformed": True}} ) with self.assertRaises(dispatch.ExecutionDecisionError): dispatch.persisted_execution_decision(store, healthy, stage="worker") finally: store.close() class DispatcherConvergenceSimulationTest(unittest.IsolatedAsyncioTestCase): def write_task( self, workspace: Path, task_name: str, source_path: str, ) -> None: directory = workspace / "agent-task" / task_name directory.mkdir(parents=True) header = f"\n" (directory / "PLAN-local-G05.md").write_text( header + "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" + f"| `{source_path}` | SIM-1 |\n", encoding="utf-8", ) (directory / "CODE_REVIEW-local-G05.md").write_text( header, encoding="utf-8", ) async def test_parallel_multi_task_followup_dependency_and_terminal_completion(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() (workspace / "agent-task").mkdir() self.write_task(workspace, "sim/01_alpha", "src/alpha.go") self.write_task(workspace, "sim/02_beta", "src/beta.go") self.write_task(workspace, "sim/03+01,02_join", "src/join.go") self.write_task(workspace, "sim/04_conflict", "./src/alpha.go") work_log = workspace / "agent-task" / "sim" / dispatch.WORK_LOG_NAME work_log.write_text("final timeline\n", encoding="utf-8") active = {"worker": set(), "selfcheck": set(), "review": set()} maximum = {"worker": 0, "selfcheck": 0, "review": 0} overlap_violations: list[tuple[str, set[str]]] = [] review_attempts: dict[str, int] = {} def enter(stage: str, task_name: str) -> None: active[stage].add(task_name) maximum[stage] = max(maximum[stage], len(active[stage])) if {"sim/01_alpha", "sim/04_conflict"} <= active[stage]: overlap_violations.append((stage, set(active[stage]))) def leave(stage: str, task_name: str) -> None: active[stage].remove(task_name) async def fake_worker(workspace_path, store, task, *args, **kwargs): enter("worker", task.name) try: await asyncio.sleep(0.005) completing_decision = { "work_unit_id": dispatch.work_unit_id_from_file(task.plan), "stage": "worker", "selected": { "adapter": "pi", "target": "iop/ornith:35b", "execution_class": "local_model", "selfcheck_required": True, }, } store.update_task( task, worker_done=True, worker_cli="pi", worker_model="ornith:35b", completing_decision=completing_decision, execution_class="local_model", selfcheck_done=False, blocked=None, ) finally: leave("worker", task.name) async def fake_selfcheck(workspace_path, store, task, *args, **kwargs): enter("selfcheck", task.name) try: await asyncio.sleep(0.005) store.update_task(task, selfcheck_done=True, blocked=None) finally: leave("selfcheck", task.name) async def fake_review(workspace_path, store, task, *args, **kwargs): enter("review", task.name) try: await asyncio.sleep( 0.002 if task.name == "sim/02_beta" else 0.015 ) attempt = review_attempts.get(task.name, 0) + 1 review_attempts[task.name] = attempt if task.name == "sim/01_alpha" and attempt == 1: for path in (task.plan, task.review): assert path is not None path.write_text( path.read_text(encoding="utf-8").replace( "plan=0", "plan=1" ), encoding="utf-8", ) return None archive = ( workspace_path / "agent-task" / "archive" / "2026" / "07" / task.name ) archive.parent.mkdir(parents=True, exist_ok=True) (task.directory / "complete.log").write_text( "simulation complete\n", encoding="utf-8" ) task.directory.rename(archive) return str(archive) finally: leave("review", task.name) args = SimpleNamespace( workspace=str(workspace), task_group="sim", dry_run=False, retry_blocked=False, ) with ( mock.patch.object(dispatch, "run_worker", new=fake_worker), mock.patch.object(dispatch, "run_selfcheck", new=fake_selfcheck), mock.patch.object(dispatch, "run_review", new=fake_review), mock.patch.object(dispatch, "ensure_review_shared_state"), mock.patch.object( dispatch, "scan_tasks", wraps=dispatch.scan_tasks ) as scan_tasks, ): result = await asyncio.wait_for(dispatch.dispatch(args), timeout=2) self.assertEqual(result, 0) self.assertGreaterEqual(maximum["worker"], 2) self.assertGreaterEqual(maximum["selfcheck"], 2) self.assertGreaterEqual(maximum["review"], 2) self.assertEqual( {stage for stage, _ in overlap_violations}, {"worker", "selfcheck", "review"}, ) self.assertGreater(scan_tasks.call_count, 1) self.assertLessEqual(scan_tasks.call_count, 5) self.assertTrue( any( call.kwargs.get("exclude_names") for call in scan_tasks.call_args_list ), "completion-triggered scans must exclude still-running tasks", ) self.assertEqual(review_attempts["sim/01_alpha"], 2) self.assertEqual(review_attempts["sim/02_beta"], 1) self.assertEqual(review_attempts["sim/03+01,02_join"], 1) self.assertEqual(review_attempts["sim/04_conflict"], 1) archive_root = workspace / "agent-task" / "archive" / "2026" / "07" / "sim" for subtask in ("01_alpha", "02_beta", "03+01,02_join", "04_conflict"): self.assertTrue((archive_root / subtask / "complete.log").is_file()) self.assertFalse(work_log.exists()) self.assertEqual( (archive_root / "work_log_0.log").read_text(encoding="utf-8"), "final timeline\n", ) class DynamicFailoverBudgetTest(unittest.TestCase): def make_task(self, workspace: Path): directory = workspace / "agent-task" / "budget/01_unit" directory.mkdir(parents=True) header = "\n" (directory / "PLAN-local-G07.md").write_text(header, encoding="utf-8") (directory / "CODE_REVIEW-local-G07.md").write_text(header, encoding="utf-8") return dispatch.scan_tasks(workspace, None)[0] def test_context_package_keeps_artifacts_and_blocks_cross_adapter_native_session(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) attempt = workspace / "attempt" attempt.mkdir() raw, normalized, native = attempt / "stream.log", attempt / "normalized-output.log", attempt / "session.jsonl" raw.write_text("raw\n", encoding="utf-8") normalized.write_text("normalized\n", encoding="utf-8") native.write_text("{}\n", encoding="utf-8") locator = attempt / "locator.json" record = {"task": task.name, "workspace": str(workspace), "plan_path": str(task.plan), "stream_log": str(raw), "normalized_output_log": str(normalized), "native_session_path": str(native)} locator.write_text(json.dumps(record), encoding="utf-8") pi = dispatch.AgentSpec("pi", "ornith:35b", "pi/iop/ornith:35b", local_pi=True) codex = dispatch.AgentSpec("codex", "gpt-5.6-sol", "codex/gpt-5.6-sol xhigh") logical = dispatch.build_context_package(workspace, task, locator, previous_spec=pi, next_spec=codex) self.assertEqual(logical["resume_mode"], "logical") self.assertNotIn("native_session_path", logical) native_package = dispatch.build_context_package(workspace, task, locator, previous_spec=pi, next_spec=pi) self.assertEqual(native_package["native_session_path"], str(native.resolve())) external = workspace / "external.log" external.write_text("outside\n", encoding="utf-8") record["stream_log"] = str(external) locator.write_text(json.dumps(record), encoding="utf-8") with self.assertRaises(dispatch.ExecutionDecisionError): dispatch.build_context_package(workspace, task, locator, previous_spec=pi, next_spec=codex) record["stream_log"] = str(raw) record["workspace"] = "" locator.write_text(json.dumps(record), encoding="utf-8") with self.assertRaises(dispatch.ExecutionDecisionError): dispatch.build_context_package(workspace, task, locator, previous_spec=pi, next_spec=codex) record["workspace"] = str(workspace) locator.write_text(json.dumps(record), encoding="utf-8") normalized.unlink() with self.assertRaises(dispatch.ExecutionDecisionError): dispatch.build_context_package(workspace, task, locator, previous_spec=pi, next_spec=codex) def test_primary_and_alternate_share_budget_across_reopen(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)") locator_gemini = self.make_attempt_locator(workspace, task, gemini_spec) laguna_spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True) locator_laguna = self.make_attempt_locator(workspace, task, laguna_spec) initial_store = dispatch.StateStore(workspace) try: dispatch.persisted_execution_decision(initial_store, task, stage="worker", evaluated_at=daytime) finally: initial_store.close() store = dispatch.StateStore(workspace) try: invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if spec.cli == "agy": return (1, "provider-quota", locator_gemini) if len(invoked_specs) == 2: return (1, "generic-error", locator_laguna) raise asyncio.CancelledError() with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): try: asyncio.run(dispatch.run_escalating(workspace, store, task, "worker", gemini_spec)) except asyncio.CancelledError: pass state1 = store.task_state(task) decisions1 = state1["execution_decisions"] history1 = state1["route_transition_history"] worker_budget1 = dispatch.StageFailureBudget.from_decision(store, task, decisions1["worker"]) self.assertEqual([s.cli for s in invoked_specs[:2]], ["agy", "pi"]) self.assertEqual([h["transition"] for h in history1], ["initial", "provider-quota"]) self.assertEqual(worker_budget1.count(), 2) finally: store.close() reopened = dispatch.StateStore(workspace) try: worker_budget_reopened = dispatch.StageFailureBudget.from_decision(reopened, task, decisions1["worker"]) current_count = worker_budget_reopened.count() needed_failures = 10 - current_count locators = [workspace / f"failure-{i}.json" for i in range(needed_failures)] with ( mock.patch.object( dispatch, "invoke", new=mock.AsyncMock( side_effect=[(1, "generic-error", loc) for loc in locators] ), ) as invoke, mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): success, final_loc = asyncio.run( dispatch.run_escalating(workspace, reopened, task, "worker", laguna_spec) ) self.assertFalse(success) self.assertEqual(invoke.await_count, 8) self.assertTrue(all(call.args[4] == laguna_spec for call in invoke.await_args_list)) self.assertEqual(final_loc, locators[-1]) self.assertIn("recovery failure limit exhausted", reopened.task_state(task)["blocked"]) state2 = reopened.task_state(task) worker_budget2 = dispatch.StageFailureBudget.from_decision(reopened, task, decisions1["worker"]) review_decision = dispatch.select_execution_decision(task, stage="review", evaluated_at=daytime) review_budget2 = dispatch.StageFailureBudget.from_decision(reopened, task, review_decision) self.assertEqual(worker_budget2.count(), 10) self.assertEqual(review_budget2.count(), 0) raw_entry = state2.get("stage_failure_budgets", {}).get(worker_budget2.key, {}) self.assertEqual(raw_entry.get("last_target"), {"adapter": "pi", "target": "iop/laguna-s:2.1"}) self.assertEqual(raw_entry.get("last_transition"), "provider-quota") self.assertEqual(state2["execution_decisions"], decisions1) self.assertEqual([h["transition"] for h in state2["route_transition_history"]], ["initial", "provider-quota"]) finally: reopened.close() def test_success_resets_only_current_stage_budget(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)") locator = self.make_attempt_locator(workspace, task, gemini_spec) worker_decision = dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime)[0] review_decision = dispatch.select_execution_decision(task, stage="review", evaluated_at=daytime) worker_budget = dispatch.StageFailureBudget.from_decision(store, task, worker_decision) review_budget = dispatch.StageFailureBudget.from_decision(store, task, review_decision) worker_budget.record_failure(target={"adapter": "agy", "target": "Gemini 3.6 Flash (Medium)"}, transition="generic-error") review_budget.record_failure(target={"adapter": "codex", "target": "gpt-5.6-sol"}, transition="generic-error") self.assertEqual(worker_budget.count(), 1) self.assertEqual(review_budget.count(), 1) async def mock_invoke(*args, **kwargs): return (0, None, locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): success, final_loc = asyncio.run( dispatch.run_escalating(workspace, store, task, "worker", gemini_spec) ) self.assertTrue(success) self.assertEqual(final_loc, locator) self.assertEqual(worker_budget.count(), 0) self.assertEqual(review_budget.count(), 1) finally: store.close() def make_attempt_locator(self, workspace: Path, task: dispatch.Task, spec: dispatch.AgentSpec) -> Path: attempt = workspace / f"attempt-{spec.cli}" attempt.mkdir(parents=True, exist_ok=True) raw, normalized = attempt / "stream.log", attempt / "normalized-output.log" raw.write_text("raw log\n", encoding="utf-8") normalized.write_text("normalized output\n", encoding="utf-8") locator = attempt / "locator.json" record = { "task": task.name, "workspace": str(workspace), "plan_path": str(task.plan), "stream_log": str(raw), "normalized_output_log": str(normalized), "cli": spec.cli, "model": spec.model, "spec": {"adapter": spec.cli, "target": spec.model}, } locator.write_text(json.dumps(record), encoding="utf-8") return locator class DispatcherCanonicalFailoverIntegrationTest(unittest.IsolatedAsyncioTestCase): def make_task(self, workspace: Path, lane: str = "local", grade: int = 8) -> dispatch.Task: directory = workspace / "agent-task" / "failover/01_unit" directory.mkdir(parents=True, exist_ok=True) header = "\n" (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") tasks = dispatch.scan_tasks(workspace, None) return tasks[0] def make_attempt_locator(self, workspace: Path, task: dispatch.Task, spec: dispatch.AgentSpec) -> Path: attempt = workspace / f"attempt-{spec.cli}" attempt.mkdir(parents=True, exist_ok=True) raw, normalized = attempt / "stream.log", attempt / "normalized-output.log" raw.write_text("raw log\n", encoding="utf-8") normalized.write_text("normalized output\n", encoding="utf-8") locator = attempt / "locator.json" record = { "task": task.name, "workspace": str(workspace), "plan_path": str(task.plan), "stream_log": str(raw), "normalized_output_log": str(normalized), "cli": spec.cli, "model": spec.model, "spec": {"adapter": spec.cli, "target": spec.model}, } locator.write_text(json.dumps(record), encoding="utf-8") return locator async def test_invalid_logical_context_does_not_commit_or_promote(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) cases = [ ("no_locator", lambda loc, ws, tk: ws / "nonexistent.json"), ("invalid_json", lambda loc, ws, tk: (loc.write_text("invalid json", encoding="utf-8"), loc)[1]), ("workspace_mismatch", lambda loc, ws, tk: ( loc.write_text(json.dumps({ "task": tk.name, "workspace": str(ws / "other"), "plan_path": str(tk.plan), "stream_log": str(loc.parent / "stream.log"), "normalized_output_log": str(loc.parent / "normalized-output.log"), }), encoding="utf-8"), loc )[1]), ("task_mismatch", lambda loc, ws, tk: ( loc.write_text(json.dumps({ "task": "other/task", "workspace": str(ws), "plan_path": str(tk.plan), "stream_log": str(loc.parent / "stream.log"), "normalized_output_log": str(loc.parent / "normalized-output.log"), }), encoding="utf-8"), loc )[1]), ("plan_mismatch", lambda loc, ws, tk: ( loc.write_text(json.dumps({ "task": tk.name, "workspace": str(ws), "plan_path": str(ws / "other.md"), "stream_log": str(loc.parent / "stream.log"), "normalized_output_log": str(loc.parent / "normalized-output.log"), }), encoding="utf-8"), loc )[1]), ("missing_raw_artifact", lambda loc, ws, tk: ( (loc.parent / "stream.log").unlink(), loc )[1]), ("missing_normalized_artifact", lambda loc, ws, tk: ( (loc.parent / "normalized-output.log").unlink(), loc )[1]), ] for name, modifier in cases: with self.subTest(variant=name), tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)") base_locator = self.make_attempt_locator(workspace, task, gemini_spec) target_locator = modifier(base_locator, workspace, task) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) return (1, "provider-quota", target_locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) initial_decisions = store.task_state(task)["execution_decisions"]["worker"] initial_history = list(store.task_state(task)["route_transition_history"]) success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", gemini_spec) self.assertFalse(success) self.assertEqual(len(invoked_specs), 1) self.assertEqual(invoked_specs[0].cli, "agy") state = store.task_state(task) self.assertIn("worker selector decision 실패", state.get("blocked", "")) self.assertEqual(state["execution_decisions"]["worker"]["selected"], initial_decisions["selected"]) self.assertEqual(state["route_transition_history"], initial_history) finally: store.close() async def test_day_gemini_zero_exit_quota_continues_on_laguna_with_logical_context(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)") locator = self.make_attempt_locator(workspace, task, gemini_spec) invoked_specs = [] invoked_prompts = [] async def mock_invoke(*args, **kwargs): spec = args[4] prompt = args[5] invoked_specs.append(spec) invoked_prompts.append(prompt) if spec.cli == "agy": return (0, "provider-quota", locator) return (0, None, locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", gemini_spec) self.assertTrue(success) self.assertEqual(len(invoked_specs), 2) self.assertEqual(invoked_specs[0].cli, "agy") self.assertEqual(invoked_specs[1].cli, "pi") self.assertTrue(invoked_specs[1].local_pi) laguna_prompt = invoked_prompts[1] self.assertIn(str(task.plan.resolve()), laguna_prompt) self.assertIn(str(locator.resolve()), laguna_prompt) self.assertIn(str(workspace.resolve()), laguna_prompt) self.assertIn(str((locator.parent / "stream.log").resolve()), laguna_prompt) self.assertIn(str((locator.parent / "normalized-output.log").resolve()), laguna_prompt) state = store.task_state(task) decisions = state["execution_decisions"]["worker"] self.assertEqual(decisions["selected"]["adapter"], "pi") self.assertEqual(decisions["transition"]["trigger"], "provider-quota") finally: store.close() async def test_cloud_g07_provider_quota_promotes_claude_to_codex_without_no_failover_block(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: claude_spec = dispatch.AgentSpec("claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh") terra_spec = dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ) claude_locator = self.make_attempt_locator( workspace, task, claude_spec ) terra_locator = self.make_attempt_locator( workspace, task, terra_spec ) invoked_specs = [] invoked_prompts = [] transition_budget_counts = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) invoked_prompts.append(args[5]) if spec.cli == "claude": return (1, "provider-quota", claude_locator) state = store.task_state(task) decision = state["execution_decisions"]["worker"] budget = dispatch.StageFailureBudget.from_decision( store, task, decision ) transition_budget_counts.append(budget.count()) return (0, None, terra_locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", claude_spec) self.assertTrue(success) self.assertEqual(final_loc, terra_locator) self.assertEqual(len(invoked_specs), 2) self.assertEqual(invoked_specs, [claude_spec, terra_spec]) self.assertEqual(transition_budget_counts, [1]) continuation = invoked_prompts[1] self.assertIn(str(task.plan.resolve()), continuation) self.assertIn(str(claude_locator.resolve()), continuation) self.assertIn(str(workspace.resolve()), continuation) self.assertIn( str((claude_locator.parent / "stream.log").resolve()), continuation, ) self.assertIn( str( (claude_locator.parent / "normalized-output.log").resolve() ), continuation, ) state = store.task_state(task) decision = state["execution_decisions"]["worker"] self.assertEqual(decision["selected"]["adapter"], "codex") self.assertEqual(decision["selected"]["target"], "gpt-5.6-terra") self.assertEqual(decision["transition"]["kind"], "promotion") self.assertEqual( decision["transition"]["trigger"], "provider-quota" ) self.assertEqual( [entry["transition"] for entry in state["route_transition_history"]], ["initial", "provider-quota"], ) budget = dispatch.StageFailureBudget.from_decision( store, task, decision ) self.assertEqual(budget.count(), 0) self.assertNotIn("no_failover_candidate", state.get("blocked") or "") finally: store.close() async def test_cloud_agy_promotion_chain_commits_each_transition(self): daytime = datetime( 2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9)) ) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="cloud", grade=5) store = dispatch.StateStore(workspace) try: agy_spec = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)", ) claude_spec = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh", ) terra_spec = dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ) locators = { spec.cli: self.make_attempt_locator(workspace, task, spec) for spec in (agy_spec, claude_spec, terra_spec) } invoked_specs = [] invoked_prompts = [] transition_budget_counts = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) invoked_prompts.append(args[5]) if spec.cli == "agy": return (1, "provider-quota", locators["agy"]) state = store.task_state(task) decision = state["execution_decisions"]["worker"] budget = dispatch.StageFailureBudget.from_decision( store, task, decision ) transition_budget_counts.append(budget.count()) if spec.cli == "claude": return (1, "context-limit", locators["claude"]) return (0, None, locators["codex"]) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object( dispatch.asyncio, "sleep", new=mock.AsyncMock() ), ): dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=daytime ) success, final_locator = await dispatch.run_escalating( workspace, store, task, "worker", agy_spec ) self.assertTrue(success) self.assertEqual(final_locator, locators["codex"]) self.assertEqual( invoked_specs, [agy_spec, claude_spec, terra_spec] ) self.assertEqual(transition_budget_counts, [1, 2]) self.assertIn( str(locators["agy"].resolve()), invoked_prompts[1] ) self.assertIn( str(locators["claude"].resolve()), invoked_prompts[2] ) state = store.task_state(task) decision = state["execution_decisions"]["worker"] self.assertEqual( decision["promotion_path"], [ { "adapter": "agy", "target": "Gemini 3.6 Flash (High)", }, {"adapter": "claude", "target": "claude-opus-4-8"}, {"adapter": "codex", "target": "gpt-5.6-terra"}, ], ) self.assertEqual( [entry["transition"] for entry in state["route_transition_history"]], ["initial", "provider-quota", "context-limit"], ) self.assertEqual( dispatch.StageFailureBudget.from_decision( store, task, decision ).count(), 0, ) finally: store.close() async def test_night_laguna_failure_continues_on_available_gemini(self): nighttime = datetime(2026, 7, 26, 1, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: laguna_spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True) locator = self.make_attempt_locator(workspace, task, laguna_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if spec.cli == "pi": return (1, "provider-stream-disconnect", locator) return (0, None, locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=nighttime) success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", laguna_spec) self.assertTrue(success) self.assertEqual(len(invoked_specs), 2) self.assertEqual(invoked_specs[0].cli, "pi") self.assertEqual(invoked_specs[1].cli, "agy") state = store.task_state(task) decisions = state["execution_decisions"]["worker"] self.assertEqual(decisions["selected"]["adapter"], "agy") self.assertEqual(decisions["transition"]["trigger"], "provider-stream-disconnect") finally: store.close() async def test_night_gemini_quota_exhaustion_blocks_without_bounce(self): nighttime = datetime(2026, 7, 26, 1, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: laguna_spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True) locator = self.make_attempt_locator(workspace, task, laguna_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) return (1, "provider-quota", locator) quota_snap = { "targets": [ {"adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "status": "exhausted"} ] } store.update_task(task, quota_snapshot=quota_snap) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=nighttime) success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", laguna_spec) self.assertFalse(success) state = store.task_state(task) self.assertIn("no_failover_candidate", state.get("blocked", "")) finally: store.close() async def test_recovered_primary_quota_does_not_reverse_failover(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: store.update_task( task, quota_snapshot={ "snapshot_id": "gemini-exhausted", "source": "iop-node quota-probe", "checked_at": "2026-07-25T03:00:00+09:00", "targets": [ {"adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "status": "exhausted"} ], }, ) laguna_spec = dispatch.AgentSpec("pi", "iop/laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True) decision, spec = dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) self.assertEqual(spec.cli, "pi") locator = self.make_attempt_locator(workspace, task, laguna_spec) store.update_task( task, quota_snapshot={ "snapshot_id": "gemini-recovered", "source": "iop-node quota-probe", "checked_at": "2026-07-25T04:00:00+09:00", "targets": [ {"adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "status": "available"} ], }, ) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) return (1, "provider-quota", locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", laguna_spec) self.assertFalse(success) self.assertEqual(len(invoked_specs), 1) self.assertEqual(invoked_specs[0].cli, "pi") state = store.task_state(task) self.assertIn("no_failover_candidate", state.get("blocked", "")) finally: store.close() async def test_generic_failure_stays_on_same_target(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)") locator = self.make_attempt_locator(workspace, task, gemini_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if len(invoked_specs) == 1: return (1, "generic-error", locator) return (0, None, locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) success, final_loc = await dispatch.run_escalating(workspace, store, task, "worker", gemini_spec) self.assertTrue(success) self.assertEqual(len(invoked_specs), 2) self.assertEqual(invoked_specs[0].cli, "agy") self.assertEqual(invoked_specs[1].cli, "agy") finally: store.close() async def test_no_promotion_target_keeps_same_target_and_persists_state(self): """local-G08 daytime: provider-connection x2 → success. Same AGY target, delay [2, 4], budget reset. The selector-backed worker must NOT fall through to legacy promoted_spec(). Invocation target, persisted selected, history, and terminal recovery backoff must all agree on AGY with bounded exponential backoff. """ daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace) store = dispatch.StateStore(workspace) try: gemini_spec = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)", ) locator = self.make_attempt_locator(workspace, task, gemini_spec) invoked_specs = [] sleep_delays = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if len(invoked_specs) <= 2: return (1, "provider-connection", locator) return (0, None, locator) async def observe_sleep(delay): sleep_delays.append(delay) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object( dispatch.asyncio, "sleep", new=mock.AsyncMock(side_effect=observe_sleep), ), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) initial_state = store.task_state(task) initial_selected = dict(initial_state["execution_decisions"]["worker"]["selected"]) initial_history = list(initial_state["route_transition_history"]) success, final_loc = await dispatch.run_escalating( workspace, store, task, "worker", gemini_spec ) self.assertTrue(success) # Two failures then one success = 3 invocations self.assertEqual(len(invoked_specs), 3) # All invocations must be on AGY — no legacy AGY→Claude fallthrough for i, spec in enumerate(invoked_specs): self.assertEqual(spec.cli, "agy", f"invocation {i} target mismatch") self.assertEqual(spec.model, "Gemini 3.6 Flash (Medium)", f"invocation {i} model mismatch") # Terminal backoff: retries go 0→1→2, delays = [2**1, 2**2] = [2, 4] self.assertEqual(len(sleep_delays), 2, f"expected 2 sleep calls, got {len(sleep_delays)}") self.assertEqual(sleep_delays[0], 2) self.assertEqual(sleep_delays[1], 4) state = store.task_state(task) # Persisted selected must NOT change from initial AGY self.assertEqual( state["execution_decisions"]["worker"]["selected"], initial_selected, ) # History must NOT have a promotion entry self.assertEqual( state["route_transition_history"], initial_history, ) # No block should be set after success self.assertIsNone(state.get("blocked")) # After success, stage failure budget count must be 0 worker_decision = state["execution_decisions"]["worker"] worker_budget = dispatch.StageFailureBudget.from_decision(store, task, worker_decision) self.assertEqual(worker_budget.count(), 0) finally: store.close() async def test_promotion_chain_exhaustion_stays_on_last_target(self): """Cloud promotion chain: AGY→Claude→Terra exhausted. When the last canonical target (Terra) fails with a promotable failure and no promotion target remains, the worker must stay on Terra for same-target recovery rather than falling through to legacy promoted_spec(). """ daytime = datetime( 2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9)) ) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="cloud", grade=5) store = dispatch.StateStore(workspace) try: agy_spec = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)" ) claude_spec = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh" ) terra_spec = dispatch.AgentSpec( "codex", "gpt-5.6-terra", "codex/gpt-5.6-terra high", reasoning_effort="high", ) loc_agy = self.make_attempt_locator(workspace, task, agy_spec) loc_claude = self.make_attempt_locator(workspace, task, claude_spec) loc_terra = self.make_attempt_locator(workspace, task, terra_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if spec.cli == "agy": return (1, "provider-quota", loc_agy) if spec.cli == "claude": return (1, "context-limit", loc_claude) # Terra fails once, then succeeds — chain exhaustion keeps it on Terra terra_count = sum(1 for s in invoked_specs if s.cli == "codex") if terra_count == 1: return (1, "provider-quota", loc_terra) return (0, None, loc_terra) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=daytime ) success, final_loc = await dispatch.run_escalating( workspace, store, task, "worker", agy_spec ) # Chain: AGY → Claude → Terra, then Terra retries on same target (no legacy fallthrough) self.assertEqual( [s.cli for s in invoked_specs], ["agy", "claude", "codex", "codex"], ) # The third and fourth invocations are both Terra (same-target recovery) self.assertEqual(invoked_specs[2].cli, "codex") self.assertEqual(invoked_specs[2].model, "gpt-5.6-terra") self.assertEqual(invoked_specs[3].cli, "codex") self.assertEqual(invoked_specs[3].model, "gpt-5.6-terra") state = store.task_state(task) decision = state["execution_decisions"]["worker"] # Promotion path should include all three transitions self.assertEqual( len(decision["promotion_path"]), 3, ) # History should show the promotions but no legacy recovery transitions = [h["transition"] for h in state["route_transition_history"]] self.assertIn("provider-quota", transitions) self.assertIn("context-limit", transitions) # No legacy promoted_spec() fallthrough: selected stays on Terra self.assertEqual( decision["selected"]["adapter"], "codex" ) self.assertEqual( decision["selected"]["target"], "gpt-5.6-terra" ) finally: store.close() async def test_legacy_promoted_spec_still_works_for_non_selector_worker(self): """Ensure legacy promoted_spec() path is preserved for non-selector workers.""" daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() # Use a task that does NOT have a persisted selector decision # so the selector promotion block is skipped entirely. directory = workspace / "agent-task" / "legacy_recovery_test" directory.mkdir(parents=True, exist_ok=True) header = "\n" (directory / "PLAN-local-G08.md").write_text(header, encoding="utf-8") (directory / "CODE_REVIEW-local-G08.md").write_text(header, encoding="utf-8") tasks = dispatch.scan_tasks(workspace, None) task = tasks[0] store = dispatch.StateStore(workspace) try: agy_spec = dispatch.AgentSpec( "agy", "Gemini 3.6 Flash (High)", "agy/Gemini 3.6 Flash (High)" ) claude_spec = dispatch.AgentSpec( "claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh" ) locator = self.make_attempt_locator(workspace, task, agy_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if spec.cli == "agy": return (1, "provider-quota", locator) return (0, None, locator) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): # Do NOT persist a selector decision — legacy path only success, final_loc = await dispatch.run_escalating( workspace, store, task, "worker", agy_spec ) self.assertTrue(success) # Legacy path: AGY → Claude (via promoted_spec) self.assertEqual(len(invoked_specs), 2) self.assertEqual(invoked_specs[0].cli, "agy") self.assertEqual(invoked_specs[1].cli, "claude") finally: store.close() class SelectorDispatcherIntegrationTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): await super().asyncSetUp() invoke_patcher = mock.patch.object( dispatch, "invoke", side_effect=AssertionError("Real provider invocation forbidden in test simulation"), ) build_cmd_patcher = mock.patch.object( dispatch, "build_command", side_effect=AssertionError("Real provider command construction forbidden in test simulation"), ) self.invoke_deny_guard = invoke_patcher.start() self.build_cmd_deny_guard = build_cmd_patcher.start() self.addCleanup(invoke_patcher.stop) self.addCleanup(build_cmd_patcher.stop) def make_task( self, workspace: Path, lane: str = "local", grade: int = 8, unit: str = "01_unit" ) -> dispatch.Task: directory = workspace / "agent-task" / "selector_dispatch_integration" / unit directory.mkdir(parents=True, exist_ok=True) header = f"\n" (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text(header, encoding="utf-8") tasks = dispatch.scan_tasks(workspace, None) for task in tasks: if task.name.endswith(unit): return task return tasks[0] def make_attempt_locator( self, workspace: Path, task: dispatch.Task, spec: dispatch.AgentSpec ) -> Path: attempt = workspace / f"attempt-{spec.cli}" attempt.mkdir(parents=True, exist_ok=True) raw, normalized = attempt / "stream.log", attempt / "normalized-output.log" raw.write_text("raw log\n", encoding="utf-8") normalized.write_text("normalized output\n", encoding="utf-8") locator = attempt / "locator.json" record = { "task": task.name, "workspace": str(workspace), "plan_path": str(task.plan), "stream_log": str(raw), "normalized_output_log": str(normalized), "cli": spec.cli, "model": spec.model, "spec": {"adapter": spec.cli, "target": spec.model}, } locator.write_text(json.dumps(record), encoding="utf-8") return locator async def test_worker_and_review_initial_invocation_uses_selector(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) store = dispatch.StateStore(workspace) try: with mock.patch.object(dispatch, "run_escalating") as run_escalating_mock, \ mock.patch.object(dispatch, "datetime") as datetime_mock: datetime_mock.now.return_value = daytime run_escalating_mock.side_effect = lambda ws, st, t, stage, spec, **kwargs: ( True, self.make_attempt_locator(ws, t, spec) ) await dispatch.run_worker(workspace, store, task) self.assertEqual(run_escalating_mock.call_count, 1) call_args = run_escalating_mock.call_args[0] self.assertEqual(call_args[3], "worker") spec_worker = call_args[4] self.assertEqual(spec_worker.cli, "agy") self.assertEqual(spec_worker.model, "Gemini 3.6 Flash (Medium)") state_after_worker = store.task_state(task) self.assertTrue(state_after_worker.get("worker_done")) self.assertIn("worker", state_after_worker.get("execution_decisions", {})) self.assertEqual( state_after_worker["execution_decisions"]["worker"]["selected"]["target"], "Gemini 3.6 Flash (Medium)", ) await dispatch.run_review(workspace, store, task) self.assertEqual(run_escalating_mock.call_count, 2) call_args2 = run_escalating_mock.call_args[0] self.assertEqual(call_args2[3], "review") spec_review = call_args2[4] self.assertEqual(spec_review.cli, "codex") self.assertEqual(spec_review.model, "gpt-5.6-sol") self.assertEqual(spec_review.display, "codex/gpt-5.6-sol xhigh") state_after_review = store.task_state(task) self.assertIn("review", state_after_review.get("execution_decisions", {})) self.assertNotEqual( state_after_review["execution_decisions"]["worker"]["selected"], state_after_review["execution_decisions"]["review"]["selected"], ) finally: store.close() async def test_dry_run_statelessness_initial_and_resume_previews(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) store = dispatch.StateStore(workspace) try: # 1. Non-persisted dry-run (initial preview) args = dispatch.argparse.Namespace( workspace=str(workspace), task_group=None, retry_blocked=False, dry_run=True, ) with mock.patch.object(dispatch, "datetime") as datetime_mock: datetime_mock.now.return_value = daytime result = await dispatch.dispatch_with_store(args, workspace, store) self.assertEqual(result, 0) state_initial = store.task_state(task) self.assertEqual(state_initial.get("execution_decisions"), {}) self.assertEqual(state_initial.get("route_transition_history"), []) # 2. Persist decision and test dry-run (read-only resume preview) dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=daytime ) state_persisted = store.task_state(task) history_before = list(state_persisted.get("route_transition_history", [])) self.assertEqual(len(history_before), 1) with mock.patch.object(dispatch, "datetime") as datetime_mock: datetime_mock.now.return_value = daytime result2 = await dispatch.dispatch_with_store(args, workspace, store) self.assertEqual(result2, 0) state_after_dry_run = store.task_state(task) history_after = list(state_after_dry_run.get("route_transition_history", [])) self.assertEqual(history_before, history_after) finally: store.close() async def test_dry_run_multiple_ready_tasks_isolation_and_statelessness(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) nighttime = datetime(2026, 7, 26, 23, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task1 = self.make_task(workspace, lane="local", grade=8, unit="01_unit1") task2 = self.make_task(workspace, lane="local", grade=8, unit="02_unit2") store = dispatch.StateStore(workspace) try: # Task 1 is pinned during daytime dec1, spec1 = dispatch.persisted_execution_decision( store, task1, stage="worker", evaluated_at=daytime ) self.assertEqual(spec1.cli, "agy") state1_before = dict(store.task_state(task1)) state2_before = dict(store.task_state(task2)) banners = [] def capture_banner(event, name, lines): banners.append((event, name, lines)) args = dispatch.argparse.Namespace( workspace=str(workspace), task_group=None, retry_blocked=False, dry_run=True, ) with mock.patch.object(dispatch, "datetime") as datetime_mock, \ mock.patch.object(dispatch, "banner", side_effect=capture_banner): datetime_mock.now.return_value = nighttime result = await dispatch.dispatch_with_store(args, workspace, store) self.assertEqual(result, 0) task1_banners = [b for b in banners if b[1] == task1.name] task2_banners = [b for b in banners if b[1] == task2.name] self.assertTrue(any("model=agy/" in line for b in task1_banners for line in b[2])) self.assertTrue(any("model=pi/" in line for b in task2_banners for line in b[2])) state1_after = store.task_state(task1) state2_after = store.task_state(task2) self.assertEqual( state1_before.get("route_transition_history"), state1_after.get("route_transition_history"), ) self.assertEqual( state2_before.get("route_transition_history"), state2_after.get("route_transition_history"), ) self.assertEqual(state2_after.get("execution_decisions"), {}) finally: store.close() async def test_resume_pins_target_across_time_and_body_changes_and_resets_on_new_generation(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) nighttime = datetime(2026, 7, 26, 23, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) store = dispatch.StateStore(workspace) try: # 1. Initial decision daytime (KST 14:00) -> agy Gemini Medium dec1, spec1 = dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=daytime ) self.assertEqual(spec1.cli, "agy") self.assertEqual(spec1.model, "Gemini 3.6 Flash (Medium)") # 2. Resuming at nighttime (KST 23:00) keeps pinned Gemini Medium dec2, spec2 = dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=nighttime ) self.assertEqual(spec2.cli, "agy") self.assertEqual(spec2.model, "Gemini 3.6 Flash (Medium)") # 3. Body edit (header intact) keeps pinned Gemini Medium plan_file = task.plan header = f"\n" plan_file.write_text(header + "\n# Modified Body Content\n", encoding="utf-8") dec3, spec3 = dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=nighttime ) self.assertEqual(spec3.cli, "agy") # 4. New generation header (plan=1) re-evaluates initial decision at nighttime -> pi Laguna plan_file.write_text("\n\n# New Plan\n", encoding="utf-8") task_new = dispatch.scan_tasks(workspace, None)[0] dec4, spec4 = dispatch.persisted_execution_decision( store, task_new, stage="worker", evaluated_at=nighttime ) self.assertEqual(spec4.cli, "pi") self.assertEqual(spec4.model, "laguna-s:2.1") finally: store.close() async def test_qualified_failover_and_blocker_scenarios(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) store = dispatch.StateStore(workspace) try: # 1. Initial decision local G08 -> agy Gemini Medium (primary) & pi Laguna (fallback) dec1, spec1 = dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=daytime ) self.assertEqual(spec1.cli, "agy") # 2. Qualified failover (provider-quota) -> transitions to pi Laguna dec2 = dispatch.select_execution_decision( task, stage="worker", prior_decision=dec1, evaluated_at=daytime, transition="failover", failure_class="provider-quota" ) self.assertEqual(dec2["transition"]["trigger"], "provider-quota") self.assertEqual(dec2["selected"]["adapter"], "pi") # 3. Subsequent failover when no candidate remains -> raises no_failover_candidate with self.assertRaises(dispatch.ExecutionDecisionError) as ctx: dispatch.select_execution_decision( task, stage="worker", prior_decision=dec2, evaluated_at=daytime, transition="failover", failure_class="provider-quota" ) self.assertIn("no_failover_candidate", str(ctx.exception)) finally: store.close() async def test_context_budget_and_retry_blocked_lifecycle(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) store = dispatch.StateStore(workspace) try: init_snap = { "schema_version": "1.0", "snapshot_id": "snap-init", "source": "fake_probe", "checked_at": daytime.isoformat(), "targets": [ { "adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "status": "available", "reason_codes": [], } ], "required_caps": [], "reason_codes": [], } # 1. Primary initial execution (agy/Gemini Medium) dec1, spec1 = dispatch.persisted_execution_decision( store, task, stage="worker", evaluated_at=daytime, quota_snapshot=init_snap ) self.assertEqual(spec1.cli, "agy") # 2. Record primary failure (count=1) -> failover to alternate (pi/laguna) budget = dispatch.StageFailureBudget.from_decision(store, task, dec1) count1 = budget.record_failure(target=dec1["selected"], transition="provider-quota") self.assertEqual(count1, 1) dec2 = dispatch.select_execution_decision( task, stage="worker", prior_decision=dec1, evaluated_at=daytime, transition="failover", failure_class="provider-quota" ) dispatch.commit_execution_decision(store, task, "worker", dec2) self.assertEqual(dec2["selected"]["adapter"], "pi") # 3. Alternate fails 9 times -> budget count reaches 10, task is blocked budget2 = dispatch.StageFailureBudget.from_decision(store, task, dec2) for _ in range(9): c = budget2.record_failure(target=dec2["selected"], transition="generic-failure") self.assertEqual(c, 10) store.update_task( task, blocked="worker recovery failure limit exhausted: 10/10 locator=/tmp/loc.json", blocker_evidence={ "role": "worker", "failure_class": "provider-quota", "locator": "/tmp/loc.json", "selected": dec2["selected"], "work_unit_id": dec2["work_unit_id"], } ) self.assertIsNotNone(store.task_state(task).get("blocked")) # 4. Retry blocked clears blocked & budget, preserves decision & transition history args = dispatch.argparse.Namespace( workspace=str(workspace), task_group=None, retry_blocked=True, dry_run=False, ) selector = dispatch._selector_module() with mock.patch.object(dispatch, "scan_tasks", return_value=[]), \ mock.patch.object(dispatch, "datetime") as datetime_mock, \ mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected subprocess")) as mock_sub: datetime_mock.now.return_value = daytime result = await dispatch.dispatch_with_store(args, workspace, store) self.assertEqual(result, 0) self.invoke_deny_guard.assert_not_called() self.build_cmd_deny_guard.assert_not_called() mock_sub.assert_not_called() state_after_retry = store.task_state(task) self.assertIsNone(state_after_retry.get("blocked")) self.assertEqual(state_after_retry.get("stage_failure_budgets"), {}) self.assertIn("worker", state_after_retry.get("execution_decisions", {})) self.assertTrue(len(state_after_retry.get("route_transition_history", [])) >= 2) # 5. Success resets stage failure budget budget3 = dispatch.StageFailureBudget.from_decision(store, task, dec2) budget3.reset_on_success() self.assertEqual(store.task_state(task).get("stage_failure_budgets"), {}) finally: store.close() async def test_review_recovery_and_runtime_audit_evidence(self): daytime = datetime( 2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9)), ) nighttime = datetime( 2026, 7, 26, 23, 0, 0, tzinfo=timezone(timedelta(hours=9)), ) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) historical_header = ( f"\n" ) (task.directory / "plan_local_G07_9.log").write_text( historical_header, encoding="utf-8" ) (task.directory / "code_review_cloud_G07_9.log").write_text( historical_header + "\n## 코드리뷰 결과\n- 종합 판정: FAIL\n", encoding="utf-8", ) assert task.review is not None task.review.rename(task.directory / "CODE_REVIEW-cloud-G09.md") task = next( item for item in dispatch.scan_tasks(workspace, None) if item.name == task.name ) self.assertTrue(task.recovery) store = dispatch.StateStore(workspace) try: # 1. Active review uses the PLAN generation/route but the fixed # official-review policy and a complete canonical schema. dec_rev, spec_rev = dispatch.persisted_execution_decision( store, task, stage="review", evaluated_at=daytime ) self.assertEqual(spec_rev.cli, "codex") self.assertEqual(spec_rev.model, "gpt-5.6-sol") self.assertEqual(dec_rev["lane"], "local") self.assertEqual(dec_rev["grade"], 8) self.assertEqual( dec_rev["decision"]["rule_id"], "official-review-codex" ) self.assertEqual(dec_rev["decision"]["policy_priority"], 10) self.assertEqual( dec_rev["decision"]["reason_codes"], ["official_review_fixed"], ) self.assertEqual(dec_rev["decision"]["timezone"], "Asia/Seoul") self.assertFalse(dec_rev["decision"]["pinned"]) self.assertEqual( dec_rev["quota"], { "snapshot_id": None, "mode": "bounded", "status": "unknown", "source": "official_review_fixed_policy", "checked_at": None, "targets": [], }, ) self.assertEqual( dec_rev["transition"], { "previous_target": None, "next_target": None, "trigger": "initial", "context_transfer": "none", }, ) self.assertNotIn("rule_id", dec_rev) self.assertNotIn("priority", dec_rev) self.assertNotIn("quota_snapshot", dec_rev) self.assertEqual( dispatch.agent_spec_from_decision(dec_rev), spec_rev ) # 2. Persisted canonical review decisions are reused after # policy/identity validation rather than being reselected. reused, reused_spec = dispatch.persisted_execution_decision( store, task, stage="review", evaluated_at=nighttime, ) self.assertEqual(reused, dec_rev) self.assertEqual(reused_spec, spec_rev) # A qualified cloud failure restarts the same fixed Codex target # without selector failover, promotion, quota probe, or local CLI. retry_locator = self.make_attempt_locator( workspace, task, spec_rev ) invoked_specs = [] async def fake_review_invoke(*args, **kwargs): invoked_specs.append(args[4]) if len(invoked_specs) == 1: return 1, "provider-quota", retry_locator return 0, None, retry_locator with ( mock.patch.object( dispatch, "invoke", new=fake_review_invoke ), mock.patch.object( dispatch, "select_execution_decision", side_effect=AssertionError( "fixed review recovery must not reselect" ), ) as selector_mock, mock.patch.object( dispatch.asyncio, "sleep", new=mock.AsyncMock() ), ): success, final_locator = await dispatch.run_escalating( workspace, store, task, "review", spec_rev, ) self.assertTrue(success) self.assertEqual(final_locator, retry_locator) self.assertEqual(invoked_specs, [spec_rev, spec_rev]) selector_mock.assert_not_called() self.assertEqual( store.task_state(task)["execution_decisions"]["review"], dec_rev, ) # 3. Review failure budget is independent from worker budget. budget_worker = dispatch.StageFailureBudget( store, task, dec_rev["work_unit_id"], "worker" ) budget_worker.record_failure( target={ "adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", }, transition="initial", ) budget_review = dispatch.StageFailureBudget.from_decision( store, task, dec_rev ) self.assertEqual(budget_review.count(), 0) # 4. Audit consumers read canonical nested decision/quota and # only expose legacy flat fields through read-only fallback. evidence = dispatch.selector_evidence_lines(dec_rev) self.assertIn("rule_id=official-review-codex", evidence) self.assertIn("priority=10", evidence) self.assertIn("transition=initial", evidence) self.assertIn("quota_status=unknown", evidence) status = dispatch.status_lines( task, "review", "ready", decision=dec_rev ) self.assertIn("rule_id=official-review-codex", status) runtime_evidence = dispatch.selector_runtime_evidence(dec_rev) self.assertIn("decision", runtime_evidence) self.assertIn("quota", runtime_evidence) self.assertNotIn("rule_id", runtime_evidence) self.assertNotIn("priority", runtime_evidence) self.assertNotIn("quota_snapshot", runtime_evidence) active_history = store.task_state(task)[ "route_transition_history" ][-1] self.assertIn("decision", active_history) self.assertIn("quota", active_history) self.assertNotIn("rule_id", active_history) self.assertNotIn("priority", active_history) self.assertNotIn("quota_snapshot", active_history) legacy_decision = { "schema_version": "1.0", "work_unit_id": dec_rev["work_unit_id"], "stage": "review", "rule_id": "official-review-codex", "priority": 10, "candidates": [{ "candidate_rank": 1, "adapter": "codex", "target": "gpt-5.6-sol", "execution_class": "cloud_model", "eligibility": "eligible", "reason_codes": ["official_review_fixed_target"], "selfcheck_required": False, }], "selected": { "adapter": "codex", "target": "gpt-5.6-sol", "execution_class": "cloud_model", "selfcheck_required": False, "reason_codes": ["official_review_fixed_target"], }, "quota_snapshot": { "id": "fixed", "status": "not_applicable" }, "transition": {"trigger": "resume"}, } legacy_evidence = dispatch.selector_evidence_lines( legacy_decision ) self.assertIn("rule_id=official-review-codex", legacy_evidence) self.assertIn("quota_status=not_applicable", legacy_evidence) # 5. Legacy finalization recovery restores only the matching # archived PLAN route/identity, then writes a canonical decision. assert task.plan is not None and task.review is not None task.review.write_text( task.review.read_text(encoding="utf-8") + "\n## 코드리뷰 결과\n- 종합 판정: FAIL\n", encoding="utf-8", ) archived_plan = task.directory / "plan_local_G08_10.log" archived_review = task.directory / "code_review_cloud_G09_10.log" task.plan.rename(archived_plan) task.review.rename(archived_review) non_verdict_review = ( task.directory / "code_review_cloud_G09_11.log" ) non_verdict_review.write_text( f"\n", encoding="utf-8", ) malformed_review = ( task.directory / "code_review_cloud_G09_99_extra.log" ) malformed_review.write_text( f"\n" "\n## 코드리뷰 결과\n- 종합 판정: PASS\n", encoding="utf-8", ) leading_zero_review = ( task.directory / "code_review_cloud_G09_099.log" ) leading_zero_review.write_text( archived_review.read_text(encoding="utf-8"), encoding="utf-8", ) leading_zero_plan = task.directory / "plan_local_G08_010.log" leading_zero_plan.write_text( archived_plan.read_text(encoding="utf-8"), encoding="utf-8", ) non_file_plan = task.directory / "plan_local_G08_12.log" non_file_plan.mkdir() historical_review = ( task.directory / "code_review_cloud_G07_9.log" ) os.utime(archived_review, (100, 100)) os.utime(non_verdict_review, (200, 200)) os.utime(malformed_review, (300, 300)) os.utime(historical_review, (400, 400)) os.utime(leading_zero_review, (500, 500)) os.utime(leading_zero_plan, (600, 600)) self.assertIsNone( dispatch.REVIEW_LOG_RE.fullmatch(leading_zero_review.name) ) self.assertIsNone( dispatch.REVIEW_LOG_RE.fullmatch( "code_review_cloud_G09_١.log" ) ) self.assertIsNone( dispatch.PLAN_LOG_RE.fullmatch(leading_zero_plan.name) ) self.assertIsNone( dispatch.PLAN_LOG_RE.fullmatch("plan_local_G08_١.log") ) self.assertIsNotNone( dispatch.REVIEW_LOG_RE.fullmatch( "code_review_cloud_G09_0.log" ) ) self.assertIsNotNone( dispatch.PLAN_LOG_RE.fullmatch("plan_local_G08_10.log") ) self.assertEqual( dispatch.latest_verdict_log(task.directory), archived_review, ) recovery_task = next( item for item in dispatch.scan_tasks(workspace, None) if item.name == task.name ) self.assertTrue(recovery_task.recovery) self.assertEqual( dispatch.official_review_plan_source(recovery_task), archived_plan, ) store.update_task( recovery_task, execution_decisions={"review": legacy_decision}, ) recovered, recovered_spec = dispatch.persisted_execution_decision( store, recovery_task, stage="review", evaluated_at=nighttime, ) self.assertEqual(recovered_spec, spec_rev) self.assertEqual(recovered["lane"], "local") self.assertEqual(recovered["grade"], 8) self.assertEqual( recovered["work_unit_id"], dec_rev["work_unit_id"] ) self.assertTrue(recovered["decision"]["pinned"]) self.assertEqual(recovered["transition"]["trigger"], "resume") self.assertNotIn("rule_id", recovered) self.assertNotIn("quota_snapshot", recovered) recovered_history = store.task_state(recovery_task)[ "route_transition_history" ][-1] self.assertIn("decision", recovered_history) self.assertIn("quota", recovered_history) self.assertNotIn("rule_id", recovered_history) self.assertNotIn("quota_snapshot", recovered_history) # 6. An identity-matching archive with a non-canonical route # filename fails closed instead of inventing plan-0/lane/grade. invalid_plan = task.directory / "plan_legacy_G08_0.log" archived_plan.rename(invalid_plan) invalid_recovery_task = next( item for item in dispatch.scan_tasks(workspace, None) if item.name == task.name ) with self.assertRaises(dispatch.ExecutionDecisionError) as ctx: dispatch.read_or_preview_stage_decision( invalid_recovery_task, {}, stage="review", evaluated_at=daytime, ) self.assertIn( "matching archived PLAN identity", str(ctx.exception), ) self.invoke_deny_guard.assert_not_called() self.build_cmd_deny_guard.assert_not_called() finally: store.close() async def test_completing_target_controls_selfcheck_and_reuses_pin(self): daytime = datetime(2026, 7, 26, 14, 0, 0, tzinfo=timezone(timedelta(hours=9))) nighttime = datetime(2026, 7, 26, 1, 0, 0, tzinfo=timezone(timedelta(hours=9))) # Case 1: Day local G08 completion on Laguna requires selfcheck with pinned Laguna with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) store = dispatch.StateStore(workspace) try: gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)") laguna_spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True) loc_gemini = self.make_attempt_locator(workspace, task, gemini_spec) loc_laguna = self.make_attempt_locator(workspace, task, laguna_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if spec.cli == "agy": return (1, "provider-quota", loc_gemini) return (0, None, loc_laguna) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) await dispatch.run_worker(workspace, store, task) self.assertEqual([s.cli for s in invoked_specs], ["agy", "pi"]) state = store.task_state(task) self.assertEqual(state["execution_class"], "local_model") self.assertFalse(state["selfcheck_done"]) self.assertEqual(dispatch.task_stage(task, state), "selfcheck") self.assertEqual(state["execution_decisions"]["worker"]["selected"]["adapter"], "pi") self.assertEqual( state["completing_decision"]["selected"]["execution_class"], "local_model" ) hist1 = list(state["route_transition_history"]) self.assertEqual([h["transition"] for h in hist1], ["initial", "resume", "provider-quota"]) selfcheck_specs = [] async def mock_invoke_selfcheck(*args, **kwargs): spec = args[4] selfcheck_specs.append(spec) return (0, None, loc_laguna) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke_selfcheck), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), mock.patch.object(dispatch, "implementation_review_errors", return_value=[]), ): await dispatch.run_selfcheck(workspace, store, task) self.assertEqual([s.cli for s in selfcheck_specs], ["pi"]) state2 = store.task_state(task) self.assertTrue(state2["selfcheck_done"]) self.assertEqual(dispatch.task_stage(task, state2), "review") hist2 = state2["route_transition_history"] self.assertEqual([h["transition"] for h in hist2], ["initial", "resume", "provider-quota"]) finally: store.close() # Case 2: Night local G08 completion on Gemini skips selfcheck with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="local", grade=8) store = dispatch.StateStore(workspace) try: gemini_spec = dispatch.AgentSpec("agy", "Gemini 3.6 Flash (Medium)", "agy/Gemini 3.6 Flash (Medium)") laguna_spec = dispatch.AgentSpec("pi", "laguna-s:2.1", "pi/iop/laguna-s:2.1", local_pi=True) loc_gemini = self.make_attempt_locator(workspace, task, gemini_spec) loc_laguna = self.make_attempt_locator(workspace, task, laguna_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) if spec.cli == "pi": return (1, "provider-stream-disconnect", loc_laguna) return (0, None, loc_gemini) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=nighttime) await dispatch.run_worker(workspace, store, task) self.assertEqual([s.cli for s in invoked_specs], ["pi", "agy"]) state = store.task_state(task) self.assertEqual(state["execution_class"], "cloud_model") self.assertTrue(state["selfcheck_done"]) self.assertEqual(dispatch.task_stage(task, state), "review") self.assertEqual([h["transition"] for h in state["route_transition_history"]], ["initial", "resume", "provider-stream-disconnect"]) finally: store.close() # Case 3: Cloud G07 completion on Claude skips selfcheck with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = self.make_task(workspace, lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: claude_spec = dispatch.AgentSpec("claude", "claude-opus-4-8", "claude/claude-opus-4-8 xhigh") loc_claude = self.make_attempt_locator(workspace, task, claude_spec) invoked_specs = [] async def mock_invoke(*args, **kwargs): spec = args[4] invoked_specs.append(spec) return (0, None, loc_claude) with ( mock.patch.object(dispatch, "invoke", new=mock_invoke), mock.patch.object(dispatch.asyncio, "sleep", new=mock.AsyncMock()), ): dispatch.persisted_execution_decision(store, task, stage="worker", evaluated_at=daytime) await dispatch.run_worker(workspace, store, task) self.assertEqual([s.cli for s in invoked_specs], ["claude"]) state = store.task_state(task) self.assertEqual(state["execution_class"], "cloud_model") self.assertTrue(state["selfcheck_done"]) self.assertEqual(dispatch.task_stage(task, state), "review") self.assertEqual([h["transition"] for h in state["route_transition_history"]], ["initial", "resume"]) finally: store.close() class ThroughputQuotaBatchTest(unittest.TestCase): def make_task( self, workspace: Path, name: str = "route/01_unit", *, lane: str = "cloud", grade: int = 7, ): directory = workspace / "agent-task" / name directory.mkdir(parents=True) header = f"\n" (directory / f"PLAN-{lane}-G{grade:02d}.md").write_text( header + "## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n" + "| src/route.py | ROUTE-1 |\n", encoding="utf-8", ) (directory / f"CODE_REVIEW-{lane}-G{grade:02d}.md").write_text( header, encoding="utf-8", ) return next(t for t in dispatch.scan_tasks(workspace, None) if t.name == name) def test_same_target_n_tasks_single_probe(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t1 = self.make_task(workspace, "route/01_task1", lane="cloud", grade=7) t2 = self.make_task(workspace, "route/02_task2", lane="cloud", grade=7) t3 = self.make_task(workspace, "route/03_task3", lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: probe_calls = [] def mock_probe(*args, **kwargs): probe_calls.append(kwargs) target = kwargs["target"] adapter = kwargs["adapter"] checked_at_iso = kwargs["checked_at"].astimezone(dispatch.KST).isoformat() return { "schema_version": "1.0", "snapshot_id": f"child-{adapter}-{target}", "source": "iop-node quota-probe", "checked_at": checked_at_iso, "targets": [{"adapter": adapter, "target": target, "status": "available"}], "required_caps": [{"name": "overall", "status": "available", "remaining_percent": 80.0}], "reason_codes": ["ok"], } selector = dispatch._selector_module() with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe): now = datetime.now(dispatch.KST) ready = [(t1, "worker"), (t2, "worker"), (t3, "worker")] batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) self.assertIsNotNone(batch_snap) # Cloud G7 candidate target: claude/claude-opus-4-8 # Total unique probe keys = 1. Probed EXACTLY 1 time across all 3 tasks! self.assertEqual(len(probe_calls), 1) # Evaluate decisions for all tasks using batch_snap d1, _ = dispatch.persisted_execution_decision(store, t1, stage="worker", quota_snapshot=batch_snap) d2, _ = dispatch.persisted_execution_decision(store, t2, stage="worker", quota_snapshot=batch_snap) d3, _ = dispatch.persisted_execution_decision(store, t3, stage="worker", quota_snapshot=batch_snap) # All decisions share the exact same snapshot_id and checked_at self.assertEqual(d1["quota"]["snapshot_id"], batch_snap["snapshot_id"]) self.assertEqual(d2["quota"]["snapshot_id"], batch_snap["snapshot_id"]) self.assertEqual(d3["quota"]["snapshot_id"], batch_snap["snapshot_id"]) self.assertEqual(d1["quota"]["checked_at"], batch_snap["checked_at"]) self.assertEqual(d2["quota"]["checked_at"], batch_snap["checked_at"]) self.assertEqual(d3["quota"]["checked_at"], batch_snap["checked_at"]) # Child evidence preserved in batch_snap targets for target_entry in batch_snap["targets"]: self.assertIn("child_snapshot_id", target_entry) finally: store.close() def test_mixed_targets_unique_key_probing(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t1 = self.make_task(workspace, "route/01_cloud7", lane="cloud", grade=7) t2 = self.make_task(workspace, "route/02_cloud9", lane="cloud", grade=9) store = dispatch.StateStore(workspace) try: probed_keys = [] def mock_probe(*args, **kwargs): probed_keys.append((kwargs["adapter"], kwargs["target"])) target = kwargs["target"] adapter = kwargs["adapter"] checked_at_iso = kwargs["checked_at"].astimezone(dispatch.KST).isoformat() return { "schema_version": "1.0", "snapshot_id": f"snap-{adapter}-{target}", "source": "iop-node quota-probe", "checked_at": checked_at_iso, "targets": [{"adapter": adapter, "target": target, "status": "available"}], "required_caps": [{"name": "overall", "status": "available", "remaining_percent": 100.0}], "reason_codes": ["ok"], } selector = dispatch._selector_module() with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe): now = datetime.now(dispatch.KST) ready = [(t1, "worker"), (t2, "worker")] batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) self.assertIsNotNone(batch_snap) # Ensure no duplicate probes were called and exactly 2 unique targets were probed self.assertEqual(len(probed_keys), 2) self.assertEqual(len(probed_keys), len(set(probed_keys))) d1, _ = dispatch.persisted_execution_decision(store, t1, stage="worker", quota_snapshot=batch_snap) d2, _ = dispatch.persisted_execution_decision(store, t2, stage="worker", quota_snapshot=batch_snap) self.assertEqual(d1["quota"]["snapshot_id"], batch_snap["snapshot_id"]) self.assertEqual(d2["quota"]["snapshot_id"], batch_snap["snapshot_id"]) self.assertEqual(d1["quota"]["status"], "available") self.assertEqual(d2["quota"]["status"], "available") finally: store.close() def test_local_and_resume_zero_probe_count(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t_local = self.make_task(workspace, "route/01_local", lane="local", grade=5) t_resume = self.make_task(workspace, "route/02_resume", lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: # Give t_resume a prior decision store.update_task( t_resume, execution_decisions={ "worker": { "work_unit_id": dispatch.work_unit_id_from_file(t_resume.plan), "stage": "worker", "selected": { "adapter": "agy", "target": "gemini-2.5-flash", "execution_class": "cloud_model", "selfcheck_required": False, }, "quota": { "snapshot_id": "prior-snap", "mode": "bounded", "status": "available", "source": "iop-node quota-probe", "checked_at": datetime.now(dispatch.KST).isoformat(), "targets": [], }, } }, ) probe_calls = [] selector = dispatch._selector_module() with mock.patch.object(selector, "probe_candidate_quota", side_effect=lambda **kw: probe_calls.append(kw)): now = datetime.now(dispatch.KST) ready = [(t_local, "worker"), (t_resume, "worker")] batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) # Local task candidate is local_model, resume task has prior decision -> 0 probes needed! self.assertIsNone(batch_snap) self.assertEqual(len(probe_calls), 0) finally: store.close() def test_night_local_and_official_review_zero_probe_count(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t_night = self.make_task(workspace, "route/01_night", lane="local", grade=8) t_review = self.make_task(workspace, "route/02_review", lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: probe_calls = [] selector = dispatch._selector_module() with ( mock.patch.object( selector, "probe_candidate_quota", side_effect=lambda **kw: probe_calls.append(kw), ), mock.patch("subprocess.run", side_effect=AssertionError("subprocess called")), ): now = datetime(2026, 7, 26, 23, 30, 0, tzinfo=dispatch.KST) ready = [(t_night, "worker"), (t_review, "review")] batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) # Night local-G08 candidate is local_model first, official review stage is not worker -> 0 probes needed! self.assertIsNone(batch_snap) self.assertEqual(len(probe_calls), 0) finally: store.close() def make_attempt_locator( self, workspace: Path, task: dispatch.Task, spec: dispatch.AgentSpec ) -> Path: attempt = workspace / f"attempt-{spec.cli}" attempt.mkdir(parents=True, exist_ok=True) raw, normalized = attempt / "stream.log", attempt / "normalized-output.log" raw.write_text("raw log\n", encoding="utf-8") normalized.write_text("normalized output\n", encoding="utf-8") locator = attempt / "locator.json" record = { "task": task.name, "workspace": str(workspace), "plan_path": str(task.plan), "stream_log": str(raw), "normalized_output_log": str(normalized), "cli": spec.cli, "model": spec.model, "spec": {"adapter": spec.cli, "target": spec.model}, } locator.write_text(json.dumps(record), encoding="utf-8") return locator def test_same_target_tasks_unbound_admission_without_cap(self): async def run(): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() bin_dir = workspace / "agent-ops" / "bin" bin_dir.mkdir(parents=True, exist_ok=True) ai_ignore = bin_dir / "ai-ignore.sh" ai_ignore.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") ai_ignore.chmod(0o755) tasks = [ self.make_task(workspace, f"route/0{i}_task", lane="local", grade=8) for i in range(1, 6) ] store = dispatch.StateStore(workspace) try: barrier = asyncio.Barrier(5) completed_archive = workspace / "completed" completed_archive.mkdir() (completed_archive / "complete.log").write_text("complete\n", encoding="utf-8") async def fake_invoke(*args, **kwargs): role = args[3] task = args[2] spec = args[4] loc = self.make_attempt_locator(workspace, task, spec) if role == "worker": await asyncio.wait_for(barrier.wait(), timeout=2.0) elif role == "review": group, leaf = task.name.split("/", 1) archive_dir = ( workspace / "agent-task" / "archive" / "2026" / "07" / group / leaf ) archive_dir.mkdir(parents=True, exist_ok=True) (archive_dir / "complete.log").write_text("complete\n", encoding="utf-8") (archive_dir / "code_review_cloud_G07_0.log").write_text( "## 코드리뷰 결과\n\n- 종합 판정: PASS\n", encoding="utf-8" ) import shutil shutil.rmtree(task.directory, ignore_errors=True) return (0, None, loc) def mock_probe(*args, **kwargs): target = kwargs.get("target", "ornith:35b") adapter = kwargs.get("adapter", "pi") return { "schema_version": "1.0", "snapshot_id": f"snap-{adapter}-{target}", "source": "iop-node quota-probe", "checked_at": datetime.now(dispatch.KST).isoformat(), "targets": [{"adapter": adapter, "target": target, "status": "available"}], "required_caps": [{"name": "overall", "status": "available", "remaining_percent": 100.0}], "reason_codes": ["ok"], } selector = dispatch._selector_module() with ( mock.patch.object(dispatch, "invoke", side_effect=fake_invoke), mock.patch.object(dispatch, "implementation_review_errors", return_value=[]), mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe), ): args = SimpleNamespace(task_group="route", retry_blocked=False, dry_run=False) exit_code = await asyncio.wait_for( dispatch.dispatch_with_store(args, workspace, store), timeout=5.0, ) self.assertEqual(exit_code, 0) self.assertEqual(barrier.n_waiting, 0) finally: store.close() asyncio.run(run()) def test_batch_key_unknown_isolation(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t1 = self.make_task(workspace, "route/01_cloud7", lane="cloud", grade=7) t2 = self.make_task(workspace, "route/02_cloud9", lane="cloud", grade=9) store = dispatch.StateStore(workspace) try: def mock_probe(*args, **kwargs): adapter = kwargs["adapter"] target = kwargs["target"] if adapter == "claude": raise RuntimeError("Quota probe unexpected failure") checked_at_iso = kwargs["checked_at"].astimezone(dispatch.KST).isoformat() return { "schema_version": "1.0", "snapshot_id": f"snap-{adapter}-{target}", "source": "iop-node quota-probe", "checked_at": checked_at_iso, "targets": [{"adapter": adapter, "target": target, "status": "available"}], "required_caps": [{"name": "overall", "status": "available", "remaining_percent": 100.0}], "reason_codes": ["ok"], } selector = dispatch._selector_module() with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe): now = datetime.now(dispatch.KST) ready = [(t1, "worker"), (t2, "worker")] batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) self.assertIsNotNone(batch_snap) statuses = {t["adapter"]: t["status"] for t in batch_snap["targets"]} # Failed probe key is isolated as unknown, while other key succeeded as available self.assertIn("unknown", list(statuses.values())) self.assertIn("available", list(statuses.values())) d2, _ = dispatch.persisted_execution_decision(store, t2, stage="worker", quota_snapshot=batch_snap) self.assertEqual(d2["quota"]["status"], "available") finally: store.close() def test_same_work_unit_resume_zero_probe_and_pin_preserved(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t1 = self.make_task(workspace, "route/01_pinned", lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: selector = dispatch._selector_module() deterministic_snapshot = { "schema_version": "1.0", "snapshot_id": "snap-init", "source": "fake_probe", "checked_at": datetime.now(dispatch.KST).isoformat(), "targets": [ { "adapter": "codex", "target": "gpt-5.6-sol", "status": "available", "reason_codes": [], } ], "required_caps": [], "reason_codes": [], } with mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run: init_d, _ = dispatch.persisted_execution_decision( store, t1, stage="worker", quota_snapshot=deterministic_snapshot ) run.assert_not_called() self.assertIsNotNone(init_d) probe_calls = [] with mock.patch.object(selector, "probe_candidate_quota", side_effect=lambda **kw: probe_calls.append(kw)): now = datetime.now(dispatch.KST) ready = [(t1, "worker")] batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) # Persisted work unit for same work_unit_id -> 0 probe calls self.assertIsNone(batch_snap) self.assertEqual(len(probe_calls), 0) with mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run: d, spec = dispatch.persisted_execution_decision(store, t1, stage="worker") run.assert_not_called() self.assertIs(d["decision"]["pinned"], True) self.assertEqual(d["work_unit_id"], init_d["work_unit_id"]) finally: store.close() def test_new_generation_next_batch_available_recovery(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t1 = self.make_task(workspace, "route/01_gen", lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: # Store prior decision with an OLD work_unit_id store.update_task( t1, execution_decisions={ "worker": { "work_unit_id": "old_task::plan-0::tag-OLD", "stage": "worker", "selected": { "adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "execution_class": "cloud_model", "selfcheck_required": False, }, "quota": { "snapshot_id": "old-snap", "mode": "bounded", "status": "exhausted", "source": "iop-node quota-probe", "checked_at": datetime.now(dispatch.KST).isoformat(), "targets": [], }, } }, ) probe_calls = [] def mock_probe(*args, **kwargs): probe_calls.append(kwargs) target, adapter = kwargs["target"], kwargs["adapter"] checked_at_iso = kwargs["checked_at"].astimezone(dispatch.KST).isoformat() return { "schema_version": "1.0", "snapshot_id": f"fresh-snap-{adapter}-{target}", "source": "iop-node quota-probe", "checked_at": checked_at_iso, "targets": [{"adapter": adapter, "target": target, "status": "available"}], "required_caps": [{"name": "overall", "status": "available", "remaining_percent": 100.0}], "reason_codes": ["ok"], } selector = dispatch._selector_module() with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe): now = datetime.now(dispatch.KST) ready = [(t1, "worker")] batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) # New generation -> probe runs, fresh snapshot returned self.assertIsNotNone(batch_snap) self.assertGreater(len(probe_calls), 0) d, _ = dispatch.persisted_execution_decision(store, t1, stage="worker", quota_snapshot=batch_snap) self.assertEqual(d["quota"]["status"], "available") finally: store.close() def test_confirmed_provider_quota_task_local_derived_exhausted(self): current_decision = { "work_unit_id": "route/01_unit::plan-0::tag-ROUTE", "stage": "worker", "selected": {"adapter": "agy", "target": "Gemini 3.6 Flash (Medium)"}, "quota": { "snapshot_id": "shared-batch-123", "mode": "bounded", "status": "available", "source": "iop-node quota-probe", "checked_at": "2026-07-26T18:00:00+09:00", "targets": [ {"adapter": "agy", "target": "Gemini 3.6 Flash (Medium)", "status": "available"}, {"adapter": "codex", "target": "gpt-5.6-sol", "status": "available"}, ], "required_caps": [{"name": "overall", "status": "available"}], "reason_codes": ["ok"], }, } derived = dispatch.derive_work_unit_quota_evidence( current_decision, status="exhausted", reason="confirmed_runtime_provider_quota" ) # Observation identity preserved self.assertEqual(derived["snapshot_id"], "shared-batch-123") self.assertEqual(derived["checked_at"], "2026-07-26T18:00:00+09:00") self.assertEqual(derived["source"], "iop-node quota-probe") self.assertIn("confirmed_runtime_provider_quota", derived["reason_codes"]) # Selected target status updated to exhausted selected_entry = next( t for t in derived["targets"] if t["adapter"] == "agy" and t["target"] == "Gemini 3.6 Flash (Medium)" ) self.assertEqual(selected_entry["status"], "exhausted") # Original shared decision quota targets NOT mutated original_entry = next( t for t in current_decision["quota"]["targets"] if t["adapter"] == "agy" and t["target"] == "Gemini 3.6 Flash (Medium)" ) self.assertEqual(original_entry["status"], "available") def test_retry_blocked_quota_refresh_lifecycle(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t1 = self.make_task(workspace, "route/01_retry", lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: selector = dispatch._selector_module() init_snap = { "schema_version": "1.0", "snapshot_id": "snap-initial", "source": "fake_probe", "checked_at": datetime.now(dispatch.KST).isoformat(), "targets": [ { "adapter": "codex", "target": "gpt-5.6-sol", "status": "available", "reason_codes": [], } ], "required_caps": [], "reason_codes": [], } with mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run: init_d, _ = dispatch.persisted_execution_decision( store, t1, stage="worker", quota_snapshot=init_snap ) run.assert_not_called() self.assertEqual(init_d["quota"]["snapshot_id"], "snap-initial") # Block the task locator = workspace / "retry-locator.json" locator.write_text("{}", encoding="utf-8") store.update_task( t1, blocked="worker failure provider-quota", blocker_evidence={ "role": "worker", "failure_class": "provider-quota", "locator": str(locator), "selected": init_d["selected"], "work_unit_id": init_d["work_unit_id"], }, ) self.assertIsNotNone(store.task_state(t1).get("blocked")) # Mark retry quota refresh (simulating --retry-blocked) store.mark_retry_quota_refresh("route/01_retry") self.assertIsNone(store.task_state(t1).get("blocked")) self.assertTrue(store.task_state(t1).get("retry_quota_refresh_pending")) # Admission batch snapshot now triggers a fresh probe because refresh is pending probe_calls = [] def mock_probe(*args, **kwargs): probe_calls.append(kwargs) adapter = kwargs["adapter"] target = kwargs["target"] checked_at_iso = kwargs["checked_at"].astimezone(dispatch.KST).isoformat() return { "schema_version": "1.0", "snapshot_id": "fresh-retry-snap", "source": "iop-node quota-probe", "checked_at": checked_at_iso, "targets": [{"adapter": adapter, "target": target, "status": "available"}], "required_caps": [], "reason_codes": ["ok"], } with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe): now = datetime.now(dispatch.KST) ready = [(t1, "worker")] retry_batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) self.assertIsNone(retry_batch_snap) self.assertEqual(len(probe_calls), 0) # With no persisted unused alternate, retry consumes no quota snapshot and resumes. with mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run: d, spec = dispatch.persisted_execution_decision( store, t1, stage="worker", quota_snapshot=retry_batch_snap ) run.assert_not_called() self.assertEqual(store.task_state(t1).get("quota_snapshot")["snapshot_id"], "snap-initial") self.assertFalse(store.task_state(t1).get("retry_quota_refresh_pending")) # Subsequent admission pass (ordinary resume) -> 0 probe calls probe_calls.clear() with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe): now = datetime.now(dispatch.KST) ready = [(t1, "worker")] resume_batch_snap = dispatch.build_admission_batch_snapshot(store, ready, now) self.assertIsNone(resume_batch_snap) self.assertEqual(len(probe_calls), 0) finally: store.close() def test_retry_blocked_scopes_to_blocked_worker_and_refreshes_pinned_alternate(self): async def _async_run(): nighttime = datetime(2026, 7, 26, 23, 0, 0, tzinfo=timezone(timedelta(hours=9))) with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() t_blocked = self.make_task(workspace, "route/01_blocked", lane="local", grade=8) t_normal = self.make_task(workspace, "route/02_normal", lane="cloud", grade=7) store = dispatch.StateStore(workspace) try: selector = dispatch._selector_module() normal_snap = { "schema_version": "1.0", "snapshot_id": "snap-normal", "source": "fake_probe", "checked_at": nighttime.isoformat(), "targets": [ { "adapter": "codex", "target": "gpt-5.6-sol", "status": "available", "reason_codes": [], } ], "required_caps": [], "reason_codes": [], } with mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run: d_normal, spec_normal = dispatch.persisted_execution_decision( store, t_normal, stage="worker", evaluated_at=nighttime, quota_snapshot=normal_snap ) run.assert_not_called() with mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run: d_blocked, spec_blocked = dispatch.persisted_execution_decision( store, t_blocked, stage="worker", evaluated_at=nighttime, quota_snapshot=normal_snap ) run.assert_not_called() self.assertEqual(d_blocked["selected"]["adapter"], "pi") self.assertEqual(d_blocked["selected"]["target"], "iop/laguna-s:2.1") loc_path = workspace / "attempt-loc.json" loc_path.write_text("{}", encoding="utf-8") store.update_task( t_blocked, blocked=f"worker failure provider-quota locator={loc_path}", blocker_evidence={ "role": "worker", "failure_class": "provider-quota", "locator": str(loc_path), "selected": d_blocked["selected"], "work_unit_id": d_blocked["work_unit_id"], } ) self.assertIsNotNone(store.task_state(t_blocked).get("blocked")) state_normal_before = dict(store.task_state(t_normal)) probe_calls = [] def mock_probe(*args, **kwargs): probe_calls.append(kwargs) adapter = kwargs["adapter"] target = kwargs["target"] checked_at_iso = kwargs["checked_at"].astimezone(dispatch.KST).isoformat() return { "schema_version": "1.0", "snapshot_id": "fresh-retry-alternate-snap", "source": "iop-node quota-probe", "checked_at": checked_at_iso, "targets": [{"adapter": adapter, "target": target, "status": "available"}], "required_caps": [], "reason_codes": ["ok"], } invoke_calls = [] async def fake_invoke(ws, st, task, role, spec, prompt, resume_locator=None): locator = ws / f"{task.name.replace('/', '_')}-{role}.json" locator.write_text("{}", encoding="utf-8") invoke_calls.append((task.name, role, spec, prompt, resume_locator)) return 0, None, locator async def fake_run_review(ws, st, task, **kwargs): archive = ws / "agent-task" / "archive" / "2026" / "07" / task.name archive.parent.mkdir(parents=True, exist_ok=True) (task.directory / "complete.log").write_text("simulation complete\n", encoding="utf-8") task.directory.rename(archive) return str(archive) args = dispatch.argparse.Namespace( workspace=str(workspace), task_group="route", retry_blocked=True, dry_run=False, ) with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe), \ mock.patch.object(dispatch, "run_review", side_effect=fake_run_review), \ mock.patch.object(dispatch, "ensure_review_shared_state"), \ mock.patch.object(dispatch, "invoke", side_effect=fake_invoke), \ mock.patch.object(dispatch, "datetime") as datetime_mock, \ mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run_sub: datetime_mock.now.return_value = nighttime res = await dispatch.dispatch_with_store(args, workspace, store) run_sub.assert_not_called() self.assertEqual(len(probe_calls), 1) self.assertEqual(probe_calls[0]["adapter"], "agy") self.assertEqual(probe_calls[0]["target"], "Gemini 3.6 Flash (Medium)") st_blocked_after = store.task_state(t_blocked) self.assertIsNone(st_blocked_after.get("blocked")) self.assertFalse(st_blocked_after.get("retry_quota_refresh_pending")) dec_after = st_blocked_after["execution_decisions"]["worker"] self.assertEqual(dec_after["selected"]["adapter"], "agy") self.assertEqual(dec_after["selected"]["target"], "Gemini 3.6 Flash (Medium)") self.assertEqual(dec_after["transition"]["trigger"], "provider-quota") self.assertEqual(dec_after["work_unit_id"], d_blocked["work_unit_id"]) used = dec_after.get("used_candidates", []) used_adapters = [u.get("adapter") for u in used] self.assertIn("pi", used_adapters) self.assertIn("agy", used_adapters) self.assertTrue(len(st_blocked_after.get("route_transition_history", [])) >= 2) blocked_invocations = [call for call in invoke_calls if call[0] == t_blocked.name] self.assertEqual(len(blocked_invocations), 1) self.assertEqual(blocked_invocations[0][1], "worker") self.assertEqual(blocked_invocations[0][4], loc_path) st_normal_after = store.task_state(t_normal) self.assertFalse(st_normal_after.get("retry_quota_refresh_pending")) self.assertEqual( st_normal_after["execution_decisions"]["worker"]["selected"], state_normal_before["execution_decisions"]["worker"]["selected"], ) probe_calls.clear() invoke_calls.clear() args_normal = dispatch.argparse.Namespace( workspace=str(workspace), task_group="route", retry_blocked=False, dry_run=False, ) with mock.patch.object(selector, "probe_candidate_quota", side_effect=mock_probe), \ mock.patch.object(dispatch, "run_review", side_effect=fake_run_review), \ mock.patch.object(dispatch, "ensure_review_shared_state"), \ mock.patch.object(dispatch, "invoke", side_effect=fake_invoke), \ mock.patch.object(dispatch, "datetime") as datetime_mock, \ mock.patch.object(selector.subprocess, "run", side_effect=AssertionError("unexpected quota subprocess")) as run_sub: datetime_mock.now.return_value = nighttime res2 = await dispatch.dispatch_with_store(args_normal, workspace, store) run_sub.assert_not_called() self.assertEqual(len(probe_calls), 0) finally: store.close() asyncio.run(_async_run()) def test_generic_stderr_unknown_preservation(self): selector = dispatch._selector_module() with mock.patch("subprocess.run") as mock_run: mock_run.return_value = SimpleNamespace( returncode=1, stdout="", stderr="Error: connection timeout to quota service\n" ) now = datetime.now(dispatch.KST) snapshot = selector.probe_candidate_quota( target="Gemini 3.6 Flash (Medium)", adapter="agy", required_caps=["overall"], checked_at=now, ) self.assertIsNotNone(snapshot) self.assertEqual(snapshot["targets"][0]["status"], "unknown") self.assertIn("probe_error", snapshot["reason_codes"]) if __name__ == "__main__": unittest.main()