From fd7e841cc2545d3f87ead6ffab997e98c399e0ed Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Tue, 11 Aug 2026 20:08:09 +0200 Subject: [PATCH 1/9] fix(steer): preserve terminal delivery ordering Seal steer acceptance atomically at terminal result boundaries, preserve rejected CLI and TUI steers as next-turn work, and keep gateway adapter slots distinct from durable FIFO state. Add race, ingress, profile-key, command-provenance, and depth-cap regressions. --- agent/agent_init.py | 5 + agent/conversation_loop.py | 173 ++++--- agent/turn_finalizer.py | 7 +- cli.py | 7 +- gateway/platforms/base.py | 19 +- gateway/run.py | 460 +++++++++++++----- gateway/slash_commands.py | 45 +- run_agent.py | 37 +- tests/cli/test_cli_init.py | 13 + tests/gateway/test_busy_session_ack.py | 40 ++ tests/gateway/test_queue_consumption.py | 30 +- tests/gateway/test_terminal_steer_fifo.py | 405 +++++++++++++++ tests/run_agent/test_413_compression.py | 26 + .../test_partial_stream_finish_reason.py | 32 ++ tests/run_agent/test_run_agent.py | 110 +++++ .../test_run_agent_codex_responses.py | 20 + tests/run_agent/test_steer.py | 43 ++ tests/test_tui_gateway_server.py | 37 ++ tui_gateway/methods_session.py | 11 +- 19 files changed, 1319 insertions(+), 201 deletions(-) create mode 100644 tests/gateway/test_terminal_steer_fifo.py diff --git a/agent/agent_init.py b/agent/agent_init.py index ca5996636bd0..2d92327fec15 100644 --- a/agent/agent_init.py +++ b/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 diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 951b87470102..ffbbdcc8cb21 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -460,6 +460,28 @@ def _is_stale_copilot_credential_error(status_code: Optional[int], error_message ) +def _terminal_result_with_pending_steer( + agent: Any, result: Dict[str, Any] +) -> Dict[str, Any]: + """Preserve a late /steer when a terminal exit bypasses finalization. + + ``finalize_turn`` normally drains this queue after the loop. Early terminal + exits must provide the same caller contract, otherwise a steer that arrived + while the provider request was in flight is silently lost. Hard-interrupt + exits intentionally remain separate because ``clear_interrupt`` discards a + pending steer by design. + """ + 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 + return result + + def _image_error_max_dimension(error: Exception) -> Optional[int]: """Extract a provider-reported image dimension ceiling, if present.""" parts = [] @@ -1800,6 +1822,12 @@ def run_conversation( Returns: Dict: Complete conversation result with final response and message history """ + # Cached agents span turns. Open a distinct steer-acceptance generation + # before any terminal branch can return; its terminal drain seals the same + # token atomically with the pending slot. + begin_steer = getattr(agent, "_begin_steer_acceptance", None) + if callable(begin_steer): + begin_steer() if moa_config is None: try: from hermes_cli.moa_config import decode_moa_turn @@ -2866,7 +2894,7 @@ def run_conversation( # so user sees the rate-limit message that led here. agent._flush_status_buffer() agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": ( f"⏳ {_nous_msg}\n\n" "No fallback provider available. " @@ -2878,7 +2906,7 @@ def run_conversation( "completed": False, "failed": True, "error": _nous_msg, - } + }) except ImportError: pass except Exception: @@ -3420,14 +3448,14 @@ def _perform_api_call(next_api_kwargs): logger.error("%sInvalid API response after %d retries.", agent.log_prefix, max_retries) agent._persist_session(messages, conversation_history) _final_response = f"Invalid API response after {max_retries} retries: {_failure_hint}" - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, "api_calls": api_call_count, "error": _final_response, "failed": True # Mark as failure for filtering - } + }) # Backoff before retry — jittered exponential: 5s base, 120s cap wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0) @@ -3623,11 +3651,14 @@ def _perform_api_call(next_api_kwargs): agent._cleanup_task_resources(effective_task_id) agent._persist_session(messages, conversation_history) - return _content_policy_blocked_result( - messages, - api_call_count, - final_response=_refusal_response, - error_detail=_refusal_text or "model declined (content_filter)", + return _terminal_result_with_pending_steer( + agent, + _content_policy_blocked_result( + messages, + api_call_count, + final_response=_refusal_response, + error_detail=_refusal_text or "model declined (content_filter)", + ), ) if finish_reason == "length": @@ -3716,14 +3747,14 @@ def _perform_api_call(next_api_kwargs): ) agent._cleanup_task_resources(effective_task_id) agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _exhaust_response, "messages": messages, "api_calls": api_call_count, "completed": False, "partial": True, "error": _exhaust_error, - } + }) # ── Detect repetition-dominated truncation (#86581) ── # A model in a degenerate repetition loop can spend its @@ -3937,7 +3968,7 @@ def _perform_api_call(next_api_kwargs): agent._session_messages = messages agent._cleanup_task_resources(effective_task_id) agent._persist_session(messages, conversation_history) - return { + result = { "final_response": partial_response or None, "messages": messages, "api_calls": api_call_count, @@ -3945,6 +3976,7 @@ def _perform_api_call(next_api_kwargs): "partial": True, "error": "Response remained truncated after 4 continuation attempts", } + return _terminal_result_with_pending_steer(agent, result) if agent.api_mode in {"chat_completions", "bedrock_converse", "anthropic_messages"}: assistant_message = _trunc_msg @@ -4007,14 +4039,14 @@ def _perform_api_call(next_api_kwargs): # never reaches finalize_turn (#48879 class). close_interrupted_tool_sequence(messages, _final_response) agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "api_calls": api_call_count, "completed": False, "partial": True, "error": _final_response, - } + }) # If we have prior messages, roll back to last complete state if len(messages) > 1: @@ -4024,27 +4056,27 @@ def _perform_api_call(next_api_kwargs): agent._cleanup_task_resources(effective_task_id) agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": "Response truncated due to output length limit", "messages": rolled_back_messages, "api_calls": api_call_count, "completed": False, "partial": True, "error": "Response truncated due to output length limit" - } + }) else: # First message was truncated - mark as failed agent._flush_status_buffer() agent._vprint(f"{agent.log_prefix}❌ First response truncated - cannot recover", force=True) agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": "First response truncated due to output length limit", "messages": messages, "api_calls": api_call_count, "completed": False, "failed": True, "error": "First response truncated due to output length limit" - } + }) # Track actual token usage from response for context management if hasattr(response, 'usage') and response.usage: @@ -5249,7 +5281,7 @@ def _perform_api_call(next_api_kwargs): "(compression.enabled: false). Run /compress to compact manually, " "/new to start fresh, or switch to a larger-context model." ) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, @@ -5258,7 +5290,7 @@ def _perform_api_call(next_api_kwargs): "partial": True, "failed": True, "compaction_disabled": True, - } + }) # ── Anthropic Sonnet long-context tier gate ─────────── # Anthropic returns HTTP 429 "Extra usage is required for @@ -5567,7 +5599,7 @@ def _perform_api_call(next_api_kwargs): logger.error("%s413 compression failed after %d attempts.", agent.log_prefix, max_compression_attempts) agent._persist_session(messages, conversation_history) _final_response = f"Request payload too large: max compression attempts ({max_compression_attempts}) reached." - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, @@ -5576,7 +5608,7 @@ def _perform_api_call(next_api_kwargs): "partial": True, "failed": True, "compression_exhausted": True, - } + }) agent._buffer_status(f"⚠️ Request payload too large (413) — compression attempt {compression_attempts}/{max_compression_attempts}...") original_len = len(messages) @@ -5598,8 +5630,11 @@ def _perform_api_call(next_api_kwargs): # NOT auto-reset the session (#9893/#35809). compression_attempts -= 1 agent._persist_session(messages, conversation_history) - return _compression_deferred_result( - agent, messages, api_call_count + return _terminal_result_with_pending_steer( + agent, + _compression_deferred_result( + agent, messages, api_call_count + ), ) conversation_history = conversation_history_after_compression( agent, messages, conversation_history @@ -5639,7 +5674,7 @@ def _perform_api_call(next_api_kwargs): logger.error("%s413 payload too large. Cannot compress further.", agent.log_prefix) agent._persist_session(messages, conversation_history) _final_response = "Request payload too large (413). Cannot compress further." - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, @@ -5648,7 +5683,7 @@ def _perform_api_call(next_api_kwargs): "partial": True, "failed": True, "compression_exhausted": True, - } + }) # Check for context-length errors BEFORE generic 4xx handler. # The classifier detects context overflow from: explicit error @@ -5717,7 +5752,7 @@ def _perform_api_call(next_api_kwargs): logger.error("%sContext compression failed after %d attempts.", agent.log_prefix, max_compression_attempts) agent._persist_session(messages, conversation_history) _final_response = f"Context length exceeded: max compression attempts ({max_compression_attempts}) reached." - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, @@ -5726,7 +5761,7 @@ def _perform_api_call(next_api_kwargs): "partial": True, "failed": True, "compression_exhausted": True, - } + }) # Also compress the message history so the output-cap # retry does not just spin on max_tokens alone. The # compressor drops the middle window, freeing enough @@ -5744,8 +5779,11 @@ def _perform_api_call(next_api_kwargs): if messages is _overflow_input and compression_skipped_due_to_lock(agent): compression_attempts -= 1 agent._persist_session(messages, conversation_history) - return _compression_deferred_result( - agent, messages, api_call_count + return _terminal_result_with_pending_steer( + agent, + _compression_deferred_result( + agent, messages, api_call_count + ), ) conversation_history = conversation_history_after_compression( agent, messages, conversation_history @@ -5796,7 +5834,7 @@ def _perform_api_call(next_api_kwargs): "max_tokens exceeds the provider's output cap for this model. " "Lower model.max_tokens in config.yaml." ) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, @@ -5804,7 +5842,7 @@ def _perform_api_call(next_api_kwargs): "error": _final_response, "partial": True, "failed": True, - } + }) # Error is about the INPUT being too large. Only reduce # context_length when the provider explicitly reports the @@ -5871,7 +5909,7 @@ def _perform_api_call(next_api_kwargs): logger.error("%sContext compression failed after %d attempts.", agent.log_prefix, max_compression_attempts) agent._persist_session(messages, conversation_history) _final_response = f"Context length exceeded: max compression attempts ({max_compression_attempts}) reached." - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, @@ -5880,7 +5918,7 @@ def _perform_api_call(next_api_kwargs): "partial": True, "failed": True, "compression_exhausted": True, - } + }) agent._buffer_status(COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE.format(tokens=approx_tokens, attempt=compression_attempts, cap=max_compression_attempts)) original_len = len(messages) @@ -5904,8 +5942,11 @@ def _perform_api_call(next_api_kwargs): # NOT auto-reset the session (#9893/#35809). compression_attempts -= 1 agent._persist_session(messages, conversation_history) - return _compression_deferred_result( - agent, messages, api_call_count + return _terminal_result_with_pending_steer( + agent, + _compression_deferred_result( + agent, messages, api_call_count + ), ) conversation_history = conversation_history_after_compression( agent, messages, conversation_history @@ -5934,7 +5975,7 @@ def _perform_api_call(next_api_kwargs): logger.error("%sContext length exceeded: %s tokens. Cannot compress further.", agent.log_prefix, f"{new_tokens:,}") agent._persist_session(messages, conversation_history) _final_response = f"Context length exceeded ({new_tokens:,} tokens). Cannot compress further." - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "completed": False, @@ -5943,7 +5984,7 @@ def _perform_api_call(next_api_kwargs): "partial": True, "failed": True, "compression_exhausted": True, - } + }) # Check for non-retryable client errors. The classifier # already accounts for 413, 429, 529 (transient), context @@ -6220,34 +6261,40 @@ def _perform_api_call(next_api_kwargs): f"Provider message: {_nonretryable_summary}\n\n" f"{_CONTENT_POLICY_RECOVERY_HINT}" ) - return _content_policy_blocked_result( - messages, - api_call_count, - final_response=_policy_response, - error_detail=_nonretryable_summary, + return _terminal_result_with_pending_steer( + agent, + _content_policy_blocked_result( + messages, + api_call_count, + final_response=_policy_response, + error_detail=_nonretryable_summary, + ), ) # Billing walls are the common non-retryable abort: enrich # the result with the same structured recovery descriptor as # the max-retries path so every surface (CLI, TUI, desktop) # renders one consistent billing signal. if classified.reason == FailoverReason.billing: - return _billing_failure_result( - classified=classified, - summary=_nonretryable_summary, - messages=messages, - api_call_count=api_call_count, - provider=_provider, - base_url=_base, - model=_model, + return _terminal_result_with_pending_steer( + agent, + _billing_failure_result( + classified=classified, + summary=_nonretryable_summary, + messages=messages, + api_call_count=api_call_count, + provider=_provider, + base_url=_base, + model=_model, + ), ) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _nonretryable_summary, "messages": messages, "api_calls": api_call_count, "completed": False, "failed": True, "error": _nonretryable_summary, - } + }) if retry_count >= max_retries: # Before falling back, try rebuilding the primary @@ -6451,7 +6498,7 @@ def _perform_api_call(next_api_kwargs): "execute_code with Python's open() for large " "files, or to write in smaller sections." ) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "api_calls": api_call_count, @@ -6470,7 +6517,7 @@ def _perform_api_call(next_api_kwargs): # Present only for billing walls: structured recovery # descriptor (provider, billing_url, is_nous, message). "billing_block": _billing_block, - } + }) # For rate limits, respect the Retry-After header if present _retry_after = None @@ -6777,14 +6824,14 @@ def _perform_api_call(next_api_kwargs): agent._cleanup_task_resources(effective_task_id) agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": "Incomplete REASONING_SCRATCHPAD after 2 retries", "messages": rolled_back_messages, "api_calls": api_call_count, "completed": False, "partial": True, "error": "Incomplete REASONING_SCRATCHPAD after 2 retries" - } + }) # Reset incomplete scratchpad counter on clean response agent._incomplete_scratchpad_retries = 0 @@ -6919,14 +6966,14 @@ def _perform_api_call(next_api_kwargs): agent._codex_incomplete_retries = 0 agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": "Codex response remained incomplete after 3 continuation attempts", "messages": messages, "api_calls": api_call_count, "completed": False, "partial": True, "error": "Codex response remained incomplete after 3 continuation attempts", - } + }) elif hasattr(agent, "_codex_incomplete_retries"): agent._codex_incomplete_retries = 0 @@ -7008,14 +7055,14 @@ def _perform_api_call(next_api_kwargs): # turn is not tool→user for strict providers. close_interrupted_tool_sequence(messages, _final_response) agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "api_calls": api_call_count, "completed": False, "partial": True, "error": _final_response - } + }) assistant_msg = agent._build_assistant_message(assistant_message, finish_reason) append_message(messages, assistant_msg) @@ -7092,14 +7139,14 @@ def _perform_api_call(next_api_kwargs): # exhaustion — this path never reaches finalize_turn. close_interrupted_tool_sequence(messages, _final_response) agent._persist_session(messages, conversation_history) - return { + return _terminal_result_with_pending_steer(agent, { "final_response": _final_response, "messages": messages, "api_calls": api_call_count, "completed": False, "partial": True, "error": _final_response, - } + }) # Track retries for invalid JSON arguments agent._invalid_json_retries += 1 diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index fbdf19cd91b4..a1e40d539390 100644 --- a/agent/turn_finalizer.py +++ b/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 diff --git a/cli.py b/cli.py index b53906386ac3..9b7d397a6856 100644 --- a/cli.py +++ b/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) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 6bae26aa052b..5c70ceeac619 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -6042,6 +6042,18 @@ 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 + ), + ) + async def handle_message(self, event: MessageEvent) -> None: """ Process an incoming message. @@ -6067,7 +6079,8 @@ 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( + session_key = self.session_key_for_source(event.source) + derived_state_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), @@ -6076,11 +6089,11 @@ async def handle_message(self, event: MessageEvent) -> None: expected_session_key = str( (event.metadata or {}).get("gateway_session_key") or "" ).strip() - if expected_session_key and session_key != expected_session_key: + if expected_session_key and derived_state_key != expected_session_key: logger.warning( "Dropping internally routed event: expected session=%s derived=%s", expected_session_key, - session_key, + derived_state_key, ) return diff --git a/gateway/run.py b/gateway/run.py index 84604b303b7b..f42fe45f4fea 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -8972,38 +8972,63 @@ def _queue_during_drain_enabled( # it up. Clearing happens on /new and /reset via # _handle_reset_command. - def _enqueue_fifo(self, session_key: str, queued_event: "MessageEvent", adapter: Any) -> None: - """Append a /queue event to the FIFO chain for a session.""" + @staticmethod + def _adapter_key_for_source( + adapter: Any, + source: Optional[SessionSource], + *, + fallback: str, + ) -> str: + """Resolve the transport owner's physical slot key, never durable state.""" + resolver = getattr(adapter, "session_key_for_source", None) + if callable(resolver) and source is not None: + try: + adapter_key = resolver(source) + if isinstance(adapter_key, str) and adapter_key: + return adapter_key + except Exception: + logger.debug("Adapter slot-key resolution failed", exc_info=True) + return fallback + + def _enqueue_fifo( + self, + state_key: str, + queued_event: "MessageEvent", + adapter: Any, + *, + adapter_key: Optional[str] = None, + ) -> None: + """Append an event across durable overflow and adapter-local slot keys.""" if adapter is None: return pending_slot = getattr(adapter, "_pending_messages", None) if pending_slot is None: return - if session_key in pending_slot: - self._session_state(session_key).conversation.queued_events.append( + slot_key = adapter_key or state_key + if slot_key in pending_slot: + self._session_state(state_key).conversation.queued_events.append( queued_event ) else: - pending_slot[session_key] = queued_event + pending_slot[slot_key] = queued_event def _promote_queued_event( self, - session_key: str, + state_key: str, adapter: Any, pending_event: Optional["MessageEvent"], + *, + adapter_key: Optional[str] = None, ) -> Optional["MessageEvent"]: - """Promote the next overflow item after the slot was drained. - - Called at the drain site after _dequeue_pending_event consumed - (or failed to consume) the slot. If there's an overflow item: - - When pending_event is None (slot was empty), return the - overflow head as the new pending_event. - - When pending_event already exists (slot was populated by an - interrupt follow-up or similar), stage the overflow head in - the slot so the NEXT recursion picks it up. - Returns the (possibly updated) pending_event for drain to use. + """Promote durable overflow after the adapter-local slot was drained. + + If there's an overflow item: + - When pending_event is None (slot was empty), return the overflow + head as the new pending_event. + - When pending_event already exists, stage the overflow head in the + physical adapter slot so the next recursion picks it up. """ - _q_state = self._peek_session_state(session_key) + _q_state = self._peek_session_state(state_key) overflow = _q_state.conversation.queued_events if _q_state else None if not overflow: return pending_event @@ -9011,17 +9036,42 @@ def _promote_queued_event( if pending_event is None: return next_queued if adapter is not None and hasattr(adapter, "_pending_messages"): - adapter._pending_messages[session_key] = next_queued + adapter._pending_messages[adapter_key or state_key] = next_queued else: # No adapter — push back so we don't silently drop the item. overflow.insert(0, next_queued) return pending_event - def _queue_depth(self, session_key: str, *, adapter: Any = None) -> int: - """Total pending /queue items for a session — slot + overflow.""" - _q_state = self._peek_session_state(session_key) + def _requeue_fifo_head( + self, + state_key: str, + adapter: Any, + adapter_key: str, + pending_event: "MessageEvent", + ) -> None: + """Restore a capped event ahead of a slot item without losing either.""" + pending_slot = getattr(adapter, "_pending_messages", None) + if not isinstance(pending_slot, dict): + return + displaced = pending_slot.get(adapter_key) + pending_slot[adapter_key] = pending_event + if displaced is not None and displaced is not pending_event: + self._session_state(state_key).conversation.queued_events.insert( + 0, displaced + ) + + def _queue_depth( + self, + state_key: str, + *, + adapter: Any = None, + adapter_key: Optional[str] = None, + ) -> int: + """Total FIFO depth across a physical slot and durable overflow.""" + _q_state = self._peek_session_state(state_key) depth = len(_q_state.conversation.queued_events) if _q_state else 0 - if adapter is not None and session_key in getattr(adapter, "_pending_messages", {}): + slot_key = adapter_key or state_key + if adapter is not None and slot_key in getattr(adapter, "_pending_messages", {}): depth += 1 return depth @@ -9036,22 +9086,33 @@ def _is_goal_continuation_event(event_or_text: Any) -> bool: text = getattr(event_or_text, "text", event_or_text) or "" return str(text).startswith("[Continuing toward your standing goal]\nGoal:") - def _clear_goal_pending_continuations(self, session_key: str, adapter: Any) -> int: + def _clear_goal_pending_continuations( + self, + state_key: str, + adapter: Any, + *, + source: Optional[SessionSource] = None, + ) -> int: """Remove queued synthetic /goal continuations for one session. User-issued /goal pause/clear can race with a continuation already - queued by the judge. Remove only synthetic goal continuations while + queued by the judge. Remove only synthetic goal continuations while preserving normal /queue and user follow-up events. """ removed = 0 + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=state_key, + ) pending_slot = getattr(adapter, "_pending_messages", None) if adapter is not None else None if isinstance(pending_slot, dict): - pending_event = pending_slot.get(session_key) + pending_event = pending_slot.get(adapter_key) if self._is_goal_continuation_event(pending_event): - pending_slot.pop(session_key, None) + pending_slot.pop(adapter_key, None) removed += 1 - _q_state = self._peek_session_state(session_key) + _q_state = self._peek_session_state(state_key) overflow = _q_state.conversation.queued_events if _q_state else [] if overflow: kept = [] @@ -10052,10 +10113,15 @@ def _lookup_session_id_under_store_lock(session_store, session_key: str): # still small enough to never threaten memory. _BUSY_QUEUE_MAX_PENDING = 32 - def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) -> None: + def _queue_or_replace_pending_event(self, state_key: str, event: MessageEvent) -> None: adapter = self._adapter_for_source(event.source) if not adapter: return + adapter_key = self._adapter_key_for_source( + adapter, + event.source, + fallback=state_key, + ) # #28503 — Previously this called ``merge_pending_message_event`` # with the default ``merge_text=False``, which silently OVERWROTE # the single pending slot when consecutive text messages arrived @@ -10065,7 +10131,7 @@ def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) # the head slot via ``merge_pending_message_event`` (album # semantics); everything else appends to the overflow tail. pending_slot = getattr(adapter, "_pending_messages", None) - existing = pending_slot.get(session_key) if isinstance(pending_slot, dict) else None + existing = pending_slot.get(adapter_key) if isinstance(pending_slot, dict) else None security_metadata_keys = ( "hermes_plugin_id", "hermes_plugin_injection", @@ -10092,21 +10158,30 @@ def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) # Preserve photo-burst / media-merge semantics for the head slot. merge_pending_message_event( adapter._pending_messages, - session_key, + adapter_key, event, merge_text=event.message_type == MessageType.TEXT, ) return - if self._queue_depth(session_key, adapter=adapter) >= self._BUSY_QUEUE_MAX_PENDING: + if self._queue_depth( + state_key, + adapter=adapter, + adapter_key=adapter_key, + ) >= self._BUSY_QUEUE_MAX_PENDING: logger.warning( "Dropping busy-mode follow-up for session %s — pending queue at cap (%d).", - session_key, + state_key, self._BUSY_QUEUE_MAX_PENDING, ) return - self._enqueue_fifo(session_key, event, adapter) + self._enqueue_fifo( + state_key, + event, + adapter, + adapter_key=adapter_key, + ) async def _prepare_busy_steer_text(self, event: MessageEvent) -> str: """Return steerable text for a busy follow-up, transcribing voice first. @@ -16373,8 +16448,24 @@ async def _busy_queue_command(self, event: MessageEvent, quick_key: str, source) internal=event.internal, timestamp=event.timestamp, ) - self._enqueue_fifo(quick_key, queued_event, adapter) - depth = self._queue_depth(quick_key, adapter=self._adapter_for_source(source)) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=quick_key, + ) + self._enqueue_fifo( + quick_key, + queued_event, + adapter, + adapter_key=adapter_key, + ) + else: + adapter_key = quick_key + depth = self._queue_depth( + quick_key, + adapter=self._adapter_for_source(source), + adapter_key=adapter_key, + ) if depth <= 1: return "Queued for the next turn." return f"Queued for the next turn. ({depth} queued)" @@ -16388,21 +16479,37 @@ async def _busy_steer_command(self, event: MessageEvent, quick_key: str, source) steer_text = event.get_command_args().strip() if not steer_text: return "Usage: /steer " + + def queue_steer_fallback() -> bool: + adapter = self._adapter_for_source(source) + if not adapter: + return False + queued_event = MessageEvent( + text=steer_text, + message_type=MessageType.TEXT, + source=event.source, + message_id=event.message_id, + channel_prompt=event.channel_prompt, + channel_context=event.channel_context, + ) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=quick_key, + ) + self._enqueue_fifo( + quick_key, + queued_event, + adapter, + adapter_key=adapter_key, + ) + return True + _steer_state = self._peek_session_state(quick_key) running_agent = _steer_state.turn.agent if _steer_state else None if running_agent is _AGENT_PENDING_SENTINEL: # Agent hasn't started yet — queue as turn-boundary fallback. - adapter = self._adapter_for_source(source) - if adapter: - queued_event = MessageEvent( - text=steer_text, - message_type=MessageType.TEXT, - source=event.source, - message_id=event.message_id, - channel_prompt=event.channel_prompt, - channel_context=event.channel_context, - ) - self._enqueue_fifo(quick_key, queued_event, adapter) + queue_steer_fallback() return "Agent still starting — /steer queued for the next turn." if running_agent and hasattr(running_agent, "steer"): try: @@ -16413,19 +16520,13 @@ async def _busy_steer_command(self, event: MessageEvent, quick_key: str, source) if accepted: preview = steer_text[:60] + ("..." if len(steer_text) > 60 else "") return f"⏩ Steer queued — arrives after the next tool call: '{preview}'" - return "Steer rejected (empty payload)." + # The active turn atomically sealed before steer acquired its lock. + # This is valid non-acceptance, not an empty payload: preserve it as + # FIFO next-turn work behind every older adapter/overflow event. + queue_steer_fallback() + return "Turn already completed — /steer queued for the next turn." # Running agent is missing or lacks steer() — fall back to queue. - adapter = self._adapter_for_source(source) - if adapter: - queued_event = MessageEvent( - text=steer_text, - message_type=MessageType.TEXT, - source=event.source, - message_id=event.message_id, - channel_prompt=event.channel_prompt, - channel_context=event.channel_context, - ) - self._enqueue_fifo(quick_key, queued_event, adapter) + queue_steer_fallback() return "No active agent — /steer queued for the next turn." async def _busy_goal_command(self, event: MessageEvent, quick_key: str, source): @@ -17055,7 +17156,16 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: logger.debug("PRIORITY photo follow-up for session %s — queueing without interrupt", _quick_key) adapter = self._adapter_for_source(source) if adapter: - merge_pending_message_event(adapter._pending_messages, _quick_key, event) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=_quick_key, + ) + merge_pending_message_event( + adapter._pending_messages, + adapter_key, + event, + ) return None effective_busy_input_mode = self._effective_busy_input_mode(source) @@ -17079,11 +17189,16 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: adapter = self._adapter_for_source(source) if adapter: if effective_busy_input_mode == "queue": - self._enqueue_fifo(_quick_key, event, adapter) + self._queue_or_replace_pending_event(_quick_key, event) else: + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=_quick_key, + ) merge_pending_message_event( adapter._pending_messages, - _quick_key, + adapter_key, event, merge_text=True, ) @@ -17102,9 +17217,14 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: # agent starts. adapter = self._adapter_for_source(source) if adapter: + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=_quick_key, + ) merge_pending_message_event( adapter._pending_messages, - _quick_key, + adapter_key, event, merge_text=True, ) @@ -20134,6 +20254,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g # same run that registered them. self._bind_adapter_run_generation( self._adapter_for_source(source), + source, session_key, run_generation, ) @@ -21388,7 +21509,17 @@ async def _poll_loop(): message_id=None, channel_prompt=None, ) - self._enqueue_fifo(quick_key, hb_event, adapter) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=quick_key, + ) + self._enqueue_fifo( + quick_key, + hb_event, + adapter, + adapter_key=adapter_key, + ) except Exception as exc: logger.debug("heartbeat poll for %s failed: %s", quick_key, exc) @@ -21450,18 +21581,24 @@ async def _deliver() -> None: logger.warning("goal continuation: status send failed: %s", exc, exc_info=True) try: - session_key = self._session_key_for_source(source) + state_key = self._session_key_for_source(source) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=state_key, + ) except Exception: - session_key = None + state_key = None + adapter_key = None - if session_key and hasattr(adapter, "register_post_delivery_callback"): + if adapter_key and hasattr(adapter, "register_post_delivery_callback"): try: generation = None - active = getattr(adapter, "_active_sessions", {}).get(session_key) + active = getattr(adapter, "_active_sessions", {}).get(adapter_key) if active is not None: generation = getattr(active, "_hermes_run_generation", None) adapter.register_post_delivery_callback( - session_key, + adapter_key, _deliver, generation=generation, ) @@ -21556,6 +21693,11 @@ async def _post_turn_goal_continuation( adapter = self._adapter_for_source(source) _quick_key = self._session_key_for_source(source) if adapter and _quick_key: + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=_quick_key, + ) cont_event = MessageEvent( text=prompt, message_type=MessageType.TEXT, @@ -21563,7 +21705,12 @@ async def _post_turn_goal_continuation( message_id=None, channel_prompt=None, ) - self._enqueue_fifo(_quick_key, cont_event, adapter) + self._enqueue_fifo( + _quick_key, + cont_event, + adapter, + adapter_key=adapter_key, + ) except Exception as exc: logger.debug("goal continuation: enqueue failed: %s", exc) @@ -26616,14 +26763,20 @@ def _is_session_run_current(self, session_key: str, generation: int) -> bool: def _bind_adapter_run_generation( self, adapter: Any, - session_key: str, + source: Optional[SessionSource], + state_key: str, generation: int | None, ) -> None: """Bind a gateway run generation to the adapter's active-session event.""" - if not adapter or not session_key or generation is None: + if not adapter or not state_key or generation is None: return try: - interrupt_event = getattr(adapter, "_active_sessions", {}).get(session_key) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=state_key, + ) + interrupt_event = getattr(adapter, "_active_sessions", {}).get(adapter_key) if interrupt_event is not None: setattr(interrupt_event, "_hermes_run_generation", int(generation)) except Exception: @@ -26677,6 +26830,11 @@ async def _interrupt_and_clear_session( daemon=True, ).start() adapter = self._adapter_for_source(source) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=session_key, + ) interrupt_session_activity = getattr( type(adapter), "interrupt_session_activity", None ) @@ -26692,12 +26850,12 @@ async def _interrupt_and_clear_session( accepts_metadata = False if accepts_metadata: await adapter.interrupt_session_activity( - session_key, source.chat_id, metadata=metadata + adapter_key, source.chat_id, metadata=metadata ) else: - await adapter.interrupt_session_activity(session_key, source.chat_id) + await adapter.interrupt_session_activity(adapter_key, source.chat_id) if adapter and hasattr(adapter, "get_pending_message"): - adapter.get_pending_message(session_key) # consume and discard + adapter.get_pending_message(adapter_key) # consume and discard if _iac_state is not None: _iac_state.persistent.pending_command_text = None if release_running_state: @@ -28726,11 +28884,15 @@ async def monitor_for_interrupt(): _adapter = self._adapter_for_source(source) if not _adapter: continue - # Check if adapter has a pending interrupt for this session. - # Must use session_key (build_session_key output) — NOT - # source.chat_id — because the adapter stores interrupt events - # under the full session key. - if hasattr(_adapter, 'has_pending_interrupt') and _adapter.has_pending_interrupt(session_key): + # Adapter-owned interrupt/pending slots use the transport's + # physical key, which can differ from durable session state + # under profile multiplexing. + _adapter_key = self._adapter_key_for_source( + _adapter, + source, + fallback=session_key, + ) + if hasattr(_adapter, 'has_pending_interrupt') and _adapter.has_pending_interrupt(_adapter_key): agent = agent_holder[0] if agent: # Peek at the pending message text WITHOUT consuming it. @@ -28741,7 +28903,7 @@ async def monitor_for_interrupt(): # before checking _interrupt_requested, and the message # is lost — neither the interrupt path nor the dequeue # path finds it. - _peek_event = _adapter._pending_messages.get(session_key) + _peek_event = _adapter._pending_messages.get(_adapter_key) pending_text = None if _peek_event is not None: pending_text = _peek_event.text or "" @@ -29027,10 +29189,15 @@ def _run_sync_with_timeout_lifecycle(): if not _interrupt_detected.is_set() and session_key: _backup_adapter = self._adapter_for_source(source) _backup_agent = agent_holder[0] + _backup_adapter_key = self._adapter_key_for_source( + _backup_adapter, + source, + fallback=session_key, + ) if (_backup_adapter and _backup_agent and hasattr(_backup_adapter, 'has_pending_interrupt') - and _backup_adapter.has_pending_interrupt(session_key)): - _bp_event = _backup_adapter._pending_messages.get(session_key) + and _backup_adapter.has_pending_interrupt(_backup_adapter_key)): + _bp_event = _backup_adapter._pending_messages.get(_backup_adapter_key) _bp_text = _bp_event.text if _bp_event else None if _bp_event is not None: _bp_media_urls = getattr(_bp_event, "media_urls", None) or [] @@ -29129,10 +29296,15 @@ def _run_sync_with_timeout_lifecycle(): if not _interrupt_detected.is_set() and session_key: _backup_adapter = self._adapter_for_source(source) _backup_agent = agent_holder[0] + _backup_adapter_key = self._adapter_key_for_source( + _backup_adapter, + source, + fallback=session_key, + ) if (_backup_adapter and _backup_agent and hasattr(_backup_adapter, 'has_pending_interrupt') - and _backup_adapter.has_pending_interrupt(session_key)): - _bp_event = _backup_adapter._pending_messages.get(session_key) + and _backup_adapter.has_pending_interrupt(_backup_adapter_key)): + _bp_event = _backup_adapter._pending_messages.get(_backup_adapter_key) _bp_text = _bp_event.text if _bp_event else None if _bp_event is not None: _bp_media_urls = getattr(_bp_event, "media_urls", None) or [] @@ -29262,6 +29434,11 @@ def _run_sync_with_timeout_lifecycle(): # Check if we were interrupted OR have a queued message (/queue). result = result_holder[0] adapter = self._adapter_for_source(source) + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=session_key, + ) # Finalize the streaming-TTS consumer (#60671). # @@ -29290,19 +29467,25 @@ def _run_sync_with_timeout_lifecycle(): if callable(_mark_turn): _mark_turn(session_key, run_generation) - # Get pending message from adapter. - # Use session_key (not source.chat_id) to match adapter's storage keys. + # Get pending transport work from the physical adapter key while + # retaining session_key for durable overflow/state ownership. pending_event = None pending = None + pending_is_terminal_steer = False if result and adapter and session_key: - pending_event = _dequeue_pending_event(adapter, session_key) - # /queue overflow: after consuming the adapter's "next-up" - # slot, promote the next queued event into it so the - # recursive run's drain will see it. This keeps the slot - # occupied for the full FIFO chain, which (a) preserves - # order, and (b) causes any mid-chain /queue to correctly - # route to overflow rather than jumping the queue. - pending_event = self._promote_queued_event(session_key, adapter, pending_event) + pending_event = _dequeue_pending_event(adapter, adapter_key) + # /queue overflow: after consuming the adapter's physical + # "next-up" slot, promote durable state under session_key. + pending_event = self._promote_queued_event( + session_key, + adapter, + pending_event, + adapter_key=adapter_key, + ) + pending_is_terminal_steer = bool( + pending_event is not None + and getattr(pending_event, "_gateway_terminal_steer", False) + ) if result.get("interrupted") and not pending_event and result.get("interrupt_message"): interrupt_message = result.get("interrupt_message") if _is_control_interrupt_message(interrupt_message): @@ -29340,20 +29523,51 @@ def _run_sync_with_timeout_lifecycle(): # Leftover /steer: if a steer arrived after the last tool batch # (e.g. during the final API call), the agent couldn't inject it - # and returned it in result["pending_steer"]. Deliver it as the - # next user turn so it isn't silently dropped. - if result and not pending and not pending_event: - _leftover_steer = result.get("pending_steer") - if _leftover_steer: + # and returned it in result["pending_steer"]. Take ownership of + # that terminal result exactly once. Older transport work keeps + # the head; the steer is appended to the same FIFO without + # setting the adapter's interrupt event. + _leftover_steer = result.pop("pending_steer", None) if result else None + if isinstance(_leftover_steer, str) and _leftover_steer.strip(): + if pending or pending_event: + _steer_event = MessageEvent( + text=_leftover_steer, + message_type=MessageType.TEXT, + source=source, + internal=True, + ) + # Accepted steers already passed inbound preprocessing on + # their original command path. Mark this synthetic event + # so the later FIFO drain preserves the exact text. + setattr(_steer_event, "_gateway_terminal_steer", True) + self._enqueue_fifo( + session_key, + _steer_event, + adapter, + adapter_key=adapter_key, + ) + logger.debug( + "Queued leftover /steer behind existing work: '%s...'", + _leftover_steer[:40], + ) + else: pending = _leftover_steer - logger.debug("Delivering leftover /steer as next turn: '%s...'", pending[:40]) + pending_is_terminal_steer = True + logger.debug( + "Delivering leftover /steer as next turn: '%s...'", + pending[:40], + ) # Safety net: if the pending text is a slash command (e.g. "/stop", # "/new"), discard it — commands should never be passed to the agent # as user input. The primary fix is in base.py (commands bypass the # active-session guard), but this catches edge cases where command # text leaks through the interrupt_message fallback. - if pending and pending.strip().startswith("/"): + if ( + pending + and not pending_is_terminal_steer + and pending.strip().startswith("/") + ): _pending_parts = pending.strip().split(None, 1) _pending_cmd_word = _pending_parts[0][1:].lower() if _pending_parts else "" if _pending_cmd_word: @@ -29385,8 +29599,13 @@ def _run_sync_with_timeout_lifecycle(): # Clear the adapter's interrupt event so the next _run_agent call # doesn't immediately re-trigger the interrupt before the new agent # even makes its first API call (this was causing an infinite loop). - if adapter and hasattr(adapter, '_active_sessions') and session_key and session_key in adapter._active_sessions: - adapter._active_sessions[session_key].clear() + if ( + adapter + and hasattr(adapter, "_active_sessions") + and adapter_key + and adapter_key in adapter._active_sessions + ): + adapter._active_sessions[adapter_key].clear() # Cap recursion depth to prevent resource exhaustion when the # user sends multiple messages while the agent keeps failing. (#816) @@ -29396,11 +29615,23 @@ def _run_sync_with_timeout_lifecycle(): "queueing message instead of recursing.", _interrupt_depth, session_key, ) - adapter = self._adapter_for_source(source) + adapter = self._adapter_for_source( + getattr(pending_event, "source", None) or source + ) if adapter and pending_event: - merge_pending_message_event(adapter._pending_messages, session_key, pending_event) + capped_adapter_key = self._adapter_key_for_source( + adapter, + getattr(pending_event, "source", None) or source, + fallback=adapter_key, + ) + self._requeue_fifo_head( + session_key, + adapter, + capped_adapter_key, + pending_event, + ) elif adapter and hasattr(adapter, 'queue_message'): - adapter.queue_message(session_key, pending) + adapter.queue_message(adapter_key, pending) return result_holder[0] or {"final_response": response, "messages": history} was_interrupted = result.get("interrupted") @@ -29532,12 +29763,15 @@ def _run_sync_with_timeout_lifecycle(): session_key or "?", exc_info=True, ) - next_message = await self._prepare_profile_scoped_inbound_message_text( - event=pending_event, - source=next_source, - history=updated_history, - session_key=next_session_key, - ) + if getattr(pending_event, "_gateway_terminal_steer", False): + next_message = pending_event.text + else: + next_message = await self._prepare_profile_scoped_inbound_message_text( + event=pending_event, + source=next_source, + history=updated_history, + session_key=next_session_key, + ) if next_message is None: return result next_message_id = self._reply_anchor_for_event(pending_event) diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index cd54164bd0d9..01bb6c5bd7ac 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -589,9 +589,18 @@ async def _handle_status_command(self, event: MessageEvent) -> str: agent = self._running_agents.get(session_key) is_running = agent is not None and agent is not _AGENT_PENDING_SENTINEL - # Count pending /queue follow-ups (slot + overflow). - adapter = self.adapters.get(source.platform) if source else None - queue_depth = self._queue_depth(session_key, adapter=adapter) + # Count pending /queue follow-ups across physical slot + durable overflow. + adapter = self._adapter_for_source(source) if source else None + adapter_key = self._adapter_key_for_source( + adapter, + source, + fallback=session_key, + ) + queue_depth = self._queue_depth( + session_key, + adapter=adapter, + adapter_key=adapter_key, + ) def _clean_str(value: Any) -> str: return value.strip() if isinstance(value, str) and value.strip() else "" @@ -2717,10 +2726,14 @@ async def _handle_goal_command(self, event: "MessageEvent") -> str: if state is None: return t("gateway.goal.no_goal_set") try: - adapter = self.adapters.get(event.source.platform) if event.source else None + adapter = self._adapter_for_source(event.source) if event.source else None _quick_key = self._session_key_for_source(event.source) if event.source else None if adapter and _quick_key: - self._clear_goal_pending_continuations(_quick_key, adapter) + self._clear_goal_pending_continuations( + _quick_key, + adapter, + source=event.source, + ) except Exception as exc: logger.debug("goal pause: pending continuation cleanup failed: %s", exc) return t("gateway.goal.paused", goal=state.goal) @@ -2756,10 +2769,14 @@ async def _handle_goal_command(self, event: "MessageEvent") -> str: had = mgr.has_goal() mgr.clear() try: - adapter = self.adapters.get(event.source.platform) if event.source else None + adapter = self._adapter_for_source(event.source) if event.source else None _quick_key = self._session_key_for_source(event.source) if event.source else None if adapter and _quick_key: - self._clear_goal_pending_continuations(_quick_key, adapter) + self._clear_goal_pending_continuations( + _quick_key, + adapter, + source=event.source, + ) except Exception as exc: logger.debug("goal clear: pending continuation cleanup failed: %s", exc) return t("gateway.goal_cleared") if had else t("gateway.no_active_goal") @@ -2857,10 +2874,15 @@ async def _handle_goal_command(self, event: "MessageEvent") -> str: # Queue the goal text as an immediate first turn so the agent # starts making progress. The post-turn hook takes over after. - adapter = self.adapters.get(event.source.platform) if event.source else None + adapter = self._adapter_for_source(event.source) if event.source else None _quick_key = self._session_key_for_source(event.source) if event.source else None if adapter and _quick_key: try: + adapter_key = self._adapter_key_for_source( + adapter, + event.source, + fallback=_quick_key, + ) kickoff_event = MessageEvent( text=state.goal, message_type=MessageType.TEXT, @@ -2868,7 +2890,12 @@ async def _handle_goal_command(self, event: "MessageEvent") -> str: message_id=event.message_id, channel_prompt=event.channel_prompt, ) - self._enqueue_fifo(_quick_key, kickoff_event, adapter) + self._enqueue_fifo( + _quick_key, + kickoff_event, + adapter, + adapter_key=adapter_key, + ) except Exception as exc: logger.debug("goal kickoff enqueue failed: %s", exc) diff --git a/run_agent.py b/run_agent.py index f5c4de1274cf..e1ae7390ea10 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3430,6 +3430,8 @@ def clear_interrupt(self, *, preserve_redirect: bool = False) -> bool: if _steer_lock is not None: with _steer_lock: self._pending_steer = None + if not preserve_redirect: + self._steer_acceptance_generation = None return True def steer(self, text: str) -> bool: @@ -3448,7 +3450,8 @@ def steer(self, text: str) -> bool: text: The user text to inject. Empty strings are ignored. Returns: - True if the steer was accepted, False if the text was empty. + True if the steer was accepted by the active turn, False if the + text was empty or that turn already sealed its terminal result. """ if not text or not text.strip(): return False @@ -3458,16 +3461,48 @@ def steer(self, text: str) -> bool: # Test stubs that built AIAgent via object.__new__ skip __init__. # Fall back to direct attribute set; no concurrent callers expected # in those stubs. + if getattr(self, "_steer_acceptance_generation", 0) is None: + return False existing = getattr(self, "_pending_steer", None) self._pending_steer = (existing + "\n" + cleaned) if existing else cleaned return True with _lock: + if getattr(self, "_steer_acceptance_generation", 0) is None: + return False if self._pending_steer: self._pending_steer = self._pending_steer + "\n" + cleaned else: self._pending_steer = cleaned return True + def _begin_steer_acceptance(self) -> int: + """Open a new turn generation for thread-safe steer acceptance.""" + _lock = getattr(self, "_pending_steer_lock", None) + if _lock is None: + generation = int(getattr(self, "_steer_generation_counter", 0)) + 1 + self._steer_generation_counter = generation + self._steer_acceptance_generation = generation + return generation + with _lock: + generation = int(getattr(self, "_steer_generation_counter", 0)) + 1 + self._steer_generation_counter = generation + self._steer_acceptance_generation = generation + return generation + + def _seal_pending_steer(self) -> Optional[str]: + """Atomically close steer acceptance and drain this turn's pending text.""" + _lock = getattr(self, "_pending_steer_lock", None) + if _lock is None: + text = getattr(self, "_pending_steer", None) + self._pending_steer = None + self._steer_acceptance_generation = None + return text + with _lock: + text = self._pending_steer + self._pending_steer = None + self._steer_acceptance_generation = None + return text + def redirect(self, text: str) -> bool: """Redirect the active turn without converting it into a new task. diff --git a/tests/cli/test_cli_init.py b/tests/cli/test_cli_init.py index cca40f831e76..70b250f0a7cc 100644 --- a/tests/cli/test_cli_init.py +++ b/tests/cli/test_cli_init.py @@ -137,6 +137,19 @@ def test_queue_command_works_while_busy(self): cli.process_command("/queue follow up") assert cli._pending_input.get_nowait() == "follow up" + def test_explicit_steer_rejection_queues_behind_older_next_turn_work(self): + """A terminal-sealed turn must not lose a non-empty explicit /steer.""" + cli = _make_cli() + cli._agent_running = True + cli.agent = SimpleNamespace(steer=lambda _text: False) + cli._pending_input.put("older queued work") + + cli.process_command("/steer preserve this message") + + assert cli._pending_input.get_nowait() == "older queued work" + assert cli._pending_input.get_nowait() == "preserve this message" + assert cli._pending_input.empty() + diff --git a/tests/gateway/test_busy_session_ack.py b/tests/gateway/test_busy_session_ack.py index c9b5442fc5a1..87b4febf141c 100644 --- a/tests/gateway/test_busy_session_ack.py +++ b/tests/gateway/test_busy_session_ack.py @@ -294,6 +294,46 @@ async def test_steer_mode_falls_back_to_queue_when_agent_rejects(self): assert "Queued for the next turn" in content assert "Steered" not in content + @pytest.mark.asyncio + async def test_explicit_steer_rejection_queues_behind_existing_fifo_head(self): + """A terminal-sealed agent rejects /steer; the command becomes next-turn work.""" + runner, _sentinel = _make_runner() + runner._queued_events = {} + adapter = _make_adapter() + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="explicit-steer-boundary", + chat_type="dm", + user_id="user1", + ) + state_key = build_session_key(source) + runner.adapters[source.platform] = adapter + older = MessageEvent( + text="older queued work", + message_type=MessageType.TEXT, + source=source, + message_id="older", + ) + adapter._pending_messages[state_key] = older + agent = MagicMock() + agent.steer.return_value = False + runner._running_agents[state_key] = agent + command = MessageEvent( + text="/steer run this next", + message_type=MessageType.TEXT, + source=source, + message_id="steer", + ) + + response = await runner._busy_steer_command(command, state_key, source) + + agent.steer.assert_called_once_with("run this next") + assert "queued for the next turn" in response.lower() + assert adapter._pending_messages[state_key] is older + assert [event.text for event in runner._queued_events[state_key]] == [ + "run this next" + ] + @pytest.mark.asyncio async def test_steer_mode_falls_back_to_queue_when_agent_pending(self): """If agent is still starting (sentinel), steer mode falls back to queue.""" diff --git a/tests/gateway/test_queue_consumption.py b/tests/gateway/test_queue_consumption.py index ad258b00233e..c3167630f36d 100644 --- a/tests/gateway/test_queue_consumption.py +++ b/tests/gateway/test_queue_consumption.py @@ -17,6 +17,7 @@ PlatformConfig, Platform, ) +from gateway.session import SessionSource # --------------------------------------------------------------------------- @@ -190,9 +191,13 @@ def _make_runner_and_adapter(self): return runner, adapter def _text_event(self, text: str) -> MessageEvent: - # profile=None: a MagicMock auto-attribute reads as a truthy stamped - # profile and trips fail-closed adapter resolution (AGENTS.md #17). - source = MagicMock(chat_id="c1", platform=Platform.TELEGRAM, profile=None) + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="c1", + chat_type="dm", + user_id="user1", + profile=None, + ) return MessageEvent( text=text, message_type=MessageType.TEXT, @@ -206,17 +211,24 @@ def test_rapid_text_followups_are_queued_in_fifo_order(self): session_key = "telegram:user:fifo" texts = ["one", "two", "three", "four", "five"] - for text in texts: - runner._queue_or_replace_pending_event(session_key, self._text_event(text)) - - # Head slot keeps the first; overflow keeps the rest in order. - assert adapter._pending_messages[session_key].text == "one" + events = [self._text_event(text) for text in texts] + adapter_key = adapter.session_key_for_source(events[0].source) + assert adapter_key != session_key + for event in events: + runner._queue_or_replace_pending_event(session_key, event) + + # Head slot keeps the first; durable overflow keeps the rest in order. + assert adapter._pending_messages[adapter_key].text == "one" assert [e.text for e in runner._queued_events[session_key]] == [ "two", "three", "four", "five", ] - assert runner._queue_depth(session_key, adapter=adapter) == len(texts) + assert runner._queue_depth( + session_key, + adapter=adapter, + adapter_key=adapter_key, + ) == len(texts) diff --git a/tests/gateway/test_terminal_steer_fifo.py b/tests/gateway/test_terminal_steer_fifo.py new file mode 100644 index 000000000000..4a0264dbe217 --- /dev/null +++ b/tests/gateway/test_terminal_steer_fifo.py @@ -0,0 +1,405 @@ +"""Gateway regressions for terminal /steer FIFO delivery. + +A terminal agent result can carry ``pending_steer`` when the turn exits before +its finalizer can inject an accepted steer. The gateway must deliver older +transport work first, then the steer exactly once, without turning the queued +fallback into an interrupt. +""" + +from __future__ import annotations + +import asyncio +import importlib +import sys +import threading +import types +from types import SimpleNamespace + +import pytest + +from gateway.config import Platform +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + PlatformConfig, + SendResult, +) +from gateway.session import SessionSource, build_session_key + + +class _TerminalSteerAdapter(BasePlatformAdapter): + def __init__(self) -> None: + super().__init__( + PlatformConfig(enabled=True, token="test"), + Platform.TELEGRAM, + ) + self.sent: list[str] = [] + + async def connect(self, *, is_reconnect: bool = False) -> bool: + return True + + async def disconnect(self) -> None: + return None + + async def send(self, chat_id, content, reply_to=None, metadata=None): + self.sent.append(content) + return SendResult(success=True, message_id=f"sent-{len(self.sent)}") + + async def get_chat_info(self, chat_id): + return {"id": chat_id, "type": "dm"} + + +class _RecordingInterruptEvent(asyncio.Event): + def __init__(self) -> None: + super().__init__() + self.set_calls = 0 + + def set(self) -> None: + self.set_calls += 1 + super().set() + + +class _TerminalSteerAgent: + started = threading.Event() + release_terminal_result = threading.Event() + messages: list[str] = [] + interrupts: list[str] = [] + terminal_steer = "terminal steer" + + def __init__(self, **kwargs) -> None: + self.tools = [] + + def interrupt(self, reason="") -> bool: + type(self).interrupts.append(reason) + return True + + def run_conversation(self, message, conversation_history=None, task_id=None): + type(self).messages.append(message) + if len(type(self).messages) == 1: + type(self).started.set() + if not type(self).release_terminal_result.wait(timeout=5): + raise AssertionError("test did not release terminal result") + return { + "final_response": "Response truncated at the output cap", + "messages": [], + "api_calls": 1, + "completed": False, + "partial": True, + "pending_steer": type(self).terminal_steer, + } + return { + "final_response": f"processed: {message}", + "messages": [], + "api_calls": 1, + "completed": True, + } + + +def _make_runner(monkeypatch, tmp_path): + fake_dotenv = types.ModuleType("dotenv") + setattr(fake_dotenv, "load_dotenv", lambda *args, **kwargs: None) + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + fake_run_agent = types.ModuleType("run_agent") + setattr(fake_run_agent, "AIAgent", _TerminalSteerAgent) + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + + gateway_run = importlib.import_module("gateway.run") + adapter = _TerminalSteerAdapter() + runner = object.__new__(gateway_run.GatewayRunner) + runner.adapters = {adapter.platform: adapter} + runner._voice_mode = {} + runner._prefill_messages = [] + runner._ephemeral_system_prompt = "" + runner._reasoning_config = None + runner._provider_routing = {} + runner._fallback_model = None + runner._session_db = None + runner._running_agents = {} + runner._session_run_generation = {} + runner._queued_events = {} + runner.session_store = SimpleNamespace(_entries={}, _save=lambda: None) + runner.hooks = SimpleNamespace(loaded_hooks=False) + runner.config = SimpleNamespace( + thread_sessions_per_user=False, + group_sessions_per_user=False, + stt_enabled=False, + multiplex_profiles=False, + ) + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr( + gateway_run, + "_resolve_runtime_agent_kwargs", + lambda: {"api_key": "***"}, + ) + return gateway_run, runner, adapter + + +def _queued_event(source: SessionSource) -> MessageEvent: + return MessageEvent( + text="older queued work", + message_type=MessageType.TEXT, + source=source, + message_id="queued-before-terminal", + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "queue_before_run", + [True, False], + ids=["queue-before-agent-start", "queue-while-terminal-result-blocked"], +) +async def test_terminal_steer_runs_once_behind_older_transport_work_without_interrupt( + monkeypatch, + tmp_path, + queue_before_run, +): + """Control both queue/result orderings with events, not scheduler timing.""" + _TerminalSteerAgent.started = threading.Event() + _TerminalSteerAgent.release_terminal_result = threading.Event() + _TerminalSteerAgent.messages = [] + _TerminalSteerAgent.interrupts = [] + _TerminalSteerAgent.terminal_steer = "terminal steer" + + _gateway_run, runner, adapter = _make_runner(monkeypatch, tmp_path) + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-terminal-steer", + chat_type="dm", + user_id="user-1", + ) + session_key = build_session_key(source) + queued = _queued_event(source) + interrupt_event = _RecordingInterruptEvent() + adapter._active_sessions[session_key] = interrupt_event + + if queue_before_run: + runner._enqueue_fifo(session_key, queued, adapter) + + task = asyncio.create_task( + runner._run_agent( + message="original turn", + context_prompt="", + history=[], + source=source, + session_id="sess-terminal-steer", + session_key=session_key, + ) + ) + assert await asyncio.to_thread(_TerminalSteerAgent.started.wait, 2) + + if not queue_before_run: + runner._enqueue_fifo(session_key, queued, adapter) + + active_event = adapter._active_sessions[session_key] + assert not active_event.is_set(), "queue fallback must not request an interrupt" + + _TerminalSteerAgent.release_terminal_result.set() + result = await asyncio.wait_for(task, timeout=5) + + assert _TerminalSteerAgent.messages == [ + "original turn", + "older queued work", + "terminal steer", + ] + assert _TerminalSteerAgent.messages.count("terminal steer") == 1 + assert _TerminalSteerAgent.interrupts == [] + assert interrupt_event.set_calls == 0 + assert session_key not in adapter._pending_messages + assert runner._queued_events.get(session_key, []) == [] + assert result["final_response"] == "processed: terminal steer" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("older_head", [False, True], ids=["direct", "behind-older-head"]) +async def test_command_shaped_terminal_steer_keeps_accepted_provenance( + monkeypatch, + tmp_path, + older_head, +): + """Trusted terminal steer text is user input even when it resembles /stop.""" + _TerminalSteerAgent.started = threading.Event() + _TerminalSteerAgent.release_terminal_result = threading.Event() + _TerminalSteerAgent.messages = [] + _TerminalSteerAgent.interrupts = [] + _TerminalSteerAgent.terminal_steer = "/stop" + + _gateway_run, runner, adapter = _make_runner(monkeypatch, tmp_path) + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-command-steer", + chat_type="dm", + user_id="user-command", + ) + state_key = runner._session_key_for_source(source) + adapter_key = adapter.session_key_for_source(source) + adapter._active_sessions[adapter_key] = _RecordingInterruptEvent() + if older_head: + runner._enqueue_fifo( + state_key, + _queued_event(source), + adapter, + adapter_key=adapter_key, + ) + + task = asyncio.create_task( + runner._run_agent( + message="original turn", + context_prompt="", + history=[], + source=source, + session_id="sess-command-steer", + session_key=state_key, + ) + ) + assert await asyncio.to_thread(_TerminalSteerAgent.started.wait, 2) + _TerminalSteerAgent.release_terminal_result.set() + result = await asyncio.wait_for(task, timeout=5) + + expected = ["original turn"] + if older_head: + expected.append("older queued work") + expected.append("/stop") + assert _TerminalSteerAgent.messages == expected + assert result["final_response"] == "processed: /stop" + assert adapter_key not in adapter._pending_messages + assert runner._queued_events.get(state_key, []) == [] + + +@pytest.mark.asyncio +async def test_named_profile_keeps_adapter_slot_and_durable_fifo_keys_separate( + monkeypatch, + tmp_path, +): + _TerminalSteerAgent.started = threading.Event() + _TerminalSteerAgent.release_terminal_result = threading.Event() + _TerminalSteerAgent.messages = [] + _TerminalSteerAgent.interrupts = [] + _TerminalSteerAgent.terminal_steer = "terminal steer" + + _gateway_run, runner, adapter = _make_runner(monkeypatch, tmp_path) + runner.config.multiplex_profiles = True + runner._profile_adapters = {"coder": {Platform.TELEGRAM: adapter}} + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-multiplex-steer", + chat_type="dm", + user_id="user-coder", + profile="coder", + ) + state_key = runner._session_key_for_source(source) + adapter_key = adapter.session_key_for_source(source) + assert state_key != adapter_key + adapter._active_sessions[adapter_key] = _RecordingInterruptEvent() + runner._enqueue_fifo( + state_key, + _queued_event(source), + adapter, + adapter_key=adapter_key, + ) + + task = asyncio.create_task( + runner._run_agent( + message="original turn", + context_prompt="", + history=[], + source=source, + session_id="sess-multiplex-steer", + session_key=state_key, + ) + ) + assert await asyncio.to_thread(_TerminalSteerAgent.started.wait, 2) + _TerminalSteerAgent.release_terminal_result.set() + result = await asyncio.wait_for(task, timeout=5) + + assert _TerminalSteerAgent.messages == [ + "original turn", + "older queued work", + "terminal steer", + ] + assert result["final_response"] == "processed: terminal steer" + assert adapter_key not in adapter._pending_messages + assert state_key not in adapter._pending_messages + assert runner._queued_events.get(state_key, []) == [] + assert runner._queued_events.get(adapter_key, []) == [] + + +@pytest.mark.asyncio +async def test_depth_cap_requeues_current_head_without_displacing_terminal_steer( + monkeypatch, + tmp_path, +): + """MAX_DEPTH + 1 older events and terminal steer retain exact FIFO order.""" + _TerminalSteerAgent.started = threading.Event() + _TerminalSteerAgent.release_terminal_result = threading.Event() + _TerminalSteerAgent.messages = [] + _TerminalSteerAgent.interrupts = [] + _TerminalSteerAgent.terminal_steer = "terminal steer" + + _gateway_run, runner, adapter = _make_runner(monkeypatch, tmp_path) + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-depth-steer", + chat_type="dm", + user_id="user-depth", + ) + state_key = runner._session_key_for_source(source) + adapter_key = adapter.session_key_for_source(source) + adapter._active_sessions[adapter_key] = _RecordingInterruptEvent() + older_texts = [f"older-{idx}" for idx in range(runner._MAX_INTERRUPT_DEPTH + 1)] + for idx, text in enumerate(older_texts): + runner._enqueue_fifo( + state_key, + MessageEvent( + text=text, + message_type=MessageType.TEXT, + source=source, + message_id=f"older-{idx}", + ), + adapter, + adapter_key=adapter_key, + ) + + first_chain = asyncio.create_task( + runner._run_agent( + message="original turn", + context_prompt="", + history=[], + source=source, + session_id="sess-depth-steer", + session_key=state_key, + ) + ) + assert await asyncio.to_thread(_TerminalSteerAgent.started.wait, 2) + _TerminalSteerAgent.release_terminal_result.set() + await asyncio.wait_for(first_chain, timeout=5) + + assert _TerminalSteerAgent.messages == ["original turn", *older_texts[:-1]] + assert adapter._pending_messages[adapter_key].text == older_texts[-1] + assert [event.text for event in runner._queued_events[state_key]] == [ + "terminal steer" + ] + + current = adapter.get_pending_message(adapter_key) + assert current is not None + await asyncio.wait_for( + runner._run_agent( + message=current.text, + context_prompt="", + history=[], + source=current.source, + session_id="sess-depth-steer", + session_key=state_key, + ), + timeout=5, + ) + + assert _TerminalSteerAgent.messages == [ + "original turn", + *older_texts, + "terminal steer", + ] + assert adapter_key not in adapter._pending_messages + assert runner._queued_events.get(state_key, []) == [] diff --git a/tests/run_agent/test_413_compression.py b/tests/run_agent/test_413_compression.py index fcedeca0e176..c73eb53c1c08 100644 --- a/tests/run_agent/test_413_compression.py +++ b/tests/run_agent/test_413_compression.py @@ -143,6 +143,32 @@ def _provider_crash(*_args, **_kwargs): assert isinstance(persisted_messages[-1]["timestamp"], float) +def test_provider_output_cap_failure_returns_mid_turn_steer(agent): + """The max_tokens-cap error exits before TurnFinalizer, so a steer that + arrives during the rejected provider call must be returned to the caller.""" + class _OutputCapError(Exception): + status_code = 400 + + error = _OutputCapError("Error code: 400 - max_tokens should be less than or equal to 16384") + + def _output_cap_error(*_args, **_kwargs): + agent.steer("retry with a shorter answer") + raise error + + agent.client.chat.completions.create.side_effect = _output_cap_error + + with ( + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("write a long answer") + + assert result["completed"] is False + assert "max_tokens exceeds the provider's output cap" in result["error"] + assert result["pending_steer"] == "retry with a shorter answer" + + class TestHTTP413Compression: """413 errors should trigger compression, not abort as generic 4xx.""" diff --git a/tests/run_agent/test_partial_stream_finish_reason.py b/tests/run_agent/test_partial_stream_finish_reason.py index 129edeb7a788..d670727e8b00 100644 --- a/tests/run_agent/test_partial_stream_finish_reason.py +++ b/tests/run_agent/test_partial_stream_finish_reason.py @@ -384,6 +384,38 @@ def test_partial_stream_stub_does_not_exit_loop_immediately(self, loop_agent): assert "first half of" in result["final_response"] assert "forty-two" in result["final_response"] + def test_length_retry_exhaustion_returns_mid_turn_steer(self, loop_agent): + """A /steer received while the model is repeatedly length-truncated + must survive the early return so the CLI/gateway can make it the next + user turn. The normal turn finalizer does this, but the exhaustion + branch returns before it runs. + """ + from tests.run_agent.test_run_agent import _mock_response + + calls = 0 + + def _always_truncated(*_args, **_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + # Simulate a steer arriving while the first API request is in + # flight. No tool result exists for the pre-call drain to use. + loop_agent.steer("break the answer into smaller files") + return _mock_response(content=f"partial-{calls}", finish_reason="length") + + loop_agent.client.chat.completions.create.side_effect = _always_truncated + + with ( + patch.object(loop_agent, "_persist_session"), + patch.object(loop_agent, "_save_trajectory"), + patch.object(loop_agent, "_cleanup_task_resources"), + ): + result = loop_agent.run_conversation("write a very long answer") + + assert loop_agent.client.chat.completions.create.call_count == 4 + assert result["completed"] is False + assert result["pending_steer"] == "break the answer into smaller files" + class TestContentFilterStallActivatesFallback: """Regression for #32421: a provider output-layer content safety filter diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 88c18fa12e57..b00881db1c73 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -4271,6 +4271,41 @@ def test_length_with_tool_calls_returns_partial_without_executing_tools(self, ag assert "truncated due to output length limit" in result["error"] mock_handle_function_call.assert_not_called() + def test_truncated_tool_call_exhaustion_returns_mid_turn_steer(self, agent): + """A steer received while truncated tool calls retry must survive the + terminal return, just like a text-only output truncation.""" + self._setup_agent(agent) + bad_tc = _mock_tool_call( + name="write_file", + arguments='{"path":"report.md","content":"partial', + call_id="c1", + ) + calls = 0 + + def _always_truncated(*_args, **_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + agent.steer("use the small report template") + return _mock_response( + content="", finish_reason="length", tool_calls=[bad_tc], + ) + + agent.client.chat.completions.create.side_effect = _always_truncated + + with ( + patch("run_agent.handle_function_call") as mock_handle_function_call, + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("write the report") + + assert agent.client.chat.completions.create.call_count == 5 + assert result["completed"] is False + assert result["pending_steer"] == "use the small report template" + mock_handle_function_call.assert_not_called() + def test_truncated_tool_call_retries_once_before_refusing(self, agent): """When tool call args are truncated, the agent retries the API call (up to 3 times). If a retry succeeds (valid JSON args), tool execution @@ -4803,6 +4838,51 @@ def test_output_cap_retry_triggers_compression_and_recovers(self, agent): # context_length was NOT mutated by an output-cap error. assert agent.context_compressor.context_length == 200_000 + def test_output_cap_lock_defer_returns_mid_turn_steer(self, agent): + """A lock-contended output-cap compression keeps a late steer.""" + self._setup_agent(agent) + agent.api_mode = "chat_completions" + agent.provider = "openrouter" + agent.model = "some/model" + agent.max_tokens = 65_536 + agent.compression_enabled = True + agent.context_compressor.context_length = 200_000 + agent.context_compressor.should_compress = MagicMock(return_value=True) + + error_msg = ( + "max_tokens: 65536 > context_window: 200000 " + "- input_tokens: 199000 = available_tokens: 1000" + ) + + class _OutputCapError(Exception): + status_code = 400 + code = 400 + + def _reject_with_steer(*_args, **_kwargs): + agent.steer("wait for compression, then keep the concise plan") + raise _OutputCapError(error_msg) + + agent.client.chat.completions.create.side_effect = _reject_with_steer + + def _lock_contended(messages, system_message, **_kwargs): + agent._compression_skipped_due_to_lock = "pid=4242:tid=1" + return messages, system_message + + with ( + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + patch.object(agent.context_compressor, "update_model"), + patch.object(agent, "_compress_context", side_effect=_lock_contended), + ): + result = agent.run_conversation("hello") + + assert result["completed"] is False + assert result["compression_deferred"] is True + assert result["pending_steer"] == ( + "wait for compression, then keep the concise plan" + ) + def test_output_cap_retry_compression_no_progress_terminates_bounded(self, agent): """Regression: when the compressor cannot reduce the request (zero progress AND no images to strip), the output-cap retry must terminate @@ -4965,6 +5045,36 @@ def test_invalid_response_returns_error_not_crash(self, agent): assert "Invalid API response" in result["error"] assert result.get("final_response") == result["error"] + def test_invalid_response_retry_exhaustion_returns_mid_turn_steer(self, agent): + """Generic provider retry exhaustion has the same leftover-steer contract.""" + self._setup_agent(agent) + bad_resp = SimpleNamespace(choices=[], model="test/model", usage=None) + calls = 0 + + def _always_invalid(*_args, **_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + agent.steer("switch to the concise fallback plan") + return bad_resp + + agent.client.chat.completions.create.side_effect = _always_invalid + from agent import conversation_loop as _conv_loop + + with ( + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + patch("run_agent.time", self._make_fast_time_mock()), + patch.object(_conv_loop, "time", self._make_fast_time_mock()), + patch.object(_conv_loop, "jittered_backoff", lambda *a, **k: 0.0), + ): + result = agent.run_conversation("hello") + + assert calls == agent._api_max_retries + assert result["completed"] is False + assert result["pending_steer"] == "switch to the concise fallback plan" + def test_invalid_response_retry_completes_one_logical_call(self, agent): self._setup_agent(agent) agent.client.chat.completions.create.side_effect = [ diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index 97e1ed1679ad..cca847ff011e 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/tests/run_agent/test_run_agent_codex_responses.py @@ -1659,6 +1659,26 @@ def _fake_execute_tool_calls(assistant_message, messages, effective_task_id, *_a +def test_run_conversation_codex_max_output_exhaustion_returns_mid_turn_steer(monkeypatch): + agent = _build_agent(monkeypatch) + calls = 0 + + def _always_incomplete(_api_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + agent.steer("stop and provide only the summary") + return _codex_max_output_incomplete_response("Partial final answer") + + monkeypatch.setattr(agent, "_interruptible_api_call", _always_incomplete) + + result = agent.run_conversation("write a long final answer") + + assert calls == 3 + assert result["completed"] is False + assert result["pending_steer"] == "stop and provide only the summary" + + def test_run_conversation_compresses_mid_turn_before_output_budget_exhaustion(monkeypatch): """Long tool-heavy turns should compact before the next API request. diff --git a/tests/run_agent/test_steer.py b/tests/run_agent/test_steer.py index cd5266ef3afe..c307ca294ff9 100644 --- a/tests/run_agent/test_steer.py +++ b/tests/run_agent/test_steer.py @@ -23,6 +23,7 @@ def _bare_agent() -> AIAgent: agent = object.__new__(AIAgent) agent._pending_steer = None agent._pending_steer_lock = threading.Lock() + agent._steer_acceptance_generation = 1 agent._pending_redirect = None agent._pending_redirect_lock = threading.Lock() agent._model_request_active = threading.Event() @@ -62,6 +63,48 @@ def test_drain_returns_and_clears(self): assert agent._drain_pending_steer() == "hello" assert agent._pending_steer is None + def test_steer_wins_lock_before_terminal_seal_and_is_returned(self): + from agent.conversation_loop import _terminal_result_with_pending_steer + + agent = _bare_agent() + steer_committed = threading.Event() + outcome = {} + + def steer_first(): + outcome["accepted"] = agent.steer("include the migration note") + steer_committed.set() + + worker = threading.Thread(target=steer_first) + worker.start() + assert steer_committed.wait(timeout=1) + result = _terminal_result_with_pending_steer(agent, {}) + worker.join(timeout=1) + + assert outcome["accepted"] is True + assert result["pending_steer"] == "include the migration note" + assert agent._pending_steer is None + + def test_terminal_seal_wins_lock_and_later_steer_is_rejected(self): + from agent.conversation_loop import _terminal_result_with_pending_steer + + agent = _bare_agent() + terminal_sealed = threading.Event() + result_holder = {} + + def seal_first(): + result_holder.update(_terminal_result_with_pending_steer(agent, {})) + terminal_sealed.set() + + worker = threading.Thread(target=seal_first) + worker.start() + assert terminal_sealed.wait(timeout=1) + accepted = agent.steer("run this as the next turn") + worker.join(timeout=1) + + assert accepted is False + assert result_holder == {} + assert agent._pending_steer is None + class TestActiveTurnRedirect: diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 2f15a65015c3..0f1f7003cbd6 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -10252,6 +10252,43 @@ def interrupt(self, *args, **kwargs): assert "interrupt_called" not in calls # must NOT interrupt +def test_session_steer_rejection_preserves_message_without_hard_interrupt(): + """A terminal-sealed TUI steer becomes FIFO next-turn work, not a stop.""" + calls = {} + + class _Agent: + def steer(self, text): + calls["steer_text"] = text + return False + + def interrupt(self, *args, **kwargs): + calls["interrupt_called"] = True + + session = _session(agent=_Agent(), running=True) + server._enqueue_prompt(session, "older queued work", "older-transport") + server._sessions["sid"] = session + try: + resp = server.handle_request( + { + "id": "1", + "method": "session.steer", + "params": {"session_id": "sid", "text": "preserve this message"}, + } + ) + finally: + server._sessions.pop("sid", None) + + assert resp["result"] == { + "status": "queued", + "text": "preserve this message", + } + assert calls["steer_text"] == "preserve this message" + assert "interrupt_called" not in calls + assert session["queued_prompt"]["text"] == ( + "older queued work\n\npreserve this message" + ) + + def test_session_steer_rejects_empty_text(): server._sessions["sid"] = _session( agent=types.SimpleNamespace(steer=lambda t: True) diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py index 564271ea130c..b4315ba21379 100644 --- a/tui_gateway/methods_session.py +++ b/tui_gateway/methods_session.py @@ -3480,7 +3480,16 @@ def _(rid, params: dict) -> dict: # settle (same class as redirect). _drop_queued_duplicates_of_inflight_user(session) session["last_active"] = time.time() - return _ok(rid, {"status": "queued" if accepted else "rejected", "text": text}) + status = "queued" + else: + # The active turn sealed steer acceptance before this RPC acquired the + # agent lock. Keep the user's text as next-turn work; do not route this + # normal terminal race through the hard-interrupt busy-submit fallback. + with session["history_lock"]: + _enqueue_prompt(session, text, current_transport() or _stdio_transport) + session["last_active"] = time.time() + status = "queued" + return _ok(rid, {"status": status, "text": text}) @method("session.redirect") From 45f6ac75485dba7353cbaa08414c6a4c1610d592 Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Tue, 11 Aug 2026 20:37:27 +0200 Subject: [PATCH 2/9] test(steer): align CI fixtures with ownership seams --- tests/gateway/test_multiplex_busy_input_mode.py | 17 +++++++++++++---- .../test_telegram_voice_v0_regressions.py | 10 +++++++--- tests/run_agent/test_tool_batch_segmentation.py | 4 ++++ 3 files changed, 24 insertions(+), 7 deletions(-) diff --git a/tests/gateway/test_multiplex_busy_input_mode.py b/tests/gateway/test_multiplex_busy_input_mode.py index 5e3d8af1bd92..822a8fcde8ce 100644 --- a/tests/gateway/test_multiplex_busy_input_mode.py +++ b/tests/gateway/test_multiplex_busy_input_mode.py @@ -162,6 +162,8 @@ async def test_secondary_profile_busy_mode_controls_priority_path( ) event = _event(profile="research") session_key = runner._session_key_for_source(event.source) + adapter_key = adapter.session_key_for_source(event.source) + assert adapter_key != session_key agent = MagicMock() agent._active_children = [] agent.steer.return_value = True @@ -172,10 +174,11 @@ async def test_secondary_profile_busy_mode_controls_priority_path( agent.interrupt.assert_not_called() if secondary_mode == "queue": agent.steer.assert_not_called() - assert adapter._pending_messages[session_key] is event + assert adapter._pending_messages[adapter_key] is event + assert session_key not in adapter._pending_messages else: agent.steer.assert_called_once_with("follow up") - assert session_key not in adapter._pending_messages + assert adapter_key not in adapter._pending_messages @pytest.mark.asyncio @@ -202,9 +205,12 @@ async def test_secondary_profile_busy_mode_controls_busy_handler_restart_drain( runner._restart_requested = True event = _event(profile="research") session_key = runner._session_key_for_source(event.source) + adapter_key = adapter.session_key_for_source(event.source) + assert adapter_key != session_key assert await runner._handle_active_session_busy_message(event, session_key) is True - assert (session_key in adapter._pending_messages) is queued + assert (adapter_key in adapter._pending_messages) is queued + assert session_key not in adapter._pending_messages @pytest.mark.asyncio @@ -223,6 +229,8 @@ async def test_secondary_profile_busy_mode_controls_priority_restart_drain( runner._restart_requested = True event = _event(profile="research") session_key = runner._session_key_for_source(event.source) + adapter_key = adapter.session_key_for_source(event.source) + assert adapter_key != session_key agent = MagicMock() agent._active_children = [] runner._running_agents[session_key] = agent @@ -231,7 +239,8 @@ async def test_secondary_profile_busy_mode_controls_priority_restart_drain( assert isinstance(response, str) assert "queued" in response - assert adapter._pending_messages[session_key] is event + assert adapter._pending_messages[adapter_key] is event + assert session_key not in adapter._pending_messages agent.interrupt.assert_not_called() diff --git a/tests/gateway/test_telegram_voice_v0_regressions.py b/tests/gateway/test_telegram_voice_v0_regressions.py index 05c2fbcffe03..a1b7237c5245 100644 --- a/tests/gateway/test_telegram_voice_v0_regressions.py +++ b/tests/gateway/test_telegram_voice_v0_regressions.py @@ -185,6 +185,8 @@ async def test_monitor_to_drain_transcribes_and_echoes_pending_voice_once( runner = _run_agent_runner(adapter) source = _source() session_key = "telegram:dm:12345" + adapter_key = adapter.session_key_for_source(source) + assert adapter_key != session_key event = MessageEvent( text="", message_type=MessageType.VOICE, @@ -192,9 +194,9 @@ async def test_monitor_to_drain_transcribes_and_echoes_pending_voice_once( media_urls=["/tmp/telegram-pending-voice.ogg"], media_types=["audio/ogg"], ) - adapter._pending_messages[session_key] = event - adapter._active_sessions[session_key] = asyncio.Event() - adapter._active_sessions[session_key].set() + adapter._pending_messages[adapter_key] = event + adapter._active_sessions[adapter_key] = asyncio.Event() + adapter._active_sessions[adapter_key].set() _PendingVoiceAgent.messages = [] with ( @@ -218,6 +220,8 @@ async def test_monitor_to_drain_transcribes_and_echoes_pending_voice_once( assert _PendingVoiceAgent.messages == ["initial turn", '"hello once"'] mock_transcribe.assert_called_once_with("/tmp/telegram-pending-voice.ogg", None, "gateway") assert adapter.sent == [("12345", '🎙️ "hello once"', None)] + assert adapter_key not in adapter._pending_messages + assert session_key not in adapter._pending_messages @pytest.mark.asyncio diff --git a/tests/run_agent/test_tool_batch_segmentation.py b/tests/run_agent/test_tool_batch_segmentation.py index 6f367878b915..3a478188a74b 100644 --- a/tests/run_agent/test_tool_batch_segmentation.py +++ b/tests/run_agent/test_tool_batch_segmentation.py @@ -397,6 +397,10 @@ def agent(): skip_memory=True, ) a.client = MagicMock() + # These tests invoke the tool-dispatch seam directly instead of entering + # through run_conversation(), which normally opens the active turn's + # steer-acceptance generation before any tool batch can execute. + a._begin_steer_acceptance() return a From 39ae8b556cb149e8d01d561fe1133e6c7ccae4f7 Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Sun, 16 Aug 2026 19:06:11 +0200 Subject: [PATCH 3/9] fix(steer): seal repetition abort results --- agent/conversation_loop.py | 19 +++++++++++-------- .../test_continuation_repetition_guard.py | 15 +++++++++++++++ 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index ffbbdcc8cb21..c3db4cb63550 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -3801,14 +3801,17 @@ def _perform_api_call(next_api_kwargs): ) agent._cleanup_task_resources(effective_task_id) agent._persist_session(messages, conversation_history) - return { - "final_response": _rep_response, - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": _rep_error, - } + return _terminal_result_with_pending_steer( + agent, + { + "final_response": _rep_response, + "messages": messages, + "api_calls": api_call_count, + "completed": False, + "partial": True, + "error": _rep_error, + }, + ) if agent.api_mode in {"chat_completions", "bedrock_converse", "anthropic_messages"}: assistant_message = _trunc_msg diff --git a/tests/run_agent/test_continuation_repetition_guard.py b/tests/run_agent/test_continuation_repetition_guard.py index 80223e21f0ec..e781dc055a49 100644 --- a/tests/run_agent/test_continuation_repetition_guard.py +++ b/tests/run_agent/test_continuation_repetition_guard.py @@ -86,6 +86,21 @@ def test_repetition_dominated_truncation_aborts(self, loop_agent): # Exactly one API call — no continuation was attempted. assert loop_agent.client.chat.completions.create.call_count == 1 + def test_repetition_abort_preserves_and_seals_late_steer(self, loop_agent): + echo = _INCIDENT_ECHO * 2000 + + def respond_with_late_steer(*_args, **_kwargs): + assert loop_agent.steer("include the migration note") is True + return _stub(echo) + + loop_agent.client.chat.completions.create.side_effect = respond_with_late_steer + + result = _run(loop_agent, "write me a long report") + + assert result["pending_steer"] == "include the migration note" + assert loop_agent._pending_steer is None + assert loop_agent.steer("too late for this turn") is False + def test_legit_truncation_still_continues(self, loop_agent): # Ordinary short truncated fragments still get continuation retries. loop_agent.client.chat.completions.create.side_effect = [ From f6a8ebf4ec94c5300abb36d7e61bd1d2204750ab Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Sun, 16 Aug 2026 19:29:47 +0200 Subject: [PATCH 4/9] fix(gateway): align post-delivery callback ownership --- gateway/run.py | 35 +++++-- tests/gateway/test_run_cleanup_progress.py | 107 ++++++++++++++++++++- tests/gateway/test_run_progress_topics.py | 59 +++++++++++- tests/gateway/test_status_command.py | 55 ++++++++--- 4 files changed, 229 insertions(+), 27 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index f42fe45f4fea..57a1f5dbe669 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5925,16 +5925,21 @@ def _bg_review_send(message: str) -> None: # Register the release hook on the adapter so base.py's finally # block can fire it after delivering the main response. if ctx._status_adapter and ctx.session_key: + _post_delivery_key = self._runner._adapter_key_for_source( + ctx._status_adapter, + ctx.source, + fallback=ctx.session_key, + ) if getattr(type(ctx._status_adapter), "register_post_delivery_callback", None) is not None: ctx._status_adapter.register_post_delivery_callback( - ctx.session_key, + _post_delivery_key, _release_bg_review_messages, generation=ctx.run_generation, ) else: _pdc = getattr(ctx._status_adapter, "_post_delivery_callbacks", None) if _pdc is not None: - _pdc[ctx.session_key] = _release_bg_review_messages + _pdc[_post_delivery_key] = _release_bg_review_messages # Memory update notifications in chat. Config: display.memory_notifications # off — no chat notification (still logged to stdout) # on — generic "💾 Memory updated" (default) @@ -20324,13 +20329,21 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g run_generation, ) _stale_adapter = self._adapter_for_source(source) + _stale_adapter_key = self._adapter_key_for_source( + _stale_adapter, + source, + fallback=_quick_key, + ) if getattr(type(_stale_adapter), "pop_post_delivery_callback", None) is not None: _stale_adapter.pop_post_delivery_callback( - _quick_key, + _stale_adapter_key, generation=run_generation, ) elif _stale_adapter and hasattr(_stale_adapter, "_post_delivery_callbacks"): - _stale_adapter._post_delivery_callbacks.pop(_quick_key, None) + _stale_adapter._post_delivery_callbacks.pop( + _stale_adapter_key, + None, + ) return None response = agent_result.get("final_response") or "" @@ -29709,7 +29722,7 @@ def _run_sync_with_timeout_lifecycle(): # base.py's finally block) and call it. if getattr(type(adapter), "pop_post_delivery_callback", None) is not None: _bg_cb = adapter.pop_post_delivery_callback( - session_key, + adapter_key, generation=run_generation, ) if callable(_bg_cb): @@ -29720,7 +29733,10 @@ def _run_sync_with_timeout_lifecycle(): except Exception: pass elif adapter and hasattr(adapter, "_post_delivery_callbacks"): - _bg_cb = adapter._post_delivery_callbacks.pop(session_key, None) + _bg_cb = adapter._post_delivery_callbacks.pop( + adapter_key, + None, + ) if callable(_bg_cb): try: _bg_result = _bg_cb() @@ -30096,8 +30112,13 @@ async def _delete_all() -> None: pass try: + _cleanup_adapter_key = self._adapter_key_for_source( + _cleanup_adapter, + source, + fallback=session_key, + ) _cleanup_adapter.register_post_delivery_callback( - session_key, + _cleanup_adapter_key, _cleanup_temp_bubbles, generation=run_generation, ) diff --git a/tests/gateway/test_run_cleanup_progress.py b/tests/gateway/test_run_cleanup_progress.py index 620f76a8590d..2b0780f8b07d 100644 --- a/tests/gateway/test_run_cleanup_progress.py +++ b/tests/gateway/test_run_cleanup_progress.py @@ -121,6 +121,24 @@ def run_conversation(self, message, conversation_history=None, task_id=None): return {"final_response": "done", "messages": [], "api_calls": 1} +class ReviewingProgressAgent(ProgressAgent): + """Emits progress plus a background-review notice in one production turn.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.background_review_callback = kwargs.get("background_review_callback") + + def run_conversation(self, message, conversation_history=None, task_id=None): + result = super().run_conversation( + message, + conversation_history=conversation_history, + task_id=task_id, + ) + if self.background_review_callback is not None: + self.background_review_callback("💾 Memory updated") + return result + + class FailingAgent: def __init__(self, **kwargs): self.tool_progress_callback = kwargs.get("tool_progress_callback") @@ -262,7 +280,9 @@ async def test_cleanup_chains_with_existing_callback(monkeypatch, tmp_path): monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) source = SessionSource(platform=Platform.TELEGRAM, chat_id="-1001") - session_key = "agent:main:telegram:group:-1001" + session_key = runner._session_key_for_source(source) + adapter_key = adapter.session_key_for_source(source) + assert adapter_key == session_key pre_existing_fired = [] @@ -271,7 +291,7 @@ def _preexisting_callback() -> None: # Pre-register a callback with the same generation the run will use # (run_generation=None in this test path — matches the default slot). - adapter.register_post_delivery_callback(session_key, _preexisting_callback) + adapter.register_post_delivery_callback(adapter_key, _preexisting_callback) result = await runner._run_agent( message="hello", @@ -283,7 +303,7 @@ def _preexisting_callback() -> None: ) assert result["final_response"] == "done" - cb = adapter.pop_post_delivery_callback(session_key) + cb = adapter.pop_post_delivery_callback(adapter_key) assert callable(cb) await _fire_post_delivery_cb(cb) for _ in range(20): @@ -295,3 +315,84 @@ def _preexisting_callback() -> None: # deletes at least one progress bubble. assert pre_existing_fired == [True] assert len(adapter.deleted) >= 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("profile", "keys_differ"), + [("research", True), ("default", False)], + ids=["named-profile", "default-profile"], +) +async def test_profiled_production_delivery_consumes_adapter_owned_callbacks( + monkeypatch, + tmp_path, + profile, + keys_differ, +): + """The real adapter lifecycle consumes review + cleanup callbacks once.""" + adapter = CleanupCaptureAdapter() + runner = _make_runner(adapter) + runner.config.multiplex_profiles = True + runner._profile_adapters = {} + if profile != "default": + runner._profile_adapters[profile] = {Platform.TELEGRAM: adapter} + gateway_run = _install_fakes( + monkeypatch, + ReviewingProgressAgent, + cleanup_on=True, + ) + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="profiled-chat", + chat_type="dm", + profile=profile, + ) + event = gateway_run.MessageEvent( + text="hello", + message_type=gateway_run.MessageType.TEXT, + source=source, + message_id="profiled-message", + ) + state_key = runner._session_key_for_source(source) + adapter_key = adapter.session_key_for_source(source) + assert (state_key != adapter_key) is keys_differ + generation = runner._begin_session_run_generation(state_key) + + async def _production_handler(inbound_event): + runner._bind_adapter_run_generation( + adapter, + inbound_event.source, + state_key, + generation, + ) + result = await runner._run_agent( + message=inbound_event.text, + context_prompt="", + history=[], + source=inbound_event.source, + session_id=f"sess-{profile}", + session_key=state_key, + run_generation=generation, + ) + return result["final_response"] + + adapter.set_message_handler(_production_handler) + await adapter.handle_message(event) + task = adapter._session_tasks[adapter_key] + await asyncio.wait_for(task, timeout=5) + assert runner._is_session_run_current(state_key, generation) + + for _ in range(50): + sent_text = [item["content"] for item in adapter.sent] + if "💾 Memory updated" in sent_text and adapter.deleted: + break + await asyncio.sleep(0.01) + + sent_text = [item["content"] for item in adapter.sent] + assert "done" in sent_text + assert sent_text.count("💾 Memory updated") == 1 + assert sent_text.index("done") < sent_text.index("💾 Memory updated") + assert adapter.deleted + assert adapter._post_delivery_callbacks == {} diff --git a/tests/gateway/test_run_progress_topics.py b/tests/gateway/test_run_progress_topics.py index da3dd2efc7bf..96e0a6f649b8 100644 --- a/tests/gateway/test_run_progress_topics.py +++ b/tests/gateway/test_run_progress_topics.py @@ -1006,6 +1006,7 @@ async def _run_with_agent( adapter_cls=ProgressCaptureAdapter, user_id=None, scope_id=None, + profile=None, ): if config_data: import yaml @@ -1023,6 +1024,10 @@ async def _run_with_agent( adapter = adapter_cls(platform=platform) runner = _make_runner(adapter) gateway_run = importlib.import_module("gateway.run") + runner.config.multiplex_profiles = profile is not None + runner._profile_adapters = {} + if profile and profile != "default": + runner._profile_adapters[profile] = {platform: adapter} if config_data and "streaming" in config_data: runner.config.streaming = StreamingConfig.from_dict(config_data["streaming"]) monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) @@ -1034,12 +1039,12 @@ async def _run_with_agent( thread_id=thread_id, user_id=user_id, scope_id=scope_id, + profile=profile, ) - session_key = f"agent:main:{platform.value}:{chat_type}:{chat_id}" - if thread_id: - session_key = f"{session_key}:{thread_id}" + session_key = runner._session_key_for_source(source) + adapter_key = adapter.session_key_for_source(source) if pending_text is not None: - adapter._pending_messages[session_key] = MessageEvent( + adapter._pending_messages[adapter_key] = MessageEvent( text=pending_text, message_type=MessageType.TEXT, source=source, @@ -1395,6 +1400,52 @@ async def test_run_agent_defers_background_review_notification_until_release(mon assert adapter.sent == [] +@pytest.mark.asyncio +async def test_named_profile_queued_delivery_pops_physical_callback_once( + monkeypatch, + tmp_path, +): + """The in-band first-response send uses the same slot as base delivery.""" + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + BackgroundReviewAgent, + session_id="sess-profiled-queued-review", + pending_text="queued follow-up", + profile="research", + ) + + adapter_key = "agent:main:telegram:group:-1001:17585" + state_key = "agent:research:telegram:group:-1001:17585" + assert result["final_response"] == "done" + for _ in range(50): + if [item["content"] for item in adapter.sent].count( + "💾 Skill 'prospect-scanner' created." + ) == 1: + break + await asyncio.sleep(0.01) + + sent_text = [item["content"] for item in adapter.sent] + assert sent_text.count("💾 Skill 'prospect-scanner' created.") == 1 + assert adapter_key in adapter._post_delivery_callbacks + assert state_key not in adapter._post_delivery_callbacks + + callback = adapter.pop_post_delivery_callback(adapter_key) + assert callable(callback) + callback() + for _ in range(50): + if [item["content"] for item in adapter.sent].count( + "💾 Skill 'prospect-scanner' created." + ) == 2: + break + await asyncio.sleep(0.01) + + assert [item["content"] for item in adapter.sent].count( + "💾 Skill 'prospect-scanner' created." + ) == 2 + assert adapter._post_delivery_callbacks == {} + + @pytest.mark.asyncio async def test_base_processing_releases_post_delivery_callback_after_main_send(): """Post-delivery callbacks on the adapter fire after the main response.""" diff --git a/tests/gateway/test_status_command.py b/tests/gateway/test_status_command.py index ec179816c767..571358f15052 100644 --- a/tests/gateway/test_status_command.py +++ b/tests/gateway/test_status_command.py @@ -13,20 +13,30 @@ from gateway.session import SessionEntry, SessionSource, build_session_key -def _make_source(platform: Platform = Platform.TELEGRAM) -> SessionSource: +def _make_source( + platform: Platform = Platform.TELEGRAM, + *, + profile: str | None = None, +) -> SessionSource: return SessionSource( platform=platform, user_id="u1", chat_id="c1", user_name="tester", chat_type="dm", + profile=profile, ) -def _make_event(text: str, *, platform: Platform = Platform.TELEGRAM) -> MessageEvent: +def _make_event( + text: str, + *, + platform: Platform = Platform.TELEGRAM, + profile: str | None = None, +) -> MessageEvent: return MessageEvent( text=text, - source=_make_source(platform), + source=_make_source(platform, profile=profile), message_id="m1", ) @@ -303,7 +313,11 @@ async def test_first_run_slack_home_channel_onboarding_uses_parent_command(monke @pytest.mark.asyncio -async def test_handle_message_stale_result_keeps_newer_generation_callback(monkeypatch): +@pytest.mark.parametrize("newer_callback", [False, True]) +async def test_handle_message_stale_result_cleans_only_owned_physical_callback( + monkeypatch, + newer_callback, +): import gateway.run as gateway_run class _Adapter: @@ -313,6 +327,9 @@ def __init__(self): async def send(self, *args, **kwargs): return None + def session_key_for_source(self, source): + return build_session_key(source) + def pop_post_delivery_callback(self, session_key, *, generation=None): entry = self._post_delivery_callbacks.get(session_key) if entry is None: @@ -327,8 +344,12 @@ def pop_post_delivery_callback(self, session_key, *, generation=None): return None return self._post_delivery_callbacks.pop(session_key, None) + source = _make_source(profile="research") + session_key = build_session_key(source, profile="research") + adapter_key = build_session_key(source) + assert adapter_key != session_key session_entry = SessionEntry( - session_key=build_session_key(_make_source()), + session_key=session_key, session_id="sess-1", created_at=datetime.now(), updated_at=datetime.now(), @@ -336,15 +357,20 @@ def pop_post_delivery_callback(self, session_key, *, generation=None): chat_type="dm", ) runner = _make_runner(session_entry) + runner.config.multiplex_profiles = True runner.session_store.load_transcript.return_value = [{"role": "user", "content": "earlier"}] - session_key = session_entry.session_key adapter = _Adapter() - runner.adapters[Platform.TELEGRAM] = adapter + runner._profile_adapters = {"research": {Platform.TELEGRAM: adapter}} async def _stale_result(**kwargs): - # Simulate a newer run claiming the callback slot before the stale run unwinds. - runner._session_run_generation[session_key] = 2 - adapter._post_delivery_callbacks[session_key] = (2, lambda: None) + # Simulate a newer run claiming the physical callback slot before the + # stale run unwinds from its profile-qualified durable state slot. + newer_generation = runner._begin_session_run_generation(session_key) + assert newer_generation > kwargs["run_generation"] + adapter._post_delivery_callbacks[adapter_key] = ( + newer_generation if newer_callback else kwargs["run_generation"], + lambda: None, + ) return { "final_response": "late reply", "messages": [], @@ -364,11 +390,14 @@ async def _stale_result(**kwargs): lambda *_args, **_kwargs: 100000, ) - result = await runner._handle_message(_make_event("hello")) + result = await runner._handle_message(_make_event("hello", profile="research")) assert result is None - assert session_key in adapter._post_delivery_callbacks - assert adapter._post_delivery_callbacks[session_key][0] == 2 + assert session_key not in adapter._post_delivery_callbacks + if newer_callback: + assert adapter._post_delivery_callbacks[adapter_key][0] == 2 + else: + assert adapter_key not in adapter._post_delivery_callbacks @pytest.mark.asyncio From 1f769ead5f03e85c337bb03ad859ecc07195d467 Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Sun, 16 Aug 2026 20:05:18 +0200 Subject: [PATCH 5/9] fix(gateway): preserve FIFO across durable overflow --- gateway/run.py | 7 +- tests/gateway/test_terminal_steer_fifo.py | 87 +++++++++++++++++++++++ 2 files changed, 90 insertions(+), 4 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 57a1f5dbe669..8f5704438823 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -9010,10 +9010,9 @@ def _enqueue_fifo( if pending_slot is None: return slot_key = adapter_key or state_key - if slot_key in pending_slot: - self._session_state(state_key).conversation.queued_events.append( - queued_event - ) + overflow = self._session_state(state_key).conversation.queued_events + if slot_key in pending_slot or overflow: + overflow.append(queued_event) else: pending_slot[slot_key] = queued_event diff --git a/tests/gateway/test_terminal_steer_fifo.py b/tests/gateway/test_terminal_steer_fifo.py index 4a0264dbe217..8002702cc52a 100644 --- a/tests/gateway/test_terminal_steer_fifo.py +++ b/tests/gateway/test_terminal_steer_fifo.py @@ -145,6 +145,93 @@ def _queued_event(source: SessionSource) -> MessageEvent: ) +@pytest.mark.parametrize("profile", [None, "coder"], ids=["default", "named-profile"]) +def test_terminal_steer_enqueue_respects_durable_overflow_after_goal_clear( + monkeypatch, + tmp_path, + profile, +): + """An empty physical slot does not make a non-empty logical FIFO empty.""" + _gateway_run, runner, adapter = _make_runner(monkeypatch, tmp_path) + runner.config.multiplex_profiles = profile is not None + if profile is not None: + runner._profile_adapters = {profile: {Platform.TELEGRAM: adapter}} + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-goal-clear-terminal-steer", + chat_type="dm", + user_id="user-goal-clear", + profile=profile, + ) + state_key = runner._session_key_for_source(source) + adapter_key = adapter.session_key_for_source(source) + stale_goal = MessageEvent( + text="[Continuing toward your standing goal]\nGoal: stale goal", + message_type=MessageType.TEXT, + source=source, + ) + older_user = _queued_event(source) + terminal_steer = MessageEvent( + text="terminal steer", + message_type=MessageType.TEXT, + source=source, + internal=True, + ) + setattr(terminal_steer, "_gateway_terminal_steer", True) + + runner._enqueue_fifo( + state_key, + stale_goal, + adapter, + adapter_key=adapter_key, + ) + runner._enqueue_fifo( + state_key, + older_user, + adapter, + adapter_key=adapter_key, + ) + assert runner._clear_goal_pending_continuations( + state_key, + adapter, + source=source, + ) == 1 + assert adapter_key not in adapter._pending_messages + assert runner._queued_events[state_key] == [older_user] + + # This is the terminal-result caller's enqueue at gateway/run.py. The old + # implementation saw the empty physical slot and inserted the steer there, + # jumping it ahead of the older durable overflow item. + runner._enqueue_fifo( + state_key, + terminal_steer, + adapter, + adapter_key=adapter_key, + ) + + drained = [] + while True: + pending = adapter.get_pending_message(adapter_key) + pending = runner._promote_queued_event( + state_key, + adapter, + pending, + adapter_key=adapter_key, + ) + if pending is None: + break + drained.append(pending) + + assert [event.text for event in drained] == [ + "older queued work", + "terminal steer", + ] + assert drained[1] is terminal_steer + assert getattr(drained[1], "_gateway_terminal_steer", False) is True + assert adapter_key not in adapter._pending_messages + assert runner._queued_events.get(state_key, []) == [] + + @pytest.mark.asyncio @pytest.mark.parametrize( "queue_before_run", From 3198162ad59bc0739d0af6603b5ab4eac4cc01d7 Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Mon, 17 Aug 2026 07:09:15 +0200 Subject: [PATCH 6/9] fix(steer): seal Codex app-server terminal results --- agent/conversation_loop.py | 15 ++++++----- .../test_codex_app_server_integration.py | 25 +++++++++++++++++++ 2 files changed, 34 insertions(+), 6 deletions(-) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index c3db4cb63550..a789f341f247 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -1997,12 +1997,15 @@ def run_conversation( # See agent/transports/codex_app_server_session.py for the adapter # and references/codex-app-server-runtime.md for the rationale. if agent.api_mode == "codex_app_server": - return agent._run_codex_app_server_turn( - user_message=user_message, - original_user_message=original_user_message, - messages=messages, - effective_task_id=effective_task_id, - should_review_memory=_should_review_memory, + return _terminal_result_with_pending_steer( + agent, + agent._run_codex_app_server_turn( + user_message=user_message, + original_user_message=original_user_message, + messages=messages, + effective_task_id=effective_task_id, + should_review_memory=_should_review_memory, + ), ) while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call: diff --git a/tests/run_agent/test_codex_app_server_integration.py b/tests/run_agent/test_codex_app_server_integration.py index 7b89fa48c391..d57ab46ba736 100644 --- a/tests/run_agent/test_codex_app_server_integration.py +++ b/tests/run_agent/test_codex_app_server_integration.py @@ -86,6 +86,31 @@ def test_run_conversation_returns_codex_shape(self, fake_session): assert result["codex_thread_id"] == "thread-stub-1" assert result["codex_turn_id"] == "turn-stub-1" + def test_terminal_result_preserves_and_seals_late_steer(self): + """Codex's direct return must honor the ordinary terminal contract.""" + agent = _make_codex_agent() + + def fake_run_turn(*, user_input: str): + assert user_input == "hello" + assert agent.steer("include the migration note") is True + return TurnResult( + final_text="done", + projected_messages=[{"role": "assistant", "content": "done"}], + turn_id="turn-steer-1", + thread_id="thread-steer-1", + ) + + fake_codex_session = MagicMock() + fake_codex_session.run_turn.side_effect = fake_run_turn + setattr(agent, "_codex_session", fake_codex_session) + with patch.object(agent, "_spawn_background_review", return_value=None): + result = agent.run_conversation("hello") + + assert result["pending_steer"] == "include the migration note" + assert agent._pending_steer is None + assert agent._seal_pending_steer() is None + assert agent.steer("too late for this turn") is False + def test_codex_app_server_token_usage_updates_session_accounting(self, monkeypatch): def fake_run_turn(self, user_input: str, **kwargs): return TurnResult( From 231274447363e0ec78c2c53a3a48f24fdfe33bd4 Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Fri, 21 Aug 2026 11:20:33 +0200 Subject: [PATCH 7/9] fix(gateway): preserve profile-scoped adapter lanes --- gateway/platforms/base.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 5c70ceeac619..24ee266dcfc6 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -6079,8 +6079,7 @@ async def handle_message(self, event: MessageEvent) -> None: if needs_topic_recovery: await asyncio.to_thread(self._apply_topic_recovery, event) - session_key = self.session_key_for_source(event.source) - derived_state_key = build_session_key( + 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), @@ -6089,11 +6088,11 @@ async def handle_message(self, event: MessageEvent) -> None: expected_session_key = str( (event.metadata or {}).get("gateway_session_key") or "" ).strip() - if expected_session_key and derived_state_key != expected_session_key: + if expected_session_key and session_key != expected_session_key: logger.warning( "Dropping internally routed event: expected session=%s derived=%s", expected_session_key, - derived_state_key, + session_key, ) return From 64b405837db7eb685070bdcca5f0d14c36859b61 Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Fri, 21 Aug 2026 11:50:15 +0200 Subject: [PATCH 8/9] fix(gateway): align adapter session key namespaces --- gateway/platforms/base.py | 8 ++------ tests/gateway/test_multiplex_busy_input_mode.py | 10 ++++------ tests/gateway/test_run_cleanup_progress.py | 9 ++------- tests/gateway/test_run_progress_topics.py | 6 ++---- tests/gateway/test_terminal_steer_fifo.py | 4 ++-- 5 files changed, 12 insertions(+), 25 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 24ee266dcfc6..8ca8aa1c6491 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -6052,6 +6052,7 @@ def session_key_for_source(self, source: SessionSource) -> str: 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: @@ -6079,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() diff --git a/tests/gateway/test_multiplex_busy_input_mode.py b/tests/gateway/test_multiplex_busy_input_mode.py index 822a8fcde8ce..655f464be542 100644 --- a/tests/gateway/test_multiplex_busy_input_mode.py +++ b/tests/gateway/test_multiplex_busy_input_mode.py @@ -163,7 +163,7 @@ async def test_secondary_profile_busy_mode_controls_priority_path( event = _event(profile="research") session_key = runner._session_key_for_source(event.source) adapter_key = adapter.session_key_for_source(event.source) - assert adapter_key != session_key + assert adapter_key == session_key agent = MagicMock() agent._active_children = [] agent.steer.return_value = True @@ -175,7 +175,6 @@ async def test_secondary_profile_busy_mode_controls_priority_path( if secondary_mode == "queue": agent.steer.assert_not_called() assert adapter._pending_messages[adapter_key] is event - assert session_key not in adapter._pending_messages else: agent.steer.assert_called_once_with("follow up") assert adapter_key not in adapter._pending_messages @@ -206,11 +205,10 @@ async def test_secondary_profile_busy_mode_controls_busy_handler_restart_drain( event = _event(profile="research") session_key = runner._session_key_for_source(event.source) adapter_key = adapter.session_key_for_source(event.source) - assert adapter_key != session_key + assert adapter_key == session_key assert await runner._handle_active_session_busy_message(event, session_key) is True assert (adapter_key in adapter._pending_messages) is queued - assert session_key not in adapter._pending_messages @pytest.mark.asyncio @@ -230,7 +228,7 @@ async def test_secondary_profile_busy_mode_controls_priority_restart_drain( event = _event(profile="research") session_key = runner._session_key_for_source(event.source) adapter_key = adapter.session_key_for_source(event.source) - assert adapter_key != session_key + assert adapter_key == session_key agent = MagicMock() agent._active_children = [] runner._running_agents[session_key] = agent @@ -240,7 +238,6 @@ async def test_secondary_profile_busy_mode_controls_priority_restart_drain( assert isinstance(response, str) assert "queued" in response assert adapter._pending_messages[adapter_key] is event - assert session_key not in adapter._pending_messages agent.interrupt.assert_not_called() @@ -264,6 +261,7 @@ async def test_secondary_adapter_busy_guard_stamps_profile_before_resolving_mode # agent:main: key here asserted the pre-fix behaviour, where every profile's # adapter collapsed onto the default lane. adapter_session_key = build_session_key(event.source, profile="research") + assert adapter.session_key_for_source(event.source) == adapter_session_key adapter._active_sessions[adapter_session_key] = asyncio.Event() routed_source = _event(profile="research").source diff --git a/tests/gateway/test_run_cleanup_progress.py b/tests/gateway/test_run_cleanup_progress.py index 2b0780f8b07d..11e72242207f 100644 --- a/tests/gateway/test_run_cleanup_progress.py +++ b/tests/gateway/test_run_cleanup_progress.py @@ -318,16 +318,11 @@ def _preexisting_callback() -> None: @pytest.mark.asyncio -@pytest.mark.parametrize( - ("profile", "keys_differ"), - [("research", True), ("default", False)], - ids=["named-profile", "default-profile"], -) +@pytest.mark.parametrize("profile", ["research", "default"]) async def test_profiled_production_delivery_consumes_adapter_owned_callbacks( monkeypatch, tmp_path, profile, - keys_differ, ): """The real adapter lifecycle consumes review + cleanup callbacks once.""" adapter = CleanupCaptureAdapter() @@ -357,7 +352,7 @@ async def test_profiled_production_delivery_consumes_adapter_owned_callbacks( ) state_key = runner._session_key_for_source(source) adapter_key = adapter.session_key_for_source(source) - assert (state_key != adapter_key) is keys_differ + assert state_key == adapter_key generation = runner._begin_session_run_generation(state_key) async def _production_handler(inbound_event): diff --git a/tests/gateway/test_run_progress_topics.py b/tests/gateway/test_run_progress_topics.py index 96e0a6f649b8..27c87da513f6 100644 --- a/tests/gateway/test_run_progress_topics.py +++ b/tests/gateway/test_run_progress_topics.py @@ -1401,7 +1401,7 @@ async def test_run_agent_defers_background_review_notification_until_release(mon @pytest.mark.asyncio -async def test_named_profile_queued_delivery_pops_physical_callback_once( +async def test_named_profile_queued_delivery_pops_profiled_callback_once( monkeypatch, tmp_path, ): @@ -1415,8 +1415,7 @@ async def test_named_profile_queued_delivery_pops_physical_callback_once( profile="research", ) - adapter_key = "agent:main:telegram:group:-1001:17585" - state_key = "agent:research:telegram:group:-1001:17585" + adapter_key = "agent:research:telegram:group:-1001:17585" assert result["final_response"] == "done" for _ in range(50): if [item["content"] for item in adapter.sent].count( @@ -1428,7 +1427,6 @@ async def test_named_profile_queued_delivery_pops_physical_callback_once( sent_text = [item["content"] for item in adapter.sent] assert sent_text.count("💾 Skill 'prospect-scanner' created.") == 1 assert adapter_key in adapter._post_delivery_callbacks - assert state_key not in adapter._post_delivery_callbacks callback = adapter.pop_post_delivery_callback(adapter_key) assert callable(callback) diff --git a/tests/gateway/test_terminal_steer_fifo.py b/tests/gateway/test_terminal_steer_fifo.py index 8002702cc52a..ac7a24561777 100644 --- a/tests/gateway/test_terminal_steer_fifo.py +++ b/tests/gateway/test_terminal_steer_fifo.py @@ -356,7 +356,7 @@ async def test_command_shaped_terminal_steer_keeps_accepted_provenance( @pytest.mark.asyncio -async def test_named_profile_keeps_adapter_slot_and_durable_fifo_keys_separate( +async def test_named_profile_keeps_adapter_slot_and_durable_fifo_keys_aligned( monkeypatch, tmp_path, ): @@ -378,7 +378,7 @@ async def test_named_profile_keeps_adapter_slot_and_durable_fifo_keys_separate( ) state_key = runner._session_key_for_source(source) adapter_key = adapter.session_key_for_source(source) - assert state_key != adapter_key + assert state_key == adapter_key adapter._active_sessions[adapter_key] = _RecordingInterruptEvent() runner._enqueue_fifo( state_key, From e426eb5e460ea9a262bcbf64feedd6bf68edc8f2 Mon Sep 17 00:00:00 2001 From: Jakub Wolniewicz <4850809+frizikk@users.noreply.github.com> Date: Fri, 21 Aug 2026 12:03:18 +0200 Subject: [PATCH 9/9] fix(gateway): resume goals on profile adapter lane --- gateway/slash_commands.py | 14 ++++++++-- tests/gateway/test_goal_resume_restart.py | 34 +++++++++++++++++++++++ 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 01bb6c5bd7ac..74087d5c9a36 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -2750,9 +2750,14 @@ async def _handle_goal_command(self, event: "MessageEvent") -> str: # and pause/clear's stale-continuation cleanup recognizes it. prompt = mgr.next_continuation_prompt() try: - adapter = self.adapters.get(event.source.platform) if event.source else None + adapter = self._adapter_for_source(event.source) if event.source else None _quick_key = self._session_key_for_source(event.source) if event.source else None if prompt and adapter and _quick_key: + adapter_key = self._adapter_key_for_source( + adapter, + event.source, + fallback=_quick_key, + ) cont_event = MessageEvent( text=prompt, message_type=MessageType.TEXT, @@ -2760,7 +2765,12 @@ async def _handle_goal_command(self, event: "MessageEvent") -> str: message_id=None, channel_prompt=None, ) - self._enqueue_fifo(_quick_key, cont_event, adapter) + self._enqueue_fifo( + _quick_key, + cont_event, + adapter, + adapter_key=adapter_key, + ) except Exception as exc: logger.debug("goal resume: continuation enqueue failed: %s", exc) return t("gateway.goal.resumed", goal=state.goal) diff --git a/tests/gateway/test_goal_resume_restart.py b/tests/gateway/test_goal_resume_restart.py index bc7efa77452f..aa37b2bfcaf2 100644 --- a/tests/gateway/test_goal_resume_restart.py +++ b/tests/gateway/test_goal_resume_restart.py @@ -126,6 +126,15 @@ def __init__(self): self._pending_messages = {} +class _ProfileFakeAdapter(_FakeAdapter): + def __init__(self, physical_key): + super().__init__() + self.physical_key = physical_key + + def session_key_for_source(self, _source): + return self.physical_key + + def _make_runner() -> tuple[GatewayRunner, _FakeAdapter]: runner = object.__new__(GatewayRunner) runner.config = GatewayConfig( @@ -134,6 +143,7 @@ def _make_runner() -> tuple[GatewayRunner, _FakeAdapter]: runner.session_store = _FakeSessionStore() adapter = _FakeAdapter() runner.adapters = {Platform.DISCORD: adapter} + runner._profile_adapters = {} runner._queued_events = {} return runner, adapter @@ -185,3 +195,27 @@ async def test_resume_without_goal_enqueues_nothing(self, hermes_home): assert "No goal to resume" in response assert adapter._pending_messages == {} + + @pytest.mark.asyncio + async def test_named_profile_resume_uses_own_adapter_and_physical_slot( + self, hermes_home + ): + runner, default_adapter = _make_runner() + runner.config.multiplex_profiles = True + profile_key = "agent:coder:discord:channel:goal-resume" + profile_adapter = _ProfileFakeAdapter(profile_key) + runner._profile_adapters = { + "coder": {Platform.DISCORD: profile_adapter}, + } + event = _resume_event() + event.source.profile = "coder" + _exhaust_budget(_GW_SID) + + response = await GatewayRunner._handle_goal_command(runner, event) + + assert "resume" in response.lower() or "Goal" in response + assert default_adapter._pending_messages == {} + pending = profile_adapter._pending_messages.get(profile_key) + assert pending is not None + assert pending.source.profile == "coder" + assert pending.text.startswith("[Continuing toward your standing goal]")