From 8d163471ee10c01d741fe9aba322d2856a4b1f44 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 13 Oct 2026 12:01:13 -0700 Subject: [PATCH 1/4] row 155 (L1#3): a model's `protocol` override reaches the llm plane and is served, fail-closed (#51) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Row-159 audit plane-llm L1#3: a model entry's `protocol` (#51, THE DESIGN §2 BEDROCK-INVOKE ruling; CHANGELOG "a model entry takes `protocol`") was dropped by the root's section builder and never read by the plane, so the plane wrote that model's requests in its provider's dialect and accepted an unknown override. The root now hands the member through (`kernel_sections`), and the plane resolves model override, else provider default, else the default dialect, refusing a dialect it does not serve naming the model. RED-able: exchange_shaping::a_models_protocol_overrides_its_providers_and_an_unknown_one_is_refused, root door_steps::a_models_protocol_override_is_handed_to_the_plane. --- .../busbar-plane-llm/src/exchange/shaping.rs | 17 +++++++++-- .../tests/exchange_shaping.rs | 25 ++++++++++++++++ crates/busbar/src/root/door_steps.rs | 6 ++++ crates/busbar/src/root/tests/door_steps.rs | 30 +++++++++++++++++++ 4 files changed, 75 insertions(+), 3 deletions(-) diff --git a/crates/busbar-plane-llm/src/exchange/shaping.rs b/crates/busbar-plane-llm/src/exchange/shaping.rs index 9ab04f6d47..4fee68fa48 100644 --- a/crates/busbar-plane-llm/src/exchange/shaping.rs +++ b/crates/busbar-plane-llm/src/exchange/shaping.rs @@ -214,6 +214,10 @@ struct ModelCfg { reasoning: Option, #[serde(default)] prompt_caching: Option, + /// The model's dialect override (#51): the wire protocol this model is spoken in when it + /// differs from its provider's; absent = the provider's. + #[serde(default)] + protocol: Option, } #[derive(Deserialize)] @@ -415,12 +419,19 @@ impl Shaping { m.provider ) })?; - let asked = p.protocol.as_deref().unwrap_or(DEFAULT_PROTOCOL); + // #51: the model's own `protocol` overrides its provider's default; the plane serves the + // resolved dialect only if it knows it (fail-closed). + let asked = m + .protocol + .as_deref() + .or(p.protocol.as_deref()) + .unwrap_or(DEFAULT_PROTOCOL); let decl = DECLS .iter() .find(|d| d.name == asked && d.codec.is_some()) - .ok_or_else(|| { - format!("provider '{}' uses unknown protocol '{asked}'", m.provider) + .ok_or_else(|| match m.protocol { + Some(_) => format!("model '{name}' uses unknown protocol '{asked}'"), + None => format!("provider '{}' uses unknown protocol '{asked}'", m.provider), })?; let dialect = decl.name; let statics = match p.auth.as_deref() { diff --git a/crates/busbar-plane-llm/tests/exchange_shaping.rs b/crates/busbar-plane-llm/tests/exchange_shaping.rs index 260a41726c..f3ece552fe 100644 --- a/crates/busbar-plane-llm/tests/exchange_shaping.rs +++ b/crates/busbar-plane-llm/tests/exchange_shaping.rs @@ -112,3 +112,28 @@ fn the_glob_takes_prefix_suffix_and_middle_wildcards() { assert!(!glob_match("a*c*e", "abde")); assert!(glob_match("exact", "exact") && !glob_match("exact", "exactly")); } + +/// A MODEL'S OWN `protocol` OVERRIDES ITS PROVIDER'S (#51, THE DESIGN §2 BEDROCK-INVOKE ruling): the +/// model is spoken in the dialect it names, a sibling model on the same provider keeps the +/// provider's, and a dialect the plane does not serve is refused naming the model (fail-closed). +/// RED: a plane that reads only the provider's `protocol` writes the overriding model's requests +/// in `openai` and accepts an unknown override. +#[test] +fn a_models_protocol_overrides_its_providers_and_an_unknown_one_is_refused() { + let mut v = settings(); + v["models"]["gpt-as-ant"] = json!({ "provider": "oai", "protocol": "anthropic" }); + let s = Shaping::from_settings(&v).expect("reads"); + assert_eq!(s.lane("gpt-as-ant").expect("the lane").dialect, "anthropic"); + assert_eq!( + s.lane("gpt").expect("the sibling").dialect, + "openai", + "a model naming no protocol keeps its provider's" + ); + let mut v = settings(); + v["models"]["gpt"]["protocol"] = json!("klingon"); + let refused = Shaping::from_settings(&v).unwrap_err(); + assert!( + refused.contains("model 'gpt' uses unknown protocol 'klingon'"), + "{refused}" + ); +} diff --git a/crates/busbar/src/root/door_steps.rs b/crates/busbar/src/root/door_steps.rs index 3a7492cba9..e661a711f6 100644 --- a/crates/busbar/src/root/door_steps.rs +++ b/crates/busbar/src/root/door_steps.rs @@ -1399,6 +1399,12 @@ pub fn kernel_sections( "prompt_caching", model.prompt_caching.map(|v| json!(v)), ); + // The model's dialect override (#51), for the plane to resolve and judge. + opt( + &mut m, + MODEL_PROTOCOL_KEYS[0], + model.protocol.as_ref().map(|v| json!(v)), + ); models.insert(name.clone(), Value::Object(m)); } let mut pools = Map::new(); diff --git a/crates/busbar/src/root/tests/door_steps.rs b/crates/busbar/src/root/tests/door_steps.rs index 675d3401ad..80103c84d0 100644 --- a/crates/busbar/src/root/tests/door_steps.rs +++ b/crates/busbar/src/root/tests/door_steps.rs @@ -5871,3 +5871,33 @@ done ); } } + +/// A MODEL'S `protocol` OVERRIDE REACHES THE PLANE (#51): the section the root hands the plane +/// serving the `pools` map carries each model's own `protocol` where it names one, and nothing for a +/// model that names none (its provider's stands). RED: a root that drops the member hands the plane +/// no override, so the plane speaks the provider's dialect to that model. +#[test] +fn a_models_protocol_override_is_handed_to_the_plane() { + let def: busbar_kernel::config::ProviderDef = + serde_yaml::from_str("protocol: openai\nbase_url: http://127.0.0.1:9\n").expect("a def"); + let defs = std::collections::HashMap::from([("p".to_string(), def)]); + let deploy = busbar_kernel::config::deploy_from_yaml_str( + "store: {module: memory}\nproviders:\n p:\n api_key: { env: BUSBAR_TEST_KEY }\n\ + models:\n over:\n provider: p\n protocol: anthropic\n plain:\n provider: p\n", + ) + .expect("parses"); + let cfg = busbar_kernel::config::resolve(&deploy, &defs).expect("resolves"); + let sections = super::kernel_sections(&cfg); + let models = sections + .get(busbar_contract::section::RESERVED_MODELS_KEY) + .expect("the models section"); + assert_eq!( + models["over"]["protocol"].as_str(), + Some("anthropic"), + "{models:?}" + ); + assert!( + models["plain"].get("protocol").is_none(), + "a model naming no protocol states none: {models:?}" + ); +} From b064c27508f27dec43585b5fd817e360ba39a73e Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 13 Oct 2026 14:33:03 -0700 Subject: [PATCH 2/4] row 155 (L1#3): the root's protocol-override test names no vendor dialect (kind-isolation law0 busbar x vendor 92 -> 89) The root hands the model's override through as opaque data; its test now spells two neutral dialect words, so the root carries no vendor name (#841 gates: law0 busbar x vendor 92/89, law0-base 89/0). --- crates/busbar/src/root/tests/door_steps.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/busbar/src/root/tests/door_steps.rs b/crates/busbar/src/root/tests/door_steps.rs index 80103c84d0..e922a16ee7 100644 --- a/crates/busbar/src/root/tests/door_steps.rs +++ b/crates/busbar/src/root/tests/door_steps.rs @@ -5879,11 +5879,11 @@ done #[test] fn a_models_protocol_override_is_handed_to_the_plane() { let def: busbar_kernel::config::ProviderDef = - serde_yaml::from_str("protocol: openai\nbase_url: http://127.0.0.1:9\n").expect("a def"); + serde_yaml::from_str("protocol: dialect-a\nbase_url: http://127.0.0.1:9\n").expect("a def"); let defs = std::collections::HashMap::from([("p".to_string(), def)]); let deploy = busbar_kernel::config::deploy_from_yaml_str( "store: {module: memory}\nproviders:\n p:\n api_key: { env: BUSBAR_TEST_KEY }\n\ - models:\n over:\n provider: p\n protocol: anthropic\n plain:\n provider: p\n", + models:\n over:\n provider: p\n protocol: dialect-b\n plain:\n provider: p\n", ) .expect("parses"); let cfg = busbar_kernel::config::resolve(&deploy, &defs).expect("resolves"); @@ -5893,7 +5893,7 @@ fn a_models_protocol_override_is_handed_to_the_plane() { .expect("the models section"); assert_eq!( models["over"]["protocol"].as_str(), - Some("anthropic"), + Some("dialect-b"), "{models:?}" ); assert!( From ab5fe57c72194210a5adb624cf4e8004c2363033 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 13 Oct 2026 23:12:37 -0700 Subject: [PATCH 3/4] llm plane: protocol: bedrock-invoke serves an Anthropic model over Bedrock InvokeModel, the carried dialect inside the invoke envelope (stacked on #841) --- CHANGELOG.md | 8 + .../busbar-plane-llm/src/exchange/attempt.rs | 9 + .../busbar-plane-llm/src/exchange/invoke.rs | 269 ++++++++++++++ crates/busbar-plane-llm/src/exchange/mod.rs | 1 + crates/busbar-plane-llm/src/exchange/probe.rs | 15 +- .../src/exchange/reply/mod.rs | 43 +++ .../busbar-plane-llm/src/exchange/shaping.rs | 13 + .../busbar-plane-llm/tests/exchange_invoke.rs | 337 ++++++++++++++++++ crates/busbar/src/root/door_steps.rs | 25 +- crates/busbar/src/root/tests/door_steps.rs | 27 ++ 10 files changed, 733 insertions(+), 14 deletions(-) create mode 100644 crates/busbar-plane-llm/src/exchange/invoke.rs create mode 100644 crates/busbar-plane-llm/tests/exchange_invoke.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f12b2eac8..4e166a5223 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -782,6 +782,14 @@ each dialect translates, field by field, is listed in the generated the node sealed the record. It does not prove the chain was never rewritten by someone holding the key; that takes a head recorded off the node. **Migration:** none; every 1.5.5 audit field is unchanged. +- 1.6.0 Added: a model entry's `protocol: bedrock-invoke` serves an Anthropic model over Amazon Bedrock's `InvokeModel` and `InvokeModelWithResponseStream`. + The request and answer are the Anthropic Messages API's own bodies and events, so an Anthropic + caller's request is relayed inside the envelope, with `model` and `stream` moved off the body, + `anthropic_version: bedrock-2023-05-31` stated, and `anthropic-beta` carried as `anthropic_beta`. + The answer, its stream and its usage read exactly as on an Anthropic provider. The model's provider + is a Bedrock provider and keeps its SigV4 credential; a provider cannot default to + `bedrock-invoke`, and `protocol:` on a provider still lists 1.5.5's six protocols. **Migration:** + none. - 1.6.0 Added: `GET /api/v1/admin/ledger/checkpoints` lists sealed checkpoints, sealed every 10,000 journal records or every 60 seconds with a change, whichever comes first. The read serves the latest 1,024 seals, oldest first; the journal keeps every seal. A node that has not reached either threshold serves an empty list. **Migration:** none. diff --git a/crates/busbar-plane-llm/src/exchange/attempt.rs b/crates/busbar-plane-llm/src/exchange/attempt.rs index 77a4e80a57..c30cc5a64a 100644 --- a/crates/busbar-plane-llm/src/exchange/attempt.rs +++ b/crates/busbar-plane-llm/src/exchange/attempt.rs @@ -469,6 +469,7 @@ pub fn upstream_path(lane: &Lane, operation: OpVerb, stream: bool) -> Option p.clone(), + None if lane.invoke => super::invoke::path(lane.wire_model(), stream), None => rh.upstream_path(&EgressCtx { operation, model: lane.wire_model(), @@ -801,6 +802,14 @@ pub fn build( body_is_json, intent.wants_stream, )?; + // THE INVOKE ENVELOPE around the carried dialect's request ([`super::invoke`]). + let (fields, body, pristine) = if lane.invoke { + let (fields, beta) = super::invoke::fields(fields, intent.wants_stream); + let body = super::invoke::body(&body, &beta).ok_or_else(|| internal(ingress))?; + (fields, body, false) + } else { + (fields, body, pristine) + }; Ok(FarRequest { verb: "POST", target, diff --git a/crates/busbar-plane-llm/src/exchange/invoke.rs b/crates/busbar-plane-llm/src/exchange/invoke.rs new file mode 100644 index 0000000000..d7ef8962c0 --- /dev/null +++ b/crates/busbar-plane-llm/src/exchange/invoke.rs @@ -0,0 +1,269 @@ +//! THE INVOKE-MODEL ENVELOPE, sans I/O: an Anthropic Messages exchange carried over Amazon +//! Bedrock's `InvokeModel` / `InvokeModelWithResponseStream` operations (the owner's 2026-09-29 +//! compat scope, "Bedrock InvokeModel dialect"; ARCHITECT 2026-10-14, the envelope design). +//! +//! WHAT IT IS. Bedrock serves Anthropic's models over two APIs. Its unified Converse API is the +//! plane's `bedrock` dialect. Its model-native `InvokeModel` API takes the Anthropic Messages body +//! itself, so this is not a dialect of its own: the request body, the answer body, the stream's +//! events and the usage they carry are the Anthropic dialect's, read and written by that dialect, +//! and only the envelope around them is Bedrock's. A model selects it with +//! `models..protocol: bedrock-invoke` ([`PROTOCOL`]); a provider never defaults to it, because +//! the name is no codec protocol the kernel knows, so the provider sweep refuses it in the previous +//! release's words. The kernel names none of this. +//! +//! THE ENVELOPE, both ways: +//! - the target is `/model/{model}/invoke`, or `/model/{model}/invoke-with-response-stream` for a +//! streamed answer ([`path`]); +//! - the body loses `model` (it rides the target) and `stream` (the operation says it), and states +//! `anthropic_version: bedrock-2023-05-31`; the caller's `anthropic-beta` field becomes the body's +//! `anthropic_beta` list and the `anthropic-version` field is not sent ([`body`], [`fields`]); +//! - a streamed answer arrives as AWS event-stream frames, each `chunk` carrying one Anthropic +//! stream event base64-encoded in `bytes`; it is re-framed as the Anthropic dialect's own +//! server-sent events, the event JSON byte for byte ([`Transcoder`]); +//! - a refusal arrives as `{"message": …}` beside an `x-amzn-ErrorType` field (or, mid-stream, an +//! exception frame); it is re-stated as the Anthropic dialect's error envelope ([`error_body`]). +//! +//! The credential is the provider's: a Bedrock provider's SigV4 style, which the composition root +//! falls back to when the model's own protocol states no default style. + +use std::borrow::Cow; + +use busbar_contract::protocol::HeadFields; +use serde_json::Value; + +use crate::codec::eventstream::{drain_frames_checked, DrainStatus}; +use crate::codec::json_splice::{self, Edit}; + +/// The protocol a model names to be served over the envelope. +pub const PROTOCOL: &str = "bedrock-invoke"; + +/// The dialect the envelope carries: its bodies, events and usage are that dialect's own. +pub const CARRIED: &str = "anthropic"; + +/// The API version the envelope's body states. +const VERSION: &str = "bedrock-2023-05-31"; +/// The body member that states it. +const VERSION_KEY: &str = "anthropic_version"; +/// The body member the caller's beta flags ride in. +const BETA_KEY: &str = "anthropic_beta"; +/// The head field the caller's beta flags arrive in. +const BETA_FIELD: &str = "anthropic-beta"; +/// The head field the Messages API version arrives in; the envelope states it in the body. +const VERSION_FIELD: &str = "anthropic-version"; +/// The content type of a streamed answer on the wire. +pub const EVENTSTREAM: &str = "application/vnd.amazon.eventstream"; +/// The frame that carries one stream event. +const CHUNK: &str = "chunk"; +/// The head field a refusal names its exception in. +const ERROR_TYPE_FIELD: &str = "x-amzn-errortype"; + +/// The far end's target for `model`, as written before encoding. +#[must_use] +pub fn path(model: &str, stream: bool) -> String { + let op = if stream { + "invoke-with-response-stream" + } else { + "invoke" + }; + format!("/model/{model}/{op}") +} + +/// The far request's head fields over the envelope: the Messages API version is not sent (the body +/// states the envelope's), the caller's beta flags are taken out to ride in the body, and a streamed +/// answer is asked for in the event-stream framing. Returns the fields and the beta flags. +#[must_use] +pub fn fields( + fields: Vec<(String, Vec)>, + stream: bool, +) -> (Vec<(String, Vec)>, Vec>) { + let (beta, rest): (Vec<_>, Vec<_>) = fields + .into_iter() + .filter(|(name, _)| !name.eq_ignore_ascii_case(VERSION_FIELD)) + .partition(|(name, _)| name.eq_ignore_ascii_case(BETA_FIELD)); + let out = rest + .into_iter() + .map(|(name, value)| { + if stream && name.eq_ignore_ascii_case("accept") { + (name, EVENTSTREAM.as_bytes().to_vec()) + } else { + (name, value) + } + }) + .collect(); + (out, beta.into_iter().map(|(_, v)| v).collect()) +} + +/// The far request's body over the envelope, a byte-level edit of the Messages body (never a +/// re-serialization): `model` and `stream` removed, `anthropic_version` stated, and the caller's +/// beta flags (each field's comma-separated list, in order) as `anthropic_beta`. `None` when the body +/// is not a JSON object. +#[must_use] +pub fn body(messages: &[u8], beta: &[Vec]) -> Option> { + let flags: Vec = beta + .iter() + .flat_map(|v| { + String::from_utf8_lossy(v) + .split(',') + .map(|f| f.trim().to_string()) + .collect::>() + }) + .filter(|f| !f.is_empty()) + .map(Value::String) + .collect(); + let version = serde_json::to_vec(VERSION).ok()?; + let mut edits = vec![ + Edit::Remove("model"), + Edit::Remove("stream"), + Edit::Set(VERSION_KEY, Cow::Owned(version)), + ]; + if !flags.is_empty() { + edits.push(Edit::Set( + BETA_KEY, + Cow::Owned(serde_json::to_vec(&Value::Array(flags)).ok()?), + )); + } + json_splice::apply_top(messages, &edits) +} + +/// The Anthropic dialect's error kind for a Bedrock exception name (`ValidationException`, or the +/// event-stream form `validationException`, with or without a `:…` suffix). +fn kind_of(exception: &str) -> &'static str { + let name = exception.split(':').next().unwrap_or("").trim(); + match name.to_ascii_lowercase().as_str() { + "validationexception" => "invalid_request_error", + "accessdeniedexception" | "unrecognizedclientexception" => "permission_error", + "resourcenotfoundexception" => "not_found_error", + "throttlingexception" | "servicequotaexceededexception" => "rate_limit_error", + "serviceunavailableexception" | "modelnotreadyexception" => "overloaded_error", + "modeltimeoutexception" => "timeout_error", + _ => "api_error", + } +} + +/// The Anthropic dialect's error envelope for a refusal's `kind` and `message`. +fn envelope(kind: &str, message: &str) -> Vec { + serde_json::to_vec(&serde_json::json!({ + "type": "error", + "error": { "type": kind, "message": message }, + })) + .unwrap_or_default() +} + +/// A refused answer's body, re-stated in the Anthropic dialect's error envelope from Bedrock's +/// `{"message": …}` and its `x-amzn-ErrorType` field. A body already in that envelope is kept. +#[must_use] +pub fn error_body(head: HeadFields<'_>, body: &[u8]) -> Vec { + let parsed: Option = serde_json::from_slice(body).ok(); + if parsed + .as_ref() + .and_then(|v| v.get("type")) + .and_then(Value::as_str) + == Some("error") + { + return body.to_vec(); + } + let exception = head + .iter() + .find(|(n, _)| n.eq_ignore_ascii_case(ERROR_TYPE_FIELD.as_bytes())) + .map(|(_, v)| String::from_utf8_lossy(v).into_owned()) + .unwrap_or_default(); + let message = parsed + .as_ref() + .and_then(|v| v.get("message").or_else(|| v.get("Message"))) + .and_then(Value::as_str) + .map_or_else( + || String::from_utf8_lossy(body).into_owned(), + str::to_string, + ); + envelope(kind_of(&exception), &message) +} + +/// ONE STREAMED ANSWER'S RE-FRAMING: AWS event-stream frames in, the Anthropic dialect's own +/// server-sent events out. Each `chunk` frame's base64 `bytes` decode to one Anthropic stream event, +/// written as `event: ` and `data: `; an exception frame becomes the dialect's +/// `error` event; a malformed frame ends the stream with one. Bytes of a frame not yet whole wait. +#[derive(Debug, Default)] +pub struct Transcoder { + pending: Vec, + ended: bool, +} + +impl Transcoder { + /// A fresh re-framing. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// The server-sent events the frames completed by `bytes` re-frame to. + pub fn feed(&mut self, bytes: &[u8]) -> Vec { + let mut out = Vec::new(); + if self.ended { + return out; + } + self.pending.extend_from_slice(bytes); + let (frames, status, _) = drain_frames_checked(&mut self.pending, None); + for (event_type, payload) in frames { + if event_type == CHUNK { + let Some(event) = chunk_event(&payload) else { + sse( + &mut out, + "error", + &envelope("api_error", "malformed InvokeModel stream chunk"), + ); + self.ended = true; + return out; + }; + let kind = serde_json::from_slice::(&event) + .ok() + .and_then(|v| v.get("type").and_then(Value::as_str).map(str::to_string)) + .unwrap_or_default(); + sse(&mut out, &kind, &event); + } else if !event_type.is_empty() { + let message = serde_json::from_slice::(&payload) + .ok() + .and_then(|v| { + v.get("message") + .or_else(|| v.get("Message")) + .and_then(Value::as_str) + .map(str::to_string) + }) + .unwrap_or_else(|| String::from_utf8_lossy(&payload).into_owned()); + sse(&mut out, "error", &envelope(kind_of(&event_type), &message)); + self.ended = true; + return out; + } + } + if status == DrainStatus::MalformedPrelude { + sse( + &mut out, + "error", + &envelope("api_error", "malformed InvokeModel stream frame"), + ); + self.ended = true; + } + out + } +} + +/// One `chunk` frame's event: its `bytes` member, base64-decoded. An event that spans lines is +/// written on one, since a server-sent event's data is one line here. +fn chunk_event(payload: &[u8]) -> Option> { + let v: Value = serde_json::from_slice(payload).ok()?; + let encoded = v.get("bytes")?.as_str()?; + let event = busbar_contract::media::base64_decode(encoded)?.to_vec(); + if event.contains(&b'\n') { + let parsed: Value = serde_json::from_slice(&event).ok()?; + return serde_json::to_vec(&parsed).ok(); + } + Some(event) +} + +/// One server-sent event, the data written as it came. +fn sse(out: &mut Vec, event: &str, data: &[u8]) { + out.extend_from_slice(b"event: "); + out.extend_from_slice(event.as_bytes()); + out.extend_from_slice(b"\ndata: "); + out.extend_from_slice(data); + out.extend_from_slice(b"\n\n"); +} diff --git a/crates/busbar-plane-llm/src/exchange/mod.rs b/crates/busbar-plane-llm/src/exchange/mod.rs index ab0a6a4030..040698b3b2 100644 --- a/crates/busbar-plane-llm/src/exchange/mod.rs +++ b/crates/busbar-plane-llm/src/exchange/mod.rs @@ -4,6 +4,7 @@ pub mod arrive; pub mod attempt; +pub mod invoke; pub mod multipart; pub mod probe; pub mod project; diff --git a/crates/busbar-plane-llm/src/exchange/probe.rs b/crates/busbar-plane-llm/src/exchange/probe.rs index 762edc1fb7..a80fb50674 100644 --- a/crates/busbar-plane-llm/src/exchange/probe.rs +++ b/crates/busbar-plane-llm/src/exchange/probe.rs @@ -46,7 +46,18 @@ pub fn request(lane: &Lane) -> Option { if lane.path_base.is_some() { return None; } - let (target, _canonical) = wire_and_canonical_path(&probe_path(lane)); + // A lane over the invoke envelope is probed through it: its target, and its body's edits. + let path = match &lane.path { + None if lane.invoke => super::invoke::path(lane.wire_model(), false), + _ => probe_path(lane), + }; + let body = probe_body(lane.dialect, lane.wire_model()); + let body = if lane.invoke { + super::invoke::body(&body, &[])? + } else { + body + }; + let (target, _canonical) = wire_and_canonical_path(&path); let user_agent = decl(lane.dialect).map_or(EGRESS_UA_DEFAULT, |d| d.egress_user_agent); Some(FarRequest { verb: "POST", @@ -59,7 +70,7 @@ pub fn request(lane: &Lane) -> Option { ("user-agent".to_string(), user_agent.as_bytes().to_vec()), ("accept".to_string(), APPLICATION_JSON.as_bytes().to_vec()), ], - body: probe_body(lane.dialect, lane.wire_model()), + body, pristine: false, dropped_controls: Vec::new(), }) diff --git a/crates/busbar-plane-llm/src/exchange/reply/mod.rs b/crates/busbar-plane-llm/src/exchange/reply/mod.rs index 931c23177e..a1888f1853 100644 --- a/crates/busbar-plane-llm/src/exchange/reply/mod.rs +++ b/crates/busbar-plane-llm/src/exchange/reply/mod.rs @@ -27,6 +27,7 @@ use busbar_contract::upstream::{Disposition, StatusClass}; use super::arrive::Arrived; use super::attempt::StreamIntent; +use super::invoke; use super::refuse::Rendered; use super::shaping::Lane; use crate::codec::DECLS; @@ -185,6 +186,9 @@ enum State { /// ONE ATTEMPT'S ANSWER. pub struct Reply { state: State, + /// A streamed answer over the invoke envelope, re-framed into its carried dialect's events + /// before it is read ([`invoke::Transcoder`]). + invoke: Option, } impl std::fmt::Debug for Reply { @@ -257,9 +261,21 @@ impl Reply { head: head.iter().map(|(n, v)| (n.to_vec(), v.to_vec())).collect(), body: Vec::new(), }, + invoke: None, }; } let far_ct = wire::head_value(head, "content-type"); + // A STREAM OVER THE INVOKE ENVELOPE is read as its carried dialect's stream: its frames are + // re-framed into that dialect's events before any byte is read. + let invoke_stream = ctx.lane.invoke + && far_ct + .and_then(wire::head_text) + .is_some_and(|ct| ct.starts_with(invoke::EVENTSTREAM)); + let far_ct = if invoke_stream { + Some(busbar_contract::protocol::TEXT_EVENT_STREAM.as_bytes()) + } else { + far_ct + }; let far_is_stream = far_ct .and_then(wire::head_text) .is_some_and(wire::is_stream_content_type); @@ -272,6 +288,7 @@ impl Reply { body: Vec::new(), head: Vec::new(), }, + invoke: None, }; }; // Only a translated answer is taken whole, and a translated answer relays no far head; a @@ -283,6 +300,7 @@ impl Reply { body: Vec::new(), head: Vec::new(), }, + invoke: None, }; } let mut fields = Vec::new(); @@ -322,6 +340,7 @@ impl Reply { head: Some(Head { status, fields }), units: Units::default(), }, + invoke: invoke_stream.then(invoke::Transcoder::new), } } @@ -333,6 +352,24 @@ impl Reply { last: bool, at: At, ) -> Piece<'a> { + let Some(transcoder) = self.invoke.as_mut() else { + return self.read(ctx, bytes, last, at); + }; + let framed = transcoder.feed(bytes); + let p = self.read(ctx, &framed, last, at); + Piece { + head: p.head, + bytes: Cow::Owned(p.bytes.into_owned()), + units: p.units, + verdict: p.verdict, + fault: p.fault, + done: p.done, + dropped: p.dropped, + } + } + + /// READ one piece in the answer's own framing. + fn read<'a>(&mut self, ctx: &ReplyCtx<'_>, bytes: &'a [u8], last: bool, at: At) -> Piece<'a> { match &mut self.state { State::Done => pending(), State::Error { body, .. } => { @@ -472,6 +509,12 @@ impl Reply { .iter() .map(|(n, v)| (n.as_slice(), v.as_slice())) .collect(); + // A refusal over the invoke envelope is read in its carried dialect's error envelope. + let body = if ctx.lane.invoke { + invoke::error_body(&pairs, &body) + } else { + body + }; let far = failure::FarError { status, head: &pairs, diff --git a/crates/busbar-plane-llm/src/exchange/shaping.rs b/crates/busbar-plane-llm/src/exchange/shaping.rs index 4fee68fa48..d87f3517b6 100644 --- a/crates/busbar-plane-llm/src/exchange/shaping.rs +++ b/crates/busbar-plane-llm/src/exchange/shaping.rs @@ -79,6 +79,9 @@ pub struct Lane { /// dialect's scheme (`api-key`, `jwt-bearer`, `oauth-client-credentials`), whose declaration /// carries none, as the previous release wrote them. pub statics: &'static [(&'static str, &'static str)], + /// The model is served over the invoke envelope ([`super::invoke`]): its dialect is the one the + /// envelope carries, its target, body edits, stream framing and refusals the envelope's. + pub invoke: bool, } impl Lane { @@ -426,6 +429,14 @@ impl Shaping { .as_deref() .or(p.protocol.as_deref()) .unwrap_or(DEFAULT_PROTOCOL); + // The invoke envelope is a MODEL's choice alone: a provider defaulting to it states an + // unknown protocol, refused below in the previous release's words. + let invoke = m.protocol.as_deref() == Some(super::invoke::PROTOCOL); + let asked = if invoke { + super::invoke::CARRIED + } else { + asked + }; let decl = DECLS .iter() .find(|d| d.name == asked && d.codec.is_some()) @@ -435,6 +446,7 @@ impl Shaping { })?; let dialect = decl.name; let statics = match p.auth.as_deref() { + _ if invoke => &[][..], Some(AUTH_API_KEY | AUTH_JWT_BEARER | AUTH_OAUTH_CLIENT_CREDENTIALS) => &[][..], _ => decl.static_headers, }; @@ -455,6 +467,7 @@ impl Shaping { caps: lane_caps(p, wire), error_map: p.error_map.clone(), statics, + invoke, }; lanes.insert(name, lane); } diff --git a/crates/busbar-plane-llm/tests/exchange_invoke.rs b/crates/busbar-plane-llm/tests/exchange_invoke.rs new file mode 100644 index 0000000000..301afbd266 --- /dev/null +++ b/crates/busbar-plane-llm/tests/exchange_invoke.rs @@ -0,0 +1,337 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! The invoke envelope: an Anthropic Messages exchange carried over Bedrock's `InvokeModel` and +//! `InvokeModelWithResponseStream` (the owner's 2026-09-29 compat scope; ARCHITECT 2026-10-14). +//! +//! RED: before the envelope, `models..protocol: bedrock-invoke` was refused as an unknown +//! protocol, so none of these builds or holds. + +use busbar_contract::operation::OpVerb; +use busbar_plane_llm::codec::eventstream::{encode_exception_frame, encode_frame}; +use busbar_plane_llm::exchange::arrive::{arrive, Arrived}; +use busbar_plane_llm::exchange::attempt::{build, stream_intent, upstream_path, FarRequest}; +use busbar_plane_llm::exchange::handler_of; +use busbar_plane_llm::exchange::invoke::{self, Transcoder}; +use busbar_plane_llm::exchange::reply::{At, Reply, ReplyCtx, Verdict}; +use busbar_plane_llm::exchange::shaping::Shaping; +use serde_json::{json, Value}; + +/// A Bedrock provider (its own dialect, SigV4), one model served over it in Converse and one over +/// the invoke envelope, and an Anthropic provider for the carried dialect's own lane. +fn shaping() -> Shaping { + Shaping::from_settings(&json!({ + "providers": { + "aws": { "protocol": "bedrock", "base_url": "https://bedrock-runtime.us-east-1.amazonaws.com" }, + "ant": { "protocol": "anthropic", "base_url": "https://anthropic.example" } + }, + "models": { + "claude-converse": { "provider": "aws", "upstream_model": "anthropic.claude-3-5-sonnet-20240620-v1:0" }, + "claude-invoke": { + "provider": "aws", + "protocol": "bedrock-invoke", + "upstream_model": "anthropic.claude-3-5-sonnet-20240620-v1:0", + "default_max_tokens": 256 + }, + "claude": { "provider": "ant" } + }, + "pools": { "p": { "members": ["claude-invoke"] } } + })) + .expect("reads") +} + +fn arrived( + body: &str, + extra: &[(&'static [u8], &'static [u8])], +) -> (Arrived, Vec<(&'static [u8], &'static [u8])>) { + let mut h: Vec<(&'static [u8], &'static [u8])> = vec![ + (b"content-type", b"application/json"), + (b"anthropic-version", b"2023-06-01"), + ]; + h.extend_from_slice(extra); + let a = arrive("POST", "/v1/messages", &h, body.as_bytes(), &()).expect("arrives"); + (a, h) +} + +fn field<'a>(r: &'a FarRequest, name: &str) -> Vec<&'a [u8]> { + r.fields + .iter() + .filter(|(n, _)| n == name) + .map(|(_, v)| v.as_slice()) + .collect() +} + +#[test] +fn a_model_naming_the_envelope_is_served_in_the_carried_dialect_over_it() { + let s = shaping(); + let lane = s.lane("claude-invoke").expect("the lane"); + assert_eq!(lane.dialect, invoke::CARRIED); + assert!(lane.invoke); + assert!( + lane.statics.is_empty(), + "the envelope states the API version in the body, never as a field" + ); + let sibling = s.lane("claude-converse").expect("the sibling"); + assert_eq!( + sibling.dialect, "bedrock", + "a model naming no protocol keeps its provider's" + ); + assert!(!sibling.invoke); + assert!( + !s.lane("claude") + .expect("the carried dialect's own lane") + .invoke + ); +} + +#[test] +fn a_provider_never_defaults_to_the_envelope() { + let refused = Shaping::from_settings(&json!({ + "providers": { "aws": { "protocol": "bedrock-invoke", "base_url": "https://x.example" } }, + "models": { "m": { "provider": "aws" } } + })) + .unwrap_err(); + assert_eq!( + refused, + "provider 'aws' uses unknown protocol 'bedrock-invoke'" + ); +} + +#[test] +fn the_target_is_the_models_invoke_operation() { + let s = shaping(); + let lane = s.lane("claude-invoke").expect("the lane"); + assert_eq!( + upstream_path(lane, OpVerb::CHAT, false).as_deref(), + Some("/model/anthropic.claude-3-5-sonnet-20240620-v1:0/invoke") + ); + assert_eq!( + upstream_path(lane, OpVerb::CHAT, true).as_deref(), + Some("/model/anthropic.claude-3-5-sonnet-20240620-v1:0/invoke-with-response-stream") + ); +} + +#[test] +fn a_carried_dialect_caller_is_relayed_inside_the_envelope() { + let body = r#"{"model":"claude-invoke","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}"#; + let (a, h) = arrived( + body, + &[( + b"anthropic-beta", + b"prompt-caching-2024-07-31, output-128k-2025-02-19", + )], + ); + let r = build(&a, &h, &shaping(), "p", "claude-invoke").expect("built"); + assert!( + r.target.ends_with("/invoke-with-response-stream"), + "{}", + r.target + ); + assert!(r + .target + .starts_with("/model/anthropic.claude-3-5-sonnet-20240620-v1")); + let sent: Value = serde_json::from_slice(&r.body).expect("json"); + assert_eq!(sent.get("model"), None, "the model rides the target"); + assert_eq!(sent.get("stream"), None, "the operation says it streams"); + assert_eq!(sent["anthropic_version"], "bedrock-2023-05-31"); + assert_eq!( + sent["anthropic_beta"], + json!(["prompt-caching-2024-07-31", "output-128k-2025-02-19"]) + ); + assert_eq!(sent["max_tokens"], 64); + assert!( + String::from_utf8_lossy(&r.body).contains(r#""messages":[{"role":"user","content":"hi"}]"#), + "the caller's members go out as its own bytes" + ); + assert!(field(&r, "anthropic-version").is_empty()); + assert!(field(&r, "anthropic-beta").is_empty()); + assert_eq!(field(&r, "accept"), [invoke::EVENTSTREAM.as_bytes()]); + assert!(!r.pristine); +} + +#[test] +fn a_whole_answer_is_asked_for_on_the_plain_operation() { + let body = r#"{"model":"claude-invoke","max_tokens":64,"messages":[]}"#; + let (a, h) = arrived(body, &[]); + let r = build(&a, &h, &shaping(), "p", "claude-invoke").expect("built"); + assert!(r.target.ends_with("/invoke"), "{}", r.target); + let sent: Value = serde_json::from_slice(&r.body).expect("json"); + assert_eq!(sent.get("anthropic_beta"), None, "no beta flags, no member"); + assert_ne!(field(&r, "accept"), [invoke::EVENTSTREAM.as_bytes()]); +} + +#[test] +fn the_body_edit_keeps_every_other_byte() { + let edited = invoke::body( + br#"{"model":"m","stream":false,"system":"s", "max_tokens":1}"#, + &[], + ) + .expect("an object"); + let text = String::from_utf8(edited).expect("utf-8"); + assert!( + text.contains(r#""system":"s""#) && text.contains(r#""max_tokens":1"#), + "{text}" + ); + assert!( + !text.contains(r#""model""#) && !text.contains(r#""stream""#), + "{text}" + ); + assert!( + text.contains(r#""anthropic_version":"bedrock-2023-05-31""#), + "{text}" + ); + assert_eq!(invoke::body(b"[1]", &[]), None, "not an object"); +} + +/// One Anthropic stream event as Bedrock frames it: a `chunk` whose `bytes` are the event, base64. +fn chunk(event: &Value) -> Vec { + let raw = serde_json::to_vec(event).expect("json"); + let payload = json!({ "bytes": busbar_contract::media::base64_encode(&raw) }); + encode_frame("chunk", &serde_json::to_vec(&payload).expect("json")) +} + +fn events() -> Vec { + vec![ + json!({"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":"claude","content":[],"stop_reason":null,"usage":{"input_tokens":11,"output_tokens":1}}}), + json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}), + json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"pong"}}), + json!({"type":"content_block_stop","index":0}), + json!({"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":7}}), + json!({"type":"message_stop","amazon-bedrock-invocationMetrics":{"inputTokenCount":11,"outputTokenCount":7}}), + ] +} + +#[test] +fn a_stream_is_reframed_into_the_carried_dialects_events_byte_for_byte() { + let wire: Vec = events().iter().flat_map(chunk).collect(); + let mut want = Vec::new(); + for e in events() { + want.extend_from_slice( + format!("event: {}\ndata: ", e["type"].as_str().unwrap()).as_bytes(), + ); + want.extend_from_slice(&serde_json::to_vec(&e).unwrap()); + want.extend_from_slice(b"\n\n"); + } + for step in [1, 7, 64, wire.len()] { + let mut t = Transcoder::new(); + let got: Vec = wire.chunks(step).flat_map(|c| t.feed(c)).collect(); + assert_eq!( + String::from_utf8_lossy(&got), + String::from_utf8_lossy(&want), + "fed {step} bytes at a time" + ); + } +} + +#[test] +fn a_mid_stream_exception_ends_the_stream_with_the_carried_dialects_error_event() { + let mut wire = chunk(&events()[0]); + wire.extend(encode_exception_frame("ThrottlingException", "slow down")); + wire.extend(chunk(&events()[1])); + let out = String::from_utf8(Transcoder::new().feed(&wire)).expect("utf-8"); + assert!(out.starts_with("event: message_start\n"), "{out}"); + assert!( + out.ends_with("event: error\ndata: {\"error\":{\"message\":\"slow down\",\"type\":\"rate_limit_error\"},\"type\":\"error\"}\n\n"), + "{out}" + ); + assert!( + !out.contains("content_block_start"), + "nothing after the exception: {out}" + ); +} + +#[test] +fn a_refusal_is_restated_in_the_carried_dialects_error_envelope() { + let head: &[(&[u8], &[u8])] = &[( + b"x-amzn-ErrorType", + b"ValidationException:http://internal.amazon.com/coral/com.amazon.bedrock/", + )]; + let body = invoke::error_body(head, br#"{"message":"max_tokens: field required"}"#); + assert_eq!( + serde_json::from_slice::(&body).expect("json"), + json!({"type":"error","error":{"type":"invalid_request_error","message":"max_tokens: field required"}}) + ); + let throttled: &[(&[u8], &[u8])] = &[(b"x-amzn-errortype", b"ThrottlingException")]; + let body = invoke::error_body(throttled, br#"{"message":"Too many requests"}"#); + assert_eq!( + serde_json::from_slice::(&body).expect("json")["error"]["type"], + "rate_limit_error" + ); +} + +/// THE MONEY READ: a streamed answer over the envelope reaches the carried dialect's caller as that +/// dialect's own events, and the units it reports are the carried dialect's usage (11 in, 7 out), +/// exactly as on that dialect's own lane. +#[test] +fn a_streamed_answer_meters_the_carried_dialects_usage() { + let s = shaping(); + let body = r#"{"model":"claude-invoke","max_tokens":64,"stream":true,"messages":[]}"#; + let (a, _) = arrived(body, &[]); + let lane = s.lane("claude-invoke").expect("the lane"); + let ctx = ReplyCtx { + arrived: &a, + lane, + intent: stream_intent(handler_of(&a).expect("a handler"), a.parsed.as_ref()), + passthrough: false, + }; + let head: &[(&[u8], &[u8])] = &[(b"content-type", invoke::EVENTSTREAM.as_bytes())]; + let mut reply = Reply::new(&ctx, 200, head); + let wire: Vec = events().iter().flat_map(chunk).collect(); + let pieces: Vec<&[u8]> = wire.chunks(50).collect(); + let mut out = Vec::new(); + let mut last_piece = None; + for (i, p) in pieces.iter().enumerate() { + let last = i + 1 == pieces.len(); + let piece = reply.feed(&ctx, p, last, At::default()); + if let Some(h) = &piece.head { + let ct = h + .fields + .iter() + .find(|(n, _)| n == "content-type") + .map(|(_, v)| v.clone()); + assert_eq!( + ct.as_deref(), + Some(&b"text/event-stream"[..]), + "the carried dialect's stream" + ); + } + out.extend_from_slice(&piece.bytes); + if last { + last_piece = Some((piece.units.clone(), piece.verdict, piece.done)); + } + } + let (units, verdict, done) = last_piece.expect("a last piece"); + assert!(done); + assert_eq!(verdict, Verdict::Ok); + assert_eq!((units.tokens_in, units.tokens_out), (11, 7), "{units:?}"); + let text = String::from_utf8_lossy(&out); + assert!( + text.contains("event: content_block_delta") && text.contains("pong"), + "{text}" + ); +} + +/// THE KERNEL NEVER SEES THE ENVELOPE (ARCHITECT 2026-10-14 condition 3; Law 0): it is no codec +/// protocol this plane declares, so the protocols the composition root installs, and with them the +/// previous release's "must be one of" list a provider's unknown protocol is refused with, stay +/// exactly 1.5.5's six in 1.5.5's order (oracle cell `boot.refusal|BOOT-020|validate`; the binary's +/// own `validate_unknown_protocol_lists_only_the_configured_planes_dialects` pins the line). +#[test] +fn the_envelope_is_no_declared_protocol_and_the_1_5_5_list_is_unchanged() { + let declared: Vec<&str> = busbar_plane_llm::codec::DECLS + .iter() + .map(|d| d.name) + .collect(); + assert_eq!( + declared, + [ + "anthropic", + "openai", + "gemini", + "bedrock", + "responses", + "cohere" + ] + ); + assert!(!declared.contains(&invoke::PROTOCOL)); +} diff --git a/crates/busbar/src/root/door_steps.rs b/crates/busbar/src/root/door_steps.rs index e661a711f6..8d10bcabfe 100644 --- a/crates/busbar/src/root/door_steps.rs +++ b/crates/busbar/src/root/door_steps.rs @@ -2248,18 +2248,19 @@ pub fn member_routes( .iter() .find_map(|k| entry_text(member, k)) .unwrap_or(&provider.protocol); - let default = served - .dialects - .iter() - .position(|d| *d == dialect) - .and_then(|at| { - let at = u32::try_from(at).ok()?; - served - .dialect_auth - .iter() - .find(|(d, _, _)| *d == at) - .map(|(_, style, params)| ((*style).to_string(), *params)) - }); + let default_of = |dialect: &str| { + let at = served.dialects.iter().position(|d| *d == dialect)?; + let at = u32::try_from(at).ok()?; + served + .dialect_auth + .iter() + .find(|(d, _, _)| *d == at) + .map(|(_, style, params)| ((*style).to_string(), *params)) + }; + // A member whose own protocol states no default style (one its plane serves over another + // dialect's envelope) rides its provider's protocol's default: the credential is the + // provider's. + let default = default_of(dialect).or_else(|| default_of(&provider.protocol)); // The provider's own `auth:` binds its style alone; the dialect's default carries its // declared parameters (Q-L6-AUTHPARAMS). let (style, params) = match (&provider.style, default) { diff --git a/crates/busbar/src/root/tests/door_steps.rs b/crates/busbar/src/root/tests/door_steps.rs index e922a16ee7..2373c52362 100644 --- a/crates/busbar/src/root/tests/door_steps.rs +++ b/crates/busbar/src/root/tests/door_steps.rs @@ -557,6 +557,33 @@ fn a_member_is_bound_under_its_dialects_default_style_on_the_need_that_style_nam ); } +/// A MEMBER WHOSE OWN PROTOCOL STATES NO DEFAULT STYLE rides its provider protocol's default (the +/// envelope ruling, ARCHITECT 2026-10-14: a model a plane serves over another dialect's +/// envelope is credentialed by its provider). RED: the member's own protocol alone was looked up, +/// so the load was refused as styleless. +#[cfg(feature = "auth-header")] +#[test] +fn a_member_whose_protocol_states_no_default_style_rides_its_providers() { + let routes = resolve( + "models: {m: {provider: p, protocol: x-envelope}}", + &[("p", provider("d", None))], + ) + .expect("the member resolves under its provider's default style"); + assert!( + routes["m"].auth.is_some(), + "bound by the plugin serving `bearer`" + ); + let styleless = resolve( + "models: {m: {provider: p, protocol: x-envelope}}", + &[("p", provider("x", None))], + ); + assert!( + styleless.as_ref().is_err_and(|e| e.contains("no `auth:`")), + "neither the member's protocol nor its provider's states a default: {:?}", + styleless.err() + ); +} + #[test] fn a_member_that_cannot_be_reached_refuses_the_load_naming_it() { let unknown = resolve("models: {m: {provider: q}}", &[("p", provider("d", None))]); From feaa3266a8dc1464c7cf7fc28f5fd2611ef23b30 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 13 Oct 2026 23:24:33 -0700 Subject: [PATCH 4/4] exchange_invoke tests: head fields as slices --- crates/busbar-plane-llm/tests/exchange_invoke.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/busbar-plane-llm/tests/exchange_invoke.rs b/crates/busbar-plane-llm/tests/exchange_invoke.rs index 301afbd266..3e001c1d42 100644 --- a/crates/busbar-plane-llm/tests/exchange_invoke.rs +++ b/crates/busbar-plane-llm/tests/exchange_invoke.rs @@ -45,8 +45,8 @@ fn arrived( extra: &[(&'static [u8], &'static [u8])], ) -> (Arrived, Vec<(&'static [u8], &'static [u8])>) { let mut h: Vec<(&'static [u8], &'static [u8])> = vec![ - (b"content-type", b"application/json"), - (b"anthropic-version", b"2023-06-01"), + (&b"content-type"[..], &b"application/json"[..]), + (&b"anthropic-version"[..], &b"2023-06-01"[..]), ]; h.extend_from_slice(extra); let a = arrive("POST", "/v1/messages", &h, body.as_bytes(), &()).expect("arrives"); @@ -117,8 +117,8 @@ fn a_carried_dialect_caller_is_relayed_inside_the_envelope() { let (a, h) = arrived( body, &[( - b"anthropic-beta", - b"prompt-caching-2024-07-31, output-128k-2025-02-19", + &b"anthropic-beta"[..], + &b"prompt-caching-2024-07-31, output-128k-2025-02-19"[..], )], ); let r = build(&a, &h, &shaping(), "p", "claude-invoke").expect("built");