fix(tui_gateway): stop replaying live-turn user text after redirect

A mid-turn correction (Desktop session.redirect / busy-input interrupt redirect)
must not leave a server-queue self-copy of the live inflight user prompt.
Otherwise post-turn _drain_queued_prompt restarts that original text as a
fresh agent turn after Q completes (#84417).

Scrub text-only self-duplicates of inflight_turn.user on successful
redirect/steer, refuse admitting them in _enqueue_prompt, rewrite merged
"{P}\n\n{Q}" slots to Q-only, bump _queued_prompt_generation on compression
session rotation, and restore the claimed queue envelope when generation
cancels mid-drain. Stabilize profile-scoped agent-build unit tests under CI load.

Fixes #84417

tests/test_tui_gateway_queue_on_busy.py

@@ -79,6 +79,237 @@ def test_busy_interrupt_mode_redirects_active_turn(monkeypatch):
     assert session.get("queued_prompt") is None
 
 
+def test_successful_redirect_drops_queued_duplicate_of_inflight_user(monkeypatch):
+    """#84417: correcting a live turn must not re-fire the original prompt from queue.
+
+    When the live turn's original user text is also sitting in the server queue
+    (e.g. a second prompt.submit of the same text while redirect was not yet
+    possible), a later successful redirect of a *new* correction Q must purge
+    that self-duplicate. Otherwise post-turn ``_drain_queued_prompt`` starts a
+    second agent turn with the old prompt P after Q has already been handled.
+    """
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: True,
+        interrupt=lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("redirect must not hard-interrupt")
+        ),
+    )
+    session = _session(agent=agent, running=True)
+    original = "deepseek released a new flash model — I changed all settings to flash"
+    session["inflight_turn"] = {
+        "user": original,
+        "assistant": "partial",
+        "streaming": True,
+        "error": "",
+    }
+    # Stale self-duplicate of the live turn (would re-fire after settle).
+    session["queued_prompt"] = {"text": original, "transport": "ws-1"}
+    session["queued_prompts"] = [
+        {"text": original, "transport": "ws-1"},
+        {"text": "unrelated later task", "transport": "ws-1"},
+    ]
+
+    resp = server._handle_busy_submit(
+        "r1", "sid", session, "what about the pricing instead?", "ws-1"
+    )
+
+    assert resp["result"]["status"] == "redirected"
+    # Self-duplicates of the live original must be gone.
+    assert session.get("queued_prompt") == {
+        "text": "unrelated later task",
+        "transport": "ws-1",
+    }
+    assert not session.get("queued_prompts")
+
+
+def test_successful_redirect_preserves_unrelated_queued_followups(monkeypatch):
+    """A legitimate next-turn queue entry must survive a mid-turn redirect."""
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: True,
+        interrupt=lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("redirect must not hard-interrupt")
+        ),
+    )
+    session = _session(agent=agent, running=True)
+    session["inflight_turn"] = {
+        "user": "live turn P",
+        "assistant": "",
+        "streaming": True,
+        "error": "",
+    }
+    session["queued_prompt"] = {"text": "run this after", "transport": "ws-1"}
+
+    resp = server._handle_busy_submit("r1", "sid", session, "correction Q", "ws-1")
+
+    assert resp["result"]["status"] == "redirected"
+    assert session.get("queued_prompt") == {
+        "text": "run this after",
+        "transport": "ws-1",
+    }
+
+
+def test_enqueue_skips_text_duplicate_of_inflight_user():
+    """#84417 defense: do not admit a self-duplicate of the live user prompt."""
+    session = _session()
+    session["inflight_turn"] = {
+        "user": "live turn P",
+        "assistant": "",
+        "streaming": True,
+        "error": "",
+    }
+
+    server._enqueue_prompt(session, "live turn P", "ws-1")
+    assert session.get("queued_prompt") is None
+
+    server._enqueue_prompt(session, "different follow-up", "ws-1")
+    assert session["queued_prompt"] == {
+        "text": "different follow-up",
+        "transport": "ws-1",
+    }
+
+
+def test_enqueue_followup_does_not_merge_stale_inflight_self_duplicate():
+    """#84417: scrub P before merging so drain cannot re-fire ``P\\n\\nQ``."""
+    session = _session()
+    session["inflight_turn"] = {
+        "user": "P",
+        "assistant": "",
+        "streaming": True,
+        "error": "",
+    }
+    # Pre-existing stale self-duplicate (e.g. admitted before inflight was set).
+    session["queued_prompt"] = {"text": "P", "transport": "ws-1"}
+
+    server._enqueue_prompt(session, "Q", "ws-1")
+
+    assert session.get("queued_prompt") == {"text": "Q", "transport": "ws-1"}
+    assert not session.get("queued_prompts")
+
+
+def test_drop_rewrites_merged_inflight_prefix_to_followup_only():
+    """Already-merged ``P\\n\\nQ`` slots keep Q and drop the live original."""
+    session = _session()
+    session["inflight_turn"] = {
+        "user": "P",
+        "assistant": "",
+        "streaming": True,
+        "error": "",
+    }
+    session["queued_prompt"] = {"text": "P\n\nQ", "transport": "ws-1"}
+
+    server._drop_queued_duplicates_of_inflight_user(session)
+
+    assert session.get("queued_prompt") == {"text": "Q", "transport": "ws-1"}
+
+
+def test_hard_interrupt_queue_path_scrubs_stale_inflight_self_duplicate(monkeypatch):
+    """#84417: interrupt+queue of Q must not leave P ahead of Q in the FIFO."""
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    interrupts = []
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: False,  # force hard-interrupt fallback
+        interrupt=lambda *a, **k: interrupts.append(True),
+    )
+    session = _session(agent=agent, running=True)
+    session["inflight_turn"] = {
+        "user": "P",
+        "assistant": "",
+        "streaming": True,
+        "error": "",
+    }
+    session["queued_prompt"] = {"text": "P", "transport": "ws-1"}
+
+    resp = server._handle_busy_submit("r1", "sid", session, "Q", "ws-1")
+
+    assert resp["result"]["status"] == "queued"
+    assert session.get("queued_prompt") == {"text": "Q", "transport": "ws-1"}
+    assert not session.get("queued_prompts")
+    # Interrupt is async-threaded; policy still enqueued Q after scrubbing P.
+
+
+def test_redirect_then_drain_does_not_re_fire_original_p(monkeypatch):
+    """#84417 drain-level: after redirect(Q), settle must not start a second P."""
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    fired = []
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: True,
+        interrupt=lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("redirect must not hard-interrupt")
+        ),
+    )
+    session = _session(agent=agent, running=True)
+    session["inflight_turn"] = {
+        "user": "P",
+        "assistant": "partial",
+        "streaming": True,
+        "error": "",
+    }
+    session["queued_prompt"] = {"text": "P", "transport": "ws-1"}
+
+    resp = server._handle_busy_submit("r1", "sid", session, "Q", "ws-1")
+    assert resp["result"]["status"] == "redirected"
+    assert session.get("queued_prompt") is None
+
+    # Turn settles (running cleared in finally) — drain must be a no-op.
+    session["running"] = False
+    monkeypatch.setattr(
+        server,
+        "_run_prompt_submit",
+        lambda rid, sid, session, text, **kwargs: fired.append(text),
+    )
+    monkeypatch.setattr(server, "_session_uses_compute_host", lambda _s: False)
+
+    assert server._drain_queued_prompt("r2", "sid", session) is False
+    assert fired == []
+
+
+def test_compress_session_rotation_bumps_queued_prompt_generation(monkeypatch):
+    """#84417 belt: rotation invalidates in-flight drain claims on the parent key.
+
+    Queue *contents* survive (a legitimate follow-up must still run after
+    compression); only the generation counter advances so a drain that claimed
+    under the pre-rotation key cannot dispatch after re-anchor.
+    """
+    monkeypatch.setattr(server, "_transfer_active_session_slot", lambda *a, **k: True)
+    monkeypatch.setattr(server, "_restart_slash_worker", lambda *a, **k: None)
+    agent = types.SimpleNamespace(session_id="child-after-rotation")
+    session = _session(agent=agent, session_key="parent-before-rotation")
+    session["_queued_prompt_generation"] = 3
+    session["queued_prompt"] = {"text": "run after compress", "transport": "ws-1"}
+
+    server._sync_session_key_after_compress("sid", session, clear_pending_title=False)
+
+    assert session["session_key"] == "child-after-rotation"
+    assert session["_queued_prompt_generation"] == 4
+    # Follow-up kept — only the claim generation bumped.
+    assert session["queued_prompt"] == {
+        "text": "run after compress",
+        "transport": "ws-1",
+    }
+
+
+def test_compress_no_rotation_does_not_bump_queue_generation(monkeypatch):
+    """No-op when agent.session_id already matches session_key."""
+    monkeypatch.setattr(
+        server,
+        "_transfer_active_session_slot",
+        lambda *a, **k: (_ for _ in ()).throw(AssertionError("no transfer")),
+    )
+    agent = types.SimpleNamespace(session_id="same-key")
+    session = _session(agent=agent, session_key="same-key")
+    session["_queued_prompt_generation"] = 2
+
+    server._sync_session_key_after_compress("sid", session)
+
+    assert session["_queued_prompt_generation"] == 2
+
+
 
 
 
@@ -299,7 +530,15 @@ def test_drain_releases_running_on_dispatch_failure(monkeypatch):
 
 
 def test_drain_does_not_dispatch_a_prompt_cancelled_after_claim(monkeypatch):
-    session = _session(queued_prompt={"text": "B", "transport": None})
+    """Generation cancel aborts dispatch but must restore the claimed head.
+
+    Compress re-anchor / Stop bump generation between claim and check. Dropping
+    the envelope would silently lose a legitimate follow-up (#84417 belt).
+    """
+    session = _session(
+        queued_prompt={"text": "B", "transport": "ws-1"},
+        queued_prompts=[{"text": "C", "transport": "ws-1"}],
+    )
     monkeypatch.setattr(
         server,
         "_session_uses_compute_host",
@@ -313,6 +552,29 @@ def test_drain_does_not_dispatch_a_prompt_cancelled_after_claim(monkeypatch):
 
     assert server._drain_queued_prompt("r1", "sid", session) is True
     assert session["running"] is False
+    # Claimed B restored first; C that advanced into the slot is behind it.
+    assert session.get("queued_prompt") == {"text": "B", "transport": "ws-1"}
+    assert session.get("queued_prompts") == [{"text": "C", "transport": "ws-1"}]
+
+
+def test_drain_restores_claimed_prompt_when_generation_bumps_mid_claim(monkeypatch):
+    """Single-item queue: generation cancel must not empty the queue."""
+    session = _session(queued_prompt={"text": "follow-up Q", "transport": None})
+    monkeypatch.setattr(
+        server,
+        "_session_uses_compute_host",
+        lambda _session: session.__setitem__("_queued_prompt_generation", 1) or False,
+    )
+    monkeypatch.setattr(
+        server,
+        "_run_prompt_submit",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("must not dispatch")),
+    )
+
+    assert server._drain_queued_prompt("r1", "sid", session) is True
+    assert session["running"] is False
+    assert session.get("queued_prompt") == {"text": "follow-up Q", "transport": None}
+    assert not session.get("queued_prompts")
 
 
 def test_drain_does_not_clear_stop_after_its_final_generation_check(monkeypatch):

tests/test_tui_gateway_server.py

@@ -695,6 +695,7 @@ def test_profile_scoped_agent_build_starts_mcp_discovery_in_profile_home(
 ):
     """Agent construction must start MCP discovery under the selected profile."""
     import threading
+    import uuid
 
     from hermes_constants import get_hermes_home
 
@@ -720,19 +721,26 @@ def test_profile_scoped_agent_build_starts_mcp_discovery_in_profile_home(
     monkeypatch.setattr(server, "_SlashWorker", lambda *args: None)
     monkeypatch.setattr(server, "_attach_worker", lambda *args: None)
     monkeypatch.setattr(server, "_config_model_target", lambda: ("", ""))
+    # CI runs this huge file serially under load; a prior session's _build can
+    # still be finishing (session.info emit) when the next test starts, so a
+    # 2s Event wait flakes. Unique sid + longer bound; still fail closed.
+    monkeypatch.setattr(server, "_start_notification_poller", lambda *a, **k: None)
+    monkeypatch.setattr(server, "_schedule_mcp_late_refresh", lambda *a, **k: None)
+    monkeypatch.setattr(server, "_emit", lambda *a, **k: None)
 
     ready = threading.Event()
-    sid = "test-sid"
+    sid = f"test-sid-{uuid.uuid4().hex[:8]}"
     session = {
         "agent_ready": ready,
-        "session_key": "test-key",
+        "session_key": f"test-key-{uuid.uuid4().hex[:8]}",
         "profile_home": str(profile_home),
     }
 
     server._sessions[sid] = session
     try:
         server._start_agent_build(sid, session)
-        assert built.wait(timeout=2)
+        assert built.wait(timeout=15), "agent build thread never called _make_agent"
+        assert ready.wait(timeout=5), "agent_ready never set after build"
     finally:
         server._sessions.pop(sid, None)
 
@@ -747,6 +755,7 @@ def test_profile_scoped_agent_build_installs_secret_scope(monkeypatch, tmp_path)
     .env (#67605 item 2).
     """
     import threading
+    import uuid
 
     from agent.secret_scope import current_secret_scope
 
@@ -775,19 +784,25 @@ def test_profile_scoped_agent_build_installs_secret_scope(monkeypatch, tmp_path)
     monkeypatch.setattr(server, "_SlashWorker", lambda *args: None)
     monkeypatch.setattr(server, "_attach_worker", lambda *args: None)
     monkeypatch.setattr(server, "_config_model_target", lambda: ("", ""))
+    # Same CI flake class as the MCP profile-home test: bound wait + less work
+    # on the build thread (no poller / late MCP refresh / session.info emit).
+    monkeypatch.setattr(server, "_start_notification_poller", lambda *a, **k: None)
+    monkeypatch.setattr(server, "_schedule_mcp_late_refresh", lambda *a, **k: None)
+    monkeypatch.setattr(server, "_emit", lambda *a, **k: None)
 
     ready = threading.Event()
-    sid = "test-secret-sid"
+    sid = f"test-secret-sid-{uuid.uuid4().hex[:8]}"
     session = {
         "agent_ready": ready,
-        "session_key": "test-secret-key",
+        "session_key": f"test-secret-key-{uuid.uuid4().hex[:8]}",
         "profile_home": str(profile_home),
     }
 
     server._sessions[sid] = session
     try:
         server._start_agent_build(sid, session)
-        assert built.wait(timeout=2)
+        assert built.wait(timeout=15), "agent build thread never called _make_agent"
+        assert ready.wait(timeout=5), "agent_ready never set after build"
     finally:
         server._sessions.pop(sid, None)
 
@@ -9388,6 +9403,89 @@ def test_session_redirect_calls_capable_core_agent(monkeypatch):
     assert before is None or session["last_active"] >= before
 
 
+def test_session_redirect_rpc_drops_queued_duplicate_of_inflight_user():
+    """#84417: Desktop ``session.redirect`` must purge stale self-duplicates.
+
+    Production path: renderer steers via ``session.redirect`` (not
+    ``prompt.submit``). A self-copy of the live original user text already in
+    the server queue must not survive a successful redirect — otherwise
+    post-turn ``_drain_queued_prompt`` restarts prompt P after Q is handled.
+    Unrelated next-turn envelopes stay.
+    """
+    original = "deepseek released a new flash model — I changed all settings to flash"
+    agent = types.SimpleNamespace(
+        _supports_active_turn_redirect=True,
+        redirect=lambda text: True,
+    )
+    session = _session(agent=agent, running=True)
+    session["inflight_turn"] = {
+        "user": original,
+        "assistant": "partial",
+        "streaming": True,
+        "error": "",
+    }
+    session["queued_prompt"] = {"text": original, "transport": "ws-1"}
+    session["queued_prompts"] = [
+        {"text": original, "transport": "ws-1"},
+        {"text": "unrelated later task", "transport": "ws-1"},
+    ]
+    server._sessions["sid"] = session
+    try:
+        resp = server.handle_request(
+            {
+                "id": "1",
+                "method": "session.redirect",
+                "params": {
+                    "session_id": "sid",
+                    "text": "what about the pricing instead?",
+                },
+            }
+        )
+    finally:
+        server._sessions.pop("sid", None)
+
+    assert resp["result"]["status"] == "redirected"
+    assert session["inflight_turn"]["user"] == original
+    assert session["inflight_turn"]["corrections"] == [
+        "what about the pricing instead?"
+    ]
+    # Self-duplicates of the live original are gone; legitimate follow-up kept.
+    assert session.get("queued_prompt") == {
+        "text": "unrelated later task",
+        "transport": "ws-1",
+    }
+    assert not session.get("queued_prompts")
+
+
+def test_session_redirect_build_window_scrubs_stale_p_when_queuing_q():
+    """#84417: build-window queue of Q must not leave P ahead of Q."""
+    original = "live original P"
+    session = _session(running=True)
+    session["agent"] = None  # async agent build window
+    session["inflight_turn"] = {
+        "user": original,
+        "assistant": "",
+        "streaming": True,
+        "error": "",
+    }
+    session["queued_prompt"] = {"text": original, "transport": "ws-1"}
+    server._sessions["sid"] = session
+    try:
+        resp = server.handle_request(
+            {
+                "id": "1",
+                "method": "session.redirect",
+                "params": {"session_id": "sid", "text": "correction Q"},
+            }
+        )
+    finally:
+        server._sessions.pop("sid", None)
+
+    assert resp["result"] == {"status": "queued", "text": "correction Q"}
+    assert session["queued_prompt"]["text"] == "correction Q"
+    assert not session.get("queued_prompts")
+
+
 def test_session_redirect_records_correction_without_erasing_prompt():
     """A redirect must not overwrite the turn's original user text.
 

tui_gateway/methods_session.py

@@ -3247,6 +3247,10 @@ def _(rid, params: dict) -> dict:
         # text has no user bubble — the "my message vanished on reload" loss.
         with session["history_lock"]:
             _record_inflight_correction(session, text)
+            # #84417: steer does not cancel the live original, but a server
+            # queue self-copy of that original must still not re-fire after
+            # 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})
 
@@ -3283,6 +3287,9 @@ def _(rid, params: dict) -> dict:
     if accepted:
         with session["history_lock"]:
             _record_inflight_correction(session, text)
+            # #84417: purge server-queue self-duplicates of the live original
+            # so post-turn drain cannot restart the pre-correction prompt.
+            _drop_queued_duplicates_of_inflight_user(session)
             session["last_active"] = time.time()
     return _ok(
         rid,

tui_gateway/server.py

@@ -5120,6 +5120,16 @@ def _sync_session_key_after_compress(
         # don't keep targeting the ended row.
         session["session_key"] = new_session_id
 
+    # #84417 (belt): invalidate any in-flight ``_drain_queued_prompt`` claim
+    # that captured generation under the pre-rotation session_key. A raced
+    # drain must not dispatch on the continuation with a stale claim; the
+    # claimed envelope is restored to the queue (see ``_drain_queued_prompt``)
+    # so legitimate follow-ups still survive. Complements self-duplicate
+    # scrubbing on redirect.
+    session["_queued_prompt_generation"] = int(
+        session.get("_queued_prompt_generation", 0)
+    ) + 1
+
     if clear_pending_title:
         session["pending_title"] = None
     if restart_slash_worker:
@@ -7675,6 +7685,20 @@ def _enqueue_prompt(
     sent it even if the session transport is rebound meanwhile.
     """
     image_paths = list(image_paths or [])
+    # #84417: scrub any live-turn self-duplicates first so the consecutive-text
+    # merge below cannot glue "{original}\\n\\n{later}" and re-fire original
+    # on drain after a later correction settles.
+    _drop_queued_duplicates_of_inflight_user(session)
+    # Never queue a text-only self-copy of the live inflight user prompt. The
+    # live turn already owns that text; draining it after settle would restart
+    # the same user turn as a fresh agent invocation.
+    if not image_paths and isinstance(text, str):
+        turn = session.get("inflight_turn")
+        original = (
+            str(turn.get("user") or "").strip() if isinstance(turn, dict) else ""
+        )
+        if original and text.strip() == original:
+            return
     queued = {"text": text, "transport": transport}
     if image_paths:
         queued["image_paths"] = image_paths
@@ -7696,6 +7720,82 @@ def _enqueue_prompt(
     session["queued_prompt"] = queued
 
 
+def _sanitize_queued_entry_vs_inflight_user(
+    entry: Any, original: str
+) -> dict | None:
+    """Drop or rewrite a queue envelope that re-carries the live user text.
+
+    Returns ``None`` to drop the envelope, or a (possibly rewritten) dict to
+    keep. Text-only self-duplicates of ``original`` are dropped. A merged
+    slot ``"{original}\\n\\n{later}"`` (from ``_enqueue_prompt``'s consecutive
+    text merge) is rewritten to just ``later`` so a later correction is not
+    lost and the original is not re-fired (#84417). Image-bearing envelopes
+    are left alone — their chronology/ownership is load-bearing.
+    """
+    if not original or not isinstance(entry, dict):
+        return entry if isinstance(entry, dict) else None
+    if entry.get("image_paths"):
+        return entry
+    text = entry.get("text")
+    if not isinstance(text, str):
+        return entry
+    stripped = text.strip()
+    if not stripped:
+        return None
+    if stripped == original:
+        return None
+    # Lossless text-merge glued the live original onto a later follow-up.
+    for sep in ("\n\n", "\n"):
+        prefix = original + sep
+        if text.startswith(prefix):
+            rest = text[len(prefix) :].strip()
+            if not rest or rest == original:
+                return None
+            cleaned = dict(entry)
+            cleaned["text"] = rest
+            return cleaned
+    return entry
+
+
+def _drop_queued_duplicates_of_inflight_user(session: dict) -> None:
+    """Remove server-queue copies of the live turn's original user text.
+
+    A mid-turn ``prompt.submit`` of the same text can land in
+    ``queued_prompt`` when redirect is not yet available (model not active,
+    build window, tool boundary). If the user then corrects the turn with a
+    different prompt via redirect, that stale self-duplicate must not
+    ``_drain_queued_prompt`` after the redirected turn completes — otherwise
+    the original prompt restarts as a fresh agent turn (#84417).
+
+    Unrelated follow-ups (different text, image-bearing envelopes) stay.
+    Merged ``original + later`` slots are rewritten to ``later`` only.
+    """
+    turn = session.get("inflight_turn")
+    if not isinstance(turn, dict):
+        return
+    original = str(turn.get("user") or "").strip()
+    if not original:
+        return
+
+    head = session.get("queued_prompt")
+    rest = list(session.get("queued_prompts") or [])
+    kept: list[dict] = []
+    for entry in ([head] if head else []) + rest:
+        cleaned = _sanitize_queued_entry_vs_inflight_user(entry, original)
+        if cleaned is not None:
+            kept.append(cleaned)
+
+    if not kept:
+        session["queued_prompt"] = None
+        session.pop("queued_prompts", None)
+        return
+    session["queued_prompt"] = kept[0]
+    if len(kept) > 1:
+        session["queued_prompts"] = kept[1:]
+    else:
+        session.pop("queued_prompts", None)
+
+
 def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None:
     """Interrupt a busy turn without blocking the RPC reader or session lock.
 
@@ -7774,6 +7874,8 @@ def _handle_busy_submit(
         try:
             if agent.steer(plain_text):
                 with session["history_lock"]:
+                    _record_inflight_correction(session, plain_text)
+                    _drop_queued_duplicates_of_inflight_user(session)
                     session["last_active"] = time.time()
                 return _ok(rid, {"status": "steered"})
         except Exception:
@@ -7793,6 +7895,9 @@ def _handle_busy_submit(
             if agent.redirect(plain_text):
                 with session["history_lock"]:
                     _record_inflight_correction(session, plain_text)
+                    # #84417: do not re-fire the live turn's original user text
+                    # from a stale server-queue self-duplicate after settle.
+                    _drop_queued_duplicates_of_inflight_user(session)
                     session["last_active"] = time.time()
                 return _ok(rid, {"status": "redirected"})
         except Exception:
@@ -7837,6 +7942,21 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
     use_compute_host = _session_uses_compute_host(session)
     with session["history_lock"]:
         if int(session.get("_queued_prompt_generation", 0)) != queue_generation:
+            # Generation cancelled the claim (Stop, compress re-anchor, …).
+            # Do not dispatch — but put the claimed envelope back so a
+            # legitimate follow-up is not silently dropped. Order: claimed
+            # head first, then whatever advanced into the slot while we held
+            # the claim (#84417 belt accuracy).
+            rest: list = []
+            advanced = session.get("queued_prompt")
+            if advanced:
+                rest.append(advanced)
+            rest.extend(session.get("queued_prompts") or [])
+            session["queued_prompt"] = queued
+            if rest:
+                session["queued_prompts"] = rest
+            else:
+                session.pop("queued_prompts", None)
             session["running"] = False
             return True
     dispatch_failed = False