openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-10 20:26:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

code-mode: expose transport-neutral session runtime

Browse Source
This commit is contained in:
Channing Conger
2026-06-20 20:03:17 +00:00
parent 098e9f0a1e
commit 1b0c2f7ea6
14 changed files with 836 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
codex-rs/Cargo.lock generated
View File

@@ -2499,9 +2499,11 @@ dependencies = [
"deno_core_icudata",
"pretty_assertions",
"serde_json",
"sha2 0.10.9",
"tokio",
"tokio-util",
"tracing",
"uuid",
"v8",
]

3
codex-rs/code-mode-protocol/src/session.rs
View File

@@ -63,9 +63,6 @@ pub trait CodeModeSessionDelegate: Send + Sync {
text: String,
cancellation_token: CancellationToken,
) -> NotificationFuture<'a>;
/// Releases delegate state associated with a cell after it reaches a terminal state.
fn cell_closed(&self, cell_id: &CellId);
}
/// A stateful code-mode session owned by one Codex thread.

2
codex-rs/code-mode/Cargo.toml
View File

@@ -20,9 +20,11 @@ codex-code-mode-protocol = { workspace = true }
codex-protocol = { workspace = true }
deno_core_icudata = { workspace = true }
serde_json = { workspace = true }
sha2 = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] }
tokio-util = { workspace = true, features = ["rt"] }
tracing = { workspace = true }
uuid = { workspace = true, features = ["v4"] }
v8 = { workspace = true }
[dev-dependencies]

22
codex-rs/code-mode/src/cell_actor/callbacks.rs
View File

@@ -1,4 +1,5 @@
use std::sync::Arc;
use std::time::Duration;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
@@ -8,6 +9,8 @@ use super::CellHost;
use super::CellToolCall;
use crate::runtime::RuntimeCommand;
const CALLBACK_CANCELLATION_GRACE: Duration = Duration::from_millis(100);
#[derive(Clone, Copy)]
pub(super) enum CallbackCompletion {
DrainNotifications,
@@ -53,10 +56,13 @@ pub(super) async fn finish_callbacks(
) {
if matches!(completion, CallbackCompletion::Cancel) {
cancellation_token.cancel();
finish_cancelled_tasks(notification_tasks, "notification").await;
finish_cancelled_tasks(tool_tasks, "tool").await;
return;
}
drain_tasks(notification_tasks, "notification").await;
cancellation_token.cancel();
drain_tasks(tool_tasks, "tool").await;
finish_cancelled_tasks(tool_tasks, "tool").await;
}
pub(super) fn log_task_result(
@@ -75,3 +81,17 @@ async fn drain_tasks(tasks: &mut JoinSet<()>, description: &str) {
log_task_result(Some(result), description);
}
}
async fn abort_tasks(tasks: &mut JoinSet<()>, description: &str) {
tasks.abort_all();
drain_tasks(tasks, description).await;
}
async fn finish_cancelled_tasks(tasks: &mut JoinSet<()>, description: &str) {
if tokio::time::timeout(CALLBACK_CANCELLATION_GRACE, drain_tasks(tasks, description))
.await
.is_err()
{
abort_tasks(tasks, description).await;
}
}

4
codex-rs/code-mode/src/cell_actor/types.rs
View File

@@ -20,7 +20,6 @@ use crate::session_runtime::ToolName;
pub(crate) type CellEventFuture =
Pin<Box<dyn Future<Output = Result<CellEvent, CellError>> + Send + 'static>>;
#[allow(dead_code)]
pub(crate) type ResumeFuture =
Pin<Box<dyn Future<Output = Result<ResumeOutcome, CellError>> + Send + 'static>>;
@@ -122,7 +121,6 @@ impl CellHandle {
response_event(response_rx)
}
#[allow(dead_code)]
pub(crate) fn resume(&self, generation: PendingGeneration) -> ResumeFuture {
if !self.state.accepting_observations() {
return closed_resume();
@@ -505,7 +503,6 @@ pub(super) enum CellCommand {
mode: ObserveMode,
response_tx: oneshot::Sender<Result<CellEvent, CellError>>,
},
#[allow(dead_code)]
Resume {
generation: PendingGeneration,
response_tx: oneshot::Sender<Result<ResumeOutcome, CellError>>,
@@ -524,7 +521,6 @@ fn closed_event() -> CellEventFuture {
Box::pin(async { Err(CellError::Closed) })
}
#[allow(dead_code)]
fn closed_resume() -> ResumeFuture {
Box::pin(async { Err(CellError::Closed) })
}

3
codex-rs/code-mode/src/lib.rs
View File

@@ -1,9 +1,10 @@
mod cell_actor;
mod runtime;
mod service;
mod session_runtime;
pub mod session_runtime;
pub use codex_code_mode_protocol::*;
pub use service::CodeModeService;
pub use service::InProcessCodeModeSessionProvider;
pub use service::NoopCodeModeSessionDelegate;
pub use session_runtime::SessionRuntime;

16
codex-rs/code-mode/src/service.rs
View File

@@ -74,8 +74,6 @@ impl CodeModeSessionDelegate for NoopCodeModeSessionDelegate {
) -> NotificationFuture<'a> {
Box::pin(async { Ok(()) })
}
fn cell_closed(&self, _cell_id: &CellId) {}
}
#[derive(Default)]
@@ -112,8 +110,14 @@ impl CodeModeService {
}
pub fn with_delegate(delegate: Arc<dyn CodeModeSessionDelegate>) -> Self {
#[cfg(not(test))]
let runtime = Arc::new(SessionRuntime::new(Arc::new(ProtocolDelegate { delegate })));
#[cfg(test)]
let runtime = Arc::new(SessionRuntime::new_for_test(Arc::new(ProtocolDelegate {
delegate,
})));
Self {
runtime: Arc::new(SessionRuntime::new(Arc::new(ProtocolDelegate { delegate }))),
runtime,
observations: Mutex::new(HashMap::new()),
#[cfg(test)]
pending_generations: Mutex::new(HashMap::new()),
@@ -179,7 +183,7 @@ impl CodeModeService {
.ok_or_else(|| "observation ended before producing a result".to_string())?
}
#[allow(dead_code)]
#[cfg(test)]
async fn begin_observe(
&self,
request: ObserveRequest,
@@ -398,10 +402,6 @@ impl runtime::SessionRuntimeDelegate for ProtocolDelegate {
)
.await
}
fn cell_closed(&self, cell_id: &runtime::CellId) {
self.delegate.cell_closed(&protocol_cell_id(cell_id));
}
}
fn runtime_request(request: CreateCellRequest) -> runtime::CreateCellRequest {

202
codex-rs/code-mode/src/service_contract_tests.rs
View File

@@ -6,6 +6,7 @@ use std::time::Duration;
use codex_protocol::ToolName;
use pretty_assertions::assert_eq;
use tokio::sync::Notify;
use tokio::sync::Semaphore;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
@@ -17,9 +18,9 @@ use crate::ToolDefinition;
enum DelegateEvent {
NotificationStarted,
NotificationCancelled,
NotificationFinished,
ToolStarted,
ToolCancelled,
CellClosed(CellId),
}
struct BlockingDelegate {
@@ -34,6 +35,11 @@ struct HeldNotificationDelegate {
notification_release: Notify,
}
struct ReleasableNotificationDelegate {
events_tx: mpsc::UnboundedSender<DelegateEvent>,
notification_release: Semaphore,
}
impl HeldNotificationDelegate {
fn new() -> (Arc<Self>, mpsc::UnboundedReceiver<DelegateEvent>) {
let (events_tx, events_rx) = mpsc::unbounded_channel();
@@ -51,6 +57,23 @@ impl HeldNotificationDelegate {
}
}
impl ReleasableNotificationDelegate {
fn new() -> (Arc<Self>, mpsc::UnboundedReceiver<DelegateEvent>) {
let (events_tx, events_rx) = mpsc::unbounded_channel();
(
Arc::new(Self {
events_tx,
notification_release: Semaphore::new(/*permits*/ 0),
}),
events_rx,
)
}
fn release_notification(&self) {
self.notification_release.add_permits(/*permits*/ 1);
}
}
impl CodeModeSessionDelegate for HeldNotificationDelegate {
fn invoke_tool<'a>(
&'a self,
@@ -78,11 +101,39 @@ impl CodeModeSessionDelegate for HeldNotificationDelegate {
Ok(())
})
}
}
fn cell_closed(&self, cell_id: &CellId) {
let _ = self
.events_tx
.send(DelegateEvent::CellClosed(cell_id.clone()));
impl CodeModeSessionDelegate for ReleasableNotificationDelegate {
fn invoke_tool<'a>(
&'a self,
_invocation: CodeModeNestedToolCall,
cancellation_token: CancellationToken,
) -> ToolInvocationFuture<'a> {
Box::pin(async move {
cancellation_token.cancelled().await;
Err("cancelled".to_string())
})
}
fn notify<'a>(
&'a self,
_call_id: String,
_cell_id: CellId,
_text: String,
cancellation_token: CancellationToken,
) -> NotificationFuture<'a> {
Box::pin(async move {
let _ = self.events_tx.send(DelegateEvent::NotificationStarted);
tokio::select! {
_ = self.notification_release.acquire() => {
let _ = self.events_tx.send(DelegateEvent::NotificationFinished);
Ok(())
}
_ = cancellation_token.cancelled() => {
Err("cancelled".to_string())
}
}
})
}
}
@@ -142,12 +193,6 @@ impl CodeModeSessionDelegate for BlockingDelegate {
Err("cancelled".to_string())
})
}
fn cell_closed(&self, cell_id: &CellId) {
let _ = self
.events_tx
.send(DelegateEvent::CellClosed(cell_id.clone()));
}
}
fn cell_id(value: &str) -> CellId {
@@ -325,6 +370,63 @@ async fn yields_and_resumes() {
);
}
#[tokio::test]
async fn yield_before_first_observation_preserves_its_output_boundary() {
let (delegate, mut events_rx) = BlockingDelegate::new();
let service = CodeModeService::with_delegate(delegate.clone());
let cell_id = service
.create_cell(CreateCellRequest {
enabled_tools: vec![blocking_tool()],
source: r#"
text("before");
yield_control();
text("after");
await tools.block({});
"#
.to_string(),
..execute_request("")
})
.await
.unwrap();
assert_eq!(next_event(&mut events_rx).await, DelegateEvent::ToolStarted);
assert_eq!(
service
.observe(ObserveRequest {
idempotency_key: "yield-boundary".to_string(),
cell_id: cell_id.clone(),
yield_time_ms: 60_000,
})
.await
.unwrap(),
ObserveOutcome::Yielded {
cell_id: cell_id.clone(),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "before".to_string(),
}],
}
);
delegate.release_tool();
assert_eq!(
service
.observe(ObserveRequest {
idempotency_key: "completion-boundary".to_string(),
cell_id: cell_id.clone(),
yield_time_ms: 60_000,
})
.await
.unwrap(),
ObserveOutcome::Completed {
cell_id,
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "after".to_string(),
}],
error_text: None,
}
);
}
#[tokio::test]
async fn returns_and_resumes_from_the_pending_frontier() {
let (delegate, mut events_rx) = BlockingDelegate::new();
@@ -464,10 +566,6 @@ async fn termination_cancels_pending_callbacks_before_responding() {
next_event(&mut events_rx).await,
DelegateEvent::NotificationCancelled
);
assert_eq!(
next_event(&mut events_rx).await,
DelegateEvent::CellClosed(cell_id("1"))
);
}
#[tokio::test]
@@ -501,10 +599,6 @@ await tools.block({});
next_event(&mut events_rx).await,
DelegateEvent::ToolCancelled
);
assert_eq!(
next_event(&mut events_rx).await,
DelegateEvent::CellClosed(cell_id("1"))
);
assert_eq!(
execute_with_yield_time(
@@ -547,10 +641,6 @@ async fn shutdown_cancels_notifications_while_natural_completion_is_draining() {
delegate.release_notification();
assert_eq!(shutdown.await.unwrap(), Ok(()));
assert_eq!(
next_event(&mut events_rx).await,
DelegateEvent::CellClosed(cell_id("1"))
);
}
#[tokio::test]
@@ -597,9 +687,69 @@ async fn repeated_termination_is_rejected_while_callback_cleanup_is_pending() {
content_items: Vec::new(),
}
);
}
#[tokio::test]
async fn create_cell_returns_before_natural_completion() {
let (delegate, mut events_rx) = ReleasableNotificationDelegate::new();
let service = CodeModeService::with_delegate(delegate.clone());
let created_cell_id = service
.create_cell(execute_request(r#"notify("pending");"#))
.await
.unwrap();
assert_eq!(created_cell_id, cell_id("1"));
let mut observation = Box::pin(service.observe(ObserveRequest {
idempotency_key: "blocked-notification".to_string(),
cell_id: created_cell_id,
yield_time_ms: 60_000,
}));
assert_eq!(
next_event(&mut events_rx).await,
DelegateEvent::CellClosed(cell_id("1"))
DelegateEvent::NotificationStarted
);
std::future::poll_fn(|context| match observation.as_mut().poll(context) {
std::task::Poll::Pending => std::task::Poll::Ready(()),
std::task::Poll::Ready(result) => {
panic!("observation returned while the notification was blocked: {result:?}")
}
})
.await;
delegate.release_notification();
assert_eq!(
observation.await.unwrap(),
ObserveOutcome::Completed {
cell_id: cell_id("1"),
content_items: Vec::new(),
error_text: None,
}
);
assert_eq!(
next_event(&mut events_rx).await,
DelegateEvent::NotificationFinished
);
}
#[tokio::test]
async fn created_cell_can_be_terminated_before_observation() {
let service = CodeModeService::new();
let created_cell_id = service
.create_cell(CreateCellRequest {
source: "await new Promise(() => {});".to_string(),
..execute_request("")
})
.await
.unwrap();
assert_eq!(created_cell_id, cell_id("1"));
assert_eq!(
service.terminate(created_cell_id.clone()).await.unwrap(),
TerminateOutcome::Terminated {
cell_id: created_cell_id,
content_items: Vec::new(),
}
);
}
@@ -690,8 +840,4 @@ async fn natural_completion_cleans_up_callbacks_before_responding() {
next_event(&mut events_rx).await,
DelegateEvent::ToolCancelled
);
assert_eq!(
next_event(&mut events_rx).await,
DelegateEvent::CellClosed(cell_id("1"))
);
}

393
codex-rs/code-mode/src/service_tests.rs
View File

@@ -1,4 +1,6 @@
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use super::CellId;
@@ -11,9 +13,17 @@ use super::ObserveRequest;
use super::ObserveToPendingOutcome;
use super::ObserveToPendingRequest;
use super::PendingOutcome;
use super::ProtocolDelegate;
use super::RuntimeResponse;
use super::TerminateOutcome;
use super::ToolInvocationFuture;
use super::missing_cell_response;
use super::observe_outcome;
use super::pending_outcome;
use super::protocol_cell_id;
use super::runtime;
use super::runtime_cell_id;
use super::runtime_request;
use crate::CodeModeToolKind;
use crate::CreateCellRequest;
use crate::FunctionCallOutputContentItem;
@@ -21,6 +31,7 @@ use crate::ToolDefinition;
use codex_protocol::ToolName;
use pretty_assertions::assert_eq;
use serde_json::Value as JsonValue;
use serde_json::json;
use tokio::sync::Notify;
use tokio_util::sync::CancellationToken;
@@ -58,8 +69,88 @@ impl CodeModeSessionDelegate for ReleasableToolDelegate {
) -> NotificationFuture<'a> {
Box::pin(async { Ok(()) })
}
}
fn cell_closed(&self, _cell_id: &CellId) {}
#[derive(Debug, PartialEq)]
enum RecordedDelegateCall {
Tool {
invocation: CodeModeNestedToolCall,
cancellation_requested: bool,
},
Notification {
call_id: String,
cell_id: CellId,
text: String,
cancellation_requested: bool,
},
}
struct RecordingDelegate {
calls: Mutex<Vec<RecordedDelegateCall>>,
tool_results: Mutex<VecDeque<Result<JsonValue, String>>>,
notification_results: Mutex<VecDeque<Result<(), String>>>,
}
impl RecordingDelegate {
fn new(
tool_results: Vec<Result<JsonValue, String>>,
notification_results: Vec<Result<(), String>>,
) -> Self {
Self {
calls: Mutex::new(Vec::new()),
tool_results: Mutex::new(tool_results.into()),
notification_results: Mutex::new(notification_results.into()),
}
}
fn take_calls(&self) -> Vec<RecordedDelegateCall> {
std::mem::take(&mut *self.calls.lock().unwrap())
}
}
impl CodeModeSessionDelegate for RecordingDelegate {
fn invoke_tool<'a>(
&'a self,
invocation: CodeModeNestedToolCall,
cancellation_token: CancellationToken,
) -> ToolInvocationFuture<'a> {
Box::pin(async move {
self.calls.lock().unwrap().push(RecordedDelegateCall::Tool {
invocation,
cancellation_requested: cancellation_token.is_cancelled(),
});
self.tool_results
.lock()
.unwrap()
.pop_front()
.expect("test must provide one result per tool call")
})
}
fn notify<'a>(
&'a self,
call_id: String,
cell_id: CellId,
text: String,
cancellation_token: CancellationToken,
) -> NotificationFuture<'a> {
Box::pin(async move {
self.calls
.lock()
.unwrap()
.push(RecordedDelegateCall::Notification {
call_id,
cell_id,
text,
cancellation_requested: cancellation_token.is_cancelled(),
});
self.notification_results
.lock()
.unwrap()
.pop_front()
.expect("test must provide one result per notification")
})
}
}
fn execute_request(source: &str) -> CreateCellRequest {
@@ -999,3 +1090,303 @@ async fn observe_reports_missing_cell_separately_from_runtime_results() {
}
);
}
#[test]
fn protocol_requests_map_to_runtime_requests_field_for_field() {
let request = CreateCellRequest {
idempotency_key: "thread-3:response-call-7".to_string(),
tool_call_id: "response-call-7".to_string(),
enabled_tools: vec![
ToolDefinition {
name: "mcp__search__query".to_string(),
tool_name: ToolName::namespaced("search", "query"),
description: "Search indexed documents".to_string(),
kind: CodeModeToolKind::Function,
input_schema: Some(json!({"type": "object"})),
output_schema: None,
},
ToolDefinition {
name: "apply_patch".to_string(),
tool_name: ToolName::plain("apply_patch"),
description: "Apply a patch".to_string(),
kind: CodeModeToolKind::Freeform,
input_schema: None,
output_schema: Some(json!({"type": "string"})),
},
],
source: "await tools.mcp__search__query({ q: 'rust' });".to_string(),
};
assert_eq!(
runtime_request(request),
runtime::CreateCellRequest {
idempotency_key: "thread-3:response-call-7".to_string(),
tool_call_id: "response-call-7".to_string(),
enabled_tools: vec![
runtime::ToolDefinition {
name: "mcp__search__query".to_string(),
tool_name: runtime::ToolName {
name: "query".to_string(),
namespace: Some("search".to_string()),
},
description: "Search indexed documents".to_string(),
kind: runtime::ToolKind::Function,
},
runtime::ToolDefinition {
name: "apply_patch".to_string(),
tool_name: runtime::ToolName {
name: "apply_patch".to_string(),
namespace: None,
},
description: "Apply a patch".to_string(),
kind: runtime::ToolKind::Freeform,
},
],
source: "await tools.mcp__search__query({ q: 'rust' });".to_string(),
}
);
let protocol_id = cell_id("cell-a7");
assert_eq!(
protocol_cell_id(&runtime_cell_id(&protocol_id)),
protocol_id
);
}
#[tokio::test]
async fn protocol_delegate_maps_callbacks_cancellation_and_errors_field_for_field() {
let delegate = Arc::new(RecordingDelegate::new(
vec![
Ok(json!({"matches": 3})),
Err("freeform tool failed".to_string()),
],
vec![Err("notification failed".to_string())],
));
let adapter = ProtocolDelegate {
delegate: delegate.clone(),
};
let cancelled = CancellationToken::new();
cancelled.cancel();
assert_eq!(
runtime::SessionRuntimeDelegate::invoke_tool(
&adapter,
runtime::NestedToolCall {
cell_id: runtime::CellId::new("cell-a7"),
runtime_tool_call_id: "runtime-call-1".to_string(),
tool_name: runtime::ToolName {
name: "query".to_string(),
namespace: Some("search".to_string()),
},
tool_kind: runtime::ToolKind::Function,
input: Some(json!({"q": "rust"})),
},
CancellationToken::new(),
)
.await,
Ok(json!({"matches": 3}))
);
assert_eq!(
runtime::SessionRuntimeDelegate::invoke_tool(
&adapter,
runtime::NestedToolCall {
cell_id: runtime::CellId::new("cell-b9"),
runtime_tool_call_id: "runtime-call-2".to_string(),
tool_name: runtime::ToolName {
name: "apply_patch".to_string(),
namespace: None,
},
tool_kind: runtime::ToolKind::Freeform,
input: None,
},
cancelled.clone(),
)
.await,
Err("freeform tool failed".to_string())
);
assert_eq!(
runtime::SessionRuntimeDelegate::notify(
&adapter,
"notification-1".to_string(),
runtime::CellId::new("cell-c3"),
"progress".to_string(),
cancelled,
)
.await,
Err("notification failed".to_string())
);
assert_eq!(
delegate.take_calls(),
vec![
RecordedDelegateCall::Tool {
invocation: CodeModeNestedToolCall {
cell_id: cell_id("cell-a7"),
runtime_tool_call_id: "runtime-call-1".to_string(),
tool_name: ToolName::namespaced("search", "query"),
tool_kind: CodeModeToolKind::Function,
input: Some(json!({"q": "rust"})),
},
cancellation_requested: false,
},
RecordedDelegateCall::Tool {
invocation: CodeModeNestedToolCall {
cell_id: cell_id("cell-b9"),
runtime_tool_call_id: "runtime-call-2".to_string(),
tool_name: ToolName::plain("apply_patch"),
tool_kind: CodeModeToolKind::Freeform,
input: None,
},
cancellation_requested: true,
},
RecordedDelegateCall::Notification {
call_id: "notification-1".to_string(),
cell_id: cell_id("cell-c3"),
text: "progress".to_string(),
cancellation_requested: true,
},
]
);
}
#[test]
fn runtime_events_map_to_protocol_outcomes_field_for_field() {
let output_items = vec![
runtime::OutputItem::Text {
text: "before".to_string(),
},
runtime::OutputItem::Image {
image_url: "data:image/png;base64,auto".to_string(),
detail: Some(runtime::ImageDetail::Auto),
},
runtime::OutputItem::Image {
image_url: "data:image/png;base64,low".to_string(),
detail: Some(runtime::ImageDetail::Low),
},
runtime::OutputItem::Image {
image_url: "data:image/png;base64,high".to_string(),
detail: Some(runtime::ImageDetail::High),
},
runtime::OutputItem::Image {
image_url: "data:image/png;base64,original".to_string(),
detail: Some(runtime::ImageDetail::Original),
},
runtime::OutputItem::Image {
image_url: "data:image/png;base64,default".to_string(),
detail: None,
},
];
let expected_output_items = vec![
FunctionCallOutputContentItem::InputText {
text: "before".to_string(),
},
FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,auto".to_string(),
detail: Some(crate::ImageDetail::Auto),
},
FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,low".to_string(),
detail: Some(crate::ImageDetail::Low),
},
FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,high".to_string(),
detail: Some(crate::ImageDetail::High),
},
FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,original".to_string(),
detail: Some(crate::ImageDetail::Original),
},
FunctionCallOutputContentItem::InputImage {
image_url: "data:image/png;base64,default".to_string(),
detail: None,
},
];
assert_eq!(
observe_outcome(
&cell_id("cell-a7"),
runtime::CellEvent::Yielded {
content_items: output_items,
},
),
ObserveOutcome::Yielded {
cell_id: cell_id("cell-a7"),
content_items: expected_output_items,
}
);
assert_eq!(
observe_outcome(
&cell_id("cell-b9"),
runtime::CellEvent::Completed {
content_items: vec![runtime::OutputItem::Text {
text: "failed".to_string(),
}],
error_text: Some("tool failed".to_string()),
},
),
ObserveOutcome::Completed {
cell_id: cell_id("cell-b9"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "failed".to_string(),
}],
error_text: Some("tool failed".to_string()),
}
);
assert_eq!(
observe_outcome(
&cell_id("cell-c3"),
runtime::CellEvent::Terminated {
content_items: vec![runtime::OutputItem::Text {
text: "partial".to_string(),
}],
},
),
ObserveOutcome::Terminated {
cell_id: cell_id("cell-c3"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "partial".to_string(),
}],
}
);
assert_eq!(
pending_outcome(
&cell_id("cell-d4"),
runtime::PausableCellEvent::Pending(runtime::PendingFrontier {
generation: runtime::PendingGeneration::new(/*value*/ 1),
content_items: vec![runtime::OutputItem::Text {
text: "waiting".to_string(),
}],
pending_tool_call_ids: vec!["runtime-call-1".to_string()],
}),
),
Ok(PendingOutcome::Pending {
cell_id: cell_id("cell-d4"),
content_items: vec![FunctionCallOutputContentItem::InputText {
text: "waiting".to_string(),
}],
pending_tool_call_ids: vec!["runtime-call-1".to_string()],
})
);
assert_eq!(
pending_outcome(
&cell_id("cell-e5"),
runtime::PausableCellEvent::Completed {
content_items: Vec::new(),
error_text: None,
},
),
Ok(PendingOutcome::Completed(RuntimeResponse::Result {
cell_id: cell_id("cell-e5"),
content_items: Vec::new(),
error_text: None,
}))
);
assert_eq!(
missing_cell_response(cell_id("missing")),
RuntimeResponse::Result {
cell_id: cell_id("missing"),
content_items: Vec::new(),
error_text: Some("exec cell missing not found".to_string()),
}
);
}

128
codex-rs/code-mode/src/session_runtime/mod.rs
View File

@@ -9,30 +9,32 @@ use std::sync::atomic::Ordering;
use std::time::Duration;
use serde_json::Value as JsonValue;
use sha2::Digest;
use sha2::Sha256;
use tokio::sync::Mutex;
use tokio::sync::OnceCell;
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;
pub(crate) use self::types::Cell;
pub(crate) use self::types::CellEvent;
pub use self::types::Cell;
pub use self::types::CellEvent;
pub(crate) use self::types::CellExecutionPolicy;
pub(crate) use self::types::CellId;
pub(crate) use self::types::CellKind;
pub(crate) use self::types::CreateCellRequest;
pub(crate) use self::types::Error;
pub(crate) use self::types::ImageDetail;
pub(crate) use self::types::NestedToolCall;
pub(crate) use self::types::OutputItem;
pub(crate) use self::types::PausableCell;
pub(crate) use self::types::PausableCellEvent;
pub(crate) use self::types::PendingFrontier;
pub(crate) use self::types::PendingGeneration;
pub(crate) use self::types::ResumeOutcome;
pub(crate) use self::types::SessionRuntimeDelegate;
pub(crate) use self::types::ToolDefinition;
pub(crate) use self::types::ToolKind;
pub(crate) use self::types::ToolName;
pub use self::types::CellId;
pub use self::types::CellKind;
pub use self::types::CreateCellRequest;
pub use self::types::Error;
pub use self::types::ImageDetail;
pub use self::types::NestedToolCall;
pub use self::types::OutputItem;
pub use self::types::PausableCell;
pub use self::types::PausableCellEvent;
pub use self::types::PendingFrontier;
pub use self::types::PendingGeneration;
pub use self::types::ResumeOutcome;
pub use self::types::SessionRuntimeDelegate;
pub use self::types::ToolDefinition;
pub use self::types::ToolKind;
pub use self::types::ToolName;
use crate::cell_actor::ActorEvent;
use crate::cell_actor::CellActor;
use crate::cell_actor::CellError;
@@ -45,12 +47,14 @@ use crate::cell_actor::CompletionCommit;
use crate::cell_actor::ObserveMode;
type RuntimeEventFuture = Pin<Box<dyn Future<Output = Result<CellEvent, Error>> + Send + 'static>>;
#[allow(dead_code)]
type PausableRuntimeEventFuture =
Pin<Box<dyn Future<Output = Result<PausableCellEvent, Error>> + Send + 'static>>;
const CELL_ID_ALPHABET: &[u8; 32] = b"0123456789abcdefghjkmnpqrstvwxyz";
const CELL_ID_LENGTH: usize = 16;
/// Owns all cells and shared state for one transport-neutral code-mode session.
pub(crate) struct SessionRuntime<D: SessionRuntimeDelegate> {
pub struct SessionRuntime<D: SessionRuntimeDelegate> {
inner: Arc<Inner<D>>,
}
@@ -61,6 +65,7 @@ struct Inner<D: SessionRuntimeDelegate> {
cell_tasks: TaskTracker,
shutdown_token: CancellationToken,
delegate: Arc<D>,
cell_id_namespace: CellIdNamespace,
next_cell_id: AtomicU64,
}
@@ -91,8 +96,23 @@ impl RegisteredCell {
}
}
enum CellIdNamespace {
Runtime(uuid::Uuid),
#[cfg(test)]
Unscoped,
}
impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
pub(crate) fn new(delegate: Arc<D>) -> Self {
pub fn new(delegate: Arc<D>) -> Self {
Self::with_cell_id_namespace(delegate, CellIdNamespace::Runtime(uuid::Uuid::new_v4()))
}
#[cfg(test)]
pub(crate) fn new_for_test(delegate: Arc<D>) -> Self {
Self::with_cell_id_namespace(delegate, CellIdNamespace::Unscoped)
}
fn with_cell_id_namespace(delegate: Arc<D>, cell_id_namespace: CellIdNamespace) -> Self {
Self {
inner: Arc::new(Inner {
stored_values: Mutex::new(HashMap::new()),
@@ -101,16 +121,17 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
cell_tasks: TaskTracker::new(),
shutdown_token: CancellationToken::new(),
delegate,
cell_id_namespace,
next_cell_id: AtomicU64::new(1),
}),
}
}
pub(crate) fn is_alive(&self) -> bool {
pub fn is_alive(&self) -> bool {
!self.inner.shutdown_token.is_cancelled()
}
pub(crate) async fn create_cell(&self, request: CreateCellRequest) -> Result<Cell, Error> {
pub async fn create_cell(&self, request: CreateCellRequest) -> Result<Cell, Error> {
let id = self
.create_idempotent_cell(
request,
@@ -121,8 +142,7 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
Ok(Cell::new(id))
}
#[allow(dead_code)]
pub(crate) async fn create_pausable_cell(
pub async fn create_pausable_cell(
&self,
request: CreateCellRequest,
) -> Result<PausableCell, Error> {
@@ -136,16 +156,11 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
Ok(PausableCell::new(id))
}
#[cfg(test)]
pub(crate) async fn wait(
&self,
cell: &Cell,
yield_after: Duration,
) -> Result<CellEvent, Error> {
pub async fn wait(&self, cell: &Cell, yield_after: Duration) -> Result<CellEvent, Error> {
self.begin_wait(cell, yield_after).await?.event().await
}
pub(crate) async fn begin_wait(
pub async fn begin_wait(
&self,
cell: &Cell,
yield_after: Duration,
@@ -159,11 +174,7 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
})
}
#[allow(dead_code)]
pub(crate) async fn wait_to_pending(
&self,
cell: &PausableCell,
) -> Result<PausableCellEvent, Error> {
pub async fn wait_to_pending(&self, cell: &PausableCell) -> Result<PausableCellEvent, Error> {
let handle = self.pausable_handle(cell.id()).await?;
map_pausable_event(
cell.id().clone(),
@@ -172,8 +183,7 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
.await
}
#[allow(dead_code)]
pub(crate) async fn resume(
pub async fn resume(
&self,
cell: &PausableCell,
generation: PendingGeneration,
@@ -196,7 +206,7 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
Ok(PausableCell::new(cell_id.clone()))
}
pub(crate) async fn terminate(&self, cell_id: &CellId) -> Result<CellEvent, Error> {
pub async fn terminate(&self, cell_id: &CellId) -> Result<CellEvent, Error> {
let cell = self
.inner
.cells
@@ -212,7 +222,7 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
.map_err(|error| actor_error(cell_id, error))
}
pub(crate) async fn shutdown(&self) -> Result<(), Error> {
pub async fn shutdown(&self) -> Result<(), Error> {
self.begin_shutdown();
// Taking the registry lock ensures every cell that passed the shutdown
// check has registered its actor with the tracker before we wait.
@@ -224,12 +234,29 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
}
fn allocate_cell_id(&self) -> CellId {
CellId::new(
self.inner
.next_cell_id
.fetch_add(1, Ordering::Relaxed)
.to_string(),
)
let sequence = self.inner.next_cell_id.fetch_add(1, Ordering::Relaxed);
let value = match &self.inner.cell_id_namespace {
CellIdNamespace::Runtime(runtime_id) => {
let digest = Sha256::new()
.chain_update(runtime_id.as_bytes())
.chain_update(sequence.to_be_bytes())
.finalize();
let mut encoded = String::with_capacity(CELL_ID_LENGTH);
for chunk in digest[..10].chunks_exact(5) {
let chunk_bits = chunk
.iter()
.fold(0_u64, |value, byte| (value << 8) | u64::from(*byte));
for shift in (0..40).step_by(5).rev() {
let index = ((chunk_bits >> shift) & 0x1f) as usize;
encoded.push(char::from(CELL_ID_ALPHABET[index]));
}
}
encoded
}
#[cfg(test)]
CellIdNamespace::Unscoped => sequence.to_string(),
};
CellId::new(value)
}
async fn start_cell(
@@ -303,7 +330,6 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
self.handle_for_kind(cell_id, CellKind::Continuing).await
}
#[allow(dead_code)]
async fn pausable_handle(&self, cell_id: &CellId) -> Result<CellHandle, Error> {
self.handle_for_kind(cell_id, CellKind::Pausable).await
}
@@ -341,12 +367,12 @@ impl<D: SessionRuntimeDelegate> Drop for SessionRuntime<D> {
}
/// An admitted cell event that has not reached its requested frontier yet.
pub(crate) struct PendingEvent {
pub struct PendingEvent {
event: RuntimeEventFuture,
}
impl PendingEvent {
pub(crate) async fn event(self) -> Result<CellEvent, Error> {
pub async fn event(self) -> Result<CellEvent, Error> {
self.event.await
}
}
@@ -411,7 +437,6 @@ impl<D: SessionRuntimeDelegate> CellHost for RuntimeCellHost<D> {
async fn closed(&self) {
self.inner.cells.lock().await.remove(&self.cell_id);
self.inner.delegate.cell_closed(&self.cell_id);
}
}
@@ -434,7 +459,6 @@ fn map_cell_event(cell_id: CellId, event: CellEventFuture) -> RuntimeEventFuture
})
}
#[allow(dead_code)]
fn map_pausable_event(cell_id: CellId, event: CellEventFuture) -> PausableRuntimeEventFuture {
Box::pin(async move {
match event.await.map_err(|error| actor_error(&cell_id, error))? {

113
codex-rs/code-mode/src/session_runtime/tests.rs
View File

@@ -26,6 +26,10 @@ struct BlockingToolDelegate {
release: Arc<Semaphore>,
}
struct NonCooperativeToolDelegate {
invocations_tx: mpsc::UnboundedSender<()>,
}
impl SessionRuntimeDelegate for RecordingDelegate {
async fn invoke_tool(
&self,
@@ -44,8 +48,6 @@ impl SessionRuntimeDelegate for RecordingDelegate {
) -> Result<(), String> {
Ok(())
}
fn cell_closed(&self, _cell_id: &CellId) {}
}
impl SessionRuntimeDelegate for ImmediateToolDelegate {
@@ -67,8 +69,6 @@ impl SessionRuntimeDelegate for ImmediateToolDelegate {
) -> Result<(), String> {
Ok(())
}
fn cell_closed(&self, _cell_id: &CellId) {}
}
impl SessionRuntimeDelegate for BlockingToolDelegate {
@@ -95,8 +95,27 @@ impl SessionRuntimeDelegate for BlockingToolDelegate {
) -> Result<(), String> {
Ok(())
}
}
fn cell_closed(&self, _cell_id: &CellId) {}
impl SessionRuntimeDelegate for NonCooperativeToolDelegate {
async fn invoke_tool(
&self,
_invocation: NestedToolCall,
_cancellation_token: CancellationToken,
) -> Result<JsonValue, String> {
let _ = self.invocations_tx.send(());
std::future::pending().await
}
async fn notify(
&self,
_call_id: String,
_cell_id: CellId,
_text: String,
_cancellation_token: CancellationToken,
) -> Result<(), String> {
Ok(())
}
}
fn tool_definition(name: &str) -> ToolDefinition {
@@ -666,7 +685,6 @@ async fn drop_terminates_cells_when_the_registry_is_locked() {
.create_cell(execute_request("while (true) {}"))
.await
.unwrap();
assert_eq!(cell.id(), &CellId::new("1"));
assert_eq!(
runtime
.wait(&cell, Duration::from_millis(/*millis*/ 1))
@@ -686,3 +704,86 @@ async fn drop_terminates_cells_when_the_registry_is_locked() {
.unwrap();
assert!(inner.cell_tasks.is_empty());
}
#[tokio::test]
async fn termination_aborts_a_non_cooperative_tool_callback() {
let (invocations_tx, mut invocations_rx) = mpsc::unbounded_channel();
let runtime = SessionRuntime::new(Arc::new(NonCooperativeToolDelegate { invocations_tx }));
let cell = runtime
.create_cell(CreateCellRequest {
idempotency_key: "non-cooperative-termination".to_string(),
tool_call_id: "call-1".to_string(),
enabled_tools: vec![tool_definition("blocked")],
source: "await tools.blocked({});".to_string(),
})
.await
.unwrap();
tokio::time::timeout(Duration::from_secs(1), invocations_rx.recv())
.await
.expect("tool invocation timed out")
.expect("tool invocation channel closed");
assert_eq!(
tokio::time::timeout(Duration::from_secs(1), runtime.terminate(cell.id()))
.await
.expect("termination waited for a non-cooperative callback"),
Ok(CellEvent::Terminated {
content_items: Vec::new(),
})
);
runtime.shutdown().await.unwrap();
}
#[tokio::test]
async fn shutdown_aborts_a_non_cooperative_tool_callback() {
let (invocations_tx, mut invocations_rx) = mpsc::unbounded_channel();
let runtime = SessionRuntime::new(Arc::new(NonCooperativeToolDelegate { invocations_tx }));
runtime
.create_cell(CreateCellRequest {
idempotency_key: "non-cooperative-shutdown".to_string(),
tool_call_id: "call-1".to_string(),
enabled_tools: vec![tool_definition("blocked")],
source: "await tools.blocked({});".to_string(),
})
.await
.unwrap();
tokio::time::timeout(Duration::from_secs(1), invocations_rx.recv())
.await
.expect("tool invocation timed out")
.expect("tool invocation channel closed");
assert_eq!(
tokio::time::timeout(Duration::from_secs(1), runtime.shutdown())
.await
.expect("shutdown waited for a non-cooperative callback"),
Ok(())
);
}
#[tokio::test]
async fn cell_ids_are_unique_across_runtime_instances() {
let first_runtime = SessionRuntime::new(Arc::new(RecordingDelegate));
let second_runtime = SessionRuntime::new(Arc::new(RecordingDelegate));
let first_cell_id = first_runtime
.create_cell(execute_request("await new Promise(() => {});"))
.await
.unwrap();
let second_cell_id = second_runtime
.create_cell(execute_request("await new Promise(() => {});"))
.await
.unwrap();
assert_ne!(first_cell_id, second_cell_id);
for cell in [&first_cell_id, &second_cell_id] {
assert_eq!(cell.id().as_str().len(), CELL_ID_LENGTH);
assert!(
cell.id()
.as_str()
.bytes()
.all(|byte| CELL_ID_ALPHABET.contains(&byte))
);
}
first_runtime.shutdown().await.unwrap();
second_runtime.shutdown().await.unwrap();
}

96
codex-rs/code-mode/src/session_runtime/types.rs
View File

@@ -6,14 +6,14 @@ use tokio_util::sync::CancellationToken;
/// Identifies one execution cell within a session runtime.
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
pub(crate) struct CellId(String);
pub struct CellId(String);
impl CellId {
pub(crate) fn new(value: impl Into<String>) -> Self {
pub fn new(value: impl Into<String>) -> Self {
Self(value.into())
}
pub(crate) fn as_str(&self) -> &str {
pub fn as_str(&self) -> &str {
&self.0
}
}
@@ -31,13 +31,12 @@ pub(crate) enum CellExecutionPolicy {
#[default]
ContinueWhenUnblocked,
/// Remain paused at a pending frontier until an explicit resume advances it.
#[allow(dead_code)]
PauseAtPendingFrontier,
}
/// A cell that continues whenever external input unblocks its runtime.
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
pub(crate) struct Cell {
pub struct Cell {
id: CellId,
}
@@ -46,31 +45,29 @@ impl Cell {
Self { id }
}
pub(crate) fn id(&self) -> &CellId {
pub fn id(&self) -> &CellId {
&self.id
}
}
/// A cell that remains paused at each pending frontier until explicitly resumed.
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
#[allow(dead_code)]
pub(crate) struct PausableCell {
pub struct PausableCell {
id: CellId,
}
#[allow(dead_code)]
impl PausableCell {
pub(super) fn new(id: CellId) -> Self {
Self { id }
}
pub(crate) fn id(&self) -> &CellId {
pub fn id(&self) -> &CellId {
&self.id
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum CellKind {
pub enum CellKind {
Continuing,
Pausable,
}
@@ -86,29 +83,29 @@ impl fmt::Display for CellKind {
/// Identifies one durable pending frontier of a pausable cell.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub(crate) struct PendingGeneration(u64);
pub struct PendingGeneration(u64);
impl PendingGeneration {
pub(crate) fn new(value: u64) -> Self {
Self(value)
}
pub(crate) fn get(self) -> u64 {
pub fn get(self) -> u64 {
self.0
}
}
/// A repeatable snapshot of one paused runtime frontier.
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct PendingFrontier {
pub(crate) generation: PendingGeneration,
pub(crate) content_items: Vec<OutputItem>,
pub(crate) pending_tool_call_ids: Vec<String>,
pub struct PendingFrontier {
pub generation: PendingGeneration,
pub content_items: Vec<OutputItem>,
pub pending_tool_call_ids: Vec<String>,
}
/// An observable cell lifecycle event.
#[derive(Clone, Debug, PartialEq)]
pub(crate) enum CellEvent {
pub enum CellEvent {
Yielded {
content_items: Vec<OutputItem>,
},
@@ -123,8 +120,7 @@ pub(crate) enum CellEvent {
/// An observable lifecycle event for a pausable cell.
#[derive(Clone, Debug, PartialEq)]
#[allow(dead_code)]
pub(crate) enum PausableCellEvent {
pub enum PausableCellEvent {
Pending(PendingFrontier),
Completed {
content_items: Vec<OutputItem>,
@@ -136,14 +132,14 @@ pub(crate) enum PausableCellEvent {
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ResumeOutcome {
pub enum ResumeOutcome {
Resumed,
AlreadyRunning,
}
/// Output emitted by a cell since its preceding observation.
#[derive(Clone, Debug, PartialEq)]
pub(crate) enum OutputItem {
pub enum OutputItem {
Text {
text: String,
},
@@ -155,7 +151,7 @@ pub(crate) enum OutputItem {
/// Requested image fidelity for an output image.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ImageDetail {
pub enum ImageDetail {
Auto,
Low,
High,
@@ -165,49 +161,53 @@ pub(crate) enum ImageDetail {
/// Transport-neutral input for creating a cell.
///
/// The owning session assigns the cell ID when it admits the request.
pub(crate) struct CreateCellRequest {
pub(crate) idempotency_key: String,
pub(crate) tool_call_id: String,
pub(crate) enabled_tools: Vec<ToolDefinition>,
pub(crate) source: String,
#[derive(Debug, PartialEq)]
pub struct CreateCellRequest {
pub idempotency_key: String,
pub tool_call_id: String,
pub enabled_tools: Vec<ToolDefinition>,
pub source: String,
}
/// Tool metadata exposed to code running inside a cell.
pub(crate) struct ToolDefinition {
pub(crate) name: String,
pub(crate) tool_name: ToolName,
pub(crate) description: String,
pub(crate) kind: ToolKind,
#[derive(Debug, PartialEq)]
pub struct ToolDefinition {
pub name: String,
pub tool_name: ToolName,
pub description: String,
pub kind: ToolKind,
}
/// A tool name with an optional namespace.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct ToolName {
pub(crate) name: String,
pub(crate) namespace: Option<String>,
pub struct ToolName {
pub name: String,
pub namespace: Option<String>,
}
/// The JavaScript calling convention for a tool.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ToolKind {
pub enum ToolKind {
Function,
Freeform,
}
/// A nested tool request emitted by a running cell.
pub(crate) struct NestedToolCall {
pub(crate) cell_id: CellId,
pub(crate) runtime_tool_call_id: String,
pub(crate) tool_name: ToolName,
pub(crate) tool_kind: ToolKind,
pub(crate) input: Option<JsonValue>,
#[derive(Debug, PartialEq)]
pub struct NestedToolCall {
pub cell_id: CellId,
pub runtime_tool_call_id: String,
pub tool_name: ToolName,
pub tool_kind: ToolKind,
pub input: Option<JsonValue>,
}
/// Host callbacks used by cells owned by a [`super::SessionRuntime`].
///
/// Implementations must honor cancellation tokens. `cell_closed` is called
/// after the runtime has stopped routing requests to the cell.
pub(crate) trait SessionRuntimeDelegate: Send + Sync + 'static {
/// Implementations should forward callback cancellation tokens to downstream
/// work. After cancellation begins, the runtime allows callbacks a bounded
/// grace period to finish, then aborts their local tasks.
pub trait SessionRuntimeDelegate: Send + Sync + 'static {
fn invoke_tool(
&self,
invocation: NestedToolCall,
@@ -221,13 +221,11 @@ pub(crate) trait SessionRuntimeDelegate: Send + Sync + 'static {
text: String,
cancellation_token: CancellationToken,
) -> impl Future<Output = Result<(), String>> + Send;
fn cell_closed(&self, cell_id: &CellId);
}
/// A failure reported by a session runtime operation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum Error {
pub enum Error {
ShuttingDown,
DuplicateCell(CellId),
MissingCell(CellId),

4
codex-rs/core/src/tools/code_mode/delegate.rs
View File

@@ -238,10 +238,6 @@ impl CodeModeSessionDelegate for CodeModeDispatchBroker {
}
})
}
fn cell_closed(&self, cell_id: &CellId) {
self.close_cell(cell_id);
}
}
enum DispatchMessage {

10
codex-rs/core/tests/suite/code_mode.rs
View File

@@ -1412,7 +1412,7 @@ text("phase 3");
assert_regex_match(
concat!(
r"(?s)\A",
r"Script running with cell ID \d+\nWall time \d+\.\d seconds\nOutput:\n\z"
r"Script running with cell ID [0-9abcdefghjkmnpqrstvwxyz]{16}\nWall time \d+\.\d seconds\nOutput:\n\z"
),
text_item(&first_items, /*index*/ 0),
);
@@ -1453,7 +1453,7 @@ text("phase 3");
assert_regex_match(
concat!(
r"(?s)\A",
r"Script running with cell ID \d+\nWall time \d+\.\d seconds\nOutput:\n\z"
r"Script running with cell ID [0-9abcdefghjkmnpqrstvwxyz]{16}\nWall time \d+\.\d seconds\nOutput:\n\z"
),
text_item(&second_items, /*index*/ 0),
);
@@ -1556,7 +1556,7 @@ while (true) {}
assert_regex_match(
concat!(
r"(?s)\A",
r"Script running with cell ID \d+\nWall time \d+\.\d seconds\nOutput:\n\z"
r"Script running with cell ID [0-9abcdefghjkmnpqrstvwxyz]{16}\nWall time \d+\.\d seconds\nOutput:\n\z"
),
text_item(&first_items, /*index*/ 0),
);
@@ -2235,7 +2235,7 @@ text("session b done");
assert_regex_match(
concat!(
r"(?s)\A",
r"Script running with cell ID \d+\nWall time \d+\.\d seconds\nOutput:\n\z"
r"Script running with cell ID [0-9abcdefghjkmnpqrstvwxyz]{16}\nWall time \d+\.\d seconds\nOutput:\n\z"
),
text_item(&third_items, /*index*/ 0),
);
@@ -2357,7 +2357,7 @@ text("after yield");
assert_regex_match(
concat!(
r"(?s)\A",
r"Script running with cell ID \d+\nWall time \d+\.\d seconds\nOutput:\n\z"
r"Script running with cell ID [0-9abcdefghjkmnpqrstvwxyz]{16}\nWall time \d+\.\d seconds\nOutput:\n\z"
),
text_item(&first_items, /*index*/ 0),
);