Add exact OpenAI request lifecycle observability - #1234
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (11)
🚧 Files skipped from review as they are similar to previous changes (11)
📝 WalkthroughWalkthroughThe OpenAI frontend now propagates request identity, trust state, timing, usage, and lifecycle telemetry. The generation backend adds trusted-session identity reuse, per-session serialization, cancellation-aware queue admission, and lifecycle-aware usage emission. ChangesOpenAI request observability and generation admission
Estimated code review effort: 5 (Critical) | ~100 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Router
participant GenerationAdmission
participant Backend
participant SSE
Client->>Router: submit request with request context
Router->>GenerationAdmission: request global and trusted-session permits
GenerationAdmission->>Backend: run blocking or streaming generation
Backend-->>Router: return response or stream events
Router->>SSE: finalize stream lifecycle and usage
SSE-->>Client: return response events
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 2📝 Generate docstrings 💡
🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
This pull request is currently a draft. Reviews will not take place until the PR is marked as ready for review. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
crates/openai-frontend/src/router.rs (2)
952-973: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for the terminal-event guarantee.
StreamLifecycleandRequestObservationimplement the central claim of this PR: exactly onerequest_finishedevent per request. The only added test checksrequest_error_outcome. No test exercisesfinish_natural, the idempotentterminalflag, or theDroppath that emits"cancelled".Add unit tests that assert the following:
finish_naturalemits"success"whenstream_erroris false, and"backend_error"when it is true.- A second
finishcall after any terminal call emits nothing.- Dropping a
RequestObservationafter an explicitfinishemits nothing.
tracing-testor a customtracing::Subscriberthat records events can capture the emitted fields.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/openai-frontend/src/router.rs` around lines 952 - 973, Add unit coverage for StreamLifecycle and RequestObservation using tracing-test or a recording tracing::Subscriber: verify finish_natural emits success without stream_error and backend_error with it, repeated finish calls after any terminal completion emit only once, and dropping RequestObservation after an explicit finish emits no additional request_finished event.
206-233: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the repeated streaming instrumentation setup. The three streaming handlers now contain the same copied block: build the request-scoped context, take the cancellation token, allocate the
stream_errorflag, callbackend_call_with_cancellation, finish the observation on error, and wrap the result withobserve_backend_stream. Only the operation label and the backend future differ. A future change to the telemetry contract must be applied in three places, and a missed site silently drops a lifecycle event.Extract a helper that takes the state, the operation label, the request ID, the observation, and the backend future, and returns the observed stream plus the cancellation token and the
stream_errorhandle.
crates/openai-frontend/src/router.rs#L206-L233: replace the copied setup inchat_completionswith a call to the new helper, passing"chat_completion_stream".crates/openai-frontend/src/router.rs#L302-L328: replace the copied setup inresponseswith a call to the new helper, passing"responses_stream".crates/openai-frontend/src/router.rs#L563-L587: replace the copied setup incompletionswith a call to the new helper, passing"completion_stream".🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/openai-frontend/src/router.rs` around lines 206 - 233, Extract the duplicated streaming instrumentation into a shared helper that accepts state, operation label, request ID, observation, and backend future, performs cancellation-aware execution and error observation, then returns the observed stream, cancellation token, and stream_error handle. Update chat_completions at crates/openai-frontend/src/router.rs#L206-L233 to pass "chat_completion_stream", responses at crates/openai-frontend/src/router.rs#L302-L328 to pass "responses_stream", and completions at crates/openai-frontend/src/router.rs#L563-L587 to pass "completion_stream"; remove the copied setup from each handler.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/openai-frontend/src/router.rs`:
- Around line 762-768: Update request_error_outcome to classify 4xx OpenAiError
statuses as "client_error", preserve "timeout" for gateway timeouts, and keep
other failures as "backend_error". Extend the existing request_error_outcome
test with the OpenAiError::model_not_found case and assert the client_error
outcome.
---
Nitpick comments:
In `@crates/openai-frontend/src/router.rs`:
- Around line 952-973: Add unit coverage for StreamLifecycle and
RequestObservation using tracing-test or a recording tracing::Subscriber: verify
finish_natural emits success without stream_error and backend_error with it,
repeated finish calls after any terminal completion emit only once, and dropping
RequestObservation after an explicit finish emits no additional request_finished
event.
- Around line 206-233: Extract the duplicated streaming instrumentation into a
shared helper that accepts state, operation label, request ID, observation, and
backend future, performs cancellation-aware execution and error observation,
then returns the observed stream, cancellation token, and stream_error handle.
Update chat_completions at crates/openai-frontend/src/router.rs#L206-L233 to
pass "chat_completion_stream", responses at
crates/openai-frontend/src/router.rs#L302-L328 to pass "responses_stream", and
completions at crates/openai-frontend/src/router.rs#L563-L587 to pass
"completion_stream"; remove the copied setup from each handler.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 3403a44d-e96c-4245-90da-a6e0969510e6
📒 Files selected for processing (2)
crates/openai-frontend/src/backend.rscrates/openai-frontend/src/router.rs
|
@ndizazzo sure thing! Could you have your agent update this PR when your stuff is ready? This was an automated pr. |
|
🤖 Reviewed locally on Two things I checked that hold up well: the events are off by default (I probed the filter — One substantive issue, then small stuff. 1.
|
i386
left a comment
i386
left a comment
There was a problem hiding this comment.
Reviewed against Cacheline observability contract: the lifecycle events use the authoritative request ID, terminal outcomes are emitted once for unary and streaming paths, and the existing CI suite is green. No blocking findings from this review.
i386
left a comment
i386
left a comment
There was a problem hiding this comment.
Reviewed against Cacheline observability contract: the lifecycle events use the authoritative request ID, terminal outcomes are emitted once for unary and streaming paths, and the current CI suite is green. No blocking findings from this review.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/skippy-server/src/frontend/generation/cache_hints.rs`:
- Around line 54-68: Update request_id construction in the Self initialization
to include a process-instance nonce or authoritative globally unique request
identifier, ensuring identical agent_session_id and sequence values across
restarts or replicas cannot collide. Keep session_id derived only from the
stable session_label so prefix KV-cache reuse remains unchanged, and add
coverage for reset- or replica-equivalent request sequences.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 9e1fb86e-1f06-4954-a42d-0070e1392cd0
📒 Files selected for processing (2)
crates/openai-frontend/src/router.rscrates/skippy-server/src/frontend/generation/cache_hints.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- crates/openai-frontend/src/router.rs
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (4)
crates/openai-frontend/src/router.rs (3)
1145-1165: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider an outcome enum instead of
&'static strcomparison.
drop_outcomereturns a string, andfinish_dropcompares it to the literal"success".finish_naturalrepeats the same literal. If a future change renames an outcome, the compiler will not catch the stale comparison, and completion logging will silently stop.An enum with a
as_str()method keeps the log field values and makes the match exhaustive.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/openai-frontend/src/router.rs` around lines 1145 - 1165, Introduce an outcome enum for the values produced by drop_outcome, with an as_str() method preserving the existing log strings. Update drop_outcome, finish_drop, and finish_natural to use the enum, replacing literal success comparisons with exhaustive enum matching while passing as_str() to observation.finish.
211-217: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider extracting the streamed lifecycle scaffolding into one helper.
chat_completions,responses, andcompletionseach create the same four cells (stream_error,stream_completed,response_completed_logged,stream_usage), then buildStreamCompletionandStreamLifecyclewith identical wiring. The only per-route differences are the operation label and the event stream.A single constructor reduces the risk that one route later forgets a cell. This is a follow-up refactor; the current wiring is correct.
♻️ Sketch of a shared scaffolding type
struct StreamTelemetry { stream_error: Arc<AtomicBool>, stream_completed: Arc<AtomicBool>, response_completed_logged: Arc<AtomicBool>, stream_usage: Arc<Mutex<Option<Usage>>>, } impl StreamTelemetry { fn new() -> Self { /* all four cells */ } fn lifecycle( &self, observation: RequestObservation, context: OpenAiRequestContext, operation: &'static str, ) -> StreamLifecycle { StreamLifecycle::new( observation, self.stream_error.clone(), self.stream_completed.clone(), ) .with_completion(StreamCompletion::new( context, operation, self.stream_usage.clone(), self.response_completed_logged.clone(), )) } }Also applies to: 269-280
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/openai-frontend/src/router.rs` around lines 211 - 217, Extract the repeated streamed lifecycle setup from chat_completions, responses, and completions into a shared StreamTelemetry type with a constructor for the four telemetry cells and a lifecycle method that builds the existing StreamLifecycle/StreamCompletion wiring. Update each route to use this helper, passing only its operation label and event stream while preserving current behavior.
1384-1524: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd an assertion on
request_finishedfor the streamed lifecycle.The new tests count
response_completedevents only. No test asserts that a streamed request emits exactly onerequest_finishedevent with the expected outcome. The PR objectives name this assertion as the requested coverage for drop classification.
EventCapturealready captures theoutcomefield, so the check is small.💚 Sketch of the missing assertion
#[test] fn streamed_lifecycle_emits_one_classified_request_finished() { let events = Arc::new(Mutex::new(Vec::new())); let mut lifecycle = StreamLifecycle::new( RequestObservation::new("request-lifecycle".to_owned(), "chat_completion"), Arc::new(AtomicBool::new(false)), Arc::new(AtomicBool::new(true)), ); tracing::subscriber::with_default( EventCapture { events: events.clone() }, || { lifecycle.finish_natural(); lifecycle.finish_drop(false); }, ); let events = events.lock().expect("event capture lock poisoned"); let finished: Vec<_> = events .iter() .filter(|event| event.get("event").is_some_and(|value| value == "request_finished")) .collect(); assert_eq!(finished.len(), 1); assert_eq!(finished[0]["outcome"], "success"); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/openai-frontend/src/router.rs` around lines 1384 - 1524, Extend the streamed lifecycle coverage in the existing lifecycle tests by asserting that calling finish_natural followed by finish_drop emits exactly one request_finished event. Verify its outcome is success, using EventCapture’s captured outcome field, while preserving the existing response_completed assertions.crates/skippy-server/src/frontend/backend.rs (1)
865-945: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a test that two concurrent turns of one trusted session serialize.
The current tests cover lease counting, entry replacement, untrusted bypass, and spoofed metadata. They do not cover the main behavior of the change:
acquire_generation_admissionmust let only one turn of a trusted session run at a time, and it must respect the shared deadline and cancellation. Add an async test that acquires the session permit, then confirms a secondacquire_generation_admissionfor the sameagent_session_idwaits, and that cancellation returnsrequest cancelled.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/skippy-server/src/frontend/backend.rs` around lines 865 - 945, Add an async test alongside the existing session-registry tests that creates trusted generation IDs for the same agent_session_id, acquires the first permit through acquire_generation_admission, then starts a second acquisition and verifies it remains blocked while the first permit is held. Cancel the second request using the shared cancellation/deadline mechanism, assert it returns the "request cancelled" error, and release the first permit to cover serialized trusted-session turns.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/skippy-server/src/frontend/backend.rs`:
- Around line 531-546: Move the reserve_generation_queue call in the
session-contention path so it occurs only after session.acquire_until
successfully returns and session_permit is restored. Do not hold a global
queue_reservation while waiting for the session permit; preserve the existing
cancellation, deadline, and generation-permit acquisition behavior around the
updated ordering.
---
Nitpick comments:
In `@crates/openai-frontend/src/router.rs`:
- Around line 1145-1165: Introduce an outcome enum for the values produced by
drop_outcome, with an as_str() method preserving the existing log strings.
Update drop_outcome, finish_drop, and finish_natural to use the enum, replacing
literal success comparisons with exhaustive enum matching while passing as_str()
to observation.finish.
- Around line 211-217: Extract the repeated streamed lifecycle setup from
chat_completions, responses, and completions into a shared StreamTelemetry type
with a constructor for the four telemetry cells and a lifecycle method that
builds the existing StreamLifecycle/StreamCompletion wiring. Update each route
to use this helper, passing only its operation label and event stream while
preserving current behavior.
- Around line 1384-1524: Extend the streamed lifecycle coverage in the existing
lifecycle tests by asserting that calling finish_natural followed by finish_drop
emits exactly one request_finished event. Verify its outcome is success, using
EventCapture’s captured outcome field, while preserving the existing
response_completed assertions.
In `@crates/skippy-server/src/frontend/backend.rs`:
- Around line 865-945: Add an async test alongside the existing session-registry
tests that creates trusted generation IDs for the same agent_session_id,
acquires the first permit through acquire_generation_admission, then starts a
second acquisition and verifies it remains blocked while the first permit is
held. Cancel the second request using the shared cancellation/deadline
mechanism, assert it returns the "request cancelled" error, and release the
first permit to cover serialized trusted-session turns.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 6c40b567-45df-49d1-8e3c-62f98580baff
📒 Files selected for processing (11)
crates/openai-frontend/src/backend.rscrates/openai-frontend/src/common.rscrates/openai-frontend/src/router.rscrates/skippy-server/src/frontend/backend.rscrates/skippy-server/src/frontend/generation/cache_hints.rscrates/skippy-server/src/frontend/generation/queue.rscrates/skippy-server/src/frontend/generation/server.rscrates/skippy-server/src/frontend/generation/types.rscrates/skippy-server/src/frontend/local_generation/tests.rscrates/skippy-server/src/frontend/tests/generation.rscrates/skippy-server/src/frontend/tests/multimodal.rs
| if let Some(session) = session_permit.as_mut() { | ||
| if cancellation.is_cancelled() { | ||
| return Err(request_cancelled_error()); | ||
| } | ||
| if !session.try_acquire()? { | ||
| queue_reservation = Some( | ||
| reserve_generation_queue( | ||
| self.generation_queue_depth.clone(), | ||
| self.generation_queue_limit, | ||
| ) | ||
| .ok_or_else(generation_queue_full_error)?, | ||
| ); | ||
| let session = session_permit.take().expect("session permit was present"); | ||
| session_permit = Some(session.acquire_until(deadline, cancellation).await?); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
A queued turn of one trusted session can consume the whole global queue and reject unrelated requests.
When the session lock is contended, this code reserves a global queue slot before it waits for the session permit, and it holds that reservation through the session wait and the following global-permit wait. generation_queue_limit is set to generation_concurrency in both serve_openai and embedded_openai_backend, so the limit is small.
Sequence with generation_concurrency = 1:
- Turn A of session S takes the session permit and the single generation permit.
- Turn B of session S finds the session busy, reserves the only queue slot, and waits.
- A request for an unrelated session calls
acquire_generation_permit_until, finds no permits, callsreserve_generation_queue, and getsNone. - The unrelated request fails immediately with
generation_queue_full_error(429), although it is not blocked by session S.
The window lasts up to GENERATION_ADMISSION_TIMEOUT. Reserve global queue capacity only after the session permit is acquired, or count session waiters in a separate counter.
🛠️ Proposed change: reserve global queue capacity after the session permit is held
if let Some(session) = session_permit.as_mut() {
if cancellation.is_cancelled() {
return Err(request_cancelled_error());
}
if !session.try_acquire()? {
- queue_reservation = Some(
- reserve_generation_queue(
- self.generation_queue_depth.clone(),
- self.generation_queue_limit,
- )
- .ok_or_else(generation_queue_full_error)?,
- );
let session = session_permit.take().expect("session permit was present");
session_permit = Some(session.acquire_until(deadline, cancellation).await?);
+ queue_reservation = Some(
+ reserve_generation_queue(
+ self.generation_queue_depth.clone(),
+ self.generation_queue_limit,
+ )
+ .ok_or_else(generation_queue_full_error)?,
+ );
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if let Some(session) = session_permit.as_mut() { | |
| if cancellation.is_cancelled() { | |
| return Err(request_cancelled_error()); | |
| } | |
| if !session.try_acquire()? { | |
| queue_reservation = Some( | |
| reserve_generation_queue( | |
| self.generation_queue_depth.clone(), | |
| self.generation_queue_limit, | |
| ) | |
| .ok_or_else(generation_queue_full_error)?, | |
| ); | |
| let session = session_permit.take().expect("session permit was present"); | |
| session_permit = Some(session.acquire_until(deadline, cancellation).await?); | |
| } | |
| } | |
| if let Some(session) = session_permit.as_mut() { | |
| if cancellation.is_cancelled() { | |
| return Err(request_cancelled_error()); | |
| } | |
| if !session.try_acquire()? { | |
| let session = session_permit.take().expect("session permit was present"); | |
| session_permit = Some(session.acquire_until(deadline, cancellation).await?); | |
| queue_reservation = Some( | |
| reserve_generation_queue( | |
| self.generation_queue_depth.clone(), | |
| self.generation_queue_limit, | |
| ) | |
| .ok_or_else(generation_queue_full_error)?, | |
| ); | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/skippy-server/src/frontend/backend.rs` around lines 531 - 546, Move
the reserve_generation_queue call in the session-contention path so it occurs
only after session.acquire_until successfully returns and session_permit is
restored. Do not hold a global queue_reservation while waiting for the session
permit; preserve the existing cancellation, deadline, and generation-permit
acquisition behavior around the updated ordering.
|
Superseded by #1258 |
3 participants
What changed
Add structured request lifecycle observability to the OpenAI frontend.
request_finishedevent for success, backend error, adapter error, timeout, cancellation, and client disconnect.Why
The previous instrumentation could label a stream as finished after an error, had no terminal event for client disconnects, and did not classify backend timeouts distinctly. This makes downstream diagnostics and evidence admission ambiguous.
The new events are backend-stage measurements. They are not claims about client-visible TTFT or end-to-end latency.
Validation
cargo fmt --allcargo check -p openai-frontendcargo test -p openai-frontend(156 unit tests, 7 integration tests)cargo clippy -p openai-frontend -- -D warningsNo API shape, prompt content, model behavior, or request payload semantics are changed.
Summary by CodeRabbit