diff --git a/crates/sqlbuild-analysis/src/review_command/_helpers/sections.rs b/crates/sqlbuild-analysis/src/review_command/_helpers/sections.rs
index ffe0f1b1e6..aa40fb8496 100644
--- a/crates/sqlbuild-analysis/src/review_command/_helpers/sections.rs
+++ b/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 {
diff --git a/crates/sqlbuild-analysis/src/review_command/tests/test_render_review.rs b/crates/sqlbuild-analysis/src/review_command/tests/test_render_review.rs
index 78ebe04dd4..fbeedf125f 100644
--- a/crates/sqlbuild-analysis/src/review_command/tests/test_render_review.rs
+++ b/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!["`"],
diff --git a/crates/sqlbuild-analysis/src/sql_macros/main/sql_macro_parameter_spans.rs b/crates/sqlbuild-analysis/src/sql_macros/main/sql_macro_parameter_spans.rs
index f216488829..22fabdf5d9 100644
--- a/crates/sqlbuild-analysis/src/sql_macros/main/sql_macro_parameter_spans.rs
+++ b/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
 }
diff --git a/crates/sqlbuild-analysis/src/sql_macros/models.rs b/crates/sqlbuild-analysis/src/sql_macros/models.rs
index cec625d827..077a802b03 100644
--- a/crates/sqlbuild-analysis/src/sql_macros/models.rs
+++ b/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 {
diff --git a/crates/sqlbuild-core/src/command_output/constants.rs b/crates/sqlbuild-core/src/command_output/constants.rs
index a80ca72aa9..a1005f6290 100644
--- a/crates/sqlbuild-core/src/command_output/constants.rs
+++ b/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']';
diff --git a/crates/sqlbuild-core/src/command_output/main/mod.rs b/crates/sqlbuild-core/src/command_output/main/mod.rs
index 473c11c79b..80dc3f228a 100644
--- a/crates/sqlbuild-core/src/command_output/main/mod.rs
+++ 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;
diff --git a/crates/sqlbuild-core/src/command_output/main/short_invocation_id.rs b/crates/sqlbuild-core/src/command_output/main/short_invocation_id.rs
new file mode 100644
index 0000000000..1a20fa8f9a
--- /dev/null
+++ b/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])
+}
diff --git a/crates/sqlbuild-core/src/dagster_invocation/constants.rs b/crates/sqlbuild-core/src/dagster_invocation/constants.rs
index ba8a89311c..ffd242aec1 100644
--- a/crates/sqlbuild-core/src/dagster_invocation/constants.rs
+++ b/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;
diff --git a/crates/sqlbuild-core/src/dagster_invocation/main/failure_summary.rs b/crates/sqlbuild-core/src/dagster_invocation/main/failure_summary.rs
new file mode 100644
index 0000000000..46a9b4cff4
--- /dev/null
+++ b/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")
+}
diff --git a/crates/sqlbuild-core/src/dagster_invocation/main/mod.rs b/crates/sqlbuild-core/src/dagster_invocation/main/mod.rs
index 4c3765b5a1..042ca2af3a 100644
--- a/crates/sqlbuild-core/src/dagster_invocation/main/mod.rs
+++ b/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;
diff --git a/crates/sqlbuild-core/src/dagster_invocation/models.rs b/crates/sqlbuild-core/src/dagster_invocation/models.rs
index d182c8d052..97c90bd0a6 100644
--- a/crates/sqlbuild-core/src/dagster_invocation/models.rs
+++ b/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>,
+}
diff --git a/crates/sqlbuild-core/src/dagster_invocation/tests/helpers.rs b/crates/sqlbuild-core/src/dagster_invocation/tests/helpers.rs
index 87dddaf7fc..05c71b5f3d 100644
--- a/crates/sqlbuild-core/src/dagster_invocation/tests/helpers.rs
+++ b/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,
+    }
+}
diff --git a/crates/sqlbuild-core/src/dagster_invocation/tests/test_texts.rs b/crates/sqlbuild-core/src/dagster_invocation/tests/test_texts.rs
index 573826ed85..3cc67f2e62 100644
--- a/crates/sqlbuild-core/src/dagster_invocation/tests/test_texts.rs
+++ b/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
+        );
+    }
+}
diff --git a/crates/sqlbuild-core/src/dagster_invocation/tests/test_types.rs b/crates/sqlbuild-core/src/dagster_invocation/tests/test_types.rs
index e63e23e7d4..7718c7ee39 100644
--- a/crates/sqlbuild-core/src/dagster_invocation/tests/test_types.rs
+++ b/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,
+}
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/failures.rs b/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/failures.rs
index f54f01aac6..cbdc4fd97f 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/failures.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/failures.rs
@@ -2,18 +2,22 @@
 
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyResult, Python};
 use pyo3::pyfunction;
-use pyo3::types::{PyDict, PyDictMethods, PyString};
+use pyo3::types::{PyDict, PyDictMethods, PyList, PyListMethods, PyString};
 use sqlbuild_core::dagster_invocation::main::asset_failure_text::asset_failure_text;
 use sqlbuild_core::dagster_invocation::main::diagnostic_tail::diagnostic_tail;
-use sqlbuild_core::dagster_invocation::models::AssetFailure;
+use sqlbuild_core::dagster_invocation::main::failure_summary::failure_summary;
+use sqlbuild_core::dagster_invocation::models::{AssetFailure, FailedCheck};
 use sqlbuild_host::entries::models::ThirdPartyHost;
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::dagster_invocation::access::{
     constant_contains, get, items, text_of,
 };
+use crate::bindings::_helpers::dagster_invocation::results::{SKIPPED_CHECK_STATUS, warn_severity};
 use crate::bindings::_helpers::records::fields::field as native_field;
 
+/// Failed checks listed in failure metadata; `failed_check_count` counts them all.
+const MAX_FAILED_CHECK_METADATA: usize = 100;
 const LIVE_FAILURE_MESSAGE: &str =
     "SQLBuild resource attempt failed; see final execution diagnostics";
 const UNKNOWN: &str = "unknown";
@@ -46,6 +50,43 @@ fn asset_failure(asset: &Bound<'_, PyAny>) -> PyResult<AssetFailure> {
     })
 }
 
+/// A check of the execution document that failed with error severity, or `None`.
+fn failed_check(check: &Bound<'_, PyAny>) -> PyResult<Option<FailedCheck>> {
+    if get(check, "passed")?.is_truthy()?
+        || get(check, "status")?.eq(SKIPPED_CHECK_STATUS)?
+        || warn_severity(check)?
+    {
+        return Ok(None);
+    }
+    let row_count: Bound<'_, PyAny> = get(check, "row_count")?;
+    Ok(Some(FailedCheck {
+        kind: text_of(check, "kind")?,
+        name: text_of(check, "name")?,
+        asset_name: optional_text(&get(check, "asset_name")?)?,
+        column_name: optional_text(&get(check, "attached_column_name")?)?,
+        status: text_of(check, "status")?,
+        row_count: row_count.extract()?,
+    }))
+}
+
+fn failed_check_metadata<'py>(
+    py: Python<'py>,
+    checks: &[FailedCheck],
+) -> PyResult<Bound<'py, PyList>> {
+    let listed: Bound<'py, PyList> = PyList::empty(py);
+    for check in checks.iter().take(MAX_FAILED_CHECK_METADATA) {
+        let item: Bound<'py, PyDict> = PyDict::new(py);
+        item.set_item("kind", &check.kind)?;
+        item.set_item("name", &check.name)?;
+        item.set_item("asset", check.asset_name.as_deref())?;
+        item.set_item("column", check.column_name.as_deref())?;
+        item.set_item("status", &check.status)?;
+        item.set_item("failing_rows", check.row_count)?;
+        listed.append(item)?;
+    }
+    Ok(listed)
+}
+
 /// The Dagster failure description and metadata of a failed invocation.
 #[pyfunction]
 #[pyo3(signature = (**options))]
@@ -85,13 +126,14 @@ pub(crate) fn dagster_failure<'py>(
     let metadata: Bound<'py, PyDict> = PyDict::new(py);
     metadata.set_item("command", &command_text)?;
     metadata.set_item("project_dir", project_dir.str()?)?;
-    metadata.set_item("selection", selection.join(" "))?;
+    metadata.set_item("selection_count", selection.len())?;
     metadata.set_item("selector_file", selector_file)?;
     let mut description: String =
         format!("SQLBuild CLI command failed with exit code {returncode}: {command_text}");
     match execution_payload {
         Some(payload) => {
             let mut incomplete: Vec<String> = Vec::new();
+            let mut failures: Vec<AssetFailure> = Vec::new();
             let mut details: Vec<String> = Vec::new();
             let limits: Bound<'py, PyDict> = PyDict::new(py);
             for asset in items(&payload, "assets")? {
@@ -102,13 +144,24 @@ pub(crate) fn dagster_failure<'py>(
                 let failure: AssetFailure = asset_failure(&asset)?;
                 incomplete.push(format!("{}:{} ({status})", failure.kind, failure.name));
                 details.push(asset_failure_text(&failure));
+                failures.push(failure);
                 let microbatch: Bound<'py, PyAny> = get(&asset, "microbatch")?;
                 if microbatch.is_instance_of::<PyDict>() && !get(&microbatch, "limit")?.is_none() {
                     limits.set_item(text_of(&asset, "name")?, microbatch)?;
                 }
             }
+            let mut checks: Vec<FailedCheck> = Vec::new();
+            for check in items(&payload, "checks")? {
+                checks.extend(failed_check(&check)?);
+            }
             metadata.set_item("execution_status", text_of(&payload, "status")?)?;
             metadata.set_item("incomplete_assets", incomplete.join(", "))?;
+            metadata.set_item("failed_check_count", checks.len())?;
+            metadata.set_item("failed_checks", failed_check_metadata(py, &checks)?)?;
+            let summary: String = failure_summary(&checks, &failures);
+            if !summary.is_empty() {
+                description = format!("{summary}\n\n{description}");
+            }
             if !limits.is_empty() {
                 metadata.set_item("microbatch_limits", limits)?;
             }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/results.rs b/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/results.rs
index 0ccbc4bfee..ed1d353082 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/results.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/results.rs
@@ -29,7 +29,7 @@ const MATERIALIZATION: &str = "materialization";
 const CHECK: &str = "check";
 const AUDIT_CHECK_KIND: &str = "audit";
 /// The status of an audit that did not run; Dagster gets no check result for it.
-const SKIPPED_CHECK_STATUS: &str = "skipped";
+pub(crate) const SKIPPED_CHECK_STATUS: &str = "skipped";
 
 /// DAG nodes upstream first, or in artifact order when edges form a cycle.
 pub(crate) fn sorted_nodes<'py>(dag: &Bound<'py, PyAny>) -> PyResult<Vec<Bound<'py, PyAny>>> {
@@ -223,7 +223,8 @@ struct CheckFilter<'a> {
     emitted_asset_paths: Option<&'a HashSet<AssetPath>>,
 }
 
-fn warn_severity(check: &Bound<'_, PyAny>) -> PyResult<bool> {
+/// Whether a check result is a warning rather than an error.
+pub(crate) fn warn_severity(check: &Bound<'_, PyAny>) -> PyResult<bool> {
     let py: Python<'_> = check.py();
     let warning: String = constant(py, ThirdPartyHost::WarningCheckSeverity)?.extract()?;
     let insufficient: String = constant(py, ThirdPartyHost::InsufficientCheckStatus)?.extract()?;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_store.rs b/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_store.rs
index 4cc6440c5d..45d6700ca0 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_store.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_store.rs
@@ -43,8 +43,9 @@ use sqlbuild_recall::schema::models::{
     ColumnSpec, DdlOptions, Dialect, ObservedSchema, SchemaAge, StoreVersion, Table, VersionPlan,
 };
 use sqlbuild_recall::settings::constants::CLICKHOUSE_DATABASE_KEY;
-use sqlbuild_recall::store::constants::LAST_INVOCATION_SELECTOR;
+use sqlbuild_recall::store::constants::{LAST_INVOCATION_SELECTOR, LISTED_INVOCATION_CANDIDATES};
 use sqlbuild_recall::store::errors::{StoreError, StoreErrorKind};
+use sqlbuild_recall::store::main::chosen_invocation::chosen_invocation;
 use sqlbuild_recall::store::models::{
     DescribedColumn, DescribedRelation, InsertOutcome, OutputTranscript, QueryResult,
 };
@@ -897,34 +898,14 @@ impl ClickHouseReader {
         } else {
             format!(
                 "SELECT DISTINCT invocation_id FROM {started} \
-                 WHERE startsWith(invocation_id, {}) ORDER BY invocation_id LIMIT 3",
-                literal(selector)
+                 WHERE startsWith(invocation_id, {}) ORDER BY invocation_id LIMIT {}",
+                literal(selector),
+                LISTED_INVOCATION_CANDIDATES + 1
             )
         };
         let found = self.store.query(py, &sql, "find the invocation")?;
         let candidates: Vec<String> = found.rows.iter().map(|row| text(row.first())).collect();
-        if let Some(exact) = candidates.iter().find(|candidate| *candidate == selector) {
-            return Ok(exact.clone());
-        }
-        match candidates.as_slice() {
-            [] => Err(StoreError {
-                kind: StoreErrorKind::NotFound,
-                message: if selector == LAST_INVOCATION_SELECTOR {
-                    "no invocations are recorded yet".to_owned()
-                } else {
-                    format!("no recorded invocation matches '{selector}'")
-                },
-            }),
-            [only] => Ok(only.clone()),
-            several => Err(StoreError {
-                kind: StoreErrorKind::Ambiguous,
-                message: format!(
-                    "'{selector}' matches several invocations: {}{}",
-                    several[..2].join(", "),
-                    if several.len() > 2 { ", …" } else { "" }
-                ),
-            }),
-        }
+        chosen_invocation(selector, &candidates)
     }
 
     /// One invocation's exact output bytes, deduplicated by sequence, and its completeness.
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/recall/command.rs b/crates/sqlbuild-python/src/bindings/_helpers/recall/command.rs
index 5f9ca45f15..2e4b45435b 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/command.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/command.rs
@@ -5,6 +5,7 @@ use std::time::Instant;
 
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyErr, PyResult, Python};
 use pyo3::types::{PyDict, PyDictMethods, PyStringMethods};
+use sqlbuild_core::command_output::main::short_invocation_id::short_invocation_id;
 use sqlbuild_recall::records::models::OutputStream;
 use sqlbuild_recall::schema::constants::VIEW_NAMES;
 use sqlbuild_recall::schema::main::ddl_statements::ddl_statements;
@@ -502,7 +503,8 @@ fn run_output(py: Python<'_>, recall: &RecallRequest) -> PyResult<i32> {
         write_stderr(
             py,
             &format!(
-                "note: output of {invocation_id} may be incomplete: {}\n",
+                "note: output of {} may be incomplete: {}\n",
+                short_invocation_id(invocation_id),
                 problems.join("; ")
             ),
         )?;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs b/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs
index b8c8b12949..004722f9f0 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs
@@ -114,3 +114,7 @@ pub(crate) const RECORD_PARAMETER: &str = "record";
 /// How Recall timestamps mark UTC, and the offset `datetime.fromisoformat` reads.
 pub(crate) const UTC_SUFFIX: &str = "Z";
 pub(crate) const UTC_OFFSET: &str = "+00:00";
+/// The command whose flags name its Recall `command_mode`.
+pub(crate) const DEBUG_COMMAND: &str = "debug";
+/// The namespace attribute of `sqb debug --no-connection`.
+pub(crate) const NO_CONNECTION_ATTRIBUTE: &str = "no_connection";
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/recall/rendering.rs b/crates/sqlbuild-python/src/bindings/_helpers/recall/rendering.rs
index a0a0fe946f..84b0e96a13 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/rendering.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/rendering.rs
@@ -161,7 +161,9 @@ pub(crate) fn render_result(
 fn relation_note(name: &str) -> &'static str {
     match name {
         "invocations" => {
-            "one row per invocation; status is completed, failed, running, abandoned or unknown"
+            "one row per invocation; status is completed, failed, running, abandoned or unknown; \
+             origin is integration, ci or local; integration, integration_job, \
+             integration_run_id and integration_retry come from external_context"
         }
         "statements" => "one row per statement; outcome no_terminal means no terminal fact arrived",
         "audit_results" => "one row per evaluated audit",
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/recall/session.rs b/crates/sqlbuild-python/src/bindings/_helpers/recall/session.rs
index 9dadbec08f..7ab0b0644e 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/session.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/session.rs
@@ -11,7 +11,9 @@ use sqlbuild_cli::dispatch::main::closing_status::closing_status;
 use sqlbuild_cli::dispatch::models::CommandFacts;
 use sqlbuild_core::command_output::constants::INVOCATION_CONTEXT_ENV;
 use sqlbuild_host::entries::models::ExecutorHost;
+use sqlbuild_recall::records::main::command_mode::command_mode;
 use sqlbuild_recall::records::main::format_timestamp::format_timestamp;
+use sqlbuild_recall::records::main::invocation_origin::invocation_origin;
 use sqlbuild_recall::records::main::sha256_hex::sha256_hex;
 use sqlbuild_recall::records::models::{InvocationContext, OutputStream, RowEnvelope};
 use sqlbuild_recall::schema::main::create_database_statement::create_database_statement;
@@ -39,9 +41,10 @@ use crate::bindings::_helpers::recall::clickhouse_store::{
 };
 use crate::bindings::_helpers::recall::console::write_stderr;
 use crate::bindings::_helpers::recall::constants::{
-    CLOSE_POLL_MS, DDL_NOTICE_PREFIX, GIT_DIRTY_ENVIRONMENT, GIT_REVISION_ENVIRONMENT,
-    HEARTBEAT_ENVIRONMENT, HOST_TOKEN_LENGTH, MINIMUM_HEARTBEAT_SECONDS, NOTICE_INTERVAL_SECONDS,
-    QUARANTINE_START_WAIT_MS, RECALL_HISTORY, STDOUT_STREAM_INDEX,
+    CLOSE_POLL_MS, DDL_NOTICE_PREFIX, DEBUG_COMMAND, GIT_DIRTY_ENVIRONMENT,
+    GIT_REVISION_ENVIRONMENT, HEARTBEAT_ENVIRONMENT, HOST_TOKEN_LENGTH, MINIMUM_HEARTBEAT_SECONDS,
+    NO_CONNECTION_ATTRIBUTE, NOTICE_INTERVAL_SECONDS, QUARANTINE_START_WAIT_MS, RECALL_HISTORY,
+    STDOUT_STREAM_INDEX,
 };
 use crate::bindings::_helpers::recall::events::escaped_text;
 use crate::bindings::_helpers::recall::replacements::{RecallReplacements, in_replacement_sink};
@@ -71,8 +74,12 @@ fn text_value(value: &Bound<'_, PyAny>) -> PyResult<Option<String>> {
     escaped_text(&value.str()?).map(Some)
 }
 
+fn attribute<'py>(args: &Bound<'py, PyAny>, name: &str) -> PyResult<Bound<'py, PyAny>> {
+    args.getattr(name)
+}
+
 fn list_attribute(args: &Bound<'_, PyAny>, name: &str) -> PyResult<Vec<String>> {
-    let value: Bound<'_, PyAny> = args.getattr(name)?;
+    let value: Bound<'_, PyAny> = attribute(args, name)?;
     let mut items: Vec<String> = Vec::new();
     if value.is_none() {
         return Ok(items);
@@ -231,7 +238,12 @@ fn invocation_facts(
         settings.destination_database.as_deref().unwrap_or_default(),
         settings.destination_schema.as_deref().unwrap_or_default()
     );
+    let external_context: Option<String> = external_context(py)?;
+    let no_connection: bool = facts.command.as_deref() == Some(DEBUG_COMMAND)
+        && attribute(args, NO_CONNECTION_ATTRIBUTE)?.is_truthy()?;
     Ok(InvocationContext {
+        origin: Some(invocation_origin(external_context.is_some(), environment).to_owned()),
+        command_mode: command_mode(facts.command.as_deref(), no_connection).map(str::to_owned),
         command: facts.command.clone(),
         subcommand: facts.subcommand.clone(),
         selection: list_attribute(args, "select")?,
@@ -248,7 +260,7 @@ fn invocation_facts(
         git_dirty: recorded_git_dirty(),
         host_token: host_token(),
         pid: Some(std::process::id()),
-        external_context: external_context(py)?,
+        external_context,
         recall_filter_active: false,
         argv_digest: None,
     })
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs b/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs
index e3317161d0..78155a1522 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs
@@ -8,8 +8,9 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use pyo3::exceptions::{PyException, PyValueError};
 use pyo3::prelude::{Bound, IntoPyObject, PyAny, PyAnyMethods, PyErr, PyResult, Python};
 use pyo3::types::{PyDict, PyDictMethods, PyString, PyTuple};
+use sqlbuild_analysis::sql_macros::main::parse_sql_macro::parse_sql_macro;
 use sqlbuild_analysis::sql_macros::main::sql_macro_body_form::sql_macro_body_form;
-use sqlbuild_analysis::sql_macros::models::SqlMacroForm;
+use sqlbuild_analysis::sql_macros::models::{SqlMacroForm, SqlMacroParameter};
 use sqlbuild_core::panics::main::catch_compiler_panic::catch_compiler_panic;
 use sqlbuild_core::text::main::python_isspace::python_isspace;
 use sqlbuild_core::text::main::python_str_repr::python_str_repr;
@@ -755,9 +756,13 @@ fn plan_bodies(
         ranges = sql_body_ranges(text, headers);
     }
     let sql_test: bool = headers.iter().any(|header| header.kind == SQL_TEST_KIND);
-    let neutralize: Option<fn(&str) -> Neutralized> = if sql_macro {
-        Some(neutralize_sql_macro_parameters)
-    } else if sql_test {
+    let parameters: Vec<SqlMacroParameter> = if sql_macro {
+        sql_macro_parameters(file)
+    } else {
+        Vec::new()
+    };
+    let neutralize_macro = |body: &str| neutralize_sql_macro_parameters(body, &parameters);
+    let other: Option<fn(&str) -> Neutralized> = if sql_test {
         Some(neutralize_test_parameters)
     } else {
         neutralizer(&BodyRole {
@@ -766,6 +771,13 @@ fn plan_bodies(
             in_hook_directory,
         })
     };
+    let neutralize: Option<&dyn Fn(&str) -> Neutralized> = if sql_macro {
+        Some(&neutralize_macro)
+    } else {
+        other
+            .as_ref()
+            .map(|function| function as &dyn Fn(&str) -> Neutralized)
+    };
     let mut bodies: Vec<PlannedBody> = ranges
         .into_iter()
         .map(|range| plan_body(text, range, neutralize, compiled))
@@ -838,7 +850,15 @@ pub(crate) fn prepare_single_body(request: &SingleBodyRequest<'_, '_>) -> PyResu
     let dialect: &str = dialect;
     let py: Python<'_> = file_path.py();
     let text: Vec<char> = contents.chars().collect();
-    let planned: PlannedBody = plan_body(&text, range, neutralizer(role), None);
+    let neutralize: Option<fn(&str) -> Neutralized> = neutralizer(role);
+    let planned: PlannedBody = plan_body(
+        &text,
+        range,
+        neutralize
+            .as_ref()
+            .map(|function| function as &dyn Fn(&str) -> Neutralized),
+        None,
+    );
     let (expanded, passes) = match planned.source {
         BodySource::Expand => {
             let expansion: AuthoredExpansion =
@@ -895,10 +915,22 @@ fn not_prepared(py: Python<'_>, path: &Path) -> PyResult<PyErr> {
     ))
 }
 
+/// The declared parameters of a SQL macro file, or none when its header does not parse.
+fn sql_macro_parameters(file: &LintFile) -> Vec<SqlMacroParameter> {
+    let name: String = file
+        .path
+        .file_stem()
+        .map(|stem| stem.to_string_lossy().into_owned())
+        .unwrap_or_default();
+    parse_sql_macro(&name, &file.path.to_string_lossy(), &file.contents)
+        .map(|definition| definition.parameters)
+        .unwrap_or_default()
+}
+
 fn plan_body(
     text: &[char],
     (mut start, end): (usize, usize),
-    neutralize: Option<fn(&str) -> Neutralized>,
+    neutralize: Option<&dyn Fn(&str) -> Neutralized>,
     compiled: Option<&CompiledExpansion>,
 ) -> PlannedBody {
     let body: String = text[start..end].iter().collect();
diff --git a/crates/sqlbuild-recall/src/records/constants.rs b/crates/sqlbuild-recall/src/records/constants.rs
index a1a4494ff7..d30b2fe176 100644
--- a/crates/sqlbuild-recall/src/records/constants.rs
+++ b/crates/sqlbuild-recall/src/records/constants.rs
@@ -45,6 +45,33 @@ pub(crate) const MARCH_BASED_JANUARY_INDEX: i64 = 10;
 pub const DROPPED_BEFORE_COLUMN: &str = "dropped_before";
 /// Initial capacity of the buffer a row's content hash is computed over.
 pub(crate) const CONTENT_HASH_BUFFER_BYTES: usize = 1_024;
+/// The origin of an invocation an integration such as Dagster started.
+pub const INTEGRATION_ORIGIN: &str = "integration";
+/// The origin of an invocation a CI system ran.
+pub const CI_ORIGIN: &str = "ci";
+/// The origin of every other invocation.
+pub const LOCAL_ORIGIN: &str = "local";
+/// Environment variables a CI system sets; any non-empty value other than `false` or `0` marks a CI run.
+pub const CI_ENVIRONMENT_VARIABLES: [&str; 10] = [
+    "CI",
+    "GITHUB_ACTIONS",
+    "GITLAB_CI",
+    "BUILDKITE",
+    "JENKINS_URL",
+    "CIRCLECI",
+    "TF_BUILD",
+    "TEAMCITY_VERSION",
+    "BITBUCKET_BUILD_NUMBER",
+    "CODEBUILD_BUILD_ID",
+];
+/// Values of a CI variable that do not mark a CI run, besides an empty one.
+pub const CI_DISABLED_VALUES: [&str; 2] = ["false", "0"];
+/// The `command_mode` of `sqb debug --no-connection`: configuration checks only.
+pub const CONFIG_ONLY_MODE: &str = "config_only";
+/// The `command_mode` of `sqb debug`: configuration checks and a connection test.
+pub const CONFIG_AND_CONNECTION_MODE: &str = "config_and_connection";
+/// The command whose mode Recall records.
+pub(crate) const DEBUG_COMMAND: &str = "debug";
 /// The output column holding the bytes a truncation gap replaces.
 pub const OMITTED_BYTES_COLUMN: &str = "omitted_bytes";
 /// The output column holding the sequences a truncation gap accounts for.
diff --git a/crates/sqlbuild-recall/src/records/main/command_mode.rs b/crates/sqlbuild-recall/src/records/main/command_mode.rs
new file mode 100644
index 0000000000..4b9686901e
--- /dev/null
+++ b/crates/sqlbuild-recall/src/records/main/command_mode.rs
@@ -0,0 +1,13 @@
+//! `command_mode`: what a diagnostic command checked, named from its flags.
+
+use crate::records::constants::{CONFIG_AND_CONNECTION_MODE, CONFIG_ONLY_MODE, DEBUG_COMMAND};
+
+/// The mode of `sqb debug`, from whether `--no-connection` was given; `None` for other commands.
+#[must_use]
+pub fn command_mode(command: Option<&str>, no_connection: bool) -> Option<&'static str> {
+    (command == Some(DEBUG_COMMAND)).then_some(if no_connection {
+        CONFIG_ONLY_MODE
+    } else {
+        CONFIG_AND_CONNECTION_MODE
+    })
+}
diff --git a/crates/sqlbuild-recall/src/records/main/invocation_origin.rs b/crates/sqlbuild-recall/src/records/main/invocation_origin.rs
new file mode 100644
index 0000000000..550d6b4f5d
--- /dev/null
+++ b/crates/sqlbuild-recall/src/records/main/invocation_origin.rs
@@ -0,0 +1,27 @@
+//! `invocation_origin`: whether an integration, a CI system or a person ran an invocation.
+
+use crate::records::constants::{
+    CI_DISABLED_VALUES, CI_ENVIRONMENT_VARIABLES, CI_ORIGIN, INTEGRATION_ORIGIN, LOCAL_ORIGIN,
+};
+
+/// `integration` when an integration supplied external context, `ci` when `environment` names a CI system, else `local`.
+#[must_use]
+pub fn invocation_origin(
+    has_external_context: bool,
+    environment: impl Fn(&str) -> Option<String>,
+) -> &'static str {
+    if has_external_context {
+        return INTEGRATION_ORIGIN;
+    }
+    let set = |name: &str| {
+        environment(name).is_some_and(|value| {
+            let value: String = value.trim().to_ascii_lowercase();
+            !value.is_empty() && !CI_DISABLED_VALUES.contains(&value.as_str())
+        })
+    };
+    if CI_ENVIRONMENT_VARIABLES.iter().any(|name| set(name)) {
+        CI_ORIGIN
+    } else {
+        LOCAL_ORIGIN
+    }
+}
diff --git a/crates/sqlbuild-recall/src/records/main/invocation_row.rs b/crates/sqlbuild-recall/src/records/main/invocation_row.rs
index 53c678fd43..dc5d54213b 100644
--- a/crates/sqlbuild-recall/src/records/main/invocation_row.rs
+++ b/crates/sqlbuild-recall/src/records/main/invocation_row.rs
@@ -69,6 +69,11 @@ pub fn invocation_row(envelope: &RowEnvelope, context: &InvocationContext) -> Re
                 boolean(context.recall_filter_active),
             ),
             ("argv_digest", optional_text(context.argv_digest.as_deref())),
+            ("origin", optional_text(context.origin.as_deref())),
+            (
+                "command_mode",
+                optional_text(context.command_mode.as_deref()),
+            ),
         ],
     )
 }
diff --git a/crates/sqlbuild-recall/src/records/main/mod.rs b/crates/sqlbuild-recall/src/records/main/mod.rs
index b16f7deafb..3ccedf2f77 100644
--- a/crates/sqlbuild-recall/src/records/main/mod.rs
+++ b/crates/sqlbuild-recall/src/records/main/mod.rs
@@ -1,4 +1,5 @@
 pub mod artifact_rows;
+pub mod command_mode;
 pub mod conflict_row;
 pub mod delivery_row;
 pub mod diagnostic_input_from_json;
@@ -8,6 +9,7 @@ pub mod event_object_row;
 pub mod event_severity;
 pub mod format_timestamp;
 pub mod heartbeat_row;
+pub mod invocation_origin;
 pub mod invocation_row;
 pub mod meta_row;
 pub mod output_row;
diff --git a/crates/sqlbuild-recall/src/records/models.rs b/crates/sqlbuild-recall/src/records/models.rs
index 40146f65be..de2e666efb 100644
--- a/crates/sqlbuild-recall/src/records/models.rs
+++ b/crates/sqlbuild-recall/src/records/models.rs
@@ -121,6 +121,10 @@ pub struct InvocationContext {
     pub external_context: Option<String>,
     pub recall_filter_active: bool,
     pub argv_digest: Option<String>,
+    /// `integration`, `ci` or `local`: where the invocation came from.
+    pub origin: Option<String>,
+    /// What a diagnostic command checked, from its flags; never its raw arguments.
+    pub command_mode: Option<String>,
 }
 
 /// The correlation identifiers of a fact.
diff --git a/crates/sqlbuild-recall/src/records/tests/test_records.rs b/crates/sqlbuild-recall/src/records/tests/test_records.rs
index 0b8f7fb19c..4e963efdea 100644
--- a/crates/sqlbuild-recall/src/records/tests/test_records.rs
+++ b/crates/sqlbuild-recall/src/records/tests/test_records.rs
@@ -3,15 +3,18 @@ use std::time::{Duration, UNIX_EPOCH};
 use crate::records::_helpers::ids::content_hash;
 use crate::records::_helpers::ids::diagnostic_record_id;
 use crate::records::_helpers::ids::output_record_id;
+use crate::records::main::command_mode::command_mode;
 use crate::records::main::diagnostic_input_from_json::diagnostic_input_from_json;
 use crate::records::main::format_timestamp::format_timestamp;
+use crate::records::main::invocation_origin::invocation_origin;
 use crate::records::main::output_row::output_row;
 use crate::records::main::parse_timestamp::parse_timestamp;
 use crate::records::models::{OutputChunk, OutputStream, RecallValue};
 use crate::records::tests::helpers::{envelope, event_row, present, row_columns, text};
 use crate::records::tests::test_types::{
-    ContentHashTestCase, DiagnosticTestCase, ParseTimestampTestCase, RecordIdTestCase,
-    RejectedInputTestCase, RowColumnsTestCase, TimestampTestCase,
+    CommandModeTestCase, ContentHashTestCase, DiagnosticTestCase, OriginTestCase,
+    ParseTimestampTestCase, RecordIdTestCase, RejectedInputTestCase, RowColumnsTestCase,
+    TimestampTestCase,
 };
 
 #[test]
@@ -274,3 +277,86 @@ fn given_incomplete_diagnostic_json_when_parsed_then_it_is_rejected() {
         );
     }
 }
+
+#[test]
+fn given_context_and_environment_when_classifying_then_the_origin_is_named() {
+    let test_cases = [
+        OriginTestCase {
+            description: "external context wins over CI",
+            has_external_context: true,
+            environment: vec![("CI", "true")],
+            expected_origin: "integration",
+        },
+        OriginTestCase {
+            description: "a CI system's variable marks CI",
+            environment: vec![("GITHUB_ACTIONS", "true")],
+            has_external_context: false,
+            expected_origin: "ci",
+        },
+        OriginTestCase {
+            description: "a URL value marks CI",
+            environment: vec![("JENKINS_URL", "https://ci.example.com/")],
+            has_external_context: false,
+            expected_origin: "ci",
+        },
+        OriginTestCase {
+            description: "false, 0 and empty values are not CI",
+            environment: vec![("CI", "false"), ("GITLAB_CI", "0"), ("BUILDKITE", " ")],
+            has_external_context: false,
+            expected_origin: "local",
+        },
+        OriginTestCase {
+            description: "no variables is local",
+            environment: Vec::new(),
+            has_external_context: false,
+            expected_origin: "local",
+        },
+    ];
+    for test_case in test_cases {
+        let environment = |name: &str| {
+            test_case
+                .environment
+                .iter()
+                .find(|(key, _)| *key == name)
+                .map(|(_, value)| (*value).to_owned())
+        };
+        assert_eq!(
+            invocation_origin(test_case.has_external_context, environment),
+            test_case.expected_origin,
+            "{}",
+            test_case.description
+        );
+    }
+}
+
+#[test]
+fn given_a_command_and_its_flags_when_naming_its_mode_then_only_debug_has_one() {
+    let test_cases = [
+        CommandModeTestCase {
+            description: "debug without a connection checks configuration only",
+            command: Some("debug"),
+            no_connection: true,
+            expected_mode: Some("config_only"),
+        },
+        CommandModeTestCase {
+            description: "debug also tests the connection",
+            command: Some("debug"),
+            no_connection: false,
+            expected_mode: Some("config_and_connection"),
+        },
+        CommandModeTestCase {
+            description: "other commands record no mode",
+            command: Some("build"),
+            no_connection: false,
+            expected_mode: None,
+        },
+    ];
+    for test_case in test_cases {
+        assert_eq!(
+            command_mode(test_case.command, test_case.no_connection),
+            test_case.expected_mode,
+            "{}",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlbuild-recall/src/records/tests/test_types.rs b/crates/sqlbuild-recall/src/records/tests/test_types.rs
index 86272de086..09bd718b05 100644
--- a/crates/sqlbuild-recall/src/records/tests/test_types.rs
+++ b/crates/sqlbuild-recall/src/records/tests/test_types.rs
@@ -44,3 +44,17 @@ pub(crate) struct DiagnosticTestCase {
     pub(crate) expected_fields: &'static str,
     pub(crate) expected_run_id: Option<&'static str>,
 }
+
+pub(crate) struct OriginTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) has_external_context: bool,
+    pub(crate) environment: Vec<(&'static str, &'static str)>,
+    pub(crate) expected_origin: &'static str,
+}
+
+pub(crate) struct CommandModeTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) command: Option<&'static str>,
+    pub(crate) no_connection: bool,
+    pub(crate) expected_mode: Option<&'static str>,
+}
diff --git a/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs b/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs
index aea4bceea2..ccf1b80752 100644
--- a/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs
+++ b/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs
@@ -6,7 +6,7 @@ use crate::schema::main::table_spec::table_spec;
 use crate::schema::models::{DdlOptions, Table};
 
 /// The columns the `invocations` pass reads from its three sources, NULL (the terminal flag 0) where a source does not give them.
-const INVOCATION_SOURCE_COLUMNS: [(&str, &str, &str); 28] = [
+const INVOCATION_SOURCE_COLUMNS: [(&str, &str, &str); 30] = [
     ("ctx_command", "Nullable(String)", "NULL"),
     ("ctx_subcommand", "Nullable(String)", "NULL"),
     ("ctx_target", "Nullable(String)", "NULL"),
@@ -21,6 +21,8 @@ const INVOCATION_SOURCE_COLUMNS: [(&str, &str, &str); 28] = [
     ("ctx_pid", "Nullable(Int64)", "NULL"),
     ("ctx_external_context", "Nullable(String)", "NULL"),
     ("ctx_recall_filter_active", "Nullable(UInt8)", "NULL"),
+    ("ctx_origin", "Nullable(String)", "NULL"),
+    ("ctx_command_mode", "Nullable(String)", "NULL"),
     ("heartbeat_at", "Nullable(DateTime64(6, 'UTC'))", "NULL"),
     ("beat_stale_at", "Nullable(DateTime64(6, 'UTC'))", "NULL"),
     ("beat_command", "Nullable(String)", "NULL"),
@@ -172,6 +174,8 @@ fn invocations(options: &DdlOptions) -> String {
             ("ctx_pid", "pid"),
             ("ctx_external_context", "external_context"),
             ("ctx_recall_filter_active", "recall_filter_active"),
+            ("ctx_origin", "origin"),
+            ("ctx_command_mode", "command_mode"),
         ],
         &context,
     );
@@ -270,6 +274,8 @@ SELECT
     terminal_error_message AS error_message,
     coalesce(context_command, started_command, first_beat_command) AS command,
     coalesce(context_subcommand, first_beat_subcommand) AS subcommand,
+    context_origin AS origin,
+    context_command_mode AS command_mode,
     context_target AS target,
     context_adapter AS adapter,
     context_project_name AS project_name,
@@ -281,6 +287,18 @@ SELECT
     context_host_token AS host_token,
     context_pid AS pid,
     context_external_context AS external_context,
+    if(isValidJSON(context_external_context),
+        JSONExtract(context_external_context, 'integration', 'name', 'Nullable(String)'), NULL)
+        AS integration,
+    if(isValidJSON(context_external_context),
+        JSONExtract(context_external_context, 'integration', 'job_name', 'Nullable(String)'), NULL)
+        AS integration_job,
+    if(isValidJSON(context_external_context),
+        JSONExtract(context_external_context, 'integration', 'run_id', 'Nullable(String)'), NULL)
+        AS integration_run_id,
+    if(isValidJSON(context_external_context),
+        JSONExtract(context_external_context, 'integration', 'retry_number', 'Nullable(Int64)'),
+        NULL) AS integration_retry,
     context_recall_filter_active AS recall_filter_active,
     last_heartbeat_at
 FROM folded",
diff --git a/crates/sqlbuild-recall/src/schema/_helpers/sqlite_views.rs b/crates/sqlbuild-recall/src/schema/_helpers/sqlite_views.rs
index 122a62434e..c679a80e08 100644
--- a/crates/sqlbuild-recall/src/schema/_helpers/sqlite_views.rs
+++ b/crates/sqlbuild-recall/src/schema/_helpers/sqlite_views.rs
@@ -287,6 +287,8 @@ projected AS (
         terminals.error_message,
         COALESCE(context.command, started.command, beats.command) AS command,
         COALESCE(context.subcommand, beats.subcommand) AS subcommand,
+        context.origin,
+        context.command_mode,
         context.target,
         context.adapter,
         context.project_name,
@@ -298,6 +300,18 @@ projected AS (
         context.host_token,
         context.pid,
         context.external_context,
+        CASE WHEN json_valid(context.external_context)
+            THEN json_extract(context.external_context, '$.integration.name') END
+            AS integration,
+        CASE WHEN json_valid(context.external_context)
+            THEN json_extract(context.external_context, '$.integration.job_name') END
+            AS integration_job,
+        CASE WHEN json_valid(context.external_context)
+            THEN json_extract(context.external_context, '$.integration.run_id') END
+            AS integration_run_id,
+        CASE WHEN json_valid(context.external_context)
+            THEN json_extract(context.external_context, '$.integration.retry_number') END
+            AS integration_retry,
         context.recall_filter_active,
         beats.last_heartbeat_at
     FROM ids
@@ -328,6 +342,8 @@ SELECT
     error_message,
     command,
     subcommand,
+    origin,
+    command_mode,
     target,
     adapter,
     project_name,
@@ -339,6 +355,10 @@ SELECT
     host_token,
     pid,
     external_context,
+    integration,
+    integration_job,
+    integration_run_id,
+    integration_retry,
     recall_filter_active,
     last_heartbeat_at
 FROM projected"
diff --git a/crates/sqlbuild-recall/src/schema/constants.rs b/crates/sqlbuild-recall/src/schema/constants.rs
index 5d95836afd..3e846c7b5d 100644
--- a/crates/sqlbuild-recall/src/schema/constants.rs
+++ b/crates/sqlbuild-recall/src/schema/constants.rs
@@ -3,7 +3,7 @@
 use crate::schema::models::{ColumnSpec, ColumnType, Table, TableSpec};
 
 /// The Recall schema version this build writes. Raise it when a column or view is added.
-pub const RECALL_SCHEMA_VERSION: i64 = 3;
+pub const RECALL_SCHEMA_VERSION: i64 = 4;
 /// Text that only older schema versions wrote into ClickHouse view definitions, with the last version that wrote it. Version 3 changed only view bodies, so view columns cannot tell version 2 views, which set `join_use_nulls`, from current ones.
 pub const SUPERSEDED_CLICKHOUSE_VIEW_MARKERS: [(&str, i64); 1] = [("join_use_nulls", 2)];
 /// The oldest SQLite library the store's views run on (ordered `group_concat`).
@@ -39,7 +39,7 @@ const CORRELATIONS: [ColumnSpec; 5] = [
     ColumnSpec::optional("statement_id", ColumnType::Text),
 ];
 
-const INVOCATION_COLUMNS: [ColumnSpec; 19] = [
+const INVOCATION_COLUMNS: [ColumnSpec; 21] = [
     ColumnSpec::optional("command", ColumnType::Text),
     ColumnSpec::optional("subcommand", ColumnType::Text),
     ColumnSpec::optional("selection", ColumnType::Json),
@@ -59,6 +59,8 @@ const INVOCATION_COLUMNS: [ColumnSpec; 19] = [
     ColumnSpec::optional("external_context", ColumnType::Json),
     ColumnSpec::optional("recall_filter_active", ColumnType::Boolean),
     ColumnSpec::optional("argv_digest", ColumnType::Text),
+    ColumnSpec::optional("origin", ColumnType::Text),
+    ColumnSpec::optional("command_mode", ColumnType::Text),
 ];
 
 const OUTPUT_COLUMNS: [ColumnSpec; 11] = [
diff --git a/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs b/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs
index efade64b1e..d43869eb41 100644
--- a/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs
+++ b/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs
@@ -154,7 +154,7 @@ fn given_clickhouse_statement_sets_when_rendered_then_every_statement_has_the_sh
                 .collect(),
             required_fragments: &[
                 "INSERT INTO `observability`.`recall_meta`",
-                "SELECT 'schema', 'schema_version', '3'",
+                "SELECT 'schema', 'schema_version', '4'",
             ],
             expected_count: 1,
         },
diff --git a/crates/sqlbuild-recall/src/schema/tests/test_version.rs b/crates/sqlbuild-recall/src/schema/tests/test_version.rs
index 0ef34b26e2..b65eba6bc7 100644
--- a/crates/sqlbuild-recall/src/schema/tests/test_version.rs
+++ b/crates/sqlbuild-recall/src/schema/tests/test_version.rs
@@ -1,7 +1,9 @@
 use std::collections::BTreeMap;
 
 use crate::schema::_helpers::clickhouse_views::clickhouse_view_bodies;
-use crate::schema::constants::{RECALL_SCHEMA_VERSION, VIEW_NAMES};
+use crate::schema::constants::{
+    RECALL_SCHEMA_VERSION, SUPERSEDED_CLICKHOUSE_VIEW_MARKERS, VIEW_NAMES,
+};
 use crate::schema::main::create_database_statement::create_database_statement;
 use crate::schema::main::schema_age::schema_age;
 use crate::schema::main::store_version_with_views::store_version_with_views;
@@ -270,10 +272,10 @@ fn given_view_definitions_when_versioned_then_views_an_older_version_wrote_make_
             expected_store: StoreVersion::Unrecorded(SchemaAge::Older),
         },
         ViewDefinitionsTestCase {
-            description: "previous views recorded as current count as the previous version",
+            description: "views that set join_use_nulls count as the version that wrote them",
             store: recorded(RECALL_SCHEMA_VERSION),
             definitions: vec![PREVIOUS_VIEW.to_owned()],
-            expected_store: recorded(RECALL_SCHEMA_VERSION - 1),
+            expected_store: recorded(SUPERSEDED_CLICKHOUSE_VIEW_MARKERS[0].1),
         },
         ViewDefinitionsTestCase {
             description: "current views keep the recorded version",
diff --git a/crates/sqlbuild-recall/src/store/constants.rs b/crates/sqlbuild-recall/src/store/constants.rs
index 83d14eb638..80bd1d4e04 100644
--- a/crates/sqlbuild-recall/src/store/constants.rs
+++ b/crates/sqlbuild-recall/src/store/constants.rs
@@ -12,6 +12,8 @@ pub const INCREMENTAL_VACUUM_PAGES: u32 = 2_048;
 pub const META_MAINTENANCE_OVERDUE_KEY: &str = "overdue";
 /// The selector naming the most recently started invocation.
 pub const LAST_INVOCATION_SELECTOR: &str = "last";
+/// Invocations an ambiguous prefix lists; one more is read to tell whether there are others.
+pub const LISTED_INVOCATION_CANDIDATES: usize = 10;
 /// The first bytes of every SQLite database file.
 pub const SQLITE_HEADER: &[u8; 16] = b"SQLite format 3\0";
 /// What a store moved aside for corruption is renamed to, before its UTC timestamp.
diff --git a/crates/sqlbuild-recall/src/store/main/chosen_invocation.rs b/crates/sqlbuild-recall/src/store/main/chosen_invocation.rs
new file mode 100644
index 0000000000..880ef2a625
--- /dev/null
+++ b/crates/sqlbuild-recall/src/store/main/chosen_invocation.rs
@@ -0,0 +1,37 @@
+//! `chosen_invocation`: the one invocation a selector names among the IDs it matched.
+
+use crate::store::constants::{LAST_INVOCATION_SELECTOR as LAST, LISTED_INVOCATION_CANDIDATES};
+use crate::store::errors::{StoreError, StoreErrorKind};
+
+/// The invocation `selector` names among `candidates`, the IDs it prefixes (at most [`LISTED_INVOCATION_CANDIDATES`] and one more, in ID order): an exact match, else the only candidate; none or several fail, several listing the candidates.
+pub fn chosen_invocation(selector: &str, candidates: &[String]) -> Result<String, StoreError> {
+    if let Some(exact) = candidates.iter().find(|candidate| *candidate == selector) {
+        return Ok(exact.clone());
+    }
+    match candidates {
+        [] => Err(StoreError::new(
+            StoreErrorKind::NotFound,
+            if selector == LAST {
+                "no invocations are recorded yet".to_owned()
+            } else {
+                format!("no recorded invocation matches '{selector}'")
+            },
+        )),
+        [only] => Ok(only.clone()),
+        several => {
+            let listed: &[String] = &several[..several.len().min(LISTED_INVOCATION_CANDIDATES)];
+            Err(StoreError::new(
+                StoreErrorKind::Ambiguous,
+                format!(
+                    "'{selector}' matches several invocations; use a longer prefix: {}{}",
+                    listed.join(", "),
+                    if several.len() > listed.len() {
+                        ", and more"
+                    } else {
+                        ""
+                    }
+                ),
+            ))
+        }
+    }
+}
diff --git a/crates/sqlbuild-recall/src/store/main/mod.rs b/crates/sqlbuild-recall/src/store/main/mod.rs
index 60a156dd05..69ccb1cf7e 100644
--- a/crates/sqlbuild-recall/src/store/main/mod.rs
+++ b/crates/sqlbuild-recall/src/store/main/mod.rs
@@ -1,4 +1,5 @@
 pub mod checkpoint_sqlite_store;
+pub mod chosen_invocation;
 pub mod describe_store;
 pub mod ensure_sqlite_schema;
 pub mod insert_rows;
diff --git a/crates/sqlbuild-recall/src/store/main/resolve_invocation.rs b/crates/sqlbuild-recall/src/store/main/resolve_invocation.rs
index 71245acf0b..b3df487e30 100644
--- a/crates/sqlbuild-recall/src/store/main/resolve_invocation.rs
+++ b/crates/sqlbuild-recall/src/store/main/resolve_invocation.rs
@@ -1,8 +1,9 @@
 //! `resolve_invocation`: the invocation an ID, unique prefix or `last` names.
 
 use crate::store::_helpers::sqlite::checked;
-use crate::store::constants::LAST_INVOCATION_SELECTOR as LAST;
-use crate::store::errors::{StoreError, StoreErrorKind};
+use crate::store::constants::{LAST_INVOCATION_SELECTOR as LAST, LISTED_INVOCATION_CANDIDATES};
+use crate::store::errors::StoreError;
+use crate::store::main::chosen_invocation::chosen_invocation;
 use crate::store::models::SqliteStore;
 
 /// The full invocation ID of `selector`; `last` is the most recently started invocation.
@@ -26,37 +27,19 @@ pub fn resolve_invocation(store: &SqliteStore, selector: &str) -> Result<String,
             store.connection.prepare(
                 "SELECT invocation_id FROM invocations \
                  WHERE substr(invocation_id, 1, length(?1)) = ?1 \
-                 ORDER BY invocation_id LIMIT 3",
+                 ORDER BY invocation_id LIMIT ?2",
             ),
             "find the invocation",
         )?;
         checked(
             statement
-                .query_map([selector], |row| row.get(0))
+                .query_map(
+                    rusqlite::params![selector, LISTED_INVOCATION_CANDIDATES + 1],
+                    |row| row.get(0),
+                )
                 .and_then(Iterator::collect),
             "find the invocation",
         )?
     };
-    if let Some(exact) = candidates.iter().find(|candidate| *candidate == selector) {
-        return Ok(exact.clone());
-    }
-    match candidates.as_slice() {
-        [] => Err(StoreError::new(
-            StoreErrorKind::NotFound,
-            if selector == LAST {
-                "no invocations are recorded yet".to_owned()
-            } else {
-                format!("no recorded invocation matches '{selector}'")
-            },
-        )),
-        [only] => Ok(only.clone()),
-        several => Err(StoreError::new(
-            StoreErrorKind::Ambiguous,
-            format!(
-                "'{selector}' matches several invocations: {}{}",
-                several[..2].join(", "),
-                if several.len() > 2 { ", …" } else { "" }
-            ),
-        )),
-    }
+    chosen_invocation(selector, &candidates)
 }
diff --git a/crates/sqlbuild-recall/src/store/tests/test_store.rs b/crates/sqlbuild-recall/src/store/tests/test_store.rs
index 27f7320f06..864ad66fc4 100644
--- a/crates/sqlbuild-recall/src/store/tests/test_store.rs
+++ b/crates/sqlbuild-recall/src/store/tests/test_store.rs
@@ -2,6 +2,7 @@ use crate::records::main::invocation_row::invocation_row;
 use crate::records::models::{InvocationContext, OutputStream, RecallValue};
 use crate::store::_helpers::truncation::truncation_note;
 use crate::store::errors::StoreErrorKind;
+use crate::store::main::chosen_invocation::chosen_invocation;
 use crate::store::main::describe_store::describe_store;
 use crate::store::main::query_store::query_store;
 use crate::store::main::read_output_transcript::read_output_transcript;
@@ -13,12 +14,12 @@ use crate::store::tests::helpers::{
 use crate::store::tests::helpers::{
     already_replaced, lock_held, nobody_else, open_under_locked_directory, quarantine_after,
 };
-#[cfg(unix)]
-use crate::store::tests::test_types::{LockedDirectoryTestCase, QuarantineTestCase};
 use crate::store::tests::test_types::{
-    NewerSchemaTestCase, QueryTestCase, ResolveInvocationTestCase, TranscriptTestCase,
-    TruncationNoteTestCase,
+    ChosenInvocationTestCase, NewerSchemaTestCase, QueryTestCase, ResolveInvocationTestCase,
+    TranscriptTestCase, TruncationNoteTestCase,
 };
+#[cfg(unix)]
+use crate::store::tests::test_types::{LockedDirectoryTestCase, QuarantineTestCase};
 
 #[test]
 fn given_ids_when_resolved_then_prefixes_last_and_ambiguity_are_exact() {
@@ -273,3 +274,35 @@ fn given_truncated_output_when_noted_then_sizes_read_naturally() {
         );
     }
 }
+
+#[test]
+fn given_an_ambiguous_prefix_when_chosen_then_the_message_lists_the_candidates() {
+    let listed: String = (0..10)
+        .map(|index| format!("abc{index:02}"))
+        .collect::<Vec<_>>()
+        .join(", ");
+    let test_cases = [
+        ChosenInvocationTestCase {
+            description: "every candidate is listed",
+            candidates: vec!["abc1".to_owned(), "abc2".to_owned()],
+            expected_message: "'abc' matches several invocations; use a longer prefix: abc1, abc2"
+                .to_owned(),
+        },
+        ChosenInvocationTestCase {
+            description: "a long list says there are more",
+            candidates: (0..11).map(|index| format!("abc{index:02}")).collect(),
+            expected_message: format!(
+                "'abc' matches several invocations; use a longer prefix: {listed}, and more"
+            ),
+        },
+    ];
+
+    for test_case in test_cases {
+        assert_eq!(
+            chosen_invocation("abc", &test_case.candidates).map_err(|error| error.message),
+            Err(test_case.expected_message),
+            "{}",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlbuild-recall/src/store/tests/test_types.rs b/crates/sqlbuild-recall/src/store/tests/test_types.rs
index dffbb076a4..4b16e167d9 100644
--- a/crates/sqlbuild-recall/src/store/tests/test_types.rs
+++ b/crates/sqlbuild-recall/src/store/tests/test_types.rs
@@ -15,6 +15,12 @@ pub(crate) struct ResolveInvocationTestCase {
     pub(crate) expected_invocation: Result<&'static str, StoreErrorKind>,
 }
 
+pub(crate) struct ChosenInvocationTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) candidates: Vec<String>,
+    pub(crate) expected_message: String,
+}
+
 pub(crate) struct QueryTestCase {
     pub(crate) description: &'static str,
     pub(crate) sql: &'static str,
diff --git a/crates/sqlbuild-rules/src/lint_text/main/neutralize_sql_macro_parameters.rs b/crates/sqlbuild-rules/src/lint_text/main/neutralize_sql_macro_parameters.rs
index cc1727b297..c86849ed63 100644
--- a/crates/sqlbuild-rules/src/lint_text/main/neutralize_sql_macro_parameters.rs
+++ b/crates/sqlbuild-rules/src/lint_text/main/neutralize_sql_macro_parameters.rs
@@ -1,22 +1,34 @@
 //! `neutralize_sql_macro_parameters`: a SQL macro body's `@name` parameters replaced by sentinels.
 
+use std::collections::HashMap;
+
 use sqlbuild_analysis::sql_macros::main::sql_macro_parameter_spans::sql_macro_parameter_spans;
+use sqlbuild_analysis::sql_macros::models::{ParameterSpan, SqlMacroParameter};
 
 use crate::lint_text::_helpers::sites::{ByteSite, assemble, sentinel};
 use crate::lint_text::constants::SQL_MACRO_PARAMETER_SENTINEL_PREFIX;
 use crate::lint_text::models::Neutralized;
 
-/// Replace each parameter reference expansion would replace, and nothing else, with a sentinel.
-pub fn neutralize_sql_macro_parameters(body: &str) -> Neutralized {
-    let byte_sites: Vec<ByteSite> = sql_macro_parameter_spans(body)
+/// Replace each parameter reference with a sentinel; one `RELATION` parameter's relation spans share one.
+pub fn neutralize_sql_macro_parameters(
+    body: &str,
+    parameters: &[SqlMacroParameter],
+) -> Neutralized {
+    let mut relations: HashMap<String, String> = HashMap::new();
+    let mut next: usize = 0;
+    let mut fresh = || {
+        let made: String = sentinel(SQL_MACRO_PARAMETER_SENTINEL_PREFIX, next);
+        next += 1;
+        made
+    };
+    let byte_sites: Vec<ByteSite> = sql_macro_parameter_spans(body, parameters)
         .into_iter()
-        .enumerate()
-        .map(|(index, (start, end))| {
-            (
-                start,
-                end,
-                sentinel(SQL_MACRO_PARAMETER_SENTINEL_PREFIX, index),
-            )
+        .map(|span: ParameterSpan| {
+            let replacement: String = match span.relation {
+                Some(relation) => relations.entry(relation).or_insert_with(&mut fresh).clone(),
+                None => fresh(),
+            };
+            (span.start, span.end, replacement)
         })
         .collect();
     assemble(body, byte_sites)
diff --git a/crates/sqlbuild-rules/src/lint_text/tests/test_lint_text.rs b/crates/sqlbuild-rules/src/lint_text/tests/test_lint_text.rs
index 001281c062..0139da59b2 100644
--- a/crates/sqlbuild-rules/src/lint_text/tests/test_lint_text.rs
+++ b/crates/sqlbuild-rules/src/lint_text/tests/test_lint_text.rs
@@ -17,9 +17,10 @@ use crate::lint_text::models::{ExpansionSpan, FormatFile, LintSqlBody, LintSqlOp
 use crate::lint_text::tests::helpers::directive_finding;
 use crate::lint_text::tests::test_types::{
     Finding, FormatBodiesTestCase, LintBodiesTestCase, NeutralizeTestCase, ParameterTestCase,
-    RestoreTestCase, SitePatternTestCase, SuppressionTestCase, TokenAtTestCase,
+    RestoreTestCase, SitePatternTestCase, SlotReadTestCase, SuppressionTestCase, TokenAtTestCase,
     UnwrappedExpressionTestCase,
 };
+use sqlbuild_analysis::sql_macros::main::parse_sql_macro::parse_sql_macro;
 
 const SUPPRESSED_FILE: &str =
     "-- sqb: ignore SQBRSQL005 because kept\n\n-- note\nselect 1\n-- sqb: ignore nope\nselect 2\n";
@@ -144,25 +145,25 @@ fn given_generated_parameters_when_neutralizing_then_only_bare_parameters_are_re
         ParameterTestCase {
             description: "SQL macro parameters followed by words and a parenthesis",
             body: "@value IS NULL OR TRIM(@value) = ''",
-            neutralize: neutralize_sql_macro_parameters,
+            neutralize: |body| neutralize_sql_macro_parameters(body, &[]),
             expected_text: "__sqlbuild_macro_parameter_0__ IS NULL OR TRIM(__sqlbuild_macro_parameter_1__) = ''",
         },
         ParameterTestCase {
             description: "SQL macro parameters before IN and a parenthesised predicate",
             body: "CASE WHEN @x IN ('a', 'b') THEN 1 END + CASE WHEN @x IS NULL AND (@y IS NULL) THEN 1 END",
-            neutralize: neutralize_sql_macro_parameters,
+            neutralize: |body| neutralize_sql_macro_parameters(body, &[]),
             expected_text: "CASE WHEN __sqlbuild_macro_parameter_0__ IN ('a', 'b') THEN 1 END + CASE WHEN __sqlbuild_macro_parameter_1__ IS NULL AND (__sqlbuild_macro_parameter_2__ IS NULL) THEN 1 END",
         },
         ParameterTestCase {
             description: "SQL macro slot reads hide the slot name, whole after an alias",
             body: "SELECT s.@src.amount AS amount, @src.id, @src.* FROM @src AS s",
-            neutralize: neutralize_sql_macro_parameters,
+            neutralize: |body| neutralize_sql_macro_parameters(body, &[]),
             expected_text: "SELECT s.__sqlbuild_macro_parameter_0__ AS amount, __sqlbuild_macro_parameter_1__.__sqlbuild_macro_parameter_2__, __sqlbuild_macro_parameter_3__.* FROM __sqlbuild_macro_parameter_4__ AS s",
         },
         ParameterTestCase {
             description: "SQL macro calls, enum references and quoted text keep their at-signs",
             body: "@sql.other(@v) + @enum(\"s\").A + LENGTH('@v') + @@x",
-            neutralize: neutralize_sql_macro_parameters,
+            neutralize: |body| neutralize_sql_macro_parameters(body, &[]),
             expected_text: "@sql.other(__sqlbuild_macro_parameter_0__) + @enum(\"s\").A + LENGTH('@v') + @@x",
         },
         ParameterTestCase {
@@ -190,6 +191,43 @@ fn given_generated_parameters_when_neutralizing_then_only_bare_parameters_are_re
     }
 }
 
+#[test]
+fn given_relation_slot_reads_when_neutralizing_then_they_stand_as_their_expansion_does() {
+    let header: &str = "MACRO (\n  description \"Weekly rates\",\n  \
+         arguments (weekly RELATION (keys COLUMNS, refund_count BIGINT), cutoff BIGINT),\n);\n\n";
+    let test_cases = [
+        SlotReadTestCase {
+            description: "a plain group read is one bare value",
+            body: "SELECT @weekly.keys, orders FROM counted GROUP BY @weekly.keys",
+            expected_text: "SELECT __sqlbuild_macro_parameter_0__, orders FROM counted GROUP BY __sqlbuild_macro_parameter_1__",
+        },
+        SlotReadTestCase {
+            description: "plain single-slot and undeclared reads go through the one relation",
+            body: "SELECT @weekly.refund_count, @weekly.week FROM @weekly WHERE @cutoff > @cutoff",
+            expected_text: "SELECT __sqlbuild_macro_parameter_0__.__sqlbuild_macro_parameter_1__, __sqlbuild_macro_parameter_0__.__sqlbuild_macro_parameter_2__ FROM __sqlbuild_macro_parameter_0__ WHERE __sqlbuild_macro_parameter_3__ > __sqlbuild_macro_parameter_4__",
+        },
+        SlotReadTestCase {
+            description: "aliased reads stay whole after their alias",
+            body: "SELECT s.@weekly.keys, s.@weekly.refund_count FROM @weekly AS s",
+            expected_text: "SELECT s.__sqlbuild_macro_parameter_0__, s.__sqlbuild_macro_parameter_1__ FROM __sqlbuild_macro_parameter_2__ AS s",
+        },
+    ];
+
+    for test_case in test_cases {
+        let source: String = format!("{header}{}\n", test_case.body);
+        let parameters = parse_sql_macro("weekly_rates", "macros/sql/weekly_rates.sql", &source)
+            .map(|definition| definition.parameters)
+            .unwrap_or_default();
+
+        assert_eq!(
+            neutralize_sql_macro_parameters(test_case.body, &parameters).text,
+            test_case.expected_text,
+            "{}",
+            test_case.description
+        );
+    }
+}
+
 #[test]
 fn given_formatted_sql_when_restoring_then_each_sentinel_must_appear_once()
 -> Result<(), LintTextError> {
diff --git a/crates/sqlbuild-rules/src/lint_text/tests/test_types.rs b/crates/sqlbuild-rules/src/lint_text/tests/test_types.rs
index f4f2481ba4..60c28b778c 100644
--- a/crates/sqlbuild-rules/src/lint_text/tests/test_types.rs
+++ b/crates/sqlbuild-rules/src/lint_text/tests/test_types.rs
@@ -8,6 +8,12 @@ pub(crate) struct NeutralizeTestCase {
     pub(crate) expected_sites: Vec<(&'static str, usize, usize)>,
 }
 
+pub(crate) struct SlotReadTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) body: &'static str,
+    pub(crate) expected_text: &'static str,
+}
+
 pub(crate) struct ParameterTestCase {
     pub(crate) description: &'static str,
     pub(crate) body: &'static str,
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md b/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md
index 2a97cece0e..db4f9017dc 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md
@@ -102,6 +102,8 @@ further processing, not just to print rows more compactly.
 Every command except `sqb recall` is recorded in the store of the target it runs against,
 including `sqb debug` and failed commands. To check a shared target, prefer `sqb recall` reads:
 anything else you run from your machine with `--target <shared>` appears in that target's history.
+`invocations.origin` (`integration`, `ci` or `local`) and `command_mode` tell such local checks
+from scheduled runs; filter on them or on `command` when looking for scheduled runs.
 
 Recall creates and upgrades its own store (database, tables, views and schema version) on the first
 write, so there is usually nothing to set up, even for a new shared store. Run the statements from
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/recall.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/recall.md
index b58f94671e..dfb94e424f 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/recall.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/recall.md
@@ -124,6 +124,10 @@ sqb recall output last --tail 20             # only the last 20 lines
 | `--tail` | Print only the last N lines |
 | `--target` | Target whose store to read |
 
+Output shows an invocation ID by its first eight characters; pass that or any other unique prefix.
+A prefix that matches several invocations fails with `C973` and lists them, so you can pass a longer
+one.
+
 Where output past [`max_output_bytes`](../concepts/observability/recall.md#output-limit) was left
 out, a note line marks the gap in place. Output that Recall could not record completely for any
 other reason is followed by a note on stderr naming the gap.
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/observability/recall.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/observability/recall.md
index b16583b2b9..aa69e1b5b4 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/observability/recall.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/observability/recall.md
@@ -11,6 +11,7 @@ Online: https://sqlbuild.com/docs/concepts/observability/recall/
 - An investigation, start to finish
 - What is recorded
 - Git revision
+- Where a run came from
 - Settings
 - Output limit
 - Retention
@@ -109,6 +110,8 @@ For every recorded invocation, Recall keeps:
   `error_message`, also when the command reported the failure only through its exit code (for
   example `sqb debug` failing to connect records `ConnectionError` and `C980`);
 - the commit it ran from (`git_revision`), see [Git revision](#git-revision);
+- where it came from (`origin`) and, for an orchestrated run, the job and run that started it, see
+  [Where a run came from](#where-a-run-came-from);
 - its [lifecycle events](../observability.md): runs, resource attempts, statements,
   retries, and audit results;
 - the SQL text of every executed statement, stored once per distinct statement and shared across
@@ -146,6 +149,29 @@ export SQLBUILD_GIT_DIRTY=false                        # optional: recorded as g
 `SQLBUILD_GIT_REVISION` takes precedence over the repository. Without `SQLBUILD_GIT_DIRTY`,
 `git_dirty` is `NULL`.
 
+## Where a run came from
+
+Every command is recorded in the store of the target it ran against, including a quick
+`sqb debug --target prod --no-connection` on a laptop. Each invocation therefore says where it came
+from, so a local check reads differently from a failed scheduled build:
+
+- `origin` is `integration` when an integration such as Dagster started the command (it records
+  `external_context`), `ci` when the command ran in CI, and `local` otherwise. CI is detected from
+  the variables CI systems set: `CI`, `GITHUB_ACTIONS`, `GITLAB_CI`, `BUILDKITE`, `JENKINS_URL`,
+  `CIRCLECI`, `TF_BUILD`, `TEAMCITY_VERSION`, `BITBUCKET_BUILD_NUMBER` and `CODEBUILD_BUILD_ID`
+  (any value other than empty, `false` or `0`).
+- `command_mode` says what a diagnostic command checked: `sqb debug` records `config_only` with
+  `--no-connection` and `config_and_connection` otherwise. Recall never stores a command's raw
+  arguments, which can hold secrets.
+- `external_context` is the JSON object the integration supplied. The `invocations` view exposes
+  its common fields as columns: `integration` (for example `dagster`), `integration_job`,
+  `integration_run_id` and `integration_retry`. The Dagster integration also records `step_key`
+  and `partition_key`, which you read from `external_context` with the store's JSON functions.
+
+Invocations recorded by an older SQLBuild version have `origin` and `command_mode` set to `NULL`.
+Older versions also stored `external_context` in a form that is not JSON; for those runs the
+`integration_*` columns are `NULL` and the query does not fail.
+
 ## Settings
 
 Configure Recall in `[recall]` in `sqlbuild_project.toml`. A `[targets.<name>.recall]` table
@@ -394,6 +420,11 @@ in the store's SQL dialect, against these views:
 | `output_tail` | Recent output chunk |
 | `conflicts` | Record written twice with different content |
 
+Output shows an invocation ID by its first eight characters, such as `38ecc95e`. Any unique prefix
+names the invocation: `sqb recall output 38ecc95e` works, and a prefix that matches more than one
+invocation fails and lists the matches. In SQL, match a prefix with
+`invocation_id LIKE '38ecc95e%'` on SQLite or `startsWith(invocation_id, '38ecc95e')` on ClickHouse.
+
 Run `sqb recall schema` for every column, and check the values a column holds before filtering on
 it, for example `SELECT DISTINCT resource_kind, purpose FROM statements`: a model's
 `resource_kind` is its materialization, such as `table` or `view`, and its build statements have
@@ -405,21 +436,26 @@ the last day reads that part of the store instead of folding the whole retention
 
 ### Common questions
 
-These run as written on the local SQLite store; on ClickHouse, read JSON with
-`JSONExtractString(external_context, 'integration', 'job_name')`.
+These run as written on the local SQLite store and on a ClickHouse store.
 
-The latest runs of an orchestrated job, by the context its integration records (the Dagster
-integration records `run_id`, `job_name`, `step_key`, `retry_number` and `partition_key`). Runs
-recorded before this context was stored as JSON hold it in a form the JSON functions cannot read,
-so this query skips them without an error:
+The latest runs of an orchestrated job, by the job name its integration records:
 
 ```bash
-sqb recall query "SELECT invocation_id, invocation_started_at, status, duration_ms
-  FROM invocations
-  WHERE json_extract(external_context, '$.integration.job_name') = 'daily_models'
+sqb recall query "SELECT invocation_id, invocation_started_at, status, duration_ms,
+    integration_run_id, integration_retry
+  FROM invocations WHERE integration_job = 'orders_job'
   ORDER BY invocation_started_at DESC LIMIT 5"
 ```
 
+The latest scheduled and CI builds, leaving out local diagnostics such as `sqb debug`:
+
+```bash
+sqb recall query "SELECT invocation_id, invocation_started_at, command, origin, status
+  FROM invocations
+  WHERE command IN ('build', 'run', 'test', 'audit', 'seed') AND origin IN ('integration', 'ci')
+  ORDER BY invocation_started_at DESC LIMIT 10"
+```
+
 The slowest models in a run:
 
 ```bash
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/integrations/dagster-reference.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/integrations/dagster-reference.md
index 0fbe0b5d99..62e40aa8fc 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/integrations/dagster-reference.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/integrations/dagster-reference.md
@@ -119,6 +119,46 @@ A running or completed SQLBuild CLI subprocess.
 5. If the DAG artifact is available, results are matched to the correct Dagster asset keys
 6. Raises `Failure` if the process exited non-zero and `raise_on_error` is set
 
+### Logs
+
+The command is logged at INFO. When Dagster selected a subset of assets, the selection is logged
+as a count at INFO, with the selector file path when SQLBuild read the selection from one; the
+full list of selectors is logged at DEBUG. Each check that did not pass is logged at ERROR, or at
+WARNING for a warn-severity check, with its failing row count when the check reports one, so it
+shows in the default run log view.
+
+### Failure output
+
+The `Failure` that `get_error()` returns, and that `wait()` and `stream()` raise, starts its
+description with what failed: up to 10 failed checks and assets, each with its name, model and
+failing row count (or status), followed by "and N more" when there are more. The command and its
+exit code follow, then the details of each failed asset.
+
+```
+Failed: 1 check
+  audit not_null on orders.order_id: 2 failing rows
+
+SQLBuild CLI command failed with exit code 1: sqb audit --json-output ...
+```
+
+Checks that only warned are not listed; they do not fail the command. The failure metadata holds:
+
+| Key | Description |
+|-----|-------------|
+| `command` | The command that ran |
+| `project_dir` | The project directory |
+| `selection_count` | How many selectors Dagster passed to SQLBuild |
+| `selector_file` | The selector file the selection was passed in, or empty |
+| `execution_status` | The status of the execution JSON |
+| `incomplete_assets` | Assets that did not complete, as `kind:name (status)` |
+| `failed_checks` | Up to 100 failed checks, each with `kind`, `name`, `asset`, `column`, `status` and `failing_rows` |
+| `failed_check_count` | How many checks failed |
+| `microbatch_limits` | Microbatch limit evidence of incomplete assets, when there is any |
+
+The metadata no longer lists the selection itself: `selection` was replaced by `selection_count`
+and `selector_file`. Without an execution JSON, the metadata holds the non-empty `stdout_tail` and
+`stderr_tail` instead of the execution keys.
+
 ## sqlbuild_assets
 
 Decorator that creates a Dagster multi-asset definition from a SQLBuild DAG artifact.
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/investigating-runs.md b/src/sqlbuild/.agents/skills/sqlbuild/references/investigating-runs.md
index 97918e64e6..6e1d004192 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/investigating-runs.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/investigating-runs.md
@@ -51,6 +51,9 @@ what they printed. `sqb recall query` runs one read-only `SELECT` over a set of
 | Column | Values |
 |--------|--------|
 | `invocations.status` | `completed`, `failed`, `running`, `abandoned`, `unknown` |
+| `invocations.origin` | `integration` (an orchestrator such as Dagster started it), `ci` (CI environment variables were set), `local`; `NULL` on runs older SQLBuild versions recorded |
+| `invocations.command_mode` | What a diagnostic checked: `sqb debug` records `config_only` (`--no-connection`) or `config_and_connection`; `NULL` for other commands |
+| `invocations.integration`, `integration_job`, `integration_run_id`, `integration_retry` | The orchestrator, job, run and retry from `external_context`, for example `dagster` and its job name; `NULL` when there is none or an older version stored it as non-JSON |
 | `invocations.error_type`, `error_code`, `error_message` | The cause of a failed invocation, also when it failed only through its exit code (`sqb debug`: `C980` connection, `C981` query, `C982` other check) |
 | `statements.outcome` | `completed`, `failed`, `no_terminal` |
 | `audit_results.outcome` | `pass`, `warn`, `error`, `insufficient`; `execution_error` is set when the audit query itself failed |
@@ -75,6 +78,13 @@ what they printed. `sqb recall query` runs one read-only `SELECT` over a set of
   `bytes_read` and `elapsed_ms` on ClickHouse. A failing `--json` query prints
   `{"error": {"code", "message", "help"}}` on stdout and exits `1`.
 - The default `--limit` is 100 rows; `truncated: true` means there were more.
+- Every command is recorded, including someone's local `sqb debug --target prod`. When looking for
+  scheduled runs, filter by purpose: `command IN ('build', 'run', 'test', 'audit', 'seed')`, or
+  `origin = 'integration'` for orchestrated runs, so local diagnostics do not look like failed
+  production runs.
+- Output shows invocation IDs by their first 8 characters. `sqb recall output` accepts any unique
+  prefix and lists the matches when one is ambiguous; in SQL use `invocation_id LIKE '<prefix>%'`
+  (SQLite) or `startsWith(invocation_id, '<prefix>')` (ClickHouse).
 
 ## Output format
 
@@ -99,8 +109,16 @@ sqb recall query "SELECT invocation_id, command, status, exit_code, error_type,
 sqb recall output <invocation_id> --stream stderr --tail 40
 ```
 
-For the last scheduled run, filter on the orchestrator's context or the target, for example
-`WHERE target = 'prod'`.
+For the last scheduled run, filter on where it came from or the target, for example
+`WHERE origin = 'integration' AND target = 'prod'`.
+
+The latest runs of an orchestrated job (the same SQL on SQLite and ClickHouse):
+
+```bash
+sqb recall query "SELECT invocation_id, invocation_started_at, status, integration_run_id, integration_retry
+  FROM invocations WHERE integration_job = 'orders_job'
+  ORDER BY invocation_started_at DESC LIMIT 5"
+```
 
 What did a model actually execute, with its SQL? Find the run, then its statements:
 
@@ -278,7 +296,7 @@ events (view): every lifecycle fact
   ingested_at TEXT
   content_hash TEXT
 
-invocations (view): one row per invocation; status is completed, failed, running, abandoned or unknown
+invocations (view): one row per invocation; status is completed, failed, running, abandoned or unknown; origin is integration, ci or local; integration, integration_job, integration_run_id and integration_retry come from external_context
   invocation_id TEXT  join key
   invocation_started_at ANY
   started_at ANY
@@ -291,6 +309,8 @@ invocations (view): one row per invocation; status is completed, failed, running
   error_message ANY
   command ANY
   subcommand ANY
+  origin TEXT
+  command_mode TEXT
   target TEXT
   adapter TEXT
   project_name TEXT
@@ -302,6 +322,10 @@ invocations (view): one row per invocation; status is completed, failed, running
   host_token TEXT
   pid INTEGER
   external_context TEXT
+  integration ANY
+  integration_job ANY
+  integration_run_id ANY
+  integration_retry ANY
   recall_filter_active INTEGER
   last_heartbeat_at ANY
 
diff --git a/src/sqlbuild/integrations/dagster/_helpers/invocation.py b/src/sqlbuild/integrations/dagster/_helpers/invocation.py
index c250b4e2f7..baba7b21ce 100644
--- a/src/sqlbuild/integrations/dagster/_helpers/invocation.py
+++ b/src/sqlbuild/integrations/dagster/_helpers/invocation.py
@@ -36,11 +36,40 @@ def _log_invocation(*, context: Any, invocation: SqlBuildCliInvocation) -> None:
     logger.info("  %s", " ".join(invocation.command))
     if invocation.selection:
         if invocation.selector_file_path:
-            logger.info("SQLBuild selector file:")
-            logger.info("  %s", invocation.selector_file_path)
-        logger.info("SQLBuild selected assets from Dagster (%s):", len(invocation.selection))
-        for line in dagster_selector_log_lines(selectors=list(invocation.selection)):
-            logger.info("  %s", line)
+            logger.info(
+                "SQLBuild selected assets from Dagster: %s (selector file: %s)",
+                len(invocation.selection),
+                invocation.selector_file_path,
+            )
+        else:
+            logger.info("SQLBuild selected assets from Dagster: %s", len(invocation.selection))
+        logger.debug(
+            "SQLBuild selectors from Dagster:\n%s",
+            "\n".join(
+                f"  {line}"
+                for line in dagster_selector_log_lines(selectors=list(invocation.selection))
+            ),
+        )
+
+
+def _log_failed_check(*, context: Any, dg: Any, result: Any) -> None:
+    """Log a check that did not pass, so it shows in the default run log view."""
+
+    if result.passed:
+        return
+    logger: Any = getattr(context, "log", None) if context is not None else None
+    if logger is None:
+        return
+    asset: str = "/".join(str(part) for part in result.asset_key.path)
+    row_count: object = (result.metadata or {}).get("row_count")
+    detail: str = ""
+    count: object = getattr(row_count, "value", row_count)
+    if isinstance(count, int) and not isinstance(count, bool):
+        detail = f": {count} failing row" if count == 1 else f": {count} failing rows"
+    if result.severity == dg.AssetCheckSeverity.WARN:
+        logger.warning("SQLBuild check %s on %s warned%s", result.check_name, asset, detail)
+        return
+    logger.error("SQLBuild check %s on %s failed%s", result.check_name, asset, detail)
 
 
 def _log_skipped_checks(*, context: Any, payload: Mapping[str, Any]) -> None:
diff --git a/src/sqlbuild/integrations/dagster/classes/sqlbuild_cli_invocation.py b/src/sqlbuild/integrations/dagster/classes/sqlbuild_cli_invocation.py
index 76a4e878c8..822ec1f785 100644
--- a/src/sqlbuild/integrations/dagster/classes/sqlbuild_cli_invocation.py
+++ b/src/sqlbuild/integrations/dagster/classes/sqlbuild_cli_invocation.py
@@ -24,6 +24,7 @@
     _build_results_from_integration_result,
     _load_execution_payload,
     _load_execution_payload_from_path,
+    _log_failed_check,
     _log_invocation,
     _start_stream_future,
 )
@@ -163,13 +164,16 @@ def stream(self) -> Iterator[Any]:
                 execution_payload = _load_execution_payload(self.stdout)
             if execution_payload is not None:
                 self.execution_payload = execution_payload
-                yield from _build_results_from_execution_payload(
+                for result in _build_results_from_execution_payload(
                     dg=dg,
                     dag=self.dag,
                     payload=execution_payload,
                     command=self.command,
                     context=self.context,
-                )
+                ):
+                    if isinstance(result, dg.AssetCheckResult):
+                        _log_failed_check(context=self.context, dg=dg, result=result)
+                    yield result
                 error: Exception | None = self.get_error()
                 if error is not None and self.raise_on_error:
                     raise error
@@ -229,6 +233,7 @@ def _stream_with_live_events(self) -> Iterator[Any]:
                                 if check_key in emitted_check_keys:
                                     continue
                                 emitted_check_keys.add(check_key)
+                                _log_failed_check(context=self.context, dg=dg, result=result)
                             else:
                                 asset_key: tuple[str, ...] = tuple(result.asset_key.path)
                                 if asset_key in emitted_asset_keys:
@@ -258,6 +263,7 @@ def _stream_with_live_events(self) -> Iterator[Any]:
                         if check_key in emitted_check_keys:
                             continue
                         emitted_check_keys.add(check_key)
+                        _log_failed_check(context=self.context, dg=dg, result=result)
                         yield result
                         continue
                     if isinstance(result, (dg.AssetMaterialization, dg.MaterializeResult)):
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/_test_types.py
index 907ae9a301..2a63bc5e25 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/_test_types.py
@@ -268,6 +268,7 @@ class ClickHouseRecallViewEquivalenceTestCase:
     replaced_views: tuple[str, ...]
     all_views: tuple[str, ...]
     expected_statuses: list[list[object]]
+    expected_added_columns: tuple[tuple[tuple[object, ...], ...], ...]
 
 
 @dataclass(frozen=True)
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/helpers.py
index b1d082211d..dcf43b1d8c 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/helpers.py
@@ -14,6 +14,7 @@
 
 from sqlbuild.adapters.clickhouse.classes.clickhouse_adapter import ClickHouseAdapter
 from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import (
+    REPO_ROOT,
     prepare_inline_project,
     prepare_waffle_shop,
     run_sqb,
@@ -28,6 +29,22 @@
 )
 
 CLICKHOUSE_IMAGE: str = "clickhouse/clickhouse-server:26.1"
+_RECALL_SCHEMA_CONSTANTS: Path = REPO_ROOT / "crates/sqlbuild-recall/src/schema/constants.rs"
+_RECALL_SCHEMA_VERSION_PATTERN: re.Pattern[str] = re.compile(
+    r"pub const RECALL_SCHEMA_VERSION: i64 = (\d+);"
+)
+
+
+def _recall_schema_version() -> int:
+    """The Recall schema version this build writes, as `RECALL_SCHEMA_VERSION` declares it."""
+
+    (version,) = _RECALL_SCHEMA_VERSION_PATTERN.findall(
+        _RECALL_SCHEMA_CONSTANTS.read_text(encoding="utf-8")
+    )
+    return int(version)
+
+
+RECALL_SCHEMA_VERSION: int = _recall_schema_version()
 
 
 def build_unique_schema_name(*, prefix: str) -> str:
@@ -915,16 +932,23 @@ def fetch_with_settings(
 
 
 def differing_rows(
-    *, config: dict[str, object], left: str, right: str, settings: dict[str, object]
+    *,
+    config: dict[str, object],
+    left: str,
+    right: str,
+    settings: dict[str, object],
+    columns: tuple[str, ...] = (),
 ) -> int:
-    """Rows of `left` missing from `right` plus rows of `right` missing from `left`."""
+    """Rows of `left` missing from `right` plus rows of `right` missing from `left`, over `columns`
+    when given, else over every column."""
 
+    selected: str = ", ".join(f"`{column}`" for column in columns) or "*"
     rows: tuple[tuple[object, ...], ...] = fetch_with_settings(
         config=config,
         sql=(
-            f"SELECT (SELECT count() FROM (SELECT * FROM {left} EXCEPT DISTINCT "
-            f"SELECT * FROM {right})) + (SELECT count() FROM (SELECT * FROM {right} "
-            f"EXCEPT DISTINCT SELECT * FROM {left}))"
+            f"SELECT (SELECT count() FROM (SELECT {selected} FROM {left} EXCEPT DISTINCT "
+            f"SELECT {selected} FROM {right})) + (SELECT count() FROM (SELECT {selected} "
+            f"FROM {right} EXCEPT DISTINCT SELECT {selected} FROM {left}))"
         ),
         settings=settings,
     )
@@ -940,6 +964,30 @@ def described(*, config: dict[str, object], relation: str) -> tuple[tuple[object
     )
 
 
+def column_names(*, config: dict[str, object], relation: str) -> tuple[str, ...]:
+    """Each column's name, in order."""
+
+    return tuple(str(name) for name, _ in described(config=config, relation=relation))
+
+
+def described_except(
+    *, config: dict[str, object], relation: str, previous: str
+) -> tuple[tuple[tuple[object, ...], ...], tuple[tuple[object, ...], ...]]:
+    """`relation`'s columns and types that `previous` also has, and those it adds, in order."""
+
+    columns: dict[object, object] = {
+        column[0]: column[1] for column in described(config=config, relation=relation)
+    }
+    kept: tuple[str, ...] = column_names(config=config, relation=previous)
+    added: dict[object, object] = dict(columns)
+    for name in kept:
+        del added[name]
+    return (
+        tuple((name, columns[name]) for name in kept),
+        tuple(added.items()),
+    )
+
+
 def create_previous_views(
     *, config: dict[str, object], database: str, names: tuple[str, ...], prefix: str
 ) -> None:
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall.py b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall.py
index 6806b488f9..a665d15cf1 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall.py
@@ -32,6 +32,7 @@
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse.helpers import (
     FIRST_RELEASE_EDITS,
+    RECALL_SCHEMA_VERSION,
     age_recorded_invocations,
     age_stored_artifacts,
     apply_printed_ddl,
@@ -903,7 +904,7 @@ def test_given_a_large_store_when_filtering_a_view_then_the_filter_reaches_the_t
             description="views of an older version are replaced when its version row expired",
             stale_view="SELECT invocation_id, statement_id FROM {database}.recall_events",
             expected_statement_columns=((1,),),
-            expected_version=((3,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
     ],
     ids=lambda case: case.description,
@@ -1039,12 +1040,12 @@ def test_given_a_newer_store_whose_version_row_expired_when_an_older_writer_buil
         ClickHouseRecallVersionRefreshTestCase(
             description="a version row past half the retention is recorded again",
             row_age_days=20,
-            expected_rows=((2, 3, 1),),
+            expected_rows=((2, RECALL_SCHEMA_VERSION, 1),),
         ),
         ClickHouseRecallVersionRefreshTestCase(
             description="a recent version row is left as it is",
             row_age_days=1,
-            expected_rows=((1, 3, 0),),
+            expected_rows=((1, RECALL_SCHEMA_VERSION, 0),),
         ),
     ],
     ids=lambda case: case.description,
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_setup.py b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_setup.py
index b0080f9766..5a582a479b 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_setup.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_setup.py
@@ -15,6 +15,7 @@
     ClickHouseRecallStaleViewsTestCase,
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse.helpers import (
+    RECALL_SCHEMA_VERSION,
     apply_printed_ddl,
     create_previous_views,
     drop_database,
@@ -197,7 +198,7 @@ def test_given_a_missing_database_when_the_writer_may_not_create_it_then_the_sta
             writer_privileges="ALL",
             expected_notices=0,
             expected_stale_views=((0,),),
-            expected_version=((3,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
         ClickHouseRecallStaleViewsTestCase(
             description="previous views recorded as current are replaced by an administrator",
@@ -205,7 +206,7 @@ def test_given_a_missing_database_when_the_writer_may_not_create_it_then_the_sta
             writer_privileges="ALL",
             expected_notices=0,
             expected_stale_views=((0,),),
-            expected_version=((3,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
         ClickHouseRecallStaleViewsTestCase(
             description="previous views without a version row make a SELECT, INSERT writer warn",
@@ -283,6 +284,16 @@ def test_given_views_of_the_previous_version_when_building_then_they_count_as_ol
             "WHERE JSONExtractString(external_context, 'integration', 'job_name') = 'daily_models'",
             expected_rows=[["completed"]],
         ),
+        ClickHouseRecallExternalContextTestCase(
+            description="an orchestrated job's runs are found by their integration columns",
+            external_context=(
+                '{"integration": {"name": "dagster", "job_name": "orders_job",'
+                ' "run_id": "run-7", "retry_number": 1}}'
+            ),
+            sql="SELECT origin, integration, integration_run_id, integration_retry "
+            "FROM invocations WHERE integration_job = 'orders_job'",
+            expected_rows=[["integration", "dagster", "run-7", 1]],
+        ),
     ],
     ids=lambda case: case.description,
 )
@@ -364,7 +375,7 @@ def test_given_a_writer_that_may_create_only_the_database_when_building_then_bot
             dropped_column=("recall_invocations", "argv_digest"),
             expected_columns=((1,),),
             expected_stale_views=((0,),),
-            expected_version=((3,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
     ],
     ids=lambda case: case.description,
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_views.py b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_views.py
index ba2d17a60a..cb76fc1447 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_views.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_views.py
@@ -17,9 +17,12 @@
     ClickHouseRecallViewEquivalenceTestCase,
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse.helpers import (
+    RECALL_SCHEMA_VERSION,
     RECALL_VIEWS,
+    column_names,
     create_previous_views,
     described,
+    described_except,
     differing_rows,
     duplicate_recall_rows,
     execute_clickhouse_sql,
@@ -206,6 +209,19 @@ def test_given_a_reader_when_querying_then_pinned_settings_apply_unless_it_may_n
                 ["failed"],
                 ["running"],
             ],
+            expected_added_columns=(
+                (
+                    ("origin", "Nullable(String)"),
+                    ("command_mode", "Nullable(String)"),
+                    ("integration", "Nullable(String)"),
+                    ("integration_job", "Nullable(String)"),
+                    ("integration_run_id", "Nullable(String)"),
+                    ("integration_retry", "Nullable(Int64)"),
+                ),
+                (),
+                (),
+                (),
+            ),
         ),
     ],
     ids=lambda case: case.description,
@@ -230,22 +246,40 @@ def test_given_representative_invocations_when_views_are_rewritten_then_rows_and
         statuses: list[list[object]] = recall_rows(
             project_dir, "SELECT status FROM invocations ORDER BY status"
         )
+        previous_columns: list[tuple[str, ...]] = [
+            column_names(
+                config=clickhouse_e2e_config, relation=f"{recall_database}.previous_{view}"
+            )
+            for view in test_case.replaced_views
+        ]
         against_previous: list[int] = [
             differing_rows(
                 config=clickhouse_e2e_config,
                 left=f"{recall_database}.{view}",
                 right=f"{recall_database}.previous_{view}",
                 settings={"join_use_nulls": 1},
+                columns=columns,
+            )
+            for view, columns in zip(test_case.replaced_views, previous_columns, strict=True)
+        ]
+        compared: list[tuple[tuple[tuple[object, ...], ...], tuple[tuple[object, ...], ...]]] = [
+            described_except(
+                config=clickhouse_e2e_config,
+                relation=f"{recall_database}.{view}",
+                previous=f"{recall_database}.previous_{view}",
             )
             for view in test_case.replaced_views
         ]
         types: list[bool] = [
-            described(config=clickhouse_e2e_config, relation=f"{recall_database}.{view}")
+            kept
             == described(
                 config=clickhouse_e2e_config, relation=f"{recall_database}.previous_{view}"
             )
-            for view in test_case.replaced_views
+            for view, (kept, _) in zip(test_case.replaced_views, compared, strict=True)
         ]
+        added: tuple[tuple[tuple[object, ...], ...], ...] = tuple(
+            added_columns for _, added_columns in compared
+        )
     finally:
         os.killpg(running.pid, signal.SIGKILL)
         _ = running.wait(timeout=30)
@@ -277,8 +311,9 @@ def test_given_representative_invocations_when_views_are_rewritten_then_rows_and
             left=f"{recall_database}.{view}",
             right=f"{recall_database}.previous_{view}",
             settings={"join_use_nulls": 1},
+            columns=columns,
         )
-        for view in test_case.replaced_views
+        for view, columns in zip(test_case.replaced_views, previous_columns, strict=True)
     ]
     counts_with_duplicates: list[bool] = [
         fetch_clickhouse_rows(
@@ -293,6 +328,7 @@ def test_given_representative_invocations_when_views_are_rewritten_then_rows_and
     assert statuses == test_case.expected_statuses
     assert against_previous == [0 for _ in test_case.replaced_views]
     assert types == [True for _ in test_case.replaced_views]
+    assert added == test_case.expected_added_columns
     assert without_join_nulls == [0 for _ in test_case.all_views]
     assert with_duplicates == [0 for _ in test_case.all_views]
     assert previous_with_duplicates == [0 for _ in test_case.replaced_views]
@@ -354,9 +390,9 @@ def test_given_one_invocation_lookup_when_planned_then_each_table_is_read_once_b
             expected_ddl_fragments=(
                 "CREATE OR REPLACE VIEW `{database}`.`invocations`",
                 "CREATE OR REPLACE VIEW `{database}`.`statements`",
-                "SELECT 'schema', 'schema_version', '3'",
+                f"SELECT 'schema', 'schema_version', '{RECALL_SCHEMA_VERSION}'",
             ),
-            expected_version=((3,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
     ],
     ids=lambda case: case.description,
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/_test_types.py
index 98ee578ab7..8facea8681 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/_test_types.py
@@ -125,3 +125,14 @@ class SqlMacroFormatPairTestCase:
     slot_text: str
     hand_text: str
     expected_check_returncode: int
+
+
+@dataclass(frozen=True)
+class SqlMacroBodyRulesTestCase:
+    """One project linting a SQL macro body, and the findings it must and must not print."""
+
+    description: str
+    files: dict[str, str]
+    expected_returncode: int
+    expected_fragments: tuple[str, ...]
+    expected_absent_fragments: tuple[str, ...]
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/helpers.py
index 380835df9e..b80afc7bce 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/helpers.py
@@ -554,3 +554,43 @@ def scoped_model(macros: dict[str, str], body: str) -> dict[str, str]:
             f"{body}\n"
         ),
     }
+
+
+SQL_RULES: dict[str, str] = {
+    "sqlbuild_project.toml": BASE_FILES["sqlbuild_project.toml"]
+    + '\n[rules]\nselect = ["SQBRSQL028", "SQBRSQL029"]\n'
+}
+
+
+def rates_with_middle_cte(middle: str) -> dict[str, str]:
+    """The docs' weekly refund rates macro with `middle` as a CTE between its two, and a model."""
+
+    return {
+        "macros/sql/weekly_rates.sql": (
+            'MACRO (\n  description "Weekly refund rate per grain",\n'
+            "  arguments (weekly RELATION (keys COLUMNS, order_count BIGINT, refund_count BIGINT)),"
+            "\n);\n\n"
+            "WITH counted AS (\n"
+            "  SELECT\n"
+            "    s.@weekly.keys,\n"
+            "    s.@weekly.order_count AS orders,\n"
+            "    s.@weekly.refund_count AS refunds\n"
+            "  FROM @weekly AS s\n"
+            "),\n\n"
+            f"{middle},\n\n"
+            "rated AS (\n"
+            "  SELECT\n"
+            "    @weekly.keys,\n"
+            "    SUM(refunds) / NULLIF(SUM(orders), 0) AS refund_rate\n"
+            "  FROM plain\n"
+            "  GROUP BY @weekly.keys\n"
+            ")\n"
+        ),
+        "models/marts/rates.sql": (
+            'MODEL (description "Refund rates");\n\n'
+            'WITH weekly AS (SELECT * FROM __seed("weekly_orders")),\n'
+            "@sql.weekly_rates(weekly, weekly.keys := (product_id, region), "
+            "weekly.order_count := orders_placed)\n"
+            "SELECT * FROM rated\n"
+        ),
+    }
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/test_sql_macro_body_rules.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/test_sql_macro_body_rules.py
new file mode 100644
index 0000000000..f2ec2a0a55
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/sql_macros/test_sql_macro_body_rules.py
@@ -0,0 +1,113 @@
+"""E2E coverage for the SQL rules that lint a SQL macro body's relation slot reads."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.compile.sql_macros._test_types import (
+    SqlMacroBodyRulesTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.compile.sql_macros.helpers import (
+    SQL_RULES,
+    WEEKLY_ORDERS,
+    output,
+    rates_with_middle_cte,
+    run_project,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SqlMacroBodyRulesTestCase(
+            description="a_plain_group_read_beside_bare_columns_passes_the_body_rules",
+            files={
+                **WEEKLY_ORDERS,
+                **SQL_RULES,
+                **rates_with_middle_cte(
+                    "plain AS (SELECT @weekly.keys, orders, refunds FROM counted)"
+                ),
+            },
+            expected_returncode=0,
+            expected_fragments=("Project compiled",),
+            expected_absent_fragments=("SQBRSQL",),
+        ),
+        SqlMacroBodyRulesTestCase(
+            description="plain_single_slot_and_undeclared_reads_through_the_relation_pass",
+            files={
+                **WEEKLY_ORDERS,
+                **SQL_RULES,
+                **rates_with_middle_cte(
+                    "named AS (\n"
+                    "  SELECT @weekly.refund_count AS refunds, @weekly.week AS week\n"
+                    "  FROM @weekly\n"
+                    "),\n\n"
+                    "plain AS (SELECT @weekly.keys, orders, refunds FROM counted)"
+                ),
+            },
+            expected_returncode=0,
+            expected_fragments=("Project compiled",),
+            expected_absent_fragments=("SQBRSQL",),
+        ),
+        SqlMacroBodyRulesTestCase(
+            description="a_real_mixed_qualification_beside_a_plain_group_read_is_reported",
+            files={
+                **WEEKLY_ORDERS,
+                **SQL_RULES,
+                **rates_with_middle_cte(
+                    "plain AS (SELECT @weekly.keys, c.orders, refunds FROM counted AS c)"
+                ),
+            },
+            expected_returncode=1,
+            expected_fragments=(
+                "error[SQBRSQL028]: Single-relation query mixes qualified and unqualified columns "
+                "(in SQL generated by @weekly)\n  --> macros/sql/weekly_rates.sql:14:18",
+                "error[SQBRSQL028]: Single-relation query mixes qualified and unqualified columns\n"
+                "  --> macros/sql/weekly_rates.sql:14:42",
+            ),
+            expected_absent_fragments=("SQBRSQL029",),
+        ),
+        SqlMacroBodyRulesTestCase(
+            description="a_qualifier_absent_from_scope_beside_a_plain_group_read_is_reported",
+            files={
+                **WEEKLY_ORDERS,
+                **SQL_RULES,
+                **rates_with_middle_cte(
+                    "plain AS (SELECT @weekly.keys, orders, refunds FROM counted)"
+                ),
+                "macros/sql/uncalled_rates.sql": (
+                    'MACRO (\n  description "Weekly orders per grain",\n'
+                    "  arguments (weekly RELATION (keys COLUMNS, order_count BIGINT)),\n);\n\n"
+                    "WITH counted AS (\n"
+                    "  SELECT s.@weekly.keys, s.@weekly.order_count AS orders FROM @weekly AS s\n"
+                    "),\n\n"
+                    "plain AS (SELECT @weekly.keys, x.orders FROM counted AS c)\n"
+                ),
+            },
+            expected_returncode=1,
+            expected_fragments=(
+                "error[SQBRSQL029]: Qualified reference uses a relation absent from this query "
+                "scope\n  --> macros/sql/uncalled_rates.sql:10:32",
+            ),
+            expected_absent_fragments=(
+                "error[SQBRSQL029]: Qualified reference uses a relation absent from this query "
+                "scope (in SQL generated by @weekly)",
+                "--> macros/sql/weekly_rates.sql",
+            ),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_slot_reads_in_a_macro_body_when_compiling_then_body_rules_judge_the_expansion(
+    test_case: SqlMacroBodyRulesTestCase, tmp_path: Path
+) -> None:
+    _, result = run_project(tmp_path=tmp_path, files=test_case.files, command=("compile",))
+    printed: str = output(result)
+
+    assert result.returncode == test_case.expected_returncode, printed
+    for fragment in test_case.expected_fragments:
+        assert fragment in printed, printed
+    for fragment in test_case.expected_absent_fragments:
+        assert fragment not in printed, printed
diff --git a/tests/e2e/src/sqlbuild/integrations/dagster/_test_types.py b/tests/e2e/src/sqlbuild/integrations/dagster/_test_types.py
index 7a3d6c2bdf..cb8e3d300f 100644
--- a/tests/e2e/src/sqlbuild/integrations/dagster/_test_types.py
+++ b/tests/e2e/src/sqlbuild/integrations/dagster/_test_types.py
@@ -161,3 +161,14 @@ class DagsterSkippedAuditCheckE2ETestCase:
     description: str
     skipped_check_name: str
     expected_log_line: str
+
+
+@dataclass(frozen=True)
+class DagsterFailureOutputE2ETestCase:
+    """Test case for what a failed SQLBuild command reports through Dagster."""
+
+    description: str
+    null_order_rows: int
+    expected_description_start: str
+    expected_failed_checks: list[dict[str, object]]
+    expected_error_log: str
diff --git a/tests/e2e/src/sqlbuild/integrations/dagster/test_failure_output.py b/tests/e2e/src/sqlbuild/integrations/dagster/test_failure_output.py
new file mode 100644
index 0000000000..51621892d8
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/integrations/dagster/test_failure_output.py
@@ -0,0 +1,128 @@
+"""E2E coverage of what a failed SQLBuild command reports through Dagster."""
+
+import json
+import logging
+import subprocess
+from collections.abc import Iterator
+from pathlib import Path
+
+import duckdb
+import pytest
+from dagster import (
+    AssetExecutionContext,
+    DagsterInstance,
+    EventLogEntry,
+    ExecuteInProcessResult,
+    materialize,
+)
+from dagster._core.execution.plan.objects import UserFailureData
+
+from sqlbuild.integrations.dagster import SqlBuildCliResource, SqlBuildProject, sqlbuild_assets
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import (
+    REPO_ROOT,
+    prepare_inline_project,
+    run_sqb,
+)
+from tests.e2e.src.sqlbuild.integrations.dagster._test_types import (
+    DagsterFailureOutputE2ETestCase,
+)
+
+_FAILING_AUDIT_PROJECT_FILES: dict[str, str] = {
+    "sqlbuild_project.toml": (
+        'name = "orders_shop"\nadapter = "duckdb"\n\n'
+        '[connection]\ndatabase = "warehouse.duckdb"\n\n[defaults]\nmaterialized = "table"\n'
+    ),
+    "models/customers.sql": (
+        "MODEL (description 'One row per customer.');\nSELECT 1 AS customer_id\n"
+    ),
+    "models/orders.sql": (
+        "MODEL (\n  description 'One row per order.',\n"
+        "  columns (order_id (audits [not_null])),\n);\n"
+        'SELECT 1 AS order_id, customer_id FROM __ref("customers")\n'
+    ),
+}
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        DagsterFailureOutputE2ETestCase(
+            description="a failing audit leads the failure and is logged as an error",
+            null_order_rows=2,
+            expected_description_start=(
+                "Failed: 1 check\n  audit not_null on orders.order_id: 2 failing rows\n\n"
+                "SQLBuild CLI command failed with exit code 1: "
+            ),
+            expected_failed_checks=[
+                {
+                    "kind": "audit",
+                    "name": "not_null",
+                    "asset": "orders",
+                    "column": "order_id",
+                    "status": "error",
+                    "failing_rows": 2,
+                }
+            ],
+            expected_error_log="SQLBuild check audit__not_null__order_id on main/orders failed: "
+            "2 failing rows",
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_a_failing_audit_when_auditing_through_dagster_then_the_failure_names_it(
+    test_case: DagsterFailureOutputE2ETestCase, tmp_path: Path
+) -> None:
+    project_dir: Path = prepare_inline_project(
+        tmp_path=tmp_path, project_name="orders_shop", repo_files=_FAILING_AUDIT_PROJECT_FILES
+    )
+    sqb: Path = REPO_ROOT / ".venv" / "bin" / "sqb"
+    built: subprocess.CompletedProcess[str] = run_sqb(command=("build",), project_dir=project_dir)
+    assert built.returncode == 0, built.stderr
+    with duckdb.connect(str(project_dir / "warehouse.duckdb")) as connection:
+        for _ in range(test_case.null_order_rows):
+            _ = connection.execute("INSERT INTO main.orders VALUES (NULL, 1)")
+    sqlbuild_project: SqlBuildProject = SqlBuildProject(
+        project_dir=project_dir, sqb_command=(str(sqb),)
+    )
+    sqlbuild_project.prepare()
+
+    @sqlbuild_assets(project=sqlbuild_project, required_resource_keys={"sqb"})
+    def orders_job_assets(context: AssetExecutionContext) -> Iterator[object]:
+        yield from context.resources.sqb.cli(["audit"], context=context).stream()
+
+    instance: DagsterInstance = DagsterInstance.ephemeral()
+    result: ExecuteInProcessResult = materialize(
+        [orders_job_assets],
+        resources={"sqb": SqlBuildCliResource(project_dir=sqlbuild_project)},
+        instance=instance,
+        raise_on_error=False,
+    )
+    failure: UserFailureData | None = result.get_step_failure_events()[
+        0
+    ].step_failure_data.user_failure_data
+    assert failure is not None
+    description: str = failure.description or ""
+    metadata: dict[str, object] = {key: value.value for key, value in failure.metadata.items()}
+    logs: list[EventLogEntry] = list(instance.all_logs(result.run_id))
+    logged: list[tuple[str | int, str]] = [(entry.level, entry.user_message) for entry in logs]
+    recalled: subprocess.CompletedProcess[str] = run_sqb(
+        command=(
+            "recall",
+            "query",
+            "--json",
+            "SELECT origin, integration FROM invocations WHERE command = 'audit'",
+        ),
+        project_dir=project_dir,
+    )
+
+    assert result.success is False
+    assert description.startswith(test_case.expected_description_start)
+    assert metadata["failed_checks"] == test_case.expected_failed_checks
+    assert metadata["failed_check_count"] == len(test_case.expected_failed_checks)
+    assert "selection" not in metadata
+    assert isinstance(metadata["selection_count"], int)
+    assert any(
+        level == logging.ERROR and test_case.expected_error_log in message
+        for level, message in logged
+    )
+    assert json.loads(recalled.stdout)["rows"] == [["integration", "dagster"]]
diff --git a/tests/e2e/src/sqlbuild/integrations/dagster/test_sqlbuild_assets.py b/tests/e2e/src/sqlbuild/integrations/dagster/test_sqlbuild_assets.py
index 7fe6fd9263..c761d1140c 100644
--- a/tests/e2e/src/sqlbuild/integrations/dagster/test_sqlbuild_assets.py
+++ b/tests/e2e/src/sqlbuild/integrations/dagster/test_sqlbuild_assets.py
@@ -370,10 +370,10 @@ def sqlbuild_waffle_shop(context: AssetExecutionContext) -> Iterator[object]:
     rendered_logs: str = capsys.readouterr().err
     assert "SQLBuild command:" in rendered_logs
     assert "build --select-file" in rendered_logs
-    assert "SQLBuild selector file:" in rendered_logs
-    assert f"SQLBuild selected assets from Dagster ({len(test_case.selected_asset_keys)}):" in (
-        rendered_logs
-    )
+    assert (
+        f"SQLBuild selected assets from Dagster: {len(test_case.selected_asset_keys)} "
+        "(selector file: "
+    ) in rendered_logs
     assert test_case.expected_selector_log_line in rendered_logs
     for table_name in test_case.expected_table_names:
         assert table_exists(db_path=project_dir / "waffle_shop.duckdb", table_name=table_name)
diff --git a/tests/integration/src/sqlbuild/cli/recall/_test_types.py b/tests/integration/src/sqlbuild/cli/recall/_test_types.py
index 03f8503d9d..3852fd9470 100644
--- a/tests/integration/src/sqlbuild/cli/recall/_test_types.py
+++ b/tests/integration/src/sqlbuild/cli/recall/_test_types.py
@@ -318,3 +318,26 @@ class DocumentedQueryTestCase:
     question: int
     columns: tuple[str, ...]
     expected_rows: list[list[object]]
+
+
+@dataclass(frozen=True)
+class InvocationOriginTestCase:
+    description: str
+    args: tuple[str, ...]
+    environment: dict[str, str]
+    expected_rows: list[list[object]]
+
+
+@dataclass(frozen=True)
+class IntegrationColumnsTestCase:
+    description: str
+    stored_context: str
+    expected_columns: list[object]
+
+
+@dataclass(frozen=True)
+class InvocationPrefixTestCase:
+    description: str
+    prefix_length: int
+    expected_exit_code: int
+    expected_error_code: str | None
diff --git a/tests/integration/src/sqlbuild/cli/recall/helpers.py b/tests/integration/src/sqlbuild/cli/recall/helpers.py
index c88a32de38..595fc40c59 100644
--- a/tests/integration/src/sqlbuild/cli/recall/helpers.py
+++ b/tests/integration/src/sqlbuild/cli/recall/helpers.py
@@ -540,6 +540,43 @@ def make_store_behind(store: Path) -> None:
     connection.close()
 
 
+CI_VARIABLES: tuple[str, ...] = (
+    "CI",
+    "GITHUB_ACTIONS",
+    "GITLAB_CI",
+    "BUILDKITE",
+    "JENKINS_URL",
+    "CIRCLECI",
+    "TF_BUILD",
+    "TEAMCITY_VERSION",
+    "BITBUCKET_BUILD_NUMBER",
+    "CODEBUILD_BUILD_ID",
+)
+NO_CI: dict[str, str] = dict.fromkeys(CI_VARIABLES, "")
+
+
+def copy_invocation_row(store: Path, *, invocation_id: str, external_context: str) -> None:
+    """Store a copy of the first context row under another ID, as an older writer left it."""
+
+    with sqlite3.connect(store) as connection:
+        cursor: sqlite3.Cursor = connection.execute("SELECT * FROM recall_invocations LIMIT 1")
+        names: list[str] = [column[0] for column in cursor.description]
+        row: dict[str, object] = dict(zip(names, cursor.fetchone(), strict=True))
+        row.update(
+            invocation_id=invocation_id,
+            external_context=external_context,
+            origin=None,
+            command_mode=None,
+            content_hash=f"copy-{invocation_id}",
+        )
+        _ = connection.execute(
+            f"INSERT INTO recall_invocations ({', '.join(names)}) "
+            f"VALUES ({', '.join('?' for _ in names)})",
+            [row[name] for name in names],
+        )
+    connection.close()
+
+
 def drop_store_views(store: Path) -> None:
     """Drop every view of a SQLite store, as a store whose creation is still under way has none."""
 
diff --git a/tests/integration/src/sqlbuild/cli/recall/test_recall_invocation_facts.py b/tests/integration/src/sqlbuild/cli/recall/test_recall_invocation_facts.py
index cb79e30414..518a44557e 100644
--- a/tests/integration/src/sqlbuild/cli/recall/test_recall_invocation_facts.py
+++ b/tests/integration/src/sqlbuild/cli/recall/test_recall_invocation_facts.py
@@ -8,10 +8,15 @@
     DocumentedQueryTestCase,
     FailureCauseTestCase,
     GitRevisionTestCase,
+    IntegrationColumnsTestCase,
+    InvocationOriginTestCase,
+    InvocationPrefixTestCase,
 )
 from tests.integration.src.sqlbuild.cli.recall.helpers import (
     DEPLOYED_REVISION,
+    NO_CI,
     checkout_project,
+    copy_invocation_row,
     deployed_project,
     documented_commands,
     recall_rows,
@@ -30,7 +35,12 @@
 )
 
 ORCHESTRATED_CONTEXT: str = (
-    '{"integration": {"name": "dagster", "job_name": "daily_models", "run_id": "run-7"}}'
+    '{"integration": {"name": "dagster", "job_name": "orders_job", "run_id": "run-7",'
+    ' "retry_number": 1}}'
+)
+LEGACY_CONTEXT: str = (
+    "{'integration': mappingproxy({'job_name': 'orders_job', 'name': 'dagster',"
+    " 'retry_number': 0, 'run_id': 'run-6'})}"
 )
 WARNING_AUDITS: str = '\n[settings]\ndefault_audit_severity = "warn"\n'
 CUSTOMERS_MODEL: str = (
@@ -44,20 +54,26 @@
     "test_case",
     [
         DocumentedQueryTestCase(
-            description="an orchestrated job's runs are found by their external context",
+            description="an orchestrated job's runs are found by their integration job",
             question=0,
-            columns=("status",),
-            expected_rows=[["completed"]],
+            columns=("status", "integration_run_id", "integration_retry"),
+            expected_rows=[["completed", "run-7", 1]],
         ),
         DocumentedQueryTestCase(
-            description="the slowest models of a run are its model builds by resource",
+            description="scheduled builds are listed by their origin",
             question=1,
+            columns=("command", "origin", "status"),
+            expected_rows=[["build", "integration", "completed"]],
+        ),
+        DocumentedQueryTestCase(
+            description="the slowest models of a run are its model builds by resource",
+            question=2,
             columns=("resource_name", "resource_kind"),
             expected_rows=[["customers", "table"], ["orders", "table"]],
         ),
         DocumentedQueryTestCase(
             description="audits that did not pass are listed",
-            question=2,
+            question=3,
             columns=("audit_name", "outcome", "violation_count"),
             expected_rows=[["not_null", "warn", 1]],
         ),
@@ -83,7 +99,7 @@ def test_given_an_orchestrated_build_when_running_the_documented_queries_then_th
     printed: subprocess.CompletedProcess[bytes] = run(project_dir, *commands[-1])
 
     assert built.returncode == 0, built.stderr
-    assert len(commands) == 4
+    assert len(commands) == 5
     assert answered.returncode == 0, answered.stderr
     # Row order follows measured durations, so compare the rows without their order.
     assert sorted(
@@ -164,3 +180,172 @@ def test_given_a_deployment_or_checkout_when_recorded_then_the_git_revision_is_k
     assert recall_rows(project_dir, "SELECT git_revision, git_dirty FROM invocations") == [
         [expected_revision, test_case.expected_dirty]
     ]
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        InvocationOriginTestCase(
+            description="a local configuration check records its mode",
+            args=("debug", "--no-connection"),
+            environment=NO_CI,
+            expected_rows=[["debug", "local", "config_only"]],
+        ),
+        InvocationOriginTestCase(
+            description="a CI connection check is recorded as CI",
+            args=("debug",),
+            environment={**NO_CI, "GITLAB_CI": "true"},
+            expected_rows=[["debug", "ci", "config_and_connection"]],
+        ),
+        InvocationOriginTestCase(
+            description="a CI variable set to false is not CI",
+            args=("build",),
+            environment={**NO_CI, "CI": "false"},
+            expected_rows=[["build", "local", None]],
+        ),
+        InvocationOriginTestCase(
+            description="an integration's run is recorded as integration, also in CI",
+            args=("build",),
+            environment={
+                **NO_CI,
+                "CI": "1",
+                "SQLBUILD_INVOCATION_CONTEXT_JSON": ORCHESTRATED_CONTEXT,
+            },
+            expected_rows=[["build", "integration", None]],
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_a_command_when_recorded_then_its_origin_and_mode_are_known(
+    test_case: InvocationOriginTestCase, tmp_path: Path
+) -> None:
+    project_dir: Path = write_project(tmp_path / "orders_shop")
+
+    completed: subprocess.CompletedProcess[bytes] = run(
+        project_dir, *test_case.args, **test_case.environment
+    )
+
+    assert completed.returncode == 0, completed.stderr.decode(errors="replace")
+    assert (
+        recall_rows(project_dir, "SELECT command, origin, command_mode FROM invocations")
+        == test_case.expected_rows
+    )
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        IntegrationColumnsTestCase(
+            description="a JSON context exposes the integration's job and run",
+            stored_context=ORCHESTRATED_CONTEXT,
+            expected_columns=["dagster", "orders_job", "run-7", 1],
+        ),
+        IntegrationColumnsTestCase(
+            description="an older writer's Python repr context reads as NULL",
+            stored_context=LEGACY_CONTEXT,
+            expected_columns=[None, None, None, None],
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_an_integration_context_when_queried_then_its_fields_are_columns(
+    test_case: IntegrationColumnsTestCase, tmp_path: Path
+) -> None:
+    project_dir: Path = write_project(tmp_path / "orders_shop")
+    built: subprocess.CompletedProcess[bytes] = run(
+        project_dir, "build", SQLBUILD_INVOCATION_CONTEXT_JSON=ORCHESTRATED_CONTEXT
+    )
+    copy_invocation_row(
+        project_dir / "logs" / "recall.sqlite3",
+        invocation_id="copied",
+        external_context=test_case.stored_context,
+    )
+
+    recorded: list[list[object]] = recall_rows(
+        project_dir,
+        "SELECT external_context, integration, integration_job, integration_run_id, "
+        "integration_retry FROM invocations WHERE invocation_id = 'copied'",
+    )
+    written: list[list[object]] = recall_rows(
+        project_dir, "SELECT external_context FROM invocations WHERE invocation_id <> 'copied'"
+    )
+
+    assert built.returncode == 0, built.stderr.decode(errors="replace")
+    assert [row[1:] for row in recorded] == [test_case.expected_columns]
+    assert json.loads(str(written[0][0])) == json.loads(ORCHESTRATED_CONTEXT)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        InvocationPrefixTestCase(
+            description="the short form names the invocation",
+            prefix_length=8,
+            expected_exit_code=0,
+            expected_error_code=None,
+        ),
+        InvocationPrefixTestCase(
+            description="a shorter unique prefix names the invocation",
+            prefix_length=3,
+            expected_exit_code=0,
+            expected_error_code=None,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_a_unique_invocation_prefix_when_printing_output_then_it_names_the_invocation(
+    test_case: InvocationPrefixTestCase, tmp_path: Path
+) -> None:
+    project_dir: Path = write_project(tmp_path / "orders_shop")
+    built: subprocess.CompletedProcess[bytes] = run(project_dir, "build")
+    invocation_id: str = str(
+        recall_rows(project_dir, "SELECT invocation_id FROM invocations")[0][0]
+    )
+
+    printed: subprocess.CompletedProcess[bytes] = run(
+        project_dir, "recall", "output", invocation_id[: test_case.prefix_length]
+    )
+    exact: subprocess.CompletedProcess[bytes] = run(project_dir, "recall", "output", invocation_id)
+
+    assert built.returncode == 0, built.stderr.decode(errors="replace")
+    assert printed.returncode == test_case.expected_exit_code, printed.stderr
+    assert printed.stdout == exact.stdout
+    assert printed.stdout.endswith(built.stdout.splitlines(keepends=True)[-1])
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        InvocationPrefixTestCase(
+            description="an ambiguous prefix fails and lists the matches",
+            prefix_length=8,
+            expected_exit_code=1,
+            expected_error_code="C973",
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_an_ambiguous_invocation_prefix_when_printing_output_then_it_lists_the_matches(
+    test_case: InvocationPrefixTestCase, tmp_path: Path
+) -> None:
+    project_dir: Path = write_project(tmp_path / "orders_shop")
+    built: subprocess.CompletedProcess[bytes] = run(project_dir, "build")
+    invocation_id: str = str(
+        recall_rows(project_dir, "SELECT invocation_id FROM invocations")[0][0]
+    )
+    prefix: str = invocation_id[: test_case.prefix_length]
+    copy_id: str = prefix + "0" * (len(invocation_id) - len(prefix))
+    copy_invocation_row(
+        project_dir / "logs" / "recall.sqlite3", invocation_id=copy_id, external_context="{}"
+    )
+
+    printed: subprocess.CompletedProcess[bytes] = run(project_dir, "recall", "output", prefix)
+
+    assert built.returncode == 0, built.stderr.decode(errors="replace")
+    assert printed.returncode == test_case.expected_exit_code
+    assert f"[{test_case.expected_error_code}]" in printed.stderr.decode()
+    assert f"'{prefix}' matches several invocations; use a longer prefix: " in (
+        printed.stderr.decode()
+    )
+    assert copy_id in printed.stderr.decode()
+    assert invocation_id in printed.stderr.decode()
diff --git a/website/src/content/docs/docs/cli/recall.mdx b/website/src/content/docs/docs/cli/recall.mdx
index 551bb6962f..b7d15976ec 100644
--- a/website/src/content/docs/docs/cli/recall.mdx
+++ b/website/src/content/docs/docs/cli/recall.mdx
@@ -115,6 +115,10 @@ sqb recall output last --tail 20             # only the last 20 lines
 | `--tail` | Print only the last N lines |
 | `--target` | Target whose store to read |
 
+Output shows an invocation ID by its first eight characters; pass that or any other unique prefix.
+A prefix that matches several invocations fails with `C973` and lists them, so you can pass a longer
+one.
+
 Where output past [`max_output_bytes`](/docs/concepts/observability/recall#output-limit) was left
 out, a note line marks the gap in place. Output that Recall could not record completely for any
 other reason is followed by a note on stderr naming the gap.
diff --git a/website/src/content/docs/docs/concepts/observability/recall.mdx b/website/src/content/docs/docs/concepts/observability/recall.mdx
index 7511bbf4c2..c6d88f50dd 100644
--- a/website/src/content/docs/docs/concepts/observability/recall.mdx
+++ b/website/src/content/docs/docs/concepts/observability/recall.mdx
@@ -92,6 +92,8 @@ For every recorded invocation, Recall keeps:
   `error_message`, also when the command reported the failure only through its exit code (for
   example `sqb debug` failing to connect records `ConnectionError` and `C980`);
 - the commit it ran from (`git_revision`), see [Git revision](#git-revision);
+- where it came from (`origin`) and, for an orchestrated run, the job and run that started it, see
+  [Where a run came from](#where-a-run-came-from);
 - its [lifecycle events](/docs/concepts/observability): runs, resource attempts, statements,
   retries, and audit results;
 - the SQL text of every executed statement, stored once per distinct statement and shared across
@@ -129,6 +131,29 @@ export SQLBUILD_GIT_DIRTY=false                        # optional: recorded as g
 `SQLBUILD_GIT_REVISION` takes precedence over the repository. Without `SQLBUILD_GIT_DIRTY`,
 `git_dirty` is `NULL`.
 
+## Where a run came from
+
+Every command is recorded in the store of the target it ran against, including a quick
+`sqb debug --target prod --no-connection` on a laptop. Each invocation therefore says where it came
+from, so a local check reads differently from a failed scheduled build:
+
+- `origin` is `integration` when an integration such as Dagster started the command (it records
+  `external_context`), `ci` when the command ran in CI, and `local` otherwise. CI is detected from
+  the variables CI systems set: `CI`, `GITHUB_ACTIONS`, `GITLAB_CI`, `BUILDKITE`, `JENKINS_URL`,
+  `CIRCLECI`, `TF_BUILD`, `TEAMCITY_VERSION`, `BITBUCKET_BUILD_NUMBER` and `CODEBUILD_BUILD_ID`
+  (any value other than empty, `false` or `0`).
+- `command_mode` says what a diagnostic command checked: `sqb debug` records `config_only` with
+  `--no-connection` and `config_and_connection` otherwise. Recall never stores a command's raw
+  arguments, which can hold secrets.
+- `external_context` is the JSON object the integration supplied. The `invocations` view exposes
+  its common fields as columns: `integration` (for example `dagster`), `integration_job`,
+  `integration_run_id` and `integration_retry`. The Dagster integration also records `step_key`
+  and `partition_key`, which you read from `external_context` with the store's JSON functions.
+
+Invocations recorded by an older SQLBuild version have `origin` and `command_mode` set to `NULL`.
+Older versions also stored `external_context` in a form that is not JSON; for those runs the
+`integration_*` columns are `NULL` and the query does not fail.
+
 ## Settings
 
 Configure Recall in `[recall]` in `sqlbuild_project.toml`. A `[targets.<name>.recall]` table
@@ -378,6 +403,11 @@ in the store's SQL dialect, against these views:
 | `output_tail` | Recent output chunk |
 | `conflicts` | Record written twice with different content |
 
+Output shows an invocation ID by its first eight characters, such as `38ecc95e`. Any unique prefix
+names the invocation: `sqb recall output 38ecc95e` works, and a prefix that matches more than one
+invocation fails and lists the matches. In SQL, match a prefix with
+`invocation_id LIKE '38ecc95e%'` on SQLite or `startsWith(invocation_id, '38ecc95e')` on ClickHouse.
+
 Run `sqb recall schema` for every column, and check the values a column holds before filtering on
 it, for example `SELECT DISTINCT resource_kind, purpose FROM statements`: a model's
 `resource_kind` is its materialization, such as `table` or `view`, and its build statements have
@@ -389,21 +419,26 @@ the last day reads that part of the store instead of folding the whole retention
 
 ### Common questions
 
-These run as written on the local SQLite store; on ClickHouse, read JSON with
-`JSONExtractString(external_context, 'integration', 'job_name')`.
+These run as written on the local SQLite store and on a ClickHouse store.
 
-The latest runs of an orchestrated job, by the context its integration records (the Dagster
-integration records `run_id`, `job_name`, `step_key`, `retry_number` and `partition_key`). Runs
-recorded before this context was stored as JSON hold it in a form the JSON functions cannot read,
-so this query skips them without an error:
+The latest runs of an orchestrated job, by the job name its integration records:
 
 ```bash
-sqb recall query "SELECT invocation_id, invocation_started_at, status, duration_ms
-  FROM invocations
-  WHERE json_extract(external_context, '$.integration.job_name') = 'daily_models'
+sqb recall query "SELECT invocation_id, invocation_started_at, status, duration_ms,
+    integration_run_id, integration_retry
+  FROM invocations WHERE integration_job = 'orders_job'
   ORDER BY invocation_started_at DESC LIMIT 5"
 ```
 
+The latest scheduled and CI builds, leaving out local diagnostics such as `sqb debug`:
+
+```bash
+sqb recall query "SELECT invocation_id, invocation_started_at, command, origin, status
+  FROM invocations
+  WHERE command IN ('build', 'run', 'test', 'audit', 'seed') AND origin IN ('integration', 'ci')
+  ORDER BY invocation_started_at DESC LIMIT 10"
+```
+
 The slowest models in a run:
 
 ```bash
diff --git a/website/src/content/docs/docs/integrations/dagster-reference.mdx b/website/src/content/docs/docs/integrations/dagster-reference.mdx
index 6f845dcf7a..a0b44dc6b9 100644
--- a/website/src/content/docs/docs/integrations/dagster-reference.mdx
+++ b/website/src/content/docs/docs/integrations/dagster-reference.mdx
@@ -107,6 +107,46 @@ A running or completed SQLBuild CLI subprocess.
 5. If the DAG artifact is available, results are matched to the correct Dagster asset keys
 6. Raises `Failure` if the process exited non-zero and `raise_on_error` is set
 
+### Logs
+
+The command is logged at INFO. When Dagster selected a subset of assets, the selection is logged
+as a count at INFO, with the selector file path when SQLBuild read the selection from one; the
+full list of selectors is logged at DEBUG. Each check that did not pass is logged at ERROR, or at
+WARNING for a warn-severity check, with its failing row count when the check reports one, so it
+shows in the default run log view.
+
+### Failure output
+
+The `Failure` that `get_error()` returns, and that `wait()` and `stream()` raise, starts its
+description with what failed: up to 10 failed checks and assets, each with its name, model and
+failing row count (or status), followed by "and N more" when there are more. The command and its
+exit code follow, then the details of each failed asset.
+
+```
+Failed: 1 check
+  audit not_null on orders.order_id: 2 failing rows
+
+SQLBuild CLI command failed with exit code 1: sqb audit --json-output ...
+```
+
+Checks that only warned are not listed; they do not fail the command. The failure metadata holds:
+
+| Key | Description |
+|-----|-------------|
+| `command` | The command that ran |
+| `project_dir` | The project directory |
+| `selection_count` | How many selectors Dagster passed to SQLBuild |
+| `selector_file` | The selector file the selection was passed in, or empty |
+| `execution_status` | The status of the execution JSON |
+| `incomplete_assets` | Assets that did not complete, as `kind:name (status)` |
+| `failed_checks` | Up to 100 failed checks, each with `kind`, `name`, `asset`, `column`, `status` and `failing_rows` |
+| `failed_check_count` | How many checks failed |
+| `microbatch_limits` | Microbatch limit evidence of incomplete assets, when there is any |
+
+The metadata no longer lists the selection itself: `selection` was replaced by `selection_count`
+and `selector_file`. Without an execution JSON, the metadata holds the non-empty `stdout_tail` and
+`stderr_tail` instead of the execution keys.
+
 ## sqlbuild_assets
 
 Decorator that creates a Dagster multi-asset definition from a SQLBuild DAG artifact.