bamboo_domain/
llm_request_diagnostic.rs

1//! A bounded failure receipt, carried by the existing Session checkpoint.
2//! These observations never authorize execution, retries or recovery.
3
4use chrono::{DateTime, Utc};
5use serde::{Deserialize, Deserializer, Serialize};
6use sha2::{Digest, Sha256};
7use uuid::Uuid;
8
9use crate::Session;
10
11pub const LAST_RUN_LLM_DIAGNOSTIC_KEY: &str = "runtime.llm_request_diagnostic";
12const MAX_DIAGNOSTIC_BYTES: usize = 4096;
13const DIAGNOSTIC_RUN_FINGERPRINT_KEY: &str = "runtime.llm_request_diagnostic_run";
14
15#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
16#[serde(rename_all = "snake_case")]
17pub enum LlmRequestStopReason {
18    AttemptsExhausted,
19    TimeBudgetExhausted,
20    RetryDelayExceedsBudget,
21    NonRetryableTransport,
22    UnclassifiedErrorBody,
23    CancelledOrDropped,
24}
25
26impl LlmRequestStopReason {
27    pub fn guidance(self) -> &'static str {
28        match self {
29            Self::AttemptsExhausted => "Automatic retry attempt limit reached",
30            Self::TimeBudgetExhausted => "Automatic retry time budget exhausted",
31            Self::RetryDelayExceedsBudget => {
32                "Required retry delay exceeds the remaining time budget"
33            }
34            Self::NonRetryableTransport => "Transport failure is not automatically retryable",
35            Self::UnclassifiedErrorBody => "Upstream error response could not be safely classified",
36            Self::CancelledOrDropped => "Request cancelled or its caller stopped waiting",
37        }
38    }
39}
40
41// This private wire shape deliberately matches the provider's stop record.
42// Validation applies on every deserialize, including cold canonical reads.
43#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
44#[serde(deny_unknown_fields)]
45struct DiagnosticFields {
46    schema_version: u8,
47    diagnostic_id: Uuid,
48    request_id: Uuid,
49    error_id: Option<Uuid>,
50    observed_at: DateTime<Utc>,
51    provider: String,
52    session_hash: Option<String>,
53    run_id: Option<Uuid>,
54    attempt: u32,
55    attempt_limit: u32,
56    event: String,
57    phase: String,
58    failure: String,
59    status: Option<u16>,
60    attempt_timeout_ms: u64,
61    total_budget_ms: u64,
62    attempt_elapsed_ms: u64,
63    total_elapsed_ms: u64,
64    remaining_budget_ms: u64,
65    retry_delay_ms: Option<u64>,
66    stop_reason: LlmRequestStopReason,
67    upstream_outcome: String,
68}
69
70/// A validated, content-free stop record. Unknown fields and free-text values
71/// fail closed instead of being echoed from editable Session metadata.
72#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
73#[serde(transparent)]
74pub struct LlmRequestDiagnostic(DiagnosticFields);
75
76impl<'de> Deserialize<'de> for LlmRequestDiagnostic {
77    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
78        let fields = DiagnosticFields::deserialize(deserializer)?;
79        if !fields.valid() {
80            return Err(serde::de::Error::custom("invalid LLM failure diagnostic"));
81        }
82        Ok(Self(fields))
83    }
84}
85
86impl DiagnosticFields {
87    fn valid(&self) -> bool {
88        self.schema_version == 1
89            && self.provider == "anthropic"
90            && self.event == "stop"
91            && self.request_id == self.diagnostic_id
92            && self.error_id == Some(self.diagnostic_id)
93            && self.session_hash.as_deref().is_some_and(|hash| {
94                hash.len() == 16
95                    && hash
96                        .bytes()
97                        .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
98            })
99            && (1..=4).contains(&self.attempt_limit)
100            && self.attempt <= self.attempt_limit
101            && self
102                .status
103                .is_none_or(|status| (100..=599).contains(&status))
104            && self.attempt_timeout_ms <= 60_000
105            && self.total_budget_ms <= 60_000
106            && self.remaining_budget_ms <= self.total_budget_ms
107            && ["send_headers", "error_body", "response_status", "backoff"]
108                .contains(&self.phase.as_str())
109            && ["unknown", "response_received"].contains(&self.upstream_outcome.as_str())
110            && [
111                "none",
112                "request deadline exceeded",
113                "request attempt limit reached",
114                "initial response headers timed out",
115                "timeout",
116                "connect",
117                "request",
118                "body",
119                "decode",
120                "status",
121                "transport",
122                "upstream error body deadline exceeded",
123                "upstream error body read failed",
124                "connector database is temporarily recovering",
125                "temporary upstream failure",
126            ]
127            .contains(&self.failure.as_str())
128    }
129}
130
131/// Same domain-separated identity as provider operational logs; never store
132/// caller-controlled session text in the diagnostic payload.
133pub fn llm_diagnostic_session_hash(session_id: &str) -> String {
134    let mut digest = Sha256::new();
135    digest.update(b"bamboo-provider-operational-log-v1\0");
136    digest.update(session_id.as_bytes());
137    hex::encode(&digest.finalize()[..8])
138}
139
140impl LlmRequestDiagnostic {
141    pub fn from_json(raw: &str) -> Option<Self> {
142        (raw.len() <= MAX_DIAGNOSTIC_BYTES)
143            .then(|| serde_json::from_str(raw).ok())
144            .flatten()
145    }
146
147    pub fn diagnostic_id(&self) -> Uuid {
148        self.0.diagnostic_id
149    }
150
151    pub fn matches_session(&self, session_id: &str) -> bool {
152        self.0.session_hash.as_deref() == Some(llm_diagnostic_session_hash(session_id).as_str())
153    }
154
155    /// Exact generated error text only. An arbitrary upstream error containing
156    /// a copied UUID cannot acquire a receipt through substring matching.
157    pub fn matches_error_message(&self, message: &str) -> bool {
158        message.strip_prefix("LLM error: ").unwrap_or(message) == self.error_message()
159    }
160
161    pub fn error_message(&self) -> String {
162        let f = &self.0;
163        let status = f
164            .status
165            .map(|s| format!("HTTP {s}"))
166            .unwrap_or_else(|| "no HTTP response".into());
167        format!(
168            "Anthropic upstream request failed ({status}): {}. {} after {} request(s) (limit {}). Diagnostic ID: {}. Try again later or contact the upstream gateway administrator.",
169            f.failure, f.stop_reason.guidance(), f.attempt, f.attempt_limit, f.diagnostic_id,
170        )
171    }
172}
173
174impl Session {
175    fn llm_diagnostic_run_fingerprint(&self) -> Option<String> {
176        self.agent_runtime_state.as_ref().map(|state| {
177            let mut digest = Sha256::new();
178            digest.update(b"bamboo-llm-diagnostic-run-v1\0");
179            digest.update((state.run_id.len() as u64).to_be_bytes());
180            digest.update(state.run_id.as_bytes());
181            if let Some(execution) = &state.execution_id {
182                digest.update([1]);
183                digest.update((execution.len() as u64).to_be_bytes());
184                digest.update(execution.as_bytes());
185            } else {
186                digest.update([0]);
187            }
188            hex::encode(&digest.finalize()[..16])
189        })
190    }
191
192    pub(crate) fn clear_llm_request_diagnostic(&mut self) {
193        self.metadata.remove(LAST_RUN_LLM_DIAGNOSTIC_KEY);
194        self.metadata.remove(DIAGNOSTIC_RUN_FINGERPRINT_KEY);
195    }
196
197    fn stored_llm_request_diagnostic(&self) -> Option<LlmRequestDiagnostic> {
198        let diagnostic =
199            LlmRequestDiagnostic::from_json(self.metadata.get(LAST_RUN_LLM_DIAGNOSTIC_KEY)?)?;
200        let run_id = self
201            .agent_runtime_state
202            .as_ref()
203            .and_then(|state| Uuid::parse_str(&state.run_id).ok());
204        let expected_run = self.llm_diagnostic_run_fingerprint();
205        (diagnostic.matches_session(&self.id)
206            && diagnostic.0.run_id == run_id
207            && self.metadata.get(DIAGNOSTIC_RUN_FINGERPRINT_KEY) == expected_run.as_ref())
208        .then_some(diagnostic)
209    }
210
211    /// Capture only the observer for this actual provider call; existing final
212    /// Session persistence carries it without a second store or background job.
213    pub fn record_llm_request_diagnostic(&mut self, mut diagnostic: LlmRequestDiagnostic) -> bool {
214        if !diagnostic.matches_session(&self.id) {
215            return false;
216        }
217        diagnostic.0.run_id = self
218            .agent_runtime_state
219            .as_ref()
220            .and_then(|state| Uuid::parse_str(&state.run_id).ok());
221        let Ok(raw) = serde_json::to_string(&diagnostic) else {
222            return false;
223        };
224        if raw.len() > MAX_DIAGNOSTIC_BYTES {
225            return false;
226        }
227        self.metadata
228            .insert(LAST_RUN_LLM_DIAGNOSTIC_KEY.into(), raw);
229        match self.llm_diagnostic_run_fingerprint() {
230            Some(fingerprint) => {
231                self.metadata
232                    .insert(DIAGNOSTIC_RUN_FINGERPRINT_KEY.into(), fingerprint);
233            }
234            None => {
235                self.metadata.remove(DIAGNOSTIC_RUN_FINGERPRINT_KEY);
236            }
237        }
238        true
239    }
240
241    pub(crate) fn retain_llm_request_diagnostic_for_error(&mut self, error: &str) {
242        if !self
243            .stored_llm_request_diagnostic()
244            .is_some_and(|d| d.matches_error_message(error))
245        {
246            self.clear_llm_request_diagnostic();
247        }
248    }
249
250    /// Public/cold-read projection requires the same Session, run and exact
251    /// canonical error. Old, foreign, malformed or overwritten evidence is absent.
252    pub fn last_run_diagnostic(&self) -> Option<LlmRequestDiagnostic> {
253        let error = self.last_run_error()?;
254        self.stored_llm_request_diagnostic()
255            .filter(|d| d.matches_error_message(&error))
256    }
257}
258
259#[cfg(test)]
260mod tests {
261    use super::*;
262
263    fn sample(session: &Session) -> LlmRequestDiagnostic {
264        let id = Uuid::new_v4();
265        LlmRequestDiagnostic::from_json(
266            &serde_json::json!({
267                "schema_version":1,"diagnostic_id":id,"request_id":id,"error_id":id,
268                "observed_at":Utc::now(),"provider":"anthropic",
269                "session_hash":llm_diagnostic_session_hash(&session.id),"run_id":null,
270                "attempt":1,"attempt_limit":4,"event":"stop","phase":"send_headers",
271                "failure":"initial response headers timed out","status":null,
272                "attempt_timeout_ms":20000,"total_budget_ms":60000,"attempt_elapsed_ms":20000,
273                "total_elapsed_ms":60000,"remaining_budget_ms":0,"retry_delay_ms":null,
274                "stop_reason":"time_budget_exhausted","upstream_outcome":"unknown",
275            })
276            .to_string(),
277        )
278        .unwrap()
279    }
280
281    #[test]
282    fn failure_receipt_survives_cold_session_round_trip_and_is_bound_to_error_and_session() {
283        let mut session = Session::new("diagnostic-current", "model");
284        let diagnostic = sample(&session);
285        assert!(session.record_llm_request_diagnostic(diagnostic.clone()));
286        assert!(session.last_run_diagnostic().is_none());
287        session.set_last_run_error(format!("LLM error: {}", diagnostic.error_message()));
288        let cold: Session =
289            serde_json::from_str(&serde_json::to_string(&session).unwrap()).unwrap();
290        assert_eq!(cold.last_run_diagnostic(), Some(diagnostic.clone()));
291        let mut foreign = cold.clone();
292        foreign.id = "other".into();
293        assert!(foreign.last_run_diagnostic().is_none());
294        let mut changed = cold.clone();
295        changed.set_last_run_error("another provider failure");
296        assert!(changed.last_run_diagnostic().is_none());
297        assert!(!changed.metadata.contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
298        let mut cleared = cold;
299        cleared.clear_last_run_error();
300        assert!(!cleared.metadata.contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
301    }
302
303    #[test]
304    fn invalid_free_text_unknown_fields_and_stale_run_fail_closed() {
305        let mut session = Session::new("diagnostic-current", "model");
306        let diagnostic = sample(&session);
307        let value = serde_json::to_value(&diagnostic).unwrap();
308        for field in [
309            "provider",
310            "phase",
311            "failure",
312            "upstream_outcome",
313            "session_hash",
314            "future_secret",
315        ] {
316            let mut invalid = value.clone();
317            invalid[field] = serde_json::json!("PRIVATE_CANARY_body_key_path");
318            assert!(LlmRequestDiagnostic::from_json(&invalid.to_string()).is_none());
319        }
320        assert!(!diagnostic.matches_error_message(&format!(
321            "arbitrary upstream body: {}",
322            diagnostic.error_message()
323        )));
324        session.record_llm_request_diagnostic(diagnostic.clone());
325        session.set_last_run_error(diagnostic.error_message());
326        session.agent_runtime_state =
327            Some(crate::AgentRuntimeState::new(Uuid::new_v4().to_string()));
328        assert!(session.last_run_diagnostic().is_none());
329    }
330
331    #[test]
332    fn non_uuid_run_and_execution_replacement_cannot_reuse_the_old_receipt() {
333        let mut session = Session::new("diagnostic-current", "model");
334        let mut runtime = crate::AgentRuntimeState::new("private-non-uuid-run-a");
335        runtime.execution_id = Some("private-execution-a".into());
336        session.agent_runtime_state = Some(runtime);
337        let diagnostic = sample(&session);
338        session.record_llm_request_diagnostic(diagnostic.clone());
339        session.set_last_run_error(diagnostic.error_message());
340        let encoded = serde_json::to_string(&session).unwrap();
341        let mut cold: Session = serde_json::from_str(&encoded).unwrap();
342        assert!(cold.last_run_diagnostic().is_some());
343        let public = serde_json::to_string(&cold.last_run_diagnostic().unwrap()).unwrap();
344        assert!(!public.contains("private-non-uuid-run-a"));
345        assert!(!public.contains("private-execution-a"));
346        cold.agent_runtime_state.as_mut().unwrap().run_id = "private-non-uuid-run-b".into();
347        assert!(cold.last_run_diagnostic().is_none());
348        let mut cold: Session = serde_json::from_str(&encoded).unwrap();
349        cold.agent_runtime_state.as_mut().unwrap().execution_id =
350            Some("private-execution-b".into());
351        assert!(cold.last_run_diagnostic().is_none());
352    }
353}
bamboo_domain/llm_request_diagnostic.rs

bamboo_domain/
llm_request_diagnostic.rs
