ci: fix TSan aborting at init on the nested-LXC runner

The ThreadSanitizer job runs on Docker nested in an unprivileged LXC container, whose kernel randomizes mmap addresses beyond the range TSan's fixed shadow mapping expects. TSan aborted at init with "unexpected memory mapping" before any test ran. Disable ASLR per-process with `setarch -R`, which needs the personality(2) syscall that Docker's default seccomp profile blocks; seccomp=unconfined on the container permits it. Verified on the runner that both are required: setarch -R alone gets EPERM, seccomp alone still aborts, both together run clean. Scoped to the tsan job, which runs only our own test binaries. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
fix: deliver EOF sentinel only when the ring is freshly empty
2026-07-17 20:01:57 +02:00 · 2026-07-14 23:14:45 +02:00 · 2026-07-14 23:14:45 +02:00 · 2026-07-14 23:14:45 +02:00 · 2026-07-04 15:07:27 +02:00 · 2026-07-04 15:07:12 +02:00
50 changed files with 2767 additions and 1294 deletions
@@ -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
@@ -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 }}
@@ -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 }}"
@@ -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
@@ -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 }}
@@ -1,6 +1,8 @@
 # Build output
 build/
+build_test/
 build_debug/
+site/
 # Python
 __pycache__/
 *.py[cod]
@@ -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
@@ -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
@@ -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.
@@ -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.
@@ -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.
@@ -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);
+```
@@ -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"
+```
@@ -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.
+```
@@ -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>());
+```
@@ -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"
+```
@@ -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.
@@ -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).
@@ -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.
@@ -0,0 +1,3 @@
+mkdocs>=1.5
+mkdocs-material>=9.5
+pymdown-extensions>=10.0
@@ -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.
@@ -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.
@@ -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]
 }
@@ -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]
 }
@@ -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]
 }
@@ -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 ────────────────────────────────────────────────────────────

@@ -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));

@@ -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
@@ -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
@@ -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;
 }
@@ -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)
@@ -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';
+}
@@ -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)
@@ -136,6 +136,36 @@ public:
            push_callback_();
    }

+    // 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 +178,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 +236,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 +266,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 +318,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 +336,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};
@@ -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,6 +29,11 @@ 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(); }

@@ -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 ───────────────────────────────────────────────
@@ -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_;
@@ -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,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 {
@@ -189,12 +226,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 +334,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 +393,13 @@ 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;
+        }
        try {
            ch->push(std::move(val));
        } catch (const ChannelOverflowError&) {
@@ -359,16 +420,18 @@ 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_;
+    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 +498,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 {
@@ -490,13 +558,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 +653,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 +699,13 @@ 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;
+        }
        try {
            ch->push(std::move(val));
        } catch (const ChannelOverflowError&) {
@@ -623,17 +714,19 @@ 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_;
+    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) ─────────────────────────────────────────────
@@ -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);
@@ -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) {
@@ -367,6 +377,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)) {
@@ -439,6 +457,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 +519,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"),
@@ -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) {
@@ -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.
@@ -259,6 +275,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};
@@ -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
@@ -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()
@@ -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.
@@ -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")
@@ -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
+}
@@ -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);
+    }
+}
@@ -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);
+}
Author	SHA1	Message	Date
dtourolleandClaude Opus 4.8	ec19137ed9	ci: fix TSan aborting at init on the nested-LXC runner 🚦 CI / changes (pull_request) Successful in 6s Details 🚦 CI / docker (pull_request) Has been skipped Details 🚦 CI / test (pull_request) Successful in 4m32s Details 🚦 CI / tsan (pull_request) Successful in 2m54s Details 🚦 CI / docs (pull_request) Has been skipped Details The ThreadSanitizer job runs on Docker nested in an unprivileged LXC container, whose kernel randomizes mmap addresses beyond the range TSan's fixed shadow mapping expects. TSan aborted at init with "unexpected memory mapping" before any test ran. Disable ASLR per-process with `setarch -R`, which needs the personality(2) syscall that Docker's default seccomp profile blocks; seccomp=unconfined on the container permits it. Verified on the runner that both are required: setarch -R alone gets EPERM, seccomp alone still aborts, both together run clean. Scoped to the tsan job, which runs only our own test binaries. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-17 20:01:57 +02:00
dtourolleandClaude Opus 4.8	a0c4bf580e	fix: deliver EOF sentinel only when the ring is freshly empty 🚦 CI / changes (pull_request) Successful in 4s Details 🚦 CI / docker (pull_request) Has been skipped Details 🚦 CI / test (pull_request) Successful in 7m30s Details 🚦 CI / tsan (pull_request) Failing after 2m21s Details 🚦 CI / docs (pull_request) Has been skipped Details Channel<T>::pop() surfaced the out-of-band sentinel from its empty branch using the tail_ snapshot taken at the top of the loop. Under contention the producer can push more values and the sentinel in the window between that snapshot and take_sentinel(), so pop() could return the sentinel while real values still sat in the ring — the sentinel jumping ahead of values pushed before it. No value was lost (a consumer that keeps draining still receives them, and approx_size() keeps counting them so a PoolNode reschedules), but a consumer treating the sentinel as a hard "last message" barrier would act on EOF early. Re-confirm emptiness against a fresh tail_ load before taking the sentinel. Costs one acquire-load on the empty-ring path only; never runs in steady state. The spin and post-spin takes already reload tail_ on the line above them; try_pop_now() already reads tail_ fresh in the same branch — both were correct and are unchanged. The two sentinel stress cases now assert the strict "sentinel is last, after every value" ordering (previously relaxed to avoid the flake this fixes). Verified TSan-clean (2606 assertions, no data races) over repeated runs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-14 23:14:45 +02:00
dtourolleandClaude Opus 4.8	3ac2242df1	docs: rewrite SPEC.md to match the implemented library The spec had drifted far from the code. Key corrections: - Execution model is reactive (PoolNode submits fire_once() to a ThreadPool when inputs are ready), not one blocking thread per node - Channel<T> is a lock-free SPSC ring buffer (atomic wait/notify + spin-before-sleep), not a mutex+CV queue - Remove latch<> ports (never implemented) - NodeErrorHandler returns bool (skip vs stop); per-node - Document new subsystems: scheduler, InterruptNode, Router/FilterNode, MainThreadNode, SharedResource, DebugHub, diagnostics/stats layer - Update StaticNetwork, Python auto_bind layer, examples 01-16 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-14 23:14:45 +02:00
dtourolleandClaude Opus 4.8	399ee4cf9b	test: add ThreadSanitizer verification for lock-free Channel<T> Add a contended SPSC stress suite (tests/test_channel_stress.cpp) that actually exercises the ring's memory-ordering pairing and spin/futex/ lost-wakeup logic, plus the CMake and CI plumbing to run it under TSan: - KPN_SANITIZER cache var + kpn_sanitizer_flags() helper (no-op when unset) - kpn_tests_stress executable, labelled "stress" for CTest - reusable tsan.yaml workflow (gcc:14 builder image, already ships libtsan) - ci.yaml gains a tsan job on the same code/dockerfile triggers as test Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-14 23:14:45 +02:00
dtourolle	66feb91821	ci: pass docker `push` input as a string, not a boolean 🚦 CI / changes (push) Successful in 6s Details 🚦 CI / docker (push) Successful in 2m57s Details 🚦 CI / test (push) Successful in 8m7s Details 🚦 CI / docs (push) Successful in 6s Details Gitea's act_runner mangles boolean workflow_call/dispatch inputs passed from an expression -- they arrive as false regardless of value. Declare `push` as a string ("true"/"false") and compare with == 'true' so the builder image is pushed on non-PR events again.	2026-07-04 15:07:27 +02:00
dtourolle	903dd4eea5	docs: update README examples table and add documentation link 07_python_network and 08_python_subport now work and run as CI smoke tests, so drop their "(pending)" markers and describe what they actually demonstrate. Also surface the hosted documentation link at the top and re-render README.md from README.md.in.	2026-07-04 15:07:12 +02:00
dtourolle	298c9e770b	examples: run Python examples 07 and 08 as CTest smoke tests Neither Python example was registered as a test, so `ctest -L examples` in CI skipped them entirely -- which is how 08's missing Python node went unnoticed. Add a kpn_python_example() helper (gated on KPN_BUILD_PYTHON) that runs each script with PYTHONPATH pointed at the freshly-built module, so it does not depend on cwd or a hard-coded build/python path, and register 07 and 08. Also del the network in 07 for deterministic teardown.	2026-07-04 14:44:30 +02:00
dtourolle	2b0873b61b	examples: make 08_python_subport run a real Python node The example was named "python subport" but its graph was entirely C++ (ProduceNode -> DoubleItNode); Python only tapped the output, leaving a dangling "#todo: return value to network". Rewrite so the only node in the graph is a pure-Python py_triple, driven from both ends via the subport taps: net.write() injects inputs and net.read() pulls results back, closing the round trip. Also del the network at the end so its callable cycle is reclaimed deterministically.	2026-07-04 14:44:24 +02:00
dtourolle	c4538f03ca	python: fix nanobind Network reference leak via GC type slots The Python Network holds each PyNode's callable, forming an uncollectable instance -> callable -> globals() -> instance cycle that tripped nanobind's leak check at interpreter shutdown. Implement tp_traverse/tp_clear type slots on the Network binding so Python's cyclic collector can see through the C++-held callables and break the cycle. PyNode exposes its callable; PyNetwork visits and clears them. Wired into both binding sites (auto_bind and the legacy register_py_network).	2026-07-04 14:43:47 +02:00
dtourolleandClaude Opus 4.8	949c8134ef	Ignore build_test/ (out-of-tree test build dir) 🚦 CI / changes (push) Successful in 3s Details 🚦 CI / docker (push) Has been skipped Details 🚦 CI / test (push) Has been skipped Details 🚦 CI / docs (push) Has been skipped Details Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-04 11:02:54 +02:00
dtourolleandClaude Opus 4.8	19f5a2b0ae	Deliver EOF sentinels out-of-band to prevent teardown deadlock 🚦 CI / changes (push) Successful in 18s Details 🚦 CI / docker (push) Has been skipped Details 🚦 CI / docs (push) Has been cancelled Details 🚦 CI / test (push) Has been cancelled Details Channel::push() drops values on overflow (the intended backpressure policy for data), and PoolNode swallows the resulting ChannelOverflowError. For a control sentinel like EOF this is fatal: a single dropped EOF under backpressure wedges every downstream pop() forever, so the pipeline never tears down. Deliver sentinels out-of-band instead. Channel::push_sentinel() stores the token in a dedicated slot that does not consume ring capacity, so it can never overflow and — crucially — never blocks the caller. That non-blocking property 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. The consumer's pop()/try_pop_now() drain the ring first, then deliver the sentinel, so it always arrives after every value pushed before it. approx_size() (which node readiness checks call) counts a pending sentinel as consumable work, so a channel carrying only a sentinel still schedules its consumer's next fire — without this the token would sit undelivered and the pipeline would still deadlock at teardown. PoolNode/PoolObjectNode route values carrying an eof flag (direct .eof or nested .source.eof) through push_sentinel via a SFINAE-safe is_sentinel_value trait; all other values keep the existing lossy throwing push. The trait compiles to false for types without an eof convention, so this is a no-op for pipelines that don't use one. Verified end-to-end: scene_analyze now reaches EOF, flushes its output, and exits cleanly instead of hanging. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-04 00:30:40 +02:00
dtourolle	a4de64ea04	Add liscence and prepare for OSS release 🚦 CI / changes (push) Successful in 39s Details 🚦 CI / docker (push) Has been skipped Details 🚦 CI / test (push) Has been skipped Details 🚦 CI / docs (push) Successful in 7s Details	2026-06-28 12:05:56 +02:00
dtourolleandClaude Opus 4.8	7c6a8be2b7	ci: provoke pipeline (touch Dockerfile + code) 🚦 CI / changes (push) Successful in 32s Details 🚦 CI / docker (push) Successful in 1m0s Details 🚦 CI / test (push) Successful in 4m47s Details 🚦 CI / docs (push) Successful in 13s Details Trivial comment changes to exercise the new CI orchestration: the docker job should rebuild+push the builder image first, then test and docs run against the fresh image. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-06-20 09:46:35 +02:00
dtourolle	4c0f1f6923	fix CI 🚦 CI / changes (push) Successful in 33s Details 🚦 CI / docker (push) Has been skipped Details 🚦 CI / test (push) Has been skipped Details 🚦 CI / docs (push) Has been skipped Details	2026-06-20 09:38:26 +02:00
dtourolle	6b52526e44	Auto-build docker when docker file changes. 🚦 CI / changes (push) Failing after 5s Details 🚦 CI / docker (push) Has been skipped Details 🚦 CI / test (push) Has been skipped Details 🚦 CI / docs (push) Has been skipped Details	2026-06-20 09:16:27 +02:00
dtourolle	7cb92a4091	Build docs with a pre-configured docker 🧪 Test / test (push) Successful in 4m41s Details	2026-06-20 08:55:48 +02:00
dtourolle	20668d6955	Fix docker image 🧪 Test / test (push) Successful in 7m31s Details	2026-06-19 22:29:46 +02:00
dtourolle	6f384dc4b5	Added callbacks for node errors and fifo overflow 📚 Docs / deploy (push) Failing after 7s Details 🧪 Test / test (push) Has been cancelled Details Add new doc system which should/might deploy to pages.	2026-06-19 22:26:39 +02:00