mirror of
https://github.com/openai/codex.git
synced 2026-10-31 19:28:49 +00:00
fix: make side thread tool disabling fail closed
This commit is contained in:
jif-oai
@@ -3916,10 +3916,6 @@
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"disableMultiAgentTools": {
|
||||
"description": "When true, disable multi-agent collaboration tools for this thread.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"ephemeral": {
|
||||
"type": [
|
||||
"boolean",
|
||||
|
||||
-4
@@ -18184,10 +18184,6 @@
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"disableMultiAgentTools": {
|
||||
"description": "When true, disable multi-agent collaboration tools for this thread.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"ephemeral": {
|
||||
"type": [
|
||||
"boolean",
|
||||
|
||||
-4
@@ -16001,10 +16001,6 @@
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"disableMultiAgentTools": {
|
||||
"description": "When true, disable multi-agent collaboration tools for this thread.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"ephemeral": {
|
||||
"type": [
|
||||
"boolean",
|
||||
|
||||
@@ -231,10 +231,6 @@
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"disableMultiAgentTools": {
|
||||
"description": "When true, disable multi-agent collaboration tools for this thread.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"ephemeral": {
|
||||
"type": [
|
||||
"boolean",
|
||||
|
||||
@@ -16,7 +16,4 @@ export type ThreadStartParams = {model?: string | null, modelProvider?: string |
|
||||
approvalsReviewer?: ApprovalsReviewer | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, serviceName?: string | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, ephemeral?: boolean | null, sessionStartSource?: ThreadStartSource | null, /**
|
||||
* Optional client-supplied analytics source classification for this thread.
|
||||
*/
|
||||
threadSource?: ThreadSource | null, /**
|
||||
* When true, disable multi-agent collaboration tools for this thread.
|
||||
*/
|
||||
disableMultiAgentTools?: boolean};
|
||||
threadSource?: ThreadSource | null};
|
||||
|
||||
@@ -143,9 +143,6 @@ pub struct ThreadStartParams {
|
||||
/// Optional client-supplied analytics source classification for this thread.
|
||||
#[ts(optional = nullable)]
|
||||
pub thread_source: Option<ThreadSource>,
|
||||
/// When true, disable multi-agent collaboration tools for this thread.
|
||||
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
|
||||
pub disable_multi_agent_tools: bool,
|
||||
/// Optional sticky environments for this thread.
|
||||
///
|
||||
/// Omitted selects the default environment when environment access is
|
||||
|
||||
@@ -130,7 +130,7 @@ Example with notification opt-out:
|
||||
|
||||
## API Overview
|
||||
|
||||
- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. Set `disableMultiAgentTools: true` to prevent the thread from receiving multi-agent collaboration tools. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd`. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots. Skills found below those roots are listed and read through the owning environment; other plugin capabilities are not activated yet.
|
||||
- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd`. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots. Skills found below those roots are listed and read through the owning environment; other plugin capabilities are not activated yet.
|
||||
- `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. Accepts the same permission override rules as `thread/start`.
|
||||
- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. The returned `thread.forkedFromId` points at the source thread when known. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread. Set `disableMultiAgentTools: true` to prevent the fork from inheriting or enabling multi-agent collaboration tools; the response sets `multiAgentToolsDisabled: true` when the forked runtime applied that restriction, and clients that require it should reject a missing or false acknowledgment. Experimental clients can pass `excludeTurns: true` when they plan to page fork history via `thread/turns/list` instead of receiving the full turn array immediately. Accepts the same permission override rules as `thread/start`.
|
||||
- `thread/start`, `thread/resume`, and `thread/fork` responses include the legacy `sandbox` compatibility projection. Experimental clients can read `runtimeWorkspaceRoots` for the thread-scoped runtime roots and `activePermissionProfile` for the named or implicit built-in profile identity/provenance when known.
|
||||
|
||||
@@ -281,6 +281,7 @@ 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,10 +17,6 @@ struct ThreadListFilters {
|
||||
use_state_db_only: bool,
|
||||
}
|
||||
|
||||
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(
|
||||
request: &ThreadResumeParams,
|
||||
config_snapshot: &ThreadConfigSnapshot,
|
||||
@@ -846,7 +842,6 @@ impl ThreadRequestProcessor {
|
||||
ephemeral,
|
||||
session_start_source,
|
||||
thread_source,
|
||||
disable_multi_agent_tools,
|
||||
environments,
|
||||
} = params;
|
||||
if sandbox.is_some() && permissions.is_some() {
|
||||
@@ -899,7 +894,6 @@ impl ThreadRequestProcessor {
|
||||
selected_capability_roots.unwrap_or_default(),
|
||||
session_start_source,
|
||||
thread_source.map(Into::into),
|
||||
disable_multi_agent_tools,
|
||||
environment_selections,
|
||||
service_name,
|
||||
experimental_raw_events,
|
||||
@@ -973,7 +967,6 @@ impl ThreadRequestProcessor {
|
||||
selected_capability_roots: Vec<SelectedCapabilityRoot>,
|
||||
session_start_source: Option<codex_app_server_protocol::ThreadStartSource>,
|
||||
thread_source: Option<codex_protocol::protocol::ThreadSource>,
|
||||
disable_multi_agent_tools: bool,
|
||||
environments: Option<Vec<TurnEnvironmentSelection>>,
|
||||
service_name: Option<String>,
|
||||
experimental_raw_events: bool,
|
||||
@@ -1095,7 +1088,6 @@ impl ThreadRequestProcessor {
|
||||
},
|
||||
session_source: None,
|
||||
thread_source,
|
||||
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,
|
||||
@@ -3275,7 +3267,11 @@ impl ThreadRequestProcessor {
|
||||
history: history_items.clone(),
|
||||
rollout_path: source_thread.rollout_path.clone(),
|
||||
}),
|
||||
multi_agent_version_override(disable_multi_agent_tools),
|
||||
if disable_multi_agent_tools {
|
||||
MultiAgentRuntimeOverride::Disabled
|
||||
} else {
|
||||
MultiAgentRuntimeOverride::Inherit
|
||||
},
|
||||
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(),
|
||||
}),
|
||||
/*multi_agent_version*/ None,
|
||||
MultiAgentRuntimeOverride::Inherit,
|
||||
/*thread_source*/ None,
|
||||
self.request_trace_context(request_id).await,
|
||||
)
|
||||
|
||||
@@ -791,7 +791,6 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<(
|
||||
ephemeral: None,
|
||||
session_start_source: None,
|
||||
thread_source: None,
|
||||
disable_multi_agent_tools: false,
|
||||
dynamic_tools: None,
|
||||
environments: None,
|
||||
selected_capability_roots: None,
|
||||
|
||||
@@ -29,8 +29,11 @@ use codex_app_server_protocol::TurnStartResponse;
|
||||
use codex_app_server_protocol::TurnStatus;
|
||||
use codex_app_server_protocol::UserInput;
|
||||
use codex_config::types::AuthCredentialsStoreMode;
|
||||
use codex_core::RolloutRecorder;
|
||||
use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::MultiAgentVersion;
|
||||
use codex_protocol::protocol::RolloutItem;
|
||||
use codex_rollout::append_rollout_item_to_path;
|
||||
@@ -83,9 +86,24 @@ async fn list_threads(mcp: &mut TestAppServer) -> Result<ThreadListResponse> {
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
|
||||
const ROOT_USAGE_HINT: &str = "disabled forks must not receive root multi-agent guidance";
|
||||
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml(codex_home.path(), &server.uri())?;
|
||||
let config_path = codex_home.path().join("config.toml");
|
||||
let mut config_toml = std::fs::read_to_string(&config_path)?;
|
||||
config_toml.push_str(&format!(
|
||||
r#"
|
||||
[features]
|
||||
enable_fanout = true
|
||||
|
||||
[features.multi_agent_v2]
|
||||
enabled = true
|
||||
root_agent_usage_hint_text = "{ROOT_USAGE_HINT}"
|
||||
"#,
|
||||
));
|
||||
std::fs::write(config_path, config_toml)?;
|
||||
|
||||
let preview = "Saved user message";
|
||||
let conversation_id = create_fake_rollout(
|
||||
@@ -173,6 +191,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
|
||||
assert_eq!(thread.preview, preview);
|
||||
assert_eq!(thread.model_provider, "mock_provider");
|
||||
assert_eq!(thread.status, ThreadStatus::Idle);
|
||||
let fork_thread_id = thread.id.clone();
|
||||
let thread_path = thread.path.clone().expect("thread path");
|
||||
assert!(thread_path.as_path().is_absolute());
|
||||
assert_ne!(thread_path.as_path(), original_path);
|
||||
@@ -180,6 +199,31 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
|
||||
assert_eq!(thread.source, SessionSource::VsCode);
|
||||
assert_eq!(thread.thread_source, Some(ThreadSource::User));
|
||||
assert_eq!(thread.name, None);
|
||||
assert_eq!(
|
||||
read_session_meta_line(&thread_path)
|
||||
.await?
|
||||
.meta
|
||||
.multi_agent_version,
|
||||
Some(MultiAgentVersion::Disabled),
|
||||
"fork should persist the disabled runtime"
|
||||
);
|
||||
let forked_history = RolloutRecorder::get_rollout_history(&thread_path).await?;
|
||||
let forked_items = forked_history.get_rollout_items();
|
||||
let interrupted_index = forked_items
|
||||
.iter()
|
||||
.position(|item| matches!(item, RolloutItem::EventMsg(EventMsg::TurnAborted(_))))
|
||||
.expect("fork should persist an interrupted-turn boundary");
|
||||
let interrupted_marker = interrupted_index
|
||||
.checked_sub(1)
|
||||
.and_then(|index| forked_items.get(index))
|
||||
.expect("interrupted-turn event should follow its model-visible marker");
|
||||
assert!(
|
||||
matches!(
|
||||
interrupted_marker,
|
||||
RolloutItem::ResponseItem(ResponseItem::Message { role, .. }) if role == "developer"
|
||||
),
|
||||
"disabled target should preserve the V2 source's developer interruption marker"
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
thread.turns.len(),
|
||||
@@ -252,6 +296,71 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
|
||||
expected_started_thread.turns.clear();
|
||||
assert_eq!(started.thread, expected_started_thread);
|
||||
|
||||
let turn_id = mcp
|
||||
.send_turn_start_request(TurnStartParams {
|
||||
thread_id: fork_thread_id,
|
||||
client_user_message_id: None,
|
||||
input: vec![UserInput::Text {
|
||||
text: "which tools are available?".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let turn_resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(turn_id)),
|
||||
)
|
||||
.await??;
|
||||
let _: TurnStartResponse = to_response::<TurnStartResponse>(turn_resp)?;
|
||||
timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_notification_message("turn/completed"),
|
||||
)
|
||||
.await??;
|
||||
|
||||
let requests = server
|
||||
.received_requests()
|
||||
.await
|
||||
.expect("mock server should record the responses request");
|
||||
let response_request = requests
|
||||
.iter()
|
||||
.find(|request| request.method == "POST" && request.url.path().ends_with("/responses"))
|
||||
.expect("turn should issue a responses request");
|
||||
let body: Value = serde_json::from_slice(&response_request.body)?;
|
||||
assert!(!body.to_string().contains(ROOT_USAGE_HINT));
|
||||
let tool_names = body
|
||||
.get("tools")
|
||||
.and_then(Value::as_array)
|
||||
.expect("responses request should include tools")
|
||||
.iter()
|
||||
.filter_map(|tool| {
|
||||
tool.get("name")
|
||||
.or_else(|| tool.get("type"))
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_string)
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert!(tool_names.contains(&"update_plan".to_string()));
|
||||
assert_eq!(
|
||||
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<_>>(),
|
||||
Vec::<String>::new()
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -28,6 +28,7 @@ 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;
|
||||
|
||||
@@ -108,6 +108,7 @@ 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;
|
||||
|
||||
@@ -154,6 +154,16 @@ impl From<usize> for ForkSnapshot {
|
||||
}
|
||||
}
|
||||
|
||||
/// Controls whether a fork inherits its multi-agent runtime or disables it.
|
||||
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
|
||||
pub enum MultiAgentRuntimeOverride {
|
||||
/// Resolve the runtime from copied history and configuration.
|
||||
#[default]
|
||||
Inherit,
|
||||
/// Force the fork to run without multi-agent collaboration.
|
||||
Disabled,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, PartialEq, Eq)]
|
||||
pub struct ThreadShutdownReport {
|
||||
pub completed: Vec<ThreadId>,
|
||||
@@ -179,7 +189,6 @@ pub struct StartThreadOptions {
|
||||
pub initial_history: InitialHistory,
|
||||
pub session_source: Option<SessionSource>,
|
||||
pub thread_source: Option<ThreadSource>,
|
||||
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>,
|
||||
@@ -578,7 +587,6 @@ impl ThreadManager {
|
||||
initial_history: InitialHistory::New,
|
||||
session_source: None,
|
||||
thread_source: None,
|
||||
multi_agent_version: None,
|
||||
dynamic_tools,
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
@@ -616,7 +624,7 @@ impl ThreadManager {
|
||||
/*parent_thread_id*/ None,
|
||||
forked_from_thread_id,
|
||||
thread_source,
|
||||
options.multi_agent_version,
|
||||
/*multi_agent_version*/ None,
|
||||
options.dynamic_tools,
|
||||
options.metrics_service_name,
|
||||
/*inherited_shell_snapshot*/ None,
|
||||
@@ -863,7 +871,7 @@ impl ThreadManager {
|
||||
snapshot,
|
||||
config,
|
||||
history,
|
||||
/*multi_agent_version*/ None,
|
||||
MultiAgentRuntimeOverride::Inherit,
|
||||
thread_source,
|
||||
parent_trace,
|
||||
)
|
||||
@@ -894,7 +902,7 @@ impl ThreadManager {
|
||||
snapshot: S,
|
||||
config: Config,
|
||||
history: InitialHistory,
|
||||
multi_agent_version: Option<MultiAgentVersion>,
|
||||
multi_agent_runtime: MultiAgentRuntimeOverride,
|
||||
thread_source: Option<ThreadSource>,
|
||||
parent_trace: Option<W3cTraceContext>,
|
||||
) -> CodexResult<NewThread>
|
||||
@@ -905,7 +913,7 @@ impl ThreadManager {
|
||||
snapshot.into(),
|
||||
config,
|
||||
history,
|
||||
multi_agent_version,
|
||||
multi_agent_runtime,
|
||||
thread_source,
|
||||
parent_trace,
|
||||
)
|
||||
@@ -917,7 +925,7 @@ impl ThreadManager {
|
||||
snapshot: ForkSnapshot,
|
||||
config: Config,
|
||||
history: InitialHistory,
|
||||
multi_agent_version: Option<MultiAgentVersion>,
|
||||
multi_agent_runtime: MultiAgentRuntimeOverride,
|
||||
thread_source: Option<ThreadSource>,
|
||||
parent_trace: Option<W3cTraceContext>,
|
||||
) -> CodexResult<NewThread> {
|
||||
@@ -928,22 +936,24 @@ impl ThreadManager {
|
||||
InitialHistory::Forked(_) => history.forked_from_id(),
|
||||
InitialHistory::New | InitialHistory::Cleared => None,
|
||||
};
|
||||
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,
|
||||
/*session_source*/ None,
|
||||
/*parent_thread_id*/ None,
|
||||
forked_from_thread_id,
|
||||
&config,
|
||||
)
|
||||
.await
|
||||
}
|
||||
let source_multi_agent_version = self
|
||||
.state
|
||||
.effective_multi_agent_version_for_spawn(
|
||||
&history,
|
||||
/*session_source*/ None,
|
||||
/*parent_thread_id*/ None,
|
||||
forked_from_thread_id,
|
||||
&config,
|
||||
)
|
||||
.await;
|
||||
let multi_agent_version = match multi_agent_runtime {
|
||||
MultiAgentRuntimeOverride::Disabled => MultiAgentVersion::Disabled,
|
||||
MultiAgentRuntimeOverride::Inherit => source_multi_agent_version,
|
||||
};
|
||||
let interrupted_marker =
|
||||
InterruptedTurnHistoryMarker::from_config_and_version(&config, multi_agent_version);
|
||||
let interrupted_marker = InterruptedTurnHistoryMarker::from_config_and_version(
|
||||
&config,
|
||||
source_multi_agent_version,
|
||||
);
|
||||
let history = fork_history_from_snapshot(snapshot, history, interrupted_marker);
|
||||
let environments = default_thread_environment_selections(
|
||||
self.state.environment_manager.as_ref(),
|
||||
|
||||
@@ -324,7 +324,6 @@ async fn start_thread_rejects_explicit_local_environment_when_default_provider_i
|
||||
initial_history: InitialHistory::New,
|
||||
session_source: None,
|
||||
thread_source: None,
|
||||
multi_agent_version: None,
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
@@ -366,7 +365,6 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() {
|
||||
InternalSessionSource::MemoryConsolidation,
|
||||
)),
|
||||
thread_source: None,
|
||||
multi_agent_version: None,
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
@@ -445,7 +443,6 @@ async fn start_thread_seeds_extension_data_before_lifecycle_contributors_run() {
|
||||
initial_history: InitialHistory::New,
|
||||
session_source: None,
|
||||
thread_source: None,
|
||||
multi_agent_version: None,
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
@@ -499,7 +496,6 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
|
||||
initial_history: InitialHistory::New,
|
||||
session_source: None,
|
||||
thread_source: None,
|
||||
multi_agent_version: None,
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
@@ -771,7 +767,6 @@ async fn resume_stopped_thread_from_rollout_preserves_thread_source() {
|
||||
initial_history: InitialHistory::New,
|
||||
session_source: None,
|
||||
thread_source: Some(ThreadSource::User),
|
||||
multi_agent_version: None,
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
|
||||
@@ -199,7 +199,7 @@ async fn fork_thread_from_history_does_not_require_source_rollout_path() {
|
||||
history: source_items.clone(),
|
||||
rollout_path: None,
|
||||
}),
|
||||
/*multi_agent_version*/ None,
|
||||
codex_core::MultiAgentRuntimeOverride::Inherit,
|
||||
/*thread_source*/ None,
|
||||
/*parent_trace*/ None,
|
||||
)
|
||||
|
||||
@@ -753,7 +753,6 @@ 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_version: None,
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
|
||||
@@ -7,7 +7,6 @@ use std::time::Instant;
|
||||
|
||||
use anyhow::Context;
|
||||
use anyhow::Result;
|
||||
use codex_core::StartThreadOptions;
|
||||
use codex_core::sandboxing::SandboxPermissions;
|
||||
use codex_features::Feature;
|
||||
use codex_protocol::models::PermissionProfile;
|
||||
@@ -17,12 +16,6 @@ use codex_protocol::permissions::FileSystemSandboxEntry;
|
||||
use codex_protocol::permissions::FileSystemSandboxPolicy;
|
||||
use codex_protocol::permissions::NetworkSandboxPolicy;
|
||||
use codex_protocol::protocol::AskForApproval;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::InitialHistory;
|
||||
use codex_protocol::protocol::MultiAgentVersion;
|
||||
use codex_protocol::protocol::Op;
|
||||
use codex_protocol::protocol::ThreadSource;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
use core_test_support::assert_regex_match;
|
||||
use core_test_support::responses::ev_assistant_message;
|
||||
use core_test_support::responses::ev_completed;
|
||||
@@ -37,8 +30,6 @@ use core_test_support::skip_if_no_network;
|
||||
use core_test_support::skip_if_sandbox;
|
||||
use core_test_support::test_codex::local;
|
||||
use core_test_support::test_codex::test_codex;
|
||||
use core_test_support::wait_for_event;
|
||||
use pretty_assertions::assert_eq;
|
||||
use regex_lite::Regex;
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
@@ -60,96 +51,6 @@ fn tool_names(body: &Value) -> Vec<String> {
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
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(()));
|
||||
|
||||
let server = start_mock_server().await;
|
||||
let response_mock = mount_sse_once(
|
||||
&server,
|
||||
sse(vec![
|
||||
ev_response_created("resp-1"),
|
||||
ev_assistant_message("msg-1", "done"),
|
||||
ev_completed("resp-1"),
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
let mut builder = test_codex().with_config(|config| {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::MultiAgentV2)
|
||||
.expect("multi-agent v2 should enable for test");
|
||||
config
|
||||
.features
|
||||
.enable(Feature::SpawnCsv)
|
||||
.expect("spawn CSV should enable for test");
|
||||
config.multi_agent_v2.root_agent_usage_hint_text = Some(ROOT_USAGE_HINT.to_string());
|
||||
});
|
||||
let test = builder.build(&server).await?;
|
||||
let thread = test
|
||||
.thread_manager
|
||||
.start_thread_with_options(StartThreadOptions {
|
||||
config: test.config.clone(),
|
||||
initial_history: InitialHistory::New,
|
||||
session_source: None,
|
||||
thread_source: Some(ThreadSource::User),
|
||||
multi_agent_version: Some(MultiAgentVersion::Disabled),
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
environments: Vec::new(),
|
||||
thread_extension_init: Default::default(),
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(
|
||||
thread.thread.multi_agent_version(),
|
||||
Some(MultiAgentVersion::Disabled)
|
||||
);
|
||||
|
||||
thread
|
||||
.thread
|
||||
.submit(Op::UserInput {
|
||||
items: vec![UserInput::Text {
|
||||
text: "which tools are available?".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
final_output_json_schema: None,
|
||||
responsesapi_client_metadata: None,
|
||||
additional_context: Default::default(),
|
||||
thread_settings: Default::default(),
|
||||
})
|
||||
.await?;
|
||||
wait_for_event(&thread.thread, |event| {
|
||||
matches!(event, EventMsg::TurnComplete(_))
|
||||
})
|
||||
.await;
|
||||
|
||||
let request = response_mock.single_request();
|
||||
assert!(!request.body_json().to_string().contains(ROOT_USAGE_HINT));
|
||||
let tools = tool_names(&request.body_json());
|
||||
assert!(tools.contains(&"update_plan".to_string()));
|
||||
let present_multi_agent_tools = tools
|
||||
.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(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn empty_turn_environments_omits_environment_backed_tools() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
@@ -247,7 +247,6 @@ impl MemoryStartupContext {
|
||||
InternalSessionSource::MemoryConsolidation,
|
||||
)),
|
||||
thread_source: Some(ThreadSource::MemoryConsolidation),
|
||||
multi_agent_version: None,
|
||||
dynamic_tools: Vec::new(),
|
||||
metrics_service_name: None,
|
||||
parent_trace: None,
|
||||
|
||||
@@ -62,6 +62,16 @@ impl App {
|
||||
app_server_client: &AppServerSession,
|
||||
notification: ServerNotification,
|
||||
) {
|
||||
let target = server_notification_thread_target(¬ification);
|
||||
if matches!(
|
||||
&target,
|
||||
ServerNotificationThreadTarget::Thread(thread_id)
|
||||
if app_server_client.is_rejected_fork_thread(*thread_id)
|
||||
) {
|
||||
tracing::debug!("ignoring notification for rejected app-server fork");
|
||||
return;
|
||||
}
|
||||
|
||||
match ¬ification {
|
||||
ServerNotification::ServerRequestResolved(notification) => {
|
||||
if let Some(request) = self
|
||||
@@ -117,7 +127,7 @@ impl App {
|
||||
_ => {}
|
||||
}
|
||||
|
||||
match server_notification_thread_target(¬ification) {
|
||||
match target {
|
||||
ServerNotificationThreadTarget::Thread(thread_id) => {
|
||||
let result = if self.primary_thread_id == Some(thread_id)
|
||||
|| self.primary_thread_id.is_none()
|
||||
@@ -158,6 +168,23 @@ impl App {
|
||||
app_server_client: &AppServerSession,
|
||||
request: ServerRequest,
|
||||
) {
|
||||
let request_thread_id = server_request_thread_id(&request);
|
||||
if request_thread_id
|
||||
.is_some_and(|thread_id| app_server_client.is_rejected_fork_thread(thread_id))
|
||||
{
|
||||
if let Err(err) = self
|
||||
.reject_app_server_request(
|
||||
app_server_client,
|
||||
request.id().clone(),
|
||||
"request belongs to a rejected app-server fork".to_string(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!("{err}");
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if let Some(unsupported) = self
|
||||
.pending_app_server_requests
|
||||
.note_server_request(&request)
|
||||
@@ -182,7 +209,7 @@ impl App {
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(thread_id) = server_request_thread_id(&request) else {
|
||||
let Some(thread_id) = request_thread_id else {
|
||||
tracing::warn!("ignoring threadless app-server request");
|
||||
return;
|
||||
};
|
||||
|
||||
@@ -2990,6 +2990,82 @@ async fn inactive_thread_started_notification_preserves_primary_model_when_path_
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rejected_fork_notifications_never_enter_thread_routing() -> Result<()> {
|
||||
let mut app = make_test_app().await;
|
||||
let mut app_server = crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref())
|
||||
.await
|
||||
.expect("embedded app server");
|
||||
let primary_thread_id = ThreadId::new();
|
||||
let rejected_thread_id = ThreadId::new();
|
||||
let accepted_thread_id = ThreadId::new();
|
||||
let primary_cwd = test_path_buf("/tmp/main").abs();
|
||||
app.primary_thread_id = Some(primary_thread_id);
|
||||
app.active_thread_id = Some(primary_thread_id);
|
||||
app.primary_session_configured = Some(test_thread_session(
|
||||
primary_thread_id,
|
||||
primary_cwd.to_path_buf(),
|
||||
));
|
||||
app_server.quarantine_rejected_fork(rejected_thread_id);
|
||||
|
||||
let rejected_thread = Thread {
|
||||
id: rejected_thread_id.to_string(),
|
||||
session_id: rejected_thread_id.to_string(),
|
||||
forked_from_id: Some(primary_thread_id.to_string()),
|
||||
parent_thread_id: None,
|
||||
preview: "rejected fork".to_string(),
|
||||
ephemeral: true,
|
||||
model_provider: "openai".to_string(),
|
||||
created_at: 1,
|
||||
updated_at: 2,
|
||||
status: codex_app_server_protocol::ThreadStatus::Idle,
|
||||
path: None,
|
||||
cwd: test_path_buf("/tmp/rejected").abs(),
|
||||
cli_version: "0.0.0".to_string(),
|
||||
source: codex_app_server_protocol::SessionSource::Unknown,
|
||||
thread_source: None,
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
git_info: None,
|
||||
name: Some("rejected fork".to_string()),
|
||||
turns: Vec::new(),
|
||||
};
|
||||
app.handle_app_server_event(
|
||||
&app_server,
|
||||
codex_app_server_client::AppServerEvent::ServerNotification(
|
||||
ServerNotification::ThreadStarted(ThreadStartedNotification {
|
||||
thread: rejected_thread.clone(),
|
||||
}),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(!app.thread_event_channels.contains_key(&rejected_thread_id));
|
||||
assert!(app.agent_navigation.get(&rejected_thread_id).is_none());
|
||||
|
||||
let accepted_thread = Thread {
|
||||
id: accepted_thread_id.to_string(),
|
||||
session_id: accepted_thread_id.to_string(),
|
||||
preview: "accepted thread".to_string(),
|
||||
name: Some("accepted thread".to_string()),
|
||||
..rejected_thread
|
||||
};
|
||||
app.handle_app_server_event(
|
||||
&app_server,
|
||||
codex_app_server_client::AppServerEvent::ServerNotification(
|
||||
ServerNotification::ThreadStarted(ThreadStartedNotification {
|
||||
thread: accepted_thread,
|
||||
}),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(app.thread_event_channels.contains_key(&accepted_thread_id));
|
||||
assert!(app.agent_navigation.get(&accepted_thread_id).is_some());
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// `thread/read` is metadata/replay hydration and does not return a fresh
|
||||
/// server-authored `PermissionProfile`, so it must not reuse the cached primary
|
||||
/// session profile after swapping in the read thread's cwd.
|
||||
|
||||
@@ -127,6 +127,7 @@ use color_eyre::eyre::ContextCompat;
|
||||
use color_eyre::eyre::Result;
|
||||
use color_eyre::eyre::WrapErr;
|
||||
use std::collections::HashMap;
|
||||
use std::collections::HashSet;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
@@ -135,6 +136,8 @@ use uuid::Uuid;
|
||||
const JSONRPC_INVALID_REQUEST: i64 = -32600;
|
||||
const JSONRPC_METHOD_NOT_FOUND: i64 = -32601;
|
||||
const THREAD_SETTINGS_UPDATE_METHOD: &str = "thread/settings/update";
|
||||
const DISABLED_FORK_UNSUPPORTED_MESSAGE: &str =
|
||||
"connected app server did not confirm that multi-agent tools were disabled for the fork";
|
||||
|
||||
fn bootstrap_request_error(context: &'static str, err: TypedRequestError) -> color_eyre::Report {
|
||||
color_eyre::eyre::eyre!("{context}: {err}")
|
||||
@@ -176,6 +179,8 @@ pub(crate) struct AppServerSession {
|
||||
thread_settings_update_supported: bool,
|
||||
default_model: Option<String>,
|
||||
available_models: Vec<ModelPreset>,
|
||||
rejected_fork_thread_ids: HashSet<ThreadId>,
|
||||
disabled_forks_unsupported: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
@@ -225,6 +230,8 @@ impl AppServerSession {
|
||||
thread_settings_update_supported: true,
|
||||
default_model: None,
|
||||
available_models: Vec::new(),
|
||||
rejected_fork_thread_ids: HashSet::new(),
|
||||
disabled_forks_unsupported: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -453,6 +460,10 @@ impl AppServerSession {
|
||||
thread_id: ThreadId,
|
||||
multi_agent_tools: MultiAgentToolsOverride,
|
||||
) -> Result<AppServerStartedThread> {
|
||||
if multi_agent_tools == MultiAgentToolsOverride::Disabled && self.disabled_forks_unsupported
|
||||
{
|
||||
return Err(color_eyre::eyre::eyre!(DISABLED_FORK_UNSUPPORTED_MESSAGE));
|
||||
}
|
||||
let request_id = self.next_request_id();
|
||||
let session_config = self.session_config_with_effective_service_tier(&config);
|
||||
let response: ThreadForkResponse = self
|
||||
@@ -474,12 +485,12 @@ impl AppServerSession {
|
||||
if multi_agent_tools == MultiAgentToolsOverride::Disabled
|
||||
&& !response.multi_agent_tools_disabled
|
||||
{
|
||||
self.disabled_forks_unsupported = true;
|
||||
if let Ok(thread_id) = ThreadId::from_string(&response.thread.id) {
|
||||
self.quarantine_rejected_fork(thread_id);
|
||||
let _ = self.thread_unsubscribe(thread_id).await;
|
||||
}
|
||||
return Err(color_eyre::eyre::eyre!(
|
||||
"connected app server did not confirm that multi-agent tools were disabled for the fork"
|
||||
));
|
||||
return Err(color_eyre::eyre::eyre!(DISABLED_FORK_UNSUPPORTED_MESSAGE));
|
||||
}
|
||||
let fork_parent_title = self
|
||||
.fork_parent_title_from_app_server(response.thread.forked_from_id.as_deref())
|
||||
@@ -490,6 +501,14 @@ impl AppServerSession {
|
||||
Ok(started)
|
||||
}
|
||||
|
||||
pub(crate) fn quarantine_rejected_fork(&mut self, thread_id: ThreadId) {
|
||||
self.rejected_fork_thread_ids.insert(thread_id);
|
||||
}
|
||||
|
||||
pub(crate) fn is_rejected_fork_thread(&self, thread_id: ThreadId) -> bool {
|
||||
self.rejected_fork_thread_ids.contains(&thread_id)
|
||||
}
|
||||
|
||||
pub(crate) fn thread_params_mode(&self) -> ThreadParamsMode {
|
||||
self.thread_params_mode
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user