diff --git a/agent-ops/.version b/agent-ops/.version index 837aff75..b1312453 100644 --- a/agent-ops/.version +++ b/agent-ops/.version @@ -1 +1 @@ -1.1.202 +1.1.203 diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/SKILL.md b/agent-ops/skills/common/orchestrate-agent-task-loop/SKILL.md index 45b4aadb..ebe2b257 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/SKILL.md +++ b/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. diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json b/agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json index fa4226b9..faec9cf2 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/assets/default-execution-catalog.json +++ b/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" } 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 8477be1f..11e06fa8 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 @@ -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) - or not record["runtime"].get("native_session_monitor") + isinstance(runtime, dict) + 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) - if native_resume_session is not None + str(resume_session_dir) + 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) - if native_resume_session is not None + str(effective_resume_session_dir) + 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) - if native_resume_session is not None + str(effective_resume_session_dir) + 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: diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/execution_target_policy.py b/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/execution_target_policy.py index 3cb8158b..d10b6437 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/scripts/execution_target_policy.py +++ b/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 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 21c86cb5..e6751408 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 @@ -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) diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatcher_observation.py b/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatcher_observation.py index 68861c26..8cc5d2dc 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_dispatcher_observation.py +++ b/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" diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_execution_target_policy.py b/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_execution_target_policy.py index 28b92e3a..1916c4c7 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_execution_target_policy.py +++ b/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" diff --git a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_select_execution_target.py b/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_select_execution_target.py index 2ac2350c..55b60fab 100644 --- a/agent-ops/skills/common/orchestrate-agent-task-loop/tests/test_select_execution_target.py +++ b/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(