fix(codex): reset the streamed-text buffer at each completed agentMessage
codex app-server emits commentary deltas + completed, then final deltas + completed. The delta buffer accumulated "commentary + final", so the completed final no longer prefix-matched its own deltas, was delivered with already_streamed=False and the gateway posted a second copy next to the streamed message (#74248 boundary 2). Drop the buffer once a completed agentMessage has been compared against its own deltas.
This commit is contained in:
@@ -337,6 +337,11 @@ def make_codex_app_server_event_bridge(agent) -> Callable[[dict], None]:
|
||||
if isinstance(text, str) and text.strip() and getattr(agent, "show_commentary", True):
|
||||
agent_cb("_emit_interim_assistant_message", "_emit_interim_assistant_message raised",
|
||||
args=({"role": "assistant", "content": text},))
|
||||
# Each agentMessage item is its own delivered message: the completed item was just compared
|
||||
# against ITS deltas, so drop them before the next item's deltas arrive. Otherwise the buffer
|
||||
# holds "commentary + final", the final agentMessage no longer prefix-matches, and it is
|
||||
# re-delivered with already_streamed=False as a second copy (#74248 boundary 2).
|
||||
agent._current_streamed_assistant_text = ""
|
||||
|
||||
def _on_item(params: dict, completed: bool) -> None:
|
||||
item = params.get("item")
|
||||
|
||||
@@ -1997,6 +1997,31 @@ def test_interim_commentary_is_not_marked_already_streamed_without_callbacks(mon
|
||||
}
|
||||
|
||||
|
||||
def test_app_server_bridge_commentary_then_final_agent_messages_are_each_already_streamed(monkeypatch):
|
||||
"""#74248 boundary 2: codex app-server emits commentary deltas + completed, then final deltas +
|
||||
completed. The completed final must compare against ITS OWN deltas, not "commentary + final", or
|
||||
it is re-delivered with already_streamed=False and the gateway posts a second copy."""
|
||||
from agent.codex_runtime import make_codex_app_server_event_bridge
|
||||
|
||||
agent = _build_agent(monkeypatch)
|
||||
agent.stream_delta_callback = lambda text: None
|
||||
deliveries = []
|
||||
agent.interim_assistant_callback = lambda text, *, already_streamed=False: deliveries.append(
|
||||
(text, already_streamed)
|
||||
)
|
||||
on_event = make_codex_app_server_event_bridge(agent)
|
||||
|
||||
def _agent_message(item_id, text, phase):
|
||||
on_event({"method": "item/agentMessage/delta", "params": {"itemId": item_id, "delta": text}})
|
||||
on_event({"method": "item/completed", "params": {
|
||||
"item": {"id": item_id, "type": "agentMessage", "text": text, "phase": phase}}})
|
||||
|
||||
_agent_message("m1", "Checking the config.", "commentary")
|
||||
_agent_message("m2", "Native compaction is active.", "final_answer")
|
||||
|
||||
assert deliveries == [("Checking the config.", True), ("Native compaction is active.", True)]
|
||||
assert agent._current_streamed_assistant_text == ""
|
||||
|
||||
|
||||
|
||||
def test_interim_content_was_streamed_matches_prefix_not_exact(monkeypatch):
|
||||
|
||||
Reference in New Issue
Block a user