From 68e1a421f55ec8a0e900c5a250aa83c13727b671 Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 29 Sep 2026 13:38:43 +0000 Subject: [PATCH] Support remote agent message boards in multi-agent sessions (#49267) ## What changed Add `features.multi_agent_v2.message_board_remote` with a board URL and a bearer token supplied directly or through an environment variable. When configured, use the remote board in preference to local or in-memory storage, including in ephemeral sessions, while preserving the existing board tools and session identity. Ignore remote board settings in project-local configuration and use the runtime HTTP client and clock for remote requests. ## Testing Add a mock-server integration test and snapshot covering an ephemeral session posting through the existing tool, with the expected bearer authentication, session identity, timestamp, and returned post. GitOrigin-RevId: 263413d90a1ee4c0bdba8c1b08ba9f68bdd7a986 --- codex-rs/Cargo.lock | 2 + codex-rs/Cargo.toml | 1 + codex-rs/config/src/loader/mod.rs | 7 + codex-rs/core/Cargo.toml | 1 + codex-rs/core/config.schema.json | 29 ++++ codex-rs/core/src/agent_message_board.rs | 34 ++++- codex-rs/core/src/config/mod.rs | 3 + .../suite/scenarios_agent_message_board.rs | 3 + .../scenarios_agent_message_board_remote.rs | 125 ++++++++++++++++++ ...age_board__remote__remote_board_tools.snap | 37 ++++++ codex-rs/features/Cargo.toml | 1 + codex-rs/features/src/feature_configs.rs | 17 +++ codex-rs/features/src/lib.rs | 1 + codex-rs/features/src/tests.rs | 1 + 14 files changed, 259 insertions(+), 3 deletions(-) create mode 100644 codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs create mode 100644 codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 732ef7f0b1..8ec0a86608 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2899,6 +2899,7 @@ dependencies = [ "bm25", "chrono", "codex-agent-graph-store", + "codex-agent-message-board-client", "codex-agent-message-board-extension", "codex-agent-roles", "codex-analytics", @@ -3370,6 +3371,7 @@ version = "0.0.0" dependencies = [ "codex-network-proxy", "codex-protocol", + "codex-utils-redacted-string", "pretty_assertions", "schemars 0.8.22", "serde", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index a8246dbcb1..4a063dc3a8 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -206,6 +206,7 @@ codex-websocket-client = { path = "websocket-client" } codex-config = { path = "config" } codex-connectors = { path = "connectors" } codex-agent-extension = { path = "ext/agent" } +codex-agent-message-board-client = { path = "agent-message-board-client" } codex-agent-message-board-extension = { path = "ext/agent-message-board" } codex-connectors-extension = { path = "ext/connectors" } codex-context-fragments = { path = "context-fragments" } diff --git a/codex-rs/config/src/loader/mod.rs b/codex-rs/config/src/loader/mod.rs index c450587985..456436df40 100644 --- a/codex-rs/config/src/loader/mod.rs +++ b/codex-rs/config/src/loader/mod.rs @@ -1157,6 +1157,13 @@ fn sanitize_project_config( { ignored_keys.push("features.shell_snapshot".to_string()); } + if let Some(multi_agent) = features + .get_mut("multi_agent_v2") + .and_then(TomlValue::as_table_mut) + && multi_agent.remove("message_board_remote").is_some() + { + ignored_keys.push("features.multi_agent_v2.message_board_remote".to_string()); + } for key in ["respect_system_proxy", "system_proxy_fallback"] { if features.remove(key).is_some() { ignored_keys.push(format!("features.{key}")); diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index d5fd95d2de..53c64fff36 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -69,6 +69,7 @@ codex-skills = { workspace = true } codex-skills-extension = { workspace = true } codex-state = { workspace = true } codex-agent-message-board-extension = { workspace = true } +codex-agent-message-board-client = { workspace = true } codex-terminal-detection = { workspace = true } codex-thread-store = { workspace = true } codex-tools = { workspace = true } diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 1e736a1a66..a3e5627aac 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -2795,6 +2795,14 @@ "description": "Keep the message board in memory for a training session, including ephemeral sessions.", "type": "boolean" }, + "message_board_remote": { + "allOf": [ + { + "$ref": "#/definitions/RemoteMessageBoardConfigToml" + } + ], + "description": "Use a session-scoped remote board instead of local storage." + }, "min_wait_timeout_ms": { "format": "int64", "maximum": 3600000.0, @@ -3842,6 +3850,27 @@ } ] }, + "RemoteMessageBoardConfigToml": { + "additionalProperties": false, + "description": "Connection to a board provisioned by the research host.", + "properties": { + "bearer_token": { + "description": "Board credential supplied directly by a runtime config override.", + "type": "string" + }, + "bearer_token_env_var": { + "description": "Read the credential from this environment variable instead of bearer_token.", + "type": "string" + }, + "url": { + "type": "string" + } + }, + "required": [ + "url" + ], + "type": "object" + }, "ResumeCwdMode": { "description": "Working directory to use when resuming or forking a session.", "oneOf": [ diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs index c004e78dd0..56ab5db27b 100644 --- a/codex-rs/core/src/agent_message_board.rs +++ b/codex-rs/core/src/agent_message_board.rs @@ -11,6 +11,8 @@ use crate::context::ContextualUserFragment; use crate::tools::MULTI_AGENT_V2_NAMESPACE_DESCRIPTION; use chrono::DateTime; use chrono::Utc; +use codex_agent_message_board_client::AccessToken; +use codex_agent_message_board_client::RemoteAgentMessageBoard; use codex_agent_message_board_extension::AgentMessageBoard; use codex_agent_message_board_extension::InMemoryMessageBoards; use codex_agent_message_board_extension::LocalAgentMessageBoard; @@ -19,6 +21,7 @@ use codex_agent_message_board_extension::NotificationDelivery; use codex_agent_message_board_extension::PostPreview; use codex_extension_api::ExtensionRegistryBuilder; use codex_features::Feature; +use codex_http_client::ClientRouteClass; use codex_protocol::AgentPath; use codex_protocol::SessionId; use codex_protocol::ThreadId; @@ -43,14 +46,18 @@ pub fn install_agent_message_board( |config: &Config| config.multi_agent_v2.tool_namespace.clone(), move |config: &Config, tree, caller| { let in_memory = config.multi_agent_v2.message_board_in_memory; - // MAv2 supplies tree paths; ephemeral runtimes must not open durable storage. + // MAv2 supplies tree paths; ephemeral runtimes must not open local SQLite. if !config.features.enabled(Feature::AgentMessageBoard) || !config.features.enabled(Feature::MultiAgentV2) - || (config.ephemeral && !in_memory) + || (config.ephemeral + && !in_memory + && config.multi_agent_v2.message_board_remote.is_none()) { return Box::pin(async { Ok(None) }); } let sqlite = config.sqlite_config().clone(); + let remote = config.multi_agent_v2.message_board_remote.clone(); + let http_factory = config.http_client_factory(); let in_memory_boards = Arc::clone(&in_memory_boards); let host = Arc::new(LocalBoardHost { manager: manager.clone(), @@ -58,7 +65,28 @@ pub fn install_agent_message_board( caller, }); Box::pin(async move { - let board: Arc = if in_memory { + let board: Arc = if let Some(remote) = remote { + let token = match remote.bearer_token_env_var { + Some(name) => std::env::var(name).map_err(|_| CodexErr::InvalidRequest( + "message-board credential environment variable is missing or invalid".into(), + ))?, + None => remote.bearer_token.ok_or_else(|| CodexErr::InvalidRequest( + "remote message board requires a credential".into(), + ))?.into_inner(), + }; + let token = AccessToken::new(token) + .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?; + let http = http_factory + .build_client(&remote.url, ClientRouteClass::Api) + .map_err(|err| CodexErr::Io(std::io::Error::other(err)))?; + let board = RemoteAgentMessageBoard::new(http, &remote.url, tree, token) + .map_err(|err| CodexErr::InvalidRequest(err.to_string()))? + .with_clock(move |caller| { + let host = host.clone(); + Box::pin(async move { host.current_time(caller).await }) + }); + Arc::new(board) + } else if in_memory { Arc::new(in_memory_boards.open(tree, host).await) } else { Arc::new(LocalAgentMessageBoard::open(&sqlite, tree, host).await?) diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 65a8aa3eae..7f30f7bd2b 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1346,6 +1346,7 @@ pub struct MultiAgentV2Config { pub wait_agent_enabled: bool, pub disable_direct_message: bool, pub message_board_in_memory: bool, + pub message_board_remote: Option, pub non_code_mode_only: bool, } @@ -1367,6 +1368,7 @@ impl MultiAgentV2Config { wait_agent_enabled: true, disable_direct_message: false, message_board_in_memory: false, + message_board_remote: None, non_code_mode_only: true, } } @@ -2838,6 +2840,7 @@ fn resolve_multi_agent_v2_config(config_toml: &ConfigToml) -> MultiAgentV2Config wait_agent_enabled, disable_direct_message, message_board_in_memory, + message_board_remote: base.and_then(|config| config.message_board_remote.clone()), non_code_mode_only, } } diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs index a0fcbfaee7..9702e31a97 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs @@ -29,6 +29,9 @@ use serde_json::Value; use serde_json::json; use tokio::sync::oneshot; +#[path = "scenarios_agent_message_board_remote.rs"] +mod remote; + enum BoardClock { Available, Unavailable, diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs new file mode 100644 index 0000000000..cf13cf8404 --- /dev/null +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs @@ -0,0 +1,125 @@ +//! Runs existing board tools against a research-provisioned remote board. + +use super::BoardClock; +use super::configure; +use super::done; +use super::tool; +use anyhow::Context; +use codex_features::RemoteMessageBoardConfigToml; +use core_test_support::context_snapshot; +use core_test_support::context_snapshot::ContextSnapshotOptions; +use core_test_support::context_snapshot::SnapshotEntry; +use core_test_support::responses; +use core_test_support::test_codex::test_codex; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::sync::Arc; +use wiremock::Mock; +use wiremock::ResponseTemplate; +use wiremock::matchers::body_partial_json; +use wiremock::matchers::header; +use wiremock::matchers::method; +use wiremock::matchers::path; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn remote_board_uses_the_existing_tools_and_session_identity() -> anyhow::Result<()> { + let server = responses::start_mock_server().await; + let board = responses::start_mock_server().await; + let token = "research-board-credential-for-runtime-test"; + let url = board.uri(); + let root = test_codex() + .with_config(move |config| { + configure(config); + config.ephemeral = true; + config.current_time_reminder = Some(codex_core::config::CurrentTimeReminderConfig { + clock_source: codex_features::CurrentTimeSource::External, + ..Default::default() + }); + config.multi_agent_v2.message_board_remote = Some(RemoteMessageBoardConfigToml { + url, + bearer_token: Some(token.into()), + bearer_token_env_var: None, + }); + }) + .with_external_time_provider(Arc::new(BoardClock::Available)) + .build_with_auto_env(&server) + .await?; + let root_id = root.session_configured.thread_id; + let post = json!({ + "message_id": "00000000-0000-4000-8000-000000000001", + "thread_id": "00000000-0000-4000-8000-000000000001", + "author": "/root", "channel_name": "design", "created_at": "2026-09-18T12:00:00Z", + }); + Mock::given(method("POST")) + .and(path(format!("/v1/boards/{root_id}/call"))) + .and(header("authorization", format!("Bearer {token}"))) + .and(body_partial_json(json!({ + "caller": root_id, + "timestamp": "2026-09-18T12:00:00Z", + "method": "post", + "params": { + "destination": {"NewChannel": "design"}, + "text": "A remote decision.", "agents_to_notify": [] + } + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(&post)) + .expect(1) + .mount(&board) + .await; + let model = responses::mount_sse_sequence( + &server, + vec![ + tool( + "remote-post", + "post", + json!({"new_channel_name":"design", "text":"A remote decision."}), + ), + done(), + ], + ) + .await; + root.submit_turn("Post the remote decision.").await?; + let requests = model.requests(); + let output: Value = serde_json::from_str( + &requests + .last() + .context("model request")? + .function_call_output_text("remote-post") + .context("tool output")?, + )?; + assert_eq!(output, post); + // Normalize tool JSON key order across Cargo and Bazel feature sets. + let mut bodies = requests + .iter() + .map(responses::ResponsesRequest::body_json) + .collect::>(); + for body in &mut bodies { + for item in body["input"].as_array_mut().context("request input")? { + if item["type"] == "function_call_output" { + let mut output: Value = + serde_json::from_str(item["output"].as_str().context("output")?)?; + output.sort_all_objects(); + item["output"] = Value::String(output.to_string()); + } + } + } + insta::assert_snapshot!( + "remote_board_tools", + context_snapshot::format_context_snapshot( + "Existing board tools use a provisioned remote board.", + &bodies.iter().map(SnapshotEntry::body).collect::>(), + &ContextSnapshotOptions::default().rewrite_known_segments(), + ) + ); + root.codex.shutdown_and_wait().await?; + assert_eq!( + board + .received_requests() + .await + .context("board requests")? + .len(), + 1 + ); + Ok(()) +} diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap new file mode 100644 index 0000000000..d62d153cf1 --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap @@ -0,0 +1,37 @@ +--- +source: core/tests/suite/scenarios_agent_message_board_remote.rs +expression: "context_snapshot::format_context_snapshot(\"Existing board tools use a provisioned remote board.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(),\n&ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: Existing board tools use a provisioned remote board. + +## Window 1 +-- request 1 (request) -- +00:message/developer: + +01:message/developer: + You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals. + + At the start of your turn, you are the active agent. + You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents. + All agents in the team, including the agents that you can assign tasks to, are equally intelligent and capable, and have access to the same set of tools. + + You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task an...message to a running agent without triggering a turn. [hash=7d40ff8b58977202] + Child agents can also spawn their own sub-agents. + + - All agents use the same current working directory. + - As a result, edits made by one agent are immediately visible to all other agents. + + When calling `wait_agent`, prefer longer waits (minutes) to avoid busy polling. + + There are 4 available concurrency slots, meaning that up to 4 agents can be active at once, including you. + + Full-history forks (`fork_turns` omitted or `"all"`) inherit the parent model and reasoning effort and d...fork_turns` to `"none"` or a positive integer string. [hash=db3545796947884f] +02:message/developer: + +03:message/user: + +04:message/user: + Post the remote decision. +-- request 2 (request) -- +05:function_call/collaboration.post:{"new_channel_name":"design","text":"A remote decision."} +06:function_call_output:{"author":"/root","channel_name":"design","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} diff --git a/codex-rs/features/Cargo.toml b/codex-rs/features/Cargo.toml index edc96b64ae..d45e70ff88 100644 --- a/codex-rs/features/Cargo.toml +++ b/codex-rs/features/Cargo.toml @@ -15,6 +15,7 @@ workspace = true [dependencies] codex-network-proxy = { workspace = true } codex-protocol = { workspace = true } +codex-utils-redacted-string = { workspace = true } schemars = { workspace = true } serde = { workspace = true, features = ["derive"] } toml = { workspace = true } diff --git a/codex-rs/features/src/feature_configs.rs b/codex-rs/features/src/feature_configs.rs index b48bc76314..f6b660d5dd 100644 --- a/codex-rs/features/src/feature_configs.rs +++ b/codex-rs/features/src/feature_configs.rs @@ -2,12 +2,26 @@ use crate::FeatureConfig; use crate::FeatureToml; use codex_network_proxy::CredentialProviderConfig; use codex_protocol::openai_models::ReasoningEffort; +use codex_utils_redacted_string::RedactedString; use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; use std::collections::BTreeMap; use std::num::NonZeroUsize; +/// Connection to a board provisioned by the research host. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct RemoteMessageBoardConfigToml { + pub url: String, + /// Board credential supplied directly by a runtime config override. + #[serde(skip_serializing_if = "Option::is_none")] + pub bearer_token: Option, + /// Read the credential from this environment variable instead of bearer_token. + #[serde(skip_serializing_if = "Option::is_none")] + pub bearer_token_env_var: Option, +} + #[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)] #[serde(deny_unknown_fields)] pub struct ToolRegistryConfigToml { @@ -303,6 +317,9 @@ pub struct MultiAgentV2ConfigToml { /// Keep the message board in memory for a training session, including ephemeral sessions. #[serde(skip_serializing_if = "Option::is_none")] pub message_board_in_memory: Option, + /// Use a session-scoped remote board instead of local storage. + #[serde(skip_serializing_if = "Option::is_none")] + pub message_board_remote: Option, #[serde(skip_serializing_if = "Option::is_none")] pub non_code_mode_only: Option, } diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 4b05bdb5be..db35cd4d44 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -32,6 +32,7 @@ pub use feature_configs::NetworkProxyDomainPermissionToml; pub use feature_configs::NetworkProxyModeToml; pub use feature_configs::NetworkProxyUnixSocketPermissionToml; pub use feature_configs::NonPrefixedMcpToolNamesConfigToml; +pub use feature_configs::RemoteMessageBoardConfigToml; use feature_configs::RemovedAppsMcpPathOverrideConfigToml; pub use feature_configs::RolloutBudgetConfigToml; pub use feature_configs::SleepToolConfigToml; diff --git a/codex-rs/features/src/tests.rs b/codex-rs/features/src/tests.rs index e11d777440..ae0a10bd7e 100644 --- a/codex-rs/features/src/tests.rs +++ b/codex-rs/features/src/tests.rs @@ -774,6 +774,7 @@ non_code_mode_only = true wait_agent_enabled: Some(false), disable_direct_message: Some(true), message_board_in_memory: Some(true), + message_board_remote: None, non_code_mode_only: Some(true), })) );