From 7041c56cdf844cd11250c6387a68945c4cdaf9c9 Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 16 Jul 2026 18:40:37 +0700 Subject: [PATCH] feat(agent): stream Codex commentary separately --- agent/agent_init.py | 4 + agent/codex_runtime.py | 53 +++- agent/conversation_loop.py | 4 + run_agent.py | 92 ++++++- .../test_run_agent_codex_responses.py | 260 ++++++++++++++++++ 5 files changed, 396 insertions(+), 17 deletions(-) diff --git a/agent/agent_init.py b/agent/agent_init.py index e1d219c62c63..b7d286af5e49 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -743,6 +743,10 @@ def init_agent( # commentary when the provider later returns it as a completed interim # assistant message. agent._current_streamed_assistant_text = "" + # Completed interim messages delivered during the current user turn. + # Unlike token-stream tracking, this spans Codex continuation/tool calls so + # repeated commentary is not re-sent before normalization can deduplicate it. + agent._delivered_interim_texts: set[str] = set() # Optional current-turn user-message override used when the API-facing # user message intentionally differs from the persisted transcript diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index c1e46e8a4604..99406ba9e0ec 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -624,6 +624,7 @@ def _consume_codex_event_stream( model: str, on_text_delta=None, on_reasoning_delta=None, + on_commentary_message=None, on_first_delta=None, on_event=None, interrupt_check=None, @@ -655,7 +656,11 @@ def _consume_codex_event_stream( * ``on_text_delta(str)`` — fires per ``response.output_text.delta``, suppressed once a function_call event is seen (so tool-call turns don't bleed text into the chat). - * ``on_reasoning_delta(str)`` — fires per ``response.reasoning.*.delta``. + * ``on_reasoning_delta(str)`` — fires per ``response.reasoning.*.delta`` and + ``phase=analysis`` message deltas. When no dedicated commentary callback + is supplied, commentary also uses this legacy fallback. + * ``on_commentary_message(str)`` — fires once per completed + ``phase=commentary`` message, before any following tool item executes. * ``on_first_delta()`` — one-shot, fires on the first text delta only. * ``on_event(event)`` — fires for every event before any other processing. Used for watchdog activity, debug logging, anything wire-shape-agnostic. @@ -666,6 +671,7 @@ def _consume_codex_event_stream( has_tool_calls = False first_delta_fired = False active_message_phase: str | None = None + commentary_text_deltas: List[str] = [] terminal_status: str = "completed" terminal_usage: Any = None terminal_response_id: str = None @@ -710,6 +716,8 @@ def _consume_codex_event_stream( if item_type == "message": phase = _item_field(item, "phase", None) active_message_phase = phase.strip().lower() if isinstance(phase, str) else None + if active_message_phase == "commentary": + commentary_text_deltas = [] else: active_message_phase = None if "function_call" in str(item_type): @@ -718,10 +726,16 @@ def _consume_codex_event_stream( if "output_text.delta" in event_type or event_type == "response.output_text.delta": delta_text = _event_field(event, "delta", "") - is_commentary_delta = active_message_phase in {"commentary", "analysis"} - if delta_text and is_commentary_delta: - # Commentary streams through the reasoning channel, not the - # visible answer stream (and stays out of output_text). + if delta_text and active_message_phase == "commentary": + commentary_text_deltas.append(delta_text) + # Preserve CLI/backward compatibility when no first-class + # commentary consumer is installed. + if on_commentary_message is None and on_reasoning_delta is not None: + try: + on_reasoning_delta(delta_text) + except Exception: + logger.debug("Codex stream on_reasoning_delta raised", exc_info=True) + elif delta_text and active_message_phase == "analysis": if on_reasoning_delta is not None: try: on_reasoning_delta(delta_text) @@ -761,6 +775,27 @@ def _consume_codex_event_stream( done_item = _event_field(event, "item") if done_item is not None: collected_output_items.append(done_item) + done_phase = _item_field(done_item, "phase", None) + done_phase = done_phase.strip().lower() if isinstance(done_phase, str) else None + if done_phase == "commentary" and on_commentary_message is not None: + commentary_text = "".join(commentary_text_deltas).strip() + if not commentary_text: + content_parts = _item_field(done_item, "content", []) + if isinstance(content_parts, list): + commentary_text = "".join( + str(_item_field(part, "text", "") or "") + for part in content_parts + if _item_field(part, "type", "") == "output_text" + ).strip() + if commentary_text: + try: + on_commentary_message(commentary_text) + except Exception: + logger.debug( + "Codex stream on_commentary_message raised", + exc_info=True, + ) + commentary_text_deltas = [] continue if event_type in _TERMINAL_EVENT_TYPES: @@ -861,6 +896,9 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta def _on_reasoning_delta(text: str) -> None: agent._fire_reasoning_delta(text) + def _on_commentary_message(text: str) -> None: + agent._fire_streamed_codex_commentary(text) + def _on_event(event: Any) -> None: # TTFB watchdog and activity touch — runs once per SSE event. agent._codex_stream_last_event_ts = time.time() @@ -900,6 +938,11 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta model=api_kwargs.get("model"), on_text_delta=_on_text_delta, on_reasoning_delta=_on_reasoning_delta, + on_commentary_message=( + _on_commentary_message + if getattr(agent, "interim_assistant_callback", None) is not None + else None + ), on_first_delta=on_first_delta, on_event=_on_event, interrupt_check=_interrupt_check, diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 6c5a171aaa84..5dd9bed319c1 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -616,6 +616,10 @@ def run_conversation( _plugin_user_context = _ctx.plugin_user_context _ext_prefetch_cache = _ctx.ext_prefetch_cache + # Commentary deduplication spans all provider continuations and tool calls + # within one user turn, but must not suppress the same phrase next turn. + agent._delivered_interim_texts = set() + # Main conversation loop counters (pure locals consumed by the loop below). api_call_count = 0 final_response = None diff --git a/run_agent.py b/run_agent.py index 6cc646603cc2..8e7947fb08d2 100644 --- a/run_agent.py +++ b/run_agent.py @@ -4754,8 +4754,11 @@ class AIAgent: ) return bool(streamed) and streamed == visible_content - def _extract_codex_interim_visible_text(self, assistant_msg: Dict[str, Any]) -> str: - """Extract visible Codex Responses commentary text. + def _extract_codex_interim_visible_parts( + self, + assistant_msg: Dict[str, Any], + ) -> List[str]: + """Extract visible Codex commentary as one string per message item. Codex Responses can keep user-facing mid-turn narration as structured ``phase=commentary`` message items while final answer text remains in @@ -4765,9 +4768,9 @@ class AIAgent: """ items = assistant_msg.get("codex_message_items") if not isinstance(items, list): - return "" + return [] - parts: List[str] = [] + messages: List[str] = [] for item in items: if not isinstance(item, dict): continue @@ -4779,6 +4782,7 @@ class AIAgent: content_parts = item.get("content") if not isinstance(content_parts, list): continue + item_parts: List[str] = [] for part in content_parts: if not isinstance(part, dict): continue @@ -4786,13 +4790,20 @@ class AIAgent: continue text = part.get("text") if isinstance(text, str) and text.strip(): - parts.append(text.strip()) + item_parts.append(text) + visible = "".join(item_parts).strip() + if visible: + visible = self._strip_think_blocks(visible).strip() + visible = redact_sensitive_text(visible) + if visible: + messages.append(visible) + return messages - visible = "\n\n".join(parts).strip() - if visible: - visible = self._strip_think_blocks(visible).strip() - visible = redact_sensitive_text(visible) - return visible + def _extract_codex_interim_visible_text(self, assistant_msg: Dict[str, Any]) -> str: + """Extract all visible Codex commentary for comparison/fallback.""" + return "\n\n".join( + self._extract_codex_interim_visible_parts(assistant_msg) + ).strip() def _interim_assistant_visible_text(self, assistant_msg: Dict[str, Any]) -> str: """Return the exact assistant text eligible for interim delivery. @@ -4808,17 +4819,74 @@ class AIAgent: content = assistant_msg.get("content") return self._strip_think_blocks(content or "").strip() + def _interim_text_was_delivered(self, text: str) -> bool: + normalized = self._normalize_interim_visible_text(text) + if not normalized: + return False + return normalized in getattr(self, "_delivered_interim_texts", set()) + + def _record_delivered_interim_text(self, text: str) -> None: + normalized = self._normalize_interim_visible_text(text) + if normalized: + delivered = getattr(self, "_delivered_interim_texts", None) + if not isinstance(delivered, set): + delivered = set() + self._delivered_interim_texts = delivered + delivered.add(normalized) + + def _fire_streamed_codex_commentary(self, text: str) -> None: + """Deliver a completed live Codex commentary message immediately.""" + cb = getattr(self, "interim_assistant_callback", None) + if cb is None or not isinstance(text, str): + return + visible = self._strip_think_blocks(text).strip() + if visible: + visible = redact_sensitive_text(visible) + if not visible or visible == "(empty)" or self._interim_text_was_delivered(visible): + return + try: + cb(visible, already_streamed=False) + self._record_delivered_interim_text(visible) + except Exception: + logger.debug("interim_assistant_callback error", exc_info=True) + def _emit_interim_assistant_message(self, assistant_msg: Dict[str, Any]) -> None: """Surface a real mid-turn assistant commentary message to the UI layer.""" cb = getattr(self, "interim_assistant_callback", None) if cb is None or not isinstance(assistant_msg, dict): return - visible = self._interim_assistant_visible_text(assistant_msg) - if not visible or visible == "(empty)": + commentary_parts = self._extract_codex_interim_visible_parts(assistant_msg) + undelivered_parts: List[str] = [] + pending_keys: set[str] = set() + for part in commentary_parts: + key = self._normalize_interim_visible_text(part) + if ( + not key + or key in pending_keys + or self._interim_text_was_delivered(part) + ): + continue + pending_keys.add(key) + undelivered_parts.append(part) + visible = ( + "\n\n".join(undelivered_parts).strip() + if commentary_parts + else self._interim_assistant_visible_text(assistant_msg) + ) + if ( + not visible + or visible == "(empty)" + or self._interim_text_was_delivered(visible) + ): return already_streamed = self._interim_content_was_streamed(visible) try: cb(visible, already_streamed=already_streamed) + if undelivered_parts: + for part in undelivered_parts: + self._record_delivered_interim_text(part) + else: + self._record_delivered_interim_text(visible) except Exception: logger.debug("interim_assistant_callback error", exc_info=True) diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index b963e50eef30..382f864634ff 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/tests/run_agent/test_run_agent_codex_responses.py @@ -704,6 +704,49 @@ def test_consume_codex_stream_routes_commentary_phase_deltas_to_reasoning(monkey assert response.output_text == "" +def test_consume_codex_stream_separates_commentary_from_analysis(monkeypatch): + from agent.codex_runtime import _consume_codex_event_stream + + commentary_item = SimpleNamespace( + type="message", + phase="commentary", + status="completed", + content=[SimpleNamespace(type="output_text", text="I'll inspect the repo first.")], + ) + streamed = [] + reasoning_streamed = [] + commentary_messages = [] + + response = _consume_codex_event_stream( + _FakeCreateStream([ + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="message", phase="commentary"), + ), + SimpleNamespace(type="response.output_text.delta", delta="I'll inspect "), + SimpleNamespace(type="response.output_text.delta", delta="the repo first."), + SimpleNamespace(type="response.output_item.done", item=commentary_item), + SimpleNamespace( + type="response.reasoning_text.delta", + delta="Need inspect files privately.", + ), + SimpleNamespace( + type="response.completed", + response=SimpleNamespace(status="completed"), + ), + ]), + model="gpt-5-codex", + on_text_delta=streamed.append, + on_reasoning_delta=reasoning_streamed.append, + on_commentary_message=commentary_messages.append, + ) + + assert commentary_messages == ["I'll inspect the repo first."] + assert reasoning_streamed == ["Need inspect files privately."] + assert streamed == [] + assert response.output == [commentary_item] + + def test_consume_codex_stream_keeps_final_answer_phase_deltas(monkeypatch): from agent.codex_runtime import _consume_codex_event_stream @@ -726,6 +769,201 @@ def test_consume_codex_stream_keeps_final_answer_phase_deltas(monkeypatch): assert response.output_text == "visible answer" +def test_run_codex_stream_delivers_redacted_commentary_once(monkeypatch): + from agent.codex_responses_adapter import _normalize_codex_response + + agent = _build_agent(monkeypatch) + monkeypatch.setattr("agent.redact._REDACT_ENABLED", True) + delivered = [] + reasoning_streamed = [] + agent.interim_assistant_callback = ( + lambda text, *, already_streamed=False: delivered.append( + (text, already_streamed) + ) + ) + agent.reasoning_callback = reasoning_streamed.append + secret = "sk-" + ("A" * 32) + commentary_text = f"Using credential {secret}. I'll inspect the repo." + commentary_item = SimpleNamespace( + type="message", + phase="commentary", + status="completed", + content=[SimpleNamespace(type="output_text", text=commentary_text)], + ) + function_item = SimpleNamespace( + type="function_call", + id="fc_1", + call_id="call_1", + name="terminal", + arguments="{}", + ) + + def _fake_create(**kwargs): + assert kwargs.get("stream") is True + return _FakeCreateStream([ + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="message", phase="commentary"), + ), + SimpleNamespace(type="response.output_text.delta", delta=commentary_text), + SimpleNamespace(type="response.output_item.done", item=commentary_item), + SimpleNamespace(type="response.reasoning_text.delta", delta="Private scratchpad."), + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="function_call"), + ), + SimpleNamespace(type="response.output_item.done", item=function_item), + SimpleNamespace( + type="response.completed", + response=SimpleNamespace(status="completed"), + ), + ]) + + agent.client = SimpleNamespace(responses=SimpleNamespace(create=_fake_create)) + + response = agent._run_codex_stream(_codex_request_kwargs()) + + assert len(delivered) == 1 + assert delivered[0][1] is False + assert secret not in delivered[0][0] + assert "Using credential" in delivered[0][0] + assert reasoning_streamed == ["Private scratchpad."] + + # The completed-response fallback sees the same preserved commentary but + # must not enqueue it again after live delivery. + normalized, finish_reason = _normalize_codex_response(response) + agent._emit_interim_assistant_message( + agent._build_assistant_message(normalized, finish_reason) + ) + assert len(delivered) == 1 + + +def test_run_codex_stream_multiple_commentary_items_are_not_reemitted(monkeypatch): + from agent.codex_responses_adapter import _normalize_codex_response + + agent = _build_agent(monkeypatch) + delivered = [] + agent.interim_assistant_callback = ( + lambda text, *, already_streamed=False: delivered.append(text) + ) + commentary_a = SimpleNamespace( + type="message", + phase="commentary", + status="completed", + content=[SimpleNamespace(type="output_text", text="First update.")], + ) + commentary_b = SimpleNamespace( + type="message", + phase="commentary", + status="completed", + content=[SimpleNamespace(type="output_text", text="Second update.")], + ) + function_item = SimpleNamespace( + type="function_call", + id="fc_1", + call_id="call_1", + name="terminal", + arguments="{}", + ) + + def _fake_create(**kwargs): + return _FakeCreateStream([ + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="message", phase="commentary"), + ), + SimpleNamespace(type="response.output_text.delta", delta="First update."), + SimpleNamespace(type="response.output_item.done", item=commentary_a), + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="message", phase="commentary"), + ), + SimpleNamespace(type="response.output_text.delta", delta="Second update."), + SimpleNamespace(type="response.output_item.done", item=commentary_b), + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="function_call"), + ), + SimpleNamespace(type="response.output_item.done", item=function_item), + SimpleNamespace( + type="response.completed", + response=SimpleNamespace(status="completed"), + ), + ]) + + agent.client = SimpleNamespace(responses=SimpleNamespace(create=_fake_create)) + response = agent._run_codex_stream(_codex_request_kwargs()) + normalized, finish_reason = _normalize_codex_response(response) + agent._emit_interim_assistant_message( + agent._build_assistant_message(normalized, finish_reason) + ) + + assert delivered == ["First update.", "Second update."] + + +def test_run_codex_stream_retry_deduplicates_multiple_commentary_items(monkeypatch): + import httpx + + agent = _build_agent(monkeypatch) + delivered = [] + agent.interim_assistant_callback = ( + lambda text, *, already_streamed=False: delivered.append(text) + ) + commentary_a = SimpleNamespace( + type="message", + phase="commentary", + status="completed", + content=[SimpleNamespace(type="output_text", text="First update.")], + ) + commentary_b = SimpleNamespace( + type="message", + phase="commentary", + status="completed", + content=[SimpleNamespace(type="output_text", text="Second update.")], + ) + + class _DroppingStream(_FakeCreateStream): + def __iter__(self): + yield from super().__iter__() + raise httpx.RemoteProtocolError("connection dropped") + + commentary_events = [ + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="message", phase="commentary"), + ), + SimpleNamespace(type="response.output_text.delta", delta="First update."), + SimpleNamespace(type="response.output_item.done", item=commentary_a), + SimpleNamespace( + type="response.output_item.added", + item=SimpleNamespace(type="message", phase="commentary"), + ), + SimpleNamespace(type="response.output_text.delta", delta="Second update."), + SimpleNamespace(type="response.output_item.done", item=commentary_b), + ] + calls = {"count": 0} + + def _fake_create(**kwargs): + calls["count"] += 1 + if calls["count"] == 1: + return _DroppingStream(commentary_events) + return _FakeCreateStream([ + *commentary_events, + SimpleNamespace( + type="response.completed", + response=SimpleNamespace(status="completed"), + ), + ]) + + agent.client = SimpleNamespace(responses=SimpleNamespace(create=_fake_create)) + + response = agent._run_codex_stream(_codex_request_kwargs()) + + assert response.status == "completed" + assert calls["count"] == 2 + assert delivered == ["First update.", "Second update."] + + def test_run_codex_stream_surfaces_failed_status_in_final_response(monkeypatch): """A ``response.failed`` terminal event is reflected on the returned object.""" agent = _build_agent(monkeypatch) @@ -2183,6 +2421,28 @@ def test_interim_commentary_redacts_secrets_from_codex_commentary_items(monkeypa assert "Using credential" in observed[0] +def test_interim_commentary_deduplicates_identical_items_in_one_response(monkeypatch): + agent = _build_agent(monkeypatch) + observed = [] + agent.interim_assistant_callback = ( + lambda text, *, already_streamed=False: observed.append(text) + ) + commentary_item = { + "type": "message", + "role": "assistant", + "phase": "commentary", + "content": [{"type": "output_text", "text": "Still working."}], + } + + agent._emit_interim_assistant_message({ + "role": "assistant", + "content": "", + "codex_message_items": [commentary_item, dict(commentary_item)], + }) + + assert observed == ["Still working."] + + def test_stream_delta_strips_leaked_memory_context(monkeypatch): agent = _build_agent(monkeypatch) observed = []