From 851bd81629040334c638a9aa6ba5281a9fa74b58 Mon Sep 17 00:00:00 2001 From: StanleyStetson Date: Wed, 12 Aug 2026 15:23:44 +0300 Subject: [PATCH 1/2] fix(tui_gateway): stop replaying live-turn user text after redirect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A mid-turn correction (Desktop session.redirect / busy 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 — duplicate active role=user rows and a second conversation turn (#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, and bump _queued_prompt_generation when compression rotates session_key so in-flight drain claims cannot dispatch against the continuation with a stale generation. Fixes #84417 --- tests/test_tui_gateway_queue_on_busy.py | 231 ++++++++++++++++++++++++ tests/test_tui_gateway_server.py | 83 +++++++++ tui_gateway/methods_session.py | 7 + tui_gateway/server.py | 104 +++++++++++ 4 files changed, 425 insertions(+) diff --git a/tests/test_tui_gateway_queue_on_busy.py b/tests/test_tui_gateway_queue_on_busy.py index e6780e2976400..df91c5f4db89c 100644 --- a/tests/test_tui_gateway_queue_on_busy.py +++ b/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 + + diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 43341730059f8..14615b837c33a 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -9033,6 +9033,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. diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py index 27adddadfa7c6..0cbff5b915be7 100644 --- a/tui_gateway/methods_session.py +++ b/tui_gateway/methods_session.py @@ -3245,6 +3245,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}) @@ -3281,6 +3285,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, diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 14d6dbc2f2fad..8fd32532c5827 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -4996,6 +4996,15 @@ 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. The queue + # contents stay (legitimate follow-ups must survive), but a drain that + # raced compression cannot dispatch against the rotated continuation + # with a stale claim. 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: @@ -7539,6 +7548,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 @@ -7560,6 +7583,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. @@ -7638,6 +7737,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: @@ -7657,6 +7758,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: From a917a4a8e4020645fe3e9d69eac43465b38b8656 Mon Sep 17 00:00:00 2001 From: StanleyStetson Date: Wed, 12 Aug 2026 17:19:21 +0300 Subject: [PATCH 2/2] fix(tui_gateway): restore claimed queue item on generation cancel When _queued_prompt_generation advances between claim and dispatch (Stop, compress re-anchor), do not drop the claimed envelope. Put it back at the head of the queue so legitimate follow-ups survive. Part of #84417 belt accuracy. --- tests/test_tui_gateway_queue_on_busy.py | 33 ++++++++++++++++++++++++- tui_gateway/server.py | 24 +++++++++++++++--- 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/tests/test_tui_gateway_queue_on_busy.py b/tests/test_tui_gateway_queue_on_busy.py index df91c5f4db89c..33c0e5ab6f1ba 100644 --- a/tests/test_tui_gateway_queue_on_busy.py +++ b/tests/test_tui_gateway_queue_on_busy.py @@ -530,7 +530,36 @@ def _boom(*a, **k): 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", + 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 + # 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", @@ -544,6 +573,8 @@ 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 + 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): diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 8fd32532c5827..77ff1c0454124 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -4997,10 +4997,11 @@ def _sync_session_key_after_compress( 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. The queue - # contents stay (legitimate follow-ups must survive), but a drain that - # raced compression cannot dispatch against the rotated continuation - # with a stale claim. Complements self-duplicate scrubbing on redirect. + # 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 @@ -7805,6 +7806,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