From 8c1c63a64d80be984eee4247d5fd321360eb06c7 Mon Sep 17 00:00:00 2001 From: Daniel Han Date: Sun, 24 May 2026 16:44:22 +0000 Subject: [PATCH] Studio: surface Codex final agent message when no deltas stream The canonical openai_codex SDK can complete a turn successfully without emitting any `message.delta` events: the final assistant text arrives only as an `ItemCompletedNotification` whose item is an `agentMessage`. Before this change `_stream_thread_run` would loop through the stream, see no delta text, return, and Studio would emit only the empty usage + stop + `[DONE]` frames -- the user sees a blank reply for what was actually a complete answer. Track agent-message texts collected during the stream loop and, if no streamed deltas came through, yield the last one before returning. The buffered `thread.run()` fallback is still gated by the existing `emitted_any` flag so it never replays a turn that already executed side effects (file writes, shell commands). `_completed_agent_message_text` accepts both the upstream object shape (`ItemCompletedNotification(item.root.text=...)`) and the dict shape pre-release builds and tests use, so it works across SDK revs without an explicit version gate. Tests: new regression `test_empty_stream_falls_back_to_completed_agent_message` exercises a fake SDK whose `turn().stream()` yields only an `item.completed` event with an `agentMessage`; the test asserts the final text reaches the visible chat output and that the buffered `run()` path is NOT re-executed. --- .../backend/core/inference/codex_provider.py | 75 ++++++++++++++++++- studio/backend/tests/test_codex_provider.py | 62 +++++++++++++++ 2 files changed, 136 insertions(+), 1 deletion(-) diff --git a/studio/backend/core/inference/codex_provider.py b/studio/backend/core/inference/codex_provider.py index a3843b30dc..bb6ab71e56 100644 --- a/studio/backend/core/inference/codex_provider.py +++ b/studio/backend/core/inference/codex_provider.py @@ -431,6 +431,60 @@ def _coerce_text(payload: Any) -> str: return "" +def _completed_agent_message_text(payload: Any) -> str: + """Return the assistant text from an ``ItemCompletedNotification``. + + The canonical openai_codex SDK sometimes finishes a turn without + emitting any ``message.delta`` events: the final answer arrives + only as ``ItemCompletedNotification(item=AgentMessage(text=...))`` + at the end of the stream. Without recognising that shape, + ``_stream_thread_run`` would loop through the stream, see no + ``delta`` text, and return an empty Chat Completions response. + + Returns the empty string for any other event shape so the caller + can ignore it. Matches by class name + structural shape so the + function works on both real upstream events and the dict / fake + shapes the tests use. + """ + if payload is None: + return "" + + # Dict shape: tests + some pre-release SDK revs. + if isinstance(payload, dict): + if payload.get("type") not in ( + "ItemCompletedNotification", + "item.completed", + "thread.item.completed", + ): + return "" + item = payload.get("item") + # The upstream model wraps the item in a discriminated-union + # `root` field; some pre-release shapes drop the wrapper. Look + # both ways. + if isinstance(item, dict): + inner = item.get("root", item) + if not isinstance(inner, dict): + return "" + if inner.get("type") not in ("agentMessage", "agent_message"): + return "" + text = inner.get("text") + return text if isinstance(text, str) else "" + return "" + + # Object shape: upstream events with `.item.root.text`. + if payload.__class__.__name__ not in ( + "ItemCompletedNotification", + "ThreadItemCompletedNotification", + ): + return "" + item = getattr(payload, "item", None) + item = getattr(item, "root", item) + if getattr(item, "type", None) not in ("agentMessage", "agent_message"): + return "" + text = getattr(item, "text", None) + return text if isinstance(text, str) else "" + + async def _stream_thread_run( thread: Any, prompt: str, @@ -455,12 +509,20 @@ async def _stream_thread_run( 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. + + Empty-delta protection: the canonical SDK can complete a turn + successfully without emitting any ``message.delta`` events -- + the final text arrives only as an ``ItemCompletedNotification`` + whose ``item`` is an ``agentMessage``. We collect those during the + stream loop and emit the last one if no deltas came through, so + Studio never returns an empty answer for a successful turn. """ emitted_any = False # 1. Canonical: thread.turn(prompt).stream() turn_factory = getattr(thread, "turn", None) if turn_factory is not None: + agent_message_texts: list[str] = [] try: turn_handle = turn_factory(prompt) if asyncio.iscoroutine(turn_handle): @@ -471,10 +533,21 @@ async def _stream_thread_run( if asyncio.iscoroutine(stream_obj): stream_obj = await stream_obj async for event in stream_obj: - text = _coerce_text(getattr(event, "payload", event)) + payload = getattr(event, "payload", event) + text = _coerce_text(payload) if text: emitted_any = True yield text + else: + final_text = _completed_agent_message_text(payload) + if final_text: + agent_message_texts.append(final_text) + if not emitted_any and agent_message_texts: + # The stream completed cleanly but only via a final + # ItemCompletedNotification -- emit the last agent + # message text so the chat reply is not blank. + yield agent_message_texts[-1] + emitted_any = True return except Exception as exc: logger.warning( diff --git a/studio/backend/tests/test_codex_provider.py b/studio/backend/tests/test_codex_provider.py index cf2aded628..372bee770e 100644 --- a/studio/backend/tests/test_codex_provider.py +++ b/studio/backend/tests/test_codex_provider.py @@ -1330,6 +1330,68 @@ class TestCodexHardenedRegressions: # Model still passed so the request is well-formed. assert kw.get("model") == "gpt-5.5" + def test_empty_stream_falls_back_to_completed_agent_message(self, monkeypatch): + """A successful turn that emits zero ``message.delta`` events + but DOES emit a final ``ItemCompletedNotification`` with an + agent message must surface that text. Without the fallback the + Chat Completions reply would be empty even though Codex + produced a complete answer. + """ + + class _CompletedEvent: + payload = { + "type": "item.completed", + "item": { + "root": { + "type": "agentMessage", + "text": "final answer from completion", + }, + }, + } + + class _Turn: + async def stream(self): + yield _CompletedEvent() + + class _ThreadEmptyDeltas: + def turn(self, prompt): + return _Turn() + + async def run(self, prompt): + raise AssertionError( + "must not fall through to buffered run() when " + "the stream completes successfully" + ) + + class _Async: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def thread_start(self, **kw): + return _ThreadEmptyDeltas() + + _install_fake_codex_sdk(monkeypatch, _Async) + from core.inference.codex_provider import stream_codex + + chunks: list[str] = [] + + async def _collect(): + async for c in stream_codex( + messages = [{"role": "user", "content": "hi"}], + model = "gpt-5.5", + parallel_calls = 1, + ): + chunks.append(c) + + asyncio.run(_collect()) + body = "".join(chunks) + assert ( + "final answer from completion" in body + ), f"agent message text from completion event was dropped; body={body!r}" + def test_synthesis_also_pins_safety_kwargs(self, monkeypatch): """The synthesis turn that unifies parallel fan-out outputs must use the same safety pins -- a fan-out tab could otherwise