refactor(steering): extract SteeringHook, remove enable_steering config

Move interruption injection logic from _LoopHook into a standalone
SteeringHook(AgentHook) that plugs into the hooks system. Steering is
now always active — InterruptionChecker is created per-dispatch and
SteeringHook is passed as an extra hook. When no interruptions arrive,
drain_all() returns empty and the hook is a no-op.

Removes enable_steering from AgentDefaults config, CLI wiring, and
AgentLoop.__init__. Removes transform_context/convert_to_llm params
that were unused after the per-tool cancellation drop.

Made-with: Cursor

Author: xzq-xu

nanobot/agent/loop.py

@@ -15,9 +15,8 @@
 from nanobot.agent.context import ContextBuilder
 from nanobot.agent.hook import AgentHook, AgentHookContext, CompositeHook
 from nanobot.agent.memory import Consolidator, Dream
-from nanobot.agent.messages import AgentMessage
 from nanobot.agent.runner import AgentRunSpec, AgentRunner
-from nanobot.agent.steering import InterruptionChecker
+from nanobot.agent.steering import InterruptionChecker, SteeringHook
 from nanobot.agent.subagent import SubagentManager
 from nanobot.agent.tools.cron import CronTool
 from nanobot.agent.skills import BUILTIN_SKILLS_DIR
@@ -41,10 +40,6 @@
     from nanobot.config.schema import ChannelsConfig, ExecToolConfig, WebToolsConfig
     from nanobot.cron.service import CronService
 
-TransformContextHook = Callable[[list[AgentMessage]], list[AgentMessage]]
-ConvertToLlmHook = Callable[[list[AgentMessage]], list[dict[str, Any]]]
-
-
 class _LoopHook(AgentHook):
     """Core hook for the main loop."""
 
@@ -58,7 +53,6 @@ def __init__(
         channel: str = "cli",
         chat_id: str = "direct",
         message_id: str | None = None,
-        interruption_checker: InterruptionChecker | None = None,
     ) -> None:
         self._loop = agent_loop
         self._on_progress = on_progress
@@ -67,7 +61,6 @@ def __init__(
         self._channel = channel
         self._chat_id = chat_id
         self._message_id = message_id
-        self._interruption_checker = interruption_checker
         self._stream_buf = ""
 
     def wants_streaming(self) -> bool:
@@ -88,20 +81,6 @@ async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> N
             await self._on_stream_end(resuming=resuming)
         self._stream_buf = ""
 
-    async def before_iteration(self, context: AgentHookContext) -> None:
-        if not self._interruption_checker:
-            return
-        pending = self._interruption_checker.drain_all()
-        if pending:
-            combined = "\n\n---\n\n".join(m.content for m in pending)
-            injection = (
-                "[The user just sent a new message while you were working. "
-                "Read it and decide: continue current work, switch to the "
-                "new request, or address both.]\n\n" + combined
-            )
-            context.messages.append({"role": "user", "content": injection})
-            logger.info("Steering: injected {} interruption(s) before LLM call", len(pending))
-
     async def before_execute_tools(self, context: AgentHookContext) -> None:
         if self._on_progress:
             if not self._on_stream:
@@ -125,23 +104,6 @@ async def after_iteration(self, context: AgentHookContext) -> None:
             u.get("completion_tokens", 0),
             u.get("cached_tokens", 0),
         )
-        # Steering: inject interruptions after tool results
-        if self._interruption_checker and context.tool_calls:
-            pending = self._interruption_checker.drain_all()
-            if pending:
-                combined = "\n\n---\n\n".join(m.content for m in pending)
-                injection = (
-                    "[The user just sent a new message while you were working. "
-                    "Read it and decide: continue current work, switch to the "
-                    "new request, or address both.]\n\n" + combined
-                )
-                context.messages.append({"role": "user", "content": injection})
-                logger.info("Steering: injected {} interruption(s)", len(pending))
-                if self._on_progress:
-                    await self._on_progress(
-                        "New message merged into current conversation",
-                        tool_hint=True,
-                    )
 
     def finalize_content(self, context: AgentHookContext, content: str | None) -> str | None:
         return self._loop._strip_think(content)
@@ -218,10 +180,6 @@ def __init__(
         channels_config: ChannelsConfig | None = None,
         timezone: str | None = None,
         hooks: list[AgentHook] | None = None,
-        *,
-        enable_steering: bool = False,
-        transform_context: TransformContextHook | None = None,
-        convert_to_llm: ConvertToLlmHook | None = None,
     ):
         from nanobot.config.schema import ExecToolConfig, WebToolsConfig
 
@@ -254,11 +212,6 @@ def __init__(
         self._last_usage: dict[str, int] = {}
         self._extra_hooks: list[AgentHook] = hooks or []
 
-        # Dual-layer architecture (opt-in)
-        self.enable_steering = enable_steering
-        self._transform_context = transform_context
-        self._convert_to_llm = convert_to_llm or AgentMessage.to_llm_list
-
         self.context = ContextBuilder(workspace, timezone=timezone)
         self.sessions = session_manager or SessionManager(workspace)
         self.tools = ToolRegistry()
@@ -389,17 +342,15 @@ async def _run_agent_loop(
         channel: str = "cli",
         chat_id: str = "direct",
         message_id: str | None = None,
-        interruption_checker: InterruptionChecker | None = None,
+        extra_hooks: list[AgentHook] | None = None,
     ) -> tuple[str | None, list[str], list[dict]]:
         """Run the agent iteration loop.
 
         *on_stream*: called with each content delta during streaming.
         *on_stream_end(resuming)*: called when a streaming session finishes.
         ``resuming=True`` means tool calls follow (spinner should restart);
         ``resuming=False`` means this is the final response.
-
-        *interruption_checker*: when steering is enabled, used to merge user
-        messages that arrive during tool execution.
+        *extra_hooks*: per-call hooks (e.g. SteeringHook) merged with instance hooks.
         """
         loop_hook = _LoopHook(
             self,
@@ -409,11 +360,11 @@ async def _run_agent_loop(
             channel=channel,
             chat_id=chat_id,
             message_id=message_id,
-            interruption_checker=interruption_checker,
         )
+        all_extra = self._extra_hooks + (extra_hooks or [])
         hook: AgentHook = (
-            _LoopHookChain(loop_hook, self._extra_hooks)
-            if self._extra_hooks
+            _LoopHookChain(loop_hook, all_extra)
+            if all_extra
             else loop_hook
         )
 
@@ -450,16 +401,14 @@ async def run(self) -> None:
         """Run the agent loop, dispatching messages as tasks to stay responsive to /stop."""
         self._running = True
         await self._connect_mcp()
-        logger.info("Agent loop started (steering={})", self.enable_steering)
+        logger.info("Agent loop started")
 
         while self._running:
             try:
                 msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
             except asyncio.TimeoutError:
                 continue
             except asyncio.CancelledError:
-                # Preserve real task cancellation so shutdown can complete cleanly.
-                # Only ignore non-task CancelledError signals that may leak from integrations.
                 if not self._running or asyncio.current_task().cancelling():
                     raise
                 continue
@@ -475,23 +424,20 @@ async def run(self) -> None:
                     await self.bus.publish_outbound(result)
                 continue
 
-            if self.enable_steering:
-                active = [t for t in self._active_tasks.get(msg.session_key, []) if not t.done()]
-                if active and msg.session_key in self._interrupt_checkers:
-                    await self._interrupt_checkers[msg.session_key].signal(msg)
-                    logger.info("Steering: signaled interruption for {}", msg.session_key)
-                    continue
+            active = [t for t in self._active_tasks.get(msg.session_key, []) if not t.done()]
+            if active and msg.session_key in self._interrupt_checkers:
+                await self._interrupt_checkers[msg.session_key].signal(msg)
+                logger.info("Steering: signaled interruption for {}", msg.session_key)
+                continue
 
             task = asyncio.create_task(self._dispatch(msg))
             self._active_tasks.setdefault(msg.session_key, []).append(task)
             task.add_done_callback(lambda t, k=msg.session_key: self._active_tasks.get(k, []) and self._active_tasks[k].remove(t) if t in self._active_tasks.get(k, []) else None)
 
     async def _dispatch(self, msg: InboundMessage) -> None:
         """Process a message: per-session serial, cross-session concurrent."""
-        checker: InterruptionChecker | None = None
-        if self.enable_steering:
-            checker = InterruptionChecker()
-            self._interrupt_checkers[msg.session_key] = checker
+        checker = InterruptionChecker()
+        self._interrupt_checkers[msg.session_key] = checker
         lock = self._session_locks.setdefault(msg.session_key, asyncio.Lock())
         gate = self._concurrency_gate or nullcontext()
         try:
@@ -531,7 +477,7 @@ async def on_stream_end(*, resuming: bool = False) -> None:
                     response = await self._process_message(
                         msg,
                         on_stream=on_stream, on_stream_end=on_stream_end,
-                        interruption_checker=checker,
+                        extra_hooks=[SteeringHook(checker)],
                     )
                     if response is not None:
                         await self.bus.publish_outbound(response)
@@ -550,8 +496,7 @@ async def on_stream_end(*, resuming: bool = False) -> None:
                         content="Sorry, I encountered an error.",
                     ))
         finally:
-            if self.enable_steering:
-                self._interrupt_checkers.pop(msg.session_key, None)
+            self._interrupt_checkers.pop(msg.session_key, None)
 
     async def close_mcp(self) -> None:
         """Drain pending background archives, then close MCP connections."""
@@ -583,7 +528,7 @@ async def _process_message(
         on_progress: Callable[[str], Awaitable[None]] | None = None,
         on_stream: Callable[[str], Awaitable[None]] | None = None,
         on_stream_end: Callable[..., Awaitable[None]] | None = None,
-        interruption_checker: InterruptionChecker | None = None,
+        extra_hooks: list[AgentHook] | None = None,
     ) -> OutboundMessage | None:
         """Process a single inbound message and return the response."""
         # System messages: parse origin from chat_id ("channel:chat_id")
@@ -607,7 +552,7 @@ async def _process_message(
             final_content, _, all_msgs = await self._run_agent_loop(
                 messages, session=session, channel=channel, chat_id=chat_id,
                 message_id=msg.metadata.get("message_id"),
-                interruption_checker=interruption_checker,
+                extra_hooks=extra_hooks,
             )
             self._save_turn(session, all_msgs, 1 + len(history))
             self._clear_runtime_checkpoint(session)
@@ -661,7 +606,7 @@ async def _bus_progress(content: str, *, tool_hint: bool = False) -> None:
             session=session,
             channel=msg.channel, chat_id=msg.chat_id,
             message_id=msg.metadata.get("message_id"),
-            interruption_checker=interruption_checker,
+            extra_hooks=extra_hooks,
         )
 
         if final_content is None or not final_content.strip():
@@ -841,13 +786,13 @@ async def process_direct(
         on_progress: Callable[[str], Awaitable[None]] | None = None,
         on_stream: Callable[[str], Awaitable[None]] | None = None,
         on_stream_end: Callable[..., Awaitable[None]] | None = None,
-        interruption_checker: InterruptionChecker | None = None,
+        extra_hooks: list[AgentHook] | None = None,
     ) -> OutboundMessage | None:
         """Process a message directly and return the outbound payload."""
         await self._connect_mcp()
         msg = InboundMessage(channel=channel, sender_id="user", chat_id=chat_id, content=content)
         return await self._process_message(
             msg, session_key=session_key, on_progress=on_progress,
             on_stream=on_stream, on_stream_end=on_stream_end,
-            interruption_checker=interruption_checker,
+            extra_hooks=extra_hooks,
         )

nanobot/agent/steering.py

@@ -7,6 +7,8 @@
 
 from loguru import logger
 
+from nanobot.agent.hook import AgentHook, AgentHookContext
+
 if TYPE_CHECKING:
     from nanobot.bus.events import InboundMessage
 
@@ -15,7 +17,7 @@ class InterruptionChecker:
     """
     Per-session interruption queue.
 
-    Written by ``AgentLoop.run()``; read by ``_run_agent_loop``.
+    Written by ``AgentLoop.run()``; read by ``SteeringHook``.
     """
 
     def __init__(self) -> None:
@@ -41,3 +43,27 @@ def drain_all(self) -> list[InboundMessage]:
     @property
     def has_pending(self) -> bool:
         return not self._queue.empty()
+
+
+class SteeringHook(AgentHook):
+    """AgentHook that injects pending user interruptions before each LLM call.
+
+    Drop this into the ``hooks`` list of ``AgentLoop`` to enable steering.
+    The hook is stateless beyond its reference to the per-session checker.
+    """
+
+    def __init__(self, checker: InterruptionChecker) -> None:
+        self._checker = checker
+
+    async def before_iteration(self, context: AgentHookContext) -> None:
+        pending = self._checker.drain_all()
+        if not pending:
+            return
+        combined = "\n\n---\n\n".join(m.content for m in pending)
+        injection = (
+            "[The user just sent a new message while you were working. "
+            "Read it and decide: continue current work, switch to the "
+            "new request, or address both.]\n\n" + combined
+        )
+        context.messages.append({"role": "user", "content": injection})
+        logger.info("Steering: injected {} interruption(s) before LLM call", len(pending))

nanobot/cli/commands.py

@@ -681,7 +681,6 @@ def gateway(
         mcp_servers=config.tools.mcp_servers,
         channels_config=config.channels,
         timezone=config.agents.defaults.timezone,
-        enable_steering=config.agents.defaults.enable_steering,
     )
 
     # Set cron callback (needs agent)
@@ -913,7 +912,6 @@ def agent(
         mcp_servers=config.tools.mcp_servers,
         channels_config=config.channels,
         timezone=config.agents.defaults.timezone,
-        enable_steering=config.agents.defaults.enable_steering,
     )
     restart_notice = consume_restart_notice_from_env()
     if restart_notice and should_show_cli_restart_notice(restart_notice, session_id):