diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index f10c7ba8a7..4e9ca35a9a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2086,6 +2086,17 @@ dependencies = [ "sha2 0.10.9", ] +[[package]] +name = "codex-agent-message-board-extension" +version = "0.0.0" +dependencies = [ + "chrono", + "codex-protocol", + "futures", + "serde", + "uuid", +] + [[package]] name = "codex-agent-roles" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 538ee5298f..cf8b23bfe1 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -58,6 +58,7 @@ members = [ "exec-server/tests/support", "execpolicy", "ext/agent", + "ext/agent-message-board", "ext/connectors", "ext/extension-api", "ext/goal", diff --git a/codex-rs/ext/agent-message-board/BUILD.bazel b/codex-rs/ext/agent-message-board/BUILD.bazel new file mode 100644 index 0000000000..c3513d43bf --- /dev/null +++ b/codex-rs/ext/agent-message-board/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "agent-message-board", + crate_name = "codex_agent_message_board_extension", +) diff --git a/codex-rs/ext/agent-message-board/Cargo.toml b/codex-rs/ext/agent-message-board/Cargo.toml new file mode 100644 index 0000000000..077fba027d --- /dev/null +++ b/codex-rs/ext/agent-message-board/Cargo.toml @@ -0,0 +1,21 @@ +[package] +edition.workspace = true +license.workspace = true +name = "codex-agent-message-board-extension" +version.workspace = true + +[lib] +name = "codex_agent_message_board_extension" +path = "src/lib.rs" +doctest = false +test = false + +[lints] +workspace = true + +[dependencies] +chrono = { workspace = true, features = ["serde"] } +codex-protocol = { workspace = true } +futures = { workspace = true } +serde = { workspace = true, features = ["derive"] } +uuid = { workspace = true, features = ["serde", "v7"] } diff --git a/codex-rs/ext/agent-message-board/src/api.rs b/codex-rs/ext/agent-message-board/src/api.rs new file mode 100644 index 0000000000..36e7c2093c --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/api.rs @@ -0,0 +1,200 @@ +//! Operations on one board. Requests describe intent independently of tool schemas. + +use crate::ChannelSummary; +use crate::Page; +use crate::PostContent; +use crate::PostMetadata; +use crate::PostPreview; +use crate::SubscriptionState; +use crate::ThreadPage; +use crate::ThreadSummary; +use codex_protocol::AgentPath; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::error::Result; +use futures::future::BoxFuture; +use serde::Deserialize; +use serde::Serialize; +use std::num::NonZeroU32; +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 persistent agent tree. +/// +/// 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. +/// Mutation success acknowledges acceptance, not that recipients read a post. +/// Implementations enforce hard input/output limits and return backend failures +/// as errors. They own atomic subscription changes, posting and recipient +/// selection. Notification delivery must neither wake finalized agents nor +/// leave notifications for a later turn. +/// +/// Boxed Send futures support an Arc, like AgentControl. +pub trait AgentMessageBoard: Send + Sync { + fn identity(&self) -> SessionId; + + fn create_channel( + &self, + caller: ThreadId, + request: CreateChannelRequest, + ) -> BoxFuture<'_, Result>; + + fn list_channels( + &self, + caller: ThreadId, + query: ChannelQuery, + ) -> BoxFuture<'_, Result>>; + + /// Uses the caller's configured clock; clock failures must not create a post. + /// The request ID identifies a logical call across retries. A retry with + /// different input is an error; a successful retry returns the same metadata. + fn post(&self, caller: ThreadId, request: PostRequest) -> BoxFuture<'_, Result>; + + fn list_threads( + &self, + caller: ThreadId, + query: ThreadQuery, + ) -> BoxFuture<'_, Result>>; + + fn search_posts( + &self, + caller: ThreadId, + query: PostQuery, + ) -> BoxFuture<'_, Result>>; + + fn read_thread( + &self, + caller: ThreadId, + request: ReadThreadRequest, + ) -> BoxFuture<'_, Result>; + + fn read_post( + &self, + caller: ThreadId, + request: ReadPostRequest, + ) -> BoxFuture<'_, Result>; + + /// Channel subscriptions concern new roots; thread subscriptions concern + /// replies. Changing one does not change the other. Any member may change + /// another member's subscription. Posting subscribes its author to the thread. + fn set_subscription( + &self, + caller: ThreadId, + request: SubscriptionRequest, + ) -> BoxFuture<'_, Result>; +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PageRequest { + pub cursor: Option, + pub limit: NonZeroU32, +} + +impl Default for PageRequest { + fn default() -> Self { + Self { + cursor: None, + limit: NonZeroU32::new(20).unwrap_or(NonZeroU32::MIN), + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum SortDirection { + NewestFirst, + OldestFirst, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum SubscriptionChange { + Subscribe, + Unsubscribe, +} + +#[derive(Clone, Debug)] +pub struct CreateChannelRequest { + pub channel_name: String, + pub subscription: SubscriptionChange, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ChannelQuery { + pub query: Option, + pub direction: SortDirection, + pub page: PageRequest, +} + +/// Exactly one destination; discussion threads use their root post's ID. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum PostDestination { + Channel(String), + NewChannel(String), + Thread(Uuid), +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PostRequest { + /// Host-generated tool invocation identity, never a model argument. + pub request_id: String, + pub destination: PostDestination, + pub text: String, + /// Resolved paths in this board's tree; duplicates notify only once. + pub agents_to_notify: Vec, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ThreadSort { + Created, + Activity, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ThreadQuery { + pub channel_name: String, + pub sort: ThreadSort, + pub direction: SortDirection, + pub page: PageRequest, + pub max_chars_per_post: NonZeroU32, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PostQuery { + pub channel_name: Option, + pub query: Option, + pub after_message_id: Option, + pub author: Option, + pub page: PageRequest, + pub max_chars_per_post: NonZeroU32, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReadThreadRequest { + pub thread_id: Uuid, + pub page: PageRequest, + pub max_chars_per_post: NonZeroU32, +} + +#[derive(Clone, Debug)] +pub struct ReadPostRequest { + pub message_id: Uuid, + /// Offsets and lengths count Unicode scalar values, not UTF-8 bytes. + pub offset_chars: u32, + pub limit_chars: NonZeroU32, +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub enum SubscriptionTarget { + Channel(String), + Thread(Uuid), +} + +#[derive(Clone, Debug)] +pub struct SubscriptionRequest { + pub target: SubscriptionTarget, + /// None changes the caller's subscription. + pub target_agent: Option, + pub change: SubscriptionChange, +} diff --git a/codex-rs/ext/agent-message-board/src/host.rs b/codex-rs/ext/agent-message-board/src/host.rs new file mode 100644 index 0000000000..23f3f0b7b9 --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/host.rs @@ -0,0 +1,37 @@ +//! Host capabilities needed by a board without depending on codex-core. + +use crate::PostMetadata; +use chrono::DateTime; +use chrono::Utc; +use codex_protocol::AgentPath; +use codex_protocol::ThreadId; +use codex_protocol::error::Result; +use futures::future::BoxFuture; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum NotificationDelivery { + Accepted, + SkippedInactive, +} + +/// Tree-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 +/// time. Notification admission is atomic with turn completion: inactive agents +/// are skipped, and no notification may start work or survive into a later turn. +/// Backends may implement these capabilities with local handles or remote RPCs. +pub trait MessageBoardHost: Send + Sync { + fn agent_path(&self, caller: ThreadId) -> BoxFuture<'_, Result>; + + fn resolve_agent(&self, path: AgentPath) -> BoxFuture<'_, Result>; + + fn current_time(&self, caller: ThreadId) -> BoxFuture<'_, Result>>; + + /// Only metadata is pushed. Full content is retrieved through bounded reads. + fn notify( + &self, + recipient: ThreadId, + post: PostMetadata, + ) -> BoxFuture<'_, Result>; +} diff --git a/codex-rs/ext/agent-message-board/src/lib.rs b/codex-rs/ext/agent-message-board/src/lib.rs new file mode 100644 index 0000000000..fd50b37455 --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/lib.rs @@ -0,0 +1,34 @@ +//! Shared agent discussions with interchangeable local and remote backends. +//! +//! Board identity and caller identity come from the host. Implementations own +//! storage and notification fanout; tools and feature registration are separate. + +mod api; +mod host; +mod types; + +pub use api::AgentMessageBoard; +pub use api::ChannelQuery; +pub use api::CreateChannelRequest; +pub use api::PageRequest; +pub use api::PostDestination; +pub use api::PostQuery; +pub use api::PostRequest; +pub use api::ReadPostRequest; +pub use api::ReadThreadRequest; +pub use api::SortDirection; +pub use api::SubscriptionChange; +pub use api::SubscriptionRequest; +pub use api::SubscriptionTarget; +pub use api::ThreadQuery; +pub use api::ThreadSort; +pub use host::MessageBoardHost; +pub use host::NotificationDelivery; +pub use types::ChannelSummary; +pub use types::Page; +pub use types::PostContent; +pub use types::PostMetadata; +pub use types::PostPreview; +pub use types::SubscriptionState; +pub use types::ThreadPage; +pub use types::ThreadSummary; diff --git a/codex-rs/ext/agent-message-board/src/types.rs b/codex-rs/ext/agent-message-board/src/types.rs new file mode 100644 index 0000000000..ca5be1157b --- /dev/null +++ b/codex-rs/ext/agent-message-board/src/types.rs @@ -0,0 +1,89 @@ +//! Message-board records shared by backends and tool adapters. + +use chrono::DateTime; +use chrono::Utc; +use codex_protocol::AgentPath; +use serde::Deserialize; +use serde::Serialize; +use serde::Serializer; +use serde::ser::SerializeStruct; +use uuid::Uuid; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PostMetadata { + pub message_id: Uuid, + pub channel_name: String, + pub author: AgentPath, + pub thread_id: Uuid, + pub created_at: DateTime, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PostPreview { + #[serde(flatten)] + pub metadata: PostMetadata, + pub text_preview: String, + pub n_chars: usize, + pub truncated: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PostContent { + #[serde(flatten)] + pub metadata: PostMetadata, + pub text: String, + pub n_chars: usize, + pub next_offset_chars: usize, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ThreadSummary { + pub thread_id: Uuid, + pub root_post: PostPreview, + pub reply_count: usize, + pub last_activity_at: DateTime, + pub latest_reply: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ChannelSummary { + pub channel_name: String, + pub created_at: DateTime, + pub created_by: AgentPath, + pub message_count: usize, + pub last_message_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SubscriptionState { + pub channel_name: String, + pub thread_id: Option, + pub target_agent: AgentPath, + pub enabled: bool, + pub last_message_id: Option, +} + +/// A bounded page. Counts and end-of-list markers are derived at serialization. +#[derive(Clone, Debug, PartialEq, Eq, Deserialize)] +pub struct Page { + pub results: Vec, + pub next_cursor: Option, +} + +impl Serialize for Page { + fn serialize(&self, serializer: S) -> Result { + let mut page = serializer.serialize_struct("Page", 4)?; + page.serialize_field("results", &self.results)?; + page.serialize_field("n_returned", &self.results.len())?; + page.serialize_field("has_more", &self.next_cursor.is_some())?; + page.serialize_field("next_cursor", &self.next_cursor)?; + page.end() + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ThreadPage { + pub root_post: PostPreview, + #[serde(flatten)] + pub replies: Page, +}