fix(agent): stop passive mode when idle

src/coworker/agent/loop.py

@@ -298,6 +298,11 @@ async def _cycle(self) -> None:
             logger.warning(
                 f"Injected {len(skill_load_warnings)} skill load warning(s) into model context"
             )
+
+        if self._should_wait_passively(events):
+            await self._rest()
+            return
+
         if self._ilog:
             self._ilog.log_system_prompt(system_prompt)
 
@@ -306,6 +311,7 @@ async def _cycle(self) -> None:
         )
         if (
             self.state.tick
+            and not self._config.agent.passive_mode
             and not events
             and not reinjected_pins
             and (not last_assistant or last_assistant.stop_reason != "tool_use")
@@ -409,7 +415,7 @@ async def _cycle(self) -> None:
             recall_query = assistant_msg.reasoning_content or assistant_msg.content_text()
             await self._auto_recall(recall_query)
             await self._task_reminder()
-        elif not events:
+        elif self._config.agent.passive_mode or not events:
             await self._rest()
 
         self.state.cycle_count += 1
@@ -423,6 +429,15 @@ async def _cycle(self) -> None:
                 tool_calls_this_cycle=len(response.tool_calls),
             )
 
+    def _should_wait_passively(self, events: list[IncomingEvent]) -> bool:
+        """Wait before thinking when passive mode has no work left to continue."""
+        if not self._config.agent.passive_mode or events:
+            return False
+        if not self._short_term.primary:
+            return True
+        last_message = self._short_term.primary[-1]
+        return last_message.role == "assistant" and last_message.stop_reason != "tool_use"
+
     @staticmethod
     def _build_content_blocks(events: list[IncomingEvent]) -> str | list[dict]:
         return build_content_blocks(events)

tests/unit/test_loop_summaries.py

@@ -30,6 +30,7 @@ def _make_loop(brain, mem, events=None):
     config = MagicMock()
     config.agent.idle_sleep_seconds = 0
     config.agent.inbox_batch_max = 10
+    config.agent.passive_mode = False
 
     state = MagicMock()
     state.tick = False
@@ -209,6 +210,82 @@ async def test_assistant_response_appended_to_primary():
     assert asst_msgs[0].content == "my reply"
 
 
+@pytest.mark.asyncio
+async def test_passive_mode_waits_before_thinking_on_clean_start():
+    mem = ShortTermMemory()
+    brain = _make_brain()
+    loop = _make_loop(brain, mem)
+    loop._config.agent.passive_mode = True
+    loop.state.tick = True
+    loop._rest = AsyncMock()
+    loop._short_term.compress_if_needed = AsyncMock()
+
+    await loop._cycle()
+
+    brain.think.assert_not_awaited()
+    loop._rest.assert_awaited_once()
+    assert mem.primary == []
+
+
+@pytest.mark.asyncio
+async def test_passive_mode_rests_after_handling_event_without_heartbeat():
+    mem = ShortTermMemory()
+    brain = _make_brain(content="done")
+    event = IncomingEvent(participant_id="alice", content="hello")
+    loop = _make_loop(brain, mem, events=[event])
+    loop._config.agent.passive_mode = True
+    loop.state.tick = True
+    loop._rest = AsyncMock()
+    loop._short_term.compress_if_needed = AsyncMock()
+
+    await loop._cycle()
+
+    brain.think.assert_awaited_once()
+    loop._rest.assert_awaited_once()
+    assert not any(m.source == "tick" for m in mem.primary)
+
+
+@pytest.mark.asyncio
+async def test_passive_mode_waits_after_restored_terminal_response():
+    mem = ShortTermMemory()
+    mem.primary.append(Message(role="assistant", content="done", stop_reason="end_turn"))
+    brain = _make_brain()
+    loop = _make_loop(brain, mem, events=[])
+    loop._config.agent.passive_mode = True
+    loop.state.tick = True
+    loop._rest = AsyncMock()
+    loop._short_term.compress_if_needed = AsyncMock()
+
+    await loop._cycle()
+
+    brain.think.assert_not_awaited()
+    loop._rest.assert_awaited_once()
+    assert not any(m.source == "tick" for m in mem.primary)
+
+
+@pytest.mark.asyncio
+async def test_passive_mode_continues_pending_tool_result_before_resting():
+    mem = ShortTermMemory()
+    mem.primary.extend(
+        [
+            Message(role="assistant", content="", stop_reason="tool_use"),
+            Message(role="tool", content="tool result", tool_call_id="tool-1"),
+        ]
+    )
+    brain = _make_brain(content="finished")
+    loop = _make_loop(brain, mem, events=[])
+    loop._config.agent.passive_mode = True
+    loop.state.tick = True
+    loop._rest = AsyncMock()
+    loop._short_term.compress_if_needed = AsyncMock()
+
+    await loop._cycle()
+
+    brain.think.assert_awaited_once()
+    loop._rest.assert_awaited_once()
+    assert mem.primary[-1].content == "finished"
+
+
 def _make_rest_loop(*, passive: bool, idle_sleep_seconds: int = 0) -> AgentLoop:
     """构造仅满足 _rest() 依赖的最小 AgentLoop。"""
     loop = AgentLoop.__new__(AgentLoop)