Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 37 additions & 17 deletions codex-rs/code-mode-protocol/src/host/codec.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,36 @@ use tokio::io::AsyncWriteExt;
/// Maximum JSON payload size accepted for one IPC frame.
pub const MAX_FRAME_BYTES: usize = 64 * 1024 * 1024;

/// A serialized IPC frame that has already passed the payload size limit.
#[derive(Clone, Debug)]
pub struct EncodedFrame {
payload: Vec<u8>,
}

impl EncodedFrame {
pub fn encode<T>(message: &T) -> io::Result<Self>
where
T: Serialize,
{
let payload = serde_json::to_vec(message).map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("failed to encode code-mode IPC frame: {err}"),
)
})?;
if payload.len() > MAX_FRAME_BYTES {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"code-mode IPC frame length {} exceeds {MAX_FRAME_BYTES} bytes",
payload.len()
),
));
}
Ok(Self { payload })
}
}

/// Decodes JSON messages prefixed by a four-byte little-endian payload length.
pub struct FramedReader<R> {
reader: R,
Expand Down Expand Up @@ -72,30 +102,20 @@ where
where
T: Serialize,
{
let payload = serde_json::to_vec(message).map_err(|err| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("failed to encode code-mode IPC frame: {err}"),
)
})?;
if payload.len() > MAX_FRAME_BYTES {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"code-mode IPC frame length {} exceeds {MAX_FRAME_BYTES} bytes",
payload.len()
),
));
}
let length = u32::try_from(payload.len()).map_err(|_| {
self.write_frame(&EncodedFrame::encode(message)?).await
}

/// Writes and flushes a frame encoded before it entered an I/O queue.
pub async fn write_frame(&mut self, frame: &EncodedFrame) -> io::Result<()> {
let length = u32::try_from(frame.payload.len()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidData,
"code-mode IPC frame length exceeds u32",
)
})?;

self.writer.write_all(&length.to_le_bytes()).await?;
self.writer.write_all(&payload).await?;
self.writer.write_all(&frame.payload).await?;
self.writer.flush().await
}
}
10 changes: 10 additions & 0 deletions codex-rs/code-mode-protocol/src/host/host_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -365,6 +365,16 @@ fn client_to_host_v1_variants_are_pinned() {
}),
);
}

assert_wire_round_trip(
ClientToHost::CancelRequest {
id: request_id(/*value*/ 9),
},
json!({
"type": "operation/cancel",
"id": 9,
}),
);
}

#[test]
Expand Down
2 changes: 2 additions & 0 deletions codex-rs/code-mode-protocol/src/host/message.rs
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,8 @@ pub enum ClientToHost {
ClientHello(ClientHello),
#[serde(rename = "operation/request")]
Request { id: RequestId, request: HostRequest },
#[serde(rename = "operation/cancel")]
CancelRequest { id: RequestId },
#[serde(rename = "delegate/response")]
DelegateResponse {
id: DelegateRequestId,
Expand Down
1 change: 1 addition & 0 deletions codex-rs/code-mode-protocol/src/host/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ mod message;
mod payload;
mod types;

pub use codec::EncodedFrame;
pub use codec::FramedReader;
pub use codec::FramedWriter;
pub use codec::MAX_FRAME_BYTES;
Expand Down
Loading