docs(infra): DuckDB concurrency/streaming strategy + issue-filing script

The OPPS re-ingest was blocked by a marimo notebook kernel holding a
read-only connection to the shared 3.2 GB aco.duckdb — DuckDB is
single-writer, so any open handle blocks a batch writer. This recurs for
every data-refresh workflow.

Add a strategy spec (options: preflight lock detection, notebook
read-replica, per-domain split, Iceberg lake context, DuckLake) with
milestones M1-M5, and a runnable filing script for the seven discrete
issues (no GITEA_TOKEN locally, so file with:
GITEA_TOKEN=… uv run python dev/scripts/file_concurrency_issues.py).

dev/scripts/file_concurrency_issues.py

@@ -0,0 +1,205 @@
+"""File the DuckDB concurrency / streaming issues to Gitea.
+
+These issues implement the plan in
+``docs/superpowers/specs/2026-07-08-duckdb-concurrency-streaming.md``.
+They could not be filed inline (no GITEA_TOKEN in the local dev env), so
+run this once with a token:
+
+    GITEA_TOKEN=<pat> uv run python dev/scripts/file_concurrency_issues.py
+    # from outside the compose network, point at the public API:
+    GITEA_TOKEN=<pat> uv run python dev/scripts/file_concurrency_issues.py \
+        --base-url https://git.fhirworx.io/api/v1
+
+    uv run python dev/scripts/file_concurrency_issues.py --dry-run   # preview
+
+Idempotency: re-running creates duplicates. Check the tracker first, or
+pass --only <key> to file a single issue.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+
+OWNER = "homelab"
+REPO = "stack"
+SPEC = "docs/superpowers/specs/2026-07-08-duckdb-concurrency-streaming.md"
+
+# Discrete, self-contained issues. Bodies cross-reference the spec by path
+# (Gitea renders it) and each other by title, since issue numbers aren't
+# known until creation.
+ISSUES: list[dict] = [
+    {
+        "key": "m1-preflight",
+        "title": "M1: ingest preflight — DuckDB lock detection, retry, and connection hygiene",
+        "labels": ["infra", "duckdb", "dx"],
+        "body": f"""\
+Batch ingests (`dev/scripts/ingest_opps.py`, PFS ingest, `aco.pipe` runs) fail
+with a raw `IOException: Could not set lock on file "data/aco.duckdb"` whenever a
+long-running reader (typically a marimo notebook kernel) holds a connection.
+DuckDB is single-writer, so even a read-only handle blocks the writer.
+
+**Scope**
+- Add a preflight to the ingest entrypoints: detect the lock (attempt RW open;
+  on failure run `lsof data/aco.duckdb`), then retry with backoff and, if still
+  held, exit with an actionable message naming the holder PID/command
+  ("marimo kernel <pid> — close the notebook tab or stop that PID").
+- Add a short-lived connection helper (context manager) to `conf.connect` so
+  callers open→use→close instead of holding a connection for the process life.
+- Document notebook connection hygiene (don't keep a module-level DuckDB
+  connection alive across cells).
+
+Background & alternatives: `{SPEC}` (option B).
+""",
+    },
+    {
+        "key": "m1-year-footgun",
+        "title": "M1: ingest_opps.py --year wipes other years (full-replace footgun)",
+        "labels": ["bug", "infra", "duckdb"],
+        "body": f"""\
+`dev/scripts/ingest_opps.py --year YYYY` processes only that year's dir, then
+runs `DROP TABLE opps.addendum_b; CREATE TABLE ... AS SELECT * FROM <that year>`
+— which **wipes every other year** from the table. The only safe way to keep all
+years is a full ingest with no `--year`.
+
+**Fix**: make `--year` a per-year merge — `DELETE FROM opps.addendum_b WHERE
+year = ?` then insert the new rows — leaving other years intact. Same for
+`apc_weight` / `skin_sub_addendum_b`.
+
+Background: `{SPEC}` (M1 / #B).
+""",
+    },
+    {
+        "key": "m2-read-replica",
+        "title": "M2: notebook read-replica (aco.ro.duckdb) to decouple readers from writers",
+        "labels": ["infra", "duckdb", "notebooks"],
+        "body": f"""\
+Decouple long-running notebook readers from batch writers: after each ingest,
+publish a read-only snapshot `aco.ro.duckdb` and point notebooks /
+`conf.connect.duckdb(read_only=True)` at the replica. Writers own the primary;
+notebooks never block them. Staleness is bounded by the refresh cadence.
+
+**Scope**
+- Post-ingest snapshot step (copy or `EXPORT`/`ATTACH` + `COPY`).
+- `conf` switch so notebook reads resolve to the replica.
+- Define/refresh cadence + document it.
+
+Background & alternatives: `{SPEC}` (option C). Complements the per-domain split
+(M2 / #D).
+""",
+    },
+    {
+        "key": "m2-split-db",
+        "title": "M2: split aco.duckdb into per-domain files + ATTACH",
+        "labels": ["infra", "duckdb"],
+        "body": f"""\
+`data/aco.duckdb` is one ~3.2 GB file for all schemas (`aco`, `opps`, `pfs`,
+`bib`, …), so a reader of any schema blocks a writer of any other. Split into
+per-domain DB files (`opps.duckdb`, `pfs.duckdb`, …) and `ATTACH` them for
+cross-domain queries. A domain ingest then only contends with readers of that
+domain.
+
+**Scope**: `conf.path`/`conf.connect` for per-domain DBs; `ATTACH` helper;
+migrate the ingest scripts; keep `aco.duckdb` as the analytics/joined store.
+
+Background: `{SPEC}` (option D).
+""",
+    },
+    {
+        "key": "m3-ducklake-iceberg-spike",
+        "title": "M3: spike DuckLake vs Iceberg for concurrent reference-data storage",
+        "labels": ["infra", "lakehouse", "spike"],
+        "body": f"""\
+Evaluate the two concurrent-read/write options for CMS reference data
+(OPPS/PFS/etc.), both on the already-deployed RustFS object store:
+
+- **Iceberg** via the existing Nessie + Trino + Polaris lake context
+  (`stack.toml [context.lake]`): snapshot isolation, optimistic concurrency.
+- **DuckLake**: DuckDB's catalog + Parquet lakehouse — ACID multi-writer while
+  keeping the DuckDB SQL interface (lighter than full Iceberg/Trino).
+
+Deliver a decision record: which becomes the concurrent store, and why.
+
+Background: `{SPEC}` (options E, F). Feeds M4.
+""",
+    },
+    {
+        "key": "m4-lake-write-pilot",
+        "title": "M4: implement aco.lake write path + pilot OPPS ingestion to the lake",
+        "labels": ["lakehouse", "feature"],
+        "body": f"""\
+Implement the `aco.lake` `Context` write path (currently `Context.load/save` and
+`_execute_transpiled` raise `NotImplementedError`) and pilot OPPS Addendum B
+ingestion to the lake store chosen in M3 (Iceberg via Nessie/RustFS, or
+DuckLake). Concurrent notebook readers + batch writers coexist via snapshot
+isolation — no file lock.
+
+Background: `{SPEC}` (option E). Depends on M3. Also closes the standing
+`aco.lake` remote-execution gap.
+""",
+    },
+    {
+        "key": "m5-notebook-cutover",
+        "title": "M5: migrate notebooks to lake reads; retire the monolith for reference data",
+        "labels": ["lakehouse", "notebooks"],
+        "body": f"""\
+Point notebooks at the lake (DuckDB iceberg extension or Trino) for OPPS/PFS
+reference data, cut ingestion over to the lake, and retire the local DuckDB
+monolith as the reference-data store. Concurrent read/write becomes the norm;
+the lock class of failures disappears.
+
+Background: `{SPEC}` (M5). Depends on M4.
+""",
+    },
+]
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument(
+        "--base-url", default="", help="Gitea API base (default: cfg.services.git)"
+    )
+    ap.add_argument("--only", help="File a single issue by key")
+    ap.add_argument("--dry-run", action="store_true", help="Print, don't post")
+    args = ap.parse_args()
+
+    issues = ISSUES if not args.only else [i for i in ISSUES if i["key"] == args.only]
+    if not issues:
+        raise SystemExit(f"no issue with key {args.only!r}")
+
+    if args.dry_run:
+        for i in issues:
+            print(f"\n=== [{i['key']}] {i['title']}  labels={i['labels']} ===")
+            print(i["body"])
+        return
+
+    token = os.environ.get("GITEA_TOKEN", "")
+    if not token:
+        try:
+            from conf import secret
+
+            token = secret("gitea.token", "GITEA_TOKEN") or ""
+        except Exception:
+            token = ""
+    if not token:
+        raise SystemExit("Set GITEA_TOKEN to file issues (or use --dry-run).")
+
+    from api.clients.gitea import GiteaClient
+
+    client = GiteaClient(token, base_url=args.base_url)
+    try:
+        for i in issues:
+            res = client.create_issue(
+                OWNER,
+                REPO,
+                {"title": i["title"], "body": i["body"], "labels": i["labels"]},
+            )
+            print(
+                f"filed #{res.get('number')}  {i['title']}\n  {res.get('html_url', '')}"
+            )
+    finally:
+        client.close()
+
+
+if __name__ == "__main__":
+    main()

docs/superpowers/specs/2026-07-08-duckdb-concurrency-streaming.md

@@ -0,0 +1,91 @@
+# DuckDB concurrency & data streaming — strategy
+
+**Status:** Draft (awaiting user approval)
+**Date:** 2026-07-08
+**Trigger:** The OPPS Addendum B re-ingest (`dev/scripts/ingest_opps.py`) failed
+with `IOException: Could not set lock on file "data/aco.duckdb": Conflicting
+lock is held ...` because a long-running marimo notebook kernel held a
+read connection to the shared DuckDB file.
+
+## Problem
+
+`data/aco.duckdb` is a single ~3.2 GB DuckDB file shared by every schema
+(`aco`, `opps`, `pfs`, `bib`, …). DuckDB's process-level locking allows
+**either** one read-write connection **or** multiple read-only connections —
+a read-write open requires exclusive access. So **any** open handle, even a
+read-only one, blocks a writer.
+
+Our readers are long-lived: the `marimo edit` server (`:2718`, uptime measured
+in **days**) spawns a kernel per open notebook, and a notebook cell that calls
+`conf.connect.duckdb()` (default `read_only=True`) keeps that connection for
+the **lifetime of the kernel**. Result: every batch ingest or pipeline write
+(`ingest_opps.py`, PFS ingest, `aco.pipe` runs) fails whenever any notebook is
+open against the DB. This is not OPPS-specific — it blocks **all** data-refresh
+workflows, and it will recur.
+
+### Root causes
+1. **Monolith + broad locking.** One file for all schemas means a reader of
+   *any* schema blocks a writer of *any other* schema.
+2. **Connection lifetime.** `conf.connect.duckdb()` returns a bare connection;
+   notebooks hold it open indefinitely instead of using a short-lived scope.
+3. **No writer coordination.** Ingest scripts open read-write directly and
+   surface DuckDB's raw `IOException` with no detection, retry, or guidance.
+
+## Options considered
+
+| # | Option | Effort | Concurrency | Notes |
+|---|--------|--------|-------------|-------|
+| A | Ingest-then-swap the whole file | M | none | Would copy 3.2 GB per ingest; readers keep stale inode until reconnect. Rejected for the monolith. |
+| B | Preflight lock detection + retry + connection hygiene | S | none | Turns cryptic failure into guidance; shrinks the lock window. Necessary regardless. |
+| C | Notebook **read-replica** (`aco.ro.duckdb`) refreshed post-ingest | M | reader/writer decoupled | Notebooks never block writers; staleness bounded by refresh cadence. Cheap, effective. |
+| D | **Split** `aco.duckdb` into per-domain files (`opps`, `pfs`, …) + `ATTACH` | M | narrows blast radius | A domain ingest only conflicts with readers of that domain. |
+| E | **Iceberg** lake context (Nessie + RustFS + Trino/Polaris — already deployed) | L | full snapshot isolation | The stack's intended concurrent read/write path. Depends on completing `aco.lake` write path (currently `NotImplementedError`). |
+| F | **DuckLake** (DuckDB catalog + Parquet on S3/RustFS) | L | ACID multi-writer | Keeps the DuckDB SQL interface; lighter than full Iceberg/Trino. Spike vs E. |
+| G | MotherDuck (hosted) | — | full | External SaaS; conflicts with the self-hosted ethos. Rejected. |
+
+## Recommended path
+
+- **Now (unblock + stop the bleeding):** **B** — a lock-aware preflight in the
+  ingest scripts (detect the holder via `lsof`, retry with backoff, and emit an
+  actionable message naming the notebook/PID) plus a short-lived
+  `conf.connect` context manager and notebook guidance.
+- **Near term (decouple readers):** **C** — publish a read-only snapshot
+  (`aco.ro.duckdb`) after each ingest and point notebooks / `conf.connect.duckdb`
+  at it. Optionally **D** to narrow write contention further.
+- **Strategic (true concurrency):** spike **F (DuckLake)** vs **E (Iceberg)**,
+  then implement the `aco.lake` write path and pilot OPPS ingestion to the lake,
+  finally migrate notebooks to read from the lake and retire the monolith for
+  reference data. This also unblocks the already-flagged
+  `aco.lake` / `_execute_transpiled` `NotImplementedError` work.
+
+## Milestones → issues
+
+- **M1 / #A** — Ingest preflight: lock detection, retry-with-backoff, actionable
+  error; `conf.connect.duckdb` context manager + notebook connection guidance. (S)
+- **M1 / #B** — Fix `ingest_opps.py --year` footgun: with `--year` it `DROP`s and
+  rebuilds `opps.addendum_b` from only that year, wiping the others. Make it a
+  per-year merge/upsert (delete-that-year + insert), not a full replace. (S)
+- **M2 / #C** — Notebook read-replica `aco.ro.duckdb`: publish post-ingest, point
+  notebooks at it, define refresh cadence. (M)
+- **M2 / #D** — Split `aco.duckdb` into per-domain files + `ATTACH` for
+  cross-domain queries; update `conf.path`/`conf.connect`. (M)
+- **M3 / #E** — Spike DuckLake vs Iceberg for concurrent reference-data storage;
+  decision record. (M)
+- **M4 / #F** — Implement `aco.lake` `Context` write path (resolve
+  `NotImplementedError`) + pilot OPPS ingestion to Iceberg/DuckLake via
+  Nessie/RustFS. (L)
+- **M5 / #G** — Migrate notebooks to read OPPS/PFS from the lake (iceberg/trino);
+  cut over ingestion; retire the monolith for reference data. (L)
+
+## Interim operating procedure (until M1 lands)
+
+Before a reference-data ingest, ensure no notebook holds the DB:
+
+```bash
+lsof data/aco.duckdb          # find the holder (usually a marimo kernel)
+# close the notebook tab, or stop the specific kernel PID, then:
+uv run python dev/scripts/ingest_opps.py
+```
+
+The discrete issues above are ready to file via
+`dev/scripts/file_concurrency_issues.py` (needs `GITEA_TOKEN`).