mirror of
https://github.com/openai/codex.git
synced 2026-11-01 19:33:35 +00:00
Expose exact per-response usage in raw app-server events (#32985)
## Why Clients that account for individual upstream Responses API calls need the exact usage for each completion rather than accumulated or estimated thread usage. ## What changed - Add the opt-in `rawResponse/completed` app-server notification with the `threadId`, `turnId`, `responseId`, and upstream `usage` payload. - Emit the transient event for normal turns and compaction requests when `experimentalRawEvents` is enabled. The event is not persisted or replayed, and `usage` is `null` when the upstream response omits it. - Export the notification in the app-server schemas and document its behavior. ## Testing - Cover exact token breakdowns for turn and compaction completions. GitOrigin-RevId: 49040a060bf1784c1ff1dd62112efb4a778428c4
This commit is contained in:
+32
@@ -14839,6 +14839,38 @@
|
||||
],
|
||||
"type": "object"
|
||||
},
|
||||
"RawResponseCompletedNotification": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"description": "Internal-only notification containing the exact usage from one upstream Responses API completion.",
|
||||
"properties": {
|
||||
"responseId": {
|
||||
"type": "string"
|
||||
},
|
||||
"threadId": {
|
||||
"type": "string"
|
||||
},
|
||||
"turnId": {
|
||||
"type": "string"
|
||||
},
|
||||
"usage": {
|
||||
"anyOf": [
|
||||
{
|
||||
"$ref": "#/definitions/v2/TokenUsageBreakdown"
|
||||
},
|
||||
{
|
||||
"type": "null"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"responseId",
|
||||
"threadId",
|
||||
"turnId"
|
||||
],
|
||||
"title": "RawResponseCompletedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
"RawResponseItemCompletedNotification": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"properties": {
|
||||
|
||||
+32
@@ -11236,6 +11236,38 @@
|
||||
],
|
||||
"type": "object"
|
||||
},
|
||||
"RawResponseCompletedNotification": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"description": "Internal-only notification containing the exact usage from one upstream Responses API completion.",
|
||||
"properties": {
|
||||
"responseId": {
|
||||
"type": "string"
|
||||
},
|
||||
"threadId": {
|
||||
"type": "string"
|
||||
},
|
||||
"turnId": {
|
||||
"type": "string"
|
||||
},
|
||||
"usage": {
|
||||
"anyOf": [
|
||||
{
|
||||
"$ref": "#/definitions/TokenUsageBreakdown"
|
||||
},
|
||||
{
|
||||
"type": "null"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"responseId",
|
||||
"threadId",
|
||||
"turnId"
|
||||
],
|
||||
"title": "RawResponseCompletedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
"RawResponseItemCompletedNotification": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"properties": {
|
||||
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
{
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"definitions": {
|
||||
"TokenUsageBreakdown": {
|
||||
"properties": {
|
||||
"cachedInputTokens": {
|
||||
"format": "int64",
|
||||
"type": "integer"
|
||||
},
|
||||
"inputTokens": {
|
||||
"format": "int64",
|
||||
"type": "integer"
|
||||
},
|
||||
"outputTokens": {
|
||||
"format": "int64",
|
||||
"type": "integer"
|
||||
},
|
||||
"reasoningOutputTokens": {
|
||||
"format": "int64",
|
||||
"type": "integer"
|
||||
},
|
||||
"totalTokens": {
|
||||
"format": "int64",
|
||||
"type": "integer"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"cachedInputTokens",
|
||||
"inputTokens",
|
||||
"outputTokens",
|
||||
"reasoningOutputTokens",
|
||||
"totalTokens"
|
||||
],
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"description": "Internal-only notification containing the exact usage from one upstream Responses API completion.",
|
||||
"properties": {
|
||||
"responseId": {
|
||||
"type": "string"
|
||||
},
|
||||
"threadId": {
|
||||
"type": "string"
|
||||
},
|
||||
"turnId": {
|
||||
"type": "string"
|
||||
},
|
||||
"usage": {
|
||||
"anyOf": [
|
||||
{
|
||||
"$ref": "#/definitions/TokenUsageBreakdown"
|
||||
},
|
||||
{
|
||||
"type": "null"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"responseId",
|
||||
"threadId",
|
||||
"turnId"
|
||||
],
|
||||
"title": "RawResponseCompletedNotification",
|
||||
"type": "object"
|
||||
}
|
||||
File diff suppressed because one or more lines are too long
+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 { TokenUsageBreakdown } from "./TokenUsageBreakdown";
|
||||
|
||||
/**
|
||||
* Internal-only notification containing the exact usage from one upstream
|
||||
* Responses API completion.
|
||||
*/
|
||||
export type RawResponseCompletedNotification = { threadId: string, turnId: string, responseId: string, usage: TokenUsageBreakdown | null, };
|
||||
@@ -343,6 +343,7 @@ export type { RateLimitResetCreditsSummary } from "./RateLimitResetCreditsSummar
|
||||
export type { RateLimitResetType } from "./RateLimitResetType";
|
||||
export type { RateLimitSnapshot } from "./RateLimitSnapshot";
|
||||
export type { RateLimitWindow } from "./RateLimitWindow";
|
||||
export type { RawResponseCompletedNotification } from "./RawResponseCompletedNotification";
|
||||
export type { RawResponseItemCompletedNotification } from "./RawResponseItemCompletedNotification";
|
||||
export type { ReasoningEffortOption } from "./ReasoningEffortOption";
|
||||
export type { ReasoningSummaryPartAddedNotification } from "./ReasoningSummaryPartAddedNotification";
|
||||
|
||||
@@ -60,7 +60,8 @@ const SPECIAL_DEFINITIONS: &[&str] = &[
|
||||
const FLAT_V2_SHARED_DEFINITIONS: &[&str] = &["ClientRequest", "ServerNotification"];
|
||||
const V1_CLIENT_REQUEST_METHODS: &[&str] =
|
||||
&["getConversationSummary", "gitDiffToRemote", "getAuthStatus"];
|
||||
const EXCLUDED_SERVER_NOTIFICATION_METHODS_FOR_JSON: &[&str] = &["rawResponseItem/completed"];
|
||||
const EXCLUDED_SERVER_NOTIFICATION_METHODS_FOR_JSON: &[&str] =
|
||||
&["rawResponseItem/completed", "rawResponse/completed"];
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct GeneratedSchema {
|
||||
|
||||
@@ -1645,6 +1645,8 @@ server_notification_definitions! {
|
||||
ItemCompleted => "item/completed" (v2::ItemCompletedNotification),
|
||||
/// This event is internal-only. Used by Codex Cloud.
|
||||
RawResponseItemCompleted => "rawResponseItem/completed" (v2::RawResponseItemCompletedNotification),
|
||||
/// This event is internal-only. Used by clients that need exact upstream usage.
|
||||
RawResponseCompleted => "rawResponse/completed" (v2::RawResponseCompletedNotification),
|
||||
AgentMessageDelta => "item/agentMessage/delta" (v2::AgentMessageDeltaNotification),
|
||||
/// EXPERIMENTAL - proposed plan streaming deltas for plan items.
|
||||
PlanDelta => "item/plan/delta" (v2::PlanDeltaNotification),
|
||||
|
||||
@@ -1375,6 +1375,18 @@ pub struct ThreadTokenUsageUpdatedNotification {
|
||||
pub token_usage: ThreadTokenUsage,
|
||||
}
|
||||
|
||||
/// Internal-only notification containing the exact usage from one upstream
|
||||
/// Responses API completion.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export_to = "v2/")]
|
||||
pub struct RawResponseCompletedNotification {
|
||||
pub thread_id: String,
|
||||
pub turn_id: String,
|
||||
pub response_id: String,
|
||||
pub usage: Option<TokenUsageBreakdown>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export_to = "v2/")]
|
||||
|
||||
@@ -1357,6 +1357,7 @@ The app-server streams JSON-RPC notifications while a turn is running. Each turn
|
||||
- `turn/completed` — `{ turn }` where `turn.status` is `completed`, `interrupted`, or `failed`; failures carry `{ error: { message, codexErrorInfo?, additionalDetails? } }`.
|
||||
- `turn/diff/updated` — `{ threadId, turnId, diff }` represents the up-to-date snapshot of the turn-level unified diff, emitted after every FileChange item. `diff` is the latest aggregated unified diff across every file change in the turn. UIs can render this to show the full "what changed" view without stitching individual `fileChange` items.
|
||||
- `turn/plan/updated` — `{ turnId, explanation?, plan }` whenever the agent shares or changes its plan; each `plan` entry is `{ step, status }` with `status` in `pending`, `inProgress`, or `completed`.
|
||||
- `rawResponse/completed` — internal-only; when `thread/start.experimentalRawEvents` is enabled, emits `{ threadId, turnId, responseId, usage }` once for each upstream Responses API completion. `usage` is the exact upstream usage payload mapped to the app-server token breakdown shape and is `null` when the upstream completion omitted usage. Unlike `thread/tokenUsage/updated`, this notification is not accumulated, estimated, persisted, or replayed.
|
||||
- `model/safetyBuffering/updated` — `{ threadId, turnId, model, useCases, reasons, showBufferingUi, fasterModel }` when a response enters safety buffering. `fasterModel` is nullable. This notification is transient and is not persisted in rollout history.
|
||||
- `model/rerouted` — `{ threadId, turnId, fromModel, toModel, reason }` when the backend reroutes a request to a different model (for example, due to high-risk cyber safety checks).
|
||||
- `model/verification` — `{ threadId, turnId, verifications }` when the backend flags additional account verification, such as `trustedAccessForCyber`.
|
||||
|
||||
@@ -46,6 +46,7 @@ use codex_app_server_protocol::NetworkPolicyAmendment as V2NetworkPolicyAmendmen
|
||||
use codex_app_server_protocol::NetworkPolicyRuleAction as V2NetworkPolicyRuleAction;
|
||||
use codex_app_server_protocol::PermissionsRequestApprovalParams;
|
||||
use codex_app_server_protocol::PermissionsRequestApprovalResponse;
|
||||
use codex_app_server_protocol::RawResponseCompletedNotification;
|
||||
use codex_app_server_protocol::RawResponseItemCompletedNotification;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
@@ -1029,6 +1030,17 @@ pub(crate) async fn apply_bespoke_event_handling(
|
||||
)
|
||||
.await;
|
||||
}
|
||||
EventMsg::RawResponseCompleted(raw_response_completed_event) => {
|
||||
let notification = RawResponseCompletedNotification {
|
||||
thread_id: conversation_id.to_string(),
|
||||
turn_id: event_turn_id,
|
||||
response_id: raw_response_completed_event.response_id,
|
||||
usage: raw_response_completed_event.token_usage.map(Into::into),
|
||||
};
|
||||
outgoing
|
||||
.send_server_notification(ServerNotification::RawResponseCompleted(notification))
|
||||
.await;
|
||||
}
|
||||
EventMsg::PatchApplyBegin(_) | EventMsg::PatchApplyEnd(_) => {
|
||||
// Core still fans out these deprecated events for raw-event and rollout compatibility
|
||||
// consumers;
|
||||
|
||||
@@ -316,7 +316,11 @@ pub(super) async fn ensure_listener_task_running(
|
||||
thread_state.track_current_turn_event(&event.id, &event.msg);
|
||||
thread_state.experimental_raw_events
|
||||
};
|
||||
if matches!(&event.msg, EventMsg::RawResponseItem(_)) && !raw_events_enabled {
|
||||
if matches!(
|
||||
&event.msg,
|
||||
EventMsg::RawResponseItem(_) | EventMsg::RawResponseCompleted(_)
|
||||
) && !raw_events_enabled
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let subscribed_connection_ids = thread_state_manager
|
||||
|
||||
@@ -16,12 +16,15 @@ use codex_app_server_protocol::ItemStartedNotification;
|
||||
use codex_app_server_protocol::JSONRPCError;
|
||||
use codex_app_server_protocol::JSONRPCNotification;
|
||||
use codex_app_server_protocol::JSONRPCResponse;
|
||||
use codex_app_server_protocol::RawResponseCompletedNotification;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::ThreadCompactStartParams;
|
||||
use codex_app_server_protocol::ThreadCompactStartResponse;
|
||||
use codex_app_server_protocol::ThreadItem;
|
||||
use codex_app_server_protocol::ThreadStartParams;
|
||||
use codex_app_server_protocol::ThreadStartResponse;
|
||||
use codex_app_server_protocol::TokenUsageBreakdown;
|
||||
use codex_app_server_protocol::TurnCompletedNotification;
|
||||
use codex_app_server_protocol::TurnStartParams;
|
||||
use codex_app_server_protocol::TurnStartResponse;
|
||||
@@ -283,7 +286,20 @@ async fn thread_compact_start_triggers_compaction_and_returns_empty_response() -
|
||||
.await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let thread_id = start_thread(&mut mcp).await?;
|
||||
let thread_req = mcp
|
||||
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
||||
model: Some("mock-model".to_string()),
|
||||
experimental_raw_events: true,
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let thread_resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
|
||||
let thread_id = thread.id;
|
||||
let compact_id = mcp
|
||||
.send_thread_compact_start_request(ThreadCompactStartParams {
|
||||
thread_id: thread_id.clone(),
|
||||
@@ -298,6 +314,15 @@ async fn thread_compact_start_triggers_compaction_and_returns_empty_response() -
|
||||
to_response::<ThreadCompactStartResponse>(compact_resp)?;
|
||||
|
||||
let started = wait_for_context_compaction_started(&mut mcp).await?;
|
||||
let raw_completed = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_notification_message("rawResponse/completed"),
|
||||
)
|
||||
.await??;
|
||||
let raw_completed: ServerNotification = raw_completed.try_into()?;
|
||||
let ServerNotification::RawResponseCompleted(raw_completed) = raw_completed else {
|
||||
anyhow::bail!("expected rawResponse/completed notification");
|
||||
};
|
||||
let completed = wait_for_context_compaction_completed(&mut mcp).await?;
|
||||
|
||||
let ThreadItem::ContextCompaction { id: started_id } = started.item else {
|
||||
@@ -310,6 +335,21 @@ async fn thread_compact_start_triggers_compaction_and_returns_empty_response() -
|
||||
assert_eq!(started.thread_id, thread_id);
|
||||
assert_eq!(completed.thread_id, thread_id);
|
||||
assert_eq!(started_id, completed_id);
|
||||
assert_eq!(
|
||||
raw_completed,
|
||||
RawResponseCompletedNotification {
|
||||
thread_id,
|
||||
turn_id: started.turn_id,
|
||||
response_id: "r1".to_string(),
|
||||
usage: Some(TokenUsageBreakdown {
|
||||
total_tokens: 200,
|
||||
input_tokens: 200,
|
||||
cached_input_tokens: 0,
|
||||
output_tokens: 0,
|
||||
reasoning_output_tokens: 0,
|
||||
}),
|
||||
}
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -36,6 +36,7 @@ use codex_app_server_protocol::JSONRPCNotification;
|
||||
use codex_app_server_protocol::JSONRPCResponse;
|
||||
use codex_app_server_protocol::PatchApplyStatus;
|
||||
use codex_app_server_protocol::PatchChangeKind;
|
||||
use codex_app_server_protocol::RawResponseCompletedNotification;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ServerRequest;
|
||||
use codex_app_server_protocol::ServerRequestResolvedNotification;
|
||||
@@ -51,6 +52,7 @@ use codex_app_server_protocol::ThreadSettingsUpdatedNotification;
|
||||
use codex_app_server_protocol::ThreadSource;
|
||||
use codex_app_server_protocol::ThreadStartParams;
|
||||
use codex_app_server_protocol::ThreadStartResponse;
|
||||
use codex_app_server_protocol::TokenUsageBreakdown;
|
||||
use codex_app_server_protocol::TurnCompletedNotification;
|
||||
use codex_app_server_protocol::TurnEnvironmentParams;
|
||||
use codex_app_server_protocol::TurnItemsView;
|
||||
@@ -765,6 +767,104 @@ async fn turn_start_sends_service_tier_id_to_model_request() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn turn_start_emits_raw_response_completed_with_upstream_usage() -> Result<()> {
|
||||
let server = responses::start_mock_server().await;
|
||||
let body = responses::sse(vec![
|
||||
responses::ev_response_created("resp-1"),
|
||||
responses::ev_assistant_message("msg-1", "Done"),
|
||||
json!({
|
||||
"type": "response.completed",
|
||||
"response": {
|
||||
"id": "resp-1",
|
||||
"usage": {
|
||||
"input_tokens": 30,
|
||||
"input_tokens_details": { "cached_tokens": 11 },
|
||||
"output_tokens": 7,
|
||||
"output_tokens_details": { "reasoning_tokens": 3 },
|
||||
"total_tokens": 37
|
||||
}
|
||||
}
|
||||
}),
|
||||
]);
|
||||
responses::mount_sse_once(&server, body).await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml(
|
||||
codex_home.path(),
|
||||
&server.uri(),
|
||||
"never",
|
||||
&BTreeMap::default(),
|
||||
)?;
|
||||
write_models_cache(codex_home.path())?;
|
||||
|
||||
let mut mcp = TestAppServer::builder()
|
||||
.with_codex_home(codex_home.path())
|
||||
.build()
|
||||
.await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let thread_req = mcp
|
||||
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
||||
experimental_raw_events: true,
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let thread_resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
|
||||
|
||||
let turn_req = mcp
|
||||
.send_turn_start_request(TurnStartParams {
|
||||
thread_id: thread.id.clone(),
|
||||
input: vec![V2UserInput::Text {
|
||||
text: "Hello".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let turn_resp = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
|
||||
)
|
||||
.await??;
|
||||
let TurnStartResponse { turn } = to_response::<TurnStartResponse>(turn_resp)?;
|
||||
|
||||
let notification = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_notification_message("rawResponse/completed"),
|
||||
)
|
||||
.await??;
|
||||
let notification: codex_app_server_protocol::ServerNotification = notification.try_into()?;
|
||||
let codex_app_server_protocol::ServerNotification::RawResponseCompleted(notification) =
|
||||
notification
|
||||
else {
|
||||
anyhow::bail!("expected rawResponse/completed notification");
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
notification,
|
||||
RawResponseCompletedNotification {
|
||||
thread_id: thread.id,
|
||||
turn_id: turn.id,
|
||||
response_id: "resp-1".to_string(),
|
||||
usage: Some(TokenUsageBreakdown {
|
||||
total_tokens: 37,
|
||||
input_tokens: 30,
|
||||
cached_input_tokens: 11,
|
||||
output_tokens: 7,
|
||||
reasoning_output_tokens: 3,
|
||||
}),
|
||||
}
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_omits_empty_instruction_overrides_from_model_request() -> Result<()> {
|
||||
let server = responses::start_mock_server().await;
|
||||
|
||||
@@ -36,6 +36,7 @@ use codex_protocol::models::ResponseInputItem;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::protocol::CompactedItem;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::RawResponseCompletedEvent;
|
||||
use codex_protocol::protocol::TurnStartedEvent;
|
||||
use codex_protocol::protocol::WarningEvent;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
@@ -698,7 +699,19 @@ async fn drain_to_completed(
|
||||
Ok(ResponseEvent::RateLimits(snapshot)) => {
|
||||
sess.update_rate_limits(turn_context, snapshot).await;
|
||||
}
|
||||
Ok(ResponseEvent::Completed { token_usage, .. }) => {
|
||||
Ok(ResponseEvent::Completed {
|
||||
response_id,
|
||||
token_usage,
|
||||
..
|
||||
}) => {
|
||||
sess.send_event(
|
||||
turn_context,
|
||||
EventMsg::RawResponseCompleted(RawResponseCompletedEvent {
|
||||
response_id,
|
||||
token_usage: token_usage.clone(),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
sess.update_token_usage_info(turn_context, token_usage.as_ref())
|
||||
.await?;
|
||||
return Ok(());
|
||||
|
||||
@@ -325,6 +325,7 @@ async fn run_remote_compact_task_inner_impl(
|
||||
|
||||
struct RemoteCompactionV2Output {
|
||||
compaction_output: ResponseItem,
|
||||
response_id: String,
|
||||
token_usage: Option<TokenUsage>,
|
||||
}
|
||||
|
||||
@@ -385,6 +386,7 @@ async fn collect_compaction_output(
|
||||
let mut compaction_count = 0usize;
|
||||
let mut compaction_output = None;
|
||||
let mut saw_completed = false;
|
||||
let mut completed_response_id = None;
|
||||
let mut completed_token_usage = None;
|
||||
while let Some(event) = stream.next().await {
|
||||
match event? {
|
||||
@@ -397,8 +399,13 @@ async fn collect_compaction_output(
|
||||
}
|
||||
}
|
||||
}
|
||||
ResponseEvent::Completed { token_usage, .. } => {
|
||||
ResponseEvent::Completed {
|
||||
response_id,
|
||||
token_usage,
|
||||
..
|
||||
} => {
|
||||
saw_completed = true;
|
||||
completed_response_id = Some(response_id);
|
||||
completed_token_usage = token_usage;
|
||||
break;
|
||||
}
|
||||
@@ -422,8 +429,12 @@ async fn collect_compaction_output(
|
||||
let Some(compaction_output) = compaction_output else {
|
||||
unreachable!("compaction output must exist when count is exactly one");
|
||||
};
|
||||
let Some(response_id) = completed_response_id else {
|
||||
unreachable!("response id must exist after response.completed");
|
||||
};
|
||||
Ok(RemoteCompactionV2Output {
|
||||
compaction_output,
|
||||
response_id,
|
||||
token_usage: completed_token_usage,
|
||||
})
|
||||
}
|
||||
@@ -837,6 +848,7 @@ mod tests {
|
||||
.expect("compaction should be collected");
|
||||
|
||||
assert_eq!(output.compaction_output, compaction);
|
||||
assert_eq!(output.response_id, "resp-compact");
|
||||
assert_eq!(
|
||||
output.token_usage,
|
||||
Some(TokenUsage {
|
||||
|
||||
@@ -13,6 +13,8 @@ use crate::session::step_context::StepContext;
|
||||
use crate::session::turn::built_tools;
|
||||
use codex_protocol::error::Result as CodexResult;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::RawResponseCompletedEvent;
|
||||
use codex_protocol::protocol::TokenUsage;
|
||||
use codex_rollout_trace::CompactionTraceContext;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
@@ -114,8 +116,19 @@ pub(super) async fn run_remote_compact_v2_attempt(
|
||||
);
|
||||
let RemoteCompactionV2Output {
|
||||
compaction_output,
|
||||
response_id,
|
||||
token_usage,
|
||||
} = compaction_output_result?;
|
||||
// TODO: Emit this before compaction output validation so malformed completed
|
||||
// responses still surface their raw upstream usage.
|
||||
sess.send_event(
|
||||
turn_context,
|
||||
EventMsg::RawResponseCompleted(RawResponseCompletedEvent {
|
||||
response_id,
|
||||
token_usage: token_usage.clone(),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
Ok(RemoteCompactV2Attempt {
|
||||
trace_input_history,
|
||||
prompt_input,
|
||||
|
||||
@@ -100,6 +100,7 @@ use codex_protocol::protocol::CodexErrorInfo;
|
||||
use codex_protocol::protocol::ErrorEvent;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::PlanDeltaEvent;
|
||||
use codex_protocol::protocol::RawResponseCompletedEvent;
|
||||
use codex_protocol::protocol::ReasoningContentDeltaEvent;
|
||||
use codex_protocol::protocol::ReasoningRawContentDeltaEvent;
|
||||
use codex_protocol::protocol::SafetyBufferingEvent;
|
||||
@@ -1591,6 +1592,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<(String, Option<
|
||||
| EventMsg::EnteredReviewMode(_)
|
||||
| EventMsg::ExitedReviewMode(_)
|
||||
| EventMsg::RawResponseItem(_)
|
||||
| EventMsg::RawResponseCompleted(_)
|
||||
| EventMsg::ItemStarted(_)
|
||||
| EventMsg::HookStarted(_)
|
||||
| EventMsg::HookCompleted(_)
|
||||
@@ -2276,9 +2278,9 @@ async fn try_run_sampling_request(
|
||||
.await;
|
||||
}
|
||||
ResponseEvent::Completed {
|
||||
response_id,
|
||||
token_usage,
|
||||
end_turn,
|
||||
..
|
||||
} => {
|
||||
flush_assistant_text_segments_all(
|
||||
&sess,
|
||||
@@ -2287,6 +2289,14 @@ async fn try_run_sampling_request(
|
||||
&mut assistant_message_stream_parsers,
|
||||
)
|
||||
.await;
|
||||
sess.send_event(
|
||||
&turn_context,
|
||||
EventMsg::RawResponseCompleted(RawResponseCompletedEvent {
|
||||
response_id,
|
||||
token_usage: token_usage.clone(),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let budget_result = sess
|
||||
.record_token_usage_info(&turn_context, token_usage.as_ref())
|
||||
.await;
|
||||
|
||||
@@ -358,6 +358,7 @@ async fn run_codex_tool_session_inner(
|
||||
| EventMsg::ImageGenerationEnd(_)
|
||||
| EventMsg::ViewImageToolCall(_)
|
||||
| EventMsg::RawResponseItem(_)
|
||||
| EventMsg::RawResponseCompleted(_)
|
||||
| EventMsg::EnteredReviewMode(_)
|
||||
| EventMsg::ItemStarted(_)
|
||||
| EventMsg::ItemCompleted(_)
|
||||
|
||||
@@ -1440,6 +1440,7 @@ pub enum EventMsg {
|
||||
ExitedReviewMode(ExitedReviewModeEvent),
|
||||
|
||||
RawResponseItem(RawResponseItemEvent),
|
||||
RawResponseCompleted(RawResponseCompletedEvent),
|
||||
|
||||
ItemStarted(ItemStartedEvent),
|
||||
ItemCompleted(ItemCompletedEvent),
|
||||
@@ -1796,6 +1797,15 @@ pub struct RawResponseItemEvent {
|
||||
pub item: ResponseItem,
|
||||
}
|
||||
|
||||
/// Exact usage reported by one upstream Responses API completion.
|
||||
///
|
||||
/// Unlike TokenCountEvent, this is not accumulated, estimated, or replayed.
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema)]
|
||||
pub struct RawResponseCompletedEvent {
|
||||
pub response_id: String,
|
||||
pub token_usage: Option<TokenUsage>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema)]
|
||||
pub struct ItemStartedEvent {
|
||||
pub thread_id: ThreadId,
|
||||
|
||||
@@ -395,6 +395,7 @@ pub(crate) fn tool_runtime_trace_event(event: &EventMsg) -> Option<ToolRuntimeTr
|
||||
| EventMsg::EnteredReviewMode(_)
|
||||
| EventMsg::ExitedReviewMode(_)
|
||||
| EventMsg::RawResponseItem(_)
|
||||
| EventMsg::RawResponseCompleted(_)
|
||||
| EventMsg::ItemStarted(_)
|
||||
| EventMsg::ItemCompleted(_)
|
||||
| EventMsg::HookStarted(_)
|
||||
@@ -468,6 +469,7 @@ pub(crate) fn wrapped_protocol_event_type(event: &EventMsg) -> Option<&'static s
|
||||
| EventMsg::EnteredReviewMode(_)
|
||||
| EventMsg::ExitedReviewMode(_)
|
||||
| EventMsg::RawResponseItem(_)
|
||||
| EventMsg::RawResponseCompleted(_)
|
||||
| EventMsg::ItemStarted(_)
|
||||
| EventMsg::ItemCompleted(_)
|
||||
| EventMsg::HookStarted(_)
|
||||
|
||||
@@ -138,6 +138,7 @@ pub fn should_persist_event_msg(ev: &EventMsg, history_mode: ThreadHistoryMode)
|
||||
| EventMsg::TurnModerationMetadata(_)
|
||||
| EventMsg::AgentReasoningSectionBreak(_)
|
||||
| EventMsg::RawResponseItem(_)
|
||||
| EventMsg::RawResponseCompleted(_)
|
||||
| EventMsg::SessionConfigured(_)
|
||||
| EventMsg::McpToolCallBegin(_)
|
||||
| EventMsg::ExecCommandBegin(_)
|
||||
|
||||
@@ -87,6 +87,9 @@ pub(super) fn server_notification_thread_target(
|
||||
ServerNotification::RawResponseItemCompleted(notification) => {
|
||||
Some(notification.thread_id.as_str())
|
||||
}
|
||||
ServerNotification::RawResponseCompleted(notification) => {
|
||||
Some(notification.thread_id.as_str())
|
||||
}
|
||||
ServerNotification::AgentMessageDelta(notification) => {
|
||||
Some(notification.thread_id.as_str())
|
||||
}
|
||||
|
||||
@@ -200,6 +200,7 @@ impl ChatWidget {
|
||||
| ServerNotification::ThreadDeleted(_)
|
||||
| ServerNotification::ThreadUnarchived(_)
|
||||
| ServerNotification::RawResponseItemCompleted(_)
|
||||
| ServerNotification::RawResponseCompleted(_)
|
||||
| ServerNotification::CommandExecOutputDelta(_)
|
||||
| ServerNotification::ProcessOutputDelta(_)
|
||||
| ServerNotification::ProcessExited(_)
|
||||
|
||||
Reference in New Issue
Block a user