Files
codex/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs
T
guinness-oai c976741124 [codex] Flush trailing realtime transcript tail (#29918)
## Why

A realtime session can end after more transcript has accumulated than
was included in its last handoff. That tail already lives in core's
active transcript state, but the stop path aborted the realtime
input/fanout tasks before routing it, so the final bit of the
conversation could disappear before `thread/realtime/closed`.

This behavior is still being evaluated, so clients must opt in per
realtime session. Omitted or false leaves shutdown behavior unchanged.

## What changed

- Add optional `flushTranscriptTailOnSessionEnd` to
`thread/realtime/start`, defaulting to false in app-server.
- Expose an idempotent `take_transcript_tail()` from the existing active
transcript state using `last_handoff_entry_count`.
- When enabled, let shutdown cancel the input owner cleanly and publish
at most one final existing `<realtime_delegation>` with the remaining
text in `<transcript_delta>`.
- Have the existing fanout drain already-parsed events before routing
that final delegation, so a queued handoff wins first and is not
duplicated in the tail.
- Flush realtime shutdown before ordinary session task abort during core
cleanup.

## Validation

- `just test -p codex-app-server-protocol`
- `just test -p codex-app-server
realtime_conversation_stop_emits_closed_notification`
- `just test -p codex-core conversation_transport_close_`
- `just test -p codex-core
conversation_close_routes_only_remaining_transcript_tail_once`
- scoped `just fix` for `codex-app-server-protocol`, `codex-app-server`,
`codex-protocol`, and `codex-core`
- `just fmt`
2026-07-06 14:12:54 -07:00

288 lines
10 KiB
Rust

use codex_protocol::protocol::ConversationTextRole;
use codex_protocol::protocol::RealtimeAudioFrame as CoreRealtimeAudioFrame;
use codex_protocol::protocol::RealtimeConversationVersion;
use codex_protocol::protocol::RealtimeOutputModality;
use codex_protocol::protocol::RealtimeVoice;
use codex_protocol::protocol::RealtimeVoicesList;
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value as JsonValue;
use ts_rs::TS;
/// EXPERIMENTAL - thread realtime audio chunk.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeAudioChunk {
pub data: String,
pub sample_rate: u32,
pub num_channels: u16,
pub samples_per_channel: Option<u32>,
pub item_id: Option<String>,
}
impl From<CoreRealtimeAudioFrame> for ThreadRealtimeAudioChunk {
fn from(value: CoreRealtimeAudioFrame) -> Self {
let CoreRealtimeAudioFrame {
data,
sample_rate,
num_channels,
samples_per_channel,
item_id,
} = value;
Self {
data,
sample_rate,
num_channels,
samples_per_channel,
item_id,
}
}
}
impl From<ThreadRealtimeAudioChunk> for CoreRealtimeAudioFrame {
fn from(value: ThreadRealtimeAudioChunk) -> Self {
let ThreadRealtimeAudioChunk {
data,
sample_rate,
num_channels,
samples_per_channel,
item_id,
} = value;
Self {
data,
sample_rate,
num_channels,
samples_per_channel,
item_id,
}
}
}
/// EXPERIMENTAL - start a thread-scoped realtime session.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeStartParams {
pub thread_id: String,
/// Leaves Codex response handoffs to the client's explicit append calls instead of forwarding
/// them automatically. Defaults to false.
#[ts(optional = nullable)]
pub client_managed_handoffs: Option<bool>,
/// Routes any transcript tail remaining at session end through Codex. Defaults to false.
/// TODO: Remove this rollout knob once transcript-tail flushing is always enabled.
#[ts(optional = nullable)]
pub flush_transcript_tail_on_session_end: Option<bool>,
/// Sends automatic Codex responses as realtime conversation items instead of handoff appends.
#[ts(optional = nullable)]
pub codex_responses_as_items: Option<bool>,
/// Optional prefix added to automatic Codex response items when `codexResponsesAsItems` is true.
#[ts(optional = nullable)]
pub codex_response_item_prefix: Option<String>,
/// Optional prefix added to automatic V1 Codex commentary sent with
/// `conversation.handoff.append` when `codexResponsesAsItems` is not true. Final answers are
/// sent without the prefix.
#[ts(optional = nullable)]
pub codex_response_handoff_prefix: Option<String>,
/// Overrides the configured realtime model for this session only.
#[ts(optional = nullable)]
pub model: Option<String>,
/// Selects text or audio output for the realtime session. Transport and voice stay
/// independent so clients can choose how they connect separately from what the model emits.
pub output_modality: RealtimeOutputModality,
/// Set to false to start without Codex's startup context. Omitted or null includes it.
#[ts(optional = nullable)]
pub include_startup_context: Option<bool>,
#[serde(
default,
deserialize_with = "crate::protocol::serde_helpers::deserialize_double_option",
serialize_with = "crate::protocol::serde_helpers::serialize_double_option",
skip_serializing_if = "Option::is_none"
)]
#[ts(optional = nullable)]
pub prompt: Option<Option<String>>,
#[ts(optional = nullable)]
pub realtime_session_id: Option<String>,
#[ts(optional = nullable)]
pub transport: Option<ThreadRealtimeStartTransport>,
/// Overrides the configured realtime protocol version for this session only.
#[ts(optional = nullable)]
pub version: Option<RealtimeConversationVersion>,
#[ts(optional = nullable)]
pub voice: Option<RealtimeVoice>,
}
/// EXPERIMENTAL - transport used by thread realtime.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(tag = "type", rename_all = "camelCase")]
#[ts(export_to = "v2/", tag = "type")]
pub enum ThreadRealtimeStartTransport {
Websocket,
Webrtc {
/// SDP offer generated by a WebRTC RTCPeerConnection after configuring audio and the
/// realtime events data channel.
sdp: String,
},
}
/// EXPERIMENTAL - response for starting thread realtime.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeStartResponse {}
/// EXPERIMENTAL - append audio input to thread realtime.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeAppendAudioParams {
pub thread_id: String,
pub audio: ThreadRealtimeAudioChunk,
}
/// EXPERIMENTAL - response for appending realtime audio input.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeAppendAudioResponse {}
/// EXPERIMENTAL - append text input to thread realtime.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeAppendTextParams {
pub thread_id: String,
pub text: String,
#[serde(default)]
pub role: ConversationTextRole,
}
/// EXPERIMENTAL - response for appending realtime text input.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeAppendTextResponse {}
/// EXPERIMENTAL - append speakable text to thread realtime.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeAppendSpeechParams {
pub thread_id: String,
pub text: String,
}
/// EXPERIMENTAL - response for appending realtime speech.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeAppendSpeechResponse {}
/// EXPERIMENTAL - stop thread realtime.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeStopParams {
pub thread_id: String,
}
/// EXPERIMENTAL - response for stopping thread realtime.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeStopResponse {}
/// EXPERIMENTAL - list voices supported by thread realtime.
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeListVoicesParams {}
/// EXPERIMENTAL - response for listing supported realtime voices.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeListVoicesResponse {
pub voices: RealtimeVoicesList,
}
/// EXPERIMENTAL - emitted when thread realtime startup is accepted.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeStartedNotification {
pub thread_id: String,
pub realtime_session_id: Option<String>,
pub version: RealtimeConversationVersion,
}
/// EXPERIMENTAL - raw non-audio thread realtime item emitted by the backend.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeItemAddedNotification {
pub thread_id: String,
pub item: JsonValue,
}
/// EXPERIMENTAL - flat transcript delta emitted whenever realtime
/// transcript text changes.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeTranscriptDeltaNotification {
pub thread_id: String,
pub role: String,
/// Live transcript delta from the realtime event.
pub delta: String,
}
/// EXPERIMENTAL - final transcript text emitted when realtime completes
/// a transcript part.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeTranscriptDoneNotification {
pub thread_id: String,
pub role: String,
/// Final complete text for the transcript part.
pub text: String,
}
/// EXPERIMENTAL - streamed output audio emitted by thread realtime.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeOutputAudioDeltaNotification {
pub thread_id: String,
pub audio: ThreadRealtimeAudioChunk,
}
/// EXPERIMENTAL - emitted with the remote SDP for a WebRTC realtime session.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeSdpNotification {
pub thread_id: String,
pub sdp: String,
}
/// EXPERIMENTAL - emitted when thread realtime encounters an error.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeErrorNotification {
pub thread_id: String,
pub message: String,
}
/// EXPERIMENTAL - emitted when thread realtime transport closes.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRealtimeClosedNotification {
pub thread_id: String,
pub reason: Option<String>,
}