From 5762dabe795fd5b1130c42560b0218018e4ef614 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 13:10:25 +0100 Subject: [PATCH] connector-reasoning-item fix: ProviderCall holds only the filtered request; a refused backend result is a recorded typed failure ProviderCall no longer holds the caller's invocation (the claim does), so no path from a call, Debug included, reaches foreign reasoning state: the call carries the filtered request and a receipt template. TransportResult::transported takes its identity from the receipt. complete no longer returns an error for a result the service refuses (wrong provider tag, wrong drop count, validate_against failing, not transported): it records a Provider-kind stand-in failure and releases the slot, so the caller gets a typed answer and a retry replays it instead of ReplayConflict/Capacity. Findings call-debug-shows-foreign-state and refused-completion-wedges-slot. Co-Authored-By: Claude Opus 5.5 --- crates/model-connector/src/contract.rs | 23 +- crates/model-connector/src/envelope.rs | 2 +- crates/model-connector/src/service.rs | 255 +++++++++++++----- .../model-connector/tests/reasoning_state.rs | 86 +++++- 4 files changed, 277 insertions(+), 89 deletions(-) diff --git a/crates/model-connector/src/contract.rs b/crates/model-connector/src/contract.rs index 148828b..bb6459e 100644 --- a/crates/model-connector/src/contract.rs +++ b/crates/model-connector/src/contract.rs @@ -326,17 +326,14 @@ impl TransportResult { } } - /// Built by `ProviderCall` in the daemon and by in-crate tests only. + /// Built by `ProviderCall` in the daemon and by in-crate tests only; the + /// result's identity is its receipt's. #[cfg(any(test, feature = "daemon"))] - pub(crate) fn transported( - invocation: &TransportInvocation, - events: Vec, - receipt: TransportReceipt, - ) -> Self { + pub(crate) fn transported(events: Vec, receipt: TransportReceipt) -> Self { Self { schema_id: RESULT_SCHEMA_ID.to_string(), - request_id: invocation.request_id().to_string(), - caller_runtime_id: invocation.caller_runtime_id.clone(), + request_id: receipt.request_id.clone(), + caller_runtime_id: receipt.caller_runtime_id.clone(), disposition: TransportDisposition::Transported { events, receipt: Box::new(receipt), @@ -769,7 +766,6 @@ mod tests { fn transported_result_binds_identity_digest_and_event_order() { let invocation = invocation(); let result = TransportResult::transported( - &invocation, vec![TransportEvent { sequence: 0, payload: TransportEventPayload::ToolCall { @@ -830,7 +826,7 @@ mod tests { ] ); let invocation = invocation(); - let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); + let result = TransportResult::transported(Vec::new(), receipt(&invocation)); let TransportDisposition::Transported { receipt: sent, .. } = &result.disposition else { unreachable!() }; @@ -876,7 +872,7 @@ mod tests { #[test] fn an_unknown_field_is_refused_not_dropped() { let invocation = invocation(); - let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); + let result = TransportResult::transported(Vec::new(), receipt(&invocation)); let TransportDisposition::Transported { receipt: sent, .. } = &result.disposition else { unreachable!() }; @@ -968,7 +964,7 @@ mod tests { #[test] fn validate_against_refuses_each_identity_field() { let invocation = invocation(); - let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); + let result = TransportResult::transported(Vec::new(), receipt(&invocation)); assert_eq!(result.validate_against(&invocation), Ok(())); let mut other_request = result.clone(); other_request.request_id = "request-2".to_string(); @@ -1031,7 +1027,7 @@ mod tests { event(4, text("three")), event(5, reasoning(&[7])), ]; - let result = TransportResult::transported(&invocation, events, receipt(&invocation)); + let result = TransportResult::transported(events, receipt(&invocation)); assert_eq!( result.transcript_items(), vec![ @@ -1072,7 +1068,6 @@ mod tests { message: "slow down".to_string(), }; let partial = TransportResult::transported( - &invocation, vec![event(0, text("half")), event(1, reasoning(&[1, 2, 3]))], failed, ); diff --git a/crates/model-connector/src/envelope.rs b/crates/model-connector/src/envelope.rs index 4d0f8e7..56bb1c2 100644 --- a/crates/model-connector/src/envelope.rs +++ b/crates/model-connector/src/envelope.rs @@ -557,7 +557,7 @@ mod tests { fn the_outer_identity_must_match_the_sealed_result_on_each_field() { let key = security("epiphany-distinct-test-key"); let invocation = invocation(); - let result = TransportResult::transported(&invocation, Vec::new(), receipt(&invocation)); + let result = TransportResult::transported(Vec::new(), receipt(&invocation)); let plaintext = wire_encode(&result).unwrap(); let seal = |caller: &str, request_id: &str| { encrypt_message(caller, request_id, EnvelopeKind::Result, &plaintext, &key).unwrap() diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index b350283..68825dc 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -107,15 +107,14 @@ struct ReplayRecord { } /// What a backend is handed: the request with every foreign reasoning item -/// already removed, and the means to answer it. The caller's own invocation, -/// the provider tag and the drop count are private, so a backend can neither -/// see foreign reasoning state nor write the identity of its receipt. +/// already removed, and the means to answer it. The caller's own invocation is +/// not held here at all, so no path from a call (Debug included) reaches +/// foreign reasoning state, and a backend cannot write the identity of its +/// receipt. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProviderCall { - invocation: TransportInvocation, request: ProviderRequest, - foreign_reasoning_dropped: u32, - provider: String, + receipt: TransportReceipt, } impl ProviderCall { @@ -127,7 +126,7 @@ impl ProviderCall { /// How many foreign reasoning items the service removed from the request. pub fn foreign_reasoning_dropped(&self) -> u32 { - self.foreign_reasoning_dropped + self.receipt.foreign_reasoning_dropped } /// The result of this call. The whole receipt (schema, ids, both digests, @@ -138,20 +137,11 @@ impl ProviderCall { events: Vec, outcome: TransportOutcome, ) -> TransportResult { - let invocation = &self.invocation; TransportResult::transported( - invocation, events, TransportReceipt { - schema_id: RECEIPT_SCHEMA_ID.to_string(), - request_id: invocation.request_id().to_string(), - caller_runtime_id: invocation.caller_runtime_id.clone(), - native_request_sha256: invocation.native_request_sha256, - provider_request_sha256: invocation.provider_request_sha256, - model: invocation.request.model.clone(), - provider: self.provider.clone(), - foreign_reasoning_dropped: self.foreign_reasoning_dropped, outcome, + ..self.receipt.clone() }, ) } @@ -160,6 +150,7 @@ impl ProviderCall { #[derive(Debug, Clone, PartialEq, Eq)] pub struct ExecutionClaim { call: ProviderCall, + invocation: TransportInvocation, invocation_sha256: [u8; 32], } @@ -327,7 +318,12 @@ impl TransportService { if fits_frame(response, max_frame_bytes)? { return Ok(TransportAdmission::Reply(response.clone())); } - let stand_in = self.seal_stand_in(&invocation, &security, max_frame_bytes)?; + let stand_in = self.seal_stand_in( + &invocation, + &security, + max_frame_bytes, + FailureKind::ResultSize, + )?; self.replay_store .put( replay_key(&identity), @@ -374,11 +370,24 @@ impl TransportService { self.live.insert(identity); Ok(TransportAdmission::Execute(Box::new(ExecutionClaim { call: ProviderCall { - invocation, + receipt: TransportReceipt { + schema_id: RECEIPT_SCHEMA_ID.to_string(), + request_id: invocation.request_id().to_string(), + caller_runtime_id: invocation.caller_runtime_id.clone(), + native_request_sha256: invocation.native_request_sha256, + provider_request_sha256: invocation.provider_request_sha256, + model: invocation.request.model.clone(), + provider: self.provider.clone(), + foreign_reasoning_dropped, + // Replaced by the backend's outcome in `transported`. + outcome: TransportOutcome::Failed { + kind: FailureKind::Provider, + message: String::new(), + }, + }, request, - foreign_reasoning_dropped, - provider: self.provider.clone(), }, + invocation, invocation_sha256, }))) } @@ -394,16 +403,7 @@ impl TransportService { result: TransportResult, max_frame_bytes: usize, ) -> Result { - let TransportDisposition::Transported { receipt, .. } = &result.disposition else { - return Err(ServiceError::InvalidCompletion); - }; - let invocation = &claim.call.invocation; - result.validate_against(invocation)?; - if receipt.provider != self.provider - || receipt.foreign_reasoning_dropped != claim.call.foreign_reasoning_dropped - { - return Err(ServiceError::InvalidCompletion); - } + let invocation = &claim.invocation; let identity = RequestIdentity { caller_runtime_id: invocation.caller_runtime_id.clone(), request_id: invocation.request_id().to_string(), @@ -426,17 +426,29 @@ impl TransportService { .get(&identity.caller_runtime_id) .ok_or(ServiceError::CallerNotAdmitted)?; let security = caller.security.clone(); - let mut response = encrypt_result(&result, &security)?; - if !fits_frame(&response, max_frame_bytes)? { - response = match self.seal_stand_in(invocation, &security, max_frame_bytes) { - Ok(stand_in) => stand_in, - Err(error) => { - // Nothing is recorded, but the caller's slot is not held - // by a request that can no longer finish. + // A result the service refuses (not a transported result, not bound to + // this invocation, or carrying a provider tag or drop count the call + // did not derive) is answered like an oversized one: a typed failure + // is recorded and the slot released, so the caller gets an answer and + // a retry replays it. + let admissible = self.admits_result(&result, &claim); + let mut response = if admissible { + encrypt_result(&result, &security)? + } else { + self.seal_stand_in(invocation, &security, max_frame_bytes, FailureKind::Provider) + .inspect_err(|_| { self.live.remove(&identity); - return Err(error); - } - }; + })? + }; + if admissible && !fits_frame(&response, max_frame_bytes)? { + // Nothing is recorded when even the stand-in cannot be carried, + // but the caller's slot is not held by a request that can no + // longer finish. + response = self + .seal_stand_in(invocation, &security, max_frame_bytes, FailureKind::ResultSize) + .inspect_err(|_| { + self.live.remove(&identity); + })?; } let caller = self .callers @@ -465,18 +477,20 @@ impl TransportService { invocation: &TransportInvocation, security: &TransportKey, max_frame_bytes: usize, + kind: FailureKind, ) -> Result { let (_, foreign_reasoning_dropped) = partition_reasoning(&invocation.request, &self.provider) .map_err(|_| ServiceError::InvalidCompletion)?; let stand_in = encrypt_result( - &result_size_failure( + &stand_in_failure( &invocation.caller_runtime_id, invocation.request_id(), &invocation.request.model, invocation.native_request_sha256, invocation.provider_request_sha256, foreign_reasoning_dropped, + kind, ), security, )?; @@ -487,6 +501,18 @@ impl TransportService { } } + /// Whether `result` is the answer to this claim's call: transported, + /// bound to the caller's invocation, and carrying the provider tag and drop + /// count the service derived. + fn admits_result(&self, result: &TransportResult, claim: &ExecutionClaim) -> bool { + let TransportDisposition::Transported { receipt, .. } = &result.disposition else { + return false; + }; + result.validate_against(&claim.invocation).is_ok() + && receipt.provider == self.provider + && receipt.foreign_reasoning_dropped == claim.call.receipt.foreign_reasoning_dropped + } + fn reply_refusal( &self, invocation: &TransportInvocation, @@ -518,6 +544,32 @@ pub(crate) fn result_size_failure( provider_request_sha256: [u8; 32], foreign_reasoning_dropped: u32, ) -> TransportResult { + stand_in_failure( + caller_runtime_id, + request_id, + model, + native_request_sha256, + provider_request_sha256, + foreign_reasoning_dropped, + FailureKind::ResultSize, + ) +} + +/// The stand-in failure for `kind`: `ResultSize` for a result that cannot be +/// carried, `Provider` for one the service refused. +fn stand_in_failure( + caller_runtime_id: &str, + request_id: &str, + model: &str, + native_request_sha256: [u8; 32], + provider_request_sha256: [u8; 32], + foreign_reasoning_dropped: u32, + kind: FailureKind, +) -> TransportResult { + let message = match kind { + FailureKind::ResultSize => "provider result exceeded its byte bound", + _ => "the service refused the backend's result", + }; TransportResult { schema_id: RESULT_SCHEMA_ID.to_string(), request_id: request_id.to_string(), @@ -534,8 +586,8 @@ pub(crate) fn result_size_failure( provider: STAND_IN_PROVIDER.to_string(), foreign_reasoning_dropped, outcome: TransportOutcome::Failed { - kind: FailureKind::ResultSize, - message: "provider result exceeded its byte bound".to_string(), + kind, + message: message.to_string(), }, }), }, @@ -680,7 +732,6 @@ mod tests { .complete( *claim, TransportResult::transported( - &invocation, vec![TransportEvent { sequence: 0, payload: TransportEventPayload::ToolCall { @@ -1198,7 +1249,7 @@ mod tests { text: text.to_string(), }, }], - receipt(&call.invocation).outcome, + receipt(&invocation_for(CALLER, "outcome")).outcome, ) } @@ -1562,39 +1613,97 @@ mod tests { } #[test] - fn complete_refuses_a_hand_built_receipt_with_another_provider_or_count() { + fn complete_answers_a_hand_built_receipt_with_another_provider_or_count_as_a_typed_failure() { let store = TestStore::new(); - let mut service = open_service(&store, 4); - let (admission, _) = admit_input( - &mut service, - "hand-built", - vec![state("other-provider", "x", 1), user("first")], - ); + let mut service = open_service(&store, 1); + let input = vec![state("other-provider", "x", 1), user("first")]; + let tamper_edits: [fn(&mut TransportReceipt); 4] = [ + |receipt| receipt.provider = "other-provider".to_string(), + |receipt| receipt.provider = STAND_IN_PROVIDER.to_string(), + |receipt| receipt.foreign_reasoning_dropped = 2, + |receipt| receipt.foreign_reasoning_dropped = 0, + ]; + for (index, edit) in tamper_edits.into_iter().enumerate() { + let id = format!("hand-built-{index}"); + let (admission, invocation) = admit_input(&mut service, &id, input.clone()); + let TransportAdmission::Execute(claim) = admission else { + panic!("a refused completion must have released the slot"); + }; + assert_eq!(claim.call().foreign_reasoning_dropped(), 1); + let mut bad = text_result(claim.call(), "done"); + let TransportDisposition::Transported { receipt, .. } = &mut bad.disposition else { + unreachable!(); + }; + edit(receipt); + let response = service.complete(*claim, bad, FRAME).unwrap(); + let answer = decrypt_result(&response, &security(KEY), &invocation).unwrap(); + let TransportDisposition::Transported { events, receipt } = &answer.disposition else { + panic!("a typed failure is transported"); + }; + assert!(events.is_empty()); + assert_eq!(receipt.provider, STAND_IN_PROVIDER); + assert_eq!(receipt.foreign_reasoning_dropped, 1); + assert!(matches!( + receipt.outcome, + TransportOutcome::Failed { + kind: FailureKind::Provider, + .. + } + )); + // A retry of the same id replays the failure, not ReplayConflict. + let TransportAdmission::Reply(replay) = service + .begin(&raw_envelope(&invocation), 1_000, FRAME) + .unwrap() + else { + panic!("a retry must replay the recorded failure"); + }; + assert_eq!(replay, response); + } + } + + #[test] + fn a_result_that_fails_validation_is_answered_as_a_typed_failure_and_releases_the_slot() { + let store = TestStore::new(); + let mut service = open_service(&store, 1); + let (claim, invocation) = claim_for(&mut service, "invalid-result"); + let mut bad = text_result(claim.call(), "done"); + bad.request_id = "someone-elses".to_string(); + let response = service.complete(*claim, bad, FRAME).unwrap(); + let answer = decrypt_result(&response, &security(KEY), &invocation).unwrap(); + let TransportDisposition::Transported { receipt, .. } = &answer.disposition else { + panic!("a typed failure is transported"); + }; + assert!(matches!( + receipt.outcome, + TransportOutcome::Failed { + kind: FailureKind::Provider, + .. + } + )); + // The caller's only slot is free for the next request. + let (next, _) = claim_for(&mut service, "next"); + drop(next); + } + + #[test] + fn a_provider_call_formats_without_foreign_reasoning_state() { + let store = TestStore::new(); + let mut service = open_service(&store, 1); + let canary = "canary-foreign-provider-tag"; + let foreign = InputItem::ReasoningState { + provider: canary.to_string(), + model: canary.to_string(), + state: vec![0xC7, 0xC8, 0xC9, 0xCA], + }; + let (admission, _) = admit_input(&mut service, "debug-leak", vec![foreign, user("first")]); let TransportAdmission::Execute(claim) = admission else { panic!("expected an execution claim"); }; assert_eq!(claim.call().foreign_reasoning_dropped(), 1); - let good = text_result(claim.call(), "done"); - let tamper = |edit: fn(&mut TransportReceipt)| { - let mut result = good.clone(); - let TransportDisposition::Transported { receipt, .. } = &mut result.disposition else { - unreachable!(); - }; - edit(receipt); - result - }; - let wrong_provider = tamper(|receipt| receipt.provider = "other-provider".to_string()); - let stand_in_provider = tamper(|receipt| receipt.provider = STAND_IN_PROVIDER.to_string()); - let too_many = tamper(|receipt| receipt.foreign_reasoning_dropped = 2); - let none = tamper(|receipt| receipt.foreign_reasoning_dropped = 0); - for bad in [wrong_provider, stand_in_provider, too_many, none] { - assert_eq!( - service.complete((*claim).clone(), bad, FRAME).unwrap_err(), - ServiceError::InvalidCompletion - ); + for text in [format!("{:?}", claim.call()), format!("{:#?}", claim.call())] { + assert!(!text.contains(canary), "{text}"); + assert!(!text.contains("199, 200, 201, 202"), "{text}"); } - // A refused completion leaves the request live, and the derived result passes. - assert!(service.complete(*claim, good, FRAME).is_ok()); } #[test] diff --git a/crates/model-connector/tests/reasoning_state.rs b/crates/model-connector/tests/reasoning_state.rs index d8e225b..559bd2f 100644 --- a/crates/model-connector/tests/reasoning_state.rs +++ b/crates/model-connector/tests/reasoning_state.rs @@ -83,6 +83,11 @@ fn now_ms() -> u64 { /// A running serve loop and a client for it. fn start(backend: Arc) -> ConnectorClient { + start_with(backend, 4) +} + +/// As `start`, with the caller's concurrent-request limit. +fn start_with(backend: Arc, max_concurrent: usize) -> ConnectorClient { static SEQUENCE: AtomicUsize = AtomicUsize::new(0); let root: PathBuf = std::env::temp_dir().join(format!( "model-connector-reasoning-{}-{}", @@ -98,7 +103,7 @@ fn start(backend: Arc) -> ConnectorClient { KEY, 1, ["claude-opus-4-5".to_string()], - 4, + max_concurrent, 64 * 1024, 32_768, ) @@ -262,3 +267,82 @@ fn replayed_transcript_reaches_the_same_provider_verbatim() { assert_eq!(receipt_of(&second).foreign_reasoning_dropped, 0); assert_eq!(backend.seen()[1].input, next); } + +/// A backend that answers with a receipt it has edited after the service +/// derived it. +struct TamperingBackend { + edit: fn(&mut TransportReceipt), +} + +impl ProviderBackend for TamperingBackend { + fn ready(&self) -> bool { + true + } + + fn provider(&self) -> &str { + PROVIDER + } + + fn execute(&self, call: &ProviderCall) -> TransportResult { + let mut result = call.transported( + vec![TransportEvent { + sequence: 0, + payload: TransportEventPayload::TextDelta { + text: "done".to_string(), + }, + }], + TransportOutcome::Completed { + provider_response_id: None, + stop_reason: StopReason::EndTurn, + usage: Usage { + input_tokens: 1, + cache_read_tokens: 0, + cache_write_tokens: 0, + output_tokens: 1, + reasoning_tokens: None, + }, + }, + ); + let TransportDisposition::Transported { receipt, .. } = &mut result.disposition else { + unreachable!(); + }; + (self.edit)(receipt); + result + } +} + +#[test] +fn a_refused_completion_reaches_the_caller_as_a_typed_failure_and_frees_the_slot() { + let edits: [fn(&mut TransportReceipt); 2] = [ + |receipt| receipt.provider = "another-provider".to_string(), + |receipt| receipt.foreign_reasoning_dropped += 1, + ]; + for edit in edits { + // One concurrent request: a wedged slot would refuse the next id. + let client = start_with(Arc::new(TamperingBackend { edit }), 1); + let first = invocation("tampered", vec![user("question")]); + let answer = client.execute(&first).unwrap(); + let TransportDisposition::Transported { events, receipt } = &answer.disposition else { + panic!("expected a typed failure, got {:?}", answer.disposition); + }; + assert!(events.is_empty()); + assert!(matches!( + receipt.outcome, + TransportOutcome::Failed { + kind: FailureKind::Provider, + .. + } + )); + // The same id replays that failure rather than conflicting. + assert_eq!(client.execute(&first).unwrap(), answer); + // The next id is not refused for capacity. + let next = client + .execute(&invocation("tampered-next", vec![user("again")])) + .unwrap(); + assert!( + !matches!(next.disposition, TransportDisposition::Refused(_)), + "{:?}", + next.disposition + ); + } +}