feat: name failed checks in Dagster failures and make Recall runs self-describing

crates/sqlbuild-analysis/src/review_command/_helpers/sections.rs

@@ -1,6 +1,7 @@
 //! The report sections the Markdown and terminal renderers share.
 
 use crate::review_command::constants::{CLONE_START, NO_COMPARISON_MODE};
+use sqlbuild_core::command_output::main::short_invocation_id::short_invocation_id;
 use std::collections::{BTreeMap, BTreeSet};
 
 use crate::review_command::_helpers::outcome::outcome;
@@ -427,7 +428,7 @@ pub(crate) fn run_lines(report: &ReviewReport, style: Style) -> Vec<String> {
     if let Some(id) = &report.invocation_id {
         lines.push(format!(
             "Recall: {}",
-            style.code(&format!("sqb recall output {id}"))
+            style.code(&format!("sqb recall output {}", short_invocation_id(id)))
         ));
     }
     if let Some(error) = &report.error {

crates/sqlbuild-analysis/src/review_command/tests/test_render_review.rs

@@ -142,7 +142,7 @@ fn given_reports_when_rendering_markdown_then_sections_follow_the_agreed_layout(
                 "## Tests and audits\n\n- Unit tests: 2 of 5 ran (3 unaffected skipped): 2 passed, 0 failed\n- Audits: 3 ran: 3 passed, 0 failed",
                 "<details><summary>order_totals: 1 difference</summary>\n\n| Column | Metric | prod | branch | Limit |\n|---|---|---|---|---|\n| amount | rows differ | 0 | 1 | exact match |",
                 "rows prod 5, branch 5; 2 other columns compared with no difference (full comparison).",
-                "## Run details\n\n- Review schema `review_20261023t101500z_3f2a9c1_a1b2c3`: dropped\n- Downstream policy: critical (critical: `daily_revenue`)\n- Recall: `sqb recall output a1b2c3d4e5f6`",
+                "## Run details\n\n- Review schema `review_20261023t101500z_3f2a9c1_a1b2c3`: dropped\n- Downstream policy: critical (critical: `daily_revenue`)\n- Recall: `sqb recall output a1b2c3d4`",
             ],
             unexpected_fragments: vec![],
         },
@@ -212,7 +212,7 @@ fn given_reports_when_rendering_text_then_summary_ends_with_outcome() {
             "Findings\n  1. order_totals.amount values differ in 1 of 5 rows",
             "What changed\n  stg_orders (model): model SQL changed",
             "Downstream impact\n  stg_orders [ok]\n  ├── order_totals [finding]",
-            "Recall: sqb recall output a1b2c3d4e5f6",
+            "Recall: sqb recall output a1b2c3d4",
             "\nReview needs a fix: 1 finding  changed=2  built=3  compared=3  tests=2  findings=1  not_validated=1  incomplete=0\n",
         ],
         unexpected_fragments: vec!["`"],

crates/sqlbuild-analysis/src/sql_macros/main/sql_macro_parameter_spans.rs

@@ -1,28 +1,62 @@
 //! The byte spans of a SQL macro body's `@name` parameter references, as expansion reads them.
 
+use crate::sql_macros::_helpers::parsing::header::unquoted;
 use crate::sql_macros::_helpers::parsing::lexing::{Token, TokenKind, tokens};
 use crate::sql_macros::_helpers::parsing::parameters::parameter_references;
 use crate::sql_macros::_helpers::parsing::slots::prefix_start;
+use crate::sql_macros::models::{ParameterSpan, SqlMacroParameter};
 
-/// The span of each `@name` expansion replaces, and of each slot name whose column a call maps.
+/// Each `@name` span expansion replaces: aliased and plain group reads whole, other plain reads as relation and slot.
 #[must_use]
-pub fn sql_macro_parameter_spans(body: &str) -> Vec<(usize, usize)> {
+pub fn sql_macro_parameter_spans(
+    body: &str,
+    parameters: &[SqlMacroParameter],
+) -> Vec<ParameterSpan> {
     let Ok(scanned) = tokens(body) else {
         return Vec::new();
     };
-    let mut spans: Vec<(usize, usize)> = Vec::new();
+    let relation = |name: &str| -> Option<&SqlMacroParameter> {
+        parameters
+            .iter()
+            .find(|parameter| parameter.name.eq_ignore_ascii_case(name))
+            .filter(|parameter| parameter.kind.declares_relation())
+    };
+    let mut spans: Vec<ParameterSpan> = Vec::new();
     for reference in parameter_references(body, &scanned) {
         let name: usize = reference.token;
+        let declared: Option<&SqlMacroParameter> = relation(&reference.name);
+        let named_relation: Option<String> =
+            declared.map(|parameter| parameter.name.to_lowercase());
         let Some(slot) = slot_word(body, &scanned, name) else {
-            spans.push((reference.start, reference.end));
+            spans.push(ParameterSpan {
+                start: reference.start,
+                end: reference.end,
+                relation: named_relation,
+            });
             continue;
         };
-        if name >= 1 && prefix_start(body, &scanned, name - 1).is_some() {
-            spans.push((reference.start, slot.end));
+        let slot_name: String = unquoted(slot.text(body));
+        let qualified: bool = name >= 1 && prefix_start(body, &scanned, name - 1).is_some();
+        let group: bool =
+            declared.is_some_and(|parameter| parameter.kind.declares_group(&slot_name));
+        if qualified || group {
+            spans.push(ParameterSpan {
+                start: reference.start,
+                end: slot.end,
+                relation: None,
+            });
             continue;
         }
-        spans.push((reference.start, reference.end));
-        spans.push((slot.start, slot.end));
+        spans.push(ParameterSpan {
+            start: reference.start,
+            end: reference.end,
+            relation: named_relation,
+        });
+        spans.push(ParameterSpan {
+            start: slot.start,
+            end: slot.end,
+            relation: None,
+        });
     }
     spans
 }

crates/sqlbuild-analysis/src/sql_macros/models.rs

@@ -77,6 +77,11 @@ impl ParameterKind {
         groups.iter().any(|group| group.eq_ignore_ascii_case(slot))
     }
 
+    /// Whether this is a `RELATION`.
+    pub fn declares_relation(&self) -> bool {
+        matches!(self, Self::Relation { .. })
+    }
+
     /// Whether this is a `RELATION` that declares a slot called `slot`.
     pub fn declares_slot(&self, slot: &str) -> bool {
         let Self::Relation { columns, groups } = self else {
@@ -468,6 +473,15 @@ pub struct SqlMacroFinding {
     pub declared_at: SqlMacroLocation,
 }
 
+/// The byte span of one parameter reference a SQL macro body's lint replaces.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ParameterSpan {
+    pub start: usize,
+    pub end: usize,
+    /// The `RELATION` parameter this span names as a relation, as `@weekly` in `FROM @weekly`.
+    pub relation: Option<String>,
+}
+
 /// What type inference knows of an expression macro body over its declared arguments.
 #[derive(Clone, Debug)]
 pub(crate) enum BodyType {

crates/sqlbuild-core/src/command_output/constants.rs

@@ -14,6 +14,8 @@ pub const RECALL_REPLACEMENT_SCHEMA_VERSION: i64 = 2;
 pub const MAX_INVOCATION_CONTEXT_BYTES: usize = 16 * 1024;
 pub const MAX_INVOCATION_CONTEXT_DEPTH: usize = 8;
 pub const INVOCATION_CONTEXT_ENV: &str = "SQLBUILD_INVOCATION_CONTEXT_JSON";
+/// Characters of an invocation ID that output shows; any unique prefix names the invocation.
+pub const SHORT_INVOCATION_ID_LENGTH: usize = 8;
 pub(crate) const ESCAPE: u8 = 0x1b;
 pub(crate) const BELL: u8 = 0x07;
 pub(crate) const OSC_INTRODUCER: u8 = b']';

crates/sqlbuild-core/src/command_output/main/mod.rs

@@ -5,4 +5,5 @@ pub mod close_output;
 pub mod flush_idle_output;
 pub mod output_loss_summary;
 pub mod queue_output_record;
+pub mod short_invocation_id;
 pub mod validate_output_record;

crates/sqlbuild-core/src/command_output/main/short_invocation_id.rs

@@ -0,0 +1,12 @@
+//! `short_invocation_id`: the prefix of an invocation ID that output shows.
+
+use crate::command_output::constants::SHORT_INVOCATION_ID_LENGTH;
+
+/// The first [`SHORT_INVOCATION_ID_LENGTH`] characters of `invocation_id`, or all of a shorter one.
+#[must_use]
+pub fn short_invocation_id(invocation_id: &str) -> &str {
+    invocation_id
+        .char_indices()
+        .nth(SHORT_INVOCATION_ID_LENGTH)
+        .map_or(invocation_id, |(end, _)| &invocation_id[..end])
+}

crates/sqlbuild-core/src/dagster_invocation/constants.rs

@@ -33,3 +33,5 @@ pub(crate) const CODE_SPAN_PADDED_EDGES: [char; 2] = [CODE_SPAN_MARK, ' '];
 pub const DEFAULT_ASSET_KIND: &str = "asset";
 /// The default translator's description name of a node without one.
 pub const DEFAULT_ASSET_NAME: &str = "unknown";
+/// Failed checks and models a failure description names before "and N more".
+pub const MAX_SUMMARIZED_FAILURES: usize = 10;

crates/sqlbuild-core/src/dagster_invocation/main/failure_summary.rs

@@ -0,0 +1,65 @@
+//! The leading summary of a failed invocation: what failed, before how the command exited.
+
+use crate::dagster_invocation::constants::MAX_SUMMARIZED_FAILURES;
+use crate::dagster_invocation::models::{AssetFailure, FailedCheck};
+
+fn check_line(check: &FailedCheck) -> String {
+    let mut line: String = format!("{} {}", check.kind, check.name);
+    match (&check.asset_name, &check.column_name) {
+        (Some(asset), Some(column)) => line.push_str(&format!(" on {asset}.{column}")),
+        (Some(asset), None) => line.push_str(&format!(" on {asset}")),
+        (None, _) => {}
+    }
+    match check.row_count {
+        Some(1) => line.push_str(": 1 failing row"),
+        Some(count) => line.push_str(&format!(": {count} failing rows")),
+        None => line.push_str(&format!(": {}", check.status)),
+    }
+    line
+}
+
+fn asset_line(failure: &AssetFailure) -> String {
+    let mut line: String = format!("{} {}: {}", failure.kind, failure.name, failure.status);
+    if let Some(phase) = &failure.failed_phase {
+        line.push_str(&format!(" during {phase}"));
+    }
+    line
+}
+
+/// `Failed: …` followed by one line per failed check, then per failed asset, at most [`MAX_SUMMARIZED_FAILURES`] of them and `and N more`; empty when nothing is known to have failed.
+#[must_use]
+pub fn failure_summary(checks: &[FailedCheck], assets: &[AssetFailure]) -> String {
+    let lines: Vec<String> = checks
+        .iter()
+        .map(check_line)
+        .chain(assets.iter().map(asset_line))
+        .collect();
+    if lines.is_empty() {
+        return String::new();
+    }
+    let mut counts: Vec<String> = Vec::new();
+    for (count, singular, plural) in [
+        (checks.len(), "check", "checks"),
+        (assets.len(), "asset", "assets"),
+    ] {
+        match count {
+            0 => {}
+            1 => counts.push(format!("1 {singular}")),
+            _ => counts.push(format!("{count} {plural}")),
+        }
+    }
+    let mut summary: Vec<String> = vec![format!("Failed: {}", counts.join(", "))];
+    summary.extend(
+        lines
+            .iter()
+            .take(MAX_SUMMARIZED_FAILURES)
+            .map(|line| format!("  {line}")),
+    );
+    if lines.len() > MAX_SUMMARIZED_FAILURES {
+        summary.push(format!(
+            "  and {} more",
+            lines.len() - MAX_SUMMARIZED_FAILURES
+        ));
+    }
+    summary.join("\n")
+}

crates/sqlbuild-core/src/dagster_invocation/main/mod.rs

@@ -4,6 +4,7 @@ pub mod caller_json_output_path;
 pub mod dagster_name_segment;
 pub mod dagster_tag_key;
 pub mod diagnostic_tail;
+pub mod failure_summary;
 pub mod fallback_asset_description;
 pub mod has_explicit_scenario_selector;
 pub mod json_output_requested;

crates/sqlbuild-core/src/dagster_invocation/models.rs

@@ -13,3 +13,14 @@ pub struct AssetFailure {
     pub failed_sql: Option<String>,
     pub last_recorded_sql: Option<String>,
 }
+
+/// The display facts of one failed check (audit, SQL test or Python check) in an execution document.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct FailedCheck {
+    pub kind: String,
+    pub name: String,
+    pub asset_name: Option<String>,
+    pub column_name: Option<String>,
+    pub status: String,
+    pub row_count: Option<i64>,
+}

crates/sqlbuild-core/src/dagster_invocation/tests/helpers.rs

@@ -1,6 +1,20 @@
 //! Shared builders for Dagster invocation tests.
 
+use crate::dagster_invocation::models::FailedCheck;
+
 /// Owned copies of string literals.
 pub(super) fn owned(values: &[&str]) -> Vec<String> {
     values.iter().map(|value| (*value).to_owned()).collect()
 }
+
+/// A failed `not_null` audit on `model.order_id`.
+pub(super) fn not_null_audit(model: &str, row_count: Option<i64>) -> FailedCheck {
+    FailedCheck {
+        kind: "audit".to_owned(),
+        name: "not_null".to_owned(),
+        asset_name: Some(model.to_owned()),
+        column_name: Some("order_id".to_owned()),
+        status: "error".to_owned(),
+        row_count,
+    }
+}

crates/sqlbuild-core/src/dagster_invocation/tests/test_texts.rs

@@ -1,11 +1,13 @@
 //! Check names, log lines and failure text match the Dagster integration's output.
 
 use crate::dagster_invocation::main::asset_failure_text::asset_failure_text;
+use crate::dagster_invocation::main::failure_summary::failure_summary;
 use crate::dagster_invocation::main::normalize_check_name::normalize_check_name;
 use crate::dagster_invocation::main::wrap_selectors::wrap_selectors;
-use crate::dagster_invocation::models::AssetFailure;
+use crate::dagster_invocation::models::{AssetFailure, FailedCheck};
+use crate::dagster_invocation::tests::helpers::not_null_audit;
 use crate::dagster_invocation::tests::test_types::{
-    CheckNameTestCase, FailureTextTestCase, WrapTestCase,
+    CheckNameTestCase, FailureSummaryTestCase, FailureTextTestCase, WrapTestCase,
 };
 
 #[test]
@@ -111,3 +113,62 @@ fn given_failed_asset_when_formatting_then_lists_error_relation_and_sql() {
         );
     }
 }
+
+#[test]
+fn given_failed_checks_and_assets_when_summarizing_then_names_them_first_and_caps_the_list() {
+    let failed_model = AssetFailure {
+        kind: "model".to_owned(),
+        name: "customers".to_owned(),
+        status: "failed".to_owned(),
+        failed_phase: Some("execute".to_owned()),
+        ..AssetFailure::default()
+    };
+    let sql_test = FailedCheck {
+        kind: "sql_test".to_owned(),
+        name: "orders_totals".to_owned(),
+        asset_name: Some("orders".to_owned()),
+        column_name: None,
+        status: "fail".to_owned(),
+        row_count: None,
+    };
+    let many: Vec<FailedCheck> = (0..12)
+        .map(|index| index + 2)
+        .map(|index: i64| not_null_audit(&format!("orders_{index:02}"), Some(index)))
+        .collect();
+    let many_lines: String = (2..12)
+        .map(|index| {
+            format!("\n  audit not_null on orders_{index:02}.order_id: {index} failing rows")
+        })
+        .collect();
+    let test_cases = [
+        FailureSummaryTestCase {
+            description: "nothing failed",
+            checks: Vec::new(),
+            assets: Vec::new(),
+            expected_summary: String::new(),
+        },
+        FailureSummaryTestCase {
+            description: "checks come before assets, with their failing rows or status",
+            checks: vec![not_null_audit("orders", Some(1)), sql_test],
+            assets: vec![failed_model],
+            expected_summary: "Failed: 2 checks, 1 asset\n  audit not_null on orders.order_id: \
+                               1 failing row\n  sql_test orders_totals on orders: fail\n  \
+                               model customers: failed during execute"
+                .to_owned(),
+        },
+        FailureSummaryTestCase {
+            description: "a long list ends with how many more failed",
+            checks: many,
+            assets: Vec::new(),
+            expected_summary: format!("Failed: 12 checks{many_lines}\n  and 2 more"),
+        },
+    ];
+    for test_case in test_cases {
+        assert_eq!(
+            failure_summary(&test_case.checks, &test_case.assets),
+            test_case.expected_summary,
+            "{}",
+            test_case.description
+        );
+    }
+}

crates/sqlbuild-core/src/dagster_invocation/tests/test_types.rs

@@ -1,6 +1,6 @@
 //! Cases for Dagster invocation assembly.
 
-use crate::dagster_invocation::models::AssetFailure;
+use crate::dagster_invocation::models::{AssetFailure, FailedCheck};
 
 pub(super) struct ArgumentsTestCase {
     pub description: &'static str,
@@ -59,3 +59,10 @@ pub(super) struct FallbackDescriptionTestCase {
     pub materialization: Option<&'static str>,
     pub expected_description: &'static str,
 }
+
+pub(super) struct FailureSummaryTestCase {
+    pub description: &'static str,
+    pub checks: Vec<FailedCheck>,
+    pub assets: Vec<AssetFailure>,
+    pub expected_summary: String,
+}