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 9ab04f6d47..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 { @@ -214,6 +217,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,15 +422,31 @@ 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); + // 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()) - .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() { + _ if invoke => &[][..], Some(AUTH_API_KEY | AUTH_JWT_BEARER | AUTH_OAUTH_CLIENT_CREDENTIALS) => &[][..], _ => decl.static_headers, }; @@ -444,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..3e001c1d42 --- /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-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..8d10bcabfe 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(); @@ -2242,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 675d3401ad..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))]); @@ -5871,3 +5898,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: 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: dialect-b\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("dialect-b"), + "{models:?}" + ); + assert!( + models["plain"].get("protocol").is_none(), + "a model naming no protocol states none: {models:?}" + ); +}