feat(agent): add explicit active-turn steering

agent/chat/service.py

@@ -183,9 +183,15 @@ class ChatService:
 
         # Register a cancel token so /cancel can abort this in-flight run.
         # IM channels key on session_id (no per-turn request_id here).
-        from agent.protocol import get_cancel_registry
+        from agent.protocol import get_cancel_registry, get_steer_registry
         registry = get_cancel_registry()
+        steer_registry = get_steer_registry()
         cancel_event = registry.register(session_id, session_id=session_id) if session_id else None
+        steer_inbox = (
+            steer_registry.register(session_id)
+            if session_id
+            else None
+        )
 
         executor = AgentStreamExecutor(
             agent=agent,
@@ -197,6 +203,7 @@ class ChatService:
             messages=messages_copy,
             max_context_turns=max_context_turns,
             cancel_event=cancel_event,
+            steer_inbox=steer_inbox,
         )
 
         try:
@@ -217,6 +224,8 @@ class ChatService:
                     registry.unregister(session_id)
                 except Exception:
                     pass
+            if session_id and steer_inbox is not None:
+                steer_registry.unregister(session_id, steer_inbox)
 
         # Sync executor messages back to agent (thread-safe).
         # The executor may have trimmed context, making its list shorter than

agent/protocol/__init__.py

@@ -8,6 +8,13 @@ from .cancel import (
     CancelTokenRegistry,
     get_cancel_registry,
 )
+from .steer import (
+    SteerInbox,
+    SteerRegistry,
+    SteerResult,
+    SteerStatus,
+    get_steer_registry,
+)
 
 __all__ = [
     'Agent', 
@@ -25,4 +32,9 @@ __all__ = [
     'AgentCancelledError',
     'CancelTokenRegistry',
     'get_cancel_registry',
+    'SteerInbox',
+    'SteerRegistry',
+    'SteerResult',
+    'SteerStatus',
+    'get_steer_registry',
 ]

agent/protocol/agent.py

@@ -381,7 +381,7 @@ class Agent:
         return action
 
     def run_stream(self, user_message: str, on_event=None, clear_history: bool = False,
-                   skill_filter=None, cancel_event=None) -> str:
+                   skill_filter=None, cancel_event=None, steer_inbox=None) -> str:
         """
         Execute single agent task with streaming (based on tool-call)
 
@@ -391,6 +391,7 @@ class Agent:
         - Event callbacks
         - Persistent conversation history across calls
         - User-initiated cancellation via ``cancel_event``
+        - Explicit active-turn guidance via ``steer_inbox``
 
         Args:
             user_message: User message
@@ -403,6 +404,8 @@ class Agent:
                 "[Interrupted by user]" assistant note, and returns the
                 partial response. ``messages`` stays in a valid state
                 (tool_use/tool_result pairs preserved).
+            steer_inbox: Optional SteerInbox drained at safe checkpoints. New
+                instructions guide this run without entering the normal queue.
 
         Returns:
             Final response text
@@ -448,6 +451,7 @@ class Agent:
             messages=messages_copy,  # Pass copied message history
             max_context_turns=max_context_turns,
             cancel_event=cancel_event,
+            steer_inbox=steer_inbox,
         )
 
         # Execute
@@ -484,4 +488,4 @@ class Agent:
     def clear_history(self):
         """Clear conversation history and captured actions"""
         self.messages = []
-        self.captured_actions = []
+        self.captured_actions = []

agent/protocol/agent_stream.py

@@ -99,6 +99,7 @@ class AgentStreamExecutor:
             messages: Optional[List[Dict]] = None,
             max_context_turns: int = 30,
             cancel_event=None,
+            steer_inbox=None,
     ):
         """
         Initialize stream executor
@@ -116,6 +117,8 @@ class AgentStreamExecutor:
                 Checked at every safe point (turn boundary, before tool execution,
                 during LLM streaming). When set, raises AgentCancelledError which
                 run_stream catches to gracefully wind down.
+            steer_inbox: Optional SteerInbox for explicit instructions sent to
+                this active run. Drained only at message-safe checkpoints.
         """
         self.agent = agent
         self.model = model
@@ -126,6 +129,7 @@ class AgentStreamExecutor:
         self.on_event = on_event
         self.max_context_turns = max_context_turns
         self.cancel_event = cancel_event
+        self.steer_inbox = steer_inbox
 
         # Message history - use provided messages or create new list
         self.messages = messages if messages is not None else []
@@ -145,6 +149,72 @@ class AgentStreamExecutor:
         if self.cancel_event is not None and self.cancel_event.is_set():
             raise AgentCancelledError("agent cancelled by user")
 
+    def _drain_steering(self) -> List[str]:
+        if self.steer_inbox is None:
+            return []
+        return self.steer_inbox.drain()
+
+    @staticmethod
+    def _steering_text(updates: List[str]) -> str:
+        if len(updates) == 1:
+            body = updates[0]
+        else:
+            body = "\n".join(f"{idx}. {text}" for idx, text in enumerate(updates, 1))
+        return (
+            "[Steering update for the active task]\n"
+            "Use this new instruction for the current task before continuing.\n\n"
+            f"{body}"
+        )
+
+    def _append_steering(
+        self,
+        updates: List[str],
+        pending_tool_calls: Optional[List[Dict]] = None,
+        content_blocks: Optional[List[Dict]] = None,
+    ) -> None:
+        """Append guidance, closing any tool_use blocks that will be skipped."""
+        if not updates:
+            return
+        blocks = content_blocks if content_blocks is not None else []
+        for tool_call in pending_tool_calls or []:
+            blocks.append({
+                "type": "tool_result",
+                "tool_use_id": tool_call["id"],
+                "content": "Skipped because the user redirected the active task.",
+                "is_error": True,
+            })
+        blocks.append({"type": "text", "text": self._steering_text(updates)})
+        if content_blocks is None:
+            self.messages.append({"role": "user", "content": blocks})
+        self._emit_event("agent_steered", {"count": len(updates)})
+        logger.info(f"[Agent] Applied {len(updates)} steering update(s)")
+
+    def _close_or_apply_final_steering(self) -> bool:
+        """Return True only when the run can finish without losing a steer."""
+        updates = self._drain_steering()
+        if updates:
+            self._append_steering(updates)
+            return False
+        if self.steer_inbox is None:
+            return True
+        if self.steer_inbox.close_if_empty():
+            return True
+        updates = self._drain_steering()
+        if updates:
+            self._append_steering(updates)
+        return False
+
+    def _drain_and_close_steering(self) -> None:
+        """Preserve any final guidance before the max-step summary call."""
+        if self.steer_inbox is None:
+            return
+        while True:
+            updates = self._drain_steering()
+            if updates:
+                self._append_steering(updates)
+            if self.steer_inbox.close_if_empty():
+                return
+
     def _handle_cancelled(self, partial_response: str) -> None:
         """Wind down ``self.messages`` after a user-initiated cancel.
 
@@ -395,6 +465,10 @@ class AgentStreamExecutor:
                 # between turns short-circuits cleanly.
                 self._check_cancelled()
 
+                steering_updates = self._drain_steering()
+                if steering_updates:
+                    self._append_steering(steering_updates)
+
                 turn += 1
                 logger.info(f"[Agent] Turn {turn}")
                 self._emit_event("turn_start", {"turn": turn})
@@ -403,6 +477,24 @@ class AgentStreamExecutor:
                 assistant_msg, tool_calls = self._call_llm_stream(retry_on_empty=True)
                 final_response = assistant_msg
 
+                # A steer that arrived while the model was streaming takes
+                # precedence over its proposed continuation. Tool calls have
+                # already been written to history, so close every one with a
+                # synthetic result before asking the model to reconsider.
+                steering_updates = self._drain_steering()
+                if steering_updates:
+                    self._append_steering(
+                        steering_updates,
+                        pending_tool_calls=tool_calls,
+                    )
+                    self._emit_event("turn_end", {
+                        "turn": turn,
+                        "has_tool_calls": bool(tool_calls),
+                        "tool_count": len(tool_calls),
+                        "steered": True,
+                    })
+                    continue
+
                 # No tool calls, end loop
                 if not tool_calls:
                     # 检查是否返回了空响应
@@ -467,6 +559,22 @@ class AgentStreamExecutor:
                     # If the explicit-response retry produced tool_calls, skip the break
                     # and continue down to the tool execution branch in this same iteration.
                     if not tool_calls:
+                        steering_updates = self._drain_steering()
+                        if steering_updates:
+                            self._append_steering(steering_updates)
+                            self._emit_event("turn_end", {
+                                "turn": turn,
+                                "has_tool_calls": False,
+                                "steered": True,
+                            })
+                            continue
+                        if not self._close_or_apply_final_steering():
+                            self._emit_event("turn_end", {
+                                "turn": turn,
+                                "has_tool_calls": False,
+                                "steered": True,
+                            })
+                            continue
                         logger.debug(f"✅ Done (no tool calls)")
                         self._emit_event("turn_end", {
                             "turn": turn,
@@ -499,9 +607,17 @@ class AgentStreamExecutor:
                 tool_result_blocks = []
 
                 try:
-                    for tool_call in tool_calls:
+                    for tool_index, tool_call in enumerate(tool_calls):
                         # Honour cancel between tool invocations within the same turn
                         self._check_cancelled()
+                        steering_updates = self._drain_steering()
+                        if steering_updates:
+                            self._append_steering(
+                                steering_updates,
+                                pending_tool_calls=tool_calls[tool_index:],
+                                content_blocks=tool_result_blocks,
+                            )
+                            break
                         result = self._execute_tool(tool_call)
                         tool_results.append(result)
                         
@@ -641,6 +757,7 @@ class AgentStreamExecutor:
 
             if turn >= self.max_turns:
                 logger.warning(f"⚠️  Reached max decision step limit: {self.max_turns}")
+                self._drain_and_close_steering()
                 
                 # Force model to summarize without tool calls
                 logger.info(f"[Agent] Requesting summary from LLM after reaching max steps...")
@@ -699,6 +816,8 @@ class AgentStreamExecutor:
             raise
 
         finally:
+            if self.steer_inbox is not None:
+                self.steer_inbox.close()
             final_response = final_response.strip() if final_response else final_response
             if cancelled:
                 # Emit before agent_end so channels can mark UI as cancelled

agent/protocol/steer.py

@@ -0,0 +1,127 @@
+"""Thread-safe active-run steering primitives.
+
+Steering is deliberately separate from the normal per-session message queue.
+An instruction is accepted only while exactly one run for the scoped session
+is active; idle sessions never start a new run as a side effect.
+"""
+
+from __future__ import annotations
+
+import threading
+from collections import deque
+from dataclasses import dataclass
+from enum import Enum
+from typing import Deque, Dict, List, Optional, Set
+
+
+class SteerStatus(str, Enum):
+    ACCEPTED = "accepted"
+    INACTIVE = "inactive"
+    AMBIGUOUS = "ambiguous"
+    INVALID = "invalid"
+    FULL = "full"
+    CLOSING = "closing"
+
+
+@dataclass(frozen=True)
+class SteerResult:
+    status: SteerStatus
+
+    @property
+    def accepted(self) -> bool:
+        return self.status == SteerStatus.ACCEPTED
+
+
+class SteerInbox:
+    """Bounded inbox owned by one active agent run."""
+
+    def __init__(self, max_pending: int = 16, max_chars: int = 8000):
+        self.max_pending = max(1, int(max_pending))
+        self.max_chars = max(1, int(max_chars))
+        self._lock = threading.Lock()
+        self._pending: Deque[str] = deque()
+        self._accepting = True
+
+    def submit(self, instruction: str) -> SteerResult:
+        text = (instruction or "").strip()
+        if not text or len(text) > self.max_chars:
+            return SteerResult(SteerStatus.INVALID)
+        with self._lock:
+            if not self._accepting:
+                return SteerResult(SteerStatus.CLOSING)
+            if len(self._pending) >= self.max_pending:
+                return SteerResult(SteerStatus.FULL)
+            self._pending.append(text)
+        return SteerResult(SteerStatus.ACCEPTED)
+
+    def drain(self) -> List[str]:
+        with self._lock:
+            items = list(self._pending)
+            self._pending.clear()
+            return items
+
+    def close_if_empty(self) -> bool:
+        """Atomically stop accepting when no instruction is pending.
+
+        This closes the race between a final empty drain and an agent run
+        returning: after this method succeeds, submitters receive CLOSING.
+        """
+        with self._lock:
+            if self._pending:
+                return False
+            self._accepting = False
+            return True
+
+    def close(self) -> None:
+        with self._lock:
+            self._accepting = False
+
+
+class SteerRegistry:
+    """Map a scoped agent/session key to its active run inboxes."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._by_session: Dict[str, Set[SteerInbox]] = {}
+
+    def register(self, session_id: str, inbox: Optional[SteerInbox] = None) -> SteerInbox:
+        inbox = inbox or SteerInbox()
+        if not session_id:
+            return inbox
+        with self._lock:
+            self._by_session.setdefault(session_id, set()).add(inbox)
+        return inbox
+
+    def unregister(self, session_id: str, inbox: Optional[SteerInbox]) -> None:
+        if not session_id or inbox is None:
+            return
+        inbox.close()
+        with self._lock:
+            bucket = self._by_session.get(session_id)
+            if bucket is None:
+                return
+            bucket.discard(inbox)
+            if not bucket:
+                self._by_session.pop(session_id, None)
+
+    def submit(self, session_id: str, instruction: str) -> SteerResult:
+        if not (instruction or "").strip():
+            return SteerResult(SteerStatus.INVALID)
+        with self._lock:
+            inboxes = list(self._by_session.get(session_id, ()))
+        if not inboxes:
+            return SteerResult(SteerStatus.INACTIVE)
+        if len(inboxes) != 1:
+            return SteerResult(SteerStatus.AMBIGUOUS)
+        return inboxes[0].submit(instruction)
+
+    def active_count(self, session_id: str) -> int:
+        with self._lock:
+            return len(self._by_session.get(session_id, ()))
+
+
+_registry = SteerRegistry()
+
+
+def get_steer_registry() -> SteerRegistry:
+    return _registry