sync: agent-ops from agentic-framework v1.1.193

agent-ops/.version

@@ -1 +1 @@
-1.1.192
+1.1.193

agent-ops/skills/common/orchestrate-agent-task-loop/SKILL.md

@@ -48,10 +48,10 @@ Each target has:
 - `execution_class`: `local_model` or `cloud_model`;
 - optional `selfcheck_required` boolean;
 - `runtime.command`: a non-empty argv template executed without a shell;
-- optional `runtime.resume_command`, `preflight_command`, `environment`, `session_path`, `native_session_monitor`, and `auxiliary_logs`;
+- optional `runtime.resume_command`, `preflight_command`, `environment`, `session_path`, `native_session_monitor`, `terminal_success`, and `auxiliary_logs`;
 - optional `runtime.output_format`: `text` or `jsonl`.
 
-Command templates may use only `{agent}`, `{model}`, `{target_id}`, `{workspace}`, `{attempt_dir}`, `{session_id}`, `{resume_session}`, and `{prompt}`. The catalog must not embed repository secrets; environment values should refer only to runtime-provided non-secret configuration.
+Command templates may use only `{agent}`, `{model}`, `{target_id}`, `{workspace}`, `{attempt_dir}`, `{session_id}`, `{resume_session}`, `{resume_session_dir}`, and `{prompt}`. `native_session_monitor=true` requires both `resume_command` and `session_path`. `terminal_success=agent_end` requires JSONL output and accepts only a non-retrying final `agent_end` whose last assistant message has `stopReason=stop`; `error`, `aborted`, a missing event, or another stop reason fails closed. The catalog must not embed repository secrets; environment values should refer only to runtime-provided non-secret configuration.
 
 Each route owns its ordered `candidates` plus optional `rule_id`, `policy_priority`, and `reason_codes`. A route may use catalog-owned `windows` instead of a fixed candidate list; every window supplies an IANA timezone, start/end time, and candidates. Exactly one window must match.
 
@@ -103,7 +103,7 @@ Accept self-check completion only when `## Implementation Checklist` or its supp
 
 - Store each attempt under the dispatcher state directory with `locator.json`, `stream.log`, `normalized-output.log`, and `heartbeat.log`.
 - Record the target id, opaque agent/model identity, execution class, runtime contract, catalog evidence, process identity, workspace identity, timestamps, result, and exact failure evidence.
-- Treat stderr as terminal diagnostic evidence. For JSONL, recognize generic terminal event fields such as error/fatal type or severity, rejected/failed status with an error code, and explicit error flags.
+- Treat stderr as terminal diagnostic evidence. For JSONL, recognize generic terminal event fields such as error/fatal type or severity, rejected/failed status with an error code, explicit error flags, and a non-retrying `agent_end` whose last assistant message ends with `error` or `aborted`.
 - Determine liveness from PID/start-token/process-marker evidence and actual stream or native-session progress. Heartbeat mtime is never agent progress.
 - Never start a duplicate attempt while owned live evidence remains.
 - Keep a 10-consecutive-failure budget per task stage. Reset only that stage's budget after success.
@@ -112,7 +112,7 @@ Accept self-check completion only when `## Implementation Checklist` or its supp
 ## Work log
 
 - Keep one dispatcher-owned `WORK_LOG.md` per task group.
-- Append chronological `START` and `FINISH` rows with KST (`Asia/Seoul`) time, task artifact, plan loop, role, attempt, selected agent/model display, result, and locator.
+- Append chronological `START` and `FINISH` rows with KST (`Asia/Seoul`) time, task artifact, plan loop, role, attempt, selected agent/model display, result, and locator. Use the PLAN artifact for worker and self-check rows; use the CODE_REVIEW artifact only for official review rows.
 - Archive the group log as the next `work_log_N.log` only after every observed task in the group is verified complete and idle.
 - Work-log write or archive failure is a retryable control-plane failure and prevents exit `0`.
 

agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json

@@ -25,7 +25,28 @@
           "{attempt_dir}/pi-sessions",
           "{prompt}"
         ],
-        "output_format": "jsonl"
+        "resume_command": [
+          "pi",
+          "-p",
+          "--mode",
+          "json",
+          "--approve",
+          "--provider",
+          "iop",
+          "--model",
+          "{model}",
+          "--thinking",
+          "high",
+          "--session",
+          "{resume_session}",
+          "--session-dir",
+          "{resume_session_dir}",
+          "{prompt}"
+        ],
+        "output_format": "jsonl",
+        "session_path": "{attempt_dir}/pi-sessions/*{session_id}*.jsonl",
+        "native_session_monitor": true,
+        "terminal_success": "agent_end"
       }
     },
     "agy-gemini-low": {

agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py

@@ -390,7 +390,7 @@ def milestone_work_log_path(task: Task) -> Path:
 
 def work_log_task_name(task: Task, role: str) -> str:
     """Return the role-specific active artifact shown in the task column."""
-    artifact = task.plan if role == "worker" else task.review
+    artifact = task.review if role == "review" else task.plan
     if artifact is None:
         return task.name
     return f"{task.name}/{artifact.name}"
@@ -2410,6 +2410,40 @@ def failure_report_lines(failure: str, locator: Path) -> list[str]:
     return lines
 
 
+def json_agent_terminal_outcome(value: object) -> str | None:
+    """Return the final agent outcome encoded by a generic JSONL event."""
+    if not isinstance(value, dict):
+        return None
+    if str(value.get("type", "")).lower() != "agent_end":
+        return None
+    if value.get("willRetry") is True:
+        return None
+    messages = value.get("messages")
+    if not isinstance(messages, list):
+        return None
+    for message in reversed(messages):
+        if (
+            not isinstance(message, dict)
+            or str(message.get("role", "")).lower() != "assistant"
+        ):
+            continue
+        stop_reason = message.get("stopReason") or message.get("stop_reason")
+        normalized = str(stop_reason or "").lower()
+        if normalized in {"error", "aborted"}:
+            return "failed"
+        if normalized == "stop":
+            return "succeeded"
+        return None
+    return None
+
+
+def json_agent_terminal_outcome_from_line(line: str) -> str | None:
+    try:
+        return json_agent_terminal_outcome(json.loads(line))
+    except json.JSONDecodeError:
+        return None
+
+
 def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None:
     if channel == "stderr":
         return line
@@ -2424,7 +2458,8 @@ def terminal_diagnostic(cli: str, channel: str, line: str) -> str | None:
     status = str(value.get("status") or "").lower()
     subtype = str(value.get("subtype") or "").lower()
     if (
-        event_type in {"error", "fatal", "request.failed", "turn.failed", "rate_limit_event"}
+        json_agent_terminal_outcome(value) == "failed"
+        or event_type in {"error", "fatal", "request.failed", "turn.failed", "rate_limit_event"}
         or severity in {"error", "fatal"}
         or subtype.startswith("error")
         or bool(value.get("is_error"))
@@ -2586,6 +2621,7 @@ def native_session_path(
         "model": spec.model,
         "prompt": "",
         "resume_session": "",
+        "resume_session_dir": "",
         "session_id": session_id,
         "target_id": str(spec.target_id or ""),
         "workspace": str(workspace),
@@ -3025,6 +3061,11 @@ def build_command(
         "model": spec.model,
         "prompt": prompt,
         "resume_session": str(native_resume_session or ""),
+        "resume_session_dir": (
+            str(native_resume_session.parent)
+            if native_resume_session is not None
+            else ""
+        ),
         "session_id": session_id,
         "target_id": str(spec.target_id or ""),
         "workspace": str(workspace),
@@ -3054,6 +3095,7 @@ def preflight_execution_catalog(
             "model": target.model,
             "prompt": "",
             "resume_session": "",
+            "resume_session_dir": "",
             "session_id": "preflight-session",
             "target_id": target_id,
             "workspace": str(checked_workspace),
@@ -3199,6 +3241,11 @@ async def invoke(
                     "model": spec.model,
                     "prompt": "",
                     "resume_session": str(native_resume_session or ""),
+                    "resume_session_dir": (
+                        str(native_resume_session.parent)
+                        if native_resume_session is not None
+                        else ""
+                    ),
                     "session_id": session_id,
                     "target_id": str(spec.target_id or ""),
                     "workspace": str(workspace),
@@ -3299,6 +3346,8 @@ async def invoke(
     diagnostics: list[str] = []
     diagnostic_origins: list[str] = []
     control_violation: str | None = None
+    terminal_success_contract = spec.runtime.get("terminal_success")
+    terminal_success_seen = False
     try:
         runtime_values = {
             "agent": spec.cli,
@@ -3306,6 +3355,11 @@ async def invoke(
             "model": spec.model,
             "prompt": prompt,
             "resume_session": str(native_resume_session or ""),
+            "resume_session_dir": (
+                str(native_resume_session.parent)
+                if native_resume_session is not None
+                else ""
+            ),
             "session_id": session_id,
             "target_id": str(spec.target_id or ""),
             "workspace": str(workspace),
@@ -3537,6 +3591,11 @@ async def invoke(
                 line = raw.decode("utf-8", errors="replace").rstrip("\n")
                 stream_log.write(f"[{channel}] {line}\n")
                 stream_log.flush()
+                if (
+                    channel == "stdout"
+                    and json_agent_terminal_outcome_from_line(line) == "succeeded"
+                ):
+                    terminal_success_seen = True
                 diagnostic = terminal_diagnostic(spec.cli, channel, line)
                 if diagnostic:
                     diagnostics.append(diagnostic)
@@ -3649,23 +3708,46 @@ async def invoke(
         classified_failure, classified_evidence = classify_failure_with_evidence(
             "\n".join(diagnostics[-50:])
         )
+        stdout_diagnostic_index = next(
+            (
+                index
+                for index in range(len(diagnostic_origins) - 1, -1, -1)
+                if diagnostic_origins[index].endswith(":stdout")
+            ),
+            None,
+        )
         if return_code != 0 or classified_evidence is not None:
             failure_class = classified_failure
             failure_evidence = classified_evidence
+        elif stdout_diagnostic_index is not None:
+            failure_class = "generic-error"
+            failure_evidence = diagnostics[stdout_diagnostic_index]
+        elif terminal_success_contract == "agent_end" and not terminal_success_seen:
+            failure_class = "generic-error"
+            failure_source = "dispatcher-terminal-contract"
+            failure_evidence = (
+                "required agent_end terminal success event was not observed"
+            )
+            failure_evidence_source = "dispatcher:terminal-contract"
         else:
             failure_class = None
-        if failure_class is not None and failure_evidence is not None:
+        if (
+            failure_class is not None
+            and failure_evidence is not None
+            and failure_evidence_source is None
+        ):
             for index in range(len(diagnostics) - 1, -1, -1):
                 if diagnostics[index] == failure_evidence:
                     failure_evidence_source = diagnostic_origins[index]
                     break
-        if failure_class in PROVIDER_TRANSPORT_FAILURES:
-            failure_source = "provider-terminal-diagnostic"
-            provider_transport_failure_confirmed = failure_evidence is not None
-        elif failure_evidence is not None:
-            failure_source = "cli-terminal-diagnostic"
-        elif return_code != 0:
-            failure_source = "cli-exit"
+        if failure_source is None:
+            if failure_class in PROVIDER_TRANSPORT_FAILURES:
+                failure_source = "provider-terminal-diagnostic"
+                provider_transport_failure_confirmed = failure_evidence is not None
+            elif failure_evidence is not None:
+                failure_source = "cli-terminal-diagnostic"
+            elif return_code != 0:
+                failure_source = "cli-exit"
     try:
         append_milestone_event(
             task,

agent-ops/skills/common/orchestrate-agent-task-loop/scripts/execution_target_policy.py

@@ -28,6 +28,7 @@ ALLOWED_TEMPLATE_FIELDS = {
     "model",
     "prompt",
     "resume_session",
+    "resume_session_dir",
     "session_id",
     "target_id",
     "workspace",
@@ -106,6 +107,7 @@ def _validate_runtime(value: object, label: str) -> dict[str, Any]:
         "output_format",
         "session_path",
         "native_session_monitor",
+        "terminal_success",
         "auxiliary_logs",
     }
     if unknown:
@@ -121,6 +123,17 @@ def _validate_runtime(value: object, label: str) -> dict[str, Any]:
         raise CatalogError(
             f"{label}.output_format must be one of {sorted(VALID_OUTPUT_FORMATS)}"
         )
+    terminal_success = value.get("terminal_success")
+    if terminal_success is not None:
+        if terminal_success != "agent_end":
+            raise CatalogError(
+                f"{label}.terminal_success must be 'agent_end'"
+            )
+        if runtime["output_format"] != "jsonl":
+            raise CatalogError(
+                f"{label}.terminal_success requires output_format='jsonl'"
+            )
+        runtime["terminal_success"] = terminal_success
     for field in ("resume_command", "preflight_command"):
         if field in value:
             template = list(
@@ -150,6 +163,16 @@ def _validate_runtime(value: object, label: str) -> dict[str, Any]:
     if not isinstance(monitor, bool):
         raise CatalogError(f"{label}.native_session_monitor must be a boolean")
     runtime["native_session_monitor"] = monitor
+    if monitor:
+        missing = [
+            field
+            for field in ("resume_command", "session_path")
+            if field not in runtime
+        ]
+        if missing:
+            raise CatalogError(
+                f"{label}.native_session_monitor requires {missing}"
+            )
     auxiliary_logs = value.get("auxiliary_logs", [])
     if not isinstance(auxiliary_logs, list) or not all(
         isinstance(item, str) and item for item in auxiliary_logs

agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatch.py

@@ -30,7 +30,14 @@ def catalog_value(command: str = "/bin/true") -> dict:
             "selfcheck_required": True,
             "runtime": {
                 "command": [command, "--workspace", "{workspace}", "--model", "{model}", "{prompt}"],
-                "resume_command": [command, "--resume", "{resume_session}", "{prompt}"],
+                "resume_command": [
+                    command,
+                    "--resume",
+                    "{resume_session}",
+                    "--session-dir",
+                    "{resume_session_dir}",
+                    "{prompt}",
+                ],
                 "environment": {"TARGET_ID": "{target_id}"},
                 "output_format": "jsonl",
                 "native_session_monitor": True,
@@ -85,7 +92,7 @@ def task_from_plan(root: Path, plan: Path) -> dispatch.Task:
         user_review=None,
         recovery=False,
         index=1,
-        write_set={"src/item.txt"},
+        write_set={str((root / "src/item.txt").resolve())},
         write_set_known=True,
         plan_hash=dispatch.sha256_file(plan),
     )
@@ -144,7 +151,13 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
             target_id="opaque-id",
             runtime={
                 "command": ["runner", "{workspace}", "{model}", "{session_id}", "{attempt_dir}", "{prompt}"],
-                "resume_command": ["runner", "resume", "{resume_session}", "{prompt}"],
+                "resume_command": [
+                    "runner",
+                    "resume",
+                    "{resume_session}",
+                    "{resume_session_dir}",
+                    "{prompt}",
+                ],
             },
         )
         command = dispatch.build_command(
@@ -163,7 +176,16 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
             native_resume_session=Path("/attempt/session.jsonl"),
         )
         self.assertEqual(command, ["runner", "/workspace", "opaque-model", "session-1", "/attempt", "do work"])
-        self.assertEqual(resumed, ["runner", "resume", "/attempt/session.jsonl", "continue"])
+        self.assertEqual(
+            resumed,
+            [
+                "runner",
+                "resume",
+                "/attempt/session.jsonl",
+                "/attempt",
+                "continue",
+            ],
+        )
 
     def test_preflight_checks_executable_and_optional_probe(self):
         with TemporaryDirectory() as tmp:
@@ -258,6 +280,169 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
             )
         )
 
+    def test_agent_end_terminal_diagnostic_respects_retry_and_stop_reason(self):
+        retrying = {
+            "type": "agent_end",
+            "willRetry": True,
+            "messages": [
+                {
+                    "role": "assistant",
+                    "stopReason": "error",
+                    "errorMessage": "provider_tunnel_error",
+                }
+            ],
+        }
+        failed = {
+            **retrying,
+            "willRetry": False,
+        }
+        succeeded = {
+            "type": "agent_end",
+            "willRetry": False,
+            "messages": [{"role": "assistant", "stopReason": "stop"}],
+        }
+
+        self.assertIsNone(
+            dispatch.terminal_diagnostic(
+                "opaque-agent", "stdout", json.dumps(retrying)
+            )
+        )
+        self.assertIn(
+            "provider_tunnel_error",
+            dispatch.terminal_diagnostic(
+                "opaque-agent", "stdout", json.dumps(failed)
+            )
+            or "",
+        )
+        self.assertEqual(dispatch.json_agent_terminal_outcome(succeeded), "succeeded")
+        self.assertIsNone(
+            dispatch.terminal_diagnostic(
+                "opaque-agent", "stdout", json.dumps(succeeded)
+            )
+        )
+
+    def _invoke_fake_json_event(
+        self,
+        event: dict,
+        *,
+        stderr_line: str | None = None,
+    ) -> tuple[int, str | None, dict, str]:
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            plan = write_plan(root)
+            task = task_from_plan(root, plan)
+            runner = root / "fake_json_runner.py"
+            runner.write_text(
+                "import json\n"
+                "import sys\n"
+                f"print(json.dumps({event!r}))\n"
+                f"print({stderr_line!r}, file=sys.stderr) if {stderr_line!r} else None\n",
+                encoding="utf-8",
+            )
+            runtime = {
+                "command": [sys.executable, str(runner)],
+                "output_format": "jsonl",
+                "terminal_success": "agent_end",
+            }
+            agent = dispatch.AgentSpec(
+                "fake-json-runner",
+                "fake-model",
+                "fake-json-runner/fake-model",
+                target_id="fake-json-target",
+                runtime=runtime,
+            )
+            with mock.patch.dict(
+                os.environ,
+                {"XDG_STATE_HOME": str(root / "state")},
+            ):
+                store = dispatch.StateStore(root)
+                try:
+                    return_code, failure, locator = asyncio.run(
+                        dispatch.invoke(
+                            root,
+                            store,
+                            task,
+                            "worker",
+                            agent,
+                            "fake prompt",
+                        )
+                    )
+                    record = json.loads(locator.read_text(encoding="utf-8"))
+                    self.assertEqual(record["cli"], "fake-json-runner")
+                    work_log = (
+                        root / "agent-task" / "group" / "WORK_LOG.md"
+                    ).read_text(encoding="utf-8")
+                finally:
+                    store.close()
+        return return_code, failure, record, work_log
+
+    def test_zero_exit_agent_error_is_not_recorded_as_success(self):
+        event = {
+            "type": "agent_end",
+            "willRetry": False,
+            "messages": [
+                {
+                    "role": "assistant",
+                    "stopReason": "error",
+                    "errorMessage": "provider_tunnel_error: recovery_failed",
+                }
+            ],
+        }
+
+        return_code, failure, record, work_log = self._invoke_fake_json_event(event)
+
+        self.assertEqual(return_code, 0)
+        self.assertEqual(failure, "provider-connection")
+        self.assertEqual(record["status"], "failed")
+        self.assertNotIn("succeeded:0", work_log)
+
+    def test_stdout_terminal_error_survives_later_stderr_diagnostic(self):
+        event = {
+            "type": "agent_end",
+            "willRetry": False,
+            "messages": [
+                {
+                    "role": "assistant",
+                    "stopReason": "error",
+                    "errorMessage": "opaque terminal failure",
+                }
+            ],
+        }
+
+        return_code, failure, record, work_log = self._invoke_fake_json_event(
+            event,
+            stderr_line="cleanup warning",
+        )
+
+        self.assertEqual(return_code, 0)
+        self.assertEqual(failure, "generic-error")
+        self.assertEqual(record["failure_evidence_source"], "fake-json-runner:stdout")
+        self.assertNotIn("succeeded:0", work_log)
+
+    def test_required_agent_end_success_event_fails_closed_when_missing(self):
+        event = {"type": "message", "text": "partial output only"}
+
+        return_code, failure, record, work_log = self._invoke_fake_json_event(event)
+
+        self.assertEqual(return_code, 0)
+        self.assertEqual(failure, "generic-error")
+        self.assertEqual(record["failure_source"], "dispatcher-terminal-contract")
+        self.assertNotIn("succeeded:0", work_log)
+
+    def test_required_agent_end_success_event_accepts_final_stop(self):
+        event = {
+            "type": "agent_end",
+            "willRetry": False,
+            "messages": [{"role": "assistant", "stopReason": "stop"}],
+        }
+
+        return_code, failure, record, work_log = self._invoke_fake_json_event(event)
+
+        self.assertEqual(return_code, 0)
+        self.assertIsNone(failure)
+        self.assertEqual(record["status"], "succeeded")
+        self.assertIn("succeeded:0", work_log)
+
     def test_catalog_source_is_in_runtime_audit_evidence(self):
         with TemporaryDirectory() as tmp:
             root = Path(tmp)
@@ -271,6 +456,31 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
 
 
 class GenericDispatcherContractTests(unittest.TestCase):
+    def test_selfcheck_work_log_uses_worker_plan_artifact(self):
+        with TemporaryDirectory() as tmp:
+            directory = Path(tmp) / "agent-task" / "group" / "01_task"
+            directory.mkdir(parents=True)
+            plan = directory / "PLAN-local-G05.md"
+            review = directory / "CODE_REVIEW-cloud-G05.md"
+            plan.touch()
+            review.touch()
+            task = dispatch.Task(
+                name="group/01_task",
+                directory=directory,
+                plan=plan,
+                review=review,
+                user_review=None,
+                recovery=False,
+            )
+
+            worker = dispatch.work_log_task_name(task, "worker")
+            selfcheck = dispatch.work_log_task_name(task, "selfcheck")
+            official_review = dispatch.work_log_task_name(task, "review")
+
+        self.assertTrue(worker.endswith("/PLAN-local-G05.md"))
+        self.assertEqual(selfcheck, worker)
+        self.assertTrue(official_review.endswith("/CODE_REVIEW-cloud-G05.md"))
+
     def test_parallel_limit_contract(self):
         self.assertEqual(dispatch.validated_max_parallel(0), 0)
         self.assertEqual(dispatch.validated_max_parallel(3), 3)

agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_execution_target_policy.py

@@ -24,9 +24,17 @@ def catalog_value(*, windows: bool = False) -> dict:
             "selfcheck_required": True,
             "runtime": {
                 "command": ["runner-a", "--model", "{model}", "{prompt}"],
-                "resume_command": ["runner-a", "--resume", "{resume_session}", "{prompt}"],
+                "resume_command": [
+                    "runner-a",
+                    "--resume",
+                    "{resume_session}",
+                    "--session-dir",
+                    "{resume_session_dir}",
+                    "{prompt}",
+                ],
                 "output_format": "jsonl",
                 "native_session_monitor": True,
+                "session_path": "sessions/*{session_id}*.jsonl",
             },
         },
         "target-b": {
@@ -148,6 +156,37 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
             with self.assertRaisesRegex(policy.CatalogError, "executable must be a literal"):
                 policy.load_catalog(write_catalog(Path(tmp), value))
 
+    def test_native_session_monitor_requires_resume_command_and_session_path(self):
+        for missing_field in ("resume_command", "session_path"):
+            value = catalog_value()
+            del value["targets"]["target-a"]["runtime"][missing_field]
+            with (
+                self.subTest(missing_field=missing_field),
+                TemporaryDirectory() as tmp,
+                self.assertRaisesRegex(policy.CatalogError, missing_field),
+            ):
+                policy.load_catalog(write_catalog(Path(tmp), value))
+
+    def test_terminal_success_contract_requires_jsonl_agent_end(self):
+        valid = catalog_value()
+        valid["targets"]["target-a"]["runtime"]["terminal_success"] = "agent_end"
+        invalid_name = catalog_value()
+        invalid_name["targets"]["target-a"]["runtime"]["terminal_success"] = "message_end"
+        invalid_format = catalog_value()
+        invalid_format["targets"]["target-a"]["runtime"]["terminal_success"] = "agent_end"
+        invalid_format["targets"]["target-a"]["runtime"]["output_format"] = "text"
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            loaded = policy.load_catalog(write_catalog(root, valid))
+            self.assertEqual(
+                loaded.targets["target-a"].runtime["terminal_success"],
+                "agent_end",
+            )
+            with self.assertRaisesRegex(policy.CatalogError, "must be 'agent_end'"):
+                policy.load_catalog(write_catalog(root, invalid_name))
+            with self.assertRaisesRegex(policy.CatalogError, "requires output_format='jsonl'"):
+                policy.load_catalog(write_catalog(root, invalid_format))
+
     def test_catalog_revision_changes_with_content(self):
         with TemporaryDirectory() as tmp:
             root = Path(tmp)

agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_select_execution_target.py

@@ -164,6 +164,11 @@ class SelectorTests(unittest.TestCase):
         self.assertEqual((pi.agent, pi.model), ("pi", "ornith:35b"))
         self.assertTrue(pi.selfcheck_required)
         self.assertIn("--thinking", pi.runtime["command"])
+        self.assertTrue(pi.runtime["native_session_monitor"])
+        self.assertEqual(pi.runtime["terminal_success"], "agent_end")
+        self.assertIn("--session", pi.runtime["resume_command"])
+        self.assertIn("{resume_session_dir}", pi.runtime["resume_command"])
+        self.assertIn("{session_id}", pi.runtime["session_path"])
         agy = catalog.targets["agy-gemini-high"]
         self.assertEqual(agy.runtime["auxiliary_logs"], ["{attempt_dir}/agy-cli.log"])
         opencode = catalog.targets["opencode-glm-max"]

agent-ops/skills/common/prepare-milestone-workspace/scripts/prepare_workspace.py

@@ -426,17 +426,14 @@ def epic_cycle_script(workspace: Path) -> Path:
 
 
 def dispatcher_script(workspace: Path) -> Path:
-    skills_root = workspace / "agent-ops" / "skills"
-    project_root = skills_root / "project" / "orchestrate-agent-task-loop"
-    project_dispatcher = project_root / "scripts" / "dispatch.py"
-    if project_dispatcher.is_file():
-        private_root = skills_root / "private" / "orchestrate-agent-task-loop"
-        private_dispatcher = private_root / "scripts" / "dispatch.py"
-        if (private_root / "SKILL.md").is_file() and private_dispatcher.is_file():
-            return private_dispatcher
-        return project_dispatcher
     common_dispatcher = (
-        skills_root / "common" / "orchestrate-agent-task-loop" / "scripts" / "dispatch.py"
+        workspace
+        / "agent-ops"
+        / "skills"
+        / "common"
+        / "orchestrate-agent-task-loop"
+        / "scripts"
+        / "dispatch.py"
     )
     if not common_dispatcher.is_file():
         raise PreparationError(f"dispatcher script not found: {common_dispatcher}")
@@ -458,17 +455,7 @@ def dispatcher_command(
         "--task-group",
         task_group,
     ]
-    common_dispatcher = (
-        workspace
-        / "agent-ops"
-        / "skills"
-        / "common"
-        / "orchestrate-agent-task-loop"
-        / "scripts"
-        / "dispatch.py"
-    )
-    if dispatcher.resolve() == common_dispatcher.resolve():
-        command.extend(["--execution-catalog", execution_catalog])
+    command.extend(["--execution-catalog", execution_catalog])
     return command
 
 

agent-ops/skills/common/prepare-milestone-workspace/tests/test_prepare_workspace.py

@@ -51,7 +51,7 @@ class PrepareWorkspaceTest(unittest.TestCase):
             Path("/tmp/example/sample-feature-worktree"),
         )
 
-    def test_dispatcher_prefers_project_override_and_private_pair(self) -> None:
+    def test_dispatcher_uses_common_runtime_only(self) -> None:
         with tempfile.TemporaryDirectory() as raw:
             workspace = Path(raw)
             common = (
@@ -70,24 +70,15 @@ class PrepareWorkspaceTest(unittest.TestCase):
                 path.parent.mkdir(parents=True, exist_ok=True)
                 path.touch()
 
-            self.assertEqual(MODULE.dispatcher_script(workspace), project)
-
             (private_root / "SKILL.md").touch()
-            self.assertEqual(MODULE.dispatcher_script(workspace), private)
-
-            project.unlink()
             self.assertEqual(MODULE.dispatcher_script(workspace), common)
 
-    def test_dispatcher_command_injects_catalog_only_for_common_runtime(self) -> None:
+    def test_dispatcher_command_always_injects_catalog(self) -> None:
         workspace = Path("/repo")
         common = (
             workspace
             / "agent-ops/skills/common/orchestrate-agent-task-loop/scripts/dispatch.py"
         )
-        project = (
-            workspace
-            / "agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py"
-        )
 
         common_command = MODULE.dispatcher_command(
             workspace=workspace,
@@ -95,15 +86,7 @@ class PrepareWorkspaceTest(unittest.TestCase):
             task_group="m-sample",
             execution_catalog="/runtime/catalog.json",
         )
-        project_command = MODULE.dispatcher_command(
-            workspace=workspace,
-            dispatcher=project,
-            task_group="m-sample",
-            execution_catalog="/runtime/catalog.json",
-        )
-
         self.assertIn("--execution-catalog", common_command)
-        self.assertNotIn("--execution-catalog", project_command)
 
     def test_epic_document_range_is_one_based_and_inclusive(self) -> None:
         epics = MODULE.parse_epics(