"""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 an assistant row's ``content`` or ``api_content`` painted raw scaffolding as the model's own prior reply (#81841). Scaffold bytes ride only in the *user correction's* ``api_content``; assistant placeholders stay clean (or ``display_kind="hidden"`` when empty). """ 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("role") == "assistant": # Scaffold must never live on an assistant row at all # — content OR api_content (API replay substitutes the # sidecar back into content). blob = ( str(msg.get("content") or "") + str(msg.get("api_content") or "") ) for marker in scaffolding: assert marker not in blob, ( f"scaffolding leaked into assistant row " f"(tail={tail_role}, streamed={bool(streamed)}): " f"{blob!r}" ) 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, # but only via the user correction's api_content sidecar. replayed = messages[-1].get("api_content") or "" 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.") placeholder = messages[-2] correction = messages[-1] # Nothing was on screen: empty hidden placeholder for alternation; # scaffold rides only on the user correction's api_content. assert placeholder["role"] == "assistant" assert placeholder["display_kind"] == "hidden" assert placeholder.get("content") == "" # Neutral provider-replay payload (#88955): keeps the row out of the # re-heal sanitizer loop; the interrupt scaffold is still never here. assert placeholder.get("api_content") == "[response interrupted]" assert correction["content"] == "New direction." assert ( "[This response was interrupted by a user correction.]" in correction["api_content"] ) def test_tool_tail_scaffold_never_on_assistant_api_content(self): """#81841: mid-tool steer must not put the interrupt scaffold on the placeholder assistant row (that is what the model echoed).""" from agent.conversation_loop import _apply_active_turn_redirect agent = _bare_agent() messages = [ {"role": "user", "content": "start"}, {"role": "assistant", "tool_calls": [{"id": "a"}]}, {"role": "tool", "content": "out", "tool_call_id": "a"}, ] _apply_active_turn_redirect(agent, messages, "Stop and do X instead.") placeholder = messages[-2] correction = messages[-1] assert placeholder["role"] == "assistant" assert placeholder.get("display_kind") == "hidden" assert placeholder.get("content") == "" # Neutral provider-replay payload (#88955), NOT the interrupt scaffold. assert placeholder.get("api_content") == "[response interrupted]" assert correction["role"] == "user" assert correction["content"] == "Stop and do X instead." assert correction["api_content"].startswith( "[Context from the interrupted assistant response]\n" "[This response was interrupted by a user correction.]" ) class TestEmptyHiddenAssistantRehealRegression: """#88955: a no-visible-text redirect persisted an empty ``display_kind="hidden"`` assistant placeholder that the pre-call sanitizer re-healed on every later call (wire copy only, so the loop never converged). The placeholder must carry a neutral provider-replay ``api_content`` so the historical API projection fills ``content`` and the sanitizer stops touching the row — while the durable transcript stays hidden and empty.""" def test_active_turn_redirect_hidden_placeholder_has_provider_replay_payload(self): from agent.conversation_loop import _apply_active_turn_redirect agent = _bare_agent() agent._current_streamed_assistant_text = "" messages = [{"role": "user", "content": "start"}] _apply_active_turn_redirect(agent, messages, "Use Postgres instead.") placeholder = messages[-2] correction = messages[-1] assert placeholder["role"] == "assistant" assert placeholder["content"] == "" assert placeholder["display_kind"] == "hidden" assert placeholder["api_content"] == "[response interrupted]" # The user correction keeps clean text in content and the interruption # context only in its own api_content sidecar. assert correction["role"] == "user" assert correction["content"] == "Use Postgres instead." assert ( "[This response was interrupted by a user correction.]" in correction["api_content"] ) # #81841: the interrupt scaffold must never reach assistant content or # api_content (API replay substitutes api_content back into content). assert ( "[This response was interrupted by a user correction.]" not in str(placeholder.get("content") or "") + str(placeholder.get("api_content") or "") ) def test_hidden_redirect_placeholder_does_not_reheal_on_repeated_projection(self): from agent.agent_runtime_helpers import ( _msg_has_payload, repair_empty_non_final_messages, ) from agent.conversation_loop import _apply_active_turn_redirect agent = _bare_agent() agent._current_streamed_assistant_text = "" messages = [{"role": "user", "content": "start"}] _apply_active_turn_redirect(agent, messages, "Do X instead.") durable = list(messages) def project(rows): """Mirror the real send-time projection (conversation_loop.py): api_content -> content for historical user/assistant rows, and the display/row bookkeeping stripped from every outgoing copy.""" out = [] for msg in rows: api_msg = dict(msg) _api_content = api_msg.pop("api_content", None) api_msg.pop("display_kind", None) api_msg.pop("display_metadata", None) api_msg.pop("_row_id", None) if ( isinstance(_api_content, str) and _api_content and msg.get("role") in ("user", "assistant") ): api_msg["content"] = _api_content out.append(api_msg) return out for _pass in range(2): projected = project(durable) hidden_assistant = next( m for m in projected if m.get("role") == "assistant" ) # The provider replay sidecar was projected into content, so the # row already carries payload and the sanitizer has nothing to heal. assert _msg_has_payload(hidden_assistant) is True assert hidden_assistant["content"] == "[response interrupted]" assert "display_kind" not in hidden_assistant assert "api_content" not in hidden_assistant healed = repair_empty_non_final_messages(projected) healed_assistant = next( m for m in healed if m.get("role") == "assistant" ) assert healed_assistant["content"] == "[response interrupted]" assert "display_kind" not in healed_assistant assert "api_content" not in healed_assistant # Durable transcript is never mutated by projection or sanitizer. assert durable == messages assert durable[1]["content"] == "" assert durable[1]["display_kind"] == "hidden" assert durable[1]["api_content"] == "[response interrupted]" # #81841 scaffold never appears on the assistant wire. assert ( "[This response was interrupted by a user correction.]" not in healed_assistant["content"] ) def test_empty_non_final_sanitizer_still_repairs_unmarked_empty_assistant(self): """Control: a genuinely empty non-final assistant with no provider-replay sidecar is still healed — the fix must not disable the generic net.""" from agent.agent_runtime_helpers import repair_empty_non_final_messages rows = [ {"role": "user", "content": "start"}, {"role": "assistant", "content": "", "display_kind": "hidden"}, {"role": "user", "content": "correction"}, ] healed = repair_empty_non_final_messages(rows) assistant = next(m for m in healed if m.get("role") == "assistant") assert assistant["content"] == "[response interrupted]" # The durable list is not mutated (wire-copy-only design). assert rows[1]["content"] == "" 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_system_prompt_scopes_freshness_to_unanswered_marker(self): """A delivered marker remains in immutable history on later API calls. The prompt contract must distinguish the unanswered tail occurrence from one followed by an assistant response, or a model can interpret a historical steer as newly delivered and repeat non-idempotent work. """ from agent.prompt_builder import STEER_CHANNEL_NOTE assert "latest tool-result batch" in STEER_CHANNEL_NOTE assert "no later assistant message follows it" in STEER_CHANNEL_NOTE assert "do not treat it as a new message" in STEER_CHANNEL_NOTE assert "repeat completed work" in STEER_CHANNEL_NOTE emitted = format_steer_marker("deploy once") assert "delivered once at this position" in emitted assert "not a new delivery when replayed" in emitted 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"]) class TestLegacyHiddenPlaceholderWireSubstitution: """Projection-side half of #88955: rows persisted BEFORE the writer-side ``api_content`` stamp are ``content=""`` + ``display_kind="hidden"`` with no sidecar. The send-time projection must give the WIRE copy the neutral ``[response interrupted]`` payload so legacy sessions converge instead of re-healing forever — while the durable row stays hidden and empty.""" def _loop_agent(self): from unittest.mock import MagicMock, patch from run_agent import AIAgent with ( patch("run_agent.get_tool_definitions", return_value=[]), patch("run_agent.check_toolset_requirements", return_value={}), patch("run_agent.OpenAI"), ): agent = AIAgent( api_key="test-key-1234567890", base_url="https://openrouter.ai/api/v1", quiet_mode=True, skip_context_files=True, skip_memory=True, ) agent.client = MagicMock() agent._cached_system_prompt = "You are helpful." agent._use_prompt_caching = False agent.tool_delay = 0 agent.compression_enabled = False agent.save_trajectories = False return agent def test_legacy_empty_hidden_assistant_row_gets_neutral_wire_payload(self): """The projection itself must fill the row — the sanitizer must have NOTHING left to heal (its per-turn warning spam IS the bug).""" from unittest.mock import patch import agent.agent_runtime_helpers as _arh from tests.run_agent.test_run_agent import _mock_response agent = self._loop_agent() agent.client.chat.completions.create.side_effect = [ _mock_response(content="ok", finish_reason="stop"), ] sanitizer_inputs = [] _real_repair = _arh.repair_empty_non_final_messages def _spy_repair(messages, *a, **k): sanitizer_inputs.append( [ (m.get("role"), m.get("content")) for m in messages if isinstance(m, dict) ] ) return _real_repair(messages, *a, **k) # Legacy pre-fix row: no api_content sidecar. history = [ {"role": "user", "content": "start"}, {"role": "assistant", "content": "", "display_kind": "hidden"}, {"role": "user", "content": "correction", "finish_reason": "stop"}, {"role": "assistant", "content": "earlier reply", "finish_reason": "stop"}, ] with ( patch.object(agent, "_flush_messages_to_session_db"), patch.object(agent, "_persist_session"), patch.object(agent, "_save_trajectory"), patch.object(agent, "_cleanup_task_resources"), patch.object( _arh, "repair_empty_non_final_messages", side_effect=_spy_repair ), ): agent.run_conversation("next question", conversation_history=history) # Precondition: the sanitizer actually ran on this call path. assert sanitizer_inputs, "sanitizer was never invoked — test is vacuous" # The projection already filled the legacy row BEFORE sanitization: # every assistant row the sanitizer saw carried payload, so it healed 0. for snapshot in sanitizer_inputs: for role, content in snapshot: if role == "assistant": assert (content or "").strip(), ( "sanitizer still received an empty assistant row — " "the re-heal loop is back (#88955)" ) wire = agent.client.chat.completions.create.call_args.kwargs["messages"] wire_assistants = [m for m in wire if m.get("role") == "assistant"] legacy = wire_assistants[0] # Substituted on the wire by the projection (not the sanitizer): assert legacy["content"] == "[response interrupted]" assert "display_kind" not in legacy # #81841: never the interrupt scaffold. assert "[This response was interrupted" not in legacy["content"] # Durable history untouched. assert history[1]["content"] == "" assert history[1]["display_kind"] == "hidden" assert "api_content" not in history[1] def test_hidden_row_with_tool_calls_or_text_is_not_touched(self): from agent.conversation_loop import _clone_message_for_send # noqa: F401 from unittest.mock import patch from tests.run_agent.test_run_agent import _mock_response agent = self._loop_agent() agent.client.chat.completions.create.side_effect = [ _mock_response(content="ok", finish_reason="stop"), ] history = [ {"role": "user", "content": "start"}, { "role": "assistant", "content": "visible text", "display_kind": "hidden", "finish_reason": "stop", }, {"role": "user", "content": "more"}, {"role": "assistant", "content": "reply", "finish_reason": "stop"}, ] with ( patch.object(agent, "_flush_messages_to_session_db"), patch.object(agent, "_persist_session"), patch.object(agent, "_save_trajectory"), patch.object(agent, "_cleanup_task_resources"), ): agent.run_conversation("next", conversation_history=history) wire = agent.client.chat.completions.create.call_args.kwargs["messages"] wire_assistants = [m for m in wire if m.get("role") == "assistant"] assert wire_assistants[0]["content"] == "visible text"