fix(tui): hold back Bot Chat stream deltas that could still be a silence marker

The completion-time emptying only takes effect when nothing streamed: the
Desktop keeps already-streamed text when message.complete carries an empty
text (#95514), so a bare NO_REPLY that arrived via message.delta still
rendered in the Bot Chat bubble. Mirror gateway.stream_consumer: for Bot
Chat sessions withhold deltas while the accumulated buffer satisfies
is_partial_silence_marker; flush the held text verbatim once the buffer
diverges from every marker, so a marker never streams and prose that merely
starts like one is delivered intact.

Review finding: streamed NO_REPLY still visible in Desktop Bot Chat because message.delta carried the marker verbatim.

tests/tui_gateway/test_bot_mode_silence_delivery.py

@@ -42,3 +42,30 @@ def test_live_bot_chat_completion_empties_marker_only_for_successful_turns(monke
     monkeypatch.setattr(srv, "_session_live_title", lambda _s, _k: "Scratch")
     payload, _, _ = srv._complete_turn_payload(session, _turn({"final_response": "NO_REPLY"}), None, 80)
     assert payload["text"] == "NO_REPLY"
+
+
+def test_live_bot_chat_stream_holds_back_partial_silence_marker(monkeypatch):
+    """Mirror of stream_consumer's hold-back: a marker never reaches message.delta, prose that
+    diverges from every marker is flushed intact once it diverges."""
+    events = []
+    monkeypatch.setattr(srv, "_emit", lambda event, _sid, payload=None: events.append((event, payload)))
+    monkeypatch.setattr(srv, "_load_interim_assistant_messages", lambda: False)
+    monkeypatch.setattr(srv, "_start_usage_ticker", lambda _sid, _agent: (SimpleNamespace(set=lambda: None), SimpleNamespace(join=lambda: None)))
+
+    def _run(final, chunks):
+        events.clear()
+
+        def run_conversation(_message, **kwargs):
+            for chunk in chunks:
+                kwargs["stream_callback"](chunk)
+            return {"final_response": final}
+
+        agent = SimpleNamespace(_session_title_hint="Bot Chat", run_conversation=run_conversation)
+        session = {"pending_title": None, "session_key": "k", "history_lock": contextlib.nullcontext(), "agent": agent}
+        st = srv._TurnRun(agent=agent, one_turn_restore=None, terminal_callback=None, receipt_committed=True)
+        srv._invoke_agent("sid", session, st, "ping", "ping", None, [], None, None)
+        return [p["text"] for e, p in events if e == "message.delta"], (session.get("inflight_turn") or {}).get("assistant", "")
+
+    assert _run("NO_REPLY", ["NO_", "REPLY"]) == ([], "")
+    assert _run("NO way, here is the answer.", ["NO", " way,", " here is the answer."]) == (
+        ["NO way,", " here is the answer."], "NO way, here is the answer.")

tui_gateway/prompt_turn.py

@@ -534,8 +534,19 @@ def _invoke_agent(
     turn_author: dict | None = None) -> None:
     """Wire the streaming callbacks and run the conversation into ``st.result``."""
     agent = st.agent
+    # Bot Chat mirrors gateway.stream_consumer: deltas are withheld while the streamed buffer
+    # could still resolve to a silence marker ("NO"->"NO_REPLY"), so a bare marker is never
+    # shown and then retracted (the client keeps streamed text when message.complete is "").
+    hold = {"buf": "", "held": ""} if _is_bot_mode_session(session) else None
 
     def _stream(delta):
+        if hold is not None and isinstance(delta, str):
+            from gateway.response_filters import is_partial_silence_marker
+            hold["buf"] += delta
+            if is_partial_silence_marker(hold["buf"]):
+                hold["held"] += delta
+                return
+            delta, hold["held"] = hold["held"] + delta, ""
         with session["history_lock"]:
             _append_inflight_delta(session, delta)
         payload = {"text": delta}