diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 2148c179ca..52bb0e3578 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2102,6 +2102,7 @@ dependencies = [ "base64 0.22.1", "caseless", "chrono", + "codex-extension-api", "codex-protocol", "codex-state", "codex-tools", @@ -3013,6 +3014,7 @@ dependencies = [ "bm25", "chrono", "codex-agent-graph-store", + "codex-agent-message-board-extension", "codex-agent-roles", "codex-analytics", "codex-api", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 94846e19c1..2f0c3a5ecd 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -201,6 +201,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-extension = { path = "ext/agent-message-board" } codex-connectors-extension = { path = "ext/connectors" } codex-context-fragments = { path = "context-fragments" } codex-core = { path = "core" } diff --git a/codex-rs/app-server/src/extensions.rs b/codex-rs/app-server/src/extensions.rs index 25c3999614..34adaff2f2 100644 --- a/codex-rs/app-server/src/extensions.rs +++ b/codex-rs/app-server/src/extensions.rs @@ -72,6 +72,7 @@ pub(crate) fn thread_extensions( codex_queue_extension::install(&mut builder, queue_service); } codex_history_notes_extension::install(&mut builder, auth_manager.clone()); + codex_core::install_agent_message_board(&mut builder, thread_manager.clone()); if let Some(state_db) = state_db { codex_goal_extension::install_with_backend( &mut builder, diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 24c6640cc1..693fbd7b78 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -67,6 +67,7 @@ codex-sandboxing = { workspace = true } codex-skills = { workspace = true } codex-skills-extension = { workspace = true } codex-state = { workspace = true } +codex-agent-message-board-extension = { 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 6e834902ec..9dffbb207f 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -646,6 +646,9 @@ "default": null, "description": "Optional feature toggles scoped to this profile.", "properties": { + "agent_message_board": { + "type": "boolean" + }, "analytics_plan_history": { "type": "boolean" }, @@ -6520,6 +6523,9 @@ "default": null, "description": "Centralized feature flags (new). Prefer this over individual toggles.", "properties": { + "agent_message_board": { + "type": "boolean" + }, "analytics_plan_history": { "type": "boolean" }, diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs new file mode 100644 index 0000000000..a9837f4c01 --- /dev/null +++ b/codex-rs/core/src/agent_message_board.rs @@ -0,0 +1,172 @@ +//! Bridges the board extension to existing tree, clock and active-turn services. +//! +//! The extension owns storage and tools. This adapter never starts or restores +//! recipients and never queues a notification for an idle agent. + +use crate::CodexThread; +use crate::ThreadManager; +use crate::config::Config; +use crate::context::AgentMessageBoardNotification; +use crate::context::ContextualUserFragment; +use chrono::DateTime; +use chrono::Utc; +use codex_agent_message_board_extension::AgentMessageBoard; +use codex_agent_message_board_extension::LocalAgentMessageBoard; +use codex_agent_message_board_extension::MessageBoardHost; +use codex_agent_message_board_extension::NotificationDelivery; +use codex_agent_message_board_extension::PostMetadata; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_features::Feature; +use codex_protocol::AgentPath; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::CodexErrorDetails; +use codex_protocol::error::Result; +use futures::future::BoxFuture; +use std::sync::Arc; +use std::sync::Weak; + +/// Registers the local board for opted-in, persistent MAv2 runtimes. +/// Shared session identity and the configured SQLite home survive runtime reloads. +pub fn install_agent_message_board( + registry: &mut ExtensionRegistryBuilder, + manager: Weak, +) { + codex_agent_message_board_extension::install(registry, move |config: &Config, tree, caller| { + // MAv2 supplies tree paths; ephemeral runtimes must not open durable storage. + if !config.features.enabled(Feature::AgentMessageBoard) + || !config.features.enabled(Feature::MultiAgentV2) + || config.ephemeral + { + return Box::pin(async { Ok(None) }); + } + let sqlite = config.sqlite_config().clone(); + let host = Arc::new(LocalBoardHost { + manager: manager.clone(), + tree, + caller, + }); + Box::pin(async move { + let board = LocalAgentMessageBoard::open(&sqlite, tree, host).await?; + Ok(Some(Arc::new(board) as Arc)) + }) + }); +} + +struct LocalBoardHost { + manager: Weak, + tree: SessionId, + caller: ThreadId, +} + +impl LocalBoardHost { + async fn actor(&self) -> Result> { + let manager = self + .manager + .upgrade() + .ok_or_else(|| CodexErr::ThreadNotFound(self.caller))?; + let actor = manager.get_thread(self.caller).await?; + if actor.session.session_id() != self.tree { + return Err(CodexErr::InvalidRequest( + "agent belongs to another message board".into(), + )); + } + // Match existing collaboration resolution: roots are registered lazily. + if self.caller == ThreadId::from(self.tree) { + actor + .session + .services + .agent_control + .register_session_root(self.caller, /*current_parent_thread_id*/ None); + } + Ok(actor) + } +} + +impl MessageBoardHost for LocalBoardHost { + fn agent_path(&self, caller: ThreadId) -> BoxFuture<'_, Result> { + Box::pin(async move { + self.actor() + .await? + .session + .services + .agent_control + .ensure_agent_known(caller)? + .agent_path + .ok_or_else(|| CodexErr::InvalidRequest("agent has no tree path".into())) + }) + } + + fn resolve_agent(&self, path: AgentPath) -> BoxFuture<'_, Result> { + Box::pin(async move { + let actor = self.actor().await?; + actor + .session + .services + .agent_control + .resolve_agent_reference(self.caller, &actor.session_source, path.as_str()) + .await + }) + } + + fn current_time(&self, caller: ThreadId) -> BoxFuture<'_, Result>> { + Box::pin(async move { + self.agent_path(caller).await?; + let manager = self + .manager + .upgrade() + .ok_or_else(|| CodexErr::ThreadNotFound(caller))?; + let actor = manager.get_thread(caller).await?; + if actor.session.session_id() != self.tree { + return Err(CodexErr::InvalidRequest( + "agent belongs to another message board".into(), + )); + } + actor + .session + .services + .time_provider + .current_time(caller) + .await + .map_err(|error| CodexErr::Io(std::io::Error::other(error))) + }) + } + + fn notify( + &self, + recipient: ThreadId, + post: PostMetadata, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + let Some(manager) = self.manager.upgrade() else { + return Ok(NotificationDelivery::SkippedInactive); + }; + let recipient = match manager.get_thread(recipient).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 notice = AgentMessageBoardNotification { + message_id: post.message_id, + thread_id: post.thread_id, + }; + Ok( + match recipient + .inject_if_running(vec![ContextualUserFragment::into(notice)]) + .await + { + Ok(()) => NotificationDelivery::Accepted, + Err(_) => NotificationDelivery::SkippedInactive, + }, + ) + }) + } +} diff --git a/codex-rs/core/src/context/agent_message_board_notification.rs b/codex-rs/core/src/context/agent_message_board_notification.rs new file mode 100644 index 0000000000..4264879b94 --- /dev/null +++ b/codex-rs/core/src/context/agent_message_board_notification.rs @@ -0,0 +1,34 @@ +//! Fixed-size discussion notices. Post text is fetched through bounded tools. + +use super::ContextualUserFragment; +use codex_protocol::models::ContentItemKind; +use uuid::Uuid; + +pub(crate) struct AgentMessageBoardNotification { + pub(crate) message_id: Uuid, + pub(crate) thread_id: Uuid, +} + +impl ContextualUserFragment for AgentMessageBoardNotification { + fn role(&self) -> &'static str { + "user" + } + fn content_kind(&self) -> ContentItemKind { + ContentItemKind("agent_message_board.notification".into()) + } + fn markers(&self) -> (&'static str, &'static str) { + Self::type_markers() + } + fn type_markers() -> (&'static str, &'static str) { + ( + "", + "", + ) + } + fn body(&self) -> String { + format!( + "\nNew post {} in discussion {}. Use collaboration.read_post or collaboration.read_thread to read it.\n", + self.message_id, self.thread_id + ) + } +} diff --git a/codex-rs/core/src/context/contextual_user_message.rs b/codex-rs/core/src/context/contextual_user_message.rs index 854dbd6471..56563aeaf6 100644 --- a/codex-rs/core/src/context/contextual_user_message.rs +++ b/codex-rs/core/src/context/contextual_user_message.rs @@ -4,6 +4,7 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use super::AdditionalContextUserFragment; +use super::AgentMessageBoardNotification; use super::ContextualUserFragment; use super::InternalModelContextFragment; use super::LegacyApplyPatchExecCommandWarning; @@ -20,6 +21,7 @@ const CONTEXTUAL_USER_FRAGMENT_MATCHERS: &[fn(&str) -> bool] = &[ UserInstructions::matches_text, EnvironmentsState::matches_text, AdditionalContextUserFragment::matches_text, + AgentMessageBoardNotification::matches_text, codex_skills_extension::is_skill_prompt_fragment, UserShellCommand::matches_text, TurnAborted::matches_text, diff --git a/codex-rs/core/src/context/mod.rs b/codex-rs/core/src/context/mod.rs index c4ff26836e..1426c633be 100644 --- a/codex-rs/core/src/context/mod.rs +++ b/codex-rs/core/src/context/mod.rs @@ -1,5 +1,6 @@ //! Context fragments injected into model input. +mod agent_message_board_notification; mod approved_command_prefix_saved; mod apps_instructions; mod available_plugins_instructions; @@ -50,6 +51,7 @@ mod user_shell_command; mod user_verification_notice; pub(crate) mod world_state; +pub(crate) use agent_message_board_notification::AgentMessageBoardNotification; pub(crate) use approved_command_prefix_saved::APPROVED_COMMAND_PREFIX_SAVED_MESSAGE_PREFIX; pub(crate) use approved_command_prefix_saved::ApprovedCommandPrefixSaved; pub(crate) use apps_instructions::AppsInstructions; diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 38b87f4dcc..eb1fc55cdb 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -78,6 +78,8 @@ pub use agent::types::SpawnAgentForkMode; pub use agent::types::SpawnAgentOptions; pub use rollout_budget::RolloutBudgetReminder; mod agent_communication; +mod agent_message_board; +pub use agent_message_board::install_agent_message_board; mod attestation; mod codex_delegate; mod command_canonicalization; diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index 1afb93a360..19153db088 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -742,6 +742,7 @@ impl TestCodexBuilder { .or_else(|| codex_utils_cargo_bin::cargo_bin("codex-code-mode-host").ok()); let thread_manager = Arc::new_cyclic(|manager| { let mut extensions = self.extensions.to_builder(); + codex_core::install_agent_message_board(&mut extensions, manager.clone()); if config.features.enabled(Feature::GuardianV2) { codex_guardian_v2::install(&mut extensions, auth_manager.clone(), manager.clone()); } else { diff --git a/codex-rs/core/tests/suite/scenarios.rs b/codex-rs/core/tests/suite/scenarios.rs index 866e7a4de0..892c319a66 100644 --- a/codex-rs/core/tests/suite/scenarios.rs +++ b/codex-rs/core/tests/suite/scenarios.rs @@ -68,6 +68,9 @@ use tokio::sync::oneshot; const ONE_PIXEL_PNG_BASE64: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg=="; +#[path = "scenarios_agent_message_board.rs"] +mod agent_message_board; + #[path = "scenarios_shared_instructions.rs"] mod shared_instructions; diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs new file mode 100644 index 0000000000..3e8d5bb6c2 --- /dev/null +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs @@ -0,0 +1,373 @@ +//! Exercises board tools through the runtime, including resume and active-only notices. + +use anyhow::Context; +use codex_core::TurnInputRequest; +use codex_features::Feature; +use codex_protocol::protocol::EventMsg; +use codex_protocol::user_input::UserInput; +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::responses::ResponsesRequest; +use core_test_support::responses::ev_assistant_message; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_function_call_with_namespace; +use core_test_support::responses::sse; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; + +enum BoardClock { + Available, + Unavailable, +} + +impl codex_core::TimeProvider for BoardClock { + fn current_time(&self, _thread_id: codex_protocol::ThreadId) -> codex_core::TimeFuture<'_> { + Box::pin(async move { + match self { + Self::Available => Ok(chrono::DateTime::parse_from_rfc3339( + "2026-09-18T12:00:00Z", + )? + .with_timezone(&chrono::Utc)), + Self::Unavailable => Err(anyhow::anyhow!("board clock unavailable")), + } + }) + } + + fn sleep( + &self, + _thread_id: codex_protocol::ThreadId, + _duration: std::time::Duration, + ) -> codex_core::SleepFuture<'_> { + Box::pin(async { Ok(()) }) + } +} + +fn tool(call: &str, name: &str, arguments: Value) -> String { + sse(vec![ + ev_function_call_with_namespace(call, "collaboration", name, &arguments.to_string()), + ev_completed(call), + ]) +} + +fn done() -> String { + sse(vec![ + ev_assistant_message("done", "Done."), + ev_completed("done"), + ]) +} + +fn configure(config: &mut codex_core::config::Config) { + super::configure_scenario_catalog(config); + config + .features + .enable(Feature::AgentMessageBoard) + .expect("enable board"); + config + .features + .enable(Feature::MultiAgentV2) + .expect("enable agent paths"); +} + +#[test_case::test_case(false, true, false; "feature_off")] +#[test_case::test_case(true, false, false; "legacy_agents")] +#[test_case::test_case(true, true, true; "ephemeral")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn board_requires_persistent_v2_runtime( + board_enabled: bool, + multi_agent_v2: bool, + ephemeral: bool, +) -> anyhow::Result<()> { + let server = responses::start_mock_server().await; + let mock = responses::mount_sse_once(&server, done()).await; + let test = test_codex() + .with_config(move |config| { + configure(config); + if !board_enabled { + config + .features + .disable(Feature::AgentMessageBoard) + .expect("disable board"); + } + if !multi_agent_v2 { + config + .features + .disable(Feature::MultiAgentV2) + .expect("disable agent paths"); + } + config.ephemeral = ephemeral; + }) + .build_with_auto_env(&server) + .await?; + test.submit_turn("Hello.").await?; + assert!( + responses::namespace_child_tool( + &mock.single_request().body_json(), + "collaboration", + "post" + ) + .is_none() + ); + assert!( + !test + .config + .sqlite_config() + .home() + .join("agent_message_board_1.sqlite") + .exists() + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn board_active_notice_and_reads_reach_model_context() -> anyhow::Result<()> { + let server = responses::start_mock_server().await; + let mock = responses::mount_sse_sequence(&server, vec![ + tool("post-decision", "post", json!({"new_channel_name":"design", "text":"A shared decision.", "agents_to_notify":["/root"]})), + tool("find-decision", "search_posts", json!({"channel_name":"design","query":"decision"})), + done(), + ]).await; + let test = test_codex() + .with_config(|config| { + configure(config); + config.current_time_reminder = Some(codex_core::config::CurrentTimeReminderConfig { + clock_source: codex_features::CurrentTimeSource::External, + ..Default::default() + }); + }) + .with_external_time_provider(std::sync::Arc::new(BoardClock::Available)) + .build_with_auto_env(&server) + .await?; + test.submit_turn("Record the decision on the board, notify me, and read it.") + .await?; + let requests = mock.requests(); + assert_eq!(requests.len(), 3); + assert!( + responses::namespace_child_tool(&requests[0].body_json(), "collaboration", "post") + .is_some() + ); + let output = requests[1] + .function_call_output_text("post-decision") + .expect("post result"); + let post: Value = + serde_json::from_str(&output).with_context(|| format!("post result: {output}"))?; + assert_eq!(post["author"], "/root"); + assert_eq!(post["created_at"], "2026-09-18T12:00:00Z"); + let result: Value = serde_json::from_str( + &requests[2] + .function_call_output_text("find-decision") + .expect("search result"), + )?; + assert_eq!(result["results"][0]["text_preview"], "A shared decision."); + let notice = requests[1] + .message_input_texts("user") + .into_iter() + .find(|text| text.contains("")) + .expect("active notification"); + assert!(notice.contains(post["message_id"].as_str().expect("post id"))); + assert!(!notice.contains("A shared decision.")); + // Cargo and Bazel can enable different serde_json ordering features. Normalize only + // the snapshot copies so the assertion compares JSON content, not object key order. + let mut bodies = requests + .iter() + .map(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("board output")?)?; + output.sort_all_objects(); + item["output"] = Value::String(output.to_string()); + } + } + } + insta::assert_snapshot!( + "agent_message_board_context", + context_snapshot::format_context_snapshot( + "An active agent posts a shared decision, receives an ID-only notice, and fetches the text.", + &bodies.iter().map(SnapshotEntry::body).collect::>(), + &ContextSnapshotOptions::default().rewrite_known_segments(), + ) + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn board_is_shared_with_children_survives_resume_and_skips_idle_notices() -> anyhow::Result<()> +{ + let server = responses::start_mock_server().await; + let mut builder = test_codex().with_config(configure); + let root = builder.build_with_auto_env(&server).await?; + responses::mount_sse_sequence( + &server, + vec![ + tool( + "create-design", + "create_channel", + json!({"channel_name":"design"}), + ), + done(), + ], + ) + .await; + root.submit_turn("Create the design channel and subscribe.") + .await?; + responses::mount_sse_sequence( + &server, + vec![ + tool( + "spawn-worker", + "spawn_agent", + json!({"task_name":"worker","message":"Say ready.","fork_turns":"none"}), + ), + done(), + done(), + ], + ) + .await; + root.submit_turn("Spawn a worker and finish your turn.") + .await?; + let child_id = root + .thread_manager + .list_thread_ids() + .await + .into_iter() + .find(|id| *id != root.session_configured.thread_id) + .expect("child runtime"); + let child = root.thread_manager.get_thread(child_id).await?; + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let child_post = responses::mount_sse_sequence( + &server, + vec![ + tool( + "worker-post", + "post", + json!({"channel_name":"design","text":"Worker's durable decision."}), + ), + done(), + ], + ) + .await; + child + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Post your decision.".into(), + text_elements: vec![], + }])) + .await?; + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let requests = child_post.requests(); + let output = requests[1] + .function_call_output_text("worker-post") + .expect("child post result"); + let post: Value = + serde_json::from_str(&output).with_context(|| format!("child post result: {output}"))?; + assert_eq!(post["author"], "/root/worker"); + assert!(matches!( + root.codex.agent_status().await, + codex_protocol::protocol::AgentStatus::Completed(_) + )); + let read = responses::mount_sse_sequence( + &server, + vec![ + tool( + "root-read", + "search_posts", + json!({"channel_name":"design"}), + ), + done(), + ], + ) + .await; + root.submit_turn("Read the worker's decision.").await?; + assert!( + read.requests() + .iter() + .all(|request| !request.body_contains_text("")) + ); + let output = read.requests()[1] + .function_call_output_text("root-read") + .expect("read result"); + let result: Value = + serde_json::from_str(&output).with_context(|| format!("root read result: {output}"))?; + assert_eq!(result["results"][0]["message_id"], post["message_id"]); + child.shutdown_and_wait().await?; + let resumed = test_codex() + .with_config(configure) + .restart(&server, &root) + .await?; + let read = responses::mount_sse_sequence( + &server, + vec![ + tool( + "resumed-read", + "read_post", + json!({"message_id":post["message_id"]}), + ), + done(), + ], + ) + .await; + resumed.submit_turn("Read the saved decision.").await?; + let output = read.requests()[1] + .function_call_output_text("resumed-read") + .expect("resumed read result"); + let result: Value = + serde_json::from_str(&output).with_context(|| format!("resumed read result: {output}"))?; + assert_eq!(result["text"], "Worker's durable decision."); + assert_eq!(result["author"], "/root/worker"); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn board_clock_failure_does_not_fall_back_or_create_a_channel() -> anyhow::Result<()> { + let server = responses::start_mock_server().await; + let mock = responses::mount_sse_sequence( + &server, + vec![ + tool( + "failed-post", + "post", + json!({"new_channel_name":"design","text":"Not accepted."}), + ), + tool("channels-after-failure", "get_channels", json!({})), + done(), + ], + ) + .await; + let test = test_codex() + .with_config(|config| { + configure(config); + config.include_environment_context = false; + config.current_time_reminder = Some(codex_core::config::CurrentTimeReminderConfig { + clock_source: codex_features::CurrentTimeSource::External, + ..Default::default() + }); + }) + .with_external_time_provider(std::sync::Arc::new(BoardClock::Unavailable)) + .build_with_auto_env(&server) + .await?; + test.submit_turn("Post the decision and list channels.") + .await?; + let requests = mock.requests(); + assert!( + requests[1] + .function_call_output_text("failed-post") + .expect("clock error") + .contains("board clock unavailable") + ); + let result: Value = serde_json::from_str( + &requests[2] + .function_call_output_text("channels-after-failure") + .expect("channels result"), + )?; + assert_eq!( + result, + json!({"results":[],"n_returned":0,"has_more":false,"next_cursor":null}) + ); + Ok(()) +} diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__agent_message_board_context.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__agent_message_board_context.snap new file mode 100644 index 0000000000..1c94c102c1 --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__agent_message_board_context.snap @@ -0,0 +1,45 @@ +--- +source: core/tests/suite/scenarios_agent_message_board.rs +assertion_line: 88 +expression: "context_snapshot::format_context_snapshot(\"An active agent posts a shared decision, receives an ID-only notice, and fetches the text.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(), &ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: An active agent posts a shared decision, receives an ID-only notice, and fetches the text. + +## 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: + Record the decision on the board, notify me, and read it. +-- request 2 (request) -- +05:function_call/collaboration.post:{"agents_to_notify":["/root"],"new_channel_name":"design","text":"A shared decision."} +06:function_call_output:{"author":"/root","channel_name":"design","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} +07:message/user: + + New post in discussion . Use collaboration.read_post or collaboration.read_thread to read it. + +-- request 3 (request) -- +08:function_call/collaboration.search_posts:{"channel_name":"design","query":"decision"} +09:function_call_output:{"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root","channel_name":"design"...ecision.","thread_id":"","truncated":false}]} [hash=facdb1e4f66cc5b9] diff --git a/codex-rs/ext/agent-message-board/Cargo.toml b/codex-rs/ext/agent-message-board/Cargo.toml index 666a3ddb70..c582ccf919 100644 --- a/codex-rs/ext/agent-message-board/Cargo.toml +++ b/codex-rs/ext/agent-message-board/Cargo.toml @@ -17,6 +17,7 @@ workspace = true base64 = { workspace = true } caseless = { workspace = true } chrono = { workspace = true, features = ["serde"] } +codex-extension-api = { workspace = true } codex-protocol = { workspace = true } codex-state = { workspace = true } codex-tools = { workspace = true } diff --git a/codex-rs/ext/agent-message-board/src/extension.rs b/codex-rs/ext/agent-message-board/src/extension.rs new file mode 100644 index 0000000000..2663161d46 --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/extension.rs @@ -0,0 +1,114 @@ +//! Installs caller-bound tools using persistent host identities. +//! +//! Backend selection is a host concern. Startup failures leave tools unavailable +//! and emit a warning; a disabled factory does not open storage. + +use crate::AgentMessageBoard; +use crate::message_board_tools; +use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionEventSink; +use codex_extension_api::ExtensionFuture; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ExtensionWarning; +use codex_extension_api::ThreadLifecycleContributor; +use codex_extension_api::ThreadStartInput; +use codex_extension_api::ToolContributor; +use codex_protocol::AgentPath; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result; +use codex_tools::ToolCall; +use codex_tools::ToolExecutor; +use futures::future::BoxFuture; +use std::sync::Arc; + +type BoardFactory = dyn Fn(&C, SessionId, ThreadId) -> BoxFuture<'static, Result>>> + + Send + + Sync; + +struct BoardExtension { + open: Box>, + events: Arc, +} +struct Binding { + board: Arc, + caller: ThreadId, + path: AgentPath, +} + +impl ThreadLifecycleContributor for BoardExtension { + fn on_thread_start<'a>(&'a self, input: ThreadStartInput<'a, C>) -> ExtensionFuture<'a, ()> { + Box::pin(async move { + let result = async { + let tree = + SessionId::from_string(input.session_store.level_id()).map_err(|_| { + CodexErr::InvalidRequest("invalid board session identity".into()) + })?; + let caller = + ThreadId::from_string(input.thread_store.level_id()).map_err(|_| { + CodexErr::InvalidRequest("invalid board caller identity".into()) + })?; + if let Some(board) = (self.open)(input.config, tree, caller).await? { + if board.identity() != tree { + return Err(CodexErr::InvalidRequest( + "message-board factory returned another tree".into(), + )); + } + input.thread_store.insert(Binding { + board, + caller, + path: input + .session_source + .get_agent_path() + .unwrap_or_else(AgentPath::root), + }); + } + Ok::<(), CodexErr>(()) + } + .await; + if result.is_err() { + self.events.emit_warning(ExtensionWarning { + thread_id: input.thread_store.level_id().into(), turn_id: None, + message: "Agent message-board initialization failed; its tools are unavailable for this runtime.".into(), + }); + } + }) + } +} + +impl ToolContributor for BoardExtension { + fn tools( + &self, + _session_store: &ExtensionData, + thread_store: &ExtensionData, + ) -> Vec ToolExecutor>>> { + thread_store + .get::() + .map_or_else(Vec::new, |binding| { + message_board_tools(binding.board.clone(), binding.caller, binding.path.clone()) + }) + } +} + +/// Installs message-board lifecycle and tool contributions. The factory selects +/// a local or remote backend, or returns None when disabled. Configuration is +/// read at runtime startup, including resume; no board is created by installation. +pub fn install( + registry: &mut ExtensionRegistryBuilder, + open: impl Fn( + &C, + SessionId, + ThreadId, + ) -> BoxFuture<'static, Result>>> + + Send + + Sync + + 'static, +) { + let extension = Arc::new(BoardExtension { + open: Box::new(open), + events: registry.event_sink(), + }); + registry.thread_lifecycle_contributor(extension.clone()); + registry.tool_contributor(extension); +} diff --git a/codex-rs/ext/agent-message-board/src/lib.rs b/codex-rs/ext/agent-message-board/src/lib.rs index 9886abeeaf..509ee24476 100644 --- a/codex-rs/ext/agent-message-board/src/lib.rs +++ b/codex-rs/ext/agent-message-board/src/lib.rs @@ -4,6 +4,7 @@ //! storage and notification fanout; tools and feature registration are separate. mod api; +mod extension; mod host; mod local; mod tools; @@ -24,6 +25,7 @@ pub use api::SubscriptionRequest; pub use api::SubscriptionTarget; pub use api::ThreadQuery; pub use api::ThreadSort; +pub use extension::install; pub use host::MessageBoardHost; pub use host::NotificationDelivery; pub use local::LocalAgentMessageBoard; diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index ad635f05e9..30318ae47b 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -205,6 +205,8 @@ pub enum Feature { Collab, /// Enable task-path-based multi-agent routing. MultiAgentV2, + /// Enable shared discussion tools for an agent tree. + AgentMessageBoard, /// Removed compatibility flag retained as a no-op. MultiAgentMode, /// Removed compatibility flag for the deleted agent-job tools. @@ -1319,6 +1321,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::Stable, default_enabled: false, }, + FeatureSpec { + id: Feature::AgentMessageBoard, + key: "agent_message_board", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::MultiAgentMode, key: "multi_agent_mode",