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

agent-ops/.version

@@ -1 +1 @@
-1.1.202
+1.1.203

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

@@ -50,10 +50,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`, `terminal_success`, and `auxiliary_logs`;
+- optional `runtime.resume_command`, `preflight_command`, `environment`, `session_path`, `native_session_monitor`, `session_stall_resume`, `terminal_success`, and `auxiliary_logs`;
 - optional `runtime.output_format`: `text` or `jsonl`.
 
-Command templates may use only `{agent}`, `{model}`, `{reasoning_effort}`, `{target_id}`, `{workspace}`, `{attempt_dir}`, `{session_id}`, `{resume_session}`, `{resume_session_dir}`, and `{prompt}`. A target with `reasoning_effort` must use `{reasoning_effort}` in its command and resume command when present; a target without the field cannot use that placeholder. `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`. `terminal_success=turn_completed` requires JSONL output and accepts only final `turn.completed`; `turn.failed` or a missing terminal event fails closed. When either declared success event is observed with exit 0, earlier recovered transport diagnostics do not turn the attempt into a failure. 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}`, `{reasoning_effort}`, `{target_id}`, `{workspace}`, `{attempt_dir}`, `{session_id}`, `{resume_session}`, `{resume_session_dir}`, and `{prompt}`. A target with `reasoning_effort` must use `{reasoning_effort}` in its command and resume command when present; a target without the field cannot use that placeholder. `native_session_monitor=true` requires both `resume_command` and `session_path`. `session_stall_resume=true` requires `resume_command` and JSONL output whose runtime session id can be persisted. `terminal_success=agent_end` requires JSONL output and accepts only a non-retrying final `agent_end` whose last assistant message has `stopReason=stop`. `terminal_success=turn_completed` requires JSONL output and accepts only final `turn.completed`; `turn.failed` or a missing terminal event fails closed. When either declared success event is observed with exit 0, earlier recovered transport diagnostics do not turn the attempt into a failure. 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.
 
@@ -113,7 +113,7 @@ Accept self-check completion only when `## Implementation Checklist` or its supp
 - Determine liveness from PID/start-token/process-marker evidence and actual stream or native-session progress. Heartbeat mtime is never agent progress. For Codex JSONL, an unmatched `item.started` `command_execution` is an active tool interval: suspend the model-response silence timer until its matching `item.completed`, then restore normal stall detection.
 - The dispatcher model-silence safety net is 70 seconds. Downstream provider runtimes should emit their bounded terminal before that deadline; do not extend the dispatcher budget per target to cover nested retries.
 - Treat a confirmed provider transport terminal as the end of the current dispatch. Do not resume or automatically resend the same native session; an operator may start a fresh dispatch after the provider/runtime state is corrected.
-- Retry `session-stall` only with a fresh native conversation. Preserve workspace changes and logical locator evidence, but do not carry the silent conversation context into the next attempt or a restarted dispatcher.
+- When the selected target declares `session_stall_resume=true` and its JSONL emitted a runtime session id, terminate the silent process and invoke the catalog `resume_command` once for that exact same target and session with a continuation message. Do not inject a second continuation into the same stalled session; return to the existing bounded fresh-conversation retry and failover route. If the capability or runtime session id is absent, preserve workspace changes and logical locator evidence but retry with a fresh conversation. Never apply same-session continuation to provider transport terminals.
 - 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.
 - Preserve failed attempt logs. Delete successful attempt logs only after verified archive completion and no live evidence.

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

@@ -73,7 +73,26 @@
           "--auto",
           "{prompt}"
         ],
+        "resume_command": [
+          "opencode",
+          "run",
+          "--format",
+          "json",
+          "--dir",
+          "{workspace}",
+          "--session",
+          "{resume_session}",
+          "--agent",
+          "build",
+          "--model",
+          "iop-glm/glm-5.2",
+          "--variant",
+          "{reasoning_effort}",
+          "--auto",
+          "{prompt}"
+        ],
         "output_format": "jsonl",
+        "session_stall_resume": true,
         "environment": {
           "TMPDIR": "/tmp"
         }
@@ -102,7 +121,26 @@
           "--auto",
           "{prompt}"
         ],
+        "resume_command": [
+          "opencode",
+          "run",
+          "--format",
+          "json",
+          "--dir",
+          "{workspace}",
+          "--session",
+          "{resume_session}",
+          "--agent",
+          "build",
+          "--model",
+          "iop-glm/glm-5.2",
+          "--variant",
+          "{reasoning_effort}",
+          "--auto",
+          "{prompt}"
+        ],
         "output_format": "jsonl",
+        "session_stall_resume": true,
         "environment": {
           "TMPDIR": "/tmp"
         }
@@ -131,7 +169,26 @@
           "--auto",
           "{prompt}"
         ],
+        "resume_command": [
+          "opencode",
+          "run",
+          "--format",
+          "json",
+          "--dir",
+          "{workspace}",
+          "--session",
+          "{resume_session}",
+          "--agent",
+          "build",
+          "--model",
+          "iop-glm/glm-5.2",
+          "--variant",
+          "{reasoning_effort}",
+          "--auto",
+          "{prompt}"
+        ],
         "output_format": "jsonl",
+        "session_stall_resume": true,
         "environment": {
           "TMPDIR": "/tmp"
         }

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

@@ -2879,7 +2879,11 @@ def render_json_line(cli: str, line: str) -> tuple[list[str], str | None]:
         return [line.rstrip()], None
     if not isinstance(value, dict):
         return [line.rstrip()], None
-    session_id = value.get("thread_id") or value.get("session_id")
+    session_id = (
+        value.get("thread_id")
+        or value.get("session_id")
+        or value.get("sessionID")
+    )
     rendered: list[str] = []
     for field in ("text", "result", "message", "output"):
         item = value.get(field)
@@ -2896,6 +2900,37 @@ def render_json_line(cli: str, line: str) -> tuple[list[str], str | None]:
     return rendered, str(session_id) if session_id else None
 
 
+def stall_resume_session_id(
+    spec: AgentSpec,
+    locator: Path | None,
+    record: dict[str, Any] | None = None,
+) -> str | None:
+    if not spec.runtime.get("session_stall_resume") or locator is None:
+        return None
+    locator_path = locator if locator.name == "locator.json" else locator / "locator.json"
+    if record is None:
+        try:
+            loaded = json.loads(locator_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            return None
+        record = loaded if isinstance(loaded, dict) else None
+    if not isinstance(record, dict):
+        return None
+    if (
+        record.get("target_id") != spec.target_id
+        or record.get("cli") != spec.cli
+        or record.get("model") != spec.model
+        or record.get("status") != "failed"
+        or record.get("failure_class") != "session-stall"
+        or record.get("resume_mode") == "runtime-session-id"
+        or record.get("session_id_source") != "runtime-stream"
+        or not isinstance(record.get("session_id"), str)
+        or not record.get("session_id")
+    ):
+        return None
+    return str(record["session_id"])
+
+
 def native_session_path(
     spec: AgentSpec,
     workspace: Path,
@@ -3250,9 +3285,21 @@ def native_resume_locator(
     )
     if not owned:
         return None
+    runtime = record.get("runtime")
     if (
-        not isinstance(record.get("runtime"), dict)
+        isinstance(runtime, dict)
-        or not record["runtime"].get("native_session_monitor")
+        and runtime.get("session_stall_resume")
+        and record.get("status") == "failed"
+        and record.get("failure_class") == "session-stall"
+        and record.get("resume_mode") != "runtime-session-id"
+        and record.get("session_id_source") == "runtime-stream"
+        and isinstance(record.get("session_id"), str)
+        and record.get("session_id")
+    ):
+        return locator
+    if (
+        not isinstance(runtime, dict)
+        or not runtime.get("native_session_monitor")
         or record.get("failure_class") != "context-limit"
         or record.get("status") != "failed"
     ):
@@ -3334,11 +3381,12 @@ def build_command(
     workspace: Path,
     session_id: str,
     attempt_dir: Path,
-    native_resume_session: Path | None = None,
+    resume_session: str | Path | None = None,
+    resume_session_dir: Path | None = None,
 ) -> list[str]:
     template_name = (
         "resume_command"
-        if native_resume_session is not None and spec.runtime.get("resume_command")
+        if resume_session is not None and spec.runtime.get("resume_command")
         else "command"
     )
     template = spec.runtime.get(template_name)
@@ -3352,10 +3400,10 @@ def build_command(
         "model": spec.model,
         "prompt": prompt,
         "reasoning_effort": str(spec.reasoning_effort or ""),
-        "resume_session": str(native_resume_session or ""),
+        "resume_session": str(resume_session or ""),
         "resume_session_dir": (
-            str(native_resume_session.parent)
+            str(resume_session_dir)
-            if native_resume_session is not None
+            if resume_session_dir is not None
             else ""
         ),
         "session_id": session_id,
@@ -3448,9 +3496,15 @@ async def invoke(
     normalized_output_path.touch()
     heartbeat_path.touch()
     session_id = str(uuid.uuid4())
+    session_id_source = "dispatcher-generated"
     process_marker = f"w{store.workspace_id}__{identity}__{uuid.uuid4()}"
     native_resume_session: Path | None = None
-    if spec.native_resume and resume_locator and resume_locator.is_file():
+    stall_resume_session: str | None = None
+    if (
+        (spec.native_resume or spec.runtime.get("session_stall_resume"))
+        and resume_locator
+        and resume_locator.is_file()
+    ):
         resume_locator_path = (
             resume_locator
             if resume_locator.name == "locator.json"
@@ -3478,6 +3532,13 @@ async def invoke(
             expected_runs_root=store.runs,
         )
         if owned and isinstance(prior, dict):
+            stall_resume_session = stall_resume_session_id(
+                spec, resume_locator_path, prior
+            )
+            if stall_resume_session is not None:
+                resume_locator = resume_locator_path
+                session_id = stall_resume_session
+                session_id_source = "runtime-stream"
             prior_native = prior.get("native_session_path")
             candidate = Path(str(prior_native)) if prior_native else None
             if candidate and candidate.is_dir():
@@ -3496,6 +3557,15 @@ async def invoke(
                 native_resume_session = candidate
                 resume_locator = resume_locator_path
                 session_id = str(prior.get("session_id") or candidate.stem)
+                session_id_source = str(
+                    prior.get("session_id_source") or "native-session"
+                )
+    effective_resume_session = native_resume_session or stall_resume_session
+    effective_resume_session_dir = (
+        native_resume_session.parent
+        if native_resume_session is not None
+        else None
+    )
     started_at = now_iso()
     work_log_path = milestone_work_log_path(task)
     record: dict[str, Any] = {
@@ -3521,6 +3591,7 @@ async def invoke(
         "plan_path": str(task.plan) if task.plan else None,
         "review_path": str(task.review) if task.review else None,
         "session_id": session_id,
+        "session_id_source": session_id_source,
         "native_session_path": (
             str(native_resume_session)
             if native_resume_session is not None
@@ -3538,10 +3609,10 @@ async def invoke(
                     "model": spec.model,
                     "prompt": "",
                     "reasoning_effort": str(spec.reasoning_effort or ""),
-                    "resume_session": str(native_resume_session or ""),
+                    "resume_session": str(effective_resume_session or ""),
                     "resume_session_dir": (
-                        str(native_resume_session.parent)
+                        str(effective_resume_session_dir)
-                        if native_resume_session is not None
+                        if effective_resume_session_dir is not None
                         else ""
                     ),
                     "session_id": session_id,
@@ -3554,7 +3625,18 @@ async def invoke(
         "work_log": str(work_log_path.resolve()),
         "started_at": started_at,
         "status": "running",
-        "resumed_from_locator": str(resume_locator) if native_resume_session else None,
+        "resumed_from_locator": (
+            str(resume_locator)
+            if native_resume_session is not None or stall_resume_session is not None
+            else None
+        ),
+        "resume_mode": (
+            "native-session-file"
+            if native_resume_session is not None
+            else "runtime-session-id"
+            if stall_resume_session is not None
+            else None
+        ),
     }
     stage_decision = None
     if isinstance(store, StateStore):
@@ -3639,7 +3721,8 @@ async def invoke(
         workspace,
         session_id,
         attempt_dir,
-        native_resume_session=native_resume_session,
+        resume_session=effective_resume_session,
+        resume_session_dir=effective_resume_session_dir,
     )
     diagnostics: list[str] = []
     diagnostic_origins: list[str] = []
@@ -3654,10 +3737,10 @@ async def invoke(
             "model": spec.model,
             "prompt": prompt,
             "reasoning_effort": str(spec.reasoning_effort or ""),
-            "resume_session": str(native_resume_session or ""),
+            "resume_session": str(effective_resume_session or ""),
             "resume_session_dir": (
-                str(native_resume_session.parent)
+                str(effective_resume_session_dir)
-                if native_resume_session is not None
+                if effective_resume_session_dir is not None
                 else ""
             ),
             "session_id": session_id,
@@ -3945,6 +4028,7 @@ async def invoke(
                 )
                 if discovered and record.get("session_id") != discovered:
                     record["session_id"] = discovered
+                    record["session_id_source"] = "runtime-stream"
                     if native_resume_session is None:
                         record["native_session_path"] = native_session_path(
                             spec, workspace, discovered, attempt_dir
@@ -4315,6 +4399,11 @@ def continuation_prompt(
     context: dict[str, Any] | None = None,
     unchecked_items: bool = False,
 ) -> str:
+    if resume_same_native_session and role != "selfcheck":
+        return dispatcher_child_prompt(
+            f"{REPOSITORY_LANGUAGE_PROMPT} Continue this session and complete "
+            "the current task."
+        )
     if native_resume and role == "selfcheck":
         if resume_same_native_session:
             if unchecked_items:
@@ -4728,6 +4817,10 @@ async def run_escalating(
                 ],
             )
             previous_locator = locator
+            if failure == "session-stall" and stall_resume_session_id(
+                spec, locator
+            ) is not None:
+                native_resume_locator = locator
             await asyncio.sleep(min(30, 2 ** min(generic_retries, 5)))
             continue
         if failure not in RECOVERABLE_RUNTIME_FAILURES:

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

@@ -109,6 +109,7 @@ def _validate_runtime(value: object, label: str) -> dict[str, Any]:
         "output_format",
         "session_path",
         "native_session_monitor",
+        "session_stall_resume",
         "terminal_success",
         "auxiliary_logs",
     }
@@ -176,6 +177,19 @@ def _validate_runtime(value: object, label: str) -> dict[str, Any]:
             raise CatalogError(
                 f"{label}.native_session_monitor requires {missing}"
             )
+    stall_resume = value.get("session_stall_resume", False)
+    if not isinstance(stall_resume, bool):
+        raise CatalogError(f"{label}.session_stall_resume must be a boolean")
+    runtime["session_stall_resume"] = stall_resume
+    if stall_resume:
+        if "resume_command" not in runtime:
+            raise CatalogError(
+                f"{label}.session_stall_resume requires resume_command"
+            )
+        if runtime["output_format"] != "jsonl":
+            raise CatalogError(
+                f"{label}.session_stall_resume requires output_format='jsonl'"
+            )
     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

@@ -175,7 +175,8 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
             Path("/workspace"),
             "session-1",
             Path("/attempt"),
-            native_resume_session=Path("/attempt/session.jsonl"),
+            resume_session=Path("/attempt/session.jsonl"),
+            resume_session_dir=Path("/attempt"),
         )
         self.assertEqual(command, ["runner", "/workspace", "opaque-model", "high", "session-1", "/attempt", "do work"])
         self.assertEqual(
@@ -848,6 +849,95 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
         self.assertEqual(record["active_command_execution_ids"], [])
         self.assertIn("stream_silence_inspection", record)
 
+    def test_session_stall_resume_reuses_runtime_stream_session_id(self):
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            plan = write_plan(root)
+            task = task_from_plan(root, plan)
+            runner = root / "stall_resume_runner.py"
+            capture = root / "resume-argv.json"
+            runner.write_text(
+                "import json, sys, time\n"
+                "args = sys.argv[1:]\n"
+                "if args and args[0] == 'resume':\n"
+                "    capture = args[1]\n"
+                "    open(capture, 'w', encoding='utf-8').write(json.dumps(args[2:]))\n"
+                "    print(json.dumps({'type': 'message', 'sessionID': args[2], 'text': 'continued'}), flush=True)\n"
+                "else:\n"
+                "    print(json.dumps({'type': 'step_start', 'sessionID': 'ses-runtime-1'}), flush=True)\n"
+                "    time.sleep(30)\n",
+                encoding="utf-8",
+            )
+            agent = dispatch.AgentSpec(
+                "opencode",
+                "glm-5.2",
+                "opencode/glm-5.2",
+                target_id="opencode-test",
+                runtime={
+                    "command": [sys.executable, str(runner)],
+                    "resume_command": [
+                        sys.executable,
+                        str(runner),
+                        "resume",
+                        str(capture),
+                        "{resume_session}",
+                        "{prompt}",
+                    ],
+                    "output_format": "jsonl",
+                    "session_stall_resume": True,
+                },
+            )
+            with (
+                mock.patch.dict(os.environ, {"XDG_STATE_HOME": str(root / "state")}),
+                mock.patch.object(dispatch, "STREAM_HEARTBEAT_SECONDS", 0.01),
+                mock.patch.object(dispatch, "MODEL_RESPONSE_STALL_SECONDS", 0.05),
+            ):
+                store = dispatch.StateStore(root)
+                try:
+                    first_rc, first_failure, first_locator = asyncio.run(
+                        dispatch.invoke(root, store, task, "worker", agent, "start")
+                    )
+                    first_record = json.loads(first_locator.read_text(encoding="utf-8"))
+                    self.assertEqual(
+                        dispatch.stall_resume_session_id(agent, first_locator),
+                        "ses-runtime-1",
+                    )
+                    second_rc, second_failure, second_locator = asyncio.run(
+                        dispatch.invoke(
+                            root,
+                            store,
+                            task,
+                            "worker",
+                            agent,
+                            dispatch.continuation_prompt(
+                                task,
+                                "worker",
+                                first_locator,
+                                resume_same_native_session=True,
+                            ),
+                            resume_locator=first_locator,
+                        )
+                    )
+                    second_record = json.loads(second_locator.read_text(encoding="utf-8"))
+                    resumed_args = json.loads(capture.read_text(encoding="utf-8"))
+                finally:
+                    store.close()
+
+        self.assertLess(first_rc, 0)
+        self.assertEqual(first_failure, "session-stall")
+        self.assertEqual(first_record["session_id"], "ses-runtime-1")
+        self.assertEqual(first_record["session_id_source"], "runtime-stream")
+        self.assertEqual(second_rc, 0)
+        self.assertIsNone(second_failure)
+        self.assertEqual(second_record["session_id"], "ses-runtime-1")
+        self.assertEqual(second_record["resume_mode"], "runtime-session-id")
+        self.assertEqual(resumed_args[0], "ses-runtime-1")
+        self.assertIn("Continue this session", resumed_args[1])
+        second_record.update(status="failed", failure_class="session-stall")
+        self.assertIsNone(
+            dispatch.stall_resume_session_id(agent, second_locator, second_record)
+        )
+
     def test_catalog_source_is_in_runtime_audit_evidence(self):
         with TemporaryDirectory() as tmp:
             root = Path(tmp)

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

@@ -130,7 +130,8 @@ class ObservationInvokeIntegrationTest(unittest.IsolatedAsyncioTestCase):
                 cwd,
                 actual_session_id,
                 attempt_dir,
-                native_resume_session=None,
+                resume_session=None,
+                resume_session_dir=None,
             ):
                 self.assertEqual(actual_session_id, session_id)
                 native = attempt_dir / "native-sessions" / f"session_{session_id}.jsonl"

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

@@ -207,6 +207,25 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
             ):
                 policy.load_catalog(write_catalog(Path(tmp), value))
 
+    def test_session_stall_resume_requires_resume_command_and_jsonl(self):
+        missing_command = catalog_value()
+        missing_command["targets"]["target-b"]["runtime"][
+            "session_stall_resume"
+        ] = True
+        text_output = catalog_value()
+        text_output["targets"]["target-b"]["runtime"].update(
+            session_stall_resume=True,
+            resume_command=["runner-b", "--session", "{resume_session}", "{prompt}"],
+        )
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            with self.assertRaisesRegex(
+                policy.CatalogError, "requires resume_command"
+            ):
+                policy.load_catalog(write_catalog(root, missing_command))
+            with self.assertRaisesRegex(policy.CatalogError, "requires output_format"):
+                policy.load_catalog(write_catalog(root, text_output))
+
     def test_terminal_success_contract_requires_jsonl_agent_end(self):
         valid = catalog_value()
         valid["targets"]["target-a"]["runtime"]["terminal_success"] = "agent_end"

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

@@ -187,10 +187,14 @@ class SelectorTests(unittest.TestCase):
             "opencode-glm-max",
         ):
             with self.subTest(target_id=target_id):
+                runtime = catalog.targets[target_id].runtime
                 self.assertEqual(
-                    catalog.targets[target_id].runtime["environment"]["TMPDIR"],
+                    runtime["environment"]["TMPDIR"],
                     "/tmp",
                 )
+                self.assertTrue(runtime["session_stall_resume"])
+                self.assertIn("--session", runtime["resume_command"])
+                self.assertIn("{resume_session}", runtime["resume_command"])
         terra = catalog.targets["codex-terra-high"]
         self.assertEqual(terra.reasoning_effort, "high")
         self.assertIn(