fix(core): a parked agent keeps its pause reason and drains queued messages on resume

BudgetPausedError carries whether the operator pause caused it, so a resume that lands between the hook's refusal and the loop's handler still parks and continues instead of exiting under the stop policy. After the resume the agent drains its mailbox into the session before retrying the call, so a message queued while it was held is not skipped.

strix/core/agents.py

@@ -75,6 +75,7 @@ class AgentCoordinator:
         self._budget_stopped = False
         self._reserve_stopped = False
         self._budget_paused = False
+        self._operator_paused = False
         self._resume_epoch = 0
         self._budget_policy: BudgetPolicy = "stop"
         self._extend_budget: Callable[[], None] | None = None
@@ -105,6 +106,10 @@ class AgentCoordinator:
     def budget_paused(self) -> bool:
         return self._budget_paused
 
+    @property
+    def operator_paused(self) -> bool:
+        return self._operator_paused
+
     @property
     def resume_epoch(self) -> int:
         """Bumped by every ``resume_budget``; an agent parks against the value it read."""
@@ -150,6 +155,7 @@ class AgentCoordinator:
         """
         async with self._lock:
             self._budget_paused = True
+            self._operator_paused = True
         logger.info("scan paused by the operator")
         await self._maybe_snapshot()
 
@@ -165,6 +171,7 @@ class AgentCoordinator:
             self._set_budget_limit(max_budget_usd)
         async with self._lock:
             self._budget_paused = False
+            self._operator_paused = False
             self._resume_epoch += 1
             woken = [aid for aid, status in self.statuses.items() if status == "budget_paused"]
             for aid in woken:
@@ -232,7 +239,7 @@ class AgentCoordinator:
             self._budget_stopped = budget_stopped
             self._reserve_stopped = reserve_stopped
             if not budget_paused:
-                self._budget_paused = False
+                self._budget_paused = self._operator_paused
                 for aid, status in self.statuses.items():
                     if status == "budget_paused":
                         self.statuses[aid] = "waiting"
@@ -424,7 +431,12 @@ class AgentCoordinator:
         unknown, or it is terminal and its loop does not park for wake-ups.
         """
         from_user = message.get("from") == "user"
-        if from_user and self._budget_paused and self._budget_policy != "pause":
+        if (
+            from_user
+            and self._budget_paused
+            and not self._operator_paused
+            and self._budget_policy != "pause"
+        ):
             await self.resume_from_budget_pause(exclude=target_agent_id)
         async with self._lock:
             if target_agent_id not in self.statuses:
@@ -447,7 +459,9 @@ class AgentCoordinator:
                 self.recovery_counts.pop(target_agent_id, None)
                 self.idle_resume_counts.pop(target_agent_id, None)
                 self._parent_notified.discard(target_agent_id)
-                self.statuses[target_agent_id] = "waiting"
+                held = self._operator_paused and self.statuses[target_agent_id] == "budget_paused"
+                if not held:
+                    self.statuses[target_agent_id] = "waiting"
             runtime.wake.set()
             stream = runtime.stream
             interrupt_on_message = runtime.interrupt_on_message
@@ -640,6 +654,7 @@ class AgentCoordinator:
                 "budget_stopped": self._budget_stopped,
                 "reserve_stopped": self._reserve_stopped,
                 "budget_paused": self._budget_paused,
+                "operator_paused": self._operator_paused,
             }
 
     async def restore(self, snap: dict[str, Any]) -> None:
@@ -662,6 +677,7 @@ class AgentCoordinator:
             self._budget_stopped = bool(snap.get("budget_stopped", False))
             self._reserve_stopped = bool(snap.get("reserve_stopped", False))
             self._budget_paused = bool(snap.get("budget_paused", False))
+            self._operator_paused = bool(snap.get("operator_paused", False))
             for aid in self.statuses:
                 self.runtimes.setdefault(aid, AgentRuntime())
 

strix/core/execution.py

@@ -599,7 +599,7 @@ async def _run_until_lifecycle(
                     hooks=hooks,
                 )
         except BudgetPausedError as exc:
-            if coordinator.budget_policy != "pause":
+            if coordinator.budget_policy != "pause" and not exc.operator_pause:
                 raise
             # The agent parked right before an LLM call; everything up to that
             # point is already in its session. Once resumed, the same call goes
@@ -612,6 +612,7 @@ async def _run_until_lifecycle(
                 # Stopped while parked (operator stop or a parent's stop_agent).
                 await coordinator.reset_recovery(agent_id)
                 return result
+            await coordinator.consume_pending(agent_id)
             if session is not None:
                 input_data = []
             continue
@@ -831,7 +832,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                 await coordinator.detach_stream(agent_id, stream)
         except BudgetPausedError as exc:
             logger.info("agent %s paused at the scan budget limit: %s", agent_id, exc)
-            if coordinator.budget_policy == "pause":
+            if coordinator.budget_policy == "pause" or exc.operator_pause:
                 await coordinator.park_for_budget(agent_id)
             else:
                 await coordinator.pause_for_budget(agent_id)

strix/core/hooks.py

@@ -65,9 +65,12 @@ class BudgetPausedError(RuntimeError):
     decided to park; the agent waits for a resume newer than that.
     """
 
-    def __init__(self, message: str, *, resume_epoch: int = 0) -> None:
+    def __init__(
+        self, message: str, *, resume_epoch: int = 0, operator_pause: bool = False
+    ) -> None:
         super().__init__(message)
         self.resume_epoch = resume_epoch
+        self.operator_pause = operator_pause
 
 
 def _validate_budget(max_budget_usd: float | None) -> None:
@@ -198,6 +201,7 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
         system_prompt: str | None,  # noqa: ARG002
         input_items: list[TResponseInputItem],
     ) -> None:
+        self._pause_if_operator_paused(context)
         if self._budget_policy == "pause":
             self._pause_if_limited(context)
         context.context[LLM_TURN_KEY] = int(context.context.get(LLM_TURN_KEY, 0)) + 1
@@ -207,6 +211,15 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
         except Exception:
             logger.exception("budget/turn warning injection failed")
 
+    def _pause_if_operator_paused(self, context: RunContextWrapper[dict[str, Any]]) -> None:
+        coordinator = coordinator_from_context(context.context)
+        if coordinator is not None and coordinator.operator_paused:
+            raise BudgetPausedError(
+                "scan paused; waiting for the operator to resume",
+                resume_epoch=coordinator.resume_epoch,
+                operator_pause=True,
+            )
+
     def _pause_if_limited(self, context: RunContextWrapper[dict[str, Any]]) -> None:
         """Park the agent before a paid call when the scan is at its limit or paused.
 

tests/test_budget_pause_policy.py

@@ -21,7 +21,7 @@ from strix.core.sessions import open_agent_session
 
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Callable
+    from collections.abc import AsyncIterator, Awaitable, Callable
     from pathlib import Path
 
 
@@ -31,6 +31,7 @@ _CALL_LATENCY_S = 0.005
 
 class _FakeLedger:
     def __init__(self) -> None:
+        self.on_pause: Callable[[], Awaitable[None]] | None = None
         self.cost = 0.0
         self.calls: list[str] = []
         self.remaining: dict[str, int] = {}
@@ -80,6 +81,8 @@ class _FakeStream:
                 await self._hooks.on_llm_start(ctx_wrapper, self._agent, None, input_items)
             except BudgetPausedError as exc:
                 self.run_loop_exception = exc
+                if self._ledger.on_pause is not None:
+                    await self._ledger.on_pause()
                 return
             self._ledger.warned_inputs.append(input_items)
             if self._ledger.gate is not None:
@@ -150,13 +153,16 @@ class _Scan:
         monkeypatch: pytest.MonkeyPatch,
         *,
         max_budget_usd: float | None,
+        budget_policy: str = "pause",
     ) -> None:
         self.ledger = _FakeLedger()
         self.hooks = ReportUsageHooks(
-            model="test-model", max_budget_usd=max_budget_usd, budget_policy="pause"
+            model="test-model",
+            max_budget_usd=max_budget_usd,
+            budget_policy=budget_policy,  # type: ignore[arg-type]
         )
         self.coordinator = AgentCoordinator()
-        self.coordinator.set_budget_policy("pause")
+        self.coordinator.set_budget_policy(budget_policy)  # type: ignore[arg-type]
         self.coordinator.set_budget_limit_setter(self.hooks.set_max_budget_usd)
         monkeypatch.setattr(execution, "Runner", _fake_runner(self.ledger, self.coordinator))
         monkeypatch.setattr(execution, "_compact_session", _noop_compact)
@@ -538,3 +544,140 @@ async def test_stop_policy_is_unchanged() -> None:
 def test_budget_policy_is_validated() -> None:
     with pytest.raises(ValueError, match="budget_policy"):
         ReportUsageHooks(model="m", budget_policy="later")  # type: ignore[arg-type]
+
+
+def _pause_flags(coordinator: AgentCoordinator) -> tuple[bool, bool]:
+    return coordinator.budget_paused, coordinator.operator_paused
+
+
+@pytest.mark.asyncio
+async def test_operator_pause_holds_under_the_stop_policy(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
+    agents = ("root", "child-a", "child-b")
+
+    with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
+        await scan.start_root(calls=6)
+        await scan.start_child("child-a", calls=6)
+        await scan.start_child("child-b", calls=6)
+
+        await _wait_until(lambda: scan.ledger.cost >= 3.0)
+        await scan.coordinator.pause_budget()
+        spent_at_pause = scan.ledger.cost
+        await _wait_until(lambda: _all_parked(scan.coordinator, *agents))
+        assert _pause_flags(scan.coordinator) == (True, True)
+        assert all(not task.done() for task in scan.tasks())
+
+        await scan.coordinator.send("child-a", {"from": "root", "content": "status?"})
+        await scan.coordinator.send("root", {"from": "user", "content": "keep going"})
+        await asyncio.sleep(0.05)
+        assert _pause_flags(scan.coordinator) == (True, True)
+        assert scan.ledger.cost == pytest.approx(spent_at_pause)
+        assert _all_parked(scan.coordinator, *agents)
+
+        woken = await scan.coordinator.resume_budget()
+        assert sorted(woken) == sorted(agents)
+        assert _pause_flags(scan.coordinator) == (False, False)
+        await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
+        assert scan.ledger.cost == pytest.approx(18.0)
+        assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
+            "root": "completed",
+            "child-a": "completed",
+            "child-b": "completed",
+        }
+        root_items = await scan.sessions[0].get_items()
+        assert any("keep going" in str(item) for item in root_items)
+
+    for session in scan.sessions:
+        session.close()
+
+
+@pytest.mark.asyncio
+async def test_a_resume_that_lands_before_the_park_still_continues(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
+    resumed = 0
+
+    async def resume_at_once() -> None:
+        nonlocal resumed
+        resumed += 1
+        await scan.coordinator.resume_budget()
+
+    scan.ledger.on_pause = resume_at_once
+    with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
+        await scan.start_root(calls=6)
+        await scan.start_child("child-a", calls=6)
+        await _wait_until(lambda: scan.ledger.cost >= 2.0)
+        await scan.coordinator.pause_budget()
+        await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
+        assert resumed >= 1
+        assert _pause_flags(scan.coordinator) == (False, False)
+        assert scan.ledger.cost == pytest.approx(12.0)
+        assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
+            "root": "completed",
+            "child-a": "completed",
+        }
+
+    for session in scan.sessions:
+        session.close()
+
+
+@pytest.mark.asyncio
+async def test_operator_pause_gates_the_stop_policy_hooks() -> None:
+    coordinator = AgentCoordinator()
+    hooks = ReportUsageHooks(model="m", max_budget_usd=None)
+    assert hooks.budget_policy == "stop"
+
+    await coordinator.pause_budget()
+    with pytest.raises(BudgetPausedError):
+        await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
+
+    await coordinator.resume_budget()
+    await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
+
+
+@pytest.mark.asyncio
+async def test_a_user_message_lifts_only_the_limit_pause(tmp_path: Path) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    session = open_agent_session("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)
+    extensions: list[int] = []
+    coordinator.set_budget_extender(lambda: extensions.append(1))
+
+    await coordinator.pause_budget()
+    await coordinator.park_for_budget("root")
+    assert coordinator.statuses["root"] == "budget_paused"
+    await coordinator.send("root", {"from": "user", "content": "keep going"})
+    assert _pause_flags(coordinator) == (True, True)
+    assert extensions == []
+    assert coordinator.pending_counts["root"] == 1
+
+    await coordinator.resume_budget()
+    await coordinator.pause_for_budget("root")
+    await coordinator.send("root", {"from": "user", "content": "keep going"})
+    assert _pause_flags(coordinator) == (False, False)
+    assert extensions == [1]
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_snapshot_and_reset_keep_an_operator_pause() -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.pause_budget()
+    snap = await coordinator.snapshot()
+
+    restored = AgentCoordinator()
+    await restored.restore(snap)
+    assert _pause_flags(restored) == (True, True)
+    await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
+    assert _pause_flags(restored) == (True, True)
+
+    await restored.resume_budget()
+    await restored.pause_for_budget("root")
+    await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
+    assert _pause_flags(restored) == (False, False)
+    assert restored.statuses["root"] == "waiting"