codex_app_server_client/rest/
backend.rs

1use std::{
2    collections::HashMap,
3    sync::{
4        atomic::{AtomicUsize, Ordering},
5        Arc, Mutex as StdMutex,
6    },
7    time::{Duration, Instant},
8};
9
10use tokio::sync::{Mutex, OwnedSemaphorePermit, Semaphore};
11use uuid::Uuid;
12
13use crate::{
14    protocol::{AskForApproval, SandboxMode, ThreadStartParams, TurnInterruptParams},
15    AllowAllApprovalHandler, ApprovalHandler, CodexAppServerClient, CodexSession,
16    CompatibilityReport, DenyAllApprovalHandler, Error, Event, EventCollector,
17    PendingServerRequest, ReadOnlyApprovalHandler, SessionOptions, TextTurnResult,
18};
19
20use super::types::{
21    session_options_from, RestApprovalPolicy, RestBackend, RestCallRequest, RestCallResponse,
22    RestClientOptions, RestError, RestErrorReplyRequest, RestEventResponse, RestFuture, RestLimits,
23    RestRequestReplyResponse, RestRequestReplyResultRequest, RestResult, RestSessionCreateRequest,
24    RestSessionCreateResponse, RestSessionSummary, RestStatusResponse, RestTextTurnRequest,
25    RestTextTurnResponse,
26};
27
28/// Production REST backend.
29///
30/// One-shot calls create a short-lived Codex session. Stateful bridge calls use
31/// sessions created by `POST /v1/sessions`.
32#[derive(Clone)]
33pub struct CodexRestBackend {
34    sessions: Arc<CodexRestSessions>,
35    limits: RestLimits,
36    compatibility: Arc<StdMutex<Option<CachedCompatibility>>>,
37    session_call_gate: Arc<Semaphore>,
38}
39
40impl Default for CodexRestBackend {
41    fn default() -> Self {
42        Self::with_limits(RestLimits::default())
43    }
44}
45
46struct CodexRestSessions {
47    sessions: Mutex<HashMap<String, Arc<CodexRestSession>>>,
48    slots: Arc<Semaphore>,
49    /// When [`CodexRestBackend::maybe_prune_idle_sessions`] last actually
50    /// scanned. See [`PRUNE_MIN_INTERVAL`].
51    last_prune: StdMutex<Instant>,
52}
53
54/// Floor on how often the *opportunistic* idle-session sweep on the hot path
55/// ([`CodexRestBackend::session`]) actually scans.
56///
57/// That sweep used to run on every single `session()` call - i.e. on every
58/// event poll and every session call - and each run is an O(sessions) scan
59/// behind the global session lock. A single SSE stream draining a burst of
60/// buffered events calls `session()` once per event, so the sweep's cost
61/// scaled with event throughput while doing no useful work: sessions don't
62/// become idle any faster because someone is polling a different one.
63///
64/// This only throttles the opportunistic sweep. The two callers whose
65/// correctness depends on a fresh view - `create_session` (which sweeps to
66/// reclaim a session slot before taking one) and `list_sessions` (which must
67/// not report a session it is about to drop) - still sweep unconditionally,
68/// so throttling here cannot make the backend hand out a stale slot or list a
69/// dead session.
70///
71/// The cost of the throttle is that a session can outlive its
72/// `idle_session_ttl` by up to this interval before hot-path traffic reaps
73/// it. That is immaterial against the 30-minute default, and the interval is
74/// capped by `idle_session_ttl` itself so a deliberately tiny TTL still
75/// behaves as configured.
76const PRUNE_MIN_INTERVAL: Duration = Duration::from_secs(1);
77
78struct CodexRestSession {
79    client: CodexAppServerClient,
80    session: Mutex<CodexSession>,
81    pending_requests: Mutex<HashMap<String, PendingRestRequest>>,
82    last_used: StdMutex<Instant>,
83    active_operations: AtomicUsize,
84    call_gate: Arc<Semaphore>,
85    _session_slot: OwnedSemaphorePermit,
86}
87
88struct SessionLease {
89    session: Arc<CodexRestSession>,
90}
91
92struct PendingRestRequest {
93    request: PendingServerRequest,
94    expires_at: Instant,
95}
96
97struct CachedCompatibility {
98    report: CompatibilityReport,
99    expires_at: Instant,
100}
101
102impl CodexRestBackend {
103    pub fn with_limits(limits: RestLimits) -> Self {
104        let max_sessions = limits.max_sessions;
105        let max_session_call_concurrency = limits.max_session_call_concurrency;
106        Self {
107            sessions: Arc::new(CodexRestSessions {
108                sessions: Mutex::new(HashMap::new()),
109                slots: Arc::new(Semaphore::new(max_sessions)),
110                last_prune: StdMutex::new(Instant::now()),
111            }),
112            limits,
113            compatibility: Arc::default(),
114            session_call_gate: Arc::new(Semaphore::new(max_session_call_concurrency)),
115        }
116    }
117
118    fn cached_compatibility(&self, now: Instant) -> Option<CompatibilityReport> {
119        let cached = self
120            .compatibility
121            .lock()
122            .unwrap_or_else(|poisoned| poisoned.into_inner());
123        cached
124            .as_ref()
125            .filter(|cached| cached.expires_at > now)
126            .map(|cached| cached.report.clone())
127    }
128
129    fn store_compatibility(&self, report: CompatibilityReport, now: Instant) {
130        let mut cached = self
131            .compatibility
132            .lock()
133            .unwrap_or_else(|poisoned| poisoned.into_inner());
134        *cached = Some(CachedCompatibility {
135            report,
136            expires_at: now + self.limits.compatibility_ttl,
137        });
138    }
139
140    async fn session(&self, session_id: &str) -> RestResult<SessionLease> {
141        self.maybe_prune_idle_sessions().await;
142        let session = self
143            .sessions
144            .sessions
145            .lock()
146            .await
147            .get(session_id)
148            .cloned()
149            .ok_or_else(|| RestError::NotFound(format!("session `{session_id}` was not found")))?;
150        session.touch().await;
151        session.active_operations.fetch_add(1, Ordering::AcqRel);
152        Ok(SessionLease { session })
153    }
154
155    /// [`Self::prune_idle_sessions`], but at most once per
156    /// [`PRUNE_MIN_INTERVAL`]. Use this on paths that sweep only
157    /// opportunistically; use `prune_idle_sessions` directly where a fresh
158    /// view is load-bearing.
159    async fn maybe_prune_idle_sessions(&self) {
160        let now = Instant::now();
161        // The interval is capped by the TTL itself so that an operator who
162        // deliberately sets a sub-second `idle_session_ttl` still gets
163        // sweeping at the cadence they asked for rather than this floor.
164        let interval = PRUNE_MIN_INTERVAL.min(self.limits.idle_session_ttl);
165        {
166            let mut last_prune = self
167                .sessions
168                .last_prune
169                .lock()
170                .unwrap_or_else(|poisoned| poisoned.into_inner());
171            if now.duration_since(*last_prune) < interval {
172                return;
173            }
174            // Claim the slot before releasing the lock and before doing the
175            // scan, so concurrent callers arriving in the same instant don't
176            // all decide to sweep.
177            *last_prune = now;
178        }
179        self.prune_idle_sessions().await;
180    }
181
182    async fn prune_idle_sessions(&self) {
183        let now = Instant::now();
184        let mut sessions = self.sessions.sessions.lock().await;
185        sessions.retain(|_, session| {
186            let is_active = session.active_operations.load(Ordering::Acquire) > 0;
187            let last_used = *session
188                .last_used
189                .lock()
190                .unwrap_or_else(|poisoned| poisoned.into_inner());
191            is_active || now.duration_since(last_used) < self.limits.idle_session_ttl
192        });
193    }
194}
195
196/// Builds the [`SessionOptions`] used to spawn every session this backend
197/// creates - one-shot text-turn sessions, stateful bridge sessions from
198/// `POST /v1/sessions`, and session-less raw calls alike - threading
199/// [`RestLimits::events_channel_capacity`] through via
200/// [`SessionOptions::with_events_capacity`] so that configured limit
201/// actually bounds the spawned session's event channel instead of every
202/// session silently falling back to [`crate::DEFAULT_EVENTS_CHANNEL_CAPACITY`].
203/// Centralized here (rather than inlined at each call site) so the three
204/// `RestBackend` methods that spawn sessions can't drift out of sync with
205/// this limit - see the unit test below for direct proof this actually
206/// applies it.
207fn session_options_with_limits(
208    client: Option<RestClientOptions>,
209    default_name: &str,
210    limits: &RestLimits,
211) -> SessionOptions {
212    session_options_from(client, default_name).with_events_capacity(limits.events_channel_capacity)
213}
214
215fn rest_text_turn_thread_params(model: Option<String>) -> ThreadStartParams {
216    let mut params = ThreadStartParams::new().ephemeral(true);
217    params.model = model;
218    params.approval_policy = Some(AskForApproval::Untrusted);
219    params.sandbox = Some(SandboxMode::ReadOnly);
220    params
221}
222
223async fn run_limited_text_turn<H>(
224    session: &mut CodexSession,
225    thread_params: ThreadStartParams,
226    prompt: String,
227    handler: &H,
228    limits: &RestLimits,
229) -> RestResult<TextTurnResult>
230where
231    H: ApprovalHandler,
232{
233    let thread = session.start_thread(thread_params).await?;
234    let turn = session.send_text_turn(&thread.thread.id, prompt).await?;
235    let thread_id = thread.thread.id.clone();
236    let turn_id = turn.turn.id.clone();
237    let collect_result = tokio::time::timeout(
238        limits.max_text_turn_duration,
239        collect_turn_with_output_limit(
240            session,
241            &thread_id,
242            &turn_id,
243            handler,
244            limits.max_text_turn_output_bytes,
245        ),
246    )
247    .await;
248    let events = match collect_result {
249        Ok(Ok(events)) => events,
250        Ok(Err(error)) => return Err(error),
251        Err(_elapsed) => {
252            interrupt_turn(session, &thread_id, &turn_id).await;
253            return Err(RestError::TimedOut(format!(
254                "text turn did not complete within {:?}",
255                limits.max_text_turn_duration
256            )));
257        }
258    };
259    Ok(TextTurnResult {
260        thread,
261        turn,
262        events,
263    })
264}
265
266async fn collect_turn_with_output_limit<H>(
267    session: &mut CodexSession,
268    thread_id: &str,
269    turn_id: &str,
270    handler: &H,
271    max_output_bytes: usize,
272) -> RestResult<EventCollector>
273where
274    H: ApprovalHandler,
275{
276    let mut collector = EventCollector::for_turn(thread_id, turn_id);
277    while !collector.is_complete() {
278        let Some(notification) = session.next_notification(handler).await else {
279            return Err(RestError::Client(Error::TransportClosed));
280        };
281        collector.observe_notification(&notification);
282        if collector.output_bytes() > max_output_bytes {
283            interrupt_turn(session, thread_id, turn_id).await;
284            return Err(RestError::PayloadTooLarge(format!(
285                "text turn output exceeded {max_output_bytes} bytes"
286            )));
287        }
288    }
289    Ok(collector)
290}
291
292async fn interrupt_turn(session: &CodexSession, thread_id: &str, turn_id: &str) {
293    let _ = session
294        .client()
295        .turn_interrupt(TurnInterruptParams {
296            thread_id: thread_id.to_owned(),
297            turn_id: turn_id.to_owned(),
298        })
299        .await;
300}
301
302impl CodexRestSession {
303    async fn touch(&self) {
304        *self
305            .last_used
306            .lock()
307            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Instant::now();
308    }
309
310    async fn prune_expired_pending(&self, now: Instant) {
311        prune_expired_pending_requests(&self.pending_requests, now).await;
312    }
313
314    async fn take_pending_request(&self, request_key: &str) -> RestResult<PendingServerRequest> {
315        take_pending_request(&self.pending_requests, request_key).await
316    }
317}
318
319impl std::ops::Deref for SessionLease {
320    type Target = CodexRestSession;
321
322    fn deref(&self) -> &Self::Target {
323        &self.session
324    }
325}
326
327impl Drop for SessionLease {
328    fn drop(&mut self) {
329        self.session
330            .active_operations
331            .fetch_sub(1, Ordering::AcqRel);
332        *self
333            .session
334            .last_used
335            .lock()
336            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Instant::now();
337    }
338}
339
340async fn prune_expired_pending_requests(
341    pending_requests: &Mutex<HashMap<String, PendingRestRequest>>,
342    now: Instant,
343) {
344    let mut pending = pending_requests.lock().await;
345    let expired = pending
346        .iter()
347        .filter(|(_, request)| request.expires_at <= now)
348        .map(|(key, _)| key.clone())
349        .collect::<Vec<_>>();
350    for key in expired {
351        pending.remove(&key);
352    }
353}
354
355async fn take_pending_request(
356    pending_requests: &Mutex<HashMap<String, PendingRestRequest>>,
357    request_key: &str,
358) -> RestResult<PendingServerRequest> {
359    let now = Instant::now();
360    let mut pending = pending_requests.lock().await;
361    if pending
362        .get(request_key)
363        .is_some_and(|request| request.expires_at <= now)
364    {
365        pending.remove(request_key);
366        return Err(RestError::Gone(format!(
367            "request `{request_key}` has expired"
368        )));
369    }
370
371    let expired = pending
372        .iter()
373        .filter(|(_, request)| request.expires_at <= now)
374        .map(|(key, _)| key.clone())
375        .collect::<Vec<_>>();
376    for key in expired {
377        pending.remove(&key);
378    }
379
380    pending
381        .remove(request_key)
382        .map(|request| request.request)
383        .ok_or_else(|| RestError::NotFound(format!("request `{request_key}` was not found")))
384}
385
386impl RestBackend for CodexRestBackend {
387    fn compatibility_report(&self) -> RestFuture<CompatibilityReport> {
388        let backend = self.clone();
389        Box::pin(async move {
390            let now = Instant::now();
391            if let Some(report) = backend.cached_compatibility(now) {
392                return Ok(report);
393            }
394
395            let report = tokio::task::spawn_blocking(CompatibilityReport::current)
396                .await
397                .map_err(|error| {
398                    RestError::Internal(format!("compatibility check task failed: {error}"))
399                })?;
400            backend.store_compatibility(report.clone(), now);
401            Ok(report)
402        })
403    }
404
405    fn run_text_turn(&self, request: RestTextTurnRequest) -> RestFuture<RestTextTurnResponse> {
406        let limits = self.limits.clone();
407        Box::pin(async move {
408            let session_options = session_options_with_limits(
409                request.client.clone(),
410                "codex_app_server_rest",
411                &limits,
412            );
413            let approval_policy = request.approval_policy.unwrap_or_default();
414            let thread_params = rest_text_turn_thread_params(request.model.clone());
415            let prompt = request.prompt.clone();
416            let mut session = CodexSession::spawn(session_options).await?;
417
418            let result = match approval_policy {
419                RestApprovalPolicy::DenyAll => {
420                    run_limited_text_turn(
421                        &mut session,
422                        thread_params,
423                        prompt,
424                        &DenyAllApprovalHandler::default(),
425                        &limits,
426                    )
427                    .await?
428                }
429                RestApprovalPolicy::ReadOnly => {
430                    run_limited_text_turn(
431                        &mut session,
432                        thread_params,
433                        prompt,
434                        &ReadOnlyApprovalHandler,
435                        &limits,
436                    )
437                    .await?
438                }
439                RestApprovalPolicy::AllowAll => {
440                    run_limited_text_turn(
441                        &mut session,
442                        thread_params,
443                        prompt,
444                        &AllowAllApprovalHandler,
445                        &limits,
446                    )
447                    .await?
448                }
449            };
450            Ok(RestTextTurnResponse::from(result))
451        })
452    }
453
454    fn create_session(
455        &self,
456        request: RestSessionCreateRequest,
457    ) -> RestFuture<RestSessionCreateResponse> {
458        let backend = self.clone();
459        Box::pin(async move {
460            backend.prune_idle_sessions().await;
461            let session_slot =
462                backend
463                    .sessions
464                    .slots
465                    .clone()
466                    .try_acquire_owned()
467                    .map_err(|_| {
468                        RestError::RateLimited(format!(
469                            "maximum REST session count ({}) reached",
470                            backend.limits.max_sessions
471                        ))
472                    })?;
473
474            let session = CodexSession::spawn(session_options_with_limits(
475                request.client,
476                "codex_app_server_rest_session",
477                &backend.limits,
478            ))
479            .await?;
480            let initialize_response = serde_json::to_value(session.initialize_response())?;
481            let client = session.client().clone();
482            let session_id = format!("session-{}", Uuid::new_v4().simple());
483            let rest_session = Arc::new(CodexRestSession {
484                client,
485                session: Mutex::new(session),
486                pending_requests: Mutex::new(HashMap::new()),
487                last_used: StdMutex::new(Instant::now()),
488                active_operations: AtomicUsize::new(0),
489                call_gate: Arc::new(Semaphore::new(
490                    backend.limits.max_session_call_concurrency_per_session,
491                )),
492                _session_slot: session_slot,
493            });
494            let mut sessions = backend.sessions.sessions.lock().await;
495            sessions.insert(session_id.clone(), rest_session);
496            Ok(RestSessionCreateResponse {
497                session_id,
498                initialize_response,
499            })
500        })
501    }
502
503    fn list_sessions(&self) -> RestFuture<Vec<RestSessionSummary>> {
504        let backend = self.clone();
505        Box::pin(async move {
506            backend.prune_idle_sessions().await;
507            let mut sessions = backend
508                .sessions
509                .sessions
510                .lock()
511                .await
512                .keys()
513                .map(|session_id| RestSessionSummary {
514                    session_id: session_id.clone(),
515                })
516                .collect::<Vec<_>>();
517            sessions.sort_by(|left, right| left.session_id.cmp(&right.session_id));
518            Ok(sessions)
519        })
520    }
521
522    fn delete_session(&self, session_id: String) -> RestFuture<RestStatusResponse> {
523        let backend = self.clone();
524        Box::pin(async move {
525            let removed = backend
526                .sessions
527                .sessions
528                .lock()
529                .await
530                .remove(&session_id)
531                .is_some();
532            if removed {
533                Ok(RestStatusResponse {
534                    status: "deleted".to_owned(),
535                })
536            } else {
537                Err(RestError::NotFound(format!(
538                    "session `{session_id}` was not found"
539                )))
540            }
541        })
542    }
543
544    fn call_method(&self, request: RestCallRequest) -> RestFuture<RestCallResponse> {
545        let backend = self.clone();
546        Box::pin(async move {
547            let method = request.method.clone();
548            let result = if let Some(session_id) = request.session_id.as_deref() {
549                let session = backend.session(session_id).await?;
550                let _global_permit = backend
551                    .session_call_gate
552                    .clone()
553                    .try_acquire_owned()
554                    .map_err(|_| {
555                        RestError::RateLimited(format!(
556                            "maximum REST session call concurrency ({}) reached",
557                            backend.limits.max_session_call_concurrency
558                        ))
559                    })?;
560                let _session_permit =
561                    session.call_gate.clone().try_acquire_owned().map_err(|_| {
562                        RestError::RateLimited(format!(
563                            "maximum REST session call concurrency per session ({}) reached",
564                            backend.limits.max_session_call_concurrency_per_session
565                        ))
566                    })?;
567                session
568                    .client
569                    .call_raw_method(method.clone(), request.params)
570                    .await?
571            } else {
572                let session = CodexSession::spawn(session_options_with_limits(
573                    request.client,
574                    "codex_app_server_rest_call",
575                    &backend.limits,
576                ))
577                .await?;
578                session
579                    .client()
580                    .call_raw_method(method.clone(), request.params)
581                    .await?
582            };
583            Ok(RestCallResponse { method, result })
584        })
585    }
586
587    fn poll_event(
588        &self,
589        session_id: String,
590        timeout_ms: Option<u64>,
591    ) -> RestFuture<RestEventResponse> {
592        let limits = self.limits.clone();
593        let backend = self.clone();
594        Box::pin(async move {
595            let session = backend.session(&session_id).await?;
596            let timeout = Duration::from_millis(timeout_ms.unwrap_or(30_000));
597            let event = tokio::time::timeout(timeout, async {
598                let mut session_guard = session.session.session.lock().await;
599                session_guard.next_event().await
600            })
601            .await;
602
603            match event {
604                Ok(Some(Event::Notification(notification))) => Ok(RestEventResponse::notification(
605                    serde_json::to_value(notification)?,
606                )),
607                Ok(Some(Event::Request(request))) => {
608                    let now = Instant::now();
609                    session.prune_expired_pending(now).await;
610                    let request_id = serde_json::to_value(request.id())?;
611                    let method = request.method_name().to_owned();
612                    let request_value = serde_json::to_value(&request.request)?;
613                    let mut pending = session.pending_requests.lock().await;
614                    if pending.len() >= limits.max_pending_requests_per_session {
615                        let _ = request.respond_error(
616                            -32000,
617                            "REST pending request limit reached",
618                            None,
619                        );
620                        return Err(RestError::RateLimited(format!(
621                            "maximum pending request count ({}) reached",
622                            limits.max_pending_requests_per_session
623                        )));
624                    }
625                    let reply_deadline = request.reply_deadline();
626                    if reply_deadline <= now {
627                        let _ = request.respond_error(
628                            -32000,
629                            "REST request reply deadline already expired",
630                            None,
631                        );
632                        return Err(RestError::Gone(
633                            "server request can no longer be answered".to_owned(),
634                        ));
635                    }
636                    let expires_at = (now + limits.pending_request_ttl).min(reply_deadline);
637                    let request_key = format!("request-{}", Uuid::new_v4().simple());
638                    pending.insert(
639                        request_key.clone(),
640                        PendingRestRequest {
641                            request,
642                            expires_at,
643                        },
644                    );
645                    Ok(RestEventResponse::request(
646                        request_key,
647                        request_id,
648                        method,
649                        request_value,
650                    ))
651                }
652                Ok(Some(Event::Closed)) | Ok(None) => Ok(RestEventResponse::closed()),
653                Err(_elapsed) => Ok(RestEventResponse::timeout()),
654            }
655        })
656    }
657
658    fn reply_request_result(
659        &self,
660        session_id: String,
661        request_key: String,
662        body: RestRequestReplyResultRequest,
663    ) -> RestFuture<RestRequestReplyResponse> {
664        let backend = self.clone();
665        Box::pin(async move {
666            let session = backend.session(&session_id).await?;
667            let pending = session.take_pending_request(&request_key).await?;
668            pending.respond(body.result).map_err(|_| {
669                RestError::Gone(format!("request `{request_key}` can no longer be answered"))
670            })?;
671            Ok(RestRequestReplyResponse {
672                status: "ok".to_owned(),
673            })
674        })
675    }
676
677    fn reply_request_error(
678        &self,
679        session_id: String,
680        request_key: String,
681        body: RestErrorReplyRequest,
682    ) -> RestFuture<RestRequestReplyResponse> {
683        let backend = self.clone();
684        Box::pin(async move {
685            let session = backend.session(&session_id).await?;
686            let pending = session.take_pending_request(&request_key).await?;
687            pending
688                .respond_error(body.code, body.message, body.data)
689                .map_err(|_| {
690                    RestError::Gone(format!("request `{request_key}` can no longer be answered"))
691                })?;
692            Ok(RestRequestReplyResponse {
693                status: "ok".to_owned(),
694            })
695        })
696    }
697}
698
699#[cfg(test)]
700mod tests {
701    use super::*;
702    use crate::protocol::{CurrentTimeReadParams, RequestId, ServerRequest};
703
704    /// The hot-path sweep must be throttled: `session()` runs on every event
705    /// poll and every session call, and each unthrottled sweep is an
706    /// O(sessions) scan behind the global session lock.
707    ///
708    /// Asserts against `last_prune` rather than a scan counter because the
709    /// throttle's whole job is to *not* take the session lock, which leaves
710    /// no other observable trace.
711    #[tokio::test]
712    async fn hot_path_sweep_is_throttled_but_correctness_paths_are_not() {
713        let backend = CodexRestBackend::with_limits(RestLimits::default());
714        let stamp = || {
715            *backend
716                .sessions
717                .last_prune
718                .lock()
719                .unwrap_or_else(|poisoned| poisoned.into_inner())
720        };
721
722        // A fresh backend stamps `last_prune` at construction, so the first
723        // hot-path call is inside the interval and must not sweep.
724        let before = stamp();
725        for _ in 0..64 {
726            // `session()` on a missing id still runs the sweep before failing
727            // the lookup, which is the path under test.
728            let _ = backend.session("no-such-session").await;
729        }
730        assert_eq!(
731            stamp(),
732            before,
733            "64 hot-path lookups inside the throttle interval should have swept zero times"
734        );
735
736        // Once the interval has elapsed, the next hot-path call sweeps again -
737        // the throttle delays the sweep, it does not disable it.
738        *backend
739            .sessions
740            .last_prune
741            .lock()
742            .unwrap_or_else(|poisoned| poisoned.into_inner()) =
743            Instant::now() - (PRUNE_MIN_INTERVAL * 2);
744        let stale = stamp();
745        let _ = backend.session("no-such-session").await;
746        assert!(
747            stamp() > stale,
748            "a hot-path lookup after the interval elapsed should sweep"
749        );
750
751        // `list_sessions` must never report a session it is about to drop, so
752        // it sweeps unconditionally regardless of when the last sweep ran.
753        let just_swept = stamp();
754        let sessions = backend.list_sessions().await.expect("list_sessions");
755        assert!(sessions.is_empty());
756        assert!(
757            stamp() >= just_swept,
758            "list_sessions must sweep unconditionally, not defer to the throttle"
759        );
760    }
761
762    /// Regression test: `session_options_with_limits` is the single place
763    /// all three `RestBackend` methods that spawn a session
764    /// (`run_text_turn`, `create_session`, `call_method`'s session-less
765    /// branch) go through to build `SessionOptions`. Exercises that exact
766    /// production function directly with a non-default
767    /// `events_channel_capacity` to prove the limit actually reaches the
768    /// `SessionOptions` handed to `CodexSession::spawn` - a knob nothing
769    /// reads is worse than no knob. (The complementary half of this proof -
770    /// that a given `SessionOptions::events_capacity` actually bounds the
771    /// resulting session's event channel - is covered by
772    /// `client.rs`'s `connect_streams_with_events_capacity_bounds_the_channel_to_the_requested_size`.)
773    #[test]
774    fn session_options_with_limits_applies_the_configured_events_channel_capacity() {
775        let limits = RestLimits {
776            events_channel_capacity: 7,
777            ..RestLimits::default()
778        };
779        assert_ne!(
780            limits.events_channel_capacity,
781            RestLimits::default().events_channel_capacity,
782            "test setup bug: this must be a non-default value to prove anything"
783        );
784
785        let options = session_options_with_limits(None, "test-default-name", &limits);
786
787        assert_eq!(options.events_capacity, Some(7));
788    }
789
790    #[tokio::test]
791    async fn expired_pending_request_returns_gone_and_removes_the_key() {
792        let pending_requests = Mutex::new(HashMap::from([(
793            "request-1".to_owned(),
794            PendingRestRequest {
795                request: pending_current_time_request(),
796                expires_at: Instant::now() - Duration::from_secs(1),
797            },
798        )]));
799
800        let err = take_pending_request(&pending_requests, "request-1")
801            .await
802            .expect_err("expired request key should be rejected");
803
804        assert!(matches!(err, RestError::Gone(_)));
805        assert!(pending_requests.lock().await.is_empty());
806    }
807
808    #[tokio::test]
809    async fn taking_pending_request_prunes_other_expired_keys() {
810        let pending_requests = Mutex::new(HashMap::from([
811            (
812                "expired".to_owned(),
813                PendingRestRequest {
814                    request: pending_current_time_request(),
815                    expires_at: Instant::now() - Duration::from_secs(1),
816                },
817            ),
818            (
819                "fresh".to_owned(),
820                PendingRestRequest {
821                    request: pending_current_time_request(),
822                    expires_at: Instant::now() + Duration::from_secs(60),
823                },
824            ),
825        ]));
826
827        let request = take_pending_request(&pending_requests, "fresh")
828            .await
829            .expect("fresh request key should be returned");
830
831        assert_eq!(request.method_name(), "currentTime/read");
832        assert!(pending_requests.lock().await.is_empty());
833    }
834
835    fn pending_current_time_request() -> PendingServerRequest {
836        PendingServerRequest::for_test(ServerRequest::CurrentTimeRead {
837            id: RequestId::Int64(7),
838            params: CurrentTimeReadParams {
839                thread_id: "thread-test".to_owned(),
840            },
841        })
842    }
843}
codex_app_server_client/rest/backend.rs

codex_app_server_client/rest/
backend.rs
