mirror of
https://github.com/openai/codex.git
synced 2026-10-19 16:17:49 +00:00
Deliver remote message board notifications to active turns (#49686)
## What changed - Start a turn-owned notification receiver using the same remote board client and session identity as the board tools. Deliver post previews as agent messages and cancel the receiver when the turn stops, aborts, or errors. - Wait for the initial connection attempt before inference, then retry failed or disconnected streams without waking idle agents. - Skip malformed notices, mismatched recipients or turns, and oversized previews. Reject delivery to stale or canceled turns and bound remote notices to a 1 KiB context budget. ## Testing Extend remote board tests to cover interrupted setup, connection failures, reconnection, invalid and stale notices, preview delivery into model context, and receiver shutdown after turn completion. GitOrigin-RevId: c7c8d977fe74b7a5cb3bd8216dc0146f3f297e44
This commit is contained in:
jif
Generated
+3
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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<BoardNotification>>,
|
||||
stream: BoxStream<'static, Result<eventsource_stream::Event>>,
|
||||
}
|
||||
|
||||
impl BoardNotifications {
|
||||
/// Skips invalid individual notices. Transport and framing failures remain errors.
|
||||
pub async fn next(&mut self) -> Result<Option<BoardNotification>> {
|
||||
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::<BoardNotification>(&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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<RemoteAgentMessageBoard>, Arc<dyn MessageBoardHost>);
|
||||
type ResolveBinding = dyn Fn(&TurnStartInput<'_>) -> Option<Binding> + Send + Sync;
|
||||
|
||||
struct Notifications(Box<ResolveBinding>);
|
||||
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::<Receiver>();
|
||||
})
|
||||
}
|
||||
|
||||
fn on_turn_abort<'a>(&'a self, input: TurnAbortInput<'a>) -> ExtensionFuture<'a, ()> {
|
||||
Box::pin(async move {
|
||||
input.turn_store.remove::<Receiver>();
|
||||
})
|
||||
}
|
||||
|
||||
fn on_turn_error<'a>(&'a self, input: TurnErrorInput<'a>) -> ExtensionFuture<'a, ()> {
|
||||
Box::pin(async move {
|
||||
input.turn_store.remove::<Receiver>();
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<C: Sync>(
|
||||
registry: &mut ExtensionRegistryBuilder<C>,
|
||||
resolve: impl Fn(&TurnStartInput<'_>) -> Option<Binding> + Send + Sync + 'static,
|
||||
) {
|
||||
registry.turn_lifecycle_contributor(Arc::new(Notifications(Box::new(resolve))));
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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<Config>,
|
||||
manager: Weak<ThreadManager>,
|
||||
) {
|
||||
codex_agent_message_board_client::install_notifications(registry, |input| {
|
||||
let board = input.thread_store.get::<RemoteAgentMessageBoard>()?;
|
||||
let host = input.thread_store.get::<LocalBoardHost>()?;
|
||||
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<dyn AgentMessageBoard> = 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<ThreadManager>,
|
||||
tree: SessionId,
|
||||
caller: ThreadId,
|
||||
expected_turn_id: Option<String>,
|
||||
}
|
||||
|
||||
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
|
||||
{
|
||||
|
||||
@@ -143,9 +143,16 @@ impl InputQueue {
|
||||
&self,
|
||||
active_turn: &Mutex<Option<ActiveTurn>>,
|
||||
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;
|
||||
|
||||
@@ -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::<Vec<_>>(),
|
||||
&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(())
|
||||
}
|
||||
|
||||
+17
@@ -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"
|
||||
}
|
||||
]
|
||||
+18
-9
@@ -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::<Vec<_>>(),\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::<Vec<_>>(),\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:
|
||||
<PERMISSIONS_INSTRUCTIONS>
|
||||
00:message/user:
|
||||
This turn will be interrupted during notification setup.
|
||||
01:message/developer:
|
||||
<turn_aborted>
|
||||
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]
|
||||
</turn_aborted>
|
||||
02:message/developer:
|
||||
<PERMISSIONS_INSTRUCTIONS>
|
||||
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:
|
||||
<MULTI_AGENT_MODE>
|
||||
03:message/user:
|
||||
05:message/user:
|
||||
<ENVIRONMENT_CONTEXT>
|
||||
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":"<UUID 1>","thread_id":"<UUID 1>"}
|
||||
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":"<UUID 1>","thread_id":"<UUID 1>"}
|
||||
10:function_call_output:{"message":"Wait completed.","timed_out":false}
|
||||
11:agent_message
|
||||
|
||||
@@ -25,7 +25,11 @@ 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>>>>
|
||||
type BoardFactory<C> = dyn for<'a> Fn(
|
||||
&'a ThreadStartInput<'_, C>,
|
||||
SessionId,
|
||||
ThreadId,
|
||||
) -> BoxFuture<'a, Result<Option<Arc<dyn AgentMessageBoard>>>>
|
||||
+ Send
|
||||
+ Sync;
|
||||
type NamespaceResolver<C> = dyn Fn(&C) -> Option<String> + Send + Sync;
|
||||
@@ -55,7 +59,7 @@ impl<C: Sync> ThreadLifecycleContributor<C> for BoardExtension<C> {
|
||||
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<C> BoardExtension<C> {
|
||||
/// 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<C: Sync + 'static>(
|
||||
registry: &mut ExtensionRegistryBuilder<C>,
|
||||
namespace_description: &'static str,
|
||||
tool_namespace: impl Fn(&C) -> Option<String> + Send + Sync + 'static,
|
||||
open: impl Fn(
|
||||
&C,
|
||||
open: impl for<'a> Fn(
|
||||
&'a ThreadStartInput<'_, C>,
|
||||
SessionId,
|
||||
ThreadId,
|
||||
) -> BoxFuture<'static, Result<Option<Arc<dyn AgentMessageBoard>>>>
|
||||
) -> BoxFuture<'a, Result<Option<Arc<dyn AgentMessageBoard>>>>
|
||||
+ Send
|
||||
+ Sync
|
||||
+ 'static,
|
||||
|
||||
Reference in New Issue
Block a user