diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 3eb4d27eea..ccf6503dac 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -6141,6 +6141,25 @@ ], "type": "object" }, + "ThreadRealtimeFrameChannels": { + "description": "A frame holds one or more payloads, at most one per channel. They share one ID, observation time, and seal. V1 requires its only channel, `voice_transcript`.", + "properties": { + "voice_transcript": { + "$ref": "#/definitions/ThreadRealtimeVoiceTranscript" + } + }, + "required": [ + "voice_transcript" + ], + "type": "object" + }, + "ThreadRealtimeHandoffMode": { + "enum": [ + "start", + "steer" + ], + "type": "string" + }, "ThreadRealtimeItem": { "description": "EXPERIMENTAL - a thread-scoped realtime item in the canonical timeline.", "oneOf": [ @@ -6160,6 +6179,26 @@ "title": "RealtimeSessionStartedThreadRealtimeItem", "type": "object" }, + { + "properties": { + "channels": { + "$ref": "#/definitions/ThreadRealtimeFrameChannels" + }, + "type": { + "enum": [ + "frame" + ], + "title": "FrameThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "channels", + "type" + ], + "title": "FrameThreadRealtimeItem", + "type": "object" + }, { "properties": { "role": { @@ -6231,12 +6270,94 @@ ], "title": "RealtimeSessionClosedThreadRealtimeItem", "type": "object" + }, + { + "properties": { + "handoff_id": { + "type": "string" + }, + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "mode": { + "$ref": "#/definitions/ThreadRealtimeHandoffMode" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "turn_id": { + "type": "string" + }, + "type": { + "enum": [ + "turnInputAccepted" + ], + "title": "TurnInputAcceptedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "handoff_id", + "input", + "input_item_id", + "mode", + "source_event_ids", + "turn_id", + "type" + ], + "title": "TurnInputAcceptedThreadRealtimeItem", + "type": "object" + }, + { + "properties": { + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "type": { + "enum": [ + "sessionEndInputSaved" + ], + "title": "SessionEndInputSavedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "input", + "input_item_id", + "source_event_ids", + "type" + ], + "title": "SessionEndInputSavedThreadRealtimeItem", + "type": "object" } ], "properties": { "id": { "type": "string" }, + "observedAtMs": { + "description": "Canonical observation time in Unix milliseconds for eligible voice calls.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "realtimeSessionId": { "type": "string" } @@ -6423,6 +6544,21 @@ ], "type": "string" }, + "ThreadRealtimeVoiceTranscript": { + "properties": { + "role": { + "$ref": "#/definitions/ThreadRealtimeTranscriptRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" + }, "ThreadRevertedNotification": { "properties": { "threadId": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 7db5e5af8b..ecef4575ce 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -25579,6 +25579,25 @@ "title": "ThreadRealtimeErrorNotification", "type": "object" }, + "ThreadRealtimeFrameChannels": { + "description": "A frame holds one or more payloads, at most one per channel. They share one ID, observation time, and seal. V1 requires its only channel, `voice_transcript`.", + "properties": { + "voice_transcript": { + "$ref": "#/definitions/v2/ThreadRealtimeVoiceTranscript" + } + }, + "required": [ + "voice_transcript" + ], + "type": "object" + }, + "ThreadRealtimeHandoffMode": { + "enum": [ + "start", + "steer" + ], + "type": "string" + }, "ThreadRealtimeInitialItem": { "description": "EXPERIMENTAL - role-bearing text item included when a realtime V3 session starts.", "properties": { @@ -25614,6 +25633,26 @@ "title": "RealtimeSessionStartedThreadRealtimeItem", "type": "object" }, + { + "properties": { + "channels": { + "$ref": "#/definitions/v2/ThreadRealtimeFrameChannels" + }, + "type": { + "enum": [ + "frame" + ], + "title": "FrameThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "channels", + "type" + ], + "title": "FrameThreadRealtimeItem", + "type": "object" + }, { "properties": { "role": { @@ -25685,12 +25724,94 @@ ], "title": "RealtimeSessionClosedThreadRealtimeItem", "type": "object" + }, + { + "properties": { + "handoff_id": { + "type": "string" + }, + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "mode": { + "$ref": "#/definitions/v2/ThreadRealtimeHandoffMode" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "turn_id": { + "type": "string" + }, + "type": { + "enum": [ + "turnInputAccepted" + ], + "title": "TurnInputAcceptedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "handoff_id", + "input", + "input_item_id", + "mode", + "source_event_ids", + "turn_id", + "type" + ], + "title": "TurnInputAcceptedThreadRealtimeItem", + "type": "object" + }, + { + "properties": { + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "type": { + "enum": [ + "sessionEndInputSaved" + ], + "title": "SessionEndInputSavedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "input", + "input_item_id", + "source_event_ids", + "type" + ], + "title": "SessionEndInputSavedThreadRealtimeItem", + "type": "object" } ], "properties": { "id": { "type": "string" }, + "observedAtMs": { + "description": "Canonical observation time in Unix milliseconds for eligible voice calls.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "realtimeSessionId": { "type": "string" } @@ -25958,6 +26079,21 @@ ], "type": "string" }, + "ThreadRealtimeVoiceTranscript": { + "properties": { + "role": { + "$ref": "#/definitions/v2/ThreadRealtimeTranscriptRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" + }, "ThreadResumeInitialTurnsPageParams": { "properties": { "itemsView": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 7957f73971..e9e4e692f4 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -23275,6 +23275,25 @@ "title": "ThreadRealtimeErrorNotification", "type": "object" }, + "ThreadRealtimeFrameChannels": { + "description": "A frame holds one or more payloads, at most one per channel. They share one ID, observation time, and seal. V1 requires its only channel, `voice_transcript`.", + "properties": { + "voice_transcript": { + "$ref": "#/definitions/ThreadRealtimeVoiceTranscript" + } + }, + "required": [ + "voice_transcript" + ], + "type": "object" + }, + "ThreadRealtimeHandoffMode": { + "enum": [ + "start", + "steer" + ], + "type": "string" + }, "ThreadRealtimeInitialItem": { "description": "EXPERIMENTAL - role-bearing text item included when a realtime V3 session starts.", "properties": { @@ -23310,6 +23329,26 @@ "title": "RealtimeSessionStartedThreadRealtimeItem", "type": "object" }, + { + "properties": { + "channels": { + "$ref": "#/definitions/ThreadRealtimeFrameChannels" + }, + "type": { + "enum": [ + "frame" + ], + "title": "FrameThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "channels", + "type" + ], + "title": "FrameThreadRealtimeItem", + "type": "object" + }, { "properties": { "role": { @@ -23381,12 +23420,94 @@ ], "title": "RealtimeSessionClosedThreadRealtimeItem", "type": "object" + }, + { + "properties": { + "handoff_id": { + "type": "string" + }, + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "mode": { + "$ref": "#/definitions/ThreadRealtimeHandoffMode" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "turn_id": { + "type": "string" + }, + "type": { + "enum": [ + "turnInputAccepted" + ], + "title": "TurnInputAcceptedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "handoff_id", + "input", + "input_item_id", + "mode", + "source_event_ids", + "turn_id", + "type" + ], + "title": "TurnInputAcceptedThreadRealtimeItem", + "type": "object" + }, + { + "properties": { + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "type": { + "enum": [ + "sessionEndInputSaved" + ], + "title": "SessionEndInputSavedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "input", + "input_item_id", + "source_event_ids", + "type" + ], + "title": "SessionEndInputSavedThreadRealtimeItem", + "type": "object" } ], "properties": { "id": { "type": "string" }, + "observedAtMs": { + "description": "Canonical observation time in Unix milliseconds for eligible voice calls.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "realtimeSessionId": { "type": "string" } @@ -23654,6 +23775,21 @@ ], "type": "string" }, + "ThreadRealtimeVoiceTranscript": { + "properties": { + "role": { + "$ref": "#/definitions/ThreadRealtimeTranscriptRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" + }, "ThreadResumeInitialTurnsPageParams": { "properties": { "itemsView": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemCompletedNotification.json index b24e3169ef..6ff298dfc0 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemCompletedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemCompletedNotification.json @@ -60,6 +60,25 @@ } ] }, + "ThreadRealtimeFrameChannels": { + "description": "A frame holds one or more payloads, at most one per channel. They share one ID, observation time, and seal. V1 requires its only channel, `voice_transcript`.", + "properties": { + "voice_transcript": { + "$ref": "#/definitions/ThreadRealtimeVoiceTranscript" + } + }, + "required": [ + "voice_transcript" + ], + "type": "object" + }, + "ThreadRealtimeHandoffMode": { + "enum": [ + "start", + "steer" + ], + "type": "string" + }, "ThreadRealtimeItem": { "description": "EXPERIMENTAL - a thread-scoped realtime item in the canonical timeline.", "oneOf": [ @@ -79,6 +98,26 @@ "title": "RealtimeSessionStartedThreadRealtimeItem", "type": "object" }, + { + "properties": { + "channels": { + "$ref": "#/definitions/ThreadRealtimeFrameChannels" + }, + "type": { + "enum": [ + "frame" + ], + "title": "FrameThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "channels", + "type" + ], + "title": "FrameThreadRealtimeItem", + "type": "object" + }, { "properties": { "role": { @@ -150,12 +189,94 @@ ], "title": "RealtimeSessionClosedThreadRealtimeItem", "type": "object" + }, + { + "properties": { + "handoff_id": { + "type": "string" + }, + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "mode": { + "$ref": "#/definitions/ThreadRealtimeHandoffMode" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "turn_id": { + "type": "string" + }, + "type": { + "enum": [ + "turnInputAccepted" + ], + "title": "TurnInputAcceptedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "handoff_id", + "input", + "input_item_id", + "mode", + "source_event_ids", + "turn_id", + "type" + ], + "title": "TurnInputAcceptedThreadRealtimeItem", + "type": "object" + }, + { + "properties": { + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "type": { + "enum": [ + "sessionEndInputSaved" + ], + "title": "SessionEndInputSavedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "input", + "input_item_id", + "source_event_ids", + "type" + ], + "title": "SessionEndInputSavedThreadRealtimeItem", + "type": "object" } ], "properties": { "id": { "type": "string" }, + "observedAtMs": { + "description": "Canonical observation time in Unix milliseconds for eligible voice calls.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "realtimeSessionId": { "type": "string" } @@ -179,6 +300,21 @@ "assistant" ], "type": "string" + }, + "ThreadRealtimeVoiceTranscript": { + "properties": { + "role": { + "$ref": "#/definitions/ThreadRealtimeTranscriptRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" } }, "description": "EXPERIMENTAL - a realtime timeline item published after canonical commit.", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemStartedNotification.json index 74971aa932..7e026514a2 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRealtimeItemStartedNotification.json @@ -60,6 +60,25 @@ } ] }, + "ThreadRealtimeFrameChannels": { + "description": "A frame holds one or more payloads, at most one per channel. They share one ID, observation time, and seal. V1 requires its only channel, `voice_transcript`.", + "properties": { + "voice_transcript": { + "$ref": "#/definitions/ThreadRealtimeVoiceTranscript" + } + }, + "required": [ + "voice_transcript" + ], + "type": "object" + }, + "ThreadRealtimeHandoffMode": { + "enum": [ + "start", + "steer" + ], + "type": "string" + }, "ThreadRealtimeItem": { "description": "EXPERIMENTAL - a thread-scoped realtime item in the canonical timeline.", "oneOf": [ @@ -79,6 +98,26 @@ "title": "RealtimeSessionStartedThreadRealtimeItem", "type": "object" }, + { + "properties": { + "channels": { + "$ref": "#/definitions/ThreadRealtimeFrameChannels" + }, + "type": { + "enum": [ + "frame" + ], + "title": "FrameThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "channels", + "type" + ], + "title": "FrameThreadRealtimeItem", + "type": "object" + }, { "properties": { "role": { @@ -150,12 +189,94 @@ ], "title": "RealtimeSessionClosedThreadRealtimeItem", "type": "object" + }, + { + "properties": { + "handoff_id": { + "type": "string" + }, + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "mode": { + "$ref": "#/definitions/ThreadRealtimeHandoffMode" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "turn_id": { + "type": "string" + }, + "type": { + "enum": [ + "turnInputAccepted" + ], + "title": "TurnInputAcceptedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "handoff_id", + "input", + "input_item_id", + "mode", + "source_event_ids", + "turn_id", + "type" + ], + "title": "TurnInputAcceptedThreadRealtimeItem", + "type": "object" + }, + { + "properties": { + "input": { + "type": "string" + }, + "input_item_id": { + "type": "string" + }, + "source_event_ids": { + "items": { + "type": "string" + }, + "type": "array" + }, + "type": { + "enum": [ + "sessionEndInputSaved" + ], + "title": "SessionEndInputSavedThreadRealtimeItemType", + "type": "string" + } + }, + "required": [ + "input", + "input_item_id", + "source_event_ids", + "type" + ], + "title": "SessionEndInputSavedThreadRealtimeItem", + "type": "object" } ], "properties": { "id": { "type": "string" }, + "observedAtMs": { + "description": "Canonical observation time in Unix milliseconds for eligible voice calls.", + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "realtimeSessionId": { "type": "string" } @@ -179,6 +300,21 @@ "assistant" ], "type": "string" + }, + "ThreadRealtimeVoiceTranscript": { + "properties": { + "role": { + "$ref": "#/definitions/ThreadRealtimeTranscriptRole" + }, + "text": { + "type": "string" + } + }, + "required": [ + "role", + "text" + ], + "type": "object" } }, "description": "EXPERIMENTAL - a realtime timeline item started before its content streams.", diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index e82ec412f1..7c3428d238 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 772c99e365..0b1c8e6fcb 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeFrameChannels.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeFrameChannels.ts new file mode 100644 index 0000000000..f22b1c393e --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeFrameChannels.ts @@ -0,0 +1,10 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ThreadRealtimeVoiceTranscript } from "./ThreadRealtimeVoiceTranscript"; + +/** + * A frame holds one or more payloads, at most one per channel. They share one ID, + * observation time, and seal. V1 requires its only channel, `voice_transcript`. + */ +export type ThreadRealtimeFrameChannels = { voice_transcript: ThreadRealtimeVoiceTranscript, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeHandoffMode.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeHandoffMode.ts new file mode 100644 index 0000000000..ced9bb72d1 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeHandoffMode.ts @@ -0,0 +1,5 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type ThreadRealtimeHandoffMode = "start" | "steer"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItem.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItem.ts index 5ac5de98a2..a3cd2707a2 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItem.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeItem.ts @@ -2,10 +2,16 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { ThreadRealtimeBemItemPresentation } from "./ThreadRealtimeBemItemPresentation"; +import type { ThreadRealtimeFrameChannels } from "./ThreadRealtimeFrameChannels"; +import type { ThreadRealtimeHandoffMode } from "./ThreadRealtimeHandoffMode"; import type { ThreadRealtimeSessionOutcome } from "./ThreadRealtimeSessionOutcome"; import type { ThreadRealtimeTranscriptRole } from "./ThreadRealtimeTranscriptRole"; /** * EXPERIMENTAL - a thread-scoped realtime item in the canonical timeline. */ -export type ThreadRealtimeItem = { id: string, realtimeSessionId: string, } & ({ "type": "realtimeSessionStarted" } | { "type": "transcriptSegment", role: ThreadRealtimeTranscriptRole, text: string, } | { "type": "bemItemPromoted", turnId: string, itemId: string, presentation: ThreadRealtimeBemItemPresentation, } | { "type": "realtimeSessionClosed", outcome: ThreadRealtimeSessionOutcome, }); +export type ThreadRealtimeItem = { id: string, realtimeSessionId: string, +/** + * Canonical observation time in Unix milliseconds for eligible voice calls. + */ +observedAtMs?: number, } & ({ "type": "realtimeSessionStarted" } | { "type": "frame", channels: ThreadRealtimeFrameChannels, } | { "type": "transcriptSegment", role: ThreadRealtimeTranscriptRole, text: string, } | { "type": "bemItemPromoted", turnId: string, itemId: string, presentation: ThreadRealtimeBemItemPresentation, } | { "type": "realtimeSessionClosed", outcome: ThreadRealtimeSessionOutcome, } | { "type": "turnInputAccepted", handoffId: string, turnId: string, mode: ThreadRealtimeHandoffMode, sourceEventIds: Array, inputItemId: string, input: string, } | { "type": "sessionEndInputSaved", sourceEventIds: Array, inputItemId: string, input: string, }); diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeVoiceTranscript.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeVoiceTranscript.ts new file mode 100644 index 0000000000..20009e962f --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeVoiceTranscript.ts @@ -0,0 +1,6 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ThreadRealtimeTranscriptRole } from "./ThreadRealtimeTranscriptRole"; + +export type ThreadRealtimeVoiceTranscript = { role: ThreadRealtimeTranscriptRole, text: string, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index e3f53d94cb..c85d38ce01 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -585,6 +585,8 @@ export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk"; export type { ThreadRealtimeBemItemPresentation } from "./ThreadRealtimeBemItemPresentation"; export type { ThreadRealtimeClosedNotification } from "./ThreadRealtimeClosedNotification"; export type { ThreadRealtimeErrorNotification } from "./ThreadRealtimeErrorNotification"; +export type { ThreadRealtimeFrameChannels } from "./ThreadRealtimeFrameChannels"; +export type { ThreadRealtimeHandoffMode } from "./ThreadRealtimeHandoffMode"; export type { ThreadRealtimeInitialItem } from "./ThreadRealtimeInitialItem"; export type { ThreadRealtimeItem } from "./ThreadRealtimeItem"; export type { ThreadRealtimeItemAddedNotification } from "./ThreadRealtimeItemAddedNotification"; @@ -599,6 +601,7 @@ export type { ThreadRealtimeStartedNotification } from "./ThreadRealtimeStartedN export type { ThreadRealtimeTranscriptDeltaNotification } from "./ThreadRealtimeTranscriptDeltaNotification"; export type { ThreadRealtimeTranscriptDoneNotification } from "./ThreadRealtimeTranscriptDoneNotification"; export type { ThreadRealtimeTranscriptRole } from "./ThreadRealtimeTranscriptRole"; +export type { ThreadRealtimeVoiceTranscript } from "./ThreadRealtimeVoiceTranscript"; export type { ThreadResumeInitialTurnsPageParams } from "./ThreadResumeInitialTurnsPageParams"; export type { ThreadResumeParams } from "./ThreadResumeParams"; export type { ThreadResumeResponse } from "./ThreadResumeResponse"; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs index 31a4b10e5e..987681d90d 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs @@ -8,6 +8,7 @@ use codex_protocol::protocol::RealtimeOutputModality; use codex_protocol::protocol::RealtimeVoice; use codex_protocol::protocol::RealtimeVoicesList; use codex_protocol::realtime::BemItemPresentation as CoreBemItemPresentation; +use codex_protocol::realtime::RealtimeHandoffMode as CoreRealtimeHandoffMode; use codex_protocol::realtime::RealtimeItem as CoreRealtimeItem; use codex_protocol::realtime::RealtimeItemContent as CoreRealtimeItemContent; use codex_protocol::realtime::RealtimeSessionOutcome as CoreRealtimeSessionOutcome; @@ -24,6 +25,10 @@ use std::collections::BTreeMap; pub struct ThreadRealtimeItem { pub id: String, pub realtime_session_id: String, + /// Canonical observation time in Unix milliseconds for eligible voice calls. + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(type = "number", optional)] + pub observed_at_ms: Option, #[serde(flatten)] pub content: ThreadRealtimeItemContent, } @@ -38,6 +43,9 @@ pub struct ThreadRealtimeItem { #[ts(tag = "type", rename_all = "camelCase", export_to = "v2/")] pub enum ThreadRealtimeItemContent { RealtimeSessionStarted, + Frame { + channels: ThreadRealtimeFrameChannels, + }, TranscriptSegment { role: ThreadRealtimeTranscriptRole, text: String, @@ -50,6 +58,27 @@ pub enum ThreadRealtimeItemContent { RealtimeSessionClosed { outcome: ThreadRealtimeSessionOutcome, }, + TurnInputAccepted { + handoff_id: String, + turn_id: String, + mode: ThreadRealtimeHandoffMode, + source_event_ids: Vec, + input_item_id: String, + input: String, + }, + SessionEndInputSaved { + source_event_ids: Vec, + input_item_id: String, + input: String, + }, +} + +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase", export_to = "v2/")] +pub enum ThreadRealtimeHandoffMode { + Start, + Steer, } /// EXPERIMENTAL - how an existing agent item appears in a realtime conversation. @@ -82,11 +111,52 @@ pub enum ThreadRealtimeSessionOutcome { Failed, } +/// A frame holds one or more payloads, at most one per channel. They share one ID, +/// observation time, and seal. V1 requires its only channel, `voice_transcript`. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeFrameChannels { + pub voice_transcript: ThreadRealtimeVoiceTranscript, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[ts(export_to = "v2/")] +pub struct ThreadRealtimeVoiceTranscript { + pub role: ThreadRealtimeTranscriptRole, + pub text: String, +} + +impl From for ThreadRealtimeTranscriptRole { + fn from(role: CoreRealtimeTranscriptRole) -> Self { + match role { + CoreRealtimeTranscriptRole::User => Self::User, + CoreRealtimeTranscriptRole::Assistant => Self::Assistant, + } + } +} + +// The existing live and mixed timeline APIs present speech as transcriptSegment. impl From for ThreadRealtimeItem { fn from(item: CoreRealtimeItem) -> Self { + let mut item = Self::from_persisted(item); + if let ThreadRealtimeItemContent::Frame { channels } = item.content { + let transcript = channels.voice_transcript; + item.content = ThreadRealtimeItemContent::TranscriptSegment { + role: transcript.role, + text: transcript.text, + }; + } + item + } +} + +impl ThreadRealtimeItem { + /// Preserve the canonical shape of a sealed record on the independent read API. + pub fn from_persisted(item: CoreRealtimeItem) -> Self { let CoreRealtimeItem { id, realtime_session_id, + observed_at_ms, content, } = item; let content = match content { @@ -95,15 +165,21 @@ impl From for ThreadRealtimeItem { } CoreRealtimeItemContent::TranscriptSegment { role, text } => { ThreadRealtimeItemContent::TranscriptSegment { - role: match role { - CoreRealtimeTranscriptRole::User => ThreadRealtimeTranscriptRole::User, - CoreRealtimeTranscriptRole::Assistant => { - ThreadRealtimeTranscriptRole::Assistant - } - }, + role: role.into(), text, } } + CoreRealtimeItemContent::Frame { channels } => { + let transcript = channels.voice_transcript; + ThreadRealtimeItemContent::Frame { + channels: ThreadRealtimeFrameChannels { + voice_transcript: ThreadRealtimeVoiceTranscript { + role: transcript.role.into(), + text: transcript.text, + }, + }, + } + } CoreRealtimeItemContent::BemItemPromoted { turn_id, item_id, @@ -131,10 +207,38 @@ impl From for ThreadRealtimeItem { }, } } + CoreRealtimeItemContent::TurnInputAccepted { + handoff_id, + turn_id, + mode, + source_event_ids, + input_item_id, + input, + } => ThreadRealtimeItemContent::TurnInputAccepted { + handoff_id, + turn_id, + mode: match mode { + CoreRealtimeHandoffMode::Start => ThreadRealtimeHandoffMode::Start, + CoreRealtimeHandoffMode::Steer => ThreadRealtimeHandoffMode::Steer, + }, + source_event_ids, + input_item_id, + input, + }, + CoreRealtimeItemContent::SessionEndInputSaved { + source_event_ids, + input_item_id, + input, + } => ThreadRealtimeItemContent::SessionEndInputSaved { + source_event_ids, + input_item_id, + input, + }, }; Self { id, realtime_session_id, + observed_at_ms, content, } } diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index ef422c214f..4493166398 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -5660,3 +5660,13 @@ fn feedback_paths_preserve_native_wire_spelling_before_host_validation() { ); } } + +#[test] +fn an_undated_live_voice_item_preserves_its_wire_shape() { + let wire = serde_json::json!({ + "id": "speech", "realtimeSessionId": "voice", "type": "transcriptSegment", + "role": "assistant", "text": "hello", + }); + let item: ThreadRealtimeItem = serde_json::from_value(wire.clone()).unwrap(); + assert_eq!(serde_json::to_value(item).unwrap(), wire); +} diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index dd99b3cd41..65d3855e72 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -466,6 +466,12 @@ pub(crate) async fn apply_bespoke_event_handling( .await; } EventMsg::RealtimeConversationRealtime(event) => match event.payload { + RealtimeEvent::HistoryItemStarted(item) | RealtimeEvent::HistoryItemCompleted(item) + if matches!( + &item.content, + codex_protocol::realtime::RealtimeItemContent::TurnInputAccepted { .. } + | codex_protocol::realtime::RealtimeItemContent::SessionEndInputSaved { .. } + ) => {} RealtimeEvent::HistoryItemStarted(item) => { outgoing .send_server_notification(ServerNotification::ThreadRealtimeItemStarted( diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 3d5cb6dfd8..b84d6ee7ef 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -755,8 +755,11 @@ async fn realtime_conversation_streams_timeline_items() -> Result<()> { .await?; assert_eq!(session_started.item, session_completed.item); assert_eq!( - Uuid::parse_str(&session_started.item.id)?.get_version_num(), - 7 + session_started.item.id, + format!( + "realtime-session-start:{}", + session_started.item.realtime_session_id + ) ); assert!(matches!( session_completed.item.content, @@ -1427,7 +1430,17 @@ async fn realtime_timeline_splits_accepted_steering_and_persists_promoted_artifa ); for entry in &page.data { if let ThreadTimelineEntry::Realtime { item, .. } = entry { - assert_eq!(Uuid::parse_str(&item.id)?.get_version_num(), 7); + if matches!( + item.content, + ThreadRealtimeItemContent::RealtimeSessionStarted + ) { + assert_eq!( + item.id, + format!("realtime-session-start:{}", item.realtime_session_id) + ); + } else { + assert_eq!(Uuid::parse_str(&item.id)?.get_version_num(), 7); + } } } @@ -2400,12 +2413,19 @@ async fn webrtc_v1_start_posts_offer_returns_sdp_and_joins_sideband() -> Result< // Phase 2: start through app-server and assert the app receives both the started notification // and the answer SDP. let started = harness.start_webrtc_realtime("v=offer\r\n").await?; + let call_request = harness.call_capture.single_request(); + let session_id = call_request + .headers + .get("x-session-id") + .context("call session ID")? + .to_str()?; + assert_ne!(session_id, harness.thread_id); assert_eq!( started, StartedWebrtcRealtime { started: ThreadRealtimeStartedNotification { thread_id: harness.thread_id.clone(), - realtime_session_id: Some(harness.thread_id.clone()), + realtime_session_id: Some(session_id.to_string()), version: RealtimeConversationVersion::V1, }, sdp: ThreadRealtimeSdpNotification { @@ -2531,7 +2551,7 @@ async fn existing_call_attaches_without_reinitializing_the_client_session( started, ThreadRealtimeStartedNotification { thread_id: harness.thread_id.clone(), - realtime_session_id: realtime_session_id.map(str::to_string), + realtime_session_id: Some(realtime_session_id.unwrap_or("rtc_existing").to_string()), version: expected_version, } ); @@ -2663,12 +2683,19 @@ async fn webrtc_v3_start_posts_live_session_and_joins_without_session_update() - RealtimeConversationVersion::V3, ) .await?; + let call_request = harness.call_capture.single_request(); + let session_id = call_request + .headers + .get("x-session-id") + .context("call session ID")? + .to_str()?; + assert_ne!(session_id, harness.thread_id); assert_eq!( started, StartedWebrtcRealtime { started: ThreadRealtimeStartedNotification { thread_id: harness.thread_id.clone(), - realtime_session_id: Some(harness.thread_id.clone()), + realtime_session_id: Some(session_id.to_string()), version: RealtimeConversationVersion::V3, }, sdp: ThreadRealtimeSdpNotification { @@ -4322,12 +4349,19 @@ async fn realtime_starts_on_forked_and_resumed_threads( } let started = harness.start_webrtc_realtime("v=offer\r\n").await?; + let call_request = harness.call_capture.single_request(); + let session_id = call_request + .headers + .get("x-session-id") + .context("call session ID")? + .to_str()?; + assert_ne!(session_id, harness.thread_id); assert_eq!( started, StartedWebrtcRealtime { started: ThreadRealtimeStartedNotification { thread_id: harness.thread_id.clone(), - realtime_session_id: Some(harness.thread_id.clone()), + realtime_session_id: Some(session_id.to_string()), version: RealtimeConversationVersion::V1, }, sdp: ThreadRealtimeSdpNotification { diff --git a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs index 87286c055c..2f09a238e2 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs @@ -95,6 +95,7 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() -> RolloutItem::RealtimeItem(RealtimeItem { id: "voice-1:started".to_string(), realtime_session_id: "voice-1".to_string(), + observed_at_ms: Some(1_000), content: RealtimeItemContent::RealtimeSessionStarted, }), RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { @@ -124,6 +125,7 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() -> RolloutItem::RealtimeItem(RealtimeItem { id: "voice-1:transcript-1".to_string(), realtime_session_id: "voice-1".to_string(), + observed_at_ms: None, content: RealtimeItemContent::TranscriptSegment { role: RealtimeTranscriptRole::Assistant, text: "Checking staging".to_string(), diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 8d4e5c3663..32978b4f56 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -2719,6 +2719,7 @@ async fn spawn_agent_fork_sanitizes_inherited_compaction_metadata() { next_item_ordinal: None, rejected_audio_file_ids: Default::default(), dynamic_tools: None, + observation_clock_floor_ms: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("parent-turn".to_string()), turn_attribution: Some(codex_history::TurnAttribution { @@ -3885,6 +3886,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc next_item_ordinal: None, rejected_audio_file_ids: Default::default(), dynamic_tools: None, + observation_clock_floor_ms: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, turn_attribution: None, @@ -4022,6 +4024,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc next_item_ordinal: None, rejected_audio_file_ids: Default::default(), dynamic_tools: None, + observation_clock_floor_ms: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, turn_attribution: None, diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 4ecc5ee848..9e3d4c4f20 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -233,6 +233,7 @@ pub use state_db_bridge::init_state_db; mod thread_rollout_truncation; pub use thread_rollout_truncation::truncate_rollout_after_turn_id; pub use thread_rollout_truncation::truncate_rollout_before_turn_id; +mod observation_clock; mod tools; pub(crate) mod turn_diff_tracker; mod turn_metadata; diff --git a/codex-rs/core/src/observation_clock.rs b/codex-rs/core/src/observation_clock.rs new file mode 100644 index 0000000000..0f89195256 --- /dev/null +++ b/codex-rs/core/src/observation_clock.rs @@ -0,0 +1,48 @@ +//! Assigns first-observed timestamps to realtime records for a single thread owner. + +use codex_history::RolloutItem; + +#[derive(Debug, Default)] +pub(crate) struct ObservationClock { + last_assigned_ms: Option, +} + +impl ObservationClock { + pub(crate) fn restore(&mut self, items: &[RolloutItem]) { + let floor = items + .iter() + .filter_map(|item| match item { + RolloutItem::RealtimeItem(item) => item.observed_at_ms, + RolloutItem::Compacted(compacted) => compacted + .resume_metadata + .as_ref() + .and_then(|metadata| metadata.observation_clock_floor_ms), + _ => None, + }) + .max(); + if let Some(floor) = floor { + self.observe(floor); + } + } + + pub(crate) fn now_ms(&mut self) -> i64 { + self.observe(crate::turn_timing::now_unix_timestamp_ms()) + } + + pub(crate) fn floor_ms(&self) -> Option { + self.last_assigned_ms + } + + pub(crate) fn observe(&mut self, wall_clock_ms: i64) -> i64 { + let time = self + .last_assigned_ms + .unwrap_or(wall_clock_ms) + .max(wall_clock_ms); + self.last_assigned_ms = Some(time); + time + } +} + +#[cfg(test)] +#[path = "observation_clock_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/observation_clock_tests.rs b/codex-rs/core/src/observation_clock_tests.rs new file mode 100644 index 0000000000..60bc66e9f3 --- /dev/null +++ b/codex-rs/core/src/observation_clock_tests.rs @@ -0,0 +1,73 @@ +#![cfg(test)] + +//! Checks clock rollback and restoration from durable history checkpoints. + +use super::ObservationClock; +use codex_history::RolloutItem; +use codex_protocol::realtime::RealtimeItem; +use codex_protocol::realtime::RealtimeItemContent; +use pretty_assertions::assert_eq; + +#[test] +fn observation_time_survives_clock_rollback_and_owner_restart() { + let mut clock = ObservationClock::default(); + assert_eq!( + [ + clock.observe(/*wall_clock_ms*/ 100), + clock.observe(/*wall_clock_ms*/ 90), + clock.observe(/*wall_clock_ms*/ 110) + ], + [100, 100, 110] + ); + + let mut clock = ObservationClock::default(); + clock.restore(&[ + RolloutItem::RealtimeItem(RealtimeItem { + id: "sealed".into(), + realtime_session_id: "voice".into(), + observed_at_ms: Some(110), + content: RealtimeItemContent::RealtimeSessionStarted, + }), + RolloutItem::RealtimeItem(RealtimeItem { + id: "legacy".into(), + realtime_session_id: "legacy-voice".into(), + observed_at_ms: None, + content: RealtimeItemContent::RealtimeSessionStarted, + }), + ]); + assert_eq!( + [ + clock.observe(/*wall_clock_ms*/ 80), + clock.observe(/*wall_clock_ms*/ 120) + ], + [110, 120] + ); +} + +#[test] +fn compaction_retains_the_floor_when_older_events_are_omitted() { + let compacted = serde_json::from_value::(serde_json::json!({ + "type": "compacted", + "payload": { + "message": "checkpoint", + "replacement_history": [], + "window_number": 1, + "resume_metadata": { + "observation_clock_floor_ms": 150, + "multi_agent_version": null, + "last_started_turn_id": null, + "previous_turn_settings": null + } + } + })) + .expect("read persisted checkpoint"); + let mut clock = ObservationClock::default(); + clock.restore(&[compacted]); + assert_eq!( + [ + clock.observe(/*wall_clock_ms*/ 100), + clock.observe(/*wall_clock_ms*/ 200) + ], + [150, 200] + ); +} diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 0c72872a58..7c2086180d 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -1575,11 +1575,8 @@ async fn prepare_realtime_start( ConfiguredRealtimeVoice::Use } }; - let mut session_config = + let session_config = build_realtime_session_config(sess, ¶ms, version, configured_voice).await?; - if matches!(&transport, ConversationStartTransport::ExistingCall { .. }) { - session_config.session_id = params.realtime_session_id.clone(); - } let requested_realtime_session_id = session_config.session_id.clone(); let event_parser = session_config.event_parser; let originator = sess.originator().await; @@ -1804,12 +1801,16 @@ pub(crate) async fn build_realtime_session_config( initial_items: params.initial_items.clone(), delegation_ack_filler: params.delegation_ack_filler, model, - session_id: Some( - params - .realtime_session_id - .clone() - .unwrap_or_else(|| sess.thread_id.to_string()), - ), + session_id: Some(params.realtime_session_id.clone().unwrap_or_else(|| { + match ¶ms.transport { + Some(ConversationStartTransport::ExistingCall { call_id, .. }) => call_id.clone(), + Some( + ConversationStartTransport::Webrtc { .. } + | ConversationStartTransport::Websocket, + ) + | None => uuid::Uuid::now_v7().to_string(), + } + })), event_parser, session_mode, output_modality: params.output_modality, @@ -1893,6 +1894,9 @@ async fn handle_start_inner( session_config, transport, } = prepared_start; + sess.initialize_realtime_history() + .or_cancel(&attachment_control.token) + .await??; info!("starting realtime conversation"); let (sdp, existing_call_id) = match transport { ConversationStartTransport::Websocket => (None, None), diff --git a/codex-rs/core/src/realtime_history.rs b/codex-rs/core/src/realtime_history.rs index 8ba0a71f55..d85ae8702c 100644 --- a/codex-rs/core/src/realtime_history.rs +++ b/codex-rs/core/src/realtime_history.rs @@ -4,16 +4,19 @@ mod presentation; +use crate::observation_clock::ObservationClock; use codex_protocol::items::TurnItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RealtimeEvent; use codex_protocol::protocol::RealtimeTranscriptDelta; use codex_protocol::protocol::RealtimeTranscriptDone; use codex_protocol::realtime::BemItemPresentation; +use codex_protocol::realtime::RealtimeFrameChannels; use codex_protocol::realtime::RealtimeItem; use codex_protocol::realtime::RealtimeItemContent; use codex_protocol::realtime::RealtimeSessionOutcome; use codex_protocol::realtime::RealtimeTranscriptRole; +use codex_protocol::realtime::RealtimeVoiceTranscript; use codex_protocol::user_input::UserInput; use std::collections::HashMap; use std::collections::HashSet; @@ -26,6 +29,7 @@ struct ActiveSegment { id: String, role: RealtimeTranscriptRole, text: String, + observed_at_ms: Option, } #[derive(Debug, Default)] @@ -71,6 +75,7 @@ pub(crate) struct RealtimeTranscriptStream { #[derive(Debug, Default)] pub(crate) struct RealtimeEventEffects { + pub(crate) seal_frames: bool, pub(crate) order: RealtimeEventOrder, pub(crate) items: Vec, pub(crate) transcript_stream: Option, @@ -92,6 +97,9 @@ enum Continuation { /// Retains only live session state; durable history is served by the rollout index. #[derive(Default)] pub(crate) struct RealtimeHistoryState { + pub(crate) observation_clock: ObservationClock, + /// Resolved once at the first call. Unresolved and unenrolled threads write legacy history. + pub(crate) frame_enrollment: Option, active_session_id: Option, active_segments: ActiveTranscriptSegments, streaming_agent_message: Option, @@ -103,6 +111,10 @@ pub(crate) struct RealtimeHistoryState { } impl RealtimeHistoryState { + pub(crate) fn frames_enabled(&self) -> bool { + self.frame_enrollment == Some(true) + } + fn should_seal_user_input(&self, input: &[UserInput]) -> bool { self.active_session_id.is_some() && [&self.active_segments.user, &self.active_segments.assistant] @@ -188,8 +200,16 @@ impl RealtimeHistoryState { self.active_session_id = Some(session_id.clone()); self.failed = false; items.push(RealtimeItem { - id: Uuid::now_v7().to_string(), + // The call ID lets a reader seek directly to its opening position. + id: if self.frames_enabled() { + format!("realtime-session-start:{session_id}") + } else { + Uuid::now_v7().to_string() + }, realtime_session_id: session_id.clone(), + observed_at_ms: self + .frames_enabled() + .then(|| self.observation_clock.now_ms()), content: RealtimeItemContent::RealtimeSessionStarted, }); } @@ -284,6 +304,9 @@ impl RealtimeHistoryState { items.push(RealtimeItem { id: Uuid::now_v7().to_string(), realtime_session_id: session_id, + observed_at_ms: self + .frames_enabled() + .then(|| self.observation_clock.now_ms()), content: RealtimeItemContent::RealtimeSessionClosed { outcome: if self.failed { RealtimeSessionOutcome::Failed @@ -298,6 +321,7 @@ impl RealtimeHistoryState { _ => {} } RealtimeEventEffects { + seal_frames: self.frames_enabled(), order, items, transcript_stream, @@ -329,6 +353,9 @@ impl RealtimeHistoryState { items.push(RealtimeItem { id: Uuid::now_v7().to_string(), realtime_session_id, + observed_at_ms: self + .frames_enabled() + .then(|| self.observation_clock.now_ms()), content: RealtimeItemContent::BemItemPromoted { turn_id: turn_id.to_string(), item_id: item_id.to_string(), @@ -344,6 +371,7 @@ impl RealtimeHistoryState { ) -> Option { let session_id = self.active_session_id.clone()?; self.active_segments.first_active_role.get_or_insert(role); + let frames_enabled = self.frames_enabled(); let segment = self .active_segments .slot_mut(role) @@ -352,14 +380,18 @@ impl RealtimeHistoryState { id: Uuid::now_v7().to_string(), role, text: String::new(), + observed_at_ms: None, }); + let observed_at_ms = frames_enabled.then(|| { + *segment + .observed_at_ms + .get_or_insert_with(|| self.observation_clock.now_ms()) + }); let started_item = segment.text.is_empty().then(|| RealtimeItem { id: segment.id.clone(), realtime_session_id: segment.session_id.clone(), - content: RealtimeItemContent::TranscriptSegment { - role: segment.role, - text: String::new(), - }, + observed_at_ms, + content: speech(frames_enabled, segment.role, String::new()), }); segment.text.push_str(&delta.delta); Some(RealtimeTranscriptStream { @@ -425,10 +457,8 @@ impl RealtimeHistoryState { items.push(RealtimeItem { id: segment.id, realtime_session_id: segment.session_id.clone(), - content: RealtimeItemContent::TranscriptSegment { - role: segment.role, - text: segment.text, - }, + observed_at_ms: segment.observed_at_ms, + content: speech(self.frames_enabled(), segment.role, segment.text), }); } if matches!(continuation, Continuation::Continue) { @@ -440,11 +470,24 @@ impl RealtimeHistoryState { id: Uuid::now_v7().to_string(), role: segment.role, text: String::new(), + observed_at_ms: None, }); } } } +fn speech(frames_enabled: bool, role: RealtimeTranscriptRole, text: String) -> RealtimeItemContent { + if frames_enabled { + RealtimeItemContent::Frame { + channels: RealtimeFrameChannels { + voice_transcript: RealtimeVoiceTranscript { role, text }, + }, + } + } else { + RealtimeItemContent::TranscriptSegment { role, text } + } +} + #[cfg(test)] #[path = "realtime_history_tests.rs"] mod tests; diff --git a/codex-rs/core/src/realtime_history_tests.rs b/codex-rs/core/src/realtime_history_tests.rs index c11a087416..c881601ab5 100644 --- a/codex-rs/core/src/realtime_history_tests.rs +++ b/codex-rs/core/src/realtime_history_tests.rs @@ -345,6 +345,7 @@ fn streams_stable_segment_items_and_ignores_upstream_transcript_revisions() { done.items, vec![RealtimeItem { id: started.id, + observed_at_ms: started.observed_at_ms, realtime_session_id: "voice-1".to_string(), content: RealtimeItemContent::TranscriptSegment { role: RealtimeTranscriptRole::Assistant, @@ -372,6 +373,7 @@ fn emits_the_complete_item_lifecycle_when_only_a_final_transcript_arrives() { effects.items, vec![RealtimeItem { id: started.id, + observed_at_ms: started.observed_at_ms, realtime_session_id: "voice-1".to_string(), content: RealtimeItemContent::TranscriptSegment { role: RealtimeTranscriptRole::Assistant, @@ -566,6 +568,7 @@ fn promotes_successful_dynamic_tools_and_splits_active_transcripts() { items[0], RealtimeItem { id: first.item_id.clone(), + observed_at_ms: first.started_item.as_ref().unwrap().observed_at_ms, realtime_session_id: "voice-1".to_string(), content: RealtimeItemContent::TranscriptSegment { role: RealtimeTranscriptRole::Assistant, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 7f38ba30cb..916fe9b2e5 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -607,6 +607,7 @@ impl Session { thread_id: self.thread_id(), tools: state.persisted_dynamic_tools.to_vec(), })), + observation_clock_floor_ms: None, multi_agent_version: self.multi_agent_version(), last_started_turn_id: state.last_started_turn_id.clone(), turn_attribution: state.turn_attribution.clone(), @@ -1970,6 +1971,13 @@ impl Session { turn_context: &Arc, rollout_items: &[RolloutItem], ) -> Option { + if let Some(history) = &self.realtime_history { + history + .lock() + .await + .observation_clock + .restore(rollout_items); + } let rollout_reconstruction::RolloutReconstruction { mut history, rejected_audio_file_ids, @@ -2953,8 +2961,8 @@ impl Session { } pub(crate) async fn send_event_raw(&self, event: Event) { - self.send_event_raw_with_persistence(event, /*persist*/ true) - .await; + // Keep event persistence out of callers' stack frames. + Box::pin(self.send_event_raw_with_persistence(event, /*persist*/ true)).await; } /// Delivers an event without creating a local rollout for a thread that has not materialized. @@ -4879,6 +4887,10 @@ impl Session { // Wait for accepted updates to finish persisting, then keep later updates from // overtaking the current settings snapshot while its checkpoint is written. let _settings_guard = thread_settings::acquire_persistence_lock(self).await; + let observation_clock_floor_ms = match &self.realtime_history { + Some(history) => history.lock().await.observation_clock.floor_ms(), + None => None, + }; let (compacted_item, world_state_baseline, started_item) = { let mut state = self.state.lock().await; let complete_replacement = @@ -4958,7 +4970,10 @@ impl Session { state .history .set_world_state_baseline(world_state_baseline.clone()); - let resume_metadata = self.compaction_resume_metadata(&state); + let resume_metadata = CompactionResumeMetadata { + observation_clock_floor_ms, + ..self.compaction_resume_metadata(&state) + }; let compacted_item = CompactedItem { message: metadata.message, replacement_history: Some(replacement_history), diff --git a/codex-rs/core/src/session/realtime_history.rs b/codex-rs/core/src/session/realtime_history.rs index 5b1acf4873..33107be221 100644 --- a/codex-rs/core/src/session/realtime_history.rs +++ b/codex-rs/core/src/session/realtime_history.rs @@ -8,6 +8,25 @@ use codex_protocol::protocol::RealtimeEvent; use codex_protocol::realtime::RealtimeItemContent; impl Session { + /// Start and attach operations run sequentially in the submission loop. + /// Cache enrollment after a successful lookup so a failed start can retry. + pub(crate) async fn initialize_realtime_history(&self) -> std::io::Result { + let Some(history) = &self.realtime_history else { + return Ok(false); + }; + if let Some(enabled) = history.lock().await.frame_enrollment { + return Ok(enabled); + } + let enabled = self + .services + .thread_store + .realtime_frames_enabled(self.thread_id()) + .await + .map_err(std::io::Error::other)?; + history.lock().await.frame_enrollment = Some(enabled); + Ok(enabled) + } + pub(super) async fn send_realtime_history_effects( &self, submission_id: &str, @@ -33,18 +52,24 @@ impl Session { if effects.items.is_empty() { return Ok(()); } - self.live_thread_for_persistence("append realtime history")? - .append_items( - &effects - .items - .iter() - .cloned() - .map(RolloutItem::RealtimeItem) - .collect::>(), - ) - .await?; - for item in effects.items { - if !matches!(&item.content, RealtimeItemContent::TranscriptSegment { .. }) { + let live_thread = self.live_thread_for_persistence("append realtime history")?; + let items = effects.items; + let records = items + .iter() + .cloned() + .map(RolloutItem::RealtimeItem) + .collect::>(); + live_thread.append_items(&records).await?; + // A completed notification acknowledges durable history, including on hosts + // whose append operation only admits work to an asynchronous writer. + if effects.seal_frames { + live_thread.flush().await?; + } + for item in items { + if !matches!( + &item.content, + RealtimeItemContent::Frame { .. } | RealtimeItemContent::TranscriptSegment { .. } + ) { self.deliver_event_raw(event(RealtimeEvent::HistoryItemStarted(item.clone()))) .await; } diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 96181f9f9a..fc36c4ccfe 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -1645,6 +1645,7 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions( next_item_ordinal: None, rejected_audio_file_ids: Default::default(), dynamic_tools: None, + observation_clock_floor_ms: None, multi_agent_version: None, last_started_turn_id: Some(format!("wake-{window_number}")), turn_attribution: None, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 71dfc8f711..a6e902ed03 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1283,7 +1283,8 @@ impl Session { error!("failed to initialize thread persistence: {e:#}"); e })?; - let session_result: anyhow::Result> = async { + // Keep the large initialization future out of callers' stack frames. + let session_result: anyhow::Result> = Box::pin(async { if let InitialHistory::Resumed(resumed) = &mut initial_history && let Some(history) = resume_context { @@ -1994,7 +1995,7 @@ impl Session { realtime_history: (session_configuration.history_mode == ThreadHistoryMode::Paginated && services.live_thread.is_some()) - .then(|| Mutex::new(Default::default())), + .then(|| Mutex::new(crate::realtime_history::RealtimeHistoryState::default())), active_turn: Mutex::new(None), async_hook_results, input_queue: InputQueue::with_controller( @@ -2143,7 +2144,7 @@ impl Session { state.queue_pending_session_start_source(session_start_source); } Ok(sess) - } + }) .await; match session_result { Ok(sess) => { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 87a00abc5c..80144a67db 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6177,6 +6177,7 @@ async fn compaction_persists_resume_metadata_and_companion_records() { thread_id: session.thread_id(), tools: Vec::new(), })), + observation_clock_floor_ms: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some(turn_context.sub_id.clone()), turn_attribution: Some(codex_history::TurnAttribution { diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index 6b0d594b97..873f0d8b49 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -149,14 +149,15 @@ impl RegularTask { turn.id = %ctx.sub_id, ); loop { - let last_agent_message = run_turn( + // Leave stack room for nested event persistence while the turn is polled. + let last_agent_message = Box::pin(run_turn( Arc::clone(&sess), Arc::clone(&ctx), next_input, &mut mcp_startup_requirements, prewarmed_client_session.take(), cancellation_token.child_token(), - ) + )) .instrument(run_turn_span.clone()) .await?; // Terminal errors are already reported. Let task completion preserve pending diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 3a8442c935..ed5b22cfc9 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -476,15 +476,22 @@ async fn conversation_records_history_without_an_event_observer( ) -> Result<()> { skip_if_no_network!(Ok(())); let api_server = start_mock_server().await; - let mut events = vec![ - json!({ "type": "session.updated", "session": { "id": "voice-1" } }), - json!({ "type": "response.output_text.delta", "delta": "assistant first" }), - json!({ "type": "conversation.item.input_audio_transcription.delta", "delta": "user second" }), - ]; - if outcome == ThreadRealtimeSessionOutcome::Failed { - events.push(json!({ "type": "error", "error": { "message": "fixture failure" } })); - } - let realtime_server = start_websocket_server(vec![vec![events.clone()], vec![events]]).await; + let session_ids = ["voice-1", "voice-2"]; + let calls = session_ids + .iter() + .map(|session_id| { + let mut events = vec![ + json!({ "type": "session.updated", "session": { "id": session_id } }), + json!({ "type": "response.output_text.delta", "delta": "assistant first" }), + json!({ "type": "conversation.item.input_audio_transcription.delta", "delta": "user second" }), + ]; + if outcome == ThreadRealtimeSessionOutcome::Failed { + events.push(json!({ "type": "error", "error": { "message": "fixture failure" } })); + } + vec![events] + }) + .collect(); + let realtime_server = start_websocket_server(calls).await; let mut builder = test_codex() .with_history_mode(ThreadHistoryMode::Paginated) .with_config({ @@ -497,7 +504,8 @@ async fn conversation_records_history_without_an_event_observer( let test = builder.build_with_auto_env(&api_server).await?; test.codex.ensure_rollout_materialized().await; let mut expected = Vec::new(); - for session_count in 1..=2 { + for previous_calls in 0..session_ids.len() { + let first_item = expected.len(); test.codex .submit(Op::RealtimeConversationStart(ConversationStartParams { client_managed_handoffs: false, @@ -516,7 +524,7 @@ async fn conversation_records_history_without_an_event_observer( realtime_start_instructions: None, realtime_end_instructions: None, prompt: Some(Some("fixture".to_string())), - realtime_session_id: Some("voice-1".to_string()), + realtime_session_id: None, voice_session_id: None, transport: None, version: None, @@ -553,7 +561,7 @@ async fn conversation_records_history_without_an_event_observer( ) }) .count() - == session_count + == previous_calls + 1 { break Ok::<_, anyhow::Error>(items); } @@ -586,10 +594,17 @@ async fn conversation_records_history_without_an_event_observer( .map(|item| item.id.as_str()) .collect::>(); assert_eq!(ids.len(), items.len()); + let session_id = &items[first_item].realtime_session_id; + Uuid::parse_str(session_id)?; + assert_ne!(session_id, &test.session_configured.thread_id.to_string()); + assert_eq!( + items[first_item].id, + format!("realtime-session-start:{session_id}") + ); assert!( - items + items[first_item..] .iter() - .all(|item| item.realtime_session_id == "voice-1") + .all(|item| &item.realtime_session_id == session_id) ); } test.codex.submit(Op::Shutdown).await?; @@ -739,22 +754,20 @@ async fn conversation_websocket_transports_send_codex_headers_without_creating_a }) .await .expect("realtime websocket connection failed"); - let (expected_uri, expected_realtime_session_id) = match transport { - ConversationStartTransport::Websocket => ( - "/v1/live?model=gpt-live-1.1-codex", - Some(conversation.thread_id.to_string()), - ), - ConversationStartTransport::ExistingCall { .. } => ("/v1/live/rtc_existing", None), + assert_eq!(started.version, RealtimeConversationVersion::V3); + let session_id = started.realtime_session_id.context("started session ID")?; + let expected_uri = match transport { + ConversationStartTransport::Websocket => { + Uuid::parse_str(&session_id)?; + assert_ne!(session_id, conversation.thread_id.to_string()); + "/v1/live?model=gpt-live-1.1-codex" + } + ConversationStartTransport::ExistingCall { call_id, .. } => { + assert_eq!(session_id, call_id); + "/v1/live/rtc_existing" + } ConversationStartTransport::Webrtc { .. } => unreachable!("websocket transports only"), }; - assert_eq!( - (started.version, started.realtime_session_id), - ( - RealtimeConversationVersion::V3, - expected_realtime_session_id.clone() - ) - ); - let handshake = realtime_server.single_handshake(); assert_eq!(handshake.uri(), expected_uri); assert_eq!( @@ -762,8 +775,8 @@ async fn conversation_websocket_transports_send_codex_headers_without_creating_a Some("Bearer dummy") ); assert_eq!( - handshake.header("x-session-id"), - expected_realtime_session_id + handshake.header("x-session-id").as_deref(), + Some(session_id.as_str()) ); assert_eq!( ( @@ -3981,17 +3994,27 @@ async fn conversation_user_text_turn_is_not_sent_to_realtime(ephemeral: bool) -> .await; assert_eq!(session_updated, "sess_user_text"); + let mut transcript_started = None; + let mut output_seen = false; wait_for_event(&test.codex, |event| { if let EventMsg::RealtimeConversationRealtime(event) = event { - match event.payload { - RealtimeEvent::HistoryItemStarted(_) - | RealtimeEvent::HistoryTranscriptDelta { .. } + match &event.payload { + RealtimeEvent::HistoryItemStarted(item) => { + assert!(!ephemeral); + if matches!( + item.content, + codex_protocol::realtime::RealtimeItemContent::Frame { .. } + ) { + transcript_started = Some(item.clone()); + } + } + RealtimeEvent::HistoryTranscriptDelta { .. } | RealtimeEvent::HistoryItemCompleted(_) => assert!(!ephemeral), - RealtimeEvent::OutputTranscriptDelta(_) => return true, + RealtimeEvent::OutputTranscriptDelta(_) => output_seen = true, _ => {} } } - false + output_seen && (ephemeral || transcript_started.is_some()) }) .await; @@ -4003,6 +4026,44 @@ async fn conversation_user_text_turn_is_not_sent_to_realtime(ephemeral: bool) -> }])) .await?; + if let Some(started) = transcript_started { + assert!(started.observed_at_ms.is_some()); + let sealed = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::HistoryItemCompleted(item), + }) if item.id == started.id => Some(item.clone()), + _ => None, + }) + .await; + assert_eq!(sealed.observed_at_ms, started.observed_at_ms); + assert!(matches!( + &sealed.content, + codex_protocol::realtime::RealtimeItemContent::Frame { channels } + if channels.voice_transcript.text == "spoken before typed input" + )); + // Completion is the persistence barrier; no additional flush is needed to read it. + let timeline = test + .thread_store + .list_timeline(ListTimelineParams { + thread_id: test.session_configured.thread_id, + cursor: None, + page_size: 100, + }) + .await?; + let persisted = timeline + .items + .iter() + .find_map(|entry| match entry { + ThreadTimelineEntry::Realtime { item, .. } if item.id == sealed.id => Some(item), + _ => None, + }) + .expect("completed realtime item is already readable"); + assert_eq!( + persisted, + &codex_app_server_protocol::ThreadRealtimeItem::from(sealed) + ); + } + let turn_complete = wait_for_event_match(&test.codex, |event| match event { EventMsg::TurnComplete(turn_complete) => Some(turn_complete.clone()), _ => None, @@ -4014,9 +4075,6 @@ async fn conversation_user_text_turn_is_not_sent_to_realtime(ephemeral: bool) -> assert!(model_user_texts.iter().any(|text| text == user_text)); if !ephemeral { - test.thread_store - .flush_thread(test.session_configured.thread_id) - .await?; let timeline = test .thread_store .list_timeline(ListTimelineParams { diff --git a/codex-rs/history/src/compaction_resume_metadata.rs b/codex-rs/history/src/compaction_resume_metadata.rs index 94768545af..b274e00c81 100644 --- a/codex-rs/history/src/compaction_resume_metadata.rs +++ b/codex-rs/history/src/compaction_resume_metadata.rs @@ -27,6 +27,9 @@ pub struct CompactionResumeMetadata { /// Catalog owned by the producing thread; absent in older checkpoints. #[serde(default, skip_serializing_if = "Option::is_none")] pub dynamic_tools: Option>, + /// Thread clock floor retained when older history is omitted from resume context. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub observation_clock_floor_ms: Option, /// Selected runtime, including when the companion turn context is absent. pub multi_agent_version: Option, /// Turn identity used to admit continuations after cold resume. diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index 7e9fadb84d..267a732fb3 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -699,6 +699,7 @@ fn compacted_resume_metadata_presence_round_trips_empty_values() -> Result<()> { next_item_ordinal: None, rejected_audio_file_ids: Default::default(), dynamic_tools: None, + observation_clock_floor_ms: None, multi_agent_version: None, last_started_turn_id: None, turn_attribution: None, @@ -813,6 +814,7 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() -> next_item_ordinal: None, rejected_audio_file_ids: Default::default(), dynamic_tools: None, + observation_clock_floor_ms: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("turn-1".to_string()), turn_attribution: None, diff --git a/codex-rs/protocol/src/realtime.rs b/codex-rs/protocol/src/realtime.rs index 8c35c3efc9..8da13879c2 100644 --- a/codex-rs/protocol/src/realtime.rs +++ b/codex-rs/protocol/src/realtime.rs @@ -8,16 +8,38 @@ use ts_rs::TS; pub struct RealtimeItem { pub id: String, pub realtime_session_id: String, + /// Owner-assigned Unix milliseconds at first observation, retained when sealed. + /// Absent for history recorded before canonical observation times were introduced. + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(type = "number | null", optional)] + pub observed_at_ms: Option, #[serde(flatten)] pub content: RealtimeItemContent, } +/// A frame holds one or more payloads, at most one per channel. They share one ID, +/// observation time, and seal. V1 requires its only channel, `voice_transcript`. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, TS)] +pub struct RealtimeFrameChannels { + pub voice_transcript: RealtimeVoiceTranscript, +} + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, TS)] +pub struct RealtimeVoiceTranscript { + pub role: RealtimeTranscriptRole, + pub text: String, +} + /// The minimum facts needed to interleave realtime speech and agent work. #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, TS)] #[serde(tag = "type", rename_all = "snake_case")] #[ts(tag = "type", rename_all = "snake_case")] pub enum RealtimeItemContent { RealtimeSessionStarted, + Frame { + channels: RealtimeFrameChannels, + }, + /// Original rollout spelling, retained for older histories and unenrolled writers. TranscriptSegment { role: RealtimeTranscriptRole, text: String, @@ -30,6 +52,31 @@ pub enum RealtimeItemContent { RealtimeSessionClosed { outcome: RealtimeSessionOutcome, }, + /// Durable acceptance of the exact bounded input used by an agent turn. + /// The enclosing item ID is the acceptance identity; source IDs never use + /// provisional rollout ordinals. + TurnInputAccepted { + handoff_id: String, + turn_id: String, + mode: RealtimeHandoffMode, + source_event_ids: Vec, + input_item_id: String, + input: String, + }, + /// End-of-call input recorded in model history without starting or steering a turn. + SessionEndInputSaved { + source_event_ids: Vec, + input_item_id: String, + input: String, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, JsonSchema, TS)] +#[serde(rename_all = "snake_case")] +#[ts(rename_all = "snake_case")] +pub enum RealtimeHandoffMode { + Start, + Steer, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, JsonSchema, TS)] @@ -57,3 +104,7 @@ pub enum BemItemPresentation { InlineMarkdown, InlineVisualization { index: u32 }, } + +#[cfg(test)] +#[path = "realtime_tests.rs"] +mod tests; diff --git a/codex-rs/protocol/src/realtime_tests.rs b/codex-rs/protocol/src/realtime_tests.rs new file mode 100644 index 0000000000..74bd40bfcf --- /dev/null +++ b/codex-rs/protocol/src/realtime_tests.rs @@ -0,0 +1,22 @@ +#![cfg(test)] + +//! Validates the sparse channel envelope written by new voice calls. + +use super::RealtimeItem; +use pretty_assertions::assert_eq; + +#[test] +fn frame_roundtrips_and_requires_a_voice_payload() { + let frame = serde_json::json!({ + "id": "speech", + "realtime_session_id": "voice", + "observed_at_ms": 1050, + "type": "frame", + "channels": { "voice_transcript": { "role": "assistant", "text": "hello" } }, + }); + let item = serde_json::from_value::(frame.clone()).unwrap(); + assert_eq!(serde_json::to_value(item).unwrap(), frame); + let mut empty = frame; + empty["channels"] = serde_json::json!({}); + assert!(serde_json::from_value::(empty).is_err()); +} diff --git a/codex-rs/rollout/src/persistence_metrics.rs b/codex-rs/rollout/src/persistence_metrics.rs index 2566567b8b..7cf5047bfe 100644 --- a/codex-rs/rollout/src/persistence_metrics.rs +++ b/codex-rs/rollout/src/persistence_metrics.rs @@ -270,6 +270,9 @@ fn rollout_item_type(item: &RolloutItem) -> String { codex_protocol::realtime::RealtimeItemContent::RealtimeSessionStarted => { "realtime.session_started".to_string() } + codex_protocol::realtime::RealtimeItemContent::Frame { .. } => { + "realtime.frame".to_string() + } codex_protocol::realtime::RealtimeItemContent::TranscriptSegment { .. } => { "realtime.transcript_segment".to_string() } @@ -279,6 +282,12 @@ fn rollout_item_type(item: &RolloutItem) -> String { codex_protocol::realtime::RealtimeItemContent::RealtimeSessionClosed { .. } => { "realtime.session_closed".to_string() } + codex_protocol::realtime::RealtimeItemContent::TurnInputAccepted { .. } => { + "realtime.turn_input_accepted".to_string() + } + codex_protocol::realtime::RealtimeItemContent::SessionEndInputSaved { .. } => { + "realtime.session_end_input_saved".to_string() + } }, RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => { format!("event.item_completed.{}", turn_item_type(&event.item)) diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 7469620457..44fffde8e5 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -1021,6 +1021,9 @@ impl RolloutRecorder { writer: None, deferred_creation: true, pending_items: Vec::new(), + realtime_sync_pending: false, + #[cfg(unix)] + directories_to_sync: Vec::new(), meta: Some(session_meta), cwd: cwd.clone(), rollout_path: path, @@ -1035,6 +1038,9 @@ impl RolloutRecorder { writer: Some(JsonlWriter { file }), deferred_creation: false, pending_items: Vec::new(), + realtime_sync_pending: false, + #[cfg(unix)] + directories_to_sync: rollout_directories_to_sync(&path)?, meta: None, cwd: cwd.clone(), rollout_path: path, @@ -1716,6 +1722,21 @@ fn precompute_new_rollout_path( Ok((path, timestamp)) } +#[cfg(unix)] +fn rollout_directories_to_sync(path: &Path) -> std::io::Result> { + // Creating the file changes its parent. Creating a date directory changes its parent too. + let mut directories = Vec::new(); + if let Some(parent) = path.parent() { + for directory in parent.ancestors() { + directories.push(directory.to_owned()); + if directory.try_exists()? { + break; + } + } + } + Ok(directories) +} + fn open_log_file(path: &Path) -> std::io::Result { let refresh_modified_time = !compression::plain_rollout_path(path).try_exists()?; let path = compression::materialize_rollout_for_append_blocking(path)?; @@ -1730,7 +1751,7 @@ fn open_log_file(path: &Path) -> std::io::Result { .read(true) .append(true) .create(true) - .open(path)?; + .open(&path)?; if refresh_modified_time { refresh_rollout_modified_time(&file)?; } @@ -1758,6 +1779,11 @@ struct RolloutWriterState { /// True until a newly created rollout is first materialized. deferred_creation: bool, pending_items: Vec, + /// A sealed V1 event requires a disk barrier, even if background append drained its item. + realtime_sync_pending: bool, + /// Directory entries that must reach disk before the first realtime seal is acknowledged. + #[cfg(unix)] + directories_to_sync: Vec, meta: Option, cwd: PathBuf, rollout_path: PathBuf, @@ -1767,6 +1793,9 @@ struct RolloutWriterState { impl RolloutWriterState { fn add_items(&mut self, items: Vec) { + self.realtime_sync_pending |= items.iter().any( + |item| matches!(item, RolloutItem::RealtimeItem(item) if item.observed_at_ms.is_some()), + ); self.pending_items.extend(items); } @@ -1774,7 +1803,7 @@ impl RolloutWriterState { if self.is_deferred() { return; } - if let Err(err) = self.flush().await { + if let Err(err) = self.write_pending_with_recovery("append").await { self.enter_recovery_mode(&err); } } @@ -1787,7 +1816,21 @@ impl RolloutWriterState { if self.is_deferred() && self.pending_items.is_empty() { return Ok(()); } - self.write_pending_with_recovery("flush").await + self.write_pending_with_recovery("flush").await?; + if self.realtime_sync_pending + && let Some(writer) = self.writer.as_mut() + { + writer.file.sync_data().await?; + #[cfg(unix)] + { + for directory in &self.directories_to_sync { + tokio::fs::File::open(directory).await?.sync_all().await?; + } + self.directories_to_sync.clear(); + } + self.realtime_sync_pending = false; + } + Ok(()) } async fn shutdown(&mut self) -> std::io::Result<()> { @@ -1848,6 +1891,10 @@ impl RolloutWriterState { return Ok(()); } + #[cfg(unix)] + if self.directories_to_sync.is_empty() { + self.directories_to_sync = rollout_directories_to_sync(&self.rollout_path)?; + } let file = open_log_file(self.rollout_path.as_path())?; self.writer = Some(JsonlWriter { file: tokio::fs::File::from_std(file), diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index a0968d3cdd..efcc6d4dcb 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -1029,6 +1029,59 @@ async fn persist_reports_filesystem_error_and_retries_buffered_items() -> std::i Ok(()) } +#[cfg(unix)] +#[tokio::test] +async fn first_realtime_seal_requires_directory_sync_after_ordinary_writes() -> std::io::Result<()> +{ + use std::os::unix::fs::PermissionsExt; + + let home = TempDir::new()?; + let config = test_config(home.path()); + let recorder = RolloutRecorder::new( + &config, + RolloutRecorderParams::new( + ThreadId::new(), + /*forked_from_id*/ None, + /*parent_thread_id*/ None, + SessionSource::Exec, + /*thread_source*/ None, + "test_originator".to_string(), + BaseInstructions::default(), + Vec::new(), + ), + ) + .await?; + let parent = recorder.rollout_path().parent().expect("rollout directory"); + fs::create_dir_all(parent)?; + // Creating and appending a file only needs write/search access to the directory. + let permissions = fs::metadata(parent)?.permissions(); + fs::set_permissions(parent, fs::Permissions::from_mode(0o300))?; + recorder + .record_canonical_items(&[agent_message_item("typed")]) + .await?; + let ordinary_flush = recorder.flush().await; + fs::set_permissions(parent, permissions)?; + ordinary_flush?; + + let record = RolloutItem::RealtimeItem(codex_protocol::realtime::RealtimeItem { + id: "opening".to_string(), + realtime_session_id: "voice".to_string(), + observed_at_ms: Some(1), + content: codex_protocol::realtime::RealtimeItemContent::RealtimeSessionStarted, + }); + recorder.record_canonical_items(&[record]).await?; + // The open file still accepts writes. Only the directory barrier can see this failure. + let moved = home.path().join("moved"); + fs::rename(parent, &moved)?; + assert_eq!( + recorder.flush().await.unwrap_err().kind(), + std::io::ErrorKind::NotFound + ); + fs::rename(moved, parent)?; + recorder.flush().await?; + recorder.shutdown().await +} + #[tokio::test] async fn writer_state_retries_write_error_before_reporting_flush_success() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir"); @@ -1040,6 +1093,9 @@ async fn writer_state_retries_write_error_before_reporting_flush_success() -> st file: tokio::fs::File::from_std(read_only_file), }), deferred_creation: false, + realtime_sync_pending: false, + #[cfg(unix)] + directories_to_sync: Vec::new(), pending_items: Vec::new(), meta: None, cwd: home.path().to_path_buf(), diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 90d9b9db32..2e1f515ff7 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -798,6 +798,10 @@ impl ThreadStore for LocalThreadStore { Box::pin(async move { projects::delete_project(self, project_id).await }) } + fn realtime_frames_enabled(&self, _thread_id: ThreadId) -> ThreadStoreFuture<'_, bool> { + Box::pin(async { Ok(self.state_db.is_some()) }) + } + fn supports_paginated_history_lists(&self) -> bool { self.state_db.is_some() } diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index 472d8b811f..54533ab0e4 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -283,6 +283,7 @@ async fn fork_version_stops_before_older_segments_once_resolved() { next_item_ordinal: None, rejected_audio_file_ids: Default::default(), dynamic_tools: None, + observation_clock_floor_ms: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, turn_attribution: None, diff --git a/codex-rs/thread-store/src/local/thread_history/read_tests.rs b/codex-rs/thread-store/src/local/thread_history/read_tests.rs index 9c22fe4b4f..2184b5a1b3 100644 --- a/codex-rs/thread-store/src/local/thread_history/read_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history/read_tests.rs @@ -430,6 +430,7 @@ async fn timeline_interleaves_items_and_restores_page_boundary_session_state() { RealtimeItem { id: "voice-1:started".to_string(), realtime_session_id: "voice-1".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionStarted, }, ), @@ -438,6 +439,7 @@ async fn timeline_interleaves_items_and_restores_page_boundary_session_state() { RealtimeItem { id: "voice-1:source:0".to_string(), realtime_session_id: "voice-1".to_string(), + observed_at_ms: None, content: RealtimeItemContent::TranscriptSegment { role: RealtimeTranscriptRole::Assistant, text: "Before the artifact".to_string(), @@ -449,6 +451,7 @@ async fn timeline_interleaves_items_and_restores_page_boundary_session_state() { RealtimeItem { id: "artifact".to_string(), realtime_session_id: "voice-1".to_string(), + observed_at_ms: None, content: RealtimeItemContent::BemItemPromoted { turn_id: "turn-1".to_string(), item_id: "middle".to_string(), @@ -461,6 +464,7 @@ async fn timeline_interleaves_items_and_restores_page_boundary_session_state() { RealtimeItem { id: "voice-1:closed".to_string(), realtime_session_id: "voice-1".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionClosed { outcome: RealtimeSessionOutcome::Ended, }, @@ -727,6 +731,7 @@ async fn timeline_page_state_excludes_realtime_events_after_fork_cutoff() { RealtimeItem { id: "voice:started".to_string(), realtime_session_id: "voice".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionStarted, }, ), @@ -735,6 +740,7 @@ async fn timeline_page_state_excludes_realtime_events_after_fork_cutoff() { RealtimeItem { id: "voice:closed".to_string(), realtime_session_id: "voice".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionClosed { outcome: RealtimeSessionOutcome::Ended, }, diff --git a/codex-rs/thread-store/src/local/thread_history/realtime.rs b/codex-rs/thread-store/src/local/thread_history/realtime.rs index cfd3a8ebde..f7e0907689 100644 --- a/codex-rs/thread-store/src/local/thread_history/realtime.rs +++ b/codex-rs/thread-store/src/local/thread_history/realtime.rs @@ -130,6 +130,7 @@ ORDER BY rollout_ordinal DESC, item_id DESC LIMIT ?7 SELECT rollout_ordinal, 2 AS kind, item_id AS id, NULL AS turn_id, item_json FROM thread_realtime_items WHERE thread_id = ?1 AND rollout_ordinal >= ?2 + AND item_type NOT IN ('turn_input_accepted', 'session_end_input_saved') AND rollout_ordinal < ?3 AND rollout_ordinal <= ?4 AND (rollout_ordinal, 2, item_id) < (?4, ?5, ?6) ORDER BY rollout_ordinal DESC, item_id DESC LIMIT ?7 diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index c4a4fd7be9..7447bfc4a5 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -626,11 +626,13 @@ async fn paginated_realtime_items_materialize_separately_in_rollout_order() { RolloutItem::RealtimeItem(RealtimeItem { id: "voice:started".to_string(), realtime_session_id: "voice".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionStarted, }), RolloutItem::RealtimeItem(RealtimeItem { id: "voice:transcript:0".to_string(), realtime_session_id: "voice".to_string(), + observed_at_ms: None, content: RealtimeItemContent::TranscriptSegment { role: RealtimeTranscriptRole::Assistant, text: "Here is the result.".to_string(), @@ -640,6 +642,7 @@ async fn paginated_realtime_items_materialize_separately_in_rollout_order() { RolloutItem::RealtimeItem(RealtimeItem { id: "voice:promoted:agent-1".to_string(), realtime_session_id: "voice".to_string(), + observed_at_ms: None, content: RealtimeItemContent::BemItemPromoted { turn_id: "turn-1".to_string(), item_id: "agent-1".to_string(), @@ -664,6 +667,7 @@ async fn paginated_realtime_items_materialize_separately_in_rollout_order() { RolloutItem::RealtimeItem(RealtimeItem { id: "voice:closed".to_string(), realtime_session_id: "voice".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionClosed { outcome: RealtimeSessionOutcome::Ended, }, @@ -797,6 +801,7 @@ async fn paginated_realtime_items_materialize_separately_in_rollout_order() { items: vec![RolloutItem::RealtimeItem(RealtimeItem { id: "legacy:started".to_string(), realtime_session_id: "legacy".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionStarted, })], }) @@ -1487,6 +1492,7 @@ async fn subagent_prefix_advances_projection_without_materializing_history() { RolloutItem::RealtimeItem(RealtimeItem { id: "parent:started".to_string(), realtime_session_id: "parent".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionStarted, }), turn_started("parent-turn"), @@ -1504,6 +1510,7 @@ async fn subagent_prefix_advances_projection_without_materializing_history() { RolloutItem::RealtimeItem(RealtimeItem { id: "child:started".to_string(), realtime_session_id: "child".to_string(), + observed_at_ms: None, content: RealtimeItemContent::RealtimeSessionStarted, }), completed_item( diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index b8e428d6e0..b1d0300901 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -547,6 +547,11 @@ pub trait ThreadStore: Any + Send + Sync { }) } + /// Whether this live thread writes canonical realtime frames and handoff receipts. + fn realtime_frames_enabled(&self, _thread_id: ThreadId) -> ThreadStoreFuture<'_, bool> { + Box::pin(async { Ok(false) }) + } + /// Whether paginated threads can hydrate durable history through turn and item lists. fn supports_paginated_history_lists(&self) -> bool { false diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 2aeb4b0595..20219f7b98 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -6423,6 +6423,11 @@ class ThreadRealtimeErrorNotification(BaseModel): thread_id: Annotated[str, Field(alias="threadId")] +class ThreadRealtimeHandoffMode(Enum): + start = "start" + steer = "steer" + + class ThreadRealtimeInitialItem(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -6436,6 +6441,13 @@ class RealtimeSessionStartedThreadRealtimeItem(BaseModel): populate_by_name=True, ) id: str + observed_at_ms: Annotated[ + int | None, + Field( + alias="observedAtMs", + description="Canonical observation time in Unix milliseconds for eligible voice calls.", + ), + ] = None realtime_session_id: Annotated[str, Field(alias="realtimeSessionId")] type: Annotated[ Literal["realtimeSessionStarted"], @@ -6448,6 +6460,13 @@ class BemItemPromotedThreadRealtimeItem(BaseModel): populate_by_name=True, ) id: str + observed_at_ms: Annotated[ + int | None, + Field( + alias="observedAtMs", + description="Canonical observation time in Unix milliseconds for eligible voice calls.", + ), + ] = None realtime_session_id: Annotated[str, Field(alias="realtimeSessionId")] item_id: str presentation: ThreadRealtimeBemItemPresentation @@ -6457,6 +6476,51 @@ class BemItemPromotedThreadRealtimeItem(BaseModel): ] +class TurnInputAcceptedThreadRealtimeItem(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + id: str + observed_at_ms: Annotated[ + int | None, + Field( + alias="observedAtMs", + description="Canonical observation time in Unix milliseconds for eligible voice calls.", + ), + ] = None + realtime_session_id: Annotated[str, Field(alias="realtimeSessionId")] + handoff_id: str + input: str + input_item_id: str + mode: ThreadRealtimeHandoffMode + source_event_ids: list[str] + turn_id: str + type: Annotated[ + Literal["turnInputAccepted"], Field(title="TurnInputAcceptedThreadRealtimeItemType") + ] + + +class SessionEndInputSavedThreadRealtimeItem(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + id: str + observed_at_ms: Annotated[ + int | None, + Field( + alias="observedAtMs", + description="Canonical observation time in Unix milliseconds for eligible voice calls.", + ), + ] = None + realtime_session_id: Annotated[str, Field(alias="realtimeSessionId")] + input: str + input_item_id: str + source_event_ids: list[str] + type: Annotated[ + Literal["sessionEndInputSaved"], Field(title="SessionEndInputSavedThreadRealtimeItemType") + ] + + class ThreadRealtimeItemAddedNotification(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -6578,6 +6642,14 @@ class ThreadRealtimeTranscriptRole(Enum): assistant = "assistant" +class ThreadRealtimeVoiceTranscript(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + role: ThreadRealtimeTranscriptRole + text: str + + class ThreadResumeParams(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -11238,11 +11310,42 @@ class ThreadReadStateChangedNotification(BaseModel): thread_id: Annotated[str, Field(alias="threadId")] +class ThreadRealtimeFrameChannels(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + voice_transcript: ThreadRealtimeVoiceTranscript + + +class FrameThreadRealtimeItem(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + id: str + observed_at_ms: Annotated[ + int | None, + Field( + alias="observedAtMs", + description="Canonical observation time in Unix milliseconds for eligible voice calls.", + ), + ] = None + realtime_session_id: Annotated[str, Field(alias="realtimeSessionId")] + channels: ThreadRealtimeFrameChannels + type: Annotated[Literal["frame"], Field(title="FrameThreadRealtimeItemType")] + + class TranscriptSegmentThreadRealtimeItem(BaseModel): model_config = ConfigDict( populate_by_name=True, ) id: str + observed_at_ms: Annotated[ + int | None, + Field( + alias="observedAtMs", + description="Canonical observation time in Unix milliseconds for eligible voice calls.", + ), + ] = None realtime_session_id: Annotated[str, Field(alias="realtimeSessionId")] role: ThreadRealtimeTranscriptRole text: str @@ -11256,6 +11359,13 @@ class RealtimeSessionClosedThreadRealtimeItem(BaseModel): populate_by_name=True, ) id: str + observed_at_ms: Annotated[ + int | None, + Field( + alias="observedAtMs", + description="Canonical observation time in Unix milliseconds for eligible voice calls.", + ), + ] = None realtime_session_id: Annotated[str, Field(alias="realtimeSessionId")] outcome: ThreadRealtimeSessionOutcome type: Annotated[ @@ -11266,16 +11376,22 @@ class RealtimeSessionClosedThreadRealtimeItem(BaseModel): class ThreadRealtimeItem( RootModel[ RealtimeSessionStartedThreadRealtimeItem + | FrameThreadRealtimeItem | TranscriptSegmentThreadRealtimeItem | BemItemPromotedThreadRealtimeItem | RealtimeSessionClosedThreadRealtimeItem + | TurnInputAcceptedThreadRealtimeItem + | SessionEndInputSavedThreadRealtimeItem ] ): root: Annotated[ RealtimeSessionStartedThreadRealtimeItem + | FrameThreadRealtimeItem | TranscriptSegmentThreadRealtimeItem | BemItemPromotedThreadRealtimeItem - | RealtimeSessionClosedThreadRealtimeItem, + | RealtimeSessionClosedThreadRealtimeItem + | TurnInputAcceptedThreadRealtimeItem + | SessionEndInputSavedThreadRealtimeItem, Field( description="EXPERIMENTAL - a thread-scoped realtime item in the canonical timeline." ),