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>

.gitea/workflows/ci.yaml

@@ -63,7 +63,8 @@ jobs:
     if: ${{ needs.changes.outputs.dockerfile == 'true' }}
     uses: ./.gitea/workflows/docker.yaml
     with:
-      push: ${{ github.event_name != 'pull_request' }}
+      # 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.
@@ -72,6 +73,13 @@ jobs:
     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') }}

.gitea/workflows/docker.yaml

@@ -4,19 +4,22 @@ name: '🐳 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'
-        type: boolean
-        default: true
+        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'
-        type: boolean
-        default: true
+        description: 'Push the built image to the registry ("true"/"false")'
+        type: string
+        default: 'true'
 
 jobs:
   build:
@@ -48,7 +51,7 @@ jobs:
             .
 
       - name: Push builder image
-        if: ${{ inputs.push }}
+        if: ${{ inputs.push == 'true' }}
         run: |
           docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
           docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:${{ github.sha }}

.gitea/workflows/tsan.yaml

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

.gitignore

@@ -1,5 +1,6 @@
 # Build output
 build/
+build_test/
 build_debug/
 site/
 # Python

CMakeLists.txt

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

Dockerfile.builder

@@ -1,4 +1,4 @@
-# KPN++ Builder Image  (CI: pipeline trigger)
+# 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

LICENSE

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

README.md

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

README.md.in

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

examples/07_python_network/example.py

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

examples/08_python_subport/example.py

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

examples/CMakeLists.txt

@@ -13,6 +13,25 @@ function(kpn_example name)
     )
 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)
 kpn_example(02_named_ports)
 kpn_example(03_multi_output)
@@ -31,7 +50,9 @@ if(KPN_WEB_DEBUG)
     target_link_libraries(14_debug_hub PRIVATE kpn)
     kpn_target_enable_web_debug(14_debug_hub)
 endif()
-# 07 and 08 are Python scripts — no compiled target needed.
+# 07 and 08 are Python scripts — no compiled target, but run as smoke tests.
+kpn_python_example(07_python_network)
+kpn_python_example(08_python_subport)
 
 # 09 requires OpenCV — only build if found
 find_package(OpenCV QUIET COMPONENTS core imgproc highgui videoio)

include/kpn/channel.hpp

@@ -136,6 +136,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};

include/kpn/pool_node.hpp

@@ -22,6 +22,38 @@
 
 namespace kpn {
 
+// ── Sentinel detection ────────────────────────────────────────────────────────
+// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
+// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
+// source Frame) or nested one level under a `.source` member (`v.source.eof`,
+// as on the pipeline's SceneFrame/…/MatchedSceneFrame message types, which wrap
+// the originating Frame). Sentinels are delivered losslessly and non-blockingly
+// via Channel::push_sentinel() instead of the throwing push(), so backpressure
+// can never drop the token that unblocks downstream teardown.
+//
+// Types with neither shape are never treated as sentinels — both traits are
+// SFINAE-safe and the runtime check compiles away to `false` for them, so this
+// stays a no-op for pipelines that don't use an eof convention.
+template<typename T, typename = void>
+struct has_eof_field : std::false_type {};
+template<typename T>
+struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
+    : std::true_type {};
+
+template<typename T, typename = void>
+struct has_source_eof_field : std::false_type {};
+template<typename T>
+struct has_source_eof_field<T,
+    std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
+    : std::true_type {};
+
+template<typename T>
+constexpr bool is_sentinel_value(const T& v) {
+    if constexpr (has_eof_field<T>::value)             return static_cast<bool>(v.eof);
+    else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
+    else                                               return false;
+}
+
 // ── PoolNode ──────────────────────────────────────────────────────────────────
 //
 // Reactive alternative to Node<>. Instead of owning a blocked thread, the node
@@ -361,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&) {
@@ -660,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&) {

include/kpn/python/auto_bind.hpp

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

include/kpn/python/bindings.hpp

@@ -35,6 +35,16 @@ public:
     using VNode    = IVariantNode<Variant>;
     using VChannel = IVariantChannel<Variant>;
 
+    // ── GC support ────────────────────────────────────────────────────────────
+    // Visit every Python object this network transitively holds (currently the
+    // callable of each PyNode). Used by the Network type's tp_traverse slot so
+    // Python's cyclic GC can discover instance → callable → globals() cycles.
+    // Defined out-of-line below, once PyNode is a complete type.
+    template<typename Fn>
+    void visit_python_objects(Fn&& visit) const;
+    // Drop all Python references held by nodes, breaking any cycle (tp_clear).
+    void clear_python_objects();
+
     // ── Builder API ───────────────────────────────────────────────────────────
 
     void add(std::string name, std::shared_ptr<VNode> node) {
@@ -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"),

mkdocs.yml

@@ -1,7 +1,7 @@
 site_name: KPN++
 site_description: A C++20 Kahn Process Network library
-repo_url: https://github.com/yourusername/kpn
-repo_name: kpn
+repo_url: https://gitea.tourolle.paris/dtourolle/KPN
+repo_name: dtourolle/KPN
 
 theme:
   name: material

scripts/render_readme.py

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

tests/CMakeLists.txt

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

tests/test_channel.cpp

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

tests/test_channel_stress.cpp

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