mirror of
https://github.com/openai/codex.git
synced 2026-11-13 21:30:47 +00:00
pakrym-oai
3df5087f75
## What changed - Add a `NoiseStreamHandler` abstraction that owns payload encoding, decoding, and connection processing while the virtual stream handles framing, encryption, and multiplexing. - Reassemble authenticated payloads as opaque bytes, with the existing JSON-RPC processor supplied as a handler implementation. - Have the physical relay send a reset after the current stream instance closes, including when its processor exits before the writer task. ## Testing - Verify local JSON-RPC decoding creates the queued request span before queue admission. - Verify a processor exit resets the corresponding Noise harness stream. GitOrigin-RevId: 270befcc8bfbf27e46c84d6a2c6e349cb98ea3f8
42 lines
1.3 KiB
Rust
42 lines
1.3 KiB
Rust
use std::time::Instant;
|
|
|
|
use anyhow::Result;
|
|
use codex_exec_server_protocol::JSONRPCMessage;
|
|
use codex_exec_server_protocol::JSONRPCRequest;
|
|
use codex_exec_server_protocol::RequestId;
|
|
use pretty_assertions::assert_eq;
|
|
|
|
use super::NoiseStreamHandler;
|
|
use crate::connection::JsonRpcConnectionEvent;
|
|
use crate::server::ConnectionProcessor;
|
|
|
|
#[test]
|
|
fn local_decode_starts_request_span_before_queueing() -> Result<()> {
|
|
let _subscriber = tracing::subscriber::set_default(tracing_subscriber::registry());
|
|
tracing::callsite::rebuild_interest_cache();
|
|
|
|
let request = JSONRPCRequest {
|
|
id: RequestId::Integer(1),
|
|
method: "test/queued".to_string(),
|
|
params: None,
|
|
trace: None,
|
|
};
|
|
let before_decode = Instant::now();
|
|
let event = <ConnectionProcessor as NoiseStreamHandler>::decode(
|
|
serde_json::to_vec(&JSONRPCMessage::Request(request.clone()))?.into(),
|
|
)?;
|
|
let after_decode = Instant::now();
|
|
let JsonRpcConnectionEvent::QueuedRequest {
|
|
request: actual_request,
|
|
request_span,
|
|
queued_at,
|
|
} = event
|
|
else {
|
|
panic!("local ingress must create the queued request synchronously");
|
|
};
|
|
assert_eq!(actual_request, request);
|
|
assert!(!request_span.is_disabled());
|
|
assert!((before_decode..=after_decode).contains(&queued_at));
|
|
Ok(())
|
|
}
|