From 74f59cc4f8e712a6336b0ac25b5f35fb5cfa90f8 Mon Sep 17 00:00:00 2001 From: zyycn Date: Mon, 27 Jul 2026 01:51:19 +0800 Subject: [PATCH] fix(core): correct attempt-observation reset, success status, and error precision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reset_uncommitted_observations left provider_processing_ms set when discarding a failed attempt's observations, leaking one attempt's upstream processing time into the next; clear it with the other per-attempt fields. finish_success fabricated an upstream_status_code of 200 when none was observed — persist the absent status as NULL instead. Give the "no active attempt" internal state its own EngineError::NoActiveAttempt instead of overloading EmptyRoutingPlan, and map it to an internal gateway error. Remove the unused free function commit_requirement. Co-authored-by: Claude --- .../crates/gateway-api/src/openai/error.rs | 1 + .../gateway-core/src/engine/coordinator.rs | 12 ++--- .../gateway-core/src/engine/execution.rs | 13 ++--- backend/crates/gateway-core/src/engine/mod.rs | 8 ++- .../gateway-core/tests/engine/coordinator.rs | 50 +++++++++++++++++-- 5 files changed, 60 insertions(+), 24 deletions(-) diff --git a/backend/crates/gateway-api/src/openai/error.rs b/backend/crates/gateway-api/src/openai/error.rs index 7234dad35..703f38681 100644 --- a/backend/crates/gateway-api/src/openai/error.rs +++ b/backend/crates/gateway-api/src/openai/error.rs @@ -98,6 +98,7 @@ pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError { | EngineError::ContinuationPinMismatch | EngineError::RequiredAccountMismatch | EngineError::DownstreamCommitRequired + | EngineError::NoActiveAttempt | EngineError::InvalidDeliveryState => { GatewayError::new(GatewayErrorKind::Internal, "gateway execution failed") } diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index c4e0fff6b..a24669541 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -381,7 +381,7 @@ where } let boundary = { - let current = self.current.as_mut().ok_or(EngineError::EmptyRoutingPlan)?; + let current = self.current.as_mut().ok_or(EngineError::NoActiveAttempt)?; poll_stream_item( &mut current.stream, self.cancellation.clone(), @@ -678,7 +678,7 @@ where if let GatewayEvent::ProviderCost(observed) = event { self.cost = observed.into_estimate(); } - let current = self.current.as_mut().ok_or(EngineError::EmptyRoutingPlan)?; + let current = self.current.as_mut().ok_or(EngineError::NoActiveAttempt)?; if !current.send_observed { self.engine .store() @@ -691,7 +691,7 @@ where } async fn observe_wire_event(&mut self) -> Result<(), EngineError> { - let current = self.current.as_mut().ok_or(EngineError::EmptyRoutingPlan)?; + let current = self.current.as_mut().ok_or(EngineError::NoActiveAttempt)?; if !current.send_observed { self.engine .store() @@ -779,7 +779,7 @@ where /// 返回 `true` 表示调用方必须丢弃本 attempt 已收集的未提交事件。 async fn handle_stream_error(&mut self, error: ProviderError) -> Result { - let current = self.current.take().ok_or(EngineError::EmptyRoutingPlan)?; + let current = self.current.take().ok_or(EngineError::NoActiveAttempt)?; if matches!( error.kind(), ProviderErrorKind::RateLimited | ProviderErrorKind::QuotaExhausted @@ -930,6 +930,7 @@ where self.timings.first_reasoning_ms = None; self.timings.first_text_ms = None; self.timings.first_token_ms = None; + self.timings.provider_processing_ms = None; self.upstream_complete = false; } @@ -959,8 +960,7 @@ where .current .as_ref() .and_then(|current| current.response_observation.as_ref()) - .and_then(ProviderResponseObservation::status_code) - .or(Some(200)); + .and_then(ProviderResponseObservation::status_code); let (upstream_transport, http_version, websocket_pool) = self.current_transport_observation(); let provider_metadata_json = self.current_provider_metadata_json(); diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index f9fd749de..bc709a158 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -20,10 +20,10 @@ use crate::engine::coordinator::ResponseExecutionSession; use crate::engine::probe::{AccountProbe, AccountProbeRequest, AccountProbeResult}; use crate::engine::provider::ProviderRegistry; use crate::engine::{ - AttemptCoordinator, AttemptRecord, CancellationToken, CommitRequirement, CoordinatedEvent, - EngineError, ExecutionStore, GatewayEngine, IntermediateFailure, ModelRequestFinalization, - ModelRequestId, NewModelRequest, ProbeFailure, ProviderAccountId, ProviderAttemptOutcome, - RecoveryReport, UpstreamSendState, + AttemptCoordinator, AttemptRecord, CancellationToken, CoordinatedEvent, EngineError, + ExecutionStore, GatewayEngine, IntermediateFailure, ModelRequestFinalization, ModelRequestId, + NewModelRequest, ProbeFailure, ProviderAccountId, ProviderAttemptOutcome, RecoveryReport, + UpstreamSendState, }; use crate::error::{GatewayError, GatewayErrorKind, ProviderErrorKind, StoreError}; use crate::event::{GatewayEvent, ProviderEvent, ProviderResponseHeader}; @@ -943,8 +943,3 @@ pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError { _ => GatewayError::new(GatewayErrorKind::Internal, "request execution failed"), } } - -#[must_use] -pub const fn commit_requirement(event: &CoordinatedEvent) -> CommitRequirement { - event.commit_requirement() -} diff --git a/backend/crates/gateway-core/src/engine/mod.rs b/backend/crates/gateway-core/src/engine/mod.rs index 9e679d75a..7f246382b 100644 --- a/backend/crates/gateway-core/src/engine/mod.rs +++ b/backend/crates/gateway-core/src/engine/mod.rs @@ -673,11 +673,7 @@ impl CoordinatedEvent { } } - /// 创建一个必须在同一下游提交边界内按序编码的事件批次。 - /// - /// # Errors - /// - /// 空批次无法建立交付边界。 + /// 将一批 provider 事件合并为单个提交单元；空批次是无效投递状态。 pub fn try_batch( events: Vec, commit_requirement: CommitRequirement, @@ -728,6 +724,8 @@ pub enum EngineError { Deadline, #[error("routing plan has no candidate")] EmptyRoutingPlan, + #[error("no provider attempt is active")] + NoActiveAttempt, #[error("downstream delivery must be committed before execution can continue")] DownstreamCommitRequired, #[error("downstream delivery cannot be committed in the current state")] diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 810d8edf6..ed933b6e0 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -15,18 +15,20 @@ use gateway_core::engine::continuation::{ use gateway_core::engine::credential::{ AccountSelectionPolicy, CredentialRevision, ProviderAccountId, RotationStrategy, }; +use gateway_core::engine::execution::gateway_error_from_engine; use gateway_core::engine::provider::{ Provider, ProviderCallMetadata, ProviderCatalogGeneration, ProviderModelCapabilities, ProviderRegistry, ProviderRequest, ProviderResource, ProviderStream, UpstreamTransport, }; use gateway_core::engine::{ AttemptContext, AttemptCoordinator, AttemptRecord, CancellationToken, CommitRequirement, - ContinuationAttempt, ExecutionOutcome, ExecutionStore, GatewayEngine, IntermediateFailure, - ModelRequestFinalization, ModelRequestId, NewModelRequest, ProviderAttemptOutcome, - RecoveryReport, UpstreamSendState, + ContinuationAttempt, EngineError, ExecutionOutcome, ExecutionStore, GatewayEngine, + IntermediateFailure, ModelRequestFinalization, ModelRequestId, NewModelRequest, + ProviderAttemptOutcome, RecoveryReport, UpstreamSendState, }; use gateway_core::error::{ - ContinuationFailure, ProviderError, ProviderErrorKind, SafeUpstreamValue, StoreError, + ContinuationFailure, GatewayErrorKind, ProviderError, ProviderErrorKind, SafeUpstreamValue, + StoreError, }; use gateway_core::event::{ ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent, @@ -855,6 +857,7 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { connect_ms: Some(11), headers_ms: Some(13), first_event_ms: Some(987_654), + provider_processing_ms: Some(41), ..ProviderResponseTimings::default() }); let second_observation = ProviderResponseObservation::new( @@ -910,6 +913,7 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { assert_eq!(finalization.transport_decision_wait_ms, None); assert_eq!(finalization.connect_ms, None); assert_eq!(finalization.headers_ms, None); + assert_eq!(finalization.provider_processing_ms, None); assert!( finalization .first_event_ms @@ -948,6 +952,44 @@ fn websocket_success_keeps_client_http_status_absent() { assert_eq!(state.finalizations[0].client_status_code, None); } +#[test] +fn success_without_response_observation_persists_no_upstream_status() { + let operation = operation(RetrySafety::Idempotent); + let route_plan = plan(&operation); + let (coordinator, store, _) = coordinator(vec![Script::Stream { + account_id: "acct_one", + items: complete_stream(None), + }]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .expect("start execution"); + + block_on(session.collect_uncommitted()).expect("collect response"); + block_on(session.commit_downstream(Some(200))).expect("commit response"); + + let state = store.state.lock().expect("store lock"); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded); + assert_eq!(state.finalizations[0].upstream_status_code, None); +} + +#[test] +fn missing_attempt_is_classified_internal_not_no_available_provider() { + assert_eq!( + gateway_error_from_engine(&EngineError::EmptyRoutingPlan).kind(), + GatewayErrorKind::NoAvailableProvider + ); + assert_eq!( + gateway_error_from_engine(&EngineError::NoActiveAttempt).kind(), + GatewayErrorKind::Internal + ); +} + #[test] fn authenticated_native_continuation_reaches_every_attempt_context() { let operation = generate_operation();