feat(notebooks): quality gates — headless integration test, FE smoke gate, dedup issue auto-filer

Three gates so notebook breakage can't ship or linger silently again
(spec: docs/superpowers/specs/2026-07-09-notebook-quality-gates-design.md):

1. nb_integration.py: runs notebooks headless via 'marimo export session',
   parses the JSON snapshots for cell errors, emits a report, exits 1 on
   unexpected failures. New ci.yml notebooks-smoke job runs the
   data-independent [ci_smoke] set (infra/marimo/nb-tests.toml) on every
   push; new nightly notebooks-integration.yml runs the full set inside
   the prod container against real data, filing failures as issues.

2. nb_fe_smoke.py: headless-browser gate that loads the editor and fails
   on any console/page error — the test that would have blocked the
   'd is not a constructor' bundle. Wired into infra-ci.yml after the
   notebooks image build (all traffic over the docker socket; -v bind
   mounts silently arrive empty in CI). Verified: exit 0 on the fixed
   image and live prod, exit 1 on a synthetic crashing page.

3. nb_issue_filer.py + nb-watcher compose sidecar: one issue per error
   signature (notebook + ename + normalized message), rate-limited
   recurrence comments, auto-close after 24h quiet — the watcher is the
   single closing authority. Tails container logs via the docker socket
   (stdlib unix-socket HTTP, stream demux) and parses live session
   snapshots. First production tick filed 9 real deduplicated issues
   (#546-#554: a real skin_subs_explorer bug, missing pyzotero/trino,
   nessie/api connectivity) under the new 'notebooks' label.

Also: notebooks.Dockerfile now lets corepack honor marimo's pinned
packageManager instead of 'pnpm@latest' — the last floating input to the
frontend build after the lockfile fix.

Tests: 6 new unit-test groups (snapshot parsing, signature stability,
dedup decisions); full suite green including notebook-layout policy
(config placed in infra/marimo/, not notebooks/).

.gitea/workflows/ci.yml

@@ -64,3 +64,36 @@ jobs:
             --run "${{ github.run_number }}" \
             --sha "${{ github.sha }}" \
             --ref "${{ github.ref }}" || true
+
+  notebooks-smoke:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: https://github.com/actions/checkout@v4
+
+      - name: Set up uv
+        run: curl -LsSf https://astral.sh/uv/install.sh | sh
+        env:
+          UV_INSTALL_DIR: /usr/local/bin
+
+      - name: Install dependencies
+        run: uv sync --dev
+
+      - name: Run data-independent notebooks headless
+        # Executes the [ci_smoke] set from notebooks/nb-tests.toml via
+        # `marimo export session` and fails on any cell error. Runs on
+        # every push (not path-gated): notebooks import src/ modules, so
+        # src changes can break them too.
+        run: uv run python dev/scripts/nb_integration.py --set ci-smoke
+
+      - name: File failure issue
+        if: failure()
+        env:
+          GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }}
+        run: |
+          uv sync --no-dev --quiet 2>/dev/null || true
+          uv run python -m api.diag.ci \
+            --workflow "CI" --job "notebooks-smoke" \
+            --run "${{ github.run_number }}" \
+            --sha "${{ github.sha }}" \
+            --ref "${{ github.ref }}" || true

.gitea/workflows/infra-ci.yml

@@ -50,6 +50,9 @@ jobs:
       - name: Build notebooks
         run: docker build -f infra/images/notebooks.Dockerfile -t local/notebooks:build .
 
+      - name: Frontend smoke gate
+        run: python3 dev/scripts/nb_fe_smoke.py --image local/notebooks:build --network ci
+
       - name: File failure issue
         if: failure()
         env:

.gitea/workflows/notebooks-integration.yml

@@ -0,0 +1,52 @@
+# DO NOT EDIT — generated by gen_config.py from stack.toml
+# Re-generate: uv run python dev/scripts/gen_config.py
+
+name: Notebooks Integration
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "30 3 * * *"
+
+jobs:
+  notebooks-integration:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: https://github.com/actions/checkout@v4
+
+      - name: Run full notebook set in prod container
+        env:
+          GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }}
+        run: |
+          set -euo pipefail
+          docker exec notebooks mkdir -p /tmp/nbtest
+          docker cp dev/scripts/nb_integration.py notebooks:/tmp/nbtest/
+          docker cp dev/scripts/nb_issue_filer.py notebooks:/tmp/nbtest/
+          docker cp infra/marimo/nb-tests.toml notebooks:/tmp/nbtest/
+          docker exec \
+            -e GITEA_TOKEN \
+            -e GITEA_API_BASE=http://git:3000/api/v1 \
+            notebooks \
+            uv run --project /home/kert/workspace python /tmp/nbtest/nb_integration.py \
+              --set all --file-issues \
+              --nb-dir /home/kert/notebooks \
+              --config /tmp/nbtest/nb-tests.toml \
+              --report /tmp/nbtest/report.json
+          docker cp notebooks:/tmp/nbtest/report.json nb-integration-report.json
+          cat nb-integration-report.json
+
+      - name: Frontend smoke against live service
+        run: python3 dev/scripts/nb_fe_smoke.py --url http://notebooks:2718 --network gateway
+
+      - name: File failure issue
+        if: failure()
+        env:
+          GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }}
+        run: |
+          uv sync --no-dev --quiet 2>/dev/null || true
+          uv run python -m api.diag.ci \
+            --workflow "Notebooks Integration" --job "notebooks-integration" \
+            --run "${{ github.run_number }}" \
+            --sha "${{ github.sha }}" \
+            --ref "${{ github.ref }}" || true

compose.yml

@@ -288,6 +288,44 @@ services:
       - "promtail=true"
     restart: unless-stopped
 
+  # Watches the notebooks service for errors (container logs via the
+  # docker socket + __marimo__/session snapshots) and files deduplicated,
+  # auto-closing Gitea issues via dev/scripts/nb_issue_filer.py. See
+  # docs/superpowers/specs/2026-07-09-notebook-quality-gates-design.md.
+  nb-watcher:
+    image: python:3.13-alpine
+    container_name: nb-watcher
+    # `gateway` to reach git:3000 for the issue API.
+    networks:
+      - gateway
+    environment:
+      - PYTHONUNBUFFERED=1
+      - GITEA_TOKEN=${GITEA_TOKEN}
+      - GITEA_API_BASE=http://git:3000/api/v1
+      - NB_POLL_S=${NB_POLL_S:-60}
+    volumes:
+      - ./dev/scripts:/scripts:ro
+      - ./notebooks:/notebooks:ro
+      - ./.state:/state
+      - ${DOCKER_SOCK:-/run/user/1000/docker.sock}:/var/run/docker.sock:ro
+    command: python /scripts/nb_watcher.py
+    depends_on:
+      notebooks:
+        condition: service_started
+    healthcheck:
+      test:
+        - "CMD-SHELL"
+        - "test -f /tmp/heartbeat && test $$(( $$(date +%s) - $$(stat -c %Y /tmp/heartbeat) )) -lt $$(( $${NB_POLL_S:-60} * 5 ))"
+      interval: 60s
+      timeout: 5s
+      retries: 3
+      start_period: 120s
+    security_opt:
+      - no-new-privileges:true
+    labels:
+      - "promtail=true"
+    restart: unless-stopped
+
   zotero:
     image: ${IMAGE_PREFIX:-fhirworx}/zotero:${HEAVY_TAG:-latest}
     pull_policy: if_not_present

dev/scripts/backends/gitea.py

@@ -185,6 +185,25 @@ jobs:
         run: uv run pytest tests/ -x --cov=src --cov-report=term-missing --cov-fail-under={coverage_threshold} -q -n auto
 
 {_failure_step("CI", "test")}
+
+  notebooks-smoke:
+    runs-on: {runner}
+    steps:
+{_checkout_step()}
+
+{_setup_uv_step(uv_version)}
+
+      - name: Install dependencies
+        run: uv sync --dev
+
+      - name: Run data-independent notebooks headless
+        # Executes the [ci_smoke] set from notebooks/nb-tests.toml via
+        # `marimo export session` and fails on any cell error. Runs on
+        # every push (not path-gated): notebooks import src/ modules, so
+        # src changes can break them too.
+        run: uv run python dev/scripts/nb_integration.py --set ci-smoke
+
+{_failure_step("CI", "notebooks-smoke")}
 """
     return (".gitea/workflows/ci.yml", content)
 
@@ -438,6 +457,17 @@ def _gen_infra_ci(
         if not img.get("hadolint", True):
             continue
         name = img["name"]
+        # The notebooks image gets a frontend smoke gate: boot the freshly
+        # built image and load the editor in a headless browser, failing on
+        # any console/page error. Guards against runtime-broken frontend
+        # bundles that compile cleanly (the 2026-07-09 "d is not a
+        # constructor" incident shipped through a green build).
+        fe_smoke_step = ""
+        if name == "notebooks":
+            fe_smoke_step = f"""
+
+      - name: Frontend smoke gate
+        run: python3 dev/scripts/nb_fe_smoke.py --image local/{name}:build --network ci"""
         jobs_parts.append(f"""\
   {name}:
     runs-on: {runner}
@@ -451,7 +481,7 @@ def _gen_infra_ci(
 
 {_setup_buildx_step()}
 
-{_build_push_step(img, f"ci-test-{name}", "", "", load_only=True)}
+{_build_push_step(img, f"ci-test-{name}", "", "", load_only=True)}{fe_smoke_step}
 
 {_failure_step("Infra CI", name)}""")
 
@@ -475,6 +505,60 @@ jobs:
     return (".gitea/workflows/infra-ci.yml", content)
 
 
+def _gen_notebooks_integration(runner: str, **_kw: object) -> tuple[str, str]:
+    """Nightly full-set notebook run inside the production container.
+
+    The full notebook set needs real data (aco.duckdb, ./data mounts) that
+    only the prod container has, so the scripts are docker-cp'd in and run
+    there. Failures are routed through nb_issue_filer (dedup + auto-close),
+    not exit codes — a red nightly run should page via the issue tracker,
+    not accumulate api.diag.ci duplicates. 03:30 sits before the 06:00
+    pkg-supply-chain run and outside interactive hours (duckdb lock, #508).
+    """
+    content = f"""\
+{_HEADER}
+name: Notebooks Integration
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "30 3 * * *"
+
+jobs:
+  notebooks-integration:
+    runs-on: {runner}
+    steps:
+{_checkout_step()}
+
+      - name: Run full notebook set in prod container
+        env:
+          GITEA_TOKEN: ${{{{ secrets.DEPLOY_TOKEN }}}}
+        run: |
+          set -euo pipefail
+          docker exec notebooks mkdir -p /tmp/nbtest
+          docker cp dev/scripts/nb_integration.py notebooks:/tmp/nbtest/
+          docker cp dev/scripts/nb_issue_filer.py notebooks:/tmp/nbtest/
+          docker cp infra/marimo/nb-tests.toml notebooks:/tmp/nbtest/
+          docker exec \\
+            -e GITEA_TOKEN \\
+            -e GITEA_API_BASE=http://git:3000/api/v1 \\
+            notebooks \\
+            uv run --project /home/kert/workspace python /tmp/nbtest/nb_integration.py \\
+              --set all --file-issues \\
+              --nb-dir /home/kert/notebooks \\
+              --config /tmp/nbtest/nb-tests.toml \\
+              --report /tmp/nbtest/report.json
+          docker cp notebooks:/tmp/nbtest/report.json nb-integration-report.json
+          cat nb-integration-report.json
+
+      - name: Frontend smoke against live service
+        run: python3 dev/scripts/nb_fe_smoke.py --url http://notebooks:2718 --network gateway
+
+{_failure_step("Notebooks Integration", "notebooks-integration")}
+"""
+    return (".gitea/workflows/notebooks-integration.yml", content)
+
+
 def _gen_release(runner: str, uv_version: str, **_kw: object) -> tuple[str, str]:
     content = f"""\
 {_HEADER}
@@ -540,6 +624,7 @@ def emit(
         _gen_harden,
         _gen_rebuild_all,
         _gen_infra_ci,
+        _gen_notebooks_integration,
         _gen_release,
     ):
         path, content = gen_fn(**common)  # type: ignore[arg-type]

dev/scripts/nb_fe_smoke.py

@@ -0,0 +1,256 @@
+"""Frontend smoke test for the notebooks (marimo) image/service.
+
+Loads the editor in a headless browser and fails on any console error or
+page error. This is the gate that would have caught the 2026-07-09 incident:
+a frontend bundle that compiled cleanly but threw "d is not a constructor"
+on every notebook open, while the container healthcheck and home page stayed
+green — no Python-level test can see that class of breakage.
+
+Two modes:
+
+  # Gate a freshly built image (CI, after `docker buildx --load`):
+  python dev/scripts/nb_fe_smoke.py --image ci-test-notebooks --network ci
+
+  # Probe a running service (post-deploy, nightly):
+  python dev/scripts/nb_fe_smoke.py --url http://notebooks:2718 --network gateway
+
+Everything crosses the docker socket (docker cp / stdin), never bind mounts:
+in CI the job's filesystem lives in a volume the host daemon can't -v mount,
+so mounted paths silently arrive empty (the trap that broke the apt mirror).
+
+Stdlib only.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+import tempfile
+import time
+import uuid
+from pathlib import Path
+
+PLAYWRIGHT_IMAGE = "mcr.microsoft.com/playwright/python:v1.61.0-noble"
+PLAYWRIGHT_PKG = "playwright==1.61.0"
+READY_TIMEOUT_S = 120
+
+# Dependency-free notebook shipped into the throwaway container. Markdown
+# only, so it renders identically under any runtime config (the bare image
+# has no marimo.toml, and marimo's default auto_instantiate=true would run
+# code cells).
+PROBE_NB = """\
+import marimo
+
+app = marimo.App()
+
+@app.cell
+def _():
+    import marimo as mo
+    mo.md("fe-smoke-canary-cell")
+    return
+
+if __name__ == "__main__":
+    app.run()
+"""
+
+# Runs inside the playwright container; URL substituted via argv.
+PROBE_SCRIPT = """
+import json, sys
+from playwright.sync_api import sync_playwright
+
+url = sys.argv[1]
+events = []
+with sync_playwright() as p:
+    browser = p.chromium.launch()
+    page = browser.new_page()
+    page.on("console", lambda m: events.append({"type": m.type, "text": m.text}))
+    page.on("pageerror", lambda e: events.append({"type": "pageerror", "text": str(e)}))
+    page.goto(url, wait_until="networkidle", timeout=60000)
+    page.wait_for_timeout(6000)
+    body = page.inner_text("body")
+    browser.close()
+print("FE_SMOKE_RESULT " + json.dumps({"events": events, "body": body[:3000]}))
+"""
+
+
+def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
+    return subprocess.run(cmd, capture_output=True, text=True, timeout=300, **kw)
+
+
+def wait_ready(url: str, network: str) -> bool:
+    """Poll the target URL from a curl container on the same network."""
+    deadline = time.time() + READY_TIMEOUT_S
+    while time.time() < deadline:
+        r = _run(
+            [
+                "docker",
+                "run",
+                "--rm",
+                "--network",
+                network,
+                "curlimages/curl:8.11.1",
+                "-s",
+                "-o",
+                "/dev/null",
+                "-w",
+                "%{http_code}",
+                "--max-time",
+                "5",
+                url,
+            ]
+        )
+        if r.stdout.strip() == "200":
+            return True
+        time.sleep(3)
+    return False
+
+
+def probe(url: str, network: str) -> dict:
+    """Load `url` headlessly; return {events, body}."""
+    shell = (
+        f"pip install -q {PLAYWRIGHT_PKG} 2>/dev/null; "
+        f"python3 - \"$0\" <<'PYEOF'\n{PROBE_SCRIPT}\nPYEOF"
+    )
+    r = subprocess.run(
+        [
+            "docker",
+            "run",
+            "--rm",
+            "-i",
+            "--network",
+            network,
+            "--entrypoint",
+            "bash",
+            PLAYWRIGHT_IMAGE,
+            "-c",
+            shell,
+            url,
+        ],
+        capture_output=True,
+        text=True,
+        timeout=600,
+    )
+    for line in r.stdout.splitlines():
+        if line.startswith("FE_SMOKE_RESULT "):
+            return json.loads(line[len("FE_SMOKE_RESULT ") :])
+    raise RuntimeError(
+        f"probe produced no result (rc={r.returncode}):\n"
+        f"{r.stdout[-1000:]}\n{r.stderr[-1000:]}"
+    )
+
+
+def check(result: dict, *, expect_text: str | None) -> list[str]:
+    problems = [
+        f"[{e['type']}] {e['text'][:300]}"
+        for e in result["events"]
+        if e["type"] in ("error", "pageerror")
+    ]
+    if expect_text and expect_text not in result["body"]:
+        problems.append(
+            f"expected editor content {expect_text!r} not found in page body "
+            f"(got: {result['body'][:200]!r})"
+        )
+    return problems
+
+
+def smoke_image(image: str, network: str) -> int:
+    """Boot a throwaway container from `image`, probe '/', probe the editor."""
+    name = f"nb-fe-smoke-{uuid.uuid4().hex[:8]}"
+    with tempfile.TemporaryDirectory() as td:
+        nb_dir = Path(td) / "notebooks"
+        nb_dir.mkdir()
+        (nb_dir / "fe_smoke_probe.py").write_text(PROBE_NB)
+        try:
+            r = _run(["docker", "create", "--name", name, "--network", network, image])
+            if r.returncode != 0:
+                print(f"docker create failed: {r.stderr[:500]}", file=sys.stderr)
+                return 2
+            # docker cp streams the client's file over the socket — works
+            # from inside CI where -v bind mounts cannot.
+            _run(["docker", "cp", str(nb_dir), f"{name}:/home/kert/"])
+            _run(["docker", "start", name])
+            _run(
+                [
+                    "docker",
+                    "exec",
+                    "-u",
+                    "0",
+                    name,
+                    "chmod",
+                    "-R",
+                    "a+rwX",
+                    "/home/kert/notebooks",
+                ]
+            )
+
+            base = f"http://{name}:2718"
+            if not wait_ready(f"{base}/", network):
+                logs = _run(["docker", "logs", "--tail", "30", name])
+                print(
+                    f"service never became ready at {base}\n{logs.stdout}{logs.stderr}",
+                    file=sys.stderr,
+                )
+                return 2
+            return _verdict(
+                base,
+                network,
+                "?file=fe_smoke_probe.py",
+                expect_text="fe-smoke-canary-cell",
+            )
+        finally:
+            _run(["docker", "rm", "-f", name])
+
+
+def smoke_url(url: str, network: str, notebook: str | None) -> int:
+    base = url.rstrip("/")
+    if not wait_ready(f"{base}/", network):
+        print(f"service not responding at {base}", file=sys.stderr)
+        return 2
+    query = f"?file={notebook}" if notebook else ""
+    # Live service: don't assert specific content (notebook set varies),
+    # only require an error-free load.
+    return _verdict(base, network, query, expect_text=None)
+
+
+def _verdict(
+    base: str, network: str, editor_query: str, expect_text: str | None
+) -> int:
+    failed = False
+    for label, target, expect in (
+        ("home", f"{base}/", None),
+        ("editor", f"{base}/{editor_query}", expect_text),
+    ):
+        result = probe(target, network)
+        problems = check(result, expect_text=expect)
+        if problems:
+            failed = True
+            print(f"✗ {label} {target}")
+            for p in problems:
+                print(f"    {p}")
+        else:
+            print(f"✓ {label} {target} — no console/page errors")
+    return 1 if failed else 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    target = parser.add_mutually_exclusive_group(required=True)
+    target.add_argument("--image", help="image tag to boot and gate")
+    target.add_argument("--url", help="running service base URL to probe")
+    parser.add_argument(
+        "--network", required=True, help="docker network shared with the target"
+    )
+    parser.add_argument(
+        "--notebook", default="sample.py", help="notebook to open in --url mode"
+    )
+    args = parser.parse_args()
+
+    if args.image:
+        return smoke_image(args.image, args.network)
+    return smoke_url(args.url, args.network, args.notebook)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

dev/scripts/nb_integration.py

@@ -0,0 +1,246 @@
+"""Headless notebook integration test.
+
+Runs `marimo export session` over a set of notebooks (executing them and
+writing JSON session snapshots), parses each snapshot for cell errors, and
+emits a consolidated JSON report. Exits 1 if any notebook outside the
+expected-failures allowlist errored.
+
+Why: the pytest suite only exercises src/ — nothing ran the notebooks
+themselves, so kernel-level breakage (dead imports, schema drift, service
+API changes) only surfaced when a human opened the notebook.
+
+Config: infra/marimo/nb-tests.toml
+    [ci_smoke]
+    notebooks = ["sample.py"]          # data-independent, runs in CI
+    [expected_failures]
+    "acodb_explorer.py" = "reason"     # reported but non-fatal
+
+Usage:
+    uv run python dev/scripts/nb_integration.py --set ci-smoke
+    uv run python dev/scripts/nb_integration.py --set all --file-issues
+"""
+
+from __future__ import annotations
+
+import argparse
+import html
+import json
+import re
+import subprocess
+import sys
+import time
+import tomllib
+from pathlib import Path
+
+# In the repo this file sits at dev/scripts/, two levels below the root.
+# Deployed contexts (nb-watcher mounts it at /scripts, the nightly run
+# docker-cps it to /tmp/nbtest) are shallower — fall back to "/" there;
+# those contexts always override paths via --nb-dir / env anyway.
+_parents = Path(__file__).resolve().parents
+ROOT = _parents[2] if len(_parents) > 2 else Path("/")
+NB_DIR = ROOT / "notebooks"
+# Config lives in infra/marimo/ (notebooks/ holds only .py content — see
+# tests/test_notebook_layout.py). Overridable via --config for contexts
+# where the repo layout doesn't exist (nightly docker exec).
+CONFIG_PATH = ROOT / "infra" / "marimo" / "nb-tests.toml"
+REPORT_PATH = ROOT / "data" / "nb-integration-report.json"
+TIMEOUT_S = 600
+
+
+# ── snapshot parsing (pure) ──────────────────────────────────────
+
+
+def _strip_html(text: str) -> str:
+    return html.unescape(re.sub(r"<[^>]+>", "", text))
+
+
+def parse_snapshot(path: Path) -> list[dict]:
+    """Extract cell errors from a __marimo__/session/<nb>.py.json snapshot."""
+    if not path.exists():
+        return [
+            {
+                "cell": "",
+                "ename": "snapshot-missing",
+                "evalue": f"no session snapshot at {path.name}",
+                "detail": "",
+            }
+        ]
+    try:
+        snap = json.loads(path.read_text())
+    except (OSError, json.JSONDecodeError) as e:
+        return [
+            {
+                "cell": "",
+                "ename": "snapshot-unreadable",
+                "evalue": str(e)[:200],
+                "detail": "",
+            }
+        ]
+
+    errors: list[dict] = []
+    for cell in snap.get("cells", []):
+        for out in cell.get("outputs", []):
+            if out.get("type") != "error":
+                continue
+            detail = ""
+            for con in cell.get("console", []):
+                if con.get("name") == "stderr":
+                    detail += _strip_html(con.get("text", "")) + "\n"
+            errors.append(
+                {
+                    "cell": cell.get("id", ""),
+                    "ename": out.get("ename", "error"),
+                    "evalue": out.get("evalue", ""),
+                    "detail": detail.strip(),
+                }
+            )
+    return errors
+
+
+def classify(notebook: str, errors: list[dict], expected: dict[str, str]) -> str:
+    if not errors:
+        return "pass"
+    return "expected-fail" if notebook in expected else "fail"
+
+
+def exit_code(results: dict[str, dict]) -> int:
+    return 1 if any(r["status"] == "fail" for r in results.values()) else 0
+
+
+# ── execution ────────────────────────────────────────────────────
+
+
+def load_config(path: Path) -> dict:
+    if not path.exists():
+        return {"ci_smoke": {"notebooks": []}, "expected_failures": {}}
+    return tomllib.loads(path.read_text())
+
+
+def select_notebooks(which: str, cfg: dict) -> list[str]:
+    if which == "ci-smoke":
+        return list(cfg.get("ci_smoke", {}).get("notebooks", []))
+    return sorted(
+        p.name for p in NB_DIR.glob("*.py") if not p.name.startswith((".", "__"))
+    )
+
+
+def run_notebook(name: str) -> tuple[int, str]:
+    """Execute one notebook via marimo export session; snapshot lands in
+    notebooks/__marimo__/session/<name>.json regardless of exit code."""
+    try:
+        proc = subprocess.run(
+            [
+                "marimo",
+                "export",
+                "session",
+                name,
+                "--no-sandbox",
+                "--force-overwrite",
+                "--continue-on-error",
+            ],
+            cwd=NB_DIR,
+            capture_output=True,
+            text=True,
+            timeout=TIMEOUT_S,
+        )
+        return proc.returncode, (proc.stderr or "")[-1500:]
+    except subprocess.TimeoutExpired:
+        return -1, f"timed out after {TIMEOUT_S}s"
+    except FileNotFoundError:
+        return -2, "marimo CLI not found on PATH"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--set", dest="which", choices=["ci-smoke", "all"], default="ci-smoke"
+    )
+    parser.add_argument(
+        "--file-issues",
+        action="store_true",
+        help="report failures via nb_issue_filer (dedup) "
+        "and sweep resolved ones, instead of only exiting 1",
+    )
+    parser.add_argument("--report", default=str(REPORT_PATH))
+    parser.add_argument("--nb-dir", help="notebooks directory (default: repo)")
+    parser.add_argument("--config", default=str(CONFIG_PATH), help="nb-tests.toml path")
+    args = parser.parse_args()
+
+    if args.nb_dir:
+        global NB_DIR
+        NB_DIR = Path(args.nb_dir).resolve()
+
+    cfg = load_config(Path(args.config))
+    expected = {k: str(v) for k, v in cfg.get("expected_failures", {}).items()}
+    notebooks = select_notebooks(args.which, cfg)
+    if not notebooks:
+        print(f"no notebooks selected for --set {args.which}")
+        return 0
+
+    results: dict[str, dict] = {}
+    for name in notebooks:
+        print(f"→ {name}", flush=True)
+        rc, stderr_tail = run_notebook(name)
+        if rc == -1:
+            errors = [
+                {"cell": "", "ename": "timeout", "evalue": stderr_tail, "detail": ""}
+            ]
+        elif rc == -2:
+            print(stderr_tail, file=sys.stderr)
+            return 2
+        else:
+            errors = parse_snapshot(NB_DIR / "__marimo__" / "session" / f"{name}.json")
+        status = classify(name, errors, expected)
+        results[name] = {"status": status, "errors": errors}
+        tag = {"pass": "ok", "fail": "FAIL", "expected-fail": "expected-fail"}[status]
+        print(
+            f"  {tag}"
+            + (
+                f" — {errors[0]['ename']}: {errors[0]['evalue'][:120]}"
+                if errors
+                else ""
+            )
+        )
+
+    report = {
+        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "set": args.which,
+        "marimo_set_size": len(notebooks),
+        "results": results,
+    }
+    report_path = Path(args.report)
+    report_path.parent.mkdir(parents=True, exist_ok=True)
+    report_path.write_text(json.dumps(report, indent=2) + "\n")
+    print(f"report → {report_path}")
+
+    counts: dict[str, int] = {}
+    for r in results.values():
+        counts[r["status"]] = counts.get(r["status"], 0) + 1
+    print("summary:", ", ".join(f"{k}={v}" for k, v in sorted(counts.items())))
+
+    if args.file_issues:
+        sys.path.insert(0, str(Path(__file__).parent))
+        import nb_issue_filer  # noqa: PLC0415 — sibling script, lazy by design
+
+        findings = [
+            {"notebook": nb, **e}
+            for nb, r in results.items()
+            if r["status"] in ("fail", "expected-fail")
+            for e in r["errors"]
+        ]
+        source = f"nb-integration/{args.which}"
+        if findings:
+            nb_issue_filer.cmd_report(findings, source)
+        else:
+            print("no findings — nothing to file")
+        # No sweep here: closing is the nb-watcher's job (24h staleness).
+        # Sweeping with only this run's signatures would close issues for
+        # errors this run can't reproduce (e.g. server-log-only errors the
+        # watcher filed).
+        return 0  # issue filing is the failure channel for scheduled runs
+
+    return exit_code(results)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

dev/scripts/nb_issue_filer.py

@@ -0,0 +1,294 @@
+"""Dedup/auto-close Gitea issue filer for notebook errors.
+
+One open issue per error signature (notebook + ename + normalized message).
+Recurrences add a rate-limited comment instead of a new issue; signatures
+that stop occurring get auto-closed. This replaces the api.diag.ci
+one-issue-per-run pattern that accumulated 23 duplicates for a single
+workflow.
+
+Stdlib only — runs on a bare python image (nb-watcher sidecar), in CI, and
+on the host.
+
+Usage:
+    # file/refresh issues for findings on stdin (JSON list, see FINDING)
+    python nb_issue_filer.py report --source nightly-integration < findings.json
+
+    # close open nb issues whose signature is NOT in the active set
+    python nb_issue_filer.py sweep --active-sigs sig1,sig2 --source nightly-integration
+    # or close signatures unseen since a state file's cutoff
+    python nb_issue_filer.py sweep --state data/nb-watcher-state.json --max-age-h 24
+
+FINDING = {"notebook": "x.py", "ename": "exception", "evalue": "...",
+           "detail": "traceback/console excerpt"}
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import sys
+import time
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+# Repo root when run in-tree; "/" in shallow deployed contexts (sidecar
+# mount, docker cp) where ROOT is only a .env-fallback location anyway.
+_parents = Path(__file__).resolve().parents
+ROOT = _parents[2] if len(_parents) > 2 else Path("/")
+
+API_BASE = os.environ.get("GITEA_API_BASE", "https://git.fhirworx.io/api/v1")
+OWNER_REPO = os.environ.get("GITEA_OWNER_REPO", "homelab/stack")
+LABEL_NAME = os.environ.get("NB_ISSUE_LABEL", "notebooks")
+MARKER = "nb-sig:"
+COOLDOWN_S = int(os.environ.get("NB_ISSUE_COOLDOWN_S", str(6 * 3600)))
+
+
+# ── pure logic ───────────────────────────────────────────────────
+
+
+def normalize_evalue(evalue: str) -> str:
+    """Collapse volatile details so recurring errors hash identically."""
+    s = evalue.strip()
+    s = re.sub(r"0x[0-9a-fA-F]+", "0xN", s)  # addresses
+    s = re.sub(r"(/[\w.\-]+)+", "/PATH", s)  # absolute paths
+    s = re.sub(r"\d+(\.\d+)?", "N", s)  # counts, line numbers, durations
+    s = re.sub(r"\s+", " ", s)
+    return s.lower()[:300]
+
+
+def signature(notebook: str, ename: str, evalue: str) -> str:
+    key = f"{notebook}|{ename}|{normalize_evalue(evalue)}"
+    return hashlib.sha1(key.encode()).hexdigest()[:12]
+
+
+def decide(
+    existing: dict | None, last_comment_age_s: float | None, cooldown_s: int
+) -> str:
+    """create | comment | skip for one observed signature."""
+    if existing is None:
+        return "create"
+    if last_comment_age_s is None or last_comment_age_s >= cooldown_s:
+        return "comment"
+    return "skip"
+
+
+def build_body(
+    *, notebook: str, ename: str, evalue: str, source: str, detail: str, sig: str
+) -> str:
+    detail = detail.strip()[:4000]
+    return (
+        f"Auto-filed by the notebook error filer (`{MARKER}{sig}` — do not edit "
+        f"this marker; dedup and auto-close key on it).\n\n"
+        f"- **Notebook:** `{notebook}`\n"
+        f"- **Error:** `{ename}`: {evalue[:500]}\n"
+        f"- **Source:** {source}\n\n"
+        f"```\n{detail or '(no traceback captured)'}\n```\n\n"
+        f"This issue is closed automatically when the error stops occurring."
+    )
+
+
+def extract_sig(body: str) -> str | None:
+    m = re.search(rf"{MARKER}([0-9a-f]{{12}})", body or "")
+    return m.group(1) if m else None
+
+
+# ── Gitea API (stdlib) ───────────────────────────────────────────
+
+
+def _token() -> str:
+    tok = os.environ.get("GITEA_TOKEN", "")
+    if not tok:
+        env_file = ROOT / ".env"
+        if env_file.exists():
+            for line in env_file.read_text().splitlines():
+                if line.startswith("GITEA_TOKEN="):
+                    tok = line.split("=", 1)[1].strip().strip('"').strip("'")
+    if not tok:
+        print("ERROR: GITEA_TOKEN not set and not found in .env", file=sys.stderr)
+        raise SystemExit(2)
+    return tok
+
+
+def api(method: str, path: str, data: dict | None = None) -> tuple[int, object]:
+    url = f"{API_BASE}/{path}"
+    body = json.dumps(data).encode() if data is not None else None
+    req = urllib.request.Request(
+        url,
+        data=body,
+        method=method,
+        headers={
+            "Authorization": f"token {_token()}",
+            "Content-Type": "application/json",
+        },
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            raw = resp.read().decode()
+            return resp.status, (json.loads(raw) if raw else None)
+    except urllib.error.HTTPError as e:  # noqa: PERF203 — single call site
+        return e.code, e.read().decode()[:300]
+    except urllib.error.URLError as e:
+        return 0, str(e)
+
+
+def _open_nb_issues() -> list[dict]:
+    """All open issues carrying our marker, keyed lookup done by caller."""
+    issues: list[dict] = []
+    page = 1
+    while True:
+        status, batch = api(
+            "GET",
+            f"repos/{OWNER_REPO}/issues?state=open&type=issues"
+            f"&q={urllib.parse.quote(MARKER)}&page={page}&limit=50",
+        )
+        if status != 200 or not isinstance(batch, list) or not batch:
+            break
+        issues.extend(batch)
+        if len(batch) < 50:
+            break
+        page += 1
+    return [i for i in issues if extract_sig(i.get("body", ""))]
+
+
+def _label_id() -> int | None:
+    status, labels = api("GET", f"repos/{OWNER_REPO}/labels?limit=50")
+    if status == 200 and isinstance(labels, list):
+        for lab in labels:
+            if lab.get("name") == LABEL_NAME:
+                return lab.get("id")
+    return None
+
+
+def _last_comment_age_s(issue: dict) -> float | None:
+    ts = issue.get("updated_at")
+    if not ts:
+        return None
+    try:
+        then = time.mktime(time.strptime(ts[:19], "%Y-%m-%dT%H:%M:%S"))
+        return max(0.0, time.time() - time.timezone - (then - time.timezone))
+    except ValueError:
+        return None
+
+
+# ── commands ─────────────────────────────────────────────────────
+
+
+def cmd_report(findings: list[dict], source: str) -> int:
+    open_by_sig = {extract_sig(i["body"]): i for i in _open_nb_issues()}
+    label = _label_id()
+    seen: set[str] = set()
+    failures = 0
+
+    for f in findings:
+        sig = signature(f["notebook"], f["ename"], f.get("evalue", ""))
+        if sig in seen:  # same error many times in one run → one action
+            continue
+        seen.add(sig)
+        existing = open_by_sig.get(sig)
+        age = _last_comment_age_s(existing) if existing else None
+        action = decide(existing, age, COOLDOWN_S)
+
+        if action == "create":
+            payload: dict = {
+                "title": f"[nb] {f['notebook']}: {f['ename']}: "
+                f"{f.get('evalue', '')[:120]}",
+                "body": build_body(
+                    notebook=f["notebook"],
+                    ename=f["ename"],
+                    evalue=f.get("evalue", ""),
+                    source=source,
+                    detail=f.get("detail", ""),
+                    sig=sig,
+                ),
+            }
+            if label:
+                payload["labels"] = [label]
+            status, resp = api("POST", f"repos/{OWNER_REPO}/issues", payload)
+            if status == 201 and isinstance(resp, dict):
+                print(f"  created #{resp.get('number')} [{sig}] {f['notebook']}")
+            else:
+                print(f"  CREATE FAILED ({status}) [{sig}]: {resp}", file=sys.stderr)
+                failures += 1
+        elif action == "comment":
+            n = existing["number"]
+            status, _ = api(
+                "POST",
+                f"repos/{OWNER_REPO}/issues/{n}/comments",
+                {
+                    "body": f"Still occurring via {source} at "
+                    f"{time.strftime('%Y-%m-%d %H:%M UTC', time.gmtime())}."
+                },
+            )
+            print(
+                f"  commented #{n} [{sig}]"
+                if status == 201
+                else f"  COMMENT FAILED ({status}) #{n}"
+            )
+        else:
+            print(f"  skip [{sig}] (within cooldown)")
+
+    return 1 if failures else 0
+
+
+def cmd_sweep(active_sigs: set[str], source: str) -> int:
+    closed = 0
+    for issue in _open_nb_issues():
+        sig = extract_sig(issue["body"])
+        if sig in active_sigs:
+            continue
+        n = issue["number"]
+        api(
+            "POST",
+            f"repos/{OWNER_REPO}/issues/{n}/comments",
+            {"body": f"No longer occurring ({source}); auto-closing."},
+        )
+        status, _ = api("PATCH", f"repos/{OWNER_REPO}/issues/{n}", {"state": "closed"})
+        if status in (200, 201):
+            print(f"  closed #{n} [{sig}]")
+            closed += 1
+        else:
+            print(f"  CLOSE FAILED ({status}) #{n}", file=sys.stderr)
+    print(f"sweep: closed {closed}")
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    sub = parser.add_subparsers(dest="cmd", required=True)
+
+    p_report = sub.add_parser("report", help="file/refresh issues for findings")
+    p_report.add_argument("--source", required=True)
+    p_report.add_argument("--findings", help="path to findings JSON (default: stdin)")
+
+    p_sweep = sub.add_parser("sweep", help="close issues for inactive signatures")
+    p_sweep.add_argument("--source", required=True)
+    p_sweep.add_argument(
+        "--active-sigs", default="", help="comma-separated signatures still failing"
+    )
+    p_sweep.add_argument("--state", help="watcher state JSON: {sig: last_seen_epoch}")
+    p_sweep.add_argument("--max-age-h", type=float, default=24.0)
+
+    args = parser.parse_args()
+
+    if args.cmd == "report":
+        raw = Path(args.findings).read_text() if args.findings else sys.stdin.read()
+        findings = json.loads(raw) if raw.strip() else []
+        if not findings:
+            print("no findings — nothing to file")
+            return 0
+        return cmd_report(findings, args.source)
+
+    active: set[str] = {s for s in args.active_sigs.split(",") if s}
+    if args.state and Path(args.state).exists():
+        state = json.loads(Path(args.state).read_text())
+        cutoff = time.time() - args.max_age_h * 3600
+        active |= {sig for sig, ts in state.items() if ts >= cutoff}
+    return cmd_sweep(active, args.source)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

dev/scripts/nb_watcher.py

@@ -0,0 +1,235 @@
+"""Watch the running notebooks (marimo) service for errors and auto-file
+deduplicated Gitea issues.
+
+Two streams, one signature space (shared with nb_integration.py):
+
+1. Container stdout/stderr via the docker socket (server errors, kernel
+   tracebacks) — read with stdlib http.client over the unix socket, no
+   docker CLI needed.
+2. Session snapshots at <notebooks>/__marimo__/session/*.json (cell-level
+   errors from live sessions and nightly integration runs alike).
+
+Findings go through nb_issue_filer (create/comment with cooldown). This
+watcher is the single closing authority: signatures unseen for
+NB_CLOSE_AFTER_H hours are swept closed. State (last-seen per signature,
+log cursor) persists in NB_STATE_FILE.
+
+Runs as the `nb-watcher` compose sidecar with dev/scripts mounted at
+/scripts and the notebooks dir at /notebooks (both ro).
+"""
+
+from __future__ import annotations
+
+import http.client
+import json
+import os
+import re
+import socket
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+import nb_integration  # noqa: E402 — sibling: parse_snapshot
+import nb_issue_filer  # noqa: E402 — sibling: cmd_report/cmd_sweep/signature
+
+DOCKER_SOCK = os.environ.get("DOCKER_SOCK", "/var/run/docker.sock")
+CONTAINER = os.environ.get("NB_CONTAINER", "notebooks")
+NB_DIR = Path(os.environ.get("NB_DIR", "/notebooks"))
+STATE_FILE = Path(os.environ.get("NB_STATE_FILE", "/state/nb-watcher-state.json"))
+HEARTBEAT = Path(os.environ.get("NB_HEARTBEAT", "/tmp/heartbeat"))
+POLL_S = int(os.environ.get("NB_POLL_S", "60"))
+SWEEP_EVERY_S = int(os.environ.get("NB_SWEEP_EVERY_S", "3600"))
+CLOSE_AFTER_H = float(os.environ.get("NB_CLOSE_AFTER_H", "24"))
+SOURCE = "nb-watcher"
+
+_TRACEBACK_START = "Traceback (most recent call last):"
+
+
+# ── docker logs via unix socket ──────────────────────────────────
+
+
+class _UnixConn(http.client.HTTPConnection):
+    def __init__(self, path: str):
+        super().__init__("localhost", timeout=30)
+        self._unix_path = path
+
+    def connect(self) -> None:
+        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        s.settimeout(30)
+        s.connect(self._unix_path)
+        self.sock = s
+
+
+def fetch_logs(since_epoch: int) -> str:
+    """Return decoded log text since `since_epoch` (demuxing the docker
+    multiplexed stream format when present)."""
+    conn = _UnixConn(DOCKER_SOCK)
+    try:
+        conn.request(
+            "GET",
+            f"/containers/{CONTAINER}/logs?stdout=1&stderr=1&since={since_epoch}",
+        )
+        resp = conn.getresponse()
+        raw = resp.read()
+    finally:
+        conn.close()
+    if not raw:
+        return ""
+    # Multiplexed frames: [stream:1][pad:3][len:4 BE][payload]. TTY
+    # containers emit raw text instead — detect by header shape.
+    if raw[0] in (0, 1, 2) and raw[1:4] == b"\x00\x00\x00":
+        out = bytearray()
+        i = 0
+        while i + 8 <= len(raw):
+            size = int.from_bytes(raw[i + 4 : i + 8], "big")
+            out += raw[i + 8 : i + 8 + size]
+            i += 8 + size
+        return out.decode(errors="replace")
+    return raw.decode(errors="replace")
+
+
+def parse_log_errors(text: str) -> list[dict]:
+    """Extract tracebacks (and their final ExcType: message) from log text."""
+    findings: list[dict] = []
+    lines = text.splitlines()
+    i = 0
+    while i < len(lines):
+        if _TRACEBACK_START in lines[i]:
+            block = [lines[i]]
+            i += 1
+            while i < len(lines) and (
+                lines[i].startswith((" ", "\t")) or not lines[i].strip()
+            ):
+                block.append(lines[i])
+                i += 1
+            # final "ExcType: message" line belongs to the traceback
+            tail = lines[i] if i < len(lines) else ""
+            m = re.match(
+                r"^([A-Za-z_][\w.]*(?:Error|Exception|Interrupt|Warning))\b:?\s*(.*)",
+                tail.strip(),
+            )
+            if m:
+                block.append(tail)
+                i += 1
+                ename, evalue = m.group(1), m.group(2) or "(no message)"
+            else:
+                ename, evalue = "traceback", block[-1].strip()[:200] or "(unknown)"
+            findings.append(
+                {
+                    "notebook": "(service)",
+                    "ename": ename,
+                    "evalue": evalue,
+                    "detail": "\n".join(block)[-3500:],
+                }
+            )
+        else:
+            i += 1
+    return findings
+
+
+# ── session snapshots ────────────────────────────────────────────
+
+
+def scan_snapshots(since_mtime: float) -> tuple[list[dict], float]:
+    findings: list[dict] = []
+    newest = since_mtime
+    session_dir = NB_DIR / "__marimo__" / "session"
+    if not session_dir.is_dir():
+        return findings, newest
+    for snap in session_dir.glob("*.json"):
+        try:
+            mtime = snap.stat().st_mtime
+        except OSError:
+            continue
+        if mtime <= since_mtime:
+            continue
+        newest = max(newest, mtime)
+        notebook = snap.name.removesuffix(".json")
+        for err in nb_integration.parse_snapshot(snap):
+            if err["ename"] in ("snapshot-missing", "snapshot-unreadable"):
+                continue
+            findings.append({"notebook": notebook, **err})
+    return findings, newest
+
+
+# ── state ────────────────────────────────────────────────────────
+
+
+def load_state() -> dict:
+    if STATE_FILE.exists():
+        try:
+            return json.loads(STATE_FILE.read_text())
+        except (OSError, json.JSONDecodeError):
+            pass
+    return {"sigs": {}, "log_cursor": int(time.time()), "snap_cursor": 0.0}
+
+
+def save_state(state: dict) -> None:
+    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    tmp = STATE_FILE.with_suffix(".tmp")
+    tmp.write_text(json.dumps(state))
+    tmp.replace(STATE_FILE)
+
+
+# ── main loop ────────────────────────────────────────────────────
+
+
+def tick(state: dict) -> list[dict]:
+    """One poll: gather findings from both streams, advance cursors."""
+    findings: list[dict] = []
+
+    cursor = state.get("log_cursor", int(time.time()))
+    now = int(time.time())
+    try:
+        findings += parse_log_errors(fetch_logs(cursor))
+        state["log_cursor"] = now
+    except (OSError, http.client.HTTPException) as e:
+        print(f"log fetch failed (will retry): {e}", file=sys.stderr)
+
+    snaps, newest = scan_snapshots(state.get("snap_cursor", 0.0))
+    findings += snaps
+    state["snap_cursor"] = newest
+
+    for f in findings:
+        sig = nb_issue_filer.signature(f["notebook"], f["ename"], f["evalue"])
+        state.setdefault("sigs", {})[sig] = time.time()
+    return findings
+
+
+def main() -> int:
+    state = load_state()
+    last_sweep = 0.0
+    print(
+        f"nb-watcher: container={CONTAINER} nb_dir={NB_DIR} "
+        f"poll={POLL_S}s close_after={CLOSE_AFTER_H}h"
+    )
+    while True:
+        findings = tick(state)
+        if findings:
+            print(f"{len(findings)} finding(s)")
+            try:
+                nb_issue_filer.cmd_report(findings, SOURCE)
+            except Exception as e:  # noqa: BLE001 — keep the watcher alive
+                print(f"filer error (continuing): {e}", file=sys.stderr)
+
+        if time.time() - last_sweep >= SWEEP_EVERY_S:
+            cutoff = time.time() - CLOSE_AFTER_H * 3600
+            active = {s for s, ts in state.get("sigs", {}).items() if ts >= cutoff}
+            try:
+                nb_issue_filer.cmd_sweep(active, SOURCE)
+            except Exception as e:  # noqa: BLE001
+                print(f"sweep error (continuing): {e}", file=sys.stderr)
+            last_sweep = time.time()
+            # drop long-stale sigs so state doesn't grow unbounded
+            state["sigs"] = {
+                s: ts for s, ts in state.get("sigs", {}).items() if ts >= cutoff
+            }
+
+        save_state(state)
+        HEARTBEAT.touch()
+        time.sleep(POLL_S)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

docs/superpowers/specs/2026-07-09-notebook-quality-gates-design.md

@@ -62,8 +62,10 @@ Wired in two places:
   build — boots a throwaway container from the just-built image on the CI
   docker network, probes it, tears it down. A bundle that crashes the editor
   can no longer ship.
-- **Post-deploy check:** `deploy.yml` notebooks job probes the live service
+- **Live-service check:** the nightly workflow probes the running service.
-  after recreate.
+  (`deploy.yml` only builds/pushes images — container recreation is a manual
+  `docker compose up -d` — so there is no in-workflow "post-deploy" moment
+  to hook.)
 
 ### 3. `dev/scripts/nb_issue_filer.py` + `nb-watcher` sidecar — dedup/auto-close filer
 

infra/images/notebooks.Dockerfile

@@ -26,8 +26,11 @@ WORKDIR /src
 RUN apt-get update \
     && apt-get install -y --no-install-recommends git ca-certificates python3 jq \
     && rm -rf /var/lib/apt/lists/* \
-    && corepack enable \
+    && corepack enable
-    && corepack prepare pnpm@latest --activate
+# No `corepack prepare pnpm@latest`: the corepack shim resolves the exact
+# pnpm version from marimo's package.json `packageManager` field at first
+# use, so the package manager can't float between builds (the lockfile fix
+# in apply-overlay.sh covers the dependency graph; this covers the tool).
 
 # Shallow clone marimo at the pinned tag.
 RUN git clone --depth 1 --branch ${MARIMO_VERSION} --filter=blob:none \

infra/marimo/nb-tests.toml

@@ -0,0 +1,16 @@
+# Notebook integration-test config — consumed by dev/scripts/nb_integration.py.
+#
+# [ci_smoke].notebooks: run in CI on every relevant push. Must execute clean
+# from a bare `uv sync --dev` checkout: no compose services, no ./data, no
+# GPU. (sample.py is excluded because it imports vega_datasets, which only
+# the notebooks-container venv carries.)
+#
+# [expected_failures]: notebook -> reason. Errors are still reported (and
+# auto-filed with dedup on scheduled runs) but don't fail the run.
+
+[ci_smoke]
+notebooks = ["_template.py", "sql_generator.py"]
+
+[expected_failures]
+"acodb_explorer.py" = "aco.duckdb single-writer lock until #508-514 lands"
+"gpu_test.py" = "requires GPU; nightly run inside the container may race other GPU users"

tests/dev/test_nb_integration.py

@@ -0,0 +1,112 @@
+"""Tests for dev/scripts/nb_integration.py — session snapshot parsing and
+pass/fail classification.
+
+The snapshot fixture mirrors the real shape marimo 0.23.13 writes to
+__marimo__/session/<nb>.py.json (verified empirically): cell errors appear
+as outputs with type "error" plus ename/evalue.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+_SCRIPT = Path(__file__).resolve().parents[2] / "dev" / "scripts" / "nb_integration.py"
+_spec = importlib.util.spec_from_file_location("_nb_integration", _SCRIPT)
+assert _spec and _spec.loader
+nbi = importlib.util.module_from_spec(_spec)
+sys.modules["_nb_integration"] = nbi
+_spec.loader.exec_module(nbi)
+
+
+SNAPSHOT_OK = {
+    "version": "1",
+    "metadata": {"marimo_version": "0.23.13"},
+    "cells": [
+        {
+            "id": "Hbol",
+            "outputs": [{"type": "data", "data": {"text/html": "<pre>2</pre>"}}],
+            "console": [],
+        }
+    ],
+}
+
+SNAPSHOT_ERR = {
+    "version": "1",
+    "metadata": {"marimo_version": "0.23.13"},
+    "cells": [
+        {
+            "id": "Hbol",
+            "outputs": [{"type": "data", "data": {"text/html": "<pre>2</pre>"}}],
+            "console": [],
+        },
+        {
+            "id": "MJUe",
+            "outputs": [
+                {
+                    "type": "error",
+                    "ename": "exception",
+                    "evalue": "intentional failure 2",
+                    "traceback": None,
+                }
+            ],
+            "console": [
+                {
+                    "type": "stream",
+                    "name": "stderr",
+                    "text": "<pre>ValueError: intentional failure 2</pre>",
+                    "mimetype": "application/vnd.marimo+traceback",
+                }
+            ],
+        },
+    ],
+}
+
+
+def _write(tmp_path: Path, name: str, snap: dict) -> Path:
+    p = tmp_path / name
+    p.write_text(json.dumps(snap))
+    return p
+
+
+def test_parse_snapshot_clean(tmp_path):
+    errors = nbi.parse_snapshot(_write(tmp_path, "ok.py.json", SNAPSHOT_OK))
+    assert errors == []
+
+
+def test_parse_snapshot_extracts_errors_with_console_detail(tmp_path):
+    errors = nbi.parse_snapshot(_write(tmp_path, "err.py.json", SNAPSHOT_ERR))
+    assert len(errors) == 1
+    e = errors[0]
+    assert e["cell"] == "MJUe"
+    assert e["ename"] == "exception"
+    assert e["evalue"] == "intentional failure 2"
+    assert "intentional failure" in e["detail"]
+
+
+def test_parse_snapshot_missing_file_reports_export_error(tmp_path):
+    errors = nbi.parse_snapshot(tmp_path / "never-written.py.json")
+    assert len(errors) == 1
+    assert errors[0]["ename"] == "snapshot-missing"
+
+
+def test_classify_pass_fail_and_expected():
+    expected = {"known_bad.py": "duckdb lock (#508)"}
+    assert nbi.classify("clean.py", [], expected) == "pass"
+    err = [{"cell": "x", "ename": "exception", "evalue": "y", "detail": ""}]
+    assert nbi.classify("clean.py", err, expected) == "fail"
+    assert nbi.classify("known_bad.py", err, expected) == "expected-fail"
+    # an expected-failure notebook that passes should surface as pass
+    assert nbi.classify("known_bad.py", [], expected) == "pass"
+
+
+def test_report_exit_code():
+    results = {
+        "a.py": {"status": "pass", "errors": []},
+        "b.py": {"status": "expected-fail", "errors": [{"ename": "e"}]},
+    }
+    assert nbi.exit_code(results) == 0
+    results["c.py"] = {"status": "fail", "errors": [{"ename": "e"}]}
+    assert nbi.exit_code(results) == 1

tests/dev/test_nb_issue_filer.py

@@ -0,0 +1,123 @@
+"""Tests for dev/scripts/nb_issue_filer.py — signature stability and
+dedup/auto-close decisions.
+
+The filer exists because api.diag.ci files one issue per failure and never
+closes them (23 duplicates accumulated for pkg-supply-chain alone). The
+invariants under test: identical failures collapse to one signature across
+runs, volatile details (paths, line numbers, addresses, counts) don't
+change the signature, and the decide/close logic never files a duplicate
+for an already-open signature.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+_SCRIPT = Path(__file__).resolve().parents[2] / "dev" / "scripts" / "nb_issue_filer.py"
+_spec = importlib.util.spec_from_file_location("_nb_issue_filer", _SCRIPT)
+assert _spec and _spec.loader
+filer = importlib.util.module_from_spec(_spec)
+sys.modules["_nb_issue_filer"] = filer
+_spec.loader.exec_module(filer)
+
+
+# ── signature ────────────────────────────────────────────────────
+
+
+def test_signature_stable_for_identical_error():
+    a = filer.signature("acodb_explorer.py", "exception", "division by zero")
+    b = filer.signature("acodb_explorer.py", "exception", "division by zero")
+    assert a == b
+
+
+def test_signature_differs_across_notebooks_and_errors():
+    base = filer.signature("a.py", "exception", "boom")
+    assert filer.signature("b.py", "exception", "boom") != base
+    assert filer.signature("a.py", "interruption", "boom") != base
+    assert filer.signature("a.py", "exception", "other") != base
+
+
+def test_signature_ignores_volatile_details():
+    a = filer.signature(
+        "nb.py",
+        "exception",
+        'IO Error: could not open "/tmp/marimo_2487488/cell_MJUe.py" at 0x7f3a2c1b'
+        " (attempt 3 of 5)",
+    )
+    b = filer.signature(
+        "nb.py",
+        "exception",
+        'IO Error: could not open "/tmp/marimo_9911223/cell_MJUe.py" at 0x559e00aa'
+        " (attempt 4 of 5)",
+    )
+    assert a == b
+
+
+def test_signature_is_short_hex():
+    sig = filer.signature("nb.py", "exception", "x")
+    assert len(sig) == 12
+    int(sig, 16)  # parses as hex
+
+
+# ── normalize ────────────────────────────────────────────────────
+
+
+def test_normalize_collapses_paths_hex_and_numbers():
+    s = filer.normalize_evalue(
+        "failed /home/kert/notebooks/x.py line 42 addr 0xDEADBEEF took 3.14s"
+    )
+    assert "0x" not in s.lower() or "0xN" in s
+    assert "/home/kert" not in s
+    assert "42" not in s
+    assert "3.14" not in s
+
+
+def test_normalize_keeps_error_identity():
+    s = filer.normalize_evalue("Could not remove file: No such file or directory")
+    assert "no such file or directory" in s.lower()
+
+
+# ── decide ───────────────────────────────────────────────────────
+
+
+def test_decide_creates_when_no_open_issue():
+    action = filer.decide(existing=None, last_comment_age_s=None, cooldown_s=3600)
+    assert action == "create"
+
+
+def test_decide_comments_after_cooldown():
+    action = filer.decide(
+        existing={"number": 7}, last_comment_age_s=7200, cooldown_s=3600
+    )
+    assert action == "comment"
+
+
+def test_decide_skips_within_cooldown():
+    action = filer.decide(
+        existing={"number": 7}, last_comment_age_s=60, cooldown_s=3600
+    )
+    assert action == "skip"
+
+
+# ── issue body marker round-trip ─────────────────────────────────
+
+
+def test_marker_embeds_and_extracts_signature():
+    sig = filer.signature("nb.py", "exception", "boom")
+    body = filer.build_body(
+        notebook="nb.py",
+        ename="exception",
+        evalue="boom",
+        source="nightly-integration",
+        detail="Traceback ...",
+        sig=sig,
+    )
+    assert filer.extract_sig(body) == sig
+    assert "nb.py" in body
+    assert "nightly-integration" in body
+
+
+def test_extract_sig_none_when_absent():
+    assert filer.extract_sig("no marker here") is None