fix: negotiate disabled fork support before creation

This commit is contained in:
jif-oai
2026-06-11 15:24:49 +01:00
parent d7da47715b
commit ca90e60290
18 changed files with 324 additions and 21 deletions
+78
View File
@@ -43,6 +43,7 @@ use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::Result as JsonRpcResult;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ThreadCapabilitiesReadResponse;
use codex_arg0::Arg0DispatchPaths;
use codex_config::CloudConfigBundleLoader;
use codex_config::LoaderOverrides;
@@ -93,6 +94,9 @@ pub mod legacy_core {
}
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
const JSONRPC_INVALID_REQUEST: i64 = -32600;
const JSONRPC_METHOD_NOT_FOUND: i64 = -32601;
const THREAD_CAPABILITIES_READ_METHOD: &str = "thread/capabilities/read";
/// Runs the embedded app-server personality migration.
///
@@ -850,6 +854,40 @@ impl AppServerRequestHandle {
}
impl AppServerClient {
/// Returns whether the connected server supports disabling multi-agent
/// collaboration when forking a thread.
///
/// In-process clients always match this crate version. Remote clients
/// probe the additive capability RPC and treat older servers as
/// unsupported before any mutating fork request is sent.
pub async fn supports_thread_fork_disable_multi_agent_tools(
&self,
request_id: RequestId,
) -> Result<bool, TypedRequestError> {
let Self::Remote(client) = self else {
return Ok(true);
};
match client
.request_typed::<ThreadCapabilitiesReadResponse>(
ClientRequest::ThreadCapabilitiesRead {
request_id,
params: None,
},
)
.await
{
Ok(response) => Ok(response.fork_disable_multi_agent_tools),
Err(TypedRequestError::Server { source, .. })
if source.code == JSONRPC_METHOD_NOT_FOUND
|| (source.code == JSONRPC_INVALID_REQUEST
&& source.message.contains(THREAD_CAPABILITIES_READ_METHOD)) =>
{
Ok(false)
}
Err(err) => Err(err),
}
}
pub async fn request(&self, request: ClientRequest) -> IoResult<RequestResult> {
match self {
Self::InProcess(client) => client.request(request).await,
@@ -938,6 +976,7 @@ mod tests {
use codex_app_server_protocol::AccountUpdatedNotification;
use codex_app_server_protocol::ConfigRequirementsReadResponse;
use codex_app_server_protocol::GetAccountResponse;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCRequest;
use codex_app_server_protocol::JSONRPCResponse;
@@ -1465,6 +1504,45 @@ mod tests {
client.shutdown().await.expect("shutdown should complete");
}
#[tokio::test]
async fn remote_disabled_fork_capability_probe_fails_closed_on_old_server() {
let websocket_url = start_test_remote_server(|mut websocket| async move {
expect_remote_initialize(&mut websocket).await;
let JSONRPCMessage::Request(request) = read_websocket_message(&mut websocket).await
else {
panic!("expected thread capability request");
};
assert_eq!(request.method, THREAD_CAPABILITIES_READ_METHOD);
write_websocket_message(
&mut websocket,
JSONRPCMessage::Error(JSONRPCError {
id: request.id,
error: JSONRPCErrorError {
code: JSONRPC_METHOD_NOT_FOUND,
message: "method not found".to_string(),
data: None,
},
}),
)
.await;
})
.await;
let client = AppServerClient::Remote(
RemoteAppServerClient::connect(test_remote_connect_args(websocket_url))
.await
.expect("remote client should connect"),
);
assert!(
!client
.supports_thread_fork_disable_multi_agent_tools(RequestId::Integer(1))
.await
.expect("old-server capability probe should fail closed")
);
client.shutdown().await.expect("shutdown should complete");
}
#[tokio::test]
async fn remote_unix_socket_typed_request_roundtrip_works() {
let socket_dir = TempDir::new().expect("socket dir");
@@ -4505,6 +4505,29 @@
"title": "Thread/forkRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/capabilities/read"
],
"title": "Thread/capabilities/readRequestMethod",
"type": "string"
},
"params": {
"type": "null"
}
},
"required": [
"id",
"method"
],
"title": "Thread/capabilities/readRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -300,6 +300,29 @@
"title": "Thread/forkRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/v2/RequestId"
},
"method": {
"enum": [
"thread/capabilities/read"
],
"title": "Thread/capabilities/readRequestMethod",
"type": "string"
},
"params": {
"type": "null"
}
},
"required": [
"id",
"method"
],
"title": "Thread/capabilities/readRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -16069,6 +16092,20 @@
"title": "ThreadArchivedNotification",
"type": "object"
},
"ThreadCapabilitiesReadResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"forkDisableMultiAgentTools": {
"description": "Whether `thread/fork` supports disabling multi-agent collaboration.",
"type": "boolean"
}
},
"required": [
"forkDisableMultiAgentTools"
],
"title": "ThreadCapabilitiesReadResponse",
"type": "object"
},
"ThreadClosedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
@@ -1242,6 +1242,29 @@
"title": "Thread/forkRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/capabilities/read"
],
"title": "Thread/capabilities/readRequestMethod",
"type": "string"
},
"params": {
"type": "null"
}
},
"required": [
"id",
"method"
],
"title": "Thread/capabilities/readRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -13886,6 +13909,20 @@
"title": "ThreadArchivedNotification",
"type": "object"
},
"ThreadCapabilitiesReadResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"forkDisableMultiAgentTools": {
"description": "Whether `thread/fork` supports disabling multi-agent collaboration.",
"type": "boolean"
}
},
"required": [
"forkDisableMultiAgentTools"
],
"title": "ThreadCapabilitiesReadResponse",
"type": "object"
},
"ThreadClosedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
@@ -0,0 +1,14 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"forkDisableMultiAgentTools": {
"description": "Whether `thread/fork` supports disabling multi-agent collaboration.",
"type": "boolean"
}
},
"required": [
"forkDisableMultiAgentTools"
],
"title": "ThreadCapabilitiesReadResponse",
"type": "object"
}
File diff suppressed because one or more lines are too long
@@ -0,0 +1,9 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type ThreadCapabilitiesReadResponse = {
/**
* Whether `thread/fork` supports disabling multi-agent collaboration.
*/
forkDisableMultiAgentTools: boolean, };
@@ -371,6 +371,7 @@ export type { ThreadApproveGuardianDeniedActionResponse } from "./ThreadApproveG
export type { ThreadArchiveParams } from "./ThreadArchiveParams";
export type { ThreadArchiveResponse } from "./ThreadArchiveResponse";
export type { ThreadArchivedNotification } from "./ThreadArchivedNotification";
export type { ThreadCapabilitiesReadResponse } from "./ThreadCapabilitiesReadResponse";
export type { ThreadClosedNotification } from "./ThreadClosedNotification";
export type { ThreadCompactStartParams } from "./ThreadCompactStartParams";
export type { ThreadCompactStartResponse } from "./ThreadCompactStartResponse";
@@ -480,6 +480,11 @@ client_request_definitions! {
serialization: thread_or_path(params.thread_id, params.path),
response: v2::ThreadForkResponse,
},
ThreadCapabilitiesRead => "thread/capabilities/read" {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
serialization: None,
response: v2::ThreadCapabilitiesReadResponse,
},
ThreadArchive => "thread/archive" {
params: v2::ThreadArchiveParams,
serialization: thread_id(params.thread_id),
@@ -589,6 +589,14 @@ pub struct ThreadForkResponse {
pub reasoning_effort: Option<ReasoningEffort>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadCapabilitiesReadResponse {
/// Whether `thread/fork` supports disabling multi-agent collaboration.
pub fork_disable_multi_agent_tools: bool,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
+2 -1
View File
@@ -132,7 +132,8 @@ Example with notification opt-out:
- `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/capabilities/read` — read optional thread behavior supported by this app-server. Clients must require `forkDisableMultiAgentTools: true` before sending a `thread/fork` request with `disableMultiAgentTools: true`; older servers reject this capability request without creating a thread.
- `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; supporting servers return `multiAgentToolsDisabled: true` to acknowledge that the forked runtime applied the restriction. 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.
- `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. Subagent threads also include `parentThreadId` when the immediate control/spawn parent is known.
- `thread/loaded/list` — list the thread ids currently loaded in memory.
@@ -1043,6 +1043,9 @@ impl MessageProcessor {
)
.await
}
ClientRequest::ThreadCapabilitiesRead { params: _, .. } => Ok(Some(
self.thread_processor.thread_capabilities_read().into(),
)),
ClientRequest::ThreadUnsubscribe { params, .. } => {
self.thread_processor
.thread_unsubscribe(&request_id, params)
@@ -178,6 +178,7 @@ use codex_app_server_protocol::ThreadBackgroundTerminalsListParams;
use codex_app_server_protocol::ThreadBackgroundTerminalsListResponse;
use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateParams;
use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateResponse;
use codex_app_server_protocol::ThreadCapabilitiesReadResponse;
use codex_app_server_protocol::ThreadClosedNotification;
use codex_app_server_protocol::ThreadCompactStartParams;
use codex_app_server_protocol::ThreadCompactStartResponse;
@@ -409,6 +409,12 @@ impl ThreadRequestProcessor {
.map(|response| Some(response.into()))
}
pub(crate) fn thread_capabilities_read(&self) -> ThreadCapabilitiesReadResponse {
ThreadCapabilitiesReadResponse {
fork_disable_multi_agent_tools: true,
}
}
pub(crate) async fn thread_resume(
&self,
request_id: ConnectionRequestId,
@@ -3341,7 +3347,7 @@ impl ThreadRequestProcessor {
..
} = self
.thread_manager
.fork_thread_from_history(
.fork_thread_from_history_with_runtime_override(
ForkSnapshot::Interrupted,
config,
InitialHistory::Resumed(ResumedHistory {
@@ -1132,7 +1132,6 @@ impl TurnRequestProcessor {
history: parent_history.items,
rollout_path: parent_thread.rollout_path(),
}),
MultiAgentRuntimeOverride::Inherit,
/*thread_source*/ None,
self.request_trace_context(request_id).await,
)
+26 -9
View File
@@ -864,15 +864,8 @@ impl ThreadManager {
{
let snapshot = snapshot.into();
let history = self.initial_history_from_rollout_path(path).await?;
self.fork_thread_from_history(
snapshot,
config,
history,
MultiAgentRuntimeOverride::Inherit,
thread_source,
parent_trace,
)
.await
self.fork_thread_from_history(snapshot, config, history, thread_source, parent_trace)
.await
}
async fn initial_history_from_rollout_path(
@@ -895,6 +888,30 @@ impl ThreadManager {
/// Fork an existing thread from already-loaded store history.
pub async fn fork_thread_from_history<S>(
&self,
snapshot: S,
config: Config,
history: InitialHistory,
thread_source: Option<ThreadSource>,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread>
where
S: Into<ForkSnapshot>,
{
self.fork_thread_from_history_with_runtime_override(
snapshot,
config,
history,
MultiAgentRuntimeOverride::Inherit,
thread_source,
parent_trace,
)
.await
}
/// Fork an existing thread from already-loaded store history with an
/// explicit multi-agent runtime override.
pub async fn fork_thread_from_history_with_runtime_override<S>(
&self,
snapshot: S,
config: Config,
-1
View File
@@ -199,7 +199,6 @@ async fn fork_thread_from_history_does_not_require_source_rollout_path() {
history: source_items.clone(),
rollout_path: None,
}),
codex_core::MultiAgentRuntimeOverride::Inherit,
/*thread_source*/ None,
/*parent_trace*/ None,
)
+72 -7
View File
@@ -186,7 +186,7 @@ pub(crate) struct AppServerSession {
default_model: Option<String>,
available_models: Vec<ModelPreset>,
rejected_fork_thread_ids: HashSet<ThreadId>,
disabled_forks_unsupported: bool,
disabled_fork_support: DisabledForkSupport,
external_agent_config_import_completion_pending: AtomicBool,
}
@@ -202,6 +202,13 @@ pub(crate) enum MultiAgentToolsOverride {
Disabled,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DisabledForkSupport {
Unknown,
Supported,
Unsupported,
}
impl ThreadParamsMode {
fn model_provider_from_config(self, config: &Config) -> Option<String> {
match self {
@@ -229,6 +236,10 @@ pub(crate) enum TurnPermissionsOverride {
impl AppServerSession {
pub(crate) fn new(client: AppServerClient, thread_params_mode: ThreadParamsMode) -> Self {
let disabled_fork_support = match &client {
AppServerClient::InProcess(_) => DisabledForkSupport::Supported,
AppServerClient::Remote(_) => DisabledForkSupport::Unknown,
};
Self {
client,
next_request_id: 1,
@@ -238,7 +249,7 @@ impl AppServerSession {
default_model: None,
available_models: Vec::new(),
rejected_fork_thread_ids: HashSet::new(),
disabled_forks_unsupported: false,
disabled_fork_support,
external_agent_config_import_completion_pending: AtomicBool::new(false),
}
}
@@ -499,9 +510,27 @@ 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));
if multi_agent_tools == MultiAgentToolsOverride::Disabled {
if self.disabled_fork_support == DisabledForkSupport::Unknown {
let request_id = self.next_request_id();
self.disabled_fork_support = if self
.client
.supports_thread_fork_disable_multi_agent_tools(request_id)
.await
.map_err(|err| {
bootstrap_request_error(
"thread/capabilities/read failed before TUI fork",
err,
)
})? {
DisabledForkSupport::Supported
} else {
DisabledForkSupport::Unsupported
};
}
if self.disabled_fork_support == DisabledForkSupport::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);
@@ -524,10 +553,14 @@ impl AppServerSession {
if multi_agent_tools == MultiAgentToolsOverride::Disabled
&& !response.multi_agent_tools_disabled
{
self.disabled_forks_unsupported = true;
self.disabled_fork_support = DisabledForkSupport::Unsupported;
if let Ok(thread_id) = ThreadId::from_string(&response.thread.id) {
self.quarantine_rejected_fork(thread_id);
let _ = self.thread_unsubscribe(thread_id).await;
if let Err(err) = self.thread_unsubscribe(thread_id).await {
tracing::warn!(
"failed to unsubscribe rejected multi-agent-enabled fork {thread_id}: {err}"
);
}
}
return Err(color_eyre::eyre::eyre!(DISABLED_FORK_UNSUPPORTED_MESSAGE));
}
@@ -2349,6 +2382,38 @@ mod tests {
assert!(params.disable_multi_agent_tools);
}
#[tokio::test]
async fn unsupported_disabled_fork_is_rejected_before_creating_a_thread() {
let temp_dir = tempfile::tempdir().expect("tempdir");
let config = build_config(&temp_dir).await;
let mut app_server = crate::start_embedded_app_server_for_picker(&config)
.await
.expect("embedded app server");
app_server.disabled_fork_support = DisabledForkSupport::Unsupported;
let before = app_server
.thread_loaded_list(ThreadLoadedListParams::default())
.await
.expect("loaded threads before rejected fork");
let error = app_server
.fork_thread(config, ThreadId::new(), MultiAgentToolsOverride::Disabled)
.await
.expect_err("unsupported disabled fork should fail");
assert!(
error
.to_string()
.contains(DISABLED_FORK_UNSUPPORTED_MESSAGE)
);
assert_eq!(
app_server
.thread_loaded_list(ThreadLoadedListParams::default())
.await
.expect("loaded threads after rejected fork"),
before
);
}
#[tokio::test]
async fn terminal_visualization_instructions_are_gated_for_all_tui_thread_flows() {
let temp_dir = tempfile::tempdir().expect("tempdir");