fix(acp): run /steer as a regular prompt on idle sessions (#18258)

When a user types /steer <text> on an ACP session that isn't actively
running a turn (and there's no interrupted-prompt salvage available),
_cmd_steer silently appended to state.queued_prompts and replied
"No active turn — queued for the next turn". That looks identical to
/queue output even though the user never typed /queue — @EddyLeeKhane
reported this as "/steer never works, gets queued instead".

Rewrite the payload to a plain user prompt before the slash-intercept
fires, matching the gateway's idle-/steer fallthrough in
gateway/run.py ~L4898.

acp_adapter/server.py

@@ -669,24 +669,38 @@ class HermesACPAgent(acp.Agent):
         if not has_content:
             return PromptResponse(stop_reason="end_turn")
 
-        # Zed currently interrupts an active ACP request before delivering a
-        # follow-up slash command. If that follow-up is /steer, there may be no
-        # live AIAgent left to steer by the time this method runs. Salvage that
-        # UX by replaying the interrupted prompt with the steer text attached as
-        # explicit correction/guidance.
+        # /steer on an idle session has no in-flight tool call to inject into.
+        # Rewrite it so the payload runs as a normal user prompt, matching the
+        # gateway's behavior (gateway/run.py ~L4898). Two sub-cases:
+        #   1. Zed-interrupt salvage — a prior prompt was cancelled by the
+        #      client right before /steer arrived; replay it with the steer
+        #      text attached as explicit correction/guidance so the user's
+        #      in-flight work isn't lost.
+        #   2. Plain idle — no prior work to salvage; just run the steer
+        #      payload as a regular prompt. Without this, _cmd_steer would
+        #      silently append to state.queued_prompts and respond with
+        #      "No active turn — queued for the next turn", which looks like
+        #      /queue even though the user never typed /queue.
         if isinstance(user_content, str) and user_text.startswith("/steer"):
             steer_text = user_text.split(maxsplit=1)[1].strip() if len(user_text.split(maxsplit=1)) > 1 else ""
             interrupted_prompt = ""
+            rewrite_idle = False
             with state.runtime_lock:
-                if not state.is_running and steer_text and state.interrupted_prompt_text:
+                if not state.is_running and steer_text:
+                    if state.interrupted_prompt_text:
                         interrupted_prompt = state.interrupted_prompt_text
                         state.interrupted_prompt_text = ""
+                    else:
+                        rewrite_idle = True
             if interrupted_prompt:
                 user_text = (
                     f"{interrupted_prompt}\n\n"
                     f"User correction/guidance after interrupt: {steer_text}"
                 )
                 user_content = user_text
+            elif rewrite_idle:
+                user_text = steer_text
+                user_content = steer_text
 
         # Intercept slash commands — handle locally without calling the LLM.
         # Slash commands are text-only; if the client included images/resources,

tests/acp_adapter/test_acp_commands.py

@@ -99,6 +99,26 @@ async def test_acp_steer_after_zed_interrupt_replays_interrupted_prompt_with_gui
     assert state.interrupted_prompt_text == ""
 
 
+@pytest.mark.asyncio
+async def test_acp_steer_on_idle_session_runs_as_regular_prompt():
+    # /steer on an idle session (no running turn, nothing to salvage) should
+    # run the steer payload as a normal user prompt — NOT silently append it
+    # to state.queued_prompts. Without this, users on Zed / other ACP clients
+    # see their /steer turn into "queued for the next turn" when they never
+    # typed /queue. Matches gateway/run.py ~L4898 idle-/steer behavior.
+    acp_agent, state, fake, _conn = make_agent_and_state()
+
+    response = await acp_agent.prompt(
+        session_id=state.session_id,
+        prompt=[TextContentBlock(type="text", text="/steer summarize the README")],
+    )
+
+    assert response.stop_reason == "end_turn"
+    assert fake.steers == []
+    assert fake.runs == ["summarize the README"]
+    assert state.queued_prompts == []
+
+
 @pytest.mark.asyncio
 async def test_acp_queue_slash_command_adds_next_turn_without_running_now():
     acp_agent, state, fake, _conn = make_agent_and_state()