From e7fbdf1cfec877ec9d0a467f88ee004c4b339fd9 Mon Sep 17 00:00:00 2001 From: Gabriel Amazonas Date: Sat, 19 Sep 2026 18:13:36 -0300 Subject: [PATCH 1/2] fix: redact advisor fail-open errors in logs and the audit trail Signed-off-by: Gabriel Amazonas --- crates/libsy/src/algorithms/advisor_gate.rs | 8 ++- .../src/algorithms/advisor_gate/tests.rs | 71 +++++++++++++++++++ 2 files changed, 77 insertions(+), 2 deletions(-) diff --git a/crates/libsy/src/algorithms/advisor_gate.rs b/crates/libsy/src/algorithms/advisor_gate.rs index d638864f8..4551ac703 100644 --- a/crates/libsy/src/algorithms/advisor_gate.rs +++ b/crates/libsy/src/algorithms/advisor_gate.rs @@ -52,6 +52,7 @@ mod trigger; mod turn; use super::util::buffered_response::{BufferedResponse, buffer_response}; +use super::util::robustness::safe_error_summary; use budget::{ReviewBudget, ScopeKey, budget_scope, stall_key}; use signals::{GateSignalProcessor, GateSignals}; use telemetry::{ @@ -440,14 +441,17 @@ impl AdvisorGate { "advisor consult failed (fail_open = false): {error}" ))); } + // An upstream error's Display can quote request content back, + // so only its redacted summary reaches logs or the audit trail. + let summary = safe_error_summary(&error); tracing::warn!( target: "libsy", - error = %error, + error = %summary, "advisor gate: consult failed; passing the turn through (fail open)" ); emit_review_audit(ReviewAudit { verdict: "APPROVE", - error: Some(error.to_string()), + error: Some(summary), latency_ms, reply_head: None, usage: None, diff --git a/crates/libsy/src/algorithms/advisor_gate/tests.rs b/crates/libsy/src/algorithms/advisor_gate/tests.rs index 72d39615b..4e5687bb8 100644 --- a/crates/libsy/src/algorithms/advisor_gate/tests.rs +++ b/crates/libsy/src/algorithms/advisor_gate/tests.rs @@ -676,6 +676,77 @@ async fn fail_closed_propagates_refunds_and_counts() { assert_eq!(completion_text(&agg_of(response).await), "recovered"); } +/// Collects everything the `fmt` subscriber renders, so assertions can run +/// against the final log sink rather than a field mid-pipeline. +#[derive(Clone, Default)] +struct LogCapture(Arc>>); + +impl std::io::Write for LogCapture { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[tokio::test] +async fn fail_open_logs_redact_the_upstream_error_body() { + let script = Script::new(); + let gate = gate(AdvisorGateConfig::default()); + const MARKER: &str = "SECRET-UPSTREAM-QUOTE-the-user-prompt"; + let calls = Arc::clone(&script.calls); + let serve = move |model: ModelId, request: Request| { + let calls = Arc::clone(&calls); + Box::pin(async move { + let model = model.to_string(); + calls.lock().push((model.clone(), request)); + if model == ADVISOR { + // UpstreamHttp's Display interpolates the raw body, which is + // exactly the content that must not reach a sink. + Err(LlmClientError::UpstreamHttp { + status: http::StatusCode::INTERNAL_SERVER_ERROR, + body: format!("{MARKER}: validation failed"), + }) + } else { + Ok(reply("done")) + } + }) + }; + + // Installed as the process default so the capture sees events from every + // thread the drive touches: the suite runs tests in parallel, and a + // thread-local default does not cover them all. + let capture = LogCapture::default(); + let writer = capture.clone(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_max_level(tracing_subscriber::filter::LevelFilter::INFO) + .with_writer(move || writer.clone()) + .finish(); + tracing::subscriber::set_global_default(subscriber).expect("unused global subscriber"); + + let (_, response) = test_drive(gate, task_request(), serve) + .await + .expect("fail-open run"); + assert_eq!(completion_text(&agg_of(response).await), "done"); + + let logs = String::from_utf8(capture.0.lock().unwrap().clone()).expect("logs are utf-8"); + assert!( + !logs.contains(MARKER), + "the upstream error body reached a log sink: {logs}" + ); + assert!(logs.contains("upstream HTTP 500"), "{logs}"); + // The audit record still renders, with the redacted summary only. + assert!(logs.contains("advisor_review="), "{logs}"); + assert!( + logs.contains("consult failed; passing the turn through"), + "{logs}" + ); +} + #[tokio::test] async fn unparseable_verdict_refunds_and_approves() { let script = Script::new(); From adaf0f8b1558df63a9c13427bb0fb466cb651b78 Mon Sep 17 00:00:00 2001 From: Gabriel Amazonas Date: Sat, 19 Sep 2026 19:31:06 -0300 Subject: [PATCH 2/2] test: key the advisor redaction capture to this drive Signed-off-by: Gabriel Amazonas --- .../src/algorithms/advisor_gate/tests.rs | 38 +++++++++++-------- 1 file changed, 23 insertions(+), 15 deletions(-) diff --git a/crates/libsy/src/algorithms/advisor_gate/tests.rs b/crates/libsy/src/algorithms/advisor_gate/tests.rs index 4e5687bb8..10bb968d4 100644 --- a/crates/libsy/src/algorithms/advisor_gate/tests.rs +++ b/crates/libsy/src/algorithms/advisor_gate/tests.rs @@ -694,16 +694,21 @@ impl std::io::Write for LogCapture { #[tokio::test] async fn fail_open_logs_redact_the_upstream_error_body() { - let script = Script::new(); - let gate = gate(AdvisorGateConfig::default()); + // A target name unique to this test keys the captured warn line and audit + // record to this drive: the process-wide capture also sees parallel + // advisor-gate tests' events. + const ADVISOR_TARGET: &str = "advisor-redaction-probe"; const MARKER: &str = "SECRET-UPSTREAM-QUOTE-the-user-prompt"; - let calls = Arc::clone(&script.calls); - let serve = move |model: ModelId, request: Request| { - let calls = Arc::clone(&calls); + let gate = gate(AdvisorGateConfig::default()); + let models: HashMap> = [ + (Category::Efficient, vec![target(EXECUTOR)]), + (Category::Judge, vec![target(ADVISOR_TARGET)]), + ] + .into(); + let serve = move |model: ModelId, _request: Request| { Box::pin(async move { let model = model.to_string(); - calls.lock().push((model.clone(), request)); - if model == ADVISOR { + if model == ADVISOR_TARGET { // UpstreamHttp's Display interpolates the raw body, which is // exactly the content that must not reach a sink. Err(LlmClientError::UpstreamHttp { @@ -728,7 +733,7 @@ async fn fail_open_logs_redact_the_upstream_error_body() { .finish(); tracing::subscriber::set_global_default(subscriber).expect("unused global subscriber"); - let (_, response) = test_drive(gate, task_request(), serve) + let (_, response) = test_drive_with_models(gate, task_request(), models, serve) .await .expect("fail-open run"); assert_eq!(completion_text(&agg_of(response).await), "done"); @@ -738,13 +743,16 @@ async fn fail_open_logs_redact_the_upstream_error_body() { !logs.contains(MARKER), "the upstream error body reached a log sink: {logs}" ); - assert!(logs.contains("upstream HTTP 500"), "{logs}"); - // The audit record still renders, with the redacted summary only. - assert!(logs.contains("advisor_review="), "{logs}"); - assert!( - logs.contains("consult failed; passing the turn through"), - "{logs}" - ); + // Only this drive's decision target names the redacted summary, so its + // presence proves this drive's warn rendered with the safe summary. + let summary = r#"client call to target "advisor-redaction-probe" failed: upstream HTTP 500"#; + assert!(logs.contains(summary), "{logs}"); + // The audit record still renders, keyed to this drive's target, with the + // redacted summary rather than the body. + let audit = logs + .lines() + .find(|line| line.contains("advisor_review=") && line.contains(ADVISOR_TARGET)); + assert!(audit.is_some(), "{logs}"); } #[tokio::test]