sync: from iop-s0 v1.1.195

agent-ops/.version

@@ -1 +1 @@
-1.1.194
+1.1.195

agent-ops/skills/common/prepare-epic-work-items/scripts/run_epic_cycle.py

@@ -621,7 +621,7 @@ def cycle(args: argparse.Namespace) -> int:
         emit(
             "EPIC_BATCH_VALIDATED",
             identity=identity,
-            event="EPIC_COMPLETED" if not epic.incomplete_ids else "EPIC_WORK_ITEMS_READY",
+            terminal="EPIC_COMPLETED" if not epic.incomplete_ids else "EPIC_WORK_ITEMS_READY",
             plans=len(pairs),
         )
         return 0

agent-ops/skills/common/prepare-epic-work-items/tests/test_run_epic_cycle.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import importlib.util
+import io
+import json
 import os
 from pathlib import Path
 import subprocess
@@ -141,6 +143,85 @@ class EpicCycleContractTest(unittest.TestCase):
         self.assertEqual(selected.task_ids, ("done-task", "open-task"))
         self.assertEqual(selected.incomplete_ids, ("open-task",))
 
+    def test_validate_only_emits_batch_event_with_terminal_state(self) -> None:
+        cases = (
+            ("- [x] [task-one] completed", set(), [], "EPIC_COMPLETED"),
+            (
+                "- [ ] [task-one] pending",
+                {"task-one"},
+                [(Path("PLAN-local-G01.md"), Path("CODE_REVIEW-local-G01.md"))],
+                "EPIC_WORK_ITEMS_READY",
+            ),
+        )
+        with tempfile.TemporaryDirectory() as raw:
+            workspace = Path(raw)
+            milestone = (
+                workspace
+                / "agent-roadmap"
+                / "phase"
+                / "phase-one"
+                / "milestones"
+                / "sample-milestone.md"
+            )
+            milestone.parent.mkdir(parents=True)
+            current = workspace / "agent-roadmap" / "current.md"
+            current.parent.mkdir(parents=True, exist_ok=True)
+            current.write_text(
+                "phase/phase-one/milestones/sample-milestone.md\n",
+                encoding="utf-8",
+            )
+
+            for task_line, task_union, pairs, terminal in cases:
+                with self.subTest(terminal=terminal):
+                    milestone.write_text(
+                        "# Milestone: [sample-01] Sample\n\n"
+                        "## 상태\n\n[계획]\n\n"
+                        "## 구현 잠금\n\n- 상태: 해제\n- 결정 필요: 없음\n\n"
+                        "## 기능\n\n"
+                        "### Epic: [sample-epic] Sample Epic\n\n"
+                        f"{task_line}\n",
+                        encoding="utf-8",
+                    )
+                    args = MODULE.parser().parse_args(
+                        [
+                            "--workspace",
+                            str(workspace),
+                            "--milestone",
+                            str(milestone.relative_to(workspace)),
+                            "--epic",
+                            "sample-epic",
+                            "--execution-catalog",
+                            "/runtime/catalog.json",
+                            "--planner-target",
+                            "planner-primary",
+                            "--validate-only",
+                        ]
+                    )
+                    output = io.StringIO()
+                    with (
+                        mock.patch.object(MODULE, "resolve_workspace", return_value=workspace),
+                        mock.patch.object(
+                            MODULE,
+                            "git",
+                            side_effect=lambda _workspace, *arguments, **_kwargs: {
+                                ("config", "--get", "gitflow.prefix.feature"): "feature/",
+                                ("branch", "--show-current"): "feature/sample-milestone",
+                            }[arguments],
+                        ),
+                        mock.patch.object(
+                            MODULE,
+                            "validate_pairs",
+                            return_value=(pairs, task_union),
+                        ),
+                        mock.patch("sys.stdout", output),
+                    ):
+                        self.assertEqual(MODULE.cycle(args), 0)
+
+                    payload = json.loads(output.getvalue())
+                    self.assertEqual(payload["event"], "EPIC_BATCH_VALIDATED")
+                    self.assertEqual(payload["terminal"], terminal)
+                    self.assertEqual(payload["plans"], len(pairs))
+
     def test_validate_pair_rejects_task_outside_epic(self) -> None:
         with tempfile.TemporaryDirectory() as raw:
             workspace = Path(raw)