fix!: batch of refactor, sink, audit, CLI and rules fixes

crates/sqlbuild-analysis/src/column_references/_helpers/references.rs

@@ -346,17 +346,52 @@ impl Analysis<'_> {
         let Some(body) = select_body(expression) else {
             return;
         };
+        let shadowed: HashSet<(usize, usize)> = self.lambda_parameter_uses(body);
         for node in walk_in_scope(body, false) {
             let Expression::Column(column) = node else {
                 continue;
             };
             if column.name.name == STAR || column.name.name.to_lowercase() != self.column {
                 continue;
             }
+            if column.table.is_none()
+                && column
+                    .span
+                    .is_some_and(|span| shadowed.contains(&(span.start, span.end)))
+            {
+                continue;
+            }
             self.record_column(column, select, chain, role);
         }
     }
 
+    /// Spans of unqualified uses of a lambda parameter named like the column, which read the parameter.
+    fn lambda_parameter_uses(&self, body: &Expression) -> HashSet<(usize, usize)> {
+        let mut shadowed: HashSet<(usize, usize)> = HashSet::new();
+        for node in walk_in_scope(body, false) {
+            let Expression::Lambda(lambda) = node else {
+                continue;
+            };
+            if !lambda
+                .parameters
+                .iter()
+                .any(|parameter| parameter.name.to_lowercase() == self.column)
+            {
+                continue;
+            }
+            for inner in walk_in_scope(&lambda.body, false) {
+                if let Expression::Column(column) = inner
+                    && column.table.is_none()
+                    && column.name.name.to_lowercase() == self.column
+                    && let Some(span) = column.span
+                {
+                    shadowed.insert((span.start, span.end));
+                }
+            }
+        }
+        shadowed
+    }
+
     fn record_column(
         &mut self,
         column: &polyglot_sql::expressions::Column,

crates/sqlbuild-analysis/src/column_references/tests/test_column_references.rs

@@ -76,6 +76,21 @@ fn given_column_uses_when_analyzing_then_reports_rename_sites() -> Result<(), St
             expected_output_aliases: &[],
             expected_outputs: 0,
         },
+        ColumnReferenceTestCase {
+            description: "a lambda parameter named like the column shadows it in the body",
+            sql: "SELECT list_filter([amount], amount -> amount > 0) AS kept, \
+                  list_filter([1], lambda amount: amount > 0) AS other FROM orders_fact",
+            output_ctes: &[],
+            expected_parsed: true,
+            expected_names: &["amount"],
+            expected_scopes: &["root"],
+            expected_bare_projections: 0,
+            expected_aliases: &[],
+            expected_star_scopes: &[],
+            expected_joins: 0,
+            expected_output_aliases: &[],
+            expected_outputs: 0,
+        },
         ColumnReferenceTestCase {
             description: "output scopes report aliased and bare projections of every branch",
             sql: "WITH fixture AS (SELECT 1 AS amount UNION ALL SELECT 2 AS amount) \

crates/sqlbuild-attachments/src/audits/_helpers/rendering.rs

@@ -1,6 +1,8 @@
 //! Python's argument merge, SQL rendering and policy resolution for one attachment.
 
-use crate::audits::_helpers::parameters::{RenderStop, render_parameterized_sql};
+use crate::audits::_helpers::parameters::{
+    RenderStop, parameter_matches, render_parameterized_sql,
+};
 use crate::audits::_helpers::policies::{resolve_run_scope, resolve_severity};
 use crate::audits::constants::{MODEL_RESOURCE, SEED_RESOURCE, SOURCE_RESOURCE};
 use crate::audits::models::{ArgumentValue, AuditAttachment, PolicySource};
@@ -94,3 +96,36 @@ pub(crate) fn merged_arguments(
     }
     Ok(merged)
 }
+
+/// The error for the first explicit argument neither the audit SQL nor its evidence SQL reads.
+pub(crate) fn unused_argument(attachment: &AuditAttachment) -> Option<String> {
+    let mut used: Vec<&str> = parameter_matches(&attachment.sql_body)
+        .into_iter()
+        .map(|item| item.name)
+        .collect();
+    if let Some(evidence) = &attachment.evidence_sql {
+        used.extend(
+            parameter_matches(evidence)
+                .into_iter()
+                .map(|item| item.name),
+        );
+    }
+    let implicit: Vec<&str> = attachment
+        .implicit_arguments
+        .iter()
+        .map(|(name, _)| name.as_str())
+        .collect();
+    let (name, _) = attachment
+        .explicit_arguments
+        .iter()
+        .find(|(name, _)| !used.contains(&name.as_str()) && !implicit.contains(&name.as_str()))?;
+    let definition: String = attachment
+        .definition_path
+        .as_deref()
+        .map_or_else(|| "its built-in definition".to_owned(), str::to_owned);
+    Some(format!(
+        "{} audit '{}' passes argument '{name}', which the audit's SQL in {definition} never \
+         uses; remove '{name}' from the attachment",
+        attachment.owner_label, attachment.definition_name
+    ))
+}

crates/sqlbuild-attachments/src/audits/main/render_attached_audit.rs

@@ -1,6 +1,6 @@
 //! Attached generic audit rendering: argument merge, SQL, severity and run scope.
 
-use crate::audits::_helpers::rendering::{merged_arguments, policies, rendered};
+use crate::audits::_helpers::rendering::{merged_arguments, policies, rendered, unused_argument};
 use crate::audits::models::{ArgumentValue, AuditAttachment, AuditRendering, RenderedAudit};
 
 /// Render one attachment, or the error rendering raises.
@@ -14,6 +14,9 @@ pub fn render_attached_audit(attachment: &AuditAttachment) -> AuditRendering {
             .as_deref()
             .map(|evidence| rendered(attachment, evidence, &arguments))
             .transpose()?;
+        if let Some(message) = unused_argument(attachment) {
+            return Err(message);
+        }
         Ok(RenderedAudit {
             sql_body,
             evidence_sql,

crates/sqlbuild-attachments/src/audits/models.rs

@@ -28,6 +28,8 @@ pub struct AuditAttachment {
     /// Python's `str(owner_file)` and the generic audit name, which its error messages quote.
     pub owner_label: String,
     pub definition_name: String,
+    /// The project file defining the audit, or `None` for a built-in audit.
+    pub definition_path: Option<String>,
     pub sql_body: String,
     pub evidence_sql: Option<String>,
     /// Arguments from the attached resource, such as `model` and `column`; always text.

crates/sqlbuild-attachments/src/audits/tests/helpers.rs

@@ -40,6 +40,7 @@ pub(super) fn attachment(
     AuditAttachment {
         owner_label: "models/schema.yml".to_owned(),
         definition_name: "accepted_values".to_owned(),
+        definition_path: None,
         sql_body: "SELECT * FROM t WHERE @column NOT IN (@'values')".to_owned(),
         evidence_sql: Some("SELECT @column FROM t".to_owned()),
         implicit_arguments: vec![("column".to_owned(), "status".to_owned())],

crates/sqlbuild-attachments/src/audits/tests/test_render_attached_audit.rs

@@ -80,6 +80,28 @@ fn given_attached_audits_when_rendering_then_python_rendering_is_returned() {
                  implicit column from attached context",
             ),
         },
+        AttachedAuditTestCase {
+            description: "an argument the audit SQL never reads names itself for removal",
+            attachment: AuditAttachment {
+                definition_path: Some("audits/generic/accepted_values.sql".to_owned()),
+                ..attachment(
+                    vec![
+                        values.clone(),
+                        (
+                            "where_condition".to_owned(),
+                            ArgumentValue::Text("status <> 'void'".to_owned()),
+                        ),
+                    ],
+                    None,
+                    None,
+                )
+            },
+            expected_rendering: failed(
+                "models/schema.yml audit 'accepted_values' passes argument 'where_condition', \
+                 which the audit's SQL in audits/generic/accepted_values.sql never uses; remove \
+                 'where_condition' from the attachment",
+            ),
+        },
         AttachedAuditTestCase {
             description: "a missing argument is Python's error before the policies",
             attachment: attachment(Vec::new(), Some("fatal"), None),

crates/sqlbuild-cli/src/dispatch/main/event_export_warning.rs

@@ -1,5 +1,7 @@
 //! The stderr warning for lifecycle events dropped or failed at command exit.
 
+use sqlbuild_core::event_export::constants::DEFAULT_EVENT_EXPORT_INVOCATION_TIMEOUT_SECONDS;
+
 use crate::dispatch::constants::SHUTDOWN_TIMEOUT_SETTING;
 use crate::dispatch::models::{ExportSummary, ExporterLoss};
 
@@ -32,7 +34,9 @@ pub fn event_export_warning(summary: &ExportSummary, command_succeeded: bool) ->
     };
     Some(format!(
         "{prefix}: {} of {} {unit} dropped, {} failed ({sink_label} {names}). \
-         Increase {SHUTDOWN_TIMEOUT_SETTING} or check sink health.",
+         Check sink health: a sink call slower than {DEFAULT_EVENT_EXPORT_INVOCATION_TIMEOUT_SECONDS}s \
+         drops that sink's events until it returns; events still queued at exit need a longer \
+         {SHUTDOWN_TIMEOUT_SETTING}.",
         summary.dropped, summary.accepted, summary.failed
     ))
 }

crates/sqlbuild-cli/src/dispatch/tests/test_event_export_warning.rs

@@ -18,7 +18,9 @@ fn given_final_export_summary_when_formatting_warning_then_reports_loss_only() {
             expected_warning: Some(
                 "Warning: command completed successfully, but lifecycle event export was \
                  incomplete: 140 of 248 events dropped, 0 failed (sink 'orders_events'). \
-                 Increase sinks.lifecycle.shutdown_timeout or check sink health.",
+                 Check sink health: a sink call slower than 1s drops that sink's events until it \
+                 returns; events still queued at exit need a longer \
+                 sinks.lifecycle.shutdown_timeout.",
             ),
         },
         EventExportWarningTestCase {
@@ -28,7 +30,9 @@ fn given_final_export_summary_when_formatting_warning_then_reports_loss_only() {
             expected_warning: Some(
                 "Warning: lifecycle event export was incomplete: 0 of 10 events dropped, \
                  1 failed (sink 'orders_events'). \
-                 Increase sinks.lifecycle.shutdown_timeout or check sink health.",
+                 Check sink health: a sink call slower than 1s drops that sink's events until it \
+                 returns; events still queued at exit need a longer \
+                 sinks.lifecycle.shutdown_timeout.",
             ),
         },
         EventExportWarningTestCase {
@@ -41,7 +45,9 @@ fn given_final_export_summary_when_formatting_warning_then_reports_loss_only() {
             expected_warning: Some(
                 "Warning: command completed successfully, but lifecycle event export was \
                  incomplete: 4 of 40 event deliveries dropped, 1 failed (sink 'orders_events'). \
-                 Increase sinks.lifecycle.shutdown_timeout or check sink health.",
+                 Check sink health: a sink call slower than 1s drops that sink's events until it \
+                 returns; events still queued at exit need a longer \
+                 sinks.lifecycle.shutdown_timeout.",
             ),
         },
     ];

crates/sqlbuild-model-config/src/header_metadata/_helpers/audits.rs

@@ -90,12 +90,12 @@ fn configured_audit<N: AuthoredNode>(
     let minimum_samples = optional_count(
         entry(options, "minimum_samples"),
         option_site,
-        (definition, "minimum_samples"),
+        (definition, "minimum_samples", options),
     )?;
     let evidence_limit = optional_count(
         entry(options, "evidence_limit"),
         option_site,
-        (definition, "evidence_limit"),
+        (definition, "evidence_limit", options),
     )?;
     let meta = optional_meta(entry(options, "meta"), option_site)?;
     let arguments: Vec<(N, N)> = options

crates/sqlbuild-model-config/src/header_metadata/_helpers/text.rs

@@ -57,13 +57,13 @@ pub(crate) fn optional_bool<N: AuthoredNode>(
     }
 }
 
-/// Read a non-negative non-boolean integer option of audit `definition` that fits in 64 bits.
+/// Read a non-negative integer option of audit `definition` that fits in 64 bits; help repeats `options`.
 pub(crate) fn optional_count<N: AuthoredNode>(
     node: Option<&N>,
     site: Site<'_>,
-    option: (&str, &str),
+    option: (&str, &str, &[(N, N)]),
 ) -> Result<Option<N>, ConfigError> {
-    let (definition, key) = option;
+    let (definition, key, options) = option;
     match (
         node.map(AuthoredNode::kind),
         node.and_then(AuthoredNode::integer),
@@ -75,14 +75,29 @@ pub(crate) fn optional_count<N: AuthoredNode>(
                 "'{key}' {} is larger than a 64-bit integer",
                 node.map(AuthoredNode::python_str).unwrap_or_default()
             ))
-            .with_help(count_help(site, definition, key))),
+            .with_help(count_help(site, (definition, options), key))),
         (Some(_), _) => Err(site.error(&format!("'{key}' must be a non-negative integer"))),
     }
 }
 
-/// The MODEL header help that sets audit option `key` to the largest 64-bit integer.
-fn count_help(site: Site<'_>, definition: &str, key: &str) -> String {
-    let audits: String = format!("audits [{definition} ({key} {})]", i64::MAX);
+/// The MODEL header help repeating the audit call with option `key` set to the largest 64-bit integer.
+fn count_help<N: AuthoredNode>(
+    site: Site<'_>,
+    (definition, options): (&str, &[(N, N)]),
+    key: &str,
+) -> String {
+    let rendered: Vec<String> = options
+        .iter()
+        .map(|(name, value)| {
+            let name: String = name.text().unwrap_or_else(|| name.python_str());
+            if name == key {
+                format!("{name} {}", i64::MAX)
+            } else {
+                format!("{name} {}", header_value(value))
+            }
+        })
+        .collect();
+    let audits: String = format!("audits [{definition} ({})]", rendered.join(", "));
     let entry: String = site.column.map_or_else(
         || audits.clone(),
         |column| format!("columns ({column} ({audits}))"),
@@ -94,6 +109,34 @@ fn count_help(site: Site<'_>, definition: &str, key: &str) -> String {
     )
 }
 
+/// One authored value as MODEL header syntax: quoted strings, `[...]` lists and `(key value)` maps.
+fn header_value<N: AuthoredNode>(node: &N) -> String {
+    match node.kind() {
+        NodeKind::Null => "null".to_owned(),
+        NodeKind::Bool(value) => value.to_string(),
+        NodeKind::Str => {
+            let text: String = node.text().unwrap_or_default();
+            format!("\"{}\"", text.replace('\\', "\\\\").replace('"', "\\\""))
+        }
+        NodeKind::List | NodeKind::Tuple => {
+            let items: Vec<String> = node.items().iter().map(header_value).collect();
+            format!("[{}]", items.join(", "))
+        }
+        NodeKind::Map => {
+            let entries: Vec<String> = node
+                .entries()
+                .iter()
+                .map(|(key, value)| {
+                    let key: String = key.text().unwrap_or_else(|| key.python_str());
+                    format!("{key} {}", header_value(value))
+                })
+                .collect();
+            format!("({})", entries.join(", "))
+        }
+        NodeKind::Int { .. } | NodeKind::Other => node.python_str(),
+    }
+}
+
 /// Return the mapping entry whose key is the string `key`.
 pub(crate) fn entry<'entries, N: AuthoredNode>(
     entries: &'entries [(N, N)],

crates/sqlbuild-python/src/bindings/_helpers/cli/entry.rs

@@ -28,6 +28,7 @@ use crate::bindings::_helpers::diff_report::output::render_diff_error_json;
 use crate::bindings::_helpers::native_host::host;
 use crate::bindings::_helpers::output_capture::records::invocation_context;
 use crate::bindings::_helpers::presentation_text::terminal::stdout_supports_color;
+use crate::bindings::_helpers::recall::events::escaped_text;
 use crate::bindings::_helpers::recall::session::{
     InvocationRecall, finish_invocation_recall, start_invocation_recall,
 };
@@ -113,23 +114,27 @@ fn print_object(value: &Bound<'_, PyAny>, stream: Option<&str>, end: Option<&str
     Ok(())
 }
 
-/// `(code, message, help)` of an expected error as Python objects, as the CLI renders it.
+/// `(code, message, help)` of an expected error as printable strings, lone surrogates escaped.
 fn error_parts<'py>(
     error: &Bound<'py, PyAny>,
     fallback_code: &str,
 ) -> PyResult<(Bound<'py, PyAny>, Bound<'py, PyAny>, Bound<'py, PyAny>)> {
     let py: Python<'py> = error.py();
+    let printable = |text: Bound<'py, PyString>| -> PyResult<Bound<'py, PyAny>> {
+        Ok(PyString::new(py, &escaped_text(&text)?).into_any())
+    };
     let code: Bound<'py, PyAny> = match error.getattr_opt("code")? {
-        Some(code) => code.str()?.into_any(),
+        Some(code) => printable(code.str()?)?,
         None => PyString::new(py, fallback_code).into_any(),
     };
     let message: Bound<'py, PyAny> = match error.getattr_opt("message")? {
-        Some(message) => message.str()?.into_any(),
-        None => error.str()?.into_any(),
+        Some(message) => printable(message.str()?)?,
+        None => printable(error.str()?)?,
+    };
+    let help: Bound<'py, PyAny> = match error.getattr_opt("help")? {
+        Some(help) if !help.is_none() => printable(help.str()?)?,
+        _ => py.None().into_bound(py),
     };
-    let help: Bound<'py, PyAny> = error
-        .getattr_opt("help")?
-        .unwrap_or_else(|| py.None().into_bound(py));
     Ok((code, message, help))
 }
 
@@ -144,7 +149,7 @@ fn print_expected_error(
     let help: Option<String> = if help.is_none() {
         None
     } else {
-        Some(help.str()?.extract()?)
+        Some(help.extract()?)
     };
     let text: String = format_coded_error(
         (

crates/sqlbuild-python/src/bindings/_helpers/event_exporting/constants.rs

@@ -11,6 +11,8 @@ pub(crate) const INVOCATION_THREAD_PREFIX: &str = "sqlbuild-event-exporter-";
 pub(crate) const WORKER_ATTRIBUTE: &str = "_sqlbuild_event_exporter_worker";
 /// The longest single wait of a polling delivery loop, as the Python dispatcher polled.
 pub(crate) const POLL_SECONDS: f64 = 0.01;
+/// The most one dispatch round charges a sink call, so the dispatcher's GIL waits are not counted.
+pub(crate) const CHARGE_LIMIT_SECONDS: f64 = 2.0 * POLL_SECONDS;
 /// Extra time `shutdown` waits for the dispatcher thread beyond the drain deadline.
 pub(crate) const JOIN_GRACE_SECONDS: f64 = 0.05;
 /// How long shutdown and interpreter exit wait for each stopped dispatcher thread.