mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Wire agent message boards into persistent multi-agent runtimes (#47017)
## What changed - Register the existing message-board tools with the app server through a lifecycle extension, gated on `agent_message_board`, multi-agent v2, and non-ephemeral sessions. The new feature flag is disabled by default. - Bind tools to each agent's identity and share the SQLite-backed board across the agent tree and runtime resumes. - Use the runtime clock for posts and deliver ID-only notifications to running agents, with tools to retrieve post contents. Skip inactive recipients without starting them or queuing notices. - Emit a warning and leave board tools unavailable if initialization fails. ## Testing Add runtime scenarios and a context snapshot covering feature gating without creating storage, posting and searching, active-only notifications, sharing with child agents, persistence across resume, and clock failure without channel creation. GitOrigin-RevId: a81bd85cd372b0d122ba375ed1a1bfd3ba3a3b7c
This commit is contained in:
Generated
+2
@@ -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",
|
||||
|
||||
@@ -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" }
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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"
|
||||
},
|
||||
|
||||
@@ -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<Config>,
|
||||
manager: Weak<ThreadManager>,
|
||||
) {
|
||||
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<dyn AgentMessageBoard>))
|
||||
})
|
||||
});
|
||||
}
|
||||
|
||||
struct LocalBoardHost {
|
||||
manager: Weak<ThreadManager>,
|
||||
tree: SessionId,
|
||||
caller: ThreadId,
|
||||
}
|
||||
|
||||
impl LocalBoardHost {
|
||||
async fn actor(&self) -> Result<Arc<CodexThread>> {
|
||||
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<AgentPath>> {
|
||||
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<ThreadId>> {
|
||||
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<DateTime<Utc>>> {
|
||||
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<NotificationDelivery>> {
|
||||
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,
|
||||
},
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
(
|
||||
"<agent_message_board_notification>",
|
||||
"</agent_message_board_notification>",
|
||||
)
|
||||
}
|
||||
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
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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("<agent_message_board_notification>"))
|
||||
.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::<Vec<_>>();
|
||||
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::<Vec<_>>(),
|
||||
&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("<agent_message_board_notification>"))
|
||||
);
|
||||
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(())
|
||||
}
|
||||
+45
@@ -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::<Vec<_>>(), &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:
|
||||
<PERMISSIONS_INSTRUCTIONS>
|
||||
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.
|
||||
<OMITTED 16 LINES; ~234 TOKENS; hash=b2e94283a8c22327>
|
||||
- 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:
|
||||
<MULTI_AGENT_MODE>
|
||||
03:message/user:
|
||||
<ENVIRONMENT_CONTEXT>
|
||||
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":"<UUID 1>","thread_id":"<UUID 1>"}
|
||||
07:message/user:
|
||||
<agent_message_board_notification>
|
||||
New post <UUID 1> in discussion <UUID 1>. Use collaboration.read_post or collaboration.read_thread to read it.
|
||||
</agent_message_board_notification>
|
||||
-- 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":"<UUID 1>","truncated":false}]} [hash=facdb1e4f66cc5b9]
|
||||
@@ -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 }
|
||||
|
||||
@@ -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<C> = dyn Fn(&C, SessionId, ThreadId) -> BoxFuture<'static, Result<Option<Arc<dyn AgentMessageBoard>>>>
|
||||
+ Send
|
||||
+ Sync;
|
||||
|
||||
struct BoardExtension<C> {
|
||||
open: Box<BoardFactory<C>>,
|
||||
events: Arc<dyn ExtensionEventSink>,
|
||||
}
|
||||
struct Binding {
|
||||
board: Arc<dyn AgentMessageBoard>,
|
||||
caller: ThreadId,
|
||||
path: AgentPath,
|
||||
}
|
||||
|
||||
impl<C: Sync> ThreadLifecycleContributor<C> for BoardExtension<C> {
|
||||
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<C: Sync> ToolContributor for BoardExtension<C> {
|
||||
fn tools(
|
||||
&self,
|
||||
_session_store: &ExtensionData,
|
||||
thread_store: &ExtensionData,
|
||||
) -> Vec<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> {
|
||||
thread_store
|
||||
.get::<Binding>()
|
||||
.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<C: Sync + 'static>(
|
||||
registry: &mut ExtensionRegistryBuilder<C>,
|
||||
open: impl Fn(
|
||||
&C,
|
||||
SessionId,
|
||||
ThreadId,
|
||||
) -> BoxFuture<'static, Result<Option<Arc<dyn AgentMessageBoard>>>>
|
||||
+ 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);
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user