From b17765b5ed0e7bdd83cac91c0aa9644a63a2beaa Mon Sep 17 00:00:00 2001 From: Daniel Han Date: Mon, 25 May 2026 15:24:25 +0000 Subject: [PATCH] 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. --- .../backend/core/inference/codex_provider.py | 55 ++++++--- studio/backend/tests/test_codex_provider.py | 108 ++++++++++++++++-- .../features/chat/chat-providers-dialog.tsx | 15 ++- 3 files changed, 156 insertions(+), 22 deletions(-) diff --git a/studio/backend/core/inference/codex_provider.py b/studio/backend/core/inference/codex_provider.py index 902ec12e3f..6d6e9adb96 100644 --- a/studio/backend/core/inference/codex_provider.py +++ b/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: diff --git a/studio/backend/tests/test_codex_provider.py b/studio/backend/tests/test_codex_provider.py index 92ef48d439..423b0a10a7 100644 --- a/studio/backend/tests/test_codex_provider.py +++ b/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"] diff --git a/studio/frontend/src/features/chat/chat-providers-dialog.tsx b/studio/frontend/src/features/chat/chat-providers-dialog.tsx index faa06afef4..64cb177804 100644 --- a/studio/frontend/src/features/chat/chat-providers-dialog.tsx +++ b/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"); }