"""Tests for AIAgent.steer() — mid-run user message injection. /steer lets the user add a note to the agent's next tool result without interrupting the current tool call. The agent sees the note inline with tool output on its next iteration, preserving message-role alternation and prompt-cache integrity. """ from __future__ import annotations import threading import pytest from agent.prompt_builder import STEER_MARKER_OPEN, format_steer_marker from run_agent import AIAgent def _bare_agent() -> AIAgent: """Build an AIAgent without running __init__, then install the steer state manually — matches the existing object.__new__ stub pattern used elsewhere in the test suite. """ agent = object.__new__(AIAgent) agent._pending_steer = None agent._pending_steer_lock = threading.Lock() agent._pending_redirect = None agent._pending_redirect_lock = threading.Lock() agent._model_request_active = threading.Event() agent._executing_tools = False agent._execution_thread_id = None agent._interrupt_thread_signal_pending = False agent._interrupt_requested = False agent._interrupt_message = None agent._active_children = [] agent._active_children_lock = threading.Lock() agent._tool_worker_threads = None agent._tool_worker_threads_lock = None agent._current_streamed_assistant_text = "" agent._stream_needs_break = False agent._strip_think_blocks = lambda content: content agent.quiet_mode = True agent.api_mode = "chat_completions" return agent class TestSteerAcceptance: def test_accepts_non_empty_text(self): agent = _bare_agent() assert agent.steer("go ahead and check the logs") is True assert agent._pending_steer == "go ahead and check the logs" class TestSteerDrain: def test_drain_returns_and_clears(self): agent = _bare_agent() agent.steer("hello") assert agent._drain_pending_steer() == "hello" assert agent._pending_steer is None class TestActiveTurnRedirect: def test_rejects_when_no_turn_is_active(self): agent = _bare_agent() assert agent.redirect("change course") is False assert agent._pending_redirect is None def test_cancels_only_an_active_model_request(self): agent = _bare_agent() agent._model_request_active.set() assert agent.redirect("use Postgres") is True assert agent._pending_redirect == "use Postgres" assert agent._interrupt_requested is True assert agent._interrupt_message is None def test_multiple_redirects_preserve_message_boundaries(self): agent = _bare_agent() agent._model_request_active.set() assert agent.redirect("first correction") is True assert agent.redirect("second correction") is True assert agent._pending_redirect == ( "first correction\n\n" "[Additional user correction]\n" "second correction" ) def test_hard_interrupt_wins_over_new_redirect(self): agent = _bare_agent() agent._model_request_active.set() agent._interrupt_requested = True assert agent.redirect("too late") is False assert agent._pending_redirect is None def test_reasoning_deltas_are_display_only(self): """Streamed reasoning must never accumulate into replayable transcript state — an assistant checkpoint that inlines chain-of-thought trips Anthropic's output classifier and permanently bricks the session (deterministic empty-response storms on every replay).""" agent = _bare_agent() seen = [] agent.reasoning_callback = seen.append agent._fire_reasoning_delta("visible provider thinking") # Displayed to the surface, but never checkpointed anywhere. assert seen == ["visible provider thinking"] assert not getattr(agent, "_current_streamed_reasoning_text", "") def test_response_completion_before_redirect_lock_rejects_correction(self): agent = _bare_agent() agent._model_request_active.set() started = threading.Event() outcome = {} def redirect(): started.set() outcome["accepted"] = agent.redirect("late correction") with agent._pending_redirect_lock: worker = threading.Thread(target=redirect) worker.start() assert started.wait(timeout=1) # Mirrors conversation_loop clearing the request-active marker # under this same lock before redirect can commit its slot. agent._model_request_active.clear() worker.join(timeout=1) assert outcome["accepted"] is False assert agent._pending_redirect is None def test_hard_stop_wins_concurrent_redirect(self): agent = _bare_agent() agent._model_request_active.set() start = threading.Barrier(3) outcome = {} def redirect(): start.wait() outcome["redirect"] = agent.redirect("change course") def hard_stop(): start.wait() agent.interrupt("stop requested") redirect_thread = threading.Thread(target=redirect) stop_thread = threading.Thread(target=hard_stop) redirect_thread.start() stop_thread.start() start.wait() redirect_thread.join(timeout=1) stop_thread.join(timeout=1) assert redirect_thread.is_alive() is False assert stop_thread.is_alive() is False assert agent._interrupt_requested is True assert agent._interrupt_message == "stop requested" assert agent._pending_redirect is None def test_codex_app_server_hard_stop_reaches_native_session(self): agent = _bare_agent() calls = [] agent.api_mode = "codex_app_server" agent._codex_session = type( "_CodexSession", (), {"request_interrupt": lambda self: calls.append("interrupt")}, )() agent.interrupt() assert calls == ["interrupt"] def test_redirect_during_tool_execution_uses_safe_steer_boundary(self): agent = _bare_agent() agent._executing_tools = True assert agent.redirect("also check migrations") is True assert agent._pending_redirect is None assert agent._pending_steer == "also check migrations" assert agent._interrupt_requested is False class TestActiveTurnRedirectCheckpoint: def test_assistant_tail_puts_correction_last(self): from agent.conversation_loop import _apply_active_turn_redirect agent = _bare_agent() agent._current_streamed_assistant_text = "Visible draft." messages = [ {"role": "user", "content": "start"}, {"role": "assistant", "content": "committed assistant item"}, ] _apply_active_turn_redirect(agent, messages, "Use Postgres instead.") assert [m["role"] for m in messages] == ["user", "assistant", "user"] assert messages[-1]["role"] == "user" assert messages[-1]["content"] == "Use Postgres instead." assert sum(1 for m in messages if m["role"] == "assistant") == 1 # Scaffolding is provider-replay text, carried in the sidecar so the # model still sees the interrupted context — never in the transcript. replayed = messages[-1]["api_content"] assert "Visible draft." in replayed assert "Context from the interrupted assistant response" in replayed assert replayed.endswith("Use Postgres instead.") def test_scaffolding_never_lands_in_transcript_content(self): """The checkpoint machinery is for the MODEL, not the transcript. Persisting ``[This response was interrupted by a user correction.]`` into ``content`` painted raw scaffolding as an assistant bubble on every reload. It must ride in ``api_content`` (replayed to the provider) while ``content`` stays clean, or be marked ``display_kind="hidden"`` when there is no clean form at all. """ from agent.conversation_loop import _apply_active_turn_redirect scaffolding = ( "[This response was interrupted by a user correction.]", "Visible response before the interruption:", "[Context from the interrupted assistant response]", ) for tail_role in ("tool", "assistant"): for streamed in ("Partial reply on screen.", ""): agent = _bare_agent() agent._current_streamed_assistant_text = streamed messages = [{"role": "user", "content": "start"}] if tail_role == "assistant": messages.append({"role": "assistant", "content": "committed"}) else: messages.append( {"role": "assistant", "tool_calls": [{"id": "a"}]} ) messages.append( {"role": "tool", "content": "out", "tool_call_id": "a"} ) _apply_active_turn_redirect(agent, messages, "New direction.") for msg in messages: if msg.get("display_kind") == "hidden": continue # dropped by every transcript surface content = str(msg.get("content", "")) for marker in scaffolding: assert marker not in content, ( f"scaffolding leaked into visible content " f"(tail={tail_role}, streamed={bool(streamed)}): {content!r}" ) # The user's correction is always shown verbatim. assert messages[-1]["content"] == "New direction." # ...and the model still receives the interrupted context. replayed = "".join( str(m.get("api_content") or m.get("content", "")) for m in messages ) assert "[This response was interrupted by a user correction.]" in replayed if streamed: assert streamed in replayed def test_checkpoint_never_replays_chain_of_thought(self): """Raw CoT serialized into checkpoint content reads to Anthropic's output classifier as reasoning-injection; because the checkpoint is persisted and replayed on every later call, one redirect during a thinking phase permanently bricked sessions with deterministic empty-response storms (July 2026). Reasoning must never appear in replayable content — in either the assistant-checkpoint or the merged-user-correction shape.""" from agent.conversation_loop import _apply_active_turn_redirect for tail_role in ("user", "assistant"): agent = _bare_agent() # Simulate a surface having displayed reasoning this turn. agent._current_streamed_reasoning_text = "SECRET chain of thought." agent._current_streamed_assistant_text = "Visible draft." messages = [{"role": "user", "content": "start"}] if tail_role == "assistant": messages.append({"role": "assistant", "content": "committed"}) _apply_active_turn_redirect(agent, messages, "Change course.") # Check BOTH the transcript content and the replayed sidecar — # the sidecar is what actually reaches the provider. serialized = "".join( str(m.get("content", "")) + str(m.get("api_content") or "") for m in messages ) assert "SECRET chain of thought." not in serialized assert "Reasoning shown before the interruption" not in serialized assert "Visible draft." in serialized def test_checkpoint_omits_reasoning_label_when_nothing_visible(self): from agent.conversation_loop import _apply_active_turn_redirect agent = _bare_agent() agent._current_streamed_reasoning_text = "thinking only, no text yet" messages = [{"role": "user", "content": "start"}] _apply_active_turn_redirect(agent, messages, "New direction.") checkpoint_row = messages[-2] # Nothing was on screen, so the row exists only for the model: hidden # from every transcript surface, scaffolding replayed via the sidecar. assert checkpoint_row["display_kind"] == "hidden" assert ( checkpoint_row["api_content"] == "[This response was interrupted by a user correction.]" ) assert messages[-1]["content"] == "New direction." class TestSteerInjection: def test_appends_to_last_tool_result(self): agent = _bare_agent() agent.steer("please also check auth.log") messages = [ {"role": "user", "content": "what's in /var/log?"}, {"role": "assistant", "tool_calls": [{"id": "a"}, {"id": "b"}]}, {"role": "tool", "content": "ls output A", "tool_call_id": "a"}, {"role": "tool", "content": "ls output B", "tool_call_id": "b"}, ] agent._apply_pending_steer_to_tool_results(messages, num_tool_msgs=2) # The LAST tool result is modified; earlier ones are untouched. assert messages[2]["content"] == "ls output A" assert "ls output B" in messages[3]["content"] assert STEER_MARKER_OPEN in messages[3]["content"] assert "please also check auth.log" in messages[3]["content"] # And pending_steer is consumed. assert agent._pending_steer is None def test_no_op_when_no_steer_pending(self): agent = _bare_agent() messages = [ {"role": "assistant", "tool_calls": [{"id": "a"}]}, {"role": "tool", "content": "output", "tool_call_id": "a"}, ] agent._apply_pending_steer_to_tool_results(messages, num_tool_msgs=1) assert messages[-1]["content"] == "output" # unchanged def test_marker_labels_text_as_out_of_band_user_message(self): """The injection marker must attribute the appended text to the user via the explicit out-of-band marker (which the system prompt tells the model to trust) — otherwise the model reads it as untrusted tool output and refuses it as suspected prompt injection. Cache-safe: it only rewrites existing tool content, never the message-role sequence. """ agent = _bare_agent() agent.steer("stop after next step") messages = [{"role": "tool", "content": "x", "tool_call_id": "1"}] agent._apply_pending_steer_to_tool_results(messages, num_tool_msgs=1) content = messages[-1]["content"] assert STEER_MARKER_OPEN in content assert "stop after next step" in content def test_multimodal_content_list_preserved(self): """Anthropic-style list content should be preserved, with the steer appended as a text block.""" agent = _bare_agent() agent.steer("extra note") original_blocks = [{"type": "text", "text": "existing output"}] messages = [ {"role": "tool", "content": list(original_blocks), "tool_call_id": "1"} ] agent._apply_pending_steer_to_tool_results(messages, num_tool_msgs=1) new_content = messages[-1]["content"] assert isinstance(new_content, list) assert len(new_content) == 2 assert new_content[0] == {"type": "text", "text": "existing output"} assert new_content[1]["type"] == "text" assert "extra note" in new_content[1]["text"] class TestSteerThreadSafety: def test_concurrent_steer_calls_preserve_all_text(self): agent = _bare_agent() N = 200 def worker(idx: int) -> None: agent.steer(f"note-{idx}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(N)] for t in threads: t.start() for t in threads: t.join() text = agent._drain_pending_steer() assert text is not None # Every single note must be preserved — none dropped by the lock. lines = text.split("\n") assert len(lines) == N assert set(lines) == {f"note-{i}" for i in range(N)} class TestSteerClearedOnInterrupt: def test_clear_interrupt_drops_pending_steer(self): """A hard interrupt supersedes any pending steer — the agent's next tool iteration won't happen, so delivering the steer later would be surprising.""" agent = _bare_agent() # Minimal surface needed by clear_interrupt() agent._interrupt_requested = True agent._interrupt_message = None agent._interrupt_thread_signal_pending = False agent._execution_thread_id = None agent._tool_worker_threads = None agent._tool_worker_threads_lock = None agent.steer("will be dropped") agent._pending_redirect = "also drop this" assert agent._pending_steer == "will be dropped" agent.clear_interrupt() assert agent._pending_steer is None assert agent._pending_redirect is None class TestPreApiCallSteerDrain: """Test that steers arriving during an API call are drained before the next API call — not deferred until the next tool batch. This is the fix for the scenario where /steer sent during model thinking only lands after the agent is completely done.""" def test_pre_api_drain_injects_into_last_tool_result(self): """If a steer is pending when the main loop starts building api_messages, it should be injected into the last tool result in the messages list.""" agent = _bare_agent() # Simulate messages after a tool batch completed messages = [ {"role": "user", "content": "do something"}, {"role": "assistant", "content": "ok", "tool_calls": [ {"id": "tc1", "function": {"name": "terminal", "arguments": "{}"}} ]}, {"role": "tool", "content": "output here", "tool_call_id": "tc1"}, ] # Steer arrives during API call (set after tool execution) agent.steer("focus on error handling") # Simulate what the pre-API-call drain does: _pre_api_steer = agent._drain_pending_steer() assert _pre_api_steer == "focus on error handling" # Inject into last tool msg (mirrors the new code in run_conversation) for _si in range(len(messages) - 1, -1, -1): if messages[_si].get("role") == "tool": messages[_si]["content"] += format_steer_marker(_pre_api_steer) break assert STEER_MARKER_OPEN in messages[-1]["content"] assert "focus on error handling" in messages[-1]["content"] assert agent._pending_steer is None def test_pre_api_drain_restashes_when_no_tool_message(self): """If there are no tool results yet (first iteration), the steer should be put back into _pending_steer for the post-tool drain.""" agent = _bare_agent() messages = [ {"role": "user", "content": "hello"}, ] agent.steer("early steer") _pre_api_steer = agent._drain_pending_steer() assert _pre_api_steer == "early steer" # No tool message found — put it back found = False for _si in range(len(messages) - 1, -1, -1): if messages[_si].get("role") == "tool": found = True break assert not found # Restash agent._pending_steer = _pre_api_steer assert agent._pending_steer == "early steer" class TestSteerMarkerContract: def test_system_prompt_note_describes_the_real_marker(self): """The system-prompt note tells the model which marker to trust; it must reference the exact open/close the injector emits, or the model trusts a marker that never appears (and vice-versa).""" from agent.prompt_builder import STEER_CHANNEL_NOTE, STEER_MARKER_CLOSE emitted = format_steer_marker("hi") assert STEER_MARKER_OPEN in emitted and STEER_MARKER_CLOSE in emitted assert STEER_MARKER_OPEN in STEER_CHANNEL_NOTE and STEER_MARKER_CLOSE in STEER_CHANNEL_NOTE def test_marker_no_longer_uses_the_distrusted_label(self): """Regression: the bare 'User guidance:' line read as tool content and got refused as injection — it must not come back.""" assert "User guidance:" not in format_steer_marker("hi") class TestSteerCommandRegistry: def test_steer_in_command_registry(self): """The /steer slash command must be registered so it reaches all platforms (CLI, gateway, TUI autocomplete, Telegram/Slack menus). """ from hermes_cli.commands import resolve_command cmd = resolve_command("steer") assert cmd is not None assert cmd.name == "steer" assert cmd.category == "Session" assert cmd.args_hint == "" def test_steer_in_bypass_set(self): """When the agent is running, /steer MUST bypass the Level-1 base-adapter queue so it reaches the gateway runner's /steer handler. Otherwise it would be queued as user text and only delivered at turn end — defeating the whole point. """ from hermes_cli.commands import ACTIVE_SESSION_BYPASS_COMMANDS, should_bypass_active_session assert "steer" in ACTIVE_SESSION_BYPASS_COMMANDS assert should_bypass_active_session("steer") is True if __name__ == "__main__": # pragma: no cover pytest.main([__file__, "-v"])