hermes-agent/tests/agent/test_steer.py

"""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
from tools.registry import registry

# Registry handler for the end-to-end steer-survival test below (module level,
# like the built-in tool files — dispatch looks the tool up here at run time).
_STEER_SURVIVAL_TOOL = "steer_survival_probe"


def _steer_survival_tool(args, **_kwargs):
    return "probe ok"


registry.register(
    name=_STEER_SURVIVAL_TOOL,
    toolset="utility",
    schema={
        "name": _STEER_SURVIVAL_TOOL,
        "description": "probe tool for the steer-survival regression test",
        "parameters": {"type": "object", "properties": {}, "required": []},
    },
    handler=_steer_survival_tool,
    override=True,
)


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 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
        pending = agent._pending_redirect
        assert "first correction" in pending and "second correction" in pending
        assert pending.index("first correction") < pending.index("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_repetition_dominated_partial_is_not_replayed(self):
        """A looped partial must not seed the correction's next API request (#112764): neither
        the replayed correction nor the alternation placeholder carries the bytes; the model
        is told the reply degenerated instead."""
        from agent.conversation_loop import _apply_active_turn_redirect
        from agent.repetition_guard import REPETITION_LOOP_INTERRUPTED

        repeated = ("The same degenerate partial response keeps repeating without progress.\n" * 10)
        agent = _bare_agent()
        agent._current_streamed_assistant_text = repeated
        messages = [{"role": "user", "content": "start"}]

        _apply_active_turn_redirect(agent, messages, "Change course.")

        placeholder, correction = messages[-2], messages[-1]
        replayed = correction["api_content"]
        assert repeated not in replayed
        assert "Visible response before the interruption:" not in replayed
        assert REPETITION_LOOP_INTERRUPTED in replayed
        assert placeholder["role"] == "assistant"
        assert placeholder["display_kind"] == "hidden"
        assert repeated not in (placeholder.get("content") or "")
        assert repeated not in (placeholder.get("api_content") or "")

    def test_ordinary_partial_remains_replayable(self):
        """Useful interrupted text remains available to the corrected request."""
        from agent.conversation_loop import _apply_active_turn_redirect

        visible = "I found the migration entry point and was about to inspect it."
        agent = _bare_agent()
        agent._current_streamed_assistant_text = visible
        messages = [{"role": "user", "content": "start"}]

        _apply_active_turn_redirect(agent, messages, "Change course.")

        assert visible in messages[-1]["api_content"]

    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




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_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_standalone_user_message_after_tool_results(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)
        # Existing tool rows are untouched (append-only persistence contract);
        # the steer becomes a NEW user message at the tail.
        assert messages[2]["content"] == "ls output A"
        assert messages[3]["content"] == "ls output B"
        assert messages[-1]["role"] == "user"
        assert STEER_MARKER_OPEN in messages[-1]["content"]
        assert "please also check auth.log" in messages[-1]["content"]
        # Role-alternation pattern: assistant(tool_calls) → tool → user is
        # the documented legal "user jumped in mid-run" shape.
        # And pending_steer is consumed.
        assert agent._pending_steer is None

    def test_appended_user_message_is_persistable(self):
        """The appended user dict carries no _DB_PERSISTED_MARKER yet, so the
        next _flush_messages_to_session_db writes it to state.db — the steer
        text lands in the durable transcript (messages.content, role=user)."""
        from agent.context_compressor import _DB_PERSISTED_MARKER

        agent = _bare_agent()
        agent.steer("remember this decision")
        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]["role"] == "user"
        assert _DB_PERSISTED_MARKER not in messages[-1]

    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_persisted_steer_row_is_never_merged_with_the_next_prompt(self):
        """A run that ends right after a steered batch leaves user(steer) as the persisted tail.
        The next real prompt makes two consecutive user rows; the alternation repair must leave
        the steer row byte-identical (append-only persistence cannot follow an in-place merge)."""
        from agent.agent_runtime_helpers import _merge_consecutive_users
        from agent.prompt_builder import steer_user_row

        steer = steer_user_row("focus on error handling")
        before = dict(steer)
        merged, repairs = _merge_consecutive_users([steer, {"role": "user", "content": "next question"}])
        assert steer == before
        assert repairs == 0
        assert [m["role"] for m in merged] == ["user", "user"]
        assert merged[-1]["content"] == "next question"

    def test_multimodal_tool_content_untouched_steer_lands_as_user_row(self):
        """Anthropic-style list content on tool results is left untouched —
        the steer is appended as a standalone user message instead of being
        merged into the content blocks."""
        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)
        assert messages[0]["content"] == original_blocks       # untouched
        assert messages[-1]["role"] == "user"
        assert "extra note" in messages[-1]["content"]



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_hard_cancel_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. Only the explicit hard-cancel clear drops it."""
        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(hard_cancel=True)
        assert agent._pending_steer is None
        assert agent._pending_redirect is None

    def test_soft_clear_preserves_pending_steer(self):
        """A soft clear (redirect rebuild, error recovery, turn-boundary
        hygiene) keeps the session alive, so an already-accepted steer must
        survive it — the existing drains deliver it on the continued run.
        Dropping it here silently lost a user message the surface had
        already acknowledged as delivered."""
        agent = _bare_agent()
        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("must survive the rebuild")
        agent._pending_redirect = "correction"

        agent.clear_interrupt(preserve_redirect=True)
        assert agent._pending_steer == "must survive the rebuild"
        # preserve_redirect semantics unchanged: the correction survives too.
        assert agent._pending_redirect == "correction"

        # A plain soft clear (no flags) preserves the steer as well.
        agent.clear_interrupt()
        assert agent._pending_steer == "must survive the rebuild"
        assert agent._pending_redirect is None


class TestSteerSurvivesRedirectRebuild:
    """A steer accepted while a redirect lands must still reach a later API
    payload. Regression for the busy-redirect message loss: the rebuild's
    ``clear_interrupt(preserve_redirect=True)`` used to wipe ``_pending_steer``
    unconditionally, so a user message the surface had already acknowledged
    as delivered evaporated with no trace (no payload, no leftover, no log)."""

    STEER_TEXT = "STEER_TEXT_ONE"
    REDIRECT_TEXT = "REDIRECT_TEXT_TWO"

    def _loop_agent(self):
        from unittest.mock import MagicMock, patch

        from run_agent import AIAgent

        tool_schema = {
            "type": "function",
            "function": {
                "name": _STEER_SURVIVAL_TOOL,
                "description": "probe tool for the steer-survival regression test",
                "parameters": {"type": "object", "properties": {}, "required": []},
            },
        }
        with (
            patch("model_tools.get_tool_definitions", return_value=[tool_schema]),
            patch("model_tools.check_toolset_requirements", return_value={}),
            patch("agent.process_bootstrap.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._disable_streaming = True
        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_steer_accepted_before_redirect_lands_in_rebuilt_payload(self):
        from types import SimpleNamespace
        from unittest.mock import patch

        from tests.agent.test_run_agent import _mock_response

        agent = self._loop_agent()
        payloads = []

        def model_call(api_kwargs):
            payloads.append([dict(m) for m in api_kwargs["messages"]])
            if len(payloads) == 1:
                tool_call = SimpleNamespace(
                    id="call_1", type="function",
                    function=SimpleNamespace(name=_STEER_SURVIVAL_TOOL, arguments="{}"),
                )
                return _mock_response(
                    content=None, finish_reason="tool_calls", tool_calls=[tool_call]
                )
            if len(payloads) == 2:
                # Both mid-turn user messages race the in-flight request:
                # the steer is accepted first, then the redirect kills the
                # request and arms the rebuild.
                assert agent.steer(self.STEER_TEXT) is True
                assert agent.redirect(self.REDIRECT_TEXT) is True
                raise InterruptedError("redirect cancelled the in-flight request")
            return _mock_response(content="rebuilt reply", finish_reason="stop")

        agent._interruptible_api_call = model_call

        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"),
        ):
            result = agent.run_conversation("start something")

        blob = "\n".join(
            str(m.get("content"))
            for call in payloads
            for m in call
            if isinstance(m, dict)
        )
        # The rebuild reached the wire: the redirect correction is a real user message.
        assert self.REDIRECT_TEXT in blob
        # The steer accepted just before the redirect must ride the same rebuild
        # (injected into the newest tool result by the pre-API drain).
        assert self.STEER_TEXT in blob
        # Fully consumed: nothing left for the finalizer's leftover handoff.
        assert result.get("pending_steer") is None
        assert result["completed"] is True


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_appends_user_row_and_leaves_tool_row_untouched(self):
        """A steer pending when the loop builds api_messages lands THIS iteration as a
        standalone user row after the newest tool result; the (already persisted, append-only)
        tool row is byte-identical afterwards so replay cannot diverge from the live request."""
        from agent.turn_iteration_prep import _inject_steer_after_newest_tool_result

        agent = _bare_agent()
        tool_row = {"role": "tool", "content": "output here", "tool_call_id": "tc1"}
        before = dict(tool_row)
        messages = [
            {"role": "user", "content": "do something"},
            {"role": "assistant", "content": "ok", "tool_calls": [
                {"id": "tc1", "function": {"name": "terminal", "arguments": "{}"}}
            ]},
            tool_row,
        ]
        agent.steer("focus on error handling")
        _inject_steer_after_newest_tool_result(agent, messages, agent._drain_pending_steer())
        assert tool_row == before
        assert messages[-2] is tool_row
        assert messages[-1]["role"] == "user"
        assert STEER_MARKER_OPEN in messages[-1]["content"]
        assert "focus on error handling" in messages[-1]["content"]
        assert agent._pending_steer is None




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





class TestSteerRowIsHumanInput:
    def test_steer_row_counts_as_a_user_originated_turn(self, tmp_path):
        """The steer row is typed (renderer label, alternation-repair guard) but it carries full user
        authority: every "is this human input" predicate — in memory and the DB pick /undo uses — must
        agree with the anchor-restoration predicate that already accepts it."""
        from agent.context_compressor import ContextCompressor, is_user_originated_turn
        from agent.conversation_compression import _is_real_user_message
        from agent.prompt_builder import steer_user_row
        from hermes_state import SessionDB

        row = steer_user_row("focus on the error handling")
        assert _is_real_user_message(row)
        assert is_user_originated_turn(row)
        assert ContextCompressor._is_actionable_user_turn(row)

        db = SessionDB(db_path=tmp_path / "state.db")
        try:
            db.create_session(session_id="s1", source="cli")
            db.append_message("s1", role="user", content="first question")
            db.append_message("s1", role="assistant", content="first answer")
            db.append_message("s1", role="user", content=row["content"], display_kind=row["display_kind"])
            recents = db.list_recent_user_messages("s1", limit=5)
            assert [r["preview"][:5] for r in recents] == [row["content"][:5], "first"]
        finally:
            db.close()


class TestSteerCommandRegistry:

    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("model_tools.get_tool_definitions", return_value=[]),
            patch("model_tools.check_toolset_requirements", return_value={}),
            patch("agent.process_bootstrap.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.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.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"