fix(streaming): probe on_stream_discarded for pre-existing UIs and format the hoisted fake

PR feedback round:

- on_stream_discarded now follows on_compaction's compat pattern for a
  hook added after UIs exist in the wild: the protocol member carries a
  REAL no-op default (an explicit subclass inherits a correct
  implementation — a UI without server-side turn buffers has nothing to
  truncate), and both call sites route through a getattr probe, so a
  duck-typed UI predating the hook degrades to no-truncate instead of
  raising an AttributeError from the very arm that is handling a stream
  death — which would replace the wire failure with the attribute error
  in the retry gate. Pinned with a hook-less-UI retry test.
- tests/_session_helpers.py gains the formatting pass the RecordingUI
  hoist bypassed (the CI lint failure).

tests/_session_helpers.py

@@ -348,6 +348,7 @@ def as_stream(result: Any) -> list[StreamChunk]:
         )
     ]
 
+
 class RecordingUI:
     """UI adapter recording the ordered event stream ``send()`` emits."""
 

tests/test_midstream_retry.py

@@ -597,6 +597,30 @@ class TestMidStreamRetry:
         assistant = _assistant_msgs(session)
         assert assistant and assistant[-1]["content"].startswith("a dead attempt")
 
+    def test_pre937_ui_without_discard_hook_survives_retry(self, tmp_db):
+        # A duck-typed UI predating on_stream_discarded must degrade to
+        # "no server-buffer truncate", not crash the retry arm with an
+        # AttributeError that replaces the stream death being handled.
+        class _Pre937UI(RecordingUI):
+            # property() with no getter raises AttributeError on access —
+            # simulating the hook's absence on an inheriting fake.
+            on_stream_discarded = property()
+
+        ui = _Pre937UI()
+        session = _make_session(ui)
+        streams = [
+            _dying_stream("x", exc=httpx.ReadError("wire died")),
+            _good_stream("ok"),
+        ]
+        with (
+            patch.object(session, "_create_stream_with_retry", side_effect=streams),
+            patch.object(session, "_full_messages", return_value=[]),
+        ):
+            session.send("test")
+
+        assert _assistant_msgs(session)[-1]["content"] == "ok"
+        assert ("state", "idle") in ui.events
+
     def test_overflow_recovery_discards_dead_text_from_turn_buffer(self, tmp_db):
         # A mid-consumption overflow is TERMINAL for the retry ladder but
         # RECOVERED by send()'s compact-and-retry — the dead attempt's text