fix(core): correct attempt-observation reset, success status, and error precision

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 <noreply@anthropic.com>
This commit is contained in:
zyycn
2026-07-27 01:51:19 +08:00
co-authored by Claude
parent e979454457
commit 74f59cc4f8
5 changed files with 60 additions and 24 deletions
@@ -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")
}
@@ -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<bool, EngineError> {
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();
@@ -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()
}
@@ -673,11 +673,7 @@ impl CoordinatedEvent {
}
}
/// 创建一个必须在同一下游提交边界内按序编码的事件批次。
///
/// # Errors
///
/// 空批次无法建立交付边界。
/// 将一批 provider 事件合并为单个提交单元;空批次是无效投递状态。
pub fn try_batch(
events: Vec<ProviderEvent>,
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")]
@@ -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();