diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index a8e086b492..f799dfda14 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1945,6 +1945,7 @@ dependencies = [ "anyhow", "chrono", "codex-agent-message-board-extension", + "codex-extension-api", "codex-http-client", "codex-protocol", "eventsource-stream", @@ -1954,6 +1955,8 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-util", + "tracing", "url", "wiremock", ] diff --git a/codex-rs/agent-message-board-client/Cargo.toml b/codex-rs/agent-message-board-client/Cargo.toml index b8b77a9245..52a9c9c272 100644 --- a/codex-rs/agent-message-board-client/Cargo.toml +++ b/codex-rs/agent-message-board-client/Cargo.toml @@ -15,6 +15,7 @@ workspace = true anyhow = { workspace = true } chrono = { workspace = true, features = ["serde"] } codex-agent-message-board-extension = { workspace = true } +codex-extension-api = { workspace = true } codex-http-client = { workspace = true } codex-protocol = { workspace = true } eventsource-stream = { workspace = true } @@ -22,7 +23,9 @@ futures = { workspace = true } http = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } -tokio = { workspace = true, features = ["time"] } +tokio = { workspace = true, features = ["rt", "time"] } +tokio-util = { workspace = true } +tracing = { workspace = true } url = { workspace = true } [dev-dependencies] diff --git a/codex-rs/agent-message-board-client/src/client.rs b/codex-rs/agent-message-board-client/src/client.rs index 761a166a9e..052c634ba3 100644 --- a/codex-rs/agent-message-board-client/src/client.rs +++ b/codex-rs/agent-message-board-client/src/client.rs @@ -169,6 +169,11 @@ impl RemoteAgentMessageBoard { .map_err(transport_error)? .map_err(transport_error)?; if !response.status().is_success() { + tracing::warn!( + %caller, %turn_id, + http_status = response.status().as_u16(), + "Remote board notification request rejected" + ); tokio::time::timeout_at(deadline, decode::<()>(response)) .await .map_err(transport_error)??; @@ -245,15 +250,7 @@ impl RemoteAgentMessageBoard { "notification stream did not acknowledge readiness", )); } - let stream = events - .map(|event| { - let event = event.map_err(transport_error)?; - if event.event != "notification" { - return Err(transport_error("invalid board notification")); - } - serde_json::from_str(&event.data).map_err(CodexErr::from) - }) - .boxed(); + let stream = events.map(|event| event.map_err(transport_error)).boxed(); Ok(BoardNotifications { caller, turn_id, @@ -347,25 +344,35 @@ impl AgentMessageBoard for RemoteAgentMessageBoard { pub struct BoardNotifications { caller: ThreadId, turn_id: String, - stream: BoxStream<'static, Result>, + stream: BoxStream<'static, Result>, } impl BoardNotifications { + /// Skips invalid individual notices. Transport and framing failures remain errors. pub async fn next(&mut self) -> Result> { - let Some(notice) = self.stream.next().await.transpose()? else { - return Ok(None); - }; - if notice.recipient != self.caller || notice.turn_id != self.turn_id { - return Err(transport_error( - "notification belongs to another agent or turn", - )); + while let Some(event) = self.stream.next().await.transpose()? { + let error_kind = if event.event != "notification" { + "unexpected_event" + } else { + match serde_json::from_str::(&event.data) { + Ok(notice) => { + if notice.recipient != self.caller || notice.turn_id != self.turn_id { + "wrong_recipient_or_turn" + } else if notice.post.text_preview.chars().count() > 150 { + "oversized_preview" + } else { + return Ok(Some(notice)); + } + } + Err(_) => "invalid_json", + } + }; + tracing::warn!( + caller = %self.caller, turn_id = %self.turn_id, error_kind, + "Skipping invalid remote board notification" + ); } - if notice.post.text_preview.chars().count() > 150 { - return Err(transport_error( - "board notification preview exceeds 150 characters", - )); - } - Ok(Some(notice)) + Ok(None) } } diff --git a/codex-rs/agent-message-board-client/src/lib.rs b/codex-rs/agent-message-board-client/src/lib.rs index 1e181da375..c6d0c5487a 100644 --- a/codex-rs/agent-message-board-client/src/lib.rs +++ b/codex-rs/agent-message-board-client/src/lib.rs @@ -2,9 +2,11 @@ //! Runtime credentials are scoped to one session; tools never see them. mod client; +mod notifications; mod protocol; pub use client::BoardNotifications; pub use client::RemoteAgentMessageBoard; +pub use notifications::install_notifications; pub use protocol::AccessToken; pub use protocol::BoardNotification; diff --git a/codex-rs/agent-message-board-client/src/notifications.rs b/codex-rs/agent-message-board-client/src/notifications.rs new file mode 100644 index 0000000000..0f80aa3d4f --- /dev/null +++ b/codex-rs/agent-message-board-client/src/notifications.rs @@ -0,0 +1,122 @@ +//! Receives live board previews for an active turn using the extension's turn hooks. +//! Provisioning and durable board ownership belong to the research host. + +use crate::RemoteAgentMessageBoard; +use codex_agent_message_board_extension::MessageBoardHost; +use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionFuture; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::TurnAbortInput; +use codex_extension_api::TurnErrorInput; +use codex_extension_api::TurnLifecycleContributor; +use codex_extension_api::TurnStartInput; +use codex_extension_api::TurnStartPhase; +use codex_extension_api::TurnStopInput; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErrKind; +use futures::channel::oneshot; +use std::sync::Arc; +use std::time::Duration; +use tokio_util::task::AbortOnDropHandle; + +type Binding = (Arc, Arc); +type ResolveBinding = dyn Fn(&TurnStartInput<'_>) -> Option + Send + Sync; + +struct Notifications(Box); +struct Receiver { + _task: AbortOnDropHandle<()>, +} + +impl TurnLifecycleContributor for Notifications { + fn turn_start_phase(&self, _thread_store: &ExtensionData) -> TurnStartPhase { + TurnStartPhase::RegularTaskStart + } + + fn on_turn_start<'a>(&'a self, input: TurnStartInput<'a>) -> ExtensionFuture<'a, ()> { + Box::pin(async move { + let Some((board, host)) = (self.0)(&input) else { + return; + }; + let Ok(caller) = ThreadId::from_string(input.thread_store.level_id()) else { + return; + }; + let turn_id = input.turn_id.to_owned(); + let (first_attempt_tx, first_attempt_rx) = oneshot::channel(); + input.turn_store.insert(Receiver { + _task: AbortOnDropHandle::new(tokio::spawn(async move { + let mut first_attempt_tx = Some(first_attempt_tx); + loop { + let connection = board.notifications(caller, turn_id.clone()).await; + if let Some(first_attempt_tx) = first_attempt_tx.take() { + let _ = first_attempt_tx.send(()); + } + match connection { + Ok(mut stream) => loop { + match stream.next().await { + Ok(Some(notice)) => { + if let Err(error) = host.notify(caller, notice.post).await { + tracing::warn!( + %caller, %turn_id, + error_kind = ?CodexErrKind::from(&error), + "Remote board notification delivery failed" + ); + } + } + Ok(None) => break, + Err(error) => { + tracing::warn!( + %caller, %turn_id, + error_kind = ?CodexErrKind::from(&error), + "Remote board notification stream failed" + ); + break; + } + } + }, + Err(error) => { + tracing::warn!( + %caller, %turn_id, + error_kind = ?CodexErrKind::from(&error), + "Remote board notification connection failed" + ); + } + } + // Reopen for future previews; the service does not replay missed ones. + tokio::time::sleep(Duration::from_secs(/*secs*/ 1)).await; + } + })), + }); + // A healthy service is listening before inference. A failed first attempt + // releases startup while the same turn-owned receiver keeps retrying. + let _ = first_attempt_rx.await; + }) + } + + fn on_turn_stop<'a>(&'a self, input: TurnStopInput<'a>) -> ExtensionFuture<'a, ()> { + Box::pin(async move { + input.turn_store.remove::(); + }) + } + + fn on_turn_abort<'a>(&'a self, input: TurnAbortInput<'a>) -> ExtensionFuture<'a, ()> { + Box::pin(async move { + input.turn_store.remove::(); + }) + } + + fn on_turn_error<'a>(&'a self, input: TurnErrorInput<'a>) -> ExtensionFuture<'a, ()> { + Box::pin(async move { + input.turn_store.remove::(); + }) + } +} + +/// Installs live notification delivery. The host resolves the existing remote +/// client and a delivery host bound to this exact turn; idle agents are never woken. +/// The turn store owns the receiver and cancels it when the turn ends. +pub fn install_notifications( + registry: &mut ExtensionRegistryBuilder, + resolve: impl Fn(&TurnStartInput<'_>) -> Option + Send + Sync + 'static, +) { + registry.turn_lifecycle_contributor(Arc::new(Notifications(Box::new(resolve)))); +} diff --git a/codex-rs/agent-message-board-client/tests/remote_board.rs b/codex-rs/agent-message-board-client/tests/remote_board.rs index 9a57d0efe1..21cc7e2fe4 100644 --- a/codex-rs/agent-message-board-client/tests/remote_board.rs +++ b/codex-rs/agent-message-board-client/tests/remote_board.rs @@ -200,8 +200,7 @@ async fn remote_board_preserves_the_host_contract() -> anyhow::Result<()> { .await; let mut receiver = client.notifications(caller, "turn-1".into()).await?; assert_eq!(receiver.next().await?, Some(notice)); - assert!(receiver.next().await.is_err()); - assert!(receiver.next().await.is_err()); + assert_eq!(receiver.next().await?, None); for body in [ format!( "event: ready\nid: {}\ndata: {{}}\n\n", diff --git a/codex-rs/core/src/agent_message_board.rs b/codex-rs/core/src/agent_message_board.rs index 56ab5db27b..62e60c92b6 100644 --- a/codex-rs/core/src/agent_message_board.rs +++ b/codex-rs/core/src/agent_message_board.rs @@ -20,6 +20,7 @@ use codex_agent_message_board_extension::MessageBoardHost; use codex_agent_message_board_extension::NotificationDelivery; use codex_agent_message_board_extension::PostPreview; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ThreadStartInput; use codex_features::Feature; use codex_http_client::ClientRouteClass; use codex_protocol::AgentPath; @@ -39,12 +40,24 @@ pub fn install_agent_message_board( registry: &mut ExtensionRegistryBuilder, manager: Weak, ) { + codex_agent_message_board_client::install_notifications(registry, |input| { + let board = input.thread_store.get::()?; + let host = input.thread_store.get::()?; + Some(( + board, + Arc::new(LocalBoardHost { + expected_turn_id: Some(input.turn_id.into()), + ..host.as_ref().clone() + }), + )) + }); let in_memory_boards = Arc::new(InMemoryMessageBoards::default()); codex_agent_message_board_extension::install( registry, MULTI_AGENT_V2_NAMESPACE_DESCRIPTION, |config: &Config| config.multi_agent_v2.tool_namespace.clone(), - move |config: &Config, tree, caller| { + move |input: &ThreadStartInput<'_, Config>, tree, caller| { + let config = input.config; 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) @@ -63,6 +76,7 @@ pub fn install_agent_message_board( manager: manager.clone(), tree, caller, + expected_turn_id: None, }); Box::pin(async move { let board: Arc = if let Some(remote) = remote { @@ -79,13 +93,14 @@ pub fn install_agent_message_board( let http = http_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()); let board = RemoteAgentMessageBoard::new(http, &remote.url, tree, token) .map_err(|err| CodexErr::InvalidRequest(err.to_string()))? .with_clock(move |caller| { let host = host.clone(); Box::pin(async move { host.current_time(caller).await }) }); - Arc::new(board) + input.thread_store.get_or_init(|| board) } else if in_memory { Arc::new(in_memory_boards.open(tree, host).await) } else { @@ -97,10 +112,12 @@ pub fn install_agent_message_board( ); } +#[derive(Clone)] struct LocalBoardHost { manager: Weak, tree: SessionId, caller: ThreadId, + expected_turn_id: Option, } impl LocalBoardHost { @@ -214,6 +231,18 @@ impl MessageBoardHost for LocalBoardHost { notice.render(), /*trigger_turn*/ false, ); + // Remote metadata is untrusted. Skip an oversized notice without closing + // the receiver; the post remains available through the board tools. + if self.expected_turn_id.is_some() + && communication.content.len() + + communication.author.as_str().len() + + communication.recipient.as_str().len() + > 1024 + { + return Err(CodexErr::InvalidRequest( + "remote board notice exceeds the context budget".into(), + )); + } Ok( if recipient .session @@ -221,6 +250,7 @@ impl MessageBoardHost for LocalBoardHost { .deliver_mailbox_communication_to_current_turn( &recipient.session.active_turn, communication, + self.expected_turn_id.as_deref(), ) .await { diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 2212dc09f6..c73b75f8eb 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -143,9 +143,16 @@ impl InputQueue { &self, active_turn: &Mutex>, communication: InterAgentCommunication, + expected_turn_id: Option<&str>, ) -> bool { let active = active_turn.lock().await; - let Some(active_turn) = active.as_ref().filter(|turn| turn.task.is_some()) else { + let Some(active_turn) = active.as_ref().filter(|turn| { + turn.task.as_ref().is_some_and(|task| { + expected_turn_id.is_none_or(|id| { + task.turn_context.sub_id == id && !task.cancellation_token.is_cancelled() + }) + }) + }) else { return false; }; let mut turn_state = active_turn.turn_state.lock().await; diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs index cf13cf8404..2a9873c1fd 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board_remote.rs @@ -3,18 +3,26 @@ use super::BoardClock; use super::configure; use super::done; -use super::tool; use anyhow::Context; +use codex_core::TurnInputRequest; use codex_features::RemoteMessageBoardConfigToml; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; use core_test_support::context_snapshot::SnapshotEntry; use core_test_support::responses; use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use std::sync::Arc; +use std::sync::OnceLock; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::time::Duration; +use tokio::sync::Notify; use wiremock::Mock; use wiremock::ResponseTemplate; use wiremock::matchers::body_partial_json; @@ -42,6 +50,9 @@ async fn remote_board_uses_the_existing_tools_and_session_identity() -> anyhow:: bearer_token_env_var: None, }); }) + .with_model_info_override("gpt-5.5", |model| { + model.multi_agent_version = Some(codex_protocol::protocol::MultiAgentVersion::V2); + }) .with_external_time_provider(Arc::new(BoardClock::Available)) .build_with_auto_env(&server) .await?; @@ -67,28 +78,105 @@ async fn remote_board_uses_the_existing_tools_and_session_identity() -> anyhow:: .expect(1) .mount(&board) .await; + let connection = AtomicUsize::default(); + let interrupted_turn = OnceLock::new(); + let connecting = Arc::new(Notify::new()); + let connected = connecting.clone(); + let (release_handshake, receive_release) = std::sync::mpsc::channel(); + let receive_release = std::sync::Mutex::new(receive_release); + Mock::given(method("POST")) + .and(path(format!("/v1/boards/{root_id}/notifications"))) + .and(header("authorization", format!("Bearer {token}"))) + .and(body_partial_json(json!({"caller": root_id}))) + .respond_with(move |request: &wiremock::Request| { + let attempt = connection.fetch_add(/*val*/ 1, Ordering::Relaxed); + // Interrupt the first handshake. On the next turn, fail setup, + // disconnect, fail a reconnect, then deliver on the same receiver. + match attempt { + 0 | 4 => {} + 2 => { + return ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string("event: ready\ndata: {}\n\n"); + } + _ => return ResponseTemplate::new(503), + } + let watch: Value = request.body_json().expect("notification watch"); + let notice = json!({ + "recipient": root_id, "turn_id": watch["turn_id"], + "post": { + "message_id": "00000000-0000-4000-8000-000000000002", + "thread_id": "00000000-0000-4000-8000-000000000002", + "author": "/root/worker", "channel_name": "design", + "created_at": "2026-09-18T12:00:00Z", + "text_preview": "Worker's remote decision.", "n_chars": 25, "truncated": false + } + }); + let mut stale = notice.clone(); + stale["turn_id"] = interrupted_turn.get_or_init(|| watch["turn_id"].clone()).clone(); + let response = ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(format!( + "event: ready\ndata: {{}}\n\nevent: notification\ndata: invalid-json\n\nevent: notification\ndata: {stale}\n\nevent: notification\ndata: {notice}\n\n" + )); + if attempt == 0 { + connected.notify_one(); + // Wiremock runs on its own thread; keep the handshake pending + // until the test confirms interruption, regardless of runner speed. + let _ = receive_release.lock().expect("handshake gate").recv(); + } + response + }) + .expect(5..) + .mount(&board) + .await; let model = responses::mount_sse_sequence( &server, vec![ - tool( - "remote-post", - "post", - json!({"new_channel_name":"design", "text":"A remote decision."}), - ), + // Wait in the same model step so pending mail cannot be drained before the wait. + responses::sse(vec![ + responses::ev_function_call_with_namespace( + "remote-post", + "collaboration", + "post", + &json!({"new_channel_name":"design", "text":"A remote decision."}).to_string(), + ), + responses::ev_function_call_with_namespace( + "await-notification", + "collaboration", + "wait_agent", + "{}", + ), + responses::ev_completed("remote-post"), + ]), done(), ], ) .await; + root.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![super::super::text( + "This turn will be interrupted during notification setup.", + )])) + .await?; + connecting.notified().await; + root.codex.submit(Op::Interrupt).await?; + wait_for_event(&root.codex, |event| { + matches!(event, EventMsg::TurnAborted(_)) + }) + .await; + release_handshake.send(())?; root.submit_turn("Post the remote decision.").await?; let requests = model.requests(); - let output: Value = serde_json::from_str( - &requests - .last() - .context("model request")? - .function_call_output_text("remote-post") - .context("tool output")?, - )?; - assert_eq!(output, post); + let last_request = requests.last().context("model request")?; + let notices = last_request.inputs_of_type("agent_message"); + let mut notification = responses::strip_metadata_from_json( + responses::strip_response_item_ids_from_json(json!(notices)), + ); + notification.sort_all_objects(); + insta::assert_snapshot!( + "remote_board_notification", + serde_json::to_string_pretty(¬ification)? + ); // Normalize tool JSON key order across Cargo and Bazel feature sets. let mut bodies = requests .iter() @@ -107,19 +195,26 @@ async fn remote_board_uses_the_existing_tools_and_session_identity() -> anyhow:: insta::assert_snapshot!( "remote_board_tools", context_snapshot::format_context_snapshot( - "Existing board tools use a provisioned remote board.", + "Remote board tools and active-turn notifications share the existing session identity.", &bodies.iter().map(SnapshotEntry::body).collect::>(), &ContextSnapshotOptions::default().rewrite_known_segments(), ) ); - root.codex.shutdown_and_wait().await?; + let request_count = board + .received_requests() + .await + .context("board requests")? + .len(); + // A completed turn must stop the receiver, including its next scheduled retry. + tokio::time::sleep(Duration::from_millis(/*millis*/ 1_100)).await; assert_eq!( board .received_requests() .await .context("board requests")? .len(), - 1 + request_count ); + root.codex.shutdown_and_wait().await?; Ok(()) } diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_notification.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_notification.snap new file mode 100644 index 0000000000..37ebc834d1 --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_notification.snap @@ -0,0 +1,17 @@ +--- +source: core/tests/suite/scenarios_agent_message_board_remote.rs +expression: "serde_json::to_string_pretty(¬ification)?" +--- +[ + { + "author": "/root/worker", + "content": [ + { + "text": "Message Type: CHANNEL_POST\nSender: /root/worker\nChannel: design\nMessage ID: 00000000-0000-4000-8000-000000000002\nThread ID: 00000000-0000-4000-8000-000000000002\nPayload:\nWorker's remote decision.", + "type": "input_text" + } + ], + "recipient": "/root", + "type": "agent_message" + } +] diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap index d62d153cf1..9fb63c8087 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__remote__remote_board_tools.snap @@ -1,14 +1,20 @@ --- source: core/tests/suite/scenarios_agent_message_board_remote.rs -expression: "context_snapshot::format_context_snapshot(\"Existing board tools use a provisioned remote board.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(),\n&ContextSnapshotOptions::default().rewrite_known_segments(),)" +expression: "context_snapshot::format_context_snapshot(\"Remote board tools and active-turn notifications share the existing session identity.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(),\n&ContextSnapshotOptions::default().rewrite_known_segments(),)" --- -Scenario: Existing board tools use a provisioned remote board. +Scenario: Remote board tools and active-turn notifications share the existing session identity. ## Window 1 -- request 1 (request) -- -00:message/developer: - +00:message/user: + This turn will be interrupted during notification setup. 01:message/developer: + + The previous turn was interrupted on purpose. Any running unified exec processes may still be running in...mands were aborted, they may have partially executed. [hash=cee256d92b55ad65] + +02:message/developer: + +03: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. @@ -26,12 +32,15 @@ Scenario: Existing board tools use a provisioned remote board. 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: +04:message/developer: -03:message/user: +05:message/user: -04:message/user: +06:message/user: Post the remote decision. -- request 2 (request) -- -05:function_call/collaboration.post:{"new_channel_name":"design","text":"A remote decision."} -06:function_call_output:{"author":"/root","channel_name":"design","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} +07:function_call/collaboration.post:{"new_channel_name":"design","text":"A remote decision."} +08:function_call/collaboration.wait_agent:{} +09:function_call_output:{"author":"/root","channel_name":"design","created_at":"2026-09-18T12:00:00Z","message_id":"","thread_id":""} +10:function_call_output:{"message":"Wait completed.","timed_out":false} +11:agent_message diff --git a/codex-rs/ext/agent-message-board/src/extension.rs b/codex-rs/ext/agent-message-board/src/extension.rs index 4c3b7f0cde..3d39c3d7d9 100644 --- a/codex-rs/ext/agent-message-board/src/extension.rs +++ b/codex-rs/ext/agent-message-board/src/extension.rs @@ -25,7 +25,11 @@ use codex_tools::ToolExecutor; use futures::future::BoxFuture; use std::sync::Arc; -type BoardFactory = dyn Fn(&C, SessionId, ThreadId) -> BoxFuture<'static, Result>>> +type BoardFactory = dyn for<'a> Fn( + &'a ThreadStartInput<'_, C>, + SessionId, + ThreadId, + ) -> BoxFuture<'a, Result>>> + Send + Sync; type NamespaceResolver = dyn Fn(&C) -> Option + Send + Sync; @@ -55,7 +59,7 @@ impl ThreadLifecycleContributor for BoardExtension { 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 let Some(board) = (self.open)(&input, tree, caller).await? { if board.identity() != tree { return Err(CodexErr::InvalidRequest( "message-board factory returned another tree".into(), @@ -129,16 +133,17 @@ impl BoardExtension { /// a local or remote backend, or returns None when disabled. Configuration is /// read at runtime startup, including resume; no board is created by installation. /// The host supplies its shared namespace description and resolves the namespace -/// name from the runtime's startup configuration. +/// name from the runtime's startup configuration. The factory may retain its +/// backend in the existing thread store for other extension contributions. pub fn install( registry: &mut ExtensionRegistryBuilder, namespace_description: &'static str, tool_namespace: impl Fn(&C) -> Option + Send + Sync + 'static, - open: impl Fn( - &C, + open: impl for<'a> Fn( + &'a ThreadStartInput<'_, C>, SessionId, ThreadId, - ) -> BoxFuture<'static, Result>>> + ) -> BoxFuture<'a, Result>>> + Send + Sync + 'static,