5088 lines
199 KiB
Python
5088 lines
199 KiB
Python
import asyncio
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
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 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("<!-- task=test plan=0 tag=TEST -->\n", encoding="utf-8")
|
|
review.write_text(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_starts_review(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, {}), "review")
|
|
|
|
def test_pi_worker_success_requires_selfcheck_before_review(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
task = self.make_task(root)
|
|
self.assertEqual(
|
|
dispatch.task_stage(task, {"worker_done": True, "selfcheck_done": False}),
|
|
"selfcheck",
|
|
)
|
|
self.assertEqual(
|
|
dispatch.task_stage(task, {"worker_done": True, "selfcheck_done": True}),
|
|
"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),
|
|
7: ("agy", "Gemini 3.6 Flash (Medium)", False),
|
|
8: ("agy", "Gemini 3.6 Flash (Medium)", False),
|
|
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):
|
|
task.grade = grade
|
|
spec = dispatch.route_agent(task)
|
|
self.assertEqual(spec.cli, cli)
|
|
self.assertEqual(spec.model, model)
|
|
self.assertEqual(spec.local_pi, local_pi)
|
|
|
|
def test_selfcheck_requires_all_implementation_review_fields(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"
|
|
"- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 채운다.\n\n"
|
|
"## 계획 대비 변경 사항\n\n_미작성_\n\n"
|
|
"## 주요 설계 결정\n\n_미작성_\n\n"
|
|
"## 검증 결과\n\n결과:\n_미실행_\n",
|
|
)
|
|
self.assertEqual(
|
|
dispatch.implementation_review_errors(task),
|
|
[
|
|
"구현 체크리스트 미완료",
|
|
"구현 항목 완료 여부 미작성",
|
|
"계획 대비 변경 사항 미작성",
|
|
"주요 설계 결정 미작성",
|
|
"검증 결과 미작성",
|
|
"CODE_REVIEW 동기화 체크 미완료",
|
|
],
|
|
)
|
|
|
|
def test_selfcheck_accepts_filled_implementation_review_fields(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
task = self.make_task(
|
|
Path(temporary),
|
|
"## 구현 항목별 완료 여부\n\n"
|
|
"| 항목 | 완료 여부 |\n|---|---|\n| TEST-1 | [x] |\n\n"
|
|
"## 구현 체크리스트\n\n- [x] TEST-1\n"
|
|
"- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 채운다.\n\n"
|
|
"## 계획 대비 변경 사항\n\n계획과 동일하게 구현했다.\n\n"
|
|
"## 주요 설계 결정\n\n상태 전이를 직렬화했다.\n\n"
|
|
"## 검증 결과\n\n결과: PASS (exit 0)\n",
|
|
)
|
|
self.assertEqual(dispatch.implementation_review_errors(task), [])
|
|
|
|
|
|
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_kst_offset(self):
|
|
self.assertRegex(dispatch.work_log_now_iso(), r"\+09:00$")
|
|
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+ \| [^|\n]+\+09:00 \| START \|")
|
|
self.assertRegex(log, r"\| \d+ \| [^|\n]+\+09:00 \| 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_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_caller_lifecycle_is_not_bound_to_child_dispatcher_exit(self):
|
|
skill = (
|
|
Path(__file__).parents[1] / "SKILL.md"
|
|
).read_text(encoding="utf-8")
|
|
self.assertIn(
|
|
"기준 주체는 child dispatcher process가 아니라 이 스킬을 실행한 caller agent 자신",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"child의 종료, yield, session/cell 유실만으로 caller가 `final`을 보내지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"exit code `3`은 다른 dispatcher의 workspace lock, 살아 있는 외부 agent, "
|
|
"예상하지 못한 dispatcher 중단을 포함하는 non-terminal 추적 상태",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"모든 CLI의 health/progress는 실제 stdout/stderr `stream.log`를 우선으로, "
|
|
"native session event가 있는 CLI는 그 event도 함께 본다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"dispatcher PID·agent PID와 각 process start token, "
|
|
"attempt별 process environment marker를 기록",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"모든 모델에서 실제 terminal 오류나 확인된 process 종료만 "
|
|
"복구 근거로 사용한다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"직전 assistant의 모든 `toolCall.id`와 이후 `toolResult.toolCallId`가 일치할 때만",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"tool 실행 구간 밖에서 stream이 3분 멈추면 dispatcher는 마지막 stream 일부를",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"agent PID가 locator에 없으면 로그 시간이 아무리 오래되어도 "
|
|
"stale로 판정하거나 중복 복구하지 않고",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"원래 예외가 persistent-state 오류 유형이어도 실행 중 agent가 하나라도 있었다면 exit `2`로 바꾸지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"남은 attempt 디렉터리가 있으면 성공 exit `0`을 반환하지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"동일 task stage의 연속 자동 복구 실패 예산 10회를 공유한다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"10번째 실패에서 해당 task를 차단하고 자동 cooldown 재개하지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"legacy `session-stall`은 provider 오류가 아니라 과거 dispatcher timeout 정책의 기록",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"실제 provider terminal evidence가 없는 exit code `143`을 provider 오류로 분류하지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"`pi -p` fresh/isolated session의 한 attempt 결과를 Pi TUI 또는 provider 전체 장애로 일반화하지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn("provider_transport_failure_confirmed", skill)
|
|
self.assertIn(
|
|
"일반 tool/test stderr의 `connection refused`, `dial tcp`, `curl` peer failure만으로 provider 오류를 만들지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"실행 중인 Python dispatcher는 source hot edit를 반영하지 않는다",
|
|
skill,
|
|
)
|
|
self.assertIn("dispatcher_source_sha256", skill)
|
|
self.assertIn("`dispatcher_source_matches_loaded=false`", skill)
|
|
self.assertIn(
|
|
"새 route에서 제외된 legacy `laguna-s` locator의 `context-limit`/`session-stall`만 Prompt 계약의 same-session 재개 규칙을 우선",
|
|
skill,
|
|
)
|
|
self.assertIn(
|
|
"그 외 legacy Pi `session-stall` 복구만 fresh session",
|
|
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(
|
|
"dispatcher가 마지막 `FINISH` 기록 후 생성된 `WORK_LOG.md`",
|
|
dispatcher_skill,
|
|
)
|
|
self.assertIn("work_log_N.log", dispatcher_skill)
|
|
self.assertIn(
|
|
"`reconciled:verified-complete-archive` FINISH를 append",
|
|
dispatcher_skill,
|
|
)
|
|
self.assertIn(
|
|
"PID가 없을 때의 stream/native evidence 중 하나라도 살아 있는 동안 "
|
|
"종료·archive하지 않고 추적",
|
|
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",
|
|
)
|
|
try:
|
|
with (
|
|
mock.patch.object(
|
|
dispatch,
|
|
"route_agent",
|
|
return_value=initial_route,
|
|
),
|
|
mock.patch.object(
|
|
dispatch,
|
|
"run_escalating",
|
|
new=mock.AsyncMock(return_value=(True, locator)),
|
|
),
|
|
):
|
|
await dispatch.run_worker(
|
|
root,
|
|
store,
|
|
task,
|
|
{"agy": asyncio.Semaphore(1)},
|
|
)
|
|
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"])
|
|
|
|
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
|
|
]
|
|
semaphore = asyncio.Semaphore(1)
|
|
sleep_observations = []
|
|
|
|
async def observe_sleep(delay):
|
|
sleep_observations.append((delay, semaphore.locked()))
|
|
|
|
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,
|
|
invocation_semaphore=semaphore,
|
|
)
|
|
|
|
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
|
|
)
|
|
self.assertTrue(all(not locked for _, locked in sleep_observations))
|
|
|
|
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)
|
|
store.update_task(
|
|
task, selfcheck_incomplete=dispatch.SELF_CHECK_INCOMPLETE_LIMIT
|
|
)
|
|
try:
|
|
with mock.patch.object(
|
|
dispatch, "run_escalating", new=mock.AsyncMock()
|
|
) as run_escalating:
|
|
await dispatch.run_selfcheck(
|
|
root,
|
|
store,
|
|
task,
|
|
{"pi:ornith:35b": asyncio.Semaphore(1)},
|
|
)
|
|
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)
|
|
store.update_task(
|
|
task,
|
|
selfcheck_incomplete=dispatch.SELF_CHECK_INCOMPLETE_LIMIT - 1,
|
|
)
|
|
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,
|
|
{"pi:ornith:35b": asyncio.Semaphore(1)},
|
|
)
|
|
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"<!-- task={name} plan=0 tag=TEST -->\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, semaphores):
|
|
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, semaphores):
|
|
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,
|
|
)
|
|
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 = "<!-- task=recovery plan=2 tag=REVIEW_TEST -->\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(
|
|
"<!-- task=recovery plan=3 tag=OTHER -->\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(
|
|
"<!-- task=recovery plan=2 tag=REVIEW_TEST -->\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(
|
|
"<!-- task=group/01_done plan=0 tag=TEST -->\n",
|
|
encoding="utf-8",
|
|
)
|
|
review.write_text(
|
|
"<!-- task=group/01_done plan=0 tag=TEST -->\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"<!-- task={name} plan=0 tag=TEST -->\n"
|
|
"## 수정 파일 요약\n\n| 파일 | 항목 |\n|---|---|\n"
|
|
"| `src/task.go` | TEST-1 |\n",
|
|
encoding="utf-8",
|
|
)
|
|
review.write_text(
|
|
f"<!-- task={name} plan=0 tag=TEST -->\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 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"<!-- task={task_name} plan=0 tag=SIM -->\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, semaphores):
|
|
enter("worker", task.name)
|
|
try:
|
|
await asyncio.sleep(0.005)
|
|
store.update_task(
|
|
task,
|
|
worker_done=True,
|
|
worker_cli="pi",
|
|
worker_model="ornith:35b",
|
|
selfcheck_done=False,
|
|
blocked=None,
|
|
)
|
|
finally:
|
|
leave("worker", task.name)
|
|
|
|
async def fake_selfcheck(workspace_path, store, task, semaphores):
|
|
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):
|
|
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",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|