feat(lake): M4 — DuckLakeContext write path + OPPS lake pilot (closes #513)

aco.lake.DuckLakeContext: the concrete lake context for the M3
decision (#512). Subclasses DuckDBContext — load/save are inherited
untouched; only the connection wiring differs (ducklake+httpfs
extensions, S3 secret for RustFS, ATTACH the catalog, USE it as the
default database so schema.table refs resolve in the lake). Postgres
DSN for multi-writer prod, plain file path for dev/tests. Explicit
close() added to the context family (releases locks/attachments).

Pilot: dev/scripts/publish_opps_to_lake.py reads the OPPS tables from
the replica and writes them through Context.save(mode=replace). Run
against prod (postgres catalog db 'ducklake' + RustFS): addendum_b
219,665 rows in 2.6s, apc_weight 7,776, skin_sub_addendum_b 2,520 —
all read-back verified through Context.load.

Config: stack.toml [lake.ducklake] (catalog DSN sans password, data
path, s3 endpoint). CI: ducklake added to the duckdb extension
preinstall (same xdist install race class as #515). 6 new tests use a
file catalog + local data path — same code paths, no services needed.

.gitea/workflows/ci.yml

@@ -51,7 +51,8 @@ jobs:
         # worker; concurrent installs race the extension-file rename in
         # ~/.duckdb ("Could not remove file ... sqlite_scanner", #515).
         # Installing once up front makes the in-test INSTALL a no-op.
-        run: uv run python -c "import duckdb; duckdb.connect().execute('INSTALL sqlite')"
+        # ducklake: same race class, used by tests/aco DuckLakeContext.
+        run: uv run python -c "import duckdb; duckdb.connect().execute('INSTALL sqlite; INSTALL ducklake')"
 
       - name: Pytest
         # -n auto parallelizes across runner cores. Coverage combining

dev/scripts/backends/gitea.py

@@ -182,7 +182,8 @@ jobs:
         # worker; concurrent installs race the extension-file rename in
         # ~/.duckdb ("Could not remove file ... sqlite_scanner", #515).
         # Installing once up front makes the in-test INSTALL a no-op.
-        run: uv run python -c "import duckdb; duckdb.connect().execute('INSTALL sqlite')"
+        # ducklake: same race class, used by tests/aco DuckLakeContext.
+        run: uv run python -c "import duckdb; duckdb.connect().execute('INSTALL sqlite; INSTALL ducklake')"
 
       - name: Pytest
         # -n auto parallelizes across runner cores. Coverage combining

dev/scripts/publish_opps_to_lake.py

@@ -0,0 +1,69 @@
+"""M4 pilot (#513): publish OPPS reference tables to the DuckLake lakehouse.
+
+Reads the OPPS tables from the local DuckDB (read replica when present)
+and writes them to DuckLake via ``aco.lake.DuckLakeContext.save`` — the
+Context write path this pilot exists to exercise. Config comes from
+``stack.toml [lake.ducklake]``; the postgres catalog password from the
+``POSTGRES_PASSWORD`` env var.
+
+The catalog host and RustFS are compose-internal, so this runs inside a
+data-network container:
+
+    docker exec -e POSTGRES_PASSWORD=... notebooks \\
+      env PYTHONPATH=/home/kert/src uv run --project /home/kert/workspace \\
+      python /tmp/publish_opps_to_lake.py
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import time
+
+TABLES = ("addendum_b", "apc_weight", "skin_sub_addendum_b")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--tables", nargs="*", default=list(TABLES), help="opps tables to publish"
+    )
+    args = parser.parse_args()
+
+    import narwhals as nw
+
+    from aco.lake import DuckLakeContext
+    from conf import cfg, connect
+
+    pw = os.environ.get("POSTGRES_PASSWORD", "")
+    dsn = cfg.lake.ducklake.catalog
+    if dsn.startswith("postgres:") and pw:
+        dsn = f"{dsn} password={pw}"
+
+    ctx = DuckLakeContext(
+        catalog_dsn=dsn,
+        data_path=cfg.lake.ducklake.data_path,
+        s3_endpoint=cfg.lake.ducklake.s3_endpoint,
+        read_only=False,
+    )
+
+    src = connect.duckdb("aco")  # read-only; resolves to the replica
+    for table in args.tables:
+        t = time.time()
+        df = src.execute(f"SELECT * FROM opps.{table}").pl()  # noqa: S608
+        ctx.save(f"opps.{table}", nw.from_native(df), mode="replace")
+        back = nw.to_native(ctx.load(f"opps.{table}"))
+        status = "OK" if back.height == df.height else "MISMATCH"
+        print(
+            f"  opps.{table}: {df.height} rows → lake ({back.height} read back) "
+            f"[{status}] in {time.time() - t:.1f}s"
+        )
+        if status != "OK":
+            return 1
+    src.close()
+    print("lake publish complete")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/aco/lake/__init__.py

@@ -44,6 +44,7 @@ Four contexts, one interface::
 from .catalog import Catalog as Catalog
 from .context import Context as Context
 from .context import DuckDBContext as DuckDBContext
+from .context import DuckLakeContext as DuckLakeContext
 from .context import EnterpriseContext as EnterpriseContext
 from .context import IcebergContext as IcebergContext
 from .context import ParquetContext as ParquetContext

src/aco/lake/context.py

@@ -335,13 +335,117 @@ class DuckDBContext(Context):
 
         return self._connection
 
-    def __del__(self):
+    def close(self) -> None:
-        """Close connection on context destruction."""
+        """Close the cached connection, releasing locks/attachments."""
         if self._connection is not None:
             try:
                 self._connection.close()
             except Exception:
                 pass
+            self._connection = None
+
+    def __del__(self):
+        """Close connection on context destruction."""
+        self.close()
+
+
+# ── DuckLake context ─────────────────────────────────────────────────
+
+
+class DuckLakeContext(DuckDBContext):
+    """DuckLake — ACID multi-writer lakehouse through plain DuckDB SQL.
+
+    Chosen over Iceberg for concurrent reference-data storage in the M3
+    spike (`docs/superpowers/specs/2026-07-10-ducklake-vs-iceberg-
+    decision.md`, #512): catalog-database transactions serialize
+    concurrent writers with zero client-side retries, and readers keep
+    the exact SQL interface notebooks already use. Data lives as
+    Parquet under ``data_path`` (RustFS/S3); table state lives in the
+    catalog database (postgres in prod, a local file in tests).
+
+    Inherits ``load``/``save`` from :class:`DuckDBContext` — only the
+    connection wiring differs: the connection LOADs the ducklake
+    extension, creates the S3 secret, ATTACHes the catalog, and makes
+    it the default database so ``schema.table`` refs resolve inside
+    the lake.
+
+    Usage::
+
+        from aco.lake import DuckLakeContext
+
+        # prod: postgres catalog + RustFS data (creds from env)
+        ctx = DuckLakeContext(
+            catalog_dsn="postgres:dbname=ducklake host=postgres "
+                        "user=postgres password=...",
+            data_path="s3://lakehouse/ducklake/",
+            read_only=False,
+        )
+        ctx.save("opps.addendum_b", df, mode="replace")
+
+        # tests/dev: file catalog + local data dir, no services
+        ctx = DuckLakeContext(
+            catalog_dsn="/tmp/cat.ducklake",
+            data_path="/tmp/lake-data/",
+            read_only=False,
+        )
+    """
+
+    database: str = ""
+    """Unused — the DuckLake catalog defines storage; kept for the
+    parent's field contract."""
+
+    catalog_dsn: str
+    """DuckLake catalog: a postgres DSN (``postgres:dbname=… host=…``)
+    for multi-writer prod use, or a local file path for dev/tests."""
+
+    data_path: str
+    """Where table data (Parquet) lives — ``s3://…`` or a local dir."""
+
+    alias: str = "lake"
+    """Attach alias; becomes the connection's default database."""
+
+    s3_endpoint: str = ""
+    """S3 endpoint host:port for ``s3://`` data paths (e.g.
+    ``rustfs:9000``). Ignored for local data paths."""
+
+    s3_access_key: str = ""
+    s3_secret_key: str = ""
+    """S3 credentials; default to ``RUSTFS_ACCESS_KEY`` /
+    ``RUSTFS_SECRET_KEY`` env vars when empty."""
+
+    def _get_connection(self) -> Any:
+        if self._connection is not None:
+            return self._connection
+        import os
+
+        import duckdb
+
+        con = duckdb.connect()
+        con.execute("INSTALL ducklake; LOAD ducklake")
+        if self.data_path.startswith("s3://"):
+            con.execute("INSTALL httpfs; LOAD httpfs")
+
+            def q(v: str) -> str:  # CREATE SECRET takes no bound params
+                return "'" + v.replace("'", "''") + "'"
+
+            key = self.s3_access_key or os.environ.get("RUSTFS_ACCESS_KEY", "")
+            secret = self.s3_secret_key or os.environ.get("RUSTFS_SECRET_KEY", "")
+            con.execute(
+                f"CREATE SECRET ducklake_s3 (TYPE S3, "
+                f"KEY_ID {q(key)}, SECRET {q(secret)}, "
+                f"ENDPOINT {q(self.s3_endpoint)}, "
+                f"USE_SSL false, URL_STYLE 'path')"
+            )
+        if self.catalog_dsn.startswith("postgres:"):
+            con.execute("INSTALL postgres; LOAD postgres")
+        ro = ", READ_ONLY" if self.read_only else ""
+        con.execute(
+            f"ATTACH 'ducklake:{self.catalog_dsn}' AS {self.alias} "
+            f"(DATA_PATH '{self.data_path}'{ro})"
+        )
+        con.execute(f"USE {self.alias}")
+        self._connection = con
+        return con
 
 
 # ── Parquet context ──────────────────────────────────────────────────

stack.toml

@@ -123,6 +123,17 @@ host = "trino"
 port = 8080
 catalog = "iceberg"
 
+# DuckLake — the concurrent reference-data store chosen in the M3 spike
+# (docs/superpowers/specs/2026-07-10-ducklake-vs-iceberg-decision.md).
+# The postgres catalog serializes concurrent writers; data is Parquet on
+# RustFS. Password comes from the POSTGRES_PASSWORD env var; the catalog
+# host is compose-internal, so lake writers run inside a data-network
+# container.
+[lake.ducklake]
+catalog = "postgres:dbname=ducklake host=postgres user=postgres"
+data_path = "s3://lakehouse/ducklake/"
+s3_endpoint = "rustfs:9000"
+
 # ── Databricks Asset Bundle generation ────────────────────────────
 # gen_config.py reads this to produce databricks.yml.
 # Add a new pipeline module → commit → databricks.yml updates automatically.

tests/aco/test_lake_ducklake.py

@@ -0,0 +1,83 @@
+"""Tests for aco.lake.DuckLakeContext — file catalog + local data dir.
+
+No postgres/S3 required: DuckLake accepts a local file catalog and a
+local DATA_PATH, which exercises the exact same load/save code paths
+as the prod postgres+RustFS configuration.
+"""
+
+from __future__ import annotations
+
+import narwhals as nw
+import polars as pl
+import pytest
+
+from aco.lake import DuckLakeContext
+
+
+def _writer(tmp_path) -> DuckLakeContext:
+    return DuckLakeContext(
+        catalog_dsn=str(tmp_path / "catalog.ducklake"),
+        data_path=str(tmp_path / "lake-data") + "/",
+        read_only=False,
+    )
+
+
+def _frame() -> nw.DataFrame:
+    return nw.from_native(
+        pl.DataFrame(
+            {
+                "hcpcs": ["Q4186", "C5271"],
+                "year": [2026, 2026],
+                "payment_rate": [127.28, 127.28],
+            }
+        )
+    )
+
+
+class TestDuckLakeContext:
+    def test_save_replace_and_load_roundtrip(self, tmp_path):
+        ctx = _writer(tmp_path)
+        ctx.save("opps.addendum_b", _frame(), mode="replace")
+        back = ctx.load("opps.addendum_b")
+        native = nw.to_native(back)
+        assert native.height == 2
+        assert set(native.columns) == {"hcpcs", "year", "payment_rate"}
+
+    def test_append_creates_then_appends(self, tmp_path):
+        ctx = _writer(tmp_path)
+        ctx.save("opps.addendum_b", _frame())  # append creates
+        ctx.save("opps.addendum_b", _frame())  # append adds
+        assert nw.to_native(ctx.load("opps.addendum_b")).height == 4
+
+    def test_replace_resets(self, tmp_path):
+        ctx = _writer(tmp_path)
+        ctx.save("opps.addendum_b", _frame())
+        ctx.save("opps.addendum_b", _frame(), mode="replace")
+        assert nw.to_native(ctx.load("opps.addendum_b")).height == 2
+
+    def test_read_only_guard(self, tmp_path):
+        writer = _writer(tmp_path)
+        writer.save("opps.addendum_b", _frame(), mode="replace")
+        writer.close()  # release the file-catalog attachment
+        reader = DuckLakeContext(
+            catalog_dsn=str(tmp_path / "catalog.ducklake"),
+            data_path=str(tmp_path / "lake-data") + "/",
+            read_only=True,
+        )
+        assert nw.to_native(reader.load("opps.addendum_b")).height == 2
+        with pytest.raises(RuntimeError, match="read-only"):
+            reader.save("opps.addendum_b", _frame())
+
+    def test_unqualified_ref_rejected(self, tmp_path):
+        ctx = _writer(tmp_path)
+        with pytest.raises(ValueError, match="qualified"):
+            ctx.load("addendum_b")
+        with pytest.raises(ValueError, match="qualified"):
+            ctx.save("addendum_b", _frame())
+
+    def test_second_context_sees_committed_data(self, tmp_path):
+        """Commits are visible to a separate connection through the shared
+        catalog — the property the monolith could never provide."""
+        _writer(tmp_path).save("opps.apc_weight", _frame(), mode="replace")
+        other = _writer(tmp_path)
+        assert nw.to_native(other.load("opps.apc_weight")).height == 2