agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[codex-analytics] rework thread_source for thread analytics (#20949)

Browse Source 
## Summary
- make `thread_source` an explicit optional thread-level field on
`thread/start`, `thread/fork`, and returned thread payloads
- persist `thread_source` in rollout/session metadata so resumed live
threads retain the original value
- replace the old best-effort `session_source` -> `thread_source`
mapping with an explicit caller-supplied analytics classification

## Why
Before this change, analytics `thread_source` was populated by a
best-effort mapping from `session_source`. `session_source` describes
the runtime/client surface, not the actual thread-level origin, so that
projection was not accurate enough to distinguish cases such as `user`,
`subagent`, `memory_consolidation`, and future thread origins reliably.

Making `thread_source` explicit keeps one thread-level analytics field
while letting callers provide the real classification directly instead
of recovering it indirectly from `session_source`.

## Impact
For new analytics events, `thread_source` now reflects the explicit
thread-level classification supplied by the caller rather than an
inferred value derived from `session_source`. Existing protocol fields
remain optional; callers that omit `threadSource` now produce `null`
instead of a best-effort inferred value.

## Validation
- `just write-app-server-schema`
- `cargo test -p codex-analytics -p codex-core -p
codex-app-server-protocol --no-run`
- `cargo test -p codex-app-server-protocol
generated_ts_optional_nullable_fields_only_in_params`
- `cargo test -p codex-analytics
thread_initialized_event_serializes_expected_shape`
- `cargo test -p codex-core
resume_stopped_thread_from_rollout_preserves_thread_source`
This commit is contained in:
rhan-oai
2026-05-05 19:12:31 -07:00
committed by GitHub
Unverified
parent 94db03d5af
commit b3d4f1a9f0
98 changed files with 896 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+3
View File
@@ -27,6 +27,7 @@ use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::user_input::UserInput;
use codex_thread_store::ReadThreadParams;
@@ -234,6 +235,7 @@ impl AgentControl {
config.clone(),
self.clone(),
session_source,
/*thread_source*/ Some(ThreadSource::Subagent),
/*persist_extended_history*/ false,
/*metrics_service_name*/ None,
inherited_shell_snapshot,
@@ -420,6 +422,7 @@ impl AgentControl {
InitialHistory::Forked(forked_rollout_items),
self.clone(),
session_source,
/*thread_source*/ Some(ThreadSource::Subagent),
/*persist_extended_history*/ false,
inherited_shell_snapshot,
inherited_exec_policy,
codex-rs/core/src/codex_delegate.rs
+2
View File
@@ -16,6 +16,7 @@ use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::Submission;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::request_permissions::PermissionGrantScope;
use codex_protocol::request_permissions::RequestPermissionsArgs;
use codex_protocol::request_permissions::RequestPermissionsEvent;
@@ -84,6 +85,7 @@ pub(crate) async fn run_codex_thread_interactive(
skills_watcher: Arc::clone(&parent_session.services.skills_watcher),
conversation_history: initial_history.unwrap_or(InitialHistory::New),
session_source: SessionSource::SubAgent(subagent_source.clone()),
thread_source: Some(ThreadSource::Subagent),
agent_control: parent_session.services.agent_control.clone(),
dynamic_tools: Vec::new(),
persist_extended_history: false,
codex-rs/core/src/codex_thread.rs
+2
View File
@@ -30,6 +30,7 @@ use codex_protocol::protocol::SessionConfiguredEvent;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::Submission;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TokenUsageInfo;
use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::user_input::UserInput;
@@ -62,6 +63,7 @@ pub struct ThreadConfigSnapshot {
pub reasoning_effort: Option<ReasoningEffort>,
pub personality: Option<Personality>,
pub session_source: SessionSource,
pub thread_source: Option<ThreadSource>,
}
impl ThreadConfigSnapshot {
codex-rs/core/src/personality_migration_tests.rs
+1
View File
@@ -73,6 +73,7 @@ async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::R
originator: "test_originator".to_string(),
cli_version: "test_version".to_string(),
source: SessionSource::Cli,
thread_source: None,
agent_path: None,
agent_nickname: None,
agent_role: None,
codex-rs/core/src/realtime_context_tests.rs
+1
View File
@@ -49,6 +49,7 @@ fn stored_thread(cwd: &str, title: &str, first_user_message: &str) -> StoredThre
cwd: PathBuf::from(cwd),
cli_version: "test".to_string(),
source: SessionSource::Cli,
thread_source: None,
agent_nickname: None,
agent_role: None,
agent_path: None,
codex-rs/core/src/session/mod.rs
+4
View File
@@ -110,6 +110,7 @@ use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::TurnContextNetworkItem;
@@ -391,6 +392,7 @@ pub(crate) struct CodexSpawnArgs {
pub(crate) skills_watcher: Arc<SkillsWatcher>,
pub(crate) conversation_history: InitialHistory,
pub(crate) session_source: SessionSource,
pub(crate) thread_source: Option<ThreadSource>,
pub(crate) agent_control: AgentControl,
pub(crate) dynamic_tools: Vec<DynamicToolSpec>,
pub(crate) persist_extended_history: bool,
@@ -453,6 +455,7 @@ impl Codex {
skills_watcher,
conversation_history,
session_source,
thread_source,
agent_control,
dynamic_tools,
persist_extended_history,
@@ -612,6 +615,7 @@ impl Codex {
app_server_client_name: None,
app_server_client_version: None,
session_source,
thread_source,
dynamic_tools,
persist_extended_history,
inherited_shell_snapshot,
codex-rs/core/src/session/review.rs
+2 -1
View File
@@ -103,7 +103,7 @@ pub(super) async fn spawn_review_thread(
let review_turn_id = sub_id.to_string();
let turn_metadata_state = Arc::new(TurnMetadataState::new(
sess.conversation_id.to_string(),
&session_source,
parent_turn_context.thread_source,
review_turn_id.clone(),
parent_turn_context.cwd.clone(),
&parent_turn_context.permission_profile,
@@ -123,6 +123,7 @@ pub(super) async fn spawn_review_thread(
reasoning_effort,
reasoning_summary,
session_source,
thread_source: parent_turn_context.thread_source,
environments: parent_turn_context.environments.clone(),
tools_config,
features: parent_turn_context.features.clone(),
codex-rs/core/src/session/session.rs
+6
View File
@@ -2,6 +2,7 @@ use super::*;
use crate::goals::GoalRuntimeState;
use codex_protocol::permissions::FileSystemPath;
use codex_protocol::permissions::FileSystemSpecialPath;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnEnvironmentSelection;
use tokio::sync::Semaphore;
@@ -86,6 +87,8 @@ pub(crate) struct SessionConfiguration {
pub(super) app_server_client_version: Option<String>,
/// Source of the session (cli, vscode, exec, mcp, ...)
pub(super) session_source: SessionSource,
/// Optional analytics source classification for this thread.
pub(super) thread_source: Option<ThreadSource>,
pub(super) dynamic_tools: Vec<DynamicToolSpec>,
pub(super) persist_extended_history: bool,
pub(super) inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
@@ -141,6 +144,7 @@ impl SessionConfiguration {
reasoning_effort: self.collaboration_mode.reasoning_effort(),
personality: self.personality,
session_source: self.session_source.clone(),
thread_source: self.thread_source,
}
}
@@ -393,6 +397,7 @@ impl Session {
thread_id: conversation_id,
forked_from_id,
source: session_source,
thread_source: session_configuration.thread_source,
base_instructions: BaseInstructions {
text: session_configuration.base_instructions.clone(),
},
@@ -875,6 +880,7 @@ impl Session {
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id: conversation_id,
forked_from_id,
thread_source: session_configuration.thread_source,
thread_name: session_configuration.thread_name.clone(),
model: session_configuration.collaboration_mode.model().to_string(),
model_provider_id: config.model_provider_id.clone(),
codex-rs/core/src/session/tests.rs
+10
View File
@@ -1722,6 +1722,7 @@ async fn fork_startup_context_then_first_turn_diff_snapshot() -> anyhow::Result<
usize::MAX,
fork_config.clone(),
rollout_path,
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
)
@@ -2411,6 +2412,7 @@ async fn set_rate_limits_retains_previous_credits() {
app_server_client_name: None,
app_server_client_version: None,
session_source: SessionSource::Exec,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
@@ -2514,6 +2516,7 @@ async fn set_rate_limits_updates_plan_type_when_present() {
app_server_client_name: None,
app_server_client_version: None,
session_source: SessionSource::Exec,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
@@ -2766,6 +2769,7 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf {
thread_id: session.conversation_id,
forked_from_id: None,
source: SessionSource::Exec,
thread_source: None,
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
metadata: ThreadPersistenceMetadata {
@@ -2972,6 +2976,7 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati
app_server_client_name: None,
app_server_client_version: None,
session_source: SessionSource::Exec,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
@@ -3495,6 +3500,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_zsh_path() {
app_server_client_name: None,
app_server_client_version: None,
session_source: SessionSource::Exec,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
@@ -3607,6 +3613,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
app_server_client_name: None,
app_server_client_version: None,
session_source: SessionSource::Exec,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
@@ -3829,6 +3836,7 @@ async fn make_session_with_config_and_rx(
app_server_client_name: None,
app_server_client_version: None,
session_source: SessionSource::Exec,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
@@ -4793,6 +4801,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() {
thread_id: session.conversation_id,
forked_from_id: None,
source: SessionSource::Exec,
thread_source: None,
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
metadata: ThreadPersistenceMetadata {
@@ -5116,6 +5125,7 @@ where
app_server_client_name: None,
app_server_client_version: None,
session_source: SessionSource::Exec,
thread_source: None,
dynamic_tools,
persist_extended_history: false,
inherited_shell_snapshot: None,
codex-rs/core/src/session/tests/guardian_tests.rs
+1
View File
@@ -752,6 +752,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
session_source: SessionSource::SubAgent(SubAgentSource::Other(
GUARDIAN_REVIEWER_NAME.to_string(),
)),
thread_source: None,
agent_control: AgentControl::default(),
dynamic_tools: Vec::new(),
persist_extended_history: false,
codex-rs/core/src/session/turn_context.rs
+5 -1
View File
@@ -5,6 +5,7 @@ use crate::environment_selection::ResolvedTurnEnvironments;
use codex_model_provider::SharedModelProvider;
use codex_model_provider::create_model_provider;
use codex_protocol::models::AdditionalPermissionProfile;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_sandboxing::compatibility_sandbox_policy_for_permission_profile;
use codex_sandboxing::policy_transforms::effective_file_system_sandbox_policy;
@@ -63,6 +64,7 @@ pub(crate) struct TurnContext {
pub(crate) reasoning_effort: Option<ReasoningEffortConfig>,
pub(crate) reasoning_summary: ReasoningSummaryConfig,
pub(crate) session_source: SessionSource,
pub(crate) thread_source: Option<ThreadSource>,
pub(crate) environments: ResolvedTurnEnvironments,
/// The session's absolute working directory. All relative paths provided
/// by the model as well as sandbox policies are resolved against this path
@@ -248,6 +250,7 @@ impl TurnContext {
reasoning_effort,
reasoning_summary: self.reasoning_summary,
session_source: self.session_source.clone(),
thread_source: self.thread_source,
environments: self.environments.clone(),
cwd: self.cwd.clone(),
current_date: self.current_date.clone(),
@@ -520,7 +523,7 @@ impl Session {
let per_turn_config = Arc::new(per_turn_config);
let turn_metadata_state = Arc::new(TurnMetadataState::new(
conversation_id.to_string(),
&session_source,
session_configuration.thread_source,
sub_id.clone(),
cwd.clone(),
&session_configuration.permission_profile(),
@@ -540,6 +543,7 @@ impl Session {
reasoning_effort,
reasoning_summary,
session_source,
thread_source: session_configuration.thread_source,
environments,
cwd,
current_date: Some(current_date),
codex-rs/core/src/thread_manager.rs
+29
View File
@@ -48,6 +48,7 @@ use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionConfiguredEvent;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnEnvironmentSelection;
@@ -219,6 +220,7 @@ pub struct StartThreadOptions {
pub config: Config,
pub initial_history: InitialHistory,
pub session_source: Option<SessionSource>,
pub thread_source: Option<ThreadSource>,
pub dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
pub persist_extended_history: bool,
pub metrics_service_name: Option<String>,
@@ -620,6 +622,7 @@ impl ThreadManager {
config,
initial_history: InitialHistory::New,
session_source: None,
thread_source: None,
dynamic_tools,
persist_extended_history,
metrics_service_name: None,
@@ -636,12 +639,16 @@ impl ThreadManager {
let session_source = options
.session_source
.unwrap_or_else(|| self.state.session_source.clone());
let thread_source = options
.thread_source
.or_else(|| options.initial_history.get_resumed_thread_source());
Box::pin(self.state.spawn_thread_with_source(
options.config,
options.initial_history,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
session_source,
thread_source,
options.dynamic_tools,
options.persist_extended_history,
options.metrics_service_name,
@@ -684,11 +691,13 @@ impl ThreadManager {
self.state.environment_manager.as_ref(),
&config.cwd,
);
let thread_source = initial_history.get_resumed_thread_source();
Box::pin(self.state.spawn_thread(
config,
initial_history,
auth_manager,
self.agent_control(),
thread_source,
Vec::new(),
persist_extended_history,
/*metrics_service_name*/ None,
@@ -713,6 +722,7 @@ impl ThreadManager {
InitialHistory::New,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
/*thread_source*/ None,
Vec::new(),
/*persist_extended_history*/ false,
/*metrics_service_name*/ None,
@@ -735,11 +745,13 @@ impl ThreadManager {
self.state.environment_manager.as_ref(),
&config.cwd,
);
let thread_source = initial_history.get_resumed_thread_source();
Box::pin(self.state.spawn_thread(
config,
initial_history,
auth_manager,
self.agent_control(),
thread_source,
Vec::new(),
/*persist_extended_history*/ false,
/*metrics_service_name*/ None,
@@ -817,6 +829,7 @@ impl ThreadManager {
snapshot: S,
config: Config,
path: PathBuf,
thread_source: Option<ThreadSource>,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread>
@@ -829,6 +842,7 @@ impl ThreadManager {
snapshot,
config,
history,
thread_source,
persist_extended_history,
parent_trace,
)
@@ -841,6 +855,7 @@ impl ThreadManager {
snapshot: S,
config: Config,
history: InitialHistory,
thread_source: Option<ThreadSource>,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread>
@@ -851,6 +866,7 @@ impl ThreadManager {
snapshot.into(),
config,
history,
thread_source,
persist_extended_history,
parent_trace,
)
@@ -862,6 +878,7 @@ impl ThreadManager {
snapshot: ForkSnapshot,
config: Config,
history: InitialHistory,
thread_source: Option<ThreadSource>,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread> {
@@ -876,6 +893,7 @@ impl ThreadManager {
history,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
thread_source,
Vec::new(),
persist_extended_history,
/*metrics_service_name*/ None,
@@ -991,6 +1009,7 @@ impl ThreadManagerState {
config,
agent_control,
self.session_source.clone(),
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*metrics_service_name*/ None,
/*inherited_shell_snapshot*/ None,
@@ -1006,6 +1025,7 @@ impl ThreadManagerState {
config: Config,
agent_control: AgentControl,
session_source: SessionSource,
thread_source: Option<ThreadSource>,
persist_extended_history: bool,
metrics_service_name: Option<String>,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
@@ -1021,6 +1041,7 @@ impl ThreadManagerState {
Arc::clone(&self.auth_manager),
agent_control,
session_source,
thread_source,
Vec::new(),
persist_extended_history,
metrics_service_name,
@@ -1047,12 +1068,14 @@ impl ThreadManagerState {
} = options;
let environments =
default_thread_environment_selections(self.environment_manager.as_ref(), &config.cwd);
let thread_source = initial_history.get_resumed_thread_source();
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
thread_source,
Vec::new(),
/*persist_extended_history*/ false,
/*metrics_service_name*/ None,
@@ -1072,6 +1095,7 @@ impl ThreadManagerState {
initial_history: InitialHistory,
agent_control: AgentControl,
session_source: SessionSource,
thread_source: Option<ThreadSource>,
persist_extended_history: bool,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
@@ -1086,6 +1110,7 @@ impl ThreadManagerState {
Arc::clone(&self.auth_manager),
agent_control,
session_source,
thread_source,
Vec::new(),
persist_extended_history,
/*metrics_service_name*/ None,
@@ -1106,6 +1131,7 @@ impl ThreadManagerState {
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
thread_source: Option<ThreadSource>,
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
persist_extended_history: bool,
metrics_service_name: Option<String>,
@@ -1119,6 +1145,7 @@ impl ThreadManagerState {
auth_manager,
agent_control,
self.session_source.clone(),
thread_source,
dynamic_tools,
persist_extended_history,
metrics_service_name,
@@ -1139,6 +1166,7 @@ impl ThreadManagerState {
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
session_source: SessionSource,
thread_source: Option<ThreadSource>,
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
persist_extended_history: bool,
metrics_service_name: Option<String>,
@@ -1202,6 +1230,7 @@ impl ThreadManagerState {
skills_watcher: Arc::clone(&self.skills_watcher),
conversation_history: initial_history,
session_source,
thread_source,
agent_control,
dynamic_tools,
persist_extended_history,
codex-rs/core/src/thread_manager_tests.rs
+90
View File
@@ -15,6 +15,7 @@ use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::InternalSessionSource;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::UserMessageEvent;
use core_test_support::PathBufExt;
@@ -178,6 +179,7 @@ fn fork_thread_accepts_legacy_usize_snapshot_argument() {
usize::MAX,
config,
path,
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
);
@@ -333,6 +335,7 @@ async fn start_thread_accepts_explicit_environment_when_default_environment_is_d
config: config.clone(),
initial_history: InitialHistory::New,
session_source: None,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
metrics_service_name: None,
@@ -370,6 +373,7 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() {
session_source: Some(SessionSource::Internal(
InternalSessionSource::MemoryConsolidation,
)),
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
metrics_service_name: None,
@@ -424,6 +428,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
config: config.clone(),
initial_history: InitialHistory::New,
session_source: None,
thread_source: None,
dynamic_tools: Vec::new(),
persist_extended_history: false,
metrics_service_name: None,
@@ -480,6 +485,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
ForkSnapshot::Interrupted,
config,
rollout_path,
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
)
@@ -620,6 +626,86 @@ async fn resume_stopped_thread_from_rollout_spawns_new_thread() {
.expect("shutdown resumed thread");
}
#[tokio::test]
async fn resume_stopped_thread_from_rollout_preserves_thread_source() {
let temp_dir = tempdir().expect("tempdir");
let mut config = test_config().await;
config.codex_home = temp_dir.path().join("codex-home").abs();
config.cwd = config.codex_home.abs();
std::fs::create_dir_all(&config.codex_home).expect("create codex home");
let auth_manager =
AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
let (state_db, thread_store, agent_graph_store) = state_backed_stores(&config).await;
let manager = ThreadManager::new(
&config,
auth_manager.clone(),
SessionSource::Exec,
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
/*analytics_events_client*/ None,
state_db,
thread_store,
agent_graph_store,
);
let source = manager
.start_thread_with_options(StartThreadOptions {
config: config.clone(),
initial_history: InitialHistory::New,
session_source: None,
thread_source: Some(ThreadSource::User),
dynamic_tools: Vec::new(),
persist_extended_history: false,
metrics_service_name: None,
parent_trace: None,
environments: Vec::new(),
})
.await
.expect("start source thread");
source.thread.ensure_rollout_materialized().await;
source
.thread
.flush_rollout()
.await
.expect("flush source rollout");
let rollout_path = source
.thread
.rollout_path()
.expect("source rollout path should exist");
source
.thread
.shutdown_and_wait()
.await
.expect("shutdown source thread before resume");
let _ = manager.remove_thread(&source.thread_id).await;
let resumed = manager
.resume_thread_from_rollout(
config,
rollout_path,
auth_manager,
/*parent_trace*/ None,
)
.await
.expect("resume source thread");
assert_eq!(
resumed
.thread
.config_snapshot()
.await
.thread_source
.as_ref(),
Some(&ThreadSource::User)
);
resumed
.thread
.shutdown_and_wait()
.await
.expect("shutdown resumed thread");
}
#[tokio::test]
async fn new_uses_active_provider_for_model_refresh() {
let server = MockServer::start().await;
@@ -891,6 +977,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
ForkSnapshot::Interrupted,
config.clone(),
source_path,
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
)
@@ -1007,6 +1094,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
ForkSnapshot::Interrupted,
config.clone(),
source_path,
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
)
@@ -1088,6 +1176,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
ForkSnapshot::Interrupted,
config.clone(),
source_path,
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
)
@@ -1128,6 +1217,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
ForkSnapshot::Interrupted,
config.clone(),
forked_path,
/*thread_source*/ None,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
)
codex-rs/core/src/turn_metadata.rs
+5 -5
View File
@@ -17,7 +17,7 @@ use codex_git_utils::get_head_commit_hash;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadSource;
use codex_utils_absolute_path::AbsolutePathBuf;
const MODEL_KEY: &str = "model";
@@ -69,7 +69,7 @@ pub(crate) struct TurnMetadataBag {
#[serde(default, skip_serializing_if = "Option::is_none")]
session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
thread_source: Option<&'static str>,
thread_source: Option<ThreadSource>,
#[serde(default, skip_serializing_if = "Option::is_none")]
turn_id: Option<String>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
@@ -115,7 +115,7 @@ fn merge_turn_metadata(
fn build_turn_metadata_bag(
session_id: Option<String>,
thread_source: Option<&'static str>,
thread_source: Option<ThreadSource>,
turn_id: Option<String>,
sandbox: Option<String>,
repo_root: Option<String>,
@@ -187,7 +187,7 @@ pub(crate) struct TurnMetadataState {
impl TurnMetadataState {
pub(crate) fn new(
session_id: String,
session_source: &SessionSource,
thread_source: Option<ThreadSource>,
turn_id: String,
cwd: AbsolutePathBuf,
permission_profile: &PermissionProfile,
@@ -205,7 +205,7 @@ impl TurnMetadataState {
);
let base_metadata = build_turn_metadata_bag(
Some(session_id),
session_source.thread_source_name(),
thread_source,
Some(turn_id),
sandbox,
/*repo_root*/ None,
codex-rs/core/src/turn_metadata_tests.rs
+8 -11
View File
@@ -4,8 +4,7 @@ use crate::sandbox_tags::sandbox_tag;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadSource;
use core_test_support::PathBufExt;
use core_test_support::PathExt;
use pretty_assertions::assert_eq;
@@ -95,7 +94,7 @@ fn turn_metadata_state_uses_platform_sandbox_tag() {
let state = TurnMetadataState::new(
"session-a".to_string(),
&SessionSource::Exec,
Some(ThreadSource::User),
"turn-a".to_string(),
cwd,
&permission_profile,
@@ -117,15 +116,13 @@ fn turn_metadata_state_uses_platform_sandbox_tag() {
}
#[test]
fn turn_metadata_state_classifies_subagent_thread_source() {
fn turn_metadata_state_uses_explicit_subagent_thread_source() {
let temp_dir = TempDir::new().expect("temp dir");
let cwd = temp_dir.path().abs();
let permission_profile = PermissionProfile::read_only();
let session_source = SessionSource::SubAgent(SubAgentSource::Review);
let state = TurnMetadataState::new(
"session-a".to_string(),
&session_source,
Some(ThreadSource::Subagent),
"turn-a".to_string(),
cwd,
&permission_profile,
@@ -148,7 +145,7 @@ fn turn_metadata_state_includes_turn_started_at_unix_ms_after_start() {
let state = TurnMetadataState::new(
"session-a".to_string(),
&SessionSource::Exec,
Some(ThreadSource::User),
"turn-a".to_string(),
cwd,
&permission_profile,
@@ -174,7 +171,7 @@ fn turn_metadata_state_includes_model_and_reasoning_effort_only_in_request_meta(
let state = TurnMetadataState::new(
"session-a".to_string(),
&SessionSource::Exec,
/*thread_source*/ None,
"turn-a".to_string(),
cwd,
&permission_profile,
@@ -218,7 +215,7 @@ fn turn_metadata_state_ignores_client_turn_started_at_unix_ms_before_start() {
let state = TurnMetadataState::new(
"session-a".to_string(),
&SessionSource::Exec,
Some(ThreadSource::User),
"turn-a".to_string(),
cwd,
&permission_profile,
@@ -244,7 +241,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
let state = TurnMetadataState::new(
"session-a".to_string(),
&SessionSource::Exec,
Some(ThreadSource::User),
"turn-a".to_string(),
cwd,
&permission_profile,