219 lines
8.9 KiB
Python
219 lines
8.9 KiB
Python
import ast
|
|
import asyncio
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts" / "dispatch.py"
|
|
loaded = sys.modules.get("agent_task_dispatch")
|
|
if loaded is not None:
|
|
dispatch = loaded
|
|
else:
|
|
SPEC = importlib.util.spec_from_file_location("agent_task_dispatch", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
dispatch = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = dispatch
|
|
SPEC.loader.exec_module(dispatch)
|
|
|
|
|
|
def make_test_task(root: Path) -> dispatch.Task:
|
|
plan = root / "PLAN-local-G05.md"
|
|
review = root / "CODE_REVIEW-local-G05.md"
|
|
plan.write_text("<!-- task=test plan=0 tag=TEST -->\n", encoding="utf-8")
|
|
review.write_text("<!-- task=test plan=0 tag=TEST -->\n", encoding="utf-8")
|
|
return dispatch.Task(
|
|
name="test",
|
|
directory=root,
|
|
plan=plan,
|
|
review=review,
|
|
user_review=None,
|
|
recovery=False,
|
|
lane="local",
|
|
grade=5,
|
|
)
|
|
|
|
|
|
class ObservationOutputTest(unittest.TestCase):
|
|
def test_banner_preserves_existing_format_and_nested_task_identity(self):
|
|
buffer = io.StringIO()
|
|
with mock.patch("sys.stdout", buffer):
|
|
dispatch.banner("START", "group/subtask/task_name", ["line 1", "line 2"])
|
|
output = buffer.getvalue()
|
|
expected = (
|
|
"------------------------------------------\n"
|
|
"START: task_name\n"
|
|
"------------------------------------------\n"
|
|
"task=group/subtask/task_name\n"
|
|
"line 1\n"
|
|
"line 2\n"
|
|
)
|
|
self.assertEqual(output, expected)
|
|
|
|
buffer_flat = io.StringIO()
|
|
with mock.patch("sys.stdout", buffer_flat):
|
|
dispatch.banner("START", "task_name")
|
|
output_flat = buffer_flat.getvalue()
|
|
expected_flat = (
|
|
"------------------------------------------\n"
|
|
"START: task_name\n"
|
|
"------------------------------------------\n"
|
|
)
|
|
self.assertEqual(output_flat, expected_flat)
|
|
|
|
def test_attempt_event_is_one_flushed_stdout_line(self):
|
|
buffer = io.StringIO()
|
|
with mock.patch("sys.stdout", buffer):
|
|
dispatch.attempt_event("[test-prefix]", "event message detail")
|
|
output = buffer.getvalue()
|
|
self.assertEqual(output, "[test-prefix] event message detail\n")
|
|
|
|
def test_dispatch_compatibility_aliases_point_to_observation_module(self):
|
|
self.assertEqual(dispatch.SEP, dispatch.observation.SEP)
|
|
self.assertIs(dispatch.banner, dispatch.observation.banner)
|
|
self.assertIs(dispatch.attempt_event, dispatch.observation.attempt_event)
|
|
|
|
def test_observation_module_identity_is_reused(self):
|
|
module1 = dispatch.load_sibling_observation_module()
|
|
module2 = dispatch.load_sibling_observation_module()
|
|
self.assertIs(module1, module2)
|
|
self.assertIs(module1, sys.modules["agent_task_dispatcher_observation"])
|
|
|
|
def test_dispatch_has_no_direct_stdout_print_calls(self):
|
|
source = SCRIPT.read_text(encoding="utf-8")
|
|
tree = ast.parse(source, filename=str(SCRIPT))
|
|
stdout_prints = []
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Call):
|
|
func = node.func
|
|
if isinstance(func, ast.Name) and func.id == "print":
|
|
is_stderr = False
|
|
for kw in node.keywords:
|
|
if kw.arg == "file":
|
|
val = kw.value
|
|
if (
|
|
isinstance(val, ast.Attribute)
|
|
and isinstance(val.value, ast.Name)
|
|
and val.value.id == "sys"
|
|
and val.attr == "stderr"
|
|
):
|
|
is_stderr = True
|
|
break
|
|
if not is_stderr:
|
|
stdout_prints.append(node.lineno)
|
|
self.assertEqual(
|
|
stdout_prints,
|
|
[],
|
|
f"found direct stdout print() calls on lines: {stdout_prints}",
|
|
)
|
|
|
|
|
|
class ObservationInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_heartbeat_and_child_output_stay_in_logs_not_user_event_stream(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
workspace = Path(temporary)
|
|
(workspace / ".git").mkdir()
|
|
task = make_test_task(workspace)
|
|
store = dispatch.StateStore(workspace)
|
|
session_id = "11111111-1111-1111-1111-111111111111"
|
|
|
|
def command_for(
|
|
spec,
|
|
prompt,
|
|
cwd,
|
|
actual_session_id,
|
|
attempt_dir,
|
|
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()
|