diff --git a/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_calls.rs b/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_calls.rs
index a8a190f140..3278baa93c 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_calls.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_calls.rs
@@ -1,7 +1,7 @@
 //! Calls into the ClickHouse adapter's Recall store operations, through the native host.
 
 use pyo3::prelude::{Bound, FromPyObject, PyAny, PyAnyMethods, PyErr, PyResult, Python};
-use pyo3::types::{PyBytes, PyDict, PyDictMethods, PyInt, PyList, PyString};
+use pyo3::types::{PyBool, PyBytes, PyDict, PyDictMethods, PyInt, PyList, PyString};
 use sqlbuild_config::models::ConfigValue;
 use sqlbuild_host::entries::models::AdapterHost;
 use sqlbuild_recall::clickhouse::main::error_kind::error_kind;
@@ -85,10 +85,17 @@ fn reply<'py>(
     }
 }
 
-/// Connect with the store's connection options; a `quiet` caller, Recall's delivery thread, never has its output captured.
+/// How a store connection is used: the delivery thread writes and is never captured; a reader skips the session settings under `readonly = 1`, so that user can read.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum ConnectionRole {
+    Writer,
+    Reader,
+}
+
+/// Connect with the store's connection options for `role`.
 pub(crate) fn connect(
     py: Python<'_>,
-    (config, quiet): (&[(String, ConfigValue)], bool),
+    (config, role): (&[(String, ConfigValue)], ConnectionRole),
     action: &str,
 ) -> Result<StoreHandle, StoreError> {
     let options = PyDict::new(py);
@@ -104,12 +111,18 @@ pub(crate) fn connect(
             ("config", options.into_any()),
             (
                 "quiet_thread_attribute",
-                if quiet {
+                if role == ConnectionRole::Writer {
                     PyString::new(py, OUTPUT_SUPPRESSED_ATTRIBUTE).into_any()
                 } else {
                     py.None().into_bound(py)
                 },
             ),
+            (
+                "read_only",
+                PyBool::new(py, role == ConnectionRole::Reader)
+                    .to_owned()
+                    .into_any(),
+            ),
         ],
         action,
     )?
@@ -132,11 +145,11 @@ fn cell_value(cell: Option<Cell>) -> RecallValue {
     }
 }
 
-/// The columns, rows and statistics of `sql` run with `settings`.
+/// The columns, at most `row_cap` rows and the statistics of `sql`, run with those of `settings` the session does not already hold.
 pub(crate) fn query(
     py: Python<'_>,
     handle: StoreHandle,
-    (sql, settings): (&str, &[(&str, i64)]),
+    (sql, settings, row_cap): (&str, &[(&str, i64)], Option<usize>),
     action: &str,
 ) -> Result<(QueryResult, QueryStatistics), StoreError> {
     let query_settings = PyDict::new(py);
@@ -152,6 +165,13 @@ pub(crate) fn query(
             ("handle", handle_value(py, handle)),
             ("sql", PyString::new(py, sql).into_any()),
             ("settings", query_settings.into_any()),
+            (
+                "row_cap",
+                row_cap.map_or_else(
+                    || py.None().into_bound(py),
+                    |cap| PyInt::new(py, cap).into_any(),
+                ),
+            ),
         ],
         action,
     )?;
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 31b43d292c..51f6942e87 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_store.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/clickhouse_store.rs
@@ -46,9 +46,11 @@ use sqlbuild_recall::store::models::{
 use sqlbuild_recall::store::types::{RecallStore, RecallStoreOpener};
 
 use crate::bindings::_helpers::recall::clickhouse_calls::{
-    self as calls, QueryStatistics, StoreHandle,
+    self as calls, ConnectionRole, QueryStatistics, StoreHandle,
+};
+use crate::bindings::_helpers::recall::constants::{
+    READ_ONLY_QUERIES, READONLY_SETTING, SECONDS_PER_DAY, UNKNOWN_TABLE_ERROR,
 };
-use crate::bindings::_helpers::recall::constants::{SECONDS_PER_DAY, UNKNOWN_TABLE_ERROR};
 
 /// Where a ClickHouse store lives and how long it keeps history.
 #[derive(Clone, Debug)]
@@ -143,7 +145,7 @@ impl ClickHouseStore {
     }
 
     fn query(&self, py: Python<'_>, sql: &str, action: &str) -> Result<QueryResult, StoreError> {
-        calls::query(py, self.connection()?, (sql, &[]), action).map(|(result, _)| result)
+        calls::query(py, self.connection()?, (sql, &[], None), action).map(|(result, _)| result)
     }
 
     fn command(&self, py: Python<'_>, sql: &str, action: &str) -> Result<(), StoreError> {
@@ -591,7 +593,7 @@ impl RecallStoreOpener for ClickHouseStoreOpener {
                 location: self.location.clone(),
                 connection: Some(calls::connect(
                     py,
-                    (&self.location.config, true),
+                    (&self.location.config, ConnectionRole::Writer),
                     &self.location.connect_action(),
                 )?),
                 columns: HashMap::new(),
@@ -645,7 +647,7 @@ impl ClickHouseReader {
             location: location.clone(),
             connection: Some(calls::connect(
                 py,
-                (&location.config, false),
+                (&location.config, ConnectionRole::Reader),
                 &location.connect_action(),
             )?),
             columns: HashMap::new(),
@@ -683,7 +685,7 @@ impl ClickHouseReader {
             location: location.clone(),
             connection: Some(calls::connect(
                 py,
-                (&location.config, false),
+                (&location.config, ConnectionRole::Reader),
                 &location.connect_action(),
             )?),
             columns: HashMap::new(),
@@ -719,15 +721,13 @@ impl ClickHouseReader {
         sql: &str,
         limit: Option<usize>,
     ) -> Result<(QueryResult, QueryStatistics), StoreError> {
-        let mut settings: Vec<(&str, i64)> = vec![("readonly", 2)];
-        if let Some(limit) = limit {
-            settings.push(("limit", i64::try_from(limit).unwrap_or(i64::MAX - 1) + 1));
-        }
+        let settings: [(&str, i64); 1] = [(READONLY_SETTING, READ_ONLY_QUERIES)];
+        let row_cap: Option<usize> = limit.map(|limit| limit.saturating_add(1));
         let run = || {
             calls::query(
                 py,
                 self.store.connection()?,
-                (sql, &settings),
+                (sql, &settings, row_cap),
                 "run the Recall query",
             )
         };
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs b/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs
index f2765ac651..2b15714f83 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/recall/constants.rs
@@ -100,6 +100,9 @@ pub(crate) const CLOSE_POLL_MS: u64 = 50;
 pub(crate) const QUARANTINE_START_WAIT_MS: u64 = 200;
 /// The ClickHouse error name of a query naming a relation that does not exist yet.
 pub(crate) const UNKNOWN_TABLE_ERROR: &str = "UNKNOWN_TABLE";
+/// `sqb recall query` runs read-only: the setting, and the value that still allows query settings. Not sent when the session is already read-only.
+pub(crate) const READONLY_SETTING: &str = "readonly";
+pub(crate) const READ_ONLY_QUERIES: i64 = 2;
 /// Seconds in a day, for the artifact refresh cutoff.
 pub(crate) const SECONDS_PER_DAY: u64 = 24 * 60 * 60;
 /// The parameter a lifecycle-event sink takes its event in.
diff --git a/crates/sqlbuild-recall/src/clickhouse/constants.rs b/crates/sqlbuild-recall/src/clickhouse/constants.rs
index d4a540e63a..d11b50dc61 100644
--- a/crates/sqlbuild-recall/src/clickhouse/constants.rs
+++ b/crates/sqlbuild-recall/src/clickhouse/constants.rs
@@ -4,8 +4,8 @@ use crate::schema::models::Table;
 
 /// Rows one insert batch takes; ClickHouse prefers few large inserts to many small ones.
 pub const BATCH_ROWS: usize = 5_000;
-/// Longest wait for a batch to fill, so each table gets at most about one part per second per writer.
-pub const BATCH_WINDOW_MS: u64 = 1_000;
+/// Longest wait for a batch to fill, so a long command adds at most one part per table every five seconds. A closing session sends what it holds at once, and heartbeats never wait for it.
+pub const BATCH_WINDOW_MS: u64 = 5_000;
 /// ClickHouse `READONLY`.
 pub const READONLY_CODE: u64 = 164;
 /// ClickHouse `DATABASE_ACCESS_DENIED`.
diff --git a/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs b/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs
index e042973bfc..aea4bceea2 100644
--- a/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs
+++ b/crates/sqlbuild-recall/src/schema/_helpers/clickhouse_views.rs
@@ -1,10 +1,85 @@
-//! The bodies of the supported views in ClickHouse: rows deduplicated by key, within retention.
+//! The bodies of the supported views in ClickHouse: each key counted once, within retention.
 
 use crate::schema::_helpers::rendering::clickhouse_table;
 use crate::schema::constants::{OUTPUT_TAIL_RECORDS, REPLACING_ENGINE};
 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] = [
+    ("ctx_command", "Nullable(String)", "NULL"),
+    ("ctx_subcommand", "Nullable(String)", "NULL"),
+    ("ctx_target", "Nullable(String)", "NULL"),
+    ("ctx_adapter", "Nullable(String)", "NULL"),
+    ("ctx_project_name", "Nullable(String)", "NULL"),
+    ("ctx_selection", "Nullable(String)", "NULL"),
+    ("ctx_exclusion", "Nullable(String)", "NULL"),
+    ("ctx_sqlbuild_version", "Nullable(String)", "NULL"),
+    ("ctx_git_revision", "Nullable(String)", "NULL"),
+    ("ctx_git_dirty", "Nullable(UInt8)", "NULL"),
+    ("ctx_host_token", "Nullable(String)", "NULL"),
+    ("ctx_pid", "Nullable(Int64)", "NULL"),
+    ("ctx_external_context", "Nullable(String)", "NULL"),
+    ("ctx_recall_filter_active", "Nullable(UInt8)", "NULL"),
+    ("heartbeat_at", "Nullable(DateTime64(6, 'UTC'))", "NULL"),
+    ("beat_stale_at", "Nullable(DateTime64(6, 'UTC'))", "NULL"),
+    ("beat_command", "Nullable(String)", "NULL"),
+    ("beat_subcommand", "Nullable(String)", "NULL"),
+    ("lifecycle_event", "Nullable(String)", "NULL"),
+    (
+        "event_occurred_at",
+        "Nullable(DateTime64(6, 'UTC'))",
+        "NULL",
+    ),
+    ("event_sequence", "Nullable(Int64)", "NULL"),
+    ("event_command", "Nullable(String)", "NULL"),
+    ("event_exit_code", "Nullable(Int64)", "NULL"),
+    ("event_duration_ms", "Nullable(Float64)", "NULL"),
+    ("event_error_type", "Nullable(String)", "NULL"),
+    ("event_error_code", "Nullable(String)", "NULL"),
+    ("event_error_message", "Nullable(String)", "NULL"),
+    ("event_is_terminal", "UInt8", "0"),
+];
+
+/// The statement columns each statement event carries into the second fold, in tuple order.
+const STATEMENT_FACT_COLUMNS: [&str; 17] = [
+    "statement_id",
+    "event_type",
+    "occurred_at",
+    "invocation_sequence",
+    "run_id",
+    "resource_attempt_id",
+    "operation_id",
+    "intent",
+    "statement_kind",
+    "purpose",
+    "sql_digest",
+    "query_id",
+    "error_type",
+    "error_code",
+    "duration_ms",
+    "row_count",
+    "affected_rows",
+];
+
+/// The columns the `output_invocations` pass reads: the chunk's own values default to what adds nothing to a sum, and the delivery and context values to NULL.
+const OUTPUT_SOURCE_COLUMNS: [(&str, &str, &str); 14] = [
+    ("chunk_stream", "String", "''"),
+    ("chunk_bytes", "Int64", "0"),
+    ("chunk_sequence", "Nullable(Int64)", "NULL"),
+    ("chunk_dropped_before", "Int64", "0"),
+    ("chunk_spans", "Int64", "0"),
+    ("chunk_omitted_bytes", "Int64", "0"),
+    ("delivery_accepted", "Nullable(Int64)", "NULL"),
+    ("delivery_delivered", "Nullable(Int64)", "NULL"),
+    ("delivery_filtered", "Nullable(Int64)", "NULL"),
+    ("delivery_dropped", "Nullable(Int64)", "NULL"),
+    ("delivery_failed", "Nullable(Int64)", "NULL"),
+    ("delivery_flush_complete", "Nullable(UInt8)", "NULL"),
+    ("context_filter_active", "Nullable(UInt8)", "NULL"),
+    ("is_listed", "UInt8", "0"),
+];
+
 /// `table`'s rows within its TTL, one copy per key: `FINAL` for replacing tables, so filters on the sorting key reach the read, else the first ingested copy. SQL text outlives retention like its TTL, so every reference still within retention finds it.
 fn deduplicated(table: Table, options: &DdlOptions) -> String {
     let spec = table_spec(table);
@@ -25,112 +100,236 @@ fn deduplicated(table: Table, options: &DdlOptions) -> String {
     )
 }
 
-/// Every fold groups by `invocation_id` and `invocation_started_at`, which an invocation's rows share, so a filter on either reaches the table reads instead of folding the whole retention window.
+/// `table`'s rows within its TTL that match `filter`, every stored copy included. Only for folds whose result is the same however often a row repeats (`any`, `min`, `max`, `argMin`, `argMax`, a count compared with zero): copies of one key hold the same content, since a write with other content is refused as a conflict.
+fn retained(table: Table, options: &DdlOptions, filter: &str) -> String {
+    let spec = table_spec(table);
+    let days: u32 = options.retention_days.saturating_add(spec.ttl_extra_days);
+    let filter: String = if filter.is_empty() {
+        String::new()
+    } else {
+        format!(" AND {filter}")
+    };
+    format!(
+        "(SELECT * FROM {name} WHERE {expiry} >= now64(6) - INTERVAL {days} DAY{filter})",
+        name = clickhouse_table(spec.name, options),
+        expiry = spec.expiry_column,
+    )
+}
+
+/// The value `values` gives column `name`, if any.
+fn branch_value<'a>(values: &[(&str, &'a str)], name: &str) -> Option<&'a str> {
+    values
+        .iter()
+        .find(|(column, _)| *column == name)
+        .map(|(_, value)| *value)
+}
+
+/// One branch of a single-pass `UNION ALL` reading `from`: `invocation_id`, `invocation_started_at`, the `source` label, then every column of `columns` (name, type, default), taken from `values` where given and the default elsewhere. Column names differ from the tables' own, so no alias hides a table column.
+fn union_branch(
+    source: &str,
+    columns: &[(&str, &str, &str)],
+    values: &[(&str, &str)],
+    from: &str,
+) -> String {
+    let selected: Vec<String> = columns
+        .iter()
+        .map(|(name, column_type, default)| {
+            let value: &str = branch_value(values, name).unwrap_or(*default);
+            format!("        CAST({value} AS {column_type}) AS {name}")
+        })
+        .collect();
+    format!(
+        "    SELECT invocation_id, invocation_started_at, '{source}' AS source,\n{}\n    FROM {from}",
+        selected.join(",\n")
+    )
+}
+
+/// One pass over the context rows, heartbeats and lifecycle events, folded by `invocation_id` and `invocation_started_at`, which an invocation's rows share: each table is read once, a filter on either column reaches every read, and no join needs `join_use_nulls`.
 fn invocations(options: &DdlOptions) -> String {
-    let context = deduplicated(Table::Invocations, options);
-    let events = deduplicated(Table::Events, options);
-    let beats = deduplicated(Table::Heartbeats, options);
+    let context = retained(Table::Invocations, options, "");
+    let events = retained(
+        Table::Events,
+        options,
+        "event_type IN ('invocation_started', 'invocation_completed', 'invocation_failed')",
+    );
+    let beats = retained(Table::Heartbeats, options, "");
+    let columns = &INVOCATION_SOURCE_COLUMNS;
+    let context_rows = union_branch(
+        "context",
+        columns,
+        &[
+            ("ctx_command", "command"),
+            ("ctx_subcommand", "subcommand"),
+            ("ctx_target", "target"),
+            ("ctx_adapter", "adapter"),
+            ("ctx_project_name", "project_name"),
+            ("ctx_selection", "selection"),
+            ("ctx_exclusion", "exclusion"),
+            ("ctx_sqlbuild_version", "sqlbuild_version"),
+            ("ctx_git_revision", "git_revision"),
+            ("ctx_git_dirty", "git_dirty"),
+            ("ctx_host_token", "host_token"),
+            ("ctx_pid", "pid"),
+            ("ctx_external_context", "external_context"),
+            ("ctx_recall_filter_active", "recall_filter_active"),
+        ],
+        &context,
+    );
+    let beat_rows = union_branch(
+        "beat",
+        columns,
+        &[
+            ("heartbeat_at", "beat_at"),
+            ("beat_stale_at", "stale_at"),
+            ("beat_command", "command"),
+            ("beat_subcommand", "subcommand"),
+        ],
+        &beats,
+    );
+    let event_rows = union_branch(
+        "event",
+        columns,
+        &[
+            ("lifecycle_event", "event_type"),
+            ("event_occurred_at", "occurred_at"),
+            ("event_sequence", "invocation_sequence"),
+            ("event_command", "command"),
+            ("event_exit_code", "exit_code"),
+            ("event_duration_ms", "duration_ms"),
+            ("event_error_type", "error_type"),
+            ("event_error_code", "error_code"),
+            (
+                "event_error_message",
+                "JSONExtract(payload, 'error_message', 'Nullable(String)')",
+            ),
+            (
+                "event_is_terminal",
+                "event_type IN ('invocation_completed', 'invocation_failed')",
+            ),
+        ],
+        &events,
+    );
+    let order = "tuple(event_sequence, event_occurred_at)";
+    let context_fields: String = INVOCATION_SOURCE_COLUMNS
+        .iter()
+        .filter_map(|(name, _, _)| name.strip_prefix("ctx_"))
+        .map(|field| format!("        anyIf(ctx_{field}, source = 'context') AS context_{field},"))
+        .collect::<Vec<_>>()
+        .join("\n");
+    let terminal = |column: &str, alias: &str| {
+        format!("        argMaxIf({column}, {order}, event_is_terminal = 1) AS {alias},")
+    };
     format!(
         "WITH
-ids AS (
-    SELECT invocation_id, invocation_started_at
-    FROM (
-        SELECT invocation_id, invocation_started_at FROM {context}
-        UNION ALL SELECT invocation_id, invocation_started_at FROM {beats}
-        UNION ALL SELECT invocation_id, invocation_started_at FROM {events}
-            WHERE event_type IN ('invocation_started', 'invocation_completed', 'invocation_failed')
-    )
-    GROUP BY invocation_id, invocation_started_at
-),
-started AS (
-    SELECT invocation_id, invocation_started_at, min(occurred_at) AS started_at,
-        any(command) AS command
-    FROM {events} WHERE event_type = 'invocation_started'
-    GROUP BY invocation_id, invocation_started_at
+source_rows AS (
+{context_rows}
+    UNION ALL
+{beat_rows}
+    UNION ALL
+{event_rows}
 ),
-terminals AS (
+folded AS (
     SELECT invocation_id, invocation_started_at,
-        argMax(event_type, tuple(invocation_sequence, occurred_at)) AS terminal_type,
-        argMax(occurred_at, tuple(invocation_sequence, occurred_at)) AS terminal_at,
-        argMax(exit_code, tuple(invocation_sequence, occurred_at)) AS terminal_exit_code,
-        argMax(duration_ms, tuple(invocation_sequence, occurred_at)) AS terminal_duration_ms,
-        argMax(error_type, tuple(invocation_sequence, occurred_at)) AS terminal_error_type,
-        argMax(error_code, tuple(invocation_sequence, occurred_at)) AS terminal_error_code,
-        argMax(JSONExtract(payload, 'error_message', 'Nullable(String)'),
-            tuple(invocation_sequence, occurred_at)) AS terminal_error_message
-    FROM {events}
-    WHERE event_type IN ('invocation_completed', 'invocation_failed')
+{context_fields}
+        minIf(event_occurred_at, lifecycle_event = 'invocation_started') AS started_occurred_at,
+        anyIf(event_command, lifecycle_event = 'invocation_started') AS started_command,
+{terminal_type}
+{terminal_at}
+{terminal_exit_code}
+{terminal_duration_ms}
+{terminal_error_type}
+{terminal_error_code}
+{terminal_error_message}
+        maxIf(heartbeat_at, source = 'beat') AS last_heartbeat_at,
+        maxIf(beat_stale_at, source = 'beat') AS stale_at,
+        minIf(beat_command, source = 'beat') AS first_beat_command,
+        minIf(beat_subcommand, source = 'beat') AS first_beat_subcommand
+    FROM source_rows
     GROUP BY invocation_id, invocation_started_at
-),
-beats AS (
-    SELECT invocation_id, invocation_started_at, max(beat_at) AS last_heartbeat_at, max(stale_at) AS stale_at,
-        min(command) AS command, min(subcommand) AS subcommand
-    FROM {beats} GROUP BY invocation_id, invocation_started_at
 )
 SELECT
-    ids.invocation_id AS invocation_id,
-    ids.invocation_started_at AS invocation_started_at,
-    coalesce(started.started_at, ids.invocation_started_at) AS started_at,
+    invocation_id,
+    invocation_started_at,
+    coalesce(started_occurred_at, invocation_started_at) AS started_at,
     multiIf(
-        terminals.terminal_type = 'invocation_completed', 'completed',
-        terminals.terminal_type = 'invocation_failed', 'failed',
-        beats.stale_at IS NULL, 'unknown',
-        beats.stale_at >= now64(6), 'running',
+        terminal_type = 'invocation_completed', 'completed',
+        terminal_type = 'invocation_failed', 'failed',
+        stale_at IS NULL, 'unknown',
+        stale_at >= now64(6), 'running',
         'abandoned'
     ) AS status,
-    if(terminals.terminal_type IS NOT NULL, terminals.terminal_at,
-        if(beats.stale_at < now64(6), beats.last_heartbeat_at, NULL)) AS ended_at,
-    if(terminals.terminal_type IS NOT NULL, terminals.terminal_duration_ms,
-        if(beats.stale_at < now64(6),
-            toFloat64(dateDiff('millisecond', started_at, beats.last_heartbeat_at)), NULL))
+    if(terminal_type IS NOT NULL, terminal_at,
+        if(stale_at < now64(6), last_heartbeat_at, NULL)) AS ended_at,
+    if(terminal_type IS NOT NULL, terminal_duration_ms,
+        if(stale_at < now64(6),
+            toFloat64(dateDiff('millisecond', started_at, last_heartbeat_at)), NULL))
         AS duration_ms,
-    terminals.terminal_exit_code AS exit_code,
-    terminals.terminal_error_type AS error_type,
-    terminals.terminal_error_code AS error_code,
-    terminals.terminal_error_message AS error_message,
-    coalesce(context.command, started.command, beats.command) AS command,
-    coalesce(context.subcommand, beats.subcommand) AS subcommand,
-    context.target AS target,
-    context.adapter AS adapter,
-    context.project_name AS project_name,
-    context.selection AS selection,
-    context.exclusion AS exclusion,
-    context.sqlbuild_version AS sqlbuild_version,
-    context.git_revision AS git_revision,
-    context.git_dirty AS git_dirty,
-    context.host_token AS host_token,
-    context.pid AS pid,
-    context.external_context AS external_context,
-    context.recall_filter_active AS recall_filter_active,
-    beats.last_heartbeat_at AS last_heartbeat_at
-FROM ids
-LEFT JOIN {context} AS context ON context.invocation_id = ids.invocation_id
-    AND context.invocation_started_at = ids.invocation_started_at
-LEFT JOIN started ON started.invocation_id = ids.invocation_id
-    AND started.invocation_started_at = ids.invocation_started_at
-LEFT JOIN terminals ON terminals.invocation_id = ids.invocation_id
-    AND terminals.invocation_started_at = ids.invocation_started_at
-LEFT JOIN beats ON beats.invocation_id = ids.invocation_id
-    AND beats.invocation_started_at = ids.invocation_started_at
-SETTINGS join_use_nulls = 1"
+    terminal_exit_code AS exit_code,
+    terminal_error_type AS error_type,
+    terminal_error_code AS error_code,
+    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_target AS target,
+    context_adapter AS adapter,
+    context_project_name AS project_name,
+    context_selection AS selection,
+    context_exclusion AS exclusion,
+    context_sqlbuild_version AS sqlbuild_version,
+    context_git_revision AS git_revision,
+    context_git_dirty AS git_dirty,
+    context_host_token AS host_token,
+    context_pid AS pid,
+    context_external_context AS external_context,
+    context_recall_filter_active AS recall_filter_active,
+    last_heartbeat_at
+FROM folded",
+        terminal_type = terminal("lifecycle_event", "terminal_type"),
+        terminal_at = terminal("event_occurred_at", "terminal_at"),
+        terminal_exit_code = terminal("event_exit_code", "terminal_exit_code"),
+        terminal_duration_ms = terminal("event_duration_ms", "terminal_duration_ms"),
+        terminal_error_type = terminal("event_error_type", "terminal_error_type"),
+        terminal_error_code = terminal("event_error_code", "terminal_error_code"),
+        terminal_error_message = terminal("event_error_message", "terminal_error_message"),
     )
 }
 
-/// One fold per statement, grouped by its invocation as well, so a filter on `invocation_id` or `invocation_started_at` reaches the table reads.
+/// Two folds over one read of the events: by resource, which names the resource and gathers its statement events, then by statement. Both group by `invocation_id` and `invocation_started_at`, so a filter on either reaches the read; `sql_available` looks digests up in the stored SQL, which is shared across invocations.
 fn statements(options: &DdlOptions) -> String {
-    let events = deduplicated(Table::Events, options);
+    let events = retained(Table::Events, options, "");
     let artifacts = clickhouse_table(table_spec(Table::Artifacts).name, options);
     let order = "tuple(invocation_sequence, occurred_at)";
     let terminal = "event_type IN ('statement_completed', 'statement_failed')";
+    let statement_event = "statement_id IS NOT NULL AND startsWith(event_type, 'statement_')";
+    let facts: String = STATEMENT_FACT_COLUMNS.join(", ");
+    let unpacked: String = STATEMENT_FACT_COLUMNS
+        .iter()
+        .enumerate()
+        .map(|(index, column)| format!("fact.{} AS {column}", index + 1))
+        .collect::<Vec<_>>()
+        .join(", ");
     format!(
         "WITH
-facts AS (
-    SELECT * FROM {events} WHERE statement_id IS NOT NULL AND startsWith(event_type, 'statement_')
+by_resource AS (
+    SELECT invocation_id, invocation_started_at, resource_id,
+        min(resource_name) AS group_resource_name,
+        minIf(resource_kind, resource_name IS NOT NULL) AS group_resource_kind,
+        groupArrayIf(tuple({facts}), {statement_event}) AS facts
+    FROM {events}
+    WHERE ({statement_event}) OR (resource_id IS NOT NULL AND resource_name IS NOT NULL)
+    GROUP BY invocation_id, invocation_started_at, resource_id
+),
+statement_events AS (
+    SELECT invocation_id, invocation_started_at, resource_id, group_resource_name, group_resource_kind,
+        {unpacked}
+    FROM by_resource ARRAY JOIN facts AS fact
 ),
 folded AS (
     SELECT invocation_id, invocation_started_at, statement_id,
         argMin(run_id, {order}) AS first_run_id,
         argMin(resource_id, {order}) AS first_resource_id,
+        argMinIf(tuple(group_resource_name, group_resource_kind), {order}, resource_id IS NOT NULL)
+            AS first_resource,
         argMin(resource_attempt_id, {order}) AS first_resource_attempt_id,
         argMin(operation_id, {order}) AS first_operation_id,
         argMin(intent, {order}) AS first_intent,
@@ -140,7 +339,7 @@ folded AS (
         minIf(occurred_at, event_type = 'statement_started') AS started_at,
         minIf(occurred_at, event_type = 'statement_submitted') AS submitted_at,
         max(query_id) AS any_query_id,
-        countIf({terminal}) AS terminal_count,
+        countIf({terminal}) > 0 AS has_terminal,
         argMaxIf(event_type, {order}, {terminal}) AS last_event_type,
         argMaxIf(occurred_at, {order}, {terminal}) AS last_ended_at,
         argMaxIf(query_id, {order}, {terminal}) AS last_query_id,
@@ -149,51 +348,38 @@ folded AS (
         argMaxIf(duration_ms, {order}, {terminal}) AS last_duration_ms,
         argMaxIf(row_count, {order}, {terminal}) AS last_row_count,
         argMaxIf(affected_rows, {order}, {terminal}) AS last_affected_rows
-    FROM facts GROUP BY invocation_id, invocation_started_at, statement_id
-),
-resources AS (
-    SELECT invocation_id, invocation_started_at, resource_id,
-        min(resource_name) AS first_resource_name, min(resource_kind) AS first_resource_kind
-    FROM {events}
-    WHERE resource_id IS NOT NULL AND resource_name IS NOT NULL
-    GROUP BY invocation_id, invocation_started_at, resource_id
+    FROM statement_events GROUP BY invocation_id, invocation_started_at, statement_id
 )
 SELECT
-    folded.invocation_id AS invocation_id,
-    folded.invocation_started_at AS invocation_started_at,
-    folded.statement_id AS statement_id,
-    folded.first_run_id AS run_id,
-    folded.first_resource_id AS resource_id,
-    resources.first_resource_name AS resource_name,
-    resources.first_resource_kind AS resource_kind,
-    folded.first_resource_attempt_id AS resource_attempt_id,
-    folded.first_operation_id AS operation_id,
-    folded.first_intent AS intent,
-    folded.first_statement_kind AS statement_kind,
-    folded.first_purpose AS purpose,
-    if(folded.terminal_count > 0, coalesce(folded.last_query_id, folded.any_query_id),
-        folded.any_query_id) AS query_id,
-    folded.started_at AS started_at,
-    folded.submitted_at AS submitted_at,
-    if(folded.terminal_count > 0, folded.last_ended_at, NULL) AS ended_at,
+    invocation_id,
+    invocation_started_at,
+    statement_id,
+    first_run_id AS run_id,
+    first_resource_id AS resource_id,
+    tupleElement(first_resource, 1) AS resource_name,
+    tupleElement(first_resource, 2) AS resource_kind,
+    first_resource_attempt_id AS resource_attempt_id,
+    first_operation_id AS operation_id,
+    first_intent AS intent,
+    first_statement_kind AS statement_kind,
+    first_purpose AS purpose,
+    if(has_terminal, coalesce(last_query_id, any_query_id), any_query_id) AS query_id,
+    started_at,
+    submitted_at,
+    if(has_terminal, last_ended_at, NULL) AS ended_at,
     multiIf(
-        folded.terminal_count = 0, 'no_terminal',
-        folded.last_event_type = 'statement_completed', 'completed',
+        NOT has_terminal, 'no_terminal',
+        last_event_type = 'statement_completed', 'completed',
         'failed'
     ) AS outcome,
-    if(folded.terminal_count > 0, folded.last_error_type, NULL) AS error_type,
-    if(folded.terminal_count > 0, folded.last_error_code, NULL) AS error_code,
-    if(folded.terminal_count > 0, folded.last_duration_ms, NULL) AS duration_ms,
-    if(folded.terminal_count > 0, folded.last_row_count, NULL) AS row_count,
-    if(folded.terminal_count > 0, folded.last_affected_rows, NULL) AS affected_rows,
-    folded.first_sql_digest AS sql_digest,
-    concat('sha256:', folded.first_sql_digest) IN (SELECT digest FROM {artifacts}) AS sql_available
-FROM folded
-LEFT JOIN resources
-    ON resources.invocation_id = folded.invocation_id
-    AND resources.invocation_started_at = folded.invocation_started_at
-    AND resources.resource_id = folded.first_resource_id
-SETTINGS join_use_nulls = 1"
+    if(has_terminal, last_error_type, NULL) AS error_type,
+    if(has_terminal, last_error_code, NULL) AS error_code,
+    if(has_terminal, last_duration_ms, NULL) AS duration_ms,
+    if(has_terminal, last_row_count, NULL) AS row_count,
+    if(has_terminal, last_affected_rows, NULL) AS affected_rows,
+    first_sql_digest AS sql_digest,
+    concat('sha256:', first_sql_digest) IN (SELECT digest FROM {artifacts}) AS sql_available
+FROM folded"
     )
 }
 
@@ -230,59 +416,99 @@ WHERE event_type = 'audit_completed'"
     )
 }
 
+/// One pass over the output chunks, the output delivery row and the context row, folded by invocation without joins. Chunks are counted and summed, so they keep `FINAL`; the delivery and context rows are only picked with `any`.
 fn output_invocations(options: &DdlOptions) -> String {
     let output = deduplicated(Table::Output, options);
-    let delivery = deduplicated(Table::Delivery, options);
-    let context = deduplicated(Table::Invocations, options);
+    let delivery = retained(Table::Delivery, options, "stream = 'output'");
+    let context = retained(Table::Invocations, options, "");
+    let columns = &OUTPUT_SOURCE_COLUMNS;
+    let chunk_rows = union_branch(
+        "output",
+        columns,
+        &[
+            ("chunk_stream", "stream"),
+            ("chunk_bytes", "byte_count"),
+            ("chunk_sequence", "sequence"),
+            ("chunk_dropped_before", "dropped_before"),
+            ("chunk_spans", "coalesce(omitted_sequences, 1)"),
+            ("chunk_omitted_bytes", "coalesce(omitted_bytes, 0)"),
+            ("is_listed", "1"),
+        ],
+        &output,
+    );
+    let delivery_rows = union_branch(
+        "delivery",
+        columns,
+        &[
+            ("delivery_accepted", "accepted"),
+            ("delivery_delivered", "delivered"),
+            ("delivery_filtered", "filtered"),
+            ("delivery_dropped", "dropped"),
+            ("delivery_failed", "failed"),
+            ("delivery_flush_complete", "flush_complete"),
+            ("is_listed", "1"),
+        ],
+        &delivery,
+    );
+    let context_rows = union_branch(
+        "context",
+        columns,
+        &[("context_filter_active", "recall_filter_active")],
+        &context,
+    );
     format!(
         "WITH
-received AS (
-    SELECT invocation_id,
-        min(invocation_started_at) AS invocation_started_at,
-        count() AS chunks,
-        sum(byte_count) AS bytes,
-        sumIf(byte_count, stream = 'stdout') AS stdout_bytes,
-        sumIf(byte_count, stream = 'stderr') AS stderr_bytes,
-        max(sequence) AS max_sequence,
-        sum(dropped_before) AS queue_drops,
-        sum(coalesce(omitted_sequences, 1)) AS spans,
-        sum(coalesce(omitted_bytes, 0)) AS truncated_bytes
-    FROM {output} GROUP BY invocation_id
+source_rows AS (
+{chunk_rows}
+    UNION ALL
+{delivery_rows}
+    UNION ALL
+{context_rows}
 ),
-delivery AS (SELECT * FROM {delivery} WHERE stream = 'output'),
-ids AS (
-    SELECT DISTINCT invocation_id, invocation_started_at
-    FROM (
-        SELECT invocation_id, invocation_started_at FROM received
-        UNION ALL SELECT invocation_id, invocation_started_at FROM delivery
-    )
+folded AS (
+    SELECT invocation_id, invocation_started_at,
+        countIf(source = 'output') AS chunks,
+        sumIf(chunk_bytes, source = 'output') AS bytes,
+        sumIf(chunk_bytes, source = 'output' AND chunk_stream = 'stdout') AS stdout_bytes,
+        sumIf(chunk_bytes, source = 'output' AND chunk_stream = 'stderr') AS stderr_bytes,
+        maxIf(chunk_sequence, source = 'output') AS max_sequence,
+        sumIf(chunk_dropped_before, source = 'output') AS queue_drops,
+        sumIf(chunk_spans, source = 'output') AS spans,
+        sumIf(chunk_omitted_bytes, source = 'output') AS truncated_bytes,
+        anyIf(delivery_accepted, source = 'delivery') AS accepted,
+        anyIf(delivery_delivered, source = 'delivery') AS delivered,
+        anyIf(delivery_filtered, source = 'delivery') AS filtered_count,
+        anyIf(delivery_dropped, source = 'delivery') AS dropped,
+        anyIf(delivery_failed, source = 'delivery') AS failed_count,
+        anyIf(delivery_flush_complete, source = 'delivery') AS flush_flag,
+        countIf(source = 'delivery') > 0 AS delivery_reported,
+        anyIf(context_filter_active, source = 'context') AS filter_flag
+    FROM source_rows
+    GROUP BY invocation_id, invocation_started_at
+    HAVING max(is_listed) = 1
 ),
 counted AS (
     SELECT
-        ids.invocation_id AS invocation_id,
-        ids.invocation_started_at AS invocation_started_at,
-        coalesce(received.chunks, 0) AS chunks,
-        coalesce(received.bytes, 0) AS bytes,
-        coalesce(received.stdout_bytes, 0) AS stdout_bytes,
-        coalesce(received.stderr_bytes, 0) AS stderr_bytes,
-        coalesce(received.truncated_bytes, 0) AS truncated_bytes,
-        received.max_sequence AS max_sequence,
+        invocation_id,
+        invocation_started_at,
+        chunks,
+        bytes,
+        stdout_bytes,
+        stderr_bytes,
+        truncated_bytes,
+        max_sequence,
         greatest(
-            coalesce(toInt64(received.max_sequence) + 1 - toInt64(received.spans), 0),
-            coalesce(toInt64(delivery.accepted), 0) - coalesce(toInt64(received.chunks), 0)
+            coalesce(toInt64(max_sequence) + 1 - toInt64(spans), 0),
+            coalesce(toInt64(accepted), 0) - coalesce(toInt64(chunks), 0)
         ) AS missing_sequences,
-        greatest(coalesce(received.queue_drops, 0), coalesce(delivery.dropped, 0)) AS reported_drops,
-        coalesce(delivery.filtered, 0) AS filtered,
-        coalesce(delivery.failed, 0) AS failed,
-        coalesce(context.recall_filter_active, 0) AS filter_active,
-        coalesce(delivery.flush_complete, 0) AS flush_complete,
-        delivery.invocation_id IS NOT NULL AS delivery_reported,
-        coalesce(delivery.delivered, 0) AS delivered
-    FROM ids
-    LEFT JOIN received ON received.invocation_id = ids.invocation_id
-    LEFT JOIN delivery ON delivery.invocation_id = ids.invocation_id
-    LEFT JOIN {context} AS context ON context.invocation_id = ids.invocation_id
-    SETTINGS join_use_nulls = 1
+        greatest(coalesce(queue_drops, 0), coalesce(dropped, 0)) AS reported_drops,
+        coalesce(filtered_count, 0) AS filtered,
+        coalesce(failed_count, 0) AS failed,
+        coalesce(filter_flag, 0) AS filter_active,
+        coalesce(flush_flag, 0) AS flush_complete,
+        delivery_reported,
+        coalesce(delivered, 0) AS delivered
+    FROM folded
 )
 SELECT
     invocation_id, invocation_started_at, chunks, bytes, stdout_bytes, stderr_bytes,
@@ -335,6 +561,7 @@ fn passthrough(table: Table, options: &DdlOptions) -> String {
     format!("SELECT * FROM {}", deduplicated(table, options))
 }
 
+/// Every reference with what is stored for its digest. The stored side is nullable before the join, so a reference without stored SQL reads NULL whether or not the session sets `join_use_nulls`.
 fn artifacts(options: &DdlOptions) -> String {
     let refs = deduplicated(Table::ArtifactRefs, options);
     let stored = deduplicated(Table::Artifacts, options);
@@ -347,16 +574,24 @@ fn artifacts(options: &DdlOptions) -> String {
     refs.role AS role,
     refs.statement_id AS statement_id,
     refs.resource_id AS resource_id,
-    artifact.kind AS kind,
-    artifact.media_type AS media_type,
-    artifact.size AS size,
-    artifact.digest IS NOT NULL AS available,
-    if(artifact.digest IS NULL, 'expired_or_not_captured', NULL) AS unavailable_reason,
-    if(startsWith(artifact.media_type, 'text/'), toValidUTF8(artifact.content), NULL) AS text,
-    artifact.content AS content
+    artifact.stored_kind AS kind,
+    artifact.stored_media_type AS media_type,
+    artifact.stored_size AS size,
+    artifact.stored_digest IS NOT NULL AS available,
+    if(artifact.stored_digest IS NULL, 'expired_or_not_captured', NULL) AS unavailable_reason,
+    if(startsWith(artifact.stored_media_type, 'text/'), toValidUTF8(artifact.stored_content), NULL)
+        AS text,
+    artifact.stored_content AS content
 FROM {refs} AS refs
-LEFT JOIN {stored} AS artifact ON artifact.digest = refs.digest
-SETTINGS join_use_nulls = 1"
+LEFT JOIN (
+    SELECT digest,
+        CAST(digest AS Nullable(String)) AS stored_digest,
+        CAST(kind AS Nullable(String)) AS stored_kind,
+        CAST(media_type AS Nullable(String)) AS stored_media_type,
+        CAST(size AS Nullable(Int64)) AS stored_size,
+        CAST(content AS Nullable(String)) AS stored_content
+    FROM {stored}
+) AS artifact ON artifact.digest = refs.digest"
     )
 }
 
diff --git a/crates/sqlbuild-recall/src/schema/constants.rs b/crates/sqlbuild-recall/src/schema/constants.rs
index b48373d847..7a4976da9f 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 = 2;
+pub const RECALL_SCHEMA_VERSION: i64 = 3;
 /// The oldest SQLite library the store's views run on (ordered `group_concat`).
 pub const MINIMUM_SQLITE_VERSION: (u32, u32) = (3, 44);
 /// The `recall_meta` component of schema rows.
diff --git a/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs b/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs
index a8365c6e04..efade64b1e 100644
--- a/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs
+++ b/crates/sqlbuild-recall/src/schema/tests/test_ddl.rs
@@ -19,6 +19,19 @@ const NEW_COLUMN: ColumnSpec = ColumnSpec {
     column_type: ColumnType::Integer,
     nullable: false,
 };
+/// Views whose rows or counts would repeat with a key's stored copies.
+const FINAL_VIEWS: [&str; 8] = [
+    "audit_results",
+    "output_invocations",
+    "output_tail",
+    "output_transcripts",
+    "events",
+    "diagnostics",
+    "artifacts",
+    "conflicts",
+];
+/// Views that fold every copy of a key to the same result.
+const FOLDING_VIEWS: [&str; 2] = ["invocations", "statements"];
 
 #[test]
 fn given_sqlite_script_when_run_then_every_table_view_and_retention_statement_prepares() {
@@ -112,6 +125,13 @@ fn given_clickhouse_statement_sets_when_rendered_then_every_statement_has_the_sh
             .map(|(_, statement)| statement)
             .collect()
     };
+    let named_views = |names: &[&str]| -> Vec<String> {
+        view_statements(Dialect::ClickHouse, &options(None))
+            .into_iter()
+            .filter(|(name, _)| names.contains(name))
+            .map(|(_, statement)| statement)
+            .collect()
+    };
     let test_cases = [
         StatementSetTestCase {
             description: "every clustered table is created on the cluster",
@@ -134,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', '2'",
+                "SELECT 'schema', 'schema_version', '3'",
             ],
             expected_count: 1,
         },
@@ -150,9 +170,33 @@ fn given_clickhouse_statement_sets_when_rendered_then_every_statement_has_the_sh
             expected_count: 1,
         },
         StatementSetTestCase {
-            description: "views deduplicate rows and filter by retention",
+            description: "views filter by retention",
             statements: views(None),
-            required_fragments: &["now64(6) - INTERVAL 14 DAY", " FINAL WHERE "],
+            required_fragments: &["now64(6) - INTERVAL 14 DAY"],
+            expected_count: VIEW_NAMES.len(),
+        },
+        StatementSetTestCase {
+            description: "views that list or count rows read each key once",
+            statements: named_views(&FINAL_VIEWS),
+            required_fragments: &[" FINAL WHERE "],
+            expected_count: FINAL_VIEWS.len(),
+        },
+        StatementSetTestCase {
+            description: "views that only fold a key's copies read the tables without FINAL",
+            statements: named_views(&FOLDING_VIEWS)
+                .into_iter()
+                .filter(|statement| !statement.contains(" FINAL "))
+                .collect(),
+            required_fragments: &["GROUP BY invocation_id, invocation_started_at"],
+            expected_count: FOLDING_VIEWS.len(),
+        },
+        StatementSetTestCase {
+            description: "no view sets query settings or depends on join_use_nulls",
+            statements: views(None)
+                .into_iter()
+                .filter(|statement| !statement.contains("SETTINGS"))
+                .collect(),
+            required_fragments: &["CREATE OR REPLACE VIEW"],
             expected_count: VIEW_NAMES.len(),
         },
     ];
diff --git a/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs b/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs
index e5ea0d1823..976362ed7f 100644
--- a/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs
+++ b/crates/sqlbuild-recall/src/writer/_helpers/delivery.rs
@@ -24,9 +24,9 @@ use crate::store::types::{RecallStore, RecallStoreOpener};
 use crate::writer::_helpers::materialize::{Built, Materializer};
 use crate::writer::_helpers::queue::{Queued, WriterState};
 use crate::writer::constants::{
-    BATCH_ROWS, BATCH_WINDOW_MS, CLOCK_CHECK_MS, FINAL_ROWS_BUDGET_MS, HEARTBEAT_STALE_INTERVALS,
-    MAX_RETRY_BACKOFF_MS, MINIMUM_MAINTENANCE_MS, OUTPUT_COALESCE_MS, RUNNING_RETRY_LIMIT_MS,
-    SETUP_RETRY_MS,
+    BATCH_ROWS, BATCH_WINDOW_MS, CLOCK_CHECK_MS, EARLY_BATCH_BYTES, FINAL_ROWS_BUDGET_MS,
+    HEARTBEAT_STALE_INTERVALS, MAX_RETRY_BACKOFF_MS, MINIMUM_MAINTENANCE_MS, OUTPUT_COALESCE_MS,
+    RUNNING_RETRY_LIMIT_MS, SETUP_RETRY_MS,
 };
 use crate::writer::models::{DeliveryStream, SessionConfig, SessionStore, SetupOutcome};
 use crate::writer::types::OutputReplacement;
@@ -238,7 +238,7 @@ fn batching(config: &SessionConfig) -> Batching {
     }
 }
 
-/// Wait until a batch is due: its row limit, the batch window, closing, or a timed duty.
+/// Wait until a batch is due: its row or byte limit, the batch window, closing, or a timed duty.
 fn next_batch(
     shared: &Shared,
     next_heartbeat: SystemTime,
@@ -260,7 +260,10 @@ fn next_batch(
         }
         let window_over =
             first_seen.is_some_and(|seen| now.duration_since(seen) >= batching.window);
-        if state.ready() >= batching.rows || (state.ready() > 0 && (window_over || state.closing)) {
+        if state.ready() >= batching.rows
+            || state.ready_bytes() >= EARLY_BATCH_BYTES
+            || (state.ready() > 0 && (window_over || state.closing))
+        {
             return (state.take_batch(batching.rows), state.closing);
         }
         if state.closing && state.queued() == 0 && !state.has_pending_output() {
diff --git a/crates/sqlbuild-recall/src/writer/_helpers/queue.rs b/crates/sqlbuild-recall/src/writer/_helpers/queue.rs
index 50016bdc7b..d2d55a7f7d 100644
--- a/crates/sqlbuild-recall/src/writer/_helpers/queue.rs
+++ b/crates/sqlbuild-recall/src/writer/_helpers/queue.rs
@@ -274,6 +274,19 @@ impl WriterState {
         }
     }
 
+    /// Bytes a batch may take now: held output stays queued.
+    pub(crate) fn ready_bytes(&self) -> usize {
+        if self.holding_output() {
+            let held: usize = self.buckets[Priority::Output as usize]
+                .iter()
+                .map(|record| record.bytes)
+                .sum();
+            self.bytes - held
+        } else {
+            self.bytes
+        }
+    }
+
     /// Up to `limit` rows, highest priority first; held output stays queued.
     pub(crate) fn take_batch(&mut self, limit: usize) -> Vec<Queued> {
         let mut batch: Vec<Queued> = Vec::new();
diff --git a/crates/sqlbuild-recall/src/writer/constants.rs b/crates/sqlbuild-recall/src/writer/constants.rs
index 653bab6c91..4d013df6c3 100644
--- a/crates/sqlbuild-recall/src/writer/constants.rs
+++ b/crates/sqlbuild-recall/src/writer/constants.rs
@@ -2,6 +2,8 @@
 
 pub const QUEUE_RECORD_CAPACITY: usize = 10_000;
 pub const QUEUE_BYTE_CAPACITY: usize = 32 * 1024 * 1024;
+/// A batch is sent before its window once this many bytes are ready, so a long window never fills the queue.
+pub const EARLY_BATCH_BYTES: usize = QUEUE_BYTE_CAPACITY / 4;
 pub const BATCH_ROWS: usize = 500;
 pub const BATCH_WINDOW_MS: u64 = 250;
 /// Contiguous writes to one stream coalesce into one chunk for at most this long.
diff --git a/crates/sqlbuild-recall/src/writer/tests/helpers.rs b/crates/sqlbuild-recall/src/writer/tests/helpers.rs
index 07ebb6d109..c26f04538d 100644
--- a/crates/sqlbuild-recall/src/writer/tests/helpers.rs
+++ b/crates/sqlbuild-recall/src/writer/tests/helpers.rs
@@ -17,7 +17,8 @@ use crate::writer::models::{
     DeliveryStream, DeliverySummary, Priority, SessionConfig, SessionStore, SetupOutcome,
 };
 use crate::writer::tests::test_types::{
-    CopyingReplacement, PartialStore, PrefixReplacement, RecordedSession, RecordingReplacement,
+    CopyingReplacement, OutputBytesStore, PartialStore, PrefixReplacement, RecordedSession,
+    RecordingReplacement,
 };
 
 /// The envelope of invocation `invocation_id`.
@@ -378,6 +379,29 @@ pub(crate) fn replaced_session(
     (transcript.data, facts.rows)
 }
 
+/// Output bytes a store with a ten-minute batch window received `wait` after `written_bytes` were printed, and the output records dropped by close.
+pub(crate) fn sent_before_the_window(written_bytes: usize, wait: Duration) -> (usize, u64) {
+    let store = OutputBytesStore::default();
+    let session = start_recall_session(SessionConfig {
+        store: SessionStore::Hosted(Arc::new(store.clone())),
+        ..config(PathBuf::from("unused"), "burst")
+    });
+    let line: Vec<u8> = vec![b'x'; 4096];
+    for _ in 0..written_bytes / line.len() {
+        session.append_output(OutputStream::Stdout, &line);
+    }
+    std::thread::sleep(wait);
+    let sent: usize = *store
+        .received
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let summary = session.close();
+    (
+        sent,
+        stream_counts(&summary, DeliveryStream::Output).dropped,
+    )
+}
+
 /// `(accepted, delivered, failed)` of events and output after one batch whose insert into `failing` fails.
 pub(crate) fn partially_inserted(failing: Table) -> [(u64, u64, u64); 2] {
     let session = start_recall_session(SessionConfig {
diff --git a/crates/sqlbuild-recall/src/writer/tests/test_session.rs b/crates/sqlbuild-recall/src/writer/tests/test_session.rs
index 2438b84bc4..ab99a7a55e 100644
--- a/crates/sqlbuild-recall/src/writer/tests/test_session.rs
+++ b/crates/sqlbuild-recall/src/writer/tests/test_session.rs
@@ -10,15 +10,15 @@ use crate::writer::tests::helpers::{
     artifact_counts, close_under_held_lock, event_counts, insert_heartbeat,
     invocation_during_discovery, output_stored_while_running, partially_inserted,
     pending_event_counts, record_run, recorded_session, replaced_session, replacement_order,
-    session_on_corrupt_store, written_to_core_release_store,
+    sent_before_the_window, session_on_corrupt_store, written_to_core_release_store,
 };
 #[cfg(unix)]
 use crate::writer::tests::test_types::UnwritableTestCase;
 use crate::writer::tests::test_types::{
     ArtifactSizeTestCase, CoreReleaseStoreTestCase, CorruptStoreTestCase, DiscoveryTestCase,
-    EventBuildTestCase, HeldLockTestCase, PartialInsertTestCase, PendingEventTestCase,
-    RecordedSession, ReplacementOrderTestCase, ReplacementTestCase, SessionTestCase,
-    StoredWhileRunningTestCase,
+    EarlyBatchTestCase, EventBuildTestCase, HeldLockTestCase, PartialInsertTestCase,
+    PendingEventTestCase, RecordedSession, ReplacementOrderTestCase, ReplacementTestCase,
+    SessionTestCase, StoredWhileRunningTestCase,
 };
 
 const CLOSE_SLACK_SECONDS: f64 = 0.3;
@@ -378,3 +378,24 @@ fn given_events_of_mixed_priority_in_one_batch_when_replaced_then_the_sink_sees_
         );
     }
 }
+
+#[test]
+fn given_a_byte_heavy_burst_when_the_window_is_long_then_it_is_sent_early_and_nothing_is_dropped() {
+    let test_cases = [EarlyBatchTestCase {
+        description: "12 MiB of output is sent long before a ten-minute window",
+        written_bytes: 12 * 1024 * 1024,
+        wait: Duration::from_secs(2),
+        expected_minimum_sent_bytes: 8 * 1024 * 1024,
+        expected_dropped: 0,
+    }];
+
+    for test_case in test_cases {
+        let (sent, dropped) = sent_before_the_window(test_case.written_bytes, test_case.wait);
+        assert_eq!(
+            (sent >= test_case.expected_minimum_sent_bytes, dropped),
+            (true, test_case.expected_dropped),
+            "{}: {sent} bytes sent",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlbuild-recall/src/writer/tests/test_types.rs b/crates/sqlbuild-recall/src/writer/tests/test_types.rs
index 079a4b2457..f240a46e1d 100644
--- a/crates/sqlbuild-recall/src/writer/tests/test_types.rs
+++ b/crates/sqlbuild-recall/src/writer/tests/test_types.rs
@@ -1,3 +1,4 @@
+use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use crate::records::models::{OutputChunk, OutputStream, RecallRow, RecallValue, RowEnvelope};
@@ -166,6 +167,56 @@ impl RecallStoreOpener for PartialStore {
     }
 }
 
+pub(crate) struct EarlyBatchTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) written_bytes: usize,
+    pub(crate) wait: Duration,
+    pub(crate) expected_minimum_sent_bytes: usize,
+    pub(crate) expected_dropped: u64,
+}
+
+/// A hosted store with a batch window far longer than any test, counting the output bytes it receives.
+#[derive(Clone, Debug, Default)]
+pub(crate) struct OutputBytesStore {
+    pub(crate) received: Arc<Mutex<usize>>,
+}
+
+impl RecallStore for OutputBytesStore {
+    fn insert(&self, rows: &[RecallRow], _give_up: Instant) -> Result<InsertOutcome, StoreError> {
+        let bytes: usize = rows
+            .iter()
+            .filter(|row| row.table == Table::Output)
+            .map(|row| match row.value("data") {
+                Some(RecallValue::Blob(data)) => data.len(),
+                _ => 0,
+            })
+            .sum();
+        *self
+            .received
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner) += bytes;
+        Ok(InsertOutcome::default())
+    }
+}
+
+impl RecallStoreOpener for OutputBytesStore {
+    fn open(
+        &self,
+        _envelope: &RowEnvelope,
+        _give_up: Instant,
+    ) -> Result<Box<dyn RecallStore>, StoreError> {
+        Ok(Box::new(self.clone()))
+    }
+
+    fn batch_rows(&self) -> usize {
+        1_000_000
+    }
+
+    fn batch_window(&self) -> Duration {
+        Duration::from_secs(600)
+    }
+}
+
 pub(crate) struct DiscoveryTestCase {
     pub(crate) description: &'static str,
     pub(crate) wait: Duration,
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters/clickhouse.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters/clickhouse.md
index 80a71cf9f5..7e497fa203 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters/clickhouse.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters/clickhouse.md
@@ -175,6 +175,8 @@ Every query runs with these settings so SQL behaves the same as on other adapter
 
 The connection's `settings` may repeat these values but not change them: a different value fails to connect with `A710`, because SQLBuild's generated SQL depends on them.
 
+A [Recall](../observability/recall.md) `read_connection` whose user is `readonly = 1`, and so may not change settings, sends none of them: its `sqb recall query` SQL runs with the server's own settings. Every other reader gets them.
+
 ClickHouse returns a type's default value (`0`, the epoch, an empty string) from `lag` and `lead` past the edge of a window frame unless the argument is `Nullable`. SQLBuild's generated snapshot SQL wraps those arguments in `toNullable` so they return `NULL` as on other adapters. Write the same in your own models when you rely on `NULL` there.
 
 ## SQL differences to watch
@@ -197,5 +199,6 @@ A ClickHouse connection can also hold [Recall](../observability/recall.md)'s com
 history for a whole team, whichever adapter the project builds with. Set
 `[recall] write_connection` and `read_connection` to it, one connection or separate writer and
 reader users. Recall creates its tables on first use, keeps every output byte by default, and needs
-only `SELECT` and `INSERT` once the tables exist; readers need only `SELECT`. See
+only `SELECT` and `INSERT` once the tables exist; readers need only `SELECT` and may be
+`readonly = 1` users. See
 [Sharing history in ClickHouse](../observability/recall.md#sharing-history-in-clickhouse).
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 3f9fdc3d71..40bad05a42 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
@@ -277,6 +277,16 @@ GRANT SELECT, INSERT ON observability.* TO recall_writer;
 GRANT SELECT ON observability.* TO recall_reader;
 ```
 
+The reader may be a standard read-only user (`SETTINGS readonly = 1`, or a settings profile with
+it), which may not change any setting. Such a user's queries run with the server's own settings;
+every other reader gets the
+[settings SQLBuild pins](../adapters/clickhouse.md#settings-sqlbuild-pins). The views
+return the same rows either way. `sqb recall query` runs with
+`readonly = 2`, which refuses writes but still allows query settings, unless the user is already
+read-only; a user that may not set the row limit gets the same truncated result, cut off while
+reading. A setting that Recall needs and the user may not change fails with an error naming it; set
+it in the user's profile, or give the user `readonly = 2`, which keeps it read-only.
+
 Then, as the administrator, create the store with the DDL that `sqb recall schema --ddl` prints
 (any connection that can read the database works, so the reader's is enough):
 
@@ -303,12 +313,14 @@ writing to an upgraded store. `--ddl` prints SQL, so it cannot be combined with
   recorded. When it has expired anyway, a writer compares the stored tables' and views' columns
   with its own: it upgrades an older store, records its version on a current one, and never
   replaces the views of a store a newer SQLBuild version created.
-- **Writes.** History is sent in batches of up to 5,000 rows about once a second, one insert per
-  table, with the same queue, shutdown budget and exit-code guarantees as the local store. Many
-  concurrent commands can write one store.
+- **Writes.** History is sent in batches of up to 5,000 rows about every five seconds, one insert
+  per table, so a long command adds few parts; everything still queued is sent when it ends, with
+  the same queue, shutdown budget and exit-code guarantees as the local store. Many concurrent
+  commands can write one store.
 - **Duplicates.** Every row has a deterministic key. A retried batch is deduplicated by ClickHouse,
-  and the views read each key once (`FINAL` on `ReplacingMergeTree` tables), so a repeated delivery
-  is counted once. Before writing, Recall looks up the batch's keys: a key already stored with
+  and the views read each key once, so a repeated delivery is counted once: views that list or count
+  rows read the `ReplacingMergeTree` tables with `FINAL`, and `invocations` and `statements`, which
+  only take the first, last, smallest or largest value of a key's copies, read them without it. Before writing, Recall looks up the batch's keys: a key already stored with
   different content keeps the stored row, and the rejected copy is listed in `conflicts`.
 - **Retention.** Every table carries a TTL of `retention` (SQL text a week longer, and refreshed
   while it is still referenced), and the views hide rows past their TTL at once. ClickHouse drops
diff --git a/src/sqlbuild/adapter/contract/_helpers/raw_statement_audit.py b/src/sqlbuild/adapter/contract/_helpers/raw_statement_audit.py
index bb79d702e1..8a172a4149 100644
--- a/src/sqlbuild/adapter/contract/_helpers/raw_statement_audit.py
+++ b/src/sqlbuild/adapter/contract/_helpers/raw_statement_audit.py
@@ -7,7 +7,9 @@
 
 _AUDITED_ROOTS: tuple[str, ...] = ("adapter", "adapters")
 _RAW_RECEIVERS: frozenset[str] = frozenset({"client", "raw_connection", "raw_cursor"})
-_STATEMENT_METHODS: frozenset[str] = frozenset({"execute", "executemany", "query"})
+_STATEMENT_METHODS: frozenset[str] = frozenset(
+    {"execute", "executemany", "query", "query_row_block_stream"}
+)
 _APPROVED_BOUNDARIES: dict[str, frozenset[tuple[str, str]]] = {
     "adapter/contract/classes/observed_connection.py": frozenset(
         {
@@ -24,7 +26,9 @@
     "adapters/bigquery/classes/bigquery_connection.py": frozenset(
         {("_BigQueryConnection.query_job", "query")}
     ),
-    "adapters/clickhouse/_helpers/recall_store.py": frozenset({("run_recall_query", "query")}),
+    "adapters/clickhouse/_helpers/recall_store.py": frozenset(
+        {("run_recall_query", "query_row_block_stream")}
+    ),
     "adapters/clickhouse/classes/clickhouse_connection.py": frozenset(
         {("_ClickHouseConnection._run", "query")}
     ),
diff --git a/src/sqlbuild/adapters/clickhouse/_helpers/connect_clickhouse.py b/src/sqlbuild/adapters/clickhouse/_helpers/connect_clickhouse.py
index cba2ff7730..ac59da7d0a 100644
--- a/src/sqlbuild/adapters/clickhouse/_helpers/connect_clickhouse.py
+++ b/src/sqlbuild/adapters/clickhouse/_helpers/connect_clickhouse.py
@@ -13,15 +13,19 @@
     DEFAULT_HTTPS_PORT,
     DSN_OPTION,
     MINIMUM_SERVER_VERSION,
+    READONLY_SETTING,
     SESSION_SETTINGS,
     SETTING_CONFLICT_CODE,
+    SETTINGS_LOCKED_READONLY,
     TIMEOUT_OPTIONS,
     VERIFY_OPTION,
 )
 
 
-def connect_clickhouse(config: dict[str, Any]) -> _ClickHouseConnection:
-    """A connection for `config`: validated options, required session settings, supported server."""
+def connect_clickhouse(
+    *, config: dict[str, Any], settings_if_allowed: bool = False
+) -> _ClickHouseConnection:
+    """A connection for `config` with SQLBuild's session settings, or only where allowed."""
 
     try:
         import clickhouse_connect
@@ -39,7 +43,9 @@ def connect_clickhouse(config: dict[str, Any]) -> _ClickHouseConnection:
             help="Supported options: " + ", ".join(sorted(CONNECTION_OPTION_KEYS)),
         )
     secure: bool = bool(config.get("secure", False))
-    settings: dict[str, object] = _session_settings(dict(config.get("settings") or {}))
+    user_settings: dict[str, object] = dict(config.get("settings") or {})
+    pinned: dict[str, object] = _session_settings(user_settings)
+    settings: dict[str, object] = user_settings if settings_if_allowed else pinned
     client_options: dict[str, Any] = {
         "host": str(config.get("host", "localhost")),
         "port": int(config.get("port", DEFAULT_HTTPS_PORT if secure else DEFAULT_HTTP_PORT)),
@@ -71,9 +77,19 @@ def create_client() -> Any:
             code="A703",
             help=f"Upgrade the server to ClickHouse {minimum} or newer.",
         )
+    if settings_if_allowed and _may_change_settings(client):
+        for name, value in pinned.items():
+            client.set_client_setting(name, value)
     return _ClickHouseConnection(client=client, interrupt_client_factory=create_client)
 
 
+def _may_change_settings(client: Any) -> bool:
+    """Whether the session may change settings: not `readonly = 1`."""
+
+    readonly: Any = client.server_settings.get(READONLY_SETTING)
+    return readonly is None or str(readonly.value) != SETTINGS_LOCKED_READONLY
+
+
 def _version_tuple(version: str) -> tuple[int, ...]:
     parts: list[int] = []
     for part in version.split(".")[:2]:
diff --git a/src/sqlbuild/adapters/clickhouse/_helpers/recall_store.py b/src/sqlbuild/adapters/clickhouse/_helpers/recall_store.py
index 75e9bf19f1..ae367239ac 100644
--- a/src/sqlbuild/adapters/clickhouse/_helpers/recall_store.py
+++ b/src/sqlbuild/adapters/clickhouse/_helpers/recall_store.py
@@ -22,6 +22,9 @@
 _ROW_BINARY: str = "RowBinary"
 _INT64_MIN: int = -(2**63)
 _INT64_MAX: int = 2**63 - 1
+_READONLY_SETTING: str = "readonly"
+_WRITABLE_READONLY_VALUE: str = "0"
+_LIMIT_SETTING: str = "limit"
 
 
 def _failure(error: Exception) -> tuple[bool, str]:
@@ -62,39 +65,84 @@ def _number(*, summary: dict[str, Any], key: str) -> int | None:
 
 
 def connect_recall_store(
-    *, config: dict[str, object], quiet_thread_attribute: str | None
+    *, config: dict[str, object], quiet_thread_attribute: str | None, read_only: bool
 ) -> tuple[RecallFailure, int]:
-    """Connect for Recall and return the handle; only the delivery thread asks to stay quiet."""
+    """Connect for Recall; a `read_only` connection skips session settings under `readonly = 1`."""
 
     if quiet_thread_attribute is not None:
         setattr(threading.current_thread(), quiet_thread_attribute, True)
     options: dict[str, object] = {_CONNECT_TIMEOUT_KEY: _DEFAULT_CONNECT_TIMEOUT_SECONDS, **config}
     try:
-        connection: _ClickHouseConnection = connect_clickhouse(options)
+        connection: _ClickHouseConnection = connect_clickhouse(
+            config=options, settings_if_allowed=read_only
+        )
     except Exception as error:
         return _failure(error), 0
     return None, SharedRecallConnections.connections.add(connection=connection)
 
 
+def _server_value(*, client: Any, name: str) -> tuple[str | None, bool]:
+    """The session's value of setting `name`, and whether this user may change it."""
+
+    setting: Any = client.server_settings.get(name)
+    if setting is None:
+        return None, True
+    return str(setting.value), not setting.readonly
+
+
+def _query_settings(
+    *, client: Any, settings: dict[str, int], row_cap: int | None
+) -> tuple[dict[str, int], str | None]:
+    """The settings the session does not already hold, or why the user may not set one."""
+
+    sent: dict[str, int] = {}
+    for name, value in settings.items():
+        current, writable = _server_value(client=client, name=name)
+        if current == str(value) or (
+            name == _READONLY_SETTING and current not in (None, _WRITABLE_READONLY_VALUE)
+        ):
+            continue
+        if not writable:
+            return {}, (
+                f"this ClickHouse user may not change the setting `{name}`, which Recall "
+                f"queries set to {value}; set it in the user's settings profile, or give the "
+                "user `readonly = 2`, which keeps it read-only and lets it set query settings"
+            )
+        sent[name] = value
+    if row_cap is not None and _server_value(client=client, name=_LIMIT_SETTING)[1]:
+        sent[_LIMIT_SETTING] = row_cap
+    return sent, None
+
+
 def run_recall_query(
-    *, handle: int, sql: str, settings: dict[str, int]
+    *, handle: int, sql: str, settings: dict[str, int], row_cap: int | None
 ) -> tuple[RecallFailure, tuple[list[str], list[list[RecallCell]], RecallStatistics] | None]:
-    """Columns, rows of plain values and server statistics; strings not in UTF-8 stay bytes."""
+    """Columns, at most `row_cap` rows of plain values, and server statistics."""
 
     try:
-        result: Any = SharedRecallConnections.connections.get(handle=handle).client.query(
-            sql, settings=settings, query_formats=_BYTES_FORMATS
-        )
+        client: Any = SharedRecallConnections.connections.get(handle=handle).client
+        sent, refused = _query_settings(client=client, settings=settings, row_cap=row_cap)
+        if refused is not None:
+            return (False, refused), None
+        rows: list[list[RecallCell]] = []
+        with client.query_row_block_stream(
+            sql, settings=sent, query_formats=_BYTES_FORMATS
+        ) as stream:
+            result: Any = stream.source
+            for block in stream:
+                rows.extend(_row(row) for row in block)
+                if row_cap is not None and len(rows) >= row_cap:
+                    break
+            columns: list[str] = list(result.column_names)
+            summary: dict[str, Any] = dict(result.summary or {})
     except Exception as error:
         return _failure(error), None
-    summary: dict[str, Any] = dict(result.summary or {})
     statistics: RecallStatistics = (
         _number(summary=summary, key="read_rows"),
         _number(summary=summary, key="read_bytes"),
         summary.get("query_id"),
     )
-    rows: list[list[RecallCell]] = [_row(row) for row in result.result_rows]
-    return None, (list(result.column_names), rows, statistics)
+    return None, (columns, rows[:row_cap] if row_cap is not None else rows, statistics)
 
 
 def run_recall_command(*, handle: int, sql: str) -> tuple[RecallFailure, None]:
diff --git a/src/sqlbuild/adapters/clickhouse/classes/clickhouse_adapter.py b/src/sqlbuild/adapters/clickhouse/classes/clickhouse_adapter.py
index 53cb57e93f..ba9b34eaf0 100644
--- a/src/sqlbuild/adapters/clickhouse/classes/clickhouse_adapter.py
+++ b/src/sqlbuild/adapters/clickhouse/classes/clickhouse_adapter.py
@@ -175,7 +175,7 @@ class ClickHouseAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter):
     _relation_locks: ClassVar[RelationWriteLocks] = RelationWriteLocks()
 
     def connect(self, config: dict[str, Any]) -> _ClickHouseConnection:
-        return connect_clickhouse(config)
+        return connect_clickhouse(config=config)
 
     def _execute(self, *, connection: _ClickHouseConnection, sql: str) -> Any:
         log_sql(logger=logging.getLogger("sqlbuild.adapter.clickhouse"), sql=sql)
diff --git a/src/sqlbuild/adapters/clickhouse/constants.py b/src/sqlbuild/adapters/clickhouse/constants.py
index 7b9047e237..bc91e48c48 100644
--- a/src/sqlbuild/adapters/clickhouse/constants.py
+++ b/src/sqlbuild/adapters/clickhouse/constants.py
@@ -22,6 +22,8 @@
     "insert_null_as_default": 0,
 }
 SETTING_CONFLICT_CODE: str = "A710"
+READONLY_SETTING: str = "readonly"
+SETTINGS_LOCKED_READONLY: str = "1"
 CONNECTION_OPTION_KEYS: frozenset[str] = frozenset(
     {
         "host",
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 fe3dfe2402..5cb2a1152f 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
@@ -253,3 +253,49 @@ class ClickHouseRecallUnrecordedBehindTestCase:
     stale_view: str
     expected_notice: str
     expected_version_rows: tuple[tuple[object, ...], ...]
+
+
+@dataclass(frozen=True)
+class ClickHouseRecallReadOnlyUserTestCase:
+    description: str
+    view_names: tuple[str, ...]
+    expected_status: list[list[object]]
+
+
+@dataclass(frozen=True)
+class ClickHouseRecallViewEquivalenceTestCase:
+    description: str
+    replaced_views: tuple[str, ...]
+    all_views: tuple[str, ...]
+    expected_statuses: list[list[object]]
+
+
+@dataclass(frozen=True)
+class ClickHouseRecallTableReadsTestCase:
+    description: str
+    view: str
+    expected_tables: tuple[str, ...]
+    expected_primary_key_reads: int
+
+
+@dataclass(frozen=True)
+class ClickHouseRecallPreviousVersionTestCase:
+    description: str
+    replaced_views: tuple[str, ...]
+    expected_notice: str
+    expected_ddl_fragments: tuple[str, ...]
+    expected_version: tuple[tuple[object, ...], ...]
+
+
+@dataclass(frozen=True)
+class ClickHouseRecallRefusedSettingTestCase:
+    description: str
+    user_settings: str
+    expected_fragments: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class ClickHouseRecallReaderSettingsTestCase:
+    description: str
+    user_settings: str
+    expected_rows: list[list[object]]
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 73f089dd06..5fbce5caae 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/helpers.py
@@ -4,6 +4,7 @@
 
 import functools
 import re
+import subprocess
 import uuid
 from collections.abc import Iterator
 from contextlib import contextmanager
@@ -13,7 +14,14 @@
 
 from sqlbuild.adapters.clickhouse.classes.clickhouse_adapter import ClickHouseAdapter
 from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import prepare_waffle_shop
-from tests.integration.src.sqlbuild.cli.recall.helpers import write_project
+from tests.integration.src.sqlbuild.cli.recall.helpers import (
+    killed_build,
+    killed_during_discovery,
+    run,
+    start,
+    wait_for,
+    write_project,
+)
 
 CLICKHOUSE_IMAGE: str = "clickhouse/clickhouse-server:26.1"
 
@@ -466,14 +474,13 @@ def apply_printed_retention_fix(*, config: dict[str, object], stderr: str) -> in
         "    `subcommand` Nullable(String),",
         "    `pid` Nullable(Int64),",
     ),
-    (",\n        min(command) AS command, min(subcommand) AS subcommand", ""),
     (
-        "coalesce(context.command, started.command, beats.command)",
-        "coalesce(context.command, started.command)",
+        "CAST(command AS Nullable(String)) AS beat_command",
+        "CAST(NULL AS Nullable(String)) AS beat_command",
     ),
     (
-        "coalesce(context.subcommand, beats.subcommand) AS subcommand",
-        "context.subcommand AS subcommand",
+        "CAST(subcommand AS Nullable(String)) AS beat_subcommand",
+        "CAST(NULL AS Nullable(String)) AS beat_subcommand",
     ),
 )
 
@@ -593,3 +600,430 @@ def write_recall_project(
         ),
         targets=targets,
     )
+
+
+_PREVIOUS_ORDER: str = "tuple(invocation_sequence, occurred_at)"
+_PREVIOUS_STATEMENT_TERMINAL: str = "event_type IN ('statement_completed', 'statement_failed')"
+
+
+def _previous_read(*, database: str, table: str, expiry: str, days: int) -> str:
+    return (
+        f"(SELECT * FROM `{database}`.`{table}` FINAL "
+        f"WHERE {expiry} >= now64(6) - INTERVAL {days} DAY)"
+    )
+
+
+def previous_view_bodies(*, database: str, retention_days: int = 30) -> dict[str, str]:
+    """The bodies of the views the previous schema version created, which this one replaces."""
+
+    def read(table: str, expiry: str = "invocation_started_at", extra_days: int = 0) -> str:
+        return _previous_read(
+            database=database, table=table, expiry=expiry, days=retention_days + extra_days
+        )
+
+    context: str = read("recall_invocations")
+    events: str = read("recall_events")
+    beats: str = read("recall_heartbeats")
+    output: str = read("recall_output")
+    delivery: str = read("recall_delivery")
+    refs: str = read("recall_artifact_refs")
+    stored: str = read("recall_artifacts", expiry="seen_at", extra_days=7)
+    order: str = _PREVIOUS_ORDER
+    terminal: str = _PREVIOUS_STATEMENT_TERMINAL
+    return {
+        "invocations": f"""WITH
+ids AS (
+    SELECT invocation_id, invocation_started_at
+    FROM (
+        SELECT invocation_id, invocation_started_at FROM {context}
+        UNION ALL SELECT invocation_id, invocation_started_at FROM {beats}
+        UNION ALL SELECT invocation_id, invocation_started_at FROM {events}
+            WHERE event_type IN ('invocation_started', 'invocation_completed', 'invocation_failed')
+    )
+    GROUP BY invocation_id, invocation_started_at
+),
+started AS (
+    SELECT invocation_id, invocation_started_at, min(occurred_at) AS started_at,
+        any(command) AS command
+    FROM {events} WHERE event_type = 'invocation_started'
+    GROUP BY invocation_id, invocation_started_at
+),
+terminals AS (
+    SELECT invocation_id, invocation_started_at,
+        argMax(event_type, {order}) AS terminal_type,
+        argMax(occurred_at, {order}) AS terminal_at,
+        argMax(exit_code, {order}) AS terminal_exit_code,
+        argMax(duration_ms, {order}) AS terminal_duration_ms,
+        argMax(error_type, {order}) AS terminal_error_type,
+        argMax(error_code, {order}) AS terminal_error_code,
+        argMax(JSONExtract(payload, 'error_message', 'Nullable(String)'),
+            {order}) AS terminal_error_message
+    FROM {events}
+    WHERE event_type IN ('invocation_completed', 'invocation_failed')
+    GROUP BY invocation_id, invocation_started_at
+),
+beats AS (
+    SELECT invocation_id, invocation_started_at, max(beat_at) AS last_heartbeat_at,
+        max(stale_at) AS stale_at, min(command) AS command, min(subcommand) AS subcommand
+    FROM {beats} GROUP BY invocation_id, invocation_started_at
+)
+SELECT
+    ids.invocation_id AS invocation_id,
+    ids.invocation_started_at AS invocation_started_at,
+    coalesce(started.started_at, ids.invocation_started_at) AS started_at,
+    multiIf(
+        terminals.terminal_type = 'invocation_completed', 'completed',
+        terminals.terminal_type = 'invocation_failed', 'failed',
+        beats.stale_at IS NULL, 'unknown',
+        beats.stale_at >= now64(6), 'running',
+        'abandoned'
+    ) AS status,
+    if(terminals.terminal_type IS NOT NULL, terminals.terminal_at,
+        if(beats.stale_at < now64(6), beats.last_heartbeat_at, NULL)) AS ended_at,
+    if(terminals.terminal_type IS NOT NULL, terminals.terminal_duration_ms,
+        if(beats.stale_at < now64(6),
+            toFloat64(dateDiff('millisecond', started_at, beats.last_heartbeat_at)), NULL))
+        AS duration_ms,
+    terminals.terminal_exit_code AS exit_code,
+    terminals.terminal_error_type AS error_type,
+    terminals.terminal_error_code AS error_code,
+    terminals.terminal_error_message AS error_message,
+    coalesce(context.command, started.command, beats.command) AS command,
+    coalesce(context.subcommand, beats.subcommand) AS subcommand,
+    context.target AS target,
+    context.adapter AS adapter,
+    context.project_name AS project_name,
+    context.selection AS selection,
+    context.exclusion AS exclusion,
+    context.sqlbuild_version AS sqlbuild_version,
+    context.git_revision AS git_revision,
+    context.git_dirty AS git_dirty,
+    context.host_token AS host_token,
+    context.pid AS pid,
+    context.external_context AS external_context,
+    context.recall_filter_active AS recall_filter_active,
+    beats.last_heartbeat_at AS last_heartbeat_at
+FROM ids
+LEFT JOIN {context} AS context ON context.invocation_id = ids.invocation_id
+    AND context.invocation_started_at = ids.invocation_started_at
+LEFT JOIN started ON started.invocation_id = ids.invocation_id
+    AND started.invocation_started_at = ids.invocation_started_at
+LEFT JOIN terminals ON terminals.invocation_id = ids.invocation_id
+    AND terminals.invocation_started_at = ids.invocation_started_at
+LEFT JOIN beats ON beats.invocation_id = ids.invocation_id
+    AND beats.invocation_started_at = ids.invocation_started_at
+SETTINGS join_use_nulls = 1""",
+        "statements": f"""WITH
+facts AS (
+    SELECT * FROM {events} WHERE statement_id IS NOT NULL AND startsWith(event_type, 'statement_')
+),
+folded AS (
+    SELECT invocation_id, invocation_started_at, statement_id,
+        argMin(run_id, {order}) AS first_run_id,
+        argMin(resource_id, {order}) AS first_resource_id,
+        argMin(resource_attempt_id, {order}) AS first_resource_attempt_id,
+        argMin(operation_id, {order}) AS first_operation_id,
+        argMin(intent, {order}) AS first_intent,
+        argMin(statement_kind, {order}) AS first_statement_kind,
+        argMin(purpose, {order}) AS first_purpose,
+        argMin(sql_digest, {order}) AS first_sql_digest,
+        minIf(occurred_at, event_type = 'statement_started') AS started_at,
+        minIf(occurred_at, event_type = 'statement_submitted') AS submitted_at,
+        max(query_id) AS any_query_id,
+        countIf({terminal}) AS terminal_count,
+        argMaxIf(event_type, {order}, {terminal}) AS last_event_type,
+        argMaxIf(occurred_at, {order}, {terminal}) AS last_ended_at,
+        argMaxIf(query_id, {order}, {terminal}) AS last_query_id,
+        argMaxIf(error_type, {order}, {terminal}) AS last_error_type,
+        argMaxIf(error_code, {order}, {terminal}) AS last_error_code,
+        argMaxIf(duration_ms, {order}, {terminal}) AS last_duration_ms,
+        argMaxIf(row_count, {order}, {terminal}) AS last_row_count,
+        argMaxIf(affected_rows, {order}, {terminal}) AS last_affected_rows
+    FROM facts GROUP BY invocation_id, invocation_started_at, statement_id
+),
+resources AS (
+    SELECT invocation_id, invocation_started_at, resource_id,
+        min(resource_name) AS first_resource_name, min(resource_kind) AS first_resource_kind
+    FROM {events}
+    WHERE resource_id IS NOT NULL AND resource_name IS NOT NULL
+    GROUP BY invocation_id, invocation_started_at, resource_id
+)
+SELECT
+    folded.invocation_id AS invocation_id,
+    folded.invocation_started_at AS invocation_started_at,
+    folded.statement_id AS statement_id,
+    folded.first_run_id AS run_id,
+    folded.first_resource_id AS resource_id,
+    resources.first_resource_name AS resource_name,
+    resources.first_resource_kind AS resource_kind,
+    folded.first_resource_attempt_id AS resource_attempt_id,
+    folded.first_operation_id AS operation_id,
+    folded.first_intent AS intent,
+    folded.first_statement_kind AS statement_kind,
+    folded.first_purpose AS purpose,
+    if(folded.terminal_count > 0, coalesce(folded.last_query_id, folded.any_query_id),
+        folded.any_query_id) AS query_id,
+    folded.started_at AS started_at,
+    folded.submitted_at AS submitted_at,
+    if(folded.terminal_count > 0, folded.last_ended_at, NULL) AS ended_at,
+    multiIf(
+        folded.terminal_count = 0, 'no_terminal',
+        folded.last_event_type = 'statement_completed', 'completed',
+        'failed'
+    ) AS outcome,
+    if(folded.terminal_count > 0, folded.last_error_type, NULL) AS error_type,
+    if(folded.terminal_count > 0, folded.last_error_code, NULL) AS error_code,
+    if(folded.terminal_count > 0, folded.last_duration_ms, NULL) AS duration_ms,
+    if(folded.terminal_count > 0, folded.last_row_count, NULL) AS row_count,
+    if(folded.terminal_count > 0, folded.last_affected_rows, NULL) AS affected_rows,
+    folded.first_sql_digest AS sql_digest,
+    concat('sha256:', folded.first_sql_digest)
+        IN (SELECT digest FROM `{database}`.`recall_artifacts`) AS sql_available
+FROM folded
+LEFT JOIN resources
+    ON resources.invocation_id = folded.invocation_id
+    AND resources.invocation_started_at = folded.invocation_started_at
+    AND resources.resource_id = folded.first_resource_id
+SETTINGS join_use_nulls = 1""",
+        "output_invocations": f"""WITH
+received AS (
+    SELECT invocation_id,
+        min(invocation_started_at) AS invocation_started_at,
+        count() AS chunks,
+        sum(byte_count) AS bytes,
+        sumIf(byte_count, stream = 'stdout') AS stdout_bytes,
+        sumIf(byte_count, stream = 'stderr') AS stderr_bytes,
+        max(sequence) AS max_sequence,
+        sum(dropped_before) AS queue_drops,
+        sum(coalesce(omitted_sequences, 1)) AS spans,
+        sum(coalesce(omitted_bytes, 0)) AS truncated_bytes
+    FROM {output} GROUP BY invocation_id
+),
+delivery AS (SELECT * FROM {delivery} WHERE stream = 'output'),
+ids AS (
+    SELECT DISTINCT invocation_id, invocation_started_at
+    FROM (
+        SELECT invocation_id, invocation_started_at FROM received
+        UNION ALL SELECT invocation_id, invocation_started_at FROM delivery
+    )
+),
+counted AS (
+    SELECT
+        ids.invocation_id AS invocation_id,
+        ids.invocation_started_at AS invocation_started_at,
+        coalesce(received.chunks, 0) AS chunks,
+        coalesce(received.bytes, 0) AS bytes,
+        coalesce(received.stdout_bytes, 0) AS stdout_bytes,
+        coalesce(received.stderr_bytes, 0) AS stderr_bytes,
+        coalesce(received.truncated_bytes, 0) AS truncated_bytes,
+        received.max_sequence AS max_sequence,
+        greatest(
+            coalesce(toInt64(received.max_sequence) + 1 - toInt64(received.spans), 0),
+            coalesce(toInt64(delivery.accepted), 0) - coalesce(toInt64(received.chunks), 0)
+        ) AS missing_sequences,
+        greatest(coalesce(received.queue_drops, 0), coalesce(delivery.dropped, 0))
+            AS reported_drops,
+        coalesce(delivery.filtered, 0) AS filtered,
+        coalesce(delivery.failed, 0) AS failed,
+        coalesce(context.recall_filter_active, 0) AS filter_active,
+        coalesce(delivery.flush_complete, 0) AS flush_complete,
+        delivery.invocation_id IS NOT NULL AS delivery_reported,
+        coalesce(delivery.delivered, 0) AS delivered
+    FROM ids
+    LEFT JOIN received ON received.invocation_id = ids.invocation_id
+    LEFT JOIN delivery ON delivery.invocation_id = ids.invocation_id
+    LEFT JOIN {context} AS context ON context.invocation_id = ids.invocation_id
+    SETTINGS join_use_nulls = 1
+)
+SELECT
+    invocation_id, invocation_started_at, chunks, bytes, stdout_bytes, stderr_bytes,
+    truncated_bytes, max_sequence, missing_sequences, reported_drops, filtered, failed,
+    filter_active, flush_complete,
+    multiIf(
+        missing_sequences = 0 AND truncated_bytes > 0, 'truncated',
+        missing_sequences = 0, NULL,
+        filtered >= missing_sequences, 'filtered',
+        filtered + reported_drops + failed >= missing_sequences, 'dropped',
+        'unexplained'
+    ) AS gap_reason,
+    toUInt8(delivery_reported AND flush_complete = 1 AND missing_sequences = 0
+        AND reported_drops = 0 AND filtered = 0 AND failed = 0 AND truncated_bytes = 0
+        AND chunks >= delivered)
+        AS complete
+FROM counted""",
+        "artifacts": f"""SELECT
+    refs.invocation_id AS invocation_id,
+    refs.invocation_started_at AS invocation_started_at,
+    refs.digest AS digest,
+    refs.ref_id AS ref_id,
+    refs.role AS role,
+    refs.statement_id AS statement_id,
+    refs.resource_id AS resource_id,
+    artifact.kind AS kind,
+    artifact.media_type AS media_type,
+    artifact.size AS size,
+    artifact.digest IS NOT NULL AS available,
+    if(artifact.digest IS NULL, 'expired_or_not_captured', NULL) AS unavailable_reason,
+    if(startsWith(artifact.media_type, 'text/'), toValidUTF8(artifact.content), NULL) AS text,
+    artifact.content AS content
+FROM {refs} AS refs
+LEFT JOIN {stored} AS artifact ON artifact.digest = refs.digest
+SETTINGS join_use_nulls = 1""",
+    }
+
+
+_PLAN_READ: re.Pattern[str] = re.compile(r"ReadFromMergeTree \(([^)]+)\)")
+_PRIMARY_KEY_INDEX: re.Pattern[str] = re.compile(
+    r"^\s*PrimaryKey\n\s*Keys:\n(?:\s*\w+\n)*?\s*invocation_id\n", re.MULTILINE
+)
+RECALL_VIEWS: tuple[str, ...] = (
+    "invocations",
+    "statements",
+    "audit_results",
+    "output_invocations",
+    "output_tail",
+    "output_transcripts",
+    "events",
+    "diagnostics",
+    "artifacts",
+    "conflicts",
+)
+
+
+def fetch_with_settings(
+    *, config: dict[str, object], sql: str, settings: dict[str, object]
+) -> tuple[tuple[object, ...], ...]:
+    """Rows of `sql` from a plain client sending only `settings`, none of SQLBuild's."""
+
+    import clickhouse_connect
+
+    client: Any = clickhouse_connect.get_client(
+        host=str(config["host"]),
+        port=int(cast(int, config["port"])),
+        username=str(config["user"]),
+        password=str(config["password"]),
+        settings=settings,
+    )
+    try:
+        return tuple(tuple(row) for row in client.query(sql).result_rows)
+    finally:
+        client.close()
+
+
+def differing_rows(
+    *, config: dict[str, object], left: str, right: str, settings: dict[str, object]
+) -> int:
+    """Rows of `left` missing from `right` plus rows of `right` missing from `left`."""
+
+    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}))"
+        ),
+        settings=settings,
+    )
+    return int(cast(int, rows[0][0]))
+
+
+def described(*, config: dict[str, object], relation: str) -> tuple[tuple[object, ...], ...]:
+    """Each column's name and type, in order."""
+
+    return tuple(
+        (row[0], row[1])
+        for row in fetch_clickhouse_rows(config=config, sql=f"DESCRIBE TABLE {relation}")
+    )
+
+
+def create_previous_views(
+    *, config: dict[str, object], database: str, names: tuple[str, ...], prefix: str
+) -> None:
+    """Create the previous version's bodies of `names`, each under `prefix` + its name."""
+
+    bodies: dict[str, str] = previous_view_bodies(database=database)
+    execute_clickhouse_sql(
+        config=config,
+        statements=tuple(
+            f"CREATE OR REPLACE VIEW `{database}`.`{prefix}{name}` AS\n{bodies[name]}"
+            for name in names
+        ),
+    )
+
+
+def snapshot_views(*, config: dict[str, object], database: str, names: tuple[str, ...]) -> None:
+    """Copy each view's current rows into a `snapshot_` memory table."""
+
+    execute_clickhouse_sql(
+        config=config,
+        statements=tuple(
+            f"CREATE TABLE `{database}`.`snapshot_{name}` ENGINE = Memory AS "
+            f"SELECT * FROM `{database}`.`{name}`"
+            for name in names
+        ),
+    )
+
+
+def table_reads(*, config: dict[str, object], sql: str) -> tuple[str, ...]:
+    """The tables `sql` reads, once per read, from its plan."""
+
+    plan: tuple[tuple[object, ...], ...] = fetch_clickhouse_rows(
+        config=config, sql=f"EXPLAIN {sql}"
+    )
+    return tuple(sorted(_PLAN_READ.findall("\n".join(str(row[0]) for row in plan))))
+
+
+def primary_key_reads(*, config: dict[str, object], sql: str) -> int:
+    """How many of `sql`'s reads use the primary key by `invocation_id`."""
+
+    plan: tuple[tuple[object, ...], ...] = fetch_clickhouse_rows(
+        config=config, sql=f"EXPLAIN indexes = 1 {sql}"
+    )
+    return len(_PRIMARY_KEY_INDEX.findall("\n".join(str(row[0]) for row in plan)))
+
+
+@contextmanager
+def readonly_user(
+    *, config: dict[str, object], database: str, settings: str = "readonly = 1"
+) -> Iterator[dict[str, object]]:
+    """A user with `settings`, by default the standard `readonly = 1`, and `SELECT` on it."""
+
+    name: str = build_unique_schema_name(prefix="recall_reader")
+    password: str = uuid.uuid4().hex
+    execute_clickhouse_sql(
+        config=config,
+        statements=(
+            f"CREATE USER {name} IDENTIFIED BY '{password}' SETTINGS {settings}",
+            f"GRANT SELECT ON {database}.* TO {name}",
+        ),
+    )
+    try:
+        yield {**config, "user": name, "password": password}
+    finally:
+        execute_clickhouse_sql(config=config, statements=(f"DROP USER IF EXISTS {name}",))
+
+
+_FAILING_MODEL: str = (
+    "MODEL (\n  description 'An order whose amount does not parse',\n  materialized table,\n);\n\n"
+    "SELECT CAST('not a number' AS INTEGER) AS order_amount\n"
+)
+
+
+def representative_store(project_dir: Path) -> subprocess.Popen[bytes]:
+    """A completed, a failed, an abandoned and a heartbeat-only invocation, and a running one."""
+
+    assert run(project_dir, "build").returncode == 0
+    failing: Path = project_dir / "models" / "unparsed_orders.sql"
+    _ = failing.write_text(_FAILING_MODEL, encoding="utf-8")
+    assert run(project_dir, "build").returncode != 0
+    failing.unlink()
+    _ = killed_build(project_dir, heartbeat_seconds="0.5")
+    _ = killed_during_discovery(project_dir, heartbeat_seconds="0.5")
+    running: subprocess.Popen[bytes] = start(project_dir, "build", ORDERS_TASK_SLEEP_SECONDS="600")
+    _ = wait_for(
+        "SELECT invocation_id FROM invocations WHERE status = 'running' "
+        "AND invocation_id IN (SELECT invocation_id FROM recall_output)",
+        project_dir,
+        timeout=120,
+    )
+    return running
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 f851ad2c6b..6806b488f9 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
@@ -903,7 +903,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=((2,),),
+            expected_version=((3,),),
         ),
     ],
     ids=lambda case: case.description,
@@ -1039,12 +1039,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, 2, 1),),
+            expected_rows=((2, 3, 1),),
         ),
         ClickHouseRecallVersionRefreshTestCase(
             description="a recent version row is left as it is",
             row_age_days=1,
-            expected_rows=((1, 2, 0),),
+            expected_rows=((1, 3, 0),),
         ),
     ],
     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
new file mode 100644
index 0000000000..ba2d17a60a
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_views.py
@@ -0,0 +1,424 @@
+from __future__ import annotations
+
+import json
+import os
+import signal
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse._test_types import (
+    ClickHouseRecallPreviousVersionTestCase,
+    ClickHouseRecallReaderSettingsTestCase,
+    ClickHouseRecallReadOnlyUserTestCase,
+    ClickHouseRecallRefusedSettingTestCase,
+    ClickHouseRecallTableReadsTestCase,
+    ClickHouseRecallViewEquivalenceTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse.helpers import (
+    RECALL_VIEWS,
+    create_previous_views,
+    described,
+    differing_rows,
+    duplicate_recall_rows,
+    execute_clickhouse_sql,
+    fetch_clickhouse_rows,
+    limited_user,
+    primary_key_reads,
+    readonly_user,
+    representative_store,
+    snapshot_views,
+    table_reads,
+    write_recall_project,
+)
+from tests.integration.src.sqlbuild.cli.recall.helpers import recall_rows, run
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ClickHouseRecallReadOnlyUserTestCase(
+            description="a readonly = 1 user queries every view, the schema and the output",
+            view_names=RECALL_VIEWS,
+            expected_status=[["completed"]],
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_a_readonly_1_reader_when_recalling_then_no_setting_is_needed(
+    test_case: ClickHouseRecallReadOnlyUserTestCase,
+    tmp_path: Path,
+    clickhouse_e2e_config: dict[str, object],
+    recall_database: str,
+) -> None:
+    writer: Path = write_recall_project(
+        parent=tmp_path / "writer", config=clickhouse_e2e_config, database=recall_database
+    )
+    built: subprocess.CompletedProcess[bytes] = run(writer, "build")
+
+    with readonly_user(config=clickhouse_e2e_config, database=recall_database) as reader:
+        project_dir: Path = write_recall_project(
+            parent=tmp_path / "reader", config=reader, database=recall_database
+        )
+        counted: list[subprocess.CompletedProcess[bytes]] = [
+            run(project_dir, "recall", "query", "--json", f"SELECT count() FROM {view}")
+            for view in test_case.view_names
+        ]
+        limited: subprocess.CompletedProcess[bytes] = run(
+            project_dir, "recall", "query", "--json", "--limit", "1", "SELECT * FROM events"
+        )
+        schema: subprocess.CompletedProcess[bytes] = run(project_dir, "recall", "schema")
+        ddl: subprocess.CompletedProcess[bytes] = run(project_dir, "recall", "schema", "--ddl")
+        output: subprocess.CompletedProcess[bytes] = run(
+            project_dir, "recall", "output", "last", "--stream", "stdout"
+        )
+        status: list[list[object]] = recall_rows(project_dir, "SELECT status FROM invocations")
+        refused: subprocess.CompletedProcess[bytes] = run(
+            project_dir, "recall", "query", f"DROP TABLE {recall_database}.recall_meta"
+        )
+
+    assert built.returncode == 0, built.stderr
+    assert [completed.returncode for completed in counted] == [0 for _ in counted], [
+        completed.stderr for completed in counted
+    ]
+    assert [len(json.loads(completed.stdout)["rows"]) for completed in counted] == [
+        1 for _ in counted
+    ]
+    document: dict[str, object] = json.loads(limited.stdout)
+    assert (limited.returncode, len(document["rows"]), document["truncated"]) == (0, 1, True)  # type: ignore[arg-type]
+    assert schema.returncode == 0, schema.stderr
+    assert b"invocations" in schema.stdout
+    assert (ddl.returncode, ddl.stdout) == (0, b""), ddl.stderr
+    assert (output.returncode, output.stdout) == (0, built.stdout), output.stderr
+    assert status == test_case.expected_status
+    assert refused.returncode == 1
+    assert fetch_clickhouse_rows(
+        config=clickhouse_e2e_config,
+        sql=f"SELECT count() FROM system.tables WHERE database = '{recall_database}' "
+        "AND name = 'recall_meta'",
+    ) == ((1,),)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ClickHouseRecallRefusedSettingTestCase(
+            description="a user that may not change readonly is told to set readonly = 2",
+            user_settings="readonly = 0 CONST",
+            expected_fragments=(
+                "error[C974]",
+                "may not change the setting `readonly`",
+                "`readonly = 2`",
+            ),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_a_reader_that_may_not_set_readonly_when_querying_then_the_error_names_the_setting(
+    test_case: ClickHouseRecallRefusedSettingTestCase,
+    tmp_path: Path,
+    clickhouse_e2e_config: dict[str, object],
+    recall_database: str,
+) -> None:
+    writer: Path = write_recall_project(
+        parent=tmp_path / "writer", config=clickhouse_e2e_config, database=recall_database
+    )
+    assert run(writer, "build").returncode == 0
+
+    with readonly_user(
+        config=clickhouse_e2e_config, database=recall_database, settings=test_case.user_settings
+    ) as reader:
+        project_dir: Path = write_recall_project(
+            parent=tmp_path / "reader", config=reader, database=recall_database
+        )
+        queried: subprocess.CompletedProcess[bytes] = run(
+            project_dir, "recall", "query", "SELECT count() FROM invocations"
+        )
+        output: subprocess.CompletedProcess[bytes] = run(project_dir, "recall", "output", "last")
+
+    stderr: str = queried.stderr.decode(errors="replace")
+    assert queried.returncode == 1
+    assert [fragment in stderr for fragment in test_case.expected_fragments] == [
+        True for _ in test_case.expected_fragments
+    ]
+    assert output.returncode == 0, output.stderr
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ClickHouseRecallReaderSettingsTestCase(
+            description="a readonly = 1 reader runs with the server's own settings",
+            user_settings="readonly = 1",
+            expected_rows=[[0, "", 1]],
+        ),
+        ClickHouseRecallReaderSettingsTestCase(
+            description="a readonly = 2 reader gets the pinned settings",
+            user_settings="readonly = 2",
+            expected_rows=[[1, "DISTINCT", 1]],
+        ),
+        ClickHouseRecallReaderSettingsTestCase(
+            description="a reader allowed to change settings gets the pinned settings",
+            user_settings="readonly = 0",
+            expected_rows=[[1, "DISTINCT", 1]],
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_a_reader_when_querying_then_pinned_settings_apply_unless_it_may_not_set_them(
+    test_case: ClickHouseRecallReaderSettingsTestCase,
+    tmp_path: Path,
+    clickhouse_e2e_config: dict[str, object],
+    recall_database: str,
+) -> None:
+    writer: Path = write_recall_project(
+        parent=tmp_path / "writer", config=clickhouse_e2e_config, database=recall_database
+    )
+    assert run(writer, "build").returncode == 0
+
+    with readonly_user(
+        config=clickhouse_e2e_config, database=recall_database, settings=test_case.user_settings
+    ) as reader:
+        project_dir: Path = write_recall_project(
+            parent=tmp_path / "reader", config=reader, database=recall_database
+        )
+        rows: list[list[object]] = recall_rows(
+            project_dir,
+            "SELECT toUInt8(getSetting('join_use_nulls')), "
+            "toString(getSetting('union_default_mode')), count() FROM invocations",
+        )
+
+    assert rows == test_case.expected_rows
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ClickHouseRecallViewEquivalenceTestCase(
+            description="rewritten views match the previous ones with and without duplicates",
+            replaced_views=("invocations", "statements", "output_invocations", "artifacts"),
+            all_views=RECALL_VIEWS,
+            expected_statuses=[
+                ["abandoned"],
+                ["abandoned"],
+                ["completed"],
+                ["failed"],
+                ["running"],
+            ],
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_representative_invocations_when_views_are_rewritten_then_rows_and_types_are_unchanged(
+    test_case: ClickHouseRecallViewEquivalenceTestCase,
+    tmp_path: Path,
+    clickhouse_e2e_config: dict[str, object],
+    recall_database: str,
+) -> None:
+    project_dir: Path = write_recall_project(
+        parent=tmp_path, config=clickhouse_e2e_config, database=recall_database
+    )
+    running: subprocess.Popen[bytes] = representative_store(project_dir)
+    try:
+        create_previous_views(
+            config=clickhouse_e2e_config,
+            database=recall_database,
+            names=test_case.replaced_views,
+            prefix="previous_",
+        )
+        statuses: list[list[object]] = recall_rows(
+            project_dir, "SELECT status FROM invocations ORDER BY status"
+        )
+        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},
+            )
+            for view in test_case.replaced_views
+        ]
+        types: list[bool] = [
+            described(config=clickhouse_e2e_config, relation=f"{recall_database}.{view}")
+            == described(
+                config=clickhouse_e2e_config, relation=f"{recall_database}.previous_{view}"
+            )
+            for view in test_case.replaced_views
+        ]
+    finally:
+        os.killpg(running.pid, signal.SIGKILL)
+        _ = running.wait(timeout=30)
+    snapshot_views(
+        config=clickhouse_e2e_config, database=recall_database, names=test_case.all_views
+    )
+    without_join_nulls: list[int] = [
+        differing_rows(
+            config=clickhouse_e2e_config,
+            left=f"{recall_database}.{view}",
+            right=f"{recall_database}.snapshot_{view}",
+            settings={"join_use_nulls": 0},
+        )
+        for view in test_case.all_views
+    ]
+    duplicate_recall_rows(config=clickhouse_e2e_config, database=recall_database)
+    with_duplicates: list[int] = [
+        differing_rows(
+            config=clickhouse_e2e_config,
+            left=f"{recall_database}.{view}",
+            right=f"{recall_database}.snapshot_{view}",
+            settings={},
+        )
+        for view in test_case.all_views
+    ]
+    previous_with_duplicates: list[int] = [
+        differing_rows(
+            config=clickhouse_e2e_config,
+            left=f"{recall_database}.{view}",
+            right=f"{recall_database}.previous_{view}",
+            settings={"join_use_nulls": 1},
+        )
+        for view in test_case.replaced_views
+    ]
+    counts_with_duplicates: list[bool] = [
+        fetch_clickhouse_rows(
+            config=clickhouse_e2e_config,
+            sql=f"SELECT (SELECT count() FROM {recall_database}.{view}) "
+            f"= (SELECT count() FROM {recall_database}.snapshot_{view})",
+        )
+        == ((1,),)
+        for view in test_case.all_views
+    ]
+
+    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 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]
+    assert counts_with_duplicates == [True for _ in test_case.all_views]
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ClickHouseRecallTableReadsTestCase(
+            description="invocations reads its three tables once each",
+            view="invocations",
+            expected_tables=("recall_events", "recall_heartbeats", "recall_invocations"),
+            expected_primary_key_reads=3,
+        ),
+        ClickHouseRecallTableReadsTestCase(
+            description="statements reads the events once and the stored SQL once",
+            view="statements",
+            expected_tables=("recall_artifacts", "recall_events"),
+            expected_primary_key_reads=1,
+        ),
+        ClickHouseRecallTableReadsTestCase(
+            description="output_invocations reads its three tables once each",
+            view="output_invocations",
+            expected_tables=("recall_delivery", "recall_invocations", "recall_output"),
+            expected_primary_key_reads=3,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_one_invocation_lookup_when_planned_then_each_table_is_read_once_by_its_key(
+    test_case: ClickHouseRecallTableReadsTestCase,
+    tmp_path: Path,
+    clickhouse_e2e_config: dict[str, object],
+    recall_database: str,
+) -> None:
+    project_dir: Path = write_recall_project(
+        parent=tmp_path, config=clickhouse_e2e_config, database=recall_database
+    )
+    assert run(project_dir, "build").returncode == 0
+    sql: str = (
+        f"SELECT * FROM {recall_database}.{test_case.view} WHERE invocation_id = 'any-invocation'"
+    )
+
+    reads: tuple[str, ...] = table_reads(config=clickhouse_e2e_config, sql=sql)
+    keyed: int = primary_key_reads(config=clickhouse_e2e_config, sql=sql)
+
+    assert reads == tuple(f"{recall_database}.{table}" for table in test_case.expected_tables)
+    assert keyed == test_case.expected_primary_key_reads
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ClickHouseRecallPreviousVersionTestCase(
+            description="a store of the previous version is upgraded by whoever may",
+            replaced_views=("invocations", "statements", "output_invocations", "artifacts"),
+            expected_notice="is behind this SQLBuild version, and this user may not upgrade it",
+            expected_ddl_fragments=(
+                "CREATE OR REPLACE VIEW `{database}`.`invocations`",
+                "CREATE OR REPLACE VIEW `{database}`.`statements`",
+                "SELECT 'schema', 'schema_version', '3'",
+            ),
+            expected_version=((3,),),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_a_previous_version_store_when_built_then_limited_writers_warn_and_admins_upgrade(
+    test_case: ClickHouseRecallPreviousVersionTestCase,
+    tmp_path: Path,
+    clickhouse_e2e_config: dict[str, object],
+    recall_database: str,
+) -> None:
+    admin: Path = write_recall_project(
+        parent=tmp_path / "admin", config=clickhouse_e2e_config, database=recall_database
+    )
+    assert run(admin, "build").returncode == 0
+    create_previous_views(
+        config=clickhouse_e2e_config,
+        database=recall_database,
+        names=test_case.replaced_views,
+        prefix="",
+    )
+    execute_clickhouse_sql(
+        config=clickhouse_e2e_config,
+        statements=(
+            f"ALTER TABLE {recall_database}.recall_meta UPDATE value = '2' "
+            "WHERE component = 'schema' SETTINGS mutations_sync = 2",
+        ),
+    )
+
+    with limited_user(config=clickhouse_e2e_config, database=recall_database) as writer:
+        writer_dir: Path = write_recall_project(
+            parent=tmp_path / "writer", config=writer, database=recall_database
+        )
+        limited: subprocess.CompletedProcess[bytes] = run(writer_dir, "build")
+        printed: subprocess.CompletedProcess[bytes] = run(writer_dir, "recall", "schema", "--ddl")
+    upgraded: subprocess.CompletedProcess[bytes] = run(admin, "build")
+    repeated: subprocess.CompletedProcess[bytes] = run(admin, "recall", "schema", "--ddl")
+
+    ddl: str = printed.stdout.decode()
+    assert limited.returncode == 0, limited.stderr
+    assert test_case.expected_notice in limited.stderr.decode(errors="replace")
+    assert [
+        fragment.format(database=recall_database) in ddl
+        for fragment in test_case.expected_ddl_fragments
+    ] == [True for _ in test_case.expected_ddl_fragments]
+    assert upgraded.returncode == 0, upgraded.stderr
+    assert b"Recall" not in upgraded.stderr
+    assert (repeated.returncode, repeated.stdout) == (0, b"")
+    assert (
+        fetch_clickhouse_rows(
+            config=clickhouse_e2e_config,
+            sql=f"SELECT max(toInt64OrZero(value)) FROM {recall_database}.recall_meta "
+            "WHERE component = 'schema'",
+        )
+        == test_case.expected_version
+    )
+    assert fetch_clickhouse_rows(
+        config=clickhouse_e2e_config,
+        sql=f"SELECT countIf(position(create_table_query, 'join_use_nulls') > 0) "
+        f"FROM system.tables WHERE database = '{recall_database}' AND engine = 'View'",
+    ) == ((0,),)
+    assert recall_rows(admin, "SELECT count() FROM invocations") == [[3]]
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])
diff --git a/website/src/content/docs/docs/concepts/adapters/clickhouse.mdx b/website/src/content/docs/docs/concepts/adapters/clickhouse.mdx
index c8759345e5..91367a362d 100644
--- a/website/src/content/docs/docs/concepts/adapters/clickhouse.mdx
+++ b/website/src/content/docs/docs/concepts/adapters/clickhouse.mdx
@@ -161,6 +161,8 @@ Every query runs with these settings so SQL behaves the same as on other adapter
 
 The connection's `settings` may repeat these values but not change them: a different value fails to connect with `A710`, because SQLBuild's generated SQL depends on them.
 
+A [Recall](/docs/concepts/observability/recall) `read_connection` whose user is `readonly = 1`, and so may not change settings, sends none of them: its `sqb recall query` SQL runs with the server's own settings. Every other reader gets them.
+
 ClickHouse returns a type's default value (`0`, the epoch, an empty string) from `lag` and `lead` past the edge of a window frame unless the argument is `Nullable`. SQLBuild's generated snapshot SQL wraps those arguments in `toNullable` so they return `NULL` as on other adapters. Write the same in your own models when you rely on `NULL` there.
 
 ## SQL differences to watch
@@ -183,5 +185,6 @@ A ClickHouse connection can also hold [Recall](/docs/concepts/observability/reca
 history for a whole team, whichever adapter the project builds with. Set
 `[recall] write_connection` and `read_connection` to it, one connection or separate writer and
 reader users. Recall creates its tables on first use, keeps every output byte by default, and needs
-only `SELECT` and `INSERT` once the tables exist; readers need only `SELECT`. See
+only `SELECT` and `INSERT` once the tables exist; readers need only `SELECT` and may be
+`readonly = 1` users. See
 [Sharing history in ClickHouse](/docs/concepts/observability/recall#sharing-history-in-clickhouse).
diff --git a/website/src/content/docs/docs/concepts/observability/recall.mdx b/website/src/content/docs/docs/concepts/observability/recall.mdx
index b3b6e9c8b3..4d4c3f29d6 100644
--- a/website/src/content/docs/docs/concepts/observability/recall.mdx
+++ b/website/src/content/docs/docs/concepts/observability/recall.mdx
@@ -260,6 +260,16 @@ GRANT SELECT, INSERT ON observability.* TO recall_writer;
 GRANT SELECT ON observability.* TO recall_reader;
 ```
 
+The reader may be a standard read-only user (`SETTINGS readonly = 1`, or a settings profile with
+it), which may not change any setting. Such a user's queries run with the server's own settings;
+every other reader gets the
+[settings SQLBuild pins](/docs/concepts/adapters/clickhouse#settings-sqlbuild-pins). The views
+return the same rows either way. `sqb recall query` runs with
+`readonly = 2`, which refuses writes but still allows query settings, unless the user is already
+read-only; a user that may not set the row limit gets the same truncated result, cut off while
+reading. A setting that Recall needs and the user may not change fails with an error naming it; set
+it in the user's profile, or give the user `readonly = 2`, which keeps it read-only.
+
 Then, as the administrator, create the store with the DDL that `sqb recall schema --ddl` prints
 (any connection that can read the database works, so the reader's is enough):
 
@@ -286,12 +296,14 @@ writing to an upgraded store. `--ddl` prints SQL, so it cannot be combined with
   recorded. When it has expired anyway, a writer compares the stored tables' and views' columns
   with its own: it upgrades an older store, records its version on a current one, and never
   replaces the views of a store a newer SQLBuild version created.
-- **Writes.** History is sent in batches of up to 5,000 rows about once a second, one insert per
-  table, with the same queue, shutdown budget and exit-code guarantees as the local store. Many
-  concurrent commands can write one store.
+- **Writes.** History is sent in batches of up to 5,000 rows about every five seconds, one insert
+  per table, so a long command adds few parts; everything still queued is sent when it ends, with
+  the same queue, shutdown budget and exit-code guarantees as the local store. Many concurrent
+  commands can write one store.
 - **Duplicates.** Every row has a deterministic key. A retried batch is deduplicated by ClickHouse,
-  and the views read each key once (`FINAL` on `ReplacingMergeTree` tables), so a repeated delivery
-  is counted once. Before writing, Recall looks up the batch's keys: a key already stored with
+  and the views read each key once, so a repeated delivery is counted once: views that list or count
+  rows read the `ReplacingMergeTree` tables with `FINAL`, and `invocations` and `statements`, which
+  only take the first, last, smallest or largest value of a key's copies, read them without it. Before writing, Recall looks up the batch's keys: a key already stored with
   different content keeps the stored row, and the rejected copy is listed in `conflicts`.
 - **Retention.** Every table carries a TTL of `retention` (SQL text a week longer, and refreshed
   while it is still referenced), and the views hide rows past their TTL at once. ClickHouse drops