toki
7ae4be6ceb
프로젝트별 복사 없이 동일한 dispatcher와 PLAN/CODE_REVIEW 계약을 공식 Agent-Ops 배포 경로로 전파하기 위해 common skill로 승격한다.
294 lines
11 KiB
Python
294 lines
11 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,
|
||
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()
|