From d84c872464aca4821dc5cb548a651651aefd84fa Mon Sep 17 00:00:00 2001 From: Ruslan Nigmatullin Date: Wed, 4 Nov 2026 03:47:51 +0000 Subject: [PATCH] Route response item IDs through a shared generator (#59312) ## What changed Introduce `ItemIdGenerator` and pass the runtime-owned generator through sessions, model clients, history, hooks, tool outputs, and memory requests. Keep UUIDv7 generation as the default and replace direct calls to `ResponseItemId::new` with the shared callable. Preserve existing item IDs and backfill missing or empty IDs through the same generator, including forked history before reconstruction and persistence. ## Testing Extend coverage for UUIDv7 IDs on user messages, agent messages, and additional tools, and assert that forked model context matches the backfilled items persisted to history. GitOrigin-RevId: 0dd9ccf0f46f0f1005a6acfc740133b03ce82259 --- codex-rs/Cargo.lock | 4 + codex-rs/app-server-protocol/Cargo.toml | 1 + .../src/protocol/thread_history.rs | 11 ++- codex-rs/core/src/agent/control.rs | 3 +- codex-rs/core/src/agent/control/runtime.rs | 4 + .../core/src/agent/control/sender_context.rs | 4 +- codex-rs/core/src/agent/control_tests.rs | 17 ++-- codex-rs/core/src/client.rs | 66 ++++++++++++-- codex-rs/core/src/client_tests.rs | 23 +++-- codex-rs/core/src/codex_thread.rs | 5 ++ codex-rs/core/src/compact_remote_v2.rs | 9 +- .../context/contextual_user_message_tests.rs | 11 ++- codex-rs/core/src/context_manager/history.rs | 53 +++++++++-- .../history_rejected_audio_tests.rs | 9 +- .../core/src/context_manager/history_tests.rs | 43 +++++---- .../history_user_authorization.rs | 5 +- codex-rs/core/src/event_mapping_tests.rs | 11 ++- codex-rs/core/src/guardian/review_request.rs | 22 ++--- .../core/src/guardian/review_session_tests.rs | 2 +- .../src/session/guardian_checkpoint_tests.rs | 3 + codex-rs/core/src/session/inject.rs | 8 +- codex-rs/core/src/session/mod.rs | 76 ++++++---------- codex-rs/core/src/session/retained_context.rs | 2 +- .../src/session/rollout_reconstruction.rs | 1 + .../session/rollout_reconstruction_tests.rs | 5 +- codex-rs/core/src/session/session.rs | 16 +++- codex-rs/core/src/session/tests.rs | 88 +++++++++--------- codex-rs/core/src/session/turn.rs | 31 +++---- codex-rs/core/src/session/turn_input.rs | 2 +- codex-rs/core/src/state/session.rs | 2 +- codex-rs/core/src/tasks/review.rs | 5 +- .../core/src/tools/executed_tool_calls.rs | 6 +- .../request_metadata_tests.rs | 21 ++++- .../tools/executed_tool_calls_direct_tests.rs | 34 +++++-- codex-rs/core/src/tools/parallel.rs | 7 +- codex-rs/core/tests/responses_headers.rs | 4 + codex-rs/core/tests/suite/client.rs | 3 + .../core/tests/suite/client_websockets.rs | 1 + .../guardian-reviewer/src/review_evidence.rs | 8 +- .../src/async_scorer/classification.rs | 1 + .../src/async_scorer/extension_tests.rs | 5 +- .../guardian-v2/src/async_scorer/sampler.rs | 5 +- .../src/async_scorer/sampler_tests.rs | 4 + .../guardian-context/src/composition_tests.rs | 2 +- .../guardian-context/src/enforcement_tests.rs | 2 +- .../guardian-context/src/registry_tests.rs | 2 +- .../src/retained_instructions_tests.rs | 89 ++++++++++++------- .../guardian-context/src/reviews_tests.rs | 9 +- .../guardian-context/tests/cache_prefix.rs | 5 +- codex-rs/history/Cargo.toml | 1 + .../src/compaction_checkpoint_tests.rs | 6 +- .../src/reconciled_retained_context_tests.rs | 38 +++++--- .../src/retained_assistant_messages.rs | 4 +- codex-rs/history/src/retained_context.rs | 11 ++- .../history/src/retained_context_tests.rs | 88 +++++++++++------- codex-rs/history/src/tests.rs | 1 + codex-rs/memories/write/src/phase1.rs | 3 +- codex-rs/memories/write/src/runtime.rs | 5 ++ codex-rs/protocol/src/item_id_generator.rs | 12 +++ codex-rs/protocol/src/items.rs | 13 ++- codex-rs/protocol/src/lib.rs | 4 + .../src/models/configuration_update_tests.rs | 5 +- codex-rs/protocol/src/response_item_id.rs | 4 - .../protocol/src/response_item_id_tests.rs | 2 +- codex-rs/protocol/tests/support/Cargo.toml | 1 + codex-rs/protocol/tests/support/lib.rs | 9 +- codex-rs/thread-store/Cargo.toml | 1 + .../src/local/rollout_migration_tests.rs | 12 +-- 68 files changed, 634 insertions(+), 336 deletions(-) create mode 100644 codex-rs/protocol/src/item_id_generator.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index a86a7e1960..48b39110c7 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2249,6 +2249,7 @@ dependencies = [ "codex-extension-items", "codex-history", "codex-protocol", + "codex-protocol-test-support", "codex-rollout", "codex-secrets", "codex-shell-command", @@ -3759,6 +3760,7 @@ version = "0.0.0" dependencies = [ "anyhow", "codex-protocol", + "codex-protocol-test-support", "pretty_assertions", "schemars 0.8.22", "serde", @@ -4595,6 +4597,7 @@ name = "codex-protocol-test-support" version = "0.0.0" dependencies = [ "codex-protocol", + "uuid", ] [[package]] @@ -5030,6 +5033,7 @@ dependencies = [ "codex-install-context", "codex-otel", "codex-protocol", + "codex-protocol-test-support", "codex-rollout", "codex-state", "codex-utils-absolute-path", diff --git a/codex-rs/app-server-protocol/Cargo.toml b/codex-rs/app-server-protocol/Cargo.toml index 2122f79ba5..e82788f826 100644 --- a/codex-rs/app-server-protocol/Cargo.toml +++ b/codex-rs/app-server-protocol/Cargo.toml @@ -41,6 +41,7 @@ uuid = { workspace = true, features = ["serde", "v7"] } zstd = { workspace = true } [dev-dependencies] +codex-protocol-test-support = { workspace = true } anyhow = { workspace = true } codex-utils-cargo-bin = { workspace = true } pretty_assertions = { workspace = true } diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index bf00a82243..10aa6f6dd3 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -5459,10 +5459,13 @@ mod tests { #[test] fn rebuilds_hook_prompt_items_from_rollout_response_items() { - let hook_prompt = build_hook_prompt_message(&[ - CoreHookPromptFragment::from_single_hook("Retry with tests.", "hook-run-1"), - CoreHookPromptFragment::from_single_hook("Then summarize cleanly.", "hook-run-2"), - ]) + let hook_prompt = build_hook_prompt_message( + &[ + CoreHookPromptFragment::from_single_hook("Retry with tests.", "hook-run-1"), + CoreHookPromptFragment::from_single_hook("Then summarize cleanly.", "hook-run-2"), + ], + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ) .expect("hook prompt message"); let items = vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index b9736c61e0..5253ed41b7 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -24,7 +24,6 @@ use crate::turn_timing::now_unix_timestamp_ms; use codex_history::InitialHistory; use codex_history::ResumedHistory; use codex_protocol::AgentPath; -use codex_protocol::ResponseItemId; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; @@ -271,7 +270,7 @@ impl LocalAgentControl { .id .take() .filter(|id| !id.is_empty()) - .unwrap_or_else(|| ResponseItemId::new("amsg")); + .unwrap_or_else(|| (self.runtime.item_id_generator)("amsg")); communication.id = Some(message_id.clone()); MessageDelivery::publish(context.sender_thread_id, message_id, observers) }); diff --git a/codex-rs/core/src/agent/control/runtime.rs b/codex-rs/core/src/agent/control/runtime.rs index c473206ab4..08a8ad2efc 100644 --- a/codex-rs/core/src/agent/control/runtime.rs +++ b/codex-rs/core/src/agent/control/runtime.rs @@ -8,6 +8,7 @@ use crate::agent::api::AgentControl; use crate::agent::registry::AgentRegistry; use crate::config::RolloutBudgetConfig; use crate::rollout_budget::RolloutBudget; +use crate::session::session::Session; use crate::thread_manager::AgentTreeShutdownFailure; use crate::thread_manager::AgentTreeShutdownFailureReason; use crate::thread_manager::AgentTreeShutdownReport; @@ -16,6 +17,7 @@ use crate::thread_manager::ThreadManagerState; use arc_swap::ArcSwapOption; use codex_extension_api::ThreadInstructionsProvider; use codex_protocol::AgentPath; +use codex_protocol::ItemIdGenerator; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::error::AgentErrorContext; @@ -192,6 +194,7 @@ pub(crate) struct LocalAgentRuntime { pub(super) manager: Weak, /// Captured at construction so delegates retain their manager's allocation policy. pub(super) thread_id_generator: ThreadIdGenerator, + pub(crate) item_id_generator: ItemIdGenerator, pub(super) agent_execution_limiter: Arc, /// Session-scoped state shared by the root thread and every cloned sub-agent control handle. pub(super) rollout_budget: Arc, @@ -217,6 +220,7 @@ impl LocalAgentRuntime { let runtime = Self { manager, thread_id_generator, + item_id_generator: Session::new_item_id_generator(), registry: Arc::default(), residency: Arc::default(), mailboxes: Arc::default(), diff --git a/codex-rs/core/src/agent/control/sender_context.rs b/codex-rs/core/src/agent/control/sender_context.rs index b3ab74db8f..39afaca3d6 100644 --- a/codex-rs/core/src/agent/control/sender_context.rs +++ b/codex-rs/core/src/agent/control/sender_context.rs @@ -103,7 +103,9 @@ impl LocalAgentRuntime { ContextManager::reconstruct_rollout( &stored.items, meta.history_mode, - ContextManager::new(), + ContextManager::new(std::sync::Arc::clone( + &self.item_id_generator, + )), // Tool bodies are not sender authorization evidence. TruncationPolicy::Bytes(0), ) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index a0f3ea2e2d..c8662e89a0 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -2507,11 +2507,9 @@ async fn spawn_agent_fork_from_paginated_parent_uses_model_context_prefix( .get_thread(child_thread_id) .await .expect("child thread should be registered"); + let child_history = child_thread.session.clone_history().await; assert!( - history_contains_text( - child_thread.session.clone_history().await.raw_items(), - "paginated parent context", - ), + history_contains_text(child_history.raw_items(), "paginated parent context"), "bounded parent context should remain model-visible to the child" ); child_thread.ensure_rollout_materialized().await; @@ -2559,6 +2557,12 @@ async fn spawn_agent_fork_from_paginated_parent_uses_model_context_prefix( copied_idless_context.id().is_some_and(|id| !id.is_empty()), "copied model context should receive response item ids before persistence" ); + assert!( + child_history + .raw_items() + .any(|item| item == &copied_idless_context.item), + "the child should see the same backfilled item that was persisted" + ); let copied_parent_context_count = lines .iter() .filter(|line| { @@ -3426,7 +3430,10 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history(marke }), }; let mut retained_context = codex_history::RetainedContext::default(); - retained_context.record(&answer_event); + retained_context.record( + &answer_event, + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); retained_context.record_sender_user_messages(delivery.metadata.as_ref().unwrap()); parent_thread .session diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index e86bb5297f..dd349ef5cf 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -79,6 +79,7 @@ use codex_login::default_client::add_originator_header; use codex_otel::SessionTelemetry; use codex_otel::WEBSOCKET_CONTINUATION_COUNT_METRIC; use codex_otel::current_span_w3c_trace_context; +use codex_protocol::ItemIdGenerator; use codex_protocol::ResponseItemId; use codex_protocol::auth::AuthMode; @@ -198,9 +199,9 @@ fn session_telemetry_for_request( /// /// This is intentionally kept minimal so `ModelClient` does not need to hold a full `Config`. Most /// configuration is per turn and is passed explicitly to streaming/unary methods. -#[derive(Debug)] struct ModelClientState { thread_id: ThreadId, + item_id_generator: ItemIdGenerator, provider: SharedModelProvider, workspace_routing: WorkspaceRoutingContext, auth_env_telemetry: AuthEnvTelemetry, @@ -223,6 +224,51 @@ struct ModelClientState { last_inference_tools: StdMutex>>, } +impl std::fmt::Debug for ModelClientState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ModelClientState") + .field("thread_id", &self.thread_id) + .field("provider", &self.provider) + .field("workspace_routing", &self.workspace_routing) + .field("auth_env_telemetry", &self.auth_env_telemetry) + .field("session_source", &self.session_source) + .field("originator", &self.originator) + .field("model_verbosity", &self.model_verbosity) + .field( + "content_item_kinds_enabled", + &self.content_item_kinds_enabled, + ) + .field( + "reasoning_effort_override_enabled", + &self.reasoning_effort_override_enabled, + ) + .field( + "enable_request_compression", + &self.enable_request_compression, + ) + .field("include_timing_metrics", &self.include_timing_metrics) + .field("beta_features_header", &self.beta_features_header) + .field( + "concurrent_reasoning_summaries_enabled", + &self.concurrent_reasoning_summaries_enabled, + ) + .field( + "output_token_replay_enabled", + &self.output_token_replay_enabled, + ) + .field("include_attestation", &self.include_attestation) + .field("attestation_provider", &self.attestation_provider) + .field("disable_websockets", &self.disable_websockets) + .field( + "agent_identity_session_fallback", + &self.agent_identity_session_fallback, + ) + .field("cached_websocket_session", &self.cached_websocket_session) + .field("last_inference_tools", &self.last_inference_tools) + .finish() + } +} + enum ClientRouting { Workspace, ConfiguredProvider, @@ -538,6 +584,7 @@ impl ModelClient { model_provider: SharedModelProvider, agent_identity_policy: AgentIdentityAuthPolicy, thread_id: ThreadId, + item_id_generator: ItemIdGenerator, session_source: SessionSource, originator: String, model_verbosity: Option, @@ -573,6 +620,7 @@ impl ModelClient { Self { state: Arc::new(ModelClientState { thread_id, + item_id_generator, provider: model_provider, workspace_routing, auth_env_telemetry, @@ -1896,7 +1944,7 @@ impl ResponsesHttpExecutor<'_> { stream, request_session_telemetry, inference_trace_attempt, - Arc::clone(&self.client.state.provider), + &self.client.state, interceptors, /*last_response_tx*/ None, )); @@ -2240,7 +2288,7 @@ impl ModelClientSession { stream_result, request_session_telemetry, inference_trace_attempt, - Arc::clone(&self.client.state.provider), + &self.client.state, interceptors, Some(last_response_tx), ); @@ -2497,7 +2545,7 @@ fn map_response_stream( mut api_stream: codex_api::ResponseStream, session_telemetry: SessionTelemetry, inference_trace_attempt: InferenceTraceAttempt, - provider: SharedModelProvider, + client: &ModelClientState, interceptors: Vec>, last_response_tx: Option>, ) -> ResponseStream { @@ -2509,7 +2557,7 @@ fn map_response_stream( crate::model_request::intercept_stream(Box::pin(api_stream), interceptors), session_telemetry, inference_trace_attempt, - provider, + client, last_response_tx, ); stream.interrupt = interrupt; @@ -2522,7 +2570,7 @@ fn map_response_events( api_stream: S, session_telemetry: SessionTelemetry, inference_trace_attempt: InferenceTraceAttempt, - provider: SharedModelProvider, + client: &ModelClientState, mut last_response_tx: Option>, ) -> ResponseStream where @@ -2535,6 +2583,8 @@ where mpsc::channel::>(RESPONSE_STREAM_CHANNEL_CAPACITY); let consumer_dropped = CancellationToken::new(); let consumer_dropped_for_stream = consumer_dropped.clone(); + let provider = Arc::clone(&client.provider); + let item_id_generator = Arc::clone(&client.item_id_generator); tokio::spawn(async move { let mut logged_error = false; @@ -2571,7 +2621,7 @@ where item.set_id( active_item_id .take() - .or_else(|| item.id_prefix().map(ResponseItemId::new)), + .or_else(|| item.id_prefix().map(item_id_generator.as_ref())), ); } active_item_id = None; @@ -2630,7 +2680,7 @@ where Ok(mut event) => { if let ResponseEvent::OutputItemAdded(item) = &mut event { if item.id().is_none_or(|id| id.is_empty()) { - item.set_id(item.id_prefix().map(ResponseItemId::new)); + item.set_id(item.id_prefix().map(item_id_generator.as_ref())); } active_item_id = item.id().cloned(); } diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index d8ecd9082b..7e5ba85741 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -69,6 +69,7 @@ use codex_protocol::openai_models::ReasoningEffortPreset; use codex_protocol::protocol::InternalSessionSource; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use codex_rollout_trace::ExecutionStatus; use codex_rollout_trace::InferenceTraceAttempt; use codex_rollout_trace::InferenceTraceContext; @@ -128,6 +129,7 @@ fn test_model_client_with_thread_id( codex_model_provider::create_model_provider(provider, /*auth_manager*/ None), AgentIdentityAuthPolicy::JwtOnly, thread_id, + uuid_v7_item_id_generator_for_test(), session_source, "test_originator".to_string(), /*model_verbosity*/ None, @@ -539,7 +541,7 @@ async fn responses_request_includes_internal_metadata_for_provider_grant_or_firs use codex_features::Feature; use codex_features::Features; - let (_, turn) = crate::session::tests::make_session_and_context().await; + let (session, turn) = crate::session::tests::make_session_and_context().await; let step = StepContext::for_test(Arc::new(turn)); let mut features = Features::default(); features.enable(Feature::ExecutedToolCallMetadata); @@ -564,7 +566,11 @@ async fn responses_request_includes_internal_metadata_for_provider_grant_or_firs call_id: id.to_string(), output: FunctionCallOutputPayload::from_text(format!("result for {id}")), }); - recorder.attach_direct_call_to_output(&mut item, Some((recorded, permit))); + recorder.attach_direct_call_to_output( + &mut item, + Some((recorded, permit)), + &session.item_id_generator, + ); outputs.push(item); } let persisted_outputs = outputs.clone(); @@ -1675,7 +1681,7 @@ async fn untraced_http_stream_releases_completed_output_before_response_ends() { events, test_session_telemetry(), InferenceTraceAttempt::disabled(), - test_model_provider(), + &test_model_client(SessionSource::Cli).state, /*last_response_tx*/ None, ); @@ -1717,7 +1723,7 @@ async fn incomplete_response_traces_attempted_call_without_streaming_it() -> any api_stream, test_session_telemetry(), attempt, - test_model_provider(), + &test_model_client(SessionSource::Cli).state, /*last_response_tx*/ None, ); assert!(matches!(stream.next().await, Some(Err(_)))); @@ -1762,7 +1768,7 @@ async fn dropped_response_stream_traces_cancelled_partial_output() -> anyhow::Re api_stream, test_session_telemetry(), attempt, - test_model_provider(), + &test_model_client(SessionSource::Cli).state, /*last_response_tx*/ None, ); @@ -1819,7 +1825,7 @@ async fn response_stream_records_last_model_feedback_ids() { api_stream, test_session_telemetry(), InferenceTraceAttempt::disabled(), - test_model_provider(), + &test_model_client(SessionSource::Cli).state, /*last_response_tx*/ None, ); @@ -2043,7 +2049,7 @@ async fn dropped_backpressured_response_stream_traces_cancelled_partial_output() api_stream, test_session_telemetry(), attempt, - test_model_provider(), + &test_model_client(SessionSource::Cli).state, /*last_response_tx*/ None, ); @@ -2150,6 +2156,7 @@ fn model_client_with_counting_attestation( codex_model_provider::create_model_provider(provider, auth_manager), AgentIdentityAuthPolicy::JwtOnly, ThreadId::new(), + uuid_v7_item_id_generator_for_test(), SessionSource::Exec, "test_originator".to_string(), /*model_verbosity*/ None, @@ -2344,7 +2351,7 @@ async fn intercepted_output_reaches_trace_and_websocket_bookkeeping() -> anyhow: }, test_session_telemetry(), attempt, - test_model_provider(), + &test_model_client(SessionSource::Cli).state, vec![Box::new(ReplaceOutput)], Some(last_response_tx), ); diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 5b47ffe4c2..9a89ccc68a 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -342,6 +342,11 @@ impl CodexThread { &self.session.services.thread_extension_data } + /// Generator used for response items produced on behalf of this thread. + pub fn item_id_generator(&self) -> &codex_protocol::ItemIdGenerator { + &self.session.item_id_generator + } + /// Returns tools from this thread’s ready, unshared Apps MCP client. /// Returns an empty list if the client is absent, still starting, or uses /// a shared catalog. Does not start or wait for discovery. diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index bbdc304984..7e1fa4668d 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -949,9 +949,12 @@ mod tests { #[test] fn build_v2_compacted_history_filters_to_installed_retention_shape() { - let hook = codex_protocol::items::build_hook_prompt_message(&[ - codex_protocol::items::HookPromptFragment::from_single_hook("hook", "hook-run"), - ]) + let hook = codex_protocol::items::build_hook_prompt_message( + &[codex_protocol::items::HookPromptFragment::from_single_hook( + "hook", "hook-run", + )], + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ) .expect("hook prompt"); let input = vec![ message("developer", "dev", /*phase*/ None), diff --git a/codex-rs/core/src/context/contextual_user_message_tests.rs b/codex-rs/core/src/context/contextual_user_message_tests.rs index 3a67332aaa..8463567723 100644 --- a/codex-rs/core/src/context/contextual_user_message_tests.rs +++ b/codex-rs/core/src/context/contextual_user_message_tests.rs @@ -148,10 +148,13 @@ fn ignores_regular_user_text() { #[test] fn detects_hook_prompt_fragment_and_roundtrips_escaping() { - let message = build_hook_prompt_message(&[HookPromptFragment::from_single_hook( - r#"Retry with "waves" & "#, - "hook-run-1", - )]) + let message = build_hook_prompt_message( + &[HookPromptFragment::from_single_hook( + r#"Retry with "waves" & "#, + "hook-run-1", + )], + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ) .expect("hook prompt message"); let ResponseItem::Message { content, .. } = message else { diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 7824aa19a6..56430e3f67 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -51,6 +51,7 @@ use codex_history::RetainedContextEvent; use codex_history::RetainedInputSource; use codex_prompts::render_model_instructions; use codex_protocol::DEFAULT_FUNCTION_NAMESPACE; +use codex_protocol::ItemIdGenerator; use codex_protocol::items::TurnItem; use codex_protocol::models::AgentMessageInputContent; use codex_protocol::models::AudioReference; @@ -97,8 +98,9 @@ fn next_guardian_review_context_revision() -> u64 { } /// Transcript of thread history -#[derive(Debug, Clone)] +#[derive(Clone)] pub(crate) struct ContextManager { + item_id_generator: ItemIdGenerator, /// The oldest items are at the beginning of the vector. Snapshots share the vector until a /// caller needs to mutate it, avoiding deep copies for read-only history consumers. items: Arc>, @@ -144,6 +146,44 @@ pub(crate) struct ContextManager { world_state_baseline: Option, } +impl std::fmt::Debug for ContextManager { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ContextManager") + .field("items", &self.items) + .field("review_history", &self.review_history) + .field("retained_context", &self.retained_context) + .field("rejected_audio_file_ids", &self.rejected_audio_file_ids) + .field("guardian_review_mode", &self.guardian_review_mode) + .field( + "guardian_include_additional_context", + &self.guardian_include_additional_context, + ) + .field( + "retain_inherited_user_messages", + &self.retain_inherited_user_messages, + ) + .field("history_version", &self.history_version) + .field("reset_version", &self.reset_version) + .field("user_message_revision", &self.user_message_revision) + .field( + "guardian_review_context_revision", + &self.guardian_review_context_revision, + ) + .field("token_info", &self.token_info) + .field( + "last_server_reported_usage_item_count", + &self.last_server_reported_usage_item_count, + ) + .field( + "last_model_output_item_count", + &self.last_model_output_item_count, + ) + .field("reference_context_item", &self.reference_context_item) + .field("world_state_baseline", &self.world_state_baseline) + .finish() + } +} + struct SharedConversationHistory { items: Arc>, review_history: Option, @@ -261,15 +301,17 @@ impl SharedConversationHistory { } } +#[cfg(test)] impl Default for ContextManager { fn default() -> Self { - Self::new() + Self::new(codex_protocol_test_support::uuid_v7_item_id_generator_for_test()) } } impl ContextManager { - pub(crate) fn new() -> Self { + pub(crate) fn new(item_id_generator: ItemIdGenerator) -> Self { Self { + item_id_generator, items: Arc::new(Vec::new()), review_history: None, retained_context: Arc::default(), @@ -311,6 +353,7 @@ impl ContextManager { pub(crate) fn for_session( root_capabilities_restricted: bool, features: &ManagedFeatures, + item_id_generator: ItemIdGenerator, ) -> Self { let independent = !features.enabled(Feature::GuardianReuseParentCompaction); Self { @@ -323,7 +366,7 @@ impl ContextManager { GuardianContextMode::ThreadOwned }, review_history: independent.then(TranscriptHistory::default), - ..Self::new() + ..Self::new(item_id_generator) } } @@ -332,7 +375,7 @@ impl ContextManager { } pub(crate) fn record_retained_context(&mut self, event: &RetainedContextEvent) -> bool { - if !Arc::make_mut(&mut self.retained_context).record(event) { + if !Arc::make_mut(&mut self.retained_context).record(event, &self.item_id_generator) { return false; } match event { diff --git a/codex-rs/core/src/context_manager/history_rejected_audio_tests.rs b/codex-rs/core/src/context_manager/history_rejected_audio_tests.rs index 4ec0467901..6609c63468 100644 --- a/codex-rs/core/src/context_manager/history_rejected_audio_tests.rs +++ b/codex-rs/core/src/context_manager/history_rejected_audio_tests.rs @@ -57,9 +57,12 @@ fn previously_accepted_file_only_disambiguates_clip_specific_errors() { std::slice::from_ref(&good), ) .expect("new audio"); - context.record(&codex_history::RetainedContextEvent::AcceptedFileAudio { - evidence: evidence.into(), - }); + context.record( + &codex_history::RetainedContextEvent::AcceptedFileAudio { + evidence: evidence.into(), + }, + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); assert!( context .accepted_file_audio("another model", "provider") diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index 36b7a37795..cdf79822a7 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -135,11 +135,12 @@ fn root_history() -> ContextManager { ContextManager::for_session( /*root_capabilities_restricted*/ false, &ManagedFeatures::from(codex_features::Features::with_defaults()), + codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ) } fn create_history_with_items(items: Vec) -> ContextManager { - let mut h = ContextManager::new(); + let mut h = ContextManager::default(); // Use a generous but fixed token budget; tests only rely on truncation // behavior, not on a specific model's token limit. h.record_items(items.iter(), TruncationPolicy::Tokens(10_000)); @@ -185,7 +186,7 @@ fn delivered_assistant_context_invalidates_reviews_without_changing_authorizatio ); assert!(!history.record_retained_context(&event)); assert_eq!(revisions(&history), (delivered_revision, authorization)); - let mut resumed = ContextManager::new(); + let mut resumed = ContextManager::default(); resumed.restore_retained_context(Some(history.retained_context())); assert_ne!(revisions(&resumed).0, delivered_revision); @@ -551,8 +552,11 @@ fn guardian_checkpoint_applies_additional_context_policy_after_resume(include: b features .set_enabled(Feature::GuardianIncludeAdditionalContext, !include) .unwrap(); - let mut history = - ContextManager::for_session(/*root_capabilities_restricted*/ false, &features); + let mut history = ContextManager::for_session( + /*root_capabilities_restricted*/ false, + &features, + codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); let additional_context: ResponseItem = crate::context::AdditionalContextToolOutput::new( "browser_info", /*name*/ None, @@ -590,8 +594,11 @@ fn guardian_checkpoint_applies_additional_context_policy_after_resume(include: b features .set_enabled(Feature::GuardianIncludeAdditionalContext, include) .unwrap(); - let mut resumed = - ContextManager::for_session(/*root_capabilities_restricted*/ false, &features); + let mut resumed = ContextManager::for_session( + /*root_capabilities_restricted*/ false, + &features, + codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); resumed.replace_annotated(compacted); resumed.restore_review_context( /*retained_context*/ None, @@ -654,7 +661,7 @@ fn world_state_baseline_deduplicates_until_history_is_replaced() { state.add_section(TestWorldStateSection); state }; - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); let (initial_fragments, initial_item) = history.update_world_state(&world_state()); assert_eq!(1, initial_fragments.len()); @@ -690,7 +697,7 @@ fn world_state_transitions_persist_changed_state_and_skip_unchanged_state() { })); state }; - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); let (_, full) = history.update_world_state(&world_state("before")); let full = full.expect("full checkpoint"); assert!(full.full); @@ -910,7 +917,7 @@ fn filters_non_api_messages() { #[test] fn retains_only_harness_authored_configuration_updates() { - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); let update = ResponseItem::ConfigurationUpdate { reasoning: ConfigurationReasoning { effort: ReasoningEffort::High, @@ -961,7 +968,7 @@ fn retains_only_harness_authored_configuration_updates() { #[test] fn drop_last_n_user_turns_removes_post_input_configuration_update_with_its_turn() { - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); let updates = [ReasoningEffort::Low, ReasoningEffort::High].map(|effort| ResponseItemEnvelope { item: ResponseItem::ConfigurationUpdate { @@ -1028,7 +1035,7 @@ fn total_token_usage_respects_saved_reasoning_accounting( includes_reasoning: Option, expected: i64, ) { - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); history.replace(vec![ reasoning_with_encrypted_content(/*len*/ 1_000), user_msg("second"), @@ -1143,7 +1150,7 @@ fn annotated_history_apis_preserve_envelopes() { item: assistant_msg("first"), metadata: Some(CodexHarnessMetadata::default()), }; - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); history.replace_annotated(vec![first_envelope.clone()]); @@ -1184,7 +1191,7 @@ fn record_annotated_items_preserves_metadata_while_processing_item( ..Default::default() }), }; - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); history.record_annotated_items( std::slice::from_mut(&mut envelope), @@ -1233,7 +1240,7 @@ async fn for_prompt_annotated_preserves_metadata_while_normalizing_item() { }, metadata: Some(CodexHarnessMetadata::default()), }; - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); history.replace_annotated(vec![envelope.clone()]); let normalized = history @@ -2481,7 +2488,7 @@ async fn normalization_retains_local_shell_outputs() { #[test] fn record_items_truncates_function_call_output_content() { - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); // Any reasonably small token budget works; the test only cares that // truncation happens and the marker is present. let policy = TruncationPolicy::Tokens(1_000); @@ -2525,7 +2532,7 @@ fn record_items_truncates_function_call_output_content() { #[test] fn record_items_truncates_custom_tool_call_output_content() { - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); let policy = TruncationPolicy::Tokens(1_000); let line = "custom output that is very long\n"; let long_output = line.repeat(2_500); @@ -2559,7 +2566,7 @@ fn record_items_truncates_custom_tool_call_output_content() { #[test] fn record_items_respects_custom_token_limit() { - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); let policy = TruncationPolicy::Tokens(10); let long_output = "tokenized content repeated many times ".repeat(200); let item = ResponseItem::FunctionCallOutput { @@ -3552,7 +3559,7 @@ fn record_items_omits_audio_that_exceeds_the_output_budget() { }, internal_chat_message_metadata_passthrough: None, }; - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); history.record_items([&item], TruncationPolicy::Tokens(50)); diff --git a/codex-rs/core/src/context_manager/history_user_authorization.rs b/codex-rs/core/src/context_manager/history_user_authorization.rs index 06cf19e485..096b044a31 100644 --- a/codex-rs/core/src/context_manager/history_user_authorization.rs +++ b/codex-rs/core/src/context_manager/history_user_authorization.rs @@ -145,6 +145,7 @@ impl ContextManager { complete: true, }, source, + &self.item_id_generator, ); return None; } @@ -208,9 +209,9 @@ impl ContextManager { }; let retained = Arc::make_mut(&mut self.retained_context); captured = if is_assistant { - retained.record_assistant_message(message, metadata.into()) + retained.record_assistant_message(message, metadata.into(), &self.item_id_generator) } else { - retained.record_user_message(message, metadata.into()) + retained.record_user_message(message, metadata.into(), &self.item_id_generator) }; } if !is_assistant { diff --git a/codex-rs/core/src/event_mapping_tests.rs b/codex-rs/core/src/event_mapping_tests.rs index fc2ffbe048..cfac56efc4 100644 --- a/codex-rs/core/src/event_mapping_tests.rs +++ b/codex-rs/core/src/event_mapping_tests.rs @@ -452,10 +452,13 @@ fn skips_user_instructions_and_env() { #[test] fn parses_hook_prompt_message_as_distinct_turn_item() { - let item = build_hook_prompt_message(&[HookPromptFragment::from_single_hook( - "Retry with exactly the phrase meow meow meow.", - "hook-run-1", - )]) + let item = build_hook_prompt_message( + &[HookPromptFragment::from_single_hook( + "Retry with exactly the phrase meow meow meow.", + "hook-run-1", + )], + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ) .expect("hook prompt message"); let turn_item = parse_turn_item(&item).expect("expected hook prompt turn item"); diff --git a/codex-rs/core/src/guardian/review_request.rs b/codex-rs/core/src/guardian/review_request.rs index 48c5012517..443ffab7ba 100644 --- a/codex-rs/core/src/guardian/review_request.rs +++ b/codex-rs/core/src/guardian/review_request.rs @@ -22,8 +22,7 @@ pub(in crate::guardian) struct ApprovalEvidence { store: Arc, authorization_version: GuardianAuthorizationVersion, review_context_revision: u64, - root_authorization_version: Option, - root_review_context_revision: Option, + root_versions: Option<(GuardianAuthorizationVersion, u64)>, } /// Authorization captured before an attempt, checked while the reviewer is still reserved. @@ -245,11 +244,13 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { None, ); } - let root_authorization_version = root_snapshot - .as_ref() - .map(|snapshot| snapshot.authorization_version); - let root_review_context_revision = - root_snapshot.map(|snapshot| snapshot.review_context_revision); + let root_versions = root_snapshot.map(|snapshot| { + ( + snapshot.authorization_version, + snapshot.review_context_revision, + ) + }); + let (root_authorization_version, root_review_context_revision) = root_versions.unzip(); // Keep the authorization revision even when no cacheable review evidence exists. let history = session.conversation_history_snapshot().await; let user_message_revision = history.user_message_revision(); @@ -262,8 +263,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { authorization_version: store.authorization_version(history.as_ref()), review_context_revision, store, - root_authorization_version, - root_review_context_revision, + root_versions, }); drop(history); let (outcome, analytics) = run_guardian_review_session_before_deadline( @@ -331,8 +331,8 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { action, evidence.authorization_version, evidence.review_context_revision, - evidence.root_authorization_version, - evidence.root_review_context_revision, + evidence.root_versions, + &self.session.item_id_generator, ); } } diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 7d211f24eb..c64a3b73b4 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -518,7 +518,7 @@ async fn encrypted_parent_compaction_requires_original_item_id(mode: GuardianCon internal_chat_message_metadata_passthrough: None, }; - let mut history = ContextManager::new(); + let mut history = ContextManager::default(); history.replace_annotated(vec![ResponseItemEnvelope { item: item.clone(), metadata: Some(CodexHarnessMetadata { diff --git a/codex-rs/core/src/session/guardian_checkpoint_tests.rs b/codex-rs/core/src/session/guardian_checkpoint_tests.rs index 0f7061ff6e..db86aebef2 100644 --- a/codex-rs/core/src/session/guardian_checkpoint_tests.rs +++ b/codex-rs/core/src/session/guardian_checkpoint_tests.rs @@ -6,6 +6,7 @@ use codex_history::InitialHistory; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::TokenUsage; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use pretty_assertions::assert_eq; use serde_json::json; use test_case::test_case; @@ -37,6 +38,7 @@ async fn guardian_checkpoint_preserves_live_context_without_storage( state.history = ContextManager::for_session( /*root_capabilities_restricted*/ false, &crate::config::ManagedFeatures::from(codex_features::Features::with_defaults()), + uuid_v7_item_id_generator_for_test(), ); state .history @@ -108,6 +110,7 @@ async fn guardian_checkpoint_preserves_live_context_without_storage( fork.state.lock().await.history = ContextManager::for_session( /*root_capabilities_restricted*/ false, &crate::config::ManagedFeatures::from(codex_features::Features::with_defaults()), + uuid_v7_item_id_generator_for_test(), ); fork.record_initial_history(InitialHistory::Forked(items)) .await; diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs index da412d1d70..dab028d7c3 100644 --- a/codex-rs/core/src/session/inject.rs +++ b/codex-rs/core/src/session/inject.rs @@ -184,11 +184,9 @@ impl Session { image_preparations.extend(prepared_images); let mut metadata = envelope.metadata; - annotated_items.extend(prepared_items.into_owned().into_iter().map(|item| { - ResponseItemEnvelope { - item, - metadata: metadata.take(), - } + annotated_items.extend(prepared_items.into_iter().map(|item| ResponseItemEnvelope { + item, + metadata: metadata.take(), })); } (annotated_items, image_preparations) diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 60f8fc46ec..08e68e45c2 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3,7 +3,6 @@ pub(crate) mod startup; use crate::PromptCacheKey; use crate::WithTurnExtensionData; use crate::runtime_auth::RuntimeAuthSlot; -use std::borrow::Cow; use std::collections::HashMap; use std::collections::HashSet; use std::fmt::Debug; @@ -583,6 +582,17 @@ pub(crate) const SUBMISSION_CHANNEL_CAPACITY: usize = 512; const CYBER_VERIFY_URL: &str = "https://chatgpt.com/cyber"; const CYBER_SAFETY_URL: &str = "https://developers.openai.com/codex/concepts/cyber-safety"; +pub(crate) fn assign_missing_item_id( + item: &mut ResponseItem, + item_id_generator: &codex_protocol::ItemIdGenerator, +) { + if item.id().is_none_or(|id| id.is_empty()) + && let Some(prefix) = item.id_prefix() + { + item.set_id(Some(item_id_generator(prefix))); + } +} + impl Session { /// Spawn and initialize a new session. /// Hide the concrete startup future from callers while keeping initialization lazy. @@ -1821,7 +1831,11 @@ impl Session { } InitialHistory::Forked(mut rollout_items) => { let turn_context = self.new_default_turn().await; - Self::assign_missing_rollout_response_item_ids(&mut rollout_items); + for record in &mut rollout_items { + if let RolloutItem::ResponseItem(envelope) = record { + self.assign_missing_response_item_id(&mut envelope.item); + } + } self.apply_rollout_reconstruction(&turn_context, &rollout_items) .await; @@ -3822,12 +3836,12 @@ impl Session { } /// Prepares media using the originating model and preserves existing item identity. - pub(crate) async fn prepare_conversation_items_for_history<'a>( + pub(crate) async fn prepare_conversation_items_for_history( &self, turn_context: &TurnContext, model_info: &ModelInfo, - items: &'a [ResponseItem], - ) -> (Cow<'a, [ResponseItem]>, Vec) { + items: &[ResponseItem], + ) -> (Vec, Vec) { let mut items = items.to_vec(); let image_preparation_mode = if unified_image_budget_enabled(&turn_context.config.features, model_info) { @@ -3858,44 +3872,13 @@ impl Session { // Most response items get their passthrough turn ID at the durable history boundary. for item in &mut items { Self::stamp_response_item_for_history(item, &turn_context.sub_id); + self.assign_missing_response_item_id(item); } - let items = Cow::Owned(items); - ( - Self::assign_missing_response_item_ids(items), - image_preparations, - ) + (items, image_preparations) } - fn assign_missing_response_item_ids(items: Cow<'_, [ResponseItem]>) -> Cow<'_, [ResponseItem]> { - if items - .iter() - .all(|item| item.id().is_some_and(|id| !id.is_empty())) - { - return items; - } - let mut items = items; - for item in items.to_mut() { - Self::assign_missing_response_item_id(item); - } - items - } - - pub(crate) fn assign_missing_response_item_id(item: &mut ResponseItem) { - if item.id().is_some_and(|id| !id.is_empty()) { - return; - } - let Some(prefix) = item.id_prefix() else { - return; - }; - item.set_id(Some(ResponseItemId::new(prefix))); - } - - fn assign_missing_rollout_response_item_ids(items: &mut [RolloutItem]) { - for item in items { - if let RolloutItem::ResponseItem(response_item) = item { - Self::assign_missing_response_item_id(&mut response_item.item); - } - } + pub(crate) fn assign_missing_response_item_id(&self, item: &mut ResponseItem) { + assign_missing_item_id(item, &self.item_id_generator); } pub(crate) fn response_item_from_user_input(&self, input: Vec) -> ResponseItem { @@ -3955,11 +3938,7 @@ impl Session { let (items, image_preparations) = self .prepare_conversation_items_for_history(turn_context, model_info, items) .await; - let items = items - .into_owned() - .into_iter() - .map(ResponseItemEnvelope::new) - .collect(); + let items = items.into_iter().map(ResponseItemEnvelope::new).collect(); self.record_prepared_conversation_items( turn_context, model_info, @@ -4541,7 +4520,6 @@ impl Session { std::slice::from_ref(&response_item), ) .await; - let items = items.as_ref(); let mut response_item = ResponseItemEnvelope::new(items[0].clone()); // A send confirmed after the pending snapshot must not reach the rollout // before this boundary; older readers assign deliveries by physical order. @@ -4569,7 +4547,7 @@ impl Session { } { let mut state = self.state.lock().await; - state.current_time_reminder.note_recorded_items(items); + state.current_time_reminder.note_recorded_items(&items); state.history.record_annotated_items( std::slice::from_mut(&mut response_item), model_info.truncation_policy.into(), @@ -4586,7 +4564,7 @@ impl Session { ]) .await; drop(boundary); - self.send_raw_response_items(turn_context, items).await; + self.send_raw_response_items(turn_context, &items).await; } async fn maybe_warn_on_server_model_mismatch( @@ -4771,7 +4749,7 @@ impl Session { } }; for envelope in &mut items { - Self::assign_missing_response_item_id(&mut envelope.item); + self.assign_missing_response_item_id(&mut envelope.item); } // Goal edits are published outside the running task. Keep edits accepted after // the compaction input snapshot, in their original order, after its replacement. diff --git a/codex-rs/core/src/session/retained_context.rs b/codex-rs/core/src/session/retained_context.rs index 8c4203bf13..6572cbb811 100644 --- a/codex-rs/core/src/session/retained_context.rs +++ b/codex-rs/core/src/session/retained_context.rs @@ -93,7 +93,7 @@ impl Session { let _guard = thread_settings::acquire_persistence_lock(self).await; let mut item = ContextualUserFragment::into(update); Self::stamp_response_item_for_history(&mut item, &context.sub_id); - Self::assign_missing_response_item_id(&mut item); + self.assign_missing_response_item_id(&mut item); let mut item = ResponseItemEnvelope { item, metadata: Some(CodexHarnessMetadata { diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 888981f782..56d931e03c 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -199,6 +199,7 @@ impl Session { ContextManager::for_session( turn_context.root_capabilities_restricted, &turn_context.config.features, + std::sync::Arc::clone(&self.item_id_generator), ), turn_context.model_info().truncation_policy.into(), ) diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 27c657d4c4..280691b867 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -21,6 +21,7 @@ use codex_protocol::protocol::ThreadRolledBackEvent; use codex_protocol::protocol::WorldStateItem; use codex_protocol::security_risk::SecurityRiskScore; use codex_protocol::turn_input::CyberAccessProgram; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use codex_rollout::ModelContextScan; use codex_rollout::ModelContextScanProgress; use core_test_support::responses::strip_metadata_from_items; @@ -128,6 +129,7 @@ async fn sender_context_follows_its_delivery_through_checkpoint_and_rollback() { let mut live = ContextManager::for_session( /*root_capabilities_restricted*/ false, &crate::config::ManagedFeatures::from(codex_features::Features::with_defaults()), + uuid_v7_item_id_generator_for_test(), ); let mut items = Vec::new(); let mut snapshots = Vec::new(); @@ -2024,8 +2026,9 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_ complete: false, }, codex_history::RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); - retained.record(&answer); + retained.record(&answer, &uuid_v7_item_id_generator_for_test()); let rollout_items = vec![ RolloutItem::ResponseItem(user_message("before compact").into()), RolloutItem::ResponseItem(assistant_message("assistant reply").into()), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 58c07de277..e94527c80a 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -33,6 +33,7 @@ use codex_http_client::RouteAwareClientPool; use codex_login::auth::AgentIdentityAuthPolicy; use codex_model_provider::SharedModelProvider; use codex_prompts::render_model_instructions; +use codex_protocol::ItemIdGenerator; use codex_protocol::SessionId; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::config_types::ShellEnvironmentPolicy; @@ -63,6 +64,7 @@ type McpToolApprovalMetadataMap = /// A session has at most 1 running task at a time, and can be interrupted by user input. pub(crate) struct Session { pub(crate) thread_id: ThreadId, + pub(crate) item_id_generator: ItemIdGenerator, pub(crate) agent_runtime: AgentRuntimeServices, pub(crate) installation_id: String, pub(super) tx_event: Sender, @@ -708,6 +710,10 @@ async fn warm_plugins_and_skills_for_session_init( } impl Session { + pub(crate) fn new_item_id_generator() -> ItemIdGenerator { + Arc::new(|prefix| ResponseItemId::with_suffix(prefix, uuid::Uuid::now_v7())) + } + pub(super) async fn effective_thread_config_snapshot( &self, configuration: &SessionConfiguration, @@ -964,8 +970,13 @@ impl Session { session_configuration.thread_source.as_ref(), Some(ThreadSource::Subagent | ThreadSource::GuardianReview) ); + let item_id_generator = Arc::clone(&agent_control.runtime().item_id_generator); if let InitialHistory::Forked(items) = &mut initial_history { - Self::assign_missing_rollout_response_item_ids(items); + for record in items { + if let RolloutItem::ResponseItem(item) = record { + assign_missing_item_id(&mut item.item, &item_id_generator); + } + } } let multi_agent_version = multi_agent_version.map(OnceLock::from).unwrap_or_default(); let initial_multi_agent_version = multi_agent_version.get().copied(); @@ -1624,6 +1635,7 @@ impl Session { ContextManager::for_session( session_configuration.root_capabilities_restricted, &config.features, + Arc::clone(&item_id_generator), ), ); if matches!(initial_history, InitialHistory::Resumed(_)) { @@ -1832,6 +1844,7 @@ impl Session { AgentIdentityAuthPolicy::JwtOnly }, thread_id, + Arc::clone(&item_id_generator), session_configuration.session_source.clone(), session_configuration.originator.clone(), config.model_verbosity, @@ -1953,6 +1966,7 @@ impl Session { let (mcp_prewarm_tx, mcp_prewarm_rx) = async_channel::bounded(1); let sess = Arc::new(Session { thread_id, + item_id_generator, agent_runtime, installation_id, tx_event: tx_event.clone(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 98da3f4737..9edd5643ec 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -357,11 +357,13 @@ fn user_message(text: &str) -> ResponseItem { } } -#[test] -fn assign_missing_response_item_ids_assigns_agent_message_ids() { - let items = Cow::Owned(vec![ +#[tokio::test] +async fn default_turn_context_assigns_missing_response_item_ids() { + let (session, turn_context) = make_session_and_context().await; + let items = [ + user_message("hello"), ResponseItem::AgentMessage { - id: None, + id: Some(ResponseItemId::from_server(String::new())), author: "worker".to_string(), recipient: "root".to_string(), content: vec![AgentMessageInputContent::InputText { @@ -369,46 +371,21 @@ fn assign_missing_response_item_ids_assigns_agent_message_ids() { }], internal_chat_message_metadata_passthrough: None, }, - user_message("hello"), - ]); - - let items = Session::assign_missing_response_item_ids(items); - - assert!(items[0].id().is_some_and(|id| id.starts_with("amsg_"))); - assert!(items[1].id().is_some_and(|id| id.starts_with("msg_"))); -} - -#[test] -fn assign_missing_response_item_ids_assigns_additional_tools_ids() { - let items = Cow::Owned(vec![ResponseItem::AdditionalTools { - id: None, - role: "developer".to_string(), - tools: Arc::from([]), - }]); - - let items = Session::assign_missing_response_item_ids(items); - - assert!(items[0].id().is_some_and(|id| id.starts_with("at_"))); -} - -#[tokio::test] -async fn default_turn_context_assigns_missing_response_item_ids() { - let (session, turn_context) = make_session_and_context().await; - let response_item = user_message("hello"); - + ResponseItem::AdditionalTools { + id: None, + role: "developer".to_string(), + tools: Arc::from([]), + }, + ]; let (items, _) = session - .prepare_conversation_items_for_history( - &turn_context, - turn_context.model_info(), - std::slice::from_ref(&response_item), - ) + .prepare_conversation_items_for_history(&turn_context, turn_context.model_info(), &items) .await; - assert!( - items[0] - .id() - .is_some_and(|item_id| item_id.starts_with("msg_")) - ); + assert_eq!(items.len(), 3); + for (item, prefix) in items.iter().zip(["msg_", "amsg_", "at_"]) { + let suffix = item.id().unwrap().strip_prefix(prefix).unwrap(); + assert_eq!(Uuid::parse_str(suffix).unwrap().get_version_num(), 7); + } } fn assistant_message(text: &str) -> ResponseItem { @@ -915,6 +892,7 @@ fn test_model_client_session() -> crate::client::ModelClientSession { ), AgentIdentityAuthPolicy::JwtOnly, thread_id, + codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), codex_protocol::protocol::SessionSource::Exec, "test_originator".to_string(), /*model_verbosity*/ None, @@ -2627,12 +2605,14 @@ async fn record_conversation_items_stamps_missing_turn_id_and_preserves_existing #[tokio::test] async fn record_response_item_and_emit_turn_item_emits_hook_prompt_lifecycle() { let (session, turn_context, rx) = make_session_and_context_with_rx().await; - let response_item = build_hook_prompt_message(&[HookPromptFragment::from_single_hook( - "Retry with tests.", - "hook-run-1", - )]) + let response_item = build_hook_prompt_message( + &[HookPromptFragment::from_single_hook( + "Retry with tests.", + "hook-run-1", + )], + &session.item_id_generator, + ) .expect("hook prompt message"); - let response_item_id = response_item.id().expect("hook prompt id").to_string(); session .record_response_item_and_emit_turn_item( @@ -2643,7 +2623,11 @@ async fn record_response_item_and_emit_turn_item_emits_hook_prompt_lifecycle() { .await; let raw_response = rx.recv().await.expect("raw response item event"); - assert!(matches!(raw_response.msg, EventMsg::RawResponseItem(_))); + let EventMsg::RawResponseItem(raw) = raw_response.msg else { + panic!("expected raw hook prompt"); + }; + let response_item_id = raw.item.id().expect("hook prompt id").to_string(); + assert!(response_item_id.starts_with("msg_")); let started = rx.recv().await.expect("started hook prompt event"); assert!(matches!( @@ -6937,6 +6921,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { config.model_provider.clone(), ); let agent_control = LocalAgentControl::default(); + let item_id_generator = Arc::clone(&agent_control.runtime.item_id_generator); let exec_policy = Arc::new(ExecPolicyManager::default()); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = get_model_offline_for_tests(config.model.as_deref()); @@ -7010,6 +6995,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { state.history = ContextManager::for_session( session_configuration.root_capabilities_restricted, &config.features, + Arc::clone(&item_id_generator), ); let (environment_manager, resolved_environments) = resolved_environments_for_configuration(&session_configuration, &default_environments) @@ -7122,6 +7108,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { ), AgentIdentityAuthPolicy::JwtOnly, thread_id, + Arc::clone(&item_id_generator), session_configuration.session_source.clone(), session_configuration.originator.clone(), config.model_verbosity, @@ -7153,6 +7140,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { let session = Session { thread_id, + item_id_generator, agent_runtime: crate::agent::host::AgentRuntimeServices::Controller, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, @@ -9276,6 +9264,7 @@ where config.model_provider.clone(), ); let agent_control = LocalAgentControl::default(); + let item_id_generator = Arc::clone(&agent_control.runtime.item_id_generator); let exec_policy = Arc::new(ExecPolicyManager::default()); let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = get_model_offline_for_tests(config.model.as_deref()); @@ -9349,6 +9338,7 @@ where state.history = ContextManager::for_session( session_configuration.root_capabilities_restricted, &config.features, + Arc::clone(&item_id_generator), ); let (environment_manager, resolved_turn_environments) = resolved_environments_for_configuration(&session_configuration, &default_environments) @@ -9460,6 +9450,7 @@ where ), AgentIdentityAuthPolicy::JwtOnly, thread_id, + Arc::clone(&item_id_generator), session_configuration.session_source.clone(), session_configuration.originator.clone(), config.model_verbosity, @@ -9491,6 +9482,7 @@ where let session = Arc::new(Session { thread_id, + item_id_generator, agent_runtime: crate::agent::host::AgentRuntimeServices::Controller, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, @@ -13530,7 +13522,7 @@ async fn sample_rollout( _turn_context: &TurnContext, ) -> (Vec, Vec) { let mut rollout_items = Vec::new(); - let mut live_history = ContextManager::new(); + let mut live_history = ContextManager::default(); // Use the same turn_context source as record_initial_history so model_info matches reconstruction. let reconstruction_turn = session.new_default_turn().await; diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 26edef6519..97e5ba96ed 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -775,9 +775,10 @@ pub(crate) async fn run_turn( )); } if stop_outcome.should_block { - if let Some(hook_prompt_message) = - build_hook_prompt_message(&stop_outcome.continuation_fragments) - { + if let Some(hook_prompt_message) = build_hook_prompt_message( + &stop_outcome.continuation_fragments, + &sess.item_id_generator, + ) { sess.record_response_item_and_emit_turn_item( &turn_context, &step_context.settings.model_info, @@ -2805,21 +2806,6 @@ async fn drain_in_flight( Ok(()) } -fn assign_missing_streamed_response_item_id( - item: &mut ResponseItem, - active_item: Option<&TurnItem>, -) { - if item.id().is_some_and(|id| !id.is_empty()) { - return; - } - - let active_item_id = active_item - .map(|item| ResponseItemId::from_server(item.id())) - .filter(|item_id| !item_id.is_empty()); - item.set_id(active_item_id); - Session::assign_missing_response_item_id(item); -} - #[allow(clippy::too_many_arguments)] #[instrument(level = "trace", skip_all, @@ -3124,7 +3110,12 @@ async fn try_run_sampling_request( authorization: authorization.clone(), }) }); - assign_missing_streamed_response_item_id(&mut item, active_item.as_ref()); + if item.id().is_none_or(|id| id.is_empty()) + && let Some(active_item) = &active_item + { + item.set_id(Some(ResponseItemId::from_server(active_item.id()))); + } + sess.assign_missing_response_item_id(&mut item); sess.reserve_assistant_message_order(&turn_context, &item) .await; if analytics_tool_call_ids.len() < MAX_ANALYTICS_TOOL_CALL_IDS_PER_RESPONSE { @@ -3283,7 +3274,7 @@ async fn try_run_sampling_request( } } ResponseEvent::OutputItemAdded(mut item) => { - assign_missing_streamed_response_item_id(&mut item, /*active_item*/ None); + sess.assign_missing_response_item_id(&mut item); sess.reserve_assistant_message_order(&turn_context, &item) .await; if !*draining_for_mailbox diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index fec9bd7a1a..7c2ac89938 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -1018,7 +1018,7 @@ async fn capture_sender_user_messages( if !matches!(item, ResponseItem::FunctionCallOutput { call_id: None, .. }) { return None; } - Session::assign_missing_response_item_id(item); + session.assign_missing_response_item_id(item); session .services .local_agent_runtime diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index f35bb2a188..972cf5092b 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -120,7 +120,7 @@ impl SessionState { Self::new_with_auto_compact_window_ids( session_configuration, AutoCompactWindowIds::new_initial(), - ContextManager::new(), + ContextManager::default(), ) } diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 7d9bf760a8..b65b67fbb3 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -2,7 +2,6 @@ use std::sync::Arc; use codex_prompts::render_review_exit_interrupted; use codex_prompts::render_review_exit_success; -use codex_protocol::ResponseItemId; use codex_protocol::config_types::WebSearchMode; use codex_protocol::items::ExitedReviewModeItem; use codex_protocol::items::TurnItem; @@ -242,7 +241,7 @@ pub(crate) async fn exit_review_mode( &[ResponseItem::Message { status: None, encrypted_content: None, - id: Some(ResponseItemId::new("msg")), + id: Some((session.item_id_generator)("msg")), role: "user".to_string(), content: vec![ContentItem::InputText { text: user_message }], phase: None, @@ -264,7 +263,7 @@ pub(crate) async fn exit_review_mode( ResponseItem::Message { status: None, encrypted_content: None, - id: Some(ResponseItemId::new("msg")), + id: Some((session.item_id_generator)("msg")), role: "assistant".to_string(), content: vec![ContentItem::OutputText { annotations: None, diff --git a/codex-rs/core/src/tools/executed_tool_calls.rs b/codex-rs/core/src/tools/executed_tool_calls.rs index 853c88a972..43c62916b7 100644 --- a/codex-rs/core/src/tools/executed_tool_calls.rs +++ b/codex-rs/core/src/tools/executed_tool_calls.rs @@ -33,7 +33,6 @@ use codex_protocol::openai_models::ToolMode; use indexmap::IndexMap; use serde_json::Value as JsonValue; -use crate::session::session::Session; use crate::session::step_context::StepContext; use crate::tools::context::ToolCallSource; use crate::tools::context::ToolOutput; @@ -385,6 +384,7 @@ impl ExecutedToolCalls { &self, item: &mut ResponseItem, prepared: Option<(ExecutedToolCall, DirectCallPermit)>, + item_id_generator: &codex_protocol::ItemIdGenerator, ) { let Some((call, permit)) = prepared else { return; @@ -396,9 +396,7 @@ impl ExecutedToolCalls { if !permit.recording.ptr_eq(&Arc::downgrade(&state.recording)) { return; } - // Assign the output's ordinary history ID here so reused model call IDs - // cannot associate this observation with a different output. - Session::assign_missing_response_item_id(item); + crate::session::assign_missing_item_id(item, item_id_generator); if let Some(id) = item.id() { state.direct_calls.insert(id.clone(), call.clone()); } diff --git a/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs b/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs index b068d9026e..3aa1384a67 100644 --- a/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs +++ b/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs @@ -154,6 +154,7 @@ fn large_direct_and_code_mode_metadata_survive_sampling_and_compaction() { recorder.attach_direct_call_to_output( &mut direct, Some((direct_call, recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); assert!(direct.executed_tool_call_metadata().is_none()); @@ -225,6 +226,7 @@ fn failed_compaction_preserves_observations_absent_from_trimmed_retries() { recorder.attach_direct_call_to_output( &mut direct, Some((direct_call, recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); assert!(direct.executed_tool_call_metadata().is_none()); let mut history = vec![direct.clone()]; @@ -280,6 +282,7 @@ fn compaction_keeps_direct_outputs_in_code_mode_binding_validation() { recorder.attach_direct_call_to_output( &mut direct, Some((call.clone(), recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); let mut compact = vec![exec_input("exec"), exec_output("exec"), direct.clone()]; recorder.attach_to_compaction_prompt(&mut compact); @@ -302,6 +305,7 @@ fn direct_retained_metadata_budget_sheds_results_then_calls_across_refresh() { recorder.attach_direct_call_to_output( &mut first, Some((call.clone(), recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); let mut expected_first = output("first"); expected_first.set_id(first.id().cloned()); @@ -337,6 +341,7 @@ fn direct_retained_metadata_budget_sheds_results_then_calls_across_refresh() { recorder.clone().attach_direct_call_to_output( &mut resource_only_output, Some((call.clone(), recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); expected_resource_only.set_id(resource_only_output.id().cloned()); assert_eq!(resource_only_output, expected_resource_only); @@ -361,6 +366,7 @@ fn direct_retained_metadata_budget_sheds_results_then_calls_across_refresh() { recorder.clone().attach_direct_call_to_output( &mut second, Some((call.clone(), recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); let mut expected_second = output("second"); expected_second.set_id(second.id().cloned()); @@ -400,6 +406,7 @@ fn direct_retained_metadata_budget_sheds_results_then_calls_across_refresh() { recorder.attach_direct_call_to_output( &mut third, Some((call, recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); assert!(third.executed_tool_call_metadata().is_none()); assert_eq!( @@ -509,13 +516,21 @@ async fn recorder_refreshes_without_changing_execution_features_or_claiming_miss if enabled { if let Some(stale) = prior_generation.take() { let mut stale_output = output("stale"); - calls.attach_direct_call_to_output(&mut stale_output, Some(stale)); + calls.attach_direct_call_to_output( + &mut stale_output, + Some(stale), + &session.item_id_generator, + ); assert!(stale_output.executed_tool_call_metadata().is_none()); } prior_generation = calls.prepare_direct_call(&call, &ToolCallSource::Direct, &step); } let mut direct_output = output(&call.call_id); - calls.attach_direct_call_to_output(&mut direct_output, prepared); + calls.attach_direct_call_to_output( + &mut direct_output, + prepared, + &session.item_id_generator, + ); assert_eq!( direct_output .executed_tool_call_metadata() @@ -537,7 +552,7 @@ async fn recorder_refreshes_without_changing_execution_features_or_claiming_miss let mut replay = output(&call.call_id); // Replay the same history item across the configuration refresh. replay.set_id(direct_output.id().cloned()); - calls.attach_direct_call_to_output(&mut replay, prepared); + calls.attach_direct_call_to_output(&mut replay, prepared, &session.item_id_generator); assert_eq!( replay.executed_tool_call_metadata(), direct_output.executed_tool_call_metadata() diff --git a/codex-rs/core/src/tools/executed_tool_calls_direct_tests.rs b/codex-rs/core/src/tools/executed_tool_calls_direct_tests.rs index bcf53892de..7514674d39 100644 --- a/codex-rs/core/src/tools/executed_tool_calls_direct_tests.rs +++ b/codex-rs/core/src/tools/executed_tool_calls_direct_tests.rs @@ -32,6 +32,7 @@ fn direct_request_uses_live_window_after_persisted_budget_is_exhausted() { recorder.attach_direct_call_to_output( &mut item, Some((call, recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); item }; @@ -99,7 +100,11 @@ fn direct_metadata_follows_output_ids_with_reused_call_ids_and_reverse_completio } // Completion order and prompt order must not choose which observation belongs here. for index in [1, 0] { - recorder.attach_direct_call_to_output(&mut outputs[index], prepared[index].take()); + recorder.attach_direct_call_to_output( + &mut outputs[index], + prepared[index].take(), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); assert!(outputs[index].executed_tool_call_metadata().is_none()); if supplied_ids { assert_eq!(outputs[index].id(), Some(&expected_ids[index])); @@ -161,6 +166,7 @@ fn direct_budget_keeps_an_omission_marker_when_it_fits() { recorder.attach_direct_call_to_output( &mut item, Some((call, recorder.reserve_direct_call().unwrap())), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); expected.set_id(item.id().cloned()); @@ -199,7 +205,11 @@ async fn direct_pending_limit_releases_on_completion_or_dropped_future() { let cloned = recorder.clone(); let mut overflow = output("overflow"); let before = serde_json::to_value(&overflow).expect("serializable output"); - cloned.attach_direct_call_to_output(&mut overflow, prepare(&cloned)); + cloned.attach_direct_call_to_output( + &mut overflow, + prepare(&cloned), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); assert_eq!( serde_json::to_value(&overflow).expect("serializable output"), before, @@ -216,7 +226,11 @@ async fn direct_pending_limit_releases_on_completion_or_dropped_future() { assert!(prepare(&recorder).is_none()); let mut completed = output("completed"); - recorder.attach_direct_call_to_output(&mut completed, pending.pop()); + recorder.attach_direct_call_to_output( + &mut completed, + pending.pop(), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); assert_eq!( completed .executed_tool_call_metadata() @@ -229,11 +243,20 @@ async fn direct_pending_limit_releases_on_completion_or_dropped_future() { recorder.refresh(&features); assert!(prepare(&recorder).is_none()); let mut stale = output("stale"); - recorder.attach_direct_call_to_output(&mut stale, Some(replacement)); + recorder.attach_direct_call_to_output( + &mut stale, + Some(replacement), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); + assert!(stale.id().is_none()); assert!(stale.executed_tool_call_metadata().is_none()); let current = prepare(&recorder).expect("stale call released its slot"); let mut current_output = output("current"); - recorder.attach_direct_call_to_output(&mut current_output, Some(current)); + recorder.attach_direct_call_to_output( + &mut current_output, + Some(current), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), + ); assert_eq!( current_output .executed_tool_call_metadata() @@ -269,6 +292,7 @@ async fn direct_recording_preserves_encoded_arguments_in_history() { recorder.attach_direct_call_to_output( &mut item, recorder.prepare_direct_call(&call, &ToolCallSource::Direct, &step), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); let wire = serde_json::to_value(&item).unwrap(); assert_eq!(wire["output"], ordinary_output); diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index 8696ad4e66..f11e64547b 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -113,6 +113,7 @@ impl ToolCallRuntime { let error_call = call.clone(); let source = call.direct_source(); let recorder = self.session.services.executed_tool_calls.clone(); + let item_id_generator = Arc::clone(&self.session.item_id_generator); let recorded_call = recorder.prepare_direct_call(&call, &source, &self.step_context); let step_context = Arc::clone(&self.step_context); let call_state = Arc::new(ToolCallState::default()); @@ -147,7 +148,11 @@ impl ToolCallRuntime { .get_or_insert_default() .delivered_assistant_message = Some(text.clone()); } - recorder.attach_direct_call_to_output(&mut response.item, recorded_call); + recorder.attach_direct_call_to_output( + &mut response.item, + recorded_call, + &item_id_generator, + ); Ok(response) } } diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index bcb28d6f1c..82a93700f4 100644 --- a/codex-rs/core/tests/responses_headers.rs +++ b/codex-rs/core/tests/responses_headers.rs @@ -17,6 +17,7 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use core_test_support::TestCodexResponsesRequestKind; use core_test_support::load_default_config_for_test; use core_test_support::responses; @@ -132,6 +133,7 @@ async fn responses_stream_includes_subagent_header_on_review() { codex_model_provider::create_model_provider(provider.clone(), /*auth_manager*/ None), AgentIdentityAuthPolicy::JwtOnly, thread_id, + uuid_v7_item_id_generator_for_test(), session_source.clone(), "test_originator".to_string(), config.model_verbosity, @@ -281,6 +283,7 @@ async fn responses_stream_includes_subagent_header_on_other() { codex_model_provider::create_model_provider(provider.clone(), /*auth_manager*/ None), AgentIdentityAuthPolicy::JwtOnly, thread_id, + uuid_v7_item_id_generator_for_test(), session_source.clone(), "test_originator".to_string(), config.model_verbosity, @@ -415,6 +418,7 @@ async fn responses_respects_model_info_overrides_from_config() { codex_model_provider::create_model_provider(provider.clone(), /*auth_manager*/ None), AgentIdentityAuthPolicy::JwtOnly, thread_id, + uuid_v7_item_id_generator_for_test(), session_source.clone(), "test_originator".to_string(), config.model_verbosity, diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 8455498749..7420519311 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -62,6 +62,7 @@ use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::user_input::UserInput; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use core_test_support::TestCodexResponsesRequestKind; use core_test_support::apps_test_server::AppsTestServer; use core_test_support::load_default_config_for_test; @@ -1832,6 +1833,7 @@ async fn send_request_with_provider(provider: ModelProviderInfo) { ), AgentIdentityAuthPolicy::JwtOnly, thread_id, + uuid_v7_item_id_generator_for_test(), SessionSource::Exec, "test_originator".to_string(), config.model_verbosity, @@ -3338,6 +3340,7 @@ async fn azure_responses_request_does_not_store_and_preserves_prefixed_item_ids( codex_model_provider::create_model_provider(provider.clone(), /*auth_manager*/ None), AgentIdentityAuthPolicy::JwtOnly, thread_id, + uuid_v7_item_id_generator_for_test(), SessionSource::Exec, "test_originator".to_string(), config.model_verbosity, diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 5880b9e151..8a48eb40cc 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -2957,6 +2957,7 @@ async fn websocket_harness_with_provider_factory( provider_factory(provider.clone(), client_auth_manager), AgentIdentityAuthPolicy::JwtOnly, thread_id, + codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), SessionSource::Exec, "test_originator".to_string(), config.model_verbosity, diff --git a/codex-rs/ext/guardian-reviewer/src/review_evidence.rs b/codex-rs/ext/guardian-reviewer/src/review_evidence.rs index bbba2205c1..dee93801f2 100644 --- a/codex-rs/ext/guardian-reviewer/src/review_evidence.rs +++ b/codex-rs/ext/guardian-reviewer/src/review_evidence.rs @@ -1,6 +1,7 @@ //! Selects Guardian answer evidence from the review's history snapshot and retains reviews. //! Retained answers survive restart even while incompatible checkpoints use legacy review. +use codex_protocol::ItemIdGenerator; use std::collections::BTreeSet; use std::collections::VecDeque; use std::sync::Arc; @@ -182,14 +183,15 @@ impl GuardianReviewEvidence { action: &str, authorization_version: GuardianAuthorizationVersion, review_context_revision: u64, - root_authorization_version: Option, - root_review_context_revision: Option, + root_versions: Option<(GuardianAuthorizationVersion, u64)>, + item_id_generator: &ItemIdGenerator, ) { let Some(completed_at_ms) = assessment.completed_at_ms else { return; }; + let (root_authorization_version, root_review_context_revision) = root_versions.unzip(); let review = Arc::new(GuardianReviewEvidenceRecord { - delivery_id: codex_protocol::ResponseItemId::new("guardian_review"), + delivery_id: item_id_generator("guardian_review"), completed_at_ms, authorization_version, review_context_revision, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs b/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs index 1a4776bd01..9da901f0fe 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs @@ -319,6 +319,7 @@ impl Classification { let extra_policy = config.guardian_extra_policy.as_deref().unwrap_or_default(); let instructions = guardian_config.render_classifier_instructions(policy, extra_policy); let sampling = LunaSamplingRequest { + item_id_generator: Arc::clone(thread.item_id_generator()), cyber_access_program, parent_response_id, instructions, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs index ea73bd791d..a6d20d3ab6 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs @@ -57,6 +57,7 @@ use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::TruncationPolicy; use codex_protocol::security_risk::SecurityRiskScore; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use core_test_support::ThreadIdle; use core_test_support::apps_test_server::HostedMessagingServer; use core_test_support::responses; @@ -479,7 +480,7 @@ fn user_instruction(text: &str) -> ResponseItem { ResponseItem::Message { status: None, encrypted_content: None, - id: Some(ResponseItemId::new("msg")), + id: Some(uuid_v7_item_id_generator_for_test()("msg")), role: "user".to_owned(), content: vec![ContentItem::InputText { text: text.to_owned(), @@ -2066,7 +2067,7 @@ async fn contributor_samples_tool_calls_with_the_existing_luna_pool() -> Result< ResponseItem::Message { status: None, encrypted_content: None, - id: Some(ResponseItemId::new("assistant")), + id: Some(uuid_v7_item_id_generator_for_test()("assistant")), role: "assistant".to_owned(), content: vec![ContentItem::OutputText { annotations: None, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs index 2dad92695d..a2fa24e1c5 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs @@ -25,7 +25,6 @@ use codex_http_client::HttpClientFactory; use codex_login::AgentIdentityAuthPolicy; use codex_model_provider::SharedModelProvider; use codex_model_provider::WorkspaceRoutingContext; -use codex_protocol::ResponseItemId; use codex_protocol::error::CodexErr; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; @@ -75,6 +74,8 @@ pub struct LunaSamplerConfig { /// One tool-less Luna classification request. pub struct LunaSamplingRequest { + /// Generator selected by the owning session for response items. + pub item_id_generator: codex_protocol::ItemIdGenerator, /// Cyber selection captured from the owning turn; Daybreak reviews use Blue. pub cyber_access_program: Option, /// ID of the response handling the classified tool. @@ -211,7 +212,7 @@ impl LunaSampler { if item.id().is_none() && let Some(prefix) = item.id_prefix() { - item.set_id(Some(ResponseItemId::new(prefix))); + item.set_id(Some((request.item_id_generator)(prefix))); } } let tokens = input diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs index 69ea60efe3..ff1069f077 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs @@ -25,6 +25,7 @@ use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::SessionSource; use codex_protocol::turn_input::CyberAccessProgram; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use core_test_support::responses; use core_test_support::responses::WebSocketConnectionConfig; use core_test_support::responses::ev_assistant_message; @@ -305,6 +306,7 @@ fn assert_classifier_instructions(request: &serde_json::Value) { pub(in crate::async_scorer) fn sample_request(parent_turn_id: &str) -> LunaSamplingRequest { LunaSamplingRequest { + item_id_generator: uuid_v7_item_id_generator_for_test(), cyber_access_program: None, parent_response_id: None, instructions: classifier_instructions(), @@ -702,6 +704,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications let first = sampler .sample(LunaSamplingRequest { + item_id_generator: uuid_v7_item_id_generator_for_test(), cyber_access_program: None, parent_response_id: None, instructions: classifier_instructions(), @@ -748,6 +751,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications sampler.prewarm().await; let second = sampler .sample(LunaSamplingRequest { + item_id_generator: uuid_v7_item_id_generator_for_test(), cyber_access_program: None, parent_response_id: None, instructions: classifier_instructions(), diff --git a/codex-rs/guardian-context/src/composition_tests.rs b/codex-rs/guardian-context/src/composition_tests.rs index 862bf57c0a..e18df72fde 100644 --- a/codex-rs/guardian-context/src/composition_tests.rs +++ b/codex-rs/guardian-context/src/composition_tests.rs @@ -6,7 +6,7 @@ use pretty_assertions::assert_eq; #[test] fn delivery_preserves_arbitrary_message_boundaries_and_rejects_them_for_sync() { let message = crate::PreviousReviews::try_from_fragments(vec![crate::PreviousReview { - id: codex_protocol::ResponseItemId::new("review"), + id: codex_protocol_test_support::uuid_v7_item_id_generator_for_test()("review"), fragment: "host-attested review".into(), }]) .unwrap() diff --git a/codex-rs/guardian-context/src/enforcement_tests.rs b/codex-rs/guardian-context/src/enforcement_tests.rs index ac5a3e46ad..39f88ea1f4 100644 --- a/codex-rs/guardian-context/src/enforcement_tests.rs +++ b/codex-rs/guardian-context/src/enforcement_tests.rs @@ -395,7 +395,7 @@ fn planned_action_budget_omits_descriptions_without_changing_arguments() { #[test] fn budget_reserves_existing_context_and_preserves_required_messages() { let trusted = crate::PreviousReviews::try_from_fragments(vec![crate::PreviousReview { - id: codex_protocol::ResponseItemId::new("review"), + id: codex_protocol_test_support::uuid_v7_item_id_generator_for_test()("review"), fragment: "verified review".to_owned(), }]) .unwrap() diff --git a/codex-rs/guardian-context/src/registry_tests.rs b/codex-rs/guardian-context/src/registry_tests.rs index e03396c7c2..54a933f48f 100644 --- a/codex-rs/guardian-context/src/registry_tests.rs +++ b/codex-rs/guardian-context/src/registry_tests.rs @@ -204,7 +204,7 @@ fn reused_registry_preserves_section_identity_and_source_roles() { ]; let answers = ["assistant: Publish?\nuser: No.\n".to_string()]; let reviews = super::PreviousReviews::try_from_fragments(vec![super::PreviousReview { - id: codex_protocol::ResponseItemId::new("review"), + id: codex_protocol_test_support::uuid_v7_item_id_generator_for_test()("review"), fragment: "debug-secret review".to_string(), }]) .unwrap(); diff --git a/codex-rs/guardian-context/src/retained_instructions_tests.rs b/codex-rs/guardian-context/src/retained_instructions_tests.rs index 3f79536aea..6f88426fbf 100644 --- a/codex-rs/guardian-context/src/retained_instructions_tests.rs +++ b/codex-rs/guardian-context/src/retained_instructions_tests.rs @@ -4,6 +4,7 @@ use codex_history::RetainedInputSource; use codex_history::RetainedUserMessage; use codex_history::VerifiedAnswer; use codex_history::VerifiedQuestionAnswer; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use pretty_assertions::assert_eq; #[test] @@ -11,17 +12,20 @@ fn instructions_preserve_source_order_and_whole_records() { let mut context = RetainedContext::default(); // An answer at the existing 900-token byte limit still fits with order framing. let answer = "x".repeat(3_600 - "assistant: Publish?\nuser: \n".len()); - context.record(&RetainedContextEvent::VerifiedAnswer { - answer: VerifiedAnswer { - turn_id: "grant".to_owned(), - call_id: "ask".to_owned(), - questions: vec![VerifiedQuestionAnswer { - question: "Publish?".to_owned(), - answer: answer.clone(), - }], + context.record( + &RetainedContextEvent::VerifiedAnswer { + answer: VerifiedAnswer { + turn_id: "grant".to_owned(), + call_id: "ask".to_owned(), + questions: vec![VerifiedQuestionAnswer { + question: "Publish?".to_owned(), + answer: answer.clone(), + }], + }, + acceptance_order: None, }, - acceptance_order: None, - }); + &uuid_v7_item_id_generator_for_test(), + ); context.record_user_message( RetainedUserMessage { phase: None, @@ -32,6 +36,7 @@ fn instructions_preserve_source_order_and_whole_records() { complete: true, }, RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); let rendered = render_retained_instructions( &context, @@ -65,6 +70,7 @@ fn instructions_preserve_source_order_and_whole_records() { complete: true, }, RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); let rendered = render_retained_instructions( &context, @@ -108,6 +114,7 @@ fn ordinary_exchanges_keep_roles_and_drop_assistant_context_before_restrictions( complete: true, }, RetainedInputSource::Local(Some(0)), + &uuid_v7_item_id_generator_for_test(), ); context.record_user_message( RetainedUserMessage { @@ -119,6 +126,7 @@ fn ordinary_exchanges_keep_roles_and_drop_assistant_context_before_restrictions( complete: true, }, RetainedInputSource::Local(Some(1)), + &uuid_v7_item_id_generator_for_test(), ); for presentation in [ crate::ContextPresentation::SyncFull { session_id: "test" }, @@ -208,6 +216,7 @@ fn ordinary_exchanges_keep_roles_and_drop_assistant_context_before_restrictions( complete: true, }, RetainedInputSource::Local(Some(2)), + &uuid_v7_item_id_generator_for_test(), ); let rendered = render_retained_instructions( &context, @@ -258,17 +267,20 @@ fn legacy_verified_answers_keep_distinct_source_order() { "Retained source order: 1\nassistant: Still publish?\nuser: No.\n".to_owned(), ] ); - context.record(&RetainedContextEvent::VerifiedAnswer { - answer: VerifiedAnswer { - turn_id: "large".into(), - call_id: "large".into(), - questions: vec![VerifiedQuestionAnswer { - question: "Publish?".into(), - answer: "Only if conditions hold. ".repeat(200), - }], + context.record( + &RetainedContextEvent::VerifiedAnswer { + answer: VerifiedAnswer { + turn_id: "large".into(), + call_id: "large".into(), + questions: vec![VerifiedQuestionAnswer { + question: "Publish?".into(), + answer: "Only if conditions hold. ".repeat(200), + }], + }, + acceptance_order: None, }, - acceptance_order: None, - }); + &uuid_v7_item_id_generator_for_test(), + ); let raw = crate::collect_verified_answer_candidates(&context); let old = crate::render_verified_answers(&context); assert!(!old.complete); @@ -316,7 +328,11 @@ fn delivery_uses_source_revision_and_complete_host_metadata() { ) .retained_instructions() }; - retained.record_user_message(message.clone(), RetainedInputSource::Local(Some(4))); + retained.record_user_message( + message.clone(), + RetainedInputSource::Local(Some(4)), + &uuid_v7_item_id_generator_for_test(), + ); let original = compose(&retained); let delivered = original.clone().into_annotated_messages(); let mut next = original.clone(); @@ -367,7 +383,11 @@ fn delivery_uses_source_revision_and_complete_host_metadata() { // A correction keeps its source ID and acceptance order but gets a new revision. message.text = "Do not draft or send.".to_owned(); - retained.record_user_message(message, RetainedInputSource::Local(Some(4))); + retained.record_user_message( + message, + RetainedInputSource::Local(Some(4)), + &uuid_v7_item_id_generator_for_test(), + ); let retained: RetainedContext = serde_json::from_value(serde_json::to_value(retained).unwrap()).unwrap(); let mut corrected = compose(&retained); @@ -503,6 +523,7 @@ fn transcript_original_requires_complete_source_proof_and_survives_budgeting() { phase: None, }, RetainedInputSource::Local(Some(6)), + &uuid_v7_item_id_generator_for_test(), ); let source = retained .source(retained.ordered_entries().next().unwrap().1) @@ -512,17 +533,20 @@ fn transcript_original_requires_complete_source_proof_and_survives_budgeting() { "original".to_owned(), ))); // The later verified grant renders before the earlier transcript restriction. - retained.record(&RetainedContextEvent::VerifiedAnswer { - answer: VerifiedAnswer { - turn_id: "grant".to_owned(), - call_id: "approve-send".to_owned(), - questions: vec![VerifiedQuestionAnswer { - question: "May I send now?".to_owned(), - answer: "Yes.".to_owned(), - }], + retained.record( + &RetainedContextEvent::VerifiedAnswer { + answer: VerifiedAnswer { + turn_id: "grant".to_owned(), + call_id: "approve-send".to_owned(), + questions: vec![VerifiedQuestionAnswer { + question: "May I send now?".to_owned(), + answer: "Yes.".to_owned(), + }], + }, + acceptance_order: Some(7), }, - acceptance_order: Some(7), - }); + &uuid_v7_item_id_generator_for_test(), + ); let mut history = RetainedTranscript { retained, messages: vec![ResponseItemEnvelope { @@ -607,6 +631,7 @@ fn coalesced_repl_text_cannot_attest_to_host_omission_delivery() { complete: true, }, RetainedInputSource::Local(Some(0)), + &uuid_v7_item_id_generator_for_test(), ); let profile = crate::ContextProfile::synchronous(); let mut previous = transcript_context(&history, &profile); diff --git a/codex-rs/guardian-context/src/reviews_tests.rs b/codex-rs/guardian-context/src/reviews_tests.rs index f3b33a0b16..820dcb17e3 100644 --- a/codex-rs/guardian-context/src/reviews_tests.rs +++ b/codex-rs/guardian-context/src/reviews_tests.rs @@ -1,6 +1,7 @@ //! Constructor boundaries preserve valid evidence and reject oversize sections. use super::*; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use pretty_assertions::assert_eq; #[test] @@ -9,7 +10,7 @@ fn previous_reviews_checks_count_and_size_without_rewriting_fragments() { let fragment = "é".repeat(max_bytes / 2); let fragments = vec![ PreviousReview { - id: ResponseItemId::new("review"), + id: uuid_v7_item_id_generator_for_test()("review"), fragment: fragment.clone(), }; MAX_PREVIOUS_REVIEWS @@ -37,7 +38,7 @@ fn previous_reviews_checks_count_and_size_without_rewriting_fragments() { invalid .into_iter() .map(|fragment| PreviousReview { - id: ResponseItemId::new("review"), + id: uuid_v7_item_id_generator_for_test()("review"), fragment, }) .collect() @@ -52,12 +53,12 @@ fn previous_reviews_checks_count_and_size_without_rewriting_fragments() { #[test] fn review_delivery_requires_complete_host_metadata_and_preserves_distinct_completions() { let first = PreviousReview { - id: ResponseItemId::new("review"), + id: uuid_v7_item_id_generator_for_test()("review"), fragment: "Approved reading README.md.".to_owned(), }; // A different completion remains new evidence even when its text is identical. let second = PreviousReview { - id: ResponseItemId::new("review"), + id: uuid_v7_item_id_generator_for_test()("review"), fragment: first.fragment.clone(), }; let first_message = PreviousReviews::try_from_fragments(vec![first.clone()]) diff --git a/codex-rs/guardian-context/tests/cache_prefix.rs b/codex-rs/guardian-context/tests/cache_prefix.rs index 1a6f9c9fca..eb4a88e9ab 100644 --- a/codex-rs/guardian-context/tests/cache_prefix.rs +++ b/codex-rs/guardian-context/tests/cache_prefix.rs @@ -21,6 +21,7 @@ use codex_history::RetainedUserMessage; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use codex_protocol::user_input::UserInput; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use pretty_assertions::assert_eq; struct History { @@ -67,6 +68,7 @@ fn changing_retained_context_and_attestations_preserves_history_before_the_curre complete: true, }, RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); // Legacy review history and Felix's retained, thread-owned history use the // same composer. Retained assistant growth and eviction must also preserve its prefix. @@ -98,11 +100,12 @@ fn changing_retained_context_and_attestations_preserves_history_before_the_curre complete: true, }, RetainedInputSource::Local(Some(generation + 1)), + &uuid_v7_item_id_generator_for_test(), ); } let reviews = PreviousReviews::try_from_fragments(vec![codex_guardian_context::PreviousReview { - id: codex_protocol::ResponseItemId::new("review"), + id: uuid_v7_item_id_generator_for_test()("review"), fragment: format!("Host-attested decision {generation}: denied."), }]) .unwrap(); diff --git a/codex-rs/history/Cargo.toml b/codex-rs/history/Cargo.toml index bd938d2c2e..984bf70647 100644 --- a/codex-rs/history/Cargo.toml +++ b/codex-rs/history/Cargo.toml @@ -19,5 +19,6 @@ serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } [dev-dependencies] +codex-protocol-test-support = { workspace = true } anyhow = { workspace = true } pretty_assertions = { workspace = true } diff --git a/codex-rs/history/src/compaction_checkpoint_tests.rs b/codex-rs/history/src/compaction_checkpoint_tests.rs index acbbd2007d..6cae38cd87 100644 --- a/codex-rs/history/src/compaction_checkpoint_tests.rs +++ b/codex-rs/history/src/compaction_checkpoint_tests.rs @@ -2,14 +2,14 @@ use super::*; use crate::CodexHarnessMetadata; -use codex_protocol::ResponseItemId; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use pretty_assertions::assert_eq; #[test] fn latest_checkpoint_keeps_its_own_producer_even_when_unusable() { let older = ResponseItemEnvelope { item: ResponseItem::Compaction { - id: Some(ResponseItemId::new("old")), + id: Some(uuid_v7_item_id_generator_for_test()("old")), encrypted_content: "older checkpoint".to_owned(), internal_chat_message_metadata_passthrough: None, }, @@ -21,7 +21,7 @@ fn latest_checkpoint_keeps_its_own_producer_even_when_unusable() { for content in [None, Some(String::new()), Some("new checkpoint".to_owned())] { let usable = content.as_ref().is_some_and(|text| !text.is_empty()); let latest = ResponseItem::ContextCompaction { - id: Some(ResponseItemId::new("new")), + id: Some(uuid_v7_item_id_generator_for_test()("new")), encrypted_content: content, internal_chat_message_metadata_passthrough: None, }; diff --git a/codex-rs/history/src/reconciled_retained_context_tests.rs b/codex-rs/history/src/reconciled_retained_context_tests.rs index 25568d4677..23a236e140 100644 --- a/codex-rs/history/src/reconciled_retained_context_tests.rs +++ b/codex-rs/history/src/reconciled_retained_context_tests.rs @@ -3,6 +3,7 @@ use crate::RetainedContextEvent; use crate::RetainedInputSource; use crate::VerifiedAnswer; use crate::VerifiedQuestionAnswer; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use pretty_assertions::assert_eq; fn instruction(text: &str) -> RetainedUserMessage { @@ -28,19 +29,30 @@ fn recovery_preserves_source_identity_acceptance_order_and_checkpoint_gaps() { ..retained_excerpt.clone() }; let mut retained = RetainedContext::default(); - retained.record_user_message(initial.clone(), RetainedInputSource::Local(Some(0))); - retained.record_user_message(retained_excerpt, RetainedInputSource::Local(Some(1))); - retained.record(&RetainedContextEvent::VerifiedAnswer { - answer: VerifiedAnswer { - turn_id: "answer-turn".to_owned(), - call_id: "publish-question".to_owned(), - questions: vec![VerifiedQuestionAnswer { - question: "Publish?".to_owned(), - answer: "Never publicly.".to_owned(), - }], + retained.record_user_message( + initial.clone(), + RetainedInputSource::Local(Some(0)), + &uuid_v7_item_id_generator_for_test(), + ); + retained.record_user_message( + retained_excerpt, + RetainedInputSource::Local(Some(1)), + &uuid_v7_item_id_generator_for_test(), + ); + retained.record( + &RetainedContextEvent::VerifiedAnswer { + answer: VerifiedAnswer { + turn_id: "answer-turn".to_owned(), + call_id: "publish-question".to_owned(), + questions: vec![VerifiedQuestionAnswer { + question: "Publish?".to_owned(), + answer: "Never publicly.".to_owned(), + }], + }, + acceptance_order: Some(3), }, - acceptance_order: Some(3), - }); + &uuid_v7_item_id_generator_for_test(), + ); retained.mark_user_messages_incomplete(); let checkpoint = retained.clone(); @@ -112,6 +124,7 @@ fn recovery_marks_missing_and_conflicting_orders_incomplete() { retained.record_user_message( instruction("Inspect the deployment."), RetainedInputSource::Local(Some(1)), + &uuid_v7_item_id_generator_for_test(), ); assert!(!retained.has_missing_user_messages()); let checkpoint = retained.clone(); @@ -180,6 +193,7 @@ fn heartbeat_versions_survive_retention_restore_and_reconciliation() { retained.record_user_message( message.clone(), RetainedInputSource::Local(Some(index as u64)), + &uuid_v7_item_id_generator_for_test(), ); } let expected = [0, 1, 31, 32, 33] diff --git a/codex-rs/history/src/retained_assistant_messages.rs b/codex-rs/history/src/retained_assistant_messages.rs index f366551ba2..b942f1decf 100644 --- a/codex-rs/history/src/retained_assistant_messages.rs +++ b/codex-rs/history/src/retained_assistant_messages.rs @@ -9,6 +9,7 @@ use super::RetainedSource; use super::RetainedSourceRole; use super::RetainedUserMessage; use super::bound_family; +use codex_protocol::ItemIdGenerator; impl RetainedContext { /// Reports observed storage loss or a retained message that cannot be delivered whole. @@ -38,6 +39,7 @@ impl RetainedContext { &mut self, mut message: RetainedUserMessage, source: RetainedInputSource, + item_id_generator: &ItemIdGenerator, ) -> Option { if source == RetainedInputSource::Local(None) { // Leave unsequenced sources to legacy transcript selection. Missing @@ -73,7 +75,7 @@ impl RetainedContext { let revision = message .message_id .as_ref() - .map(|_| codex_protocol::ResponseItemId::new("retained")); + .map(|_| item_id_generator("retained")); let entry = Ordered { revision, inherited, diff --git a/codex-rs/history/src/retained_context.rs b/codex-rs/history/src/retained_context.rs index f827425649..f883348845 100644 --- a/codex-rs/history/src/retained_context.rs +++ b/codex-rs/history/src/retained_context.rs @@ -11,6 +11,7 @@ pub use source::RetainedSource; pub use source::RetainedSourceId; pub use source::RetainedSourceRole; +use codex_protocol::ItemIdGenerator; use std::collections::BTreeSet; use std::collections::VecDeque; use std::sync::Arc; @@ -386,6 +387,7 @@ impl RetainedContext { &mut self, mut message: RetainedUserMessage, source: RetainedInputSource, + item_id_generator: &ItemIdGenerator, ) -> Option { message.bound(); let inherited = source == RetainedInputSource::Inherited; @@ -437,7 +439,7 @@ impl RetainedContext { let revision = message .message_id .as_ref() - .map(|_| codex_protocol::ResponseItemId::new("retained")); + .map(|_| item_id_generator("retained")); let entry = Ordered { revision, inherited, @@ -470,7 +472,11 @@ impl RetainedContext { } /// Same-event delivery is idempotent; changed contents replace that source's record. - pub fn record(&mut self, event: &RetainedContextEvent) -> bool { + pub fn record( + &mut self, + event: &RetainedContextEvent, + item_id_generator: &ItemIdGenerator, + ) -> bool { let mut event = event.clone(); event.bound(); match event { @@ -524,6 +530,7 @@ impl RetainedContext { self.record_assistant_message( message, RetainedInputSource::Local(Some(acceptance_order)), + item_id_generator, ); true } diff --git a/codex-rs/history/src/retained_context_tests.rs b/codex-rs/history/src/retained_context_tests.rs index 4d569d1102..8d71cf59fa 100644 --- a/codex-rs/history/src/retained_context_tests.rs +++ b/codex-rs/history/src/retained_context_tests.rs @@ -2,6 +2,7 @@ use super::*; use crate::CodexHarnessMetadata; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use pretty_assertions::assert_eq; #[test] @@ -15,8 +16,9 @@ fn accepted_audio_checkpoint_discards_oversized_and_rolled_back_evidence() { file_ids: BTreeSet::from([id]), }), }; - assert!(retained.record(&proof("file_a".into()))); - assert!(!retained.record(&proof("x".repeat(MAX_RECORD_BYTES)))); + let item_id_generator = uuid_v7_item_id_generator_for_test(); + assert!(retained.record(&proof("file_a".into()), &item_id_generator)); + assert!(!retained.record(&proof("x".repeat(MAX_RECORD_BYTES)), &item_id_generator)); let bytes = serde_json::to_vec(&retained).expect("encode checkpoint"); let mut restored: RetainedContext = serde_json::from_slice(&bytes).expect("decode checkpoint"); let get_ids = |state: &RetainedContext| { @@ -65,7 +67,7 @@ fn delivered_message(id: &str, text: &str, acceptance_order: u64) -> RetainedCon fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() { let mut context = RetainedContext::default(); let first = publish_answer(); - assert!(context.record(&first)); + assert!(context.record(&first, &uuid_v7_item_id_generator_for_test())); let before_restriction = context.clone(); context.record_user_message( RetainedUserMessage { @@ -77,6 +79,7 @@ fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() complete: false, }, RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); let mut expected = context.clone(); expected.user_messages[0].value.text = "Do not publish after all.".to_owned(); @@ -87,7 +90,7 @@ fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() assert_eq!(context, expected); context.recover_user_message_excerpts(|_| panic!("existing text must not be replaced")); let snapshot = context.clone(); - assert!(!context.record(&first)); + assert!(!context.record(&first, &uuid_v7_item_id_generator_for_test())); assert_eq!(context, snapshot); let checkpoint = serde_json::from_str(&serde_json::to_string(&context).expect("retained answer fixture")) @@ -124,21 +127,24 @@ fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() fn retained_families_enforce_storage_limits_without_changing_snapshots() { let mut context = RetainedContext::default(); let first = publish_answer(); - context.record(&first); + context.record(&first, &uuid_v7_item_id_generator_for_test()); let snapshot = context.clone(); for index in 2..=10 { - context.record(&RetainedContextEvent::VerifiedAnswer { - answer: VerifiedAnswer { - turn_id: "turn-2".to_owned(), - call_id: format!("ask-{index}"), - questions: vec![VerifiedQuestionAnswer { - question: "Continue?".to_owned(), - answer: "Yes".to_owned(), - }], + context.record( + &RetainedContextEvent::VerifiedAnswer { + answer: VerifiedAnswer { + turn_id: "turn-2".to_owned(), + call_id: format!("ask-{index}"), + questions: vec![VerifiedQuestionAnswer { + question: "Continue?".to_owned(), + answer: "Yes".to_owned(), + }], + }, + acceptance_order: None, }, - acceptance_order: None, - }); + &uuid_v7_item_id_generator_for_test(), + ); } assert!(!context.verified_answers_complete()); assert_eq!(context.verified_answers().count(), MAX_FAMILY_RECORDS); @@ -156,7 +162,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { panic!("verified answer fixture"); }; answer.questions[0].answer = "a".repeat(MAX_RECORD_BYTES); - restored.record(&oversized); + restored.record(&oversized, &uuid_v7_item_id_generator_for_test()); assert!(!restored.verified_answers_complete()); assert!( restored @@ -186,6 +192,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { complete: true, }, RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); } assert!(!restored.user_messages_complete()); @@ -207,6 +214,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { complete: true, }, RetainedInputSource::Local(Some(0)), + &uuid_v7_item_id_generator_for_test(), ); assert_eq!( restored, recent, @@ -222,6 +230,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { complete: true, }, RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); let Some((_, RetainedContextEntry::UserMessage(message))) = restored.ordered_entries().next_back() @@ -241,18 +250,19 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { ..restrictions.back().unwrap().value.clone() }, RetainedInputSource::Local(Some(order)), + &uuid_v7_item_id_generator_for_test(), ); } assert!(!restored.has_omitted_assistant_messages()); let evicted = delivered_message("evicted", "Already replaced?", /*acceptance_order*/ 0); - assert!(restored.record(&evicted)); + assert!(restored.record(&evicted, &uuid_v7_item_id_generator_for_test())); // The old-order delivery is immediately evicted, so the bounded buffer // cannot prove a subsequent recording with this ID is a duplicate. - assert!(restored.record(&evicted)); + assert!(restored.record(&evicted, &uuid_v7_item_id_generator_for_test())); let order = restored.reserve_order(); let oversized = delivered_message("oversized", &"x".repeat(MAX_RECORD_BYTES), order); - assert!(restored.record(&oversized)); - assert!(!restored.record(&oversized)); + assert!(restored.record(&oversized, &uuid_v7_item_id_generator_for_test())); + assert!(!restored.record(&oversized, &uuid_v7_item_id_generator_for_test())); assert_eq!(restored.user_messages, restrictions); assert_eq!(restored.verified_answers, authorization); assert_eq!(restored.assistant_messages.len(), MAX_FAMILY_RECORDS); @@ -271,8 +281,8 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { // Once the omission marker is set, another confirmed delivery still needs // its own rollout record even when the bounded window stays incomplete. let later = delivered_message("later", "A later question?", order + 1); - assert!(restored.record(&later)); - assert!(!restored.record(&later)); + assert!(restored.record(&later, &uuid_v7_item_id_generator_for_test())); + assert!(!restored.record(&later, &uuid_v7_item_id_generator_for_test())); } #[test] @@ -289,6 +299,7 @@ fn recovered_excerpts_obey_record_and_family_limits() { complete: false, }, RetainedInputSource::Local(None), + &uuid_v7_item_id_generator_for_test(), ); } let unchanged = context.clone(); @@ -381,7 +392,7 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { answer, acceptance_order: Some(answer_order), }; - context.record(&event); + context.record(&event, &uuid_v7_item_id_generator_for_test()); let instruction = RetainedUserMessage { phase: None, origin: crate::UserInputOrigin::User, @@ -393,16 +404,21 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { // Assistant delivery after accepted steering must not move the steering past it. let assistant_order = context.reserve_order(); let delivered = delivered_message("assistant", "Publish publicly?", assistant_order); - assert!(context.record(&delivered)); + assert!(context.record(&delivered, &uuid_v7_item_id_generator_for_test())); let checkpoint = serde_json::from_value(serde_json::to_value(&context).unwrap()).unwrap(); context.record_user_message( instruction.clone(), RetainedInputSource::Local(Some(steer_order)), + &uuid_v7_item_id_generator_for_test(), ); let mut resumed = RetainedContext::default(); resumed.restore(Some(&checkpoint), &[]); - assert!(!resumed.record(&delivered)); - resumed.record_user_message(instruction, RetainedInputSource::Local(Some(steer_order))); + assert!(!resumed.record(&delivered, &uuid_v7_item_id_generator_for_test())); + resumed.record_user_message( + instruction, + RetainedInputSource::Local(Some(steer_order)), + &uuid_v7_item_id_generator_for_test(), + ); // This suffix has no captured version metadata, so replay creates a fresh revision. // Its original contents and acceptance order still reconstruct identically. let mut expected = context.clone(); @@ -471,7 +487,7 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { #[test] fn adopted_instructions_preserve_local_order_and_rollback_scope() { let mut context = RetainedContext::default(); - context.record(&publish_answer()); + context.record(&publish_answer(), &uuid_v7_item_id_generator_for_test()); for index in 0..2 { let message = RetainedUserMessage { phase: None, @@ -481,9 +497,21 @@ fn adopted_instructions_preserve_local_order_and_rollback_scope() { text: format!("Parent instruction {index}"), complete: true, }; - context.record_user_message(message.clone(), RetainedInputSource::Inherited); - context.record_user_message(message.clone(), RetainedInputSource::Inherited); - context.record_assistant_message(message, RetainedInputSource::Inherited); + context.record_user_message( + message.clone(), + RetainedInputSource::Inherited, + &uuid_v7_item_id_generator_for_test(), + ); + context.record_user_message( + message.clone(), + RetainedInputSource::Inherited, + &uuid_v7_item_id_generator_for_test(), + ); + context.record_assistant_message( + message, + RetainedInputSource::Inherited, + &uuid_v7_item_id_generator_for_test(), + ); } assert_eq!(context.reserve_order(), 1); assert!(!context.verified_answers_complete()); diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index f7a8357cef..6c8784fe5f 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -326,6 +326,7 @@ fn delivered_assistant_rollout_survives_an_earlier_read_and_rewrite() -> Result< phase: None, }, RetainedInputSource::Local(metadata.user_input_order), + &codex_protocol_test_support::uuid_v7_item_id_generator_for_test(), ); let rewritten_checkpoint = serde_json::from_value(serde_json::to_value(earlier_checkpoint)?)?; diff --git a/codex-rs/memories/write/src/phase1.rs b/codex-rs/memories/write/src/phase1.rs index f5a5b42044..27dc570e6a 100644 --- a/codex-rs/memories/write/src/phase1.rs +++ b/codex-rs/memories/write/src/phase1.rs @@ -12,7 +12,6 @@ use codex_config::types::MemoriesConfig; use codex_core::Prompt; use codex_core::config::Config; use codex_protocol::MemoryVersion; -use codex_protocol::ResponseItemId; use codex_protocol::error::CodexErr; use codex_protocol::models::BaseInstructions; use codex_protocol::models::ContentItem; @@ -292,7 +291,7 @@ mod job { MemoryVersion::V1 => vec![ResponseItem::Message { status: None, encrypted_content: None, - id: Some(ResponseItemId::new("msg")), + id: Some(context.item_id_generator()("msg")), role: "user".to_string(), content: vec![ContentItem::InputText { text: input_text }], phase: None, diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index 1b995b9098..d13cf3d415 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -139,6 +139,10 @@ fn build_session_telemetry( } impl MemoryStartupContext { + pub(crate) fn item_id_generator(&self) -> &codex_protocol::ItemIdGenerator { + self.thread.item_id_generator() + } + pub(crate) fn new( thread_manager: Arc, auth_manager: Arc, @@ -336,6 +340,7 @@ impl MemoryStartupContext { Arc::clone(&self.provider), AgentIdentityAuthPolicy::JwtOnly, self.thread_id, + Arc::clone(self.item_id_generator()), session_source.clone(), config_snapshot.originator, config.model_verbosity, diff --git a/codex-rs/protocol/src/item_id_generator.rs b/codex-rs/protocol/src/item_id_generator.rs new file mode 100644 index 0000000000..60472a3c50 --- /dev/null +++ b/codex-rs/protocol/src/item_id_generator.rs @@ -0,0 +1,12 @@ +//! Response item ID generation shared by sessions and their consumers. + +use crate::ResponseItemId; +use std::sync::Arc; + +/// Generates response item IDs from a type prefix for an owning session. Clones share the same callable. +pub type ItemIdGenerator = Arc ResponseItemId + Send + Sync>; + +#[cfg(test)] +pub(super) fn uuid_v7_item_id_generator_for_test() -> ItemIdGenerator { + Arc::new(|prefix| ResponseItemId::with_suffix(prefix, uuid::Uuid::now_v7())) +} diff --git a/codex-rs/protocol/src/items.rs b/codex-rs/protocol/src/items.rs index 4905590176..3736095275 100644 --- a/codex-rs/protocol/src/items.rs +++ b/codex-rs/protocol/src/items.rs @@ -1,5 +1,5 @@ use crate::AgentPath; -use crate::ResponseItemId; +use crate::ItemIdGenerator; use crate::ThreadId; use crate::dynamic_tools::DynamicToolCallOutputContentItem; use crate::mcp::CallToolResult; @@ -727,7 +727,10 @@ impl HookPromptFragment { } } -pub fn build_hook_prompt_message(fragments: &[HookPromptFragment]) -> Option { +pub fn build_hook_prompt_message( + fragments: &[HookPromptFragment], + item_id_generator: &ItemIdGenerator, +) -> Option { let mut content = Vec::new(); let mut metadata = Vec::new(); for fragment in fragments { @@ -744,7 +747,7 @@ pub fn build_hook_prompt_message(fragments: &[HookPromptFragment]) -> Option Self { - Self::with_suffix(prefix, uuid::Uuid::now_v7()) - } - pub fn with_suffix(prefix: &str, suffix: impl fmt::Display) -> Self { Self(format!("{prefix}_{suffix}")) } diff --git a/codex-rs/protocol/src/response_item_id_tests.rs b/codex-rs/protocol/src/response_item_id_tests.rs index 40c8349578..2a7f17d170 100644 --- a/codex-rs/protocol/src/response_item_id_tests.rs +++ b/codex-rs/protocol/src/response_item_id_tests.rs @@ -5,7 +5,7 @@ use super::ResponseItemId; #[test] fn creates_prefixed_uuid_v7_ids() { - let id = ResponseItemId::new("msg"); + let id = crate::uuid_v7_item_id_generator_for_test()("msg"); let uuid = id.strip_prefix("msg_").expect("message prefix"); assert_eq!( Uuid::parse_str(uuid).expect("UUID suffix").get_version(), diff --git a/codex-rs/protocol/tests/support/Cargo.toml b/codex-rs/protocol/tests/support/Cargo.toml index 674cd71d8a..95cb65ed18 100644 --- a/codex-rs/protocol/tests/support/Cargo.toml +++ b/codex-rs/protocol/tests/support/Cargo.toml @@ -14,3 +14,4 @@ workspace = true [dependencies] codex-protocol = { workspace = true } +uuid = { workspace = true, features = ["v7"] } diff --git a/codex-rs/protocol/tests/support/lib.rs b/codex-rs/protocol/tests/support/lib.rs index 14671e8c0c..b4b363968f 100644 --- a/codex-rs/protocol/tests/support/lib.rs +++ b/codex-rs/protocol/tests/support/lib.rs @@ -1,6 +1,13 @@ -//! Text assertions for protocol content, shared only by test targets. +//! Protocol fixtures and text assertions shared by test targets. +use codex_protocol::ItemIdGenerator; +use codex_protocol::ResponseItemId; use codex_protocol::models::ContentItem; +use std::sync::Arc; + +pub fn uuid_v7_item_id_generator_for_test() -> ItemIdGenerator { + Arc::new(|prefix| ResponseItemId::with_suffix(prefix, uuid::Uuid::now_v7())) +} /// Assertion helpers for tests that require text rather than media content. /// Implementations should fail at the calling assertion for non-text content. diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml index 31e16dd72a..e0ba6c7df0 100644 --- a/codex-rs/thread-store/Cargo.toml +++ b/codex-rs/thread-store/Cargo.toml @@ -37,6 +37,7 @@ tracing = { workspace = true } zstd = { workspace = true } [dev-dependencies] +codex-protocol-test-support = { workspace = true } codex-utils-absolute-path = { workspace = true } pretty_assertions = { workspace = true } tempfile = { workspace = true } diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 920ed3e27a..f5a99ed3a5 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -48,6 +48,7 @@ use codex_protocol::protocol::UserMessageAudioKind; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::UserMessageImageKind; use codex_protocol::user_input::UserInput; +use codex_protocol_test_support::uuid_v7_item_id_generator_for_test; use codex_rollout::CompactedItem; use codex_rollout::RetainedContextEvent; use codex_rollout::RolloutConfig; @@ -1330,8 +1331,8 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() { .retained_context .as_mut() .expect("retained context"); - context.record(&answer_event); - context.record(&wake_answer_event); + context.record(&answer_event, &uuid_v7_item_id_generator_for_test()); + context.record(&wake_answer_event, &uuid_v7_item_id_generator_for_test()); let path = write_rollout( home.path(), thread_id, @@ -1429,7 +1430,7 @@ async fn migration_preserves_answers_before_a_rolled_back_steer() { .retained_context .as_mut() .expect("retained context") - .record(&answer); + .record(&answer, &uuid_v7_item_id_generator_for_test()); answers.push(answer); } // Delayed delivery must use the original call boundary, not the turn's latest steer. @@ -1551,12 +1552,13 @@ async fn migration_rolls_back_nested_deliveries_by_acceptance_order() { })) .expect("retained instruction"), codex_rollout::RetainedInputSource::Local(Some(order)), + &uuid_v7_item_id_generator_for_test(), ); } - context.record(&deliveries[0]); + context.record(&deliveries[0], &uuid_v7_item_id_generator_for_test()); let first_retained_message = serde_json::to_value(&*context).expect("retained context")["assistant_messages"][0].clone(); - context.record(&deliveries[1]); + context.record(&deliveries[1], &uuid_v7_item_id_generator_for_test()); let outer_call = serde_json::from_value(json!({ "type": "function_call", "call_id": "exec", "name": "exec", "arguments": "{}" }))