From a5e7c6c5bf10f39573def10d1324175c4df7aa04 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 16:47:06 +0100
Subject: [PATCH 1/8] perf(planner): read fingerprint state once per plan

Full-project plans and projects declaring migrations read the latest
fingerprint per identity without a node filter, apply the selection in
memory, and share the rows with migration inspection instead of reading
the state table a second time. Narrow selections keep the filtered read.
---
 .../compiler/fingerprints/_helpers/latest.py  | 25 ++++++
 .../fingerprints/main/_select_latest.py       | 28 ++++++
 .../compiler/fingerprints/main/read.py        | 15 +---
 .../planner/_helpers/migrations/planning.py   |  1 +
 .../planner/_helpers/warehouse/snapshot.py    | 42 ++++++++-
 .../classes/migration_state_inspection.py     | 40 ++++++---
 src/sqlbuild/compiler/planner/models.py       |  4 +
 .../main/model_migrations/_test_types.py      | 11 +++
 .../commands/main/model_migrations/helpers.py | 51 +++++++++++
 .../test_fingerprint_state_reads.py           | 88 +++++++++++++++++++
 .../compiler/fingerprints/main/_test_types.py |  8 ++
 .../compiler/fingerprints/main/helpers.py     | 38 ++++++++
 .../fingerprints/main/test_read_write.py      | 78 ++++++++++++++++
 13 files changed, 401 insertions(+), 28 deletions(-)
 create mode 100644 src/sqlbuild/compiler/fingerprints/_helpers/latest.py
 create mode 100644 src/sqlbuild/compiler/fingerprints/main/_select_latest.py
 create mode 100644 tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_state_reads.py

diff --git a/src/sqlbuild/compiler/fingerprints/_helpers/latest.py b/src/sqlbuild/compiler/fingerprints/_helpers/latest.py
new file mode 100644
index 0000000000..7a360e9f9e
--- /dev/null
+++ b/src/sqlbuild/compiler/fingerprints/_helpers/latest.py
@@ -0,0 +1,25 @@
+"""Latest fingerprint set assembly shared by warehouse reads and in-memory selection."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_MODEL
+from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet
+
+
+def build_latest_fingerprint_set(*, schema: str, latest: Iterable[Fingerprint]) -> FingerprintSet:
+    """Index latest rows by identity and by node name, preferring models on name collisions."""
+
+    fingerprints: dict[str, Fingerprint] = {}
+    fingerprints_by_identity: dict[tuple[str, str], Fingerprint] = {}
+    fingerprint: Fingerprint
+    for fingerprint in latest:
+        fingerprints_by_identity[(fingerprint.node_type, fingerprint.node_name)] = fingerprint
+        if fingerprint.node_type == NODE_TYPE_MODEL or fingerprint.node_name not in fingerprints:
+            fingerprints[fingerprint.node_name] = fingerprint
+    return FingerprintSet(
+        schema=schema,
+        fingerprints=fingerprints,
+        fingerprints_by_identity=fingerprints_by_identity,
+    )
diff --git a/src/sqlbuild/compiler/fingerprints/main/_select_latest.py b/src/sqlbuild/compiler/fingerprints/main/_select_latest.py
new file mode 100644
index 0000000000..656c02090b
--- /dev/null
+++ b/src/sqlbuild/compiler/fingerprints/main/_select_latest.py
@@ -0,0 +1,28 @@
+"""In-memory node selection over an unfiltered latest fingerprint read."""
+
+from __future__ import annotations
+
+from sqlbuild.compiler.fingerprints._helpers.latest import build_latest_fingerprint_set
+from sqlbuild.compiler.fingerprints.models import FingerprintSet
+
+
+def select_latest_fingerprints(
+    *,
+    fingerprint_set: FingerprintSet,
+    node_names: tuple[str, ...],
+    filtered_node_types: tuple[str, ...] = (),
+) -> FingerprintSet:
+    """Apply the ``read_latest_fingerprints`` node filter to an unfiltered latest read."""
+
+    if not node_names and not filtered_node_types:
+        return FingerprintSet(schema=fingerprint_set.schema, fingerprints={})
+    names: frozenset[str] = frozenset(node_names)
+    types: frozenset[str] = frozenset(filtered_node_types)
+    return build_latest_fingerprint_set(
+        schema=fingerprint_set.schema,
+        latest=(
+            fingerprint
+            for fingerprint in (fingerprint_set.fingerprints_by_identity or {}).values()
+            if fingerprint.node_name in names or (types and fingerprint.node_type not in types)
+        ),
+    )
diff --git a/src/sqlbuild/compiler/fingerprints/main/read.py b/src/sqlbuild/compiler/fingerprints/main/read.py
index f22200c28e..df1a170199 100644
--- a/src/sqlbuild/compiler/fingerprints/main/read.py
+++ b/src/sqlbuild/compiler/fingerprints/main/read.py
@@ -9,10 +9,10 @@
 from typing import Any
 
 from sqlbuild.adapter.contract.types import AdapterExecute
+from sqlbuild.compiler.fingerprints._helpers.latest import build_latest_fingerprint_set
 from sqlbuild.compiler.fingerprints._helpers.sql import (
     build_qualified_table_name,
 )
-from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_MODEL
 from sqlbuild.compiler.fingerprints.exceptions import FingerprintInputError
 from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet
 
@@ -75,18 +75,9 @@ def read_latest_fingerprints(
             "fingerprint table to regenerate fingerprints."
         ) from error
     rows: list[tuple[Any, ...]] = result.fetchall()
-    fingerprints: dict[str, Fingerprint] = {}
-    fingerprints_by_identity: dict[tuple[str, str], Fingerprint] = {}
-    row: tuple[Any, ...]
-    for row in rows:
-        fingerprint: Fingerprint = _row_to_fingerprint(row=row, qualified_name=qualified_name)
-        fingerprints_by_identity[(fingerprint.node_type, fingerprint.node_name)] = fingerprint
-        if fingerprint.node_type == NODE_TYPE_MODEL or fingerprint.node_name not in fingerprints:
-            fingerprints[fingerprint.node_name] = fingerprint
-    return FingerprintSet(
+    return build_latest_fingerprint_set(
         schema=schema,
-        fingerprints=fingerprints,
-        fingerprints_by_identity=fingerprints_by_identity,
+        latest=(_row_to_fingerprint(row=row, qualified_name=qualified_name) for row in rows),
     )
 
 
diff --git a/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py b/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py
index df0491ea04..4b1638a0cc 100644
--- a/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py
+++ b/src/sqlbuild/compiler/planner/_helpers/migrations/planning.py
@@ -147,6 +147,7 @@ def plan_model_migrations(
         adapter=runtime.adapter,
         connection=runtime.connection,
         database=planning_database(runtime=runtime),
+        known_fingerprints=snapshot.fingerprints,
     )
     declares_migrations: bool = any(
         _declared_request(model) is not None for model in runtime.project.models
diff --git a/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py b/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py
index 1bb31e2133..aebb030fd9 100644
--- a/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py
+++ b/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py
@@ -24,6 +24,7 @@
     run_recorded_inspection_query,
 )
 from sqlbuild.adapter.relations.models import InspectionQueryRecord
+from sqlbuild.compiler.compile.constants import MIGRATE_FROM_CONFIG_KEY
 from sqlbuild.compiler.compile.main._cursor_roles import resolve_cursor_input_roles
 from sqlbuild.compiler.compile.models import (
     CompiledFunction,
@@ -42,6 +43,7 @@
     NODE_TYPE_MODEL,
     NODE_TYPE_SEED,
 )
+from sqlbuild.compiler.fingerprints.main._select_latest import select_latest_fingerprints
 from sqlbuild.compiler.fingerprints.main.read import read_latest_fingerprints
 from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet
 from sqlbuild.compiler.graph.main.transitive_closure_many import transitive_closure_many
@@ -297,6 +299,9 @@ def gather_warehouse_snapshot(
         schemas=query_schemas,
         fingerprint_state_schemas=fingerprint_state_schemas,
         node_names=_selected_node_names(relevant_keys),
+        read_unfiltered=_shares_unfiltered_fingerprints(
+            project=project, selected_keys=relevant_keys
+        ),
     )
 
     effective_full_refresh_names: frozenset[str] = (
@@ -695,8 +700,9 @@ def _gather_fingerprints(
     schemas: tuple[str, ...] | None,
     fingerprint_state_schemas: frozenset[str],
     node_names: tuple[str, ...] | None,
+    read_unfiltered: bool,
 ) -> WarehouseFingerprints:
-    """Read latest fingerprints across all target schemas grouped by node type."""
+    """Read latest fingerprints per target schema, keeping unfiltered reads for migrations."""
 
     if schemas is None:
         return WarehouseFingerprints()
@@ -704,6 +710,8 @@ def _gather_fingerprints(
     function_fingerprints: dict[str, Fingerprint] = {}
     seed_fingerprints: dict[str, Fingerprint] = {}
     python_fingerprints: dict[tuple[str, str], Fingerprint] = {}
+    unfiltered_schemas: dict[str, tuple[Fingerprint, ...]] = {}
+    filtered_node_types: tuple[str, ...] = (NODE_TYPE_MODEL, *FUNCTION_NODE_TYPES, NODE_TYPE_SEED)
     schema: str
     for schema in schemas:
         fingerprint_set: FingerprintSet = read_latest_fingerprints(
@@ -714,9 +722,19 @@ def _gather_fingerprints(
             schema=schema,
             render_qualified_name=adapter.render_qualified_name,
             render_read_latest_sql=adapter.render_read_latest_fingerprints_sql,
-            node_names=node_names,
-            filtered_node_types=(NODE_TYPE_MODEL, *FUNCTION_NODE_TYPES, NODE_TYPE_SEED),
+            node_names=None if read_unfiltered else node_names,
+            filtered_node_types=filtered_node_types,
         )
+        if read_unfiltered:
+            unfiltered_schemas[schema.lower()] = tuple(
+                (fingerprint_set.fingerprints_by_identity or {}).values()
+            )
+            if node_names is not None:
+                fingerprint_set = select_latest_fingerprints(
+                    fingerprint_set=fingerprint_set,
+                    node_names=node_names,
+                    filtered_node_types=filtered_node_types,
+                )
         node_name: str
         fingerprint: Fingerprint
         for node_name, fingerprint in fingerprint_set.fingerprints.items():
@@ -740,6 +758,8 @@ def _gather_fingerprints(
         functions=function_fingerprints,
         seeds=seed_fingerprints,
         python_nodes=python_fingerprints,
+        unfiltered_schemas=unfiltered_schemas,
+        unfiltered_database=database,
     )
 
 
@@ -751,6 +771,22 @@ def _selected_node_names(
     return tuple(sorted({key.name for key in selected_keys}))
 
 
+def _shares_unfiltered_fingerprints(
+    *,
+    project: CompiledProject,
+    selected_keys: frozenset[CompiledObjectKey] | None,
+) -> bool:
+    """Return whether one unfiltered read can serve this selection and migration inspection."""
+
+    if selected_keys is None:
+        return True
+    selected: frozenset[str] = frozenset(key.name for key in selected_keys)
+    return all(model.name in selected for model in project.models) or any(
+        isinstance(model.config.values.get(MIGRATE_FROM_CONFIG_KEY), str)
+        for model in project.models
+    )
+
+
 def _gather_cursor_snapshots(
     *,
     project: CompiledProject,
diff --git a/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py b/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py
index a612b03206..0a63ed1c45 100644
--- a/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py
+++ b/src/sqlbuild/compiler/planner/classes/migration_state_inspection.py
@@ -35,15 +35,28 @@
     OldNameViewEvent,
     OldNameViewHistory,
 )
+from sqlbuild.compiler.planner.models import WarehouseFingerprints
 
 
 class MigrationStateInspection:
     """Relations, fingerprints, and migration events for one planning database."""
 
-    def __init__(self, *, adapter: BaseAdapter, connection: Any, database: str | None) -> None:
+    def __init__(
+        self,
+        *,
+        adapter: BaseAdapter,
+        connection: Any,
+        database: str | None,
+        known_fingerprints: WarehouseFingerprints | None = None,
+    ) -> None:
         self._adapter: BaseAdapter = adapter
         self._connection: Any = connection
         self._database: str | None = database
+        self._known_fingerprints: dict[str, tuple[Fingerprint, ...]] = (
+            known_fingerprints.unfiltered_schemas
+            if known_fingerprints is not None and known_fingerprints.unfiltered_database == database
+            else {}
+        )
         self._read_schemas: set[str] = set()
         self._fingerprints: dict[str, tuple[Fingerprint, ...]] = {}
         self._events: list[MigrationEvent] = []
@@ -210,19 +223,20 @@ def inspect_schemas(self, *, schemas: set[str]) -> None:
                 self._fingerprints[schema.lower()] = self._read_model_fingerprints(schema=schema)
 
     def _read_model_fingerprints(self, *, schema: str) -> tuple[Fingerprint, ...]:
-        fingerprint_set: FingerprintSet = read_latest_fingerprints(
-            connection=self._connection,
-            execute=self._adapter.execute,
-            table_exists=True,
-            database=self._database,
-            schema=schema,
-            render_qualified_name=self._adapter.render_qualified_name,
-            render_read_latest_sql=self._adapter.render_read_latest_fingerprints_sql,
-        )
+        latest: tuple[Fingerprint, ...] | None = self._known_fingerprints.get(schema.lower())
+        if latest is None:
+            fingerprint_set: FingerprintSet = read_latest_fingerprints(
+                connection=self._connection,
+                execute=self._adapter.execute,
+                table_exists=True,
+                database=self._database,
+                schema=schema,
+                render_qualified_name=self._adapter.render_qualified_name,
+                render_read_latest_sql=self._adapter.render_read_latest_fingerprints_sql,
+            )
+            latest = tuple((fingerprint_set.fingerprints_by_identity or {}).values())
         return tuple(
-            fingerprint
-            for fingerprint in (fingerprint_set.fingerprints_by_identity or {}).values()
-            if fingerprint.node_type == NODE_TYPE_MODEL
+            fingerprint for fingerprint in latest if fingerprint.node_type == NODE_TYPE_MODEL
         )
 
     def inspect_relations(self, *, locations: tuple[CompiledRelationLocation, ...]) -> None:
diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py
index 2c2c510c49..019bdc3b13 100644
--- a/src/sqlbuild/compiler/planner/models.py
+++ b/src/sqlbuild/compiler/planner/models.py
@@ -601,6 +601,10 @@ class WarehouseFingerprints:
     functions: dict[str, Fingerprint] = field(default_factory=dict)
     seeds: dict[str, Fingerprint] = field(default_factory=dict)
     python_nodes: dict[tuple[str, str], Fingerprint] = field(default_factory=dict)
+    unfiltered_schemas: dict[str, tuple[Fingerprint, ...]] = field(
+        default_factory=dict, compare=False, repr=False
+    )
+    unfiltered_database: str | None = field(default=None, compare=False, repr=False)
 
 
 @dataclass(frozen=True)
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py
index b79aa64a37..c690e2e4e3 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/_test_types.py
@@ -143,3 +143,14 @@ class MigrationFingerprintWorkTestCase:
     expected_computations: int
     expected_migrations: int
     expected_stored_models: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class FingerprintStateReadTestCase:
+    description: str
+    prior_models: tuple[Callable[[], dict[str, str]], ...]
+    models: Callable[[], dict[str, str]]
+    plan_args: tuple[str, ...]
+    expected_reads: int
+    expected_rows_read: Callable[[int], int]
+    expected_migrations: tuple[tuple[str | None, str, str, str], ...]
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py
index c0e7c0a8c9..72d48a5e89 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/helpers.py
@@ -18,6 +18,7 @@
 from sqlbuild.adapter.contract.types import MigrationTransfer
 from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter
 from sqlbuild.cli.commands.main.entrypoint.entry import main
+from sqlbuild.compiler.fingerprints.constants import FINGERPRINT_TABLE_NAME
 from sqlbuild.executor.build._helpers import scheduler as scheduler_module
 
 DATABASE_FILE: str = "orders.duckdb"
@@ -806,3 +807,53 @@ def build_each(
     for models in model_sets:
         write_project(project_dir=project_dir, models=models())
         _ = build_ok(project_dir=project_dir, capsys=capsys)
+
+
+def _is_latest_fingerprint_read(sql: str) -> bool:
+    return FINGERPRINT_TABLE_NAME in sql and "__sqlbuild_latest_rank" in sql
+
+
+def record_fingerprint_state_reads(monkeypatch: pytest.MonkeyPatch) -> list[int]:
+    """Record the row count of every latest-fingerprint read the DuckDB adapter executes."""
+
+    rows_read: list[int] = []
+    original: Callable[..., Any] = DuckDbAdapter._execute
+
+    def recording(self: DuckDbAdapter, *, connection: Any, sql: str) -> Any:
+        rows_read.extend(
+            int(connection.execute(f"SELECT count(*) FROM ({read_sql})").fetchone()[0])
+            for read_sql in filter(_is_latest_fingerprint_read, (sql,))
+        )
+        return original(self, connection=connection, sql=sql)
+
+    monkeypatch.setattr(DuckDbAdapter, "_execute", recording)
+    return rows_read
+
+
+def fingerprint_history(*, project_dir: Path) -> tuple[int, int]:
+    """Return (stored rows, distinct node identities) of the fingerprint state table."""
+
+    row: tuple[Any, ...] = query(
+        project_dir=project_dir,
+        sql=(
+            "SELECT count(*), count(DISTINCT (node_type, node_name)) "
+            f"FROM main.{FINGERPRINT_TABLE_NAME}"
+        ),
+    )[0]
+    return int(row[0]), int(row[1])
+
+
+def edited_original_order_models() -> dict[str, str]:
+    """Return the original connected models with a comment edit that rebuilds each one."""
+
+    return {name: f"{sql}-- revised\n" for name, sql in original_order_models().items()}
+
+
+def declared_rename_models() -> dict[str, str]:
+    """Return the original models with the staging model renamed through migrate_from."""
+
+    return {
+        DESTINATION_MODEL: incremental_orders_sql(migrate_from=ORIGIN_MODEL),
+        "orders_enriched": enriched_view_sql(upstream=DESTINATION_MODEL, alias="o"),
+        "daily_order_totals": daily_totals_sql(upstream="orders_enriched", cte="orders_enriched"),
+    }
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_state_reads.py b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_state_reads.py
new file mode 100644
index 0000000000..ce275f1177
--- /dev/null
+++ b/tests/integration/src/sqlbuild/cli/commands/main/model_migrations/test_fingerprint_state_reads.py
@@ -0,0 +1,88 @@
+"""Integration coverage for how often planning reads the fingerprint state table."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.integration.src.sqlbuild.cli.commands.main.model_migrations._test_types import (
+    FingerprintStateReadTestCase,
+)
+from tests.integration.src.sqlbuild.cli.commands.main.model_migrations.helpers import (
+    build_each,
+    declared_rename_models,
+    edited_original_order_models,
+    fingerprint_history,
+    load_raw_orders,
+    original_order_models,
+    plan_json,
+    planned_migrations,
+    record_fingerprint_state_reads,
+    renamed_order_models,
+    write_project,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FingerprintStateReadTestCase(
+            description="full plan discovering renames reads latest state once",
+            prior_models=(original_order_models, edited_original_order_models),
+            models=renamed_order_models,
+            plan_args=(),
+            expected_reads=1,
+            expected_rows_read=lambda identities: identities,
+            expected_migrations=(
+                ("daily_order_totals", "customer_daily_order_totals", "automatic", "migrate"),
+                ("orders_enriched", "customer_orders_enriched", "automatic", "renamed"),
+                ("stg_orders", "stg_customer_orders", "automatic", "migrate"),
+            ),
+        ),
+        FingerprintStateReadTestCase(
+            description="selected plan of a project declaring migrations reads latest state once",
+            prior_models=(original_order_models, edited_original_order_models),
+            models=declared_rename_models,
+            plan_args=("-s", "stg_customer_orders"),
+            expected_reads=1,
+            expected_rows_read=lambda identities: identities,
+            expected_migrations=(("stg_orders", "stg_customer_orders", "manual", "migrate"),),
+        ),
+        FingerprintStateReadTestCase(
+            description="selected plan without migration work reads only the selected state",
+            prior_models=(original_order_models, edited_original_order_models),
+            models=original_order_models,
+            plan_args=("-s", "stg_orders"),
+            expected_reads=1,
+            expected_rows_read=lambda identities: 1,
+            expected_migrations=(),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_fingerprint_history_when_planning_then_reads_latest_state_once(
+    test_case: FingerprintStateReadTestCase,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    load_raw_orders(project_dir=tmp_path, first_day=1, last_day=5)
+    build_each(project_dir=tmp_path, model_sets=test_case.prior_models, capsys=capsys)
+    stored_rows: int
+    identities: int
+    stored_rows, identities = fingerprint_history(project_dir=tmp_path)
+    write_project(project_dir=tmp_path, models=test_case.models())
+    rows_read: list[int] = record_fingerprint_state_reads(monkeypatch)
+
+    migrations: tuple[tuple[str | None, str, str, str], ...] = planned_migrations(
+        plan_json(project_dir=tmp_path, capsys=capsys, args=test_case.plan_args)
+    )
+
+    assert stored_rows > identities
+    assert migrations == test_case.expected_migrations
+    assert rows_read == [test_case.expected_rows_read(identities)] * test_case.expected_reads
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])
diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py
index 4015e53c80..495329ab8c 100644
--- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py
+++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py
@@ -87,3 +87,11 @@ class PruneFingerprintHistoryTestCase:
     fingerprints: tuple[Fingerprint, ...]
     expected_run_ids_by_identity: dict[tuple[str, str], tuple[str, ...]]
     expected_latest_run_id: str
+
+
+@dataclass(frozen=True)
+class SelectLatestTestCase:
+    description: str
+    node_names: tuple[str, ...]
+    filtered_node_types: tuple[str, ...]
+    expected_identities: frozenset[tuple[str, str]]
diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py
index 100c220b4c..1ac1db016c 100644
--- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py
+++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/helpers.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import itertools
 import threading
 from datetime import UTC, datetime
 from pathlib import Path
@@ -83,3 +84,40 @@ def run_concurrent_fingerprint_write_round(
     ).fetchone()
     check_connection.close()
     return int(getattr(row, "__getitem__", lambda _index: 0)(0))
+
+
+_MIXED_IDENTITIES: tuple[tuple[str, str], ...] = (
+    ("model", "orders"),
+    ("model", "customers"),
+    ("seed", "orders"),
+    ("udf", "is_large_order"),
+    ("python", "export_orders"),
+)
+_HISTORY_RUNS: tuple[tuple[str, int], ...] = (("run_001", 10), ("run_002", 12), ("run_003", 12))
+
+
+def _history_fingerprint(*, identity: tuple[str, str], run: tuple[str, int]) -> Fingerprint:
+    node_type, node_name = identity
+    run_id, hour = run
+    return Fingerprint(
+        node_type=node_type,
+        node_name=node_name,
+        target_database=None,
+        target_schema=None,
+        target_name=node_name,
+        run_id=run_id,
+        definition_hash=f"{node_name}_{run_id}",
+        version_hash=f"{node_name}_{run_id}",
+        schema_fingerprint="",
+        definition=f"SELECT '{run_id}'",
+        ts=datetime(2026, 1, 15, hour, 0, 0),
+    )
+
+
+def mixed_fingerprint_history() -> tuple[Fingerprint, ...]:
+    """Return three runs of history for models, a seed, a function, and a Python node."""
+
+    return tuple(
+        _history_fingerprint(identity=identity, run=run)
+        for identity, run in itertools.product(_MIXED_IDENTITIES, _HISTORY_RUNS)
+    )
diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py
index 2259c049d9..27d56ed8d0 100644
--- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py
+++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py
@@ -10,6 +10,7 @@
 from sqlbuild.adapter.contract.types import FrameworkType
 from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter
 from sqlbuild.compiler.fingerprints.constants import FINGERPRINT_TABLE_NAME
+from sqlbuild.compiler.fingerprints.main._select_latest import select_latest_fingerprints
 from sqlbuild.compiler.fingerprints.main.read import read_latest_fingerprints
 from sqlbuild.compiler.fingerprints.main.write import write_fingerprint
 from sqlbuild.compiler.fingerprints.main.write_many import write_fingerprints
@@ -21,9 +22,13 @@
     OldFingerprintSchemaTestCase,
     PruneFingerprintHistoryTestCase,
     ReadNonExistentTableTestCase,
+    SelectLatestTestCase,
     WriteAndReadTestCase,
     WriteCreatesTableTestCase,
 )
+from tests.integration.src.sqlbuild.compiler.fingerprints.main.helpers import (
+    mixed_fingerprint_history,
+)
 
 RENDER_QUALIFIED_NAME: Callable[..., str | None] = DuckDbAdapter().render_qualified_name
 RENDER_FRAMEWORK_TYPE: Callable[[FrameworkType], str] = DuckDbAdapter().render_framework_type
@@ -897,3 +902,76 @@ def test_given_invalid_definition_storage_when_reading_then_raises_contextual_er
     fragment: str
     for fragment in test_case.expected_error_fragments:
         assert fragment in message
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SelectLatestTestCase(
+            description="keeps named nodes of every type and unfiltered node types",
+            node_names=("orders",),
+            filtered_node_types=("model", "udf", "seed"),
+            expected_identities=frozenset(
+                {("model", "orders"), ("seed", "orders"), ("python", "export_orders")}
+            ),
+        ),
+        SelectLatestTestCase(
+            description="keeps only unfiltered node types without names",
+            node_names=(),
+            filtered_node_types=("model", "udf", "seed"),
+            expected_identities=frozenset({("python", "export_orders")}),
+        ),
+        SelectLatestTestCase(
+            description="keeps only named nodes without filtered types",
+            node_names=("customers", "is_large_order"),
+            filtered_node_types=(),
+            expected_identities=frozenset({("model", "customers"), ("udf", "is_large_order")}),
+        ),
+        SelectLatestTestCase(
+            description="keeps nothing without names or filtered types",
+            node_names=(),
+            filtered_node_types=(),
+            expected_identities=frozenset(),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_unfiltered_latest_read_when_selecting_then_matches_filtered_read(
+    test_case: SelectLatestTestCase,
+    connection: Any,
+    execute: Any,
+) -> None:
+    write_fingerprints(
+        connection=connection,
+        execute=execute,
+        database=None,
+        schema="test_schema",
+        fingerprints=mixed_fingerprint_history(),
+        render_qualified_name=RENDER_QUALIFIED_NAME,
+        render_framework_type=RENDER_FRAMEWORK_TYPE,
+    )
+
+    def read(node_names: tuple[str, ...] | None) -> FingerprintSet:
+        return read_latest_fingerprints(
+            connection=connection,
+            execute=execute,
+            table_exists=True,
+            database=None,
+            schema="test_schema",
+            render_qualified_name=RENDER_QUALIFIED_NAME,
+            render_read_latest_sql=RENDER_READ_LATEST_SQL,
+            node_names=node_names,
+            filtered_node_types=test_case.filtered_node_types,
+        )
+
+    filtered: FingerprintSet = read(test_case.node_names)
+    selected: FingerprintSet = select_latest_fingerprints(
+        fingerprint_set=read(None),
+        node_names=test_case.node_names,
+        filtered_node_types=test_case.filtered_node_types,
+    )
+
+    assert selected.fingerprints_by_identity == filtered.fingerprints_by_identity
+    assert selected.fingerprints == filtered.fingerprints
+    assert frozenset(selected.fingerprints_by_identity or {}) == test_case.expected_identities
+    assert {fingerprint.run_id for fingerprint in selected.fingerprints.values()} <= {"run_003"}

From 41cc32daa2d231b49d556392b3a332b3eee9da44 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 17:54:54 +0100
Subject: [PATCH 2/8] perf(observability): cut per-statement monitoring
 overhead

Statement monitors without a query-ID provider now start their thread
only once a statement outlives the heartbeat threshold, through one
shared deferred-start daemon, instead of starting and joining a thread
for every statement. Lifecycle events are validated once at
construction rather than again on publish and on sequencing, JSON
freezing builds error paths only on failure, and statements hash their
SQL once. Recorded events, diagnostics, and -v output are unchanged.
---
 .../observability/_helpers/identity.py        |  7 +-
 .../observability/_helpers/validation.py      | 27 ++++--
 .../classes/deferred_thread_starts.py         | 53 ++++++++++++
 .../observability/classes/event_dispatcher.py |  6 +-
 .../classes/statement_lifecycle.py            | 11 ++-
 .../classes/statement_monitor.py              | 44 +++++++---
 src/sqlbuild/runtime/observability/models.py  | 12 +++
 .../runtime/observability/_test_types.py      | 30 +++++++
 .../sqlbuild/runtime/observability/helpers.py | 64 +++++++++++++-
 .../runtime/observability/test_models.py      | 84 ++++++++++++++++++-
 .../observability/test_statement_monitor.py   | 80 ++++++++++++++++++
 11 files changed, 392 insertions(+), 26 deletions(-)
 create mode 100644 src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py
 create mode 100644 tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py

diff --git a/src/sqlbuild/runtime/observability/_helpers/identity.py b/src/sqlbuild/runtime/observability/_helpers/identity.py
index 84d8b3f36b..599e68933e 100644
--- a/src/sqlbuild/runtime/observability/_helpers/identity.py
+++ b/src/sqlbuild/runtime/observability/_helpers/identity.py
@@ -5,7 +5,7 @@
 from collections.abc import Iterator, Mapping
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
-from dataclasses import asdict, replace
+from dataclasses import fields, replace
 from types import MappingProxyType
 from typing import cast
 from uuid import uuid4
@@ -20,6 +20,9 @@
 )
 
 
+_IDENTITY_FIELD_NAMES: tuple[str, ...] = tuple(field.name for field in fields(ExecutionIdentity))
+
+
 _CURRENT_INVOCATION_EXTERNAL_CONTEXT: ContextVar[Mapping[str, JSONValue]] = ContextVar(
     "sqlbuild_invocation_external_context", default=MappingProxyType({})
 )
@@ -39,7 +42,7 @@ def identity_scope(identity: ExecutionIdentity) -> Iterator[ExecutionIdentity]:
 
 
 def execution_identity_to_dict(identity: ExecutionIdentity) -> dict[str, str | None]:
-    return asdict(identity)
+    return {name: getattr(identity, name) for name in _IDENTITY_FIELD_NAMES}
 
 
 def _required_current_identity() -> ExecutionIdentity:
diff --git a/src/sqlbuild/runtime/observability/_helpers/validation.py b/src/sqlbuild/runtime/observability/_helpers/validation.py
index 6e752841a3..1dac7d85e3 100644
--- a/src/sqlbuild/runtime/observability/_helpers/validation.py
+++ b/src/sqlbuild/runtime/observability/_helpers/validation.py
@@ -65,25 +65,42 @@ def validate_schema_version(*, value: object) -> None:
 def freeze_json(*, value: object, path: str) -> JSONValue:
     """Validate and recursively freeze a JSON-compatible value."""
 
+    return _freeze_json(value=value, path=(path,))
+
+
+def _freeze_json(*, value: object, path: tuple[str | int, ...]) -> JSONValue:
     if value is None or isinstance(value, (str, bool, int)):
         return value
     if isinstance(value, float):
         if not math.isfinite(value):
-            raise ObservabilityValidationError(f"{path} must not contain NaN or infinity")
+            raise ObservabilityValidationError(
+                f"{_render_path(path)} must not contain NaN or infinity"
+            )
         return value
     if isinstance(value, Mapping):
         frozen: dict[str, JSONValue] = {}
         for key, item in value.items():
             if not isinstance(key, str):
-                raise ObservabilityValidationError(f"{path} keys must be strings")
-            frozen[key] = freeze_json(value=item, path=f"{path}.{key}")
+                raise ObservabilityValidationError(f"{_render_path(path)} keys must be strings")
+            frozen[key] = (
+                item
+                if item is None or isinstance(item, (str, bool, int))
+                else _freeze_json(value=item, path=(*path, key))
+            )
         return MappingProxyType(frozen)
     if isinstance(value, (list, tuple)):
         return tuple(
-            freeze_json(value=item, path=f"{path}[{index}]") for index, item in enumerate(value)
+            _freeze_json(value=item, path=(*path, index)) for index, item in enumerate(value)
         )
     raise ObservabilityValidationError(
-        f"{path} contains non-JSON value of type {type(value).__name__}"
+        f"{_render_path(path)} contains non-JSON value of type {type(value).__name__}"
+    )
+
+
+def _render_path(path: tuple[str | int, ...]) -> str:
+    return "".join(
+        f"[{segment}]" if isinstance(segment, int) else f".{segment}" if index else segment
+        for index, segment in enumerate(path)
     )
 
 
diff --git a/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py b/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py
new file mode 100644
index 0000000000..1e9302ee62
--- /dev/null
+++ b/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py
@@ -0,0 +1,53 @@
+"""One shared daemon that runs deferred thread starts once their deadline passes."""
+
+from __future__ import annotations
+
+import heapq
+import itertools
+import threading
+import time
+from collections.abc import Callable
+
+type _DeferredStart = tuple[float, int, Callable[[], None]]
+
+
+class DeferredThreadStarts:
+    """Run each scheduled start callback at its monotonic deadline on one shared thread."""
+
+    def __init__(self, *, name: str) -> None:
+        self._name: str = name
+        self._condition: threading.Condition = threading.Condition()
+        self._pending: list[_DeferredStart] = []
+        self._order: itertools.count[int] = itertools.count()
+        self._thread: threading.Thread | None = None
+
+    def schedule(self, *, delay_seconds: float, start: Callable[[], None]) -> None:
+        """Call ``start`` after ``delay_seconds``; the callback must tolerate a stopped owner."""
+
+        entry: _DeferredStart = (time.monotonic() + delay_seconds, next(self._order), start)
+        with self._condition:
+            heapq.heappush(self._pending, entry)
+            if self._thread is None:
+                self._thread = threading.Thread(target=self._run, name=self._name, daemon=True)
+                self._thread.start()
+            elif self._pending[0] is entry:
+                self._condition.notify()
+
+    def _run(self) -> None:
+        while True:
+            start: Callable[[], None] = self._next_due()
+            try:
+                start()
+            except BaseException:
+                pass
+
+    def _next_due(self) -> Callable[[], None]:
+        with self._condition:
+            while True:
+                if not self._pending:
+                    _ = self._condition.wait()
+                    continue
+                remaining: float = self._pending[0][0] - time.monotonic()
+                if remaining <= 0:
+                    return heapq.heappop(self._pending)[2]
+                _ = self._condition.wait(remaining)
diff --git a/src/sqlbuild/runtime/observability/classes/event_dispatcher.py b/src/sqlbuild/runtime/observability/classes/event_dispatcher.py
index b1742c8446..4bffd0c4ff 100644
--- a/src/sqlbuild/runtime/observability/classes/event_dispatcher.py
+++ b/src/sqlbuild/runtime/observability/classes/event_dispatcher.py
@@ -4,7 +4,6 @@
 
 from collections import deque
 from contextvars import ContextVar, Token
-from dataclasses import replace
 from functools import partial
 from threading import RLock
 from typing import Literal, cast, overload
@@ -14,7 +13,6 @@
     _safe_error_type,
     _safe_subscriber_name,
 )
-from sqlbuild.runtime.observability._helpers.validation import validate_known_lifecycle_event
 from sqlbuild.runtime.observability.constants import (
     LIFECYCLE_INVOCATION_METADATA_SCHEMA_VERSION,
 )
@@ -98,8 +96,6 @@ def publish_lifecycle(self, event: LifecycleEvent | OpaqueLifecycleEvent) -> Non
             raise ObservabilityValidationError(
                 "lifecycle publication requires LifecycleEvent or OpaqueLifecycleEvent"
             )
-        if isinstance(event, LifecycleEvent):
-            validate_known_lifecycle_event(event=event)
         self._publish(channel=_LIFECYCLE_CHANNEL, value=event)
 
     def publish_diagnostic(self, log: DiagnosticLog) -> None:
@@ -130,7 +126,7 @@ def _publish(
                 ):
                     sequence: int = self._lifecycle_sequences.get(lifecycle_value.invocation_id, 0)
                     self._lifecycle_sequences[lifecycle_value.invocation_id] = sequence + 1
-                    lifecycle_value = replace(lifecycle_value, invocation_sequence=sequence)
+                    lifecycle_value = lifecycle_value.with_invocation_sequence(sequence)
                 self._pending_publications.append((channel, lifecycle_value, suppress_health))
             else:
                 diagnostic_value: DiagnosticLog = cast(DiagnosticLog, value)
diff --git a/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py b/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py
index b4e034e447..10dc1dbb38 100644
--- a/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py
+++ b/src/sqlbuild/runtime/observability/classes/statement_lifecycle.py
@@ -81,6 +81,7 @@ def __init__(
             None
         )
         self._pending_failure: tuple[BaseException, str | None, str | None] | None = None
+        self._statement_facts: tuple[str, str] | None = None
 
     @staticmethod
     @contextmanager
@@ -268,11 +269,17 @@ def __exit__(
             notify_statement_finished(sql=self._sql)
 
     def _base_payload(self) -> dict[str, JSONValue]:
+        if self._statement_facts is None:
+            self._statement_facts = (
+                hashlib.sha256(self._sql.encode("utf-8")).hexdigest(),
+                _statement_kind(sql=self._sql),
+            )
+        sql_digest, statement_kind = self._statement_facts
         payload: dict[str, JSONValue] = {
             "adapter": self._adapter,
             "intent": self._intent,
-            "sql_digest": hashlib.sha256(self._sql.encode("utf-8")).hexdigest(),
-            "statement_kind": _statement_kind(sql=self._sql),
+            "sql_digest": sql_digest,
+            "statement_kind": statement_kind,
         }
         if self._batch_size is not None:
             payload["batch_size"] = self._batch_size
diff --git a/src/sqlbuild/runtime/observability/classes/statement_monitor.py b/src/sqlbuild/runtime/observability/classes/statement_monitor.py
index 438024d213..95fcadb5f6 100644
--- a/src/sqlbuild/runtime/observability/classes/statement_monitor.py
+++ b/src/sqlbuild/runtime/observability/classes/statement_monitor.py
@@ -7,9 +7,13 @@
 from collections.abc import Callable
 from contextvars import Context, copy_context
 
+from sqlbuild.runtime.observability.classes.deferred_thread_starts import DeferredThreadStarts
 from sqlbuild.runtime.observability.constants import STATEMENT_HEARTBEAT_THRESHOLD_SECONDS
 
 _STATEMENT_QUERY_ID_POLL_SECONDS: float = 0.1
+_DEFERRED_MONITOR_STARTS: DeferredThreadStarts = DeferredThreadStarts(
+    name="sqlbuild-statement-monitor-starter"
+)
 
 
 class StatementMonitor:
@@ -38,6 +42,9 @@ def __init__(
         self._query_id_provider: Callable[[], str | None] | None = None
         self._query_id: str | None = None
         self._thread: threading.Thread | None = None
+        self._thread_lock: threading.Lock = threading.Lock()
+        self._context: Context | None = None
+        self._stopped: bool = False
 
     @property
     def query_id(self) -> str | None:
@@ -46,22 +53,23 @@ def query_id(self) -> str | None:
         return self._query_id
 
     def start(self) -> None:
-        """Start monitoring in a daemon thread with the current execution context."""
-
-        context: Context = copy_context()
-        self._thread = threading.Thread(
-            target=context.run,
-            args=(self._run,),
-            name="sqlbuild-statement-monitor",
-            daemon=True,
+        """Monitor in a daemon thread, deferred to the first heartbeat without a query-ID poll."""
+
+        self._context = copy_context()
+        if self._has_query_id_provider() or self._threshold_seconds <= 0:
+            self._start_thread()
+            return
+        _DEFERRED_MONITOR_STARTS.schedule(
+            delay_seconds=self._threshold_seconds, start=self._start_thread
         )
-        self._thread.start()
 
     def set_query_id_provider(self, provider: Callable[[], str | None]) -> None:
         """Install a non-blocking adapter query-ID reader."""
 
         with self._provider_lock:
             self._query_id_provider = provider
+        if self._context is not None:
+            self._start_thread()
         self._wake_event.set()
 
     def stop(self) -> str | None:
@@ -70,11 +78,27 @@ def stop(self) -> str | None:
         self._capture_query_id()
         self._stop_event.set()
         self._wake_event.set()
-        thread: threading.Thread | None = self._thread
+        with self._thread_lock:
+            self._stopped = True
+            self._context = None
+            thread: threading.Thread | None = self._thread
         if thread is not None and thread is not threading.current_thread():
             thread.join()
         return self._query_id
 
+    def _start_thread(self) -> None:
+        with self._thread_lock:
+            context: Context | None = self._context
+            if self._stopped or self._thread is not None or context is None:
+                return
+            self._thread = threading.Thread(
+                target=context.run,
+                args=(self._run,),
+                name="sqlbuild-statement-monitor",
+                daemon=True,
+            )
+            self._thread.start()
+
     def _run(self) -> None:
         next_heartbeat: float = self._started_at + self._threshold_seconds
         while not self._stop_event.is_set():
diff --git a/src/sqlbuild/runtime/observability/models.py b/src/sqlbuild/runtime/observability/models.py
index 0e937361d4..1efbb621d0 100644
--- a/src/sqlbuild/runtime/observability/models.py
+++ b/src/sqlbuild/runtime/observability/models.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import copy
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -190,6 +191,17 @@ def __post_init__(self) -> None:
         object.__setattr__(self, "payload", frozen_payload)
         validate_known_lifecycle_event(event=self)
 
+    def with_invocation_sequence(self, sequence: int) -> LifecycleEvent:
+        """Return this validated event with a dispatcher-assigned invocation sequence."""
+
+        if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 0:
+            raise ObservabilityValidationError(
+                "invocation_sequence must be a non-negative integer excluding bool"
+            )
+        sequenced: LifecycleEvent = copy.copy(self)
+        object.__setattr__(sequenced, "invocation_sequence", sequence)
+        return sequenced
+
 
 @dataclass(frozen=True)
 class OpaqueLifecycleEvent:
diff --git a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py
index d272b95ae8..b7d0563611 100644
--- a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py
+++ b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py
@@ -321,3 +321,33 @@ class ErrorCodePrivacyCase:
     code: object
     expected_error_code: str | None
     private_fragments: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class FastStatementMonitorCase:
+    description: str
+    statement_count: int
+    expected_monitor_threads: int
+
+
+@dataclass(frozen=True)
+class DeferredMonitorCase:
+    description: str
+    threshold_seconds: float
+    query_id: str
+    expected_submissions: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class FreezeJsonErrorCase:
+    description: str
+    payload: object
+    expected_error: str
+
+
+@dataclass(frozen=True)
+class InvocationSequenceCase:
+    description: str
+    sequence: object
+    expected_error: str = ""
+    expected_sequence: int | None = None
diff --git a/tests/unit/src/sqlbuild/runtime/observability/helpers.py b/tests/unit/src/sqlbuild/runtime/observability/helpers.py
index 0a9495eed4..c422392c28 100644
--- a/tests/unit/src/sqlbuild/runtime/observability/helpers.py
+++ b/tests/unit/src/sqlbuild/runtime/observability/helpers.py
@@ -1,19 +1,24 @@
 """Test builders for runtime observability contracts."""
 
 import asyncio
-from collections.abc import Mapping
+import threading
+import time
+from collections.abc import Callable, Mapping
 from concurrent.futures import ThreadPoolExecutor
 from contextvars import Context, copy_context
 from datetime import UTC, datetime
 from threading import Lock
 from types import MappingProxyType
 
+import pytest
+
 from sqlbuild.runtime.observability._helpers.dispatcher import dispatcher_scope
 from sqlbuild.runtime.observability._helpers.factory import create_lifecycle_event
 from sqlbuild.runtime.observability._helpers.identity import invocation_scope
 from sqlbuild.runtime.observability.classes.event_dispatcher import EventDispatcher
 from sqlbuild.runtime.observability.classes.microbatch_lifecycle import MicrobatchLifecycle
 from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle
+from sqlbuild.runtime.observability.classes.statement_monitor import StatementMonitor
 from sqlbuild.runtime.observability.models import (
     DiagnosticLog,
     LifecycleEvent,
@@ -208,3 +213,60 @@ def diagnostic_log() -> DiagnosticLog:
         message="diagnostic",
         invocation_id="inv-1",
     )
+
+
+def count_monitor_thread_starts(*, monkeypatch: pytest.MonkeyPatch, statement_count: int) -> int:
+    """Run fast statements without query-ID providers and count monitor threads started."""
+
+    started: list[str] = []
+    original_start: Callable[[threading.Thread], None] = threading.Thread.start
+
+    def recording_start(thread: threading.Thread) -> None:
+        started.append(thread.name)
+        original_start(thread)
+
+    monkeypatch.setattr(threading.Thread, "start", recording_start)
+    dispatcher: EventDispatcher
+    dispatcher, _ = capture_lifecycle_events()
+    with invocation_scope("inv-fast-statements"), dispatcher_scope(dispatcher):
+        for index in range(statement_count):
+            with StatementLifecycle(adapter="duckdb", sql=f"SELECT {index}", intent="execute"):
+                pass
+    return started.count("sqlbuild-statement-monitor")
+
+
+def submissions_after_late_provider(*, threshold_seconds: float, query_id: str) -> tuple[str, ...]:
+    """Install a query-ID provider after a deferred start and return published submissions."""
+
+    submitted: threading.Event = threading.Event()
+    submissions: list[str] = []
+
+    def record_submission(value: str) -> None:
+        submissions.append(value)
+        submitted.set()
+
+    monitor: StatementMonitor = StatementMonitor(
+        on_submitted=record_submission,
+        on_heartbeat=lambda elapsed, current_query_id: None,
+        threshold_seconds=threshold_seconds,
+    )
+    monitor.start()
+    monitor.set_query_id_provider(lambda: query_id)
+    _ = submitted.wait(timeout=5.0)
+    _ = monitor.stop()
+    return tuple(submissions)
+
+
+def heartbeats_after_early_stop(*, threshold_seconds: float) -> int:
+    """Stop a deferred monitor before its threshold and count heartbeats after the deadline."""
+
+    heartbeats: list[float] = []
+    monitor: StatementMonitor = StatementMonitor(
+        on_submitted=lambda value: None,
+        on_heartbeat=lambda elapsed, current_query_id: heartbeats.append(elapsed),
+        threshold_seconds=threshold_seconds,
+    )
+    monitor.start()
+    _ = monitor.stop()
+    time.sleep(threshold_seconds * 4)
+    return len(heartbeats)
diff --git a/tests/unit/src/sqlbuild/runtime/observability/test_models.py b/tests/unit/src/sqlbuild/runtime/observability/test_models.py
index b197e7cb8c..25e67c29bf 100644
--- a/tests/unit/src/sqlbuild/runtime/observability/test_models.py
+++ b/tests/unit/src/sqlbuild/runtime/observability/test_models.py
@@ -1,13 +1,16 @@
 """Tests for immutable observability envelope models."""
 
-from dataclasses import FrozenInstanceError
+from dataclasses import FrozenInstanceError, replace
 
 import pytest
 
+from sqlbuild.runtime.observability._helpers.validation import freeze_json
 from sqlbuild.runtime.observability.exceptions import ObservabilityValidationError
 from sqlbuild.runtime.observability.models import DiagnosticLog, LifecycleEvent
 from tests.unit.src.sqlbuild.runtime.observability._test_types import (
+    FreezeJsonErrorCase,
     ImmutabilityCase,
+    InvocationSequenceCase,
     SchemaVersionCase,
 )
 from tests.unit.src.sqlbuild.runtime.observability.helpers import OCCURRED_AT, lifecycle_event
@@ -143,3 +146,82 @@ def test_given_malformed_schema_version_when_constructing_diagnostic_then_it_is_
             source="runtime",
             message="working",
         )
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FreezeJsonErrorCase(
+            description="nested non-finite number names its full path",
+            payload={"metadata": {"rows": [1, float("nan")]}},
+            expected_error="payload.metadata.rows[1] must not contain NaN or infinity",
+        ),
+        FreezeJsonErrorCase(
+            description="nested non-string key names its parent path",
+            payload={"metadata": [{"ok": True}, {1: "orders"}]},
+            expected_error="payload.metadata[1] keys must be strings",
+        ),
+        FreezeJsonErrorCase(
+            description="nested non-JSON value names its full path",
+            payload={"metadata": {"owner": object()}},
+            expected_error="payload.metadata.owner contains non-JSON value of type object",
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_invalid_nested_payload_when_freezing_then_error_names_full_path(
+    test_case: FreezeJsonErrorCase,
+) -> None:
+    with pytest.raises(ObservabilityValidationError) as error_info:
+        freeze_json(value=test_case.payload, path="payload")
+
+    assert str(error_info.value) == test_case.expected_error
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        InvocationSequenceCase(
+            description="negative sequence is rejected",
+            sequence=-1,
+            expected_error="invocation_sequence must be a non-negative integer excluding bool",
+        ),
+        InvocationSequenceCase(
+            description="boolean sequence is rejected",
+            sequence=True,
+            expected_error="invocation_sequence must be a non-negative integer excluding bool",
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_invalid_sequence_when_sequencing_event_then_it_is_rejected(
+    test_case: InvocationSequenceCase,
+) -> None:
+    event: LifecycleEvent = lifecycle_event(payload={"command": "plan"})
+
+    with pytest.raises(ObservabilityValidationError) as error_info:
+        event.with_invocation_sequence(test_case.sequence)  # ty: ignore[invalid-argument-type]
+
+    assert str(error_info.value) == test_case.expected_error
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        InvocationSequenceCase(
+            description="valid sequence replaces only the sequence",
+            sequence=7,
+            expected_sequence=7,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_event_when_sequencing_then_only_sequence_changes(
+    test_case: InvocationSequenceCase,
+) -> None:
+    event: LifecycleEvent = lifecycle_event(payload={"command": "plan"})
+
+    sequenced: LifecycleEvent = event.with_invocation_sequence(test_case.sequence)  # ty: ignore[invalid-argument-type]
+
+    assert sequenced.invocation_sequence == test_case.expected_sequence
+    assert replace(sequenced, invocation_sequence=event.invocation_sequence) == event
diff --git a/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py b/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py
new file mode 100644
index 0000000000..35275e62de
--- /dev/null
+++ b/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py
@@ -0,0 +1,80 @@
+"""Tests for statement monitoring cost and deferred monitor threads."""
+
+import pytest
+
+from tests.unit.src.sqlbuild.runtime.observability._test_types import (
+    DeferredMonitorCase,
+    FastStatementMonitorCase,
+)
+from tests.unit.src.sqlbuild.runtime.observability.helpers import (
+    count_monitor_thread_starts,
+    heartbeats_after_early_stop,
+    submissions_after_late_provider,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FastStatementMonitorCase(
+            description="fast statements without query IDs start no monitor thread",
+            statement_count=200,
+            expected_monitor_threads=0,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_fast_statements_without_query_ids_when_executing_then_no_monitor_thread_starts(
+    test_case: FastStatementMonitorCase, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    started: int = count_monitor_thread_starts(
+        monkeypatch=monkeypatch, statement_count=test_case.statement_count
+    )
+
+    assert started == test_case.expected_monitor_threads
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        DeferredMonitorCase(
+            description="provider installed after a deferred start is polled at once",
+            threshold_seconds=30.0,
+            query_id="01-query-orders",
+            expected_submissions=("01-query-orders",),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_deferred_monitor_when_provider_arrives_then_query_id_is_submitted_once(
+    test_case: DeferredMonitorCase,
+) -> None:
+    submissions: tuple[str, ...] = submissions_after_late_provider(
+        threshold_seconds=test_case.threshold_seconds, query_id=test_case.query_id
+    )
+
+    assert submissions == test_case.expected_submissions
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        DeferredMonitorCase(
+            description="monitor stopped before its threshold never heartbeats",
+            threshold_seconds=0.02,
+            query_id="",
+            expected_submissions=(),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_monitor_stopped_before_threshold_when_deadline_passes_then_no_heartbeat(
+    test_case: DeferredMonitorCase,
+) -> None:
+    heartbeats: int = heartbeats_after_early_stop(threshold_seconds=test_case.threshold_seconds)
+
+    assert heartbeats == len(test_case.expected_submissions)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])

From c11be86e1925434dc090a4a2f7e7c29de96669fd Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 18:49:45 +0100
Subject: [PATCH 3/8] perf(planner): resolve graph identity hashes in linear
 time

Expected and write identity resolution copied the whole hash cache for
every resolved node, which made identity context building and change
reconciliation quadratic in project size. A memoized resolver now
records each node once with the same resolution order, cycle handling,
and results.
---
 .../planner/_helpers/identity/graph.py        | 86 +++----------------
 .../classes/graph_identity_resolver.py        | 85 ++++++++++++++++++
 .../compiler/planner/_helpers/_test_types.py  |  1 +
 .../compiler/planner/_helpers/helpers.py      | 24 ++++++
 .../planner/_helpers/test_graph_identity.py   | 67 +++++++++++++++
 5 files changed, 191 insertions(+), 72 deletions(-)
 create mode 100644 src/sqlbuild/compiler/planner/classes/graph_identity_resolver.py

diff --git a/src/sqlbuild/compiler/planner/_helpers/identity/graph.py b/src/sqlbuild/compiler/planner/_helpers/identity/graph.py
index 1b7c8b26dd..be520d01e0 100644
--- a/src/sqlbuild/compiler/planner/_helpers/identity/graph.py
+++ b/src/sqlbuild/compiler/planner/_helpers/identity/graph.py
@@ -4,6 +4,7 @@
 
 from collections.abc import Mapping
 
+from sqlbuild.compiler.planner.classes.graph_identity_resolver import GraphIdentityResolver
 from sqlbuild.compiler.planner.models import GraphIdentityNode, GraphNodeKey
 from sqlbuild.compiler.planner.types import GraphIdentityComposer
 
@@ -14,37 +15,13 @@ def build_expected_graph_identity_hashes(
     execution_order: tuple[GraphNodeKey, ...],
     compose_identity: GraphIdentityComposer,
 ) -> dict[GraphNodeKey, str | None]:
-    hashes: dict[GraphNodeKey, str | None] = {}
-
-    def resolve(
-        *,
-        key: GraphNodeKey,
-        visiting: frozenset[GraphNodeKey],
-        cache: dict[GraphNodeKey, str | None],
-    ) -> tuple[str | None, dict[GraphNodeKey, str | None]]:
-        if key in cache:
-            return cache[key], cache
-        node: GraphIdentityNode | None = nodes.get(key)
-        if node is None or node.local_hash is None:
-            return None, {**cache, key: None}
-        if key in visiting:
-            return node.local_hash, {**cache, key: node.local_hash}
-        upstream_hashes: list[tuple[GraphNodeKey, str]] = []
-        upstream_key: GraphNodeKey
-        for upstream_key in node.upstream_keys:
-            upstream_hash, cache = resolve(key=upstream_key, visiting=visiting | {key}, cache=cache)
-            if upstream_hash is not None:
-                upstream_hashes.append((upstream_key, upstream_hash))
-        composed: str = compose_identity(
-            local_hash=node.local_hash,
-            upstream_hashes=tuple(upstream_hashes),
-        )
-        return composed, {**cache, key: composed}
-
+    resolver: GraphIdentityResolver = GraphIdentityResolver(
+        nodes=nodes, compose_identity=compose_identity, base_hashes={}, selected_keys=None
+    )
     key: GraphNodeKey
     for key in execution_order:
-        _, hashes = resolve(key=key, visiting=frozenset(), cache=hashes)
-    return hashes
+        _ = resolver.resolve_expected(key)
+    return resolver.hashes
 
 
 def build_graph_write_identity_hashes(
@@ -55,48 +32,13 @@ def build_graph_write_identity_hashes(
     base_identity_hashes: Mapping[GraphNodeKey, str],
     compose_identity: GraphIdentityComposer,
 ) -> dict[GraphNodeKey, str]:
-    hashes: dict[GraphNodeKey, str] = dict(base_identity_hashes)
-    resolved_selected: dict[GraphNodeKey, str] = {}
-
-    def resolve(
-        *,
-        key: GraphNodeKey,
-        visiting: frozenset[GraphNodeKey],
-        cache: dict[GraphNodeKey, str],
-        selected_cache: dict[GraphNodeKey, str],
-    ) -> tuple[str | None, dict[GraphNodeKey, str], dict[GraphNodeKey, str]]:
-        if key not in selected_keys:
-            return cache.get(key), cache, selected_cache
-        if key in selected_cache:
-            return selected_cache[key], cache, selected_cache
-        node: GraphIdentityNode | None = nodes.get(key)
-        if node is None or node.local_hash is None:
-            return cache.get(key), cache, selected_cache
-        if key in visiting:
-            return node.local_hash, cache, selected_cache
-        upstream_hashes: list[tuple[GraphNodeKey, str]] = []
-        upstream_key: GraphNodeKey
-        for upstream_key in node.upstream_keys:
-            upstream_hash, cache, selected_cache = resolve(
-                key=upstream_key,
-                visiting=visiting | {key},
-                cache=cache,
-                selected_cache=selected_cache,
-            )
-            if upstream_hash is not None:
-                upstream_hashes.append((upstream_key, upstream_hash))
-        composed: str = compose_identity(
-            local_hash=node.local_hash,
-            upstream_hashes=tuple(upstream_hashes),
-        )
-        return composed, {**cache, key: composed}, {**selected_cache, key: composed}
-
+    resolver: GraphIdentityResolver = GraphIdentityResolver(
+        nodes=nodes,
+        compose_identity=compose_identity,
+        base_hashes=base_identity_hashes,
+        selected_keys=selected_keys,
+    )
     key: GraphNodeKey
     for key in execution_order:
-        _, hashes, resolved_selected = resolve(
-            key=key,
-            visiting=frozenset(),
-            cache=hashes,
-            selected_cache=resolved_selected,
-        )
-    return hashes
+        _ = resolver.resolve_write(key)
+    return {hashed_key: value for hashed_key, value in resolver.hashes.items() if value is not None}
diff --git a/src/sqlbuild/compiler/planner/classes/graph_identity_resolver.py b/src/sqlbuild/compiler/planner/classes/graph_identity_resolver.py
new file mode 100644
index 0000000000..7d2f6e1fcf
--- /dev/null
+++ b/src/sqlbuild/compiler/planner/classes/graph_identity_resolver.py
@@ -0,0 +1,85 @@
+"""Memoized graph identity resolution over one dependency graph."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+
+from sqlbuild.compiler.planner.models import GraphIdentityNode, GraphNodeKey
+from sqlbuild.compiler.planner.types import GraphIdentityComposer
+
+type _Resolve = Callable[[GraphNodeKey], str | None]
+
+
+class GraphIdentityResolver:
+    """Resolve expected or selected-write identity hashes, each node once, in linear time."""
+
+    def __init__(
+        self,
+        *,
+        nodes: Mapping[GraphNodeKey, GraphIdentityNode],
+        compose_identity: GraphIdentityComposer,
+        base_hashes: Mapping[GraphNodeKey, str | None],
+        selected_keys: frozenset[GraphNodeKey] | None,
+    ) -> None:
+        self._nodes: Mapping[GraphNodeKey, GraphIdentityNode] = nodes
+        self._compose_identity: GraphIdentityComposer = compose_identity
+        self._selected_keys: frozenset[GraphNodeKey] | None = selected_keys
+        self._hashes: dict[GraphNodeKey, str | None] = dict(base_hashes)
+        self._resolved_selected: dict[GraphNodeKey, str] = {}
+        self._visiting: set[GraphNodeKey] = set()
+
+    @property
+    def hashes(self) -> dict[GraphNodeKey, str | None]:
+        """Return every hash recorded so far, in first-recorded order."""
+
+        return self._hashes
+
+    def resolve_expected(self, key: GraphNodeKey) -> str | None:
+        """Compose a node's expected hash from its upstream expected hashes."""
+
+        if key in self._hashes:
+            return self._hashes[key]
+        node: GraphIdentityNode | None = self._nodes.get(key)
+        if node is None or node.local_hash is None:
+            self._hashes[key] = None
+            return None
+        if key in self._visiting:
+            self._hashes[key] = node.local_hash
+            return node.local_hash
+        composed: str = self._compose(
+            key=key, local_hash=node.local_hash, node=node, resolve=self.resolve_expected
+        )
+        self._hashes[key] = composed
+        return composed
+
+    def resolve_write(self, key: GraphNodeKey) -> str | None:
+        """Recompute a selected node's hash from upstream write hashes; others keep theirs."""
+
+        if self._selected_keys is None or key not in self._selected_keys:
+            return self._hashes.get(key)
+        if key in self._resolved_selected:
+            return self._resolved_selected[key]
+        node: GraphIdentityNode | None = self._nodes.get(key)
+        if node is None or node.local_hash is None:
+            return self._hashes.get(key)
+        if key in self._visiting:
+            return node.local_hash
+        composed: str = self._compose(
+            key=key, local_hash=node.local_hash, node=node, resolve=self.resolve_write
+        )
+        self._hashes[key] = composed
+        self._resolved_selected[key] = composed
+        return composed
+
+    def _compose(
+        self, *, key: GraphNodeKey, local_hash: str, node: GraphIdentityNode, resolve: _Resolve
+    ) -> str:
+        self._visiting.add(key)
+        upstream_hashes: list[tuple[GraphNodeKey, str]] = []
+        upstream_key: GraphNodeKey
+        for upstream_key in node.upstream_keys:
+            upstream_hash: str | None = resolve(upstream_key)
+            if upstream_hash is not None:
+                upstream_hashes.append((upstream_key, upstream_hash))
+        self._visiting.discard(key)
+        return self._compose_identity(local_hash=local_hash, upstream_hashes=tuple(upstream_hashes))
diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py
index 4428fb9bb9..61f81e3fba 100644
--- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py
+++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/_test_types.py
@@ -239,6 +239,7 @@ class GraphIdentityWritePerfTestCase:
     description: str
     layer_count: int
     expected_max_seconds: float
+    layer_width: int = 2
 
 
 @dataclass(frozen=True)
diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py
index 9bbd8e628f..78881f8001 100644
--- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py
+++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/helpers.py
@@ -193,6 +193,30 @@ def node_key(name: str) -> GraphNodeKey:
     return nodes, tuple(order)
 
 
+def build_wide_layered_identity_nodes(
+    *, layer_count: int, layer_width: int
+) -> tuple[dict[GraphNodeKey, GraphIdentityNode], tuple[GraphNodeKey, ...]]:
+    """Build a wide project graph where each node reads two nodes of the prior layer."""
+
+    nodes: dict[GraphNodeKey, GraphIdentityNode] = {}
+    previous: tuple[GraphNodeKey, ...] = ()
+    layer: int
+    for layer in range(layer_count):
+        current: list[GraphNodeKey] = []
+        column: int
+        for column in range(layer_width):
+            key: GraphNodeKey = GraphNodeKey(node_type="model", node_name=f"L{layer}_{column}")
+            nodes[key] = GraphIdentityNode(
+                key=key,
+                resource_kind=GraphResourceKind.MODEL,
+                upstream_keys=previous[column : column + 2],
+                local_hash=f"local_{key.node_name}",
+            )
+            current.append(key)
+        previous = tuple(current)
+    return nodes, tuple(nodes)
+
+
 def _append_relation_info(listed: list[RelationInfo], relation: RelationInfo) -> None:
     listed.append(relation)
 
diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py
index 0dd6420da2..e1d04af0ef 100644
--- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py
+++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/test_graph_identity.py
@@ -19,6 +19,7 @@
 )
 from tests.unit.src.sqlbuild.compiler.planner._helpers.helpers import (
     build_diamond_ladder_identity_nodes,
+    build_wide_layered_identity_nodes,
     compose_hashed_identity,
     compose_readable_identity,
 )
@@ -190,3 +191,69 @@ def test_given_fully_selected_dense_graph_when_building_write_hashes_then_stays_
 
     assert len(result) == len(nodes)
     assert elapsed < test_case.expected_max_seconds
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        GraphIdentityWritePerfTestCase(
+            description="twenty thousand node project graph resolves in linear time",
+            layer_count=40,
+            layer_width=500,
+            expected_max_seconds=1.5,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_wide_project_graph_when_building_expected_hashes_then_stays_linear(
+    test_case: GraphIdentityWritePerfTestCase,
+) -> None:
+    nodes: dict[GraphNodeKey, GraphIdentityNode]
+    execution_order: tuple[GraphNodeKey, ...]
+    nodes, execution_order = build_wide_layered_identity_nodes(
+        layer_count=test_case.layer_count, layer_width=test_case.layer_width
+    )
+
+    start: float = time.monotonic()
+    result: dict[GraphNodeKey, str | None] = build_expected_graph_identity_hashes(
+        nodes=nodes, execution_order=execution_order, compose_identity=compose_hashed_identity
+    )
+    elapsed: float = time.monotonic() - start
+
+    assert len(result) == len(nodes)
+    assert elapsed < test_case.expected_max_seconds
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        GraphIdentityWritePerfTestCase(
+            description="twenty thousand node project graph resolves in linear time",
+            layer_count=40,
+            layer_width=500,
+            expected_max_seconds=1.5,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_wide_selected_project_graph_when_building_write_hashes_then_stays_linear(
+    test_case: GraphIdentityWritePerfTestCase,
+) -> None:
+    nodes: dict[GraphNodeKey, GraphIdentityNode]
+    execution_order: tuple[GraphNodeKey, ...]
+    nodes, execution_order = build_wide_layered_identity_nodes(
+        layer_count=test_case.layer_count, layer_width=test_case.layer_width
+    )
+
+    start: float = time.monotonic()
+    result: dict[GraphNodeKey, str] = build_graph_write_identity_hashes(
+        nodes=nodes,
+        execution_order=execution_order,
+        selected_keys=frozenset(nodes.keys()),
+        base_identity_hashes={},
+        compose_identity=compose_hashed_identity,
+    )
+    elapsed: float = time.monotonic() - start
+
+    assert len(result) == len(nodes)
+    assert elapsed < test_case.expected_max_seconds

From 512d08b81db35c574516471f2514c901d227fd25 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 19:07:13 +0100
Subject: [PATCH 4/8] perf(planner): overlay per-test fixture types instead of
 copying them

SQL-test fixture completion copied the project-wide expected-type map
for every test, which made planning test-heavy projects quadratic. Each
test now overlays only the models of its chain on the shared map.
---
 .../compiler/planner/_helpers/fixtures/completion.py  | 11 ++++++-----
 src/sqlbuild/compiler/planner/models.py               |  4 ++--
 2 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py b/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py
index d95c9230d8..bdd11253a8 100644
--- a/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py
+++ b/src/sqlbuild/compiler/planner/_helpers/fixtures/completion.py
@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import re
-from collections.abc import Sequence
+from collections import ChainMap
+from collections.abc import Mapping, Sequence
 from functools import lru_cache
 from typing import Any
 
@@ -91,7 +92,7 @@ def build_relation_fixture_completion(
         authoritative_columns=context.authoritative_columns,
         adapter=adapter,
     )
-    expected_types: dict[FixtureKey, dict[str, str]] = _expected_fixture_types(
+    expected_types: Mapping[FixtureKey, dict[str, str]] = _expected_fixture_types(
         base_expected_types=context.expected_types,
         ordered_model_names=ordered_model_names,
         model_map=model_map,
@@ -327,7 +328,7 @@ def complete_empty_fixture_sql(
 def _typed_null_fixture_types(
     *,
     null_literal_names_by_fixture: dict[FixtureKey, frozenset[str]],
-    expected_types: dict[FixtureKey, dict[str, str]],
+    expected_types: Mapping[FixtureKey, dict[str, str]],
 ) -> dict[FixtureKey, dict[str, str]]:
     result: dict[FixtureKey, dict[str, str]] = {}
     for key, null_literal_names in null_literal_names_by_fixture.items():
@@ -665,8 +666,8 @@ def _expected_fixture_types(
     base_expected_types: dict[FixtureKey, dict[str, str]],
     ordered_model_names: tuple[str, ...],
     model_map: dict[str, CompiledModel],
-) -> dict[FixtureKey, dict[str, str]]:
-    expected_types: dict[FixtureKey, dict[str, str]] = dict(base_expected_types)
+) -> Mapping[FixtureKey, dict[str, str]]:
+    expected_types: ChainMap[FixtureKey, dict[str, str]] = ChainMap({}, base_expected_types)
     for model_name in ordered_model_names:
         model: CompiledModel | None = model_map.get(model_name)
         if model is None:
diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py
index 019bdc3b13..9acbb34cc1 100644
--- a/src/sqlbuild/compiler/planner/models.py
+++ b/src/sqlbuild/compiler/planner/models.py
@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from decimal import Decimal, InvalidOperation
@@ -120,7 +120,7 @@ class RelationFixtureCompletion:
 
     fixture_sql_by_key: dict[FixtureKey, str]
     inferred_by_fixture: dict[FixtureKey, tuple[InferredColumn, ...]]
-    expected_types: dict[FixtureKey, dict[str, str]]
+    expected_types: Mapping[FixtureKey, dict[str, str]]
     diagnostics: tuple[RelationFixtureDiagnostic, ...]
 
 

From bb6aa966d5ad0f885ec02c9f6ad25968dc54a16c Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 19:11:10 +0100
Subject: [PATCH 5/8] fix(cli): report rules phase wall time on its progress
 line

The rules completion line reported only built-in and custom rule time,
which run concurrently and are not phase wall time, so users could not
see how long the rules phase took next to the compile line. It now leads
with the phase wall time in the same form as other phase lines.
---
 .../cli/commands/_helpers/compile/pipeline.py |  7 +--
 .../compiler/pipeline/main/compile.py         | 14 ++---
 .../rule_engine/main/render_rules_progress.py | 21 ++++++++
 .../main/plan/test_terminal_progress.py       |  3 +-
 .../main/render_rules_progress/_test_types.py | 14 +++++
 .../test_render_rules_progress.py             | 54 +++++++++++++++++++
 6 files changed, 102 insertions(+), 11 deletions(-)
 create mode 100644 src/sqlbuild/rule_engine/main/render_rules_progress.py
 create mode 100644 tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/_test_types.py
 create mode 100644 tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/test_render_rules_progress.py

diff --git a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py
index cede009293..5654563290 100644
--- a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py
+++ b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py
@@ -53,6 +53,7 @@
 from sqlbuild.presentation.classes.transient_status_reporter import TransientStatusReporter
 from sqlbuild.rule_engine.classes.early_sql_lint import EarlySqlLint
 from sqlbuild.rule_engine.main.load_config import load_rules_config
+from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress
 from sqlbuild.rule_engine.main.run_rules import run_rules
 from sqlbuild.rule_engine.models import RulesConfig, RulesRunResult
 from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle
@@ -204,6 +205,7 @@ def _analyze_compile_project(
         ):
             prepared_artifacts.start(project=graph.project, adapter=adapter)
         _ = start_compile_phase(status=status, message="Evaluating built-in and custom rules...")
+        rules_start: float = time.monotonic()
         rules_result = run_rules(
             graph=graph,
             discovered_inputs=discovered_inputs,
@@ -217,9 +219,8 @@ def _analyze_compile_project(
         )
         _ = complete_compile_phase(
             status=status,
-            message=(
-                f"Evaluated rules. (built-in {rules_result.built_in_ms / 1000:.2f}s, "
-                f"custom {rules_result.custom_ms / 1000:.2f}s)"
+            message=format_rules_progress(
+                elapsed_seconds=time.monotonic() - rules_start, result=rules_result
             ),
         )
     rule_diagnostics: tuple[CompilerDiagnostic, ...] = tuple(
diff --git a/src/sqlbuild/compiler/pipeline/main/compile.py b/src/sqlbuild/compiler/pipeline/main/compile.py
index 1a4f08017c..b4a7800262 100644
--- a/src/sqlbuild/compiler/pipeline/main/compile.py
+++ b/src/sqlbuild/compiler/pipeline/main/compile.py
@@ -66,7 +66,7 @@
 from sqlbuild.rule_engine.exceptions import RulesError
 from sqlbuild.rule_engine.main.load_config import load_rules_config
 from sqlbuild.rule_engine.main.render_result import format_result
-from sqlbuild.rule_engine.main.render_skipped_rules import format_skipped_type_proof_rules
+from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress
 from sqlbuild.rule_engine.main.run_rules import run_rules
 from sqlbuild.rule_engine.models import RulesCacheConfig, RulesConfig, RulesResult, RulesRunResult
 from sqlbuild.runtime.contracts.main.open_connection import open_connection_with_hooks
@@ -106,6 +106,7 @@ def run_compile_pipeline(
         rules_config = replace(rules_config, cache=RulesCacheConfig(enabled=False))
     if on_progress is not None:
         on_progress("Evaluating configured rules...")
+    rules_start: float = time.monotonic()
     rules_result: RulesRunResult = run_rules(
         graph=graph,
         discovered_inputs=discovered_inputs,
@@ -120,13 +121,12 @@ def run_compile_pipeline(
         no_sql_analysis=resolved_options.no_sql_validation,
     )
     if on_progress is not None:
-        skipped_rules_note: str | None = format_skipped_type_proof_rules(
-            codes=rules_result.skipped_type_proof_rules
-        )
         on_progress(
-            f"Evaluated rules. (built-in {rules_result.built_in_ms / 1000:.2f}s, "
-            f"custom {rules_result.custom_ms / 1000:.2f}s)"
-            + ("" if skipped_rules_note is None else f"; {skipped_rules_note}")
+            format_rules_progress(
+                elapsed_seconds=time.monotonic() - rules_start,
+                result=rules_result,
+                note_skipped_rules=True,
+            )
         )
     if rules_result.findings:
         rendered: str = format_result(
diff --git a/src/sqlbuild/rule_engine/main/render_rules_progress.py b/src/sqlbuild/rule_engine/main/render_rules_progress.py
new file mode 100644
index 0000000000..11f4bc5531
--- /dev/null
+++ b/src/sqlbuild/rule_engine/main/render_rules_progress.py
@@ -0,0 +1,21 @@
+"""Rules phase completion line entrypoint."""
+
+from sqlbuild.rule_engine.main.render_skipped_rules import format_skipped_type_proof_rules
+from sqlbuild.rule_engine.models import RulesRunResult
+
+
+def format_rules_progress(
+    *, elapsed_seconds: float, result: RulesRunResult, note_skipped_rules: bool = False
+) -> str:
+    """Report the rules phase wall time first, then built-in and custom rule time."""
+
+    message: str = (
+        f"Evaluated rules. ({elapsed_seconds:.2f}s; built-in {result.built_in_ms / 1000:.2f}s, "
+        f"custom {result.custom_ms / 1000:.2f}s)"
+    )
+    skipped_rules_note: str | None = (
+        format_skipped_type_proof_rules(codes=result.skipped_type_proof_rules)
+        if note_skipped_rules
+        else None
+    )
+    return message if skipped_rules_note is None else f"{message}; {skipped_rules_note}"
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py b/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py
index d89032a16c..4850001e11 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_terminal_progress.py
@@ -32,7 +32,8 @@
             columns=100,
             expected_screen_fragments=(
                 "Compiled project.",
-                "Evaluated rules.",
+                "Evaluated rules. (",
+                "s; built-in ",
                 "Connecting to duckdb...",
                 "✓ Warehouse connected  duckdb",
                 "Generated plan.",
diff --git a/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/_test_types.py b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/_test_types.py
new file mode 100644
index 0000000000..928e4a1882
--- /dev/null
+++ b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/_test_types.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class RulesProgressTestCase:
+    description: str
+    elapsed_seconds: float
+    built_in_ms: int
+    custom_ms: int
+    expected_message: str
+    skipped_type_proof_rules: tuple[str, ...] = ()
+    note_skipped_rules: bool = False
diff --git a/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/test_render_rules_progress.py b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/test_render_rules_progress.py
new file mode 100644
index 0000000000..0e28eb9de1
--- /dev/null
+++ b/tests/unit/src/sqlbuild/rule_engine/main/render_rules_progress/test_render_rules_progress.py
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import pytest
+
+from sqlbuild.rule_engine.main.render_rules_progress import format_rules_progress
+from sqlbuild.rule_engine.models import RulesRunResult
+from tests.unit.src.sqlbuild.rule_engine.main.render_rules_progress._test_types import (
+    RulesProgressTestCase,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        RulesProgressTestCase(
+            description="wall time leads concurrent built-in and custom rule time",
+            elapsed_seconds=14.5,
+            built_in_ms=5_600,
+            custom_ms=13_400,
+            expected_message="Evaluated rules. (14.50s; built-in 5.60s, custom 13.40s)",
+        ),
+        RulesProgressTestCase(
+            description="skipped type-proof rules follow the timings when requested",
+            elapsed_seconds=0.25,
+            built_in_ms=100,
+            custom_ms=0,
+            skipped_type_proof_rules=("SQBRCONTRACT105",),
+            note_skipped_rules=True,
+            expected_message=(
+                "Evaluated rules. (0.25s; built-in 0.10s, custom 0.00s); skipped type-proof "
+                "rules (SQBRCONTRACT105) because SQL analysis is disabled"
+            ),
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_rules_run_when_formatting_progress_then_reports_phase_wall_time_first(
+    test_case: RulesProgressTestCase,
+) -> None:
+    result: RulesRunResult = RulesRunResult(
+        findings=(),
+        evaluated_models=0,
+        built_in_ms=test_case.built_in_ms,
+        custom_ms=test_case.custom_ms,
+        skipped_type_proof_rules=test_case.skipped_type_proof_rules,
+    )
+
+    message: str = format_rules_progress(
+        elapsed_seconds=test_case.elapsed_seconds,
+        result=result,
+        note_skipped_rules=test_case.note_skipped_rules,
+    )
+
+    assert message == test_case.expected_message

From e30040de922be558183e64415417ee3a4ec9e40d Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 19:16:29 +0100
Subject: [PATCH 6/8] test(planner): guard plan phase scaling on the inspection
 benchmark

Plan the inspection benchmark at 1,000 and 3,000 models without the
compile cache and fail when warehouse inspection or plan generation
grows more than 1.5 times faster than the project, or when either
phase exceeds a loose ceiling.
---
 .../cli/commands/main/compile/_test_types.py  | 10 ++++
 .../cli/commands/main/compile/helpers.py      | 28 +++++++++
 .../test_inspection_command_performance.py    | 60 +++++++++++++++++++
 3 files changed, 98 insertions(+)

diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
index 9f3215492c..604db55122 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
@@ -331,6 +331,16 @@ class PlanScalingGuardTestCase:
     expected_max_linear_factor: float
 
 
+@dataclass(frozen=True)
+class PlanPhaseScalingGuardTestCase:
+    description: str
+    small_model_count: int
+    runs: int
+    expected_phases: tuple[str, ...]
+    expected_max_linear_factor: float
+    expected_max_large_phase_seconds: float
+
+
 @dataclass(frozen=True)
 class BuildPerformanceGuardTestCase:
     description: str
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
index 6d30815bdf..237943741e 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
@@ -643,6 +643,7 @@ class _TimedSqbRun(NamedTuple):
     minor_page_faults: int
     payload: object
     output_path: Path
+    stderr_path: Path
 
 
 def _run_timed_sqb(
@@ -703,6 +704,7 @@ def _run_timed_sqb(
         minor_page_faults=int(minor_text),
         payload=payload,
         output_path=output_path,
+        stderr_path=stderr_path,
     )
 
 
@@ -712,6 +714,7 @@ class InspectionCommandMeasurement(NamedTuple):
     cpu_seconds: float
     payload: object
     output: str
+    stderr: str = ""
 
 
 def run_fresh_process_inspection_command(
@@ -735,9 +738,34 @@ def run_fresh_process_inspection_command(
         cpu_seconds=run.cpu_seconds,
         payload=run.payload,
         output=run.output_path.read_text(encoding="utf-8"),
+        stderr=run.stderr_path.read_text(encoding="utf-8"),
     )
 
 
+_PLAN_PHASE_LINE: re.Pattern[str] = re.compile(
+    r"^(Inspected warehouse state|Generated plan)\. \((\d+\.\d+)s\)$", re.MULTILINE
+)
+
+
+def fastest_plan_phase_seconds(*, project_dir: Path, label: str, runs: int) -> dict[str, float]:
+    """Plan without the compile cache several times and keep each phase's fastest time."""
+
+    fastest: dict[str, float] = {}
+    index: int
+    for index in range(runs):
+        measurement: InspectionCommandMeasurement = run_fresh_process_inspection_command(
+            project_dir=project_dir,
+            label=f"{label}-{index}",
+            sqb_args=("plan", "--json", "--no-cache"),
+            expected_max_wall_seconds=120.0,
+        )
+        phase: str
+        seconds: str
+        for phase, seconds in _PLAN_PHASE_LINE.findall(measurement.stderr):
+            fastest[phase] = min(float(seconds), fastest.get(phase, float(seconds)))
+    return fastest
+
+
 INSPECTION_DIAMOND_LAYERS: int = 16
 INSPECTION_DIAMOND_WIDTH: int = 8
 INSPECTION_BENCHMARK_MODEL_COUNT: int = (
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py
index a7aae1f4a8..c7e463afc8 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py
@@ -18,15 +18,20 @@
 from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import (
     InspectionCommandPerformanceGuardTestCase,
     PlanComparedToCompileGuardTestCase,
+    PlanPhaseScalingGuardTestCase,
     PlanScalingGuardTestCase,
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import (
     INSPECTION_BENCHMARK_MODEL_COUNT,
+    INSPECTION_DIAMOND_LAYERS,
+    INSPECTION_DIAMOND_WIDTH,
     SHARED_DIAMOND_HUB,
     SHARED_DIAMOND_ROLLUP,
     InspectionCommandMeasurement,
+    fastest_plan_phase_seconds,
     prepare_small_inspection_project,
     run_fresh_process_inspection_command,
+    write_inspection_benchmark_project,
 )
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -285,3 +290,58 @@ def test_given_benchmarks_of_two_sizes_when_planning_then_planning_work_scales_l
         linear_cpu,
     )
     assert planning_cpu["large"] < test_case.expected_max_linear_factor * linear_cpu
+
+
+@pytest.mark.performance
+@pytest.mark.cold_compile_performance
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        PlanPhaseScalingGuardTestCase(
+            description="models_3000_plan_phases_scale_linearly_from_1000_models",
+            small_model_count=1_000,
+            runs=2,
+            expected_phases=("Inspected warehouse state", "Generated plan"),
+            expected_max_linear_factor=1.5,
+            expected_max_large_phase_seconds=20.0,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_benchmarks_of_two_sizes_when_planning_then_each_plan_phase_scales_linearly(
+    inspection_benchmark_project: Path,
+    test_case: PlanPhaseScalingGuardTestCase,
+    tmp_path: Path,
+) -> None:
+    small_project: Path = tmp_path / "semantic_small"
+    write_inspection_benchmark_project(
+        project_dir=small_project, model_count=test_case.small_model_count
+    )
+    lattice_models: int = INSPECTION_DIAMOND_LAYERS * INSPECTION_DIAMOND_WIDTH + 2
+    size_ratio: float = _MODEL_COUNT / (test_case.small_model_count + lattice_models)
+    fastest: dict[str, dict[str, float]] = {
+        label: fastest_plan_phase_seconds(
+            project_dir=project_dir,
+            label=f"{test_case.description}-{label}-plan",
+            runs=test_case.runs,
+        )
+        for label, project_dir in (
+            ("small", small_project),
+            ("large", inspection_benchmark_project),
+        )
+    }
+
+    _LOGGER.info(
+        "plan phase scaling %s size_ratio=%.2f small=%s large=%s",
+        test_case.description,
+        size_ratio,
+        fastest["small"],
+        fastest["large"],
+    )
+    ratios: dict[str, float] = {
+        phase: seconds / max(fastest["small"].get(phase, 0.0), 0.01)
+        for phase, seconds in fastest["large"].items()
+    }
+    assert set(fastest["large"]) == set(fastest["small"]) == set(test_case.expected_phases)
+    assert max(ratios.values()) < test_case.expected_max_linear_factor * size_ratio, ratios
+    assert max(fastest["large"].values()) < test_case.expected_max_large_phase_seconds

From 86506d296372c38bde5da93b5649a07f8edf2379 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 19:40:21 +0100
Subject: [PATCH 7/8] fix(observability): release finished statements from
 deferred monitor starts

A stopped monitor now cancels its pending deferred start and drops its
callbacks and query-ID provider, so a finished statement and its SQL are
no longer retained until the heartbeat deadline. A monitor thread is
recorded only after it starts, so a refused start no longer makes stop()
join an unstarted thread and fail an otherwise successful statement.
---
 .../classes/deferred_thread_starts.py         | 54 ++++++++---
 .../classes/statement_monitor.py              | 21 ++++-
 .../runtime/observability/_test_types.py      | 23 +++++
 .../sqlbuild/runtime/observability/helpers.py | 87 ++++++++++++++++++
 .../observability/test_statement_monitor.py   | 91 +++++++++++++++++++
 5 files changed, 261 insertions(+), 15 deletions(-)

diff --git a/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py b/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py
index 1e9302ee62..62fbb013db 100644
--- a/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py
+++ b/src/sqlbuild/runtime/observability/classes/deferred_thread_starts.py
@@ -8,7 +8,9 @@
 import time
 from collections.abc import Callable
 
-type _DeferredStart = tuple[float, int, Callable[[], None]]
+type _DeferredDeadline = tuple[float, int]
+
+_COMPACTION_SLACK: int = 64
 
 
 class DeferredThreadStarts:
@@ -17,21 +19,44 @@ class DeferredThreadStarts:
     def __init__(self, *, name: str) -> None:
         self._name: str = name
         self._condition: threading.Condition = threading.Condition()
-        self._pending: list[_DeferredStart] = []
+        self._deadlines: list[_DeferredDeadline] = []
+        self._starts: dict[int, Callable[[], None]] = {}
         self._order: itertools.count[int] = itertools.count()
         self._thread: threading.Thread | None = None
 
-    def schedule(self, *, delay_seconds: float, start: Callable[[], None]) -> None:
-        """Call ``start`` after ``delay_seconds``; the callback must tolerate a stopped owner."""
+    @property
+    def pending_count(self) -> int:
+        """Return how many scheduled starts have neither run nor been cancelled."""
+
+        with self._condition:
+            return len(self._starts)
+
+    def schedule(self, *, delay_seconds: float, start: Callable[[], None]) -> int:
+        """Call ``start`` after ``delay_seconds`` unless cancelled; return its cancel token."""
 
-        entry: _DeferredStart = (time.monotonic() + delay_seconds, next(self._order), start)
+        token: int = next(self._order)
+        deadline: _DeferredDeadline = (time.monotonic() + delay_seconds, token)
         with self._condition:
-            heapq.heappush(self._pending, entry)
+            self._starts[token] = start
+            heapq.heappush(self._deadlines, deadline)
             if self._thread is None:
                 self._thread = threading.Thread(target=self._run, name=self._name, daemon=True)
                 self._thread.start()
-            elif self._pending[0] is entry:
+            elif self._deadlines[0] is deadline:
                 self._condition.notify()
+        return token
+
+    def cancel(self, token: int) -> None:
+        """Drop a scheduled start so nothing it references outlives its owner."""
+
+        with self._condition:
+            if self._starts.pop(token, None) is None:
+                return
+            if len(self._deadlines) > 2 * len(self._starts) + _COMPACTION_SLACK:
+                self._deadlines = [
+                    deadline for deadline in self._deadlines if deadline[1] in self._starts
+                ]
+                heapq.heapify(self._deadlines)
 
     def _run(self) -> None:
         while True:
@@ -44,10 +69,15 @@ def _run(self) -> None:
     def _next_due(self) -> Callable[[], None]:
         with self._condition:
             while True:
-                if not self._pending:
+                if not self._deadlines:
                     _ = self._condition.wait()
                     continue
-                remaining: float = self._pending[0][0] - time.monotonic()
-                if remaining <= 0:
-                    return heapq.heappop(self._pending)[2]
-                _ = self._condition.wait(remaining)
+                due_at, token = self._deadlines[0]
+                remaining: float = due_at - time.monotonic()
+                if remaining > 0:
+                    _ = self._condition.wait(remaining)
+                    continue
+                heapq.heappop(self._deadlines)
+                start: Callable[[], None] | None = self._starts.pop(token, None)
+                if start is not None:
+                    return start
diff --git a/src/sqlbuild/runtime/observability/classes/statement_monitor.py b/src/sqlbuild/runtime/observability/classes/statement_monitor.py
index 95fcadb5f6..d662dd2000 100644
--- a/src/sqlbuild/runtime/observability/classes/statement_monitor.py
+++ b/src/sqlbuild/runtime/observability/classes/statement_monitor.py
@@ -45,6 +45,7 @@ def __init__(
         self._thread_lock: threading.Lock = threading.Lock()
         self._context: Context | None = None
         self._stopped: bool = False
+        self._deferred_start: int | None = None
 
     @property
     def query_id(self) -> str | None:
@@ -59,7 +60,7 @@ def start(self) -> None:
         if self._has_query_id_provider() or self._threshold_seconds <= 0:
             self._start_thread()
             return
-        _DEFERRED_MONITOR_STARTS.schedule(
+        self._deferred_start = _DEFERRED_MONITOR_STARTS.schedule(
             delay_seconds=self._threshold_seconds, start=self._start_thread
         )
 
@@ -82,22 +83,36 @@ def stop(self) -> str | None:
             self._stopped = True
             self._context = None
             thread: threading.Thread | None = self._thread
+            deferred_start: int | None = self._deferred_start
+            self._deferred_start = None
+        if deferred_start is not None:
+            _DEFERRED_MONITOR_STARTS.cancel(deferred_start)
         if thread is not None and thread is not threading.current_thread():
             thread.join()
+        self._release_callbacks()
         return self._query_id
 
+    def _release_callbacks(self) -> None:
+        """Drop references back to the statement so a stopped monitor retains nothing."""
+
+        with self._provider_lock:
+            self._query_id_provider = None
+        self._on_submitted = lambda query_id: None
+        self._on_heartbeat = lambda elapsed_seconds, query_id: None
+
     def _start_thread(self) -> None:
         with self._thread_lock:
             context: Context | None = self._context
             if self._stopped or self._thread is not None or context is None:
                 return
-            self._thread = threading.Thread(
+            thread: threading.Thread = threading.Thread(
                 target=context.run,
                 args=(self._run,),
                 name="sqlbuild-statement-monitor",
                 daemon=True,
             )
-            self._thread.start()
+            thread.start()
+            self._thread = thread
 
     def _run(self) -> None:
         next_heartbeat: float = self._started_at + self._threshold_seconds
diff --git a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py
index b7d0563611..5105567e38 100644
--- a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py
+++ b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py
@@ -351,3 +351,26 @@ class InvocationSequenceCase:
     sequence: object
     expected_error: str = ""
     expected_sequence: int | None = None
+
+
+@dataclass(frozen=True)
+class FinishedStatementRetentionCase:
+    description: str
+    statement_count: int
+    sql_bytes: int
+    expected_live_lifecycles: int
+
+
+@dataclass(frozen=True)
+class DeferredStartCancellationCase:
+    description: str
+    delay_seconds: float
+    expected_pending: int
+    expected_runs: int
+
+
+@dataclass(frozen=True)
+class FailedMonitorStartCase:
+    description: str
+    threshold_seconds: float
+    expected_event_types: tuple[str, ...]
diff --git a/tests/unit/src/sqlbuild/runtime/observability/helpers.py b/tests/unit/src/sqlbuild/runtime/observability/helpers.py
index c422392c28..7a483738dc 100644
--- a/tests/unit/src/sqlbuild/runtime/observability/helpers.py
+++ b/tests/unit/src/sqlbuild/runtime/observability/helpers.py
@@ -1,8 +1,10 @@
 """Test builders for runtime observability contracts."""
 
 import asyncio
+import gc
 import threading
 import time
+import weakref
 from collections.abc import Callable, Mapping
 from concurrent.futures import ThreadPoolExecutor
 from contextvars import Context, copy_context
@@ -15,6 +17,7 @@
 from sqlbuild.runtime.observability._helpers.dispatcher import dispatcher_scope
 from sqlbuild.runtime.observability._helpers.factory import create_lifecycle_event
 from sqlbuild.runtime.observability._helpers.identity import invocation_scope
+from sqlbuild.runtime.observability.classes.deferred_thread_starts import DeferredThreadStarts
 from sqlbuild.runtime.observability.classes.event_dispatcher import EventDispatcher
 from sqlbuild.runtime.observability.classes.microbatch_lifecycle import MicrobatchLifecycle
 from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle
@@ -270,3 +273,87 @@ def heartbeats_after_early_stop(*, threshold_seconds: float) -> int:
     _ = monitor.stop()
     time.sleep(threshold_seconds * 4)
     return len(heartbeats)
+
+
+def live_lifecycles_after_large_statements(*, statement_count: int, sql_bytes: int) -> int:
+    """Run statements with large SQL and count lifecycles still reachable after collection."""
+
+    references: list[weakref.ref[StatementLifecycle]] = []
+    dispatcher: EventDispatcher
+    dispatcher, _ = capture_lifecycle_events()
+    with invocation_scope("inv-large-statements"), dispatcher_scope(dispatcher):
+        for index in range(statement_count):
+            lifecycle: StatementLifecycle = StatementLifecycle(
+                adapter="duckdb", sql=f"SELECT {index}, '{'x' * sql_bytes}'", intent="execute"
+            )
+            with lifecycle:
+                pass
+            references.append(weakref.ref(lifecycle))
+        del lifecycle
+    _ = gc.collect()
+    return len(references) - [reference() for reference in references].count(None)
+
+
+def run_cancelled_deferred_start(*, delay_seconds: float) -> tuple[int, int]:
+    """Schedule then cancel one start and return pending starts and runs after its deadline."""
+
+    runs: list[int] = []
+    starts: DeferredThreadStarts = DeferredThreadStarts(name="sqlbuild-test-deferred-starts")
+    token: int = starts.schedule(delay_seconds=delay_seconds, start=lambda: runs.append(1))
+    starts.cancel(token)
+    time.sleep(delay_seconds * 4)
+    return starts.pending_count, len(runs)
+
+
+def statement_events_when_monitor_start_fails(
+    *, monkeypatch: pytest.MonkeyPatch, threshold_seconds: float
+) -> tuple[str, ...]:
+    """Fail every monitor thread start and return the statement's published event types."""
+
+    original_start: Callable[[threading.Thread], None] = threading.Thread.start
+    refusals: dict[str, Callable[[threading.Thread], None]] = {
+        "sqlbuild-statement-monitor": _refuse_thread_start
+    }
+
+    def refusing_start(thread: threading.Thread) -> None:
+        refusals.get(thread.name, original_start)(thread)
+
+    monkeypatch.setattr(threading.Thread, "start", refusing_start)
+    dispatcher: EventDispatcher
+    events: list[LifecycleEvent]
+    dispatcher, events = capture_lifecycle_events()
+    with invocation_scope("inv-refused-monitor"), dispatcher_scope(dispatcher):
+        with StatementLifecycle(
+            adapter="duckdb",
+            sql="CREATE TABLE orders AS SELECT 1",
+            intent="execute",
+            heartbeat_threshold_seconds=threshold_seconds,
+        ):
+            time.sleep(threshold_seconds * 5)
+    return tuple(event.event_type for event in events)
+
+
+def _refuse_thread_start(thread: threading.Thread) -> None:
+    raise RuntimeError(f"cannot start {thread.name}")
+
+
+def heartbeats_past_threshold(*, threshold_seconds: float) -> int:
+    """Run a deferred monitor past its threshold and count heartbeats before stopping it."""
+
+    heartbeat: threading.Event = threading.Event()
+    heartbeats: list[float] = []
+
+    def record_heartbeat(elapsed: float, query_id: str | None) -> None:
+        del query_id
+        heartbeats.append(elapsed)
+        heartbeat.set()
+
+    monitor: StatementMonitor = StatementMonitor(
+        on_submitted=lambda value: None,
+        on_heartbeat=record_heartbeat,
+        threshold_seconds=threshold_seconds,
+    )
+    monitor.start()
+    _ = heartbeat.wait(timeout=5.0)
+    _ = monitor.stop()
+    return len(heartbeats)
diff --git a/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py b/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py
index 35275e62de..d2b07a4723 100644
--- a/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py
+++ b/tests/unit/src/sqlbuild/runtime/observability/test_statement_monitor.py
@@ -4,11 +4,18 @@
 
 from tests.unit.src.sqlbuild.runtime.observability._test_types import (
     DeferredMonitorCase,
+    DeferredStartCancellationCase,
+    FailedMonitorStartCase,
     FastStatementMonitorCase,
+    FinishedStatementRetentionCase,
 )
 from tests.unit.src.sqlbuild.runtime.observability.helpers import (
     count_monitor_thread_starts,
     heartbeats_after_early_stop,
+    heartbeats_past_threshold,
+    live_lifecycles_after_large_statements,
+    run_cancelled_deferred_start,
+    statement_events_when_monitor_start_fails,
     submissions_after_late_provider,
 )
 
@@ -76,5 +83,89 @@ def test_given_monitor_stopped_before_threshold_when_deadline_passes_then_no_hea
     assert heartbeats == len(test_case.expected_submissions)
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FinishedStatementRetentionCase(
+            description="finished statements with large SQL are not retained",
+            statement_count=200,
+            sql_bytes=100_000,
+            expected_live_lifecycles=0,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_finished_statements_when_collecting_then_no_lifecycle_is_retained(
+    test_case: FinishedStatementRetentionCase,
+) -> None:
+    live: int = live_lifecycles_after_large_statements(
+        statement_count=test_case.statement_count, sql_bytes=test_case.sql_bytes
+    )
+
+    assert live == test_case.expected_live_lifecycles
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        DeferredStartCancellationCase(
+            description="cancelled start is forgotten and never runs",
+            delay_seconds=0.02,
+            expected_pending=0,
+            expected_runs=0,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_cancelled_deferred_start_when_deadline_passes_then_it_never_runs(
+    test_case: DeferredStartCancellationCase,
+) -> None:
+    pending, runs = run_cancelled_deferred_start(delay_seconds=test_case.delay_seconds)
+
+    assert pending == test_case.expected_pending
+    assert runs == test_case.expected_runs
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        DeferredMonitorCase(
+            description="statement outliving a lowered threshold still heartbeats",
+            threshold_seconds=0.02,
+            query_id="",
+            expected_submissions=("heartbeat",),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_statement_past_threshold_when_deferred_start_runs_then_it_heartbeats(
+    test_case: DeferredMonitorCase,
+) -> None:
+    heartbeats: int = heartbeats_past_threshold(threshold_seconds=test_case.threshold_seconds)
+
+    assert heartbeats >= len(test_case.expected_submissions)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FailedMonitorStartCase(
+            description="refused monitor thread still completes the statement once",
+            threshold_seconds=0.01,
+            expected_event_types=("statement_started", "statement_completed"),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_monitor_thread_cannot_start_when_statement_finishes_then_it_completes(
+    test_case: FailedMonitorStartCase, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    event_types: tuple[str, ...] = statement_events_when_monitor_start_fails(
+        monkeypatch=monkeypatch, threshold_seconds=test_case.threshold_seconds
+    )
+
+    assert event_types == test_case.expected_event_types
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-vv"])

From 9313ec15b63168a499a5daa8f38186802f40189a Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 1 Oct 2026 19:42:45 +0100
Subject: [PATCH 8/8] test(pipeline): account for the rules wall-time clock
 read in phase timings

The rules progress line reads the clock once before rules run, between
the compile and planning windows. The phase-timing test now supplies
that read, gives rules a measurable duration, and proves compile and
planning stay disjoint from it, with and without progress reporting.
---
 .../compiler/pipeline/main/_test_types.py     |  2 ++
 .../main/test_compile_phase_timings.py        | 28 ++++++++++++++++---
 2 files changed, 26 insertions(+), 4 deletions(-)

diff --git a/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py b/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py
index 898d3db9ab..1fb2794b64 100644
--- a/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py
+++ b/tests/unit/src/sqlbuild/compiler/pipeline/main/_test_types.py
@@ -40,3 +40,5 @@ class PipelinePhaseTimingTestCase:
     expected_compile_seconds: float
     expected_planning_seconds: float
     expected_total_seconds: float
+    report_progress: bool = False
+    expected_progress_fragments: tuple[str, ...] = ()
diff --git a/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py b/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py
index cfb4afec0a..98da979be6 100644
--- a/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py
+++ b/tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py
@@ -14,6 +14,7 @@
 from sqlbuild.diagnostics.classes.build_phase_timing_tracker import BuildPhaseTimingTracker
 from sqlbuild.diagnostics.models import PartialBuildPhaseTimings
 from sqlbuild.rule_engine.models import RulesConfig, RulesRunResult
+from sqlbuild.runtime.contracts.models import ConnectionHooks
 from tests.unit.src.sqlbuild.compiler.pipeline.main._test_types import PipelinePhaseTimingTestCase
 
 
@@ -22,11 +23,23 @@
     [
         PipelinePhaseTimingTestCase(
             description="slow planning connection is excluded from direct compilation",
-            clock_values=(0.0, 0.0, 2.0, 2.0, 12.0, 12.0),
+            clock_values=(0.0, 0.0, 2.0, 2.0, 3.0, 13.0, 13.0),
             expected_compile_seconds=2.0,
             expected_planning_seconds=10.0,
-            expected_total_seconds=12.0,
-        )
+            expected_total_seconds=13.0,
+        ),
+        PipelinePhaseTimingTestCase(
+            description="rules wall time is reported and excluded from compile and planning",
+            clock_values=(0.0, 0.0, 2.0, 2.0, 3.0, 3.0, 13.0, 13.0),
+            expected_compile_seconds=2.0,
+            expected_planning_seconds=10.0,
+            expected_total_seconds=13.0,
+            report_progress=True,
+            expected_progress_fragments=(
+                "Compiled project. (2.00s)",
+                "Evaluated rules. (1.00s; built-in 0.00s, custom 0.00s)",
+            ),
+        ),
     ],
     ids=lambda case: case.description,
 )
@@ -65,18 +78,25 @@ def test_given_slow_planning_connection_when_compiling_then_direct_phases_are_di
         ),
     )
     tracker: BuildPhaseTimingTracker = BuildPhaseTimingTracker(monotonic=monotonic)
+    progress: list[str] = []
+    progress_hooks: dict[bool, ConnectionHooks] = {
+        True: ConnectionHooks(on_progress=progress.append),
+        False: ConnectionHooks(),
+    }
 
     with tracker.scope():
         result: CompilePipelineResult = compile_module.run_compile_pipeline(
             discovered_inputs=Mock(),
             adapter=adapter,
             options=CompilePipelineOptions(connection_config={}),
+            hooks=progress_hooks[test_case.report_progress],
         )
     timings: PartialBuildPhaseTimings = tracker.snapshot()
 
     assert result.compile_seconds == test_case.expected_compile_seconds
     assert result.planning_seconds == test_case.expected_planning_seconds
     assert timings.total_seconds == test_case.expected_total_seconds
-    assert (result.compile_seconds or 0) + (result.planning_seconds or 0) <= (
+    assert (result.compile_seconds or 0) + (result.planning_seconds or 0) < (
         timings.total_seconds or 0
     )
+    assert set(test_case.expected_progress_fragments) <= set(progress)