diff --git a/AGENTS.md b/AGENTS.md
index ece9f03f92..79f0386051 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -201,6 +201,13 @@
 - Members marked `[forced]` are adapter contract overrides required by `test_strict_adapter.py`. Record other intentional mirrors in `[dupes]` in `fensu.toml` with a reason.
 - Public adapter contract methods are duplicated per adapter on purpose: `test_strict_adapter.py` requires every first-class adapter to define them, and `[dupes]` exempts them. The private helpers behind them must not be copied. Keep them in shared builders, such as the snapshot SQL builders in `src/sqlbuild/adapter/contract/classes/`, and let each contract method delegate. Express real dialect differences as explicit parameters, such as a `SnapshotSqlDialect` field, not as a forked copy of the helper.
 
+## Rust-Hosted Migration
+
+- Rust is moving to host compile, planning and non-warehouse CLI work. Python stays only for the modules listed in `python-boundary.toml`: user Python code, adapters, the executor and public Python APIs.
+- A migration change deletes the Python it replaces in the same pull request and lowers `remaining_lines`. Never keep a parallel Python implementation, and never fall back to Python after a native failure; surface the failure as an error.
+- Adding a `stays` or `handled_native_errors` entry, or raising `remaining_lines`, requires a stated reason in the pull request and explicit user approval. Difficulty or a small measured speed gain is not a reason.
+- Moving code into a path listed in `stays`, or behind a reviewed handler, is not migration progress; only code that now runs in Rust lowers `remaining_lines`.
+
 ## Bounded Review Process
 
 Orchestrating agents must run reviews as a single bounded cycle, not an open-ended loop:
diff --git a/Makefile b/Makefile
index 0e799e32a1..1b9ccf6c88 100644
--- a/Makefile
+++ b/Makefile
@@ -1,6 +1,6 @@
 SHELL := /bin/bash
 
-.PHONY: verify verify-quick verify-pg coverage cli-preview rust-check check-native-layering \
+.PHONY: verify verify-quick verify-pg coverage cli-preview rust-check check-native-layering check-python-boundary \
 	check-e2e-shards test-e2e-duckdb test-e2e-duckdb-build-core \
 	test-e2e-duckdb-build-incremental \
 	test-e2e-duckdb-cli-data test-e2e-duckdb-cli test-e2e-duckdb-inspection \
@@ -604,7 +604,7 @@ verify-quick:
 	exit $$status
 
 
-check-ci: check-e2e-shards check-native-layering
+check-ci: check-e2e-shards check-native-layering check-python-boundary
 	uv run ruff format --check .
 	uv run ruff check .
 	uv run ty check src tests
@@ -616,6 +616,10 @@ check-native-layering:
 	uv run python -m scripts.check_native_layering
 
 
+check-python-boundary:
+	uv run python -m scripts.check_python_boundary
+
+
 check-pr-metadata:
 	@test -n "$(PR_TITLE)" || { echo "PR_TITLE is required" >&2; exit 2; }
 	@test -n "$(PR_BODY_FILE)" || { echo "PR_BODY_FILE is required" >&2; exit 2; }
diff --git a/python-boundary.toml b/python-boundary.toml
new file mode 100644
index 0000000000..205db07533
--- /dev/null
+++ b/python-boundary.toml
@@ -0,0 +1,221 @@
+# Python that stays once Rust hosts SQLBuild. Every other Python file under src/sqlbuild is
+# migration work that moves to Rust. `remaining_lines` must equal the current count of that work:
+# a PR that moves code lowers it, and adding movable Python fails the check.
+# Native failures must surface as errors; `handled_native_errors` lists the reviewed exceptions.
+remaining_lines = 146314
+
+[[stays]]
+path = "executor"
+reason = "executor"
+note = "Runs plans against warehouses; a separate later decision."
+
+[[stays]]
+path = "adapters"
+reason = "adapter"
+note = "Built-in warehouse adapters that project adapters subclass and override."
+
+[[stays]]
+path = "adapter/contract"
+reason = "adapter"
+note = "BaseAdapter and StrictAdapter, the documented classes project adapters subclass."
+
+[[stays]]
+path = "adapter/discovery"
+reason = "user-code"
+note = "Imports project adapter modules."
+
+[[stays]]
+path = "adapter/relations"
+reason = "adapter"
+note = "Relation inspection wrappers adapters call, with per-connection caches."
+
+[[stays]]
+path = "adapter/state_sql"
+reason = "adapter"
+note = "State-table SQL rendered through adapter callbacks."
+
+[[stays]]
+path = "adapter/type_system"
+reason = "adapter"
+note = "Type normalisation adapters import; already a thin native wrapper."
+
+[[stays]]
+path = "provider"
+reason = "user-code"
+note = "Provider base class users subclass, and injection into user functions."
+
+[[stays]]
+path = "python_nodes"
+reason = "user-code"
+note = "Decorators that run when user modules are imported."
+
+[[stays]]
+path = "microbatches"
+reason = "executor"
+note = "Microbatch state store the executor and adapters use."
+
+[[stays]]
+path = "rules"
+reason = "public-api"
+note = "Public custom-rule API and test helper."
+
+[[stays]]
+path = "rule_engine/_helpers/host"
+reason = "user-code"
+note = "Subprocess host that loads and runs user rule modules."
+
+[[stays]]
+path = "rule_engine/classes/rule_context.py"
+reason = "user-code"
+note = "The context object user rules receive."
+
+[[stays]]
+path = "runtime/observability"
+reason = "public-api"
+note = "Documented observability API used throughout the executor."
+
+[[stays]]
+path = "runtime/execution_history"
+reason = "public-api"
+note = "Documented SQLite and Postgres execution-history backends."
+
+[[stays]]
+path = "runtime/execution_limits"
+reason = "executor"
+note = "Deadline the executor scheduler and warehouse cursors enforce."
+
+[[stays]]
+path = "integrations/dagster"
+reason = "public-api"
+note = "Documented orchestrator integration."
+
+[[stays]]
+path = "integrations/rivers"
+reason = "public-api"
+note = "Documented orchestrator integration."
+
+[[stays]]
+path = "integrations/dlt"
+reason = "user-code"
+note = "Loader runtime that runs inside user loaders."
+
+[[stays]]
+path = "integrations/ingestr"
+reason = "user-code"
+note = "Loader runtime that runs inside the executor load phase."
+
+[[stays]]
+path = "compiler/python_nodes/_helpers/identity.py"
+reason = "user-code"
+note = "Identity hashing over live user functions (source and globals)."
+
+[[stays]]
+path = "compiler/discovery/_helpers/filesystem/command_output_sinks.py"
+reason = "user-code"
+note = "Inspects loaded user sink functions."
+
+[[stays]]
+path = "compiler/pipeline/_helpers/materializations.py"
+reason = "user-code"
+note = "Imports user materialization modules."
+
+[[stays]]
+path = "assets.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "audits.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "checks.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "compute_logs.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "execution_history.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "factories.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "functions.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "hooks.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "loaders.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "observability.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "postgres_history.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "providers.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "refs.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "retries.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "sinks.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "sqlite_history.py"
+reason = "public-api"
+note = "Public facade."
+
+[[stays]]
+path = "tasks.py"
+reason = "public-api"
+note = "Public facade."
+
+[[handled_native_errors]]
+location = "compiler/macro_bridge/_helpers/store_keys.py::macro_store_token"
+note = "A macro without a stable key is not stored; no Python implementation runs."
+
+[[handled_native_errors]]
+location = "compiler/macro_bridge/_helpers/store_keys.py::context_store_token"
+note = "A context without a stable key is not stored; no Python implementation runs."
+
+[[handled_native_errors]]
+location = "lint/_helpers/native_sql.py::_native_responses"
+note = "A native lint failure is reported as a lint error result, not recomputed in Python."
+
+[[handled_native_errors]]
+location = "rule_engine/_helpers/run/rules.py::_cached_sql_findings"
+note = "An undecodable cached finding is a cache miss; the rules run again natively."
diff --git a/scripts/check_python_boundary.py b/scripts/check_python_boundary.py
new file mode 100644
index 0000000000..5eab7ccd58
--- /dev/null
+++ b/scripts/check_python_boundary.py
@@ -0,0 +1,14 @@
+"""Direct wrapper for the Python boundary ratchet."""
+
+from __future__ import annotations
+
+from scripts.python_boundary.main.check import run_python_boundary_check
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Check the Python boundary ratchet."""
+    return run_python_boundary_check(argv)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/python_boundary/__init__.py b/scripts/python_boundary/__init__.py
new file mode 100644
index 0000000000..bc5b06eae8
--- /dev/null
+++ b/scripts/python_boundary/__init__.py
@@ -0,0 +1 @@
+"""Python boundary ratchet for the Rust-hosted migration."""
diff --git a/scripts/python_boundary/_helpers/__init__.py b/scripts/python_boundary/_helpers/__init__.py
new file mode 100644
index 0000000000..01077d23c8
--- /dev/null
+++ b/scripts/python_boundary/_helpers/__init__.py
@@ -0,0 +1 @@
+"""Python boundary ratchet helpers."""
diff --git a/scripts/python_boundary/_helpers/lines.py b/scripts/python_boundary/_helpers/lines.py
new file mode 100644
index 0000000000..fb5c1874ed
--- /dev/null
+++ b/scripts/python_boundary/_helpers/lines.py
@@ -0,0 +1,26 @@
+"""Count the Python lines that remain to move to Rust."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from scripts.python_boundary.constants import SOURCE_ROOT
+from scripts.python_boundary.models import StayEntry
+
+
+def tracked_python_files(*, root: Path, stays: tuple[StayEntry, ...]) -> tuple[Path, ...]:
+    """Every Python file under the source root outside the modules that stay Python."""
+    source: Path = root / SOURCE_ROOT
+    staying: tuple[Path, ...] = tuple(source / entry.path for entry in stays)
+    return tuple(
+        path for path in sorted(source.rglob("*.py")) if not _stays(path=path, staying=staying)
+    )
+
+
+def _stays(*, path: Path, staying: tuple[Path, ...]) -> bool:
+    return any(path == stay or stay in path.parents for stay in staying)
+
+
+def count_lines(paths: tuple[Path, ...]) -> int:
+    """The total physical line count of `paths`."""
+    return sum(len(path.read_text(encoding="utf-8").splitlines()) for path in paths)
diff --git a/scripts/python_boundary/_helpers/native_errors.py b/scripts/python_boundary/_helpers/native_errors.py
new file mode 100644
index 0000000000..ec186c6c98
--- /dev/null
+++ b/scripts/python_boundary/_helpers/native_errors.py
@@ -0,0 +1,129 @@
+"""Find code that calls native functions and swallows their failures."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+from scripts.python_boundary.constants import (
+    MODULE_SCOPE,
+    NATIVE_MARKER,
+    NATIVE_PACKAGE,
+    SUPPRESS_NAME,
+)
+from scripts.python_boundary.models import NativeErrorHandler
+
+
+def find_native_error_handlers(
+    *, source: Path, paths: tuple[Path, ...]
+) -> tuple[NativeErrorHandler, ...]:
+    """Each handler or `suppress` around a native call that does not unconditionally raise."""
+    handlers: list[NativeErrorHandler] = []
+    for path in paths:
+        tree: ast.Module = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        native: frozenset[str] = _native_names(tree)
+        if not native:
+            continue
+        relative: str = path.relative_to(source).as_posix()
+        for scope, node in _guarded_blocks(tree):
+            for line in _swallowing_lines(node=node, native=native):
+                handlers.append(NativeErrorHandler(location=f"{relative}::{scope}", line=line))
+    return tuple(handlers)
+
+
+def _native_names(tree: ast.Module) -> frozenset[str]:
+    """Names bound to native code: imports from native modules or of native-named functions."""
+    names: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and node.module is not None:
+            native_module: bool = _is_native_module(node.module)
+            names.update(
+                alias.asname or alias.name
+                for alias in node.names
+                if native_module or _names_native(alias.name)
+            )
+        elif isinstance(node, ast.Import):
+            names.update(
+                (alias.asname or alias.name).split(".")[0]
+                for alias in node.names
+                if _is_native_module(alias.name)
+            )
+    return frozenset(names)
+
+
+def _is_native_module(module: str) -> bool:
+    parts: list[str] = module.split(".")
+    return parts[0] == NATIVE_PACKAGE and any(_names_native(part) for part in parts[1:])
+
+
+def _names_native(name: str) -> bool:
+    return NATIVE_MARKER in name.split("_")
+
+
+def _guarded_blocks(tree: ast.Module) -> list[tuple[str, ast.stmt]]:
+    """Every `try` and `with` statement, named by its innermost enclosing function."""
+    return _collect(node=tree, scope=MODULE_SCOPE, owner=None)
+
+
+def _collect(*, node: ast.AST, scope: str, owner: str | None) -> list[tuple[str, ast.stmt]]:
+    found: list[tuple[str, ast.stmt]] = []
+    for child in ast.iter_child_nodes(node):
+        if isinstance(child, ast.ClassDef):
+            found.extend(_collect(node=child, scope=scope, owner=child.name))
+        elif isinstance(child, ast.FunctionDef | ast.AsyncFunctionDef):
+            name: str = child.name if owner is None else f"{owner}.{child.name}"
+            found.extend(_collect(node=child, scope=name, owner=None))
+        else:
+            if isinstance(child, ast.Try | ast.TryStar | ast.With | ast.AsyncWith):
+                found.append((scope, child))
+            found.extend(_collect(node=child, scope=scope, owner=owner))
+    return found
+
+
+def _swallowing_lines(*, node: ast.stmt, native: frozenset[str]) -> list[int]:
+    if isinstance(node, ast.Try | ast.TryStar):
+        if not _calls_native(body=node.body, native=native):
+            return []
+        return [handler.lineno for handler in node.handlers if not _always_raises(handler)]
+    if isinstance(node, ast.With | ast.AsyncWith) and any(
+        _is_suppress(item.context_expr) for item in node.items
+    ):
+        return [node.lineno] if _calls_native(body=node.body, native=native) else []
+    return []
+
+
+def _always_raises(handler: ast.ExceptHandler) -> bool:
+    return isinstance(handler.body[-1], ast.Raise) and not any(
+        _exits_early(statement) for statement in handler.body
+    )
+
+
+def _exits_early(node: ast.AST) -> bool:
+    if isinstance(node, ast.Return | ast.Break | ast.Continue):
+        return True
+    if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda | ast.ClassDef):
+        return False
+    return any(_exits_early(child) for child in ast.iter_child_nodes(node))
+
+
+def _is_suppress(expression: ast.expr) -> bool:
+    if not isinstance(expression, ast.Call):
+        return False
+    function: ast.expr = expression.func
+    if isinstance(function, ast.Attribute):
+        return function.attr == SUPPRESS_NAME
+    return isinstance(function, ast.Name) and function.id == SUPPRESS_NAME
+
+
+def _calls_native(*, body: list[ast.stmt], native: frozenset[str]) -> bool:
+    for statement in body:
+        for node in ast.walk(statement):
+            if isinstance(node, ast.Call) and _root_name(node.func) in native:
+                return True
+    return False
+
+
+def _root_name(expression: ast.expr) -> str | None:
+    while isinstance(expression, ast.Attribute):
+        expression = expression.value
+    return expression.id if isinstance(expression, ast.Name) else None
diff --git a/scripts/python_boundary/_helpers/policy.py b/scripts/python_boundary/_helpers/policy.py
new file mode 100644
index 0000000000..a3c2538b46
--- /dev/null
+++ b/scripts/python_boundary/_helpers/policy.py
@@ -0,0 +1,62 @@
+"""Read and validate the Python boundary policy file."""
+
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+from typing import Any
+
+from scripts.python_boundary.constants import ESCAPING_PATH_PARTS, SOURCE_ROOT, STAY_REASONS
+from scripts.python_boundary.models import BoundaryPolicy, ReviewedNativeHandler, StayEntry
+
+
+def read_boundary_policy(path: Path) -> BoundaryPolicy:
+    """Return the policy stored at `path`."""
+    with path.open("rb") as handle:
+        document: dict[str, Any] = tomllib.load(handle)
+    return BoundaryPolicy(
+        remaining_lines=int(document.get("remaining_lines", -1)),
+        stays=tuple(
+            StayEntry(
+                path=str(entry.get("path", "")),
+                reason=str(entry.get("reason", "")),
+                note=str(entry.get("note", "")),
+            )
+            for entry in document.get("stays", ())
+        ),
+        handled_native_errors=tuple(
+            ReviewedNativeHandler(
+                location=str(entry.get("location", "")), note=str(entry.get("note", ""))
+            )
+            for entry in document.get("handled_native_errors", ())
+        ),
+    )
+
+
+def get_policy_errors(*, root: Path, policy: BoundaryPolicy) -> tuple[str, ...]:
+    """Return every malformed or stale policy entry."""
+    source: Path = root / SOURCE_ROOT
+    errors: list[str] = []
+    if policy.remaining_lines < 0:
+        errors.append("remaining_lines must be a non-negative integer.")
+    seen: set[str] = set()
+    for entry in policy.stays:
+        if entry.path in seen:
+            errors.append(f"stays lists {entry.path!r} more than once.")
+        seen.add(entry.path)
+        parts: tuple[str, ...] = Path(entry.path).parts
+        if not parts or Path(entry.path).is_absolute() or set(parts) & ESCAPING_PATH_PARTS:
+            errors.append(f"stays entry {entry.path!r} must name a module under {SOURCE_ROOT}.")
+        elif not (source / entry.path).exists():
+            errors.append(f"stays entry {entry.path!r} does not exist under {SOURCE_ROOT}.")
+        if entry.reason not in STAY_REASONS:
+            errors.append(
+                f"stays entry {entry.path!r} has reason {entry.reason!r}; "
+                f"use one of {', '.join(STAY_REASONS)}."
+            )
+        if not entry.note.strip():
+            errors.append(f"stays entry {entry.path!r} needs a note explaining why it stays.")
+    for handler in policy.handled_native_errors:
+        if not handler.note.strip():
+            errors.append(f"handled_native_errors entry {handler.location!r} needs a note.")
+    return tuple(errors)
diff --git a/scripts/python_boundary/constants.py b/scripts/python_boundary/constants.py
new file mode 100644
index 0000000000..ce4aa5a72d
--- /dev/null
+++ b/scripts/python_boundary/constants.py
@@ -0,0 +1,10 @@
+"""Stable Python boundary policy constants."""
+
+POLICY_FILE: str = "python-boundary.toml"
+SOURCE_ROOT: str = "src/sqlbuild"
+NATIVE_PACKAGE: str = "sqlbuild"
+NATIVE_MARKER: str = "native"
+SUPPRESS_NAME: str = "suppress"
+MODULE_SCOPE: str = "<module>"
+STAY_REASONS: tuple[str, ...] = ("user-code", "adapter", "executor", "public-api")
+ESCAPING_PATH_PARTS: frozenset[str] = frozenset({".", ".."})
diff --git a/scripts/python_boundary/main/__init__.py b/scripts/python_boundary/main/__init__.py
new file mode 100644
index 0000000000..d04c6ea786
--- /dev/null
+++ b/scripts/python_boundary/main/__init__.py
@@ -0,0 +1 @@
+"""Python boundary ratchet command entrypoints."""
diff --git a/scripts/python_boundary/main/check.py b/scripts/python_boundary/main/check.py
new file mode 100644
index 0000000000..cc1d5d08d3
--- /dev/null
+++ b/scripts/python_boundary/main/check.py
@@ -0,0 +1,60 @@
+"""Python boundary ratchet command entrypoint."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+from scripts.python_boundary._helpers.lines import count_lines, tracked_python_files
+from scripts.python_boundary._helpers.native_errors import find_native_error_handlers
+from scripts.python_boundary._helpers.policy import get_policy_errors, read_boundary_policy
+from scripts.python_boundary.constants import POLICY_FILE, SOURCE_ROOT
+from scripts.python_boundary.models import BoundaryPolicy, NativeErrorHandler
+
+
+def run_python_boundary_check(argv: list[str] | None = None) -> int:
+    """Fail when movable Python changes without the policy or swallows native failures."""
+    parser: argparse.ArgumentParser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, default=Path.cwd())
+    arguments: argparse.Namespace = parser.parse_args(argv)
+    root: Path = arguments.root
+    policy: BoundaryPolicy = read_boundary_policy(root / POLICY_FILE)
+    errors: list[str] = list(get_policy_errors(root=root, policy=policy))
+    tracked: tuple[Path, ...] = tracked_python_files(root=root, stays=policy.stays)
+    remaining: int = count_lines(tracked)
+    if remaining > policy.remaining_lines:
+        errors.append(
+            f"Movable Python grew to {remaining} lines (policy: {policy.remaining_lines}). "
+            "Move the new code to Rust, or justify it in python-boundary.toml."
+        )
+    elif remaining < policy.remaining_lines:
+        errors.append(
+            f"Movable Python shrank to {remaining} lines; "
+            f"set remaining_lines = {remaining} in {POLICY_FILE}."
+        )
+    reviewed: frozenset[str] = frozenset(
+        handler.location for handler in policy.handled_native_errors
+    )
+    found: tuple[NativeErrorHandler, ...] = find_native_error_handlers(
+        source=root / SOURCE_ROOT, paths=tracked
+    )
+    for handler in found:
+        if handler.location not in reviewed:
+            errors.append(
+                f"{SOURCE_ROOT}/{handler.location.split('::')[0]}:{handler.line} swallows a "
+                "native failure; re-raise it, or list the function in handled_native_errors "
+                "with a reason. Native failures must not fall back to Python."
+            )
+    present: frozenset[str] = frozenset(handler.location for handler in found)
+    errors.extend(
+        f"handled_native_errors entry {location!r} no longer matches any handler; remove it."
+        for location in sorted(reviewed - present)
+    )
+    if errors:
+        print("Python boundary check failed:", file=sys.stderr)
+        for error in errors:
+            print(f"- {error}", file=sys.stderr)
+        return 1
+    print(f"Python boundary is valid: {remaining} movable Python lines remain.")
+    return 0
diff --git a/scripts/python_boundary/models.py b/scripts/python_boundary/models.py
new file mode 100644
index 0000000000..0abf0f0f5b
--- /dev/null
+++ b/scripts/python_boundary/models.py
@@ -0,0 +1,39 @@
+"""Python boundary policy models."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class StayEntry:
+    """A module or package that stays Python, and why."""
+
+    path: str
+    reason: str
+    note: str
+
+
+@dataclass(frozen=True)
+class ReviewedNativeHandler:
+    """A reviewed function that handles a native failure without re-raising it."""
+
+    location: str
+    note: str
+
+
+@dataclass(frozen=True)
+class BoundaryPolicy:
+    """The committed remaining-line count, the Python that stays, and reviewed handlers."""
+
+    remaining_lines: int
+    stays: tuple[StayEntry, ...]
+    handled_native_errors: tuple[ReviewedNativeHandler, ...]
+
+
+@dataclass(frozen=True)
+class NativeErrorHandler:
+    """A function that calls the native module inside `try` and swallows the failure."""
+
+    location: str
+    line: int
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/adapters/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/adapters/_test_types.py
index 90a8610fee..0a2aac1999 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/adapters/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/adapters/_test_types.py
@@ -7,3 +7,10 @@ class ProjectLocalAdapterCliTestCase:
     command: tuple[str, ...]
     expected_stdout_fragment: str
     expected_return_code: int = 0
+
+
+@dataclass(frozen=True)
+class ProjectLocalRenderingTestCase:
+    description: str
+    expected_marker_rows: list[tuple[object, ...]]
+    expected_model_rows: list[tuple[object, ...]]
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/adapters/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/adapters/helpers.py
index 905da18bf9..2954fd46dc 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/adapters/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/adapters/helpers.py
@@ -32,3 +32,35 @@ def prepare_project_with_local_adapter(*, tmp_path: Path) -> Path:
             ),
         },
     )
+
+
+def prepare_project_with_rendering_adapter(*, tmp_path: Path) -> Path:
+    """Prepare a project whose project-local adapter overrides table rendering."""
+
+    return prepare_inline_project(
+        tmp_path=tmp_path,
+        project_name="rendering_adapter",
+        repo_files={
+            "sqlbuild_project.toml": (
+                'name = "rendering_adapter"\n'
+                'adapter = "duckdb_marked"\n\n'
+                "[connection]\n"
+                'database = "local.duckdb"\n'
+            ),
+            "adapters/duckdb_marked.py": (
+                "from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter\n\n\n"
+                "class DuckDbMarkedAdapter(DuckDbAdapter):\n"
+                "    adapter_name = 'duckdb_marked'\n\n"
+                "    def render_create_table_as(self, *, destination, sql):\n"
+                "        return (\n"
+                "            *super().render_create_table_as(destination=destination, sql=sql),\n"
+                "            'CREATE OR REPLACE TABLE main.rendering_override_marker AS '\n"
+                "            'SELECT 1 AS marked',\n"
+                "        )\n"
+            ),
+            "models/orders.sql": (
+                'MODEL (description "Orders", materialized table);\n\n'
+                "SELECT 1 AS order_id, 'shipped' AS status\n"
+            ),
+        },
+    )
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/adapters/test_project_local_adapters.py b/tests/e2e/src/sqlbuild/cli/commands/main/adapters/test_project_local_adapters.py
index 0747401a59..14c807538c 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/adapters/test_project_local_adapters.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/adapters/test_project_local_adapters.py
@@ -7,11 +7,13 @@
 
 from tests.e2e.src.sqlbuild.cli.commands.main.adapters._test_types import (
     ProjectLocalAdapterCliTestCase,
+    ProjectLocalRenderingTestCase,
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.adapters.helpers import (
     prepare_project_with_local_adapter,
+    prepare_project_with_rendering_adapter,
 )
-from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import query_duckdb, run_sqb
 
 
 @pytest.mark.parametrize(
@@ -38,3 +40,36 @@ def test_given_project_local_adapter_when_running_query_then_uses_local_adapter(
 
     assert result.returncode == test_case.expected_return_code, result.stdout + result.stderr
     assert test_case.expected_stdout_fragment in result.stdout
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ProjectLocalRenderingTestCase(
+            description="table DDL comes from the overriding project local adapter",
+            expected_marker_rows=[(1,)],
+            expected_model_rows=[(1, "shipped")],
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_project_local_rendering_override_when_building_then_override_renders_ddl(
+    tmp_path: Path,
+    test_case: ProjectLocalRenderingTestCase,
+) -> None:
+    project_dir: Path = prepare_project_with_rendering_adapter(tmp_path=tmp_path)
+
+    result: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "build"), project_dir=project_dir, working_dir=project_dir
+    )
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    database: Path = project_dir / "local.duckdb"
+    assert (
+        query_duckdb(db_path=database, sql="SELECT marked FROM main.rendering_override_marker")
+        == test_case.expected_marker_rows
+    )
+    assert (
+        query_duckdb(db_path=database, sql="SELECT order_id, status FROM main.orders")
+        == test_case.expected_model_rows
+    )
diff --git a/tests/unit/scripts/python_boundary/__init__.py b/tests/unit/scripts/python_boundary/__init__.py
new file mode 100644
index 0000000000..16c47c2972
--- /dev/null
+++ b/tests/unit/scripts/python_boundary/__init__.py
@@ -0,0 +1 @@
+"""Python boundary ratchet tests."""
diff --git a/tests/unit/scripts/python_boundary/main/__init__.py b/tests/unit/scripts/python_boundary/main/__init__.py
new file mode 100644
index 0000000000..8a1bccca31
--- /dev/null
+++ b/tests/unit/scripts/python_boundary/main/__init__.py
@@ -0,0 +1 @@
+"""Python boundary command tests."""
diff --git a/tests/unit/scripts/python_boundary/main/_test_types.py b/tests/unit/scripts/python_boundary/main/_test_types.py
new file mode 100644
index 0000000000..0ae8745de3
--- /dev/null
+++ b/tests/unit/scripts/python_boundary/main/_test_types.py
@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class PythonBoundaryTestCase:
+    description: str
+    source_files: dict[str, str]
+    policy: str
+    expected_messages: tuple[str, ...]
diff --git a/tests/unit/scripts/python_boundary/main/helpers.py b/tests/unit/scripts/python_boundary/main/helpers.py
new file mode 100644
index 0000000000..639cc6f8ad
--- /dev/null
+++ b/tests/unit/scripts/python_boundary/main/helpers.py
@@ -0,0 +1,12 @@
+from pathlib import Path
+
+from tests.unit.scripts.python_boundary.main._test_types import PythonBoundaryTestCase
+
+
+def write_boundary_project(root: Path, test_case: PythonBoundaryTestCase) -> None:
+    """Write a source tree under src/sqlbuild and its python-boundary.toml."""
+    for relative_path, contents in test_case.source_files.items():
+        path: Path = root / "src" / "sqlbuild" / relative_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(contents, encoding="utf-8")
+    (root / "python-boundary.toml").write_text(test_case.policy, encoding="utf-8")
diff --git a/tests/unit/scripts/python_boundary/main/test_check.py b/tests/unit/scripts/python_boundary/main/test_check.py
new file mode 100644
index 0000000000..cd5071b1a3
--- /dev/null
+++ b/tests/unit/scripts/python_boundary/main/test_check.py
@@ -0,0 +1,189 @@
+from pathlib import Path
+
+import pytest
+
+from scripts.python_boundary.main.check import run_python_boundary_check
+from tests.unit.scripts.python_boundary.main._test_types import PythonBoundaryTestCase
+from tests.unit.scripts.python_boundary.main.helpers import write_boundary_project
+
+SWALLOWED: str = (
+    "from sqlbuild import _native\n"
+    "\n"
+    "\n"
+    "def digest(text):\n"
+    "    try:\n"
+    "        return _native.content_digest([text])\n"
+    "    except ValueError:\n"
+    "        return None\n"
+)
+RERAISED: str = (
+    "from sqlbuild._native import content_digest\n"
+    "\n"
+    "\n"
+    "def digest(text):\n"
+    "    try:\n"
+    "        return content_digest([text])\n"
+    "    except ValueError as error:\n"
+    "        raise RuntimeError(text) from error\n"
+)
+CONDITIONAL: str = (
+    "from sqlbuild._native import content_digest\n"
+    "\n"
+    "\n"
+    "def digest(text, strict):\n"
+    "    try:\n"
+    "        return content_digest([text])\n"
+    "    except ValueError:\n"
+    "        if not strict:\n"
+    "            return None\n"
+    "        raise\n"
+)
+SUPPRESSED: str = (
+    "import contextlib\n"
+    "\n"
+    "from sqlbuild.compiler.graph.main._native_project_graph import native_project_graph\n"
+    "\n"
+    "\n"
+    "class Graph:\n"
+    "    def build(self, project):\n"
+    "        with contextlib.suppress(ValueError):\n"
+    "            return native_project_graph(project)\n"
+    "        return None\n"
+)
+STAYS: str = '[[stays]]\npath = "executor"\nreason = "executor"\nnote = "Runs plans."\n'
+REVIEWED: str = (
+    '[[handled_native_errors]]\nlocation = "compiler/keys.py::digest"\nnote = "No key."\n'
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        PythonBoundaryTestCase(
+            description="staying modules are not counted",
+            source_files={"compiler/plan.py": "a = 1\nb = 2\n", "executor/run.py": "c = 3\n"},
+            policy="remaining_lines = 2\n" + STAYS,
+            expected_messages=("2 movable Python lines remain",),
+        ),
+        PythonBoundaryTestCase(
+            description="names that only contain the letters native are not native calls",
+            source_files={
+                "compiler/plan.py": (
+                    "from sqlbuild.compiler.alternatives import alternative_plan\n"
+                    "\n"
+                    "\n"
+                    "def plan():\n"
+                    "    try:\n"
+                    "        return alternative_plan()\n"
+                    "    except ValueError:\n"
+                    "        return None\n"
+                )
+            },
+            policy="remaining_lines = 8\n",
+            expected_messages=("8 movable Python lines remain",),
+        ),
+        PythonBoundaryTestCase(
+            description="re-raised native failures are allowed",
+            source_files={"compiler/keys.py": RERAISED},
+            policy="remaining_lines = 8\n",
+            expected_messages=("8 movable Python lines remain",),
+        ),
+        PythonBoundaryTestCase(
+            description="a reviewed native error handler is allowed",
+            source_files={"compiler/keys.py": SWALLOWED},
+            policy="remaining_lines = 8\n" + REVIEWED,
+            expected_messages=("8 movable Python lines remain",),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_policy_matching_source_when_checking_then_passes(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], test_case: PythonBoundaryTestCase
+) -> None:
+    write_boundary_project(tmp_path, test_case)
+
+    assert run_python_boundary_check(["--root", str(tmp_path)]) == 0
+    output: str = capsys.readouterr().out
+    assert all(message in output for message in test_case.expected_messages)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        PythonBoundaryTestCase(
+            description="movable Python grew",
+            source_files={"compiler/plan.py": "a = 1\nb = 2\nc = 3\n"},
+            policy="remaining_lines = 2\n",
+            expected_messages=("Movable Python grew to 3 lines (policy: 2)",),
+        ),
+        PythonBoundaryTestCase(
+            description="movable Python shrank without lowering the baseline",
+            source_files={"compiler/plan.py": "a = 1\n"},
+            policy="remaining_lines = 2\n",
+            expected_messages=("set remaining_lines = 1",),
+        ),
+        PythonBoundaryTestCase(
+            description="an unreviewed native failure is swallowed",
+            source_files={"compiler/keys.py": SWALLOWED},
+            policy="remaining_lines = 8\n",
+            expected_messages=("src/sqlbuild/compiler/keys.py:7 swallows a native failure",),
+        ),
+        PythonBoundaryTestCase(
+            description="a handler that raises only conditionally",
+            source_files={"compiler/keys.py": CONDITIONAL},
+            policy="remaining_lines = 10\n",
+            expected_messages=("src/sqlbuild/compiler/keys.py:7 swallows a native failure",),
+        ),
+        PythonBoundaryTestCase(
+            description="a native wrapper call inside suppress in a method",
+            source_files={"compiler/graph.py": SUPPRESSED},
+            policy=(
+                "remaining_lines = 10\n"
+                '[[handled_native_errors]]\nlocation = "compiler/graph.py::build"\nnote = "Old."\n'
+            ),
+            expected_messages=(
+                "src/sqlbuild/compiler/graph.py:8 swallows a native failure",
+                "'compiler/graph.py::build' no longer matches any handler",
+            ),
+        ),
+        PythonBoundaryTestCase(
+            description="a staying path escapes the source root",
+            source_files={"compiler/plan.py": "a = 1\n"},
+            policy=(
+                'remaining_lines = 0\n[[stays]]\npath = ".."\nreason = "executor"\nnote = "No."\n'
+            ),
+            expected_messages=("stays entry '..' must name a module under src/sqlbuild.",),
+        ),
+        PythonBoundaryTestCase(
+            description="a reviewed handler no longer exists",
+            source_files={"compiler/keys.py": RERAISED},
+            policy="remaining_lines = 8\n" + REVIEWED,
+            expected_messages=("'compiler/keys.py::digest' no longer matches any handler",),
+        ),
+        PythonBoundaryTestCase(
+            description="a staying module is missing, unexplained and has an unknown reason",
+            source_files={"compiler/plan.py": "a = 1\n"},
+            policy=(
+                'remaining_lines = 1\n[[stays]]\npath = "planner"\nreason = "too hard"\nnote = ""\n'
+            ),
+            expected_messages=(
+                "stays entry 'planner' does not exist under src/sqlbuild.",
+                "stays entry 'planner' has reason 'too hard'",
+                "stays entry 'planner' needs a note explaining why it stays.",
+            ),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_policy_violation_when_checking_then_fails_with_reason(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], test_case: PythonBoundaryTestCase
+) -> None:
+    write_boundary_project(tmp_path, test_case)
+
+    assert run_python_boundary_check(["--root", str(tmp_path)]) == 1
+    errors: str = capsys.readouterr().err
+    assert all(message in errors for message in test_case.expected_messages)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])