import ast import asyncio import importlib.util import io import json import os import re import sys import tempfile import unittest from pathlib import Path from unittest import mock SCRIPT = Path(__file__).parents[1] / "scripts" / "dispatch.py" loaded = sys.modules.get("agent_task_dispatch") if loaded is not None: dispatch = loaded else: SPEC = importlib.util.spec_from_file_location("agent_task_dispatch", SCRIPT) assert SPEC and SPEC.loader dispatch = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = dispatch SPEC.loader.exec_module(dispatch) def make_test_task(root: Path) -> dispatch.Task: plan = root / "PLAN-local-G05.md" review = root / "CODE_REVIEW-local-G05.md" plan.write_text("\n", encoding="utf-8") review.write_text("\n", encoding="utf-8") return dispatch.Task( name="test", directory=root, plan=plan, review=review, user_review=None, recovery=False, lane="local", grade=5, ) class ObservationOutputTest(unittest.TestCase): def test_banner_preserves_existing_format_and_nested_task_identity(self): buffer = io.StringIO() with mock.patch("sys.stdout", buffer): dispatch.banner("START", "group/subtask/task_name", ["line 1", "line 2"]) output = buffer.getvalue() expected = ( "------------------------------------------\n" "START: task_name\n" "------------------------------------------\n" "task=group/subtask/task_name\n" "line 1\n" "line 2\n" ) self.assertEqual(output, expected) buffer_flat = io.StringIO() with mock.patch("sys.stdout", buffer_flat): dispatch.banner("START", "task_name") output_flat = buffer_flat.getvalue() expected_flat = ( "------------------------------------------\n" "START: task_name\n" "------------------------------------------\n" ) self.assertEqual(output_flat, expected_flat) def test_attempt_event_is_one_flushed_stdout_line(self): buffer = io.StringIO() with mock.patch("sys.stdout", buffer): dispatch.attempt_event("[test-prefix]", "event message detail") output = buffer.getvalue() self.assertEqual(output, "[test-prefix] event message detail\n") def test_dispatch_compatibility_aliases_point_to_observation_module(self): self.assertEqual(dispatch.SEP, dispatch.observation.SEP) self.assertIs(dispatch.banner, dispatch.observation.banner) self.assertIs(dispatch.attempt_event, dispatch.observation.attempt_event) def test_observation_module_identity_is_reused(self): module1 = dispatch.load_sibling_observation_module() module2 = dispatch.load_sibling_observation_module() self.assertIs(module1, module2) self.assertIs(module1, sys.modules["agent_task_dispatcher_observation"]) def test_dispatch_has_no_direct_stdout_print_calls(self): source = SCRIPT.read_text(encoding="utf-8") tree = ast.parse(source, filename=str(SCRIPT)) stdout_prints = [] for node in ast.walk(tree): if isinstance(node, ast.Call): func = node.func if isinstance(func, ast.Name) and func.id == "print": is_stderr = False for kw in node.keywords: if kw.arg == "file": val = kw.value if ( isinstance(val, ast.Attribute) and isinstance(val.value, ast.Name) and val.value.id == "sys" and val.attr == "stderr" ): is_stderr = True break if not is_stderr: stdout_prints.append(node.lineno) self.assertEqual( stdout_prints, [], f"found direct stdout print() calls on lines: {stdout_prints}", ) class ObservationInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase): async def test_heartbeat_and_child_output_stay_in_logs_not_user_event_stream(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) (workspace / ".git").mkdir() task = make_test_task(workspace) store = dispatch.StateStore(workspace) session_id = "11111111-1111-1111-1111-111111111111" def command_for( spec, prompt, cwd, actual_session_id, attempt_dir, 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), mock.patch("builtins.print") as print_mock, ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "review", spec, "Reply briefly." ) finally: store.close() self.assertEqual(rc, 0) self.assertIsNone(failure) record = json.loads(locator.read_text(encoding="utf-8")) self.assertTrue(record["native_session_path"].endswith(f"{session_id}.jsonl")) self.assertIsInstance(record["native_session_mtime_ns"], int) heartbeat = Path(record["heartbeat_log"]).read_text(encoding="utf-8") self.assertIn("[heartbeat] 작업중...", heartbeat) self.assertIn("native_session=", heartbeat) self.assertIn("native_mtime_ns=", heartbeat) stream = Path(record["stream_log"]).read_text(encoding="utf-8") self.assertIn("[stdout] done", stream) self.assertNotIn("[heartbeat]", stream) normalized = Path(record["normalized_output_log"]).read_text( encoding="utf-8" ) self.assertIn("done", normalized) visible_output = "\n".join( " ".join(str(value) for value in call.args) for call in print_mock.call_args_list ) self.assertIn("locator=", visible_output) self.assertNotIn("작업중...", visible_output) self.assertNotIn("done", visible_output) class SkillObservationContractTest(unittest.TestCase): def test_dispatcher_owns_observation_and_caller_wakes_only_for_attention(self): skill = ( Path(__file__).parents[1] / "SKILL.md" ).read_text(encoding="utf-8") self.assertIn( "dispatcher as the execution lifecycle and observation owner", skill, ) self.assertIn( "without caller-LLM supervision", skill, ) self.assertIn( "The caller never monitors", skill, ) self.assertIn( "Wake the caller LLM only for an attention event that the dispatcher cannot resolve autonomously", skill, ) self.assertIn( "Exit code `3` is a non-terminal tracking state, including another dispatcher 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, ) if __name__ == "__main__": unittest.main()