From af749de610c5cd9c8a93d8c8ab271c465016bab6 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 28 Aug 2026 17:18:35 +0200 Subject: [PATCH] Retry timed out model transports once --- crates/ghostlight-dungeon/src/model.rs | 181 ++++++++++++++++++--- crates/ghostlight-dungeon/src/scheduler.rs | 2 +- notes/ghostlight-current-system-map.md | 6 +- 3 files changed, 159 insertions(+), 30 deletions(-) diff --git a/crates/ghostlight-dungeon/src/model.rs b/crates/ghostlight-dungeon/src/model.rs index 967faea..3b93d5f 100644 --- a/crates/ghostlight-dungeon/src/model.rs +++ b/crates/ghostlight-dungeon/src/model.rs @@ -202,38 +202,42 @@ pub async fn run_validated_stage_with_timeout( let mut transport_attempt = 0; loop { let started = Instant::now(); - match tokio::time::timeout(timeout, port.run_observed(&attempt_request)).await { + let (failure_kind, diagnostic) = match tokio::time::timeout( + timeout, + port.run_observed(&attempt_request), + ) + .await + { Ok(Ok(output)) => break (output, started.elapsed().as_millis() as u64), - Ok(Err(error)) => { - let diagnostic = bounded_validation_error(&error); - provider_attempts.push(ModelProviderAttemptReceipt { - provider_request_id: None, - system_fingerprint: None, - finish_reason: Some("provider_error".into()), - latency_ms: started.elapsed().as_millis() as u64, - token_usage: None, - transport_features: Vec::new(), - local_validation_result: "provider_error".into(), - local_validation_error: Some(diagnostic.clone()), - }); - if transport_attempt == 0 { - transport_attempt += 1; - continue; - } - return Err(anyhow!( - "model stage {} provider failed twice against the same snapshot: {diagnostic}", - request.stage - )); - } - Err(_) => { - return Err(anyhow!( + Ok(Err(error)) => ("provider_error", bounded_validation_error(&error)), + Err(_) => ( + "timeout", + format!( "model stage {} timed out after {} seconds with {} input characters", request.stage, timeout.as_secs(), attempt_request.lived_stream.chars().count() - )); - } + ), + ), + }; + provider_attempts.push(ModelProviderAttemptReceipt { + provider_request_id: None, + system_fingerprint: None, + finish_reason: Some(failure_kind.into()), + latency_ms: started.elapsed().as_millis() as u64, + token_usage: None, + transport_features: Vec::new(), + local_validation_result: failure_kind.into(), + local_validation_error: Some(diagnostic.clone()), + }); + if transport_attempt == 0 { + transport_attempt += 1; + continue; } + return Err(anyhow!( + "model stage {} transport failed on both permitted attempts against the same snapshot; latest failure ({failure_kind}): {diagnostic}", + request.stage + )); } }; let ModelProviderOutput { @@ -510,7 +514,10 @@ fn nearest_rejected_object<'a>( #[cfg(test)] mod tests { use super::*; - use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::{ + Mutex, + atomic::{AtomicUsize, Ordering}, + }; fn receipt_fixture() -> ModelStageReceipt { ModelStageReceipt { @@ -569,6 +576,14 @@ mod tests { struct TransportFailureThenValid { calls: AtomicUsize, } + struct TimeoutThenValid { + calls: AtomicUsize, + requests: Mutex>, + } + struct MixedTransportFailures { + calls: AtomicUsize, + timeout_first: bool, + } struct PhysicallyRoutedModel; #[async_trait] @@ -660,6 +675,38 @@ mod tests { } } + #[async_trait] + impl ModelPort for TimeoutThenValid { + async fn run(&self, request: &ModelStageRequest) -> Result { + self.requests.lock().unwrap().push(request.clone()); + if self.calls.fetch_add(1, Ordering::SeqCst) == 0 { + std::future::pending().await + } else { + Ok(r#"{"answer":"ready"}"#.into()) + } + } + + fn provider(&self) -> &'static str { + "transient-timeout-fixture" + } + } + + #[async_trait] + impl ModelPort for MixedTransportFailures { + async fn run(&self, _: &ModelStageRequest) -> Result { + let first_attempt = self.calls.fetch_add(1, Ordering::SeqCst) == 0; + if first_attempt == self.timeout_first { + std::future::pending().await + } else { + Err(anyhow!("connection reset by fixture peer")) + } + } + + fn provider(&self) -> &'static str { + "mixed-transport-fixture" + } + } + #[async_trait] impl ModelPort for PhysicallyRoutedModel { async fn run(&self, _: &ModelStageRequest) -> Result { @@ -774,6 +821,86 @@ mod tests { ); } + #[tokio::test] + async fn transient_timeout_retries_the_exact_request_and_receipts_both_attempts() { + let port = TimeoutThenValid { + calls: AtomicUsize::new(0), + requests: Mutex::new(Vec::new()), + }; + let request = ModelStageRequest { + stage: "typed-stage".into(), + model: "fixture".into(), + snapshot_binding: "campaign:one:revision:4".into(), + lived_stream: "fixture".into(), + output_schema: Some(serde_json::json!({ + "type":"object", + "additionalProperties":false, + "required":["answer"], + "properties":{"answer":{"type":"string"}} + })), + source_receipt_ids: vec![], + temperature: None, + max_output_tokens: None, + }; + + let output = + run_validated_stage_with_timeout(&port, &request, std::time::Duration::from_millis(5)) + .await + .unwrap(); + + assert_eq!(port.calls.load(Ordering::SeqCst), 2); + let requests = port.requests.lock().unwrap(); + assert_eq!( + serde_json::to_value(&requests[0]).unwrap(), + serde_json::to_value(&requests[1]).unwrap() + ); + assert_eq!(output.structured.unwrap()["answer"], "ready"); + assert_eq!(output.receipt.snapshot_binding, request.snapshot_binding); + assert_eq!(output.receipt.provider_attempts.len(), 2); + assert_eq!( + output.receipt.provider_attempts[0].local_validation_result, + "timeout" + ); + assert_eq!( + output.receipt.provider_attempts[1].local_validation_result, + "valid" + ); + } + + #[tokio::test] + async fn mixed_transport_failures_report_the_latest_kind_without_rewriting_history() { + let request = ModelStageRequest { + stage: "typed-stage".into(), + model: "fixture".into(), + snapshot_binding: "campaign:one:revision:4".into(), + lived_stream: "fixture".into(), + output_schema: None, + source_receipt_ids: vec![], + temperature: None, + max_output_tokens: None, + }; + for (timeout_first, latest_kind) in [(false, "timeout"), (true, "provider_error")] { + let port = MixedTransportFailures { + calls: AtomicUsize::new(0), + timeout_first, + }; + let error = run_validated_stage_with_timeout( + &port, + &request, + std::time::Duration::from_millis(5), + ) + .await + .unwrap_err() + .to_string(); + + assert_eq!(port.calls.load(Ordering::SeqCst), 2); + assert!(error.contains("transport failed on both permitted attempts")); + assert!(error.contains(&format!("latest failure ({latest_kind})"))); + assert!(!error.contains("failed twice")); + assert!(!error.contains("timed out twice")); + } + } + #[tokio::test] async fn repeated_schema_failure_names_stage_and_exact_instance_path() { let request = ModelStageRequest { diff --git a/crates/ghostlight-dungeon/src/scheduler.rs b/crates/ghostlight-dungeon/src/scheduler.rs index b729d4e..8afeb2d 100644 --- a/crates/ghostlight-dungeon/src/scheduler.rs +++ b/crates/ghostlight-dungeon/src/scheduler.rs @@ -2225,7 +2225,7 @@ mod tests { assert_eq!(failure.pipeline_errors.len(), 1); assert!(failure.pipeline_errors[0].contains( - "cell Persona model stage failed: model stage cell_persona provider failed twice against the same snapshot" + "cell Persona model stage failed: model stage cell_persona transport failed on both permitted attempts against the same snapshot" )); assert_eq!( model.failures_remaining.load(Ordering::SeqCst), diff --git a/notes/ghostlight-current-system-map.md b/notes/ghostlight-current-system-map.md index 670ff47..f630cf1 100644 --- a/notes/ghostlight-current-system-map.md +++ b/notes/ghostlight-current-system-map.md @@ -212,8 +212,10 @@ owns no cell retry. The cell engine projects one exact lived moment. If the Persona supplies no explicit decision, the engine retries only Persona and Interpreter against that same projected stream; Projector does not rerun. The second Persona cites the first Persona and Interpreter attempt as causal -ancestry. Provider transport retains its separately bounded same-request retry, -and Interpreter draft repair remains workbench-local. Actor and cell membranes +ancestry. Provider transport gives both connection failures and attempt +timeouts one separately bounded retry of the exact same request and snapshot; +a later successful receipt preserves the failed attempt's diagnostic. +Interpreter draft repair remains workbench-local. Actor and cell membranes share one causal-source derivation primitive for Projector to Persona to Interpreter, while canonical evidence remains a separate namespace.