From b115e870f0ff6ebc03c2818fd03d215cac42114a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 12:43:25 +0100 Subject: [PATCH 1/5] connector-reasoning-item (B3b): provider reasoning state is one opaque transcript item; the service refuses it foreign in flight and drops it elsewhere InputItem::ReasoningState and TransportEventPayload::ReasoningState carry the connector's own bytes as MessagePack bin. TransportService owns the provider tag: partition_reasoning is the one place foreignness is decided, begin refuses foreign state in the in-flight exchange (ForeignReasoningState), drops and counts it earlier (receipt.foreign_reasoning_dropped), and complete gates the receipt's provider and count. A backend now receives a ProviderCall holding only the filtered request and builds its result through call.transported, so it can neither see foreign state nor write receipt identity. TransportResult::transcript_items hands a completed result back as items the caller replays verbatim. Co-Authored-By: Claude Opus 5.5 --- README.md | 3 + crates/model-connector/src/contract.rs | 211 +++++++- crates/model-connector/src/envelope.rs | 7 +- crates/model-connector/src/fixtures.rs | 2 + crates/model-connector/src/serve.rs | 117 +++-- crates/model-connector/src/service.rs | 465 ++++++++++++++++-- .../tests/fixtures/wire_v3/README.md | 3 +- .../fixtures/wire_v3/input_reasoning_state.mp | 1 + .../wire_v3/transport_receipt_failed.mp | Bin 358 -> 385 bytes .../wire_v3/transport_result_transported.mp | Bin 715 -> 914 bytes .../model-connector/tests/reasoning_state.rs | 264 ++++++++++ .../model-connector/tests/wire_golden_v3.rs | 68 ++- schemas/provider_request.json | 32 ++ schemas/transport_envelope.json | 2 +- schemas/transport_invocation.json | 32 ++ schemas/transport_receipt.json | 7 + schemas/transport_result.json | 66 ++- 17 files changed, 1143 insertions(+), 137 deletions(-) create mode 100644 crates/model-connector/tests/fixtures/wire_v3/input_reasoning_state.mp create mode 100644 crates/model-connector/tests/reasoning_state.rs diff --git a/README.md b/README.md index cbcccb5..596f571 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,9 @@ Crate `crates/model-connector`: - `src/contract.rs` - the v3 wire contract (`gamecult.model.*.v3`): `ProviderRequest`, `TransportInvocation`, `TransportResult`, `TransportReceipt`, the typed `StopReason`, `FailureKind` and `Refusal`, + `InputItem::ReasoningState` (a provider's opaque reasoning block, replayed by callers + verbatim via `TransportResult::transcript_items`; the service refuses another provider's + state in the in-flight tool exchange and drops it from earlier turns), and the canonical byte and digest functions that bind a result to its request. Every document is named MessagePack with unknown fields refused. - `src/envelope.rs` - the AES-256-GCM `TransportEnvelope` (binary ciphertext), its diff --git a/crates/model-connector/src/contract.rs b/crates/model-connector/src/contract.rs index d60fab5..148828b 100644 --- a/crates/model-connector/src/contract.rs +++ b/crates/model-connector/src/contract.rs @@ -7,7 +7,7 @@ use sha2::{Digest, Sha256}; use thiserror::Error; use zeroize::Zeroize; -use crate::envelope::ServiceError; +use crate::envelope::{ServiceError, binary_schema}; pub const PROVIDER_REQUEST_SCHEMA_ID: &str = "gamecult.model.provider_request.v3"; pub const INVOCATION_SCHEMA_ID: &str = "gamecult.model.transport_invocation.v3"; @@ -124,6 +124,17 @@ impl ProviderRequest { require_provider_call_id(call_id, "input.call_id")?; require_content(output, "input.output")?; } + InputItem::ReasoningState { + provider, + model, + state, + } => { + require_id(provider, "input.provider")?; + require_id(model, "input.model")?; + if state.is_empty() { + return Err(ContractError::Invalid("input.state")); + } + } } } Ok(()) @@ -210,6 +221,18 @@ pub enum InputItem { call_id: String, output: String, }, + /// One provider reasoning block, opaque to every caller: `state` is the + /// connector's own encoding of the whole block. Callers replay it + /// verbatim in transcript order and never inspect or branch on it; only + /// the connector of `provider` reads it, and the service drops or refuses + /// the state of any other provider before a backend sees the request. + ReasoningState { + provider: String, + model: String, + #[serde(with = "serde_bytes")] + #[schemars(schema_with = "binary_schema")] + state: Vec, + }, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] @@ -303,7 +326,9 @@ impl TransportResult { } } - pub fn transported( + /// Built by `ProviderCall` in the daemon and by in-crate tests only. + #[cfg(any(test, feature = "daemon"))] + pub(crate) fn transported( invocation: &TransportInvocation, events: Vec, receipt: TransportReceipt, @@ -319,6 +344,53 @@ impl TransportResult { } } + /// The transcript items a completed result adds, ready to append to the + /// next request verbatim: consecutive text deltas become one assistant + /// text, a tool call stays a tool call and reasoning state is stamped with + /// the receipt's provider and model. A refused or failed result yields + /// none, because a partial reasoning block must never be replayed. + pub fn transcript_items(&self) -> Vec { + let TransportDisposition::Transported { events, receipt } = &self.disposition else { + return Vec::new(); + }; + if !matches!(receipt.outcome, TransportOutcome::Completed { .. }) { + return Vec::new(); + } + let mut items: Vec = Vec::new(); + let mut text = String::new(); + for event in events { + if !matches!(event.payload, TransportEventPayload::TextDelta { .. }) && !text.is_empty() + { + items.push(InputItem::AssistantText { + text: std::mem::take(&mut text), + }); + } + match &event.payload { + TransportEventPayload::TextDelta { text: delta } => text.push_str(delta), + TransportEventPayload::ToolCall { + call_id, + name, + arguments, + } => items.push(InputItem::ToolCall { + call_id: call_id.clone(), + name: name.clone(), + arguments: arguments.clone(), + }), + TransportEventPayload::ReasoningState { state } => { + items.push(InputItem::ReasoningState { + provider: receipt.provider.clone(), + model: receipt.model.clone(), + state: state.clone(), + }) + } + } + } + if !text.is_empty() { + items.push(InputItem::AssistantText { text }); + } + items + } + pub fn validate_against(&self, invocation: &TransportInvocation) -> Result<(), ContractError> { if self.schema_id != RESULT_SCHEMA_ID { return Err(ContractError::Schema); @@ -353,6 +425,14 @@ pub enum TransportEventPayload { name: String, arguments: String, }, + /// One reasoning block of the provider that answered. It carries no + /// provider or model: the receipt names them, so a backend cannot tag its + /// own state. + ReasoningState { + #[serde(with = "serde_bytes")] + #[schemars(schema_with = "binary_schema")] + state: Vec, + }, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] @@ -375,6 +455,12 @@ impl TransportEvent { require_id(name, "event.tool_name")?; require_content(arguments, "event.arguments") } + TransportEventPayload::ReasoningState { state } => { + if state.is_empty() { + return Err(ContractError::Invalid("event.state")); + } + Ok(()) + } } } } @@ -389,6 +475,9 @@ pub struct TransportReceipt { pub provider_request_sha256: [u8; 32], pub model: String, pub provider: String, + /// How many foreign `ReasoningState` items the service removed from the + /// request before the backend saw it. + pub foreign_reasoning_dropped: u32, pub outcome: TransportOutcome, } @@ -454,6 +543,9 @@ pub enum Refusal { Indeterminate, ReplayConflict, Malformed, + /// A `ReasoningState` of another provider sits in the in-flight tool + /// exchange, where dropping it would break the provider's own rules. + ForeignReasoningState, } /// The one wire codec of every v3 document: named MessagePack (struct fields @@ -903,4 +995,119 @@ mod tests { ); } } + + fn reasoning(state: &[u8]) -> TransportEventPayload { + TransportEventPayload::ReasoningState { + state: state.to_vec(), + } + } + + fn event(sequence: u64, payload: TransportEventPayload) -> TransportEvent { + TransportEvent { sequence, payload } + } + + fn text(delta: &str) -> TransportEventPayload { + TransportEventPayload::TextDelta { + text: delta.to_string(), + } + } + + fn tool_call() -> TransportEventPayload { + TransportEventPayload::ToolCall { + call_id: "call-9".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + } + } + + #[test] + fn transcript_items_follow_event_order() { + let invocation = invocation(); + let events = vec![ + event(0, text("one ")), + event(1, text("two")), + event(2, reasoning(&[0x80, 0xff])), + event(3, tool_call()), + event(4, text("three")), + event(5, reasoning(&[7])), + ]; + let result = TransportResult::transported(&invocation, events, receipt(&invocation)); + assert_eq!( + result.transcript_items(), + vec![ + InputItem::AssistantText { + text: "one two".to_string() + }, + InputItem::ReasoningState { + provider: "fixture-sse".to_string(), + model: invocation.request.model.clone(), + state: vec![0x80, 0xff], + }, + InputItem::ToolCall { + call_id: "call-9".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + InputItem::AssistantText { + text: "three".to_string() + }, + InputItem::ReasoningState { + provider: "fixture-sse".to_string(), + model: invocation.request.model.clone(), + state: vec![7], + }, + ] + ); + } + + #[test] + fn transcript_items_of_refused_or_failed_are_empty() { + let invocation = invocation(); + let refused = TransportResult::refused(&invocation, Refusal::OutputBound); + assert_eq!(refused.transcript_items(), Vec::new()); + + let mut failed = receipt(&invocation); + failed.outcome = TransportOutcome::Failed { + kind: FailureKind::RateLimited, + message: "slow down".to_string(), + }; + let partial = TransportResult::transported( + &invocation, + vec![event(0, text("half")), event(1, reasoning(&[1, 2, 3]))], + failed, + ); + assert_eq!(partial.transcript_items(), Vec::new()); + } + + #[test] + fn reasoning_state_validation() { + let item = |provider: &str, model: &str, state: &[u8]| InputItem::ReasoningState { + provider: provider.to_string(), + model: model.to_string(), + state: state.to_vec(), + }; + let validate = |item: InputItem| { + let mut request = request(); + request.input.push(item); + request.validate() + }; + assert_eq!(validate(item("anthropic", "claude", &[1])), Ok(())); + assert_eq!( + validate(item("", "claude", &[1])), + Err(ContractError::Invalid("input.provider")) + ); + assert_eq!( + validate(item("anthropic", "", &[1])), + Err(ContractError::Invalid("input.model")) + ); + assert_eq!( + validate(item("anthropic", "claude", &[])), + Err(ContractError::Invalid("input.state")) + ); + assert_eq!(event(0, reasoning(&[1])).validate(), Ok(())); + assert_eq!( + event(0, reasoning(&[])).validate(), + Err(ContractError::Invalid("event.state")) + ); + } } diff --git a/crates/model-connector/src/envelope.rs b/crates/model-connector/src/envelope.rs index 0f87546..4d0f8e7 100644 --- a/crates/model-connector/src/envelope.rs +++ b/crates/model-connector/src/envelope.rs @@ -37,12 +37,13 @@ pub struct TransportEnvelope { pub ciphertext: Vec, } -/// The ciphertext is MessagePack `bin` on the wire, never an array of integers. -fn binary_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema { +/// Raw bytes are MessagePack `bin` on the wire, never an array of integers. +/// Serves the ciphertext and both reasoning states. +pub(crate) fn binary_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema { schemars::json_schema!({ "type": "string", "format": "binary", - "description": "MessagePack bin: the raw ciphertext bytes, never an array of integers" + "description": "MessagePack bin: raw bytes, never an array of integers" }) } diff --git a/crates/model-connector/src/fixtures.rs b/crates/model-connector/src/fixtures.rs index e94ee52..b96fea8 100644 --- a/crates/model-connector/src/fixtures.rs +++ b/crates/model-connector/src/fixtures.rs @@ -69,6 +69,7 @@ pub(crate) fn receipt(invocation: &TransportInvocation) -> TransportReceipt { provider_request_sha256: invocation.provider_request_sha256, model: invocation.request.model.clone(), provider: "fixture-sse".to_string(), + foreign_reasoning_dropped: 0, outcome: TransportOutcome::Completed { provider_response_id: Some("response-1".to_string()), stop_reason: StopReason::EndTurn, @@ -104,6 +105,7 @@ pub(crate) fn caller( pub(crate) fn open_service(store: &TestStore, max_concurrent_requests: usize) -> TransportService { TransportService::open( &store.path, + "fixture-sse", [ caller( "epiphany-yggdrasil", diff --git a/crates/model-connector/src/serve.rs b/crates/model-connector/src/serve.rs index cef686a..40ece2d 100644 --- a/crates/model-connector/src/serve.rs +++ b/crates/model-connector/src/serve.rs @@ -10,12 +10,12 @@ use serde::{Deserialize, Serialize}; use thiserror::Error; use zeroize::Zeroizing; -use crate::contract::{TransportInvocation, TransportResult}; +use crate::contract::TransportResult; use crate::envelope::{ ServiceError, TransportFrameError, decode_invocation_frame, encode_result_frame, frame_bytes_for_payload, read_transport_frame, write_transport_frame, }; -use crate::service::{CallerAdmission, TransportAdmission, TransportService}; +use crate::service::{CallerAdmission, ProviderCall, TransportAdmission, TransportService}; /// The only provider seam of the connector kit. A connector binary implements /// it over one model provider; the kit names no provider. @@ -24,9 +24,14 @@ pub trait ProviderBackend: Send + Sync { /// opens the replay store or accepts a connection; the request path does /// not consult it. fn ready(&self) -> bool; - /// Executes one admitted invocation. The service has already bound the - /// invocation to its caller and recorded its replay claim. - fn execute(&self, invocation: &TransportInvocation) -> TransportResult; + /// The tag of the provider this backend fronts. The service owns it from + /// `serve` on: it names the provider on every receipt and decides which + /// reasoning state is foreign. + fn provider(&self) -> &str; + /// Executes one admitted call. The service has already bound it to its + /// caller and recorded its replay claim. The call cannot show a foreign + /// reasoning item and builds the result's receipt itself. + fn execute(&self, call: &ProviderCall) -> TransportResult; } /// Bounds of one serve loop. The connector binary's own document owns where @@ -157,6 +162,7 @@ pub fn serve( } let service = Arc::new(Mutex::new(TransportService::open( replay_store, + backend.provider(), admissions, limits.max_expiry_skew_ms, )?)); @@ -230,7 +236,7 @@ fn execute_inference_frame( let response = match admission { TransportAdmission::Reply(response) => response, TransportAdmission::Execute(claim) => { - let result = backend.execute(claim.invocation()); + let result = backend.execute(claim.call()); service .lock() .map_err(|_| ServeError::ServicePoisoned)? @@ -328,7 +334,7 @@ mod tests { use crate::client::ConnectorClient; use crate::contract::{ FailureKind, Refusal, TransportDisposition, TransportEvent, TransportEventPayload, - TransportOutcome, TransportResult, + TransportInvocation, TransportOutcome, TransportResult, }; use crate::fixtures::{TestStore, caller, invocation_for, receipt, request}; @@ -360,17 +366,20 @@ mod tests { self.ready } - fn execute(&self, invocation: &TransportInvocation) -> TransportResult { + fn provider(&self) -> &str { + "fixture-sse" + } + + fn execute(&self, call: &ProviderCall) -> TransportResult { self.executions.fetch_add(1, Ordering::SeqCst); - TransportResult::transported( - invocation, + call.transported( vec![TransportEvent { sequence: 0, payload: TransportEventPayload::TextDelta { text: "done".to_string(), }, }], - receipt(invocation), + receipt(&fresh("sizing")).outcome, ) } } @@ -568,8 +577,13 @@ mod tests { // The wide completion goes straight through the service: serve holds // the replay store for as long as it runs, so a second serve cannot // open it while the first lives. - let mut service = - TransportService::open(&store.path, vec![caller(CALLER, KEY, 4)], 300_000).unwrap(); + let mut service = TransportService::open( + &store.path, + "fixture-sse", + vec![caller(CALLER, KEY, 4)], + 300_000, + ) + .unwrap(); let envelope = crate::envelope::encrypt_invocation(&invocation, &crate::fixtures::security(KEY)) .unwrap(); @@ -578,13 +592,8 @@ mod tests { else { panic!("a fresh request must execute"); }; - service - .complete( - *claim, - sized_result(&invocation, 150_000, "fixture-sse"), - 1 << 20, - ) - .unwrap(); + let wide = sized_result(claim.call(), 150_000); + service.complete(*claim, wide, 1 << 20).unwrap(); drop(service); let narrow = 140_000; @@ -684,7 +693,7 @@ mod tests { fn the_smallest_frame_also_carries_the_failure_that_stands_in_for_a_result() { let long = "x".repeat(crate::envelope::MAX_IDENTITY_BYTES); let plaintext = crate::contract::wire_encode(&crate::service::result_size_failure( - &long, &long, &long, [0xff; 32], [0xff; 32], + &long, &long, &long, [0xff; 32], [0xff; 32], 0, )) .unwrap() .len(); @@ -913,19 +922,13 @@ mod tests { /// Answers every invocation with one text event of a chosen size. struct SizedBackend { text_len: usize, - transport: String, executions: AtomicUsize, } impl SizedBackend { fn new(text_len: usize) -> Arc { - Self::with_transport(text_len, "fixture-sse".to_string()) - } - - fn with_transport(text_len: usize, transport: String) -> Arc { Arc::new(Self { text_len, - transport, executions: AtomicUsize::new(0), }) } @@ -936,28 +939,25 @@ mod tests { true } - fn execute(&self, invocation: &TransportInvocation) -> TransportResult { + fn provider(&self) -> &str { + "fixture-sse" + } + + fn execute(&self, call: &ProviderCall) -> TransportResult { self.executions.fetch_add(1, Ordering::SeqCst); - sized_result(invocation, self.text_len, &self.transport) + sized_result(call, self.text_len) } } - fn sized_result( - invocation: &TransportInvocation, - text_len: usize, - transport: &str, - ) -> TransportResult { - let mut receipt = receipt(invocation); - receipt.provider = transport.to_string(); - TransportResult::transported( - invocation, + fn sized_result(call: &ProviderCall, text_len: usize) -> TransportResult { + call.transported( vec![TransportEvent { sequence: 0, payload: TransportEventPayload::TextDelta { text: "y".repeat(text_len), }, }], - receipt, + receipt(&fresh("sizing")).outcome, ) } @@ -997,7 +997,25 @@ mod tests { /// The encoded size of a result carrying no text, for this request id. fn empty_result_len(request_id: &str) -> usize { - encoded_len(&sized_result(&fresh(request_id), 0, "fixture-sse")) + let store = TestStore::new(); + let mut service = TransportService::open( + &store.path, + "fixture-sse", + vec![caller(CALLER, KEY, 4)], + 300_000, + ) + .unwrap(); + let envelope = crate::envelope::encrypt_invocation( + &fresh(request_id), + &crate::fixtures::security(KEY), + ) + .unwrap(); + let TransportAdmission::Execute(claim) = + service.begin(&envelope, now_ms(), 1 << 20).unwrap() + else { + panic!("a fresh request must execute"); + }; + encoded_len(&sized_result(claim.call(), 0)) } #[test] @@ -1030,17 +1048,6 @@ mod tests { assert_eq!(executions, 1); } - #[test] - fn a_stand_in_failure_never_carries_the_backends_transport() { - let (first, second, executions) = call_twice( - SizedBackend::with_transport(FRAME, "t".repeat(FRAME)), - "request-transport", - ); - assert!(failed_result_size(&first)); - assert_eq!(first, second); - assert_eq!(executions, 1); - } - /// The largest result plaintext whose sealed frame fits at the worst encoding. fn widest_guaranteed_plaintext(base: usize) -> usize { (base..=FRAME) @@ -1093,7 +1100,13 @@ mod tests { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let endpoint = listener.local_addr().unwrap(); let service = Arc::new(Mutex::new( - TransportService::open(&store.path, vec![caller(CALLER, KEY, 4)], 300_000).unwrap(), + TransportService::open( + &store.path, + "fixture-sse", + vec![caller(CALLER, KEY, 4)], + 300_000, + ) + .unwrap(), )); let served = backend.clone(); thread::spawn(move || { diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 157ab20..b350283 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -7,9 +7,9 @@ use sha2::{Digest, Sha256}; use zeroize::Zeroizing; use crate::contract::{ - ContractError, FailureKind, RECEIPT_SCHEMA_ID, RESULT_SCHEMA_ID, Refusal, TransportDisposition, - TransportInvocation, TransportKey, TransportOutcome, TransportReceipt, TransportResult, - canonical_invocation_bytes, require_id, + ContractError, FailureKind, InputItem, ProviderRequest, RECEIPT_SCHEMA_ID, RESULT_SCHEMA_ID, + Refusal, TransportDisposition, TransportEvent, TransportInvocation, TransportKey, + TransportOutcome, TransportReceipt, TransportResult, canonical_invocation_bytes, require_id, }; use crate::envelope::{ EnvelopeKind, MAX_IDENTITY_BYTES, ServiceError, TransportEnvelope, decrypt_invocation, @@ -106,15 +106,66 @@ struct ReplayRecord { state: ReplayState, } +/// 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. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProviderCall { + invocation: TransportInvocation, + request: ProviderRequest, + foreign_reasoning_dropped: u32, + provider: String, +} + +impl ProviderCall { + /// The request as the provider may see it: foreign reasoning items absent, + /// every other item in the caller's order. + pub fn request(&self) -> &ProviderRequest { + &self.request + } + + /// How many foreign reasoning items the service removed from the request. + pub fn foreign_reasoning_dropped(&self) -> u32 { + self.foreign_reasoning_dropped + } + + /// The result of this call. The whole receipt (schema, ids, both digests, + /// model, provider tag and drop count) is derived here, so a backend + /// writes only the events and the outcome. + pub fn transported( + &self, + 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, + }, + ) + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct ExecutionClaim { - invocation: TransportInvocation, + call: ProviderCall, invocation_sha256: [u8; 32], } impl ExecutionClaim { - pub fn invocation(&self) -> &TransportInvocation { - &self.invocation + pub fn call(&self) -> &ProviderCall { + &self.call } } @@ -129,15 +180,23 @@ pub struct TransportService { live: HashSet, replay_store: CultCache, max_expiry_skew_ms: u64, + provider: String, } impl TransportService { + /// `provider` is the tag of the backend this service fronts. The service + /// owns it from here on: it decides which reasoning state is foreign and + /// what provider every receipt names. pub fn open( replay_store_path: &Path, + provider: &str, admissions: impl IntoIterator, max_expiry_skew_ms: u64, ) -> Result { - if max_expiry_skew_ms == 0 { + if max_expiry_skew_ms == 0 + || require_id(provider, "provider").is_err() + || provider == STAND_IN_PROVIDER + { return Err(ServiceError::InvalidAdmission); } let mut callers = HashMap::new(); @@ -195,6 +254,7 @@ impl TransportService { live: HashSet::new(), replay_store, max_expiry_skew_ms, + provider: provider.to_string(), }) } @@ -242,6 +302,11 @@ impl TransportService { if invocation.request.max_output_tokens > caller.max_output_tokens { return self.reply_refusal(&invocation, Refusal::OutputBound, &security); } + let (request, foreign_reasoning_dropped) = + match partition_reasoning(&invocation.request, &self.provider) { + Ok(partition) => partition, + Err(refusal) => return self.reply_refusal(&invocation, refusal, &security), + }; let identity = RequestIdentity { caller_runtime_id: invocation.caller_runtime_id.clone(), @@ -308,7 +373,12 @@ impl TransportService { .map_err(replay_store_error)?; self.live.insert(identity); Ok(TransportAdmission::Execute(Box::new(ExecutionClaim { - invocation, + call: ProviderCall { + invocation, + request, + foreign_reasoning_dropped, + provider: self.provider.clone(), + }, invocation_sha256, }))) } @@ -324,16 +394,19 @@ impl TransportService { result: TransportResult, max_frame_bytes: usize, ) -> Result { - if !matches!( - &result.disposition, - TransportDisposition::Transported { .. } - ) { + 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); } - result.validate_against(&claim.invocation)?; let identity = RequestIdentity { - caller_runtime_id: claim.invocation.caller_runtime_id.clone(), - request_id: claim.invocation.request_id().to_string(), + caller_runtime_id: invocation.caller_runtime_id.clone(), + request_id: invocation.request_id().to_string(), }; if !self.live.contains(&identity) { return Err(ServiceError::NoActiveRequest); @@ -355,7 +428,7 @@ impl TransportService { 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(&claim.invocation, &security, 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 @@ -374,7 +447,7 @@ impl TransportService { request_id: identity.request_id.clone(), connection_key_epoch: caller.connection_key_epoch, invocation_sha256: claim.invocation_sha256, - expires_at_unix_ms: claim.invocation.expires_at_unix_ms, + expires_at_unix_ms: invocation.expires_at_unix_ms, state: ReplayState::Completed { response: response.clone(), }, @@ -393,6 +466,9 @@ impl TransportService { security: &TransportKey, max_frame_bytes: usize, ) -> Result { + let (_, foreign_reasoning_dropped) = + partition_reasoning(&invocation.request, &self.provider) + .map_err(|_| ServiceError::InvalidCompletion)?; let stand_in = encrypt_result( &result_size_failure( &invocation.caller_runtime_id, @@ -400,6 +476,7 @@ impl TransportService { &invocation.request.model, invocation.native_request_sha256, invocation.provider_request_sha256, + foreign_reasoning_dropped, ), security, )?; @@ -439,6 +516,7 @@ pub(crate) fn result_size_failure( model: &str, native_request_sha256: [u8; 32], provider_request_sha256: [u8; 32], + foreign_reasoning_dropped: u32, ) -> TransportResult { TransportResult { schema_id: RESULT_SCHEMA_ID.to_string(), @@ -454,6 +532,7 @@ pub(crate) fn result_size_failure( provider_request_sha256, model: model.to_string(), provider: STAND_IN_PROVIDER.to_string(), + foreign_reasoning_dropped, outcome: TransportOutcome::Failed { kind: FailureKind::ResultSize, message: "provider result exceeded its byte bound".to_string(), @@ -463,6 +542,40 @@ pub(crate) fn result_size_failure( } } +/// The one place foreignness is decided. A reasoning item of another provider +/// in the in-flight exchange (after the last user text; the whole input when +/// there is none) is refused, because dropping it would break the provider's +/// own tool-exchange rules. Earlier foreign items are removed and counted. +/// Foreignness is the provider tag alone: another model of this provider +/// passes through for the provider to judge. +fn partition_reasoning( + request: &ProviderRequest, + provider: &str, +) -> Result<(ProviderRequest, u32), Refusal> { + let in_flight_from = request + .input + .iter() + .rposition(|item| matches!(item, InputItem::UserText { .. })) + .map_or(0, |last_user_text| last_user_text + 1); + let mut input = Vec::with_capacity(request.input.len()); + let mut dropped = 0u32; + for (position, item) in request.input.iter().enumerate() { + if let InputItem::ReasoningState { provider: tag, .. } = item + && tag != provider + { + if position >= in_flight_from { + return Err(Refusal::ForeignReasoningState); + } + dropped += 1; + continue; + } + input.push(item.clone()); + } + let mut filtered = request.clone(); + filtered.input = input; + Ok((filtered, dropped)) +} + fn replay_key(identity: &RequestIdentity) -> String { let mut hasher = Sha256::new(); hasher.update((identity.caller_runtime_id.len() as u64).to_be_bytes()); @@ -509,6 +622,7 @@ mod tests { assert!(matches!( TransportService::open( &store.path, + "fixture-sse", [ caller("epiphany-yggdrasil", "shared-key", 1), caller("ghostlight-yggdrasil", "shared-key", 1), @@ -710,6 +824,7 @@ mod tests { assert!(matches!( TransportService::open( &store.path, + "fixture-sse", [CallerAdmission::new( "epiphany-yggdrasil", "rotated-connection-key", @@ -818,7 +933,7 @@ mod tests { 32_768, ) .unwrap(); - TransportService::open(&store.path, [admission], 5_000) + TransportService::open(&store.path, "fixture-sse", [admission], 5_000) .unwrap() .begin(&envelope, 1_000, FRAME) }; @@ -887,7 +1002,15 @@ mod tests { fn restart_refuses_each_malformed_replay_record() { let control = TestStore::new(); seed(&control.path, &seeded_key(), |_| {}); - assert!(TransportService::open(&control.path, [caller(CALLER, KEY, 1)], 5_000).is_ok()); + assert!( + TransportService::open( + &control.path, + "fixture-sse", + [caller(CALLER, KEY, 1)], + 5_000 + ) + .is_ok() + ); type Edit = fn(&mut ReplayRecord); let cases: [(&str, Edit); 3] = [ @@ -905,7 +1028,12 @@ mod tests { seed(&store.path, &key, edit); assert!( matches!( - TransportService::open(&store.path, [caller(CALLER, KEY, 1)], 5_000), + TransportService::open( + &store.path, + "fixture-sse", + [caller(CALLER, KEY, 1)], + 5_000 + ), Err(ServiceError::InvalidReplayRecord) ), "{name}" @@ -921,16 +1049,7 @@ mod tests { let TransportAdmission::Execute(claim) = admission else { panic!("first invocation must execute"); }; - let result = TransportResult::transported( - claim.invocation(), - vec![TransportEvent { - sequence: 0, - payload: TransportEventPayload::TextDelta { - text: "done".to_string(), - }, - }], - receipt(claim.invocation()), - ); + let result = text_result(claim.call(), "done"); let mut forged = (*claim).clone(); forged.invocation_sha256 = [1; 32]; @@ -1071,22 +1190,15 @@ mod tests { } } - fn text_result( - invocation: &TransportInvocation, - text: &str, - transport: &str, - ) -> TransportResult { - let mut receipt = receipt(invocation); - receipt.provider = transport.to_string(); - TransportResult::transported( - invocation, + fn text_result(call: &ProviderCall, text: &str) -> TransportResult { + call.transported( vec![TransportEvent { sequence: 0, payload: TransportEventPayload::TextDelta { text: text.to_string(), }, }], - receipt, + receipt(&call.invocation).outcome, ) } @@ -1095,7 +1207,7 @@ mod tests { let store = TestStore::new(); let mut service = open_service(&store, 8); let (claim, invocation) = claim_for(&mut service, "stand-in"); - let big = text_result(&invocation, &"y".repeat(4000), &"t".repeat(5000)); + let big = text_result(claim.call(), &"y".repeat(4000)); let response = service.complete(*claim, big, 2000).unwrap(); assert!(encode_result_frame(&response).unwrap().len() <= 2000); let stand_in = decrypt_result(&response, &security(KEY), &invocation).unwrap(); @@ -1107,6 +1219,7 @@ mod tests { &invocation.request.model, invocation.native_request_sha256, invocation.provider_request_sha256, + 0, ) ); let TransportDisposition::Transported { receipt, .. } = &stand_in.disposition else { @@ -1129,7 +1242,7 @@ mod tests { let store = TestStore::new(); let mut service = open_service(&store, 8); let (claim, invocation) = claim_for(&mut service, "fits"); - let small = text_result(&invocation, "done", "fixture-sse"); + let small = text_result(claim.call(), "done"); let response = service.complete(*claim, small.clone(), 1 << 20).unwrap(); assert_eq!( decrypt_result(&response, &security(KEY), &invocation).unwrap(), @@ -1142,7 +1255,7 @@ mod tests { let store = TestStore::new(); let mut service = open_service(&store, 1); let (claim, invocation) = claim_for(&mut service, "wedge"); - let big = text_result(&invocation, "y", "fixture-sse"); + let big = text_result(claim.call(), "y"); assert_eq!( service.complete(*claim, big, 10).unwrap_err(), ServiceError::FrameBound @@ -1169,7 +1282,7 @@ mod tests { let store = TestStore::new(); let mut service = open_service(&store, 8); let (claim, invocation) = claim_for(&mut service, "restart"); - let big = text_result(&invocation, &"y".repeat(4000), "fixture-sse"); + let big = text_result(claim.call(), &"y".repeat(4000)); let recorded = service.complete(*claim, big, FRAME).unwrap(); assert!(encode_result_frame(&recorded).unwrap().len() > 2000); drop(service); @@ -1200,6 +1313,7 @@ mod tests { &invocation.request.model, invocation.native_request_sha256, invocation.provider_request_sha256, + 0, ) ); // The stand-in is what is recorded now, byte for byte, at any frame. @@ -1217,7 +1331,7 @@ mod tests { let store = TestStore::new(); let mut service = open_service(&store, 8); let (claim, invocation) = claim_for(&mut service, "no-room"); - let big = text_result(&invocation, "y", "fixture-sse"); + let big = text_result(claim.call(), "y"); service.complete(*claim, big, FRAME).unwrap(); assert_eq!( service.begin(&raw_envelope(&invocation), 1_000, 10).err(), @@ -1230,7 +1344,7 @@ mod tests { let store = TestStore::new(); let mut service = open_service(&store, 8); let (claim, invocation) = claim_for(&mut service, "nothing-fits"); - let big = text_result(&invocation, "y", "fixture-sse"); + let big = text_result(claim.call(), "y"); assert_eq!( service.complete(*claim, big, 10).unwrap_err(), ServiceError::FrameBound @@ -1271,4 +1385,263 @@ mod tests { assert_eq!(contract_refusal(&other), Refusal::Malformed, "{other:?}"); } } + + fn user(text: &str) -> InputItem { + InputItem::UserText { + text: text.to_string(), + } + } + + fn state(provider: &str, model: &str, byte: u8) -> InputItem { + InputItem::ReasoningState { + provider: provider.to_string(), + model: model.to_string(), + state: vec![byte], + } + } + + fn admit_input( + service: &mut TransportService, + request_id: &str, + input: Vec, + ) -> (TransportAdmission, TransportInvocation) { + run(service, request_id, |request| request.input = input) + } + + #[test] + fn foreign_reasoning_in_flight_is_refused_and_never_executed() { + let store = TestStore::new(); + let mut service = open_service(&store, 1); + let (admission, invocation) = admit_input( + &mut service, + "in-flight", + vec![ + user("question"), + state("fixture-sse", "gpt-5.4", 1), + state("other-provider", "x", 2), + InputItem::ToolResult { + call_id: "call-1".to_string(), + output: "evidence".to_string(), + }, + ], + ); + assert_eq!( + refusal_of(admission, &invocation), + Refusal::ForeignReasoningState + ); + // Nothing was recorded or held: the caller's one slot is free and the + // same request id is not a replay. + let (admission, _) = admit_input(&mut service, "in-flight", vec![user("question")]); + assert!(matches!(admission, TransportAdmission::Execute(_))); + } + + #[test] + fn foreign_reasoning_with_no_user_text_is_in_flight() { + let store = TestStore::new(); + let mut service = open_service(&store, 4); + let (admission, invocation) = admit_input( + &mut service, + "no-user-text", + vec![ + state("other-provider", "x", 2), + InputItem::AssistantText { + text: "earlier".to_string(), + }, + ], + ); + assert_eq!( + refusal_of(admission, &invocation), + Refusal::ForeignReasoningState + ); + // The boundary is the position after the last user text: a foreign item + // directly after it is in flight too. + let (admission, invocation) = admit_input( + &mut service, + "just-after", + vec![user("a"), state("other-provider", "x", 2)], + ); + assert_eq!( + refusal_of(admission, &invocation), + Refusal::ForeignReasoningState + ); + } + + #[test] + fn foreign_reasoning_in_completed_turns_is_dropped_and_counted() { + let store = TestStore::new(); + let mut service = open_service(&store, 4); + let own = state("fixture-sse", "gpt-5.4", 3); + let tool_call = InputItem::ToolCall { + call_id: "call-1".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }; + let (admission, invocation) = admit_input( + &mut service, + "completed-turns", + vec![ + state("other-provider", "x", 1), + user("first"), + own.clone(), + state("third-provider", "y", 2), + tool_call.clone(), + user("second"), + tool_call.clone(), + own.clone(), + ], + ); + let TransportAdmission::Execute(claim) = admission else { + panic!("foreign state in completed turns must not stop the request"); + }; + assert_eq!( + claim.call().request().input, + vec![ + user("first"), + own.clone(), + tool_call.clone(), + user("second"), + tool_call, + own, + ] + ); + assert_eq!(claim.call().foreign_reasoning_dropped(), 2); + // The caller's digest binds what the caller sent, not the filtered view, + // and the completion below is accepted against it. + assert_ne!( + invocation.provider_request_sha256, + provider_request_sha256(claim.call().request()).unwrap() + ); + let result = text_result(claim.call(), "done"); + let response = service.complete(*claim, result, FRAME).unwrap(); + let TransportDisposition::Transported { receipt, .. } = + decrypt_result(&response, &security(KEY), &invocation) + .unwrap() + .disposition + else { + panic!("completed"); + }; + assert_eq!(receipt.foreign_reasoning_dropped, 2); + } + + #[test] + fn same_provider_other_model_reasoning_passes_through() { + let store = TestStore::new(); + let mut service = open_service(&store, 4); + let other_model = state("fixture-sse", "some-other-model", 9); + let (admission, _) = admit_input( + &mut service, + "other-model", + vec![other_model.clone(), user("first"), other_model.clone()], + ); + let TransportAdmission::Execute(claim) = admission else { + panic!("an own-provider item is never foreign"); + }; + assert_eq!( + claim.call().request().input, + vec![other_model.clone(), user("first"), other_model] + ); + assert_eq!(claim.call().foreign_reasoning_dropped(), 0); + } + + #[test] + fn open_refuses_a_blank_or_stand_in_provider_tag() { + let store = TestStore::new(); + for tag in ["", " ", STAND_IN_PROVIDER] { + assert!( + matches!( + TransportService::open(&store.path, tag, [caller(CALLER, KEY, 1)], 5_000), + Err(ServiceError::InvalidAdmission) + ), + "{tag:?}" + ); + } + assert!( + TransportService::open(&store.path, "fixture-sse", [caller(CALLER, KEY, 1)], 5_000) + .is_ok() + ); + } + + #[test] + fn complete_refuses_a_hand_built_receipt_with_another_provider_or_count() { + 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 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 + ); + } + // A refused completion leaves the request live, and the derived result passes. + assert!(service.complete(*claim, good, FRAME).is_ok()); + } + + #[test] + fn stand_in_carries_the_drop_count() { + let store = TestStore::new(); + let mut service = open_service(&store, 4); + let input = vec![state("other-provider", "x", 1), user("first")]; + let (admission, invocation) = admit_input(&mut service, "stand-in-count", input.clone()); + let TransportAdmission::Execute(claim) = admission else { + panic!("expected an execution claim"); + }; + let big = text_result(claim.call(), &"y".repeat(4000)); + let response = service.complete(*claim, big, 2000).unwrap(); + let expected = result_size_failure( + CALLER, + "stand-in-count", + &invocation.request.model, + invocation.native_request_sha256, + invocation.provider_request_sha256, + 1, + ); + assert_eq!( + decrypt_result(&response, &security(KEY), &invocation).unwrap(), + expected + ); + + // A recorded wide result replayed at a smaller frame keeps the count too. + let (admission, invocation) = admit_input(&mut service, "replay-count", input); + let TransportAdmission::Execute(claim) = admission else { + panic!("expected an execution claim"); + }; + let big = text_result(claim.call(), &"y".repeat(4000)); + service.complete(*claim, big, FRAME).unwrap(); + let TransportAdmission::Reply(smaller) = service + .begin(&raw_envelope(&invocation), 1_000, 2000) + .unwrap() + else { + panic!("expected a replay"); + }; + let TransportDisposition::Transported { receipt, .. } = + decrypt_result(&smaller, &security(KEY), &invocation) + .unwrap() + .disposition + else { + panic!("expected the stand-in"); + }; + assert_eq!(receipt.provider, STAND_IN_PROVIDER); + assert_eq!(receipt.foreign_reasoning_dropped, 1); + } } diff --git a/crates/model-connector/tests/fixtures/wire_v3/README.md b/crates/model-connector/tests/fixtures/wire_v3/README.md index 918df33..2751c8f 100644 --- a/crates/model-connector/tests/fixtures/wire_v3/README.md +++ b/crates/model-connector/tests/fixtures/wire_v3/README.md @@ -5,7 +5,8 @@ produced by the kit at the revision that added them. They pin drift; the types i `src/contract.rs` own the shape, and `schemas/` publishes it. - `provider_request.mp`, `transport_invocation.mp`, `transport_result_transported.mp`, - `transport_result_refused.mp`, `transport_receipt_failed.mp`, `transport_envelope.mp`: + `transport_result_refused.mp`, `transport_receipt_failed.mp`, `transport_envelope.mp`, + `input_reasoning_state.mp` (an `InputItem::ReasoningState`, its state as MessagePack `bin`): the plaintext documents. - `carriage_invocation.frame`: a `cultnet.operation_request.v0` for service `gamecult.model.connector`, operation `invoke`, whose payload is the base64 of an envelope diff --git a/crates/model-connector/tests/fixtures/wire_v3/input_reasoning_state.mp b/crates/model-connector/tests/fixtures/wire_v3/input_reasoning_state.mp new file mode 100644 index 0000000..c6eae0c --- /dev/null +++ b/crates/model-connector/tests/fixtures/wire_v3/input_reasoning_state.mp @@ -0,0 +1 @@ +ReasoningStateprovideranthropic-messagesmodelclaude-opus-4-5stateĀ \ No newline at end of file diff --git a/crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp b/crates/model-connector/tests/fixtures/wire_v3/transport_receipt_failed.mp index 4e0f1b1e9fd852297b60da82763d620e0cd9e8d0..0c5feb28476fc1102939f7a750b06510305c5243 100644 GIT binary patch delta 42 xcmaFH)X2=(xsg$hQF>=ueo<;>dR}}{YGQGIUS?i;d`eM%K|yK?(`0!@cK|~&4)g#3 delta 14 VcmZon;k=;`X~?CtLF z@bU8V^!4`l`1$(#{Qds_f~a$4adl;GV`T!7!$<*%vkU, + executions: AtomicUsize, + seen: Mutex>, +} + +impl RecordingBackend { + fn new(events: Vec) -> Arc { + Arc::new(Self { + events, + executions: AtomicUsize::new(0), + seen: Mutex::new(Vec::new()), + }) + } + + fn seen(&self) -> Vec { + self.seen.lock().unwrap().clone() + } +} + +impl ProviderBackend for RecordingBackend { + fn ready(&self) -> bool { + true + } + + fn provider(&self) -> &str { + PROVIDER + } + + fn execute(&self, call: &ProviderCall) -> TransportResult { + self.executions.fetch_add(1, Ordering::SeqCst); + self.seen.lock().unwrap().push(call.request().clone()); + call.transported( + self.events + .iter() + .cloned() + .enumerate() + .map(|(sequence, payload)| TransportEvent { + sequence: sequence as u64, + payload, + }) + .collect(), + TransportOutcome::Completed { + provider_response_id: Some("msg-1".to_string()), + stop_reason: StopReason::ToolUse, + usage: Usage { + input_tokens: 1, + cache_read_tokens: 0, + cache_write_tokens: 0, + output_tokens: 1, + reasoning_tokens: None, + }, + }, + ) + } +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as u64 +} + +/// A running serve loop and a client for it. +fn start(backend: Arc) -> ConnectorClient { + static SEQUENCE: AtomicUsize = AtomicUsize::new(0); + let root: PathBuf = std::env::temp_dir().join(format!( + "model-connector-reasoning-{}-{}", + std::process::id(), + SEQUENCE.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir_all(&root).unwrap(); + let store = root.join("replay.cc"); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = listener.local_addr().unwrap(); + let admission = CallerAdmission::new( + CALLER, + KEY, + 1, + ["claude-opus-4-5".to_string()], + 4, + 64 * 1024, + 32_768, + ) + .unwrap(); + let limits = ServeLimits { + max_frame_bytes: FRAME, + max_connections: 8, + socket_timeout_ms: 5_000, + max_expiry_skew_ms: 300_000, + }; + thread::spawn(move || serve(listener, limits, vec![admission], &store, backend)); + ConnectorClient::new(endpoint, KEY, FRAME, Some(Duration::from_secs(5))).unwrap() +} + +fn invocation(request_id: &str, input: Vec) -> TransportInvocation { + let mut request = ProviderRequest::new( + request_id, + "conversation-1", + "claude-opus-4-5", + "Return the typed result.", + 4_096, + ); + request.input = input; + TransportInvocation::new(CALLER, now_ms() + 60_000, [7; 32], request).unwrap() +} + +fn user(text: &str) -> InputItem { + InputItem::UserText { + text: text.to_string(), + } +} + +fn state(provider: &str, byte: u8) -> InputItem { + InputItem::ReasoningState { + provider: provider.to_string(), + model: "claude-opus-4-5".to_string(), + state: vec![0x80 | byte, byte], + } +} + +fn receipt_of(result: &TransportResult) -> &TransportReceipt { + let TransportDisposition::Transported { receipt, .. } = &result.disposition else { + panic!( + "expected a transported result, got {:?}", + result.disposition + ); + }; + receipt +} + +#[test] +fn foreign_reasoning_in_flight_is_refused_and_never_executed() { + let backend = RecordingBackend::new(vec![TransportEventPayload::TextDelta { + text: "done".to_string(), + }]); + let client = start(backend.clone()); + let result = client + .execute(&invocation( + "foreign-in-flight", + vec![ + user("question"), + InputItem::ToolCall { + call_id: "call-1".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + state("another-provider", 1), + InputItem::ToolResult { + call_id: "call-1".to_string(), + output: "evidence".to_string(), + }, + ], + )) + .unwrap(); + assert_eq!( + result.disposition, + TransportDisposition::Refused(Refusal::ForeignReasoningState) + ); + assert_eq!(backend.executions.load(Ordering::SeqCst), 0); +} + +#[test] +fn a_backend_never_sees_foreign_reasoning_state() { + let backend = RecordingBackend::new(vec![TransportEventPayload::TextDelta { + text: "done".to_string(), + }]); + let client = start(backend.clone()); + let own = state(PROVIDER, 2); + let result = client + .execute(&invocation( + "foreign-earlier", + vec![ + state("another-provider", 1), + user("first"), + own.clone(), + state("a-third-provider", 3), + user("second"), + ], + )) + .unwrap(); + assert_eq!(receipt_of(&result).foreign_reasoning_dropped, 2); + assert_eq!(receipt_of(&result).provider, PROVIDER); + assert_eq!( + backend.seen()[0].input, + vec![user("first"), own, user("second")] + ); +} + +#[test] +fn replayed_transcript_reaches_the_same_provider_verbatim() { + let thinking = vec![0x80, 0xfe, 0x00, 0xff]; + let backend = RecordingBackend::new(vec![ + TransportEventPayload::TextDelta { + text: "let me ".to_string(), + }, + TransportEventPayload::TextDelta { + text: "look".to_string(), + }, + TransportEventPayload::ReasoningState { + state: thinking.clone(), + }, + TransportEventPayload::ToolCall { + call_id: "call-1".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + ]); + let client = start(backend.clone()); + let first = client + .execute(&invocation("turn-1", vec![user("question")])) + .unwrap(); + let added = first.transcript_items(); + assert_eq!( + added, + vec![ + InputItem::AssistantText { + text: "let me look".to_string() + }, + InputItem::ReasoningState { + provider: PROVIDER.to_string(), + model: "claude-opus-4-5".to_string(), + state: thinking, + }, + InputItem::ToolCall { + call_id: "call-1".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + ] + ); + + // The caller appends the items untouched and answers the tool call; the + // next call is in flight on the same provider and is admitted as it is. + let mut next = vec![user("question")]; + next.extend(added.clone()); + next.push(InputItem::ToolResult { + call_id: "call-1".to_string(), + output: "evidence".to_string(), + }); + let second = client.execute(&invocation("turn-2", next.clone())).unwrap(); + assert_eq!(receipt_of(&second).foreign_reasoning_dropped, 0); + assert_eq!(backend.seen()[1].input, next); +} diff --git a/crates/model-connector/tests/wire_golden_v3.rs b/crates/model-connector/tests/wire_golden_v3.rs index 5f91513..dddf9ef 100644 --- a/crates/model-connector/tests/wire_golden_v3.rs +++ b/crates/model-connector/tests/wire_golden_v3.rs @@ -94,6 +94,7 @@ fn receipt(outcome: TransportOutcome) -> TransportReceipt { provider_request_sha256: invocation.provider_request_sha256, model: invocation.request.model.clone(), provider: "anthropic-messages".to_string(), + foreign_reasoning_dropped: 2, outcome, } } @@ -118,26 +119,46 @@ fn wire_golden_v3() { assert_wire("provider_request.mp", &invocation.request); assert_wire("transport_invocation.mp", &invocation); - let transported = TransportResult::transported( - &invocation, - vec![ - TransportEvent { - sequence: 0, - payload: TransportEventPayload::TextDelta { - text: "thinking".to_string(), - }, - }, - TransportEvent { - sequence: 1, - payload: TransportEventPayload::ToolCall { - call_id: "call-2".to_string(), - name: "read_source".to_string(), - arguments: "{}".to_string(), - }, - }, - ], - receipt(completed()), + // The reasoning bytes are all >= 0x80 so a byte-array encoding cannot pass + // for MessagePack bin. + let reasoning = || (0x80..=0xff_u8).collect::>(); + assert_wire( + "input_reasoning_state.mp", + &InputItem::ReasoningState { + provider: "anthropic-messages".to_string(), + model: "claude-opus-4-5".to_string(), + state: reasoning(), + }, ); + + let transported = TransportResult { + schema_id: RESULT_SCHEMA_ID.to_string(), + request_id: invocation.request_id().to_string(), + caller_runtime_id: CALLER.to_string(), + disposition: TransportDisposition::Transported { + events: vec![ + TransportEvent { + sequence: 0, + payload: TransportEventPayload::TextDelta { + text: "thinking".to_string(), + }, + }, + TransportEvent { + sequence: 1, + payload: TransportEventPayload::ReasoningState { state: reasoning() }, + }, + TransportEvent { + sequence: 2, + payload: TransportEventPayload::ToolCall { + call_id: "call-2".to_string(), + name: "read_source".to_string(), + arguments: "{}".to_string(), + }, + }, + ], + receipt: Box::new(receipt(completed())), + }, + }; assert_eq!(transported.validate_against(&invocation), Ok(())); assert_wire("transport_result_transported.mp", &transported); @@ -212,3 +233,12 @@ fn the_golden_carriage_frame_decrypts_and_validates_under_the_fixed_key() { assert_eq!(decrypted, invocation); assert_eq!(decrypted.validate(1_799_999_999_000, 5_000), Ok(())); } + +/// The reasoning bytes travel as MessagePack `bin` (0xc4 length marker), never +/// as an array of integers. +#[test] +fn reasoning_state_is_messagepack_bin() { + let golden = std::fs::read(fixtures().join("input_reasoning_state.mp")).unwrap(); + let marker = golden.windows(3).any(|window| window == [0xc4, 0x80, 0x80]); + assert!(marker, "state must be a bin 8 of 128 bytes starting 0x80"); +} diff --git a/schemas/provider_request.json b/schemas/provider_request.json index d4fb346..0ecf545 100644 --- a/schemas/provider_request.json +++ b/schemas/provider_request.json @@ -186,6 +186,38 @@ "required": [ "ToolResult" ] + }, + { + "description": "One provider reasoning block, opaque to every caller: `state` is the\nconnector's own encoding of the whole block. Callers replay it\nverbatim in transcript order and never inspect or branch on it; only\nthe connector of `provider` reads it, and the service drops or refuses\nthe state of any other provider before a backend sees the request.", + "type": "object", + "properties": { + "ReasoningState": { + "type": "object", + "properties": { + "model": { + "type": "string" + }, + "provider": { + "type": "string" + }, + "state": { + "description": "MessagePack bin: raw bytes, never an array of integers", + "type": "string", + "format": "binary" + } + }, + "additionalProperties": false, + "required": [ + "provider", + "model", + "state" + ] + } + }, + "additionalProperties": false, + "required": [ + "ReasoningState" + ] } ] }, diff --git a/schemas/transport_envelope.json b/schemas/transport_envelope.json index 1983423..69a97d3 100644 --- a/schemas/transport_envelope.json +++ b/schemas/transport_envelope.json @@ -7,7 +7,7 @@ "type": "string" }, "ciphertext": { - "description": "MessagePack bin: the raw ciphertext bytes, never an array of integers", + "description": "MessagePack bin: raw bytes, never an array of integers", "type": "string", "format": "binary" }, diff --git a/schemas/transport_invocation.json b/schemas/transport_invocation.json index 9e899b1..5435b2e 100644 --- a/schemas/transport_invocation.json +++ b/schemas/transport_invocation.json @@ -154,6 +154,38 @@ "required": [ "ToolResult" ] + }, + { + "description": "One provider reasoning block, opaque to every caller: `state` is the\nconnector's own encoding of the whole block. Callers replay it\nverbatim in transcript order and never inspect or branch on it; only\nthe connector of `provider` reads it, and the service drops or refuses\nthe state of any other provider before a backend sees the request.", + "type": "object", + "properties": { + "ReasoningState": { + "type": "object", + "properties": { + "model": { + "type": "string" + }, + "provider": { + "type": "string" + }, + "state": { + "description": "MessagePack bin: raw bytes, never an array of integers", + "type": "string", + "format": "binary" + } + }, + "additionalProperties": false, + "required": [ + "provider", + "model", + "state" + ] + } + }, + "additionalProperties": false, + "required": [ + "ReasoningState" + ] } ] }, diff --git a/schemas/transport_receipt.json b/schemas/transport_receipt.json index 68916ed..722a42a 100644 --- a/schemas/transport_receipt.json +++ b/schemas/transport_receipt.json @@ -6,6 +6,12 @@ "caller_runtime_id": { "type": "string" }, + "foreign_reasoning_dropped": { + "description": "How many foreign `ReasoningState` items the service removed from the\nrequest before the backend saw it.", + "type": "integer", + "format": "uint32", + "minimum": 0 + }, "model": { "type": "string" }, @@ -53,6 +59,7 @@ "provider_request_sha256", "model", "provider", + "foreign_reasoning_dropped", "outcome" ], "$defs": { diff --git a/schemas/transport_result.json b/schemas/transport_result.json index ceb1e42..25d97ab 100644 --- a/schemas/transport_result.json +++ b/schemas/transport_result.json @@ -38,19 +38,28 @@ ] }, "Refusal": { - "type": "string", - "enum": [ - "Expired", - "IdentitySubstitution", - "ProviderDigestSubstitution", - "ModelNotAllowed", - "OutputBound", - "PayloadBound", - "Capacity", - "InFlight", - "Indeterminate", - "ReplayConflict", - "Malformed" + "oneOf": [ + { + "type": "string", + "enum": [ + "Expired", + "IdentitySubstitution", + "ProviderDigestSubstitution", + "ModelNotAllowed", + "OutputBound", + "PayloadBound", + "Capacity", + "InFlight", + "Indeterminate", + "ReplayConflict", + "Malformed" + ] + }, + { + "description": "A `ReasoningState` of another provider sits in the in-flight tool\nexchange, where dropping it would break the provider's own rules.", + "type": "string", + "const": "ForeignReasoningState" + } ] }, "StopReason": { @@ -178,6 +187,30 @@ "required": [ "ToolCall" ] + }, + { + "description": "One reasoning block of the provider that answered. It carries no\nprovider or model: the receipt names them, so a backend cannot tag its\nown state.", + "type": "object", + "properties": { + "ReasoningState": { + "type": "object", + "properties": { + "state": { + "description": "MessagePack bin: raw bytes, never an array of integers", + "type": "string", + "format": "binary" + } + }, + "additionalProperties": false, + "required": [ + "state" + ] + } + }, + "additionalProperties": false, + "required": [ + "ReasoningState" + ] } ] }, @@ -247,6 +280,12 @@ "caller_runtime_id": { "type": "string" }, + "foreign_reasoning_dropped": { + "description": "How many foreign `ReasoningState` items the service removed from the\nrequest before the backend saw it.", + "type": "integer", + "format": "uint32", + "minimum": 0 + }, "model": { "type": "string" }, @@ -294,6 +333,7 @@ "provider_request_sha256", "model", "provider", + "foreign_reasoning_dropped", "outcome" ] }, From 5762dabe795fd5b1130c42560b0218018e4ef614 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 13:10:25 +0100 Subject: [PATCH 2/5] 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 + ); + } +} From c3f503e16056e5031d35b316fa2877c9d05f5d2c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 13:10:26 +0100 Subject: [PATCH 3/5] connector-reasoning-item fix: Refusal publishes as one flat string enum again; a test pins the shape The doc comment on ForeignReasoningState made schemars emit a oneOf; a plain comment keeps the closed set one enum. Finding refusal-schema-split. Co-Authored-By: Claude Opus 5.5 --- crates/model-connector/src/contract.rs | 5 +-- .../tests/published_schemas.rs | 15 ++++++++ schemas/transport_result.json | 36 ++++++++----------- 3 files changed, 32 insertions(+), 24 deletions(-) diff --git a/crates/model-connector/src/contract.rs b/crates/model-connector/src/contract.rs index bb6459e..3062fe4 100644 --- a/crates/model-connector/src/contract.rs +++ b/crates/model-connector/src/contract.rs @@ -540,8 +540,9 @@ pub enum Refusal { Indeterminate, ReplayConflict, Malformed, - /// A `ReasoningState` of another provider sits in the in-flight tool - /// exchange, where dropping it would break the provider's own rules. + // A `ReasoningState` of another provider sits in the in-flight tool + // exchange, where dropping it would break the provider's own rules. A plain + // comment: a doc comment would split the published schema into a union. ForeignReasoningState, } diff --git a/crates/model-connector/tests/published_schemas.rs b/crates/model-connector/tests/published_schemas.rs index 4634280..8a1bf47 100644 --- a/crates/model-connector/tests/published_schemas.rs +++ b/crates/model-connector/tests/published_schemas.rs @@ -98,3 +98,18 @@ fn the_published_ciphertext_is_what_the_wire_sends() { .unwrap(); assert!(matches!(wire, rmpv::Value::Binary(bytes) if bytes.len() == 40)); } + +#[test] +fn the_refusal_schema_is_one_flat_string_enum() { + let schema: serde_json::Value = serde_json::from_str(&render::()).unwrap(); + assert!(schema.get("oneOf").is_none(), "{schema}"); + assert_eq!(schema["type"], "string"); + let variants: Vec<&str> = schema["enum"] + .as_array() + .expect("a closed set of strings") + .iter() + .map(|value| value.as_str().unwrap()) + .collect(); + assert_eq!(variants.len(), 12); + assert!(variants.contains(&"ForeignReasoningState")); +} diff --git a/schemas/transport_result.json b/schemas/transport_result.json index 25d97ab..acdf6b6 100644 --- a/schemas/transport_result.json +++ b/schemas/transport_result.json @@ -38,28 +38,20 @@ ] }, "Refusal": { - "oneOf": [ - { - "type": "string", - "enum": [ - "Expired", - "IdentitySubstitution", - "ProviderDigestSubstitution", - "ModelNotAllowed", - "OutputBound", - "PayloadBound", - "Capacity", - "InFlight", - "Indeterminate", - "ReplayConflict", - "Malformed" - ] - }, - { - "description": "A `ReasoningState` of another provider sits in the in-flight tool\nexchange, where dropping it would break the provider's own rules.", - "type": "string", - "const": "ForeignReasoningState" - } + "type": "string", + "enum": [ + "Expired", + "IdentitySubstitution", + "ProviderDigestSubstitution", + "ModelNotAllowed", + "OutputBound", + "PayloadBound", + "Capacity", + "InFlight", + "Indeterminate", + "ReplayConflict", + "Malformed", + "ForeignReasoningState" ] }, "StopReason": { From 72a99b6ef3db619c5ba349461339bd76e7802e86 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 13:11:43 +0100 Subject: [PATCH 4/5] connector-reasoning-item fix: result_size_failure is a test spelling of stand_in_failure The default feature set builds service.rs for tests only; clippy -D warnings flagged the wrapper dead there. Co-Authored-By: Claude Opus 5.5 --- crates/model-connector/src/service.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 68825dc..4d59f75 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -536,6 +536,7 @@ const STAND_IN_PROVIDER: &str = "model-connector"; /// The failed result that stands in for one whose sealed frame cannot be /// carried: the same shape a provider failure takes, with nothing a provider /// controls but the identities the invocation already bound. +#[cfg(test)] pub(crate) fn result_size_failure( caller_runtime_id: &str, request_id: &str, From d915f6757f34e82268127f368292727a16c8f2d7 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 13:15:30 +0100 Subject: [PATCH 5/5] connector-reasoning-item fix: cargo fmt Co-Authored-By: Claude Opus 5.5 --- crates/model-connector/src/service.rs | 25 +++++++++++++++++++------ 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 4d59f75..9a2ea2a 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -435,17 +435,27 @@ impl TransportService { 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); - })? + self.seal_stand_in( + invocation, + &security, + max_frame_bytes, + FailureKind::Provider, + ) + .inspect_err(|_| { + self.live.remove(&identity); + })? }; 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) + .seal_stand_in( + invocation, + &security, + max_frame_bytes, + FailureKind::ResultSize, + ) .inspect_err(|_| { self.live.remove(&identity); })?; @@ -1701,7 +1711,10 @@ mod tests { panic!("expected an execution claim"); }; assert_eq!(claim.call().foreign_reasoning_dropped(), 1); - for text in [format!("{:?}", claim.call()), format!("{:#?}", claim.call())] { + for text in [ + format!("{:?}", claim.call()), + format!("{:#?}", claim.call()), + ] { assert!(!text.contains(canary), "{text}"); assert!(!text.contains("199, 200, 201, 202"), "{text}"); }