toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(agent-ops): dispatcher 리뷰 재진입 경계를 바로잡는다

Browse source 
공식 리뷰가 후속 PLAN 검증을 금지된 중첩 실행으로 오인해 같은 리뷰를 반복하지 않도록 child 경계를 명확히 한다. 영문 USER_REVIEW 계약도 dispatcher 파서와 일치시킨다.
This commit is contained in:
toki 2026-07-30 16:13:39 +09:00
parent 213eee4e28
commit ad4993a9e0
4 changed files with 276 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
agent-ops/skills/common/code-review/templates/user-review-template.md
View file

@ -1,50 +1,50 @@
# User Review Required - {task_name}
## 요청 일시
## Requested At
{YYYY-MM-DD or ISO-8601}
## 상태
## Status
USER_REVIEW
## 사유
## Reason
- 유형: {milestone-lock | external-execution}
- 연결 대상: {agent-roadmap/phase/<phase>/milestones/<milestone>.md | exact runner/device/service/access target}
- 현재 리뷰 회차: {review-number}
- 최종 판정: {WARN or FAIL}
- 요약: {Milestone 결정 또는 user-controlled external execution이 다음 안전한 단계를 차단한 이유}
- Type: {milestone-lock | external-execution}
- Target: {agent-roadmap/phase/<phase>/milestones/<milestone>.md | exact runner/device/service/access target}
- Current review number: {review-number}
- Final verdict: {WARN or FAIL}
- Summary: {why the Milestone decision or user-controlled external execution blocks the next safe step}
## 루프 이력
## Loop History
| Plan | Review | Verdict | 메모 |
| Plan | Review | Verdict | Note |
|------|--------|---------|------|
| `{plan-log-0}` | `{code-review-log-0}` | {PASS/WARN/FAIL/unknown} | {main issue or blocking reason} |
| `{current-archived-plan-log}` | `{current-archived-review-log}` | {WARN/FAIL} | {main issue or blocking reason} |
## 차단 근거
## Blocking Evidence
- 문제: {review finding summary}
- 현재 archive plan: `{current-archived-plan-log}`
- 현재 archive review: `{current-archived-review-log}`
- 검증 명령: `{command or 없음}`
- 실제 출력: {stdout/stderr excerpt or saved output path}
- 차단 판단 근거: {Milestone 결정과 일치하는 근거 | declared runner/transport를 확인하고도 자동 실행할 수 없으며 사용자 조치가 필요한 근거}
- Problem: {review finding summary}
- Current archived plan: `{current-archived-plan-log}`
- Current archived review: `{current-archived-review-log}`
- Verification command: {command or none}
- Actual output: {stdout/stderr excerpt or saved output path}
- Blocking rationale: {evidence matching the Milestone decision | evidence that the declared runner/transport was checked but automatic execution remains unsafe without user action}
## 사용자 조치 또는 결정
## Required User Action
- [ ] {Milestone `구현 잠금 > 결정 필요` 항목 | exact access/authorization/environment/evidence action}
- [ ] {Milestone `구현 잠금 > 결정 필요` item | exact access/authorization/environment/evidence action}
## 재개 조건
## Resume Condition
- {위 사용자 조치 또는 결정이 충족되었음을 확인하는 구체적인 evidence와 후속 review/plan 진입 조건}
- {concrete evidence proving the required action or decision is resolved, plus the next review/plan entry condition}
## 다음 실행 힌트
## Next Execution Hint
- {resolve-review, update-roadmap, external verification replan 중 맞는 진입점과 대상 경로}
- {the correct resolve-review, update-roadmap, or external-verification replan entry point and target path}
## 종료 규칙
## Closure Rules
- 기록된 사용자 조치 또는 결정과 evidence가 이 stop state를 완료/PASS로 해소하면 `USER_REVIEW.md`를 해소 상태로 갱신하고, `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.
- 새 구현이 필요하면 `plan` 스킬이 `USER_REVIEW.md`를 `user_review_N.log`로 아카이브한 뒤 새 `PLAN-*-G??.md` / `CODE_REVIEW-*-G??.md`를 작성한다.
- If the recorded user action and evidence resolve this stop as complete/PASS, update `USER_REVIEW.md` to the resolved state, write `complete.log` from `agent-ops/skills/common/code-review/templates/complete-log-template.md`, and move the task directory to the archive.
- If new implementation is required, the `plan` skill archives `USER_REVIEW.md` as `user_review_N.log` before writing a new `PLAN-*-G??.md` / `CODE_REVIEW-*-G??.md` pair.

10
agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
View file

@ -63,6 +63,12 @@ Treat Korean text inside code spans or fenced examples as exact runtime or file-
- [ ] Verify that no other dispatcher is running in the same workspace. Never bypass a workspace-lock failure.
- [ ] Run `--dry-run` before the first live run to inspect active-task classification and dependency state.
Dispatcher-child re-entry boundary:
- If `IOP_AGENT_TASK_EXECUTION_ID` is present, the current process is already a dispatcher-launched worker, self-check, or reviewer. Do not run this skill's `--dry-run` or live dispatcher procedure, inspect or monitor the parent dispatcher, or wait for its PID. Execute only the assigned PLAN/self-check/review role directly.
- `dispatch.py --validate-plan` remains allowed inside a dispatcher child because it validates one candidate PLAN without starting or monitoring orchestration.
- The dispatcher rejects every other nested invocation before workspace-lock inspection.
## Routing Contract
| PLAN route | Worker |
@ -94,6 +100,10 @@ Concurrency limits:
Keep control prompts in English, insert absolute paths only, and do not expand these sentences unnecessarily.
Prefix every worker, self-check, review, escalation, and recovery prompt exactly with:
`You are a child agent already launched by the dispatcher, not the orchestration caller. Execute only the assigned role directly. Do not start, monitor, or wait for orchestration through dispatch.py or orchestrate-agent-task-loop. You may run dispatch.py --validate-plan only when required by plan or code-review finalization because that mode validates one candidate PLAN without starting or monitoring orchestration.`
- Cloud worker: `Read {PLAN_PATH} and complete the task. Keep artifact content in English. Final in Korean.`
- Pi worker: `Think in English. Keep artifact content in English. Final in Korean. Read {PLAN_PATH} and complete the task.`
- Pi self-check: `Think in English. Keep artifact content in English. Final in Korean. Read {CODE_REVIEW_PATH} and fill every missing implementation field. Do not finish until all implementation fields are complete. This is a self-check of completed work, not a review. Read {PLAN_PATH} and finish any missing work. Recheck and fix your work.`

126
agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py
View file

@ -74,6 +74,28 @@ CODE_REVIEW_RESULT_SCHEMAS = (
("Code Review Result", "Overall Verdict"),
("코드리뷰 결과", "종합 판정"),
)
USER_REVIEW_SCHEMAS = (
{
"status_heading": "Status",
"reason_heading": "Reason",
"type_label": "Type",
"target_label": "Target",
"evidence_heading": "Blocking Evidence",
"evidence_label": "Blocking rationale",
"decision_headings": ("Required User Action",),
"resume_heading": "Resume Condition",
},
{
"status_heading": "상태",
"reason_heading": "사유",
"type_label": "유형",
"target_label": "연결 대상",
"evidence_heading": "차단 근거",
"evidence_label": "차단 판단 근거",
"decision_headings": ("사용자 조치 또는 결정", "연결 결정 필요"),
"resume_heading": "재개 조건",
},
)
VERDICT_SCHEMA_MATCHERS = tuple(
(
re.compile(rf"^##\s*{re.escape(heading)}[ \t]*$", re.MULTILINE),
@ -97,6 +119,14 @@ IMPLEMENTATION_CHECKBOX_RE = re.compile(
WORK_LOG_NAME = "WORK_LOG.md"
WORK_LOG_ARCHIVE_RE = re.compile(r"^work_log_(\d+)\.log$")
AGENT_PROCESS_MARKER_ENV = "IOP_AGENT_TASK_EXECUTION_ID"
DISPATCHER_CHILD_BOUNDARY_PROMPT = (
"You are a child agent already launched by the dispatcher, not the "
"orchestration caller. Execute only the assigned role directly. Do not "
"start, monitor, or wait for orchestration through dispatch.py or "
"orchestrate-agent-task-loop. You may run dispatch.py --validate-plan only "
"when required by plan or code-review finalization because that mode "
"validates one candidate PLAN without starting or monitoring orchestration."
)
KST = timezone(timedelta(hours=9), name="KST")
STREAM_HEARTBEAT_SECONDS = 30
PI_MODEL_RESPONSE_STALL_SECONDS = 3 * 60
@ -2255,18 +2285,34 @@ def user_review_blocker_state(path: Path) -> tuple[bool, str]:
text = path.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
return False, f"파일을 읽을 수 없다: {exc}"
status = markdown_section(text, "상태").strip().strip("`")
matched_schemas = [
schema
for schema in USER_REVIEW_SCHEMAS
if re.search(
rf"^##\s*{re.escape(schema['status_heading'])}[ \t]*$",
text,
re.MULTILINE,
)
]
if len(matched_schemas) != 1:
return False, "지원하는 USER_REVIEW schema가 정확히 하나가 아니다"
schema = matched_schemas[0]
status = markdown_section(text, schema["status_heading"]).strip().strip("`")
if status != "USER_REVIEW":
return False, "상태가 USER_REVIEW가 아니다"
reason = markdown_section(text, "사유")
reason = markdown_section(text, schema["reason_heading"])
gate_type_matches = re.findall(
r"(?m)^-\s*유형:\s*(milestone-lock|external-execution)\s*$",
rf"(?m)^-\s*{re.escape(schema['type_label'])}:\s*"
r"(milestone-lock|external-execution)\s*$",
reason,
)
if len(gate_type_matches) != 1:
return False, "지원하는 user-review 유형이 정확히 하나가 아니다"
gate_type = gate_type_matches[0]
target = re.search(r"(?m)^-\s*연결 대상:\s*(.+?)\s*$", reason)
target = re.search(
rf"(?m)^-\s*{re.escape(schema['target_label'])}:\s*(.+?)\s*$",
reason,
)
target_value = target.group(1) if target else ""
if not concrete_user_review_value(target_value):
return False, "구체적인 연결 대상이 없다"
@ -2276,18 +2322,16 @@ def user_review_blocker_state(path: Path) -> tuple[bool, str]:
or ".md" not in target_value
):
return False, "구체적인 Milestone 연결 대상이 없다"
evidence = markdown_section(text, "차단 근거")
evidence_line = re.search(r"(?m)^-\s*차단 판단 근거:\s*(.+?)\s*$", evidence)
evidence = markdown_section(text, schema["evidence_heading"])
evidence_line = re.search(
rf"(?m)^-\s*{re.escape(schema['evidence_label'])}:\s*(.+?)\s*$",
evidence,
)
if evidence_line is None or not concrete_user_review_value(
evidence_line.group(1)
):
return False, "구체적인 차단 판단 근거가 없다"
decision = markdown_section(text, "사용자 조치 또는 결정")
legacy_decision = markdown_section(text, "연결 결정 필요")
if decision.strip() and legacy_decision.strip():
return False, "사용자 조치 또는 결정 섹션이 중복됐다"
if not decision.strip():
decision = legacy_decision
decision = markdown_section(text, schema["decision_headings"])
unresolved = [
value
for value in re.findall(r"(?m)^-\s*\[\s\]\s+(.+?)\s*$", decision)
@ -2295,7 +2339,7 @@ def user_review_blocker_state(path: Path) -> tuple[bool, str]:
]
if not unresolved:
return False, "미해결 사용자 조치 또는 결정 항목이 없다"
resume = markdown_section(text, "재개 조건")
resume = markdown_section(text, schema["resume_heading"])
resume_conditions = [
line
for line in resume.splitlines()
@ -4069,27 +4113,43 @@ async def invoke(
return return_code, failure_class, locator_path
def dispatcher_child_prompt(body: str) -> str:
return f"{DISPATCHER_CHILD_BOUNDARY_PROMPT} {body}"
def base_prompt(task: Task, role: str, spec: AgentSpec) -> str:
if role == "review":
target = task.review or task.directory
if task.review:
return f"Read {target.resolve()} and start the review. Keep artifact content in English. Final in Korean."
return f"Continue the review for {target.resolve()}. Keep artifact content in English. Final in Korean."
return dispatcher_child_prompt(
f"Read {target.resolve()} and start the review. Keep artifact "
"content in English. Final in Korean."
)
return dispatcher_child_prompt(
f"Continue the review for {target.resolve()}. Keep artifact content "
"in English. Final in Korean."
)
if task.plan is None:
raise RuntimeError("worker PLAN이 없다")
target = task.plan.resolve()
if role == "selfcheck":
if task.review is None:
raise RuntimeError("selfcheck CODE_REVIEW 파일이 없다")
return (
return dispatcher_child_prompt(
f"Think in English. Keep artifact content 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 completed work, not a review. "
f"Read {target} and finish any missing work. Recheck and fix your work."
)
if spec.local_pi:
return f"Think in English. Keep artifact content in English. Final in Korean. Read {target} and complete the task."
return f"Read {target} and complete the task. Keep artifact content in English. Final in Korean."
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in "
f"Korean. Read {target} and complete the task."
)
return dispatcher_child_prompt(
f"Read {target} and complete the task. Keep artifact content in English. "
"Final in Korean."
)
@ -4164,7 +4224,7 @@ def logical_context_prompt(context: dict[str, Any]) -> str:
workspace = context["workspace"]
raw_log = context["raw_log"]
normalized_output = context["normalized_output"]
return (
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in Korean. "
f"Read plan={plan}, locator={locator}, workspace={workspace}, "
f"raw_log={raw_log}, normalized_output={normalized_output} and complete the task."
@ -4178,7 +4238,7 @@ def continuation_prompt_from_package(
native_resume: bool = False,
) -> str:
if native_resume or context_package.get("resume_mode") == "native":
return (
return dispatcher_child_prompt(
"Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete "
"the current task."
)
@ -4187,7 +4247,7 @@ def continuation_prompt_from_package(
workspace = context_package["workspace"]
raw_log = context_package["raw_log"]
normalized_output = context_package["normalized_output"]
return (
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in Korean. "
f"Read plan={plan}, locator={locator}, workspace={workspace}, "
f"raw_log={raw_log}, normalized_output={normalized_output} and complete the task."
@ -4210,12 +4270,12 @@ def continuation_prompt(
)
if local_pi:
if resume_same_pi_session:
return (
return dispatcher_child_prompt(
"Think in English. Keep artifact content in English. Final in Korean. Continue this session and complete "
"the current task."
)
if role == "selfcheck" and task.plan and task.review:
return (
return dispatcher_child_prompt(
f"Think in English. Keep artifact content 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 completed work, not a review. "
@ -4223,10 +4283,16 @@ def continuation_prompt(
"your work."
)
target = task.plan or task.directory
return f"Think in English. Keep artifact content in English. Final in Korean. Read {target.resolve()} and complete the task."
return dispatcher_child_prompt(
f"Think in English. Keep artifact content in English. Final in "
f"Korean. Read {target.resolve()} and complete the task."
)
if role == "review":
return f"Continue the review for {task.directory.resolve()}. Keep artifact content in English. Final in Korean."
return (
return dispatcher_child_prompt(
f"Continue the review for {task.directory.resolve()}. Keep artifact "
"content in English. Final in Korean."
)
return dispatcher_child_prompt(
f"Continue from {locator.resolve() if locator else task.directory.resolve()}. Check the saved context and current "
"workspace. Keep artifact content in English. Final in Korean."
)
@ -6473,6 +6539,14 @@ def main() -> int:
for path in sorted(write_set):
validation_claim(path)
return 0
if os.environ.get(AGENT_PROCESS_MARKER_ENV):
print(
"nested dispatcher invocation rejected: this process is already a "
"dispatcher child; continue the assigned role directly and do not "
"wait for the parent dispatcher",
file=sys.stderr,
)
return 4
try:
return asyncio.run(dispatch(args))
except KeyboardInterrupt:

149
agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py
View file

@ -179,6 +179,22 @@ class TaskStageTest(unittest.TestCase):
"- Verify SSH access or the supplied evidence before resuming review.\n"
)
@staticmethod
def blocking_english_external_user_review_text():
return (
"# User Review Required - test\n\n"
"## Status\n\nUSER_REVIEW\n\n"
"## Reason\n\n"
"- Type: external-execution\n"
"- Target: ssh toki@toki-labs.com:/Users/toki/agent-work/iop-dev\n\n"
"## Blocking Evidence\n\n"
"- Blocking rationale: Required dev smoke needs a user-controlled runner and renewed authorization.\n\n"
"## Required User Action\n\n"
"- [ ] Authorize one replacement live invocation.\n\n"
"## Resume Condition\n\n"
"- Verify idle provider capacity before resuming.\n"
)
def test_default_or_arbitrary_status_text_does_not_start_review(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
@ -235,6 +251,37 @@ class TaskStageTest(unittest.TestCase):
task.recovery = True
self.assertEqual(dispatch.task_stage(task, {}), "user-review")
def test_english_external_execution_user_review_stops_the_loop(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_english_external_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_rejects_mixed_language_schema(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_english_external_user_review_text().replace(
"## Status\n\nUSER_REVIEW\n\n",
"## Status\n\nUSER_REVIEW\n\n## 상태\n\nUSER_REVIEW\n\n",
),
encoding="utf-8",
)
task.plan = None
task.review = None
task.recovery = True
self.assertEqual(dispatch.task_stage(task, {}), "blocked")
def test_external_execution_user_review_requires_concrete_target(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
@ -3417,15 +3464,20 @@ class ReviewControlTest(unittest.TestCase):
self.assertNotIn("user review", selfcheck.lower())
self.assertEqual(
selfcheck,
f"Think in English. Keep artifact content 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.",
dispatch.dispatcher_child_prompt(
f"Think in English. Keep artifact content 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. Keep artifact content in English. Final in Korean.",
dispatch.dispatcher_child_prompt(
f"Read {task.review.resolve()} and start the review. Keep "
"artifact content in English. Final in Korean."
),
)
def test_local_review_stub_has_no_user_review_control_plane_content(self):
@ -4033,8 +4085,11 @@ class ReviewRetryTest(unittest.IsolatedAsyncioTestCase):
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. Keep artifact content in English. Final in Korean. Continue this session and "
"complete the current task.",
dispatch.dispatcher_child_prompt(
"Think in English. Keep artifact content in English. Final "
"in Korean. Continue this session and complete the current "
"task."
),
)
self.assertEqual(
invoke.await_args_list[1].kwargs["resume_locator"],
@ -6858,6 +6913,59 @@ class OrchestrationPersistenceTest(unittest.TestCase):
finally:
owner.close()
def test_dispatcher_child_rejects_nested_orchestration_before_lock(self):
args = SimpleNamespace(
workspace=".",
task_group="m-test",
dry_run=True,
retry_blocked=False,
validate_plan=None,
)
with (
mock.patch.object(dispatch, "parse_args", return_value=args),
mock.patch.dict(
os.environ,
{dispatch.AGENT_PROCESS_MARKER_ENV: "owned-worker"},
),
mock.patch.object(dispatch.asyncio, "run") as run,
mock.patch("sys.stderr", new_callable=io.StringIO) as stderr,
):
result = dispatch.main()
self.assertEqual(result, 4)
run.assert_not_called()
self.assertIn("nested dispatcher invocation rejected", stderr.getvalue())
self.assertIn("do not wait for the parent dispatcher", stderr.getvalue())
def test_dispatcher_child_can_validate_one_plan(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
plan = workspace / "PLAN-cloud-G10.md"
target = workspace / "src" / "target.go"
plan.write_text(
"## Modified Files Summary\n\n"
"| File | Items |\n|---|---|\n"
f"| `{target}` | TEST-1 |\n",
encoding="utf-8",
)
args = SimpleNamespace(
workspace=str(workspace),
task_group=None,
dry_run=False,
retry_blocked=False,
validate_plan=str(plan),
)
with (
mock.patch.object(dispatch, "parse_args", return_value=args),
mock.patch.dict(
os.environ,
{dispatch.AGENT_PROCESS_MARKER_ENV: "owned-review"},
),
):
result = dispatch.main()
self.assertEqual(result, 0)
def test_unexpected_dispatcher_exception_is_non_terminal_exit_three(self):
args = SimpleNamespace(
workspace=".",
@ -11110,7 +11218,7 @@ class ArtifactLanguageContractTest(unittest.TestCase):
"{milestone-lock | external-execution}",
user_review_template,
)
self.assertIn("## 사용자 조치 또는 결정", user_review_template)
self.assertIn("## Required User Action", user_review_template)
def test_templates_and_prompts_separate_artifact_and_final_languages(self):
documents = self.contract_documents()
@ -11276,9 +11384,32 @@ class ArtifactLanguageContractTest(unittest.TestCase):
}
for name, prompt in prompts.items():
with self.subTest(prompt=name):
self.assertTrue(
prompt.startswith(
dispatch.DISPATCHER_CHILD_BOUNDARY_PROMPT
)
)
self.assertIn("Keep artifact content in English.", prompt)
self.assertIn("Final in Korean.", prompt)
self.assertIn(
"`IOP_AGENT_TASK_EXECUTION_ID` is present",
orchestrator_skill,
)
self.assertIn(
dispatch.DISPATCHER_CHILD_BOUNDARY_PROMPT,
orchestrator_skill,
)
self.assertIn(
"You may run dispatch.py --validate-plan only when required by "
"plan or code-review finalization",
dispatch.DISPATCHER_CHILD_BOUNDARY_PROMPT,
)
self.assertNotIn(
"Do not invoke, monitor, or wait for dispatch.py",
dispatch.DISPATCHER_CHILD_BOUNDARY_PROMPT,
)
if __name__ == "__main__":
unittest.main()