diff --git a/src/llm.rs b/src/llm.rs index 1cfd3f4..518820e 100644 --- a/src/llm.rs +++ b/src/llm.rs @@ -55,6 +55,15 @@ impl ModelError { pub fn is_provider(&self) -> bool { self.error.downcast_ref::().is_some() } + + fn is_timeout(&self) -> bool { + self.error.chain().any(|cause| { + cause + .downcast_ref::() + .is_some_and(reqwest::Error::is_timeout) + || cause.to_string() == "LLM total timeout exceeded" + }) + } } impl std::fmt::Display for ModelError { @@ -78,6 +87,26 @@ fn add_usage(total: &mut Usage, usage: Usage) { total.completion_tokens += usage.completion_tokens; } +fn elapsed_text(elapsed: Duration) -> String { + if elapsed < Duration::from_secs(1) { + format!("{}ms", elapsed.as_millis()) + } else { + format!("{:.1}s", elapsed.as_secs_f64()) + } +} + +fn log_text(value: &str) -> String { + let mut sanitized = String::with_capacity(value.len()); + for character in value.chars() { + if character.is_control() { + sanitized.extend(character.escape_default()); + } else { + sanitized.push(character); + } + } + sanitized +} + /// Raw shape we ask the model for. Findings are validated leniently here and /// strictly (grounding, ranges) by the caller. #[derive(Debug, Deserialize)] @@ -224,24 +253,45 @@ impl LlmClient { .map(|m| { let client = self.clone(); let (model, system, user) = (m.clone(), system.to_string(), user.to_string()); - tokio::spawn( - async move { client.review_with_model(&model, &system, &user).await }, - ) + let task_model = model.clone(); + let handle = tokio::spawn(async move { + let model_log = log_text(&task_model); + eprintln!("postil: attempting consensus model: {model_log}"); + let started_at = Instant::now(); + let result = client.review_with_model(&task_model, &system, &user).await; + let elapsed = elapsed_text(started_at.elapsed()); + match &result { + Ok(_) => eprintln!( + "postil: consensus model {model_log} responded in {elapsed}" + ), + Err(error) if error.is_timeout() => eprintln!( + "postil: consensus model {model_log} timed out after {elapsed}" + ), + Err(error) => eprintln!( + "postil: consensus model {model_log} failed after {elapsed}: {}", + log_text(&format!("{error:#}")) + ), + } + result + }); + (model, handle) }) .collect(); let mut ok: Vec = Vec::new(); let mut failed_usage = Usage::default(); let mut last_err: Option = None; - for h in handles { - match h.await { + for (model, handle) in handles { + let model_log = log_text(&model); + match handle.await { Ok(Ok(r)) => ok.push(r), Ok(Err(mut e)) => { - eprintln!("postil: consensus model failed: {e:#}"); add_usage(&mut failed_usage, e.usage); e.usage = failed_usage; last_err = Some(e); } - Err(e) => eprintln!("postil: consensus task panicked: {e}"), + Err(e) => { + eprintln!("postil: consensus model {model_log} task panicked: {e}") + } } } match ok.len() { @@ -270,14 +320,43 @@ impl LlmClient { } else { let mut failed_usage = Usage::default(); let mut last_err = None; - for model in &chain { + for (index, model) in chain.iter().enumerate() { + let model_log = log_text(model); + eprintln!("postil: attempting model: {model_log}"); + let started_at = Instant::now(); match self.review_with_model(model, system, user).await { Ok(mut r) => { + eprintln!( + "postil: model {model_log} responded in {}", + elapsed_text(started_at.elapsed()) + ); add_usage(&mut r.usage, failed_usage); return Ok(r); } Err(mut e) => { - eprintln!("postil: model {model} failed: {e:#}"); + let elapsed = elapsed_text(started_at.elapsed()); + let has_fallback = index + 1 < chain.len(); + if e.is_timeout() { + if has_fallback { + eprintln!( + "postil: model {model_log} timed out after {elapsed}, falling back to next model" + ); + } else { + eprintln!( + "postil: model {model_log} timed out after {elapsed}; no fallback models remain" + ); + } + } else if has_fallback { + eprintln!( + "postil: model {model_log} failed after {elapsed}, falling back to next model: {}", + log_text(&format!("{e:#}")) + ); + } else { + eprintln!( + "postil: model {model_log} failed after {elapsed}; no fallback models remain: {}", + log_text(&format!("{e:#}")) + ); + } add_usage(&mut failed_usage, e.usage); e.usage = failed_usage; last_err = Some(e); @@ -315,17 +394,36 @@ impl LlmClient { ) -> std::result::Result { let mut failed_usage = Usage::default(); let mut last_err = None; - for model in cfg.scorer_chain() { + let chain = cfg.scorer_chain(); + for (index, model) in chain.iter().enumerate() { + let model_log = log_text(model); + eprintln!("postil: running scorer with {model_log}"); + let started_at = Instant::now(); match self - .score_with_model(&model, system, user, expected_len) + .score_with_model(model, system, user, expected_len) .await { Ok(mut r) => { + eprintln!( + "postil: scorer {model_log} completed successfully in {}", + elapsed_text(started_at.elapsed()) + ); add_usage(&mut r.usage, failed_usage); return Ok(r); } Err(mut e) => { - eprintln!("postil: scorer model {model} failed: {e:#}"); + let elapsed = elapsed_text(started_at.elapsed()); + if index + 1 < chain.len() { + eprintln!( + "postil: scorer {model_log} failed after {elapsed}, falling back to next scorer: {}", + log_text(&format!("{e:#}")) + ); + } else { + eprintln!( + "postil: scorer {model_log} failed after {elapsed}; no fallback scorers remain: {}", + log_text(&format!("{e:#}")) + ); + } add_usage(&mut failed_usage, e.usage); e.usage = failed_usage; last_err = Some(e); @@ -490,6 +588,7 @@ impl LlmClient { let mut attempt = 0u32; let text = loop { attempt += 1; + let attempt_started_at = Instant::now(); let request = self .http .post(format!("{}/chat/completions", self.api_base)) @@ -524,8 +623,10 @@ impl LlmClient { if retryable_status(status.as_u16()) && attempt <= TRANSIENT_RETRIES { let wait = std::time::Duration::from_secs(2 * attempt as u64); eprintln!( - "postil: {model} returned {status}, retrying in {}s \ - (attempt {attempt}/{TRANSIENT_RETRIES})", + "postil: model {} returned retryable HTTP {status} after {}, retrying in {}s \ + (retry {attempt}/{TRANSIENT_RETRIES})", + log_text(model), + elapsed_text(attempt_started_at.elapsed()), wait.as_secs() ); self.sleep_with_total_budget(wait).await?; @@ -536,10 +637,15 @@ impl LlmClient { // Connection-level failures retry too; timeouts do not (the // request already waited the full budget). Err(e) if e.is_connect() && attempt <= TRANSIENT_RETRIES => { - self.sleep_with_total_budget(std::time::Duration::from_secs( - 2 * attempt as u64, - )) - .await?; + let wait = std::time::Duration::from_secs(2 * attempt as u64); + eprintln!( + "postil: model {} hit a retryable connection error after {}, retrying in {}s \ + (retry {attempt}/{TRANSIENT_RETRIES})", + log_text(model), + elapsed_text(attempt_started_at.elapsed()), + wait.as_secs() + ); + self.sleep_with_total_budget(wait).await?; } Err(e) => { return Err(anyhow::Error::from(e).context("request to model endpoint failed")); @@ -890,6 +996,14 @@ mod tests { } } + #[test] + fn log_text_escapes_line_and_terminal_controls() { + assert_eq!( + log_text("model\nnext\r\t\u{1b}[31m"), + r"model\nnext\r\t\u{1b}[31m" + ); + } + impl Drop for EnvRestore { fn drop(&mut self) { for (name, value) in &self.saved { diff --git a/src/review.rs b/src/review.rs index e5b05f8..7fad3a0 100644 --- a/src/review.rs +++ b/src/review.rs @@ -703,7 +703,9 @@ async fn review_diff(cfg: &Config, args: &ReviewArgs, input: ReviewInput<'_>) -> } Ok(Err(e)) => { let detail = format!("{e:#}"); - eprintln!("postil: scorer failed open: {detail}"); + eprintln!( + "postil: scorer failed open after all scorer models failed" + ); let scorer_usage = e.usage(); usage.prompt_tokens += scorer_usage.prompt_tokens; usage.completion_tokens += scorer_usage.completion_tokens; diff --git a/tests/e2e.rs b/tests/e2e.rs index 5fc54e2..5891a44 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -234,6 +234,7 @@ async fn scorer_lowers_confidence_and_stores_both_values() { .code(0); let stdout = String::from_utf8(out.get_output().stdout.clone()).unwrap(); + let stderr = String::from_utf8(out.get_output().stderr.clone()).unwrap(); let env: Value = serde_json::from_str(&stdout).unwrap(); let finding = &env["findings"][0]; assert_eq!(env["scorerModel"], "anthropic/claude-haiku-4.5"); @@ -245,6 +246,10 @@ async fn scorer_lowers_confidence_and_stores_both_values() { assert_eq!(finding["generatorKind"], "risk"); assert_eq!(finding["scorerKind"], "risk"); assert_eq!(env["usage"]["promptTokens"], 130); + assert!(stderr.contains("postil: attempting model: generator-model")); + assert!(stderr.contains("postil: model generator-model responded in")); + assert!(stderr.contains("postil: running scorer with anthropic/claude-haiku-4.5")); + assert!(stderr.contains("postil: scorer anthropic/claude-haiku-4.5 completed successfully in")); let requests = server.received_requests().await.unwrap(); let scorer_request: Value = requests @@ -427,6 +432,7 @@ async fn scorer_error_fails_open_and_preserves_generator_values() { .code(0); let stdout = String::from_utf8(out.get_output().stdout.clone()).unwrap(); + let stderr = String::from_utf8(out.get_output().stderr.clone()).unwrap(); let env: Value = serde_json::from_str(&stdout).unwrap(); let finding = &env["findings"][0]; assert_eq!(env["gate"]["failing"], false); @@ -444,6 +450,49 @@ async fn scorer_error_fails_open_and_preserves_generator_values() { assert!(finding.get("scorerConfidence").is_none()); assert!(finding.get("generatorKind").is_none()); assert!(finding.get("scorerKind").is_none()); + assert!(stderr.contains("postil: scorer anthropic/claude-haiku-4.5 failed after")); + assert!(stderr.contains("falling back to next scorer")); + assert!(stderr.contains("postil: running scorer with openai/gpt-5-mini")); + assert!(stderr.contains("no fallback scorers remain")); +} + +#[tokio::test] +async fn scorer_provider_error_cannot_inject_stderr_lines() { + let server = MockServer::start().await; + mock_review_model( + &server, + "generator-model", + json!([finding_at(41, "warn", 0.92)]), + ) + .await; + for scorer_model in ["anthropic/claude-haiku-4.5", "openai/gpt-5-mini"] { + Mock::given(method("POST")) + .and(path("/chat/completions")) + .and(body_string_contains(scorer_model)) + .respond_with( + ResponseTemplate::new(400).set_body_string("bad\n[stderr] forged\u{1b}[31m"), + ) + .mount(&server) + .await; + } + + let dir = tempfile::tempdir().unwrap(); + let diff = write_diff(dir.path()); + let out = postil() + .current_dir(dir.path()) + .env("POSTIL_API_BASE", server.uri()) + .env("REVIEW_MODEL", "generator-model") + .args(["review", "--diff-file"]) + .arg(&diff) + .args(["--output", "json"]) + .assert() + .code(0); + + let stderr = String::from_utf8(out.get_output().stderr.clone()).unwrap(); + assert!(!stderr.contains("\n[stderr] forged")); + assert!(!stderr.contains('\u{1b}')); + assert!(stderr.contains(r"bad\n[stderr] forged\u{1b}[31m")); + assert!(stderr.contains("postil: scorer failed open after all scorer models failed")); } #[tokio::test] @@ -1276,6 +1325,42 @@ async fn cascade_falls_back_to_next_model() { let env: Value = serde_json::from_str(&String::from_utf8(out.get_output().stdout.clone()).unwrap()).unwrap(); assert_eq!(env["modelUsed"], "backup-model"); + let stderr = String::from_utf8(out.get_output().stderr.clone()).unwrap(); + assert!(stderr.contains("postil: attempting model: primary-model")); + assert!(stderr.contains("postil: model primary-model returned retryable HTTP 500")); + assert!(stderr.contains("retrying in 2s")); + assert!(stderr.contains("postil: model primary-model failed after")); + assert!(stderr.contains("falling back to next model")); + assert!(stderr.contains("postil: attempting model: backup-model")); + assert!(stderr.contains("postil: model backup-model responded in")); +} + +#[tokio::test] +async fn consensus_logs_each_model_outcome() { + let server = MockServer::start().await; + for model in ["consensus-a", "consensus-b"] { + mock_review_model(&server, model, json!([])).await; + } + + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join(".postil.yaml"), "model:\n consensus: 2\n").unwrap(); + let diff = write_diff(dir.path()); + let out = postil() + .current_dir(dir.path()) + .env("POSTIL_API_BASE", server.uri()) + .env("REVIEW_MODEL", "consensus-a") + .env("REVIEW_MODEL_CASCADE", "consensus-b") + .args(["review", "--diff-file"]) + .arg(&diff) + .args(["--output", "json"]) + .assert() + .code(0); + + let stderr = String::from_utf8(out.get_output().stderr.clone()).unwrap(); + for model in ["consensus-a", "consensus-b"] { + assert!(stderr.contains(&format!("postil: attempting consensus model: {model}"))); + assert!(stderr.contains(&format!("postil: consensus model {model} responded in"))); + } } #[tokio::test] @@ -1315,6 +1400,11 @@ async fn slow_model_request_times_out_and_falls_back() { let env: Value = serde_json::from_str(&String::from_utf8(out.get_output().stdout.clone()).unwrap()).unwrap(); assert_eq!(env["modelUsed"], "backup-model"); + let stderr = String::from_utf8(out.get_output().stderr.clone()).unwrap(); + assert!(stderr.contains("postil: model primary-model timed out after")); + assert!(stderr.contains("falling back to next model")); + assert!(stderr.contains("postil: attempting model: backup-model")); + assert!(stderr.contains("postil: model backup-model responded in")); } #[tokio::test]