Trim side thread compatibility fix

This commit is contained in:
jif-oai
2026-06-10 14:56:38 +01:00
parent 0f20f9a7a6
commit 1f5c87a397
14 changed files with 33 additions and 210 deletions
@@ -281,7 +281,6 @@ use codex_core::CodexThread;
use codex_core::CodexThreadSettingsOverrides;
use codex_core::ForkSnapshot;
use codex_core::McpManager;
use codex_core::MultiAgentRuntimeOverride;
use codex_core::NewThread;
#[cfg(test)]
use codex_core::SessionMeta;
@@ -17,12 +17,8 @@ struct ThreadListFilters {
use_state_db_only: bool,
}
fn multi_agent_runtime_override(disable_multi_agent_tools: bool) -> MultiAgentRuntimeOverride {
if disable_multi_agent_tools {
MultiAgentRuntimeOverride::Disabled
} else {
MultiAgentRuntimeOverride::Inherit
}
fn multi_agent_version_override(disable_multi_agent_tools: bool) -> Option<MultiAgentVersion> {
disable_multi_agent_tools.then_some(MultiAgentVersion::Disabled)
}
fn collect_resume_override_mismatches(
@@ -1099,7 +1095,7 @@ impl ThreadRequestProcessor {
},
session_source: None,
thread_source,
multi_agent_runtime: multi_agent_runtime_override(disable_multi_agent_tools),
multi_agent_version: multi_agent_version_override(disable_multi_agent_tools),
dynamic_tools: core_dynamic_tools,
metrics_service_name: service_name,
parent_trace: request_trace,
@@ -3279,7 +3275,7 @@ impl ThreadRequestProcessor {
history: history_items.clone(),
rollout_path: source_thread.rollout_path.clone(),
}),
multi_agent_runtime_override(disable_multi_agent_tools),
multi_agent_version_override(disable_multi_agent_tools),
thread_source.map(Into::into),
self.request_trace_context(&request_id).await,
)
@@ -1130,7 +1130,7 @@ impl TurnRequestProcessor {
history: parent_history.items,
rollout_path: parent_thread.rollout_path(),
}),
MultiAgentRuntimeOverride::Inherit,
/*multi_agent_version*/ None,
/*thread_source*/ None,
self.request_trace_context(request_id).await,
)
@@ -51,7 +51,6 @@ mod safety_check_downgrade;
mod skills_list;
mod thread_archive;
mod thread_fork;
mod thread_fork_multi_agent;
mod thread_inject_items;
mod thread_list;
mod thread_loaded_list;
@@ -112,7 +112,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
original_path.display()
);
let mut session_meta = read_session_meta_line(&original_path).await?;
session_meta.meta.multi_agent_version = Some(MultiAgentVersion::V1);
session_meta.meta.multi_agent_version = Some(MultiAgentVersion::V2);
append_rollout_item_to_path(&original_path, &RolloutItem::SessionMeta(session_meta)).await?;
let original_contents = std::fs::read_to_string(&original_path)?;
@@ -123,6 +123,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
thread_source: Some(ThreadSource::User),
disable_multi_agent_tools: true,
..Default::default()
})
.await?;
@@ -132,7 +133,12 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
)
.await??;
let fork_result = fork_resp.result.clone();
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let ThreadForkResponse {
thread,
multi_agent_tools_disabled,
..
} = to_response::<ThreadForkResponse>(fork_resp)?;
assert!(multi_agent_tools_disabled);
// Wire contract: thread title field is `name`, serialized as null when unset.
let thread_json = fork_result
@@ -1,157 +0,0 @@
use anyhow::Context;
use anyhow::Result;
use app_test_support::TestAppServer;
use app_test_support::create_fake_rollout;
use app_test_support::rollout_path;
use app_test_support::to_response;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::UserInput;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::RolloutItem;
use codex_rollout::append_rollout_item_to_path;
use codex_rollout::read_session_meta_line;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
async fn thread_fork_disable_multi_agent_tools_overrides_v2_history() -> Result<()> {
const ROOT_USAGE_HINT: &str = "disabled forks must not receive root multi-agent guidance";
let server = responses::start_mock_server().await;
let response_mock = responses::mount_sse_once(
&server,
responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "done"),
responses::ev_completed("resp-1"),
]),
)
.await;
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join("config.toml"),
format!(
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "read-only"
model_provider = "mock_provider"
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
[features.multi_agent_v2]
enabled = true
root_agent_usage_hint_text = "{ROOT_USAGE_HINT}"
"#,
server_uri = server.uri(),
),
)?;
let source_thread_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"parent message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let source_rollout_path =
rollout_path(codex_home.path(), "2025-01-05T12-00-00", &source_thread_id);
let mut source_meta = read_session_meta_line(&source_rollout_path).await?;
source_meta.meta.multi_agent_version = Some(MultiAgentVersion::V2);
append_rollout_item_to_path(&source_rollout_path, &RolloutItem::SessionMeta(source_meta))
.await?;
let mut app_server = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, app_server.initialize()).await??;
let fork_request_id = app_server
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id,
disable_multi_agent_tools: true,
..Default::default()
})
.await?;
let fork_response: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(fork_request_id)),
)
.await??;
let ThreadForkResponse {
thread,
multi_agent_tools_disabled,
..
} = to_response::<ThreadForkResponse>(fork_response)?;
assert!(multi_agent_tools_disabled);
let fork_rollout_path = thread.path.as_ref().context("fork rollout path")?;
let fork_meta = read_session_meta_line(fork_rollout_path.as_path()).await?;
assert_eq!(
fork_meta.meta.multi_agent_version,
Some(MultiAgentVersion::Disabled)
);
let turn_request_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id: thread.id,
input: vec![UserInput::Text {
text: "which tools are available?".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(turn_request_id)),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
app_server.read_stream_until_notification_message("turn/completed"),
)
.await??;
let request = response_mock.single_request();
assert!(!request.body_contains_text(ROOT_USAGE_HINT));
let tool_names = request.body_json()["tools"]
.as_array()
.into_iter()
.flatten()
.filter_map(|tool| tool.get("name").and_then(serde_json::Value::as_str))
.map(str::to_string)
.collect::<Vec<_>>();
assert!(tool_names.contains(&"update_plan".to_string()));
let present_multi_agent_tools = tool_names
.into_iter()
.filter(|tool| {
matches!(
tool.as_str(),
"spawn_agent"
| "send_message"
| "followup_task"
| "wait_agent"
| "interrupt_agent"
| "list_agents"
| "spawn_agents_on_csv"
)
})
.collect::<Vec<_>>();
assert_eq!(present_multi_agent_tools, Vec::<String>::new());
Ok(())
}
-1
View File
@@ -28,7 +28,6 @@ pub use codex_core::CodexThread;
pub use codex_core::ForkSnapshot;
pub use codex_core::LoadedAgentsMd;
pub use codex_core::McpManager;
pub use codex_core::MultiAgentRuntimeOverride;
pub use codex_core::NewThread;
pub use codex_core::StartThreadOptions;
pub use codex_core::StateDbHandle;
-1
View File
@@ -108,7 +108,6 @@ mod thread_manager;
pub(crate) mod web_search;
pub(crate) mod windows_sandbox_read_grants;
pub use thread_manager::ForkSnapshot;
pub use thread_manager::MultiAgentRuntimeOverride;
pub use thread_manager::NewThread;
pub use thread_manager::StartThreadOptions;
pub use thread_manager::ThreadManager;
+10 -24
View File
@@ -174,22 +174,12 @@ pub struct ThreadManager {
_test_codex_home_guard: Option<TempCodexHomeGuard>,
}
/// Controls whether a new thread inherits its multi-agent runtime or disables it.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum MultiAgentRuntimeOverride {
/// Resolve the runtime from copied history, parent state, and configuration.
#[default]
Inherit,
/// Force the thread to run without multi-agent collaboration.
Disabled,
}
pub struct StartThreadOptions {
pub config: Config,
pub initial_history: InitialHistory,
pub session_source: Option<SessionSource>,
pub thread_source: Option<ThreadSource>,
pub multi_agent_runtime: MultiAgentRuntimeOverride,
pub multi_agent_version: Option<MultiAgentVersion>,
pub dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
pub metrics_service_name: Option<String>,
pub parent_trace: Option<W3cTraceContext>,
@@ -588,7 +578,7 @@ impl ThreadManager {
initial_history: InitialHistory::New,
session_source: None,
thread_source: None,
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools,
metrics_service_name: None,
parent_trace: None,
@@ -617,10 +607,6 @@ impl ThreadManager {
.unwrap_or_else(|| (self.state.session_source.clone(), None));
let session_source = options.session_source.unwrap_or(resumed_session_source);
let thread_source = options.thread_source.or(resumed_thread_source);
let multi_agent_version = match options.multi_agent_runtime {
MultiAgentRuntimeOverride::Inherit => None,
MultiAgentRuntimeOverride::Disabled => Some(MultiAgentVersion::Disabled),
};
Box::pin(self.state.spawn_thread_with_source(
options.config,
options.initial_history,
@@ -630,7 +616,7 @@ impl ThreadManager {
/*parent_thread_id*/ None,
forked_from_thread_id,
thread_source,
multi_agent_version,
options.multi_agent_version,
options.dynamic_tools,
options.metrics_service_name,
/*inherited_shell_snapshot*/ None,
@@ -877,7 +863,7 @@ impl ThreadManager {
snapshot,
config,
history,
MultiAgentRuntimeOverride::Inherit,
/*multi_agent_version*/ None,
thread_source,
parent_trace,
)
@@ -908,7 +894,7 @@ impl ThreadManager {
snapshot: S,
config: Config,
history: InitialHistory,
multi_agent_runtime: MultiAgentRuntimeOverride,
multi_agent_version: Option<MultiAgentVersion>,
thread_source: Option<ThreadSource>,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread>
@@ -919,7 +905,7 @@ impl ThreadManager {
snapshot.into(),
config,
history,
multi_agent_runtime,
multi_agent_version,
thread_source,
parent_trace,
)
@@ -931,7 +917,7 @@ impl ThreadManager {
snapshot: ForkSnapshot,
config: Config,
history: InitialHistory,
multi_agent_runtime: MultiAgentRuntimeOverride,
multi_agent_version: Option<MultiAgentVersion>,
thread_source: Option<ThreadSource>,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread> {
@@ -942,9 +928,9 @@ impl ThreadManager {
InitialHistory::Forked(_) => history.forked_from_id(),
InitialHistory::New | InitialHistory::Cleared => None,
};
let multi_agent_version = match multi_agent_runtime {
MultiAgentRuntimeOverride::Disabled => MultiAgentVersion::Disabled,
MultiAgentRuntimeOverride::Inherit => {
let multi_agent_version = match multi_agent_version {
Some(multi_agent_version) => multi_agent_version,
None => {
self.state
.effective_multi_agent_version_for_spawn(
&history,
+5 -5
View File
@@ -324,7 +324,7 @@ async fn start_thread_rejects_explicit_local_environment_when_default_provider_i
initial_history: InitialHistory::New,
session_source: None,
thread_source: None,
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
@@ -366,7 +366,7 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() {
InternalSessionSource::MemoryConsolidation,
)),
thread_source: None,
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
@@ -445,7 +445,7 @@ async fn start_thread_seeds_extension_data_before_lifecycle_contributors_run() {
initial_history: InitialHistory::New,
session_source: None,
thread_source: None,
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
@@ -499,7 +499,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
initial_history: InitialHistory::New,
session_source: None,
thread_source: None,
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
@@ -771,7 +771,7 @@ async fn resume_stopped_thread_from_rollout_preserves_thread_source() {
initial_history: InitialHistory::New,
session_source: None,
thread_source: Some(ThreadSource::User),
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
+1 -2
View File
@@ -1,5 +1,4 @@
use codex_core::ForkSnapshot;
use codex_core::MultiAgentRuntimeOverride;
use codex_core::NewThread;
use codex_core::parse_turn_item;
use codex_protocol::items::TurnItem;
@@ -200,7 +199,7 @@ async fn fork_thread_from_history_does_not_require_source_rollout_path() {
history: source_items.clone(),
rollout_path: None,
}),
MultiAgentRuntimeOverride::Inherit,
/*multi_agent_version*/ None,
/*thread_source*/ None,
/*parent_trace*/ None,
)
@@ -1,5 +1,4 @@
use anyhow::Result;
use codex_core::MultiAgentRuntimeOverride;
use codex_core::StartThreadOptions;
use codex_core::ThreadConfigSnapshot;
use codex_core::config::AgentRoleConfig;
@@ -754,7 +753,7 @@ async fn subagent_stop_replaces_stop_and_skips_internal_subagents() -> Result<()
initial_history: InitialHistory::New,
session_source: Some(SessionSource::SubAgent(SubAgentSource::Review)),
thread_source: None,
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
+2 -3
View File
@@ -7,7 +7,6 @@ use std::time::Instant;
use anyhow::Context;
use anyhow::Result;
use codex_core::MultiAgentRuntimeOverride;
use codex_core::StartThreadOptions;
use codex_core::sandboxing::SandboxPermissions;
use codex_features::Feature;
@@ -62,7 +61,7 @@ fn tool_names(body: &Value) -> Vec<String> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn multi_agent_runtime_override_omits_multi_agent_tools_from_requests() -> Result<()> {
async fn multi_agent_version_override_omits_multi_agent_tools_from_requests() -> Result<()> {
const ROOT_USAGE_HINT: &str = "disabled threads must not receive root multi-agent guidance";
skip_if_no_network!(Ok(()));
@@ -96,7 +95,7 @@ async fn multi_agent_runtime_override_omits_multi_agent_tools_from_requests() ->
initial_history: InitialHistory::New,
session_source: None,
thread_source: Some(ThreadSource::User),
multi_agent_runtime: MultiAgentRuntimeOverride::Disabled,
multi_agent_version: Some(MultiAgentVersion::Disabled),
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
+1 -2
View File
@@ -1,6 +1,5 @@
use codex_core::CodexThread;
use codex_core::ModelClient;
use codex_core::MultiAgentRuntimeOverride;
use codex_core::NewThread;
use codex_core::Prompt;
use codex_core::ResponseEvent;
@@ -248,7 +247,7 @@ impl MemoryStartupContext {
InternalSessionSource::MemoryConsolidation,
)),
thread_source: Some(ThreadSource::MemoryConsolidation),
multi_agent_runtime: MultiAgentRuntimeOverride::Inherit,
multi_agent_version: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,