From 97239fd052db455bd08c41afcdab738e0fe83b87 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 12 Oct 2026 14:35:40 +0000 Subject: [PATCH] Add explicit local board connections with fixed runtime access (#53400) ## Why Local message boards were scoped to an agent tree. Independently created Codex runtimes need a way to share a board while receiving different access permissions. ## What changed - Add `LocalBoard` and `LocalBoardConnection` so hosts can supply read-only or read-write access through `thread_extension_init`, including for isolated, ephemeral runtimes. - Allow read-only connections to read and manage their own subscriptions; reject posting, replying, channel creation, and changes to another agent's subscriptions. Expose effective permissions in channel discovery. - Use unique runtime paths and require matching board IDs and canonical database paths when resolving agents or delivering notifications. - Document explicit extension installation and connection setup for new runtimes, cold resumes, and forks. ## Testing Add a scenario and context snapshot covering an independent writer and isolated Guardian sharing a board through a symlink alias, read-only permission denials, subscription and explicit notifications, and rejection of notifications to an unconnected runtime before a post is written. GitOrigin-RevId: f92410c3a5c6a46920c3e0e71849c85dc83d9c31 --- codex-rs/core/src/agent_message_board.rs | 68 +++-- codex-rs/core/src/thread_manager.rs | 21 ++ .../suite/scenarios_agent_message_board.rs | 260 ++++++++++++++++++ ...gent_message_board__board_connections.snap | 120 ++++++++ codex-rs/ext/agent-message-board/README.md | 27 ++ codex-rs/ext/agent-message-board/src/api.rs | 2 +- .../ext/agent-message-board/src/extension.rs | 20 +- codex-rs/ext/agent-message-board/src/host.rs | 2 +- codex-rs/ext/agent-message-board/src/lib.rs | 2 + codex-rs/ext/agent-message-board/src/local.rs | 35 ++- .../src/local/connection.rs | 113 ++++++++ .../agent-message-board/src/local/queries.rs | 8 +- .../agent-message-board/src/permissions.rs | 4 +- 13 files changed, 647 insertions(+), 35 deletions(-) create mode 100644 codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap create mode 100644 codex-rs/ext/agent-message-board/src/local/connection.rs diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs index 62e60c92b6..8af388dce5 100644 --- a/codex-rs/core/src/agent_message_board.rs +++ b/codex-rs/core/src/agent_message_board.rs @@ -16,6 +16,7 @@ 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; +use codex_agent_message_board_extension::LocalBoardConnection; use codex_agent_message_board_extension::MessageBoardHost; use codex_agent_message_board_extension::NotificationDelivery; use codex_agent_message_board_extension::PostPreview; @@ -58,28 +59,31 @@ pub fn install_agent_message_board( |config: &Config| config.multi_agent_v2.tool_namespace.clone(), move |input: &ThreadStartInput<'_, Config>, tree, caller| { let config = input.config; + let connection = input.thread_store.get::(); let in_memory = config.multi_agent_v2.message_board_in_memory; - // 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.multi_agent_v2.message_board_remote.is_none()) + // Automatic boards need MAv2 and persistent storage. An explicit host + // connection also works for isolated, ephemeral runtimes. + if connection.is_none() + && (!config.features.enabled(Feature::AgentMessageBoard) + || !config.features.enabled(Feature::MultiAgentV2) + || (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(), tree, caller, expected_turn_id: None, + connection: connection.clone(), }); Box::pin(async move { - let board: Arc = if let Some(remote) = remote { + let board: Arc = if let Some(connection) = connection { + Arc::new(connection.open(host).await?) + } else if let Some(remote) = config.multi_agent_v2.message_board_remote.clone() { 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(), @@ -90,7 +94,8 @@ pub fn install_agent_message_board( }; let token = AccessToken::new(token) .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?; - let http = http_factory + let http = config + .http_client_factory() .build_client(&remote.url, ClientRouteClass::Api) .map_err(|err| CodexErr::Io(std::io::Error::other(err)))?; input.thread_store.insert(host.as_ref().clone()); @@ -104,7 +109,9 @@ pub fn install_agent_message_board( } else if in_memory { Arc::new(in_memory_boards.open(tree, host).await) } else { - Arc::new(LocalAgentMessageBoard::open(&sqlite, tree, host).await?) + Arc::new( + LocalAgentMessageBoard::open(config.sqlite_config(), tree, host).await?, + ) }; Ok(Some(board)) }) @@ -118,6 +125,7 @@ struct LocalBoardHost { tree: SessionId, caller: ThreadId, expected_turn_id: Option, + connection: Option>, } impl LocalBoardHost { @@ -126,8 +134,16 @@ impl LocalBoardHost { .manager .upgrade() .ok_or_else(|| CodexErr::ThreadNotFound(caller))?; + if let Some(connection) = &self.connection { + return manager.get_thread_for_board(caller, connection).await; + } let actor = manager.get_thread(caller).await?; - if actor.session.session_id() != self.tree { + if actor.session.session_id() != self.tree + || actor + .thread_extension_data() + .get::() + .is_some() + { return Err(CodexErr::InvalidRequest( "agent belongs to another message board".into(), )); @@ -137,6 +153,9 @@ impl LocalBoardHost { async fn registered_path(&self, actor: &CodexThread) -> Result { let caller = actor.session.thread_id; + if let Some(connection) = &self.connection { + return connection.caller_path(caller); + } let path = actor .session_source .get_agent_path() @@ -172,6 +191,19 @@ impl MessageBoardHost for LocalBoardHost { fn resolve_agent(&self, path: AgentPath) -> BoxFuture<'_, Result> { Box::pin(async move { + if self.connection.is_some() { + let caller = path + .as_str() + .strip_prefix("/root/") + .and_then(|id| ThreadId::from_string(id).ok()) + .ok_or_else(|| { + CodexErr::InvalidRequest("invalid connected board agent path".into()) + })?; + return self + .actor(caller) + .await + .map(|actor| actor.session.thread_id); + } let actor = self.actor(self.caller).await?; actor .session @@ -207,21 +239,13 @@ impl MessageBoardHost for LocalBoardHost { post: PostPreview, ) -> BoxFuture<'_, Result> { Box::pin(async move { - let Some(manager) = self.manager.upgrade() else { - return Ok(NotificationDelivery::SkippedInactive); - }; - let recipient = match manager.get_thread(recipient_id).await { + let recipient = match self.actor(recipient_id).await { Ok(thread) => thread, Err(error) if matches!(error.details(), CodexErrorDetails::ThreadNotFound(_)) => { return Ok(NotificationDelivery::SkippedInactive); } Err(error) => return Err(error), }; - if recipient.session.session_id() != self.tree { - return Err(CodexErr::InvalidRequest( - "notification recipient belongs to another board".into(), - )); - } let recipient_path = self.registered_path(&recipient).await?; let notice = AgentMessageBoardNotification(post); let communication = InterAgentCommunication::new( diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index a0b6f169d5..4b256b1a39 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -953,6 +953,27 @@ impl ThreadManager { self.state.get_thread(thread_id).await } + /// Board hosts may reach internal runtimes only through a matching explicit connection. + pub(crate) async fn get_thread_for_board( + &self, + thread_id: ThreadId, + connection: &codex_agent_message_board_extension::LocalBoardConnection, + ) -> CodexResult> { + self.state + .threads + .read() + .await + .get(&thread_id) + .filter(|thread| { + thread + .thread_extension_data() + .get::() + .is_some_and(|other| connection.same_board(&other)) + }) + .cloned() + .ok_or(CodexErr::ThreadNotFound(thread_id)) + } + /// Updates metadata for loaded and cold threads through one entrypoint. /// /// Loaded threads route through `CodexThread`/`LiveThread`, so metadata changes stay ordered 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 ca587b80c8..918908e68c 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs @@ -931,3 +931,263 @@ async fn board_clock_failure_does_not_fall_back_or_create_a_channel() -> anyhow: ); Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn codex_instances_share_board_with_fixed_access() -> anyhow::Result<()> { + use codex_agent_message_board_extension::LocalBoard; + use codex_extension_api::ExtensionRegistryBuilder; + use codex_extension_api::IsolatedSessionExtensions; + use codex_extension_api::SessionIsolation; + use codex_extension_api::ToolName; + use codex_protocol::protocol::InternalSessionSource; + use codex_protocol::protocol::SessionSource; + use codex_state::SqliteConfig; + use std::sync::Arc; + + let server = responses::start_mock_server().await; + let response = |body| vec![StreamingSseChunk { gate: None, body }]; + let (release, gate) = oneshot::channel(); + let mut working = ev_assistant_message("watching", "Watching the policy board."); + working["item"]["phase"] = json!("commentary"); + let (streaming, _) = start_streaming_sse_server(vec![ + response(tool( + "publish", + "post", + json!({"new_channel_name":"policy", "text":"A shared policy"}), + )), + response(done()), + response(tool("permissions", "get_channels", json!({}))), + response(tool("read", "search_posts", json!({}))), + response(tool( + "denied", + "post", + json!({"channel_name":"policy", "text":"Cannot publish"}), + )), + response(tool("deny-create", "create_channel", json!({"channel_name":"denied"}))), + response(tool("deny-new", "post", json!({"new_channel_name":"denied", "text":"Cannot publish"}))), + response(tool("deny-reply", "post", json!({"thread_id":"00000000-0000-4000-8000-000000000099", "text":"Cannot reply"}))), + response(tool("deny-subscribe", "subscribe", json!({"channel_name":"policy", "target_agent":"/root/00000000000040008000000000000000"}))), + response(done()), + response(tool( + "subscribe", + "subscribe", + json!({"channel_name":"policy"}), + )), + vec![ + StreamingSseChunk { + gate: None, + body: sse(vec![ev_response_created("watching"), working]), + }, + StreamingSseChunk { + gate: Some(gate), + body: tool("checkpoint", "get_channels", json!({})), + }, + ], + response(sse(vec![ + ev_function_call_with_namespace( + "wrong-board", + "collaboration", + "post", + &json!({ + "channel_name":"policy", "text":"Must not publish", + "agents_to_notify":["/root/00000000000040008000000000000002"] + }) + .to_string(), + ), + ev_function_call_with_namespace( + "update", + "collaboration", + "post", + &json!({"channel_name":"policy", "text":"A new policy"}).to_string(), + ), + ev_function_call_with_namespace( + "explicit-update", + "collaboration", + "post", + &json!({ + "channel_name":"policy", "text":"An explicit update", + "agents_to_notify":["/root/00000000000040008000000000000001"] + }) + .to_string(), + ), + ev_completed("update"), + ])), + response(done()), + response(done()), + ]) + .await; + let base_url = format!("{}/v1", streaming.uri()); + let test = test_codex() + .with_config(move |config| { + super::configure_scenario_catalog(config); + config.ephemeral = true; + config.model_provider.base_url = Some(base_url); + config + .features + .disable(Feature::EnableRequestCompression) + .expect("capture gated requests"); + config.multi_agent_v2.tool_namespace = Some("collaboration".into()); + config.current_time_reminder = Some(codex_core::config::CurrentTimeReminderConfig { + clock_source: codex_features::CurrentTimeSource::External, + ..Default::default() + }); + }) + .with_external_time_provider(Arc::new(BoardClock::Available)) + .build_with_auto_env(&server) + .await?; + let storage = tempfile::tempdir()?; + let database_home = storage.path().join("data"); + let alias = storage.path().join("alias"); + std::fs::create_dir(&database_home)?; + core_test_support::create_directory_symlink(&database_home, &alias); + let board_id = codex_protocol::SessionId::new(); + let writer_board = LocalBoard::new( + SqliteConfig::new_for_testing(database_home.try_into()?), + board_id, + ); + let guardian_board = + LocalBoard::new(SqliteConfig::new_for_testing(alias.try_into()?), board_id); + let mut extensions = ExtensionRegistryBuilder::new(); + codex_core::install_agent_message_board(&mut extensions, Arc::downgrade(&test.thread_manager)); + let extensions = Arc::new(extensions.build()); + let mut instances = Vec::new(); + for (index, (connection, source)) in [ + (writer_board.read_write(), SessionSource::Cli), + ( + guardian_board.read_only(), + SessionSource::Internal(InternalSessionSource::Guardian), + ), + ] + .into_iter() + .enumerate() + { + let mut options = test.start_thread_options().await; + let internal = source.is_internal(); + options.session_source = Some(source); + options.reserved_thread_id = Some(codex_protocol::ThreadId::from_string(&format!( + "00000000-0000-4000-8000-00000000000{index}" + ))?); + options + .thread_extension_init + .insert(SessionIsolation::Isolated); + options + .thread_extension_init + .insert(IsolatedSessionExtensions(extensions.clone())); + let mut policy = codex_guardian_reviewer::reviewer_tool_policy(); + policy.allowed_tools.as_mut().unwrap().extend( + [ + "post", + "create_channel", + "get_channels", + "search_posts", + "subscribe", + ] + .map(|name| ToolName::namespaced("collaboration", name)), + ); + options.thread_extension_init.insert(policy); + options.thread_extension_init.insert(connection); + let instance = if internal { + test.thread_manager + .spawn_internal_session(test.session_configured.thread_id, options) + .await? + } else { + test.thread_manager.start_thread(options).await? + } + .thread; + instance + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Use the shared policy board.".into(), + text_elements: vec![], + }])) + .await?; + wait_for_event(&instance, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + instances.push(instance); + } + assert_eq!(streaming.requests().await.len(), 10); + + // Keep the Guardian active while the independent writer delivers a subscribed notice. + let writer = &instances[0]; + let guardian = &instances[1]; + // A same-tree internal runtime has no board access unless its host supplies a connection. + let writer_id = + codex_protocol::ThreadId::from_string(writer.thread_extension_data().level_id())?; + let mut options = test.start_thread_options().await; + options.session_source = Some(SessionSource::Internal(InternalSessionSource::Guardian)); + options.reserved_thread_id = Some(codex_protocol::ThreadId::from_string( + "00000000-0000-4000-8000-000000000002", + )?); + let unconnected = test + .thread_manager + .spawn_internal_session(writer_id, options) + .await?; + assert_eq!(unconnected.session_configured.session_id, writer_id.into()); + guardian + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Watch for updates.".into(), + text_elements: vec![], + }])) + .await?; + wait_for_event(guardian, |event| { + matches!(event, EventMsg::ItemCompleted(event) + if matches!(&event.item, TurnItem::AgentMessage(message) if message.id == "watching")) + }) + .await; + writer + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Publish an update.".into(), + text_elements: vec![], + }])) + .await?; + wait_for_event(writer, |event| matches!(event, EventMsg::TurnComplete(_))).await; + release + .send(()) + .expect("release Guardian after notification delivery"); + wait_for_event(guardian, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let requests = streaming.requests().await; + assert!( + requests + .iter() + .any(|body| String::from_utf8_lossy(body).contains("Message Type: CHANNEL_POST")) + ); + let last: Value = serde_json::from_slice(requests.last().unwrap())?; + let checkpoint = last["input"] + .as_array() + .unwrap() + .iter() + .find(|item| item["call_id"] == "checkpoint" && item["type"] == "function_call_output") + .context("channel checkpoint")?; + let channels: Value = serde_json::from_str(checkpoint["output"].as_str().unwrap())?; + assert_eq!(channels["results"][0]["message_count"], 3); // The unconnected runtime's notice was rejected before writing. + let mut bodies = streaming + .requests() + .await + .iter() + .map(|body| serde_json::from_slice::(body)) + .collect::>>()?; + assert_eq!(bodies.len(), 15); + for body in &mut bodies { + for item in body["input"].as_array_mut().context("request input")? { + if let Some(Value::String(output)) = item.get_mut("output") { + *output = context_snapshot::normalize_json_lines(output); + } + } + } + insta::assert_snapshot!( + "board_connections", + context_snapshot::format_context_snapshot( + "Independent Codex runtimes share a local board: a writer publishes, an isolated Guardian reads and cannot publish.", + &bodies.iter().map(SnapshotEntry::body).collect::>(), + &ContextSnapshotOptions::default().rewrite_known_segments(), + ) + ); + + unconnected.thread.shutdown_and_wait().await?; + for instance in instances { + instance.shutdown_and_wait().await?; + } + streaming.shutdown().await; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap new file mode 100644 index 0000000000..191cbcd810 --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__board_connections.snap @@ -0,0 +1,120 @@ +--- +source: core/tests/suite/scenarios_agent_message_board.rs +expression: "context_snapshot::format_context_snapshot(\"Independent Codex runtimes share a local board: a writer publishes, an isolated Guardian reads and cannot publish.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(),\n&ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: Independent Codex runtimes share a local board: a writer publishes, an isolated Guardian reads and cannot publish. + +## Window 1 +-- request 1 (request) -- +00:message/developer: + +01:message/developer: + +02:message/user: + +03:message/user: + Use the shared policy board. +-- request 2 (request) -- +04:function_call/collaboration.post:{"new_channel_name":"policy","text":"A shared policy"} +05:function_call_output:{"author":"/root/00000000000040008000000000000000","channel_name":"policy","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} + +## Window 2 (after request 2: input truncated at item 04, settings changed (prompt_cache_key)) +-- request 3 (request) -- +00:message/developer: + +01:message/developer: + +02:message/user: + +03:message/user: + Use the shared policy board. +-- request 4 (request) -- +04:function_call/collaboration.get_channels:{} +05:function_call_output:{"board_permissions":{"create_channel":false,"edit_metadata":false,"manage_permissions":false},"has_more...ata":false,"post":false,"read":true,"reply":false}}]} [hash=eac4eff4c4aa3c0d] +-- request 5 (request) -- +06:function_call/collaboration.search_posts:{} +07:function_call_output:{"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root/000000000000400080000000...d policy","thread_id":"","truncated":false}]} [hash=dba74b349a2e49af] +-- request 6 (request) -- +08:function_call/collaboration.post:{"channel_name":"policy","text":"Cannot publish"} +09:function_call_output:{"error":{"action":"channel.post","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +-- request 7 (request) -- +10:function_call/collaboration.create_channel:{"channel_name":"denied"} +11:function_call_output:{"error":{"action":"board.create_channel","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +-- request 8 (request) -- +12:function_call/collaboration.post:{"new_channel_name":"denied","text":"Cannot publish"} +13:function_call_output:{"error":{"action":"channel.post","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +-- request 9 (request) -- +14:function_call/collaboration.post:{"text":"Cannot reply","thread_id":""} +15:function_call_output:{"error":{"action":"channel.reply","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +-- request 10 (request) -- +16:function_call/collaboration.subscribe:{"channel_name":"policy","target_agent":"/root/00000000000040008000000000000000"} +17:function_call_output:{"error":{"action":"board.manage_subscriptions","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +-- request 11 (request) -- +18:message/assistant: + Done. +19:message/user: + Watch for updates. +-- request 12 (request) -- +20:function_call/collaboration.subscribe:{"channel_name":"policy"} +21:function_call_output:{"channel_name":"policy","enabled":true,"last_message_id":"","target_agent":"/root/00000000000040008000000000000001","thread_id":null} + +## Window 3 (after request 12: input diverged at item 04, settings changed (prompt_cache_key)) +-- request 13 (request) -- +00:message/developer: + +01:message/developer: + +02:message/user: + +03:message/user: + Use the shared policy board. +04:function_call/collaboration.post:{"new_channel_name":"policy","text":"A shared policy"} +05:function_call_output:{"author":"/root/00000000000040008000000000000000","channel_name":"policy","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} +06:message/assistant: + Done. +07:message/user: + Publish an update. +-- request 14 (request) -- +08:function_call/collaboration.post:{"agents_to_notify":["/root/00000000000040008000000000000002"],"channel_name":"policy","text":"Must not publish"} +09:function_call/collaboration.post:{"channel_name":"policy","text":"A new policy"} +10:function_call/collaboration.post:{"agents_to_notify":["/root/00000000000040008000000000000001"],"channel_name":"policy","text":"An explicit update"} +11:function_call_output:no thread with id: +12:function_call_output:{"author":"/root/00000000000040008000000000000000","channel_name":"policy","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} +13:function_call_output:{"author":"/root/00000000000040008000000000000000","channel_name":"policy","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} + +## Window 4 (after request 14: input diverged at item 04, settings changed (prompt_cache_key)) +-- request 15 (request) -- +00:message/developer: + +01:message/developer: + +02:message/user: + +03:message/user: + Use the shared policy board. +04:function_call/collaboration.get_channels:{} +05:function_call_output:{"board_permissions":{"create_channel":false,"edit_metadata":false,"manage_permissions":false},"has_more...ata":false,"post":false,"read":true,"reply":false}}]} [hash=eac4eff4c4aa3c0d] +06:function_call/collaboration.search_posts:{} +07:function_call_output:{"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root/000000000000400080000000...d policy","thread_id":"","truncated":false}]} [hash=dba74b349a2e49af] +08:function_call/collaboration.post:{"channel_name":"policy","text":"Cannot publish"} +09:function_call_output:{"error":{"action":"channel.post","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +10:function_call/collaboration.create_channel:{"channel_name":"denied"} +11:function_call_output:{"error":{"action":"board.create_channel","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +12:function_call/collaboration.post:{"new_channel_name":"denied","text":"Cannot publish"} +13:function_call_output:{"error":{"action":"channel.post","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +14:function_call/collaboration.post:{"text":"Cannot reply","thread_id":""} +15:function_call_output:{"error":{"action":"channel.reply","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +16:function_call/collaboration.subscribe:{"channel_name":"policy","target_agent":"/root/00000000000040008000000000000000"} +17:function_call_output:{"error":{"action":"board.manage_subscriptions","code":"permission_denied","message":"This board connection is read-only.","resource":"board:"}} +18:message/assistant: + Done. +19:message/user: + Watch for updates. +20:function_call/collaboration.subscribe:{"channel_name":"policy"} +21:function_call_output:{"channel_name":"policy","enabled":true,"last_message_id":"","target_agent":"/root/00000000000040008000000000000001","thread_id":null} +22:message/assistant: + Watching the policy board. +23:function_call/collaboration.get_channels:{} +24:function_call_output:{"board_permissions":{"create_channel":false,"edit_metadata":false,"manage_permissions":false},"has_more...ata":false,"post":false,"read":true,"reply":false}}]} [hash=7cc8dc25fc1c52b2] +25:agent_message +26:agent_message diff --git a/codex-rs/ext/agent-message-board/README.md b/codex-rs/ext/agent-message-board/README.md index ca36ab393e..7f0ff61069 100644 --- a/codex-rs/ext/agent-message-board/README.md +++ b/codex-rs/ext/agent-message-board/README.md @@ -25,3 +25,30 @@ otherwise it uses UTC now. This loads channels and descriptions only. ACLs and roles are not accepted by this version; existing agent access stays unchanged. + +## Connect Codex runtimes + +The host can give independently created Codex runtimes different access to the +same local board through the existing startup extension settings: + +```rust +let board = LocalBoard::new(sqlite_config, board_id); +guardian_options.thread_extension_init.insert(board.read_only()); +writer_options.thread_extension_init.insert(board.read_write()); +``` + +Both connections carry the same SQLite configuration and board ID. Read-only +connections can read and manage their own subscriptions; posting, replying, +creating channels and changing another agent's subscriptions return permission +errors. Explicit connections expose effective permissions in channel discovery. +Runtimes without a connection keep the existing behavior and response shapes. + +Install `codex_core::install_agent_message_board` in the runtime's extension +registry. For isolated Guardians, use their `IsolatedSessionExtensions` registry +and explicitly allow the board tools in their `ToolPolicy`. This does not enable +board access for existing Guardians automatically. Connected runtimes use unique +`/root/` paths, independently of their agent trees. + +The host supplies the connection again when creating another runtime, including +cold resume and forks. Connections are not database ACLs or inherited through +`spawn_agent`. Local filesystem writers remain trusted. diff --git a/codex-rs/ext/agent-message-board/src/api.rs b/codex-rs/ext/agent-message-board/src/api.rs index e4e0b9c34b..2f1b4f9198 100644 --- a/codex-rs/ext/agent-message-board/src/api.rs +++ b/codex-rs/ext/agent-message-board/src/api.rs @@ -24,7 +24,7 @@ use uuid::Uuid; // Local and remote implementations must remain usable through the same handle. const _: Option<&dyn AgentMessageBoard> = None; -/// Stores discussions and subscriptions for one agent tree. +/// Stores discussions and subscriptions for one board. /// /// Every operation validates caller membership. IDs are scoped to this board; /// an agent runtime's ThreadId is distinct from a discussion's root post UUID. diff --git a/codex-rs/ext/agent-message-board/src/extension.rs b/codex-rs/ext/agent-message-board/src/extension.rs index 3d39c3d7d9..9ca056bb4d 100644 --- a/codex-rs/ext/agent-message-board/src/extension.rs +++ b/codex-rs/ext/agent-message-board/src/extension.rs @@ -1,10 +1,11 @@ -//! Installs caller-bound tools using persistent host identities. +//! Installs caller-bound tools using host identities and optional local connections. //! //! Backend selection is a host concern. Startup failures leave tools unavailable //! and emit a warning; a disabled factory does not open storage. //! Tool namespaces follow the host's configuration at runtime startup. use crate::AgentMessageBoard; +use crate::LocalBoardConnection; use crate::tools::message_board_tools_with_descriptions; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; @@ -60,7 +61,11 @@ impl ThreadLifecycleContributor for BoardExtension { CodexErr::InvalidRequest("invalid board caller identity".into()) })?; if let Some(board) = (self.open)(&input, tree, caller).await? { - if board.identity() != tree { + let connection = input.thread_store.get::(); + let expected = connection + .as_ref() + .map_or(tree, |connection| connection.identity()); + if board.identity() != expected { return Err(CodexErr::InvalidRequest( "message-board factory returned another tree".into(), )); @@ -69,10 +74,13 @@ impl ThreadLifecycleContributor for BoardExtension { board, caller, namespace: (self.tool_namespace)(input.config), - path: input - .session_source - .get_agent_path() - .unwrap_or_else(AgentPath::root), + path: match connection { + Some(connection) => connection.caller_path(caller)?, + None => input + .session_source + .get_agent_path() + .unwrap_or_else(AgentPath::root), + }, }); } Ok::<(), CodexErr>(()) diff --git a/codex-rs/ext/agent-message-board/src/host.rs b/codex-rs/ext/agent-message-board/src/host.rs index a9952c1aa0..c915a41b8d 100644 --- a/codex-rs/ext/agent-message-board/src/host.rs +++ b/codex-rs/ext/agent-message-board/src/host.rs @@ -14,7 +14,7 @@ pub enum NotificationDelivery { SkippedInactive, } -/// Tree-scoped membership, clock and notification access supplied by the host. +/// Board-scoped membership, clock and notification access supplied by the host. /// /// The host remains authoritative for membership even when runtimes are unloaded. /// Clock reads use the posting agent's configured source, including simulated diff --git a/codex-rs/ext/agent-message-board/src/lib.rs b/codex-rs/ext/agent-message-board/src/lib.rs index d4221d4b63..f32bf921de 100644 --- a/codex-rs/ext/agent-message-board/src/lib.rs +++ b/codex-rs/ext/agent-message-board/src/lib.rs @@ -34,6 +34,8 @@ pub use host::NotificationDelivery; pub use in_memory::InMemoryAgentMessageBoard; pub use in_memory::InMemoryMessageBoards; pub use local::LocalAgentMessageBoard; +pub use local::connection::LocalBoard; +pub use local::connection::LocalBoardConnection; pub use local::setup::LocalBoardSetup; pub use local::setup::configure_local_board; pub use permissions::BoardPermissions; diff --git a/codex-rs/ext/agent-message-board/src/local.rs b/codex-rs/ext/agent-message-board/src/local.rs index 335d6279e1..e58ee55c23 100644 --- a/codex-rs/ext/agent-message-board/src/local.rs +++ b/codex-rs/ext/agent-message-board/src/local.rs @@ -1,12 +1,15 @@ -//! SQLite-backed boards. The tree ID scopes every read and write. +//! SQLite-backed boards. The board ID scopes every read and write. //! //! Immediate transactions serialize mutations across independently opened handles. //! Live handles in this process share one connection pool per database path. //! Accepted posts survive runtime unload and process restart, but cannot recreate //! a board after its root has been permanently deleted. //! Opening storage adds optional channel metadata without invalidating older readers. +//! Explicit connections can restrict writes; default handles retain legacy access. +use self::connection::Access; use crate::ChannelDescription; +use crate::ChannelPermissions; use crate::ChannelSummary; use crate::CreateChannelRequest; use crate::MessageBoardHost; @@ -38,6 +41,7 @@ use std::collections::HashSet; use std::sync::Arc; use uuid::Uuid; +pub(crate) mod connection; mod lifecycle; mod paging; mod pool; @@ -57,6 +61,7 @@ pub struct LocalAgentMessageBoard { identity: SessionId, pool: Arc, host: Arc, + access: Option, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -77,6 +82,7 @@ impl LocalAgentMessageBoard { identity, pool: pool::open(sqlite).await?, host, + access: None, }) } @@ -85,6 +91,9 @@ impl LocalAgentMessageBoard { caller: ThreadId, request: CreateChannelRequest, ) -> Result { + if let Some(access) = self.access { + access.require_write(self.identity, "board.create_channel")?; + } validate_channel(&request.channel_name)?; let author = self.host.agent_path(caller).await?; let now = self.host.current_time(caller).await?; @@ -116,6 +125,15 @@ impl LocalAgentMessageBoard { /// into a recipient's later turn. Once committed, notification failures are /// logged without failing the post; the persisted content remains readable. pub async fn post(&self, caller: ThreadId, request: PostRequest) -> Result { + if let Some(access) = self.access { + access.require_write( + self.identity, + match &request.destination { + PostDestination::Thread(_) => "channel.reply", + PostDestination::Channel(_) | PostDestination::NewChannel(_) => "channel.post", + }, + )?; + } let board = self.clone(); tokio::spawn(async move { board.post_inner(caller, request).await }) .await @@ -269,6 +287,11 @@ impl LocalAgentMessageBoard { let caller_path = self.host.agent_path(caller).await?; let target_path = request.target_agent.unwrap_or(caller_path); let target_agent = self.host.resolve_agent(target_path.clone()).await?; + if target_agent != caller + && let Some(access) = self.access + { + access.require_write(self.identity, "board.manage_subscriptions")?; + } let mut tx = self.begin_write().await?; let (channel, root, last) = match &request.target { SubscriptionTarget::Channel(name) => { @@ -413,7 +436,15 @@ impl LocalAgentMessageBoard { .map_err(storage_error)? .ok_or_else(|| invalid("channel not found in this board"))?; Ok(ChannelSummary { - permissions: None, + permissions: self.access.map(|access| { + let write = matches!(access, Access::ReadWrite); + ChannelPermissions { + read: true, + post: write, + reply: write, + edit_metadata: false, + } + }), channel_name: name.to_string(), description: row.get("description"), created_at: DateTime::parse_from_rfc3339(row.get("created_at")) diff --git a/codex-rs/ext/agent-message-board/src/local/connection.rs b/codex-rs/ext/agent-message-board/src/local/connection.rs new file mode 100644 index 0000000000..405b74723d --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/local/connection.rs @@ -0,0 +1,113 @@ +//! Host-supplied connections to one local board. Access is fixed per runtime, +//! independent of the board's storage or the agent's tree identity. + +use super::LocalAgentMessageBoard; +use crate::MessageBoardHost; +use crate::PermissionDenied; +use codex_protocol::AgentPath; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result; +use codex_state::SqliteConfig; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::OnceLock; + +/// Shared local storage from which a host grants individual runtime connections. +#[derive(Clone)] +pub struct LocalBoard { + sqlite: SqliteConfig, + identity: SessionId, +} + +impl LocalBoard { + pub fn new(sqlite: SqliteConfig, identity: SessionId) -> Self { + Self { sqlite, identity } + } + + pub fn read_only(&self) -> LocalBoardConnection { + LocalBoardConnection { + board: self.clone(), + access: Access::ReadOnly, + database: OnceLock::new(), + } + } + + pub fn read_write(&self) -> LocalBoardConnection { + LocalBoardConnection { + board: self.clone(), + access: Access::ReadWrite, + database: OnceLock::new(), + } + } +} + +/// Insert into `StartThreadOptions::thread_extension_init` before creating Codex. +/// Hosts must supply it again for a new runtime, including cold resume or a fork. +/// The board extension must be installed, including in an isolated session's registry. +#[derive(Clone)] +pub struct LocalBoardConnection { + board: LocalBoard, + access: Access, + // Capture the opened database, independently of caller unload or symlink changes. + database: OnceLock, +} + +impl LocalBoardConnection { + pub fn identity(&self) -> SessionId { + self.board.identity + } + + /// Compare the canonical database paths captured when the connections were opened. + pub fn same_board(&self, other: &Self) -> bool { + self.identity() == other.identity() + && self + .database + .get() + .zip(other.database.get()) + .is_some_and(|(database, other)| database == other) + } + + /// Independent runtimes have no common agent tree. Use their unique IDs as paths. + pub fn caller_path(&self, caller: ThreadId) -> Result { + AgentPath::root() + .join(&caller.to_string().replace('-', "")) + .map_err(CodexErr::InvalidRequest) + } + + pub async fn open(&self, host: Arc) -> Result { + let mut board = + LocalAgentMessageBoard::open(&self.board.sqlite, self.identity(), host).await?; + // Pool initialization canonicalizes the directory before opening SQLite. + let options = board.pool.connect_options(); + let database = options.get_filename(); + if self.database.get_or_init(|| database.to_path_buf()) != database { + return Err(CodexErr::InvalidRequest( + "board database changed; create a new board connection".into(), + )); + } + board.access = Some(self.access); + Ok(board) + } +} + +#[derive(Clone, Copy)] +pub(super) enum Access { + ReadOnly, + ReadWrite, +} + +impl Access { + pub(super) fn require_write(self, board: SessionId, action: &str) -> Result<()> { + match self { + Self::ReadWrite => Ok(()), + Self::ReadOnly => Err(PermissionDenied { + message: "This board connection is read-only.".into(), + action: Some(action.into()), + resource: Some(format!("board:{board}")), + } + .into()), + } + } +} diff --git a/codex-rs/ext/agent-message-board/src/local/queries.rs b/codex-rs/ext/agent-message-board/src/local/queries.rs index ead816a66c..c7966f85e5 100644 --- a/codex-rs/ext/agent-message-board/src/local/queries.rs +++ b/codex-rs/ext/agent-message-board/src/local/queries.rs @@ -3,6 +3,7 @@ use super::LocalAgentMessageBoard; use super::MAX_READ_CHARS; use super::StoredPost; +use super::connection::Access; use super::invalid; use super::paging::Window; use super::paging::decode_posts; @@ -10,6 +11,7 @@ use super::paging::direction; use super::paging::preview; use super::storage_error; use crate::AgentMessageBoard; +use crate::BoardPermissions; use crate::ChannelPage; use crate::ChannelQuery; use crate::ChannelSummary; @@ -109,7 +111,11 @@ impl AgentMessageBoard for LocalAgentMessageBoard { } Ok(ChannelPage { channels: window.finish(channels)?, - board_permissions: None, + board_permissions: self.access.map(|access| BoardPermissions { + create_channel: matches!(access, Access::ReadWrite), + edit_metadata: false, + manage_permissions: false, + }), }) }) } diff --git a/codex-rs/ext/agent-message-board/src/permissions.rs b/codex-rs/ext/agent-message-board/src/permissions.rs index 09804ada71..a9019f5b69 100644 --- a/codex-rs/ext/agent-message-board/src/permissions.rs +++ b/codex-rs/ext/agent-message-board/src/permissions.rs @@ -1,4 +1,4 @@ -//! Service-computed permissions and denials. Codex exposes these without evaluating ACLs. +//! Backend-computed permissions and denials exposed by the board tools. use codex_protocol::error::CodexErr; use codex_tools::FunctionCallError; @@ -25,7 +25,7 @@ pub struct ChannelPermissions { pub edit_metadata: bool, } -/// A service denial retained through the board's existing error interface. +/// A backend denial retained through the board's existing error interface. #[derive(Clone, Debug, Serialize)] #[serde(tag = "code", rename = "permission_denied")] pub struct PermissionDenied {