diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py b/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py index 12066614..49afaa12 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py +++ b/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py @@ -2309,11 +2309,59 @@ def implementation_review_errors(task: Task) -> list[str]: text = task.review.read_text(encoding="utf-8", errors="replace") checklist = markdown_section(text, IMPLEMENTATION_CHECKLIST_HEADINGS) checkbox_values = IMPLEMENTATION_CHECKBOX_RE.findall(checklist) - if not checkbox_values or any(not value.strip() for value in checkbox_values): + if not checkbox_values: + return ["구현 체크리스트 미완료"] + if ( + any(not value.strip() for value in checkbox_values) + and not implementation_blocker_evidence_complete(text) + ): return ["구현 체크리스트 미완료"] return [] +def implementation_blocker_evidence_complete(text: str) -> bool: + """Allow official review to classify an externally blocked implementation. + + Workers must not invent a verdict merely to bypass the checklist gate. The + exception therefore requires both an explicit blocked runtime/status field + and a concrete, non-placeholder resume condition. The official reviewer + still owns USER_REVIEW, follow-up, and terminal-state classification. + """ + blocked_status = re.search( + r"(?im)^\s*(?:status\s*=\s*|authorization/runtime state:\s*`?)" + r"(?:blocked|차단)\b", + text, + ) + if blocked_status is None: + return False + for match in re.finditer( + r"(?im)^\s*(?:resume condition|재개 조건)\s*:\s*(.+?)\s*$", + text, + ): + value = match.group(1).strip().strip("`").strip() + lowered = value.lower() + if ( + len(value) < 8 + or "<" in value + or ">" in value + or lowered in {"none", "n/a", "na", "없음"} + or any( + token in lowered + for token in ( + "actual state", + "exact condition", + "none or", + "placeholder", + "tbd", + "todo", + ) + ) + ): + continue + return True + return False + + def classify_failure_with_evidence(output: str) -> tuple[str, str | None]: lines = output.splitlines() for category, patterns in RUNTIME_FAILURE_PATTERNS.items(): diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py b/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py index b6fd6419..c3d1249f 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py +++ b/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py @@ -511,6 +511,42 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase): class GenericDispatcherContractTests(unittest.TestCase): + def _implementation_review_errors(self, text: str) -> list[str]: + with TemporaryDirectory() as tmp: + root = Path(tmp) + plan = write_plan(root) + task = task_from_plan(root, plan) + review = plan.parent / "CODE_REVIEW-cloud-G05.md" + review.write_text(text, encoding="utf-8") + task.review = review + return dispatch.implementation_review_errors(task) + + def test_explicit_blocker_evidence_can_reach_official_review(self): + errors = self._implementation_review_errors( + "## Overview\n\n" + "status=BLOCKED — external runtime input is missing\n\n" + "## Implementation Checklist\n\n" + "- [x] Captured available evidence.\n" + "- [ ] Run the external verification after access is ready.\n\n" + "## Verification Results\n\n" + "Resume condition: Register the missing route and provide the named runtime input.\n" + ) + + self.assertEqual(errors, []) + + def test_blocker_placeholder_does_not_bypass_checklist_gate(self): + errors = self._implementation_review_errors( + "## Overview\n\n" + "status=BLOCKED\n\n" + "## Implementation Checklist\n\n" + "- [x] Captured available evidence.\n" + "- [ ] Run the external verification.\n\n" + "## Verification Results\n\n" + "Resume condition: \n" + ) + + self.assertEqual(errors, ["구현 체크리스트 미완료"]) + def test_verdict_parser_accepts_inline_code_value(self): text = "## Code Review Result\n\n- Overall Verdict: `FAIL`\n"