agent_engine/runtime/trace/
anthropic.rs

1//! Metadata-only structural description of an Anthropic Messages request
2//! (Task 8). Every function here consumes trusted structural inputs — the
3//! already-sanitized message array, the caller-supplied system prompt, the
4//! registered tool schemas, and the exact serialized wire bytes — and
5//! produces only counts, byte lengths, enums, validated IDs, and keyed
6//! digests. No raw string is ever retained.
7
8use super::emit::{wire_meta_from_sent_bytes, RequestStructure};
9use super::key::{keyed_digest, DigestDomain, TraceDigestKey};
10use super::types::{
11    BlockKind, BlockMeta, CacheBoundaryLocation, CacheBoundaryMeta, CacheMeta, CacheTtlClass,
12    MessageMeta, MessageRole, RequestAnatomy, RetryClass, StopReason, SystemSegmentKind,
13    SystemSegmentMeta, ToolMeta, TraceId, WireName,
14};
15use serde_json::Value;
16
17/// Map an Anthropic HTTP failure status to a coarse retry class.
18pub fn retry_class_for_status(status: u16) -> RetryClass {
19    match status {
20        429 => RetryClass::RateLimited,
21        529 => RetryClass::Overloaded,
22        500 | 502 | 503 => RetryClass::ServerError,
23        401 | 403 => RetryClass::Auth,
24        408 => RetryClass::Timeout,
25        _ => RetryClass::Other,
26    }
27}
28
29/// Map an in-stream Anthropic error type (already vetted upstream) to a
30/// coarse retry class. `None` means the stream died at transport level.
31pub fn retry_class_for_stream_error(error_type: Option<&str>) -> RetryClass {
32    match error_type {
33        Some("overloaded_error") => RetryClass::Overloaded,
34        Some("rate_limit_error") => RetryClass::RateLimited,
35        Some("api_error") => RetryClass::ServerError,
36        Some(_) => RetryClass::Other,
37        None => RetryClass::Network,
38    }
39}
40
41/// Normalize an Anthropic wire `stop_reason` into the trace enum. Unknown
42/// values collapse to `Other` — the raw string is never stored.
43pub fn stop_reason_from_wire(raw: &str) -> StopReason {
44    match raw {
45        "end_turn" => StopReason::EndTurn,
46        "max_tokens" => StopReason::MaxTokens,
47        "stop_sequence" => StopReason::StopSequence,
48        "tool_use" => StopReason::ToolUse,
49        "pause_turn" => StopReason::PauseTurn,
50        "refusal" => StopReason::Refusal,
51        _ => StopReason::Other,
52    }
53}
54
55/// Build the structural description of one Anthropic request.
56///
57/// `sent_bytes` MUST be the exact buffer handed to reqwest — wire length and
58/// digest come from it directly, never from re-serialization. When `key` is
59/// `None` (digest key unavailable) all digest-bearing sections (`wire`,
60/// `system_segments`, `tools`, cache prefixes/delta) are omitted; counts in
61/// `anatomy` remain. `snapshots`, when provided, is the session's
62/// previous-component snapshot store (Task 12, spec §6.6): the cache prefix
63/// digests and per-segment previous-turn delta are computed and the
64/// snapshot atomically advanced, once per built request structure.
65/// `translation` is the Anthropic adapter's Task 9 report entries
66/// (positional IDs only, never content).
67#[allow(clippy::too_many_arguments)]
68pub fn anthropic_request_structure(
69    key: Option<&TraceDigestKey>,
70    sent_bytes: &[u8],
71    messages: &[crate::SharedMessage],
72    system_prompt: Option<&str>,
73    tools_schema: &[Value],
74    prefix_marker_ttl: Option<CacheTtlClass>,
75    has_tool_marker: bool,
76    has_system_marker: bool,
77    translation: Vec<super::types::TranslationLoss>,
78    snapshots: Option<&super::diagnostics::CacheSnapshotStore>,
79) -> RequestStructure {
80    let message_meta: Vec<MessageMeta> = messages.iter().map(|m| message_meta(m)).collect();
81    let block_count: u32 = message_meta.iter().map(|m| m.blocks.len() as u32).sum();
82
83    let system_segment_count = u32::from(system_prompt.is_some_and(|s| !s.is_empty()));
84    let anatomy = RequestAnatomy {
85        system_segment_count,
86        message_count: messages.len() as u32,
87        block_count,
88        tool_count: tools_schema.len() as u32,
89    };
90
91    let system_segments = match (key, system_prompt) {
92        (Some(key), Some(s)) if !s.is_empty() => vec![SystemSegmentMeta {
93            kind: SystemSegmentKind::Primary,
94            byte_len: s.len() as u64,
95            digest: keyed_digest(key, DigestDomain::SystemSegment, s.as_bytes()),
96        }],
97        _ => Vec::new(),
98    };
99
100    let tools = key
101        .map(|key| {
102            tools_schema
103                .iter()
104                .filter_map(|t| tool_meta(key, t))
105                .collect()
106        })
107        .unwrap_or_default();
108
109    let mut cache = cache_meta(
110        messages,
111        tools_schema.len(),
112        prefix_marker_ttl,
113        has_tool_marker,
114        has_system_marker,
115    );
116    // Task 12 (spec §6.6): keyed prefix/tail digests over canonical
117    // component bytes + previous-turn segment delta from the session
118    // snapshot store. Degrades to `None`s without a key or store.
119    if let Some(snapshots) = snapshots {
120        let diag = snapshots.compare_and_update(key, tools_schema, system_prompt, messages);
121        cache.tools_prefix = diag.tools_prefix;
122        cache.system_prefix = diag.system_prefix;
123        cache.history_tail = diag.history_tail;
124        cache.delta = diag.delta;
125    }
126
127    RequestStructure {
128        anatomy,
129        wire: key.map(|key| wire_meta_from_sent_bytes(key, sent_bytes)),
130        system_segments,
131        messages: message_meta,
132        tools,
133        cache,
134        translation,
135    }
136}
137
138fn message_meta(message: &Value) -> MessageMeta {
139    let role = match message["role"].as_str() {
140        Some("assistant") => MessageRole::Assistant,
141        Some("system") => MessageRole::System,
142        Some("tool") => MessageRole::Tool,
143        _ => MessageRole::User,
144    };
145    let blocks = match &message["content"] {
146        Value::String(s) => vec![BlockMeta {
147            kind: BlockKind::Text,
148            byte_len: s.len() as u64,
149        }],
150        Value::Array(items) => items.iter().map(block_meta).collect(),
151        other => vec![BlockMeta {
152            kind: BlockKind::Unknown,
153            byte_len: serialized_len(other),
154        }],
155    };
156    MessageMeta { role, blocks }
157}
158
159fn block_meta(block: &Value) -> BlockMeta {
160    let kind = match block["type"].as_str() {
161        Some("text") => BlockKind::Text,
162        Some("thinking") | Some("redacted_thinking") => BlockKind::Thinking,
163        Some("tool_use") => BlockKind::ToolUse,
164        Some("tool_result") => BlockKind::ToolResult,
165        Some("image") | Some("document") => BlockKind::Media,
166        _ => BlockKind::Unknown,
167    };
168    BlockMeta {
169        kind,
170        byte_len: serialized_len(block),
171    }
172}
173
174fn serialized_len(value: &Value) -> u64 {
175    serde_json::to_vec(value)
176        .map(|v| v.len() as u64)
177        .unwrap_or(0)
178}
179
180fn tool_meta(key: &TraceDigestKey, tool: &Value) -> Option<ToolMeta> {
181    let name = tool["name"].as_str()?;
182    // Validated identifiers: a tool whose name does not fit the safe
183    // grammar is omitted from the digest section (still counted in
184    // anatomy) — the raw name is never copied into the record.
185    let wire_name = WireName::new(name).ok()?;
186    let stable_id = TraceId::new(name).ok()?;
187    let schema_bytes = serde_json::to_vec(&tool["input_schema"]).ok()?;
188    Some(ToolMeta {
189        stable_id,
190        wire_name,
191        schema_byte_len: schema_bytes.len() as u64,
192        schema_digest: keyed_digest(key, DigestDomain::ToolSchema, &schema_bytes),
193    })
194}
195
196/// Cache boundaries: message-tail markers are read from the annotated
197/// message array (structural, trusted — the annotation this process applied);
198/// tools/system prefix markers are reported from the request-builder's own
199/// marker flags with the configured prefix TTL class.
200fn cache_meta(
201    messages: &[crate::SharedMessage],
202    tool_count: usize,
203    prefix_marker_ttl: Option<CacheTtlClass>,
204    has_tool_marker: bool,
205    has_system_marker: bool,
206) -> CacheMeta {
207    let mut boundaries = Vec::new();
208    if let (Some(ttl), true) = (prefix_marker_ttl, has_tool_marker) {
209        boundaries.push(CacheBoundaryMeta {
210            location: CacheBoundaryLocation::Tools,
211            index: tool_count.saturating_sub(1) as u32,
212            ttl,
213        });
214    }
215    if let (Some(ttl), true) = (prefix_marker_ttl, has_system_marker) {
216        boundaries.push(CacheBoundaryMeta {
217            location: CacheBoundaryLocation::System,
218            index: 0,
219            ttl,
220        });
221    }
222    for (index, message) in messages.iter().enumerate() {
223        let marker = message["content"]
224            .as_array()
225            .and_then(|blocks| blocks.last())
226            .and_then(|b| b.get("cache_control"));
227        if let Some(marker) = marker {
228            let ttl = if marker.get("ttl").and_then(Value::as_str) == Some("1h") {
229                CacheTtlClass::OneHour
230            } else {
231                CacheTtlClass::FiveMinutes
232            };
233            boundaries.push(CacheBoundaryMeta {
234                location: CacheBoundaryLocation::Messages,
235                index: index as u32,
236                ttl,
237            });
238        }
239    }
240    CacheMeta {
241        boundaries,
242        tools_prefix: None,
243        system_prefix: None,
244        history_tail: None,
245        delta: None,
246    }
247}
agent_engine/runtime/trace/anthropic.rs

agent_engine/runtime/trace/
anthropic.rs
