fix(steer): preserve late steer after output truncation

agent/agent_init.py

@@ -886,6 +886,11 @@ def init_agent(
     # existing tool message rather than inserting a new user turn).
     agent._pending_steer: Optional[str] = None
     agent._pending_steer_lock = threading.Lock()
+    # A steer is accepted only while one run_conversation generation is open.
+    # Terminal drains seal this token under the same lock as the pending slot,
+    # so callers can distinguish accepted in-turn work from next-turn queueing.
+    agent._steer_generation_counter = 0
+    agent._steer_acceptance_generation: Optional[int] = None
 
     # Active-turn redirect mechanism. A regular follow-up sent while the model
     # is generating is different from a hard /stop: preserve the valid turn

agent/turn_finalizer.py

@@ -752,7 +752,12 @@ def finalize_turn(
     # If a /steer landed after the final assistant turn (no more tool
     # batches to drain into), hand it back to the caller so it can be
     # delivered as the next user turn instead of being silently lost.
-    _leftover_steer = agent._drain_pending_steer()
+    _seal_steer = getattr(agent, "_seal_pending_steer", None)
+    _leftover_steer = (
+        _seal_steer()
+        if callable(_seal_steer)
+        else agent._drain_pending_steer()  # compatibility for minimal test/plugin agents
+    )
     if _leftover_steer:
         result["pending_steer"] = _leftover_steer
     agent._response_was_previewed = False

cli.py

@@ -12262,7 +12262,12 @@ def process_command(self, command: str) -> bool:
                     if accepted:
                         _cprint(f"  ⏩ Steer queued — arrives after the next tool call: {payload[:80]}{'...' if len(payload) > 80 else ''}")
                     else:
-                        _cprint("  Steer rejected (empty payload).")
+                        # The turn can seal its terminal result between the UI's
+                        # busy check and steer() acquiring the acceptance lock.
+                        # Preserve that valid non-empty message behind any older
+                        # next-turn work instead of misreporting it as empty.
+                        self._pending_input.put(payload)
+                        _cprint(f"  Turn already completed; queued for the next turn: {payload[:80]}{'...' if len(payload) > 80 else ''}")
             else:
                 # No active run — treat as a normal next-turn message.
                 self._pending_input.put(payload)

gateway/platforms/base.py

@@ -6042,6 +6042,19 @@ async def _dispatch_active_session_command(
 
         await self._drain_pending_after_session_command(session_key, command_guard)
 
+    def session_key_for_source(self, source: SessionSource) -> str:
+        """Return this adapter's physical active/pending slot key."""
+        return build_session_key(
+            source,
+            group_sessions_per_user=self.config.extra.get(
+                "group_sessions_per_user", True
+            ),
+            thread_sessions_per_user=self.config.extra.get(
+                "thread_sessions_per_user", False
+            ),
+            profile=self._session_key_profile(source),
+        )
+
     async def handle_message(self, event: MessageEvent) -> None:
         """
         Process an incoming message.
@@ -6067,12 +6080,7 @@ async def handle_message(self, event: MessageEvent) -> None:
         if needs_topic_recovery:
             await asyncio.to_thread(self._apply_topic_recovery, event)
 
-        session_key = build_session_key(
-            event.source,
-            group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
-            thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
-            profile=self._session_key_profile(event.source),
-        )
+        session_key = self.session_key_for_source(event.source)
         expected_session_key = str(
             (event.metadata or {}).get("gateway_session_key") or ""
         ).strip()