From d4eabf59d5c619d501aab0497d03acdd510a2bbc Mon Sep 17 00:00:00 2001
From: Andres Contreras <andres.contreras@soon.es>
Date: Wed, 14 Oct 2026 09:45:32 -0700
Subject: [PATCH 1/5] Confine SQL execution to registered data snapshots

---
 .../core/services/execution/ast_classifier.py | 125 +++++++++-
 .../services/execution/duckdb_executor.py     |  88 +++++--
 .../core/services/execution/scope_guard.py    |  10 +-
 .../web/controllers/sql_execute_controller.py |  16 +-
 tests/unit/test_query_read_boundary.py        | 230 ++++++++++++++++++
 tests/unit/test_query_service.py              |  37 +++
 tests/unit/test_sql_read_boundary.py          | 120 +++++++++
 7 files changed, 601 insertions(+), 25 deletions(-)
 create mode 100644 tests/unit/test_query_read_boundary.py
 create mode 100644 tests/unit/test_sql_read_boundary.py

diff --git a/src/flyquery/core/services/execution/ast_classifier.py b/src/flyquery/core/services/execution/ast_classifier.py
index a5f9cd70..1ed9ffb2 100644
--- a/src/flyquery/core/services/execution/ast_classifier.py
+++ b/src/flyquery/core/services/execution/ast_classifier.py
@@ -1,4 +1,5 @@
 # Copyright 2024-2026 Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -26,6 +27,103 @@
 
 import sqlglot
 import sqlglot.expressions  # noqa: F401 — ensures exp alias is populated on sqlglot module
+from sqlglot.optimizer.scope import Scope, traverse_scope
+
+_PURE_FUNCTIONS = frozenset(
+    {
+        "SUM",
+        "COUNT",
+        "AVG",
+        "MIN",
+        "MAX",
+        "COUNT_IF",
+        "APPROX_DISTINCT",
+        "STDDEV",
+        "STDDEV_POP",
+        "STDDEV_SAMP",
+        "VARIANCE",
+        "VAR_POP",
+        "VAR_SAMP",
+        "MEDIAN",
+        "BOOL_AND",
+        "BOOL_OR",
+        "LOGICAL_AND",
+        "LOGICAL_OR",
+        "ANY_VALUE",
+        "COALESCE",
+        "NULLIF",
+        "CAST",
+        "TRY_CAST",
+        "CASE",
+        "IF",
+        "DATE_TRUNC",
+        "TIMESTAMP_TRUNC",
+        "DATE_DIFF",
+        "DATEDIFF",
+        "EXTRACT",
+        "YEAR",
+        "MONTH",
+        "DAY",
+        "DAY_OF_MONTH",
+        "DAY_OF_WEEK",
+        "DAY_OF_YEAR",
+        "WEEK",
+        "WEEK_OF_YEAR",
+        "QUARTER",
+        "TIME_TO_STR",
+        "STR_TO_TIME",
+        "CURRENT_DATE",
+        "CURRENT_TIMESTAMP",
+        "CURRENT_TIME",
+        "ABS",
+        "ROUND",
+        "FLOOR",
+        "CEIL",
+        "CEILING",
+        "POWER",
+        "POW",
+        "SQRT",
+        "LOG",
+        "LN",
+        "EXP",
+        "SIGN",
+        "GREATEST",
+        "LEAST",
+        "LOWER",
+        "UPPER",
+        "LENGTH",
+        "CHAR_LENGTH",
+        "TRIM",
+        "LTRIM",
+        "RTRIM",
+        "CONCAT",
+        "CONCAT_WS",
+        "SUBSTRING",
+        "STR_POSITION",
+        "REPLACE",
+        "REGEXP_REPLACE",
+        "REGEXP_EXTRACT",
+        "REGEXP_LIKE",
+        "SPLIT",
+        "SPLIT_PART",
+        "STARTS_WITH",
+        "ENDS_WITH",
+        "ARRAY",
+        "ARRAY_SIZE",
+        "STRUCT",
+        "STRUCT_EXTRACT",
+        "ROW_NUMBER",
+        "RANK",
+        "DENSE_RANK",
+        "LAG",
+        "LEAD",
+        "FIRST_VALUE",
+        "LAST_VALUE",
+        "NTH_VALUE",
+        "GROUP_CONCAT",
+        "STRING_AGG",
+    }
+)
 
 
 @dataclass(frozen=True)
@@ -37,6 +135,7 @@ class AstClassification:
     table_refs: tuple[str, ...]  # unqualified table names referenced in the statement
     column_refs: tuple[str, ...]  # column names referenced in the statement
     has_subquery: bool
+    unsupported_sql: bool = False
 
 
 class AstClassifier:
@@ -76,10 +175,31 @@ def classify(self, sql: str) -> AstClassification:
         # pyright does not unify with the public ``Expression`` base below.
         kind = self._kind(stmt)  # pyright: ignore[reportArgumentType]
 
-        # Collect table refs — skip anonymous subquery aliases
-        tables = tuple(sorted({t.name for t in stmt.find_all(sqlglot.expressions.Table) if t.name}))
+        exp = sqlglot.expressions
+        try:
+            cte_refs = {
+                id(node)
+                for scope in traverse_scope(stmt)
+                for node, source in scope.selected_sources.values()
+                if isinstance(source, Scope)
+            }
+        except Exception:
+            return AstClassification("UNKNOWN", single, (), (), False, unsupported_sql=True)
+        sources = list(stmt.find_all(exp.Table))
+        tables = tuple(sorted({t.name for t in sources if t.name and id(t) not in cte_refs}))
         columns = tuple(sorted({c.name for c in stmt.find_all(sqlglot.expressions.Column) if c.name}))
         has_subquery = bool(list(stmt.find_all(sqlglot.expressions.Subquery)))
+        # Typed parser nodes can also read files; admission is by reviewed function name, never node type alone.
+        unsupported = any(
+            (function.name.upper() if isinstance(function, exp.Anonymous) else function.sql_name())
+            not in _PURE_FUNCTIONS
+            for function in stmt.find_all(exp.Func)
+        ) or any(
+            not isinstance(table.this, exp.Identifier)
+            or table.catalog
+            or (table.db and table.db.lower() != "main")
+            for table in sources
+        )
 
         return AstClassification(
             classification=kind,
@@ -87,6 +207,7 @@ def classify(self, sql: str) -> AstClassification:
             table_refs=tables,
             column_refs=columns,
             has_subquery=has_subquery,
+            unsupported_sql=bool(unsupported),
         )
 
     @staticmethod
diff --git a/src/flyquery/core/services/execution/duckdb_executor.py b/src/flyquery/core/services/execution/duckdb_executor.py
index 8ee0a72f..589afdc2 100644
--- a/src/flyquery/core/services/execution/duckdb_executor.py
+++ b/src/flyquery/core/services/execution/duckdb_executor.py
@@ -1,4 +1,5 @@
 # Copyright 2024-2026 Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -53,6 +54,8 @@
 import tempfile
 from dataclasses import dataclass
 
+from flyquery.core.services.execution.ast_classifier import AstClassifier
+
 
 @dataclass(frozen=True)
 class ExecutionResult:
@@ -110,7 +113,7 @@ class DuckDBExecutor:
     4. Return a :class:`DmlMutationResult` with the new Parquet bytes.
 
     :param settings: :class:`FlyquerySettings` — reads ``duckdb_memory_limit``,
-        ``default_statement_timeout_ms``, and ``default_row_cap``.
+        and ``default_row_cap``. Statement timeouts are not enforced by this executor.
     """
 
     def __init__(self, settings) -> None:
@@ -133,7 +136,7 @@ async def execute(
         :return: :class:`ExecutionResult` for SELECT, :class:`DmlMutationResult` for DML,
             or :class:`ExecutionError` on failure
         """
-        return await asyncio.to_thread(self._sync, sql, attached_tables, derived_tables or {})
+        return await asyncio.to_thread(self._sync, sql, dict(attached_tables), dict(derived_tables or {}))
 
     def _sync(
         self,
@@ -142,15 +145,57 @@ def _sync(
         derived_tables: dict[str, str],
     ) -> ExecutionResult | DmlMutationResult | ExecutionError:
         """Synchronous execution body run in a thread pool worker."""
+        classification = AstClassifier().classify(sql)
+        if any(attached_tables.get(name) != path for name, path in derived_tables.items()):
+            return ExecutionError(message="DERIVED snapshot is not registered for this execution")
+        # Actor scopes remain upstream; this boundary confines SQL to the supplied registered snapshots.
+        if (
+            not classification.single_statement
+            or classification.classification not in {"SELECT", "INSERT", "UPDATE", "DELETE"}
+            or classification.unsupported_sql
+        ):
+            return ExecutionError(message="unsupported SQL statement, function, or table source")
+        if any(name not in attached_tables for name in classification.table_refs):
+            return ExecutionError(message="SQL source is not a registered table")
+        if classification.classification != "SELECT" and any(
+            name not in derived_tables for name in classification.table_refs
+        ):
+            return ExecutionError(message="DML source is not a registered DERIVED table")
         try:
             import duckdb
         except ImportError as exc:
             return ExecutionError(message=f"duckdb not installed: {exc}")
 
-        if derived_tables:
+        if classification.classification in {"INSERT", "UPDATE", "DELETE"}:
+            if len(derived_tables) != 1:
+                return ExecutionError(message="DML requires exactly one registered DERIVED target")
             return self._sync_dml(sql, attached_tables, derived_tables, duckdb)
         return self._sync_select(sql, attached_tables, duckdb)
 
+    def _connection(self, duckdb, paths: list[str]):
+        conn = duckdb.connect(
+            ":memory:", config={"autoload_known_extensions": False, "autoinstall_known_extensions": False}
+        )
+        try:
+            # DuckDB freezes allowed_paths when external access is disabled, so finish the exact allowlist first.
+            conn.execute("SET allowed_paths = ?", [sorted(set(paths))])
+            conn.execute("SET enable_external_access=false")
+            conn.execute("SET memory_limit = ?", [self._memory_limit])
+            conn.execute("SET threads=2")
+            conn.execute("SET lock_configuration=true")
+            return conn
+        except Exception:
+            conn.close()
+            raise
+
+    @staticmethod
+    def _identifier(value: str) -> str:
+        return '"' + value.replace('"', '""') + '"'
+
+    @staticmethod
+    def _literal(value: str) -> str:
+        return "'" + value.replace("'", "''") + "'"
+
     def _sync_select(
         self,
         sql: str,
@@ -158,14 +203,15 @@ def _sync_select(
         duckdb,  # noqa: ANN001
     ) -> ExecutionResult | ExecutionError:
         """Execute a SELECT and return rows."""
-        conn = duckdb.connect(":memory:")
+        conn = None
         try:
-            conn.execute(f"SET memory_limit='{self._memory_limit}'")
-            conn.execute("SET threads=2")
+            conn = self._connection(duckdb, list(attached_tables.values()))
 
             # Create a VIEW for each parquet-backed table.
             for name, parquet_path in attached_tables.items():
-                conn.execute(f"CREATE VIEW {name} AS SELECT * FROM read_parquet('{parquet_path}')")
+                conn.execute(
+                    f"CREATE VIEW {self._identifier(name)} AS SELECT * FROM read_parquet({self._literal(parquet_path)})"
+                )
 
             # Wrap in LIMIT row_cap+1 to detect truncation.
             wrapped = f"SELECT * FROM ({sql.rstrip(';')}) AS q LIMIT {self._row_cap + 1}"
@@ -186,7 +232,8 @@ def _sync_select(
         except Exception as exc:  # noqa: BLE001
             return ExecutionError(message=str(exc))
         finally:
-            conn.close()
+            if conn is not None:
+                conn.close()
 
     def _sync_dml(
         self,
@@ -211,35 +258,39 @@ def _sync_dml(
         target_name = next(iter(derived_tables))
         current_parquet_path = derived_tables[target_name]
 
-        conn = duckdb.connect(":memory:")
+        conn = None
         tmp_path: str | None = None
 
         try:
-            conn.execute(f"SET memory_limit='{self._memory_limit}'")
-            conn.execute("SET threads=2")
+            fd, tmp_path = tempfile.mkstemp(suffix=".parquet")
+            os.close(fd)
+            conn = self._connection(duckdb, [*attached_tables.values(), current_parquet_path, tmp_path])
 
             # Provide read-only views for all non-target tables.
             for name, path in attached_tables.items():
                 if name == target_name:
                     continue
-                conn.execute(f"CREATE VIEW {name} AS SELECT * FROM read_parquet('{path}')")
+                conn.execute(
+                    f"CREATE VIEW {self._identifier(name)} AS SELECT * FROM read_parquet({self._literal(path)})"
+                )
 
             # Load the DERIVED table as a mutable TABLE (not a VIEW).
             conn.execute(
-                f"CREATE TABLE {target_name} AS SELECT * FROM read_parquet('{current_parquet_path}')"
+                f"CREATE TABLE {self._identifier(target_name)} AS SELECT * FROM read_parquet({self._literal(current_parquet_path)})"
             )
 
             # Execute the DML statement.
             conn.execute(sql.rstrip(";"))
 
             # Count rows in the mutated table.
-            count_cursor = conn.execute(f"SELECT COUNT(*) FROM {target_name}")
+            count_cursor = conn.execute(f"SELECT COUNT(*) FROM {self._identifier(target_name)}")
             rows_affected: int = count_cursor.fetchone()[0]  # type: ignore[index]
 
             # Export the mutated table to a temp Parquet file.
-            fd, tmp_path = tempfile.mkstemp(suffix=".parquet")
-            os.close(fd)
-            conn.execute(f"COPY {target_name} TO '{tmp_path}' (FORMAT PARQUET, COMPRESSION SNAPPY)")
+            # The destination is a new owned scratch file, so no extra atomic-replacement path is needed.
+            conn.execute(
+                f"COPY {self._identifier(target_name)} TO {self._literal(tmp_path)} (FORMAT PARQUET, COMPRESSION SNAPPY, USE_TMP_FILE false)"
+            )
 
             with open(tmp_path, "rb") as fh:
                 new_parquet_bytes = fh.read()
@@ -253,6 +304,7 @@ def _sync_dml(
         except Exception as exc:  # noqa: BLE001
             return ExecutionError(message=str(exc))
         finally:
-            conn.close()
+            if conn is not None:
+                conn.close()
             if tmp_path and os.path.exists(tmp_path):
                 os.unlink(tmp_path)
diff --git a/src/flyquery/core/services/execution/scope_guard.py b/src/flyquery/core/services/execution/scope_guard.py
index d3b70af3..cdd81ee4 100644
--- a/src/flyquery/core/services/execution/scope_guard.py
+++ b/src/flyquery/core/services/execution/scope_guard.py
@@ -1,4 +1,5 @@
 # Copyright 2024-2026 Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -64,6 +65,8 @@ def check(
         # 2. DDL never allowed via /query
         if classification.classification == "DDL":
             raise ScopeGuardError("DDL not allowed via /query (use internal :derive endpoint)")
+        if classification.classification == "UNKNOWN" or classification.unsupported_sql:
+            raise ScopeGuardError("unsupported SQL statement, function, or table source")
 
         # 3. DML rules
         if classification.classification in ("INSERT", "UPDATE", "DELETE"):
@@ -82,12 +85,17 @@ def check(
             # 4. SELECT: require read scope
             if not (scopes & {"flyquery.query:read", "flyquery.sql:execute", "*"}):
                 raise ScopeGuardError("missing flyquery.query:read scope (or flyquery.sql:execute / *)")
+            if any(
+                table_kinds_by_name.get(tbl) not in {"UPLOADED", "DERIVED"}
+                for tbl in classification.table_refs
+            ):
+                raise ScopeGuardError("SQL source is not a registered table")
 
         # 5. Dataset allowlist
         if dataset_allowlist is not None:
             for tbl in classification.table_refs:
                 ds = dataset_of_table.get(tbl)
-                if ds is not None and ds not in dataset_allowlist:
+                if ds is None or ds not in dataset_allowlist:
                     raise ScopeGuardError(
                         f"table {tbl!r} (dataset {ds!r}) is not in the caller's dataset allowlist"
                     )
diff --git a/src/flyquery/web/controllers/sql_execute_controller.py b/src/flyquery/web/controllers/sql_execute_controller.py
index 12bab9be..cf36af47 100644
--- a/src/flyquery/web/controllers/sql_execute_controller.py
+++ b/src/flyquery/web/controllers/sql_execute_controller.py
@@ -1,4 +1,5 @@
 # Copyright 2024-2026 Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -155,8 +156,11 @@ async def execute(
         await self._assert_direct_sql_enabled(workspace_id)
 
         ast = self._ast_classifier.classify(body.sql)
-        table_kinds: dict[str, str] = {}
-        dataset_of_table: dict[str, str] = {}
+        registered = await self._table_repo.list_kinds_for_dataset_by_names(
+            body.dataset_id, list(ast.table_refs)
+        )
+        table_kinds = {row["name"]: row["kind"] for row in registered}
+        dataset_of_table = {name: str(body.dataset_id) for name in table_kinds}
 
         scope_error: str | None = None
         try:
@@ -359,15 +363,19 @@ async def _stream_events(
         await self._assert_direct_sql_enabled(workspace_id)
 
         ast = self._ast_classifier.classify(request.sql)
+        registered = await self._table_repo.list_kinds_for_dataset_by_names(
+            request.dataset_id, list(ast.table_refs)
+        )
+        table_kinds = {row["name"]: row["kind"] for row in registered}
 
         scope_error: str | None = None
         try:
             self._scope_guard.check(
                 classification=ast,
                 scopes=_DEFAULT_SCOPES,
-                table_kinds_by_name={},
+                table_kinds_by_name=table_kinds,
                 dataset_allowlist=None,
-                dataset_of_table={},
+                dataset_of_table={name: str(request.dataset_id) for name in table_kinds},
             )
         except ScopeGuardError as exc:
             scope_error = str(exc)
diff --git a/tests/unit/test_query_read_boundary.py b/tests/unit/test_query_read_boundary.py
new file mode 100644
index 00000000..f1ba0a0b
--- /dev/null
+++ b/tests/unit/test_query_read_boundary.py
@@ -0,0 +1,230 @@
+# Copyright 2026 Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Native read-boundary regressions using only owned fictional files."""
+
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from threading import Thread
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from flyquery.core.services.execution.ast_classifier import AstClassification, AstClassifier
+from flyquery.core.services.execution.duckdb_executor import (
+    DmlMutationResult,
+    DuckDBExecutor,
+    ExecutionError,
+    ExecutionResult,
+)
+from flyquery.core.services.execution.scope_guard import ScopeGuard, ScopeGuardError
+
+
+class Settings:
+    duckdb_memory_limit = "256MB"
+    default_row_cap = 100
+    default_statement_timeout_ms = 5000
+
+
+def check(sql: str, kinds=None, allowlist=None, datasets=None) -> None:
+    ScopeGuard().check(
+        classification=AstClassifier().classify(sql),
+        scopes={"flyquery.query:read", "flyquery.derived:write"},
+        table_kinds_by_name=kinds or {},
+        dataset_allowlist=allowlist,
+        dataset_of_table=datasets or {},
+    )
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT * FROM read_csv_auto('/owned/unregistered.csv')",
+        "SELECT * FROM read_csv('/owned/unregistered.csv')",
+        "SELECT * FROM read_parquet('/owned/unregistered.parquet')",
+        "SELECT * FROM read_json_auto('/owned/unregistered.json')",
+        "SELECT * FROM read_blob('/owned/unregistered.txt')",
+        "SELECT * FROM glob('/owned/*')",
+        "SELECT * FROM query('SELECT 1')",
+        "SELECT * FROM query_table('orders')",
+        "SELECT * FROM read_csv_auto('http://127.0.0.1:49152/unregistered.csv')",
+        "SELECT current_setting('s3_secret_access_key')",
+        "SELECT getenv('SYNTHETIC_FIXTURE_VALUE')",
+        "SELECT * FROM duckdb_secrets()",
+        "SELECT * FROM (SELECT * FROM read_csv_auto('/owned/unregistered.csv')) x",
+        "WITH x AS (SELECT * FROM read_csv_auto('/owned/unregistered.csv')) SELECT * FROM x",
+        "SELECT 1 UNION ALL SELECT * FROM read_csv_auto('/owned/unregistered.csv')",
+    ],
+)
+def test_unadmitted_sources_fail_closed_before_execution(sql: str) -> None:
+    with pytest.raises(ScopeGuardError):
+        check(sql, {"orders": "UPLOADED"}, {"owned"}, {"orders": "owned"})
+
+
+@pytest.mark.parametrize(
+    "sql", ["PRAGMA version", "SET threads=4", "SELECT * FROM '/owned/unregistered.parquet'"]
+)
+def test_unknown_statements_and_unregistered_tables_have_no_read_authority(sql: str) -> None:
+    with pytest.raises(ScopeGuardError):
+        check(sql)
+
+
+def test_missing_dataset_mapping_cannot_bypass_an_empty_allowlist() -> None:
+    with pytest.raises(ScopeGuardError):
+        check("SELECT * FROM orders", {"orders": "UPLOADED"}, set(), {})
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT SUM(total) FROM orders",
+        "WITH x AS (SELECT SUM(total) AS total FROM orders) SELECT total FROM x",
+        "SELECT o.total FROM orders o JOIN orders p ON o.id=p.id",
+    ],
+)
+def test_registered_sources_keep_actual_dataset_authority(sql: str) -> None:
+    check(sql, {"orders": "UPLOADED"}, {"owned"}, {"orders": "owned"})
+    with pytest.raises(ScopeGuardError):
+        check(sql, {"orders": "UPLOADED"}, {"other"}, {"orders": "owned"})
+
+
+def test_constant_cte_does_not_invent_a_dataset_grant() -> None:
+    ast = AstClassifier().classify("WITH x AS (SELECT 1 AS n) SELECT n FROM x")
+    assert ast.table_refs == ()
+    check("WITH x AS (SELECT 1 AS n) SELECT n FROM x", allowlist=set())
+
+
+@pytest.mark.asyncio
+async def test_executor_itself_refuses_an_unregistered_owned_csv(tmp_path: Path) -> None:
+    file = tmp_path / "unregistered.csv"
+    file.write_text("evidence\nsynthetic-unregistered-file\n")
+    result = await DuckDBExecutor(Settings()).execute(f"SELECT * FROM read_csv_auto('{file}')", {})
+    assert isinstance(result, ExecutionError)
+
+
+@pytest.mark.asyncio
+async def test_executor_itself_refuses_implicit_file_tables(tmp_path: Path) -> None:
+    file = tmp_path / "unregistered.parquet"
+    pq.write_table(pa.Table.from_pylist([{"evidence": "synthetic-unregistered-file"}]), file)
+    result = await DuckDBExecutor(Settings()).execute(f"SELECT * FROM '{file}'", {})
+    assert isinstance(result, ExecutionError)
+
+
+@pytest.mark.asyncio
+async def test_native_path_fence_survives_a_parser_omission(tmp_path: Path, monkeypatch) -> None:
+    file = tmp_path / "unregistered.csv"
+    file.write_text("evidence\nsynthetic-unregistered-file\n")
+    monkeypatch.setattr(
+        AstClassifier, "classify", lambda self, sql: AstClassification("SELECT", True, (), (), False)
+    )
+    result = await DuckDBExecutor(Settings()).execute(f"SELECT * FROM read_csv_auto('{file}')", {})
+    assert isinstance(result, ExecutionError)
+
+
+@pytest.mark.asyncio
+async def test_registered_apostrophe_path_and_quoted_identifier_are_data_not_bootstrap_sql(
+    tmp_path: Path,
+) -> None:
+    file = tmp_path / "registered's snapshot.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 25}]), file)
+    result = await DuckDBExecutor(Settings()).execute(
+        'SELECT total FROM "orders""qa"', {'orders"qa': str(file)}
+    )
+    assert isinstance(result, ExecutionResult)
+    assert result.rows == [{"total": 25}]
+
+
+@pytest.mark.asyncio
+async def test_registered_cte_and_reviewed_pure_builtins_preserve_rows(tmp_path: Path) -> None:
+    file = tmp_path / "orders.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 10}, {"id": 2, "total": 20}]), file)
+    result = await DuckDBExecutor(Settings()).execute(
+        "WITH x AS (SELECT id, total FROM orders) SELECT SUM(total) AS total, COUNT(*) AS n, ROUND(AVG(total)) AS average FROM x",
+        {"orders": str(file)},
+    )
+    assert isinstance(result, ExecutionResult)
+    assert result.rows == [{"total": 30, "n": 2, "average": 15.0}]
+
+
+@pytest.mark.asyncio
+async def test_derived_copy_on_write_stays_scoped_and_preserves_original(tmp_path: Path) -> None:
+    file = tmp_path / "derived.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 10}]), file)
+    original = file.read_bytes()
+    result = await DuckDBExecutor(Settings()).execute(
+        "INSERT INTO derived VALUES (2, 20)",
+        {"derived": str(file)},
+        {"derived": str(file)},
+    )
+    assert isinstance(result, DmlMutationResult)
+    assert pq.read_table(pa.BufferReader(result.new_parquet_bytes)).to_pylist() == [
+        {"id": 1, "total": 10},
+        {"id": 2, "total": 20},
+    ]
+    assert file.read_bytes() == original
+
+
+@pytest.mark.asyncio
+async def test_derived_mutation_cannot_import_an_unregistered_file(tmp_path: Path) -> None:
+    file = tmp_path / "derived.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 10}]), file)
+    other = tmp_path / "unregistered.csv"
+    other.write_text("id,total\n2,20\n")
+    result = await DuckDBExecutor(Settings()).execute(
+        f"INSERT INTO derived SELECT * FROM read_csv_auto('{other}')",
+        {"derived": str(file)},
+        {"derived": str(file)},
+    )
+    assert isinstance(result, ExecutionError)
+
+
+@pytest.mark.asyncio
+async def test_selecting_a_derived_snapshot_is_read_only(tmp_path: Path) -> None:
+    file = tmp_path / "derived.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 10}]), file)
+    result = await DuckDBExecutor(Settings()).execute(
+        "SELECT * FROM derived", {"derived": str(file)}, {"derived": str(file)}
+    )
+    assert isinstance(result, ExecutionResult)
+    assert result.rows == [{"id": 1, "total": 10}]
+
+
+@pytest.mark.asyncio
+async def test_reviewed_boolean_aggregate_is_not_mistaken_for_an_external_function(tmp_path: Path) -> None:
+    file = tmp_path / "orders.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 10}, {"id": 2, "total": 20}]), file)
+    result = await DuckDBExecutor(Settings()).execute(
+        "SELECT bool_and(total > 0) AS positive FROM orders", {"orders": str(file)}
+    )
+    assert isinstance(result, ExecutionResult)
+    assert result.rows == [{"positive": True}]
+
+
+@pytest.mark.asyncio
+async def test_unregistered_loopback_url_is_denied_without_an_http_request() -> None:
+    requests = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            requests.append(self.path)
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(b"evidence\nsynthetic-unregistered-network-file\n")
+
+        def log_message(self, *args):
+            pass
+
+    with HTTPServer(("127.0.0.1", 0), Handler) as server:
+        worker = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
+        worker.start()
+        try:
+            sql = f"SELECT * FROM read_csv_auto('http://127.0.0.1:{server.server_port}/unregistered.csv')"
+            result = await DuckDBExecutor(Settings()).execute(sql, {})
+        finally:
+            server.shutdown()
+            worker.join()
+    assert isinstance(result, ExecutionError)
+    assert requests == []
diff --git a/tests/unit/test_query_service.py b/tests/unit/test_query_service.py
index 94667aad..46778b11 100644
--- a/tests/unit/test_query_service.py
+++ b/tests/unit/test_query_service.py
@@ -1,4 +1,5 @@
 # Copyright 2024-2026 Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -519,3 +520,39 @@ async def execute(self, sql, attached_tables):
     # Here we test that the service passed retries correctly.
     if learner.proposals:
         assert learner.proposals[0]["retries"] >= 1
+
+
+@pytest.mark.asyncio
+async def test_critic_cannot_replace_a_scoped_query_with_an_unregistered_native_file_read(tmp_path):
+    from flyquery.core.services.execution.duckdb_executor import DuckDBExecutor
+
+    file = tmp_path / "unregistered.csv"
+    file.write_text("evidence\nsynthetic-unregistered-file\n")
+
+    class Settings:
+        duckdb_memory_limit = "256MB"
+        default_row_cap = 100
+
+    class InitialColumnFailure(DuckDBExecutor):
+        calls = 0
+
+        async def execute(self, sql, attached_tables):
+            self.calls += 1
+            if self.calls == 1:
+                return ExecutionError(message="fixture missing column")
+            return await super().execute(sql, attached_tables)
+
+    executor = InitialColumnFailure(Settings())
+    svc = _make_service(executor=executor)
+    svc._critic_agent = _FakeAgent(
+        RefinedSql(sql=f"SELECT * FROM read_csv_auto('{file}')", reasoning="fixture rewrite", confidence=0.8)
+    )
+    result = await svc.answer(
+        tenant_id="ten-a",
+        workspace_id=uuid.uuid4(),
+        dataset_id=uuid.uuid4(),
+        question="fixture",
+        scopes={"flyquery.query:read"},
+    )
+    assert result.execution_status not in ("OK", "REFINED_OK")
+    assert executor.calls == 1
diff --git a/tests/unit/test_sql_read_boundary.py b/tests/unit/test_sql_read_boundary.py
new file mode 100644
index 00000000..d5a842bb
--- /dev/null
+++ b/tests/unit/test_sql_read_boundary.py
@@ -0,0 +1,120 @@
+# Copyright 2026 Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Direct-handler contract fixtures with native DuckDB and owned Parquet.
+
+Repository doubles represent registered metadata; these cases do not qualify PostgreSQL RLS.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from types import SimpleNamespace
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+from starlette.requests import Request
+
+from flyquery.interfaces.sql_execute import SqlExecuteRequest
+from flyquery.web.controllers.sql_execute_controller import SqlExecuteController
+from flyquery.web.conventions import TenantContext
+
+
+def controller(tmp_path):
+    dataset = uuid.uuid4()
+    workspace = uuid.uuid4()
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 25}]), tmp_path / "orders.parquet")
+
+    class Rows:
+        def mappings(self):
+            return [{"name": "orders", "parquet_object_key": "orders.parquet"}]
+
+    class Session:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            pass
+
+        async def execute(self, statement, params):
+            assert params == {"ds": dataset, "names": ["orders"]}
+            return Rows()
+
+    class TableRepository:
+        async def list_kinds_for_dataset_by_names(self, selected, names):
+            if selected != dataset:
+                return []
+            return [{"name": "orders", "kind": "UPLOADED"}] if "orders" in names else []
+
+    class QueryRepository:
+        async def create_query(self, **kwargs):
+            return uuid.uuid4()
+
+        async def upsert_result(self, **kwargs):
+            pass
+
+    class Store:
+        async def put(self, key, value, **kwargs):
+            assert key.startswith("flyquery/fixture/")
+
+    class WorkspaceService:
+        async def get(self, selected):
+            return {"allow_direct_sql": selected == workspace}
+
+    settings = SimpleNamespace(
+        duckdb_memory_limit="256MB",
+        default_row_cap=100,
+        object_store_base=str(tmp_path),
+        result_preview_max_bytes=4096,
+        result_ttl_hours=1,
+    )
+    api = SqlExecuteController(
+        settings, Session, Store(), QueryRepository(), WorkspaceService(), TableRepository(), None
+    )
+    ctx = TenantContext(tenant_id="fixture", workspace_id=str(workspace), correlation_id="fixture")
+    request = Request(
+        {
+            "type": "http",
+            "headers": [(b"x-tenant-id", b"fixture"), (b"x-workspace-id", str(workspace).encode())],
+        }
+    )
+    return api, request, dataset, workspace, ctx
+
+
+@pytest.mark.parametrize(
+    "sql", ["SELECT total FROM orders", "WITH x AS (SELECT total FROM orders) SELECT total FROM x"]
+)
+@pytest.mark.asyncio
+async def test_registered_direct_sql_preserves_native_results(tmp_path, sql):
+    api, request, dataset, _, _ = controller(tmp_path)
+    result = await api.execute(request, SqlExecuteRequest(dataset_id=dataset, sql=sql))
+    assert result.execution_status == "OK"
+    assert result.preview == [{"total": 25}]
+
+
+@pytest.mark.asyncio
+async def test_stream_uses_registered_metadata_before_returning_native_rows(tmp_path):
+    api, _, dataset, workspace, ctx = controller(tmp_path)
+    frames = [
+        frame
+        async for frame in api._stream_events(
+            ctx=ctx,
+            workspace_id=workspace,
+            request=SqlExecuteRequest(dataset_id=dataset, sql="SELECT total FROM orders"),
+        )
+    ]
+    final = json.loads(frames[-1].decode().split("data: ", 1)[1])
+    assert final["execution_status"] == "OK"
+    assert final["preview"] == [{"total": 25}]
+
+
+@pytest.mark.asyncio
+async def test_unregistered_dataset_cannot_get_a_fabricated_table_registration(tmp_path):
+    api, request, _, _, _ = controller(tmp_path)
+    result = await api.execute(
+        request, SqlExecuteRequest(dataset_id=uuid.uuid4(), sql="SELECT total FROM orders")
+    )
+    assert result.execution_status == "REJECTED_BY_FIREWALL"
+    assert result.preview is None

From d283fceef7aa73c97ac4d6f3fed21aa184d2302e Mon Sep 17 00:00:00 2001
From: Andres Contreras <andres.contreras@soon.es>
Date: Wed, 14 Oct 2026 09:58:33 -0700
Subject: [PATCH 2/5] Preserve scoped correlated EXISTS reads

---
 .../core/services/execution/ast_classifier.py |  1 +
 tests/unit/test_query_read_boundary.py        | 37 +++++++++++++++++++
 2 files changed, 38 insertions(+)

diff --git a/src/flyquery/core/services/execution/ast_classifier.py b/src/flyquery/core/services/execution/ast_classifier.py
index 1ed9ffb2..6bbe73e1 100644
--- a/src/flyquery/core/services/execution/ast_classifier.py
+++ b/src/flyquery/core/services/execution/ast_classifier.py
@@ -56,6 +56,7 @@
         "TRY_CAST",
         "CASE",
         "IF",
+        "EXISTS",
         "DATE_TRUNC",
         "TIMESTAMP_TRUNC",
         "DATE_DIFF",
diff --git a/tests/unit/test_query_read_boundary.py b/tests/unit/test_query_read_boundary.py
index f1ba0a0b..34f4134c 100644
--- a/tests/unit/test_query_read_boundary.py
+++ b/tests/unit/test_query_read_boundary.py
@@ -228,3 +228,40 @@ def log_message(self, *args):
             worker.join()
     assert isinstance(result, ExecutionError)
     assert requests == []
+
+
+@pytest.mark.asyncio
+async def test_correlated_exists_preserves_registered_parquet_authority(tmp_path: Path) -> None:
+    file = tmp_path / "orders.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 25}]), file)
+    sql = "SELECT total FROM orders WHERE EXISTS (SELECT 1 FROM orders p WHERE p.id=orders.id)"
+    result = await DuckDBExecutor(Settings()).execute(sql, {"orders": str(file)})
+    assert isinstance(result, ExecutionResult)
+    assert result.rows == [{"total": 25}]
+    ast = AstClassifier().classify(sql)
+    assert ast.table_refs == ("orders",)
+    assert not ast.unsupported_sql
+    check(sql, {"orders": "UPLOADED"}, {"owned"}, {"orders": "owned"})
+    with pytest.raises(ScopeGuardError):
+        check(sql, {"orders": "UPLOADED"}, {"other"}, {"orders": "owned"})
+
+
+@pytest.mark.parametrize("nested_source", ["read_csv_auto", "unregistered"])
+@pytest.mark.asyncio
+async def test_exists_does_not_admit_nested_unregistered_sources(tmp_path: Path, nested_source: str) -> None:
+    file = tmp_path / "orders.parquet"
+    pq.write_table(pa.Table.from_pylist([{"id": 1, "total": 25}]), file)
+    unregistered = tmp_path / "unregistered.csv"
+    unregistered.write_text("id\n1\n")
+    source = f"read_csv_auto('{unregistered}')" if nested_source == "read_csv_auto" else nested_source
+    sql = f"SELECT total FROM orders WHERE EXISTS (SELECT 1 FROM {source} p WHERE p.id=orders.id)"
+    ast = AstClassifier().classify(sql)
+    if nested_source == "read_csv_auto":
+        assert ast.unsupported_sql
+    else:
+        assert not ast.unsupported_sql
+        assert ast.table_refs == ("orders", "unregistered")
+    with pytest.raises(ScopeGuardError):
+        check(sql, {"orders": "UPLOADED"}, {"owned"}, {"orders": "owned"})
+    result = await DuckDBExecutor(Settings()).execute(sql, {"orders": str(file)})
+    assert isinstance(result, ExecutionError)

From 9ed15ef1d4bc99cd5b60acd4e48ae89166dc1551 Mon Sep 17 00:00:00 2001
From: Andres Contreras <andres.contreras@soon.es>
Date: Wed, 14 Oct 2026 10:00:24 -0700
Subject: [PATCH 3/5] Require DuckDB with scoped file access controls

---
 pyproject.toml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index f62d0f45..da0c8890 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -42,8 +42,8 @@ dependencies = [
     # Object store
     "aiofiles>=24",
 
-    # DuckDB -- query engine; pinned here so the extras boundary stays stable.
-    "duckdb>=1.1",
+    # DuckDB's scoped file reads require the allowed_paths security setting.
+    "duckdb>=1.3.2",
     "pyarrow>=17.0",
 
     # sqlglot -- SQL AST parser / validator / transpiler for the

From fd2cf21fb42116ba41bde294f5e825190bf3d39b Mon Sep 17 00:00:00 2001
From: Andres Contreras <andres.contreras@soon.es>
Date: Wed, 14 Oct 2026 10:21:28 -0700
Subject: [PATCH 4/5] Document the enforced SQL snapshot boundary and remaining
 limits

---
 README.md                     |  24 ++++---
 docs/api-reference.md         |  15 +++--
 docs/architecture.md          |  46 ++++++++------
 docs/concurrency.md           |  55 ++++++++--------
 docs/deployment-topology.md   |  10 ++-
 docs/deployment.md            |  15 ++++-
 docs/glossary.md              |  17 ++---
 docs/operations-runbook.md    |  31 +++++----
 docs/payload-reference.md     |  12 ++--
 docs/pipeline.md              |  21 ++++---
 docs/scale-and-performance.md |  31 +++++----
 docs/security-model.md        | 115 +++++++++++++++++++++++-----------
 docs/troubleshooting.md       |  26 ++++----
 13 files changed, 247 insertions(+), 171 deletions(-)

diff --git a/README.md b/README.md
index 85a8cb1a..90f33385 100644
--- a/README.md
+++ b/README.md
@@ -58,7 +58,7 @@ object storage, derives a typed schema, samples + profiles each
 column, runs a PII pass, proposes joins to your other tables, and
 fits it to a long-lived knowledge base. Then you ask in English. A
 multi-agent pipeline (Grounding → Generation → Critic → Explainer)
-produces governed DuckDB SQL, runs it in a sandboxed connection
+produces governed DuckDB SQL, runs it over registered local Parquet snapshots
 under per-token scopes, returns rows + a chart hint, and keeps a
 conversation thread for drill-down.
 
@@ -96,9 +96,10 @@ supersession.
 
 **flyquery** handles **structured tabular data**. Its model is: file → typed
 schema → Parquet on object storage → DuckDB SQL via multi-agent pipeline. The
-query language is natural-language-to-SQL. The storage engine is Parquet on
-object storage queried in-process by DuckDB. The retrieval primitive is
-schema-element retrieval + table resolution + AST firewall. The user-facing
+query language is natural-language-to-SQL. ObjectStore holds the Parquet
+snapshots, with registered local snapshots queried in-process by DuckDB. The
+retrieval primitive is schema-element retrieval + table resolution + AST
+firewall. The user-facing
 concepts are datasets, tables, snapshots, examples, and metrics.
 
 The query languages, storage engines, retrieval primitives, and user-facing
@@ -211,7 +212,7 @@ curl -s -X POST http://localhost:8520/api/v1/query \
 | **Compression** | .gz, .zip, .bz2 transparently unwrapped |
 | **10-stage pipeline** | Receive → Parse → Reconcile → Sample → Profile → Relations → Describe → PII → Embed → Publish |
 | **4-agent query pipeline** | GroundingAgent → GenerationAgent → CriticAgent → ExplainerAgent |
-| **AST firewall** | sqlglot + DuckDB double-check; single-statement SELECT only on uploaded tables |
+| **AST firewall** | Closed sqlglot admission, registered table checks, and an independent executor check; exact file allowlist with external access disabled ([SQL read boundary](docs/security-model.md#sql-read-boundary)) |
 | **Schema knowledge base** | pgvector HNSW + BM25 + cross-encoder reranker |
 | **Semantic layer** | OSI/MetricFlow YAML metrics, deterministic SQL compilation |
 | **Conversation memory** | Multi-turn drill-down with snapshot pinning |
@@ -280,11 +281,14 @@ with `-m 'not llm'`.
 
 ## Architecture quick-take
 
-flyquery uses a **two-tier storage design**: Parquet on object storage (LocalFs in
-dev; S3/GCS/Azure Blob in production) queried by DuckDB in-process per request.
-Postgres stores all metadata, embeddings, relations, audit, and agent tokens.
-There is no persistent DuckDB database — every query ATTACHes Parquet snapshots
-fresh, eliminating shared mutable state between requests.
+flyquery stores Parquet through ObjectStore adapters and queries registered
+local snapshots with DuckDB in a fresh in-memory connection per request.
+Postgres stores metadata, embeddings, relations, audit, and agent tokens.
+Cloud storage and ingestion adapters are separate from the qualified local
+query workflow. The [SQL read boundary](docs/security-model.md#sql-read-boundary)
+explains the exact file allowlist, copy-on-write, and upstream authorization.
+Statement timeouts are currently unenforced; these in-process controls do not
+constitute an operating-system SQL sandbox.
 
 The **hexagonal adapter ring** means every external boundary (object store, file
 reader, vector store, PII scanner, semantic compiler, rate limiter) is a Python
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 265e66f8..abbcd781 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -753,12 +753,16 @@ the full pipeline completes.
 {
   "question": "Show top 10 customers by revenue last month",
   "dataset_id": "01906f2a-...",
-  "conversation_id": null,
-  "row_cap": 1000,
-  "statement_timeout_ms": 30000
+  "conversation_id": null
 }
 ```
 
+The current request accepts `question`, `dataset_id`, and optional
+`conversation_id`. Row caps use the executor's global setting; statement
+timeouts are unenforced. There are no per-request `row_cap` or
+`statement_timeout_ms` overrides. See the
+[SQL read boundary](security-model.md#sql-read-boundary).
+
 **Response** `200 OK`
 ```json
 {
@@ -864,8 +868,7 @@ Available only when `workspace.allow_direct_sql = true`.
 ```json
 {
   "sql": "SELECT COUNT(*) FROM orders",
-  "dataset_id": "01906f2a-...",
-  "row_cap": 500
+  "dataset_id": "01906f2a-..."
 }
 ```
 
@@ -1015,7 +1018,7 @@ Source: `queries_controller.py:124` (get_query_result).
 
 Aggregates `flyquery_cost_events` into day / week / month buckets
 scoped to the caller's workspace. Consume
-[`GET /api/v1/cost-events`](#cost-events) for the raw per-call ledger;
+[`GET /api/v1/cost-events`](#get-apiv1cost-events) for the raw per-call ledger;
 this endpoint just rolls it up.
 
 **Query params**: `period` = `day` (default) | `week` | `month`,
diff --git a/docs/architecture.md b/docs/architecture.md
index d9a2ae29..29ba291e 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -44,8 +44,8 @@ flyquery's specific responsibilities:
   operator approves.
 - **Query execution** — multi-agent pipeline
   (Grounding → Generation → Executor → Critic → Explainer) gated by per-token
-  scopes, an AST firewall, row caps, statement timeouts, PII scanning, and
-  conversation-drill-down memory.
+  scopes, closed AST admission, registered snapshot paths, row caps, PII
+  scanning, and conversation-drill-down memory.
 - **Audit trail** — every upload, snapshot diff, question, candidate SQL,
   executed statement, and cost event, append-only.
 
@@ -219,9 +219,12 @@ flyquery uses a two-tier storage design:
    Backed by `LocalFs` (dev), `S3`, `GCS`, or `Azure Blob` via the `ObjectStore`
    hexagonal port. Selected by `FLYQUERY_OBJECT_STORE`.
 
-2. **DuckDB** — in-process per-request query engine. ATTACHes the Parquet
-   snapshots at query time; no persistent DuckDB database. This eliminates
-   shared mutable state between requests.
+2. **DuckDB** — in-process per-request query engine. Creates views over
+   registered local Parquet snapshots in a fresh in-memory connection.
+   Connections have exact file allowlists and locked external-access settings.
+   See the [SQL read boundary](security-model.md#sql-read-boundary) for admission
+   rules, copy-on-write, and the unenforced statement timeout. ObjectStore
+   provider support does not establish remote query support.
 
 3. **Postgres + pgvector** — metadata, schema knowledge base, relations,
    embeddings, audit, agent tokens. The single persistent transactional store.
@@ -468,17 +471,20 @@ GroundingAgent  →  GroundedContext {
                                                          merge ◄──┘
        │
        ▼
-AST classifier (sqlglot + DuckDB parse) + scope/firewall check
+AST classifier (sqlglot) + ScopeGuard
   single-statement only
-  reject DDL on uploaded tables
+  reject DDL and unknown statements
   reject INSERT/UPDATE/DELETE on uploaded tables
   reject SQL referencing tables outside dataset_allowlist
-  reject DuckDB host-touching functions except httpfs/parquet/json/arrow
+  require registered physical tables and reviewed pure function names
        │
        ▼
 DuckDB executor (in-process, per-request)
-  ATTACH Parquet snapshots as virtual tables
-  SET memory_limit, statement_timeout, LIMIT row_cap+1
+  recheck SQL and supplied registered snapshots
+  create views over registered local Parquet; exact allowed_paths
+  external access / extension autoload off; configuration locked
+  SET memory_limit; LIMIT default_row_cap+1 (SELECT)
+  statement timeout is unenforced
        │
   ┌────┴──────┐
   OK        ERROR
@@ -503,6 +509,10 @@ Recorder → flyquery_queries + audit_events + cost_events
 
 ### Query-pipeline agents
 
+Direct SQL skips the agents and uses the same admission policy with registered
+metadata in both synchronous and SSE handlers. See the
+[SQL read boundary](security-model.md#sql-read-boundary).
+
 | Agent | Output type | Notes |
 |-------|-------------|-------|
 | `GroundingAgent` | `GroundedContext` | Iterative schema expansion up to `EXPAND_ITERS=2` |
@@ -554,16 +564,14 @@ class FileReader(Protocol):
 Implementations: `csv_reader`, `excel_reader`, `json_reader`,
 `parquet_reader`, `avro_reader`, `orc_reader`, `arrow_reader`.
 
-### DialectAdapter port (DuckDB only in v0)
+### Query execution services (DuckDB)
 
-```python
-class DialectAdapter(Protocol):
-    dialect: Literal["duckdb"]
-    async def execute(self, sql, *, attached_tables, row_cap, ...) -> ExecutionResult: ...
-    async def explain(self, sql, *, attached_tables) -> ExplainPlan: ...
-    def ast_classify(self, sql) -> AstClassification: ...
-    def lint(self, sql) -> list[LintFinding]: ...
-```
+`AstClassifier` classifies SQL with sqlglot. `ScopeGuard` checks registered
+metadata and grants before `DuckDBExecutor.execute(sql, attached_tables,
+derived_tables=None)` runs the statement. The executor reads the global
+memory and row-cap settings when constructed. It does not accept a
+per-request timeout or row-cap argument. See the
+[SQL read boundary](security-model.md#sql-read-boundary).
 
 ### Other ports (shared with canon)
 
diff --git a/docs/concurrency.md b/docs/concurrency.md
index 65ac7a0c..0494c28c 100644
--- a/docs/concurrency.md
+++ b/docs/concurrency.md
@@ -185,46 +185,43 @@ The regression test that covers the drain-with-cancel path is in
 
 ## 5. DuckDB per-request isolation
 
-Every query that hits the DuckDB executor (`POST /query`, `POST /sql:execute`)
-creates a new in-process DuckDB connection:
+Each admitted query execution (`POST /query`, `POST /sql:execute`) creates a
+fresh in-memory DuckDB connection inside the synchronous worker body.
+The asynchronous entry point hands off the request as follows:
 
 ```python
-import duckdb
-
-async def execute(self, sql, *, attached_tables, row_cap, statement_timeout_ms, memory_limit):
-    conn = duckdb.connect(":memory:", read_only=False)  # fresh per call
-    try:
-        conn.execute(f"SET memory_limit='{memory_limit}'")
-        conn.execute(f"SET threads=1")
-        for qname, parquet_key in attached_tables.items():
-            conn.execute(f"ATTACH '{parquet_key}' AS {qname} (READ_ONLY)")
-        # ... execute sql with LIMIT row_cap+1
-    finally:
-        conn.close()
+import asyncio
+
+async def execute(self, sql, attached_tables, derived_tables=None):
+    return await asyncio.to_thread(
+        self._sync, sql, dict(attached_tables), dict(derived_tables or {})
+    )
 ```
 
 Properties of this design:
 
 - **No shared state.** There is no persistent DuckDB database on disk. Each
   connection starts empty.
-- **No shared ATTACH.** Tables are ATTACHed fresh per query. Two concurrent
-  queries for the same dataset attach the same Parquet files independently.
-- **Read-only mode on ingested tables.** `ATTACH ... (READ_ONLY)` prevents
-  writes to Parquet even if the AST firewall were bypassed.
-- **Memory bounded.** `SET memory_limit` caps each connection. With N
-  concurrent queries at 4 GB each, API pods need `N × 4 GB + OS overhead`
-  RAM.
+- **Per-request views.** Registered local Parquet paths become views in each
+  connection. Concurrent queries do not share view or mutable-table state.
+- **Confined file access.** Exact `allowed_paths`, disabled external access
+  and extension autoload/install, and locked configuration restrict the
+  connection. DERIVED DML writes a new scratch Parquet output and leaves the
+  original snapshot unchanged.
+- **Memory budget.** `SET memory_limit` configures each connection's budget.
+  Size API pods for concurrent query memory and process overhead; this
+  setting is not an operating-system process limit. The current executor
+  fixes `threads=2` per connection.
+
+The connection is not read-only. SQL admission and snapshot checks precede
+execution; see the [SQL read boundary](security-model.md#sql-read-boundary)
+for the full policy. The global statement-timeout setting is unenforced,
+so the SELECT row cap does not bound the time required to compute rows.
 
 **Concurrency concern:** DuckDB in-process with asyncio requires care. DuckDB
 operations are synchronous and CPU-bound. They block the asyncio event loop
-thread. flyquery runs DuckDB calls in a thread pool executor:
-
-```python
-import asyncio
-result = await asyncio.get_event_loop().run_in_executor(
-    _duckdb_thread_pool, _execute_sync, conn, sql
-)
-```
+thread. flyquery moves the synchronous execution body to a worker thread
+with `asyncio.to_thread`, as shown above.
 
 This keeps the asyncio event loop responsive for incoming requests while
 DuckDB computes.
diff --git a/docs/deployment-topology.md b/docs/deployment-topology.md
index 5871c197..77c1e50d 100644
--- a/docs/deployment-topology.md
+++ b/docs/deployment-topology.md
@@ -3,8 +3,8 @@
 ## Table of Contents
 
 1. [Overview](#1-overview)
-2. [Single-node (docker-compose)](#2-single-node-docker-compose)
-3. [Multi-node (separate services)](#3-multi-node-separate-services)
+2. [Single-node (docker-compose)](#2-single-node-docker-compose--dev)
+3. [Multi-node (separate services)](#3-multi-node-production--separate-services)
 4. [HA considerations](#4-ha-considerations)
 5. [Service port reference](#5-service-port-reference)
 
@@ -26,6 +26,12 @@ the ingest worker. The retention worker doesn't run pipeline stages
 either — it polls on a fixed interval and only writes recovery /
 cleanup SQL.
 
+The storage-provider choices in these topologies describe ObjectStore
+storage and ingestion. Query execution is qualified for registered local
+Parquet snapshots accessible to the API process. Remote SQL file reads and
+extension-backed query workflows remain outside that qualification; see the
+[SQL read boundary](security-model.md#sql-read-boundary).
+
 For dev / docker-compose convenience, `flyquery worker all` runs both
 workers in a single asyncio event loop — **dev only**, production must
 split them.
diff --git a/docs/deployment.md b/docs/deployment.md
index 4931c4f1..8844434c 100644
--- a/docs/deployment.md
+++ b/docs/deployment.md
@@ -140,10 +140,16 @@ is the canonical reference. Copy to `.env` and edit.
 |----------|---------|-------|
 | `FLYQUERY_DUCKDB_MEMORY_LIMIT` | `4GB` | Per-connection memory cap |
 | `FLYQUERY_DEFAULT_ROW_CAP` | `1000` | Default result row limit |
-| `FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` | `30000` | Default query timeout (30 s) |
+| `FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` | `30000` | Configuration field; currently unenforced by DuckDBExecutor |
 | `FLYQUERY_RESULT_TTL_HOURS` | `24` | Presigned URL and result object TTL |
 | `FLYQUERY_RESULT_PREVIEW_MAX_BYTES` | `131072` | Preview JSON size cap (128 KiB) |
-| `FLYQUERY_DUCKDB_HTTPFS` | `true` | Enable httpfs for cloud object storage |
+| `FLYQUERY_DUCKDB_HTTPFS` | `true` | Configuration field; does not enable httpfs in DuckDBExecutor |
+
+The qualified query workflow uses registered local Parquet with DuckDB >= 1.3.2.
+External access and extension autoload/install remain disabled; see the
+[SQL read boundary](security-model.md#sql-read-boundary). Query requests have
+no per-request row-cap or statement-timeout override, and the global timeout
+setting does not stop a long-running statement.
 
 ### Ingestion
 
@@ -314,6 +320,11 @@ logging middleware.
 
 ## 10. Object storage configuration
 
+These settings select ObjectStore adapters for stored uploads, ingestion,
+and results. Cloud storage adapter availability does not qualify cloud-backed
+SQL execution. The current query boundary is qualified for registered local
+Parquet; see the [SQL read boundary](security-model.md#sql-read-boundary).
+
 ### Local filesystem (dev)
 
 ```
diff --git a/docs/glossary.md b/docs/glossary.md
index 0e84146a..f2b239b2 100644
--- a/docs/glossary.md
+++ b/docs/glossary.md
@@ -178,13 +178,15 @@ lists: `score = 1 / (k + rank)` where `k=FLYQUERY_RRF_K` (default 60).
 Ranks from BM25 and pgvector are fused before the cross-encoder reranker.
 
 **AST classifier**
-The component (sqlglot + DuckDB parser double-check) that classifies
-generated SQL as SELECT, INSERT, UPDATE, DELETE, or DDL, and enforces the
-AST firewall rules. See [security-model.md § 6](security-model.md#6-ast-firewall).
+The sqlglot component that classifies generated, rewritten, or direct SQL
+and admits only reviewed pure function names and supported table sources.
+`ScopeGuard` checks registered metadata and grants; the executor independently
+reclassifies SQL. See the [SQL read boundary](security-model.md#sql-read-boundary).
 
 **ScopeGuard**
-The component that checks the intersection of token scopes, workspace/dataset
-allowlists, AST classification, and table kind before allowing execution.
+The component that checks SQL classification, token scopes, table kind,
+registered table metadata, and dataset allowlists before execution. Actor
+authentication, workspace restrictions, and Postgres RLS remain upstream.
 
 **MetricFlow**
 The OSI/MetricFlow-shape YAML format used for semantic metrics in flyquery.
@@ -212,8 +214,9 @@ changes ship as `Patch+1` within the same month, signalled in CHANGELOG.
 
 **httpfs**
 A DuckDB extension that enables reading remote files via HTTP/S3/GCS/Azure.
-flyquery loads httpfs (along with parquet, json, arrow) and restricts its
-paths to the object-store prefix. No other DuckDB extensions are loaded.
+The current query executor does not load httpfs and disables extension
+autoload/install and external access. Remote query workflows are unqualified;
+see the [SQL read boundary](security-model.md#sql-read-boundary).
 
 **presigned URL**
 A time-limited, single-object URL generated by the ObjectStore for result
diff --git a/docs/operations-runbook.md b/docs/operations-runbook.md
index 941b8662..03cad4b0 100644
--- a/docs/operations-runbook.md
+++ b/docs/operations-runbook.md
@@ -25,7 +25,10 @@ full teardown.
 - Redis (optional; required for `FLYQUERY_EDA_ADAPTER=redis` or
   `FLYQUERY_RATE_LIMIT_BACKEND=redis`).
 - Object storage: local directory, S3 bucket, GCS bucket, or Azure Blob
-  container. IAM/service-account credentials configured.
+  container for storage/ingestion, with IAM/service-account credentials
+  configured. Query execution is qualified for registered local Parquet
+  snapshots accessible to the API process; see the
+  [SQL read boundary](security-model.md#sql-read-boundary).
 - `ANTHROPIC_API_KEY` (or `OPENAI_API_KEY` if using GPT-4o fallback).
 
 ### Step-by-step
@@ -246,23 +249,17 @@ the `TenantContextMiddleware` log output.
 
 ### Agent timeouts
 
-**Symptom:** Query fails with `error_code=AGENT_TIMEOUT` or
-`execution_status=FAILED` after ~30 seconds.
+**Symptom:** The query reports a provider timeout or stalls during execution.
 
-**Cause:** LLM API is slow; `FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` or
-`FLYQUERY_AGENT_MAX_OUTPUT_TOKENS` is too tight.
+**Action:** Identify the stage in the query and agent logs. For a provider
+failure, inspect rate-limit and latency errors from the configured provider.
+For DuckDB, inspect the query plan and reduce expensive scans or joins.
 
-**Fix:**
-```bash
-# Increase statement timeout (ms)
-export FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS=60000
-
-# Check LLM API health
-curl https://api.anthropic.com/v1/messages \
-  -H "x-api-key: $ANTHROPIC_API_KEY" \
-  -H "Content-Type: application/json" \
-  -d '{"model":"claude-haiku-4-5","max_tokens":10,"messages":[{"role":"user","content":"hi"}]}'
-```
+`FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` is currently unenforced by
+DuckDBExecutor, and query requests have no timeout override. Raising it
+does not stop a long-running SQL statement. The SELECT row cap limits
+returned rows, not computation time; see the
+[SQL read boundary](security-model.md#sql-read-boundary).
 
 ---
 
@@ -625,7 +622,7 @@ FOLLOW-UP ACTIONS:
 ### Key runbook pointers
 
 - Worker fleet ops → [§7 Worker fleet](#7-worker-fleet) and [workers.md](workers.md)
-- Stuck worker → [§3 Stuck ingest job](#stuck-ingest-job)
+- Stuck worker → [§3 Stuck ingest job](#ingest-job-stuck-in-running)
 - RLS returning empty → [§3 RLS returning zero rows](#rls-returning-zero-rows)
 - GDPR request → [§4 GDPR purge](#4-gdpr-purge-runbook)
 - Scaling → [§8 Scaling up](#8-scaling-up-more-workers)
diff --git a/docs/payload-reference.md b/docs/payload-reference.md
index 4ea377bd..cff73995 100644
--- a/docs/payload-reference.md
+++ b/docs/payload-reference.md
@@ -292,19 +292,19 @@ Sent to `PUT /api/v1/schema-objects/{id}`:
 {
   "question": "Show top 10 customers by revenue last month",
   "dataset_id": "01906f2a-0000-7000-a000-000000000002",
-  "conversation_id": null,
-  "row_cap": 1000,
-  "statement_timeout_ms": 30000
+  "conversation_id": null
 }
 ```
 
 | Field | Type | Required | Notes |
 |-------|------|----------|-------|
 | `question` | `string` | Yes | Natural-language question |
-| `dataset_id` | `string \| null` | No | If omitted, queries across all datasets the token can access |
+| `dataset_id` | `string` | Yes | UUID of the selected dataset |
 | `conversation_id` | `string \| null` | No | For multi-turn drill-down |
-| `row_cap` | `int \| null` | No | Default `FLYQUERY_DEFAULT_ROW_CAP` (1000) |
-| `statement_timeout_ms` | `int \| null` | No | Default `FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` (30000) |
+
+The current DTO has no `row_cap` or `statement_timeout_ms` fields. The
+executor uses `FLYQUERY_DEFAULT_ROW_CAP`; statement timeouts are currently
+unenforced. See the [SQL read boundary](security-model.md#sql-read-boundary).
 
 ### AnswerResponse
 
diff --git a/docs/pipeline.md b/docs/pipeline.md
index 09657af2..78adead5 100644
--- a/docs/pipeline.md
+++ b/docs/pipeline.md
@@ -273,24 +273,25 @@ GroundingAgent   →   GroundedContext {
 
         │
         ▼
-AST classifier (sqlglot + DuckDB parser double-check)
+AST classifier (sqlglot) + ScopeGuard
   Rules:
     single-statement only                         → REJECTED_BY_FIREWALL
     no DDL (CREATE/DROP/ALTER/TRUNCATE)           → REJECTED_BY_FIREWALL
     no INSERT/UPDATE/DELETE on UPLOADED tables    → REJECTED_BY_FIREWALL
     no DML on DERIVED without flyquery.derived:write → REJECTED_BY_FIREWALL
     no tables outside dataset_allowlist           → REJECTED_BY_FIREWALL
-    no DuckDB extensions other than httpfs/parquet/json/arrow → REJECTED_BY_FIREWALL
+    registered physical tables + closed pure function names required
+    file/table functions and unknown sources are rejected
 
         │
         ▼
 DuckDB executor (in-process, per-request)
-  ATTACH Parquet snapshots as virtual tables
-    (httpfs for cloud object store; local fs for dev)
+  Reclassify SQL and verify supplied registered snapshots
+  Create views over registered local Parquet; exact allowed_paths
+  Disable external access and extension autoload/install; lock configuration
   SET memory_limit = FLYQUERY_DUCKDB_MEMORY_LIMIT (default 4GB)
-  SET statement_timeout = FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS (default 30 s)
-  LIMIT row_cap+1 to detect overflow
-  Read-only mode — defence in depth against AST firewall gaps.
+  LIMIT default_row_cap+1 to detect SELECT overflow
+  Statement timeout is unenforced; connection is not read-only
 
         │
    ┌────┴──────┐
@@ -340,6 +341,12 @@ AnswerResponse / SSE final frame
    result_preview, result_url (presigned), snapshot_pins, query_id}
 ```
 
+Direct SQL skips the agents and uses the same policy with registered
+metadata in both synchronous and SSE handlers. Initial generated SQL and
+every Critic rewrite are checked before execution. See the
+[SQL read boundary](security-model.md#sql-read-boundary) for the local
+Parquet workflow, copy-on-write, and remaining limits.
+
 ### Query-pipeline agents
 
 | Agent | Output type | Model default | Notes |
diff --git a/docs/scale-and-performance.md b/docs/scale-and-performance.md
index ca77ff43..f91ad6f7 100644
--- a/docs/scale-and-performance.md
+++ b/docs/scale-and-performance.md
@@ -92,16 +92,15 @@ Mitigations:
 
 ### Object-storage I/O (large Parquet files)
 
-Stage 2 (parse) and DuckDB query both read from object storage.
-For 1 GB Parquet files:
-- Ingest stage 2: 5–15 s (streaming read + re-materialise)
-- DuckDB query: 2–10 s (httpfs streaming with predicate pushdown)
+Ingestion reads through ObjectStore adapters. The qualified DuckDB query
+workflow reads registered local Parquet snapshots accessible to the API
+process. Measure ingestion storage I/O and local query scans separately;
+remote httpfs query timings have not been qualified.
 
-Mitigations:
-- Enable `FLYQUERY_DUCKDB_HTTPFS_METADATA_CACHE_MB=512` (already default).
-- Place flyquery and object storage in the same AWS region / GCP region.
-- Use Parquet with efficient row groups and column-level statistics for
-  better predicate pushdown.
+Use efficient Parquet row groups and column statistics for predicate
+pushdown. `FLYQUERY_DUCKDB_HTTPFS_METADATA_CACHE_MB` does not configure the
+current executor, which disables extension autoload/install and external
+access. See the [SQL read boundary](security-model.md#sql-read-boundary).
 
 ---
 
@@ -116,8 +115,8 @@ Mitigations:
 | `FLYQUERY_EXPAND_ITERS` | `2` | +1-2 s per iteration | Better grounding |
 | `FLYQUERY_TOP_K_SCHEMA` | `12` | +10-20 ms per extra row | More context for Grounding |
 | `FLYQUERY_RERANKER_TOP_N` | `30` | +50-200 ms | Better schema linking |
-| `FLYQUERY_DUCKDB_MEMORY_LIMIT` | `4GB` | N/A | Prevents OOM |
-| `FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` | `30000` | N/A | Prevents runaway queries |
+| `FLYQUERY_DUCKDB_MEMORY_LIMIT` | `4GB` | N/A | Per-connection DuckDB memory budget |
+| `FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` | `30000` | None in current executor | Unenforced; does not prevent runaway queries |
 
 ### Ingestion
 
@@ -133,7 +132,7 @@ Mitigations:
 
 | Variable | Default | Effect |
 |----------|---------|--------|
-| `FLYQUERY_DUCKDB_HTTPFS_METADATA_CACHE_MB` | `512` | Higher = fewer object-store HEAD calls |
+| `FLYQUERY_DUCKDB_HTTPFS_METADATA_CACHE_MB` | `512` | Configuration field; unused by the current query executor |
 | `FLYQUERY_OBJECT_STORE_PRESIGN_TTL_S` | `86400` | Reduce to limit exposure window |
 | `FLYQUERY_RESULT_TTL_HOURS` | `24` | Reduce to save object-storage costs |
 
@@ -195,10 +194,10 @@ With 10 concurrent queries, expect 10× CPU cores to avoid queuing.
 
 ### Network
 
-Each query downloads the relevant Parquet columns from object storage.
-With `httpfs` and column predicate pushdown, only the queried columns are
-transferred. For a 100-column table with 1M rows querying 3 columns:
-typical transfer = 5–50 MB per query.
+ObjectStore ingestion and result transfers contribute network traffic. The
+qualified query executor scans registered local Parquet; it does not load
+httpfs. Remote column-transfer estimates require separate provider and query
+qualification. See the [SQL read boundary](security-model.md#sql-read-boundary).
 
 ### Horizontal worker scaling formula
 
diff --git a/docs/security-model.md b/docs/security-model.md
index e3a9fea3..71c6979c 100644
--- a/docs/security-model.md
+++ b/docs/security-model.md
@@ -26,9 +26,9 @@ flyquery is a multi-tenant service. The primary threats it defends against:
 | Cross-tenant data leak | Postgres RLS on every multi-tenant table |
 | Cross-workspace data leak | RLS workspace filter + agent-token workspace allowlist |
 | Malicious uploaded file (zip bomb, polyglot, billion-laughs JSON) | Size cap + magic-byte format verify + per-reader hardening |
-| Generated SQL exfiltrates host filesystem | AST firewall + DuckDB read-only mode + allowlisted extensions only |
+| Generated or direct SQL reads unregistered files or URLs | Closed AST admission + registered table checks + exact DuckDB file allowlist |
 | PII leaks via schema samples | PIIScanner gate before sample persistence; late-tag scrub |
-| DuckDB cross-request contamination | New in-process connection per request; no shared ATTACH state |
+| DuckDB cross-request contamination | New in-memory connection per request; no shared views or mutable tables |
 | Object storage misconfig exposes raw URIs | Presigned URLs only; TTL-scoped; per-tenant prefix isolation |
 | GDPR purge race with in-flight queries | Cooperative cancel + 30-day tombstone window |
 | Atomic snapshot interrupted mid-flight | PARTIAL → READY only after stage 10; no half-ingested window |
@@ -182,7 +182,10 @@ The effective grant for any operation is the **intersection** of:
 4. AST classification of the generated SQL
 5. Target table kind
 
-The executor refuses if any of these dimensions denies the operation.
+Request authentication, token gates, and `ScopeGuard` enforce these grants
+before execution. The executor independently confines SQL to the snapshots
+supplied by the resolver; it does not authenticate the actor or grant access
+to a tenant, workspace, or dataset.
 
 **Table kind enforcement**:
 - `UPLOADED` tables are read-only. Any INSERT/UPDATE/DELETE on an UPLOADED
@@ -203,38 +206,76 @@ scope list.
 
 ## 6. AST firewall
 
-Every piece of SQL — whether generated by an AI agent or submitted directly —
-passes through the AST firewall before execution.
+### SQL read boundary
 
-### Rules (DuckDB dialect, via sqlglot + DuckDB parser)
+The qualified query workflow uses registered local Parquet snapshots. The
+initial generated SQL, every Critic rewrite, and direct SQL in both the
+synchronous and SSE handlers must pass the same admission policy.
 
-| Rule | Effect on violation |
-|------|-------------------|
-| Single-statement only | `REJECTED_BY_FIREWALL` |
-| No DDL (CREATE, DROP, ALTER, TRUNCATE) on any table | `REJECTED_BY_FIREWALL` |
-| No INSERT/UPDATE/DELETE on UPLOADED tables | `REJECTED_BY_FIREWALL` |
-| No DML on DERIVED tables without `flyquery.derived:write` | `REJECTED_BY_FIREWALL` |
-| No references to tables outside the dataset_allowlist | `REJECTED_BY_FIREWALL` |
-| No DuckDB extensions other than `httpfs`, `parquet`, `json`, `arrow` | `REJECTED_BY_FIREWALL` |
-| No DuckDB functions that touch the host filesystem outside our prefix | `REJECTED_BY_FIREWALL` |
-| No `COPY TO` or `EXPORT DATABASE` | `REJECTED_BY_FIREWALL` |
-
-### DuckDB execution mode
-
-The `DuckDBAdapter` opens a new in-process DuckDB connection per request in
-**read-only mode** for ingested tables. This provides defence-in-depth: even
-if the AST firewall misses a statement, DuckDB's read-only mode refuses any
-writes to the Parquet files.
-
-Only `httpfs`, `parquet`, `json`, and `arrow` extensions are loaded. The
-`httpfs` extension is restricted to URLs matching the object storage prefix
-configured in `FLYQUERY_OBJECT_STORE_BASE`.
-
-### Double-check
+```
+Actor / token / workspace checks + Postgres RLS
+                     |
+Registered metadata for the selected dataset
+                     |
+sqlglot AST + ScopeGuard (also after each Critic rewrite)
+                     |
+Executor reclassifies SQL and checks supplied snapshots
+                     |
+Fresh DuckDB connection with exact allowed_paths
+                     |
+Registered Parquet views -> capped SELECT rows
+```
 
-After sqlglot AST classification, DuckDB parses the SQL a second time. If
-DuckDB's parser disagrees with sqlglot's classification, the statement is
-rejected with `error_code=AST_PARSE_CONFLICT`.
+| Check | Current behavior |
+|-------|------------------|
+| Statement shape | One SELECT or supported INSERT/UPDATE/DELETE; unknown statements, DDL, user `COPY`, and `EXPORT` are rejected |
+| Functions | A closed list of reviewed pure function names, including aggregates and `EXISTS`; unreviewed functions and file/table functions such as `read_csv_auto`, `read_parquet`, `glob`, and `query_table` are rejected |
+| Sources | Physical table references must exist in the selected dataset's active registered metadata; CTE aliases are resolved without treating them as physical tables |
+| Qualification | Unqualified names and `main.<name>` are supported; other schemas and catalogs are rejected |
+| Dataset grants | When an allowlist is present, every physical table must have a matching dataset mapping; missing mappings fail closed |
+| Mutation | UPLOADED tables reject DML; DERIVED DML requires the upstream write scope and a matching supplied DERIVED snapshot |
+
+Initial and direct admission failures return `REJECTED_BY_FIREWALL`. An
+unsafe Critic rewrite stops the retry loop without execution. The executor
+also checks SQL independently and returns `ExecutionError` for unsupported
+statements or unregistered snapshots. Both classifier checks use sqlglot;
+there is no separate DuckDB classification comparison or
+`AST_PARSE_CONFLICT` gate.
+
+`DuckDBExecutor` requires DuckDB >= 1.3.2. It opens a fresh `:memory:`
+connection with `autoload_known_extensions=false` and
+`autoinstall_known_extensions=false`, sets `allowed_paths` to the exact
+registered snapshot paths, then sets `enable_external_access=false` and
+`lock_configuration=true`. The memory limit is configured and `threads=2`.
+Table identifiers and paths are quoted when creating Parquet-backed views.
+SELECT runs inside a `LIMIT default_row_cap + 1` wrapper to detect truncation.
+The connection is not opened in read-only mode.
+
+DERIVED mutations use an in-memory mutable table and copy-on-write. The
+executor requires exactly one registered DERIVED target, permits a new owned
+scratch output path, and exports with `USE_TMP_FILE false`. It returns new
+Parquet bytes for the caller to publish as a new snapshot. The original
+snapshot remains unchanged.
+
+Actor identity, token scopes, workspace restrictions, and Postgres RLS remain
+upstream responsibilities. Local handler tests with repository doubles do
+not establish PostgreSQL RLS acceptance. The in-process settings provide
+defense in depth; they do not constitute an operating-system SQL sandbox.
+
+Statement timeouts are currently unenforced. `FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS`
+does not interrupt DuckDB execution, and the current `QueryRequest` has no
+`statement_timeout_ms` or `row_cap` override. Result limits do not bound CPU
+time spent scanning, joining, or aggregating before rows are returned.
+
+ObjectStore upload, ingestion, and result-storage adapters are separate from
+this query boundary. Remote object-store and extension-backed query workflows
+have not been qualified here. `FLYQUERY_DUCKDB_HTTPFS` does not enable httpfs
+in this executor; do not relax external-access or extension settings to make
+remote SQL sources run.
+
+Implementation: [AST classifier](../src/flyquery/core/services/execution/ast_classifier.py),
+[scope guard](../src/flyquery/core/services/execution/scope_guard.py),
+[executor](../src/flyquery/core/services/execution/duckdb_executor.py).
 
 ---
 
@@ -283,10 +324,10 @@ replaced with `[REDACTED]` in the result preview.
 ### Per-tenant prefix isolation
 
 All blobs are stored under `flyquery/{tenant_id}/{workspace_id}/`. A
-misconfigured storage policy that grants broad bucket read access cannot expose
-one tenant's files to another, because the prefix itself encodes the tenant
-boundary. Object-store IAM policies should restrict to the service-account
-key prefix.
+prefix organizes each tenant's objects. Access isolation also requires
+request-scoped metadata/RLS checks and object-store IAM policies; a prefix
+name alone does not prevent broad bucket access. Restrict service-account
+permissions to the required key prefix.
 
 ### Presigned URLs
 
@@ -332,7 +373,7 @@ Files exceeding the per-file cap are rejected at the upload endpoint with
 | Zip bomb | Decompressed-size check during decompression; capped at 2× the compressed size or `FLYQUERY_MAX_FILE_MB`, whichever is smaller |
 | ZIP with multiple entries | Rejected immediately with `error_code=ZIP_MULTI_ENTRY` |
 | Polyglot file (valid magic bytes for two formats) | Stage 1 verifies magic bytes match declared extension; mismatch → `FORMAT_MISMATCH` |
-| Billion-laughs JSON | DuckDB parser memory cap (`FLYQUERY_DUCKDB_MEMORY_LIMIT`) + statement timeout; worker cgroup memory limit |
+| Billion-laughs JSON | Apply deployment resource limits to ingestion; DuckDB query statement timeout is unenforced (see [SQL read boundary](#sql-read-boundary)) |
 
 ### GDPR purge
 
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index 96817be4..52a64d03 100644
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -212,19 +212,12 @@ for the test role setup recipe.
 
 ### Symptom: Query fails with `execution_status=FAILED`, error contains "timeout"
 
-**Root cause A:** DuckDB statement timeout (`FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS=30000`).
-Large aggregations on big Parquet files can exceed this.
+`FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS` is currently unenforced by the query
+executor, and `QueryRequest` has no per-query timeout override. Increasing
+that setting cannot fix a DuckDB execution stall. Identify the failing stage
+from the error and logs; see the [SQL read boundary](security-model.md#sql-read-boundary).
 
-**Fix:**
-```bash
-# Increase timeout (ms)
-export FLYQUERY_DEFAULT_STATEMENT_TIMEOUT_MS=120000
-
-# Or use the per-query timeout override:
-POST /api/v1/query {"question": "...", "statement_timeout_ms": 120000}
-```
-
-**Root cause B:** LLM API latency (Anthropic / OpenAI rate limit or slow
+**Root cause A:** LLM API latency (Anthropic / OpenAI rate limit or slow
 response).
 
 **Fix:** Check LLM API status pages. Temporarily switch to fallback model:
@@ -232,7 +225,7 @@ response).
 export FLYQUERY_FALLBACK_MODEL=openai:gpt-4o
 ```
 
-**Root cause C:** `FLYQUERY_AGENT_MAX_OUTPUT_TOKENS` too small for the
+**Root cause B:** `FLYQUERY_AGENT_MAX_OUTPUT_TOKENS` too small for the
 number of tables/columns in the Grounded context.
 
 **Fix:** Increase to 16384 for large schemas.
@@ -300,6 +293,13 @@ POST /api/v1/query:explain {"question": "...", "dataset_id": "..."}
 For repeated OOM on the same query, simplify the question or add a WHERE
 clause to scope the data.
 
+For a long-running DuckDB statement, reduce the scan or join work and inspect
+the query plan. The statement-timeout setting currently does not interrupt
+execution; the result row cap limits returned rows, not computation time.
+File-source or unregistered-table rejections are expected policy failures,
+not reasons to enable httpfs or external access. See the
+[SQL read boundary](security-model.md#sql-read-boundary).
+
 ---
 
 ## 10. Wrong SQL generated

From 60d1fa3fa98fd0f69a136ac4615d2925c33098fd Mon Sep 17 00:00:00 2001
From: Andres Contreras <andres.contreras@soon.es>
Date: Wed, 14 Oct 2026 10:22:40 -0700
Subject: [PATCH 5/5] Clarify snapshot resolution does not stage remote query
 files

---
 .../core/services/execution/table_resolver.py | 21 ++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/src/flyquery/core/services/execution/table_resolver.py b/src/flyquery/core/services/execution/table_resolver.py
index 3bcd6b53..88e7ae79 100644
--- a/src/flyquery/core/services/execution/table_resolver.py
+++ b/src/flyquery/core/services/execution/table_resolver.py
@@ -15,8 +15,8 @@
 """TableResolver — map AST table names to their current parquet snapshot paths.
 
 Given a dataset_id and a list of unqualified table names (as reported by
-``AstClassifier``), looks up each table's active ``schema_snapshots`` row
-and returns a ``{name: absolute_parquet_path}`` dict. This dict is passed
+``AstClassifier``), looks up active registered tables and their current
+schema snapshots, and returns a ``{name: snapshot_path}`` dict. This dict is passed
 directly to ``DuckDBExecutor.execute`` as ``attached_tables``.
 
 Path construction
@@ -25,9 +25,16 @@
   for LocalFs or ``s3://my-bucket/flyquery`` for S3).
 * ``parquet_object_key`` is the relative path stored in ``flyquery_schema_snapshots``
   (e.g. ``flyquery/ten-a/ws-id/ds-id/tables/table-id/v3.parquet``).
-* The full path is ``{object_store_base}/{parquet_object_key}`` regardless of
-  the adapter — DuckDB's S3 httpfs support makes the s3:// scheme work
-  transparently when ``duckdb_httpfs=true``.
+* The constructed path is ``{object_store_base}/{parquet_object_key}``.
+  This resolver performs metadata lookup and string construction only; it
+  does not download snapshots, stage local files, or load DuckDB extensions.
+
+The qualified executor workflow uses registered local Parquet paths with
+external access and extension autoload/install disabled. ``duckdb_httpfs``
+does not enable remote reads in that executor. A remote object-store URI
+would need a separately implemented, owned local-snapshot staging step;
+this resolver does not provide one. See ``docs/security-model.md`` under
+"SQL read boundary" for the execution policy and qualification limits.
 """
 
 from __future__ import annotations
@@ -55,12 +62,12 @@ async def resolve(
         table_names: list[str],
         object_store_base: str | None = None,
     ) -> dict[str, str]:
-        """Return a mapping of table name → absolute parquet path.
+        """Return a mapping of table name → constructed snapshot path.
 
         Tables that are active and have a current snapshot are included.
         Tables that are not found (not active, no snapshot, wrong dataset)
         are silently omitted so the caller can decide how to handle missing
-        tables (the executor will raise an error if a VIEW is missing).
+        tables (unregistered references are rejected by admission/executor checks).
 
         :param dataset_id: dataset to scope the lookup
         :param table_names: unqualified table names from the AST