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, native_resume_session=None, ): self.assertEqual(actual_session_id, session_id) native = attempt_dir / "native-sessions" / f"session_{session_id}.jsonl" child = ( "from pathlib import Path\n" "import sys,time\n" "path = Path(sys.argv[1])\n" "path.parent.mkdir(parents=True, exist_ok=True)\n" "path.write_text(" "'{\"type\":\"session\",\"version\":3,\"id\":\"test\"," "\"timestamp\":\"2026-07-25T00:00:00.000Z\"," "\"cwd\":\"/tmp/test\"}\\n', encoding='utf-8')\n" "time.sleep(0.05)\n" "print('done', flush=True)\n" ) return [sys.executable, "-c", child, str(native)] spec = dispatch.AgentSpec( "runtime-agent", "runtime-model", "runtime-target", native_resume=True, target_id="runtime-target", execution_class="local_model", runtime={ "command": ["runtime-command", "{prompt}"], "session_path": "native-sessions/session_{session_id}.jsonl", "native_session_monitor": True, "output_format": "text", }, ) try: with ( mock.patch.object(dispatch, "build_command", side_effect=command_for), mock.patch.object(dispatch.uuid, "uuid4", return_value=session_id), mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01), mock.patch("builtins.print") as print_mock, ): rc, failure, locator = await dispatch.invoke( workspace, store, task, "review", spec, "Reply briefly." ) finally: store.close() self.assertEqual(rc, 0) self.assertIsNone(failure) record = json.loads(locator.read_text(encoding="utf-8")) self.assertTrue(record["native_session_path"].endswith(f"{session_id}.jsonl")) self.assertIsInstance(record["native_session_mtime_ns"], int) heartbeat = Path(record["heartbeat_log"]).read_text(encoding="utf-8") self.assertIn("[heartbeat] 작업중...", heartbeat) self.assertIn("native_session=", heartbeat) self.assertIn("native_mtime_ns=", heartbeat) stream = Path(record["stream_log"]).read_text(encoding="utf-8") self.assertIn("[stdout] done", stream) self.assertNotIn("[heartbeat]", stream) normalized = Path(record["normalized_output_log"]).read_text( encoding="utf-8" ) self.assertIn("done", normalized) visible_output = "\n".join( " ".join(str(value) for value in call.args) for call in print_mock.call_args_list ) self.assertIn("locator=", visible_output) self.assertNotIn("작업중...", visible_output) self.assertNotIn("done", visible_output) class SkillObservationContractTest(unittest.TestCase): def test_dispatcher_owns_observation_and_caller_wakes_only_for_attention(self): skill = ( Path(__file__).parents[1] / "SKILL.md" ).read_text(encoding="utf-8") self.assertIn("The dispatcher owns deterministic scheduling, recovery", skill) self.assertIn("Launch the live dispatcher as one persistent foreground process", skill) self.assertIn("Do not count internal helper coroutines as agent slots", skill) self.assertIn("actual stream or native-session progress", skill) self.assertIn("PID/start-token/process-marker evidence", skill) self.assertIn("never queries quota before admission", skill) self.assertIn("confirmed quota/rate-limit error advances directly", skill) self.assertIn("Use the bundled default catalog", skill) if __name__ == "__main__": unittest.main()