use crate::ollama_chat::{HttpMethod, NetworkPolicy, ReplayBoundary, ReplayDecision};
use router_protocol::{
CanonicalRequest, ContentBlock, InputItem, MessageRole, PhaseKind, RouteIdentity,
RouteIdentityError, ToolChoice, TransportKind,
};
use serde_json::{Map, Value, json};
use std::fmt;
pub const ANTHROPIC_VERSION: &str = router_protocol::ANTHROPIC_VERSION;
pub const ANTHROPIC_API_KEY_HEADER: &str = router_protocol::ANTHROPIC_API_KEY_HEADER;
pub const ANTHROPIC_VERSION_HEADER: &str = router_protocol::ANTHROPIC_VERSION_HEADER;
pub const DEFAULT_MAX_TOKENS: u32 = 1024;
pub const fn is_claude_phase(phase: PhaseKind) -> bool {
matches!(phase, PhaseKind::ClaudeSubscription | PhaseKind::ClaudeApi)
}
pub const fn capability_projection() -> router_protocol::MultiAgentCapabilityProjection {
router_protocol::MultiAgentCapabilityProjection {
namespace_tools: false,
multi_agent_version: router_protocol::MultiAgentVersion::V2,
wire_api: router_protocol::MultiAgentWireApi::Responses,
observed_namespace_calls:
router_protocol::ObservedNamespaceCallBehavior::BareNameNoNamespace,
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct AnthropicSecretReferenceMarker(String);
impl AnthropicSecretReferenceMarker {
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for AnthropicSecretReferenceMarker {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("AnthropicSecretReferenceMarker([REDACTED])")
}
}
#[derive(Clone, PartialEq)]
pub struct AnthropicRequestSpec {
pub method: HttpMethod,
pub origin: String,
pub path: String,
pub content_type: &'static str,
pub accept: &'static str,
pub api_key_secret: AnthropicSecretReferenceMarker,
pub network: NetworkPolicy,
pub version_header: &'static str,
body: Value,
}
impl AnthropicRequestSpec {
pub fn body(&self) -> &Value {
&self.body
}
pub fn body_bytes(&self) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&self.body)
}
}
impl fmt::Debug for AnthropicRequestSpec {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("AnthropicRequestSpec")
.field("method", &self.method)
.field("origin", &self.origin)
.field("path", &self.path)
.field("content_type", &self.content_type)
.field("accept", &self.accept)
.field("api_key_secret", &self.api_key_secret)
.field("network", &self.network)
.field("version_header", &self.version_header)
.field("body", &"[REDACTED]")
.finish()
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AnthropicRequestErrorKind {
InvalidRouteIdentity,
UnsupportedRoute,
InvalidCanonicalRequest,
AmbiguousToolName,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct AnthropicRequestError {
pub kind: AnthropicRequestErrorKind,
}
impl fmt::Display for AnthropicRequestError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("the Anthropic Messages request could not be prepared")
}
}
impl std::error::Error for AnthropicRequestError {}
impl From<RouteIdentityError> for AnthropicRequestError {
fn from(_: RouteIdentityError) -> Self {
Self {
kind: AnthropicRequestErrorKind::InvalidRouteIdentity,
}
}
}
pub fn build_request(
route: &RouteIdentity,
request: &CanonicalRequest,
) -> Result<AnthropicRequestSpec, AnthropicRequestError> {
route.validate()?;
if !is_claude_phase(route.phase) || route.transport != TransportKind::DirectHttps {
return Err(AnthropicRequestError {
kind: AnthropicRequestErrorKind::UnsupportedRoute,
});
}
request.validate().map_err(|_| AnthropicRequestError {
kind: AnthropicRequestErrorKind::InvalidCanonicalRequest,
})?;
let secret_ref = route.credential_ref.clone().ok_or(AnthropicRequestError {
kind: AnthropicRequestErrorKind::InvalidRouteIdentity,
})?;
Ok(AnthropicRequestSpec {
method: HttpMethod::Post,
origin: route.endpoint.origin.clone(),
path: route.endpoint.path.clone(),
content_type: "application/json",
accept: "text/event-stream",
api_key_secret: AnthropicSecretReferenceMarker(secret_ref),
network: NetworkPolicy {
allow_proxy: false,
allow_dns: true,
follow_redirects: false,
require_tls: true,
},
version_header: ANTHROPIC_VERSION,
body: build_body(route, request)?,
})
}
fn build_body(
route: &RouteIdentity,
request: &CanonicalRequest,
) -> Result<Value, AnthropicRequestError> {
let mut object = Map::new();
object.insert("model".to_owned(), Value::String(route.model_id.clone()));
object.insert(
"max_tokens".to_owned(),
Value::from(request.max_output_tokens.unwrap_or(DEFAULT_MAX_TOKENS)),
);
object.insert("stream".to_owned(), Value::Bool(true));
let system = build_system(request);
if !system.is_empty() {
object.insert("system".to_owned(), Value::String(system));
}
let emit_tools = !request.tools.is_empty() && request.tool_choice != ToolChoice::None;
if emit_tools {
object.insert("tool_choice".to_owned(), anthropic_tool_choice(request));
let mut names = std::collections::HashSet::new();
let mut tools = Vec::with_capacity(request.tools.len());
for tool in &request.tools {
let name = provider_tool_name(tool.namespace.as_deref(), &tool.name);
if !names.insert(name.clone()) {
return Err(AnthropicRequestError {
kind: AnthropicRequestErrorKind::AmbiguousToolName,
});
}
tools.push(json!({
"name": name,
"description": tool.description,
"input_schema": tool.parameters,
}));
}
object.insert("tools".to_owned(), Value::Array(tools));
}
object.insert("messages".to_owned(), Value::Array(build_messages(request)));
Ok(Value::Object(object))
}
fn build_system(request: &CanonicalRequest) -> String {
request
.input
.iter()
.filter_map(|item| match item {
InputItem::Message {
role: MessageRole::Developer | MessageRole::System,
content,
} => Some(content),
_ => None,
})
.flat_map(|content| {
content.iter().filter_map(|block| match block {
ContentBlock::Text { text } => Some(text.as_str()),
ContentBlock::ImageReference { .. } => None,
})
})
.collect::<Vec<_>>()
.join("\n")
}
fn build_messages(request: &CanonicalRequest) -> Vec<Value> {
request
.input
.iter()
.map(|item| match item {
InputItem::Message {
role: MessageRole::User,
content,
} => json!({
"role": "user",
"content": anthropic_content(content),
}),
InputItem::Message {
role: MessageRole::Assistant,
content,
} => json!({
"role": "assistant",
"content": anthropic_content(content),
}),
InputItem::Message {
role: MessageRole::Developer | MessageRole::System,
..
} => json!({
"role": "user",
"content": [],
}),
InputItem::FunctionCall {
call_id,
namespace,
name,
arguments,
} => json!({
"role": "assistant",
"content": [{
"type": "tool_use",
"id": call_id.as_str(),
"name": provider_tool_name(namespace.as_deref(), name),
"input": arguments,
}]
}),
InputItem::FunctionCallOutput { call_id, output } => json!({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": call_id.as_str(),
"content": output.to_string(),
}]
}),
})
.collect()
}
fn anthropic_content(content: &[ContentBlock]) -> Value {
if content
.iter()
.all(|block| matches!(block, ContentBlock::Text { .. }))
{
let text = content
.iter()
.filter_map(|block| match block {
ContentBlock::Text { text } => Some(text.as_str()),
ContentBlock::ImageReference { .. } => None,
})
.collect::<String>();
return Value::String(text);
}
Value::Array(
content
.iter()
.filter_map(|block| match block {
ContentBlock::Text { text } => Some(json!({"type": "text", "text": text})),
ContentBlock::ImageReference { .. } => {
None
}
})
.collect(),
)
}
fn provider_tool_name(namespace: Option<&str>, name: &str) -> String {
namespace.map_or_else(
|| name.to_owned(),
|namespace| format!("{namespace}__{name}"),
)
}
fn anthropic_tool_choice(request: &CanonicalRequest) -> Value {
match &request.tool_choice {
ToolChoice::Auto => Value::String("auto".to_owned()),
ToolChoice::None => Value::String("none".to_owned()),
ToolChoice::Required => Value::String("any".to_owned()),
ToolChoice::Named(name) => {
let provider_name = request
.tools
.iter()
.find(|tool| tool.name == *name)
.map_or_else(
|| name.clone(),
|tool| provider_tool_name(tool.namespace.as_deref(), &tool.name),
);
json!({"type": "tool", "name": provider_name})
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct AnthropicUsage {
pub input_tokens: u64,
pub output_tokens: u64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AnthropicStreamEvent {
ContentDelta { text: String },
FinishReason { reason: String },
Usage(AnthropicUsage),
Done,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AnthropicStreamErrorKind {
InvalidUtf8,
InvalidJson,
InvalidFrame,
EventTooLarge,
MissingTerminal,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct AnthropicStreamError {
pub kind: AnthropicStreamErrorKind,
}
impl fmt::Display for AnthropicStreamError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("the Anthropic stream could not be normalized")
}
}
impl std::error::Error for AnthropicStreamError {}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct AnthropicReplayGuard {
boundary: Option<ReplayBoundary>,
body_written: bool,
}
impl AnthropicReplayGuard {
pub fn mark_body_written(&mut self) {
self.body_written = true;
}
pub fn observe(&mut self, event: &AnthropicStreamEvent) {
match event {
AnthropicStreamEvent::ContentDelta { text } if !text.is_empty() => {
self.cross(ReplayBoundary::PublicContent);
}
AnthropicStreamEvent::ContentDelta { .. }
| AnthropicStreamEvent::FinishReason { .. }
| AnthropicStreamEvent::Usage(_)
| AnthropicStreamEvent::Done => {}
}
}
pub fn mark_transport_outcome_unknown(&mut self) {
if self.body_written {
self.cross(ReplayBoundary::AmbiguousDispatch);
}
}
pub fn decision(&self) -> ReplayDecision {
self.boundary
.map_or(ReplayDecision::Allowed, ReplayDecision::Denied)
}
fn cross(&mut self, boundary: ReplayBoundary) {
if self.boundary.is_none() {
self.boundary = Some(boundary);
}
}
}
pub struct AnthropicSseParser {
_attempt_id: AttemptId,
replay: AnthropicReplayGuard,
terminal: bool,
input_tokens: u64,
output_tokens: u64,
}
impl AnthropicSseParser {
pub fn new(attempt_id: &AttemptId) -> Self {
Self {
_attempt_id: attempt_id.clone(),
replay: AnthropicReplayGuard::default(),
terminal: false,
input_tokens: 0,
output_tokens: 0,
}
}
pub fn replay_decision(&self) -> ReplayDecision {
self.replay.decision()
}
pub fn mark_body_written(&mut self) {
self.replay.mark_body_written();
}
pub fn mark_transport_outcome_unknown(&mut self) {
self.replay.mark_transport_outcome_unknown();
}
pub fn feed(
&mut self,
chunk: &[u8],
) -> Result<Vec<AnthropicStreamEvent>, AnthropicStreamError> {
let mut events = Vec::new();
for frame in split_sse_frames(chunk)? {
let value: Value =
serde_json::from_slice(&frame).map_err(|_| AnthropicStreamError {
kind: AnthropicStreamErrorKind::InvalidJson,
})?;
events.extend(self.observe_json(value)?);
}
for event in &events {
self.replay.observe(event);
}
Ok(events)
}
pub fn finish(&mut self) -> Result<Vec<AnthropicStreamEvent>, AnthropicStreamError> {
if !self.terminal {
return Err(AnthropicStreamError {
kind: AnthropicStreamErrorKind::MissingTerminal,
});
}
Ok(Vec::new())
}
fn observe_json(
&mut self,
value: Value,
) -> Result<Vec<AnthropicStreamEvent>, AnthropicStreamError> {
let event_type = value.get("type").and_then(Value::as_str);
let mut events = Vec::new();
match event_type {
Some("message_start") => {
if let Some(input) = value
.pointer("/message/usage/input_tokens")
.and_then(Value::as_u64)
{
self.input_tokens = input;
}
}
Some("content_block_delta") => {
match value.pointer("/delta/type").and_then(Value::as_str) {
Some("text_delta") => {
if let Some(text) = value.pointer("/delta/text").and_then(Value::as_str) {
events.push(AnthropicStreamEvent::ContentDelta {
text: text.to_owned(),
});
}
}
Some(_) => {}
None => {
return Err(AnthropicStreamError {
kind: AnthropicStreamErrorKind::InvalidFrame,
});
}
}
}
Some("message_delta") => {
if let Some(reason) = value.pointer("/delta/stop_reason").and_then(Value::as_str) {
events.push(AnthropicStreamEvent::FinishReason {
reason: reason.to_owned(),
});
}
if let Some(output) = value
.pointer("/usage/output_tokens")
.and_then(Value::as_u64)
{
self.output_tokens = output;
}
}
Some("message_stop") => {
self.terminal = true;
events.push(AnthropicStreamEvent::Usage(AnthropicUsage {
input_tokens: self.input_tokens,
output_tokens: self.output_tokens,
}));
events.push(AnthropicStreamEvent::Done);
}
Some("ping") | Some("content_block_start") | Some("content_block_stop") => {}
_ => {
return Err(AnthropicStreamError {
kind: AnthropicStreamErrorKind::InvalidFrame,
});
}
}
Ok(events)
}
}
fn split_sse_frames(buffer: &[u8]) -> Result<Vec<Vec<u8>>, AnthropicStreamError> {
let text = std::str::from_utf8(buffer).map_err(|_| AnthropicStreamError {
kind: AnthropicStreamErrorKind::InvalidUtf8,
})?;
let mut frames = Vec::new();
let mut current = Vec::new();
for line in text.split('\n') {
let line = line.strip_suffix('\r').unwrap_or(line);
if line.is_empty() {
if !current.is_empty() {
frames.push(std::mem::take(&mut current));
}
continue;
}
if let Some(data) = line.strip_prefix("data:") {
let data = data.strip_prefix(' ').unwrap_or(data);
if !current.is_empty() {
current.push(b'\n');
}
current.extend_from_slice(data.as_bytes());
}
}
if !current.is_empty() {
frames.push(current);
}
if frames.is_empty() {
return Err(AnthropicStreamError {
kind: AnthropicStreamErrorKind::InvalidFrame,
});
}
Ok(frames)
}
use router_protocol::AttemptId;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AnthropicFailureClass {
InvalidRequest,
Authentication,
PaymentRequired,
Authorization,
NotFound,
Timeout,
Unprocessable,
RateLimited,
UpstreamServer,
UnexpectedStatus,
MismatchedContentType,
RedirectRejected,
Cancelled,
TransportTimeout,
AmbiguousDispatch,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct AnthropicFailure {
pub class: AnthropicFailureClass,
pub status: Option<u16>,
pub retry_after: Option<crate::ollama_chat::BoundedRetryAfter>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AnthropicBodyWriteState {
NotWritten,
Written,
Unknown,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AnthropicTransportFailureSignal {
Cancelled,
Timeout,
}
pub fn normalize_http_response(
status: u16,
content_type: Option<&str>,
retry_after: Option<&str>,
redirected: bool,
) -> Result<(), AnthropicFailure> {
if redirected || (300..400).contains(&status) {
return Err(anthropic_failure(
AnthropicFailureClass::RedirectRejected,
Some(status),
retry_after,
));
}
if status != 200 {
let class = match status {
400 => AnthropicFailureClass::InvalidRequest,
401 => AnthropicFailureClass::Authentication,
402 => AnthropicFailureClass::PaymentRequired,
403 => AnthropicFailureClass::Authorization,
404 => AnthropicFailureClass::NotFound,
408 => AnthropicFailureClass::Timeout,
422 => AnthropicFailureClass::Unprocessable,
429 => AnthropicFailureClass::RateLimited,
500..=599 => AnthropicFailureClass::UpstreamServer,
_ => AnthropicFailureClass::UnexpectedStatus,
};
return Err(anthropic_failure(class, Some(status), retry_after));
}
let media_type = content_type
.and_then(|value| value.split(';').next())
.map(str::trim);
if media_type != Some("text/event-stream") {
return Err(anthropic_failure(
AnthropicFailureClass::MismatchedContentType,
Some(status),
None,
));
}
Ok(())
}
pub fn normalize_transport_failure(
signal: AnthropicTransportFailureSignal,
body_write: AnthropicBodyWriteState,
) -> AnthropicFailure {
let class = match body_write {
AnthropicBodyWriteState::Written | AnthropicBodyWriteState::Unknown => {
AnthropicFailureClass::AmbiguousDispatch
}
AnthropicBodyWriteState::NotWritten => match signal {
AnthropicTransportFailureSignal::Cancelled => AnthropicFailureClass::Cancelled,
AnthropicTransportFailureSignal::Timeout => AnthropicFailureClass::TransportTimeout,
},
};
anthropic_failure(class, None, None)
}
fn anthropic_failure(
class: AnthropicFailureClass,
status: Option<u16>,
retry_after: Option<&str>,
) -> AnthropicFailure {
AnthropicFailure {
class,
status,
retry_after: crate::ollama_chat::BoundedRetryAfter::parse(retry_after),
}
}
#[cfg(test)]
mod tests {
use super::*;
use router_protocol::{
BillingAuthorityClass, CapabilityRequirements, CatalogSourceKind, CredentialSourceKind,
EndpointIdentity, ExecutionLocation, FunctionTool, HarnessInstanceId, M1_WIRE_VERSION,
MessageRole, ModelNamespace, PhaseKind, PhasePlacement, ProviderService, ReasoningIntent,
SessionId, TurnId,
};
fn claude_route(phase: PhaseKind, billing: BillingAuthorityClass) -> RouteIdentity {
RouteIdentity {
phase,
placement: PhasePlacement::Builtin,
provider: ProviderService::Claude,
transport: TransportKind::DirectHttps,
endpoint: EndpointIdentity {
origin: router_protocol::ANTHROPIC_ORIGIN.to_owned(),
path: router_protocol::ANTHROPIC_CHAT_PATH.to_owned(),
},
credential_source: CredentialSourceKind::SecretRef,
credential_ref: Some("env:STEADYMUX_CLAUDE_KEY".to_owned()),
credential_refs: vec!["env:STEADYMUX_CLAUDE_KEY".to_owned()],
catalog_source: CatalogSourceKind::ClaudeCatalog,
catalog_ref: "catalog:claude:verified".to_owned(),
billing_authority: billing,
model_namespace: ModelNamespace::Claude,
model_id: "claude-sonnet-4-5".to_owned(),
execution_location: ExecutionLocation::RemoteCloud,
account_ref: "account:claude".to_owned(),
capacity_domain_ref: "capacity:claude".to_owned(),
local_execution_proof: false,
approved_local_inventory: Vec::new(),
reasoning_modes: Vec::new(),
}
}
fn request(prompt: &str) -> CanonicalRequest {
CanonicalRequest {
wire_version: M1_WIRE_VERSION.to_owned(),
model: "router-auto".to_owned(),
stream: true,
harness_instance: HarnessInstanceId::parse("fixture-cli").expect("fixture"),
session_id: SessionId::parse("session-claude").expect("fixture"),
turn_id: TurnId::parse("turn-claude").expect("fixture"),
input: vec![InputItem::Message {
role: MessageRole::User,
content: vec![ContentBlock::Text {
text: prompt.to_owned(),
}],
}],
tools: vec![FunctionTool {
namespace: Some("weather".to_owned()),
name: "lookup".to_owned(),
description: "lookup weather".to_owned(),
parameters: json!({"type":"object"}),
strict: true,
}],
tool_choice: ToolChoice::Auto,
structured_output: None,
reasoning: ReasoningIntent::None,
max_output_tokens: Some(64),
max_spend_micro_usd: None,
capabilities: CapabilityRequirements::default(),
}
}
#[test]
fn claude_routes_build_verified_anthropic_wire_shape() {
for (phase, billing) in [
(
PhaseKind::ClaudeSubscription,
BillingAuthorityClass::ClaudeSubscription,
),
(
PhaseKind::ClaudeApi,
BillingAuthorityClass::ClaudeMeteredApi,
),
] {
let route = claude_route(phase, billing);
route.validate().expect("valid claude route");
let spec = build_request(&route, &request("claude-canary")).expect("request");
assert_eq!(spec.method, HttpMethod::Post);
assert_eq!(spec.origin, router_protocol::ANTHROPIC_ORIGIN);
assert_eq!(spec.path, router_protocol::ANTHROPIC_CHAT_PATH);
assert_eq!(spec.accept, "text/event-stream");
assert_eq!(spec.version_header, "2023-06-01");
assert_eq!(spec.api_key_secret.as_str(), "env:STEADYMUX_CLAUDE_KEY");
assert!(spec.network.require_tls);
assert!(!spec.network.follow_redirects);
assert!(!spec.network.allow_proxy);
assert_eq!(spec.body()["model"], json!("claude-sonnet-4-5"));
assert_eq!(spec.body()["max_tokens"], json!(64));
assert_eq!(spec.body()["stream"], json!(true));
assert_eq!(
spec.body()["messages"][0]["content"],
json!("claude-canary")
);
assert_eq!(spec.body()["tools"][0]["name"], json!("weather__lookup"));
assert_eq!(
spec.body()["tools"][0]["input_schema"],
json!({"type": "object"})
);
assert_eq!(spec.body()["tool_choice"], json!("auto"));
}
}
#[test]
fn claude_adapter_has_no_hardcoded_model_id_and_exposes_projection() {
let adapter_core = include_str!("anthropic_chat.rs")
.split("#[cfg(test)]")
.next()
.expect("adapter body precedes the test module");
for model_literal in [
concat!("claude", "-sonnet"),
concat!("claude", "-opus"),
concat!("claude", "-haiku"),
] {
assert!(
!adapter_core.contains(model_literal),
"adapter hardcodes the model literal {model_literal}"
);
}
assert!(is_claude_phase(PhaseKind::ClaudeSubscription));
assert!(is_claude_phase(PhaseKind::ClaudeApi));
assert!(!is_claude_phase(PhaseKind::DeepseekApi));
assert!(!capability_projection().namespace_tools);
assert_eq!(
PhaseKind::ClaudeApi.multi_agent_v2_projection(),
capability_projection()
);
}
#[test]
fn anthropic_sse_parses_text_finish_and_usage_once() {
let transcript = concat!(
"event: message_start\n",
"data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"usage\":{\"input_tokens\":12,\"output_tokens\":0}}}\n\n",
"event: content_block_delta\n",
"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"hello \"}}\n\n",
"event: content_block_delta\n",
"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"world\"}}\n\n",
"event: message_delta\n",
"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":7}}\n\n",
"event: message_stop\n",
"data: {\"type\":\"message_stop\"}\n\n",
);
let attempt = AttemptId::parse("attempt-anthropic-1").expect("attempt");
let mut parser = AnthropicSseParser::new(&attempt);
let events = parser
.feed(transcript.as_bytes())
.expect("feed anthropic stream");
assert_eq!(
events,
vec![
AnthropicStreamEvent::ContentDelta {
text: "hello ".to_owned()
},
AnthropicStreamEvent::ContentDelta {
text: "world".to_owned()
},
AnthropicStreamEvent::FinishReason {
reason: "end_turn".to_owned()
},
AnthropicStreamEvent::Usage(AnthropicUsage {
input_tokens: 12,
output_tokens: 7,
}),
AnthropicStreamEvent::Done,
]
);
assert_eq!(parser.finish(), Ok(vec![]));
assert!(parser.replay_decision() != ReplayDecision::Allowed);
}
#[test]
fn anthropic_stream_requires_terminal() {
let attempt = AttemptId::parse("attempt-anthropic-2").expect("attempt");
let mut parser = AnthropicSseParser::new(&attempt);
parser
.feed(
b"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"partial\"}}\n\n",
)
.expect("feed");
assert_eq!(
parser.finish(),
Err(AnthropicStreamError {
kind: AnthropicStreamErrorKind::MissingTerminal
})
);
}
}