refactor(llm): extract codec/anthropic_messages behind the Codec trait (#488)

## Summary

Dialect extraction in the gateway refactor series (after #481 / #485,
sibling of #487): the Anthropic Messages wire translation moves out of
`providers/anthropic.rs` into `codec/anthropic_messages/`, behind the
`Codec` / `StreamDecoder` traits. The adapter becomes a thin transport
shell owning auth, base URL, the streaming byte loop, and route config;
all translation is in the codec.

Three commits, each independently green:
1. **Add the codec** (`wire`/`encode`/`decode`/`stream`/`mod`) —
compiling but unused behind a scoped `dead_code` allow.
2. **Rewire the adapter** to it and migrate the ~70 unit tests into the
codec submodules they now cover.
3. **Port #482's Claude Fable 5 handling into the codec layout** (see
below).

Key moves:
- **Route config replaces the request-time `provider_name ==
"anthropic"` branches**: auth scheme (x-api-key vs bearer), version/beta
headers, the count-tokens availability gate, and Kimi-over-anthropic
forced streaming resolve once per call into a `RouteConfig`. Dialect
headers ride on `CodecParams` (`AnthropicVersion::Header("2023-06-01")`
+ beta-header emission for the direct route; inert defaults for Kimi).
- **`build_api_request`'s `(ApiRequest, RequestBuilder)` dual-return
dies**: codec `encode` produces body + headers as data
(`EncodedRequest`); the transport applies them. This also kills the
duplicated header rebuild in `count_input_tokens`.
- **Encode goes sync**: file-backed Image/Document attachments resolve
to inline data via the shared `attachments::resolve` (#485) in the
adapter before encode (drop-on-error preserved; audio stays a text
placeholder in the codec).
- The SSE state machine becomes `SseAccumulator` behind `StreamDecoder`:
the transport owns byte reading + `\n\n` framing; the decoder is fed
framed `RawEvent`s. `finish()` returns nothing — `message_stop` is the
only finisher, matching today's no-synthesis contract.
- json_schema synthetic-tool machinery (encode injection, decode
extraction, stream rewrite) moves intact around the shared
`SYNTHETIC_TOOL_NAME`.

### The #482 (Claude Fable 5) port

#482 modifies the old-layout `anthropic.rs` directly, so this branch
re-homes its behavior into the codec structure (commit 3):
`stop_details` on the wire type, the Fable encode gates keyed off the
deployment id (no default adaptive `thinking`, no `temperature`/`top_p`,
no legacy 1M-context beta header — which now lands **once** instead of
twice, since both routes share `build_headers`), refusal →
failover-eligible content-filter errors in decode and stream, and the
`validate_request` rejection of manual thinking configs. The port is
inert until the Fable catalog entry lands. Validated by merging #482's
head into this branch on a scratch branch: the only conflict is
`anthropic.rs` (resolved as this branch's version), and **all of #482's
Fable/refusal tests pass against the codec implementation** (521
fabro-llm tests + fabro-model/fabro-workflow 1286 green on the merged
tree). If #482 merges first, this PR's rebase resolves the same
single-file conflict the same way.

Coordination note: this PR makes the same unit→fielded `CodecParams`
change as #487 (each adds only its own fields) — whichever lands second
resolves a trivial field-union conflict in `codec/mod.rs`.

## Behavior preservation

No behavior change. The anthropic wire snapshots from #471 (direct
route, Kimi-over-anthropic bearer/no-version pin, prompt-cache with
catalog, json_schema, count-tokens wire, streaming happy path / tool
deltas / error events / no-message_stop-no-Finish) pass unmodified, and
the full fabro-llm suite is back to count (515).

## Testing

- `cargo nextest run -p fabro-llm` — 515 passed (126 wire snapshots
included)
- Scratch-merge validation against #482's head — 521 passed incl. its 6
Fable/refusal tests; `cargo nextest run -p fabro-model -p
fabro-workflow` — 1286 passed
- `cargo build --workspace`
- `cargo +nightly-2026-04-14 clippy -p fabro-llm --all-targets -- -D
warnings`
- `cargo +nightly-2026-04-14 fmt --check`

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Scott Werner 2026-06-10 17:01:03 -04:00 • committed by GitHub
parent a4e8987da8
commit 269eca719f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 3076 additions and 2856 deletions

View file

@ -6,18 +6,11 @@
//! Loads that fail drop the part silently — the long-standing contract — and
//! non-file URLs and already-inline data pass through untouched.
//!
//! Shared infra introduced ahead of its consumers: the per-dialect codecs
//! (anthropic/openai_responses/gemini) each construct their own
//! [`AttachmentPolicy`] and call [`resolve`] from their adapter shells when
//! they are wired. Until the first of those lands, nothing here is reachable.
//! Shared infra for the per-dialect codecs (anthropic/openai_responses/gemini):
//! each constructs its own [`AttachmentPolicy`] and calls [`resolve`] from its
//! adapter shell.
// Each dialect adapter that wires this in removes the allow as part of its
// rewire; harmless if it lingers when several land in parallel.
#![allow(
dead_code,
reason = "Attachment-resolution infra added ahead of the dialect codecs (PRs 3-5) that \
construct an AttachmentPolicy and call resolve from their adapter shells."
)]
use std::borrow::Cow;
use crate::providers::common;
use crate::types::{AudioData, ContentPart, DocumentData, ImageData, Request};
@ -32,9 +25,15 @@ pub(crate) struct AttachmentPolicy {
pub audio: bool,
}
/// Return a copy of `request` with file-path attachments (per `policy`)
/// resolved to inline data. Parts whose file fails to load are dropped.
pub(crate) async fn resolve(request: &Request, policy: AttachmentPolicy) -> Request {
/// Resolve file-path attachments (per `policy`) to inline data. Parts whose
/// file fails to load are dropped. Borrows the request untouched in the common
/// case where nothing needs loading; only requests with policy-matching
/// local-file parts pay for a copy.
pub(crate) async fn resolve(request: &Request, policy: AttachmentPolicy) -> Cow<'_, Request> {
if !needs_resolution(request, policy) {
return Cow::Borrowed(request);
}
let mut resolved = request.clone();
for message in &mut resolved.messages {
let mut new_content = Vec::with_capacity(message.content.len());
@ -45,7 +44,21 @@ pub(crate) async fn resolve(request: &Request, policy: AttachmentPolicy) -> Requ
}
message.content = new_content;
}
resolved
Cow::Owned(resolved)
}
/// Whether any part is a policy-matching local-file attachment.
fn needs_resolution(request: &Request, policy: AttachmentPolicy) -> bool {
request
.messages
.iter()
.flat_map(|message| &message.content)
.any(|part| match part {
ContentPart::Image(img) => policy.images && is_local_file(img.url.as_deref()),
ContentPart::Document(doc) => policy.documents && is_local_file(doc.url.as_deref()),
ContentPart::Audio(audio) => policy.audio && is_local_file(audio.url.as_deref()),
_ => false,
})
}
/// Resolve a single part. `None` means the part was dropped (load error).

View file

@ -0,0 +1,263 @@
//! Response decoding: Anthropic Messages body → canonical `Response`.
use serde::Deserialize;
use super::SYNTHETIC_TOOL_NAME;
use super::wire::{ApiResponse, ApiUsage, CountTokensResponse};
use crate::codec::CodecCtx;
use crate::error::{Error, ProviderErrorDetail, ProviderErrorKind};
use crate::types::{
ContentPart, FinishReason, Message, RateLimitInfo, Request, Response, ResponseFormatType, Role,
ThinkingData, TokenCounts, ToolCall,
};
pub(super) fn token_counts_from_api_usage(usage: &ApiUsage) -> TokenCounts {
// Anthropic does not expose a separate billed thinking/reasoning token
// count. Thinking tokens are billed as part of `output_tokens`. When
// Anthropic adds a real thinking token field, wire it through and subtract
// it here.
TokenCounts {
input_tokens: usage.input_tokens,
output_tokens: usage.output_tokens,
reasoning_tokens: 0,
cache_read_tokens: usage.cache_read_input_tokens.unwrap_or(0),
cache_write_tokens: usage.cache_creation_input_tokens.unwrap_or(0),
}
}
pub(super) fn map_finish_reason(stop_reason: Option<&str>) -> FinishReason {
match stop_reason {
Some("end_turn" | "stop_sequence") | None => FinishReason::Stop,
Some("max_tokens") => FinishReason::Length,
Some("tool_use") => FinishReason::ToolCalls,
Some(other) => FinishReason::Other(other.to_string()),
}
}
pub(super) fn parse_content_block(block: &serde_json::Value) -> Option<ContentPart> {
match block.get("type")?.as_str()? {
"text" => Some(ContentPart::text(block.get("text")?.as_str()?)),
"tool_use" => Some(ContentPart::ToolCall(ToolCall::new(
block.get("id")?.as_str()?,
block.get("name")?.as_str()?,
block.get("input")?.clone(),
))),
"thinking" => Some(ContentPart::Thinking(ThinkingData {
text: block.get("thinking")?.as_str()?.to_string(),
signature: block
.get("signature")
.and_then(serde_json::Value::as_str)
.map(String::from),
redacted: false,
})),
"redacted_thinking" => Some(ContentPart::Thinking(ThinkingData {
text: block
.get("data")
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_string(),
signature: None,
redacted: true,
})),
_ => None,
}
}
/// Convert synthetic `tool_use` content blocks back to text content parts.
///
/// When `response_format` uses `JsonSchema` mode, the model responds with a
/// `tool_use` block for our synthetic tool. We extract its arguments as a JSON
/// text string.
pub(super) fn convert_synthetic_tool_to_text(content_parts: Vec<ContentPart>) -> Vec<ContentPart> {
content_parts
.into_iter()
.map(|part| match &part {
ContentPart::ToolCall(tc) if tc.name == SYNTHETIC_TOOL_NAME => {
ContentPart::text(tc.arguments.to_string())
}
_ => part,
})
.collect()
}
/// Check if the request uses `JsonSchema` `response_format`.
pub(super) fn uses_json_schema_format(request: &Request) -> bool {
request
.response_format
.as_ref()
.is_some_and(|f| matches!(f.kind, ResponseFormatType::JsonSchema))
}
/// Map a refusal stop reason (Claude Fable 5) to a content-filter provider
/// error. Shared by the response decoder and the stream decoder; the
/// `error_code = "refusal"` marker is what makes it failover-eligible.
pub(super) fn refusal_error(
provider_name: &str,
model: &str,
raw: serde_json::Value,
stop_details: Option<&serde_json::Value>,
) -> Error {
let model_label = if model.is_empty() { "The model" } else { model };
let message = stop_details
.and_then(|details| details.get("explanation"))
.and_then(serde_json::Value::as_str)
.map_or_else(
|| format!("{model_label} refused the request"),
|explanation| format!("{model_label} refused the request: {explanation}"),
);
Error::Provider {
kind: ProviderErrorKind::ContentFilter,
detail: Box::new(ProviderErrorDetail {
message,
provider: provider_name.to_string(),
status_code: None,
error_code: Some("refusal".to_string()),
retry_after: None,
raw: Some(raw),
}),
}
}
pub(super) fn decode_response(
body: &str,
ctx: &CodecCtx<'_>,
rate_limit: Option<RateLimitInfo>,
) -> Result<Response, Error> {
let raw: serde_json::Value = serde_json::from_str(body).map_err(|e| {
Error::network(
format!("failed to parse {} response: {e}", ctx.provider_name),
e,
)
})?;
let api_resp = ApiResponse::deserialize(&raw).map_err(|e| {
Error::network(
format!("failed to parse {} response: {e}", ctx.provider_name),
e,
)
})?;
if api_resp.stop_reason.as_deref() == Some("refusal") {
return Err(refusal_error(
ctx.provider_name,
&api_resp.model,
raw,
api_resp.stop_details.as_ref(),
));
}
let content_parts: Vec<ContentPart> = api_resp
.content
.iter()
.filter_map(parse_content_block)
.collect();
// If we used JsonSchema mode, convert the synthetic tool call back to text.
let json_schema_mode = uses_json_schema_format(ctx.request);
let content_parts = if json_schema_mode {
convert_synthetic_tool_to_text(content_parts)
} else {
content_parts
};
let finish_reason = if json_schema_mode {
// The model was forced to call a tool, so stop_reason is "tool_use",
// but from the caller's perspective, the request completed normally.
FinishReason::Stop
} else {
map_finish_reason(api_resp.stop_reason.as_deref())
};
Ok(Response {
id: api_resp.id,
model: api_resp.model,
provider: ctx.provider_name.to_string(),
message: Message {
role: Role::Assistant,
content: content_parts,
name: None,
tool_call_id: None,
},
finish_reason,
usage: token_counts_from_api_usage(&api_resp.usage),
raw: Some(raw),
warnings: vec![],
rate_limit,
})
}
pub(super) fn decode_count_tokens(body: &str) -> Result<i64, Error> {
let response: CountTokensResponse =
serde_json::from_str(body).map_err(|e| Error::Configuration {
message: format!("failed to parse token count response: {e}"),
source: None,
})?;
Ok(response.input_tokens)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn api_token_counts_leaves_reasoning_zero_and_output_full() {
let body = serde_json::json!({
"id": "msg_test",
"model": "claude-sonnet-4-5",
"content": [
{ "type": "thinking", "thinking": "summary text", "signature": "" },
{ "type": "text", "text": "answer" }
],
"stop_reason": "end_turn",
"usage": {
"input_tokens": 50,
"output_tokens": 1200,
"cache_read_input_tokens": 9000,
"cache_creation_input_tokens": 1000
}
});
let api: ApiResponse = serde_json::from_value(body).unwrap();
let usage = token_counts_from_api_usage(&api.usage);
assert_eq!(usage.input_tokens, 50);
assert_eq!(usage.cache_read_tokens, 9000);
assert_eq!(usage.cache_write_tokens, 1000);
assert_eq!(usage.output_tokens, 1200);
assert_eq!(usage.reasoning_tokens, 0);
assert_eq!(usage.total_tokens(), 11_250);
}
#[test]
fn convert_synthetic_tool_to_text_replaces_synthetic_tool() {
let parts = vec![ContentPart::ToolCall(ToolCall::new(
"id1",
SYNTHETIC_TOOL_NAME,
serde_json::json!({"name": "Alice"}),
))];
let result = convert_synthetic_tool_to_text(parts);
assert_eq!(result.len(), 1);
match &result[0] {
ContentPart::Text(text) => {
assert!(text.contains("Alice"));
}
other => panic!("expected Text, got {other:?}"),
}
}
#[test]
fn convert_synthetic_tool_to_text_preserves_other_tool_calls() {
let parts = vec![ContentPart::ToolCall(ToolCall::new(
"id1",
"real_tool",
serde_json::json!({"key": "value"}),
))];
let result = convert_synthetic_tool_to_text(parts);
assert_eq!(result.len(), 1);
match &result[0] {
ContentPart::ToolCall(tc) => {
assert_eq!(tc.name, "real_tool");
}
other => panic!("expected ToolCall, got {other:?}"),
}
}
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,63 @@
//! The Anthropic Messages (`/messages`) codec.
//!
//! Serves Anthropic direct today, and (via route config + `CodecParams`)
//! Kimi-over-anthropic; the Bedrock and OpenRouter-skin routes pair the same
//! codec with different transports later. Pure translation: no HTTP, auth, or
//! base URL — the adapter shell owns those.
//!
//! HTTP error bodies use the shared `decode_error` default (anthropic uses the
//! standard `error_from_status_code` + `parse_error_body` path); streaming
//! `error` events are mapped inside the decoder (`on_event` → `Err`).
mod decode;
mod encode;
mod stream;
mod wire;
pub(crate) use encode::anthropic_option;
use crate::codec::{Codec, CodecCtx, EncodedRequest, StreamDecoder};
use crate::error::Error;
use crate::types::{RateLimitInfo, Response};
/// Synthetic tool injected to coerce structured (`JsonSchema`) output. Shared
/// across encode (injection), decode (extraction), and stream (rewrite).
pub(super) const SYNTHETIC_TOOL_NAME: &str = "json_output";
/// Codec for the Anthropic Messages wire dialect.
pub(crate) struct AnthropicMessages;
impl Codec for AnthropicMessages {
fn encode(&self, ctx: &CodecCtx<'_>, stream: bool) -> Result<EncodedRequest, Error> {
Ok(encode::encode(ctx, stream))
}
fn decode_response(
&self,
body: &str,
ctx: &CodecCtx<'_>,
rate_limit: Option<RateLimitInfo>,
) -> Result<Response, Error> {
decode::decode_response(body, ctx, rate_limit)
}
fn stream_decoder(
&self,
ctx: &CodecCtx<'_>,
rate_limit: Option<RateLimitInfo>,
) -> Box<dyn StreamDecoder> {
Box::new(stream::SseAccumulator::new(
ctx.provider_name,
decode::uses_json_schema_format(ctx.request),
rate_limit,
))
}
fn encode_count_tokens(&self, ctx: &CodecCtx<'_>) -> Option<Result<EncodedRequest, Error>> {
Some(Ok(encode::encode_count_tokens(ctx)))
}
fn decode_count_tokens(&self, body: &str) -> Result<i64, Error> {
decode::decode_count_tokens(body)
}
}

View file

@ -0,0 +1,654 @@
//! Streaming decoder: Anthropic SSE events → canonical `StreamEvent`s.
//!
//! Byte reading and SSE block framing live in the transport; this decoder is
//! fed framed `RawEvent`s (`event:` type + `data:` JSON). Anthropic never
//! synthesizes a finish on byte-stream end — `message_stop` is the finisher —
//! so `finish()` returns nothing.
use super::SYNTHETIC_TOOL_NAME;
use super::decode::{convert_synthetic_tool_to_text, map_finish_reason, refusal_error};
use crate::codec::{RawEvent, StreamDecoder};
use crate::error::{Error, ProviderErrorDetail, ProviderErrorKind};
use crate::types::{
ContentPart, FinishReason, Message, RateLimitInfo, Response, Role, StreamEvent, ThinkingData,
TokenCounts, ToolCall,
};
/// The type of the current content block being streamed.
#[derive(Clone)]
enum ContentBlockKind {
Text,
ToolUse { id: String, name: String },
Thinking { signature: Option<String> },
}
/// Accumulated state across SSE events during streaming.
pub(super) struct SseAccumulator {
id: String,
model: String,
/// Configured provider name stamped into the final `Response.provider`.
provider: String,
/// When true, synthetic-tool events are rewritten to text events.
json_schema_mode: bool,
content_parts: Vec<ContentPart>,
usage: TokenCounts,
finish_reason: FinishReason,
current_block: Option<ContentBlockKind>,
current_text: String,
current_thinking: String,
current_tool_args: String,
rate_limit: Option<RateLimitInfo>,
}
impl SseAccumulator {
pub(super) fn new(
provider: &str,
json_schema_mode: bool,
rate_limit: Option<RateLimitInfo>,
) -> Self {
Self {
id: String::new(),
model: String::new(),
provider: provider.to_string(),
json_schema_mode,
content_parts: Vec::new(),
usage: TokenCounts::default(),
finish_reason: FinishReason::Stop,
current_block: None,
current_text: String::new(),
current_thinking: String::new(),
current_tool_args: String::new(),
rate_limit,
}
}
fn take_response(&mut self) -> Response {
Response {
id: std::mem::take(&mut self.id),
model: std::mem::take(&mut self.model),
provider: self.provider.clone(),
message: Message {
role: Role::Assistant,
content: std::mem::take(&mut self.content_parts),
name: None,
tool_call_id: None,
},
finish_reason: std::mem::replace(&mut self.finish_reason, FinishReason::Stop),
usage: std::mem::take(&mut self.usage),
raw: None,
warnings: vec![],
rate_limit: self.rate_limit.take(),
}
}
fn process_event(&mut self, event_type: &str, data: &serde_json::Value) -> Vec<StreamEvent> {
match event_type {
"message_start" => self.handle_message_start(data),
"content_block_start" => self.handle_content_block_start(data),
"content_block_delta" => self.handle_content_block_delta(data),
"content_block_stop" => self.handle_content_block_stop(data),
"message_delta" => {
self.handle_message_delta(data);
vec![]
}
"message_stop" => self.handle_message_stop(),
_ => vec![],
}
}
fn handle_message_start(&mut self, data: &serde_json::Value) -> Vec<StreamEvent> {
if let Some(message) = data.get("message") {
if let Some(id) = message.get("id").and_then(serde_json::Value::as_str) {
self.id = id.to_string();
}
if let Some(model) = message.get("model").and_then(serde_json::Value::as_str) {
self.model = model.to_string();
}
if let Some(usage) = message.get("usage") {
self.usage.input_tokens = usage
.get("input_tokens")
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
self.usage.cache_read_tokens = usage
.get("cache_read_input_tokens")
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
self.usage.cache_write_tokens = usage
.get("cache_creation_input_tokens")
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
}
}
vec![StreamEvent::StreamStart]
}
fn handle_content_block_start(&mut self, data: &serde_json::Value) -> Vec<StreamEvent> {
let block_type = data
.get("content_block")
.and_then(|b| b.get("type"))
.and_then(serde_json::Value::as_str)
.unwrap_or("");
let text_id = Some(block_text_id(data));
match block_type {
"text" => {
self.current_block = Some(ContentBlockKind::Text);
self.current_text.clear();
vec![StreamEvent::TextStart { text_id }]
}
"tool_use" => {
let content_block = data.get("content_block");
let id = content_block
.and_then(|b| b.get("id"))
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_string();
let name = content_block
.and_then(|b| b.get("name"))
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_string();
self.current_block = Some(ContentBlockKind::ToolUse {
id: id.clone(),
name: name.clone(),
});
self.current_tool_args.clear();
vec![StreamEvent::ToolCallStart {
tool_call: ToolCall::new(id, name, serde_json::json!({})),
}]
}
"thinking" => {
let signature = data
.get("content_block")
.and_then(|b| b.get("signature"))
.and_then(serde_json::Value::as_str)
.map(String::from);
self.current_block = Some(ContentBlockKind::Thinking { signature });
self.current_thinking.clear();
vec![StreamEvent::ReasoningStart]
}
_ => vec![],
}
}
fn handle_content_block_delta(&mut self, data: &serde_json::Value) -> Vec<StreamEvent> {
let delta = data.get("delta");
let delta_type = delta
.and_then(|d| d.get("type"))
.and_then(serde_json::Value::as_str)
.unwrap_or("");
match delta_type {
"text_delta" => {
let text = delta
.and_then(|d| d.get("text"))
.and_then(serde_json::Value::as_str)
.unwrap_or("");
self.current_text.push_str(text);
vec![StreamEvent::TextDelta {
delta: text.to_string(),
text_id: Some(block_text_id(data)),
}]
}
"input_json_delta" => {
let partial_json = delta
.and_then(|d| d.get("partial_json"))
.and_then(serde_json::Value::as_str)
.unwrap_or("");
self.current_tool_args.push_str(partial_json);
if let Some(ContentBlockKind::ToolUse { id, name }) = &self.current_block {
vec![StreamEvent::ToolCallDelta {
tool_call: ToolCall::new(
id.clone(),
name.clone(),
serde_json::json!(partial_json),
),
}]
} else {
vec![]
}
}
"thinking_delta" => {
let thinking = delta
.and_then(|d| d.get("thinking"))
.and_then(serde_json::Value::as_str)
.unwrap_or("");
self.current_thinking.push_str(thinking);
vec![StreamEvent::ReasoningDelta {
delta: thinking.to_string(),
}]
}
"signature_delta" => {
let signature = delta
.and_then(|d| d.get("signature"))
.and_then(serde_json::Value::as_str)
.map(String::from);
if let Some(ContentBlockKind::Thinking {
signature: ref mut sig,
}) = self.current_block
{
*sig = signature;
}
vec![]
}
_ => vec![],
}
}
fn handle_content_block_stop(&mut self, data: &serde_json::Value) -> Vec<StreamEvent> {
let current_block = self.current_block.take();
match current_block {
Some(ContentBlockKind::Text) => {
let text = std::mem::take(&mut self.current_text);
self.content_parts.push(ContentPart::text(text));
vec![StreamEvent::TextEnd {
text_id: Some(block_text_id(data)),
}]
}
Some(ContentBlockKind::ToolUse { id, name }) => {
let raw_args = std::mem::take(&mut self.current_tool_args);
let arguments =
serde_json::from_str(&raw_args).unwrap_or_else(|_| serde_json::json!({}));
let mut tool_call = ToolCall::new(id, name, arguments);
tool_call.raw_arguments = Some(raw_args);
self.content_parts
.push(ContentPart::ToolCall(tool_call.clone()));
vec![StreamEvent::ToolCallEnd { tool_call }]
}
Some(ContentBlockKind::Thinking { signature }) => {
let thinking_text = std::mem::take(&mut self.current_thinking);
// Prefer signature from content_block_stop if available, fall
// back to one captured at content_block_start.
let stop_signature = data
.get("content_block")
.and_then(|b| b.get("signature"))
.and_then(serde_json::Value::as_str)
.map(String::from);
self.content_parts.push(ContentPart::Thinking(ThinkingData {
text: thinking_text,
signature: stop_signature.or(signature),
redacted: false,
}));
vec![StreamEvent::ReasoningEnd]
}
None => vec![],
}
}
fn handle_message_delta(&mut self, data: &serde_json::Value) {
if let Some(delta) = data.get("delta") {
let stop_reason = delta.get("stop_reason").and_then(serde_json::Value::as_str);
self.finish_reason = map_finish_reason(stop_reason);
}
if let Some(usage) = data.get("usage") {
self.usage.output_tokens = usage
.get("output_tokens")
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
}
}
fn handle_message_stop(&mut self) -> Vec<StreamEvent> {
let response = self.take_response();
vec![StreamEvent::Finish {
finish_reason: response.finish_reason.clone(),
usage: response.usage.clone(),
response: Box::new(response),
}]
}
}
/// The `text_id` for a content-block event: `block_<index>`.
fn block_text_id(data: &serde_json::Value) -> String {
let index = data
.get("index")
.and_then(serde_json::Value::as_u64)
.unwrap_or(0);
format!("block_{index}")
}
/// Extract the `stop_details` from a refusal `message_delta`, if present.
fn refusal_stop_details(data: &serde_json::Value) -> Option<&serde_json::Value> {
data.get("delta")
.and_then(|delta| delta.get("stop_details"))
}
/// Whether a `message_delta` event carries a refusal stop reason.
fn is_refusal_message_delta(event_type: &str, data: &serde_json::Value) -> bool {
event_type == "message_delta"
&& data
.get("delta")
.and_then(|delta| delta.get("stop_reason"))
.and_then(serde_json::Value::as_str)
== Some("refusal")
}
/// Wrap a refusal stream event in the same raw shape the non-streaming
/// refusal error carries (`stop_reason` + `stop_details` + the event).
fn refusal_stream_raw(data: &serde_json::Value) -> serde_json::Value {
serde_json::json!({
"stop_reason": "refusal",
"stop_details": refusal_stop_details(data)
.cloned()
.unwrap_or(serde_json::Value::Null),
"stream_event": data,
})
}
/// Map an Anthropic `error` stream event to a provider error.
fn stream_error_event_to_provider_error(data: &serde_json::Value, provider_name: &str) -> Error {
let error = data.get("error").unwrap_or(data);
let message = error
.get("message")
.and_then(serde_json::Value::as_str)
.or_else(|| data.get("message").and_then(serde_json::Value::as_str))
.unwrap_or("Unknown Anthropic stream error")
.to_string();
let error_code = error
.get("type")
.and_then(serde_json::Value::as_str)
.map(String::from);
let kind = match error_code.as_deref() {
Some("rate_limit_error") => ProviderErrorKind::RateLimit,
Some("authentication_error") => ProviderErrorKind::Authentication,
Some("permission_error") => ProviderErrorKind::AccessDenied,
Some("not_found_error") => ProviderErrorKind::NotFound,
Some("invalid_request_error") => ProviderErrorKind::InvalidRequest,
Some("request_too_large") => ProviderErrorKind::ContextLength,
// overloaded_error, api_error, and unknown stream errors are transient.
_ => ProviderErrorKind::Server,
};
Error::Provider {
kind,
detail: Box::new(ProviderErrorDetail {
message,
provider: provider_name.to_string(),
status_code: None,
error_code,
retry_after: None,
raw: Some(data.clone()),
}),
}
}
/// Rewrite a streaming event for `JsonSchema` mode: synthetic-tool events
/// become text events, and the Finish event's content + finish_reason are
/// adjusted.
fn convert_stream_event_for_json_schema(event: StreamEvent) -> StreamEvent {
match event {
StreamEvent::ToolCallStart { tool_call } if tool_call.name == SYNTHETIC_TOOL_NAME => {
StreamEvent::TextStart { text_id: None }
}
StreamEvent::ToolCallDelta { tool_call } if tool_call.name == SYNTHETIC_TOOL_NAME => {
let delta = match tool_call.arguments {
serde_json::Value::String(s) => s,
other => other.to_string(),
};
StreamEvent::TextDelta {
delta,
text_id: None,
}
}
StreamEvent::ToolCallEnd { tool_call } if tool_call.name == SYNTHETIC_TOOL_NAME => {
StreamEvent::TextEnd { text_id: None }
}
StreamEvent::Finish {
mut response,
usage,
..
} => {
response.message.content =
convert_synthetic_tool_to_text(std::mem::take(&mut response.message.content));
response.finish_reason = FinishReason::Stop;
StreamEvent::Finish {
finish_reason: FinishReason::Stop,
usage,
response,
}
}
other => other,
}
}
impl StreamDecoder for SseAccumulator {
fn on_event(&mut self, ev: RawEvent<'_>) -> Result<Vec<StreamEvent>, Error> {
let event_type = ev.event.unwrap_or("");
let data: serde_json::Value = serde_json::from_str(ev.data)
.map_err(|e| Error::stream_error(format!("failed to parse SSE data: {e}"), e))?;
if event_type == "error" {
return Err(stream_error_event_to_provider_error(&data, &self.provider));
}
// A refusal (Claude Fable 5) arrives as a `message_delta` stop reason;
// surface it as an error instead of letting `message_stop` emit a
// normal Finish.
if is_refusal_message_delta(event_type, &data) {
return Err(refusal_error(
&self.provider,
&self.model,
refusal_stream_raw(&data),
refusal_stop_details(&data),
));
}
let events = self.process_event(event_type, &data);
if self.json_schema_mode {
Ok(events
.into_iter()
.map(convert_stream_event_for_json_schema)
.collect())
} else {
Ok(events)
}
}
fn finish(&mut self) -> Vec<StreamEvent> {
// Anthropic relies on `message_stop` to finish; nothing to synthesize.
Vec::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn new_accumulator(provider: &str, json_schema_mode: bool) -> SseAccumulator {
SseAccumulator::new(provider, json_schema_mode, None)
}
#[test]
fn stream_token_counts_leaves_reasoning_zero_and_output_full() {
let mut acc = new_accumulator("anthropic", false);
acc.content_parts.push(ContentPart::Thinking(ThinkingData {
text: "summary text".to_string(),
signature: Some(String::new()),
redacted: false,
}));
acc.content_parts.push(ContentPart::text("answer"));
acc.usage = TokenCounts {
input_tokens: 50,
output_tokens: 1200,
reasoning_tokens: 0,
cache_read_tokens: 9000,
cache_write_tokens: 1000,
};
let events = acc.handle_message_stop();
let StreamEvent::Finish {
usage, response, ..
} = &events[0]
else {
panic!("expected finish event");
};
assert_eq!(usage.input_tokens, 50);
assert_eq!(usage.cache_read_tokens, 9000);
assert_eq!(usage.cache_write_tokens, 1000);
assert_eq!(usage.output_tokens, 1200);
assert_eq!(usage.reasoning_tokens, 0);
assert_eq!(usage.total_tokens(), 11_250);
assert_eq!(response.usage, *usage);
}
#[test]
fn stream_error_event_overloaded_becomes_retryable_server_error() {
let mut acc = new_accumulator("anthropic", false);
let data = serde_json::json!({
"type": "error",
"error": {
"type": "overloaded_error",
"message": "Overloaded"
}
});
let raw = data.to_string();
let err = acc
.on_event(RawEvent {
event: Some("error"),
data: &raw,
})
.unwrap_err();
assert!(err.retryable());
match err {
Error::Provider { kind, detail } => {
assert_eq!(kind, ProviderErrorKind::Server);
assert_eq!(detail.provider, "anthropic");
assert_eq!(detail.message, "Overloaded");
assert_eq!(detail.error_code.as_deref(), Some("overloaded_error"));
assert_eq!(detail.raw.as_ref(), Some(&data));
}
other => panic!("expected provider error, got {other:?}"),
}
}
#[test]
fn stream_error_event_invalid_request_remains_non_retryable() {
let mut acc = new_accumulator("anthropic", false);
let data = serde_json::json!({
"type": "error",
"error": {
"type": "invalid_request_error",
"message": "max_tokens is required"
}
});
let raw = data.to_string();
let err = acc
.on_event(RawEvent {
event: Some("error"),
data: &raw,
})
.unwrap_err();
assert!(!err.retryable());
match err {
Error::Provider { kind, detail } => {
assert_eq!(kind, ProviderErrorKind::InvalidRequest);
assert_eq!(detail.error_code.as_deref(), Some("invalid_request_error"));
}
other => panic!("expected provider error, got {other:?}"),
}
}
#[test]
fn unknown_sse_events_remain_ignored() {
let mut acc = new_accumulator("anthropic", false);
let data = serde_json::json!({
"type": "content_block_delta",
"delta": { "type": "text_delta", "text": "ignored" }
});
let raw = data.to_string();
let events = acc
.on_event(RawEvent {
event: Some("some_future_event"),
data: &raw,
})
.unwrap();
assert!(events.is_empty());
}
#[test]
fn convert_stream_event_converts_tool_start_for_synthetic() {
let event = StreamEvent::ToolCallStart {
tool_call: ToolCall::new("id1", SYNTHETIC_TOOL_NAME, serde_json::json!({})),
};
let result = convert_stream_event_for_json_schema(event);
assert!(matches!(result, StreamEvent::TextStart { .. }));
}
#[test]
fn convert_stream_event_preserves_real_tool_start() {
let event = StreamEvent::ToolCallStart {
tool_call: ToolCall::new("id1", "real_tool", serde_json::json!({})),
};
let result = convert_stream_event_for_json_schema(event);
assert!(matches!(result, StreamEvent::ToolCallStart { .. }));
}
#[test]
fn convert_stream_event_converts_tool_delta_for_synthetic() {
let event = StreamEvent::ToolCallDelta {
tool_call: ToolCall::new("id1", SYNTHETIC_TOOL_NAME, serde_json::json!("{\"name\"")),
};
let result = convert_stream_event_for_json_schema(event);
match result {
StreamEvent::TextDelta { delta, .. } => {
assert_eq!(delta, "{\"name\"");
}
other => panic!("expected TextDelta, got {other:?}"),
}
}
#[test]
fn convert_stream_event_converts_finish_reason() {
let response = Box::new(Response {
id: "test".to_string(),
model: "claude".to_string(),
provider: "anthropic".to_string(),
message: Message {
role: Role::Assistant,
content: vec![ContentPart::ToolCall(ToolCall::new(
"id1",
SYNTHETIC_TOOL_NAME,
serde_json::json!({"data": "value"}),
))],
name: None,
tool_call_id: None,
},
finish_reason: FinishReason::ToolCalls,
usage: TokenCounts::default(),
raw: None,
warnings: vec![],
rate_limit: None,
});
let event = StreamEvent::Finish {
finish_reason: FinishReason::ToolCalls,
usage: TokenCounts::default(),
response,
};
let result = convert_stream_event_for_json_schema(event);
match result {
StreamEvent::Finish {
finish_reason,
response,
..
} => {
assert_eq!(finish_reason, FinishReason::Stop);
assert_eq!(response.finish_reason, FinishReason::Stop);
// Content should be converted from tool call to text
assert!(matches!(&response.message.content[0], ContentPart::Text(_)));
}
other => panic!("expected Finish, got {other:?}"),
}
}
}

View file

@ -0,0 +1,154 @@
//! Serde types mirroring the Anthropic Messages wire shapes.
#[derive(serde::Serialize)]
pub(super) struct ApiRequest {
pub model: String,
pub messages: Vec<ApiMessage>,
pub max_tokens: i64,
/// System prompt: either a plain string or an array of content blocks
/// (with optional `cache_control` annotations for prompt caching).
#[serde(skip_serializing_if = "Option::is_none")]
pub system: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub temperature: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub top_p: Option<f64>,
/// Always serialized, even when empty (pinned by wire tests).
pub stop_sequences: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tools: Option<Vec<ApiToolDef>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tool_choice: Option<serde_json::Value>,
/// Extended thinking configuration (e.g. `{"type": "enabled",
/// "budget_tokens": 10000}`).
#[serde(skip_serializing_if = "Option::is_none")]
pub thinking: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub output_config: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub speed: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub metadata: Option<std::collections::HashMap<String, String>>,
#[serde(skip_serializing_if = "std::ops::Not::not")]
pub stream: bool,
}
#[derive(serde::Serialize)]
pub(super) struct CountTokensRequest {
pub model: String,
pub messages: Vec<ApiMessage>,
#[serde(skip_serializing_if = "Option::is_none")]
pub system: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tools: Option<Vec<ApiToolDef>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tool_choice: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub thinking: Option<serde_json::Value>,
}
impl From<ApiRequest> for CountTokensRequest {
fn from(request: ApiRequest) -> Self {
Self {
model: request.model,
messages: request.messages,
system: request.system,
tools: request.tools,
tool_choice: request.tool_choice,
thinking: request.thinking,
}
}
}
/// Anthropic messages use structured content blocks, not plain strings.
#[derive(serde::Serialize)]
pub(super) struct ApiMessage {
pub role: String,
pub content: Vec<serde_json::Value>,
}
/// Anthropic tool definition format.
#[derive(serde::Serialize)]
pub(super) struct ApiToolDef {
pub name: String,
pub description: String,
pub input_schema: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_control: Option<CacheControl>,
}
/// Anthropic `cache_control` annotation.
#[derive(serde::Serialize, Clone)]
pub(super) struct CacheControl {
#[serde(rename = "type")]
pub kind: String,
}
impl CacheControl {
pub(super) fn ephemeral() -> Self {
Self {
kind: "ephemeral".to_string(),
}
}
}
// --- Response types ---
#[derive(serde::Deserialize)]
pub(super) struct ApiResponse {
pub id: String,
pub model: String,
pub content: Vec<serde_json::Value>,
pub stop_reason: Option<String>,
#[serde(default)]
pub stop_details: Option<serde_json::Value>,
pub usage: ApiUsage,
}
#[derive(serde::Deserialize)]
#[allow(
clippy::struct_field_names,
reason = "Field names mirror the provider API payload."
)]
pub(super) struct ApiUsage {
pub input_tokens: i64,
pub output_tokens: i64,
#[serde(default)]
pub cache_read_input_tokens: Option<i64>,
#[serde(default)]
pub cache_creation_input_tokens: Option<i64>,
}
#[derive(serde::Deserialize)]
pub(super) struct CountTokensResponse {
pub input_tokens: i64,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn tool_serialization_includes_cache_control() {
let tool = ApiToolDef {
name: "test_tool".to_string(),
description: "A test tool".to_string(),
input_schema: serde_json::json!({"type": "object"}),
cache_control: Some(CacheControl::ephemeral()),
};
let json = serde_json::to_value(&tool).expect("should serialize");
assert_eq!(json["cache_control"]["type"], "ephemeral");
}
#[test]
fn tool_serialization_omits_cache_control_when_none() {
let tool = ApiToolDef {
name: "test_tool".to_string(),
description: "A test tool".to_string(),
input_schema: serde_json::json!({"type": "object"}),
cache_control: None,
};
let json = serde_json::to_value(&tool).expect("should serialize");
assert!(json.get("cache_control").is_none());
}
}

View file

@ -10,18 +10,7 @@
//! defaults) so the per-dialect codecs that follow only ever *override*
//! methods, never extend the contract.
// The contract is defined in full now, but `openai_compatible` (the first
// codec) is the simplest dialect and does not exercise every seam: the
// `model`/`params` context, `RawEvent.event`, and the count-tokens methods
// are consumed by the anthropic/openai/gemini codecs and the transport
// consolidation in later PRs of this series. Scoped to this trait-definition
// file; the codec impls below are fully used.
#![allow(
dead_code,
reason = "Codec contract is defined in full ahead of the dialects (PRs 3-6) that exercise \
the capability context, SSE event type, and count-tokens routes."
)]
pub(crate) mod anthropic_messages;
pub(crate) mod openai_compatible;
use fabro_model::Model;
@ -51,11 +40,30 @@ pub(crate) struct CodecCtx<'a> {
}
/// Per-route dialect knobs, expressed as data so one codec can serve several
/// routes. Starts empty; grows by adding `#[serde(default)]` fields (a
/// non-breaking change) — e.g. PR 3 adds version-placement, #459 adds
/// model-placement for Bedrock.
/// routes. The default is inert ("nothing special"); a route that needs a
/// dialect quirk sets the relevant field. Grows as codecs need it — #459 adds
/// `ModelPlacement` for Bedrock.
#[derive(Debug, Default, Clone)]
pub(crate) struct CodecParams;
pub(crate) struct CodecParams {
/// Where/whether to place the Anthropic API version. Direct Anthropic uses
/// `Header("2023-06-01")`; Kimi-over-anthropic uses `None`; the Bedrock
/// redo will add a body-field variant. Inert for non-anthropic codecs.
pub anthropic_version: AnthropicVersion,
/// Whether to emit Anthropic beta headers (prompt-caching / fast-mode /
/// 1M-context). True on the direct route, false for Kimi-over-anthropic.
pub anthropic_beta: bool,
}
/// Placement of the Anthropic API version on the wire.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) enum AnthropicVersion {
/// No version sent (Kimi-over-anthropic; also the inert default).
#[default]
None,
/// `anthropic-version` request header (direct Anthropic).
Header(&'static str),
// BodyField(&'static str) arrives with the Bedrock redo (#459).
}
/// What [`Codec::encode`] produces. The transport applies `endpoint` +
/// `headers` on top of the route's base URL and auth; the codec never touches
@ -131,9 +139,9 @@ pub(crate) trait Codec: Send + Sync {
/// Map a non-2xx response to an `Error`. `retry_after` is the
/// transport-parsed `retry-after` header value in seconds (header parsing
/// is the transport's job, like `rate_limit` on the decode methods).
/// Default = shared HTTP-status mapping (what openai_compatible uses);
/// anthropic/openai/gemini override to fold in dialect error bodies
/// (error.type, gRPC status, …).
/// Default = shared HTTP-status mapping, which openai_compatible and
/// anthropic use as-is; a codec overrides when its dialect's error bodies
/// need more (e.g. gemini's gRPC status).
fn decode_error(
&self,
status: u16,

View file

@ -107,7 +107,7 @@ mod tests {
/// Encode `request` through the codec with `deployment_id == request.model`
/// (the no-catalog case) and return the body.
fn encode_body(request: &Request, provider_name: &str, stream: bool) -> serde_json::Value {
let params = CodecParams;
let params = CodecParams::default();
let deployment_id = request.model.clone();
let ctx = CodecCtx {
request,
@ -155,7 +155,7 @@ mod tests {
#[test]
fn encode_uses_deployment_id_as_model() {
let request = minimal_request();
let params = CodecParams;
let params = CodecParams::default();
let deployment_id = "acme/model-large".to_string();
let ctx = CodecCtx {
request: &request,

View file

@ -284,7 +284,7 @@ mod tests {
metadata: None,
provider_options: None,
};
let params = CodecParams;
let params = CodecParams::default();
let ctx = CodecCtx {
request: &request,
provider_name: provider,

File diff suppressed because it is too large Load diff

View file

@ -163,7 +163,7 @@ impl ProviderAdapter for Adapter {
let codec = OpenAiCompatible;
let deployment_id = self.deployment_id(request);
let params = CodecParams;
let params = CodecParams::default();
let ctx = self.codec_ctx(request, &deployment_id, &params);
let mut req = self.encoded_request(&codec, &ctx, false)?;
@ -181,7 +181,7 @@ impl ProviderAdapter for Adapter {
let codec = OpenAiCompatible;
let deployment_id = self.deployment_id(request);
let params = CodecParams;
let params = CodecParams::default();
let ctx = self.codec_ctx(request, &deployment_id, &params);
let req = self.encoded_request(&codec, &ctx, true)?;