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            && self.attempt_timeout_ms <= 60_000
265            && ["send_headers", "error_body", "response_status", "backoff"]
266                .contains(&self.phase.as_str())
267            && ["unknown", "response_received"].contains(&self.upstream_outcome.as_str())
268            && [
269                "none",
270                "request deadline exceeded",
271                "request attempt limit reached",
272                "initial response headers timed out",
273                "timeout",
274                "connect",
275                "request",
276                "body",
277                "decode",
278                "status",
279                "transport",
280                "upstream error body deadline exceeded",
281                "upstream error body read failed",
282                "connector database is temporarily recovering",
283                "temporary upstream failure",
284            ]
285            .contains(&self.failure.as_str())
286    }
287}
288
289/// Same domain-separated identity as provider operational logs; never store
290/// caller-controlled session text in the diagnostic payload.
291pub fn llm_diagnostic_session_hash(session_id: &str) -> String {
292    let mut digest = Sha256::new();
293    digest.update(b"bamboo-provider-operational-log-v1\0");
294    digest.update(session_id.as_bytes());
295    hex::encode(&digest.finalize()[..8])
296}
297
298impl LlmRequestDiagnostic {
299    pub fn from_stream_timeout(fields: StreamTimeoutDiagnostic) -> Option<Self> {
300        fields
301            .valid()
302            .then_some(Self(DiagnosticFields::StreamTimeout(fields)))
303    }
304
305    pub fn from_json(raw: &str) -> Option<Self> {
306        (raw.len() <= MAX_DIAGNOSTIC_BYTES)
307            .then(|| serde_json::from_str(raw).ok())
308            .flatten()
309    }
310
311    pub fn diagnostic_id(&self) -> Uuid {
312        match &self.0 {
313            DiagnosticFields::InitialRequest(fields) => fields.diagnostic_id,
314            DiagnosticFields::StreamTimeout(fields) => fields.diagnostic_id,
315        }
316    }
317
318    pub fn matches_session(&self, session_id: &str) -> bool {
319        let hash = llm_diagnostic_session_hash(session_id);
320        match &self.0 {
321            DiagnosticFields::InitialRequest(fields) => {
322                fields.session_hash.as_deref() == Some(hash.as_str())
323            }
324            DiagnosticFields::StreamTimeout(fields) => fields.session_hash == hash,
325        }
326    }
327
328    /// Exact generated error text only. An arbitrary upstream error containing
329    /// a copied UUID cannot acquire a receipt through substring matching.
330    pub fn matches_error_message(&self, message: &str) -> bool {
331        let message = match &self.0 {
332            DiagnosticFields::InitialRequest(_) => {
333                message.strip_prefix("LLM error: ").unwrap_or(message)
334            }
335            DiagnosticFields::StreamTimeout(_) => message,
336        };
337        message == self.error_message()
338    }
339
340    pub fn error_message(&self) -> String {
341        let f = match &self.0 {
342            DiagnosticFields::InitialRequest(fields) => fields,
343            DiagnosticFields::StreamTimeout(fields) => return fields.error_message(),
344        };
345        let status = f
346            .status
347            .map(|s| format!("HTTP {s}"))
348            .unwrap_or_else(|| "no HTTP response".into());
349        let limit = f
350            .attempt_limit
351            .map(|n| format!(" (limit {n})"))
352            .unwrap_or_default();
353        format!(
354            "Anthropic upstream request failed ({status}): {}. {} after {} request(s){limit}. Diagnostic ID: {}. Try again later or contact the upstream gateway administrator.",
355            f.failure, f.stop_reason.guidance(), f.attempt, f.diagnostic_id,
356        )
357    }
358}
359
360impl Session {
361    fn llm_diagnostic_run_fingerprint(&self) -> Option<String> {
362        self.agent_runtime_state.as_ref().map(|state| {
363            let mut digest = Sha256::new();
364            digest.update(b"bamboo-llm-diagnostic-run-v1\0");
365            digest.update((state.run_id.len() as u64).to_be_bytes());
366            digest.update(state.run_id.as_bytes());
367            if let Some(execution) = &state.execution_id {
368                digest.update([1]);
369                digest.update((execution.len() as u64).to_be_bytes());
370                digest.update(execution.as_bytes());
371            } else {
372                digest.update([0]);
373            }
374            hex::encode(&digest.finalize()[..16])
375        })
376    }
377
378    pub(crate) fn clear_llm_request_diagnostic(&mut self) {
379        self.metadata.remove(LAST_RUN_LLM_DIAGNOSTIC_KEY);
380        self.metadata.remove(DIAGNOSTIC_RUN_FINGERPRINT_KEY);
381    }
382
383    fn stored_llm_request_diagnostic(&self) -> Option<LlmRequestDiagnostic> {
384        let diagnostic =
385            LlmRequestDiagnostic::from_json(self.metadata.get(LAST_RUN_LLM_DIAGNOSTIC_KEY)?)?;
386        let run_id = self
387            .agent_runtime_state
388            .as_ref()
389            .and_then(|state| Uuid::parse_str(&state.run_id).ok());
390        let expected_run = self.llm_diagnostic_run_fingerprint();
391        (diagnostic.matches_session(&self.id)
392            && diagnostic.0.run_id() == run_id
393            && self.metadata.get(DIAGNOSTIC_RUN_FINGERPRINT_KEY) == expected_run.as_ref())
394        .then_some(diagnostic)
395    }
396
397    /// Capture only the observer for this actual provider call; existing final
398    /// Session persistence carries it without a second store or background job.
399    pub fn record_llm_request_diagnostic(&mut self, mut diagnostic: LlmRequestDiagnostic) -> bool {
400        if !diagnostic.matches_session(&self.id) {
401            return false;
402        }
403        diagnostic.0.set_run_id(
404            self.agent_runtime_state
405                .as_ref()
406                .and_then(|state| Uuid::parse_str(&state.run_id).ok()),
407        );
408        let Ok(raw) = serde_json::to_string(&diagnostic) else {
409            return false;
410        };
411        if raw.len() > MAX_DIAGNOSTIC_BYTES {
412            return false;
413        }
414        self.metadata
415            .insert(LAST_RUN_LLM_DIAGNOSTIC_KEY.into(), raw);
416        match self.llm_diagnostic_run_fingerprint() {
417            Some(fingerprint) => {
418                self.metadata
419                    .insert(DIAGNOSTIC_RUN_FINGERPRINT_KEY.into(), fingerprint);
420            }
421            None => {
422                self.metadata.remove(DIAGNOSTIC_RUN_FINGERPRINT_KEY);
423            }
424        }
425        true
426    }
427
428    pub(crate) fn retain_llm_request_diagnostic_for_error(&mut self, error: &str) {
429        if !self
430            .stored_llm_request_diagnostic()
431            .is_some_and(|d| d.matches_error_message(error))
432        {
433            self.clear_llm_request_diagnostic();
434        }
435    }
436
437    /// Public/cold-read projection requires the same Session, run and exact
438    /// canonical error. Old, foreign, malformed or overwritten evidence is absent.
439    pub fn last_run_diagnostic(&self) -> Option<LlmRequestDiagnostic> {
440        let error = self.last_run_error()?;
441        self.stored_llm_request_diagnostic()
442            .filter(|d| d.matches_error_message(&error))
443    }
444}
445
446#[cfg(test)]
447mod tests {
448    use super::*;
449
450    fn timeout_sample(session: &Session) -> LlmRequestDiagnostic {
451        let id = Uuid::new_v4();
452        LlmRequestDiagnostic::from_stream_timeout(StreamTimeoutDiagnostic {
453            schema_version: 1,
454            kind: "stream_timeout".into(),
455            diagnostic_id: id,
456            error_id: id,
457            observed_at: Utc::now(),
458            provider: Some("a2a3cebf-8abc-45cb-8407-66c90fd9a978".into()),
459            model: Some("gpt-6-astra".into()),
460            session_hash: llm_diagnostic_session_hash(&session.id),
461            run_id: None,
462            phase: "transport_idle".into(),
463            deadline_ms: 120_000,
464            last_transport_ms_ago: 120_002,
465            last_semantic_ms_ago: Some(120_002),
466            semantic_output_started: true,
467            retry_safe: false,
468            last_http_status: None,
469            retry_delay_ms: None,
470        })
471        .unwrap()
472    }
473
474    fn sample(session: &Session) -> LlmRequestDiagnostic {
475        let id = Uuid::new_v4();
476        LlmRequestDiagnostic::from_json(
477            &serde_json::json!({
478                "schema_version":1,"diagnostic_id":id,"request_id":id,"error_id":id,
479                "observed_at":Utc::now(),"provider":"anthropic",
480                "session_hash":llm_diagnostic_session_hash(&session.id),"run_id":null,
481                "attempt":1,"attempt_limit":4,"event":"stop","phase":"send_headers",
482                "failure":"initial response headers timed out","status":null,
483                "attempt_timeout_ms":20000,"total_budget_ms":60000,"attempt_elapsed_ms":20000,
484                "total_elapsed_ms":60000,"remaining_budget_ms":0,"retry_delay_ms":null,
485                "stop_reason":"time_budget_exhausted","upstream_outcome":"unknown",
486            })
487            .to_string(),
488        )
489        .unwrap()
490    }
491
492    #[test]
493    fn failure_receipt_survives_cold_session_round_trip_and_is_bound_to_error_and_session() {
494        let mut session = Session::new("diagnostic-current", "model");
495        let diagnostic = sample(&session);
496        assert!(session.record_llm_request_diagnostic(diagnostic.clone()));
497        assert!(session.last_run_diagnostic().is_none());
498        session.set_last_run_error(format!("LLM error: {}", diagnostic.error_message()));
499        let cold: Session =
500            serde_json::from_str(&serde_json::to_string(&session).unwrap()).unwrap();
501        assert_eq!(cold.last_run_diagnostic(), Some(diagnostic.clone()));
502        let mut foreign = cold.clone();
503        foreign.id = "other".into();
504        assert!(foreign.last_run_diagnostic().is_none());
505        let mut changed = cold.clone();
506        changed.set_last_run_error("another provider failure");
507        assert!(changed.last_run_diagnostic().is_none());
508        assert!(!changed.metadata.contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
509        let mut cleared = cold;
510        cleared.clear_last_run_error();
511        assert!(!cleared.metadata.contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
512    }
513
514    #[test]
515    fn configured_and_continuous_receipts_keep_distinct_validated_limits() {
516        let session = Session::new("retry-diagnostic", "model");
517        let mut bounded = serde_json::to_value(sample(&session)).unwrap();
518        bounded["attempt_limit"] = 12.into();
519        bounded["total_budget_ms"] = 300_000.into();
520        assert!(LlmRequestDiagnostic::from_json(&bounded.to_string()).is_some());
521        let mut continuous = bounded.clone();
522        continuous["retry_mode"] = "until_cancelled".into();
523        for field in ["attempt_limit", "total_budget_ms", "remaining_budget_ms"] {
524            continuous[field] = serde_json::Value::Null;
525        }
526        continuous["stop_reason"] = "cancelled_or_dropped".into();
527        let diagnostic = LlmRequestDiagnostic::from_json(&continuous.to_string()).unwrap();
528        assert!(diagnostic.matches_error_message(&diagnostic.error_message()));
529        for field in ["attempt_limit", "total_budget_ms", "remaining_budget_ms"] {
530            let mut invalid = continuous.clone();
531            invalid[field] = 1.into();
532            assert!(LlmRequestDiagnostic::from_json(&invalid.to_string()).is_none());
533        }
534        continuous["retry_mode"] = "unrecognized".into();
535        assert!(LlmRequestDiagnostic::from_json(&continuous.to_string()).is_none());
536        bounded["total_budget_ms"] = 3_600_001.into();
537        assert!(LlmRequestDiagnostic::from_json(&bounded.to_string()).is_none());
538    }
539
540    #[test]
541    fn invalid_free_text_unknown_fields_and_stale_run_fail_closed() {
542        let mut session = Session::new("diagnostic-current", "model");
543        let diagnostic = sample(&session);
544        let value = serde_json::to_value(&diagnostic).unwrap();
545        for field in [
546            "provider",
547            "phase",
548            "failure",
549            "upstream_outcome",
550            "session_hash",
551            "future_secret",
552        ] {
553            let mut invalid = value.clone();
554            invalid[field] = serde_json::json!("PRIVATE_CANARY_body_key_path");
555            assert!(LlmRequestDiagnostic::from_json(&invalid.to_string()).is_none());
556        }
557        assert!(!diagnostic.matches_error_message(&format!(
558            "arbitrary upstream body: {}",
559            diagnostic.error_message()
560        )));
561        session.record_llm_request_diagnostic(diagnostic.clone());
562        session.set_last_run_error(diagnostic.error_message());
563        session.agent_runtime_state =
564            Some(crate::AgentRuntimeState::new(Uuid::new_v4().to_string()));
565        assert!(session.last_run_diagnostic().is_none());
566    }
567
568    #[test]
569    fn non_uuid_run_and_execution_replacement_cannot_reuse_the_old_receipt() {
570        let mut session = Session::new("diagnostic-current", "model");
571        let mut runtime = crate::AgentRuntimeState::new("private-non-uuid-run-a");
572        runtime.execution_id = Some("private-execution-a".into());
573        session.agent_runtime_state = Some(runtime);
574        let diagnostic = sample(&session);
575        session.record_llm_request_diagnostic(diagnostic.clone());
576        session.set_last_run_error(diagnostic.error_message());
577        let encoded = serde_json::to_string(&session).unwrap();
578        let mut cold: Session = serde_json::from_str(&encoded).unwrap();
579        assert!(cold.last_run_diagnostic().is_some());
580        let public = serde_json::to_string(&cold.last_run_diagnostic().unwrap()).unwrap();
581        assert!(!public.contains("private-non-uuid-run-a"));
582        assert!(!public.contains("private-execution-a"));
583        cold.agent_runtime_state.as_mut().unwrap().run_id = "private-non-uuid-run-b".into();
584        assert!(cold.last_run_diagnostic().is_none());
585        let mut cold: Session = serde_json::from_str(&encoded).unwrap();
586        cold.agent_runtime_state.as_mut().unwrap().execution_id =
587            Some("private-execution-b".into());
588        assert!(cold.last_run_diagnostic().is_none());
589    }
590
591    #[test]
592    fn stream_timeout_receipt_survives_cold_read_only_for_its_exact_run_and_error() {
593        let mut session = Session::new("timeout-session-private", "model");
594        let mut runtime = crate::AgentRuntimeState::new(Uuid::new_v4().to_string());
595        runtime.execution_id = Some("private-execution".into());
596        session.agent_runtime_state = Some(runtime);
597        let receipt = timeout_sample(&session);
598        assert!(session.record_llm_request_diagnostic(receipt.clone()));
599        assert!(session.last_run_diagnostic().is_none());
600        session.set_last_run_error(receipt.error_message());
601        let encoded = serde_json::to_string(&session).unwrap();
602        let cold: Session = serde_json::from_str(&encoded).unwrap();
603        let actual = cold.last_run_diagnostic().unwrap();
604        assert_eq!(actual.diagnostic_id(), receipt.diagnostic_id());
605        assert_eq!(actual.error_message(), receipt.error_message());
606        let public = serde_json::to_value(&actual).unwrap();
607        assert_eq!(
608            public["run_id"],
609            session.agent_runtime_state.as_ref().unwrap().run_id
610        );
611        assert!(!public.to_string().contains("timeout-session-private"));
612        assert!(!public.to_string().contains("private-execution"));
613
614        let mut wrong_session = cold.clone();
615        wrong_session.id = "other-session".into();
616        assert!(wrong_session.last_run_diagnostic().is_none());
617        let mut wrong_run = cold.clone();
618        wrong_run.agent_runtime_state.as_mut().unwrap().run_id = Uuid::new_v4().to_string();
619        assert!(wrong_run.last_run_diagnostic().is_none());
620        let mut wrong_execution = cold.clone();
621        wrong_execution
622            .agent_runtime_state
623            .as_mut()
624            .unwrap()
625            .execution_id = Some("next-execution".into());
626        assert!(wrong_execution.last_run_diagnostic().is_none());
627        let mut repeated_timeout = cold;
628        repeated_timeout.set_last_run_error(timeout_sample(&session).error_message());
629        assert!(repeated_timeout.last_run_diagnostic().is_none());
630        assert!(!repeated_timeout
631            .metadata
632            .contains_key(LAST_RUN_LLM_DIAGNOSTIC_KEY));
633    }
634
635    #[test]
636    fn timeout_receipt_rejects_inconsistent_observations_unknown_keys_and_unsafe_labels() {
637        let session = Session::new("session", "model");
638        let value = serde_json::to_value(timeout_sample(&session)).unwrap();
639        for (field, invalid) in [
640            ("schema_version", serde_json::json!(2)),
641            ("kind", serde_json::json!("another_timeout")),
642            ("error_id", serde_json::json!(Uuid::new_v4())),
643            ("phase", serde_json::json!("first_semantic")),
644            ("deadline_ms", serde_json::json!(0)),
645            ("deadline_ms", serde_json::json!(86_400_001_u64)),
646            ("last_transport_ms_ago", serde_json::json!(120_003)),
647            (
648                "last_semantic_ms_ago",
649                serde_json::json!(9_007_199_254_740_992_u64),
650            ),
651            ("semantic_output_started", serde_json::json!(false)),
652            ("retry_safe", serde_json::json!(true)),
653            ("last_http_status", serde_json::json!(429)),
654            ("retry_delay_ms", serde_json::json!(5_000)),
655            (
656                "provider",
657                serde_json::json!("https://private.invalid/token"),
658            ),
659            ("model", serde_json::json!("sk-private")),
660            ("model", serde_json::json!("Bearer_private")),
661            ("future_secret", serde_json::json!("PRIVATE_CANARY")),
662        ] {
663            let mut forged = value.clone();
664            forged[field] = invalid;
665            assert!(
666                LlmRequestDiagnostic::from_json(&forged.to_string()).is_none(),
667                "accepted invalid {field}"
668            );
669        }
670        let receipt = timeout_sample(&session);
671        assert!(
672            !receipt.matches_error_message(&format!("upstream said {}", receipt.error_message()))
673        );
674        assert!(!receipt.matches_error_message(&format!("LLM error: {}", receipt.error_message())));
675    }
676}
bamboo_domain/llm_request_diagnostic.rs

bamboo_domain/
llm_request_diagnostic.rs
