From 18ef007fca6249b66ac25c66bd3617428f2efec7 Mon Sep 17 00:00:00 2001 From: toki Date: Mon, 10 Aug 2026 07:57:39 +0900 Subject: [PATCH] =?UTF-8?q?fix(agent-ops):=20=EC=99=B8=EB=B6=80=20?= =?UTF-8?q?=EC=B0=A8=EB=8B=A8=20=EC=A6=9D=EA=B1=B0=EB=A5=BC=20=EA=B3=B5?= =?UTF-8?q?=EC=8B=9D=20=EB=A6=AC=EB=B7=B0=EB=A1=9C=20=EB=84=98=EA=B8=B4?= =?UTF-8?q?=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../scripts/dispatch.py | 50 ++++++++++++++++++- .../tests/test_dispatch.py | 36 +++++++++++++ 2 files changed, 85 insertions(+), 1 deletion(-) 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"