ACE-037: fail-closed on unscopable SQL

deploy/agami.env.example

@@ -56,3 +56,9 @@ AGAMI_ADMIN_PASSWORD=
 # AGAMI_REFRESH_TOKEN_MODE=overwrite # refresh-token storage: overwrite (default: one row/session, no
 #                                    # dead-token heap; DROPS stolen-token reuse detection) OR rotate
 #                                    # (keeps OAuth 2.1 reuse detection + prunes expired revoked rows)
+
+# Optional: SQL scopability posture. `enforce` (the default) REFUSES any query the executor can't
+# fully scope to your declared tables — unparseable SQL, or a table-function / VALUES / UNNEST /
+# LATERAL source. This is the safe default. `warn` logs and allows instead; use it only as a
+# temporary staged-rollout escape hatch, never as the shipped setting.
+# AGAMI_SQL_UNSCOPABLE_POSTURE=enforce

packages/agami-core/src/execute_sql.py

@@ -920,6 +920,14 @@ def _refusal_from_verdict(kind: str, verdict: Verdict) -> Refusal:
     return Refusal(kind=kind, reason=verdict.detail, remediation=verdict.remediation)
 
 
+def _unscopable_posture() -> str:
+    """The unscopable-SQL rollout posture: ``enforce`` (default — fail-closed) or ``warn`` (a
+    staged-rollout escape hatch that logs and allows). ``warn`` is never the shipped default; safety
+    fails closed. This is an operational rollout knob, NOT the deployment tier — safety has no tier
+    variance and enforces in every tier."""
+    return os.environ.get("AGAMI_SQL_UNSCOPABLE_POSTURE", "enforce").strip().lower()
+
+
 def _model_safety(sql: str, profile: str, area: str | None) -> tuple[str, Refusal | None]:
     """Semantic-model safety pass before execution: fan-trap / chasm-trap pre-flight
     + default_filters auto-application, over a model resolved from the DB (hosted) or disk (local).
@@ -963,6 +971,21 @@ def _model_safety(sql: str, profile: str, area: str | None) -> tuple[str, Refusa
     # returns the same verdict as one that builds its own.
     ctx = RT.build_guard_context(sql, org)
 
+    # Scopability gate — refuse a query that can't be fully scoped (unparseable, or a non-`Table`
+    # FROM/JOIN source the scope walk can't reject) rather than run it blind. Runs
+    # BEFORE the object-scope gates so an unscopable query fails closed instead of reaching their
+    # degrade-to-allow branches. Posture `warn` is a staged-rollout escape hatch (logs + allows);
+    # the default `enforce` refuses. Safety fails closed regardless of tier.
+    scop = RT.check_scopable(sql, org, ctx=ctx)
+    if scop is not None:
+        if _unscopable_posture() == "warn":
+            sys.stderr.write(
+                "[agami] unscopable SQL allowed (AGAMI_SQL_UNSCOPABLE_POSTURE=warn): "
+                f"{scop.detail}\n"
+            )
+        else:
+            return sql, _refusal_from_verdict("unscopable_sql", scop)
+
     # Table-scope guard — a query may only reference tables the semantic model
     # declares; any other table in the connected database is refused. Runs FIRST
     # so the fan/chasm and sensitive checks below only evaluate in-scope tables.

packages/agami-core/src/semantic_model/runtime.py

@@ -600,12 +600,21 @@ def check_table_scope(
     """
     if not _HAVE_SQLGLOT:
         return None
-    allow = {
-        name.lower()
-        for name in (ctx.model_table_index if ctx is not None else _model_table_index(org))
-    }
+    index = ctx.model_table_index if ctx is not None else _model_table_index(org)
+    allow = {name.lower() for name in index}
     if not allow:
         return None
+    # name -> the set of declared (non-empty) SCHEMAS for that name. Used to catch a SCHEMA-qualified
+    # reference to a same-named table in an UNDECLARED schema — `secret_schema.orders` when the model
+    # declares only `public.orders` — which bare-name matching alone would wrongly admit, letting a
+    # query read a table the model never declared if the datasource role can see other schemas.
+    # `_model_table_index` values are `(Table, area_name)`; `Table.schema_name` is the declared schema
+    # (the field is aliased from `schema` — plain `.schema` collides with Pydantic's BaseModel.schema).
+    declared_schemas: dict[str, set[str]] = {}
+    for name, val in index.items():
+        sch = (val[0].schema_name or "").lower()
+        if sch:
+            declared_schemas.setdefault(name.lower(), set()).add(sch)
     if ctx is not None:
         tree = ctx.tree
     else:
@@ -626,8 +635,20 @@ def check_table_scope(
         name = tbl.name
         if not name or name.lower() in cte_names:
             continue  # a CTE reference, not a physical table
-        if name.lower() not in allow:
+        name_l = name.lower()
+        if name_l not in allow:
             offending.add(name)
+            continue
+        # The bare name is declared. If the reference QUALIFIES a schema, and the declared table(s) of
+        # that name carry a schema, the reference's schema must be one of them — else it targets a
+        # same-named table in an undeclared schema and is refused (confinement). An UNqualified
+        # reference matches by name as before: the datasource search_path resolves it, and we don't
+        # second-guess which schema that is.
+        q_schema = (tbl.db or "").lower()
+        if q_schema:
+            schemas = declared_schemas.get(name_l)
+            if schemas and q_schema not in schemas:
+                offending.add(f"{tbl.db}.{name}")
     if not offending:
         return None
 
@@ -832,6 +853,85 @@ def _select_chain(node):
     )
 
 
+# ---------------------------------------------------------------------------
+# Scopability gate (fail-closed)
+#
+# The object-scope gates above only reject the `exp.Table` sources they FIND. A
+# query that passes the read-only guard but yields no scopable tree, or a
+# FROM/JOIN source that is not a plain named `exp.Table` — a table-function or
+# `ROWS FROM` (an `exp.Table` with an EMPTY name, the function in `.this`), or a
+# `VALUES` / `UNNEST` / `LATERAL` node — leaves the scope walk with nothing to
+# reject: a silent fail-OPEN the contract forbids. This gate closes it: a query
+# that can't be fully scoped is refused (`unscopable_sql`) rather than run blind.
+# Reuses the single parse in `ctx` (no second parser) and enforces only when the
+# model declares a table surface, consistent with the object-scope gates.
+# ---------------------------------------------------------------------------
+
+
+def _unscopable(detail: str) -> Verdict:
+    return safety_verdict(
+        "unscopable_sql",
+        "the query can't be scoped to the declared object surface: "
+        + detail
+        + " — only queries whose FROM/JOIN sources resolve to declared tables may run.",
+        "Query only declared tables (no table-functions, VALUES, UNNEST, or LATERAL "
+        "sources); add a source to the model if it should be queryable.",
+    )
+
+
+def check_scopable(
+    sql: str, org: Organization, ctx: "GuardContext | None" = None
+) -> Verdict | None:
+    """Refuse a query that passes read-only but cannot be fully parsed/scoped.
+
+    Returns ``None`` when every FROM/JOIN source is something the scope checks can
+    resolve — a named `exp.Table` (real table or CTE reference) or a derived
+    `exp.Subquery` (whose own sources are covered by the same whole-tree walk) —
+    else a safety ``Verdict`` (`rule=unscopable_sql`). Fail-closed when: sqlglot is
+    unavailable, the SQL doesn't parse (`tree is None`), the tree has no SELECT, a
+    source is a table-function / `ROWS FROM` (an `exp.Table` with an empty name),
+    or a source is a `VALUES` / `UNNEST` / `LATERAL` / any other non-`Table`,
+    non-`Subquery` node — anywhere in the tree, so every set-operation arm is
+    covered. Reuses ``ctx.tree`` (no second parser). Inert (allow) when the model
+    declares no tables — a deployment with no declared surface isn't scoping,
+    consistent with `check_table_scope`.
+    """
+    # No declared surface -> not scoping (matches the object-scope gates' empty-model no-op). Only
+    # the emptiness matters here, so don't build a membership set.
+    if not (ctx.model_table_index if ctx is not None else _model_table_index(org)):
+        return None
+    if not _HAVE_SQLGLOT:
+        return _unscopable("the SQL parser is unavailable")
+    tree = ctx.tree if ctx is not None else _parse_sql(sql)
+    if tree is None:
+        return _unscopable("the query did not parse")
+    if tree.find(exp.Select) is None:
+        return _unscopable("the query has no SELECT to scope")
+    # Table-functions and `ROWS FROM` parse to an `exp.Table` with an EMPTY name (the
+    # function lives in `.this`); the object-scope gates skip empty-name tables, so a
+    # whole-tree sweep catches them here — including inside a set-operation arm.
+    for tbl in tree.find_all(exp.Table):
+        if not (tbl.name or ""):
+            return _unscopable("a FROM/JOIN source is a table-function, not a named table")
+    # LATERAL is a table-generating source that can't be scoped; sqlglot attaches it under a
+    # From/Join (Postgres `LATERAL (...)`) OR as a Select `laterals` property (Hive `LATERAL
+    # VIEW`), so sweep the whole tree, not only From/Join `.this`.
+    if tree.find(exp.Lateral) is not None:
+        return _unscopable("a FROM/JOIN source is a LATERAL, not a table")
+    # VALUES / UNNEST and any other non-`Table`, non-derived-subquery FROM/JOIN source. A comma-join
+    # `FROM t1, <VALUES/UNNEST/...>` normalizes to a `Join` node whose `.this` is that source (so it is
+    # covered by walking From AND Join). Some sqlglot versions instead hang the additional comma-join
+    # sources off `From.expressions`, so check those too — belt-and-suspenders, so an unscopable
+    # comma-join source can't slip through on either shape (Copilot review).
+    for node in tree.find_all(exp.From, exp.Join):
+        for src in [node.this, *(node.args.get("expressions") or [])]:
+            if src is not None and not isinstance(src, (exp.Table, exp.Subquery)):
+                return _unscopable(
+                    f"a FROM/JOIN source is a {type(src).__name__.upper()}, not a table"
+                )
+    return None
+
+
 def _cardinality_index(org: Organization) -> list[Relationship]:
     rels: list[Relationship] = []
     for sa in org.subject_areas:

plugins/agami/lib/execute_sql.py

@@ -920,6 +920,14 @@ def _refusal_from_verdict(kind: str, verdict: Verdict) -> Refusal:
     return Refusal(kind=kind, reason=verdict.detail, remediation=verdict.remediation)
 
 
+def _unscopable_posture() -> str:
+    """The unscopable-SQL rollout posture: ``enforce`` (default — fail-closed) or ``warn`` (a
+    staged-rollout escape hatch that logs and allows). ``warn`` is never the shipped default; safety
+    fails closed. This is an operational rollout knob, NOT the deployment tier — safety has no tier
+    variance and enforces in every tier."""
+    return os.environ.get("AGAMI_SQL_UNSCOPABLE_POSTURE", "enforce").strip().lower()
+
+
 def _model_safety(sql: str, profile: str, area: str | None) -> tuple[str, Refusal | None]:
     """Semantic-model safety pass before execution: fan-trap / chasm-trap pre-flight
     + default_filters auto-application, over a model resolved from the DB (hosted) or disk (local).
@@ -963,6 +971,21 @@ def _model_safety(sql: str, profile: str, area: str | None) -> tuple[str, Refusa
     # returns the same verdict as one that builds its own.
     ctx = RT.build_guard_context(sql, org)
 
+    # Scopability gate — refuse a query that can't be fully scoped (unparseable, or a non-`Table`
+    # FROM/JOIN source the scope walk can't reject) rather than run it blind. Runs
+    # BEFORE the object-scope gates so an unscopable query fails closed instead of reaching their
+    # degrade-to-allow branches. Posture `warn` is a staged-rollout escape hatch (logs + allows);
+    # the default `enforce` refuses. Safety fails closed regardless of tier.
+    scop = RT.check_scopable(sql, org, ctx=ctx)
+    if scop is not None:
+        if _unscopable_posture() == "warn":
+            sys.stderr.write(
+                "[agami] unscopable SQL allowed (AGAMI_SQL_UNSCOPABLE_POSTURE=warn): "
+                f"{scop.detail}\n"
+            )
+        else:
+            return sql, _refusal_from_verdict("unscopable_sql", scop)
+
     # Table-scope guard — a query may only reference tables the semantic model
     # declares; any other table in the connected database is refused. Runs FIRST
     # so the fan/chasm and sensitive checks below only evaluate in-scope tables.

plugins/agami/skills/agami-deploy/bundle/agami.env.example

@@ -48,3 +48,9 @@ AGAMI_ADMIN_PASSWORD=
 # AGAMI_REFRESH_TOKEN_MODE=overwrite # refresh-token storage: overwrite (default: one row/session, no
 #                                    # dead-token heap; DROPS stolen-token reuse detection) OR rotate
 #                                    # (keeps OAuth 2.1 reuse detection + prunes expired revoked rows)
+
+# Optional: SQL scopability posture. `enforce` (the default) REFUSES any query the executor can't
+# fully scope to your declared tables — unparseable SQL, or a table-function / VALUES / UNNEST /
+# LATERAL source. This is the safe default. `warn` logs and allows instead; use it only as a
+# temporary staged-rollout escape hatch, never as the shipped setting.
+# AGAMI_SQL_UNSCOPABLE_POSTURE=enforce

tests/test_ace051_fail_closed.py

@@ -108,6 +108,7 @@ def test_db_sourced_model_enforces_guards(tmp_path, monkeypatch, capsys):
 
 def test_disk_db_verdict_parity(tmp_path, monkeypatch, capsys):
     # The same model sourced from disk vs the DB must yield identical guard verdicts.
+    monkeypatch.delenv("AGAMI_SQL_UNSCOPABLE_POSTURE", raising=False)  # default enforce for the rows below
     _write_disk(tmp_path / "art" / "acme")
     url = "sqlite://" + str(tmp_path / "model.db")
     _seed_db(url, "acme")
@@ -124,15 +125,28 @@ def verdict(hosted: bool, sql: str):
         return code
 
     # Query BOTH declared tables + an undeclared table + a bad column, so a lossy DB round-trip that
-    # drops a table (customers) or mangles a column can't hide behind identical verdicts.
+    # drops a table (customers) or mangles a column can't hide behind identical verdicts. The last two
+    # rows are ACE-037's SC5: an unscopable query must produce the SAME fail-closed verdict on either
+    # source — the scopability gate is path-agnostic (it reads the parsed tree + the resolved model,
+    # never the datasource), so file-served and DB-served models refuse identically.
     for sql in (
-        "SELECT id FROM sqlite_master",   # undeclared table → refuse (both)
-        "SELECT id FROM orders",          # declared → allow (both)
-        "SELECT id FROM customers",       # the OTHER declared table → allow only if it survived
-        "SELECT nope FROM orders",        # undeclared column → refuse only if the column set survived
+        "SELECT id FROM sqlite_master",  # undeclared table → refuse (both)
+        "SELECT id FROM orders",  # declared → allow (both)
+        "SELECT id FROM customers",  # the OTHER declared table → allow only if it survived
+        "SELECT nope FROM orders",  # undeclared column → refuse only if the column set survived
     ):
         assert verdict(hosted=True, sql=sql) == verdict(hosted=False, sql=sql), sql
 
+    # The unscopable rows must both REFUSE (a Refusal, not None), not merely agree — pins the
+    # fail-closed half of SC5 (a silent no-op on BOTH sources would satisfy equality alone).
+    for sql in (
+        "SELECT g FROM generate_series(1, 10) AS t(g)",  # table-function
+        "SELECT x FROM (VALUES (1)) AS v(x)",  # VALUES source
+    ):
+        h, d = verdict(hosted=True, sql=sql), verdict(hosted=False, sql=sql)
+        assert h is not None and h.kind == "unscopable_sql", sql
+        assert d is not None and d.kind == "unscopable_sql", sql
+
 
 def test_hosted_falls_back_to_disk_when_db_has_no_model(tmp_path, monkeypatch, capsys):
     # Hosted, DB configured but EMPTY, yet a disk model exists → guards run off disk (not fail-closed).
@@ -195,3 +209,77 @@ def boom(name, _globals=None, _locals=None, fromlist=(), level=0):
     _, refusal = execute_sql._model_safety("SELECT id FROM orders", "acme", None)
     assert refusal is not None and refusal.kind == "model_unavailable"  # fail closed, no DB load
     assert capsys.readouterr().err == ""
+
+
+# ── ACE-037: the scopability gate wired into _model_safety + the posture flag ─────────────────
+
+# The differential corpus — queries that parse but don't scope (a source the scope walk can't
+# resolve). Fixtures live here; the broad regression suite is ACE-040's.
+_UNSCOPABLE_CORPUS = [
+    "SELECT g FROM generate_series(1, 10) AS t(g)",  # table-function
+    "SELECT a FROM ROWS FROM (generate_series(1, 3)) AS t(a)",  # ROWS FROM
+    "SELECT x FROM (VALUES (1), (2)) AS v(x)",  # VALUES source
+    "SELECT x FROM UNNEST(ARRAY[1, 2]) AS t(x)",  # UNNEST source
+    "SELECT a FROM orders o, LATERAL (SELECT 1 AS a) l",  # LATERAL source
+    "SELECT id FROM orders UNION SELECT g FROM generate_series(1, 3) AS t(g)",  # unscopable set-op arm
+]
+
+
+@pytest.mark.parametrize("sql", _UNSCOPABLE_CORPUS)
+def test_unscopable_corpus_refused_under_enforce(tmp_path, monkeypatch, capsys, sql):
+    # Every parse-but-don't-scope construct is refused (unscopable_sql) by _model_safety under the
+    # default `enforce` posture — proving the gate is REACHED in the wired pass (no silent pass).
+    url = "sqlite://" + str(tmp_path / "model.db")
+    _seed_db(url, "acme")
+    monkeypatch.setenv("AGAMI_DB_URL", url)
+    monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(tmp_path / "no_disk"))
+    monkeypatch.delenv("AGAMI_SQL_UNSCOPABLE_POSTURE", raising=False)  # default = enforce
+
+    _, refusal = execute_sql._model_safety(sql, "acme", None)
+    assert refusal is not None and refusal.kind == "unscopable_sql"  # refused, never executed
+    assert capsys.readouterr().err == ""
+
+
+def test_unscopable_allowed_and_logged_under_warn(tmp_path, monkeypatch, capsys):
+    # `warn` is the staged-rollout escape hatch: log + allow, do NOT refuse.
+    url = "sqlite://" + str(tmp_path / "model.db")
+    _seed_db(url, "acme")
+    monkeypatch.setenv("AGAMI_DB_URL", url)
+    monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(tmp_path / "no_disk"))
+    monkeypatch.setenv("AGAMI_SQL_UNSCOPABLE_POSTURE", "warn")
+
+    _, refusal = execute_sql._model_safety("SELECT g FROM generate_series(1, 10) AS t(g)", "acme", None)
+    assert refusal is None  # allowed (not refused)
+    err = capsys.readouterr().err
+    assert "unscopable SQL allowed" in err and "warn" in err
+    assert '"unscopable_sql"' not in err  # no refusal emitted
+
+
+@pytest.mark.parametrize("posture", ["off", "disable", "warm", "", "ENFORCE ", "0"])
+def test_unknown_posture_value_fails_closed(tmp_path, monkeypatch, capsys, posture):
+    # The posture flag's core safety invariant: ONLY exact `warn` allows; every other value
+    # (typo, empty, garbage) must fail closed (refuse). Guards against a future refactor that flips
+    # the default open.
+    url = "sqlite://" + str(tmp_path / "model.db")
+    _seed_db(url, "acme")
+    monkeypatch.setenv("AGAMI_DB_URL", url)
+    monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(tmp_path / "no_disk"))
+    monkeypatch.setenv("AGAMI_SQL_UNSCOPABLE_POSTURE", posture)
+
+    _, refusal = execute_sql._model_safety("SELECT g FROM generate_series(1, 3) AS t(g)", "acme", None)
+    assert refusal is not None and refusal.kind == "unscopable_sql"
+    assert capsys.readouterr().err == ""
+
+
+def test_scopability_gate_runs_before_table_scope(tmp_path, monkeypatch, capsys):
+    # A table-function is unscopable AND references no declared table; it must refuse as
+    # `unscopable_sql` (the gate runs first), not fall through to a different verdict.
+    url = "sqlite://" + str(tmp_path / "model.db")
+    _seed_db(url, "acme")
+    monkeypatch.setenv("AGAMI_DB_URL", url)
+    monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(tmp_path / "no_disk"))
+    monkeypatch.delenv("AGAMI_SQL_UNSCOPABLE_POSTURE", raising=False)
+
+    _, refusal = execute_sql._model_safety("SELECT g FROM generate_series(1, 3) AS t(g)", "acme", None)
+    assert refusal is not None and refusal.kind == "unscopable_sql"
+    assert capsys.readouterr().err == ""