Merge branch 'main' into codex/prefix-compaction-prewarm

This commit is contained in:
rka-oai
2026-04-16 13:40:12 -07:00
committed by GitHub
44 changed files with 1450 additions and 217 deletions
@@ -4,6 +4,7 @@ use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCompactionEventRequest;
use crate::events::CodexHookRunEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
@@ -12,6 +13,7 @@ use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_hook_run_metadata;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::events::subagent_thread_started_event_request;
@@ -28,6 +30,8 @@ use crate::facts::CompactionStatus;
use crate::facts::CompactionStrategy;
use crate::facts::CompactionTrigger;
use crate::facts::CustomAnalyticsFact;
use crate::facts::HookRunFact;
use crate::facts::HookRunInput;
use crate::facts::InputError;
use crate::facts::InvocationType;
use crate::facts::PluginState;
@@ -81,6 +85,9 @@ use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::ModeKind;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookRunStatus;
use codex_protocol::protocol::HookSource;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
@@ -1282,6 +1289,109 @@ fn plugin_management_event_serializes_expected_shape() {
);
}
#[test]
fn hook_run_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-3".to_string(),
turn_id: "turn-3".to_string(),
};
let event = TrackEventRequest::HookRun(CodexHookRunEventRequest {
event_type: "codex_hook_run",
event_params: codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::PreToolUse,
hook_source: HookSource::User,
status: HookRunStatus::Completed,
},
),
});
let payload = serde_json::to_value(&event).expect("serialize hook run event");
assert_eq!(
payload,
json!({
"event_type": "codex_hook_run",
"event_params": {
"thread_id": "thread-3",
"turn_id": "turn-3",
"model_slug": "gpt-5",
"hook_name": "PreToolUse",
"hook_source": "user",
"status": "completed"
}
})
);
}
#[test]
fn hook_run_metadata_maps_sources_and_statuses() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let system = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::SessionStart,
hook_source: HookSource::System,
status: HookRunStatus::Completed,
},
))
.expect("serialize system hook");
let project = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::Stop,
hook_source: HookSource::Project,
status: HookRunStatus::Blocked,
},
))
.expect("serialize project hook");
let unknown = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::UserPromptSubmit,
hook_source: HookSource::Unknown,
status: HookRunStatus::Failed,
},
))
.expect("serialize unknown hook");
assert_eq!(system["hook_source"], "system");
assert_eq!(system["status"], "completed");
assert_eq!(project["hook_source"], "project");
assert_eq!(project["status"], "blocked");
assert_eq!(unknown["hook_source"], "unknown");
assert_eq!(unknown["status"], "failed");
}
#[test]
fn hook_run_metadata_maps_stopped_status() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let stopped = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::Stop,
hook_source: HookSource::User,
status: HookRunStatus::Stopped,
},
))
.expect("serialize stopped hook");
assert_eq!(stopped["hook_source"], "user");
assert_eq!(stopped["status"], "stopped");
}
#[test]
fn plugin_used_dedupe_is_keyed_by_turn_and_plugin() {
let (sender, _receiver) = mpsc::channel(1);
@@ -1359,6 +1469,37 @@ async fn reducer_ingests_skill_invoked_fact() {
);
}
#[tokio::test]
async fn reducer_ingests_hook_run_fact() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::HookRun(HookRunInput {
tracking: TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
},
hook: HookRunFact {
event_name: HookEventName::PostToolUse,
hook_source: HookSource::Unknown,
status: HookRunStatus::Failed,
},
})),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_hook_run");
assert_eq!(payload[0]["event_params"]["hook_name"], "PostToolUse");
assert_eq!(payload[0]["event_params"]["hook_source"], "unknown");
assert_eq!(payload[0]["event_params"]["status"], "failed");
}
#[tokio::test]
async fn reducer_ingests_app_and_plugin_facts() {
let mut reducer = AnalyticsReducer::default();
+8
View File
@@ -9,6 +9,8 @@ use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CustomAnalyticsFact;
use crate::facts::HookRunFact;
use crate::facts::HookRunInput;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::SkillInvocation;
@@ -191,6 +193,12 @@ impl AnalyticsEventsClient {
)));
}
pub fn track_hook_run(&self, tracking: TrackEventsContext, hook: HookRunFact) {
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::HookRun(
HookRunInput { tracking, hook },
)));
}
pub fn track_plugin_used(&self, tracking: TrackEventsContext, plugin: PluginTelemetryMetadata) {
if !self.queue.should_enqueue_plugin_used(&tracking, &plugin) {
return;
+66
View File
@@ -1,5 +1,6 @@
use crate::facts::AppInvocation;
use crate::facts::CodexCompactionEvent;
use crate::facts::HookRunFact;
use crate::facts::InvocationType;
use crate::facts::PluginState;
use crate::facts::SubAgentThreadStartedInput;
@@ -15,6 +16,9 @@ use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::approvals::NetworkApprovalProtocol;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::SandboxPermissions;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookRunStatus;
use codex_protocol::protocol::HookSource;
use codex_protocol::protocol::SubAgentSource;
use serde::Serialize;
@@ -39,6 +43,7 @@ pub(crate) enum TrackEventRequest {
GuardianReview(Box<GuardianReviewEventRequest>),
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
HookRun(CodexHookRunEventRequest),
Compaction(Box<CodexCompactionEventRequest>),
TurnEvent(Box<CodexTurnEventRequest>),
TurnSteer(CodexTurnSteerEventRequest),
@@ -300,6 +305,22 @@ pub(crate) struct CodexAppUsedEventRequest {
pub(crate) event_params: CodexAppMetadata,
}
#[derive(Serialize)]
pub(crate) struct CodexHookRunMetadata {
pub(crate) thread_id: Option<String>,
pub(crate) turn_id: Option<String>,
pub(crate) model_slug: Option<String>,
pub(crate) hook_name: Option<String>,
pub(crate) hook_source: Option<&'static str>,
pub(crate) status: Option<HookRunStatus>,
}
#[derive(Serialize)]
pub(crate) struct CodexHookRunEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexHookRunMetadata,
}
#[derive(Serialize)]
pub(crate) struct CodexCompactionEventParams {
pub(crate) thread_id: String,
@@ -529,6 +550,43 @@ pub(crate) fn codex_plugin_used_metadata(
}
}
pub(crate) fn codex_hook_run_metadata(
tracking: &TrackEventsContext,
hook: HookRunFact,
) -> CodexHookRunMetadata {
CodexHookRunMetadata {
thread_id: Some(tracking.thread_id.clone()),
turn_id: Some(tracking.turn_id.clone()),
model_slug: Some(tracking.model_slug.clone()),
hook_name: Some(analytics_hook_event_name(hook.event_name).to_owned()),
hook_source: Some(analytics_hook_source(hook.hook_source)),
status: Some(analytics_hook_status(hook.status)),
}
}
fn analytics_hook_event_name(event_name: HookEventName) -> &'static str {
match event_name {
HookEventName::PreToolUse => "PreToolUse",
HookEventName::PostToolUse => "PostToolUse",
HookEventName::SessionStart => "SessionStart",
HookEventName::UserPromptSubmit => "UserPromptSubmit",
HookEventName::Stop => "Stop",
}
}
fn analytics_hook_source(source: HookSource) -> &'static str {
match source {
HookSource::System => "system",
HookSource::User => "user",
HookSource::Project => "project",
HookSource::Mdm => "mdm",
HookSource::SessionFlags => "session_flags",
HookSource::LegacyManagedConfigFile => "legacy_managed_config_file",
HookSource::LegacyManagedConfigMdm => "legacy_managed_config_mdm",
HookSource::Unknown => "unknown",
}
}
pub(crate) fn current_runtime_metadata() -> CodexRuntimeMetadata {
let os_info = os_info::get();
CodexRuntimeMetadata {
@@ -586,3 +644,11 @@ pub(crate) fn subagent_parent_thread_id(subagent_source: &SubAgentSource) -> Opt
_ => None,
}
}
fn analytics_hook_status(status: HookRunStatus) -> HookRunStatus {
match status {
// Running is unexpected here and normalized defensively.
HookRunStatus::Running => HookRunStatus::Failed,
other => other,
}
}
+15
View File
@@ -15,6 +15,9 @@ use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::ServiceTier;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookRunStatus;
use codex_protocol::protocol::HookSource;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SkillScope;
@@ -298,6 +301,7 @@ pub(crate) enum CustomAnalyticsFact {
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),
HookRun(HookRunInput),
PluginUsed(PluginUsedInput),
PluginStateChanged(PluginStateChangedInput),
}
@@ -317,6 +321,17 @@ pub(crate) struct AppUsedInput {
pub app: AppInvocation,
}
pub(crate) struct HookRunInput {
pub tracking: TrackEventsContext,
pub hook: HookRunFact,
}
pub struct HookRunFact {
pub event_name: HookEventName,
pub hook_source: HookSource,
pub status: HookRunStatus,
}
pub(crate) struct PluginUsedInput {
pub tracking: TrackEventsContext,
pub plugin: PluginTelemetryMetadata,
+1
View File
@@ -29,6 +29,7 @@ pub use facts::CompactionReason;
pub use facts::CompactionStatus;
pub use facts::CompactionStrategy;
pub use facts::CompactionTrigger;
pub use facts::HookRunFact;
pub use facts::InputError;
pub use facts::InvocationType;
pub use facts::SkillInvocation;
+14
View File
@@ -3,6 +3,7 @@ use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCompactionEventRequest;
use crate::events::CodexHookRunEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
@@ -20,6 +21,7 @@ use crate::events::ThreadInitializedEventParams;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_compaction_event_params;
use crate::events::codex_hook_run_metadata;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::events::plugin_state_event_type;
@@ -32,6 +34,7 @@ use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CodexCompactionEvent;
use crate::facts::CustomAnalyticsFact;
use crate::facts::HookRunInput;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
@@ -217,6 +220,9 @@ impl AnalyticsReducer {
CustomAnalyticsFact::AppUsed(input) => {
self.ingest_app_used(input, out);
}
CustomAnalyticsFact::HookRun(input) => {
self.ingest_hook_run(input, out);
}
CustomAnalyticsFact::PluginUsed(input) => {
self.ingest_plugin_used(input, out);
}
@@ -442,6 +448,14 @@ impl AnalyticsReducer {
}));
}
fn ingest_hook_run(&mut self, input: HookRunInput, out: &mut Vec<TrackEventRequest>) {
let HookRunInput { tracking, hook } = input;
out.push(TrackEventRequest::HookRun(CodexHookRunEventRequest {
event_type: "codex_hook_run",
event_params: codex_hook_run_metadata(&tracking, hook),
}));
}
fn ingest_plugin_used(&mut self, input: PluginUsedInput, out: &mut Vec<TrackEventRequest>) {
let PluginUsedInput { tracking, plugin } = input;
out.push(TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
@@ -1534,6 +1534,14 @@
"scope": {
"$ref": "#/definitions/HookScope"
},
"source": {
"allOf": [
{
"$ref": "#/definitions/HookSource"
}
],
"default": "unknown"
},
"sourcePath": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -1572,6 +1580,19 @@
],
"type": "string"
},
"HookSource": {
"enum": [
"system",
"user",
"project",
"mdm",
"sessionFlags",
"legacyManagedConfigFile",
"legacyManagedConfigMdm",
"unknown"
],
"type": "string"
},
"HookStartedNotification": {
"properties": {
"run": {
@@ -8614,6 +8614,14 @@
"scope": {
"$ref": "#/definitions/v2/HookScope"
},
"source": {
"allOf": [
{
"$ref": "#/definitions/v2/HookSource"
}
],
"default": "unknown"
},
"sourcePath": {
"$ref": "#/definitions/v2/AbsolutePathBuf"
},
@@ -8652,6 +8660,19 @@
],
"type": "string"
},
"HookSource": {
"enum": [
"system",
"user",
"project",
"mdm",
"sessionFlags",
"legacyManagedConfigFile",
"legacyManagedConfigMdm",
"unknown"
],
"type": "string"
},
"HookStartedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
@@ -5342,6 +5342,14 @@
"scope": {
"$ref": "#/definitions/HookScope"
},
"source": {
"allOf": [
{
"$ref": "#/definitions/HookSource"
}
],
"default": "unknown"
},
"sourcePath": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -5380,6 +5388,19 @@
],
"type": "string"
},
"HookSource": {
"enum": [
"system",
"user",
"project",
"mdm",
"sessionFlags",
"legacyManagedConfigFile",
"legacyManagedConfigMdm",
"unknown"
],
"type": "string"
},
"HookStartedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
@@ -106,6 +106,14 @@
"scope": {
"$ref": "#/definitions/HookScope"
},
"source": {
"allOf": [
{
"$ref": "#/definitions/HookSource"
}
],
"default": "unknown"
},
"sourcePath": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -143,6 +151,19 @@
"turn"
],
"type": "string"
},
"HookSource": {
"enum": [
"system",
"user",
"project",
"mdm",
"sessionFlags",
"legacyManagedConfigFile",
"legacyManagedConfigMdm",
"unknown"
],
"type": "string"
}
},
"properties": {
@@ -106,6 +106,14 @@
"scope": {
"$ref": "#/definitions/HookScope"
},
"source": {
"allOf": [
{
"$ref": "#/definitions/HookSource"
}
],
"default": "unknown"
},
"sourcePath": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -143,6 +151,19 @@
"turn"
],
"type": "string"
},
"HookSource": {
"enum": [
"system",
"user",
"project",
"mdm",
"sessionFlags",
"legacyManagedConfigFile",
"legacyManagedConfigMdm",
"unknown"
],
"type": "string"
}
},
"properties": {
@@ -8,5 +8,6 @@ import type { HookHandlerType } from "./HookHandlerType";
import type { HookOutputEntry } from "./HookOutputEntry";
import type { HookRunStatus } from "./HookRunStatus";
import type { HookScope } from "./HookScope";
import type { HookSource } from "./HookSource";
export type HookRunSummary = { id: string, eventName: HookEventName, handlerType: HookHandlerType, executionMode: HookExecutionMode, scope: HookScope, sourcePath: AbsolutePathBuf, displayOrder: bigint, status: HookRunStatus, statusMessage: string | null, startedAt: bigint, completedAt: bigint | null, durationMs: bigint | null, entries: Array<HookOutputEntry>, };
export type HookRunSummary = { id: string, eventName: HookEventName, handlerType: HookHandlerType, executionMode: HookExecutionMode, scope: HookScope, sourcePath: AbsolutePathBuf, source: HookSource, displayOrder: bigint, status: HookRunStatus, statusMessage: string | null, startedAt: bigint, completedAt: bigint | null, durationMs: bigint | null, entries: Array<HookOutputEntry>, };
@@ -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 HookSource = "system" | "user" | "project" | "mdm" | "sessionFlags" | "legacyManagedConfigFile" | "legacyManagedConfigMdm" | "unknown";
@@ -140,6 +140,7 @@ export type { HookPromptFragment } from "./HookPromptFragment";
export type { HookRunStatus } from "./HookRunStatus";
export type { HookRunSummary } from "./HookRunSummary";
export type { HookScope } from "./HookScope";
export type { HookSource } from "./HookSource";
export type { HookStartedNotification } from "./HookStartedNotification";
export type { ItemCompletedNotification } from "./ItemCompletedNotification";
export type { ItemGuardianApprovalReviewCompletedNotification } from "./ItemGuardianApprovalReviewCompletedNotification";
@@ -123,6 +123,20 @@ impl ThreadHistoryBuilder {
.or_else(|| self.turns.last().cloned())
}
/// Returns the index of the active turn snapshot within the finished turn list.
///
/// When a turn is still open, this is the index it will occupy after
/// `finish`. When no turn is open, it is the index of the last finished turn.
pub fn active_turn_position(&self) -> Option<usize> {
if self.current_turn.is_some() {
Some(self.turns.len())
} else if self.turns.is_empty() {
None
} else {
Some(self.turns.len() - 1)
}
}
pub fn has_active_turn(&self) -> bool {
self.current_turn.is_some()
}
@@ -66,6 +66,7 @@ use codex_protocol::protocol::HookOutputEntryKind as CoreHookOutputEntryKind;
use codex_protocol::protocol::HookRunStatus as CoreHookRunStatus;
use codex_protocol::protocol::HookRunSummary as CoreHookRunSummary;
use codex_protocol::protocol::HookScope as CoreHookScope;
use codex_protocol::protocol::HookSource as CoreHookSource;
use codex_protocol::protocol::ModelRerouteReason as CoreModelRerouteReason;
use codex_protocol::protocol::NetworkAccess as CoreNetworkAccess;
use codex_protocol::protocol::NonSteerableTurnKind as CoreNonSteerableTurnKind;
@@ -435,6 +436,23 @@ v2_enum_from_core!(
}
);
v2_enum_from_core!(
pub enum HookSource from CoreHookSource {
System,
User,
Project,
Mdm,
SessionFlags,
LegacyManagedConfigFile,
LegacyManagedConfigMdm,
Unknown,
}
);
fn default_hook_source() -> HookSource {
HookSource::Unknown
}
v2_enum_from_core!(
pub enum HookRunStatus from CoreHookRunStatus {
Running, Completed, Failed, Blocked, Stopped
@@ -482,6 +500,8 @@ pub struct HookRunSummary {
pub execution_mode: HookExecutionMode,
pub scope: HookScope,
pub source_path: AbsolutePathBuf,
#[serde(default = "default_hook_source")]
pub source: HookSource,
pub display_order: i64,
pub status: HookRunStatus,
pub status_message: Option<String>,
@@ -500,6 +520,7 @@ impl From<CoreHookRunSummary> for HookRunSummary {
execution_mode: value.execution_mode.into(),
scope: value.scope.into(),
source_path: value.source_path,
source: value.source.into(),
display_order: value.display_order,
status: value.status.into(),
status_message: value.status_message,
+2 -2
View File
@@ -250,7 +250,7 @@ Start a fresh thread when you need a new Codex conversation.
Valid `personality` values are `"friendly"`, `"pragmatic"`, and `"none"`. When `"none"` is selected, the personality placeholder is replaced with an empty string.
To continue a stored session, call `thread/resume` with the `thread.id` you previously recorded. The response shape matches `thread/start`, and no additional notifications are emitted. You can also pass the same configuration overrides supported by `thread/start`, including `approvalsReviewer`.
To continue a stored session, call `thread/resume` with the `thread.id` you previously recorded. The response shape matches `thread/start`. When the stored session includes persisted token usage, the server emits `thread/tokenUsage/updated` immediately after the response so clients can render restored usage before the next turn starts. You can also pass the same configuration overrides supported by `thread/start`, including `approvalsReviewer`.
By default, resume uses the latest persisted `model` and `reasoningEffort` values associated with the thread. Supplying any of `model`, `modelProvider`, `config.model`, or `config.model_reasoning_effort` disables that persisted fallback and uses the explicit overrides plus normal config resolution instead.
@@ -264,7 +264,7 @@ Example:
{ "id": 11, "result": { "thread": { "id": "thr_123", … } } }
```
To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. If the source thread is actively running, the fork snapshots it as if the current turn had been interrupted first. Pass `ephemeral: true` when the fork should stay in-memory only:
To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. When the source history includes persisted token usage, the server also emits `thread/tokenUsage/updated` for the new thread immediately after the response. If the source thread is actively running, the fork snapshots it as if the current turn had been interrupted first. Pass `ephemeral: true` when the fork should stay in-memory only:
```json
{ "method": "thread/fork", "id": 12, "params": { "threadId": "thr_123", "ephemeral": true } }
@@ -363,12 +363,17 @@ use codex_app_server_protocol::ServerRequest;
mod apps_list_helpers;
mod plugin_app_helpers;
mod plugin_mcp_oauth;
mod token_usage_replay;
use crate::filters::compute_source_filters;
use crate::filters::source_kind_matches;
use crate::thread_state::ThreadListenerCommand;
use crate::thread_state::ThreadState;
use crate::thread_state::ThreadStateManager;
use token_usage_replay::latest_token_usage_turn_id_for_thread_path;
use token_usage_replay::latest_token_usage_turn_id_from_rollout_items;
use token_usage_replay::latest_token_usage_turn_id_from_rollout_path;
use token_usage_replay::send_thread_token_usage_update_to_connection;
const THREAD_LIST_DEFAULT_LIMIT: usize = 25;
const THREAD_LIST_MAX_LIMIT: usize = 100;
@@ -4103,7 +4108,7 @@ impl CodexMessageProcessor {
{
Ok(NewThread {
thread_id,
thread,
thread: codex_thread,
session_configured,
}) => {
let SessionConfiguredEvent { rollout_path, .. } = session_configured;
@@ -4132,7 +4137,7 @@ impl CodexMessageProcessor {
let mut thread = match self
.load_thread_from_resume_source_or_send_internal(
thread_id,
thread.as_ref(),
codex_thread.as_ref(),
&response_history,
rollout_path.as_path(),
fallback_model_provider.as_str(),
@@ -4184,7 +4189,25 @@ impl CodexMessageProcessor {
);
}
let connection_id = request_id.connection_id;
let token_usage_thread = response.thread.clone();
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
&response_history.get_rollout_items(),
&token_usage_thread,
);
self.outgoing.send_response(request_id, response).await;
// The client needs restored usage before it starts another turn.
// Sending after the response preserves JSON-RPC request ordering while
// still filling the status line before the next turn lifecycle begins.
send_thread_token_usage_update_to_connection(
&self.outgoing,
connection_id,
thread_id,
&token_usage_thread,
codex_thread.as_ref(),
token_usage_turn_id,
)
.await;
}
Err(err) => {
let error = JSONRPCErrorError {
@@ -4820,7 +4843,31 @@ impl CodexMessageProcessor {
);
}
let connection_id = request_id.connection_id;
let token_usage_thread = response.thread.clone();
let token_usage_turn_id = if let Some(turn_id) =
latest_token_usage_turn_id_for_thread_path(&token_usage_thread).await
{
Some(turn_id)
} else {
latest_token_usage_turn_id_from_rollout_path(
rollout_path.as_path(),
&token_usage_thread,
)
.await
};
self.outgoing.send_response(request_id, response).await;
// Mirror the resume contract for forks: the new thread is usable as soon
// as the response arrives, so restored usage must follow immediately.
send_thread_token_usage_update_to_connection(
&self.outgoing,
connection_id,
thread_id,
&token_usage_thread,
forked_thread.as_ref(),
token_usage_turn_id,
)
.await;
let notif = ThreadStartedNotification { thread };
self.outgoing
@@ -8525,7 +8572,24 @@ async fn handle_pending_thread_resume_request(
sandbox: sandbox_policy.into(),
reasoning_effort,
};
let token_usage_thread = response.thread.clone();
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_path(
pending.rollout_path.as_path(),
&token_usage_thread,
)
.await;
outgoing.send_response(request_id, response).await;
// Rejoining a loaded thread has the same UI contract as a cold resume, but
// uses the live conversation state instead of reconstructing a new session.
send_thread_token_usage_update_to_connection(
outgoing,
connection_id,
conversation_id,
&token_usage_thread,
conversation.as_ref(),
token_usage_turn_id,
)
.await;
outgoing
.replay_requests_to_connection_for_thread(connection_id, conversation_id)
.await;
@@ -0,0 +1,133 @@
//! Replays persisted token usage snapshots when a client attaches to an existing thread.
//!
//! The message processor decides when replay is allowed and preserves JSON-RPC response
//! ordering. This module owns notification construction and the attribution rules that
//! map the latest persisted `TokenCount` back to a v2 turn id.
//!
//! Rollout histories can contain explicit turn ids or generated turn ids. When explicit
//! ids do not match the rebuilt thread, replay falls back to the active turn position at
//! the time the `TokenCount` was persisted so the notification still targets the
//! corresponding rebuilt turn.
use std::path::Path;
use std::sync::Arc;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadHistoryBuilder;
use codex_app_server_protocol::ThreadTokenUsage;
use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification;
use codex_app_server_protocol::TurnStatus;
use codex_core::CodexThread;
use codex_protocol::ThreadId;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use crate::codex_message_processor::read_rollout_items_from_rollout;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::OutgoingMessageSender;
/// Sends a restored token usage update to the connection that attached to a thread.
///
/// This is lifecycle replay rather than a model event: the rollout already contains
/// the original `TokenCount`, and emitting through `send_event` here would duplicate
/// persisted usage records. Keeping this helper connection-scoped also avoids
/// surprising other subscribers with a historical usage update while they may be
/// rendering live turn events.
pub(super) async fn send_thread_token_usage_update_to_connection(
outgoing: &Arc<OutgoingMessageSender>,
connection_id: ConnectionId,
thread_id: ThreadId,
thread: &Thread,
conversation: &CodexThread,
token_usage_turn_id: Option<String>,
) {
let Some(info) = conversation.token_usage_info().await else {
return;
};
let notification = ThreadTokenUsageUpdatedNotification {
thread_id: thread_id.to_string(),
turn_id: token_usage_turn_id.unwrap_or_else(|| latest_token_usage_turn_id(thread)),
token_usage: ThreadTokenUsage::from(info),
};
outgoing
.send_server_notification_to_connections(
&[connection_id],
ServerNotification::ThreadTokenUsageUpdated(notification),
)
.await;
}
pub(super) async fn latest_token_usage_turn_id_for_thread_path(thread: &Thread) -> Option<String> {
let rollout_path = thread.path.as_deref()?;
latest_token_usage_turn_id_from_rollout_path(rollout_path, thread).await
}
pub(super) async fn latest_token_usage_turn_id_from_rollout_path(
rollout_path: &Path,
thread: &Thread,
) -> Option<String> {
let rollout_items = read_rollout_items_from_rollout(rollout_path).await.ok()?;
latest_token_usage_turn_id_from_rollout_items(&rollout_items, thread)
}
/// Identifies the turn that was active when a `TokenCount` record appeared.
///
/// The id is preferred when it still appears in the rebuilt thread. The position is a
/// fallback for histories whose implicit turn ids are regenerated during reconstruction.
struct TokenUsageTurnOwner {
id: String,
position: Option<usize>,
}
pub(super) fn latest_token_usage_turn_id_from_rollout_items(
rollout_items: &[RolloutItem],
thread: &Thread,
) -> Option<String> {
let owner = latest_token_usage_turn_owner_from_rollout_items(rollout_items)?;
if thread.turns.iter().any(|turn| turn.id == owner.id) {
return Some(owner.id);
}
owner
.position
.and_then(|position| thread.turns.get(position))
.map(|turn| turn.id.clone())
}
fn latest_token_usage_turn_owner_from_rollout_items(
rollout_items: &[RolloutItem],
) -> Option<TokenUsageTurnOwner> {
let mut builder = ThreadHistoryBuilder::new();
let mut token_usage_turn_owner = None;
for item in rollout_items {
if matches!(item, RolloutItem::EventMsg(EventMsg::TokenCount(_))) {
token_usage_turn_owner =
builder
.active_turn_snapshot()
.map(|turn| TokenUsageTurnOwner {
id: turn.id,
position: builder.active_turn_position(),
});
}
builder.handle_rollout_item(item);
}
token_usage_turn_owner
}
/// Chooses a fallback turn id that should own a replayed token usage update.
///
/// Normal replay derives the owner from the rollout position of the latest
/// `TokenCount` event. This fallback only preserves a stable wire shape for
/// unusual histories where that rollout information cannot be read.
fn latest_token_usage_turn_id(thread: &Thread) -> String {
thread
.turns
.iter()
.rev()
.find(|turn| matches!(turn.status, TurnStatus::Completed | TurnStatus::Failed))
.or_else(|| thread.turns.last())
.map(|turn| turn.id.clone())
.unwrap_or_default()
}
+1
View File
@@ -40,6 +40,7 @@ pub use responses::create_shell_command_sse_response;
pub use rollout::create_fake_rollout;
pub use rollout::create_fake_rollout_with_source;
pub use rollout::create_fake_rollout_with_text_elements;
pub use rollout::create_fake_rollout_with_token_usage;
pub use rollout::rollout_path;
use serde::de::DeserializeOwned;
@@ -1,9 +1,13 @@
use anyhow::Result;
use codex_protocol::ThreadId;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::GitInfo;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::TokenCountEvent;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use serde_json::json;
use std::fs;
use std::fs::FileTimes;
@@ -50,6 +54,61 @@ pub fn create_fake_rollout(
)
}
/// Creates a minimal rollout whose history includes a persisted token usage event.
///
/// Resume and fork tests use this fixture to verify lifecycle replay of restored
/// usage without starting a model turn. The exact token values are intentionally
/// non-zero and asymmetric so assertions catch swapped total/last fields and
/// dropped cached or reasoning counters.
pub fn create_fake_rollout_with_token_usage(
codex_home: &Path,
filename_ts: &str,
meta_rfc3339: &str,
preview: &str,
model_provider: Option<&str>,
) -> Result<String> {
let thread_id = create_fake_rollout(
codex_home,
filename_ts,
meta_rfc3339,
preview,
model_provider,
/*git_info*/ None,
)?;
let payload = serde_json::to_value(EventMsg::TokenCount(TokenCountEvent {
info: Some(TokenUsageInfo {
total_token_usage: TokenUsage {
input_tokens: 120,
cached_input_tokens: 20,
output_tokens: 30,
reasoning_output_tokens: 10,
total_tokens: 150,
},
last_token_usage: TokenUsage {
input_tokens: 70,
cached_input_tokens: 10,
output_tokens: 20,
reasoning_output_tokens: 5,
total_tokens: 90,
},
model_context_window: Some(200_000),
}),
rate_limits: None,
}))?;
let file_path = rollout_path(codex_home, filename_ts, &thread_id);
let line = json!({
"timestamp": meta_rfc3339,
"type": "event_msg",
"payload": payload
})
.to_string();
fs::write(
&file_path,
format!("{}{}\n", fs::read_to_string(&file_path)?, line),
)?;
Ok(thread_id)
}
/// Create a minimal rollout file with an explicit session source.
pub fn create_fake_rollout_with_source(
codex_home: &Path,
@@ -2,6 +2,7 @@ use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::McpProcess;
use app_test_support::create_fake_rollout;
use app_test_support::create_fake_rollout_with_token_usage;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::to_response;
use app_test_support::write_chatgpt_auth;
@@ -9,6 +10,7 @@ use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
@@ -186,6 +188,59 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_fork_emits_restored_token_usage_before_next_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let conversation_id = create_fake_rollout_with_token_usage(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::ThreadTokenUsageUpdated(notification) = parsed else {
panic!("expected thread/tokenUsage/updated notification");
};
assert_eq!(notification.thread_id, thread.id);
assert_eq!(notification.turn_id, thread.turns[0].id);
assert_eq!(notification.token_usage.total.total_tokens, 150);
assert_eq!(notification.token_usage.total.input_tokens, 120);
assert_eq!(notification.token_usage.total.cached_input_tokens, 20);
assert_eq!(notification.token_usage.total.output_tokens, 30);
assert_eq!(notification.token_usage.total.reasoning_output_tokens, 10);
assert_eq!(notification.token_usage.last.total_tokens, 90);
assert_eq!(notification.token_usage.model_context_window, Some(200_000));
Ok(())
}
#[tokio::test]
async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -3,6 +3,7 @@ use app_test_support::ChatGptAuthFixture;
use app_test_support::McpProcess;
use app_test_support::create_apply_patch_sse_response;
use app_test_support::create_fake_rollout_with_text_elements;
use app_test_support::create_fake_rollout_with_token_usage;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::create_mock_responses_server_sequence_unchecked;
@@ -23,6 +24,7 @@ use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::PatchChangeKind;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::ThreadItem;
@@ -50,6 +52,11 @@ use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource as RolloutSessionSource;
use codex_protocol::protocol::TokenCountEvent;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::user_input::ByteRange;
use codex_protocol::user_input::TextElement;
@@ -278,6 +285,268 @@ async fn thread_resume_returns_rollout_history() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_resume_emits_restored_token_usage_before_next_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let conversation_id = create_fake_rollout_with_token_usage(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::ThreadTokenUsageUpdated(notification) = parsed else {
panic!("expected thread/tokenUsage/updated notification");
};
assert_eq!(notification.thread_id, thread.id);
assert_eq!(notification.turn_id, thread.turns[0].id);
assert_eq!(notification.token_usage.total.total_tokens, 150);
assert_eq!(notification.token_usage.total.input_tokens, 120);
assert_eq!(notification.token_usage.total.cached_input_tokens, 20);
assert_eq!(notification.token_usage.total.output_tokens, 30);
assert_eq!(notification.token_usage.total.reasoning_output_tokens, 10);
assert_eq!(notification.token_usage.last.total_tokens, 90);
assert_eq!(notification.token_usage.model_context_window, Some(200_000));
Ok(())
}
#[tokio::test]
async fn thread_resume_token_usage_replay_ignores_stale_interrupted_tail_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let filename_ts = "2025-01-05T12-00-00";
let meta_rfc3339 = "2025-01-05T12:00:00Z";
let conversation_id = create_fake_rollout_with_token_usage(
codex_home.path(),
filename_ts,
meta_rfc3339,
"Saved user message",
Some("mock_provider"),
)?;
let rollout_file_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
let persisted_rollout = std::fs::read_to_string(&rollout_file_path)?;
let stale_turn_id = "incomplete-turn-after-token-usage";
let appended_rollout = [
json!({
"timestamp": meta_rfc3339,
"type": "event_msg",
"payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: stale_turn_id.to_string(),
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
}))?,
})
.to_string(),
json!({
"timestamp": meta_rfc3339,
"type": "event_msg",
"payload": serde_json::to_value(EventMsg::AgentMessage(AgentMessageEvent {
message: "Still running".to_string(),
phase: None,
memory_citation: None,
}))?,
})
.to_string(),
]
.join("\n");
std::fs::write(
&rollout_file_path,
format!("{persisted_rollout}{appended_rollout}\n"),
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(thread.turns.len(), 2);
assert_eq!(thread.turns[0].status, TurnStatus::Completed);
assert_eq!(thread.turns[1].id, stale_turn_id);
assert_eq!(thread.turns[1].status, TurnStatus::Interrupted);
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::ThreadTokenUsageUpdated(notification) = parsed else {
panic!("expected thread/tokenUsage/updated notification");
};
assert_eq!(notification.thread_id, thread.id);
assert_eq!(notification.turn_id, thread.turns[0].id);
assert_ne!(notification.turn_id, stale_turn_id);
assert_eq!(notification.token_usage.total.total_tokens, 150);
assert_eq!(notification.token_usage.last.total_tokens, 90);
Ok(())
}
#[tokio::test]
async fn thread_resume_token_usage_replay_can_belong_to_interrupted_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let filename_ts = "2025-01-05T12-00-00";
let meta_rfc3339 = "2025-01-05T12:00:00Z";
let conversation_id = create_fake_rollout_with_token_usage(
codex_home.path(),
filename_ts,
meta_rfc3339,
"Saved user message",
Some("mock_provider"),
)?;
let rollout_file_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
let persisted_rollout = std::fs::read_to_string(&rollout_file_path)?;
let interrupted_turn_id = "interrupted-turn-with-token-usage";
let appended_rollout = [
json!({
"timestamp": meta_rfc3339,
"type": "event_msg",
"payload": serde_json::to_value(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: interrupted_turn_id.to_string(),
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
}))?,
})
.to_string(),
json!({
"timestamp": meta_rfc3339,
"type": "event_msg",
"payload": serde_json::to_value(EventMsg::AgentMessage(AgentMessageEvent {
message: "Interrupted after usage".to_string(),
phase: None,
memory_citation: None,
}))?,
})
.to_string(),
json!({
"timestamp": meta_rfc3339,
"type": "event_msg",
"payload": serde_json::to_value(EventMsg::TokenCount(TokenCountEvent {
info: Some(TokenUsageInfo {
total_token_usage: TokenUsage {
input_tokens: 180,
cached_input_tokens: 40,
output_tokens: 50,
reasoning_output_tokens: 15,
total_tokens: 230,
},
last_token_usage: TokenUsage {
input_tokens: 90,
cached_input_tokens: 30,
output_tokens: 40,
reasoning_output_tokens: 12,
total_tokens: 130,
},
model_context_window: Some(200_000),
}),
rate_limits: None,
}))?,
})
.to_string(),
json!({
"timestamp": meta_rfc3339,
"type": "event_msg",
"payload": serde_json::to_value(EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: Some(interrupted_turn_id.to_string()),
reason: TurnAbortReason::Interrupted,
completed_at: None,
duration_ms: None,
}))?,
})
.to_string(),
]
.join("\n");
std::fs::write(
&rollout_file_path,
format!("{persisted_rollout}{appended_rollout}\n"),
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(thread.turns.len(), 2);
assert_eq!(thread.turns[0].status, TurnStatus::Completed);
assert_eq!(thread.turns[1].id, interrupted_turn_id);
assert_eq!(thread.turns[1].status, TurnStatus::Interrupted);
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::ThreadTokenUsageUpdated(notification) = parsed else {
panic!("expected thread/tokenUsage/updated notification");
};
assert_eq!(notification.thread_id, thread.id);
assert_eq!(notification.turn_id, interrupted_turn_id);
assert_eq!(notification.token_usage.total.total_tokens, 230);
assert_eq!(notification.token_usage.last.total_tokens, 130);
Ok(())
}
#[tokio::test]
async fn thread_resume_prefers_persisted_git_metadata_for_local_threads() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -1,14 +1,15 @@
use anyhow::Context;
use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use app_test_support::create_shell_command_sse_response;
use app_test_support::to_response;
use codex_app_server_protocol::DynamicToolCallOutputContentItem;
use codex_app_server_protocol::DynamicToolCallParams;
use codex_app_server_protocol::DynamicToolCallResponse;
use codex_app_server_protocol::DynamicToolSpec;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadLoadedListParams;
use codex_app_server_protocol::ThreadLoadedListResponse;
@@ -26,61 +27,14 @@ use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput as V2UserInput;
use core_test_support::responses;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
// Windows CI can spend most of the default read timeout launching PowerShell
// before the command finishes and the follow-up model request is sent.
const TURN_COMPLETION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
async fn wait_for_responses_request_count_to_stabilize(
server: &wiremock::MockServer,
expected_count: usize,
settle_duration: std::time::Duration,
timeout_duration: std::time::Duration,
) -> Result<()> {
timeout(timeout_duration, async {
let mut stable_since: Option<tokio::time::Instant> = None;
loop {
let requests = server
.received_requests()
.await
.context("failed to fetch received requests")?;
let responses_request_count = requests
.iter()
.filter(|request| {
request.method == "POST" && request.url.path().ends_with("/responses")
})
.count();
if responses_request_count > expected_count {
anyhow::bail!(
"expected exactly {expected_count} /responses requests, got {responses_request_count}"
);
}
if responses_request_count == expected_count {
match stable_since {
Some(stable_since) if stable_since.elapsed() >= settle_duration => {
return Ok::<(), anyhow::Error>(());
}
None => stable_since = Some(tokio::time::Instant::now()),
Some(_) => {}
}
} else {
stable_since = None;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await??;
Ok(())
}
#[tokio::test]
async fn thread_unsubscribe_keeps_thread_loaded_until_idle_timeout() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -132,14 +86,10 @@ async fn thread_unsubscribe_keeps_thread_loaded_until_idle_timeout() -> Result<(
#[tokio::test]
async fn thread_unsubscribe_during_turn_keeps_turn_running() -> Result<()> {
#[cfg(target_os = "windows")]
let shell_command = vec![
"Start-Sleep".to_string(),
"-Seconds".to_string(),
"1".to_string(),
];
#[cfg(not(target_os = "windows"))]
let shell_command = vec!["sleep".to_string(), "1".to_string()];
let call_id = "deterministic-wait-call";
let tool_name = "deterministic_wait";
let tool_args = json!({});
let tool_call_arguments = serde_json::to_string(&tool_args)?;
let tmp = TempDir::new()?;
let codex_home = tmp.path().join("codex_home");
@@ -147,28 +97,61 @@ async fn thread_unsubscribe_during_turn_keeps_turn_running() -> Result<()> {
let working_directory = tmp.path().join("workdir");
std::fs::create_dir(&working_directory)?;
let server = create_mock_responses_server_sequence_unchecked(vec![
create_shell_command_sse_response(
shell_command.clone(),
Some(&working_directory),
Some(10_000),
"call_sleep",
)?,
create_final_assistant_message_sse_response("Done")?,
let (server, mut completions) = start_streaming_sse_server(vec![
vec![StreamingSseChunk {
gate: None,
body: responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_function_call(call_id, tool_name, &tool_call_arguments),
responses::ev_completed("resp-1"),
]),
}],
vec![StreamingSseChunk {
gate: None,
body: responses::sse(vec![
responses::ev_response_created("resp-2"),
responses::ev_assistant_message("msg-1", "Done"),
responses::ev_completed("resp-2"),
]),
}],
])
.await;
create_config_toml(&codex_home, &server.uri())?;
let first_response_completed = completions.remove(0);
let final_response_completed = completions.remove(0);
create_config_toml(&codex_home, server.uri())?;
let mut mcp = McpProcess::new(&codex_home).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let thread_id = start_thread(&mut mcp).await?;
let thread_req = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
dynamic_tools: Some(vec![DynamicToolSpec {
name: tool_name.to_string(),
description: "Deterministic wait tool".to_string(),
input_schema: json!({
"type": "object",
"properties": {},
"additionalProperties": false,
}),
defer_loading: false,
}]),
..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 turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread_id.clone(),
input: vec![V2UserInput::Text {
text: "run sleep".to_string(),
text: "run deterministic tool".to_string(),
text_elements: Vec::new(),
}],
cwd: Some(working_directory),
@@ -184,9 +167,37 @@ async fn thread_unsubscribe_during_turn_keeps_turn_running() -> Result<()> {
timeout(
DEFAULT_READ_TIMEOUT,
wait_for_command_execution_item_started(&mut mcp),
server.wait_for_request_count(/*count*/ 1),
)
.await?;
timeout(DEFAULT_READ_TIMEOUT, first_response_completed).await??;
let started = timeout(
DEFAULT_READ_TIMEOUT,
wait_for_dynamic_tool_started(&mut mcp, call_id),
)
.await??;
assert_eq!(started.thread_id, thread_id);
let request = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_request_message(),
)
.await??;
let (request_id, params) = match request {
ServerRequest::DynamicToolCall { request_id, params } => (request_id, params),
other => panic!("expected DynamicToolCall request, got {other:?}"),
};
assert_eq!(
params,
DynamicToolCallParams {
thread_id: thread_id.clone(),
turn_id: started.turn_id,
call_id: call_id.to_string(),
tool: tool_name.to_string(),
arguments: tool_args,
}
);
let unsubscribe_id = mcp
.send_thread_unsubscribe_request(ThreadUnsubscribeParams {
@@ -201,22 +212,29 @@ async fn thread_unsubscribe_during_turn_keeps_turn_running() -> Result<()> {
let unsubscribe = to_response::<ThreadUnsubscribeResponse>(unsubscribe_resp)?;
assert_eq!(unsubscribe.status, ThreadUnsubscribeStatus::Unsubscribed);
assert!(
timeout(
std::time::Duration::from_millis(250),
mcp.read_stream_until_notification_message("thread/closed"),
)
.await
.is_err()
let closed_while_tool_call_blocked = timeout(
std::time::Duration::from_millis(250),
mcp.read_stream_until_notification_message("thread/closed"),
);
let closed_while_tool_call_blocked = closed_while_tool_call_blocked.await;
assert!(closed_while_tool_call_blocked.is_err());
wait_for_responses_request_count_to_stabilize(
&server,
/*expected_count*/ 2,
std::time::Duration::from_millis(200),
TURN_COMPLETION_TIMEOUT,
let response = DynamicToolCallResponse {
content_items: vec![DynamicToolCallOutputContentItem::InputText {
text: "dynamic-ok".to_string(),
}],
success: true,
};
mcp.send_response(request_id, serde_json::to_value(response)?)
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
server.wait_for_request_count(/*count*/ 2),
)
.await?;
timeout(DEFAULT_READ_TIMEOUT, final_response_completed).await??;
server.shutdown().await;
Ok(())
}
@@ -355,15 +373,20 @@ async fn thread_unsubscribe_reports_not_subscribed_before_idle_unload() -> Resul
Ok(())
}
async fn wait_for_command_execution_item_started(mcp: &mut McpProcess) -> Result<()> {
async fn wait_for_dynamic_tool_started(
mcp: &mut McpProcess,
call_id: &str,
) -> Result<ItemStartedNotification> {
loop {
let started_notif = mcp
let notification = mcp
.read_stream_until_notification_message("item/started")
.await?;
let started_params = started_notif.params.context("item/started params")?;
let started: ItemStartedNotification = serde_json::from_value(started_params)?;
if let ThreadItem::CommandExecution { .. } = started.item {
return Ok(());
let Some(params) = notification.params else {
continue;
};
let started: ItemStartedNotification = serde_json::from_value(params)?;
if matches!(&started.item, ThreadItem::DynamicToolCall { id, .. } if id == call_id) {
return Ok(started);
}
}
}
+2
View File
@@ -86,3 +86,5 @@ pub use state::ConfigLayerEntry;
pub use state::ConfigLayerStack;
pub use state::ConfigLayerStackOrdering;
pub use state::LoaderOverrides;
pub use codex_app_server_protocol::ConfigLayerSource;
+23 -10
View File
@@ -25,6 +25,7 @@ use crate::compact_remote::run_remote_prefix_compact_task;
use crate::config::ManagedFeatures;
use crate::connectors;
use crate::exec_policy::ExecPolicyManager;
use crate::hook_runtime::emit_hook_completed_events;
use crate::installation_id::resolve_installation_id;
use crate::mcp_tool_exposure::build_mcp_tool_exposure;
use crate::parse_turn_item;
@@ -2423,6 +2424,17 @@ impl Session {
state.token_info().map(|info| info.total_token_usage)
}
/// Returns the complete token usage snapshot currently cached for this session.
///
/// Resume and fork reconstruction seed this state from the last persisted rollout
/// `TokenCount` event. Callers that need to replay restored usage to a client
/// should use this accessor instead of `total_token_usage`, because the app-server
/// notification includes both total and last-turn usage.
pub(crate) async fn token_usage_info(&self) -> Option<TokenUsageInfo> {
let state = self.state.lock().await;
state.token_info()
}
pub(crate) async fn get_estimated_token_count(
&self,
turn_context: &TurnContext,
@@ -6651,10 +6663,8 @@ pub(crate) async fn run_turn(
.await;
}
let stop_outcome = sess.hooks().run_stop(stop_request).await;
for completed in stop_outcome.hook_events {
sess.send_event(&turn_context, EventMsg::HookCompleted(completed))
.await;
}
emit_hook_completed_events(&sess, &turn_context, stop_outcome.hook_events)
.await;
if stop_outcome.should_block {
if let Some(hook_prompt_message) =
build_hook_prompt_message(&stop_outcome.continuation_fragments)
@@ -6746,13 +6756,16 @@ pub(crate) async fn run_turn(
break;
}
Err(CodexErr::InvalidImageRequest()) => {
let mut state = sess.state.lock().await;
error_or_panic(
"Invalid image detected; sanitizing tool output to prevent poisoning",
);
if state.history.replace_last_turn_images("Invalid image") {
continue;
{
let mut state = sess.state.lock().await;
error_or_panic(
"Invalid image detected; sanitizing tool output to prevent poisoning",
);
if state.history.replace_last_turn_images("Invalid image") {
continue;
}
}
let event = EventMsg::Error(ErrorEvent {
message: "Invalid image in your last message. Please remove it and try again."
.to_string(),
+12
View File
@@ -22,6 +22,7 @@ use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::Submission;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::user_input::UserInput;
use codex_utils_absolute_path::AbsolutePathBuf;
@@ -144,6 +145,17 @@ impl CodexThread {
self.codex.session.total_token_usage().await
}
/// Returns the complete token usage snapshot currently cached for this thread.
///
/// This accessor is intentionally narrower than direct session access: it lets
/// app-server lifecycle paths replay restored usage after resume or fork without
/// exposing broader session mutation authority. A caller that only reads
/// `total_token_usage` would drop last-turn usage and make the v2
/// `thread/tokenUsage/updated` payload incomplete.
pub async fn token_usage_info(&self) -> Option<TokenUsageInfo> {
self.codex.session.token_usage_info().await
}
/// Records a user-role session-prefix message without creating a new user turn boundary.
pub(crate) async fn inject_user_message_without_turn(&self, message: String) {
let message = ResponseItem::Message {
+104 -1
View File
@@ -1,6 +1,8 @@
use std::future::Future;
use std::sync::Arc;
use codex_analytics::HookRunFact;
use codex_analytics::build_track_events_context;
use codex_hooks::PostToolUseOutcome;
use codex_hooks::PostToolUseRequest;
use codex_hooks::PreToolUseOutcome;
@@ -316,17 +318,52 @@ async fn emit_hook_started_events(
}
}
async fn emit_hook_completed_events(
pub(crate) async fn emit_hook_completed_events(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
completed_events: Vec<HookCompletedEvent>,
) {
for completed in completed_events {
track_hook_completed_analytics(sess, turn_context, &completed);
sess.send_event(turn_context, EventMsg::HookCompleted(completed))
.await;
}
}
fn track_hook_completed_analytics(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
completed: &HookCompletedEvent,
) {
let (tracking, hook) =
hook_run_analytics_payload(sess.conversation_id.to_string(), turn_context, completed);
sess.services
.analytics_events_client
.track_hook_run(tracking, hook);
}
fn hook_run_analytics_payload(
thread_id: String,
turn_context: &TurnContext,
completed: &HookCompletedEvent,
) -> (codex_analytics::TrackEventsContext, HookRunFact) {
(
build_track_events_context(
turn_context.model_info.slug.clone(),
thread_id,
completed
.turn_id
.clone()
.unwrap_or_else(|| turn_context.sub_id.clone()),
),
HookRunFact {
event_name: completed.run.event_name,
hook_source: completed.run.source,
status: completed.run.status,
},
)
}
fn hook_permission_mode(turn_context: &TurnContext) -> String {
match turn_context.approval_policy.value() {
AskForApproval::Never => "bypassPermissions",
@@ -341,9 +378,21 @@ fn hook_permission_mode(turn_context: &TurnContext) -> String {
#[cfg(test)]
mod tests {
use codex_protocol::models::ContentItem;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookExecutionMode;
use codex_protocol::protocol::HookHandlerType;
use codex_protocol::protocol::HookRunStatus;
use codex_protocol::protocol::HookScope;
use codex_protocol::protocol::HookSource;
use pretty_assertions::assert_eq;
use super::additional_context_messages;
use super::hook_run_analytics_payload;
use crate::codex::make_session_and_context;
use codex_protocol::protocol::HookCompletedEvent;
use codex_protocol::protocol::HookRunSummary;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
#[test]
fn additional_context_messages_stay_separate_and_ordered() {
@@ -378,4 +427,58 @@ mod tests {
],
);
}
#[tokio::test]
async fn hook_run_analytics_payload_uses_completed_turn_id() {
let (_session, turn_context) = make_session_and_context().await;
let completed = HookCompletedEvent {
turn_id: Some("turn-from-hook".to_string()),
run: sample_hook_run(HookRunStatus::Blocked, HookSource::Project),
};
let (tracking, hook) =
hook_run_analytics_payload("thread-123".to_string(), &turn_context, &completed);
assert_eq!(tracking.thread_id, "thread-123");
assert_eq!(tracking.turn_id, "turn-from-hook");
assert_eq!(tracking.model_slug, turn_context.model_info.slug);
assert_eq!(hook.event_name, HookEventName::Stop);
assert_eq!(hook.hook_source, HookSource::Project);
assert_eq!(hook.status, HookRunStatus::Blocked);
}
#[tokio::test]
async fn hook_run_analytics_payload_falls_back_to_turn_context_id() {
let (_session, turn_context) = make_session_and_context().await;
let completed = HookCompletedEvent {
turn_id: None,
run: sample_hook_run(HookRunStatus::Failed, HookSource::Unknown),
};
let (tracking, hook) =
hook_run_analytics_payload("thread-123".to_string(), &turn_context, &completed);
assert_eq!(tracking.turn_id, turn_context.sub_id);
assert_eq!(hook.hook_source, HookSource::Unknown);
assert_eq!(hook.status, HookRunStatus::Failed);
}
fn sample_hook_run(status: HookRunStatus, source: HookSource) -> HookRunSummary {
HookRunSummary {
id: "stop:0:/tmp/hooks.json".to_string(),
event_name: HookEventName::Stop,
handler_type: HookHandlerType::Command,
execution_mode: HookExecutionMode::Sync,
scope: HookScope::Turn,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source,
display_order: 0,
status,
status_message: None,
started_at: 10,
completed_at: Some(37),
duration_ms: Some(27),
entries: Vec::new(),
}
}
}
@@ -7,6 +7,7 @@ use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::sync::Mutex as TokioMutex;
use tokio::sync::Notify;
use tokio::sync::oneshot;
/// Streaming SSE chunk payload gated by a per-chunk signal.
@@ -20,6 +21,7 @@ pub struct StreamingSseChunk {
pub struct StreamingSseServer {
uri: String,
requests: Arc<TokioMutex<Vec<Vec<u8>>>>,
request_notify: Arc<Notify>,
shutdown: oneshot::Sender<()>,
task: tokio::task::JoinHandle<()>,
}
@@ -33,6 +35,15 @@ impl StreamingSseServer {
self.requests.lock().await.clone()
}
pub async fn wait_for_request_count(&self, count: usize) {
loop {
if self.requests.lock().await.len() >= count {
return;
}
self.request_notify.notified().await;
}
}
pub async fn shutdown(self) {
let _ = self.shutdown.send(());
let _ = self.task.await;
@@ -67,7 +78,9 @@ pub async fn start_streaming_sse_server(
completions: VecDeque::from(completion_senders),
}));
let requests = Arc::new(TokioMutex::new(Vec::new()));
let request_notify = Arc::new(Notify::new());
let requests_for_task = Arc::clone(&requests);
let request_notify_for_task = Arc::clone(&request_notify);
let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
let task = tokio::spawn(async move {
@@ -78,6 +91,7 @@ pub async fn start_streaming_sse_server(
let (mut stream, _) = accept_res.expect("accept streaming SSE connection");
let state = Arc::clone(&state);
let requests = Arc::clone(&requests_for_task);
let request_notify = Arc::clone(&request_notify_for_task);
tokio::spawn(async move {
let (request, body_prefix) = read_http_request(&mut stream).await;
let Some((method, path)) = parse_request_line(&request) else {
@@ -113,6 +127,7 @@ pub async fn start_streaming_sse_server(
}
};
requests.lock().await.push(body);
request_notify.notify_one();
let Some((chunks, completion)) = take_next_stream(&state).await else {
let _ = write_http_response(&mut stream, /*status*/ 500, "no responses queued", "text/plain").await;
return;
@@ -149,6 +164,7 @@ pub async fn start_streaming_sse_server(
StreamingSseServer {
uri,
requests,
request_notify,
shutdown: shutdown_tx,
task,
},
+156 -108
View File
@@ -9,6 +9,8 @@ use super::config::HooksFile;
use super::config::MatcherGroup;
use crate::events::common::matcher_pattern_for_event;
use crate::events::common::validate_matcher_pattern;
use codex_config::ConfigLayerSource;
use codex_protocol::protocol::HookSource;
pub(crate) struct DiscoveryResult {
pub handlers: Vec<ConfiguredHandler>,
@@ -93,6 +95,7 @@ pub(crate) fn discover_handlers(config_layer_stack: Option<&ConfigLayerStack>) -
&mut warnings,
&mut display_order,
&source_path,
hook_source_for_config_layer_source(&layer.name),
event_name,
groups,
);
@@ -102,95 +105,94 @@ pub(crate) fn discover_handlers(config_layer_stack: Option<&ConfigLayerStack>) -
DiscoveryResult { handlers, warnings }
}
fn append_group_handlers(
handlers: &mut Vec<ConfiguredHandler>,
warnings: &mut Vec<String>,
display_order: &mut i64,
source_path: &AbsolutePathBuf,
event_name: codex_protocol::protocol::HookEventName,
matcher: Option<&str>,
group_handlers: Vec<HookHandlerConfig>,
) {
if let Some(matcher) = matcher
&& let Err(err) = validate_matcher_pattern(matcher)
{
warnings.push(format!(
"invalid matcher {matcher:?} in {}: {err}",
source_path.display()
));
return;
}
for handler in group_handlers {
match handler {
HookHandlerConfig::Command {
command,
timeout_sec,
r#async,
status_message,
} => {
if r#async {
warnings.push(format!(
"skipping async hook in {}: async hooks are not supported yet",
source_path.display()
));
continue;
}
if command.trim().is_empty() {
warnings.push(format!(
"skipping empty hook command in {}",
source_path.display()
));
continue;
}
let timeout_sec = timeout_sec.unwrap_or(600).max(1);
handlers.push(ConfiguredHandler {
event_name,
matcher: matcher.map(ToOwned::to_owned),
command,
timeout_sec,
status_message,
source_path: source_path.clone(),
display_order: *display_order,
});
*display_order += 1;
}
HookHandlerConfig::Prompt {} => warnings.push(format!(
"skipping prompt hook in {}: prompt hooks are not supported yet",
source_path.display()
)),
HookHandlerConfig::Agent {} => warnings.push(format!(
"skipping agent hook in {}: agent hooks are not supported yet",
source_path.display()
)),
}
}
}
fn append_matcher_groups(
handlers: &mut Vec<ConfiguredHandler>,
warnings: &mut Vec<String>,
display_order: &mut i64,
source_path: &AbsolutePathBuf,
source: HookSource,
event_name: codex_protocol::protocol::HookEventName,
groups: Vec<MatcherGroup>,
) {
for group in groups {
append_group_handlers(
handlers,
warnings,
display_order,
source_path,
event_name,
matcher_pattern_for_event(event_name, group.matcher.as_deref()),
group.hooks,
);
let matcher = matcher_pattern_for_event(event_name, group.matcher.as_deref());
if let Some(matcher) = matcher
&& let Err(err) = validate_matcher_pattern(matcher)
{
warnings.push(format!(
"invalid matcher {matcher:?} in {}: {err}",
source_path.display()
));
continue;
}
for handler in group.hooks {
match handler {
HookHandlerConfig::Command {
command,
timeout_sec,
r#async,
status_message,
} => {
if r#async {
warnings.push(format!(
"skipping async hook in {}: async hooks are not supported yet",
source_path.display()
));
continue;
}
if command.trim().is_empty() {
warnings.push(format!(
"skipping empty hook command in {}",
source_path.display()
));
continue;
}
let timeout_sec = timeout_sec.unwrap_or(600).max(1);
handlers.push(ConfiguredHandler {
event_name,
matcher: matcher.map(ToOwned::to_owned),
command,
timeout_sec,
status_message,
source_path: source_path.clone(),
source,
display_order: *display_order,
});
*display_order += 1;
}
HookHandlerConfig::Prompt {} => warnings.push(format!(
"skipping prompt hook in {}: prompt hooks are not supported yet",
source_path.display()
)),
HookHandlerConfig::Agent {} => warnings.push(format!(
"skipping agent hook in {}: agent hooks are not supported yet",
source_path.display()
)),
}
}
}
}
fn hook_source_for_config_layer_source(source: &ConfigLayerSource) -> HookSource {
match source {
ConfigLayerSource::System { .. } => HookSource::System,
ConfigLayerSource::User { .. } => HookSource::User,
ConfigLayerSource::Project { .. } => HookSource::Project,
ConfigLayerSource::Mdm { .. } => HookSource::Mdm,
ConfigLayerSource::SessionFlags => HookSource::SessionFlags,
ConfigLayerSource::LegacyManagedConfigTomlFromFile { .. } => {
HookSource::LegacyManagedConfigFile
}
ConfigLayerSource::LegacyManagedConfigTomlFromMdm => HookSource::LegacyManagedConfigMdm,
}
}
#[cfg(test)]
mod tests {
use codex_config::ConfigLayerSource;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookSource;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
@@ -198,32 +200,43 @@ mod tests {
use super::ConfiguredHandler;
use super::HookHandlerConfig;
use super::append_group_handlers;
use crate::events::common::matcher_pattern_for_event;
use super::MatcherGroup;
use super::append_matcher_groups;
fn source_path() -> AbsolutePathBuf {
test_path_buf("/tmp/hooks.json").abs()
}
fn hook_source() -> HookSource {
HookSource::User
}
fn command_group(matcher: Option<&str>) -> MatcherGroup {
MatcherGroup {
matcher: matcher.map(str::to_string),
hooks: vec![HookHandlerConfig::Command {
command: "echo hello".to_string(),
timeout_sec: None,
r#async: false,
status_message: None,
}],
}
}
#[test]
fn user_prompt_submit_ignores_invalid_matcher_during_discovery() {
let mut handlers = Vec::new();
let mut warnings = Vec::new();
let mut display_order = 0;
append_group_handlers(
append_matcher_groups(
&mut handlers,
&mut warnings,
&mut display_order,
&source_path(),
hook_source(),
HookEventName::UserPromptSubmit,
matcher_pattern_for_event(HookEventName::UserPromptSubmit, Some("[")),
vec![HookHandlerConfig::Command {
command: "echo hello".to_string(),
timeout_sec: None,
r#async: false,
status_message: None,
}],
vec![command_group(Some("["))],
);
assert_eq!(warnings, Vec::<String>::new());
@@ -236,6 +249,7 @@ mod tests {
timeout_sec: 600,
status_message: None,
source_path: source_path(),
source: hook_source(),
display_order: 0,
}]
);
@@ -247,19 +261,14 @@ mod tests {
let mut warnings = Vec::new();
let mut display_order = 0;
append_group_handlers(
append_matcher_groups(
&mut handlers,
&mut warnings,
&mut display_order,
&source_path(),
hook_source(),
HookEventName::PreToolUse,
matcher_pattern_for_event(HookEventName::PreToolUse, Some("^Bash$")),
vec![HookHandlerConfig::Command {
command: "echo hello".to_string(),
timeout_sec: None,
r#async: false,
status_message: None,
}],
vec![command_group(Some("^Bash$"))],
);
assert_eq!(warnings, Vec::<String>::new());
@@ -272,6 +281,7 @@ mod tests {
timeout_sec: 600,
status_message: None,
source_path: source_path(),
source: hook_source(),
display_order: 0,
}]
);
@@ -283,19 +293,14 @@ mod tests {
let mut warnings = Vec::new();
let mut display_order = 0;
append_group_handlers(
append_matcher_groups(
&mut handlers,
&mut warnings,
&mut display_order,
&source_path(),
hook_source(),
HookEventName::PreToolUse,
matcher_pattern_for_event(HookEventName::PreToolUse, Some("*")),
vec![HookHandlerConfig::Command {
command: "echo hello".to_string(),
timeout_sec: None,
r#async: false,
status_message: None,
}],
vec![command_group(Some("*"))],
);
assert_eq!(warnings, Vec::<String>::new());
@@ -309,19 +314,14 @@ mod tests {
let mut warnings = Vec::new();
let mut display_order = 0;
append_group_handlers(
append_matcher_groups(
&mut handlers,
&mut warnings,
&mut display_order,
&source_path(),
hook_source(),
HookEventName::PostToolUse,
matcher_pattern_for_event(HookEventName::PostToolUse, Some("Edit|Write")),
vec![HookHandlerConfig::Command {
command: "echo hello".to_string(),
timeout_sec: None,
r#async: false,
status_message: None,
}],
vec![command_group(Some("Edit|Write"))],
);
assert_eq!(warnings, Vec::<String>::new());
@@ -329,4 +329,52 @@ mod tests {
assert_eq!(handlers[0].event_name, HookEventName::PostToolUse);
assert_eq!(handlers[0].matcher.as_deref(), Some("Edit|Write"));
}
#[test]
fn hook_source_for_config_layer_source_discards_source_details() {
let config_file = test_path_buf("/tmp/.codex/config.toml").abs();
let dot_codex_folder = test_path_buf("/tmp/worktree/.codex").abs();
assert_eq!(
super::hook_source_for_config_layer_source(&ConfigLayerSource::System {
file: config_file.clone(),
}),
HookSource::System,
);
assert_eq!(
super::hook_source_for_config_layer_source(&ConfigLayerSource::User {
file: config_file.clone(),
}),
HookSource::User,
);
assert_eq!(
super::hook_source_for_config_layer_source(&ConfigLayerSource::Project {
dot_codex_folder
}),
HookSource::Project,
);
assert_eq!(
super::hook_source_for_config_layer_source(&ConfigLayerSource::Mdm {
domain: "com.openai.codex".to_string(),
key: "config".to_string(),
}),
HookSource::Mdm,
);
assert_eq!(
super::hook_source_for_config_layer_source(&ConfigLayerSource::SessionFlags),
HookSource::SessionFlags,
);
assert_eq!(
super::hook_source_for_config_layer_source(
&ConfigLayerSource::LegacyManagedConfigTomlFromFile { file: config_file },
),
HookSource::LegacyManagedConfigFile,
);
assert_eq!(
super::hook_source_for_config_layer_source(
&ConfigLayerSource::LegacyManagedConfigTomlFromMdm,
),
HookSource::LegacyManagedConfigMdm,
);
}
}
+4
View File
@@ -50,6 +50,7 @@ pub(crate) fn running_summary(handler: &ConfiguredHandler) -> HookRunSummary {
execution_mode: HookExecutionMode::Sync,
scope: scope_for_event(handler.event_name),
source_path: handler.source_path.clone(),
source: handler.source,
display_order: handler.display_order,
status: HookRunStatus::Running,
status_message: handler.status_message.clone(),
@@ -95,6 +96,7 @@ pub(crate) fn completed_summary(
execution_mode: HookExecutionMode::Sync,
scope: scope_for_event(handler.event_name),
source_path: handler.source_path.clone(),
source: handler.source,
display_order: handler.display_order,
status,
status_message: handler.status_message.clone(),
@@ -118,6 +120,7 @@ fn scope_for_event(event_name: HookEventName) -> HookScope {
#[cfg(test)]
mod tests {
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookSource;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
@@ -137,6 +140,7 @@ mod tests {
timeout_sec: 5,
status_message: None,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: HookSource::User,
display_order,
}
}
+2
View File
@@ -7,6 +7,7 @@ pub(crate) mod schema_loader;
use codex_config::ConfigLayerStack;
use codex_protocol::protocol::HookRunSummary;
use codex_protocol::protocol::HookSource;
use codex_utils_absolute_path::AbsolutePathBuf;
use crate::events::post_tool_use::PostToolUseOutcome;
@@ -34,6 +35,7 @@ pub(crate) struct ConfiguredHandler {
pub timeout_sec: u64,
pub status_message: Option<String>,
pub source_path: AbsolutePathBuf,
pub source: HookSource,
pub display_order: i64,
}
@@ -525,6 +525,7 @@ mod tests {
timeout_sec: 5,
status_message: Some("running post tool use hook".to_string()),
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
}
}
@@ -514,6 +514,7 @@ mod tests {
timeout_sec: 5,
status_message: None,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
}
}
@@ -361,6 +361,7 @@ mod tests {
timeout_sec: 600,
status_message: None,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
}
}
+1
View File
@@ -528,6 +528,7 @@ mod tests {
timeout_sec: 600,
status_message: None,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
}
}
@@ -419,6 +419,7 @@ mod tests {
timeout_sec: 5,
status_message: None,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
}
}
+16
View File
@@ -1635,6 +1635,20 @@ pub enum HookScope {
Turn,
}
#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
pub enum HookSource {
System,
User,
Project,
Mdm,
SessionFlags,
LegacyManagedConfigFile,
LegacyManagedConfigMdm,
#[default]
Unknown,
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
pub enum HookRunStatus {
@@ -1671,6 +1685,8 @@ pub struct HookRunSummary {
pub execution_mode: HookExecutionMode,
pub scope: HookScope,
pub source_path: AbsolutePathBuf,
#[serde(default)]
pub source: HookSource,
pub display_order: i64,
pub status: HookRunStatus,
pub status_message: Option<String>,
@@ -1,7 +1,12 @@
use std::path::PathBuf;
fn main() -> Result<(), Box<dyn std::error::Error>> {
let proto_dir = PathBuf::from(std::env::args().nth(1).expect("proto dir"));
let Some(proto_dir_arg) = std::env::args().nth(1) else {
eprintln!("Usage: generate-proto <proto-dir>");
std::process::exit(1);
};
let proto_dir = PathBuf::from(proto_dir_arg);
let proto_file = proto_dir.join("codex.thread_store.v1.proto");
tonic_prost_build::configure()
+2
View File
@@ -9892,6 +9892,7 @@ guardian_approval = true
execution_mode: AppServerHookExecutionMode::Sync,
scope: AppServerHookScope::Turn,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_app_server_protocol::HookSource::User,
display_order: 0,
status: AppServerHookRunStatus::Running,
status_message: Some("checking go-workflow input policy".to_string()),
@@ -9914,6 +9915,7 @@ guardian_approval = true
execution_mode: AppServerHookExecutionMode::Sync,
scope: AppServerHookScope::Turn,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_app_server_protocol::HookSource::User,
display_order: 0,
status: AppServerHookRunStatus::Stopped,
status_message: Some("checking go-workflow input policy".to_string()),
+1
View File
@@ -1329,6 +1329,7 @@ fn hook_run_summary_from_notification(
execution_mode: run.execution_mode.to_core(),
scope: run.scope.to_core(),
source_path: run.source_path,
source: run.source.to_core(),
display_order: run.display_order,
status: run.status.to_core(),
status_message: run.status_message,
@@ -1001,6 +1001,7 @@ pub(super) async fn assert_hook_events_snapshot(
execution_mode: codex_protocol::protocol::HookExecutionMode::Sync,
scope: codex_protocol::protocol::HookScope::Turn,
source_path: PathBuf::from(test_path_display("/tmp/hooks.json")).abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
status: codex_protocol::protocol::HookRunStatus::Running,
status_message: Some(status_message.to_string()),
@@ -1035,6 +1036,7 @@ pub(super) async fn assert_hook_events_snapshot(
execution_mode: codex_protocol::protocol::HookExecutionMode::Sync,
scope: codex_protocol::protocol::HookScope::Turn,
source_path: PathBuf::from(test_path_display("/tmp/hooks.json")).abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
status: codex_protocol::protocol::HookRunStatus::Completed,
status_message: Some(status_message.to_string()),
@@ -1425,6 +1425,7 @@ async fn user_prompt_submit_app_server_hook_notifications_render_snapshot() {
execution_mode: AppServerHookExecutionMode::Sync,
scope: AppServerHookScope::Turn,
source_path: PathBuf::from(test_path_display("/tmp/hooks.json")).abs(),
source: codex_app_server_protocol::HookSource::User,
display_order: 0,
status: AppServerHookRunStatus::Running,
status_message: Some("checking go-workflow input policy".to_string()),
@@ -1447,6 +1448,7 @@ async fn user_prompt_submit_app_server_hook_notifications_render_snapshot() {
execution_mode: AppServerHookExecutionMode::Sync,
scope: AppServerHookScope::Turn,
source_path: PathBuf::from(test_path_display("/tmp/hooks.json")).abs(),
source: codex_app_server_protocol::HookSource::User,
display_order: 0,
status: AppServerHookRunStatus::Stopped,
status_message: Some("checking go-workflow input policy".to_string()),
@@ -1517,6 +1519,7 @@ async fn completed_hook_with_no_entries_stays_out_of_history() {
execution_mode: codex_protocol::protocol::HookExecutionMode::Sync,
scope: codex_protocol::protocol::HookScope::Turn,
source_path: PathBuf::from(test_path_display("/tmp/hooks.json")).abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
status: codex_protocol::protocol::HookRunStatus::Running,
status_message: None,
@@ -1542,6 +1545,7 @@ async fn completed_hook_with_no_entries_stays_out_of_history() {
execution_mode: codex_protocol::protocol::HookExecutionMode::Sync,
scope: codex_protocol::protocol::HookScope::Turn,
source_path: PathBuf::from(test_path_display("/tmp/hooks.json")).abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
status: codex_protocol::protocol::HookRunStatus::Completed,
status_message: None,
@@ -2017,6 +2021,7 @@ fn hook_run_summary(
execution_mode: codex_protocol::protocol::HookExecutionMode::Sync,
scope: codex_protocol::protocol::HookScope::Turn,
source_path: PathBuf::from(test_path_display("/tmp/hooks.json")).abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
status,
status_message: status_message.map(str::to_string),
@@ -768,6 +768,7 @@ mod tests {
execution_mode: codex_protocol::protocol::HookExecutionMode::Sync,
scope: codex_protocol::protocol::HookScope::Turn,
source_path: test_path_buf("/tmp/hooks.json").abs(),
source: codex_protocol::protocol::HookSource::User,
display_order: 0,
status: HookRunStatus::Running,
status_message: Some("checking output policy".to_string()),