mirror of
https://github.com/openai/codex.git
synced 2026-11-07 20:30:33 +00:00
pash-openai
56032e9732
## Why Shared hosts serve independent chats and their subagents. Larger callback workloads need room for unrelated work and control messages, while gRPC completion bursts can exhaust HTTP/2's small-DATA-frame budget and disconnect shared sessions. ## What changed - Raise the active-cell limit from 128 to 2,048 and the pending delegate callback limit from 1,024 to 4,096. Expand request and transport queues to leave room for completion bursts and control traffic. - Negotiate larger callback budgets through the stdio `delegate-capacity-4096` capability and gRPC `max_pending_delegate_calls` field. Preserve the 1,024-callback budget for legacy clients and enforce both session and shared host limits on gRPC. - Limit concurrent gRPC completion RPCs to 64 across a provider's sessions, allowing queued completions to be canceled without waiting for delivery capacity. ## Testing Add stdio and gRPC integration tests covering 256 cells and 4,096 callbacks across shared sessions while unrelated work completes. Add tests for callback budget isolation and reuse, rejection of zero capacity, and cancellation of queued completions. Include an ignored resource-intensive test for full host capacity. GitOrigin-RevId: e85bfa24cc46cc8a735b464d80adf8066a6a6d3a
79 lines
2.8 KiB
Rust
79 lines
2.8 KiB
Rust
#![cfg(test)]
|
|
//! Negotiated callback budgets protect older clients on a shared host.
|
|
|
|
use codex_code_mode_protocol::grpc as proto;
|
|
use codex_code_mode_protocol::grpc::code_mode_host_server::CodeModeHost;
|
|
use codex_code_mode_protocol::host::LEGACY_MAX_PENDING_DELEGATE_CALLS;
|
|
use futures::StreamExt;
|
|
use pretty_assertions::assert_eq;
|
|
use tonic::Code;
|
|
use tonic::Request;
|
|
|
|
use super::GrpcCodeModeHost;
|
|
use super::tests::open_session;
|
|
|
|
#[tokio::test]
|
|
async fn client_callback_capacity_is_isolated_and_reusable() {
|
|
let host = GrpcCodeModeHost::new();
|
|
let (legacy_id, _legacy_lease) = open_session(&host).await;
|
|
let legacy = host.state.session(&legacy_id).expect("legacy session");
|
|
let legacy_permits = (0..LEGACY_MAX_PENDING_DELEGATE_CALLS)
|
|
.map(|_| legacy.delegate_permit().expect("legacy callback"))
|
|
.collect::<Vec<_>>();
|
|
assert_eq!(
|
|
legacy.delegate_permit().unwrap_err(),
|
|
"code-mode session has too many pending delegate calls"
|
|
);
|
|
|
|
let mut lease = host
|
|
.open_session(Request::new(proto::OpenSessionRequest {
|
|
max_pending_delegate_calls: Some(2),
|
|
..Default::default()
|
|
}))
|
|
.await
|
|
.expect("open capacity-limited session")
|
|
.into_inner();
|
|
let event = lease.next().await.expect("opened event").expect("event");
|
|
let Some(proto::session_event::Event::Opened(opened)) = event.event else {
|
|
panic!("expected session opening");
|
|
};
|
|
let session = host.state.session(&opened.session_id).expect("session");
|
|
let first = session.delegate_permit().expect("first callback");
|
|
let second = session.delegate_permit().expect("second callback");
|
|
assert_eq!(
|
|
session.delegate_permit().unwrap_err(),
|
|
"code-mode session has too many pending delegate calls"
|
|
);
|
|
|
|
drop(first);
|
|
let replacement = session.delegate_permit().expect("reused callback capacity");
|
|
drop(legacy_permits);
|
|
let legacy_replacement = legacy.delegate_permit().expect("reused legacy capacity");
|
|
drop((second, replacement, legacy_replacement));
|
|
host.close_session(Request::new(proto::CloseSessionRequest {
|
|
session_id: opened.session_id,
|
|
}))
|
|
.await
|
|
.expect("close new session");
|
|
host.close_session(Request::new(proto::CloseSessionRequest {
|
|
session_id: legacy_id,
|
|
}))
|
|
.await
|
|
.expect("close legacy session");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn zero_callback_capacity_is_rejected_before_opening_a_session() {
|
|
let host = GrpcCodeModeHost::new();
|
|
let result = host
|
|
.open_session(Request::new(proto::OpenSessionRequest {
|
|
max_pending_delegate_calls: Some(0),
|
|
..Default::default()
|
|
}))
|
|
.await;
|
|
let Err(error) = result else {
|
|
panic!("zero callback capacity must be rejected");
|
|
};
|
|
assert_eq!(error.code(), Code::InvalidArgument);
|
|
}
|