From da8a5ecdacd412467d2965fc51012dc574bb6c8a Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Thu, 3 Sep 2026 22:58:53 -0600
Subject: [PATCH 1/4] feat(py): bind definitions to a source and wire both
 compiler phases

Last of four PRs porting the definition compiler (kata f6hz, stage 1).

Phase 1 runs when a dictionary is read, so an unusable definition fails at
data_dictionary() before any source exists, which is where R reports it too.
Phase 2 runs at data_source(), where a dialect is finally known, and fills
the compiled_definitions the registry has been rendering since #261.

Composition inlines a definition's sibling references into its SQL and
merges their notes. Each reference is renamed to a unique marker before the
SQL is emitted, rather than searching the emitted SQL for the definition's
own name afterwards, which is what makes a collision impossible. The
substitution is a scan rather than a string replace, so a string literal
holding a marker is left alone.

Grain is derived from the typed IR because the exported kind cannot answer
it: a row expression can hold an aggregate child without becoming an
aggregate. A metric whose chain mixes grain is refused, since one SQL
expression cannot express the subquery rewrite it would need.

Only DuckDB is lowered, so a source of any other dialect is refused at
construction rather than emitted for wrongly. A dictionary with no
definitions needs no emitter and is unaffected.

End to end, the compiled SQL is executed against a real DuckDB source rather
than only compared with a fixture, and every corpus definition is parsed by
DuckDB. That parse check asserts the returned json_serialize_sql payload:
it reports a parse failure in its result rather than raising, so the first
version of the test passed unconditionally.

export_spec() now accepts dictionary entries as models as well as raw
mappings, because callers construct DataDictionary both ways and phase 1
runs for both.
---
 pkg-py/src/commons/_data_dictionary.py      |  17 +-
 pkg-py/src/commons/_data_source.py          |   5 +
 pkg-py/src/commons/_definitions/__init__.py |   5 +-
 pkg-py/src/commons/_definitions/_compile.py | 273 +++++++++++++++++
 pkg-py/src/commons/_definitions/_export.py  |  28 +-
 pkg-py/tests/test_definition_compile.py     | 306 ++++++++++++++++++++
 6 files changed, 629 insertions(+), 5 deletions(-)
 create mode 100644 pkg-py/src/commons/_definitions/_compile.py
 create mode 100644 pkg-py/tests/test_definition_compile.py

diff --git a/pkg-py/src/commons/_data_dictionary.py b/pkg-py/src/commons/_data_dictionary.py
index de1402a4..893719ab 100644
--- a/pkg-py/src/commons/_data_dictionary.py
+++ b/pkg-py/src/commons/_data_dictionary.py
@@ -128,6 +128,11 @@ class DataDictionary(_Permissive):
     tables: dict[str, Table] = {}
     relationships: list[Relationship] = []
     glossary: dict[str, str] = {}
+    # Phase 1 of the compiler, keyed by table name. Source-independent, so it
+    # is produced here; the SQL needs a dialect and waits for `data_source()`.
+    # Values are `_definitions._export.DefinitionExport`, typed loosely so
+    # this module need not import the compiler's types.
+    definition_exports: dict[str, Any] = {}
 
     @model_validator(mode="before")
     @classmethod
@@ -136,6 +141,14 @@ def _normalize(cls, data: Any) -> Any:
             return data
         data = dict(data)
         data["tables"] = _key_by_name(data.get("tables"), "table")
+        # Type-checked here rather than at `data_source()` so an unusable
+        # definition is reported when the dictionary is read, before any
+        # source exists. Only the lowering to SQL needs a dialect.
+        from ._definitions._export import export_spec
+
+        data["definition_exports"] = {
+            name: table.definitions for name, table in export_spec(data).items()
+        }
         for field in ("name", "description", "details"):
             data[field] = _prose(data.get(field))
         relationships = data.get("relationships") or []
@@ -235,9 +248,7 @@ def _relationships_text(self, table: str) -> str | None:
         lines = []
         for relationship in self.relationships:
             text = " ".join(
-                part
-                for part in (relationship.join, relationship.description)
-                if part
+                part for part in (relationship.join, relationship.description) if part
             )
             if not _word_pattern(table).search(text):
                 continue
diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py
index c138b154..ce79acc2 100644
--- a/pkg-py/src/commons/_data_source.py
+++ b/pkg-py/src/commons/_data_source.py
@@ -272,6 +272,11 @@ def data_source(
         source = DataSource.from_frames(**frames)
 
     source.dictionary = resolved
+    if resolved is not None:
+        # The dialect is only known now, which is why lowering waits for it.
+        from ._definitions import attach_compiled_definitions
+
+        attach_compiled_definitions(resolved, source.dialect())
     return source
 
 
diff --git a/pkg-py/src/commons/_definitions/__init__.py b/pkg-py/src/commons/_definitions/__init__.py
index b30e236e..980d4fc6 100644
--- a/pkg-py/src/commons/_definitions/__init__.py
+++ b/pkg-py/src/commons/_definitions/__init__.py
@@ -1,4 +1,4 @@
-"""Governed definitions: the registry, and the compiler that will feed it.
+"""Governed definitions: the registry, and the compiler that feeds it.
 
 Definitions are authored in data-dict's expression language, not in the SQL
 dialect of the attached source, so they are type-checked against the
@@ -12,6 +12,7 @@
 authority, not this code.
 """
 
+from ._compile import attach_compiled_definitions, mixed_grain
 from ._registry import (
     ExportRecord,
     Registry,
@@ -28,10 +29,12 @@
     "ExportRecord",
     "Registry",
     "applied_text",
+    "attach_compiled_definitions",
     "build_registry",
     "context_chunks",
     "entry_text",
     "expand_tokens",
     "index_overflows",
     "index_text",
+    "mixed_grain",
 ]
diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py
new file mode 100644
index 00000000..4f132aee
--- /dev/null
+++ b/pkg-py/src/commons/_definitions/_compile.py
@@ -0,0 +1,273 @@
+"""Binding source-independent export records to a source.
+
+Export records stay source-independent until a source supplies the target
+dialect. This module does the rest: it derives the grain metadata
+`call_metrics` needs, refuses a metric no single SQL expression can express,
+inlines each definition's sibling references into its SQL, and produces the
+`ExportRecord` values the registry consumes.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ._emit_duckdb import emit_duckdb
+from ._export import DefinitionExport, Ir
+from ._registry import ExportRecord
+
+__all__ = ["attach_compiled_definitions", "mixed_grain"]
+
+# Only DuckDB is ported so far; Snowflake and Databricks are stage 2.
+_TARGETS = {"duckdb": "SQL(duckdb)"}
+
+
+def mixed_grain(definitions: dict[str, DefinitionExport]) -> dict[str, bool]:
+    """Which definitions mix row and aggregate grain, directly or inherited.
+
+    `call_metrics` refuses a mix of row and aggregate definitions in one call,
+    and the exported kind cannot answer this: a row expression can hold an
+    aggregate child without becoming an aggregate itself. Inheritance is
+    resolved to a fixed point because a reference chain can be any depth.
+    """
+    grain = {
+        name: definition.ir is not None
+        and definition.ir.shape == "row"
+        and _has_aggregate(definition.ir)
+        for name, definition in definitions.items()
+    }
+    while True:
+        updated = {
+            name: grain[name]
+            or any(grain.get(reference, False) for reference in definition.definitions)
+            for name, definition in definitions.items()
+        }
+        if updated == grain:
+            return grain
+        grain = updated
+
+
+def _has_aggregate(ir: Ir) -> bool:
+    if ir.shape == "agg":
+        return True
+    return any(_has_aggregate(child) for child in _children(ir))
+
+
+def _children(ir: Ir) -> list[Ir]:
+    out: list[Ir] = []
+    for value in ir.attrs.values():
+        if isinstance(value, Ir):
+            out.append(value)
+        elif isinstance(value, list):
+            out.extend(item for item in value if isinstance(item, Ir))
+        elif isinstance(value, dict):
+            # A `CASE` branch is a mapping of condition and result.
+            out.extend(item for item in value.values() if isinstance(item, Ir))
+    return out
+
+
+def attach_compiled_definitions(dictionary: Any, dialect: str) -> None:
+    """Compile every governed definition for `dialect`, onto the dictionary.
+
+    Called once a source is known, because the dialect decides the emitter.
+    A dictionary with no definitions needs no emitter, so an unsupported
+    dialect is only an error when there is something to lower.
+    """
+    exports: dict[str, dict[str, DefinitionExport]] = (
+        getattr(dictionary, "definition_exports", None) or {}
+    )
+    for table_name, entry in dictionary.tables.items():
+        definitions = exports.get(table_name) or {}
+        if not definitions:
+            entry.compiled_definitions = []
+            continue
+        target = _TARGETS.get(dialect)
+        if target is None:
+            raise ValueError(
+                f"Definitions on table {table_name!r} cannot be compiled for a "
+                f"{dialect!r} data source. commons lowers definitions to DuckDB "
+                f"only."
+            )
+        entry.compiled_definitions = _compile_table(table_name, definitions, target)
+
+
+def _compile_table(
+    table: str, definitions: dict[str, DefinitionExport], target: str
+) -> list[ExportRecord]:
+    grain = mixed_grain(definitions)
+    offenders = [
+        name
+        for name, definition in definitions.items()
+        if grain[name] and definition.kind == "metric"
+    ]
+    if offenders:
+        raise ValueError(
+            f"Metric definition {offenders[0]!r} on table {table!r} cannot be "
+            f"compiled into one SQL expression. Its dependency chain mixes row "
+            f"and aggregate grain and would need a subquery rewrite."
+        )
+    markers = _reference_markers(definitions)
+    emitted = {
+        name: emit_duckdb(
+            _mark_references(definition.ir, markers), definition.selection
+        )
+        for name, definition in definitions.items()
+        if definition.ir is not None
+    }
+    composed = _compose(table, definitions, emitted, markers)
+    return [
+        ExportRecord(
+            name=name,
+            # Filled by the registry, which knows the source and the table it
+            # was reached through.
+            table=table,
+            source="",
+            kind=definition.kind,
+            type=definition.type,
+            expression=definition.expression,
+            label=definition.label,
+            description=definition.description,
+            details=definition.details,
+            columns=list(definition.columns),
+            definitions=list(definition.definitions),
+            sql=composed[name]["code"],
+            target=target,
+            notes=composed[name]["notes"],
+            mixed_grain=grain[name],
+        )
+        for name, definition in definitions.items()
+    ]
+
+
+def _reference_markers(definitions: dict[str, DefinitionExport]) -> dict[str, str]:
+    """A unique stand-in for each definition name, safe to substitute later.
+
+    Composition works on the emitted SQL, so a reference has to be findable in
+    it without colliding with a real identifier. Marking the reference before
+    emitting rather than searching for the definition's own name afterwards is
+    what makes that collision impossible.
+    """
+    used = set(definitions)
+    for definition in definitions.values():
+        used.update(definition.columns)
+        if definition.ir is not None:
+            used.update(_ir_identifiers(definition.ir))
+    markers: dict[str, str] = {}
+    for index, name in enumerate(definitions, start=1):
+        marker = f"__commons_definition_reference_{index:03d}__"
+        while marker in used or marker in markers.values():
+            marker += "_"
+        markers[name] = marker
+    return markers
+
+
+def _ir_identifiers(ir: Ir) -> set[str]:
+    out: set[str] = set()
+    if ir.kind == "column":
+        out.update(ir.attrs.get("path") or [])
+    for child in _children(ir):
+        out |= _ir_identifiers(child)
+    return out
+
+
+def _mark_references(ir: Ir, markers: dict[str, str]) -> Ir:
+    """Rename each sibling-definition reference to its marker."""
+    attrs: dict[str, Any] = {}
+    for key, value in ir.attrs.items():
+        if isinstance(value, Ir):
+            attrs[key] = _mark_references(value, markers)
+        elif isinstance(value, list):
+            attrs[key] = [
+                _mark_references(item, markers) if isinstance(item, Ir) else item
+                for item in value
+            ]
+        elif isinstance(value, dict):
+            attrs[key] = {
+                inner_key: _mark_references(inner, markers)
+                if isinstance(inner, Ir)
+                else inner
+                for inner_key, inner in value.items()
+            }
+        else:
+            attrs[key] = value
+    if ir.kind == "column" and ir.attrs.get("reference") == "definition":
+        path = list(attrs["path"])
+        path[0] = markers[path[0]]
+        attrs["path"] = path
+    return Ir(kind=ir.kind, type=ir.type, shape=ir.shape, attrs=attrs)
+
+
+def _compose(
+    table: str,
+    definitions: dict[str, DefinitionExport],
+    emitted: dict[str, dict[str, Any]],
+    markers: dict[str, str],
+) -> dict[str, dict[str, Any]]:
+    """Inline each definition's references, in dependency order."""
+    composed: dict[str, dict[str, Any]] = {}
+    pending = list(definitions)
+    while pending:
+        ready = [
+            name
+            for name in pending
+            if all(reference in composed for reference in definitions[name].definitions)
+        ]
+        if not ready:
+            raise ValueError(
+                f"Definitions on table {table!r} cannot be composed because "
+                f"their dependency graph is unresolved: "
+                f"{', '.join(repr(name) for name in pending)}."
+            )
+        for name in ready:
+            references = definitions[name].definitions
+            replacements = {
+                markers[reference]: composed[reference]["code"]
+                for reference in references
+            }
+            notes = list(emitted[name]["notes"])
+            for reference in references:
+                notes.extend(composed[reference]["notes"])
+            composed[name] = {
+                "code": _substitute_identifiers(emitted[name]["code"], replacements),
+                "notes": sorted(set(notes)),
+            }
+        pending = [name for name in pending if name not in ready]
+    return composed
+
+
+def _substitute_identifiers(code: str, replacements: dict[str, str]) -> str:
+    """Replace quoted marker identifiers with the SQL they stand for.
+
+    A scan rather than a string replace, so a marker appearing inside a string
+    literal is left alone. The substituted SQL is parenthesised because it
+    lands in the middle of an expression whose precedence it does not know.
+    """
+    if not replacements:
+        return code
+    out: list[str] = []
+    index = 0
+    while index < len(code):
+        char = code[index]
+        if char in ("'", '"'):
+            token, index = _take_quoted(code, index, char)
+            if char == '"':
+                name = token[1:-1].replace('""', '"')
+                if name in replacements:
+                    out.append(f"({replacements[name]})")
+                    continue
+            out.append(token)
+            continue
+        out.append(char)
+        index += 1
+    return "".join(out)
+
+
+def _take_quoted(code: str, start: str | int, quote: str) -> tuple[str, int]:
+    index = int(start) + 1
+    while index < len(code):
+        if code[index] == quote:
+            if index + 1 < len(code) and code[index + 1] == quote:
+                index += 2
+                continue
+            return code[int(start) : index + 1], index + 1
+        index += 1
+    raise ValueError("Generated SQL contains an unterminated quoted value.")
diff --git a/pkg-py/src/commons/_definitions/_export.py b/pkg-py/src/commons/_definitions/_export.py
index fa74663f..660bd366 100644
--- a/pkg-py/src/commons/_definitions/_export.py
+++ b/pkg-py/src/commons/_definitions/_export.py
@@ -712,14 +712,40 @@ def _walk(node: Node, visit: Any) -> None:
         _walk(child, visit)
 
 
+def _as_mapping(value: Any) -> dict[str, Any]:
+    """The raw form of a dictionary entry, however it was supplied.
+
+    Callers construct a dictionary either from parsed YAML, where entries are
+    plain mappings, or from the reader's own models, where they are not.
+    """
+    if isinstance(value, dict):
+        return value
+    dump = getattr(value, "model_dump", None)
+    if callable(dump):
+        dumped = dump()
+        return dumped if isinstance(dumped, dict) else {}
+    return {}
+
+
 def _named_entries(entries: Any, what: str) -> dict[str, Any]:
     if not entries:
         return {}
+    if isinstance(entries, dict):
+        # data-dict authors a sequence, but the reader keys it by name before
+        # validating, so both shapes reach here.
+        keyed: dict[str, Any] = {}
+        for key, value in entries.items():
+            name = str(key)
+            if not name:
+                raise ValueError(f"Each {what} needs a non-empty name.")
+            keyed[name] = {**_as_mapping(value), "name": name}
+        return keyed
     if not isinstance(entries, list):
         raise TypeError(f"The data dictionary's {what}s must be a list.")
     out: dict[str, Any] = {}
     for entry in entries:
-        name = entry.get("name") if isinstance(entry, dict) else None
+        entry = _as_mapping(entry) if not isinstance(entry, dict) else entry
+        name = entry.get("name")
         if not isinstance(name, str) or not name:
             raise ValueError(f"Each {what} needs a non-empty name.")
         if name in out:
diff --git a/pkg-py/tests/test_definition_compile.py b/pkg-py/tests/test_definition_compile.py
new file mode 100644
index 00000000..39918919
--- /dev/null
+++ b/pkg-py/tests/test_definition_compile.py
@@ -0,0 +1,306 @@
+"""Binding export records to a source, and the two attach points.
+
+Phase 1 runs when a dictionary is read and needs no source. Phase 2 runs at
+`data_source()`, where a dialect is finally known, and fills the
+`compiled_definitions` the registry renders.
+"""
+
+import json
+
+import duckdb
+import pandas as pd
+import pytest
+import yaml
+
+from commons import data_source
+from commons._data_dictionary import DataDictionary
+from commons._definitions import build_registry, expand_tokens
+from commons._definitions._compile import attach_compiled_definitions, mixed_grain
+from commons._definitions._export import export_spec
+from tests._shared import SHARED_DIR, load_shared_fixture
+
+
+def dictionary_yaml(*definitions: dict, columns: list[dict] | None = None) -> dict:
+    return {
+        "tables": [
+            {
+                "name": "orders",
+                "columns": columns
+                if columns is not None
+                else [
+                    {"name": "amount", "type": "number(quantity)"},
+                    {"name": "region", "type": "string"},
+                ],
+                "definitions": list(definitions),
+            }
+        ]
+    }
+
+
+def compiled(*definitions: dict, **kwargs) -> list:
+    dictionary = DataDictionary.model_validate(dictionary_yaml(*definitions, **kwargs))
+    attach_compiled_definitions(dictionary, "duckdb")
+    return dictionary.tables["orders"].compiled_definitions
+
+
+def by_name(records: list) -> dict:
+    return {record.name: record for record in records}
+
+
+# --- grain -----------------------------------------------------------------
+
+
+def test_grain_matches_the_shared_contract():
+    fixture = load_shared_fixture("definitions")["mixed_grain"]
+    paths = sorted((SHARED_DIR / "definition-export" / "valid").glob("*.yaml"))
+    assert paths
+    checked = 0
+    for path in paths:
+        exported = export_spec(yaml.safe_load(path.read_text(encoding="utf-8")))
+        for key, expected in fixture[path.name].items():
+            table, name = key.split("::")
+            grain = mixed_grain(exported[table].definitions)
+            assert grain[name] is expected, key
+            checked += 1
+    assert checked == 42
+
+
+def test_a_row_expression_holding_an_aggregate_is_mixed_grain():
+    records = by_name(compiled({"name": "d", "expr": "amount > avg(amount)"}))
+    assert records["d"].mixed_grain is True
+
+
+def test_a_plain_row_expression_is_not_mixed_grain():
+    assert by_name(compiled({"name": "d", "expr": "amount > 0"}))["d"].mixed_grain is (
+        False
+    )
+
+
+def test_an_aggregate_is_not_mixed_grain():
+    assert by_name(compiled({"name": "d", "expr": "sum(amount)"}))["d"].mixed_grain is (
+        False
+    )
+
+
+def test_mixed_grain_is_inherited_through_a_reference():
+    records = by_name(
+        compiled(
+            {"name": "big", "expr": "amount > avg(amount)"},
+            {"name": "d", "expr": "big and region = 'x'"},
+        )
+    )
+    assert records["d"].mixed_grain is True
+
+
+def test_a_mixed_grain_metric_cannot_be_compiled():
+    # A metric whose chain mixes row and aggregate grain needs a subquery
+    # rewrite, which one SQL expression cannot express.
+    with pytest.raises(ValueError, match="one SQL expression"):
+        compiled(
+            {"name": "flag", "expr": "amount > avg(amount)"},
+            {"name": "d", "expr": "sum(case when flag then 1 else 0 end)"},
+        )
+
+
+# --- composition -----------------------------------------------------------
+
+
+def test_a_referenced_definition_is_inlined_and_parenthesised():
+    records = by_name(
+        compiled(
+            {"name": "big", "expr": "amount > 100"},
+            {"name": "d", "expr": "big and region = 'x'"},
+        )
+    )
+    assert records["d"].sql == '("amount" > 100) AND "region" = \'x\''
+
+
+def test_composition_follows_a_chain():
+    records = by_name(
+        compiled(
+            {"name": "a", "expr": "amount > 1"},
+            {"name": "b", "expr": "a and amount < 10"},
+            {"name": "c", "expr": "b or region = 'x'"},
+        )
+    )
+    assert records["c"].sql == (
+        '(("amount" > 1) AND "amount" < 10) OR "region" = \'x\''
+    )
+
+
+def test_composition_merges_the_notes_of_what_it_inlined():
+    records = by_name(
+        compiled(
+            {"name": "big", "expr": "amount > 100"},
+            {"name": "d", "expr": "big and region = 'x'"},
+        )
+    )
+    # The NaN note belongs to the numeric comparison inside `big`.
+    assert any("NaN" in note for note in records["d"].notes)
+
+
+def test_a_string_that_looks_like_a_definition_name_is_not_substituted():
+    records = by_name(
+        compiled(
+            {"name": "big", "expr": "amount > 100"},
+            {"name": "d", "expr": "big and region = 'big'"},
+        )
+    )
+    assert records["d"].sql.endswith("\"region\" = 'big'")
+
+
+def test_an_unreferenced_definition_keeps_its_own_sql():
+    records = by_name(compiled({"name": "d", "expr": "amount > 0"}))
+    assert records["d"].sql == '"amount" > 0'
+
+
+# --- the record ------------------------------------------------------------
+
+
+def test_the_record_carries_what_the_registry_needs():
+    record = by_name(compiled({"name": "d", "expr": "sum(amount)", "label": "Total"}))[
+        "d"
+    ]
+    assert record.name == "d"
+    assert record.kind == "metric"
+    assert record.type == "number"
+    assert record.label == "Total"
+    assert record.target == "SQL(duckdb)"
+    assert record.expression == "sum(amount)"
+    assert record.columns == ["amount"]
+
+
+# --- the dialect gate ------------------------------------------------------
+
+
+def test_a_dialect_with_no_emitter_is_refused():
+    # Only DuckDB is ported so far. A source commons cannot lower for must
+    # fail at construction rather than emit the wrong dialect's SQL.
+    with pytest.raises(ValueError, match="postgresql"):
+        dictionary = DataDictionary.model_validate(
+            dictionary_yaml({"name": "d", "expr": "amount > 0"})
+        )
+        attach_compiled_definitions(dictionary, "postgresql")
+
+
+def test_a_dictionary_with_no_definitions_needs_no_emitter():
+    dictionary = DataDictionary.model_validate(
+        {"tables": [{"name": "orders", "columns": [{"name": "amount"}]}]}
+    )
+    attach_compiled_definitions(dictionary, "postgresql")
+    assert dictionary.tables["orders"].compiled_definitions == []
+
+
+# --- phase 1: reading a dictionary ----------------------------------------
+
+
+def test_an_invalid_definition_fails_when_the_dictionary_is_read():
+    # Before any source exists, so both packages report it at the same point.
+    with pytest.raises(ValueError):
+        DataDictionary.model_validate(
+            dictionary_yaml({"name": "d", "expr": "nope > 1"})
+        )
+
+
+def test_a_valid_dictionary_reads_and_keeps_its_authored_definitions():
+    dictionary = DataDictionary.model_validate(
+        dictionary_yaml({"name": "d", "expr": "amount > 0"})
+    )
+    assert dictionary.tables["orders"].definitions["d"].expr == "amount > 0"
+    # Nothing is compiled yet: that needs a source.
+    assert dictionary.tables["orders"].compiled_definitions == []
+
+
+# --- phase 2: end to end through a real source ----------------------------
+
+
+@pytest.fixture
+def source(tmp_path):
+    path = tmp_path / "data-dict.yaml"
+    path.write_text(
+        yaml.safe_dump(
+            {
+                "tables": [
+                    {
+                        "name": "orders",
+                        "columns": [
+                            {"name": "amount", "type": "number"},
+                            {"name": "region", "type": "string"},
+                        ],
+                        "definitions": [
+                            {"name": "big", "expr": "amount > 100"},
+                            {
+                                "name": "big emea",
+                                "expr": "big and region = 'EMEA'",
+                            },
+                            {"name": "total", "expr": "sum(amount)"},
+                        ],
+                    }
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+    frame = pd.DataFrame({"amount": [50, 150, 250], "region": ["EMEA", "EMEA", "AMER"]})
+    return data_source(orders=frame, dictionary=path)
+
+
+def test_a_source_compiles_the_dictionary_it_was_given(source):
+    records = by_name(source.dictionary.tables["orders"].compiled_definitions)
+    assert sorted(records) == ["big", "big emea", "total"]
+    assert records["total"].sql == 'sum("amount")'
+
+
+def test_a_compiled_definition_reaches_the_registry(source):
+    registry = build_registry({"orders_db": source})
+    records = registry.for_source("orders_db")
+    assert sorted(record.name for record in records) == [
+        "big",
+        "big emea",
+        "total",
+    ]
+    assert all(record.table == "orders" for record in records)
+
+
+def test_an_expanded_token_runs_against_the_real_source(source):
+    registry = build_registry({"orders_db": source})
+    query, used = expand_tokens(
+        "SELECT count(*) FROM orders WHERE {{big emea}}", registry.for_source()
+    )
+    assert [record.name for record in used] == ["big emea"]
+    result = source.backend.connection.execute(query).fetchone()
+    # One row is over 100 and in EMEA.
+    assert result[0] == 1
+
+
+def test_an_expanded_metric_runs_against_the_real_source(source):
+    registry = build_registry({"orders_db": source})
+    query, _ = expand_tokens("SELECT {{total}} FROM orders", registry.for_source())
+    assert source.backend.connection.execute(query).fetchone()[0] == 450
+
+
+def test_every_corpus_definition_compiles_to_sql_duckdb_accepts():
+    """The emitter's output is parsed by DuckDB, not just compared to a string.
+
+    A fixture can only say the text matches what data-dict produced. This says
+    DuckDB can parse it. Binding it would need a real table per corpus
+    dictionary with matching column types, which is more fixture than the
+    question deserves, so this checks syntax only.
+    """
+    connection = duckdb.connect()
+    checked = 0
+    for path in sorted((SHARED_DIR / "definition-export" / "valid").glob("*.yaml")):
+        raw = yaml.safe_load(path.read_text(encoding="utf-8"))
+        dictionary = DataDictionary.model_validate(raw)
+        attach_compiled_definitions(dictionary, "duckdb")
+        for table in dictionary.tables.values():
+            for record in table.compiled_definitions:
+                # `json_serialize_sql` reports a parse failure in its result
+                # rather than raising, so the result is what gets asserted.
+                row = connection.execute(
+                    "SELECT json_serialize_sql(?)", [f"SELECT {record.sql}"]
+                ).fetchone()
+                assert row is not None
+                assert json.loads(row[0])["error"] is False, record.sql
+                checked += 1
+    assert checked == 42

From 5b4a53e63a1c1376717c3a11479a93a5e11aa80e Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Thu, 3 Sep 2026 23:00:55 -0600
Subject: [PATCH 2/4] fix(py): refuse definitions on an unexposed table at
 source construction

build_registry() already caught this, but it runs at agent construction,
by which time the compiled records have reached the dictionary's retrieval
chunks and can describe a table the agent cannot query. R checks it at bind
time for the same reason.

Prose about an unexposed table is still fine. Only a definition emits SQL
against a relation that has to be there.

Found by roborev job 323.
---
 pkg-py/src/commons/_data_source.py          |  2 +-
 pkg-py/src/commons/_definitions/_compile.py | 16 ++++++-
 pkg-py/tests/test_definition_compile.py     | 51 +++++++++++++++++++++
 3 files changed, 67 insertions(+), 2 deletions(-)

diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py
index ce79acc2..4493e830 100644
--- a/pkg-py/src/commons/_data_source.py
+++ b/pkg-py/src/commons/_data_source.py
@@ -276,7 +276,7 @@ def data_source(
         # The dialect is only known now, which is why lowering waits for it.
         from ._definitions import attach_compiled_definitions
 
-        attach_compiled_definitions(resolved, source.dialect())
+        attach_compiled_definitions(resolved, source.dialect(), set(source.tables))
     return source
 
 
diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py
index 4f132aee..fd96cad4 100644
--- a/pkg-py/src/commons/_definitions/_compile.py
+++ b/pkg-py/src/commons/_definitions/_compile.py
@@ -65,12 +65,20 @@ def _children(ir: Ir) -> list[Ir]:
     return out
 
 
-def attach_compiled_definitions(dictionary: Any, dialect: str) -> None:
+def attach_compiled_definitions(
+    dictionary: Any, dialect: str, exposed: set[str] | None = None
+) -> None:
     """Compile every governed definition for `dialect`, onto the dictionary.
 
     Called once a source is known, because the dialect decides the emitter.
     A dictionary with no definitions needs no emitter, so an unsupported
     dialect is only an error when there is something to lower.
+
+    `exposed` names the source's tables. A definition on a table outside them
+    would compile to SQL against a relation that is not there, so it is
+    refused here rather than at `build_registry()`: by then the compiled
+    records have already reached the dictionary's retrieval chunks. Prose
+    about an unexposed table is left alone, since only a definition emits SQL.
     """
     exports: dict[str, dict[str, DefinitionExport]] = (
         getattr(dictionary, "definition_exports", None) or {}
@@ -80,6 +88,12 @@ def attach_compiled_definitions(dictionary: Any, dialect: str) -> None:
         if not definitions:
             entry.compiled_definitions = []
             continue
+        if exposed is not None and table_name not in exposed:
+            raise ValueError(
+                f"The data dictionary declares definitions on table "
+                f"{table_name!r}, which the data source does not expose. "
+                f"Exposed tables: {', '.join(sorted(exposed))}."
+            )
         target = _TARGETS.get(dialect)
         if target is None:
             raise ValueError(
diff --git a/pkg-py/tests/test_definition_compile.py b/pkg-py/tests/test_definition_compile.py
index 39918919..59b6f52c 100644
--- a/pkg-py/tests/test_definition_compile.py
+++ b/pkg-py/tests/test_definition_compile.py
@@ -304,3 +304,54 @@ def test_every_corpus_definition_compiles_to_sql_duckdb_accepts():
                 assert json.loads(row[0])["error"] is False, record.sql
                 checked += 1
     assert checked == 42
+
+
+def test_definitions_on_a_table_the_source_does_not_expose_fail_construction(
+    tmp_path,
+):
+    """Caught at `data_source()`, not deferred to the registry.
+
+    Waiting for `build_registry()` would let the compiled records reach the
+    dictionary's retrieval chunks first, describing a table the agent cannot
+    query.
+    """
+    path = tmp_path / "data-dict.yaml"
+    path.write_text(
+        yaml.safe_dump(
+            {
+                "tables": [
+                    {
+                        "name": "elsewhere",
+                        "columns": [{"name": "amount", "type": "number"}],
+                        "definitions": [{"name": "big", "expr": "amount > 100"}],
+                    }
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+    with pytest.raises(ValueError, match="does not expose"):
+        data_source(orders=pd.DataFrame({"amount": [1]}), dictionary=path)
+
+
+def test_a_table_without_definitions_need_not_be_exposed(tmp_path):
+    # Prose about a table the source does not expose is the author's business;
+    # only a definition would produce SQL against something that is not there.
+    path = tmp_path / "data-dict.yaml"
+    path.write_text(
+        yaml.safe_dump(
+            {
+                "tables": [
+                    {"name": "elsewhere", "description": "Documented elsewhere."},
+                    {
+                        "name": "orders",
+                        "columns": [{"name": "amount", "type": "number"}],
+                        "definitions": [{"name": "big", "expr": "amount > 100"}],
+                    },
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+    source = data_source(orders=pd.DataFrame({"amount": [1]}), dictionary=path)
+    assert source.dictionary is not None

From 966cb21f7b5bf36cc3aaf486c57b728f99388b90 Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Sat, 5 Sep 2026 21:13:20 -0600
Subject: [PATCH 3/4] fix(py): make definition compilation atomic and pin the
 marker machinery

Review follow-ups:

- attach_compiled_definitions assigns nothing until every table
  compiles, so a refusal leaves the dictionary untouched and reusable
  for another source, and the exposed-table set is now required so the
  check cannot be skipped by omitting an argument.
- Tests now pin that a string literal holding marker text and a column
  named like a marker are both left alone; the previous test only
  covered a definition's own name, which is never a substitution
  target.
- Docstrings record where reading a dictionary and constructing a
  source can now fail, and the definition_exports comment names the
  actual value shape.
---
 pkg-py/src/commons/_data_dictionary.py      |  9 ++-
 pkg-py/src/commons/_data_source.py          |  5 ++
 pkg-py/src/commons/_definitions/_compile.py | 26 ++++---
 pkg-py/tests/test_definition_compile.py     | 76 +++++++++++++++++++--
 4 files changed, 100 insertions(+), 16 deletions(-)

diff --git a/pkg-py/src/commons/_data_dictionary.py b/pkg-py/src/commons/_data_dictionary.py
index 893719ab..0ae300ff 100644
--- a/pkg-py/src/commons/_data_dictionary.py
+++ b/pkg-py/src/commons/_data_dictionary.py
@@ -8,6 +8,10 @@
 
 Prose fields stay as authored markdown, because they reach the model verbatim.
 
+Reading a dictionary also type-checks any ``definitions:`` blocks against
+data-dict's expression language, so an unusable definition raises here,
+before any source exists. Only the lowering to SQL waits for a dialect.
+
 The three channels are methods rather than separate structures.
 ``pkg-r`` spreads the same rendering across ``R/data-dictionary.R``,
 ``R/prompt.R`` and ``R/context-layer.R``; here the dictionary owns it and the
@@ -130,8 +134,9 @@ class DataDictionary(_Permissive):
     glossary: dict[str, str] = {}
     # Phase 1 of the compiler, keyed by table name. Source-independent, so it
     # is produced here; the SQL needs a dialect and waits for `data_source()`.
-    # Values are `_definitions._export.DefinitionExport`, typed loosely so
-    # this module need not import the compiler's types.
+    # Values map each definition name to its
+    # `_definitions._export.DefinitionExport`, typed loosely so this module
+    # need not import the compiler's types.
     definition_exports: dict[str, Any] = {}
 
     @model_validator(mode="before")
diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py
index 4493e830..102bca64 100644
--- a/pkg-py/src/commons/_data_source.py
+++ b/pkg-py/src/commons/_data_source.py
@@ -247,6 +247,11 @@ def data_source(
     tables of the engine and board forms; `dictionary` attaches a data
     dictionary to any form. To use either as a frame name, call
     `DataSource.from_frames()` directly.
+
+    A dictionary's governed definitions are compiled for the source's
+    dialect here, so construction raises if the dialect has no emitter
+    (only DuckDB does today), if a definition sits on a table the source
+    does not expose, or if a metric mixes row and aggregate grain.
     """
     from ._data_dictionary import as_data_dictionary
 
diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py
index fd96cad4..d835f262 100644
--- a/pkg-py/src/commons/_definitions/_compile.py
+++ b/pkg-py/src/commons/_definitions/_compile.py
@@ -17,7 +17,7 @@
 
 __all__ = ["attach_compiled_definitions", "mixed_grain"]
 
-# Only DuckDB is ported so far; Snowflake and Databricks are stage 2.
+# Only DuckDB is supported so far; Snowflake and Databricks are stage 2.
 _TARGETS = {"duckdb": "SQL(duckdb)"}
 
 
@@ -66,7 +66,7 @@ def _children(ir: Ir) -> list[Ir]:
 
 
 def attach_compiled_definitions(
-    dictionary: Any, dialect: str, exposed: set[str] | None = None
+    dictionary: Any, dialect: str, exposed: set[str]
 ) -> None:
     """Compile every governed definition for `dialect`, onto the dictionary.
 
@@ -79,16 +79,20 @@ def attach_compiled_definitions(
     refused here rather than at `build_registry()`: by then the compiled
     records have already reached the dictionary's retrieval chunks. Prose
     about an unexposed table is left alone, since only a definition emits SQL.
+
+    Nothing is assigned until every table compiles, so a refusal leaves the
+    dictionary untouched and the caller can attach it to another source.
     """
     exports: dict[str, dict[str, DefinitionExport]] = (
         getattr(dictionary, "definition_exports", None) or {}
     )
-    for table_name, entry in dictionary.tables.items():
+    compiled: dict[str, list[ExportRecord]] = {}
+    for table_name in dictionary.tables:
         definitions = exports.get(table_name) or {}
         if not definitions:
-            entry.compiled_definitions = []
+            compiled[table_name] = []
             continue
-        if exposed is not None and table_name not in exposed:
+        if table_name not in exposed:
             raise ValueError(
                 f"The data dictionary declares definitions on table "
                 f"{table_name!r}, which the data source does not expose. "
@@ -101,7 +105,9 @@ def attach_compiled_definitions(
                 f"{dialect!r} data source. commons lowers definitions to DuckDB "
                 f"only."
             )
-        entry.compiled_definitions = _compile_table(table_name, definitions, target)
+        compiled[table_name] = _compile_table(table_name, definitions, target)
+    for table_name, entry in dictionary.tables.items():
+        entry.compiled_definitions = compiled[table_name]
 
 
 def _compile_table(
@@ -226,6 +232,8 @@ def _compose(
             if all(reference in composed for reference in definitions[name].definitions)
         ]
         if not ready:
+            # Unreachable while phase 1 rejects reference cycles; this guards
+            # a caller that builds export records by hand.
             raise ValueError(
                 f"Definitions on table {table!r} cannot be composed because "
                 f"their dependency graph is unresolved: "
@@ -275,13 +283,13 @@ def _substitute_identifiers(code: str, replacements: dict[str, str]) -> str:
     return "".join(out)
 
 
-def _take_quoted(code: str, start: str | int, quote: str) -> tuple[str, int]:
-    index = int(start) + 1
+def _take_quoted(code: str, start: int, quote: str) -> tuple[str, int]:
+    index = start + 1
     while index < len(code):
         if code[index] == quote:
             if index + 1 < len(code) and code[index + 1] == quote:
                 index += 2
                 continue
-            return code[int(start) : index + 1], index + 1
+            return code[start : index + 1], index + 1
         index += 1
     raise ValueError("Generated SQL contains an unterminated quoted value.")
diff --git a/pkg-py/tests/test_definition_compile.py b/pkg-py/tests/test_definition_compile.py
index 59b6f52c..6ad4a771 100644
--- a/pkg-py/tests/test_definition_compile.py
+++ b/pkg-py/tests/test_definition_compile.py
@@ -39,7 +39,7 @@ def dictionary_yaml(*definitions: dict, columns: list[dict] | None = None) -> di
 
 def compiled(*definitions: dict, **kwargs) -> list:
     dictionary = DataDictionary.model_validate(dictionary_yaml(*definitions, **kwargs))
-    attach_compiled_definitions(dictionary, "duckdb")
+    attach_compiled_definitions(dictionary, "duckdb", {"orders"})
     return dictionary.tables["orders"].compiled_definitions
 
 
@@ -149,6 +149,47 @@ def test_a_string_that_looks_like_a_definition_name_is_not_substituted():
     assert records["d"].sql.endswith("\"region\" = 'big'")
 
 
+def test_a_string_literal_holding_marker_text_is_not_substituted():
+    # `big` is the first definition, so its marker is
+    # `__commons_definition_reference_001__`. A string literal holding that
+    # exact text must survive: only the quoted identifier is a reference.
+    records = by_name(
+        compiled(
+            {"name": "big", "expr": "amount > 100"},
+            {
+                "name": "d",
+                "expr": "big and region = '__commons_definition_reference_001__'",
+            },
+        )
+    )
+    assert records["d"].sql == (
+        '("amount" > 100) AND "region" = \'__commons_definition_reference_001__\''
+    )
+
+
+def test_a_column_named_like_a_marker_is_not_confused_with_one():
+    # The marker pool skips names the dictionary already uses, so `big`'s
+    # marker grows a suffix and the real column is left alone.
+    marker_named_column = {
+        "name": "__commons_definition_reference_001__",
+        "type": "number",
+    }
+    records = by_name(
+        compiled(
+            {"name": "big", "expr": "amount > 100"},
+            {"name": "d", "expr": "big and __commons_definition_reference_001__ > 0"},
+            columns=[
+                {"name": "amount", "type": "number"},
+                {"name": "region", "type": "string"},
+                marker_named_column,
+            ],
+        )
+    )
+    assert records["d"].sql == (
+        '("amount" > 100) AND "__commons_definition_reference_001__" > 0'
+    )
+
+
 def test_an_unreferenced_definition_keeps_its_own_sql():
     records = by_name(compiled({"name": "d", "expr": "amount > 0"}))
     assert records["d"].sql == '"amount" > 0'
@@ -174,20 +215,20 @@ def test_the_record_carries_what_the_registry_needs():
 
 
 def test_a_dialect_with_no_emitter_is_refused():
-    # Only DuckDB is ported so far. A source commons cannot lower for must
+    # Only DuckDB is supported so far. A source commons cannot lower for must
     # fail at construction rather than emit the wrong dialect's SQL.
     with pytest.raises(ValueError, match="postgresql"):
         dictionary = DataDictionary.model_validate(
             dictionary_yaml({"name": "d", "expr": "amount > 0"})
         )
-        attach_compiled_definitions(dictionary, "postgresql")
+        attach_compiled_definitions(dictionary, "postgresql", {"orders"})
 
 
 def test_a_dictionary_with_no_definitions_needs_no_emitter():
     dictionary = DataDictionary.model_validate(
         {"tables": [{"name": "orders", "columns": [{"name": "amount"}]}]}
     )
-    attach_compiled_definitions(dictionary, "postgresql")
+    attach_compiled_definitions(dictionary, "postgresql", {"orders"})
     assert dictionary.tables["orders"].compiled_definitions == []
 
 
@@ -292,7 +333,7 @@ def test_every_corpus_definition_compiles_to_sql_duckdb_accepts():
     for path in sorted((SHARED_DIR / "definition-export" / "valid").glob("*.yaml")):
         raw = yaml.safe_load(path.read_text(encoding="utf-8"))
         dictionary = DataDictionary.model_validate(raw)
-        attach_compiled_definitions(dictionary, "duckdb")
+        attach_compiled_definitions(dictionary, "duckdb", set(dictionary.tables))
         for table in dictionary.tables.values():
             for record in table.compiled_definitions:
                 # `json_serialize_sql` reports a parse failure in its result
@@ -334,6 +375,31 @@ def test_definitions_on_a_table_the_source_does_not_expose_fail_construction(
         data_source(orders=pd.DataFrame({"amount": [1]}), dictionary=path)
 
 
+def test_a_refused_compile_leaves_the_dictionary_untouched():
+    # `orders` compiles before `elsewhere` is refused, so this only passes
+    # if nothing is assigned until every table succeeds — leaving the
+    # dictionary reusable for another source.
+    dictionary = DataDictionary.model_validate(
+        {
+            "tables": [
+                {
+                    "name": "orders",
+                    "columns": [{"name": "amount", "type": "number"}],
+                    "definitions": [{"name": "big", "expr": "amount > 100"}],
+                },
+                {
+                    "name": "elsewhere",
+                    "columns": [{"name": "amount", "type": "number"}],
+                    "definitions": [{"name": "big", "expr": "amount > 100"}],
+                },
+            ]
+        }
+    )
+    with pytest.raises(ValueError, match="does not expose"):
+        attach_compiled_definitions(dictionary, "duckdb", {"orders"})
+    assert dictionary.tables["orders"].compiled_definitions == []
+
+
 def test_a_table_without_definitions_need_not_be_exposed(tmp_path):
     # Prose about a table the source does not expose is the author's business;
     # only a definition would produce SQL against something that is not there.

From 0e56cab70705460313290580b2f83b8780f75b06 Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Sat, 5 Sep 2026 21:20:45 -0600
Subject: [PATCH 4/4] fix(py): inline references inside CASE branches, pinned
 by a shared contract
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The IR walkers that rename references and find children stepped into
lists of nodes and into mappings, but not into mappings inside lists —
which is how a CASE stores its branches. A definition that referenced a
sibling inside a CASE emitted the sibling's name as a quoted column
that does not exist, and the corpus's own enterprise_revenue compiled
to SQL no table can answer.

A shared fixture is what should have caught this, so definitions.json
gains a composed section: the composed DuckDB SQL and merged notes for
every valid corpus definition, asserted by both suites. Composition is
commons' own step, so the section is hand-maintained like mixed_grain
and the generator preserves it. The R suite's hardcoded composed-SQL
expectation is replaced by the fixture runner. Both implementations
agree on all 42 cases.
---
 pkg-py/src/commons/_definitions/_compile.py   |  22 +-
 pkg-py/tests/test_definition_compile.py       |  18 ++
 .../testthat/fixtures/shared/definitions.json | 195 ++++++++++++++++++
 .../tests/testthat/test-definition-compile.R  |  38 +++-
 scripts/generate-definitions-fixture.sh       |   8 +-
 tests/shared/README.md                        |   1 +
 tests/shared/definitions.json                 | 195 ++++++++++++++++++
 7 files changed, 463 insertions(+), 14 deletions(-)

diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py
index d835f262..8fdee34f 100644
--- a/pkg-py/src/commons/_definitions/_compile.py
+++ b/pkg-py/src/commons/_definitions/_compile.py
@@ -58,9 +58,13 @@ def _children(ir: Ir) -> list[Ir]:
         if isinstance(value, Ir):
             out.append(value)
         elif isinstance(value, list):
-            out.extend(item for item in value if isinstance(item, Ir))
+            for item in value:
+                if isinstance(item, Ir):
+                    out.append(item)
+                elif isinstance(item, dict):
+                    # A `CASE` branch is a mapping of condition and result.
+                    out.extend(v for v in item.values() if isinstance(v, Ir))
         elif isinstance(value, dict):
-            # A `CASE` branch is a mapping of condition and result.
             out.extend(item for item in value.values() if isinstance(item, Ir))
     return out
 
@@ -197,7 +201,19 @@ def _mark_references(ir: Ir, markers: dict[str, str]) -> Ir:
             attrs[key] = _mark_references(value, markers)
         elif isinstance(value, list):
             attrs[key] = [
-                _mark_references(item, markers) if isinstance(item, Ir) else item
+                _mark_references(item, markers)
+                if isinstance(item, Ir)
+                # A `CASE` branch is a mapping of condition and result.
+                else (
+                    {
+                        inner_key: _mark_references(inner, markers)
+                        if isinstance(inner, Ir)
+                        else inner
+                        for inner_key, inner in item.items()
+                    }
+                    if isinstance(item, dict)
+                    else item
+                )
                 for item in value
             ]
         elif isinstance(value, dict):
diff --git a/pkg-py/tests/test_definition_compile.py b/pkg-py/tests/test_definition_compile.py
index 6ad4a771..2ac04759 100644
--- a/pkg-py/tests/test_definition_compile.py
+++ b/pkg-py/tests/test_definition_compile.py
@@ -65,6 +65,24 @@ def test_grain_matches_the_shared_contract():
     assert checked == 42
 
 
+def test_composed_sql_matches_the_shared_contract():
+    fixture = load_shared_fixture("definitions")["composed"]
+    paths = sorted((SHARED_DIR / "definition-export" / "valid").glob("*.yaml"))
+    assert paths
+    checked = 0
+    for path in paths:
+        raw = yaml.safe_load(path.read_text(encoding="utf-8"))
+        dictionary = DataDictionary.model_validate(raw)
+        attach_compiled_definitions(dictionary, "duckdb", set(dictionary.tables))
+        for table in dictionary.tables.values():
+            for record in table.compiled_definitions:
+                expected = fixture[path.name][f"{record.table}::{record.name}"]
+                assert record.sql == expected["sql"], record.name
+                assert record.notes == expected["notes"], record.name
+                checked += 1
+    assert checked == 42
+
+
 def test_a_row_expression_holding_an_aggregate_is_mixed_grain():
     records = by_name(compiled({"name": "d", "expr": "amount > avg(amount)"}))
     assert records["d"].mixed_grain is True
diff --git a/pkg-r/tests/testthat/fixtures/shared/definitions.json b/pkg-r/tests/testthat/fixtures/shared/definitions.json
index a7e19ed3..bf279c25 100644
--- a/pkg-r/tests/testthat/fixtures/shared/definitions.json
+++ b/pkg-r/tests/testthat/fixtures/shared/definitions.json
@@ -1,4 +1,199 @@
 {
+  "composed": {
+    "core.yaml": {
+      "orders::enterprise_revenue": {
+        "notes": [
+          "DuckDB sums integers at 128 bits, so a total data-dict reports as an overflow (D09) may succeed."
+        ],
+        "sql": "sum(CASE WHEN (\"tile_size\" IN ('Mid-Market-3', 'Enterprise-1')) THEN \"order_total\" ELSE 0 END)"
+      },
+      "orders::is_enterprise": {
+        "notes": [],
+        "sql": "\"tile_size\" IN ('Mid-Market-3', 'Enterprise-1')"
+      },
+      "orders::list_price": {
+        "notes": [],
+        "sql": "\"order_total\" * 1.2"
+      },
+      "orders::net_revenue": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there.",
+          "DuckDB sums integers at 128 bits, so a total data-dict reports as an overflow (D09) may succeed."
+        ],
+        "sql": "sum(CASE WHEN \"status_cd\" = 90 THEN 0 ELSE \"order_total\" END)"
+      }
+    },
+    "functions.yaml": {
+      "values::boolean folds": {
+        "notes": [],
+        "sql": "bool_or(\"flag\") OR bool_and(\"flag\")"
+      },
+      "values::finite": {
+        "notes": [],
+        "sql": "isfinite(\"number\") AND NOT isinf(\"number\") AND NOT isnan(\"number\")"
+      },
+      "values::folds": {
+        "notes": [
+          "DuckDB sums integers at 128 bits, so a total data-dict reports as an overflow (D09) may succeed."
+        ],
+        "sql": "min(\"number\") + max(\"number\") + sum(\"number\") + avg(\"number\") + count(\"number\") + count(DISTINCT \"number\") + count(*)"
+      },
+      "values::numeric": {
+        "notes": [],
+        "sql": "abs(\"number\") + floor(\"number\") + ceil(\"number\") + round(\"number\", 2)"
+      },
+      "values::patterns": {
+        "notes": [],
+        "sql": "regexp_full_match(\"text\", '^A..*$') AND regexp_full_match(\"text\", 'A.*')"
+      },
+      "values::remainder": {
+        "notes": [
+          "DuckDB yields null for an integer modulus by zero, where data-dict yields a NaN."
+        ],
+        "sql": "mod(mod(\"number\", 3) + 3, 3)"
+      },
+      "values::strings": {
+        "notes": [],
+        "sql": "starts_with(lower(trim(\"text\")), 'a') OR ends_with(upper(\"text\"), 'Z')"
+      }
+    },
+    "language.yaml": {
+      "survey::amount band": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" <= 2 * min(\"amount\")"
+      },
+      "survey::anything missing": {
+        "notes": [],
+        "sql": "\"q1\" IS NOT NULL AND \"q2\" IS NOT NULL AND \"created\" IS NOT NULL AND \"observed\" IS NOT NULL AND \"amount\" IS NOT NULL AND \"ratio\" IS NOT NULL AND \"pattern\" IS NOT NULL AND \"category\" IS NOT NULL AND \"profile\" IS NOT NULL AND \"tags\" IS NOT NULL AND \"untyped\" IS NOT NULL"
+      },
+      "survey::case without else": {
+        "notes": [],
+        "sql": "CASE WHEN \"q1\" THEN \"amount\" END"
+      },
+      "survey::category A": {
+        "notes": [],
+        "sql": "\"category\" = 'A'"
+      },
+      "survey::complete": {
+        "notes": [],
+        "sql": "\"q1\" IS NOT NULL AND \"q2\" IS NOT NULL"
+      },
+      "survey::computed interval": {
+        "notes": [],
+        "sql": "\"observed\" + (\"amount\" + 1 * INTERVAL '1 days')"
+      },
+      "survey::constant": {
+        "notes": [],
+        "sql": "1"
+      },
+      "survey::deep score": {
+        "notes": [],
+        "sql": "\"profile\".\"geo\".\"latitude\" + length(\"profile\".\"nick names\")"
+      },
+      "survey::dynamic match": {
+        "notes": [],
+        "sql": "\"pattern\" LIKE \"profile\".\"zip\""
+      },
+      "survey::exact match": {
+        "notes": [],
+        "sql": "\"pattern\" = 'a'"
+      },
+      "survey::fractional interval": {
+        "notes": [],
+        "sql": "\"observed\" - (1.5 * INTERVAL '1 hours')"
+      },
+      "survey::fractional time": {
+        "notes": [],
+        "sql": "\"observed\" <= TIMESTAMP '2024-01-01 01:30:00.123'"
+      },
+      "survey::fresh": {
+        "notes": [],
+        "sql": "\"observed\" >= current_timestamp - INTERVAL '2 weeks'"
+      },
+      "survey::long postal": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "(length(\"profile\".\"zip\")) > 5"
+      },
+      "survey::missing tags": {
+        "notes": [],
+        "sql": "\"tags\" IS NULL"
+      },
+      "survey::missing unknown": {
+        "notes": [],
+        "sql": "\"untyped\" IS NULL"
+      },
+      "survey::mixed temporal case": {
+        "notes": [],
+        "sql": "CASE WHEN \"q1\" THEN \"created\" ELSE \"observed\" END"
+      },
+      "survey::negative infinity": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" > -CAST('Infinity' AS DOUBLE)"
+      },
+      "survey::negative quotient": {
+        "notes": [],
+        "sql": "-(\"amount\" - 2) / (\"ratio\" + 1.0)"
+      },
+      "survey::null logic": {
+        "notes": [],
+        "sql": "NOT \"q1\" OR NULL IS NULL"
+      },
+      "survey::nullable date shift": {
+        "notes": [],
+        "sql": "\"created\" + NULL"
+      },
+      "survey::offset time": {
+        "notes": [],
+        "sql": "\"observed\" >= TIMESTAMP '2024-01-01 07:30:00'"
+      },
+      "survey::postal length": {
+        "notes": [],
+        "sql": "length(\"profile\".\"zip\")"
+      },
+      "survey::precise threshold": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" > 0.12345678901234566"
+      },
+      "survey::prefix match": {
+        "notes": [],
+        "sql": "starts_with(\"pattern\", 'a')"
+      },
+      "survey::range predicate": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" BETWEEN -10 AND 10 AND \"amount\" NOT IN (0, 1)"
+      },
+      "survey::recent": {
+        "notes": [],
+        "sql": "\"created\" >= DATE '2024-01-01'"
+      },
+      "survey::right associative arithmetic": {
+        "notes": [],
+        "sql": "\"amount\" - (\"ratio\" - 1)"
+      },
+      "survey::selected dates": {
+        "notes": [],
+        "sql": "\"created\" >= '2020-01-01'"
+      },
+      "survey::suffix mismatch": {
+        "notes": [],
+        "sql": "NOT ends_with(\"pattern\", 'z')"
+      },
+      "survey::untyped count": {
+        "notes": [],
+        "sql": "count(\"untyped\")"
+      }
+    }
+  },
   "corpus_dir": "definition-export",
   "data_dict_commit": "d950c5ac90d0ab939d330600f3a5ee1bfde0f604",
   "export_records": {
diff --git a/pkg-r/tests/testthat/test-definition-compile.R b/pkg-r/tests/testthat/test-definition-compile.R
index 4c7f462e..2458e0d6 100644
--- a/pkg-r/tests/testthat/test-definition-compile.R
+++ b/pkg-r/tests/testthat/test-definition-compile.R
@@ -38,14 +38,6 @@ test_that("definitions compile and compose for a DuckDB source", {
   definitions <- definition_compiled_table(compiled)
 
   expect_equal(compiled$target, "SQL(duckdb)")
-  expect_equal(
-    definitions$enterprise_revenue$sql,
-    paste0(
-      "sum(CASE WHEN (\"tile_size\" IN ",
-      "('Mid-Market-3', 'Enterprise-1')) ",
-      "THEN \"order_total\" ELSE 0 END)"
-    )
-  )
   expect_equal(
     definition_translation(
       definitions$enterprise_revenue,
@@ -74,6 +66,36 @@ test_that("definitions compile and compose for a DuckDB source", {
   expect_equal(enterprise_revenue$value, 100)
 })
 
+test_that("composed SQL and notes match the shared contract", {
+  skip_if_not_installed("yaml")
+  fixture <- shared_fixture("definitions")$composed
+  checked <- 0L
+  for (path in definition_fixture_paths("valid")) {
+    raw <- yaml::read_yaml(path)
+    tables <- vapply(raw$tables, `[[`, "", "name")
+    frames <- stats::setNames(
+      lapply(tables, function(table) data.frame(x = 1)),
+      tables
+    )
+    source <- do.call(data_source, frames)
+    compiled <- definition_compile_source(raw, source)
+    expected <- fixture[[basename(path)]]
+    for (table in compiled$tables) {
+      for (definition in table$definitions) {
+        key <- paste0(table$name, "::", definition$name)
+        notes <- expected[[key]]$notes
+        expect_equal(definition$sql, expected[[key]]$sql)
+        expect_equal(
+          definition$notes,
+          if (length(notes)) unlist(notes) else character(0)
+        )
+        checked <- checked + 1L
+      }
+    }
+  }
+  expect_equal(checked, 42L)
+})
+
 test_that("source compilation rejects metrics over mixed-grain definitions", {
   expect_error(
     definitions_source(
diff --git a/scripts/generate-definitions-fixture.sh b/scripts/generate-definitions-fixture.sh
index 4a58c7cd..66da05dc 100755
--- a/scripts/generate-definitions-fixture.sh
+++ b/scripts/generate-definitions-fixture.sh
@@ -3,9 +3,10 @@
 # pinned data-dict binary.
 #
 # `export_records` is data-dict's own output, projected to the fields both
-# packages consume. `mixed_grain` and `invalid` are not in that output:
-# grain is derived from the typed IR, and the problem codes come from
-# validate-spec. Both are hand-maintained and this script preserves them.
+# packages consume. `mixed_grain`, `composed`, and `invalid` are not in that
+# output: grain is derived from the typed IR, composition is commons' own
+# step, and the problem codes come from validate-spec. All three are
+# hand-maintained and this script preserves them.
 #
 # The binary is the authority. Regenerating against a build from any other
 # revision would quietly bless whatever that build does.
@@ -94,6 +95,7 @@ spec = {
     "corpus_dir": "definition-export",
     "export_records": records,
     "mixed_grain": existing.get("mixed_grain", {}),
+    "composed": existing.get("composed", {}),
     "invalid": existing.get("invalid", {}),
 }
 out.write_text(json.dumps(spec, indent=2, sort_keys=True) + "\n")
diff --git a/tests/shared/README.md b/tests/shared/README.md
index 44b1d522..b519d197 100644
--- a/tests/shared/README.md
+++ b/tests/shared/README.md
@@ -28,6 +28,7 @@ The Python suite reads this directory directly. The R suite cannot. `testthat` n
 - **The definitions interface.** `definition-export/` holds the shared fixtures: 14 data dictionaries in data-dict's YAML format, each declaring table-level `definitions` whose expressions use data-dict's expression language — 3 valid files (42 definitions) and 11 invalid ones — read by both commons implementations. `definitions.json` pins what both packages agree to produce from them, in three sections:
   - `export_records` — the expected export for each valid definition: its SQL translation, its inferred kind and type, and the columns and definitions it references. Generated from the data-dict binary at the pinned commit by `scripts/generate-definitions-fixture.sh`, which refuses to run against a binary built from anything else. Never hand-edit.
   - `mixed_grain` — a per-definition boolean for `call_metrics`' mixed-grain guard: true when a definition's exported shape is `row` but its expression contains an aggregate, directly or through a definition it references. Absent from data-dict's export (it exists only in the compiler's internal parse tree), so it is hand-maintained and the generator preserves it.
+  - `composed` — the composed DuckDB SQL and merged notes for each valid definition: what the compiler produces after inlining a definition's sibling references. Composition is commons' own step, so it is absent from data-dict's export; hand-maintained and the generator preserves it.
   - `invalid` — the data-dict problem code each invalid fixture must produce (e.g. `cycle.yaml` must fail with the cycle error, not a generic parse failure). Hand-maintained; the generator preserves it.
 
   This fixture does not replace the conformance harness: the harness compares against a real binary, while this pins what both packages agree to consume.
diff --git a/tests/shared/definitions.json b/tests/shared/definitions.json
index a7e19ed3..bf279c25 100644
--- a/tests/shared/definitions.json
+++ b/tests/shared/definitions.json
@@ -1,4 +1,199 @@
 {
+  "composed": {
+    "core.yaml": {
+      "orders::enterprise_revenue": {
+        "notes": [
+          "DuckDB sums integers at 128 bits, so a total data-dict reports as an overflow (D09) may succeed."
+        ],
+        "sql": "sum(CASE WHEN (\"tile_size\" IN ('Mid-Market-3', 'Enterprise-1')) THEN \"order_total\" ELSE 0 END)"
+      },
+      "orders::is_enterprise": {
+        "notes": [],
+        "sql": "\"tile_size\" IN ('Mid-Market-3', 'Enterprise-1')"
+      },
+      "orders::list_price": {
+        "notes": [],
+        "sql": "\"order_total\" * 1.2"
+      },
+      "orders::net_revenue": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there.",
+          "DuckDB sums integers at 128 bits, so a total data-dict reports as an overflow (D09) may succeed."
+        ],
+        "sql": "sum(CASE WHEN \"status_cd\" = 90 THEN 0 ELSE \"order_total\" END)"
+      }
+    },
+    "functions.yaml": {
+      "values::boolean folds": {
+        "notes": [],
+        "sql": "bool_or(\"flag\") OR bool_and(\"flag\")"
+      },
+      "values::finite": {
+        "notes": [],
+        "sql": "isfinite(\"number\") AND NOT isinf(\"number\") AND NOT isnan(\"number\")"
+      },
+      "values::folds": {
+        "notes": [
+          "DuckDB sums integers at 128 bits, so a total data-dict reports as an overflow (D09) may succeed."
+        ],
+        "sql": "min(\"number\") + max(\"number\") + sum(\"number\") + avg(\"number\") + count(\"number\") + count(DISTINCT \"number\") + count(*)"
+      },
+      "values::numeric": {
+        "notes": [],
+        "sql": "abs(\"number\") + floor(\"number\") + ceil(\"number\") + round(\"number\", 2)"
+      },
+      "values::patterns": {
+        "notes": [],
+        "sql": "regexp_full_match(\"text\", '^A..*$') AND regexp_full_match(\"text\", 'A.*')"
+      },
+      "values::remainder": {
+        "notes": [
+          "DuckDB yields null for an integer modulus by zero, where data-dict yields a NaN."
+        ],
+        "sql": "mod(mod(\"number\", 3) + 3, 3)"
+      },
+      "values::strings": {
+        "notes": [],
+        "sql": "starts_with(lower(trim(\"text\")), 'a') OR ends_with(upper(\"text\"), 'Z')"
+      }
+    },
+    "language.yaml": {
+      "survey::amount band": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" <= 2 * min(\"amount\")"
+      },
+      "survey::anything missing": {
+        "notes": [],
+        "sql": "\"q1\" IS NOT NULL AND \"q2\" IS NOT NULL AND \"created\" IS NOT NULL AND \"observed\" IS NOT NULL AND \"amount\" IS NOT NULL AND \"ratio\" IS NOT NULL AND \"pattern\" IS NOT NULL AND \"category\" IS NOT NULL AND \"profile\" IS NOT NULL AND \"tags\" IS NOT NULL AND \"untyped\" IS NOT NULL"
+      },
+      "survey::case without else": {
+        "notes": [],
+        "sql": "CASE WHEN \"q1\" THEN \"amount\" END"
+      },
+      "survey::category A": {
+        "notes": [],
+        "sql": "\"category\" = 'A'"
+      },
+      "survey::complete": {
+        "notes": [],
+        "sql": "\"q1\" IS NOT NULL AND \"q2\" IS NOT NULL"
+      },
+      "survey::computed interval": {
+        "notes": [],
+        "sql": "\"observed\" + (\"amount\" + 1 * INTERVAL '1 days')"
+      },
+      "survey::constant": {
+        "notes": [],
+        "sql": "1"
+      },
+      "survey::deep score": {
+        "notes": [],
+        "sql": "\"profile\".\"geo\".\"latitude\" + length(\"profile\".\"nick names\")"
+      },
+      "survey::dynamic match": {
+        "notes": [],
+        "sql": "\"pattern\" LIKE \"profile\".\"zip\""
+      },
+      "survey::exact match": {
+        "notes": [],
+        "sql": "\"pattern\" = 'a'"
+      },
+      "survey::fractional interval": {
+        "notes": [],
+        "sql": "\"observed\" - (1.5 * INTERVAL '1 hours')"
+      },
+      "survey::fractional time": {
+        "notes": [],
+        "sql": "\"observed\" <= TIMESTAMP '2024-01-01 01:30:00.123'"
+      },
+      "survey::fresh": {
+        "notes": [],
+        "sql": "\"observed\" >= current_timestamp - INTERVAL '2 weeks'"
+      },
+      "survey::long postal": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "(length(\"profile\".\"zip\")) > 5"
+      },
+      "survey::missing tags": {
+        "notes": [],
+        "sql": "\"tags\" IS NULL"
+      },
+      "survey::missing unknown": {
+        "notes": [],
+        "sql": "\"untyped\" IS NULL"
+      },
+      "survey::mixed temporal case": {
+        "notes": [],
+        "sql": "CASE WHEN \"q1\" THEN \"created\" ELSE \"observed\" END"
+      },
+      "survey::negative infinity": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" > -CAST('Infinity' AS DOUBLE)"
+      },
+      "survey::negative quotient": {
+        "notes": [],
+        "sql": "-(\"amount\" - 2) / (\"ratio\" + 1.0)"
+      },
+      "survey::null logic": {
+        "notes": [],
+        "sql": "NOT \"q1\" OR NULL IS NULL"
+      },
+      "survey::nullable date shift": {
+        "notes": [],
+        "sql": "\"created\" + NULL"
+      },
+      "survey::offset time": {
+        "notes": [],
+        "sql": "\"observed\" >= TIMESTAMP '2024-01-01 07:30:00'"
+      },
+      "survey::postal length": {
+        "notes": [],
+        "sql": "length(\"profile\".\"zip\")"
+      },
+      "survey::precise threshold": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" > 0.12345678901234566"
+      },
+      "survey::prefix match": {
+        "notes": [],
+        "sql": "starts_with(\"pattern\", 'a')"
+      },
+      "survey::range predicate": {
+        "notes": [
+          "DuckDB compares a NaN as equal to itself and greater than every number, where data-dict answers false; a row holding one passes here and is reported there."
+        ],
+        "sql": "\"amount\" BETWEEN -10 AND 10 AND \"amount\" NOT IN (0, 1)"
+      },
+      "survey::recent": {
+        "notes": [],
+        "sql": "\"created\" >= DATE '2024-01-01'"
+      },
+      "survey::right associative arithmetic": {
+        "notes": [],
+        "sql": "\"amount\" - (\"ratio\" - 1)"
+      },
+      "survey::selected dates": {
+        "notes": [],
+        "sql": "\"created\" >= '2020-01-01'"
+      },
+      "survey::suffix mismatch": {
+        "notes": [],
+        "sql": "NOT ends_with(\"pattern\", 'z')"
+      },
+      "survey::untyped count": {
+        "notes": [],
+        "sql": "count(\"untyped\")"
+      }
+    }
+  },
   "corpus_dir": "definition-export",
   "data_dict_commit": "d950c5ac90d0ab939d330600f3a5ee1bfde0f604",
   "export_records": {