fix: apply lossless push in PoolObjectNode too

PoolNode and PoolObjectNode each have their own push_one_out(). The previous
commit only reached PoolNode's copy, so ObjectNode-based pipelines — the common
case — still dropped on overflow with lossless enabled.

Verified end to end: a 2300-frame source now reports exactly 2300 frames at
every stage, where it previously delivered 774 to the second node.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.gitea/workflows/ci.yaml

@@ -0,0 +1,87 @@
+name: '🚦 CI'
+
+# Single orchestrator. This is the only workflow that triggers on push/PR.
+# It decides which reusable sub-workflows to run and in what order:
+#   changes ─┬─> docker (only if the Dockerfile/requirements changed) ─┬─> test
+#            │                                                          └─> docs
+# When the builder image is rebuilt it MUST finish (and push) before test/docs
+# run, so they validate against the fresh image.
+on:
+  push:
+    branches:
+      - master
+      - develop
+  pull_request:
+    branches:
+      - master
+      - develop
+  workflow_dispatch:
+
+jobs:
+  # Detect which parts of the repo changed in this push/PR.
+  changes:
+    runs-on: linux/amd64
+    # Runs in the builder image because the host has no Node, which the
+    # JS-based checkout/paths-filter actions require.
+    container:
+      image: gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
+    outputs:
+      dockerfile: ${{ steps.filter.outputs.dockerfile }}
+      code: ${{ steps.filter.outputs.code }}
+      docs: ${{ steps.filter.outputs.docs }}
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Detect changed paths
+        id: filter
+        uses: dorny/paths-filter@v3
+        with:
+          filters: |
+            dockerfile:
+              - 'Dockerfile.builder'
+              - 'docs/requirements.txt'
+            docs:
+              - 'docs/**'
+              - 'mkdocs.yml'
+              - 'examples/**/*.cpp'
+            code:
+              - 'src/**'
+              - 'include/**'
+              - 'tests/**'
+              - 'examples/**'
+              - 'python/**'
+              - 'CMakeLists.txt'
+              - '**/*.cpp'
+              - '**/*.hpp'
+              - '**/*.h'
+
+  # Rebuild the builder image first, but only when it actually changed.
+  # On pull requests we build to validate the Dockerfile but do not push.
+  docker:
+    needs: changes
+    if: ${{ needs.changes.outputs.dockerfile == 'true' }}
+    uses: ./.gitea/workflows/docker.yaml
+    with:
+      # Explicit string, not a boolean expression (act_runner mangles bools).
+      push: ${{ github.event_name == 'pull_request' && 'false' || 'true' }}
+
+  # Runs after docker (if docker ran). A skipped docker job is fine; a failed
+  # one blocks this via !failure(). Re-run tests when code OR the image changed.
+  test:
+    needs: [changes, docker]
+    if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }}
+    uses: ./.gitea/workflows/test.yaml
+
+  # ThreadSanitizer run for the lock-free Channel<T>. Same trigger conditions as
+  # test (code or image changed); runs in parallel with test.
+  tsan:
+    needs: [changes, docker]
+    if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }}
+    uses: ./.gitea/workflows/tsan.yaml
+
+  docs:
+    needs: [changes, docker]
+    if: ${{ !failure() && !cancelled() && github.ref == 'refs/heads/master' && (needs.changes.outputs.docs == 'true' || needs.changes.outputs.dockerfile == 'true') }}
+    uses: ./.gitea/workflows/docs.yaml
+    secrets: inherit

.gitea/workflows/docker.yaml

@@ -0,0 +1,57 @@
+name: '🐳 Builder Image'
+
+# Reusable workflow: builds (and optionally pushes) the kpnpp-builder image.
+# It is called by ci.yaml only when Dockerfile.builder or docs/requirements.txt
+# change. It runs on the host runner (NOT inside the builder container) because
+# it needs the Docker CLI/daemon.
+# Note: `push` is a STRING ("true"/"false"), not a boolean. Gitea's act_runner
+# mangles boolean inputs passed from an expression (they arrive as false), so we
+# pass an explicit string and compare with == 'true' below.
+on:
+  workflow_call:
+    inputs:
+      push:
+        description: 'Push the built image to the registry ("true"/"false")'
+        type: string
+        default: 'true'
+  workflow_dispatch:
+    inputs:
+      push:
+        description: 'Push the built image to the registry ("true"/"false")'
+        type: string
+        default: 'true'
+
+jobs:
+  build:
+    runs-on: linux/amd64
+    steps:
+      # This job runs on the host (not in a container) so it can reach the
+      # host Docker daemon and reuse the cached registry credentials. The host
+      # has no Node, so the JS-based actions/checkout can't run here; do a
+      # minimal shallow fetch of this commit with plain git instead.
+      - name: Checkout repository
+        run: |
+          git init -q .
+          git remote add origin "${{ github.server_url }}/${{ github.repository }}.git"
+          git -c http.extraheader="AUTHORIZATION: basic $(printf '%s' '${{ github.actor }}:${{ github.token }}' | base64 -w0)" \
+            fetch --depth 1 origin "${{ github.sha }}"
+          git checkout -q FETCH_HEAD
+
+      # No docker login step: the host runner was authenticated to
+      # gitea.tourolle.paris with `docker login` during setup, so its cached
+      # credentials in ~/.docker/config.json cover the push below.
+      - name: Build builder image
+        # Context is the repo root because Dockerfile.builder COPYs
+        # docs/requirements.txt during the build.
+        run: |
+          docker build \
+            -f Dockerfile.builder \
+            -t gitea.tourolle.paris/dtourolle/kpnpp-builder:latest \
+            -t gitea.tourolle.paris/dtourolle/kpnpp-builder:${{ github.sha }} \
+            .
+
+      - name: Push builder image
+        if: ${{ inputs.push == 'true' }}
+        run: |
+          docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
+          docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:${{ github.sha }}

.gitea/workflows/docs.yaml

@@ -0,0 +1,34 @@
+name: '📚 Docs'
+
+# Triggering and path filtering are owned by ci.yaml (the orchestrator), which
+# calls this as a reusable workflow. workflow_dispatch is kept for manual runs.
+on:
+  workflow_call:
+  workflow_dispatch:
+
+jobs:
+  deploy:
+    runs-on: linux/amd64
+    container:
+      image: gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0   # full history needed for mkdocs gh-deploy
+
+      - name: Configure git identity
+        run: |
+          git config user.name  "Gitea Actions"
+          git config user.email "actions@gitea.tourolle.paris"
+
+      - name: Build and deploy to gitea-pages branch
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          mkdocs gh-deploy \
+            --force \
+            --remote-branch gitea-pages \
+            --remote-name origin \
+            --message "docs: deploy from ${{ github.sha }}"

.gitea/workflows/test.yaml

@@ -1,18 +1,9 @@
 name: '🧪 Test'
 
+# Triggering and path filtering are owned by ci.yaml (the orchestrator), which
+# calls this as a reusable workflow. workflow_dispatch is kept for manual runs.
 on:
-  push:
-    branches:
-      - master
-      - develop
-    paths-ignore:
-      - '**/*.md'
-  pull_request:
-    branches:
-      - master
-      - develop
-    paths-ignore:
-      - '**/*.md'
+  workflow_call:
   workflow_dispatch:
 
 jobs:
@@ -41,7 +32,7 @@ jobs:
             -G Ninja \
             -DCMAKE_BUILD_TYPE=Debug \
             -DKPN_BUILD_TESTS=ON \
-            -DKPN_BUILD_EXAMPLES=OFF \
+            -DKPN_BUILD_EXAMPLES=ON \
             -DKPN_BUILD_PYTHON=ON \
             -DFETCHCONTENT_BASE_DIR=$HOME/.cmake/fetchcontent
 
@@ -49,18 +40,26 @@ jobs:
         working-directory: test-${{ github.run_id }}
         run: cmake --build build --parallel
 
-      - name: Run tests
+      - name: Run unit tests
         working-directory: test-${{ github.run_id }}
         run: |
           cd build
-          ctest --output-on-failure --output-junit test-results.xml
+          ctest --output-on-failure --output-junit test-results.xml --label-exclude examples
+
+      - name: Run example smoke tests
+        working-directory: test-${{ github.run_id }}
+        run: |
+          cd build
+          ctest --output-on-failure --output-junit example-results.xml -L examples
 
       - name: Upload test results
         if: always()
         uses: actions/upload-artifact@v3
         with:
           name: test-results
-          path: test-${{ github.run_id }}/build/test-results.xml
+          path: |
+            test-${{ github.run_id }}/build/test-results.xml
+            test-${{ github.run_id }}/build/example-results.xml
           retention-days: 7
 
       - name: Cleanup

.gitea/workflows/tsan.yaml

@@ -0,0 +1,77 @@
+name: '🧵 ThreadSanitizer'
+
+# Reusable workflow: builds the channel stress suite with ThreadSanitizer and
+# runs it. This is the dynamic half of verifying the lock-free SPSC Channel<T>
+# (the static half is the CDSChecker model-check harness in verify/).
+#
+# Triggering and path filtering are owned by ci.yaml (the orchestrator), which
+# calls this only when code changed. workflow_dispatch is kept for manual runs.
+#
+# Runs in the prebuilt builder image (gcc:14), which already ships libtsan — no
+# package installs at job time.
+on:
+  workflow_call:
+  workflow_dispatch:
+
+jobs:
+  tsan:
+    runs-on: linux/amd64
+    container:
+      image: gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
+      # This runner is Docker nested in an unprivileged LXC container, whose
+      # kernel randomizes mmap addresses beyond the range TSan's fixed shadow
+      # mapping expects, so TSan aborts at init with "unexpected memory
+      # mapping". The fix is to disable ASLR per-process with `setarch -R`
+      # (below), which needs the personality(2) syscall that Docker's default
+      # seccomp profile blocks. seccomp=unconfined permits it. Verified on the
+      # runner: setarch -R alone gets EPERM, seccomp alone still aborts, both
+      # together run clean. Scoped to this job, which runs only our own tests.
+      options: --security-opt seccomp=unconfined
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          path: tsan-${{ github.run_id }}
+
+      - name: Cache FetchContent dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.cmake/fetchcontent
+          key: cmake-fetchcontent-${{ hashFiles('**/CMakeLists.txt') }}
+          restore-keys: cmake-fetchcontent-
+
+      - name: Configure (TSan)
+        working-directory: tsan-${{ github.run_id }}
+        run: |
+          cmake -S . -B build \
+            -G Ninja \
+            -DCMAKE_BUILD_TYPE=Debug \
+            -DKPN_SANITIZER=thread \
+            -DKPN_BUILD_TESTS=ON \
+            -DKPN_BUILD_EXAMPLES=OFF \
+            -DKPN_BUILD_PYTHON=OFF \
+            -DFETCHCONTENT_BASE_DIR=$HOME/.cmake/fetchcontent
+
+      - name: Build (TSan)
+        working-directory: tsan-${{ github.run_id }}
+        run: cmake --build build --parallel --target kpn_tests kpn_tests_stress
+
+      - name: Run stress suite under TSan
+        working-directory: tsan-${{ github.run_id }}
+        # halt_on_error=1 makes the first detected race fail the job; the report
+        # (with both stacks) is printed to the log. second_deadlock_stack gives
+        # the full picture for lock-order issues.
+        env:
+          TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1"
+        # setarch -R disables ASLR for this process; see the container comment.
+        run: setarch -R ./build/tests/kpn_tests_stress
+
+      - name: Run unit tests under TSan
+        working-directory: tsan-${{ github.run_id }}
+        env:
+          TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1"
+        run: setarch -R ./build/tests/kpn_tests
+
+      - name: Cleanup
+        if: always()
+        run: rm -rf tsan-${{ github.run_id }}

.gitignore

@@ -1,6 +1,8 @@
 # Build output
 build/
+build_test/
 build_debug/
+site/
 # Python
 __pycache__/
 *.py[cod]

CMakeLists.txt

@@ -10,6 +10,30 @@ option(KPN_BUILD_PYTHON   "Build Python bindings (requires nanobind)" ON)
 option(KPN_BUILD_EXAMPLES "Build examples"                            ON)
 option(KPN_WEB_DEBUG      "Enable web debug UI (cpp-httplib)"         OFF)
 
+# Sanitizer build. Empty = off. Accepts "thread", "address", "undefined",
+# or a combination like "address,undefined". Applied to all kpn targets via
+# the kpn_sanitizer_flags() helper below.
+#
+# The lock-free SPSC Channel<T> (include/kpn/channel.hpp) has hand-reasoned
+# acquire/release ordering; -DKPN_SANITIZER=thread + the channel stress test
+# (tests/test_channel_stress.cpp) is the dynamic half of verifying it. The
+# static half is the CDSChecker model-check harness (see verify/).
+set(KPN_SANITIZER "" CACHE STRING
+    "Build with sanitizer: thread | address | undefined | <combo> (empty = off)")
+
+# Translate KPN_SANITIZER into compile/link flags. No-op when empty.
+function(kpn_sanitizer_flags out_var)
+    if(KPN_SANITIZER)
+        set(${out_var}
+            -fsanitize=${KPN_SANITIZER}
+            -fno-omit-frame-pointer
+            -g
+            PARENT_SCOPE)
+    else()
+        set(${out_var} "" PARENT_SCOPE)
+    endif()
+endfunction()
+
 # ── Core library (header-only) ────────────────────────────────────────────────
 add_library(kpn INTERFACE)
 target_include_directories(kpn INTERFACE

Dockerfile.builder

@@ -1,4 +1,4 @@
-# KPN++ Builder Image
+# KPN++ Builder Image  (CI: pipeline trigger v2)
 # Pre-built image with GCC, CMake, Ninja, and Python dev headers for building and testing KPN++
 # Build: docker build -f Dockerfile.builder -t gitea.tourolle.paris/dtourolle/kpnpp-builder:latest .
 # Push:  docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
@@ -16,4 +16,12 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     nodejs \
     && rm -rf /var/lib/apt/lists/*
 
+# Pre-install MkDocs dependencies so the docs workflow does not need to pip
+# install at runtime. --break-system-packages is required because the Debian
+# base marks the environment as externally managed (PEP 668); this is safe in
+# a dedicated container image.
+COPY docs/requirements.txt /tmp/docs-requirements.txt
+RUN pip install --no-cache-dir --break-system-packages -r /tmp/docs-requirements.txt \
+    && rm /tmp/docs-requirements.txt
+
 WORKDIR /src

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Duncan Tourolle
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -2,6 +2,8 @@
 
 A C++20 Kahn Process Network (KPN) library. Each node wraps a function and runs in its own thread, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind.
 
+📖 **[Documentation](https://pages.tourolle.paris/dtourolle/kpn/)**
+
 ---
 
 ## Requirements
@@ -398,8 +400,8 @@ Violating the second rule deadlocks.
 | `04_storage_policy` | `channel_storage_policy` default and specialisation |
 | `05_error_handling` | `ChannelOverflowError`, `ErrorHandler` |
 | `06_watchdog` | Watchdog interval, stall detection |
-| `07_python_network` | PyNetwork, pure Python node *(pending)* |
-| `08_python_subport` | `net.read`, `net.write`, sub-port tap *(pending)* |
+| `07_python_network` | PyNetwork with a pure-Python node between a C++ source and sink |
+| `08_python_subport` | Drive a Python node from Python via `net.write`/`net.read` sub-port taps |
 | `09_opencv_cellshade` | Real-time cell-shading on webcam/pattern; requires OpenCV ≥ 4 |
 
 Run the cell-shading example:
@@ -467,7 +469,7 @@ not data.
 
 Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | 1.7 | **1.4** |
 | chain depth-4 | 2.5 | **2.2** |
@@ -479,7 +481,7 @@ Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
 Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | **2.1** | 3.5 |
 | chain depth-4 | **4.3** | 5.2 |
@@ -489,7 +491,7 @@ Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 | wide fanout-4 | 3.4 | **1.9** |
 | diamond (2×2) | **4.1** | 6.1 |
 
-KPN private pools beat TBB for every chain and diamond topology at 100 µs/node, and
+KPN++ pools beat TBB for every chain and diamond topology at 100 µs/node, and
 match TBB within ~20% at 10 µs/node for shallow chains. TBB retains an edge on wide
 fanout (serial dispatch loop vs. work-stealing pool) and at extreme oversubscription
 depths (chain-32 at 10 µs). The remaining gap at light work is the cost of
@@ -578,3 +580,33 @@ examples/
 scripts/
   render_readme.py   — regenerates README.md from README.md.in
 ```
+
+---
+
+## Contributing
+
+Contributions are welcome. This project is hosted on a self-hosted Gitea
+instance that accepts sign-in and registration with a GitHub account, so you
+can log in with your existing GitHub identity to open issues and pull requests.
+
+If you change any code that appears in a README snippet, edit `README.md.in`
+(the template) rather than `README.md` directly, then regenerate:
+
+```bash
+cmake --build build --target readme   # or: python scripts/render_readme.py
+```
+
+---
+
+## Acknowledgments
+
+AI tooling was used heavily throughout the development of this project,
+including the design, implementation, tests, and documentation. All output
+has been reviewed, but please keep this in mind when reading or building on the
+code.
+
+---
+
+## License
+
+Released under the [MIT License](LICENSE). Copyright (c) 2026 Duncan Tourolle.

README.md.in

@@ -2,6 +2,8 @@
 
 A C++20 Kahn Process Network (KPN) library. Each node wraps a function and runs in its own thread, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind.
 
+📖 **[Documentation](https://pages.tourolle.paris/dtourolle/kpn/)**
+
 ---
 
 ## Requirements
@@ -220,8 +222,8 @@ Violating the second rule deadlocks.
 | `04_storage_policy` | `channel_storage_policy` default and specialisation |
 | `05_error_handling` | `ChannelOverflowError`, `ErrorHandler` |
 | `06_watchdog` | Watchdog interval, stall detection |
-| `07_python_network` | PyNetwork, pure Python node *(pending)* |
-| `08_python_subport` | `net.read`, `net.write`, sub-port tap *(pending)* |
+| `07_python_network` | PyNetwork with a pure-Python node between a C++ source and sink |
+| `08_python_subport` | Drive a Python node from Python via `net.write`/`net.read` sub-port taps |
 | `09_opencv_cellshade` | Real-time cell-shading on webcam/pattern; requires OpenCV ≥ 4 |
 
 Run the cell-shading example:
@@ -289,7 +291,7 @@ not data.
 
 Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | 1.7 | **1.4** |
 | chain depth-4 | 2.5 | **2.2** |
@@ -301,7 +303,7 @@ Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
 Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | **2.1** | 3.5 |
 | chain depth-4 | **4.3** | 5.2 |
@@ -311,7 +313,7 @@ Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 | wide fanout-4 | 3.4 | **1.9** |
 | diamond (2×2) | **4.1** | 6.1 |
 
-KPN private pools beat TBB for every chain and diamond topology at 100 µs/node, and
+KPN++ pools beat TBB for every chain and diamond topology at 100 µs/node, and
 match TBB within ~20% at 10 µs/node for shallow chains. TBB retains an edge on wide
 fanout (serial dispatch loop vs. work-stealing pool) and at extreme oversubscription
 depths (chain-32 at 10 µs). The remaining gap at light work is the cost of
@@ -400,3 +402,33 @@ examples/
 scripts/
   render_readme.py   — regenerates README.md from README.md.in
 ```
+
+---
+
+## Contributing
+
+Contributions are welcome. This project is hosted on a self-hosted Gitea
+instance that accepts sign-in and registration with a GitHub account, so you
+can log in with your existing GitHub identity to open issues and pull requests.
+
+If you change any code that appears in a README snippet, edit `README.md.in`
+(the template) rather than `README.md` directly, then regenerate:
+
+```bash
+cmake --build build --target readme   # or: python scripts/render_readme.py
+```
+
+---
+
+## Acknowledgments
+
+AI tooling was used heavily throughout the development of this project,
+including the design, implementation, tests, and documentation. All output
+has been reviewed, but please keep this in mind when reading or building on the
+code.
+
+---
+
+## License
+
+Released under the [MIT License](LICENSE). Copyright (c) 2026 Duncan Tourolle.

docs/channels.md

@@ -0,0 +1,55 @@
+# Channels
+
+A `Channel<T>` is a lock-free SPSC (single-producer, single-consumer) ring buffer with atomic wait/notify.
+
+## Semantics
+
+- **Bounded**: fixed capacity set at construction. Default is 5 items.
+- **Backpressure**: when full, `push()` throws `ChannelOverflowError` immediately — no blocking, no spin.
+- **Blocking consumer**: `pop()` blocks until an item is available or the channel is disabled.
+- **Disable**: `channel.disable()` stops accepting pushes and unblocks any waiting `pop()` with `ChannelClosedError`.
+
+## Storage policy
+
+Small trivially-copyable types (≤ 8 bytes) are stored by value. Larger types are heap-allocated and passed via `shared_ptr<const T>` — one allocation per push, zero-copy fan-out:
+
+```cpp
+--8<-- "examples/04_storage_policy/main.cpp:storage_policy_spec"
+```
+
+Specialize `kpn::ChannelDataSize<T>` for accurate bandwidth reporting on heap-owning types:
+
+```cpp
+template<>
+struct kpn::ChannelDataSize<cv::Mat> {
+    static std::size_t bytes(const cv::Mat& m) { return m.total() * m.elemSize(); }
+};
+```
+
+## Named ports
+
+`in<"name">` and `out<"name">` tag nodes for readable wiring:
+
+```cpp
+--8<-- "examples/02_named_ports/main.cpp:named_port_creation"
+```
+
+Named ports are checked at compile time — a typo in a port name is a compile error.
+
+## Capacity tuning
+
+Set capacity per node at construction:
+
+```cpp
+auto node = make_node<my_func>(/*capacity=*/20);
+```
+
+Capacity is rounded up internally to the next power of two. Monitor fill levels via diagnostics to tune for your workload — a too-small capacity causes overflows; a too-large one wastes memory and hides producer/consumer speed mismatches.
+
+## Spin count
+
+`Channel` spins for up to ~4 µs (200 `pause` hints at ~20 ns each on x86) before sleeping on a futex. Set to 0 for power-constrained or predominantly-idle pipelines:
+
+```cpp
+Channel<int> ch(/*capacity=*/5, /*spin_count=*/0);
+```

docs/error-handling.md

@@ -0,0 +1,84 @@
+# Error Handling & Events
+
+KPN++ provides three complementary layers for observing and reacting to failures.
+
+---
+
+## 1. Per-node error handler
+
+Called when a node's function throws an unhandled exception. Return `true` to skip the failed invocation and keep running; `false` to stop the node.
+
+```cpp
+--8<-- "examples/15_node_error_handler/main.cpp:error_handler"
+```
+
+When a node stops (either from `false` return or no handler installed), it:
+
+1. Disables its **input** channels — upstream stops pushing into dead queues.
+2. Disables its **output** channels — downstream nodes receive `ChannelClosedError` on their next pop, propagating the shutdown naturally through the graph.
+
+---
+
+## 2. Per-node overflow callback
+
+Fired with a timestamp each time an output push is dropped because the channel is full. The node name is known at registration so it is not included — keeping the callback zero-overhead when unused.
+
+```cpp
+--8<-- "examples/16_event_callbacks/main.cpp:per_node_callback"
+```
+
+!!! note
+    The callback is purely informational — the node always continues after an overflow. To stop the node on overflow, call `node.stop()` from inside the callback.
+
+A matching `set_closed_callback()` fires (also with just a timestamp) when the node stops due to a closed upstream channel:
+
+```cpp
+node.set_closed_callback([](std::chrono::steady_clock::time_point ts) {
+    std::cerr << "node stopped at t=" << ts.time_since_epoch().count() << '\n';
+});
+```
+
+Each node holds two callback slots per event type — one user-set (registered above) and one injected by the network (see below). Both fire independently.
+
+---
+
+## 3. Network-level event handler
+
+One callback for the whole network. Receives the node name (captured in a closure by the network at `build()` / `start()`), a `NodeEvent`, and a timestamp:
+
+```cpp
+--8<-- "examples/16_event_callbacks/main.cpp:network_event_handler"
+```
+
+`NodeEvent` values:
+
+| Value | Meaning |
+|---|---|
+| `NodeEvent::Overflow` | An output push was dropped (channel full) |
+| `NodeEvent::Closed` | The node stopped (crash or upstream close cascade) |
+
+The network handler and any per-node callbacks are **independent** — both fire when set.
+
+---
+
+## Complete example
+
+`examples/16_event_callbacks/main.cpp` shows a fast producer overflowing a slow consumer, with both a per-node overflow callback and a network-level event handler active simultaneously.
+
+Node functions:
+
+```cpp
+--8<-- "examples/16_event_callbacks/main.cpp:node_fns"
+```
+
+Per-node overflow callback:
+
+```cpp
+--8<-- "examples/16_event_callbacks/main.cpp:per_node_callback"
+```
+
+Network-level event handler:
+
+```cpp
+--8<-- "examples/16_event_callbacks/main.cpp:network_event_handler"
+```

docs/examples.md

@@ -0,0 +1,40 @@
+# Examples
+
+All C++ examples are built by default and registered as CTest smoke tests. Run them all with:
+
+```bash
+ctest --test-dir build -L examples
+```
+
+## Index
+
+| Example | What it shows |
+|---|---|
+| `01_hello_pipeline` | Linear pipeline, index-based port wiring |
+| `02_named_ports` | `in<>`/`out<>` name tags, named port access |
+| `03_multi_output` | Tuple-returning node, per-element routing |
+| `04_storage_policy` | `channel_storage_policy` specialisation |
+| `05_error_handling` | Diagnostics handler, overflow channel stats |
+| `06_watchdog` | Watchdog interval, stall detection |
+| `10_static_hello_pipeline` | `StaticNetwork` + `make_network()` |
+| `11_static_fanout` | `StaticNetwork` with `FanoutNode` |
+| `15_node_error_handler` | `set_error_handler()` — skip or stop on exception |
+| `16_event_callbacks` | `set_overflow_callback()`, `set_event_handler()` |
+
+## OpenCV examples (optional)
+
+Built only when OpenCV ≥ 4 is found:
+
+| Example | What it shows |
+|---|---|
+| `09_opencv_cellshade` | Real-time cell-shading on webcam; `MainThreadNode` for display |
+| `12_static_cellshade` | Same pipeline as a `StaticNetwork` |
+| `13_debug_cellshade` | Web debug UI overlay on the cell-shading pipeline |
+
+Run the cell-shading example:
+
+```bash
+./build/examples/09_opencv_cellshade
+# Press 'q' or close the window to stop.
+# Falls back to an animated synthetic pattern if no webcam is found.
+```

docs/fanout.md

@@ -0,0 +1,44 @@
+# Fan-out & Routing
+
+## FanoutNode
+
+Reads one item and pushes a copy to each of N output channels. All downstream nodes receive every item.
+
+```cpp
+auto fan = make_fanout<Image, 2>(/*capacity=*/8);
+
+net.connect("src",  src.output<0>(),  "fan",   fan.input<0>())
+   .connect("fan",  fan.output<0>(),  "nodeA", nodeA.input<0>())
+   .connect("fan",  fan.output<1>(),  "nodeB", nodeB.input<0>());
+```
+
+If one downstream channel overflows, that output drops the item independently — the other outputs are unaffected.
+
+See `examples/11_static_fanout`.
+
+## RouterNode
+
+Reads one item and pushes it to exactly one of N outputs, chosen by a selector function:
+
+```cpp
+auto router = make_router<Frame, 3>(
+    [](const Frame& f) -> std::size_t { return f.stream_id % 3; });
+
+net.connect("src",    src.output<0>(),     "router", router.input<0>())
+   .connect("router", router.output<0>(),  "nodeA",  nodeA.input<0>())
+   .connect("router", router.output<1>(),  "nodeB",  nodeB.input<0>())
+   .connect("router", router.output<2>(),  "nodeC",  nodeC.input<0>());
+```
+
+If the selector returns `>= N` the item is silently dropped.
+
+## FilterNode
+
+Reads one item and passes it downstream only when a predicate returns `true`:
+
+```cpp
+auto filt = make_filter<Frame>([](const Frame& f) { return f.valid; });
+
+net.connect("src",  src.output<0>(),  "filt", filt.input<0>())
+   .connect("filt", filt.output<0>(), "dst",  dst.input<0>());
+```

docs/getting-started.md

@@ -0,0 +1,76 @@
+# Getting Started
+
+## Requirements
+
+| Dependency | Version | Notes |
+|---|---|---|
+| CMake | ≥ 3.21 | |
+| C++ compiler | GCC ≥ 11, Clang ≥ 13 | C++20 required |
+| nanobind | ≥ 2.1 | auto-fetched; Python ≥ 3.8 |
+| Catch2 | v3 | auto-fetched for tests |
+| OpenCV | ≥ 4 | optional; only for examples 09/12/13 |
+
+## Build
+
+```bash
+cmake -B build                          # core + tests + C++ examples
+cmake --build build --parallel
+ctest --test-dir build                  # run all tests including example smoke tests
+```
+
+Enable Python bindings:
+
+```bash
+cmake -B build -DKPN_BUILD_PYTHON=ON
+cmake --build build --parallel
+```
+
+Skip examples:
+
+```bash
+cmake -B build -DKPN_BUILD_EXAMPLES=OFF
+```
+
+## Your first pipeline
+
+Three functions — source, transform, sink — wired into a `Network`:
+
+```cpp
+--8<-- "examples/01_hello_pipeline/main.cpp:basic_node_fns"
+```
+
+Create nodes, connect them, build and run:
+
+```cpp
+--8<-- "examples/01_hello_pipeline/main.cpp:network_build"
+```
+
+That's it. Types are inferred from function signatures. The channel between `src` and `dbl` carries `int`; the channel between `dbl` and `prn` also carries `int`. A type mismatch is a compile error.
+
+## Named ports
+
+For nodes with multiple inputs or outputs, name the ports for clarity:
+
+```cpp
+--8<-- "examples/02_named_ports/main.cpp:named_port_creation"
+```
+
+Wire by name instead of index:
+
+```cpp
+--8<-- "examples/02_named_ports/main.cpp:named_port_network"
+```
+
+## Multi-output nodes
+
+Return a `std::tuple` to fan out to multiple downstream nodes:
+
+```cpp
+--8<-- "examples/03_multi_output/main.cpp:multi_output_fn"
+```
+
+Wire each tuple element to its own downstream node:
+
+```cpp
+--8<-- "examples/03_multi_output/main.cpp:fanout_network"
+```

docs/index.md

@@ -0,0 +1,39 @@
+# KPN++
+
+A C++20 [Kahn Process Network](https://en.wikipedia.org/wiki/Kahn_process_networks) library. Each node wraps a plain function and runs concurrently, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind.
+
+---
+
+## Why KPN++?
+
+- **Zero boilerplate** — wrap any callable as a node; types flow automatically from the function signature
+- **Bounded channels** — backpressure is structural, not bolted on
+- **Observable** — per-node and network-level callbacks for overflow and stop events; diagnostics snapshots; optional web UI
+- **Composable** — `Network` for runtime wiring, `StaticNetwork` for compile-time topology with zero overhead
+
+---
+
+## Quick example
+
+```cpp
+#include <kpn/kpn.hpp>
+using namespace kpn;
+
+--8<-- "examples/01_hello_pipeline/main.cpp:basic_node_fns"
+
+int main() {
+--8<-- "examples/01_hello_pipeline/main.cpp:network_build"
+}
+```
+
+---
+
+## Install & build
+
+```bash
+cmake -B build
+cmake --build build --parallel
+ctest --test-dir build          # unit tests + example smoke tests
+```
+
+See [Getting Started](getting-started.md) for full build options.

docs/network.md

@@ -0,0 +1,67 @@
+# Networks
+
+A `Network` wires nodes together at runtime using a builder chain.
+
+## Building a network
+
+```cpp
+--8<-- "examples/01_hello_pipeline/main.cpp:network_build"
+```
+
+The builder chain:
+
+| Method | Purpose |
+|---|---|
+| `.add(name, node)` | Register a node; assigns its name |
+| `.connect(src, port, dst, port)` | Wire one output port to one input port |
+| `.build()` | Compute topological order; inject network callbacks |
+| `.start()` | Start nodes in topological order |
+| `.stop()` | Stop all nodes immediately |
+| `.shutdown()` | Graceful drain: stop sources first, wait for channels to empty, then stop downstream |
+
+## Port access
+
+Ports are accessed by index or by name:
+
+```cpp
+// By index
+net.connect("src", src.output<0>(), "dst", dst.input<0>());
+
+// By name (requires named ports)
+--8<-- "examples/02_named_ports/main.cpp:named_port_network"
+```
+
+## Diagnostics
+
+Install a diagnostics handler to receive periodic snapshots of every node and channel:
+
+```cpp
+--8<-- "examples/05_error_handling/main.cpp:diagnostics_handler"
+```
+
+Or print a full report at any time:
+
+```cpp
+net.print_diagnostics();   // writes to stderr by default
+net.print_diagnostics(std::cout);
+```
+
+## Network-level event handler
+
+Observe overflow and node-stop events across the entire network in one place:
+
+```cpp
+--8<-- "examples/16_event_callbacks/main.cpp:network_event_handler"
+```
+
+`NodeEvent` is either `NodeEvent::Overflow` (item dropped on full channel) or `NodeEvent::Closed` (node stopped due to crash or closed upstream channel). See [Error Handling & Events](error-handling.md).
+
+## Shutdown
+
+`net.stop()` halts immediately — all nodes stop in reverse topological order.
+
+`net.shutdown()` drains gracefully: source nodes stop first; their output channels are polled until empty; then the next layer stops, and so on. This ensures no items are lost if downstream nodes are still consuming.
+
+## StaticNetwork
+
+For zero-overhead compile-time topology, see [Static Networks](static-network.md).

docs/nodes.md

@@ -0,0 +1,81 @@
+# Nodes
+
+A node wraps any callable. Its input types are inferred from the function's parameter list; its output types from the return type.
+
+## Node types
+
+| Type | Thread model | Use case |
+|---|---|---|
+| `Node<Func>` | Dedicated thread per node | Default — simplest, most isolated |
+| `PoolNode<Func>` | Shared `ThreadPool` | Many nodes, resource-bounded execution |
+| `InterruptNode<Func>` | Event-driven, no thread | Camera frame ready, timer tick, socket |
+| `FanoutNode<T, N>` | Dedicated thread | Broadcast one item to N outputs |
+| `RouterNode<T, N>` | Dedicated thread | Route one item to one of N outputs |
+| `FilterNode<T>` | Dedicated thread | Pass items matching a predicate |
+
+## Creating nodes
+
+All node types are created via factory functions that infer types from the callable:
+
+```cpp
+// Free function — simplest case
+auto node = make_node<my_func>();
+
+// Stateful functor (operator() is the function)
+MyProcessor proc;
+auto node = make_node(proc);
+
+// Pool node — shares a ThreadPool with other nodes
+auto pool = std::make_shared<ThreadPool>(4);
+auto node = make_pool_node<my_func>(pool);
+
+// Interrupt node — triggered externally
+auto sched = std::make_shared<ThreadPool>(2);
+auto node  = make_interrupt_node<produce_frame>(sched, out<"frame">{});
+camera_sdk.on_frame_ready(node.get_trigger());
+```
+
+## Channel capacity
+
+Each node's input FIFO has a configurable capacity (default 5):
+
+```cpp
+auto node = make_node<my_func>(/*capacity=*/20);
+auto node = make_pool_node<my_func>(pool, /*capacity=*/20);
+```
+
+When an upstream push would exceed capacity, `ChannelOverflowError` is thrown and the item is dropped. See [Error Handling & Events](error-handling.md) to observe and react to this.
+
+## Source nodes
+
+A node with no inputs is a source. It self-submits immediately on `start()` and re-submits after each execution:
+
+```cpp
+static int produce() {
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    return ++counter;
+}
+auto src = make_node<produce>();
+```
+
+!!! tip
+    Source nodes must sleep or yield to avoid overflowing their output channel. The channel capacity provides the only bound.
+
+## Sink nodes
+
+A node with a `void` return is a sink — it consumes items without producing output:
+
+```cpp
+static void print_it(int x) { std::cout << x << '\n'; }
+auto snk = make_node<print_it>();
+```
+
+## Error handler
+
+When a node's function throws an unhandled exception, the default behaviour is to stop the node (disabling its channels so the shutdown cascades downstream). Install a handler to override:
+
+```cpp
+--8<-- "examples/15_node_error_handler/main.cpp:error_handler"
+```
+
+See [Error Handling & Events](error-handling.md) for the full picture.

docs/requirements.txt

@@ -0,0 +1,3 @@
+mkdocs>=1.5
+mkdocs-material>=9.5
+pymdown-extensions>=10.0

docs/shared-resource.md

@@ -0,0 +1,42 @@
+# Shared Resources
+
+`SharedResource<T>` arbitrates exclusive access to a resource (ONNX session, CUDA stream, serial port) across multiple nodes using a priority-based waiter queue with starvation prevention.
+
+## Usage
+
+```cpp
+#include <kpn/shared_resource.hpp>
+using namespace kpn;
+
+SharedResource<OnnxSession> model(session_args...);
+
+static cv::Mat run_inference(cv::Mat frame) {
+    // Acquires the model; releases automatically on scope exit.
+    auto guard = model.acquire_balanced(in_channel, out_channel);
+    return guard->Run(frame);
+}
+```
+
+## Acquire modes
+
+| Method | Priority |
+|---|---|
+| `acquire()` | Equal (fair FIFO) |
+| `acquire(fn)` | Custom — `fn()` returns `float` in `[0, 1]` |
+| `acquire_balanced(in_ch, out_ch)` | `input_fill × output_headroom` — highest urgency wins |
+
+`acquire_balanced` favours nodes with full input queues and empty output queues — the node that has the most work to do and nowhere to stall wins the resource next.
+
+## Starvation prevention
+
+Each waiter's effective score grows with elapsed wait time (`0.05` per second by default), ensuring a low-priority node eventually gets served regardless of how frequently higher-priority nodes compete.
+
+## Diagnostics
+
+Register with the network for snapshot reporting:
+
+```cpp
+net.register_resource("model", &model);
+```
+
+The diagnostics table then shows acquisition count, mean wait time, and current waiter count.

docs/static-network.md

@@ -0,0 +1,46 @@
+# Static Networks
+
+`StaticNetwork` encodes the entire topology at compile time using a `make_network()` builder. Nodes and channel types are verified statically with zero runtime overhead.
+
+## Usage
+
+```cpp
+#include <kpn/kpn.hpp>
+using namespace kpn;
+
+static int  produce()        { return 42; }
+static int  double_it(int x) { return x * 2; }
+static void print_it(int x)  { std::cout << x << '\n'; }
+
+int main() {
+    auto src = make_node<produce>  ();
+    auto dbl = make_node<double_it>();
+    auto prn = make_node<print_it> ();
+
+    auto net = make_network(
+        edge(src, src.output<0>(), dbl, dbl.input<0>()),
+        edge(dbl, dbl.output<0>(), prn, prn.input<0>())
+    );
+
+    net.set_event_handler([](std::string_view name, NodeEvent ev, auto ts) {
+        // same API as Network
+    });
+
+    net.start();
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    net.stop();
+}
+```
+
+See `examples/10_static_hello_pipeline` and `examples/11_static_fanout`.
+
+## When to use
+
+| | `Network` | `StaticNetwork` |
+|---|---|---|
+| Topology known at | Runtime | Compile time |
+| Type checking | Runtime (`dynamic_cast`) | Compile time |
+| Overhead | Minimal | Zero |
+| Flexibility | Add nodes dynamically | Fixed at compile time |
+
+For most applications `Network` is sufficient. Use `StaticNetwork` when you need the absolute minimum overhead or want compile-time topology verification.

examples/01_hello_pipeline/main.cpp

@@ -7,20 +7,20 @@
 //
 //  [produce] --int--> [double_it] --int--> [print_it]
 
-// [snippet: basic_node_fns]
+// --8<-- [start:basic_node_fns]
 static int  produce()        { return 42; }
 static int  double_it(int x) { return x * 2; }
 static void print_it(int x)  { std::cout << "result: " << x << '\n'; }
-// [/snippet: basic_node_fns]
+// --8<-- [end:basic_node_fns]
 
 int main() {
     using namespace kpn;
 
-    // [snippet: index_only_nodes]
+    // --8<-- [start:index_only_nodes]
     auto src  = make_node<produce>(5);
     auto dbl  = make_node<double_it>(5);
     auto sink = make_node<print_it>(5);
-    // [/snippet: index_only_nodes]
+    // --8<-- [end:index_only_nodes]
 
     // Wire channels
     auto& dbl_in  = dbl.input_channel<0>();
@@ -28,7 +28,7 @@ int main() {
     src.set_output_channel<0>(&dbl_in);
     dbl.set_output_channel<0>(&sink_in);
 
-    // [snippet: network_build]
+    // --8<-- [start:network_build]
     Network net;
     net.add("src",  src)
        .add("dbl",  dbl)
@@ -40,5 +40,5 @@ int main() {
     net.start();
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
     net.stop();
-    // [/snippet: network_build]
+    // --8<-- [end:network_build]
 }

examples/02_named_ports/main.cpp

@@ -54,7 +54,7 @@ static void report(int count, std::vector<std::string> words) {
 int main() {
     using namespace kpn;
 
-    // [snippet: named_port_creation]
+    // --8<-- [start:named_port_creation]
     // tokenise: no inputs, one named output "words"
     auto tok = make_node<tokenise>(out<"words">{}, 4);
 
@@ -63,9 +63,9 @@ int main() {
 
     // report: two named inputs
     auto snk = make_node<report>(in<"count", "words">{}, 4);
-    // [/snippet: named_port_creation]
+    // --8<-- [end:named_port_creation]
 
-    // [snippet: named_port_network]
+    // --8<-- [start:named_port_network]
     Network net;
     net.add("tok", tok)
        .add("cnt", cnt)
@@ -78,5 +78,5 @@ int main() {
     net.start();
     std::this_thread::sleep_for(std::chrono::milliseconds(500));
     net.stop();
-    // [/snippet: named_port_network]
+    // --8<-- [end:named_port_network]
 }

examples/03_multi_output/main.cpp

@@ -33,7 +33,7 @@ static std::string generate() {
     return pairs[gen_index++ % 5];
 }
 
-// [snippet: multi_output_fn]
+// --8<-- [start:multi_output_fn]
 // Multi-output: returns (key, value) as a tuple — KPN++ routes each element
 // to its own output port automatically.
 static std::tuple<std::string, std::string> parse(std::string kv) {
@@ -41,7 +41,7 @@ static std::tuple<std::string, std::string> parse(std::string kv) {
     if (sep == std::string::npos) return {kv, ""};
     return {kv.substr(0, sep), kv.substr(sep + 1)};
 }
-// [/snippet: multi_output_fn]
+// --8<-- [end:multi_output_fn]
 
 static void print_key(std::string key) {
     std::cout << "KEY   → " << key << '\n';
@@ -56,7 +56,7 @@ static void print_value(std::string value) {
 int main() {
     using namespace kpn;
 
-    // [snippet: fanout_network]
+    // --8<-- [start:fanout_network]
     auto gen  = make_node<generate>(out<"kv">{},          4);
     auto par  = make_node<parse>   (in<"kv">{},  out<"key", "value">{}, 4);
     auto keys = make_node<print_key>  (in<"key">{},   4);
@@ -75,5 +75,5 @@ int main() {
     net.start();
     std::this_thread::sleep_for(std::chrono::milliseconds(600));
     net.stop();
-    // [/snippet: fanout_network]
+    // --8<-- [end:fanout_network]
 }

examples/04_storage_policy/main.cpp

@@ -34,14 +34,14 @@ struct Tag {
     int value = 0;
 };
 
-// [snippet: storage_policy_spec]
+// --8<-- [start:storage_policy_spec]
 // Override: store Tag by value despite being a struct
 // (it's trivially copyable and small — this just makes the policy explicit)
 template<>
 struct kpn::channel_storage_policy<Tag> {
     static constexpr bool by_value = true;
 };
-// [/snippet: storage_policy_spec]
+// --8<-- [end:storage_policy_spec]
 
 // ── Node functions ────────────────────────────────────────────────────────────
 

examples/05_error_handling/main.cpp

@@ -45,7 +45,7 @@ int main() {
 
     Network net;
 
-    // [snippet: diagnostics_handler]
+    // --8<-- [start:diagnostics_handler]
     // Custom diagnostics handler — fires on the watchdog interval.
     // Print a concise one-liner rather than the full table.
     net.set_diagnostics_handler([](const std::vector<NodeSnapshot>& nodes,
@@ -58,7 +58,7 @@ int main() {
                       << "overflows=" << c.overflows;
         std::cout << '\n';
     });
-    // [/snippet: diagnostics_handler]
+    // --8<-- [end:diagnostics_handler]
 
     net.set_watchdog_interval(std::chrono::milliseconds(200));
 

examples/07_python_network/example.py

@@ -30,3 +30,8 @@ net.build()
 net.start()
 time.sleep(0.1)
 net.stop()
+
+# Drop the network deterministically: it holds the Python callable, which forms
+# a reference cycle via globals(). Deleting the global breaks it so the network
+# is reclaimed now rather than lingering to interpreter shutdown.
+del net

examples/08_python_subport/example.py

@@ -1,38 +1,55 @@
 """
-08_python_subport — tap a C++ node's output from Python using net.read().
+08_python_subport — drive a *Python* node from Python via write()/read() taps.
 
 Graph:
-  [ProduceNode] --int--> [DoubleItNode] --int--> (tapped by net.read())
+  (fed by net.write()) --int--> [py_triple] --int--> (tapped by net.read())
 
-The sink is Python: instead of connecting a PrintItNode, we call net.read()
-to pull values out of DoubleItNode's output directly into Python.
-We also demonstrate net.write() by injecting a value into DoubleItNode's input.
+Unlike 07, there is no C++ source or sink here: the only node in the network is
+a pure-Python function, py_triple. Python plays *both* the producer and the
+consumer by using the subport taps:
+
+  * net.write("py", 0, v)  injects v into py_triple's input   (Python -> network)
+  * net.read("py", 0)      pulls py_triple's output back out   (network -> Python)
+
+This closes the loop the old version left as a "#todo": a value flows from
+Python, through a Python node running inside the network, and back to Python.
 """
 
 import sys
-import time
-import threading
-sys.path.insert(0, "build/python")
+sys.path.insert(0, "build/python")  # for `python examples/.../example.py` from repo root
 
 import kpn_python as kpn
 
+
+def py_triple(x: int) -> int:
+    return x * 3
+
+
 net = kpn.Network()
 
-net.add("src", kpn.make_produce())
-net.add("dbl", kpn.make_double_it())
+# The whole network is a single Python node with a tapped input and output.
+net.add_node("py", py_triple, inputs=["int"], outputs=["int"])
 
-net.connect("src", 0, "dbl", 0)
 net.build()
 net.start()
 
-# Collect a few values from DoubleItNode's output via Python tap
+# Push values in from Python and read the Python node's results back out.
+inputs = [1, 2, 7, 10, 100]
 results = []
-for _ in range(5):
-    val = net.read("dbl", 0)
-    results.append(val)
+for v in inputs:
+    net.write("py", 0, v)          # Python -> py_triple input
+    results.append(net.read("py", 0))  # py_triple output -> Python
 
 net.stop()
 
-print("values read from C++ DoubleItNode output:", results)
-assert all(v == 84 for v in results), f"expected all 84, got {results}"
-print("all correct (42 * 2 = 84)")
+print("inputs written from Python: ", inputs)
+print("outputs read from py_triple:", results)
+
+expected = [v * 3 for v in inputs]
+assert results == expected, f"expected {expected}, got {results}"
+print("all correct (x * 3 computed by a Python node inside the network)")
+
+# Drop the network deterministically. The network holds the Python callable,
+# which (via globals) forms a reference cycle; deleting the global breaks it so
+# the network is reclaimed promptly instead of lingering to interpreter exit.
+del net

examples/09_opencv_cellshade/main.cpp

@@ -38,7 +38,7 @@ static cv::Mat make_gradient(int W, int H) {
 
 // ── Pipeline functions ────────────────────────────────────────────────────────
 
-// [snippet: capture_fn]
+// --8<-- [start:capture_fn]
 static std::tuple<cv::Mat, cv::Mat> capture() {
     constexpr int W = 640, H = 480;
     static cv::VideoCapture cap;
@@ -75,7 +75,7 @@ static std::tuple<cv::Mat, cv::Mat> capture() {
     }
     return {frame.clone(), frame.clone()};
 }
-// [/snippet: capture_fn]
+// --8<-- [end:capture_fn]
 
 static cv::Mat to_gray(cv::Mat bgr) {
     cv::Mat gray;
@@ -120,7 +120,7 @@ static std::tuple<cv::Mat, cv::Mat> composite(cv::Mat edge_mask, cv::Mat colour)
 // The constructor opens both windows on the main thread (Wayland requirement).
 // operator() is called by step() whenever both channels have a frame ready.
 
-// [snippet: display_node]
+// --8<-- [start:display_node]
 class DisplayNode : public kpn::MainThreadNode<DisplayNode,
                                                kpn::in<"composite", "edges">,
                                                cv::Mat, cv::Mat> {
@@ -150,14 +150,14 @@ private:
         catch (const cv::Exception&) { return false; }
     }
 };
-// [/snippet: display_node]
+// --8<-- [end:display_node]
 
 // ─────────────────────────────────────────────────────────────────────────────
 
 int main() {
     using namespace kpn;
 
-    // [snippet: opencv_network]
+    // --8<-- [start:opencv_network]
     auto src       = make_node<capture>  (out<"colour","grey">{},            8);
     auto gray_node = make_node<to_gray>  (in<"bgr">{},   out<"gray">{},      8);
     auto edge_node = make_node<edges_fn> (in<"gray">{},  out<"edges">{},     8);
@@ -182,7 +182,7 @@ int main() {
        .connect("comp",  comp.template output<"result">(),     "display", disp.template input<"composite">())
        .connect("comp",  comp.template output<"edges">(),      "display", disp.template input<"edges">())
        .build();
-    // [/snippet: opencv_network]
+    // --8<-- [end:opencv_network]
 
     net.set_watchdog_interval(std::chrono::milliseconds(5000));
 #ifdef KPN_WEB_DEBUG
@@ -192,7 +192,7 @@ int main() {
     std::cout << "Cell-shading pipeline running. Press 'q' to stop.\n";
     std::cout << "Web debug UI: http://localhost:9090\n";
 
-    // [snippet: main_thread_step]
+    // --8<-- [start:main_thread_step]
     net.start();
 
     // Main thread drives display — imshow/waitKey stay on the GUI thread.
@@ -201,6 +201,6 @@ int main() {
         cv::waitKey(8); // yield event loop when no frame ready
 
     net.stop();
-    // [/snippet: main_thread_step]
+    // --8<-- [end:main_thread_step]
     return 0;
 }

examples/15_node_error_handler/main.cpp

@@ -44,6 +44,7 @@ int main() {
     auto proc = make_node<validate>();
     auto snk  = make_node<sink>   ();
 
+    // --8<-- [start:error_handler]
     // Return true  → skip this invocation, keep the node running.
     // Return false → stop the node (downstream drains then also stops).
     proc.set_error_handler([](std::string_view name, std::exception_ptr ep) {
@@ -53,6 +54,7 @@ int main() {
         }
         return true;
     });
+    // --8<-- [end:error_handler]
 
     Network net;
     net.add("source",    src)

examples/16_event_callbacks/main.cpp

@@ -0,0 +1,83 @@
+// Example 16 — Event Callbacks: overflow and node-stopped signals
+//
+// Two complementary observation mechanisms:
+//
+// 1. Per-node overflow callback   set_overflow_callback()
+//    Fired (with a timestamp) when a node's output channel is full and an
+//    item is dropped.  Useful for targeted monitoring of a specific node.
+//
+// 2. Network-level event handler  net.set_event_handler()
+//    Aggregate callback covering every node: receives the node name, a
+//    NodeEvent (Overflow or Closed), and a timestamp.  Register once and
+//    observe the whole network.
+//
+// Pipeline:  [fast_source] --int--> [slow_sink]
+//
+// fast_source produces at ~500 items/s; slow_sink consumes at ~20 items/s.
+// The channel capacity is 3, so overflows appear within milliseconds.
+
+#include <kpn/kpn.hpp>
+#include <atomic>
+#include <chrono>
+#include <iostream>
+#include <thread>
+
+using namespace kpn;
+using namespace std::chrono;
+
+// ── Node functions ────────────────────────────────────────────────────────────
+
+// --8<-- [start:node_fns]
+static std::atomic<int> g_seq{0};
+
+static int fast_source() {
+    std::this_thread::sleep_for(milliseconds(2));   // ~500/s
+    return g_seq.fetch_add(1);
+}
+
+static void slow_sink(int x) {
+    std::this_thread::sleep_for(milliseconds(50));  // ~20/s
+    std::cout << "  consumed: " << x << '\n';
+}
+// --8<-- [end:node_fns]
+
+// ── main ──────────────────────────────────────────────────────────────────────
+
+int main() {
+    auto src = make_node<fast_source>(/*capacity=*/3);
+    auto snk = make_node<slow_sink> (/*capacity=*/3);
+
+    // --8<-- [start:per_node_callback]
+    // Per-node overflow callback — no node name needed, known at registration.
+    std::atomic<int> overflow_count{0};
+    src.set_overflow_callback([&](steady_clock::time_point ts) {
+        auto ms = duration_cast<milliseconds>(ts.time_since_epoch()).count();
+        std::cerr << "[overflow] fast_source at t=" << ms << "ms\n";
+        overflow_count.fetch_add(1);
+    });
+    // --8<-- [end:per_node_callback]
+
+    Network net;
+
+    // --8<-- [start:network_event_handler]
+    // Network-level aggregate handler — covers every node, includes node name.
+    net.set_event_handler([](std::string_view name, NodeEvent ev,
+                             steady_clock::time_point ts) {
+        auto ms = duration_cast<milliseconds>(ts.time_since_epoch()).count();
+        std::string_view kind = (ev == NodeEvent::Overflow) ? "overflow" : "closed";
+        std::cerr << "[net:" << kind << "] node=" << name << " t=" << ms << "ms\n";
+    });
+    // --8<-- [end:network_event_handler]
+
+    net.add("source", src)
+       .add("sink",   snk)
+       .connect("source", src.output<0>(), "sink", snk.input<0>())
+       .build()
+       .start();
+
+    std::this_thread::sleep_for(milliseconds(300));
+    net.stop();
+
+    std::cout << "\nTotal overflows observed by per-node callback: "
+              << overflow_count.load() << '\n';
+}

examples/CMakeLists.txt

@@ -1,8 +1,35 @@
 cmake_minimum_required(VERSION 3.21)
 
+# Build an example and register it as a CTest smoke test.
+# Examples that are self-terminating (fixed sleep → net.stop()) pass when
+# they exit 0 within TIMEOUT seconds. OpenCV/UI examples are excluded.
 function(kpn_example name)
     add_executable(${name} ${name}/main.cpp)
     target_link_libraries(${name} PRIVATE kpn)
+    add_test(NAME example_${name} COMMAND ${name})
+    set_tests_properties(example_${name} PROPERTIES
+        TIMEOUT 15
+        LABELS  examples
+    )
+endfunction()
+
+# Register a Python example script as a CTest smoke test. Runs the script with
+# PYTHONPATH pointing at the freshly-built kpn_python module, so it does not
+# depend on the caller's working directory or a hard-coded "build/python" path.
+function(kpn_python_example name)
+    if(NOT KPN_BUILD_PYTHON)
+        return()
+    endif()
+    add_test(
+        NAME    example_${name}
+        COMMAND ${CMAKE_COMMAND} -E env
+                "PYTHONPATH=$<TARGET_FILE_DIR:kpn_python>"
+                ${Python_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/${name}/example.py
+    )
+    set_tests_properties(example_${name} PROPERTIES
+        TIMEOUT 15
+        LABELS  examples
+    )
 endfunction()
 
 kpn_example(01_hello_pipeline)
@@ -11,9 +38,11 @@ kpn_example(03_multi_output)
 kpn_example(04_storage_policy)
 kpn_example(05_error_handling)
 kpn_example(06_watchdog)
-kpn_example(15_node_error_handler)
+set_tests_properties(example_06_watchdog PROPERTIES TIMEOUT 40)
 kpn_example(10_static_hello_pipeline)
 kpn_example(11_static_fanout)
+kpn_example(15_node_error_handler)
+kpn_example(16_event_callbacks)
 if(KPN_WEB_DEBUG)
     kpn_target_enable_web_debug(06_watchdog)
 
@@ -21,7 +50,9 @@ if(KPN_WEB_DEBUG)
     target_link_libraries(14_debug_hub PRIVATE kpn)
     kpn_target_enable_web_debug(14_debug_hub)
 endif()
-# 07 and 08 are Python scripts — no compiled target needed.
+# 07 and 08 are Python scripts — no compiled target, but run as smoke tests.
+kpn_python_example(07_python_network)
+kpn_python_example(08_python_subport)
 
 # 09 requires OpenCV — only build if found
 find_package(OpenCV QUIET COMPONENTS core imgproc highgui videoio)

include/kpn/channel.hpp

@@ -136,6 +136,65 @@ public:
             push_callback_();
     }
 
+    // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to
+    // drain instead of dropping (the throwing push()) — the producer just runs slower.
+    // Use when every value must be delivered (e.g. replaying a dump for scoring, where
+    // a dropped frame silently corrupts the result). SPSC: only the sole producer may
+    // call it. Returns false if the channel was disabled while waiting.
+    bool push_blocking(T value) {
+        for (;;) {
+            if (!accepting_.load(std::memory_order_acquire)) {
+                stats_.record_drop();
+                return false;
+            }
+            const std::size_t t = tail_.load(std::memory_order_relaxed);
+            const std::size_t h = head_.load(std::memory_order_acquire);
+            if (t - h < capacity_) {          // space available → normal push
+                const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
+                const bool was_empty = (t == h);
+                buf_[t & ring_mask_] = make_storage(std::move(value));
+                tail_.store(t + 1, std::memory_order_release);
+                stats_.record_push(t - h + 1, data_bytes);
+                wake_.fetch_add(1, std::memory_order_release);
+                wake_.notify_one();
+                if (was_empty && push_callback_) push_callback_();
+                return true;
+            }
+            // full: yield briefly and retry (consumer will drain)
+            std::this_thread::sleep_for(std::chrono::microseconds(50));
+        }
+    }
+
+    // Lossless, non-blocking delivery for a must-deliver control token (EOF).
+    //
+    // A sentinel is stored out-of-band — in a dedicated slot that does NOT
+    // consume ring capacity — so this can never overflow and never blocks the
+    // caller. That distinction is essential: each KPN node has a single worker
+    // thread, so a *blocking* push would park that thread and stop it draining
+    // its own input, cascading into a hold-and-wait deadlock under backpressure.
+    // Setting a flag and returning keeps the worker free to keep popping.
+    //
+    // The consumer's pop() drains the ring first, then delivers this sentinel,
+    // preserving ordering (EOF arrives after all data pushed before it).
+    //
+    // Only the sole producer may call it (SPSC contract, same as push()).
+    // Returns false if the channel is already disabled (token discarded —
+    // teardown is in progress, so the sentinel is moot).
+    bool push_sentinel(T value) {
+        if (!accepting_.load(std::memory_order_acquire)) {
+            stats_.record_drop();
+            return false;
+        }
+        eof_value_ = make_storage(std::move(value));
+        has_eof_.store(true, std::memory_order_release);
+        // Wake a consumer blocked in pop(): the sentinel is now deliverable even
+        // though the ring may be empty.
+        wake_.fetch_add(1, std::memory_order_release);
+        wake_.notify_one();
+        if (push_callback_) push_callback_();
+        return true;
+    }
+
     // Blocking pop. Returns when an item is available.
     // Throws ChannelClosedError if the channel is disabled (regardless of fill).
     T pop() {
@@ -148,21 +207,37 @@ public:
             // If empty, spin before sleeping: avoids the futex when the next item
             // arrives within the spin window (~4 µs at default spin_count=200 on x86).
             if (h == t) {
+                // Ring drained — deliver any pending out-of-band sentinel (EOF)
+                // now, so it always arrives after the data pushed before it.
+                //
+                // Re-confirm emptiness against a fresh tail_ first: the snapshot
+                // at the top of the loop may be stale (the producer can push more
+                // values *and* the sentinel in the window since), and the sentinel
+                // must never jump ahead of ring values pushed before it. The spin
+                // and post-spin takes below already reload tail_ on the line above
+                // them; this is the one take that used the loop-top snapshot.
+                if (h == tail_.load(std::memory_order_acquire)) {
+                    T s; if (take_sentinel(s)) return s;
+                }
+
                 if (!accepting_.load(std::memory_order_acquire))
                     throw ChannelClosedError{};
 
-                for (std::size_t s = 0; s < spin_count_; ++s) {
+                for (std::size_t si = 0; si < spin_count_; ++si) {
                     spin_hint();
                     t = tail_.load(std::memory_order_acquire);
                     if (t != h) break;
+                    { T s; if (take_sentinel(s)) return s; }
                     if (!accepting_.load(std::memory_order_relaxed))
                         throw ChannelClosedError{};
                 }
 
                 if (h == t) {
-                    // Still empty after spin — sleep until push() or disable() fires.
-                    // Re-check tail after loading w to guard against a lost wakeup.
+                    // Still empty after spin — sleep until push()/push_sentinel()
+                    // or disable() fires. Re-check tail and the sentinel after
+                    // loading w to guard against a lost wakeup.
                     if (tail_.load(std::memory_order_acquire) != h) continue;
+                    if (has_eof_.load(std::memory_order_acquire)) continue;
                     wake_.wait(w, std::memory_order_relaxed);
                     continue;
                 }
@@ -190,9 +265,12 @@ public:
     }
 
     // Immediate non-blocking pop. Returns false if the ring is empty.
+    // Once the ring is drained, delivers any pending out-of-band sentinel (EOF)
+    // so pool nodes — which pop only via this path — still receive the token.
     bool try_pop_now(T& out) {
         const std::size_t h = head_.load(std::memory_order_relaxed);
-        if (h == tail_.load(std::memory_order_acquire)) return false;
+        if (h == tail_.load(std::memory_order_acquire))
+            return take_sentinel(out);
         out = extract(std::move(buf_[h & ring_mask_]));
         head_.store(h + 1, std::memory_order_release);
         stats_.record_pop();
@@ -217,12 +295,21 @@ public:
         push_callback_ = std::move(cb);
     }
 
-    // Size derived lazily from ring indices — no separate counter on the hot path.
+    // Ring occupancy, derived lazily from indices — no separate counter on the
+    // hot path. Excludes any out-of-band sentinel (that lives outside the ring).
     std::size_t size() const {
         return tail_.load(std::memory_order_relaxed)
              - head_.load(std::memory_order_relaxed);
     }
-    std::size_t approx_size() const { return size(); }
+
+    // A pending out-of-band sentinel (EOF) counts as consumable work here even
+    // though it holds no ring slot. This is what node readiness checks call, so
+    // a channel carrying only a sentinel still schedules its consumer's next
+    // fire — without this the sentinel would never be popped and the pipeline
+    // would deadlock at teardown.
+    std::size_t approx_size() const {
+        return size() + (has_eof_.load(std::memory_order_acquire) ? 1u : 0u);
+    }
 
     std::size_t capacity()     const { return capacity_; }
     bool        is_accepting() const { return accepting_.load(std::memory_order_relaxed); }
@@ -260,6 +347,17 @@ private:
             return *s;
     }
 
+    // Consume the out-of-band sentinel if one is pending. Consumer-only.
+    // Called only when the ring is observed empty, so the sentinel is always
+    // delivered after every value pushed before it.
+    bool take_sentinel(T& out) {
+        if (!has_eof_.load(std::memory_order_acquire)) return false;
+        out = extract(std::move(eof_value_));
+        has_eof_.store(false, std::memory_order_release);
+        stats_.record_pop();
+        return true;
+    }
+
     const std::size_t               capacity_;
     const std::size_t               spin_count_;
     std::size_t                     ring_mask_;
@@ -267,8 +365,16 @@ private:
     std::function<void()>           push_callback_;
     ChannelStats                    stats_;
 
+    // Out-of-band sentinel (EOF): stored outside the ring so its delivery never
+    // depends on ring capacity and never blocks the producer. Written by the
+    // producer (push_sentinel), read+cleared by the consumer (take_sentinel);
+    // has_eof_ is the publish/consume handshake.
+    storage_type              eof_value_{};
+    std::atomic<bool>         has_eof_{false};
+
     // Separate cache lines: head_ is written only by the consumer;
     // tail_ and wake_ are written only by the producer.
+    //   wake_  wakes a blocked pop() on enqueue or on a pending sentinel.
     alignas(64) std::atomic<std::size_t> head_{0};
     alignas(64) std::atomic<std::size_t> tail_{0};
                 std::atomic<uint32_t>    wake_{0};

include/kpn/fanout.hpp

@@ -115,6 +115,16 @@ public:
         out_channels_[I] = ch;
     }
 
+    // Lossless fanout: block until each consumer drains rather than dropping.
+    void set_lossless_output(bool on) override { lossless_ = on; }
+
+    // Opt a single output back out of blocking.  Needed when one branch may
+    // stall indefinitely — a display tap nobody is servicing, say — since
+    // blocking on it would apply backpressure to every other branch too.
+    void set_lossy_output(std::size_t i, bool lossy = true) {
+        if (i < N) lossy_out_[i] = lossy;
+    }
+
 private:
     void run_loop() {
         while (!stop_flag_.load(std::memory_order_relaxed)) {
@@ -126,8 +136,19 @@ private:
 
                 for (std::size_t i = 0; i < N; ++i) {
                     if (out_channels_[i]) {
-                        try { out_channels_[i]->push(val); }
-                        catch (const ChannelOverflowError&) {}  // drop for this output independently
+                        // Lossless: block until this consumer drains.  Note the
+                        // branches differ in more than blocking — the dropping
+                        // path discards per-output independently and silently,
+                        // so a slow consumer on one branch costs frames on that
+                        // branch only, with no diagnostic.  That is the right
+                        // default for display taps but hides frame loss from
+                        // analysis branches.
+                        if (lossless_ && !lossy_out_[i])
+                            out_channels_[i]->push_blocking(val);
+                        else {
+                            try { out_channels_[i]->push(val); }
+                            catch (const ChannelOverflowError&) {}  // drop independently
+                        }
                     }
                 }
 
@@ -142,6 +163,8 @@ private:
 
     std::string                 name_;
     std::size_t                 fifo_capacity_;
+    bool                        lossless_{false};
+    std::array<bool, N>         lossy_out_{};   // per-output opt-out of blocking
     std::shared_ptr<Channel<T>> input_ch_;
     std::array<Channel<T>*, N>  out_channels_{};
     std::atomic<bool>           stop_flag_{false};

include/kpn/inode.hpp

@@ -1,5 +1,6 @@
 #pragma once
 #include "diagnostics.hpp"
+#include <chrono>
 #include <functional>
 #include <string>
 #include <string_view>
@@ -10,6 +11,13 @@ namespace kpn {
 // invocation and keep running, false to stop the node.
 using NodeErrorHandler = std::function<bool(std::string_view node_name, std::exception_ptr)>;
 
+// Lightweight timestamp-only callback fired on per-node events.
+// The node name is known at registration time so it is not included here.
+using NodeEventCallback = std::function<void(std::chrono::steady_clock::time_point)>;
+
+// Event types reported to the network-level aggregate callback.
+enum class NodeEvent { Overflow, Closed };
+
 // ── INode — type-erased interface for Network / watchdog ─────────────────────
 
 struct INode {
@@ -21,9 +29,18 @@ struct INode {
     virtual NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const = 0;
     virtual void set_name(std::string name) = 0;
 
+    // Network-injected callbacks (slot 1 of each node's callback array).
+    // Default no-ops; overridden by PoolNode, PoolObjectNode, InterruptNode.
+    virtual void set_network_overflow_callback(NodeEventCallback) {}
+    virtual void set_network_closed_callback(NodeEventCallback)   {}
+
     // halt(): alias for stop() — immediate, discards in-flight work.
     virtual void halt() { stop(); }
 
+    // Opt into lossless (blocking) output for nodes that support it.  Default
+    // is a no-op so node types with no output channels ignore it.
+    virtual void set_lossless_output(bool) {}
+
     // shutdown(): graceful drain before stopping. Base implementation falls
     // back to stop(). Network and StaticNetwork override with topo-ordered drain.
     virtual void shutdown() { stop(); }

include/kpn/interrupt_node.hpp

@@ -84,6 +84,11 @@ public:
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
+    void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
+    void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
+    void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
+    void set_network_closed_callback(NodeEventCallback cb)   override { closed_callbacks_[1] = std::move(cb); }
+
     const NodeStats& stats() const override { return stats_; }
 
     NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
@@ -170,8 +175,8 @@ private:
             auto cpu1 = NodeStats::cpu_now();
             auto t2   = clock_t::now();
             stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
-        } catch (const ChannelOverflowError& e) {
-            std::cerr << "[kpn] interrupt node overflow: " << e.what() << "\n";
+        } catch (const ChannelOverflowError&) {
+            fire_callbacks(event_callbacks_);
         } catch (...) {
             if (!error_handler_ || !error_handler_(name_, std::current_exception()))
                 fatal = true;
@@ -180,6 +185,8 @@ private:
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
 
         if (fatal) {
+            fire_callbacks(closed_callbacks_);
+            disable_outputs(std::make_index_sequence<output_count>{});
             pending_.store(0, std::memory_order_release);
             stop_flag_.store(true, std::memory_order_relaxed);
             return;
@@ -230,15 +237,28 @@ private:
     using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
         std::make_index_sequence<output_count>{}));
 
-    std::shared_ptr<IScheduler>  scheduler_;
-    std::string                  name_;
-    std::size_t                  fifo_capacity_;
-    output_channels_t            output_channels_{};
-    std::atomic<bool>            stop_flag_{true};
-    std::atomic<int>             pending_{0};   // triggers awaiting execution
-    NodeStats                    stats_;
-    NodeErrorHandler             error_handler_;
-    std::chrono::milliseconds    max_exec_time_{0};
+    template<std::size_t... Is>
+    void disable_outputs(std::index_sequence<Is...>) {
+        auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
+        (disable_one(std::get<Is>(output_channels_)), ...);
+    }
+
+    static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
+        const auto ts = std::chrono::steady_clock::now();
+        for (auto& cb : cbs) if (cb) cb(ts);
+    }
+
+    std::shared_ptr<IScheduler>              scheduler_;
+    std::string                              name_;
+    std::size_t                              fifo_capacity_;
+    output_channels_t                        output_channels_{};
+    std::atomic<bool>                        stop_flag_{true};
+    std::atomic<int>                         pending_{0};
+    NodeStats                                stats_;
+    NodeErrorHandler                         error_handler_;
+    std::chrono::milliseconds                max_exec_time_{0};
+    std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
+    std::array<NodeEventCallback, 2>         closed_callbacks_{};
 };
 
 // ── make_interrupt_node factory ───────────────────────────────────────────────

include/kpn/network.hpp

@@ -44,6 +44,9 @@ public:
     using DiagnosticsHandler =
         std::function<void(const std::vector<NodeSnapshot>&,
                            const std::vector<ChannelSnapshot>&)>;
+    using EventHandler =
+        std::function<void(std::string_view node_name, NodeEvent,
+                           std::chrono::steady_clock::time_point)>;
 
     // ── Builder API ───────────────────────────────────────────────────────────
 
@@ -111,6 +114,19 @@ public:
         for (auto& [name, _] : nodes_)
             if (color[name] == 0)
                 dfs(name, color);
+        if (event_handler_) {
+            for (auto& name : topo_) {
+                auto* node = nodes_.at(name);
+                node->set_network_overflow_callback(
+                    [this, n = name](auto ts) {
+                        event_handler_(n, NodeEvent::Overflow, ts);
+                    });
+                node->set_network_closed_callback(
+                    [this, n = name](auto ts) {
+                        event_handler_(n, NodeEvent::Closed, ts);
+                    });
+            }
+        }
         return *this;
     }
 
@@ -194,8 +210,9 @@ public:
         watchdog_interval_ = interval;
     }
 
-    void set_error_handler(ErrorHandler h)       { error_handler_       = std::move(h); }
+    void set_error_handler(ErrorHandler h)             { error_handler_  = std::move(h); }
     void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_  = std::move(h); }
+    void set_event_handler(EventHandler h)             { event_handler_  = std::move(h); }
 
     void register_pool(const std::string& name, IPoolProbe* probe) {
         pool_probes_.emplace_back(name, probe);
@@ -430,6 +447,7 @@ private:
     std::vector<std::pair<std::string, IPoolProbe*>>       pool_probes_;
     ErrorHandler                                           error_handler_;
     DiagnosticsHandler                                     diag_handler_;
+    EventHandler                                           event_handler_;
     std::chrono::milliseconds                              watchdog_interval_{3000};
     std::jthread                                           watchdog_;
     clock_t::time_point                                    start_time_;

include/kpn/pool_node.hpp

@@ -22,6 +22,38 @@
 
 namespace kpn {
 
+// ── Sentinel detection ────────────────────────────────────────────────────────
+// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
+// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
+// source Frame) or nested one level under a `.source` member (`v.source.eof`,
+// as on the pipeline's SceneFrame/…/MatchedSceneFrame message types, which wrap
+// the originating Frame). Sentinels are delivered losslessly and non-blockingly
+// via Channel::push_sentinel() instead of the throwing push(), so backpressure
+// can never drop the token that unblocks downstream teardown.
+//
+// Types with neither shape are never treated as sentinels — both traits are
+// SFINAE-safe and the runtime check compiles away to `false` for them, so this
+// stays a no-op for pipelines that don't use an eof convention.
+template<typename T, typename = void>
+struct has_eof_field : std::false_type {};
+template<typename T>
+struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
+    : std::true_type {};
+
+template<typename T, typename = void>
+struct has_source_eof_field : std::false_type {};
+template<typename T>
+struct has_source_eof_field<T,
+    std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
+    : std::true_type {};
+
+template<typename T>
+constexpr bool is_sentinel_value(const T& v) {
+    if constexpr (has_eof_field<T>::value)             return static_cast<bool>(v.eof);
+    else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
+    else                                               return false;
+}
+
 // ── PoolNode ──────────────────────────────────────────────────────────────────
 //
 // Reactive alternative to Node<>. Instead of owning a blocked thread, the node
@@ -101,6 +133,19 @@ public:
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
+    // Lossless output: block the producer until the consumer drains rather than
+    // dropping on a full channel.  Default is drop, which suits live sources
+    // where a stale frame is worth less than a fresh one.  Enable for offline
+    // batch runs, where a dropped item leaves a gap that downstream analysis
+    // cannot recover.  Must be set before start().
+    void set_lossless_output(bool on) override { lossless_ = on; }
+    void set_lossless(bool on) { set_lossless_output(on); }
+
+    void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
+    void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
+    void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
+    void set_network_closed_callback(NodeEventCallback cb)   override { closed_callbacks_[1] = std::move(cb); }
+
     const NodeStats& stats() const override { return stats_; }
 
     NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
@@ -189,12 +234,31 @@ private:
         (std::get<Is>(input_channels_)->disable(), ...);
     }
 
+    template<std::size_t... Is>
+    void disable_outputs(std::index_sequence<Is...>) {
+        auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
+        (disable_one(std::get<Is>(output_channels_)), ...);
+    }
+
     template<std::size_t... Is>
     void register_callbacks(std::index_sequence<Is...>) {
         (std::get<Is>(input_channels_)->set_push_callback(
             [this] { on_input_ready(); }), ...);
     }
 
+    static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
+        const auto ts = std::chrono::steady_clock::now();
+        for (auto& cb : cbs) if (cb) cb(ts);
+    }
+
+    void self_stop() {
+        disable_inputs(std::make_index_sequence<input_count>{});
+        disable_outputs(std::make_index_sequence<output_count>{});
+        stats_.exec_start_us.store(0, std::memory_order_relaxed);
+        queued_.store(false, std::memory_order_release);
+        stop_flag_.store(true, std::memory_order_relaxed);
+    }
+
     template<typename Tup, std::size_t... Is>
     static auto make_input_channel_tuple(std::index_sequence<Is...>)
         -> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;
@@ -278,19 +342,17 @@ private:
             // blocked_time = 0 for pool nodes (we don't block waiting for inputs)
             stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
         } catch (const ChannelClosedError&) {
-            stats_.exec_start_us.store(0, std::memory_order_relaxed);
-            queued_.store(false, std::memory_order_release);
-            stop_flag_.store(true, std::memory_order_relaxed);
+            fire_callbacks(closed_callbacks_);
+            self_stop();
             return;
-        } catch (const ChannelOverflowError& e) {
-            std::cerr << "[kpn] pool overflow: " << e.what() << "\n";
+        } catch (const ChannelOverflowError&) {
+            fire_callbacks(event_callbacks_);
         } catch (...) {
             if (error_handler_ && error_handler_(name_, std::current_exception())) {
                 // continue — fall through to resubmit check
             } else {
-                stats_.exec_start_us.store(0, std::memory_order_relaxed);
-                queued_.store(false, std::memory_order_release);
-                stop_flag_.store(true, std::memory_order_relaxed);
+                fire_callbacks(closed_callbacks_);
+                self_stop();
                 return;
             }
         }
@@ -339,6 +401,22 @@ private:
     void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
         if (!ch) return;
+        // Sentinels (EOF) must never be dropped: a lost token wedges every
+        // downstream pop() forever. Deliver them out-of-band (push_sentinel),
+        // which never overflows and never blocks this node's worker thread.
+        if (is_sentinel_value(val)) {
+            ch->push_sentinel(std::move(val));
+            return;
+        }
+        // Lossless mode: block until the consumer drains instead of dropping.
+        // Dropping is the right default for live sources (a stale frame is
+        // worth less than a fresh one), but for offline batch work every sample
+        // matters — dropped frames leave a non-uniformly sampled series, which
+        // silently invalidates any fixed-rate spectral analysis downstream.
+        if (lossless_) {
+            ch->push_blocking(std::move(val));
+            return;
+        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {
@@ -359,16 +437,19 @@ private:
 
     // ── State ─────────────────────────────────────────────────────────────────
 
-    std::shared_ptr<IScheduler>  scheduler_;
-    std::string                  name_;
-    std::size_t                  fifo_capacity_;
-    input_channels_t             input_channels_;
-    output_channels_t            output_channels_{};
-    std::atomic<bool>            stop_flag_{true};
-    std::atomic<bool>            queued_{false};
-    NodeStats                    stats_;
-    NodeErrorHandler             error_handler_;
-    std::chrono::milliseconds    max_exec_time_{0};
+    std::shared_ptr<IScheduler>              scheduler_;
+    std::string                              name_;
+    bool                                     lossless_{false};
+    std::size_t                              fifo_capacity_;
+    input_channels_t                         input_channels_;
+    output_channels_t                        output_channels_{};
+    std::atomic<bool>                        stop_flag_{true};
+    std::atomic<bool>                        queued_{false};
+    NodeStats                                stats_;
+    NodeErrorHandler                         error_handler_;
+    std::chrono::milliseconds                max_exec_time_{0};
+    std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
+    std::array<NodeEventCallback, 2>         closed_callbacks_{};
 };
 
 // ── PoolObjectNode ────────────────────────────────────────────────────────────
@@ -435,6 +516,19 @@ public:
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
+    // Lossless output: block the producer until the consumer drains rather than
+    // dropping on a full channel.  Default is drop, which suits live sources
+    // where a stale frame is worth less than a fresh one.  Enable for offline
+    // batch runs, where a dropped item leaves a gap that downstream analysis
+    // cannot recover.  Must be set before start().
+    void set_lossless_output(bool on) override { lossless_ = on; }
+    void set_lossless(bool on) { set_lossless_output(on); }
+
+    void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
+    void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
+    void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
+    void set_network_closed_callback(NodeEventCallback cb)   override { closed_callbacks_[1] = std::move(cb); }
+
     const NodeStats& stats() const override { return stats_; }
 
     NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
@@ -490,13 +584,31 @@ private:
               std::make_shared<Channel<std::tuple_element_t<Is, args_tuple>>>(fifo_capacity_)),
          ...);
     }
-    template<std::size_t... Is> void enable_inputs(std::index_sequence<Is...>)   { (std::get<Is>(input_channels_)->enable(), ...); }
-    template<std::size_t... Is> void disable_inputs(std::index_sequence<Is...>)  { (std::get<Is>(input_channels_)->disable(), ...); }
+    template<std::size_t... Is> void enable_inputs(std::index_sequence<Is...>)  { (std::get<Is>(input_channels_)->enable(), ...); }
+    template<std::size_t... Is> void disable_inputs(std::index_sequence<Is...>) { (std::get<Is>(input_channels_)->disable(), ...); }
+    template<std::size_t... Is>
+    void disable_outputs(std::index_sequence<Is...>) {
+        auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
+        (disable_one(std::get<Is>(output_channels_)), ...);
+    }
     template<std::size_t... Is>
     void register_callbacks(std::index_sequence<Is...>) {
         (std::get<Is>(input_channels_)->set_push_callback([this] { on_input_ready(); }), ...);
     }
 
+    static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
+        const auto ts = std::chrono::steady_clock::now();
+        for (auto& cb : cbs) if (cb) cb(ts);
+    }
+
+    void self_stop() {
+        disable_inputs(std::make_index_sequence<input_count>{});
+        disable_outputs(std::make_index_sequence<output_count>{});
+        stats_.exec_start_us.store(0, std::memory_order_relaxed);
+        queued_.store(false, std::memory_order_release);
+        stop_flag_.store(true, std::memory_order_relaxed);
+    }
+
     template<typename Tup, std::size_t... Is>
     static auto make_input_channel_tuple(std::index_sequence<Is...>)
         -> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;
@@ -567,18 +679,16 @@ private:
             auto t2   = clock_t::now();
             stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
         } catch (const ChannelClosedError&) {
-            stats_.exec_start_us.store(0, std::memory_order_relaxed);
-            queued_.store(false, std::memory_order_release);
-            stop_flag_.store(true, std::memory_order_relaxed);
+            fire_callbacks(closed_callbacks_);
+            self_stop();
             return;
-        } catch (const ChannelOverflowError& e) {
-            std::cerr << "[kpn] pool overflow: " << e.what() << "\n";
+        } catch (const ChannelOverflowError&) {
+            fire_callbacks(event_callbacks_);
         } catch (...) {
             if (error_handler_ && error_handler_(name_, std::current_exception())) {
             } else {
-                stats_.exec_start_us.store(0, std::memory_order_relaxed);
-                queued_.store(false, std::memory_order_release);
-                stop_flag_.store(true, std::memory_order_relaxed);
+                fire_callbacks(closed_callbacks_);
+                self_stop();
                 return;
             }
         }
@@ -615,6 +725,18 @@ private:
     void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
         if (!ch) return;
+        // Sentinels (EOF) must never be dropped: a lost token wedges every
+        // downstream pop() forever. Deliver them out-of-band (push_sentinel),
+        // which never overflows and never blocks this node's worker thread.
+        if (is_sentinel_value(val)) {
+            ch->push_sentinel(std::move(val));
+            return;
+        }
+        // Lossless mode: block until the consumer drains instead of dropping.
+        if (lossless_) {
+            ch->push_blocking(std::move(val));
+            return;
+        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {
@@ -623,17 +745,20 @@ private:
         }
     }
 
-    Obj&                         obj_;
-    std::shared_ptr<IScheduler>  scheduler_;
-    std::string                  name_;
-    std::size_t                  fifo_capacity_;
-    input_channels_t             input_channels_;
-    output_channels_t            output_channels_{};
-    std::atomic<bool>            stop_flag_{true};
-    std::atomic<bool>            queued_{false};
-    NodeStats                    stats_;
-    NodeErrorHandler             error_handler_;
-    std::chrono::milliseconds    max_exec_time_{0};
+    Obj&                                     obj_;
+    std::shared_ptr<IScheduler>              scheduler_;
+    std::string                              name_;
+    bool                                     lossless_{false};
+    std::size_t                              fifo_capacity_;
+    input_channels_t                         input_channels_;
+    output_channels_t                        output_channels_{};
+    std::atomic<bool>                        stop_flag_{true};
+    std::atomic<bool>                        queued_{false};
+    NodeStats                                stats_;
+    NodeErrorHandler                         error_handler_;
+    std::chrono::milliseconds                max_exec_time_{0};
+    std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
+    std::array<NodeEventCallback, 2>         closed_callbacks_{};
 };
 
 // ── make_pool_node factory (NTTP) ─────────────────────────────────────────────

include/kpn/python/auto_bind.hpp

@@ -247,7 +247,7 @@ void bind_network(nb::module_& m) {
 
     nb::class_<IVariantNode<Variant>>(m, "INode");
 
-    nb::class_<Net>(m, "Network")
+    nb::class_<Net>(m, "Network", nb::type_slots(network_type_slots<Variant>()))
         .def("__init__", [](Net* self) {
             new (self) Net();
             register_all_converters<Registry>(*self);

include/kpn/python/bindings.hpp

@@ -35,6 +35,16 @@ public:
     using VNode    = IVariantNode<Variant>;
     using VChannel = IVariantChannel<Variant>;
 
+    // ── GC support ────────────────────────────────────────────────────────────
+    // Visit every Python object this network transitively holds (currently the
+    // callable of each PyNode). Used by the Network type's tp_traverse slot so
+    // Python's cyclic GC can discover instance → callable → globals() cycles.
+    // Defined out-of-line below, once PyNode is a complete type.
+    template<typename Fn>
+    void visit_python_objects(Fn&& visit) const;
+    // Drop all Python references held by nodes, breaking any cycle (tp_clear).
+    void clear_python_objects();
+
     // ── Builder API ───────────────────────────────────────────────────────────
 
     void add(std::string name, std::shared_ptr<VNode> node) {
@@ -207,6 +217,20 @@ public:
         return it->second;
     }
 
+    // Raw node handle by name — lets a binding dynamic_cast to a concrete wrapper
+    // type and call its functor's runtime setters (persistent-pipeline reuse).
+    VNode* node_ptr(const std::string& name) { return &node_at(name); }
+
+    // Per-node timing snapshot for profiling where a replay spends its time.
+    std::map<std::string, double> node_stats(const std::string& name) {
+        auto& n = node_at(name);
+        NodeSnapshot s = n.node_snapshot(name, 0.0);
+        return {{"frames", double(s.frames_processed)},
+                {"exec_ms", s.ema_exec_ms}, {"max_ms", s.max_exec_ms},
+                {"blocked_ms", s.total_blocked_ms}, {"fps", s.throughput_fps},
+                {"cpu_ms", s.total_cpu_ms}, {"cpu_util_pct", s.cpu_util_pct}};
+    }
+
 private:
     VNode& node_at(const std::string& name) {
         auto it = nodes_.find(name);
@@ -367,6 +391,14 @@ public:
         out_channels_[i] = std::move(ch);
     }
 
+    // ── GC support (tp_traverse / tp_clear on the owning Network) ──────────────
+    // The node holds a Python callable, which typically forms an
+    // instance → callable → globals() → instance cycle. Expose the callable so
+    // the Network's GC slots can traverse and clear it. See bindings.hpp's
+    // network_tp_traverse/network_tp_clear.
+    const nb::object& python_callable() const { return callable_; }
+    void clear_python_callable() { callable_ = nb::object(); }
+
 private:
     void run_loop() {
         while (!stop_flag_.load(std::memory_order_relaxed)) {
@@ -404,13 +436,17 @@ private:
 
                 for (std::size_t i = 0; i < out_channels_.size(); ++i) {
                     if (out_channels_[i])
-                        out_channels_[i]->push(std::move(outputs[i]));
+                        // Lossless: wait for space rather than drop. A dropped frame
+                        // silently corrupts a replay's score; backpressure just slows
+                        // the producer. (Was push() + "drop on overflow".)
+                        out_channels_[i]->push_blocking(std::move(outputs[i]));
                 }
 
             } catch (const ChannelClosedError&) {
                 break;
             } catch (const ChannelOverflowError&) {
-                // drop and continue
+                // no longer reachable with push_blocking, kept for safety
+                break;
             }
         }
     }
@@ -439,6 +475,60 @@ private:
     NodeStats          stats_;
 };
 
+// ── PyNetwork GC helpers (defined here: PyNode is now complete) ────────────────
+
+template<typename Variant>
+template<typename Fn>
+void PyNetwork<Variant>::visit_python_objects(Fn&& visit) const {
+    for (const auto& [name, node] : nodes_)
+        if (auto* py = dynamic_cast<const PyNode<Variant>*>(node.get()))
+            visit(py->python_callable());
+}
+
+template<typename Variant>
+void PyNetwork<Variant>::clear_python_objects() {
+    for (auto& [name, node] : nodes_)
+        if (auto* py = dynamic_cast<PyNode<Variant>*>(node.get()))
+            py->clear_python_callable();
+}
+
+// ── GC type slots for the Network binding ─────────────────────────────────────
+// The Network holds Python callables (via PyNode), forming uncollectable
+// instance → callable → globals() → instance cycles at interpreter shutdown.
+// These slots let Python's cyclic collector traverse and break them, silencing
+// nanobind's leak warnings. See the nanobind "Reference leaks" documentation.
+
+template<typename Variant>
+int network_tp_traverse(PyObject* self, visitproc visit, void* arg) {
+    Py_VISIT(Py_TYPE(self));
+    if (!nb::inst_ready(self))
+        return 0;
+    auto* net = nb::inst_ptr<PyNetwork<Variant>>(self);
+    int rv = 0;
+    net->visit_python_objects([&](const nb::object& obj) {
+        if (rv == 0 && obj.is_valid())
+            rv = visit(obj.ptr(), arg);
+    });
+    return rv;
+}
+
+template<typename Variant>
+int network_tp_clear(PyObject* self) {
+    auto* net = nb::inst_ptr<PyNetwork<Variant>>(self);
+    net->clear_python_objects();
+    return 0;
+}
+
+template<typename Variant>
+PyType_Slot* network_type_slots() {
+    static PyType_Slot slots[] = {
+        { Py_tp_traverse, reinterpret_cast<void*>(&network_tp_traverse<Variant>) },
+        { Py_tp_clear,    reinterpret_cast<void*>(&network_tp_clear<Variant>) },
+        { 0, nullptr }
+    };
+    return slots;
+}
+
 // ── register_py_network (legacy helper) ───────────────────────────────────────
 // Registers PyNetwork<Variant> with the given nanobind module.
 // Prefer bind_network<Registry> from auto_bind.hpp for new code.
@@ -447,7 +537,7 @@ template<typename Variant>
 void register_py_network(nb::module_& m, const char* class_name = "Network") {
     using Net = PyNetwork<Variant>;
 
-    nb::class_<Net>(m, class_name)
+    nb::class_<Net>(m, class_name, nb::type_slots(network_type_slots<Variant>()))
         .def(nb::init<>())
         .def("connect", &Net::connect,
              nb::arg("src"), nb::arg("out_idx"),
@@ -458,7 +548,8 @@ void register_py_network(nb::module_& m, const char* class_name = "Network") {
         .def("read",   &Net::read,
              nb::arg("node"), nb::arg("out_idx") = std::size_t(0))
         .def("write",  &Net::write,
-             nb::arg("node"), nb::arg("in_idx"), nb::arg("value"));
+             nb::arg("node"), nb::arg("in_idx"), nb::arg("value"))
+        .def("node_stats", &Net::node_stats, nb::arg("node"));
 }
 
 } // namespace kpn::python

include/kpn/python/object_variant_node.hpp

@@ -0,0 +1,149 @@
+#pragma once
+// ObjectVariantNodeWrapper — variant-node adapter for *stateful* functors.
+//
+// VariantNodeWrapper (variant_node.hpp) wraps Node<Func,...>, where Func is a
+// default-constructible NTTP callable. That doesn't fit nodes whose functor must
+// be constructed with runtime state (a Config, a loaded gallery, etc.) — those use
+// ObjectNode<Obj>, which takes `Obj& obj` at construction.
+//
+// This wrapper owns an Obj instance and exposes the same IVariantNode surface so a
+// stateful C++ node can live inside a PyNetwork. Build one via a factory that
+// constructs the functor from Python-supplied config, e.g.:
+//
+//   auto n = std::make_shared<ObjectVariantNodeWrapper<
+//                IdentityMatcherFunc, Variant, in<"tracked">, out<"matched">>>(
+//                fifo_cap, gallery, cfg);            // Obj ctor args forwarded
+//   net.add("identity_matcher", n);
+//
+// The wrapper mirrors VariantNodeWrapper's channel plumbing exactly; only the
+// underlying node type (PoolObjectNode, holding Obj&) differs.
+
+#include "../channel.hpp"
+#include "../node.hpp"
+#include "../variant_node.hpp"
+
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <tuple>
+#include <typeindex>
+#include <utility>
+#include <vector>
+
+namespace kpn {
+
+template<typename Obj, typename Variant,
+         typename InputTag  = in<>,
+         typename OutputTag = out<>>
+class ObjectVariantNodeWrapper;
+
+template<typename Obj, typename Variant,
+         fixed_string... InNames, fixed_string... OutNames>
+class ObjectVariantNodeWrapper<Obj, Variant, in<InNames...>, out<OutNames...>>
+    : public IVariantNode<Variant>
+{
+    using NodeT = ObjectNode<Obj, in<InNames...>, out<OutNames...>>;
+
+public:
+    using args_tuple   = typename NodeT::args_tuple;
+    using return_tuple = typename NodeT::return_tuple;
+
+    static constexpr std::size_t n_in  = NodeT::input_count;
+    static constexpr std::size_t n_out = NodeT::output_count;
+
+    // Owns the functor; forwards remaining args to Obj's constructor.
+    template<typename... ObjArgs>
+    explicit ObjectVariantNodeWrapper(std::size_t fifo_capacity, ObjArgs&&... obj_args)
+        : obj_(std::forward<ObjArgs>(obj_args)...)
+        , node_(obj_, fifo_capacity)
+        , in_channels_(n_in)
+        , out_channels_(n_out)
+        , out_type_indices_(n_out, std::type_index(typeid(void)))
+    {
+        init_inputs(std::make_index_sequence<n_in>{}, fifo_capacity);
+        init_out_types(std::make_index_sequence<n_out>{});
+    }
+
+    // Access the owned functor so callers can invoke its runtime setters (e.g. to
+    // change a threshold on a persistent pipeline without rebuilding the node).
+    Obj& functor() { return obj_; }
+
+    // ── INode ─────────────────────────────────────────────────────────────────
+    void start() override { node_.start(); }
+    void stop()  override { node_.stop(); }
+    bool running() const override { return node_.running(); }
+    const NodeStats& stats() const override { return node_.stats(); }
+    void set_name(std::string name) override { node_.set_name(std::move(name)); }
+    NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
+        return node_.node_snapshot(name, elapsed_s);
+    }
+
+    // ── IVariantNode ──────────────────────────────────────────────────────────
+    std::size_t input_count()  const override { return n_in; }
+    std::size_t output_count() const override { return n_out; }
+
+    std::type_index input_type(std::size_t i) const override {
+        return in_channels_[i]->type_index();
+    }
+    std::type_index output_type(std::size_t i) const override {
+        return out_type_indices_[i];
+    }
+
+    std::shared_ptr<IVariantChannel<Variant>> input_channel(std::size_t i) override {
+        return in_channels_[i];
+    }
+
+    void set_output_channel(std::size_t i,
+                            std::shared_ptr<IVariantChannel<Variant>> ch) override {
+        set_output_impl(i, std::move(ch), std::make_index_sequence<n_out>{});
+    }
+
+private:
+    template<std::size_t... Is>
+    void init_inputs(std::index_sequence<Is...>, std::size_t cap) {
+        ((init_one_input<Is>(cap)), ...);
+    }
+
+    template<std::size_t I>
+    void init_one_input(std::size_t cap) {
+        using T = std::tuple_element_t<I, args_tuple>;
+        auto shared_ch = std::make_shared<Channel<T>>(cap);
+        node_.template set_input_channel<I>(shared_ch);
+        in_channels_[I] = std::make_shared<VariantChannel<T, Variant>>(std::move(shared_ch));
+    }
+
+    template<std::size_t... Is>
+    void init_out_types(std::index_sequence<Is...>) {
+        ((out_type_indices_[Is] =
+              std::type_index(typeid(std::tuple_element_t<Is, return_tuple>))), ...);
+    }
+
+    template<std::size_t... Is>
+    void set_output_impl(std::size_t port,
+                         std::shared_ptr<IVariantChannel<Variant>> ch,
+                         std::index_sequence<Is...>) {
+        bool matched = false;
+        ((Is == port && (set_output_at<Is>(std::move(ch)), matched = true)), ...);
+        if (!matched)
+            throw std::out_of_range("set_output_channel: port index out of range");
+    }
+
+    template<std::size_t I>
+    void set_output_at(std::shared_ptr<IVariantChannel<Variant>> ch) {
+        using T = std::tuple_element_t<I, return_tuple>;
+        auto* typed = dynamic_cast<VariantChannel<T, Variant>*>(ch.get());
+        if (!typed)
+            throw std::runtime_error(
+                "set_output_channel: type mismatch at output port " + std::to_string(I));
+        node_.template set_output_channel<I>(typed->raw_ptr());
+        out_channels_[I] = std::move(ch);
+    }
+
+    Obj   obj_;    // owned; node_ holds Obj& — declaration order keeps obj_ alive first
+    NodeT node_;
+    std::vector<std::shared_ptr<IVariantChannel<Variant>>> in_channels_;
+    std::vector<std::shared_ptr<IVariantChannel<Variant>>> out_channels_;
+    std::vector<std::type_index>                           out_type_indices_;
+};
+
+} // namespace kpn

include/kpn/scheduler.hpp

@@ -69,6 +69,9 @@ public:
             while (!q->pq.empty()) q->pq.pop();
             total_.fetch_sub(discarded, std::memory_order_relaxed);
         }
+        // Lock cv_mx_ before notifying so the stop signal can't be lost in the
+        // gap between a worker's predicate check and its wait() (see submit()).
+        { std::lock_guard<std::mutex> lk(cv_mx_); }
         cv_.notify_all();
         for (auto& t : workers_) if (t.joinable()) t.join();
         workers_.clear();
@@ -91,6 +94,12 @@ public:
         }
         total_.fetch_add(1, std::memory_order_relaxed);
         submitted_.fetch_add(1, std::memory_order_relaxed);
+        // Synchronize with worker_loop's predicate evaluation: taking cv_mx_
+        // here guarantees a worker is either before its predicate check (and
+        // will observe total_ > 0) or already blocked in wait() (and will be
+        // woken). Without this, notify_one() can slip into the gap between the
+        // worker's predicate check and its wait(), and be lost — a deadlock.
+        { std::lock_guard<std::mutex> lk(cv_mx_); }
         cv_.notify_one();
     }
 
@@ -157,8 +166,12 @@ private:
         active_.fetch_sub(1, std::memory_order_relaxed);
         // Notify drain() if this was the last in-flight task.
         // acq_rel ensures the decrement is visible before any drain() load.
-        if (total_.fetch_sub(1, std::memory_order_acq_rel) == 1)
+        // Lock drain_mx_ before notifying to avoid a lost wakeup against
+        // drain()'s predicate check (same hazard as submit()/cv_mx_).
+        if (total_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
+            { std::lock_guard<std::mutex> lk(drain_mx_); }
             drain_cv_.notify_all();
+        }
     }
 
     void worker_loop(std::size_t id) {

include/kpn/static_network.hpp

@@ -112,6 +112,16 @@ public:
     void start() override {
         stop_flag_ = false;
         start_time_ = clock_t::now();
+        if (event_handler_) {
+            for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) {
+                auto* node = user_nodes_topo_[i];
+                const auto& n = user_node_names_[i];
+                node->set_network_overflow_callback(
+                    [this, n](auto ts) { event_handler_(n, NodeEvent::Overflow, ts); });
+                node->set_network_closed_callback(
+                    [this, n](auto ts) { event_handler_(n, NodeEvent::Closed, ts); });
+            }
+        }
         for (auto* n : user_nodes_topo_)  n->start();
         for (auto* n : fanout_nodes_ptr_) n->start();
 #ifdef KPN_WEB_DEBUG
@@ -165,6 +175,12 @@ public:
         return {n, 0, 0, 0, 0, 0, 0, 0};
     }
 
+    using EventHandler =
+        std::function<void(std::string_view node_name, NodeEvent,
+                           std::chrono::steady_clock::time_point)>;
+
+    void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
+
 #ifdef KPN_WEB_DEBUG
     void set_web_debug_port(uint16_t port) { web_debug_port_ = port; }
     // Called by DebugHub::register_network() so the hub owns the debug server.
@@ -203,6 +219,24 @@ public:
 
     FanoutStorage& fanouts_storage() { return *fanouts_; }
 
+    // Make every node in this network push losslessly (block until the consumer
+    // drains) instead of dropping on a full channel.  Includes the fanout nodes
+    // make_network() inserts automatically, which is the part user code cannot
+    // reach: they are unnamed, and they drop silently per-output, so a network
+    // whose own nodes are all lossless can still lose items at a fanout.
+    //
+    // Only safe when every consumer eventually drains.  A branch that can stall
+    // indefinitely — a display node nobody is servicing, say — will block the
+    // whole pipeline through backpressure.  Call before start().
+    void set_lossless(bool on = true) {
+        for (auto* n : user_nodes_topo_)  if (n) n->set_lossless_output(on);
+        for (auto* n : fanout_nodes_ptr_) if (n) n->set_lossless_output(on);
+    }
+
+    // Block until every channel is empty.  Useful before stop() so work already
+    // in flight completes rather than being discarded at teardown.
+    void drain() const { drain_all_channels(); }
+
 private:
     struct Snapshots {
         std::vector<NodeSnapshot>     nodes;
@@ -259,6 +293,7 @@ private:
     std::vector<std::unique_ptr<IChannelProbe>>         channel_probes_;
     std::vector<std::pair<std::string, IResourceProbe*>> resource_probes_;
     std::vector<std::pair<std::string, IPoolProbe*>>     pool_probes_;
+    EventHandler                                        event_handler_;
     clock_t::time_point                                 start_time_;
 #ifdef KPN_WEB_DEBUG
     uint16_t                                            web_debug_port_{9090};

include/kpn/variant_node.hpp

@@ -55,6 +55,8 @@ class IVariantChannel {
 public:
     virtual ~IVariantChannel() = default;
     virtual void            push(Variant v) = 0;
+    // Lossless push with backpressure (waits instead of dropping when full).
+    virtual void            push_blocking(Variant v) = 0;
     virtual Variant         pop()           = 0;
     virtual std::type_index type_index()    const = 0;
     virtual std::string     type_name()     const = 0;
@@ -76,6 +78,9 @@ public:
     void push(Variant v) override {
         channel_->push(std::get<T>(std::move(v)));
     }
+    void push_blocking(Variant v) override {
+        channel_->push_blocking(std::get<T>(std::move(v)));
+    }
     Variant pop() override {
         return Variant{ channel_->pop() };
     }

mkdocs.yml

@@ -0,0 +1,50 @@
+site_name: KPN++
+site_description: A C++20 Kahn Process Network library
+repo_url: https://gitea.tourolle.paris/dtourolle/KPN
+repo_name: dtourolle/KPN
+
+theme:
+  name: material
+  palette:
+    - scheme: slate
+      primary: indigo
+      accent: indigo
+  features:
+    - navigation.tabs
+    - navigation.sections
+    - navigation.top
+    - content.code.copy
+    - content.code.annotate
+
+nav:
+  - Home: index.md
+  - Getting Started: getting-started.md
+  - Concepts:
+    - Nodes: nodes.md
+    - Networks: network.md
+    - Channels: channels.md
+  - Error Handling & Events: error-handling.md
+  - Advanced:
+    - Static Networks: static-network.md
+    - Shared Resources: shared-resource.md
+    - Fan-out & Routing: fanout.md
+  - Examples: examples.md
+
+markdown_extensions:
+  - admonition
+  - toc:
+      permalink: true
+  - pymdownx.highlight:
+      anchor_linenums: true
+      line_spans: __span
+      pygments_lang_class: true
+  - pymdownx.inlinehilite
+  - pymdownx.superfences
+  - pymdownx.tabbed:
+      alternate_style: true
+  - pymdownx.snippets:
+      base_path: ['.']
+      check_paths: true
+  - pymdownx.details
+  - attr_list
+  - md_in_html

scripts/render_readme.py

@@ -6,9 +6,9 @@ Directive format (in README.md.in):
   <!-- @snippet path/to/file.cpp snippet_name -->
 
 Snippet tags (in C++ source files):
-  // [snippet: snippet_name]
+  // --8<-- [start:snippet_name]
   ...content...
-  // [/snippet: snippet_name]
+  // --8<-- [end:snippet_name]
 
 In Python files use # instead of //.
 """
@@ -33,8 +33,8 @@ def comment_prefix(path: Path) -> str:
 
 def extract_snippet(file_path: Path, name: str) -> str:
     prefix = comment_prefix(file_path)
-    open_tag  = f'{prefix} [snippet: {name}]'
-    close_tag = f'{prefix} [/snippet: {name}]'
+    open_tag  = f'{prefix} --8<-- [start:{name}]'
+    close_tag = f'{prefix} --8<-- [end:{name}]'
 
     text = file_path.read_text()
     lines = text.splitlines()

src/network.cpp

@@ -1,4 +1,4 @@
-// network.cpp — orchestrator/watchdog implementation details.
+// network.cpp — orchestrator/watchdog implementation details. (CI: pipeline trigger)
 // Most of the Network class is header-only (template-heavy).
 // Non-template implementation lives here once the watchdog grows
 // beyond the stub in network.hpp.

tests/CMakeLists.txt

@@ -43,6 +43,35 @@ target_link_libraries(kpn_tests PRIVATE
     GTest::gtest
 )
 
+# ── Channel stress suite (separate executable) ────────────────────────────────
+# Contended SPSC tests for the lock-free Channel<T>. Kept out of kpn_tests
+# because each case runs many reps / tens of thousands of items and is slow.
+# Most valuable under -DKPN_SANITIZER=thread, but correct (and run) without it.
+add_executable(kpn_tests_stress test_channel_stress.cpp)
+target_link_libraries(kpn_tests_stress PRIVATE kpn Catch2::Catch2WithMain)
+
+# ── Sanitizer flags ───────────────────────────────────────────────────────────
+# kpn_sanitizer_flags() is defined in the top-level CMakeLists and is a no-op
+# unless -DKPN_SANITIZER=... is set. Sanitizer must be on both compile and link.
+kpn_sanitizer_flags(_kpn_san)
+if(_kpn_san)
+    foreach(_t kpn_tests kpn_tests_stress)
+        target_compile_options(${_t} PRIVATE ${_kpn_san})
+        target_link_options(${_t}    PRIVATE ${_kpn_san})
+    endforeach()
+endif()
+
 include(CTest)
 include(Catch)
-catch_discover_tests(kpn_tests)
+
+# DISCOVERY_MODE PRE_TEST defers test enumeration to `ctest` run time. The
+# default (POST_BUILD) runs each test binary during the build to list its
+# cases — which fails a sanitizer build: a TSan/ASan binary needs a fixed
+# address-space layout and aborts on startup ("unexpected memory mapping")
+# under the container's ASLR, breaking the build before any test runs. The
+# tsan.yaml job invokes the binaries directly (not via ctest), so deferring
+# discovery costs nothing there and keeps `ctest` working for normal builds.
+catch_discover_tests(kpn_tests DISCOVERY_MODE PRE_TEST)
+# Register the stress suite under its own label so CI can run / time it
+# separately from the fast unit tests.
+catch_discover_tests(kpn_tests_stress DISCOVERY_MODE PRE_TEST PROPERTIES LABELS "stress")

tests/test_channel.cpp

@@ -175,3 +175,66 @@ TEST_CASE("bandwidth_mbs returns 0 when elapsed_s is zero or negative", "[channe
     REQUIRE(snap.bandwidth_mbs(0.0) == 0.0);
     REQUIRE(snap.bandwidth_mbs(-1.0) == 0.0);
 }
+
+TEST_CASE("push_sentinel never overflows even on a full channel", "[channel][sentinel]") {
+    Channel<int> ch(2);
+    ch.push(1);
+    ch.push(2);   // channel full — a plain push(3) would throw ChannelOverflowError
+
+    // The sentinel is stored out-of-band, so it neither throws nor blocks the
+    // caller — the exact property an EOF token needs under backpressure. This
+    // returns immediately with the ring still full.
+    REQUIRE(ch.push_sentinel(99));
+    REQUIRE(ch.size() == 2);   // sentinel did not consume ring capacity
+}
+
+TEST_CASE("push_sentinel is delivered after all ring data, in order", "[channel][sentinel]") {
+    Channel<int> ch(4);
+    ch.push(1);
+    ch.push(2);
+    ch.push_sentinel(99);   // enqueue EOF while data is still buffered
+
+    // Data drains first; the sentinel arrives only once the ring is empty.
+    REQUIRE(ch.pop() == 1);
+    REQUIRE(ch.pop() == 2);
+    REQUIRE(ch.pop() == 99);
+}
+
+TEST_CASE("push_sentinel wakes a blocked pop", "[channel][sentinel]") {
+    Channel<int> ch(2);   // empty
+    std::thread producer([&] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        ch.push_sentinel(99);   // must wake a consumer parked on an empty ring
+    });
+    REQUIRE(ch.pop() == 99);
+    producer.join();
+}
+
+TEST_CASE("approx_size counts a pending sentinel so consumers stay schedulable",
+          "[channel][sentinel]") {
+    Channel<int> ch(4);
+    REQUIRE(ch.approx_size() == 0);
+    ch.push_sentinel(99);
+    // Node readiness checks call approx_size(); it must report the out-of-band
+    // sentinel as consumable work even though it holds no ring slot.
+    REQUIRE(ch.approx_size() == 1);
+    REQUIRE(ch.size() == 0);       // ...but the ring itself is still empty
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));
+    REQUIRE(out == 99);
+    REQUIRE(ch.approx_size() == 0);
+}
+
+TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty",
+          "[channel][sentinel]") {
+    Channel<int> ch(2);
+    ch.push(1);
+    ch.push_sentinel(99);
+
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));   // ring data first
+    REQUIRE(out == 1);
+    REQUIRE(ch.try_pop_now(out));   // then the sentinel
+    REQUIRE(out == 99);
+    REQUIRE_FALSE(ch.try_pop_now(out));   // nothing left
+}

tests/test_channel_stress.cpp

@@ -0,0 +1,281 @@
+// Contended stress tests for the lock-free SPSC Channel<T>.
+//
+// The other channel tests (test_channel.cpp) are single-threaded or use a
+// single 20 ms sleep to order two threads — they never actually contend on the
+// ring, so they exercise neither the memory-ordering pairing nor the
+// spin/futex/lost-wakeup logic in pop().
+//
+// These tests are written to be run under ThreadSanitizer:
+//
+//   cmake -B build -DKPN_SANITIZER=thread -DKPN_BUILD_EXAMPLES=OFF -DKPN_BUILD_PYTHON=OFF
+//   cmake --build build --target kpn_tests_tsan
+//   ./build/tests/kpn_tests_tsan
+//
+// They are also valid (and meaningful) without a sanitizer: the value/sequence
+// assertions catch lost or duplicated items regardless of build flags. TSan
+// adds detection of the underlying data race even on runs where the race did
+// not corrupt observable state.
+//
+// Channel<T> is SPSC: exactly one producer thread and one consumer thread per
+// channel. Every scenario below honours that contract.
+
+#include <catch2/catch_test_macros.hpp>
+#include <atomic>
+#include <chrono>
+#include <kpn/channel.hpp>
+#include <thread>
+#include <vector>
+
+using namespace kpn;
+using namespace std::chrono_literals;
+
+namespace {
+
+// Repeat each scenario enough times that rare interleavings (spin window just
+// missing / just catching the next push, disable landing inside the futex
+// wait) actually occur across a run. Kept modest so a TSan run stays minutes,
+// not hours.
+constexpr int kReps = 200;
+
+} // namespace
+
+TEST_CASE("SPSC: every pushed item is popped exactly once, in order", "[channel][stress]") {
+    // Small capacity forces frequent full/empty transitions, so both the
+    // producer's overflow-retry and the consumer's spin->futex path are hit
+    // many times. The producer retries on overflow rather than dropping, so
+    // the consumer must observe a strictly contiguous 0..N-1 sequence.
+    constexpr int N = 50'000;
+    Channel<int> ch(/*capacity=*/4, /*spin_count=*/16);
+
+    std::thread producer([&] {
+        for (int i = 0; i < N; ++i) {
+            for (;;) {
+                try { ch.push(i); break; }
+                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+            }
+        }
+    });
+
+    int expected = 0;
+    bool in_order = true;
+    for (int i = 0; i < N; ++i) {
+        int v = ch.pop();
+        if (v != expected) in_order = false;
+        ++expected;
+    }
+    producer.join();
+
+    REQUIRE(in_order);
+    REQUIRE(expected == N);
+    REQUIRE(ch.size() == 0);
+}
+
+TEST_CASE("SPSC: tight empty<->non-empty transitions exercise spin/futex boundary",
+          "[channel][stress]") {
+    // spin_count=0 forces every empty pop() straight into atomic::wait, so this
+    // hammers the lost-wakeup guard (snapshot wake_, re-check tail_, then wait).
+    // The producer pushes one item then waits to go empty again, maximising the
+    // number of empty->non-empty edges relative to item count.
+    constexpr int N = 20'000;
+    Channel<int> ch(/*capacity=*/2, /*spin_count=*/0);
+
+    std::thread producer([&] {
+        for (int i = 0; i < N; ++i) {
+            for (;;) {
+                try { ch.push(i); break; }
+                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+            }
+        }
+    });
+
+    long sum = 0;
+    for (int i = 0; i < N; ++i) sum += ch.pop();
+    producer.join();
+
+    // Sum of 0..N-1 — detects any lost or duplicated item.
+    REQUIRE(sum == static_cast<long>(N) * (N - 1) / 2);
+}
+
+TEST_CASE("SPSC: disable() while consumer is blocked in pop() unblocks cleanly",
+          "[channel][stress]") {
+    // The data race of record: consumer blocked in pop() (spinning or parked in
+    // the futex) while the owner thread calls disable(). pop() must observe the
+    // close and throw ChannelClosedError — it must not hang and must not read
+    // past the ring. Repeated so disable() lands at many points in pop()'s loop.
+    for (int rep = 0; rep < kReps; ++rep) {
+        Channel<int> ch(/*capacity=*/4, /*spin_count=*/8);
+        std::atomic<bool> threw{false};
+        std::atomic<bool> finished{false};
+
+        std::thread consumer([&] {
+            try {
+                ch.pop();              // empty channel: will block
+            } catch (const ChannelClosedError&) {
+                threw.store(true, std::memory_order_relaxed);
+            }
+            finished.store(true, std::memory_order_relaxed);
+        });
+
+        // Give the consumer a chance to reach the wait, then close.
+        std::this_thread::sleep_for(50us);
+        ch.disable();
+
+        consumer.join();
+        REQUIRE(finished.load());
+        REQUIRE(threw.load());
+    }
+}
+
+TEST_CASE("SPSC: producer racing a disable() never throws and never hangs",
+          "[channel][stress]") {
+    // Mirror of the above from the producer side: push() racing disable() must
+    // either enqueue or silently drop, never throw ChannelClosedError and never
+    // wedge. Overflow is still a legal outcome (full accepting channel) and is
+    // tolerated here.
+    for (int rep = 0; rep < kReps; ++rep) {
+        Channel<int> ch(/*capacity=*/8, /*spin_count=*/8);
+        std::atomic<bool> bad{false};
+
+        std::thread producer([&] {
+            for (int i = 0; i < 1000; ++i) {
+                try { ch.push(i); }
+                catch (const ChannelOverflowError&) { /* legal: full */ }
+                catch (...) { bad.store(true, std::memory_order_relaxed); break; }
+            }
+        });
+
+        std::this_thread::sleep_for(20us);
+        ch.disable();           // owner closes mid-stream
+        producer.join();
+
+        REQUIRE_FALSE(bad.load());
+    }
+}
+
+TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
+          "[channel][stress]") {
+    // The empty->non-empty callback ([channel.hpp] was_empty branch) is read by
+    // the consumer-side notification path. Run it under contention to make sure
+    // the was_empty detection isn't torn by a concurrent pop().
+    Channel<int> ch(/*capacity=*/4, /*spin_count=*/4);
+    std::atomic<int> callbacks{0};
+    ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); });
+
+    constexpr int N = 10'000;
+    std::thread producer([&] {
+        for (int i = 0; i < N; ++i) {
+            for (;;) {
+                try { ch.push(i); break; }
+                catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+            }
+        }
+    });
+
+    for (int i = 0; i < N; ++i) (void)ch.pop();
+    producer.join();
+
+    // At least one transition, at most one per item; mainly we assert the run
+    // completed without TSan flagging a race on push_callback_/was_empty.
+    REQUIRE(callbacks.load() >= 1);
+    REQUIRE(callbacks.load() <= N);
+}
+
+// Ordering contract of the out-of-band sentinel under contention.
+//
+// push_sentinel() publishes has_eof_ (release) after the producer's N ring
+// pushes; a consumer that observes has_eof_ (acquire) therefore also observes
+// every value pushed before it. Both pop() and try_pop_now() only surface the
+// sentinel once the ring is *freshly* observed empty, so the sentinel is the
+// strictly last item received — it never jumps ahead of a ring value pushed
+// before it. These tests treat the sentinel as a hard "last message" barrier
+// (the consumer stops draining the moment it sees it) and assert that all N
+// values arrived, in a contiguous 0..N-1 sequence, before it.
+//
+// Regression guard: an earlier version of pop() checked emptiness against a
+// stale tail_ snapshot from the top of its loop, so under load the sentinel
+// could surface with a few real values still queued — breaking in_order /
+// values==N here. Under TSan these also cover the has_eof_/eof_value_
+// acquire/release handshake and the spin/futex wakeup on push_sentinel().
+
+TEST_CASE("SPSC: sentinel is strictly last, after every value (blocking pop)",
+          "[channel][stress]") {
+    constexpr int N        = 20'000;
+    constexpr int SENTINEL = -1;
+
+    for (int rep = 0; rep < kReps; ++rep) {
+        // Small ring + tiny spin window so the ring is frequently empty exactly
+        // when the sentinel is published — the interleaving under test.
+        Channel<int> ch(/*capacity=*/4, /*spin_count=*/8);
+
+        std::thread producer([&] {
+            for (int i = 0; i < N; ++i) {
+                for (;;) {
+                    try { ch.push(i); break; }
+                    catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+                }
+            }
+            ch.push_sentinel(SENTINEL);   // must-deliver, never overflows/blocks
+        });
+
+        int  expected = 0;
+        bool in_order = true;
+        bool saw_sentinel = false;
+        // Treat the sentinel as EOF: stop draining the instant it appears.
+        for (;;) {
+            int v = ch.pop();
+            if (v == SENTINEL) { saw_sentinel = true; break; }
+            if (v != expected) in_order = false;
+            ++expected;
+        }
+        producer.join();
+
+        REQUIRE(saw_sentinel);
+        REQUIRE(in_order);
+        REQUIRE(expected == N);        // all N values received before the sentinel
+        REQUIRE(ch.size() == 0);
+        REQUIRE(ch.approx_size() == 0);
+    }
+}
+
+TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)",
+          "[channel][stress]") {
+    // The pool-node consume path is try_pop_now(), not pop(): it must surface
+    // the out-of-band sentinel only once the ring is freshly observed empty.
+    // The consumer spins with no sleeps, racing the producer at full tilt
+    // across the empty-ring boundary where take_sentinel() is reached.
+    constexpr int N        = 20'000;
+    constexpr int SENTINEL = -1;
+
+    for (int rep = 0; rep < kReps; ++rep) {
+        Channel<int> ch(/*capacity=*/4, /*spin_count=*/0);
+
+        std::thread producer([&] {
+            for (int i = 0; i < N; ++i) {
+                for (;;) {
+                    try { ch.push(i); break; }
+                    catch (const ChannelOverflowError&) { std::this_thread::yield(); }
+                }
+            }
+            ch.push_sentinel(SENTINEL);
+        });
+
+        int  expected = 0;
+        bool in_order = true;
+        bool saw_sentinel = false;
+        int  v;
+        for (;;) {
+            if (!ch.try_pop_now(v)) { std::this_thread::yield(); continue; }
+            if (v == SENTINEL) { saw_sentinel = true; break; }
+            if (v != expected) in_order = false;
+            ++expected;
+        }
+        producer.join();
+
+        REQUIRE(saw_sentinel);
+        REQUIRE(in_order);
+        REQUIRE(expected == N);
+        // Sentinel held no ring slot; once taken the channel is fully empty.
+        REQUIRE(ch.size() == 0);
+        REQUIRE(ch.approx_size() == 0);
+    }
+}

tests/test_pool_node.cpp

@@ -4,6 +4,7 @@
 #include <kpn/interrupt_node.hpp>
 #include <atomic>
 #include <chrono>
+#include <mutex>
 #include <thread>
 
 using namespace kpn;
@@ -239,3 +240,225 @@ TEST_CASE("interrupt node: trigger after stop is ignored", "[interrupt_node]") {
     REQUIRE(out_ch.approx_size() == 0);
     pool->stop();
 }
+
+// ── Overflow callback ─────────────────────────────────────────────────────────
+
+TEST_CASE("pool node overflow callback fires on full output channel", "[pool_node][overflow]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto node = make_pool_node<double_it>(pool);
+    // Pre-fill a tiny channel so every node push overflows.
+    Channel<int> full_ch(1);
+    full_ch.push(99);
+    node.set_output_channel<0>(&full_ch);
+
+    std::atomic<int> overflow_count{0};
+    node.set_overflow_callback([&](auto) { overflow_count.fetch_add(1); });
+
+    node.start();
+    node.input_channel<0>().push(1);
+    node.input_channel<0>().push(2);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    node.stop();
+    pool->stop();
+
+    REQUIRE(overflow_count.load() > 0);
+}
+
+TEST_CASE("pool node overflow callback is independent per instance", "[pool_node][overflow]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto nodeA = make_pool_node<double_it>(pool);
+    auto nodeB = make_pool_node<double_it>(pool);
+
+    std::atomic<int> a_overflows{0}, b_overflows{0};
+    nodeA.set_overflow_callback([&](auto) { a_overflows.fetch_add(1); });
+
+    Channel<int> full_ch(1);
+    full_ch.push(0);
+    nodeA.set_output_channel<0>(&full_ch);
+
+    Channel<int> ok_ch(20);
+    nodeB.set_output_channel<0>(&ok_ch);
+
+    nodeA.start();
+    nodeB.start();
+
+    nodeA.input_channel<0>().push(1);
+    nodeA.input_channel<0>().push(2);
+    nodeB.input_channel<0>().push(10);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    nodeA.stop();
+    nodeB.stop();
+    pool->stop();
+
+    REQUIRE(a_overflows.load() > 0);
+    REQUIRE(b_overflows.load() == 0);
+}
+
+TEST_CASE("interrupt node overflow callback fires on full output", "[interrupt_node][overflow]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    g_interrupt_counter.store(0);
+    auto node = make_interrupt_node<interrupt_produce>(pool, out<>{});
+
+    Channel<int> full_ch(1);
+    full_ch.push(99);
+    node.set_output_channel<0>(&full_ch);
+
+    std::atomic<int> overflow_count{0};
+    node.set_overflow_callback([&](auto) { overflow_count.fetch_add(1); });
+
+    node.start();
+    auto trigger = node.get_trigger();
+    trigger(); trigger(); trigger();
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    node.stop();
+    pool->stop();
+
+    REQUIRE(overflow_count.load() > 0);
+}
+
+// ── self_stop: disable inputs + outputs on crash ──────────────────────────────
+
+static int always_throw(int) { throw std::runtime_error("node crashed"); return 0; }
+
+TEST_CASE("pool node self_stop disables output on crash so downstream sees closed", "[pool_node][self_stop]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto node = make_pool_node<always_throw>(pool, 5);
+    Channel<int> out_ch(10);
+    node.set_output_channel<0>(&out_ch);
+
+    node.start();
+    node.input_channel<0>().push(1);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    REQUIRE_FALSE(out_ch.is_accepting());
+
+    node.stop();
+    pool->stop();
+}
+
+TEST_CASE("pool node self_stop disables input on crash", "[pool_node][self_stop]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto node = make_pool_node<always_throw>(pool, 5);
+    Channel<int> out_ch(5);
+    node.set_output_channel<0>(&out_ch);
+
+    node.start();
+    node.input_channel<0>().push(1);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    REQUIRE_FALSE(node.input_channel<0>().is_accepting());
+
+    node.stop();
+    pool->stop();
+}
+
+TEST_CASE("pool node closed callback fires on self_stop from crash", "[pool_node][self_stop]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto node = make_pool_node<always_throw>(pool, 5);
+    Channel<int> out_ch(5);
+    node.set_output_channel<0>(&out_ch);
+
+    std::atomic<bool> closed_fired{false};
+    node.set_closed_callback([&](auto) { closed_fired.store(true); });
+
+    node.start();
+    node.input_channel<0>().push(1);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    REQUIRE(closed_fired.load());
+
+    node.stop();
+    pool->stop();
+}
+
+// ── Network-level event callbacks ─────────────────────────────────────────────
+
+TEST_CASE("network_overflow_callback fires on overflow", "[pool_node][network]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto node = make_pool_node<double_it>(pool);
+
+    Channel<int> full_ch(1);
+    full_ch.push(0);
+    node.set_output_channel<0>(&full_ch);
+
+    std::atomic<int> net_overflows{0};
+    node.set_network_overflow_callback([&](auto) { net_overflows.fetch_add(1); });
+
+    node.start();
+    node.input_channel<0>().push(1);
+    node.input_channel<0>().push(2);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    node.stop();
+    pool->stop();
+
+    REQUIRE(net_overflows.load() > 0);
+}
+
+TEST_CASE("network_closed_callback fires on crash", "[pool_node][network]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto node = make_pool_node<always_throw>(pool);
+    Channel<int> out_ch(5);
+    node.set_output_channel<0>(&out_ch);
+
+    std::atomic<bool> net_closed{false};
+    node.set_network_closed_callback([&](auto) { net_closed.store(true); });
+
+    node.start();
+    node.input_channel<0>().push(1);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    REQUIRE(net_closed.load());
+
+    node.stop();
+    pool->stop();
+}
+
+TEST_CASE("per-node and network overflow callbacks both fire independently", "[pool_node][network]") {
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    auto node = make_pool_node<double_it>(pool);
+
+    Channel<int> full_ch(1);
+    full_ch.push(0);
+    node.set_output_channel<0>(&full_ch);
+
+    std::atomic<int> per_node{0}, network{0};
+    node.set_overflow_callback([&](auto) { per_node.fetch_add(1); });
+    node.set_network_overflow_callback([&](auto) { network.fetch_add(1); });
+
+    node.start();
+    node.input_channel<0>().push(1);
+    node.input_channel<0>().push(2);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    node.stop();
+    pool->stop();
+
+    REQUIRE(per_node.load() > 0);
+    REQUIRE(network.load() > 0);
+}