fix(agent-ops): 타깃별 응답 정지 예산을 적용한다

긴 로컬 모델 응답이 하위 복구 시간보다 먼저 종료되지 않도록 실행 타깃이 무응답 예산을 소유하게 한다.

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

@@ -50,11 +50,13 @@ 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`, `model_response_stall_seconds`, `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.
 
+`runtime.model_response_stall_seconds` is a positive integer target-owned silence budget. Omission keeps the dispatcher default of 180 seconds. A runtime that can remain externally silent while performing bounded downstream recovery must set this value above its complete silent recovery window, including the initial attempt, every internal retry, and bounded cleanup overhead. Record the resolved value in each locator and never infer it from provider or model names in dispatcher code.
+
 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.
 
 The bundled review routes vary model and reasoning effort by routed grade instead of fixing every review to one target: G01-G04 use Terra/high, G05-G08 use Sol/high, and G09-G10 use Sol/xhigh. Runtime or project catalog overrides may replace this default tiering.

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

@@ -47,6 +47,7 @@
         "output_format": "jsonl",
         "session_path": "{attempt_dir}/pi-sessions/*{session_id}*.jsonl",
         "native_session_monitor": true,
+        "model_response_stall_seconds": 1500,
         "terminal_success": "agent_end"
       }
     },

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

@@ -511,6 +511,19 @@ class AgentSpec:
     reasoning_effort: str | None = None
     runtime: dict[str, Any] = field(default_factory=dict)
 
+
+def model_response_stall_seconds(spec: AgentSpec) -> float:
+    """Resolve the catalog-owned silence budget for one selected target."""
+    value = spec.runtime.get("model_response_stall_seconds")
+    if value is None:
+        return float(MODEL_RESPONSE_STALL_SECONDS)
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        raise ExecutionDecisionError(
+            "model_response_stall_seconds must be a positive integer"
+        )
+    return float(value)
+
+
 def agent_spec_from_record(record: dict[str, Any]) -> AgentSpec | None:
     cli = str(record.get("cli") or "")
     model = str(record.get("model") or "")
@@ -3433,6 +3446,7 @@ async def invoke(
     prompt: str,
     resume_locator: Path | None = None,
 ) -> tuple[int, str | None, Path]:
+    response_stall_seconds = model_response_stall_seconds(spec)
     worker_signature_before = (
         task_signature(workspace, task) if role == "worker" else None
     )
@@ -3516,6 +3530,7 @@ async def invoke(
         "selfcheck_required": spec.selfcheck_required,
         "reasoning_effort": spec.reasoning_effort,
         "runtime": spec.runtime,
+        "model_response_stall_timeout_seconds": response_stall_seconds,
         "agent_process_marker": process_marker,
         "plan_path": str(task.plan) if task.plan else None,
         "review_path": str(task.review) if task.review else None,
@@ -3819,13 +3834,13 @@ async def invoke(
                         record["native_pending_tool_call_ids"] = list(
                             native_state.pending_tool_call_ids
                         )
-                        record["native_stall_timeout_seconds"] = None
+                        record["native_stall_timeout_seconds"] = response_stall_seconds
                         record.setdefault("native_activity_state", "starting")
                     if (
                         spec.native_resume
                         and not is_native_tool_execution
                         and not is_command_execution
-                        and native_inactive_seconds >= MODEL_RESPONSE_STALL_SECONDS
+                        and native_inactive_seconds >= response_stall_seconds
                         and session_stall_seconds is None
                     ):
                         session_stall_seconds = native_inactive_seconds
@@ -3852,7 +3867,7 @@ async def invoke(
                         not spec.native_resume
                         and not is_command_execution
                         and non_native_inactive_seconds
-                        >= MODEL_RESPONSE_STALL_SECONDS
+                        >= response_stall_seconds
                         and session_stall_seconds is None
                     ):
                         session_stall_seconds = non_native_inactive_seconds

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

@@ -22,6 +22,7 @@ VALID_STAGES = {"worker", "review"}
 VALID_LANES = {"local", "cloud"}
 VALID_EXECUTION_CLASSES = {"local_model", "cloud_model"}
 VALID_OUTPUT_FORMATS = {"jsonl", "text"}
+MAX_MODEL_RESPONSE_STALL_SECONDS = 24 * 60 * 60
 ALLOWED_TEMPLATE_FIELDS = {
     "agent",
     "attempt_dir",
@@ -109,6 +110,7 @@ def _validate_runtime(value: object, label: str) -> dict[str, Any]:
         "output_format",
         "session_path",
         "native_session_monitor",
+        "model_response_stall_seconds",
         "terminal_success",
         "auxiliary_logs",
     }
@@ -176,6 +178,19 @@ def _validate_runtime(value: object, label: str) -> dict[str, Any]:
             raise CatalogError(
                 f"{label}.native_session_monitor requires {missing}"
             )
+    model_response_stall_seconds = value.get("model_response_stall_seconds")
+    if model_response_stall_seconds is not None:
+        if (
+            isinstance(model_response_stall_seconds, bool)
+            or not isinstance(model_response_stall_seconds, int)
+            or model_response_stall_seconds <= 0
+            or model_response_stall_seconds > MAX_MODEL_RESPONSE_STALL_SECONDS
+        ):
+            raise CatalogError(
+                f"{label}.model_response_stall_seconds must be an integer "
+                f"between 1 and {MAX_MODEL_RESPONSE_STALL_SECONDS}"
+            )
+        runtime["model_response_stall_seconds"] = model_response_stall_seconds
     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

@@ -754,6 +754,67 @@ class RuntimeCatalogDispatcherTests(unittest.TestCase):
         self.assertGreaterEqual(record["session_stall_seconds"], 0.05)
         self.assertIn("native_silence_inspection", record)
 
+    def test_target_silence_budget_survives_default_window_after_turn_start(self):
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            plan = write_plan(root)
+            task = task_from_plan(root, plan)
+            runner = root / "bounded_silent_runner.py"
+            runner.write_text(
+                "import json, time\n"
+                "print(json.dumps({'type': 'turn_start'}), flush=True)\n"
+                "time.sleep(0.15)\n"
+                "print(json.dumps({'type': 'agent_end', 'willRetry': False, "
+                "'messages': [{'role': 'assistant', 'stopReason': 'stop'}]}), "
+                "flush=True)\n",
+                encoding="utf-8",
+            )
+            agent = dispatch.AgentSpec(
+                "bounded-silent-runner",
+                "bounded-model",
+                "bounded-silent-runner/bounded-model",
+                native_resume=True,
+                target_id="bounded-target",
+                runtime={
+                    "command": [sys.executable, str(runner)],
+                    "output_format": "jsonl",
+                    "native_session_monitor": True,
+                    "session_path": "sessions/{session_id}.jsonl",
+                    "terminal_success": "agent_end",
+                    "model_response_stall_seconds": 1,
+                },
+            )
+            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:
+                    return_code, failure, locator = asyncio.run(
+                        dispatch.invoke(
+                            root,
+                            store,
+                            task,
+                            "worker",
+                            agent,
+                            "fake prompt",
+                        )
+                    )
+                    record = json.loads(locator.read_text(encoding="utf-8"))
+                finally:
+                    store.close()
+
+        self.assertEqual(return_code, 0)
+        self.assertIsNone(failure)
+        self.assertEqual(record["status"], "succeeded")
+        self.assertEqual(record["model_response_stall_timeout_seconds"], 1.0)
+        self.assertEqual(record["native_stall_timeout_seconds"], 1.0)
+        self.assertNotIn("session_stall_seconds", record)
+
     def test_non_native_active_command_execution_is_not_classified_as_stall(self):
         with TemporaryDirectory() as tmp:
             root = Path(tmp)

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

@@ -207,6 +207,33 @@ class ExecutionTargetPolicyTests(unittest.TestCase):
             ):
                 policy.load_catalog(write_catalog(Path(tmp), value))
 
+    def test_model_response_stall_budget_is_bounded_positive_integer(self):
+        valid = catalog_value()
+        valid["targets"]["target-a"]["runtime"][
+            "model_response_stall_seconds"
+        ] = 1200
+        invalid_values = (True, 0, -1, 1.5, 24 * 60 * 60 + 1)
+        with TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            loaded = policy.load_catalog(write_catalog(root, valid))
+            self.assertEqual(
+                loaded.targets["target-a"].runtime[
+                    "model_response_stall_seconds"
+                ],
+                1200,
+            )
+            for invalid in invalid_values:
+                with self.subTest(invalid=invalid):
+                    value = catalog_value()
+                    value["targets"]["target-a"]["runtime"][
+                        "model_response_stall_seconds"
+                    ] = invalid
+                    with self.assertRaisesRegex(
+                        policy.CatalogError,
+                        "model_response_stall_seconds must be an integer",
+                    ):
+                        policy.load_catalog(write_catalog(root, value))
+
     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

@@ -172,6 +172,7 @@ class SelectorTests(unittest.TestCase):
         self.assertIn("--thinking", pi.runtime["command"])
         self.assertIn("{reasoning_effort}", pi.runtime["command"])
         self.assertTrue(pi.runtime["native_session_monitor"])
+        self.assertEqual(pi.runtime["model_response_stall_seconds"], 1500)
         self.assertEqual(pi.runtime["terminal_success"], "agent_end")
         self.assertIn("--session", pi.runtime["resume_command"])
         self.assertIn("{resume_session_dir}", pi.runtime["resume_command"])