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    RetryDelayTooLarge,
22    NonRetryableTransport,
23    UnclassifiedErrorBody,
24    CancelledOrDropped,
25}
26
27impl LlmRequestStopReason {
28    pub fn guidance(self) -> &'static str {
29        match self {
30            Self::AttemptsExhausted => "Automatic retry attempt limit reached",
31            Self::TimeBudgetExhausted => "Automatic retry time budget exhausted",
32            Self::RetryDelayExceedsBudget => {
33                "Required retry delay exceeds the remaining time budget"
34            }
35            Self::RetryDelayTooLarge => "Required retry delay is too large to schedule safely",
36            Self::NonRetryableTransport => "Transport failure is not automatically retryable",
37            Self::UnclassifiedErrorBody => "Upstream error response could not be safely classified",
38            Self::CancelledOrDropped => "Request cancelled or its caller stopped waiting",
39        }
40    }
41}
42
43// This private wire shape deliberately matches the provider's stop record.
44// Validation applies on every deserialize, including cold canonical reads.
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(deny_unknown_fields)]
47struct InitialRequestDiagnosticFields {
48    schema_version: u8,
49    diagnostic_id: Uuid,
50    request_id: Uuid,
51    error_id: Option<Uuid>,
52    observed_at: DateTime<Utc>,
53    provider: String,
54    session_hash: Option<String>,
55    run_id: Option<Uuid>,
56    attempt: u32,
57    attempt_limit: Option<u32>,
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    retry_mode: Option<String>,
60    event: String,
61    phase: String,
62    failure: String,
63    status: Option<u16>,
64    attempt_timeout_ms: u64,
65    total_budget_ms: Option<u64>,
66    attempt_elapsed_ms: u64,
67    total_elapsed_ms: u64,
68    remaining_budget_ms: Option<u64>,
69    retry_delay_ms: Option<u64>,
70    stop_reason: LlmRequestStopReason,
71    upstream_outcome: String,
72}
73
74/// Public watchdog evidence only. The validated wrapper below is required
75/// before these fields may be persisted or attached to a client error event.
76#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
77#[serde(deny_unknown_fields)]
78pub struct StreamTimeoutDiagnostic {
79    pub schema_version: u8,
80    pub kind: String,
81    pub diagnostic_id: Uuid,
82    pub error_id: Uuid,
83    pub observed_at: DateTime<Utc>,
84    pub provider: Option<String>,
85    pub model: Option<String>,
86    pub session_hash: String,
87    pub run_id: Option<Uuid>,
88    pub phase: String,
89    pub deadline_ms: u64,
90    pub last_transport_ms_ago: u64,
91    pub last_semantic_ms_ago: Option<u64>,
92    pub semantic_output_started: bool,
93    pub retry_safe: bool,
94    pub last_http_status: Option<u16>,
95    pub retry_delay_ms: Option<u64>,
96}
97
98impl StreamTimeoutDiagnostic {
99    /// Identifiers are bounded labels, never URLs or credential-like values.
100    /// This same check applies before formatting the timeout message and on
101    /// cold metadata reads, so redaction cannot break exact error binding.
102    pub fn valid_identifier(value: &str) -> bool {
103        !value.is_empty()
104            && value.len() <= 120
105            && value
106                .bytes()
107                .all(|b| b.is_ascii_alphanumeric() || b"-_.:/@".contains(&b))
108            && !value.contains("://")
109            && !value.to_ascii_lowercase().starts_with("sk-")
110            && !value.to_ascii_lowercase().starts_with("bearer")
111    }
112
113    fn valid(&self) -> bool {
114        // JavaScript consumers must not round a correlation-relevant duration.
115        const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
116        let has_semantic = self.last_semantic_ms_ago.is_some();
117        self.schema_version == 1
118            && self.kind == "stream_timeout"
119            && self.error_id == self.diagnostic_id
120            && valid_session_hash(&self.session_hash)
121            && self.provider.as_deref().is_none_or(Self::valid_identifier)
122            && self.model.as_deref().is_none_or(Self::valid_identifier)
123            && (1_000..=86_400_000).contains(&self.deadline_ms)
124            && self.last_transport_ms_ago <= MAX_SAFE_INTEGER
125            && self.last_semantic_ms_ago.is_none_or(|elapsed| {
126                elapsed <= MAX_SAFE_INTEGER && self.last_transport_ms_ago <= elapsed
127            })
128            && self.semantic_output_started == has_semantic
129            && (!self.retry_safe || !has_semantic)
130            && match self.phase.as_str() {
131                "bootstrap" | "first_semantic" => !has_semantic,
132                "transport_idle" => true,
133                "semantic_idle" => has_semantic,
134                _ => false,
135            }
136            && match (self.last_http_status, self.retry_delay_ms) {
137                (None, None) => true,
138                (Some(status), Some(delay)) => {
139                    self.phase == "bootstrap"
140                        && (100..=599).contains(&status)
141                        && delay <= MAX_SAFE_INTEGER
142                }
143                _ => false,
144            }
145    }
146
147    fn error_message(&self) -> String {
148        let last_semantic = self
149            .last_semantic_ms_ago
150            .map(|elapsed| format!("{elapsed}ms"))
151            .unwrap_or_else(|| "never".into());
152        let mut message = format!(
153            "Stream timed out: phase={}, deadline_ms={}, provider={}, model={}, last_transport_ms_ago={}, \
154             last_semantic_ms_ago={}, semantic_output_started={}, retry_safe={}",
155            self.phase,
156            self.deadline_ms,
157            self.provider.as_deref().unwrap_or("unknown"),
158            self.model.as_deref().unwrap_or("unknown"),
159            self.last_transport_ms_ago,
160            last_semantic,
161            self.semantic_output_started,
162            self.retry_safe,
163        );
164        if let (Some(status), Some(delay)) = (self.last_http_status, self.retry_delay_ms) {
165            use std::fmt::Write;
166            let _ = write!(
167                message,
168                ", last_http_status={status}, retry_delay_ms={delay}"
169            );
170        }
171        format!("{message}. Diagnostic ID: {}.", self.diagnostic_id)
172    }
173}
174
175#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
176#[serde(untagged)]
177enum DiagnosticFields {
178    InitialRequest(InitialRequestDiagnosticFields),
179    StreamTimeout(StreamTimeoutDiagnostic),
180}
181
182impl DiagnosticFields {
183    fn valid(&self) -> bool {
184        match self {
185            Self::InitialRequest(fields) => fields.valid(),
186            Self::StreamTimeout(fields) => fields.valid(),
187        }
188    }
189
190    fn run_id(&self) -> Option<Uuid> {
191        match self {
192            Self::InitialRequest(fields) => fields.run_id,
193            Self::StreamTimeout(fields) => fields.run_id,
194        }
195    }
196
197    fn set_run_id(&mut self, run_id: Option<Uuid>) {
198        match self {
199            Self::InitialRequest(fields) => fields.run_id = run_id,
200            Self::StreamTimeout(fields) => fields.run_id = run_id,
201        }
202    }
203}
204
205fn valid_session_hash(hash: &str) -> bool {
206    hash.len() == 16
207        && hash
208            .bytes()
209            .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
210}
211
212/// A validated, content-free stop record. Unknown fields and free-text values
213/// fail closed instead of being echoed from editable Session metadata.
214#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
215#[serde(transparent)]
216pub struct LlmRequestDiagnostic(DiagnosticFields);
217
218impl<'de> Deserialize<'de> for LlmRequestDiagnostic {
219    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
220        let fields = DiagnosticFields::deserialize(deserializer)?;
221        if !fields.valid() {
222            return Err(serde::de::Error::custom("invalid LLM failure diagnostic"));
223        }
224        Ok(Self(fields))
225    }
226}
227
228impl InitialRequestDiagnosticFields {
229    fn valid(&self) -> bool {
230        self.schema_version == 1
231            && self.provider == "anthropic"
232            && self.event == "stop"
233            && self.request_id == self.diagnostic_id
234            && self.error_id == Some(self.diagnostic_id)
235            && self.session_hash.as_deref().is_some_and(valid_session_hash)
236            && match self.retry_mode.as_deref() {
237                None => {
238                    self.attempt_limit
239                        .is_some_and(|limit| limit > 0 && self.attempt <= limit)
240                        && self
241                            .total_budget_ms
242                            .is_some_and(|budget| budget <= 3_600_000)
243                        && self
244                            .remaining_budget_ms
245                            .zip(self.total_budget_ms)
246                            .is_some_and(|(remaining, total)| remaining <= total)
247                }
248                Some("until_cancelled") => {
249                    self.attempt_limit.is_none()
250                        && self.total_budget_ms.is_none()
251                        && self.remaining_budget_ms.is_none()
252                        && !matches!(
253                            self.stop_reason,
254                            LlmRequestStopReason::AttemptsExhausted
255                                | LlmRequestStopReason::TimeBudgetExhausted
256                                | LlmRequestStopReason::RetryDelayExceedsBudget
257                        )
258                }
259                _ => false,
260            }
261            && self
262                .status
263                .is_none_or(|status| (100..=599).contains(&status))
264            // Ceiling, not a default: the provider policy may legitimately
265            // configure a long headers deadline for slow first-byte routes, and
266            // rejecting that receipt would silently drop the only evidence.
267            && self.attempt_timeout_ms <= 600_000
268            && ["send_headers", "error_body", "response_status", "backoff"]
269                .contains(&self.phase.as_str())
270            && ["unknown", "response_received"].contains(&self.upstream_outcome.as_str())
271            && [
272                "none",
273                "request deadline exceeded",
274                "request attempt limit reached",
275                "initial response headers timed out",
276                "timeout",
277                "connect",
278                "request",
279                "body",
280                "decode",
281                "status",
282                "transport",
283                "upstream error body deadline exceeded",
284                "upstream error body read failed",
285                "connector database is temporarily recovering",
286                "temporary upstream failure",
287            ]
288            .contains(&self.failure.as_str())
289    }
290}
291
292/// Same domain-separated identity as provider operational logs; never store
293/// caller-controlled session text in the diagnostic payload.
294pub fn llm_diagnostic_session_hash(session_id: &str) -> String {
295    let mut digest = Sha256::new();
296    digest.update(b"bamboo-provider-operational-log-v1\0");
297    digest.update(session_id.as_bytes());
298    hex::encode(&digest.finalize()[..8])
299}
300
301impl LlmRequestDiagnostic {
302    pub fn from_stream_timeout(fields: StreamTimeoutDiagnostic) -> Option<Self> {
303        fields
304            .valid()
305            .then_some(Self(DiagnosticFields::StreamTimeout(fields)))
306    }
307
308    pub fn from_json(raw: &str) -> Option<Self> {
309        (raw.len() <= MAX_DIAGNOSTIC_BYTES)
310            .then(|| serde_json::from_str(raw).ok())
311            .flatten()
312    }
313
314    pub fn diagnostic_id(&self) -> Uuid {
315        match &self.0 {
316            DiagnosticFields::InitialRequest(fields) => fields.diagnostic_id,
317            DiagnosticFields::StreamTimeout(fields) => fields.diagnostic_id,
318        }
319    }
320
321    pub fn matches_session(&self, session_id: &str) -> bool {
322        let hash = llm_diagnostic_session_hash(session_id);
323        match &self.0 {
324            DiagnosticFields::InitialRequest(fields) => {
325                fields.session_hash.as_deref() == Some(hash.as_str())
326            }
327            DiagnosticFields::StreamTimeout(fields) => fields.session_hash == hash,
328        }
329    }
330
331    /// Exact generated error text only. An arbitrary upstream error containing
332    /// a copied UUID cannot acquire a receipt through substring matching.
333    pub fn matches_error_message(&self, message: &str) -> bool {
334        let message = match &self.0 {
335            DiagnosticFields::InitialRequest(_) => {
336                message.strip_prefix("LLM error: ").unwrap_or(message)
337            }
338            DiagnosticFields::StreamTimeout(_) => message,
339        };
340        message == self.error_message()
341    }
342
343    pub fn error_message(&self) -> String {
344        let f = match &self.0 {
345            DiagnosticFields::InitialRequest(fields) => fields,
346            DiagnosticFields::StreamTimeout(fields) => return fields.error_message(),
347        };
348        let status = f
349            .status
350            .map(|s| format!("HTTP {s}"))
351            .unwrap_or_else(|| "no HTTP response".into());
352        let limit = f
353            .attempt_limit
354            .map(|n| format!(" (limit {n})"))
355            .unwrap_or_default();
356        format!(
357            "Anthropic upstream request failed ({status}): {}. {} after {} request(s){limit}. Diagnostic ID: {}. Try again later or contact the upstream gateway administrator.",
358            f.failure, f.stop_reason.guidance(), f.attempt, f.diagnostic_id,
359        )
360    }
361}
362
363impl Session {
364    fn llm_diagnostic_run_fingerprint(&self) -> Option<String> {
365        self.agent_runtime_state.as_ref().map(|state| {
366            let mut digest = Sha256::new();
367            digest.update(b"bamboo-llm-diagnostic-run-v1\0");
368            digest.update((state.run_id.len() as u64).to_be_bytes());
369            digest.update(state.run_id.as_bytes());
370            if let Some(execution) = &state.execution_id {
371                digest.update([1]);
372                digest.update((execution.len() as u64).to_be_bytes());
373                digest.update(execution.as_bytes());
374            } else {
375                digest.update([0]);
376            }
377            hex::encode(&digest.finalize()[..16])
378        })
379    }
380
381    pub(crate) fn clear_llm_request_diagnostic(&mut self) {
382        self.metadata.remove(LAST_RUN_LLM_DIAGNOSTIC_KEY);
383        self.metadata.remove(DIAGNOSTIC_RUN_FINGERPRINT_KEY);
384    }
385
386    fn stored_llm_request_diagnostic(&self) -> Option<LlmRequestDiagnostic> {
387        let diagnostic =
388            LlmRequestDiagnostic::from_json(self.metadata.get(LAST_RUN_LLM_DIAGNOSTIC_KEY)?)?;
389        let run_id = self
390            .agent_runtime_state
391            .as_ref()
392            .and_then(|state| Uuid::parse_str(&state.run_id).ok());
393        let expected_run = self.llm_diagnostic_run_fingerprint();
394        (diagnostic.matches_session(&self.id)
395            && diagnostic.0.run_id() == run_id
396            && self.metadata.get(DIAGNOSTIC_RUN_FINGERPRINT_KEY) == expected_run.as_ref())
397        .then_some(diagnostic)
398    }
399
400    /// Capture only the observer for this actual provider call; existing final
401    /// Session persistence carries it without a second store or background job.
402    pub fn record_llm_request_diagnostic(&mut self, mut diagnostic: LlmRequestDiagnostic) -> bool {
403        if !diagnostic.matches_session(&self.id) {
404            return false;
405        }
406        diagnostic.0.set_run_id(
407            self.agent_runtime_state
408                .as_ref()
409                .and_then(|state| Uuid::parse_str(&state.run_id).ok()),
410        );
411        let Ok(raw) = serde_json::to_string(&diagnostic) else {
412            return false;
413        };
414        if raw.len() > MAX_DIAGNOSTIC_BYTES {
415            return false;
416        }
417        self.metadata
418            .insert(LAST_RUN_LLM_DIAGNOSTIC_KEY.into(), raw);
419        match self.llm_diagnostic_run_fingerprint() {
420            Some(fingerprint) => {
421                self.metadata
422                    .insert(DIAGNOSTIC_RUN_FINGERPRINT_KEY.into(), fingerprint);
423            }
424            None => {
425                self.metadata.remove(DIAGNOSTIC_RUN_FINGERPRINT_KEY);
426            }
427        }
428        true
429    }
430
431    pub(crate) fn retain_llm_request_diagnostic_for_error(&mut self, error: &str) {
432        if !self
433            .stored_llm_request_diagnostic()
434            .is_some_and(|d| d.matches_error_message(error))
435        {
436            self.clear_llm_request_diagnostic();
437        }
438    }
439
440    /// Public/cold-read projection requires the same Session, run and exact
441    /// canonical error. Old, foreign, malformed or overwritten evidence is absent.
442    pub fn last_run_diagnostic(&self) -> Option<LlmRequestDiagnostic> {
443        let error = self.last_run_error()?;
444        self.stored_llm_request_diagnostic()
445            .filter(|d| d.matches_error_message(&error))
446    }
447}
448
449#[cfg(test)]
450mod tests {
451    use super::*;
452
453    fn timeout_sample(session: &Session) -> LlmRequestDiagnostic {
454        let id = Uuid::new_v4();
455        LlmRequestDiagnostic::from_stream_timeout(StreamTimeoutDiagnostic {
456            schema_version: 1,
457            kind: "stream_timeout".into(),
458            diagnostic_id: id,
459            error_id: id,
460            observed_at: Utc::now(),
461            provider: Some("a2a3cebf-8abc-45cb-8407-66c90fd9a978".into()),
462            model: Some("gpt-6-astra".into()),
463            session_hash: llm_diagnostic_session_hash(&session.id),
464            run_id: None,
465            phase: "transport_idle".into(),
466            deadline_ms: 120_000,
467            last_transport_ms_ago: 120_002,
468            last_semantic_ms_ago: Some(120_002),
469            semantic_output_started: true,
470            retry_safe: false,
471            last_http_status: None,
472            retry_delay_ms: None,
473        })
474        .unwrap()
475    }
476
477    fn sample(session: &Session) -> LlmRequestDiagnostic {
478        let id = Uuid::new_v4();
479        LlmRequestDiagnostic::from_json(
480            &serde_json::json!({
481                "schema_version":1,"diagnostic_id":id,"request_id":id,"error_id":id,
482                "observed_at":Utc::now(),"provider":"anthropic",
483                "session_hash":llm_diagnostic_session_hash(&session.id),"run_id":null,
484                "attempt":1,"attempt_limit":4,"event":"stop","phase":"send_headers",
485                "failure":"initial response headers timed out","status":null,
486                "attempt_timeout_ms":20000,"total_budget_ms":60000,"attempt_elapsed_ms":20000,
487                "total_elapsed_ms":60000,"remaining_budget_ms":0,"retry_delay_ms":null,
488                "stop_reason":"time_budget_exhausted","upstream_outcome":"unknown",
489            })
490            .to_string(),
491        )
492        .unwrap()
493    }
494
495    #[test]
496    fn failure_receipt_survives_cold_session_round_trip_and_is_bound_to_error_and_session() {
497        let mut session = Session::new("diagnostic-current", "model");
498        let diagnostic = sample(&session);
499        assert!(session.record_llm_request_diagnostic(diagnostic.clone()));
500        assert!(session.last_run_diagnostic().is_none());
501        session.set_last_run_error(format!("LLM error: {}", diagnostic.error_message()));
502        let cold: Session =
503            serde_json::from_str(&serde_json::to_string(&session).unwrap()).unwrap();
504        assert_eq!(cold.last_run_diagnostic(), Some(diagnostic.clone()));
505        let mut foreign = cold.clone();
506        foreign.id = "other".into();
507        assert!(foreign.last_run_diagnostic().is_none());
508        let mut changed = cold.clone();
509        changed.set_last_run_error("another provider failure");
510        assert!(changed.last_run_diagnostic().is_none());
511        assert!(!changed.metadata.contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
512        let mut cleared = cold;
513        cleared.clear_last_run_error();
514        assert!(!cleared.metadata.contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
515    }
516
517    /// A slow-route policy is a supported configuration: its receipt must stay
518    /// readable instead of being dropped by a ceiling that only matched the
519    /// old short default.
520    #[test]
521    fn receipt_accepts_a_long_configured_headers_deadline_up_to_its_ceiling() {
522        let session = Session::new("slow-route", "model");
523        let mut bounded = serde_json::to_value(sample(&session)).unwrap();
524        bounded["attempt_timeout_ms"] = 180_000.into();
525        bounded["total_budget_ms"] = 600_000.into();
526        bounded["attempt_elapsed_ms"] = 180_000.into();
527        bounded["total_elapsed_ms"] = 600_000.into();
528        bounded["remaining_budget_ms"] = 0.into();
529        assert!(LlmRequestDiagnostic::from_json(&bounded.to_string()).is_some());
530        bounded["attempt_timeout_ms"] = 600_000.into();
531        assert!(LlmRequestDiagnostic::from_json(&bounded.to_string()).is_some());
532        bounded["attempt_timeout_ms"] = 600_001.into();
533        assert!(LlmRequestDiagnostic::from_json(&bounded.to_string()).is_none());
534    }
535
536    #[test]
537    fn configured_and_continuous_receipts_keep_distinct_validated_limits() {
538        let session = Session::new("retry-diagnostic", "model");
539        let mut bounded = serde_json::to_value(sample(&session)).unwrap();
540        bounded["attempt_limit"] = 12.into();
541        bounded["total_budget_ms"] = 300_000.into();
542        assert!(LlmRequestDiagnostic::from_json(&bounded.to_string()).is_some());
543        let mut continuous = bounded.clone();
544        continuous["retry_mode"] = "until_cancelled".into();
545        for field in ["attempt_limit", "total_budget_ms", "remaining_budget_ms"] {
546            continuous[field] = serde_json::Value::Null;
547        }
548        continuous["stop_reason"] = "cancelled_or_dropped".into();
549        let diagnostic = LlmRequestDiagnostic::from_json(&continuous.to_string()).unwrap();
550        assert!(diagnostic.matches_error_message(&diagnostic.error_message()));
551        for field in ["attempt_limit", "total_budget_ms", "remaining_budget_ms"] {
552            let mut invalid = continuous.clone();
553            invalid[field] = 1.into();
554            assert!(LlmRequestDiagnostic::from_json(&invalid.to_string()).is_none());
555        }
556        continuous["retry_mode"] = "unrecognized".into();
557        assert!(LlmRequestDiagnostic::from_json(&continuous.to_string()).is_none());
558        bounded["total_budget_ms"] = 3_600_001.into();
559        assert!(LlmRequestDiagnostic::from_json(&bounded.to_string()).is_none());
560    }
561
562    #[test]
563    fn invalid_free_text_unknown_fields_and_stale_run_fail_closed() {
564        let mut session = Session::new("diagnostic-current", "model");
565        let diagnostic = sample(&session);
566        let value = serde_json::to_value(&diagnostic).unwrap();
567        for field in [
568            "provider",
569            "phase",
570            "failure",
571            "upstream_outcome",
572            "session_hash",
573            "future_secret",
574        ] {
575            let mut invalid = value.clone();
576            invalid[field] = serde_json::json!("PRIVATE_CANARY_body_key_path");
577            assert!(LlmRequestDiagnostic::from_json(&invalid.to_string()).is_none());
578        }
579        assert!(!diagnostic.matches_error_message(&format!(
580            "arbitrary upstream body: {}",
581            diagnostic.error_message()
582        )));
583        session.record_llm_request_diagnostic(diagnostic.clone());
584        session.set_last_run_error(diagnostic.error_message());
585        session.agent_runtime_state =
586            Some(crate::AgentRuntimeState::new(Uuid::new_v4().to_string()));
587        assert!(session.last_run_diagnostic().is_none());
588    }
589
590    #[test]
591    fn non_uuid_run_and_execution_replacement_cannot_reuse_the_old_receipt() {
592        let mut session = Session::new("diagnostic-current", "model");
593        let mut runtime = crate::AgentRuntimeState::new("private-non-uuid-run-a");
594        runtime.execution_id = Some("private-execution-a".into());
595        session.agent_runtime_state = Some(runtime);
596        let diagnostic = sample(&session);
597        session.record_llm_request_diagnostic(diagnostic.clone());
598        session.set_last_run_error(diagnostic.error_message());
599        let encoded = serde_json::to_string(&session).unwrap();
600        let mut cold: Session = serde_json::from_str(&encoded).unwrap();
601        assert!(cold.last_run_diagnostic().is_some());
602        let public = serde_json::to_string(&cold.last_run_diagnostic().unwrap()).unwrap();
603        assert!(!public.contains("private-non-uuid-run-a"));
604        assert!(!public.contains("private-execution-a"));
605        cold.agent_runtime_state.as_mut().unwrap().run_id = "private-non-uuid-run-b".into();
606        assert!(cold.last_run_diagnostic().is_none());
607        let mut cold: Session = serde_json::from_str(&encoded).unwrap();
608        cold.agent_runtime_state.as_mut().unwrap().execution_id =
609            Some("private-execution-b".into());
610        assert!(cold.last_run_diagnostic().is_none());
611    }
612
613    #[test]
614    fn stream_timeout_receipt_survives_cold_read_only_for_its_exact_run_and_error() {
615        let mut session = Session::new("timeout-session-private", "model");
616        let mut runtime = crate::AgentRuntimeState::new(Uuid::new_v4().to_string());
617        runtime.execution_id = Some("private-execution".into());
618        session.agent_runtime_state = Some(runtime);
619        let receipt = timeout_sample(&session);
620        assert!(session.record_llm_request_diagnostic(receipt.clone()));
621        assert!(session.last_run_diagnostic().is_none());
622        session.set_last_run_error(receipt.error_message());
623        let encoded = serde_json::to_string(&session).unwrap();
624        let cold: Session = serde_json::from_str(&encoded).unwrap();
625        let actual = cold.last_run_diagnostic().unwrap();
626        assert_eq!(actual.diagnostic_id(), receipt.diagnostic_id());
627        assert_eq!(actual.error_message(), receipt.error_message());
628        let public = serde_json::to_value(&actual).unwrap();
629        assert_eq!(
630            public["run_id"],
631            session.agent_runtime_state.as_ref().unwrap().run_id
632        );
633        assert!(!public.to_string().contains("timeout-session-private"));
634        assert!(!public.to_string().contains("private-execution"));
635
636        let mut wrong_session = cold.clone();
637        wrong_session.id = "other-session".into();
638        assert!(wrong_session.last_run_diagnostic().is_none());
639        let mut wrong_run = cold.clone();
640        wrong_run.agent_runtime_state.as_mut().unwrap().run_id = Uuid::new_v4().to_string();
641        assert!(wrong_run.last_run_diagnostic().is_none());
642        let mut wrong_execution = cold.clone();
643        wrong_execution
644            .agent_runtime_state
645            .as_mut()
646            .unwrap()
647            .execution_id = Some("next-execution".into());
648        assert!(wrong_execution.last_run_diagnostic().is_none());
649        let mut repeated_timeout = cold;
650        repeated_timeout.set_last_run_error(timeout_sample(&session).error_message());
651        assert!(repeated_timeout.last_run_diagnostic().is_none());
652        assert!(!repeated_timeout
653            .metadata
654            .contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
655    }
656
657    #[test]
658    fn timeout_receipt_rejects_inconsistent_observations_unknown_keys_and_unsafe_labels() {
659        let session = Session::new("session", "model");
660        let value = serde_json::to_value(timeout_sample(&session)).unwrap();
661        for (field, invalid) in [
662            ("schema_version", serde_json::json!(2)),
663            ("kind", serde_json::json!("another_timeout")),
664            ("error_id", serde_json::json!(Uuid::new_v4())),
665            ("phase", serde_json::json!("first_semantic")),
666            ("deadline_ms", serde_json::json!(0)),
667            ("deadline_ms", serde_json::json!(86_400_001_u64)),
668            ("last_transport_ms_ago", serde_json::json!(120_003)),
669            (
670                "last_semantic_ms_ago",
671                serde_json::json!(9_007_199_254_740_992_u64),
672            ),
673            ("semantic_output_started", serde_json::json!(false)),
674            ("retry_safe", serde_json::json!(true)),
675            ("last_http_status", serde_json::json!(429)),
676            ("retry_delay_ms", serde_json::json!(5_000)),
677            (
678                "provider",
679                serde_json::json!("https://private.invalid/token"),
680            ),
681            ("model", serde_json::json!("sk-private")),
682            ("model", serde_json::json!("Bearer_private")),
683            ("future_secret", serde_json::json!("PRIVATE_CANARY")),
684        ] {
685            let mut forged = value.clone();
686            forged[field] = invalid;
687            assert!(
688                LlmRequestDiagnostic::from_json(&forged.to_string()).is_none(),
689                "accepted invalid {field}"
690            );
691        }
692        let receipt = timeout_sample(&session);
693        assert!(
694            !receipt.matches_error_message(&format!("upstream said {}", receipt.error_message()))
695        );
696        assert!(!receipt.matches_error_message(&format!("LLM error: {}", receipt.error_message())));
697    }
698}
bamboo_domain/llm_request_diagnostic.rs

bamboo_domain/
llm_request_diagnostic.rs
