add Pydantic field metadata, pipeline schema validation, col: tag namespace, and stack.toml config

- SQLTable: add column_meta(), field_descriptions(), to_ddl(), _resolve_type()
- generate_models.py: read DuckDB column comments, emit Field(description=...)
- run_pipeline: validate output columns against SQLTable contract (SchemaError)
- Pipeline.run: pass output class through to runner
- Tag.col(): new namespace for column-level descriptions
- bib/meta: collect_column_comments() and apply_column_comments() for Zotero→DuckDB flow
- stack.toml + src/conf/: centralised config loader with attribute access and path()
- 100% test coverage on all changed files

dev/scripts/generate_models.py

@@ -13,7 +13,6 @@ from __future__ import annotations
 
 import argparse
 import re
-import textwrap
 from pathlib import Path
 
 import duckdb
@@ -96,8 +95,8 @@ def get_tables(con: duckdb.DuckDBPyConnection, schema: str) -> list[str]:
 
 def get_columns(
     con: duckdb.DuckDBPyConnection, schema: str, table: str
-) -> list[tuple[str, str, bool]]:
+) -> list[tuple[str, str, bool, str]]:
-    """Return [(col_name, duckdb_type, nullable), ...]."""
+    """Return [(col_name, duckdb_type, nullable, comment), ...]."""
     rows = con.execute(
         "SELECT column_name, data_type, is_nullable "
         "FROM information_schema.columns "
@@ -105,29 +104,65 @@ def get_columns(
         "ORDER BY ordinal_position",
         [schema, table],
     ).fetchall()
-    return [(r[0], r[1], r[2] == "YES") for r in rows]
+    cols = [(r[0], r[1], r[2] == "YES") for r in rows]
+
+    # Try to read column comments from duckdb_columns()
+    comments: dict[str, str] = {}
+    try:
+        crows = con.execute(
+            "SELECT column_name, comment FROM duckdb_columns() "
+            "WHERE schema_name = ? AND table_name = ? "
+            "AND comment IS NOT NULL",
+            [schema, table],
+        ).fetchall()
+        comments = {r[0]: r[1] for r in crows}
+    except Exception:
+        pass
+
+    return [
+        (name, dtype, nullable, comments.get(name, ""))
+        for name, dtype, nullable in cols
+    ]
 
 
 # ── Code generation ─────────────────────────────────────────────────────────
 
 
 def generate_model(
-    schema: str, table: str, columns: list[tuple[str, str, bool]]
+    schema: str, table: str, columns: list[tuple[str, str, bool, str]]
 ) -> str:
     """Generate a single Pydantic model class."""
     cls = to_class_name(schema, table)
     all_imports: set[str] = set()
     field_lines: list[str] = []
+    uses_field = False
 
-    for col_name, col_type, nullable in columns:
+    for col_name, col_type, nullable, comment in columns:
         py = python_type(col_type)
         all_imports |= needs_import(py)
         fname = sanitize_field(col_name)
-        if nullable:
+
-            annotation = f"{py} | None = None"
+        if comment:
+            uses_field = True
+            escaped = comment.replace('"', '\\"')
+            if nullable:
+                field_lines.append(
+                    f'    {fname}: {py} | None = Field('
+                    f'default=None, description="{escaped}")'
+                )
+            else:
+                field_lines.append(
+                    f'    {fname}: {py} = Field('
+                    f'description="{escaped}")'
+                )
         else:
-            annotation = py
+            if nullable:
-        field_lines.append(f"    {fname}: {annotation}")
+                field_lines.append(f"    {fname}: {py} | None = None")
+            else:
+                field_lines.append(f"    {fname}: {py}")
+
+    if uses_field:
+        all_imports.add("from pydantic import Field")
 
     lines = []
     lines.append(f"class {cls}(SQLTable):")
@@ -167,39 +202,6 @@ def generate_schema_module(
     return f"{header}\n\n\n{body}\n"
 
 
-# ── Base class ──────────────────────────────────────────────────────────────
-
-BASE_MODULE = textwrap.dedent('''\
-    """Base class for all generated SQLTable models."""
-
-    from __future__ import annotations
-
-    from pydantic import BaseModel, ConfigDict
-
-
-    class SQLTable(BaseModel):
-        """Pydantic model representing a database table.
-
-        Subclasses set __schema__ and __tablename__ as class-level metadata.
-        All fields are nullable by default since DuckDB columns are nullable.
-        """
-
-        model_config = ConfigDict(from_attributes=True)
-
-        __schema__: str = ""
-        __tablename__: str = ""
-
-        @classmethod
-        def qualified_name(cls) -> str:
-            return f"{cls.__schema__}.{cls.__tablename__}"
-
-        @classmethod
-        def column_names(cls) -> list[str]:
-            return [
-                name
-                for name in cls.model_fields
-            ]
-''')
 
 
 # ── Main ────────────────────────────────────────────────────────────────────
@@ -224,11 +226,6 @@ def main() -> None:
 
     con = duckdb.connect(str(db_path), read_only=True)
 
-    # Write base module into the output directory itself
-    base_path = out_dir / "base.py"
-    base_path.write_text(BASE_MODULE)
-    print(f"Wrote {base_path}")
-
     # Generate per-schema modules
     schemas = get_schemas(con)
     all_modules: list[str] = []

src/aco/pipe/base.py

@@ -25,7 +25,7 @@ class Pipeline(BaseModel):
         """Execute expressions in order, returning cache of DataFrames."""
         from aco.pipe.runner import run_pipeline
 
-        return run_pipeline([(e.name, e.fn) for e in self.exprs], load)
+        return run_pipeline([(e.name, e.fn, e.output) for e in self.exprs], load)
 
     def names(self) -> list[str]:
         """Return qualified output names for all expressions."""

src/aco/pipe/runner.py

@@ -4,6 +4,10 @@ import inspect
 from typing import Any, Callable
 
 
+class SchemaError(Exception):
+    """Raised when a pipeline expression output doesn't match its contract."""
+
+
 def _param_to_table(param: str) -> str:
     """Convert function parameter name to qualified table reference.
 
@@ -38,20 +42,53 @@ def make_databricks_loader(spark: Any, catalog: str) -> Callable[[str], Any]:
     return _load
 
 
+def _validate_output(
+    output_name: str,
+    result: Any,
+    output_cls: type | None,
+) -> None:
+    """Check that result columns match the output contract."""
+    if output_cls is None:
+        return
+
+    expected = set(output_cls.column_names())
+    actual = set(result.columns)
+
+    missing = expected - actual
+    extra = actual - expected
+
+    if missing or extra:
+        parts = [f"{output_name} schema mismatch:"]
+        if missing:
+            parts.append(f"  missing columns: {sorted(missing)}")
+        if extra:
+            parts.append(f"  extra columns: {sorted(extra)}")
+        raise SchemaError("\n".join(parts))
+
+
 def run_pipeline(
-    exprs: list[tuple[str, Callable]],
+    exprs: list[tuple[str, Callable] | tuple[str, Callable, type | None]],
     load: Callable[[str], Any],
 ) -> dict[str, Any]:
     """Execute an ordered list of pipeline expressions.
 
-    Each expr is a (output_name, function) tuple. Dependencies are
+    Each expr is a (output_name, function) or (output_name, function,
-    resolved from the cache (prior outputs) or by calling load()
+    output_cls) tuple.  When output_cls is provided, the result columns
-    for external tables.
+    are validated against the SQLTable contract after execution.
+
+    Dependencies are resolved from the cache (prior outputs) or by
+    calling load() for external tables.
 
     Returns the cache dict mapping output_name -> DataFrame.
     """
     cache: dict[str, Any] = {}
-    for output_name, fn in exprs:
+    for expr in exprs:
+        if len(expr) == 3:
+            output_name, fn, output_cls = expr
+        else:
+            output_name, fn = expr
+            output_cls = None
+
         sig = inspect.signature(fn)
         kwargs = {}
         for param in sig.parameters:
@@ -60,5 +97,9 @@ def run_pipeline(
                 kwargs[param] = cache[table_ref]
             else:
                 kwargs[param] = load(table_ref)
-        cache[output_name] = fn(**kwargs)
+
+        result = fn(**kwargs)
+        _validate_output(output_name, result, output_cls)
+        cache[output_name] = result
+
     return cache

src/aco/table/base.py

@@ -2,14 +2,55 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from pydantic import BaseModel, ConfigDict
 
+# Python type → DuckDB type mapping for DDL generation.
+_DDL_TYPE_MAP: dict[type, str] = {
+    str: "VARCHAR",
+    int: "INTEGER",
+    float: "DOUBLE",
+    bool: "BOOLEAN",
+}
+
+# Lazy-resolved types (avoid top-level imports of date/datetime/Decimal
+# so that modules that only need the base class stay lightweight).
+_DDL_TYPE_NAME_MAP: dict[str, str] = {
+    "date": "DATE",
+    "datetime": "TIMESTAMP",
+    "Decimal": "DECIMAL(18,2)",
+}
+
+
+def _resolve_type(annotation: Any) -> str:
+    """Map a field annotation to its DuckDB SQL type."""
+    import types
+
+    # Handle T | None (union types)
+    if isinstance(annotation, types.UnionType):
+        args = [a for a in annotation.__args__ if a is not type(None)]
+        if args:
+            return _resolve_type(args[0])
+
+    # Direct type match
+    if isinstance(annotation, type):
+        if annotation in _DDL_TYPE_MAP:
+            return _DDL_TYPE_MAP[annotation]
+        if annotation.__name__ in _DDL_TYPE_NAME_MAP:
+            return _DDL_TYPE_NAME_MAP[annotation.__name__]
+
+    return "VARCHAR"
+
 
 class SQLTable(BaseModel):
     """Pydantic model representing a database table.
 
     Subclasses set __schema__ and __tablename__ as class-level metadata.
     All fields are nullable by default since DuckDB columns are nullable.
+
+    Field-level metadata is stored via ``pydantic.Field`` and can be
+    queried through the classmethods below.
     """
 
     model_config = ConfigDict(from_attributes=True)
@@ -24,3 +65,47 @@ class SQLTable(BaseModel):
     @classmethod
     def column_names(cls) -> list[str]:
         return [name for name in cls.model_fields]
+
+    @classmethod
+    def column_meta(cls) -> dict[str, dict[str, Any]]:
+        """Return per-column metadata from Field definitions.
+
+        Each key is a column name, value is a dict with:
+        - ``description``: field description (empty string if unset)
+        - ``extra``: contents of ``json_schema_extra`` (empty dict if unset)
+        - ``type``: Python type annotation
+        """
+        result: dict[str, dict[str, Any]] = {}
+        for name, info in cls.model_fields.items():
+            extra = info.json_schema_extra or {}
+            result[name] = {
+                "description": info.description or "",
+                "extra": extra if isinstance(extra, dict) else {},
+                "type": info.annotation,
+            }
+        return result
+
+    @classmethod
+    def to_ddl(cls) -> str:
+        """Generate a CREATE TABLE statement from the model definition."""
+        items = list(cls.model_fields.items())
+        lines: list[str] = []
+        for i, (name, info) in enumerate(items):
+            sql_type = _resolve_type(info.annotation)
+            comma = "," if i < len(items) - 1 else ""
+            desc = info.description
+            line = f"    {name} {sql_type}{comma}"
+            if desc:
+                line += f"  -- {desc}"
+            lines.append(line)
+        col_block = "\n".join(lines)
+        return f"CREATE TABLE {cls.qualified_name()} (\n{col_block}\n);"
+
+    @classmethod
+    def field_descriptions(cls) -> dict[str, str]:
+        """Return {column: description} for columns that have descriptions."""
+        return {
+            name: info.description
+            for name, info in cls.model_fields.items()
+            if info.description
+        }

src/bib/meta.py

@@ -32,7 +32,7 @@ Usage::
 from __future__ import annotations
 
 import re
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from bib.tag import Tag
 
@@ -280,6 +280,68 @@ def generate_docstring_refs(
     return "\n".join(lines)
 
 
+def collect_column_comments(store: Store) -> dict[str, str]:
+    """Collect column descriptions from ``col:`` tags in the bib store.
+
+    Scans all tags in the ``col`` namespace and parses the
+    ``schema.table.column=description`` format.
+
+    Returns
+    -------
+    dict[str, str]
+        Mapping of ``schema.table.column`` → description.
+    """
+    result: dict[str, str] = {}
+    for tag_info in store.list_tags(namespace="col"):
+        label = tag_info["name"]
+        # label is "col:schema.table.column=description"
+        _, _, payload = label.partition(":")
+        if "=" not in payload:
+            continue
+        column_ref, _, description = payload.partition("=")
+        if column_ref and description:
+            result[column_ref] = description
+    return result
+
+
+def apply_column_comments(
+    con: Any,
+    store: Store,
+) -> list[str]:
+    """Apply ``col:`` tag descriptions as DuckDB column comments.
+
+    Reads all ``col:`` tags from the bib store and executes
+    ``COMMENT ON COLUMN`` statements on the DuckDB connection.
+
+    Parameters
+    ----------
+    con : duckdb.DuckDBPyConnection
+        Open DuckDB connection (read-write).
+    store : Store
+        Bibliography store to read ``col:`` tags from.
+
+    Returns
+    -------
+    list[str]
+        SQL statements that were executed.
+    """
+    comments = collect_column_comments(store)
+    stmts: list[str] = []
+    for col_ref, description in sorted(comments.items()):
+        parts = col_ref.split(".", 2)
+        if len(parts) != 3:
+            continue
+        schema, table, column = parts
+        escaped = description.replace("'", "''")
+        sql = f'COMMENT ON COLUMN "{schema}"."{table}"."{column}" IS \'{escaped}\''
+        try:
+            con.execute(sql)
+            stmts.append(sql)
+        except Exception:
+            pass
+    return stmts
+
+
 def generate_pipeline_bibliography(
     pipeline: object,
     store: Store,

src/bib/tag.py

@@ -200,6 +200,24 @@ class Tag(BaseModel):
         """
         return cls(namespace="meta", value=citation_ref)
 
+    @classmethod
+    def col(cls, column_ref: str, description: str) -> Tag:
+        """Tag documenting a column with a human-readable description.
+
+        Parameters
+        ----------
+        column_ref : str
+            Qualified column path: ``schema.table.column``.
+        description : str
+            Short description of the column.
+
+        Examples::
+
+            Tag.col("core.encounter.encounter_id", "Unique encounter ID")
+            # col:core.encounter.encounter_id=Unique encounter ID
+        """
+        return cls(namespace="col", value=f"{column_ref}={description}")
+
     @classmethod
     def fn(cls, qualified_name: str) -> Tag:
         """Tag linking a bibliography item to an express function.

src/conf/__init__.py

@@ -0,0 +1,102 @@
+"""Centralised configuration loader for stack.toml.
+
+Reads ``stack.toml`` from the repository root and exposes its
+sections as plain dicts.  Sections are accessed via attribute or
+key lookup on the module-level ``cfg`` object::
+
+    from conf import cfg
+
+    cfg.db.aco          # "notebooks/aco.duckdb"
+    cfg.bcda.timeout    # 120.0
+    cfg["lake"]["trino"]  # {"host": "trino", "port": 8080, ...}
+
+Paths can be resolved to absolute ``Path`` objects::
+
+    from conf import path
+
+    path("db.aco")      # Path("/home/.../notebooks/aco.duckdb")
+    path("storage.bib") # Path("/home/.../data/bib/storage")
+"""
+
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+from typing import Any
+
+
+def _find_root() -> Path:
+    """Walk up from this file to find the directory containing stack.toml."""
+    d = Path(__file__).resolve().parent
+    for _ in range(10):
+        if (d / "stack.toml").exists():
+            return d
+        d = d.parent
+    raise FileNotFoundError("stack.toml not found in any parent directory")
+
+
+ROOT = _find_root()
+
+
+class _Cfg:
+    """Thin wrapper around a parsed TOML dict with attribute access."""
+
+    def __init__(self, data: dict[str, Any]) -> None:
+        self._data = data
+
+    def __getattr__(self, key: str) -> Any:
+        try:
+            val = self._data[key]
+        except KeyError:
+            raise AttributeError(key) from None
+        if isinstance(val, dict):
+            return _Cfg(val)
+        return val
+
+    def __getitem__(self, key: str) -> Any:
+        val = self._data[key]
+        if isinstance(val, dict):
+            return _Cfg(val)
+        return val
+
+    def __contains__(self, key: str) -> bool:
+        return key in self._data
+
+    def __repr__(self) -> str:
+        return f"Cfg({self._data!r})"
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return the raw dict."""
+        return self._data
+
+
+def _load() -> _Cfg:
+    text = (ROOT / "stack.toml").read_text()
+    return _Cfg(tomllib.loads(text))
+
+
+cfg = _load()
+
+
+def path(dotted_key: str) -> Path:
+    """Resolve a dotted config key to an absolute Path.
+
+    Examples::
+
+        path("db.aco")       # ROOT / "notebooks/aco.duckdb"
+        path("storage.bcda") # ROOT / "data/bcda"
+    """
+    parts = dotted_key.split(".")
+    val: Any = cfg._data
+    for p in parts:
+        val = val[p]
+    result = Path(val)
+    if not result.is_absolute():
+        result = ROOT / result
+    return result
+
+
+def reload() -> None:
+    """Re-read stack.toml (useful after edits)."""
+    global cfg
+    cfg = _load()

stack.toml

@@ -0,0 +1,47 @@
+# stack.toml — centralised configuration for the stack platform.
+#
+# Paths are relative to the repository root unless absolute.
+# Secrets live in .env (never here).
+
+[db]
+aco = "notebooks/aco.duckdb"
+bib = "data/bib.sqlite"
+zotero = "zotero/data/zotero.sqlite"
+
+[storage]
+bib = "data/bib/storage"
+zotero = "zotero/data/storage"
+bcda = "data/bcda"
+rex = "data/rex"
+pfs = "data/pfs"
+cms_log = "data/cms/log.jsonl"
+bcda_log = "data/bcda/log.jsonl"
+
+[generate]
+table_out = "src/aco/table"
+base_import = "aco.table.base"
+
+[bcda]
+sandbox = "https://sandbox.bcda.cms.gov"
+production = "https://api.bcda.cms.gov"
+max_retries = 3
+retry_interval = 1.0
+token_lifetime = 1200
+timeout = 120.0
+
+[lake]
+warehouse = "s3://lakehouse/"
+
+[lake.nessie]
+catalog_uri = "http://nessie:19120/iceberg/"
+
+[lake.polaris]
+catalog_uri = "http://polaris:8181/api/catalog"
+
+[lake.trino]
+host = "trino"
+port = 8080
+catalog = "iceberg"
+
+[lint]
+line_length = 88

tests/aco/test_runner.py

@@ -11,7 +11,7 @@ import pytest
 
 from aco.express.base import Expr
 from aco.pipe.base import Pipeline
-from aco.pipe.runner import _param_to_table, run_pipeline
+from aco.pipe.runner import SchemaError, _param_to_table, run_pipeline
 from aco.table.base import SQLTable
 from bib.tag import Tag
 
@@ -295,6 +295,116 @@ class TestRunPipeline:
         assert order == ["first", "second"]
 
 
+# ── make_databricks_loader ────────────────────────────────────────────────────
+
+
+class TestMakeDatabricksLoader:
+    def test_returns_callable(self) -> None:
+        from aco.pipe.runner import make_databricks_loader
+
+        spark = MagicMock()
+        load = make_databricks_loader(spark, "my_catalog")
+        assert callable(load)
+
+    def test_calls_spark_table_with_catalog_prefix(self) -> None:
+        from aco.pipe.runner import make_databricks_loader
+
+        spark = MagicMock()
+        spark.table.return_value = "fake_df"
+        load = make_databricks_loader(spark, "cat")
+
+        result = load("schema.table")
+        spark.table.assert_called_once_with("cat.schema.table")
+        assert result == "fake_df"
+
+
+# ── Schema validation ─────────────────────────────────────────────────────────
+
+
+class TestSchemaValidation:
+    """Tests for output contract validation in run_pipeline."""
+
+    def test_matching_output_passes(self) -> None:
+        class Out(SQLTable):
+            __schema__ = "t"
+            __tablename__ = "out"
+            a: str | None = None
+            b: int | None = None
+
+        @nw.narwhalify
+        def fn(t__src):
+            return t__src.select("a", "b")
+
+        src = pl.DataFrame({"a": ["x"], "b": [1], "c": [2]})
+        cache = run_pipeline([("t.out", fn, Out)], lambda _: src)
+        assert "t.out" in cache
+
+    def test_missing_column_raises_schema_error(self) -> None:
+        class Out(SQLTable):
+            __schema__ = "t"
+            __tablename__ = "out"
+            a: str | None = None
+            b: int | None = None
+
+        @nw.narwhalify
+        def fn(t__src):
+            return t__src.select("a")  # missing b
+
+        src = pl.DataFrame({"a": ["x"], "b": [1]})
+        with pytest.raises(SchemaError, match="missing columns"):
+            run_pipeline([("t.out", fn, Out)], lambda _: src)
+
+    def test_extra_column_raises_schema_error(self) -> None:
+        class Out(SQLTable):
+            __schema__ = "t"
+            __tablename__ = "out"
+            a: str | None = None
+
+        @nw.narwhalify
+        def fn(t__src):
+            return t__src  # has extra column b
+
+        src = pl.DataFrame({"a": ["x"], "b": [1]})
+        with pytest.raises(SchemaError, match="extra columns"):
+            run_pipeline([("t.out", fn, Out)], lambda _: src)
+
+    def test_none_output_skips_validation(self) -> None:
+        @nw.narwhalify
+        def fn(t__src):
+            return t__src
+
+        src = pl.DataFrame({"any": [1]})
+        cache = run_pipeline([("t.out", fn, None)], lambda _: src)
+        assert "t.out" in cache
+
+    def test_two_tuple_skips_validation(self) -> None:
+        @nw.narwhalify
+        def fn(t__src):
+            return t__src
+
+        src = pl.DataFrame({"any": [1]})
+        cache = run_pipeline([("t.out", fn)], lambda _: src)
+        assert "t.out" in cache
+
+    def test_pipeline_run_validates_output(self) -> None:
+        """Pipeline.run passes output classes to runner for validation."""
+
+        class Out(SQLTable):
+            __schema__ = "t"
+            __tablename__ = "out"
+            a: str | None = None
+
+        @nw.narwhalify
+        def fn(t__src):
+            return t__src  # has extra column
+
+        src = pl.DataFrame({"a": ["x"], "b": [1]})
+        expr = Expr(name="t.out", fn=fn, output=Out)
+        p = Pipeline(exprs=[expr])
+        with pytest.raises(SchemaError):
+            p.run(lambda _: src)
+
+
 # ── Expr ───────────────────────────────────────────────────────────────────────
 
 

tests/aco/test_table_base.py

@@ -2,7 +2,12 @@
 
 from __future__ import annotations
 
-from aco.table.base import SQLTable
+from datetime import date, datetime
+from decimal import Decimal
+
+from pydantic import Field
+
+from aco.table.base import SQLTable, _resolve_type
 
 
 class TestSQLTable:
@@ -37,3 +42,158 @@ class TestSQLTable:
         names = AlrAssignedBeneficiaries.column_names()
         assert "mbi" in names
         assert len(names) > 10
+
+
+class TestColumnMeta:
+    def test_column_meta_with_descriptions(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str = Field(description="first col")
+            b: int | None = None
+
+        meta = T.column_meta()
+        assert meta["a"]["description"] == "first col"
+        assert meta["b"]["description"] == ""
+
+    def test_column_meta_with_json_schema_extra(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str = Field(
+                description="id",
+                json_schema_extra={"source": "input_layer.claim"},
+            )
+
+        meta = T.column_meta()
+        assert meta["a"]["extra"]["source"] == "input_layer.claim"
+
+    def test_column_meta_empty_extra_when_unset(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str | None = None
+
+        meta = T.column_meta()
+        assert meta["a"]["extra"] == {}
+
+    def test_column_meta_includes_type(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: int | None = None
+
+        meta = T.column_meta()
+        assert meta["a"]["type"] is not None
+
+
+class TestFieldDescriptions:
+    def test_returns_only_described_fields(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str = Field(description="has desc")
+            b: int | None = None
+            c: str | None = Field(default=None, description="also has desc")
+
+        descs = T.field_descriptions()
+        assert descs == {"a": "has desc", "c": "also has desc"}
+
+    def test_empty_when_no_descriptions(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str | None = None
+
+        assert T.field_descriptions() == {}
+
+
+class TestResolveType:
+    def test_str(self) -> None:
+        assert _resolve_type(str) == "VARCHAR"
+
+    def test_int(self) -> None:
+        assert _resolve_type(int) == "INTEGER"
+
+    def test_float(self) -> None:
+        assert _resolve_type(float) == "DOUBLE"
+
+    def test_bool(self) -> None:
+        assert _resolve_type(bool) == "BOOLEAN"
+
+    def test_date(self) -> None:
+        assert _resolve_type(date) == "DATE"
+
+    def test_datetime(self) -> None:
+        assert _resolve_type(datetime) == "TIMESTAMP"
+
+    def test_decimal(self) -> None:
+        assert _resolve_type(Decimal) == "DECIMAL(18,2)"
+
+    def test_nullable_str(self) -> None:
+        assert _resolve_type(str | None) == "VARCHAR"
+
+    def test_nullable_int(self) -> None:
+        assert _resolve_type(int | None) == "INTEGER"
+
+    def test_unknown_falls_back_to_varchar(self) -> None:
+        assert _resolve_type(object) == "VARCHAR"
+
+
+class TestToDdl:
+    def test_basic_ddl(self) -> None:
+        class T(SQLTable):
+            __schema__ = "core"
+            __tablename__ = "demo"
+            a: str | None = None
+            b: int | None = None
+
+        ddl = T.to_ddl()
+        assert ddl.startswith("CREATE TABLE core.demo")
+        assert "a VARCHAR" in ddl
+        assert "b INTEGER" in ddl
+
+    def test_ddl_with_descriptions_as_comments(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str = Field(description="the a column")
+            b: int | None = None
+
+        ddl = T.to_ddl()
+        assert "-- the a column" in ddl
+        assert "b INTEGER" in ddl
+
+    def test_ddl_comma_placement(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str | None = None
+            b: int | None = None
+
+        ddl = T.to_ddl()
+        lines = ddl.strip().split("\n")
+        # first col line should have comma, last should not
+        assert lines[1].strip().startswith("a VARCHAR,")
+        assert not lines[2].strip().endswith(",")
+
+    def test_ddl_all_types(self) -> None:
+        class T(SQLTable):
+            __schema__ = "s"
+            __tablename__ = "t"
+            a: str | None = None
+            b: int | None = None
+            c: float | None = None
+            d: bool | None = None
+            e: date | None = None
+            f: datetime | None = None
+            g: Decimal | None = None
+
+        ddl = T.to_ddl()
+        assert "VARCHAR" in ddl
+        assert "INTEGER" in ddl
+        assert "DOUBLE" in ddl
+        assert "BOOLEAN" in ddl
+        assert "DATE" in ddl
+        assert "TIMESTAMP" in ddl
+        assert "DECIMAL" in ddl

tests/bib/test_meta.py

@@ -2,7 +2,11 @@
 
 from __future__ import annotations
 
+import duckdb
+
 from bib.meta import (
+    apply_column_comments,
+    collect_column_comments,
     expr_fn_tag,
     extract_meta_details,
     extract_meta_tags,
@@ -143,6 +147,40 @@ class TestExtractMetaDetails:
         assert details[0]["source"] == "Code of Federal Regulations"
         assert details[0]["title"] == "42 CFR § 425.502"
 
+    def test_cclf_dedup(self):
+        def fn():
+            """CCLF IP Section 2.2 (p.8):
+            Again CCLF IP Section 2.2 (p.8):"""
+
+        details = extract_meta_details(fn)
+        assert len(details) == 1
+
+    def test_cfr_dedup(self):
+        def fn():
+            """42 CFR § 425.502 and again 42 CFR § 425.502"""
+
+        details = extract_meta_details(fn)
+        assert len(details) == 1
+
+    def test_fr_details(self):
+        def fn():
+            """Published at 90 FR 86252."""
+
+        details = extract_meta_details(fn)
+        assert len(details) == 1
+        d = details[0]
+        assert d["source"] == "Federal Register"
+        assert d["title"] == "90 FR 86252"
+        assert d["page"] == "86252"
+        assert d["section"] == ""
+
+    def test_fr_dedup(self):
+        def fn():
+            """90 FR 86252 and again 90 FR 86252"""
+
+        details = extract_meta_details(fn)
+        assert len(details) == 1
+
 
 # ── expr_fn_tag ───────────────────────────────────────────────────
 
@@ -192,6 +230,39 @@ class TestTagFactories:
 
 
 class TestTagItemsFromPipeline:
+    def test_table_tag_fallback(self):
+        """When no refs match, falls back to table: tag lookup."""
+        import narwhals as nw
+
+        from bib.item import Source
+        from bib.store import Store
+
+        store = Store(":memory:")
+        item = Source(
+            title="Core Encounter Spec",
+            tags=["table:test.out"],
+        )
+        key = store.create(item)
+
+        @nw.narwhalify
+        def fn(df):
+            """CCLF IP Section 9.9 "Fake" (p.1):"""
+            return df
+
+        from aco.express.base import Expr
+        from aco.pipe.base import Pipeline
+
+        expr = Expr(name="test.out", fn=fn, refs=[])
+        pipeline = Pipeline(exprs=[expr])
+
+        report = tag_items_from_pipeline(pipeline, store)
+        tagged = {k: v for k, v in report.items() if v}
+        assert len(tagged) > 0
+
+        updated = store.get(key)
+        fn_tags = [t for t in updated.tags if t.startswith("fn:")]
+        assert len(fn_tags) > 0
+
     def test_tags_matched_items(self):
         from bib.item import Source
         from bib.store import Store
@@ -246,3 +317,161 @@ class TestGenerateDocstringRefs:
         refs = generate_docstring_refs("cclf._stg_beneficiary_xref", store)
         assert "References" in refs
         assert "CCLF Information Packet" in refs
+
+
+# ── generate_pipeline_bibliography ───────────────────────────────
+
+
+class TestGeneratePipelineBibliography:
+    def test_returns_refs_for_tagged_exprs(self):
+        import narwhals as nw
+
+        from bib.item import Source
+        from bib.meta import generate_pipeline_bibliography
+        from bib.store import Store
+
+        store = Store(":memory:")
+        item = Source(
+            title="Test Spec",
+            tags=["fn:test.passthrough"],
+            institution="CMS",
+            date_published="2025-01-01",
+        )
+        store.create(item)
+
+        @nw.narwhalify
+        def fn(df):
+            return df
+
+        from aco.express.base import Expr
+        from aco.pipe.base import Pipeline
+
+        expr = Expr(name="test._passthrough", fn=fn)
+        pipeline = Pipeline(exprs=[expr])
+
+        result = generate_pipeline_bibliography(pipeline, store)
+        assert "test._passthrough" in result
+        assert "Test Spec" in result["test._passthrough"]
+
+    def test_empty_pipeline(self):
+        from bib.meta import generate_pipeline_bibliography
+        from bib.store import Store
+
+        store = Store(":memory:")
+
+        from aco.pipe.base import Pipeline
+
+        pipeline = Pipeline(exprs=[])
+        result = generate_pipeline_bibliography(pipeline, store)
+        assert result == {}
+
+
+# ── collect_column_comments ──────────────────────────────────────
+
+
+class TestCollectColumnComments:
+    def test_collects_col_tags(self):
+        from bib.item import Source
+        from bib.store import Store
+
+        store = Store(":memory:")
+        item = Source(title="CCLF Spec")
+        key = store.create(item)
+        store.add_tag(
+            key,
+            Tag.col("core.encounter.encounter_id", "Unique encounter ID").label,
+        )
+        store.add_tag(
+            key,
+            Tag.col("core.encounter.admit_date", "Admission date").label,
+        )
+
+        comments = collect_column_comments(store)
+        assert comments["core.encounter.encounter_id"] == "Unique encounter ID"
+        assert comments["core.encounter.admit_date"] == "Admission date"
+
+    def test_empty_store_returns_empty(self):
+        from bib.store import Store
+
+        store = Store(":memory:")
+        assert collect_column_comments(store) == {}
+
+    def test_ignores_malformed_col_tags(self):
+        from bib.item import Source
+        from bib.store import Store
+
+        store = Store(":memory:")
+        item = Source(title="Bad tags")
+        key = store.create(item)
+        # Tag without = separator
+        store.add_tag(key, "col:no-equals-sign")
+
+        comments = collect_column_comments(store)
+        assert comments == {}
+
+
+# ── apply_column_comments ────────────────────────────────────────
+
+
+class TestApplyColumnComments:
+    def test_applies_comments_to_duckdb(self):
+        from bib.item import Source
+        from bib.store import Store
+
+        store = Store(":memory:")
+        item = Source(title="Spec")
+        key = store.create(item)
+        store.add_tag(
+            key,
+            Tag.col("s.t.col_a", "Column A description").label,
+        )
+
+        con = duckdb.connect(":memory:")
+        con.execute("CREATE SCHEMA s")
+        con.execute("CREATE TABLE s.t (col_a VARCHAR, col_b INTEGER)")
+
+        stmts = apply_column_comments(con, store)
+        assert len(stmts) == 1
+        assert "Column A description" in stmts[0]
+
+        # Verify comment was actually set
+        rows = con.execute(
+            "SELECT comment FROM duckdb_columns() "
+            "WHERE schema_name = 's' AND table_name = 't' "
+            "AND column_name = 'col_a'"
+        ).fetchall()
+        assert rows[0][0] == "Column A description"
+        con.close()
+
+    def test_skips_invalid_column_refs(self):
+        from bib.item import Source
+        from bib.store import Store
+
+        store = Store(":memory:")
+        item = Source(title="Bad ref")
+        key = store.create(item)
+        # Only two parts instead of three
+        store.add_tag(key, "col:schema.table=desc")
+
+        con = duckdb.connect(":memory:")
+        stmts = apply_column_comments(con, store)
+        assert stmts == []
+        con.close()
+
+    def test_skips_nonexistent_columns(self):
+        from bib.item import Source
+        from bib.store import Store
+
+        store = Store(":memory:")
+        item = Source(title="Spec")
+        key = store.create(item)
+        store.add_tag(
+            key,
+            Tag.col("s.t.nonexistent", "desc").label,
+        )
+
+        con = duckdb.connect(":memory:")
+        # Table doesn't exist — should not raise
+        stmts = apply_column_comments(con, store)
+        assert stmts == []
+        con.close()

tests/bib/test_tag.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import pytest
 
-from bib.tag import Tag, filter_tags
+from bib.tag import Tag, filter_tags, rule_slug
 
 # ── Tag construction ──────────────────────────────────────────────────────────
 
@@ -406,6 +406,71 @@ class TestTagFile:
         assert tag.value == file_type
 
 
+class TestTagSeed:
+    """Tag.seed() creates a 'seed' namespace tag."""
+
+    def test_namespace_is_seed(self) -> None:
+        assert Tag.seed().namespace == "seed"
+
+    def test_value_is_true(self) -> None:
+        assert Tag.seed().value == "true"
+
+    def test_label(self) -> None:
+        assert Tag.seed().label == "seed:true"
+
+
+class TestTagProgram:
+    """Tag.program() creates a 'program' namespace tag, lowercasing the value."""
+
+    def test_namespace_is_program(self) -> None:
+        assert Tag.program("reach").namespace == "program"
+
+    def test_value_lowercased(self) -> None:
+        assert Tag.program("MSSP").value == "mssp"
+
+    def test_label(self) -> None:
+        assert Tag.program("mips").label == "program:mips"
+
+
+class TestTagMeasure:
+    """Tag.measure() creates a 'measure' namespace tag, uppercasing the value."""
+
+    def test_namespace_is_measure(self) -> None:
+        assert Tag.measure("uamcc").namespace == "measure"
+
+    def test_value_uppercased(self) -> None:
+        assert Tag.measure("uamcc").value == "UAMCC"
+
+    def test_label(self) -> None:
+        assert Tag.measure("acr").label == "measure:ACR"
+
+
+class TestTagCol:
+    """Tag.col() creates a 'col' namespace tag with column=description format."""
+
+    def test_namespace_is_col(self) -> None:
+        tag = Tag.col("core.encounter.encounter_id", "Unique encounter ID")
+        assert tag.namespace == "col"
+
+    def test_value_format(self) -> None:
+        tag = Tag.col("core.encounter.encounter_id", "Unique encounter ID")
+        assert tag.value == "core.encounter.encounter_id=Unique encounter ID"
+
+    def test_label_format(self) -> None:
+        tag = Tag.col("core.encounter.admit_date", "Admission date")
+        assert tag.label == "col:core.encounter.admit_date=Admission date"
+
+    def test_returns_tag(self) -> None:
+        assert isinstance(Tag.col("s.t.c", "desc"), Tag)
+
+    def test_roundtrip_via_from_label(self) -> None:
+        tag = Tag.col("core.patient.birth_date", "Date of birth")
+        restored = Tag.from_label(tag.label)
+        assert restored.namespace == "col"
+        assert "birth_date" in restored.value
+        assert "Date of birth" in restored.value
+
+
 class TestTagSup:
     """Tag.sup() creates a 'sup' namespace tag linking to a parent."""
 
@@ -599,3 +664,17 @@ class TestFilterTags:
                 f"Expected {tag.value!r} in filter_tags result for namespace {ns!r}, "
                 f"got {values}"
             )
+
+
+# ── rule_slug ────────────────────────────────────────────────────────────────
+
+
+class TestRuleSlug:
+    def test_final_rule(self) -> None:
+        assert rule_slug("Medicare Program; CY 2023 PFS Final Rule") == "2023_PFS_FR"
+
+    def test_proposed_rule(self) -> None:
+        assert rule_slug("Medicare Program; CY 2026 PFS Proposed Rule") == "2026_PFS_PR"
+
+    def test_no_match(self) -> None:
+        assert rule_slug("Something else entirely") == ""

tests/conf/test_conf.py

@@ -0,0 +1,104 @@
+"""Tests for conf — centralised configuration loader."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from conf import ROOT, cfg, path, reload
+
+
+class TestCfg:
+    def test_root_is_repo_root(self) -> None:
+        assert (ROOT / "stack.toml").exists()
+        assert (ROOT / "pyproject.toml").exists()
+
+    def test_db_section_exists(self) -> None:
+        assert "db" in cfg
+
+    def test_db_aco_value(self) -> None:
+        assert cfg.db.aco == "notebooks/aco.duckdb"
+
+    def test_db_bib_value(self) -> None:
+        assert cfg.db.bib == "data/bib.sqlite"
+
+    def test_db_zotero_value(self) -> None:
+        assert cfg.db.zotero == "zotero/data/zotero.sqlite"
+
+    def test_storage_bcda(self) -> None:
+        assert cfg.storage.bcda == "data/bcda"
+
+    def test_bcda_timeout_is_float(self) -> None:
+        assert isinstance(cfg.bcda.timeout, float)
+
+    def test_bcda_max_retries_is_int(self) -> None:
+        assert isinstance(cfg.bcda.max_retries, int)
+
+    def test_lake_trino_port_is_int(self) -> None:
+        assert isinstance(cfg.lake.trino.port, int)
+
+    def test_dict_access(self) -> None:
+        val = cfg["db"]["aco"]
+        assert val == "notebooks/aco.duckdb"
+
+    def test_to_dict(self) -> None:
+        d = cfg.db.to_dict()
+        assert isinstance(d, dict)
+        assert "aco" in d
+
+    def test_contains(self) -> None:
+        assert "db" in cfg
+        assert "nonexistent_key" not in cfg
+
+    def test_repr(self) -> None:
+        r = repr(cfg.db)
+        assert "aco" in r
+
+    def test_missing_attr_raises(self) -> None:
+        import pytest
+
+        with pytest.raises(AttributeError):
+            cfg.no_such_section
+
+
+class TestPath:
+    def test_path_returns_absolute(self) -> None:
+        p = path("db.aco")
+        assert isinstance(p, Path)
+        assert p.is_absolute()
+
+    def test_path_resolves_relative_to_root(self) -> None:
+        p = path("db.aco")
+        assert p == ROOT / "notebooks/aco.duckdb"
+
+    def test_path_nested_key(self) -> None:
+        p = path("db.bib")
+        assert p == ROOT / "data/bib.sqlite"
+
+
+class TestReload:
+    def test_reload_does_not_raise(self) -> None:
+        reload()
+        assert cfg.db.aco == "notebooks/aco.duckdb"
+
+
+class TestFindRootError:
+    def test_raises_when_no_stack_toml(self, tmp_path: Path) -> None:
+        import pytest
+
+        # Monkeypatch __file__ won't work easily, but we can test
+        # the function directly with a path that has no stack.toml
+        import conf
+        from conf import _find_root
+
+        original = conf.__file__
+        try:
+            # Point to a deep tmp path with no stack.toml
+            deep = tmp_path / "a" / "b" / "c" / "d" / "e"
+            deep.mkdir(parents=True)
+            fake_init = deep / "__init__.py"
+            fake_init.write_text("")
+            conf.__file__ = str(fake_init)
+            with pytest.raises(FileNotFoundError, match="stack.toml"):
+                _find_root()
+        finally:
+            conf.__file__ = original