mirror of
https://github.com/openai/codex.git
synced 2026-11-14 21:40:39 +00:00
Persist timestamped realtime frames before acknowledging completion (#61473)
## What changed - Record voice transcripts as `frame` items with a `voice_transcript` channel for eligible threads, while retaining legacy transcript support and the existing app-server live and mixed timeline representation. - Assign first-observed timestamps that never move backward, retain them through completion, and restore the clock floor from history and compaction checkpoints. - Flush realtime records before completion notifications, including file and initial directory durability barriers for local rollouts. - Generate a distinct default session ID for each new call, use the call ID when attaching to an existing call, and derive opening record IDs from session IDs. - Add protocol types for turn-input acceptance and session-end input receipts, excluding those receipts from live notifications and mixed timelines. Update generated schemas and SDK types. ## Testing Add coverage for frame serialization and required voice payloads, clock rollback and checkpoint restoration, durability barrier failures and retries, per-call session IDs, and persisted transcript availability immediately after completion. GitOrigin-RevId: 3f6632b517d563861aee085b723475b8945c391c
This commit is contained in:
Zahan Malkani
@@ -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": {
|
||||
|
||||
+136
@@ -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": {
|
||||
|
||||
+136
@@ -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": {
|
||||
|
||||
+136
@@ -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.",
|
||||
|
||||
+136
@@ -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.",
|
||||
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
+10
@@ -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, };
|
||||
+5
@@ -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";
|
||||
@@ -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<string>, inputItemId: string, input: string, } | { "type": "sessionEndInputSaved", sourceEventIds: Array<string>, inputItemId: string, input: string, });
|
||||
|
||||
+6
@@ -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, };
|
||||
@@ -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";
|
||||
|
||||
@@ -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<i64>,
|
||||
#[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<String>,
|
||||
input_item_id: String,
|
||||
input: String,
|
||||
},
|
||||
SessionEndInputSaved {
|
||||
source_event_ids: Vec<String>,
|
||||
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<CoreRealtimeTranscriptRole> 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<CoreRealtimeItem> 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<CoreRealtimeItem> 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<CoreRealtimeItem> 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,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<i64>,
|
||||
}
|
||||
|
||||
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<i64> {
|
||||
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;
|
||||
@@ -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::<RolloutItem>(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]
|
||||
);
|
||||
}
|
||||
@@ -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),
|
||||
|
||||
@@ -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<i64>,
|
||||
}
|
||||
|
||||
#[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<RealtimeItem>,
|
||||
pub(crate) transcript_stream: Option<RealtimeTranscriptStream>,
|
||||
@@ -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<bool>,
|
||||
active_session_id: Option<String>,
|
||||
active_segments: ActiveTranscriptSegments,
|
||||
streaming_agent_message: Option<StreamingAgentMessage>,
|
||||
@@ -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<RealtimeTranscriptStream> {
|
||||
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;
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<TurnContext>,
|
||||
rollout_items: &[RolloutItem],
|
||||
) -> Option<PreviousTurnSettings> {
|
||||
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),
|
||||
|
||||
@@ -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<bool> {
|
||||
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::<Vec<_>>(),
|
||||
)
|
||||
.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::<Vec<_>>();
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -1283,7 +1283,8 @@ impl Session {
|
||||
error!("failed to initialize thread persistence: {e:#}");
|
||||
e
|
||||
})?;
|
||||
let session_result: anyhow::Result<Arc<Self>> = async {
|
||||
// Keep the large initialization future out of callers' stack frames.
|
||||
let session_result: anyhow::Result<Arc<Self>> = 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) => {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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::<std::collections::HashSet<_>>();
|
||||
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 {
|
||||
|
||||
@@ -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<Box<DynamicToolsSnapshot>>,
|
||||
/// 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<i64>,
|
||||
/// Selected runtime, including when the companion turn context is absent.
|
||||
pub multi_agent_version: Option<MultiAgentVersion>,
|
||||
/// Turn identity used to admit continuations after cold resume.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<i64>,
|
||||
#[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<String>,
|
||||
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<String>,
|
||||
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;
|
||||
|
||||
@@ -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::<RealtimeItem>(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::<RealtimeItem>(empty).is_err());
|
||||
}
|
||||
@@ -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))
|
||||
|
||||
@@ -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<Vec<PathBuf>> {
|
||||
// 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<File> {
|
||||
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<File> {
|
||||
.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<RolloutItem>,
|
||||
/// 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<PathBuf>,
|
||||
meta: Option<SessionMeta>,
|
||||
cwd: PathBuf,
|
||||
rollout_path: PathBuf,
|
||||
@@ -1767,6 +1793,9 @@ struct RolloutWriterState {
|
||||
|
||||
impl RolloutWriterState {
|
||||
fn add_items(&mut self, items: Vec<RolloutItem>) {
|
||||
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),
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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."
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user