Skip to content

Show review model progress in run logs #55

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
morgaesis merged 1 commit into from
Jul 12, 2026
Merged

Show review model progress in run logs #55

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 132 additions & 18 deletions src/llm.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,15 @@ impl ModelError {
pub fn is_provider(&self) -> bool {
self.error.downcast_ref::<ProviderError>().is_some()
}

fn is_timeout(&self) -> bool {
self.error.chain().any(|cause| {
cause
.downcast_ref::<reqwest::Error>()
.is_some_and(reqwest::Error::is_timeout)
|| cause.to_string() == "LLM total timeout exceeded"
})
}
}

impl std::fmt::Display for ModelError {
Expand All @@ -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)]
Expand Down Expand Up @@ -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<ModelReview> = Vec::new();
let mut failed_usage = Usage::default();
let mut last_err: Option<ModelError> = 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() {
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -315,17 +394,36 @@ impl LlmClient {
) -> std::result::Result<ScorerReview, ModelError> {
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);
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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?;
Expand All @@ -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"));
Expand Down Expand Up @@ -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 {
Expand Down
4 changes: 3 additions & 1 deletion src/review.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
90 changes: 90 additions & 0 deletions tests/e2e.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -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
Expand Down Expand Up @@ -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);
Expand All @@ -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]
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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]
Expand Down
Loading