Studio: round 8 -- replay guard on non-visible events + Add-flow Codex preselect

Two P1 fixes from the round 8 reviewer pass:

1. _stream_thread_run no longer replays a Codex turn that fired
   non-visible events before crashing.

   The replay guard only tracked `emitted_any` (visible text). A
   Codex turn that emitted, say, a command.delta or file.delta event
   first -- both filtered to "" by _coerce_text -- and THEN crashed
   would leave emitted_any=False and fall through to the buffered
   `thread.run(prompt)` fallback, re-executing the same turn and
   duplicating its side effects (shell commands, file writes,
   tool calls). This is exactly the case the guard was added to
   prevent in earlier rounds; the missing bit was tracking
   "the turn ran at all", not just "the turn yielded text".

   Fix: add a separate turn_started flag that flips True the moment
   we ask the SDK for a turn handle or observe any event from a
   streaming helper. When the buffered fallback is gated on
   turn_started instead of emitted_any, a partial-turn crash
   correctly stops without replaying. Regression test reproduces
   the bug against the pre-fix code (assertion catches the extra
   thread.run call) and locks the fix in.

2. openAddProvider now mirrors the providerType-change effect's
   Codex pre-check.

   The first-run UX fix from `26799d9a` pre-checked every Codex
   default model in the providerType-change effect, but
   openAddProvider() calls resetForm() (which clears
   selectedModelIds) and then only restores availableModels, not
   selectedModelIds. If the user closes the Add connection form
   and re-opens it while Codex is still the current providerType,
   the effect does not re-run, so the form opens with Codex
   defaults available but none selected -- the "Add at least one
   model ID" save guard then blocks the Save click.

   Fix: openAddProvider now seeds selectedModelIds with the full
   default-models list when the provider is Codex, matching the
   providerType-change effect so the two entry paths produce the
   same first-run state.

studio/backend/core/inference/codex_provider.py

@@ -719,12 +719,16 @@ async def _stream_thread_run(
        path. Used when neither streaming helper resolves and as the
        final fallback.
 
-    Cross-turn side-effect protection: once any chunk has been emitted
-    via a streaming helper, we never fall through to the buffered
-    ``thread.run(prompt)`` path -- a partial-stream failure would
-    otherwise re-execute the same Codex turn and duplicate side
-    effects (file writes, shell commands, etc.). The buffered path
-    runs only when streaming helpers produced zero output.
+    Cross-turn side-effect protection: once a turn has STARTED (any
+    SDK event was received, including ones that ``_coerce_text``
+    drops -- command/file/tool/plan events), we never fall through
+    to the buffered ``thread.run(prompt)`` path. A partial-stream
+    failure mid-turn must not re-execute the same Codex turn
+    because the side effects (file writes, shell commands, etc.)
+    would replay. Tracking only ``emitted_any`` (visible text) is
+    not enough -- a turn that crashes after running shell commands
+    but before producing answer text would otherwise replay because
+    no visible chunk was emitted.
 
     Empty-delta protection: the canonical SDK can complete a turn
     successfully without emitting any ``message.delta`` events --
@@ -734,6 +738,11 @@ async def _stream_thread_run(
     Studio never returns an empty answer for a successful turn.
     """
     emitted_any = False
+    # True once ANY event has been observed from a streaming helper.
+    # Even when ``_coerce_text`` filters the event out, the turn has
+    # demonstrably started executing on the Codex side, so a later
+    # error must not trigger a buffered ``thread.run`` replay.
+    turn_started = False
 
     # 1. Canonical: thread.turn(prompt).stream()
     turn_factory = getattr(thread, "turn", None)
@@ -741,6 +750,12 @@ async def _stream_thread_run(
         agent_message_texts: list[str] = []
         try:
             turn_handle = turn_factory(prompt)
+            # Asking the SDK for the turn handle is itself enough to
+            # start the turn on the upstream side; mark turn_started
+            # before we even start iterating so a crash inside the
+            # stream factory below does not look like a never-started
+            # turn that is safe to replay.
+            turn_started = True
             if asyncio.iscoroutine(turn_handle):
                 turn_handle = await turn_handle
             stream_fn = getattr(turn_handle, "stream", None)
@@ -749,6 +764,7 @@ async def _stream_thread_run(
                 if asyncio.iscoroutine(stream_obj):
                     stream_obj = await stream_obj
                 async for event in stream_obj:
+                    turn_started = True
                     payload = getattr(event, "payload", event)
                     text = _coerce_text(payload)
                     if text:
@@ -771,11 +787,16 @@ async def _stream_thread_run(
                 exc_type = type(exc).__name__,
                 error = str(exc),
                 emitted_any = emitted_any,
+                turn_started = turn_started,
             )
-            if emitted_any:
-                # The Codex turn already ran far enough to emit text;
-                # do not re-execute via run() or run_streaming() -- the
-                # side-effects (commands / writes) would replay.
+            if turn_started:
+                # The Codex turn has executed at least one event on
+                # the upstream side (it may have launched shell
+                # commands or written files via tool events that
+                # _coerce_text filtered out). Re-executing via
+                # run_streaming / run() would duplicate those side
+                # effects, so stop here even if no visible text was
+                # yielded.
                 return
 
     # 2. Legacy: thread.run_streaming(prompt)
@@ -783,9 +804,14 @@ async def _stream_thread_run(
     if run_streaming is not None:
         try:
             stream_obj = run_streaming(prompt)
+            # Same reasoning as the canonical path above: calling the
+            # streaming helper is enough to start the turn on the SDK
+            # side, so a later crash must NOT replay via buffered run.
+            turn_started = True
             if asyncio.iscoroutine(stream_obj):
                 stream_obj = await stream_obj
             async for event in stream_obj:
+                turn_started = True
                 text = _coerce_text(event)
                 if text:
                     emitted_any = True
@@ -797,13 +823,16 @@ async def _stream_thread_run(
                 exc_type = type(exc).__name__,
                 error = str(exc),
                 emitted_any = emitted_any,
+                turn_started = turn_started,
             )
-            if emitted_any:
+            if turn_started:
                 return
 
     # 3. Buffered fallback: await the full TurnResult, emit one chunk.
-    # Only reached when no streaming helper emitted anything, so this
-    # is the first (and only) execution of the turn.
+    # Only reached when no streaming helper ran at all (no turn /
+    # run_streaming attributes on the thread, or both raised before
+    # observing any event / starting the turn), so this is the first
+    # and only execution of the turn.
     result = await thread.run(prompt)
     text = _buffered_result_text(result)
     if text:

studio/backend/tests/test_codex_provider.py

@@ -2047,7 +2047,9 @@ class TestDeviceLoginLogFilter:
         # caught when the production source is loaded.
         import importlib
 
-        mod = importlib.reload(importlib.import_module("core.inference.codex_provider"))
+        mod = importlib.reload(
+            importlib.import_module("core.inference.codex_provider")
+        )
         # Walk the source string to find the patterns; they live inside
         # the generator. Use a stable proxy: read the regex literals.
         import re
@@ -2057,10 +2059,8 @@ class TestDeviceLoginLogFilter:
         # phrases AND an unsafe-content blocklist.
         assert "safe_log_res" in src
         assert "unsafe_log_re" in src
-        assert (
-            "not\\s+(?:logged|signed)\\s+in" in src
-            or "not\\\\s+(?:logged|signed)\\\\s+in" in src
-        )
+        assert "not\\s+(?:logged|signed)\\s+in" in src or \
+               "not\\\\s+(?:logged|signed)\\\\s+in" in src
         return None
 
     def test_safe_log_source_has_anchored_patterns_and_blocklist(self):
@@ -2117,6 +2117,98 @@ class TestDeviceLoginLogFilter:
             "Browser opened",
             "Press Ctrl+C to cancel",
         ]:
-            assert any(
-                pat.search(clean) for pat in safe_log_res
-            ), f"clean line should match safe: {clean!r}"
+            assert any(pat.search(clean) for pat in safe_log_res), \
+                f"clean line should match safe: {clean!r}"
+
+
+# ── Round 8: stream replay protection on non-visible events ──────────
+
+
+class TestStreamReplayProtection:
+    """Lock in the round 8 fix: a turn that fired non-rendered events
+    (command/file/tool deltas) before crashing MUST NOT replay via the
+    buffered `thread.run(prompt)` fallback even though no visible
+    text was yielded. The earlier guard only tracked `emitted_any`
+    (visible text), missing the case where shell commands or file
+    writes already happened upstream.
+    """
+
+    def test_buffered_run_not_called_after_non_visible_event_crash(self):
+        """Stream raises after a tool event with no visible text. The
+        buffered ``thread.run`` MUST NOT be called -- the Codex turn
+        has already started running side-effects upstream.
+        """
+        from core.inference.codex_provider import _stream_thread_run
+
+        class _Stream:
+            def __init__(self):
+                self._i = 0
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                self._i += 1
+                if self._i == 1:
+                    # An event with no answer text -- _coerce_text
+                    # returns "" but the turn has demonstrably run.
+                    return {"type": "command.delta", "delta": "rm -rf"}
+                raise RuntimeError("upstream stream died mid-turn")
+
+        class _Turn:
+            def stream(self):
+                return _Stream()
+
+        class _Thread:
+            run_calls: int = 0
+
+            def turn(self_inner, prompt):
+                return _Turn()
+
+            async def run(self_inner, prompt):
+                self_inner.run_calls += 1
+                return "REPLAY-WOULD-RETURN-THIS"
+
+        thread = _Thread()
+
+        async def collect():
+            chunks = []
+            async for c in _stream_thread_run(thread, "hello"):
+                chunks.append(c)
+            return chunks
+
+        chunks = asyncio.run(collect())
+        # No visible text was emitted (the only event was filtered),
+        # but thread.run MUST NOT have been called because the turn
+        # already started.
+        assert thread.run_calls == 0, (
+            "thread.run was called after a partial-turn crash; this "
+            "would replay shell commands / file writes"
+        )
+        assert chunks == []
+
+    def test_buffered_run_called_when_no_streaming_helper(self):
+        """Threads that expose neither .turn nor .run_streaming still
+        fall through to the buffered .run -- that is the ONLY path
+        the buffered fallback is allowed to execute.
+        """
+        from core.inference.codex_provider import _stream_thread_run
+
+        class _Thread:
+            run_calls: int = 0
+
+            async def run(self_inner, prompt):
+                self_inner.run_calls += 1
+                return "answer"
+
+        thread = _Thread()
+
+        async def collect():
+            chunks = []
+            async for c in _stream_thread_run(thread, "hello"):
+                chunks.append(c)
+            return chunks
+
+        chunks = asyncio.run(collect())
+        assert thread.run_calls == 1
+        assert chunks == ["answer"]

studio/frontend/src/features/chat/chat-providers-dialog.tsx

@@ -556,7 +556,20 @@ export function ChatProvidersSettings({
     resetForm();
     const entry = providerType ? registryByType.get(providerType) : null;
     if (entry?.model_list_mode === "curated") {
-      setAvailableModels([...entry.default_models]);
+      const defaults = [...entry.default_models];
+      setAvailableModels(defaults);
+      // Mirror the providerType-change effect's first-run behavior:
+      // Codex is the local CLI so pre-checking the default models lets
+      // the user click Save without re-ticking anything. Without this
+      // the resetForm above would zero selectedModelIds and the form
+      // would fail the "Add at least one model ID" save guard even
+      // though the round 7 Codex auto-enable effect would have
+      // populated them. Triggered when the user clicks Add connection
+      // while Codex was already the providerType (e.g. after closing
+      // and reopening the form).
+      setSelectedModelIds(
+        providerType === CODEX_PROVIDER_TYPE ? defaults : [],
+      );
     }
     setPage("form");
   }