realtime: conversation.item.retrieved regresses completed history items to in_progress

[shoutoutuoadi325]

Please read this first

• Have you read the docs? Agents SDK docs — yes, including the Realtime sections.
• Have you searched for related issues? Yes.

Describe the bug

conversation_item_to_realtime_message_item in src/agents/realtime/openai_realtime.py hardcodes "status": "in_progress" when converting a server ConversationItem, so the status carried by the server payload is dropped.

This is reachable in a default voice session. After the assistant produces audio, _current_item_id points at the assistant's audio item (set in _handle_audio_delta). When the user's next utterance finishes transcription, the handler for conversation.item.input_audio_transcription.completed sends conversation.item.retrieve for that item, and the server replies with conversation.item.retrieved carrying the item's actual status (completed). The hardcoded conversion rewrites that to in_progress, and the session history merge keeps the incoming status, so the completed assistant item regresses to in_progress in history after every user turn, with nothing to restore it. The same retrieve is also triggered by conversation.item.truncated.

The GA conversation item models do carry status: Optional[Literal["in_progress", "completed", "incomplete"]], so the value is available on the parsed item and just needs to be passed through.

Effect: history items flip back to in_progress permanently, and consumers watching RealtimeHistoryUpdated see stale statuses plus a spurious extra update per user turn.

Debug information

• Agents SDK version: 0.22.0 (also present on current main)
• Python version: 3.12
• Operating system: Linux
• Model and model provider: any realtime model; the repro below is offline with a stubbed websocket
• Does the issue reproduce with the latest Agents SDK release? Yes.
• Does the issue occur consistently or intermittently? Consistently — it fires on every user utterance that follows assistant audio when input transcription is enabled.

Repro steps

import asyncio
import base64
import json
from unittest.mock import AsyncMock, Mock

from agents.realtime.agent import RealtimeAgent
from agents.realtime.items import AssistantMessageItem
from agents.realtime.openai_realtime import OpenAIRealtimeWebSocketModel
from agents.realtime.session import RealtimeSession


async def main() -> None:
    model = OpenAIRealtimeWebSocketModel()
    ws = Mock()
    ws.send = AsyncMock()
    model._websocket = ws

    session = RealtimeSession(model, RealtimeAgent(name="agent"), None)
    model.add_listener(session)

    async def feed(event: dict) -> None:
        await model._handle_ws_event(event)

    # Turn 1: assistant audio response completes.
    await feed({"type": "response.created", "event_id": "e1", "response": {"id": "resp_1"}})
    await feed({
        "type": "response.output_item.added",
        "response_id": "resp_1",
        "output_index": 0,
        "item": {
            "id": "item_A",
            "type": "message",
            "role": "assistant",
            "status": "in_progress",
            "content": [{"type": "output_audio", "transcript": None}],
        },
    })
    await feed({
        "type": "response.output_audio.delta",
        "event_id": "e3",
        "response_id": "resp_1",
        "item_id": "item_A",
        "content_index": 0,
        "output_index": 0,
        "delta": base64.b64encode(b"\x00\x00" * 2400).decode(),
    })
    await feed({
        "type": "response.output_item.done",
        "response_id": "resp_1",
        "output_index": 0,
        "item": {
            "id": "item_A",
            "type": "message",
            "role": "assistant",
            "status": "completed",
            "content": [{"type": "output_audio", "transcript": "hello there"}],
        },
    })
    await feed({"type": "response.done", "event_id": "e5", "response": {"id": "resp_1"}})

    item = session._history[0]
    assert isinstance(item, AssistantMessageItem)
    print("after turn 1:", item.status)  # completed

    # Turn 2: user speaks; transcription completing triggers a retrieve of item_A.
    await feed({
        "type": "conversation.item.input_audio_transcription.completed",
        "event_id": "e6",
        "item_id": "item_B",
        "content_index": 0,
        "transcript": "user speech",
        "usage": {"type": "tokens", "input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
    })
    await feed({
        "type": "conversation.item.retrieved",
        "event_id": "e7",
        "item": {
            "id": "item_A",
            "type": "message",
            "role": "assistant",
            "status": "completed",
            "content": [{"type": "output_audio", "transcript": "hello there"}],
        },
    })

    item = session._history[0]
    print("after retrieve:", item.status)  # expected completed, actual in_progress


asyncio.run(main())

Output on 0.22.0:

after turn 1: completed
after retrieve: in_progress

The server explicitly reported status: "completed" in the retrieved payload, but the history entry ends up in_progress.

[leadingproblemsolver]

One regression I’d add beyond the direct converter fix: exercise the actual history-reconciliation path, not only _ConversionHelper.

conversation_item_to_realtime_message_item() currently hardcodes status="in_progress", but RealtimeSession then treats the converted item as an item_updated reconciliation event. That session path already has explicit protection for another retrieve regression: it preserves a previously known transcript when a retrieved item arrives without transcript fields populated.

A discriminating fixture would be:

1. history contains assistant item item_A with status="completed" and transcript "hello";
2. feed conversation.item.retrieved for item_A with server status "completed" and missing/empty transcript;
3. assert the final history item is still status="completed" and still has transcript "hello".

I’d parameterize the same test for status="incomplete", since that is also a valid server-carried status. And because the generated ConversationItem status is optional, add one explicit status=None case: either assert that conversation.item.retrieved is guaranteed to carry status, or define the fallback intentionally. The important part is not letting an omitted status silently manufacture another in_progress regression.

That gives a stronger contract than just “copy item.status”: a retrieve/reconciliation event may refresh an existing history item, but it should not discard already-known lifecycle state or content evidence unless the protocol explicitly tells us to.