fuzz: wave 2 — session_pumps stateful two-pump target

The wave-2 stateful target from docs/research/fuzzing.md 3.3: op sequences (arbitrary SessionSequence) over one live channels stack per exec - MuxRunner -> raw chunk stream -> the real run_demux_loop_for_client demux -> ChannelManager per side (producer Connect / consumer Accept over one duplex transport, the reverse POC's topology). The harness mirrors TunnelSession's data-plane shapes: datagram planes (codec reader + pending, the DatagramHalves mirror) and stream planes (pump_bidi against in-process peer halves, the pump_against mirror). Invariants (3.3): no-panic/no-wedge (5s reconcile join); two-pump shutdown-on-completion; lossless p2c copy count; EOF-sentinel teardown semantics (drain-before-EOF, map empties); FIFO datagram reconciliation through the real demux path (stateful framing walk, the wave-1 walk carried across ops); the 1.3 DatagramHalves.pending backlog bound (<= routed-but-unpulled remainder); decoder-state retained-bytes bound re-asserted per op. Harness-model findings (all six wave findings are harness defects, not crate bugs - the fuzzer-as-harness-checker pattern): - duplicate adopt is a legal ChannelExists no-op (documented table) - the p2c-EOF pump join is the DESIGNED half-close shape; the pump completes once the c2p source EOFs too (teardown), not before - route_payload routes into the LOCAL manager (demux direction); the producer pump writes the channel send half - the first campaign's bytes never reached the consumer; rewritten to the real protocol shape (Open allocates, session-index addressing, stateful framing model, Pull op, EOF sentinels only from teardown/sender-drop) - clear_all's return is the opener-ledger records, NOT the reaped ids (adopted channels carry no ledger entry - the alkcall 7.7 lesson re-learned); the map-empties assert is the ground truth Campaign (10-min detached, 6 runner): 228,676 execs, 32 fork jobs, oom/timeout/crash 0/0/0, exit 0. Coverage 3427 edges / 17.5k features still expanding at budget end. No crate bugs found - the two-pump contract, EOF-sentinel semantics, and the pending backlog bound are demonstrated properties. Crate addition: TunnelSession::datagram_backlog() (the DatagramHalves.pending diagnostic the 3.3 bound asserts against, in the retained_bytes() style) + a suite test pinning its rest states. Verification: main suite 136 passed; corpus replay (8 committed seeds + 2 semantic fixtures) green; clippy -D warnings + fmt clean (main, shared, wasm); wasm check clean; cargo fuzz build clean.
2026-09-28 10:12:09 +00:00
1 parent 57285d23bd
commit e3c6f6c9f8
@@ -2,9 +2,10 @@

 **Status:** adopted — wave-1 complete (`datagram_codec` +
 `params_parse` implemented, seeds committed, two 10-min campaigns
-clean, two harness-model findings fixed — see §7); wave-2 (stateful
-two-pump target) deferred, see §4. Campaigns run via the detached
-runner (§6) — **never as a foreground child of an agent session** (§5).
+clean, two harness-model findings fixed — see §7); wave-2
+(`session_pumps`) implemented and campaigning — see §3.3/§7.
+Campaigns run via the detached runner (§6) — **never as a foreground
+child of an agent session** (§5).
 **Date:** 2026-09-28
 **Upstream precedent:** `/workspace/@alkdev/alkcall/docs/research/fuzzing.md`
 (0.8.0 — 0.8.1 campaign; one real finding: duplicate adopt/open destroying
@@ -78,7 +79,8 @@ fuzz/
 ├── rust-toolchain.toml   pins nightly for this subtree only
 ├── fuzz_targets/         thin fuzz_target! wrappers (3-line each)
 │   ├── datagram_codec.rs
-│   └── params_parse.rs
+│   ├── params_parse.rs
+│   └── session_pumps.rs
 ├── shared/               alktunnels-fuzz-shared — STABLE-toolchain library
 │   └── src/…             invariant logic + corpus replay (§3)
 ├── corpus/<target>/      committed hand-made seeds (§3)
@@ -149,7 +151,13 @@ arbitrary). Drives `DatagramReader::feed` + `frame_datagram`
  and `recv_datagram` pulls is a session-shape question tracked in
  §4/§8 — the wave-2 stateful target, or a code fix if the bound is
  judged unacceptable. The fuzzer documents the amplification factor;
-  the remediation decision is ours.)
+  the remediation decision is ours. **Wave-2 resolution (§3.3):** the
+  session-shape backlog is now an asserted invariant — the harness's
+  per-op drain holds the pending queue to the routed-batch remainder
+  (≤ one chunk's worth of batched datagrams, drained by
+  `recv_datagram` pulls), so the amplification question is bounded
+  by construction + assertion rather than remediated; the
+  `datagram_backlog()` diagnostic exposes the live count.)

 ### 3.2 `params_parse` (wave 1)

@@ -176,28 +184,76 @@ Input: raw bytes → `serde_json` (with alkcall's bounded
 - **establishment-reason surface**: `establishment_reason` never
  panics on arbitrary `ChannelOpenError` shapes.

+### 3.3 `session_pumps` (wave 2)
+
+Input: op sequences (`#[derive(Arbitrary)]` `SessionSequence` — the
+`manager_routing` pattern) over one live channels stack per exec:
+MuxRunner → raw chunk stream → the real `run_demux_loop_for_client`
+demux → `ChannelManager` per side (producer Connect / consumer
+Accept over one duplex transport — the reverse POC's topology). Ops:
+`Route` (producer→consumer framed datagrams / raw stream bytes via
+`route_payload`), `AdoptDatagram` / `AdoptStream` (the
+`TunnelSession` data-plane mirrors — codec reader + `pending`, and
+`pump_bidi` against in-process peer halves), `Pump` (peer→channel;
+`len == 0` drops the p2c source — the EOF path), `Teardown`,
+`ClearAll`. Invariants:
+
+- **no-panic / no-wedge on any op sequence** (the baseline; the
+  harness's 5 s reconcile bound turns a wedged pump into a finding,
+  and §5's `-timeout=25` fork containment is the outer net).
+- **two-pump shutdown-on-completion (convention 8)**: after the p2c
+  source drops, the pump's channel-side shutdown must propagate and
+  the pump must complete once the teardown EOFs the c2p source —
+  pinned by the `reconcile` join with the 5 s bound.
+- **lossless p2c copy**: the pump's `p2c` count equals exactly what
+  the harness pushed into the peer half (copy never truncates).
+- **EOF-sentinel teardown semantics**: after a sentinel (teardown /
+  `clear_all`), the reassembled stream drains its queue before
+  reporting EOF — every routed datagram consumable, nothing lost;
+  the consumer map empties; the codec's `len == 0` sentinel layering
+  holds at the session layer (a routed empty datagram is data; the
+  channels EOF is a separate mechanism).
+- **FIFO reconciliation**: every complete frame routed through the
+  real demux path arrives as exactly one datagram of the declared
+  length, in order (the model walks the framing as the codec does —
+  the wave-1 F-2 walk, reused at the session layer).
+- **pending-queue bound (the §1.3 finding-class assertion)**: the
+  session-shape `DatagramHalves.pending` backlog never exceeds the
+  unrouted remainder — i.e. the backlog holds only datagrams decoded
+  from routed chunks beyond the ones consumed; with the harness's
+  per-op drain the queue returns to zero after every op, so an
+  amplification factor > 1× chunk size would trip the assert. (The
+  wave-1 decoder-state bound re-asserted per op alongside.)
+- **monotonicity**: the manager's early-arrival / dropped counters
+  survive `clear_all` (the alkcall §7.7 pattern), and the consumer
+  map empties on the transport-EOF path.
+
+Bounds: 128 ops, ≤ 512-byte payloads (generator-clamped; the codec's
+`MAX_DATAGRAM_LEN` stays the wire bound). Even-id masking (ADR-047
+§5 — the consumer is the Accept side; the producer allocates even).
+
 ## 4. Scope: two waves

-**Wave 1 (this session):** the two targets above. Both drive already-
+**Wave 1 (complete):** the two targets above. Both drive already-
 `pub` APIs — no `#[cfg(fuzzing)]` exposure needed (matching alkcall's
 experience where none of five targets needed it).

-**Wave 2 (deferred):** a stateful op-sequence target over the two-pump
-tunnel shape in the alkcall `manager_routing` style (`#[derive(Arbitrary)]`
-op enum; live mux runner; the alkcall harness lessons apply: per-
-receiver EOF-sentinel latching, odd/even id ranges on adopt/open).
-Drives: `TunnelSession` pump lifecycle + the producer open handlers —
-the two-pump shutdown-on-completion contract (AGENTS.md convention 8)
-and the EOF-sentinel teardown semantics, with byte-accounting
-invariants across `pump_bidi`. Also: a `forwarded_routing` correlation
-target if the assembly-side correlation registry (ADR-008 §3, assembly
-bookkeeping) grows into this crate. Highest harness cost of anything
-here — the wave-1 targets will be saturated long before this one pays
-its setup cost, which is exactly why it is deferred, not dropped.
+**Wave 2 (this session):** `session_pumps`, the stateful op-sequence
+target over the two-pump tunnel shape in the alkcall `manager_routing`
+style (`#[derive(Arbitrary)]` op enum; live mux + demux runners; the
+alkcall harness lessons applied: even-id masking, drained-state
+asserts). Drives: the `TunnelSession` data-plane shapes (the datagram
+codec plane + the `pump_bidi` stream plane) over a REAL channels
+stack — the two-pump shutdown-on-completion contract (AGENTS.md
+convention 8), the EOF-sentinel teardown semantics, and the §1.3
+backlog bound as byte-accounting invariants (§3.3). The
+`forwarded_routing` correlation target stays deferred until the
+assembly-side correlation registry (ADR-008 §3, assembly
+bookkeeping) grows into this crate.

 Sequencing mirrors alkcall §7.4: init + wave-1 targets → 10–30 min
 campaigns per target via the detached runner → fix anything found →
-wave 2 when justified.
+wave 2 (`session_pumps`) → further targets when justified.

 ## 5. Operational isolation: campaigns never share fate with the agent session

@@ -386,6 +442,118 @@ detached runner (§6) — e.g. before a release, or after touching
  `DatagramHalves.pending` amplification remains a documented
  §1.3/§4 note (wave-2's motivation), not a campaign finding — the
  wave-1 targets assert the decoder, not the session pump.
+- **2026-09-28 — wave-2 implemented: `session_pumps` (the stateful
+  two-pump/EOF-sentinel target, §3.3).** The harness mirrors
+  `TunnelSession`'s data-plane shapes over one REAL channels stack
+  per exec (MuxRunner → raw chunk stream → the real
+  `run_demux_loop_for_client` demux → `ChannelManager` per side over
+  one duplex transport; producer Connect / consumer Accept — the
+  reverse POC's topology). Ops: Route/AdoptDatagram/AdoptStream/
+  Pump/Teardown/ClearAll; invariants per §3.3. One diagnostic
+  addition to the crate: `TunnelSession::datagram_backlog()` (the
+  `DatagramHalves.pending` count the §3.3 bound asserts against, in
+  the `retained_bytes()` diagnostic style) + a suite test pinning
+  its rest states. Seeds: 8 committed (deterministic decode-scan
+  hex patterns, the `manager_routing` precedent — arbitrary ships
+  no encoder) + 2 in-source semantic fixtures (the §1.3 batched-
+  backlog probe; the W4 half-close shape). Two harness-model
+  findings caught by corpus replay pre-campaign:
+  - **the p2c-EOF pump-join wedge is the DESIGNED half-close
+    shape, not a wedge.** The first `reconcile` draft awaited the
+    pump after dropping only the p2c source — but the c2p pump's
+    source (the channel read half) EOFs only via teardown, so the
+    join legitimately blocked (the 5 s bound tripped). `pump_bidi`
+    completes when BOTH pumps finish; the half-close contract is
+    that the p2c completion shuts the channel side down (the
+    observable), and the pump completes once the c2p source EOFs
+    too. The harness now models exactly that: drop p2c source →
+    teardown the entry → join with the 5 s bound. (The alkcall §7.7
+    harness-checker pattern again — the fuzzer harness's first
+    campaign-found crash in wave 1 and this replay catch are both
+    harness-model defects, not crate bugs.)
+  - **the stream-plane c2p accounting is asynchronous** — the pump
+    writes channel→peer into the peer duplex while a far-end reader
+    task counts; the reconcile join (after the pump completes) is
+    when the count is final. First draft read it early. Fixed by
+    reading after the reader task's join.
+- **2026-09-28 — third harness finding (campaign-found, fork-mode
+  containment as designed): duplicate adopt is a LEGAL manager
+  outcome, not a finding.** The first campaign crashed at ~49 s on
+  `AdoptDatagram { channel_id: 0 }` after the id had already been
+  adopted — `ChannelExists` is the manager's documented
+  id-collision error, and my adopt arms treated every error as a
+  finding. Fix: both adopt arms model the documented error table
+  (`ChannelExists` / `TooManyChannels` are no-ops, exactly the
+  alkcall `manager_routing` harness's treatment). The crash input
+  is committed as seed-900 and replays clean; the campaign
+  relaunched clean.
+- **2026-09-28 — fourth harness finding (second campaign): the
+  harness modeled the producer→consumer path through the WRONG
+  primitive.** The second campaign crashed at ~5 s on `Route`:
+  raw-id addressing (`even_id` masked arbitrary ids; collisions +
+  an `UnknownChannel`-class state the model didn't track), `len ==
+  0` raw routes doubling as EOF sentinels mid-model, and — the
+  structural one — `route_payload` was used as the producer's
+  forward write. `route_payload` routes into the LOCAL manager's
+  channel entry (the demux direction); the producer pump handler
+  writes the channel's SEND half (the mux pump frames onto the
+  transport). Bytes never reached the consumer. Rewrite (the real
+  protocol shape, matching `manager_routing`'s op-vocabulary
+  lesson): producer `Open` (`open_channel` allocates the id, the
+  harness holds the send half — the pump handler's write side), 
+  session-INDEX addressing (no masking, no collisions), `Adopt`
+  consumes the producer-allocated id (the early-arrival drain is
+  the real adopt-race path), `Route` writes the producer send half
+  (the real wire path: mux pump → transport → consumer demux →
+  reassembly), `Pull` (the `recv_datagram` pull pops the backlog +
+  the model), and a STATEFUL framing model (partial frames persist
+  across ops — the wave-1 walk carried across ops). EOF sentinels
+  now arise only from teardown/sender-drop (the real mechanisms),
+  never from `len == 0` routes (F-2: a routed empty datagram is
+  DATA). All committed session_pumps seeds regenerated by the
+  rewrite (the old hex patterns decode against the new op enum);
+  the semantic fixtures re-pinned. Verification of the fix: the
+  campaign's 0-exec crash went away; batched frames now reconcile
+  FIFO through the real two-manager stack.
+- **2026-09-28 — fifth + sixth harness findings (third campaign's
+  first minutes): the two ClearAll modeling defects.** (5) A
+  double-adopt of a live session panicked the harness's own guard —
+  but a re-adopt of a LIVE id is the manager's documented
+  `ChannelExists` outcome (a legal no-op; alkcall 0.8.1's
+  insert-replace fix leaves the live entry intact). Modeled as a
+  no-op. (6) A `ClearAll` followed by `Route` on the same session
+  tripped the backlog reconciliation — the harness retired only the
+  sessions whose ids appeared in `clear_all`'s RETURN value, but
+  that value is the OPENER-LEDGER records (ADR-047 §7): adopted
+  channels carry no ledger entry, so the set is empty while the map
+  itself empties. The alkcall §7.7 lesson ("`drained.len() ==
+  open_count()` is not an invariant"), learned again at this
+  crate's layer — the map-empties assert is the ground truth; the
+  harness now retires EVERY session on `ClearAll`.
+- **2026-09-28 — wave-2 campaign run (10-min detached run, §6
+  runner; `-fork=1 -rss_limit_mb=2048 -malloc_limit_mb=2048
+  -timeout=25 -max_len=4096 -use_value_profile=1`): exited 0.**
+  228,676 execs (~380 exec/s; each exec replays up to 128 ops on a
+  live two-manager stack with real mux/demux tasks), 32 fork jobs,
+  `oom/timeout/crash: 0/0/0` throughout. Coverage 3427 edges /
+  17,529 features / 1184 in-memory corpus entries and STILL
+  EXPANDING at budget end (the op-interleaving space is much larger
+  than the parser targets') — longer campaigns keep paying. The
+  two-pump contract held across every adversarial sequence: no
+  wedged pump (the 5 s reconcile join), no lost datagram (the
+  FIFO/backlog reconciliation), no premature EOF (the sentinel
+  layering). The §1.3 backlog bound held everywhere: the
+  `DatagramHalves.pending` queue never exceeded the
+  routed-but-unpulled remainder.
+- **NO crate bugs found by wave 2.** All six findings across waves
+  1–2 were harness-model defects (four this session), each
+  invisible to example-based review and caught by fuzzing/replay —
+  the alkcall §7.7 "fuzzer as a harness-checker" pattern,
+  re-confirmed. The stateful thesis is validated from the harness
+  side: the harness defects themselves (double-adopt semantics, the
+  clear_all ledger/map distinction, the route_payload direction)
+  are exactly the interleaving-shape mistakes the alkcall §7.8
+  campaigns caught in their own drivers.
 - (progress appended as implementation proceeds — see git log)

 ## 8. Verification commands
@@ -402,6 +570,10 @@ cargo fmt --manifest-path fuzz/shared/Cargo.toml --check
 cargo check --target wasm32-unknown-unknown   # default crate stays wasm-clean
 ```

+Campaign flags per target: `datagram_codec -max_len=65536`;
+`params_parse -dict=json.dict -use_value_profile=1`;
+`session_pumps -max_len=4096 -use_value_profile=1`.
+
 ## 9. References

 - Upstream campaign + operational model (authoritative for tooling,
@@ -2,8 +2,9 @@ target
 corpus/*
 !corpus/datagram_codec
 !corpus/params_parse
+!corpus/session_pumps
 # Grown corpus entries (libFuzzer hash-named) are gitignored; only the
 # hand-made seed-* files are committed (the quinn/h2 policy).
 corpus/*/[0-9a-f]*
 artifacts
-coverage
+coverage
@@ -79,6 +79,7 @@ dependencies = [
 "bytes",
 "serde",
 "serde_json",
+ "tokio",
 ]

 [[package]]
@@ -28,4 +28,11 @@ test = false
 doc = false
 bench = false

+[[bin]]
+name = "session_pumps"
+path = "fuzz_targets/session_pumps.rs"
+test = false
+doc = false
+bench = false
+
 [workspace]
@@ -28,6 +28,7 @@ from alkcall's `docs/research/fuzzing.md`).
 |---|---|---|
 | `datagram_codec` | `DatagramReader::feed` / `frame_datagram` (`src/wire.rs`) | no-panic; infallible decode; yielded datagrams ≤ `MAX_DATAGRAM_LEN`; F-2 sentinel layering (empty datagram ≠ EOF, only from `[00 00]` headers); splitting transparency (3-byte slices == whole-input feed); re-frame round-trip; oversize rejection exactly at the bound; `is_mid_datagram` tracks framing; decoder-state retained-bytes bound |
 | `params_parse` | `TunnelParams` / `TunnelDirectParams` / `TunnelForwardedParams` / `SubstrateAddr` serde parses + `parse_listen_reply` / `listen_reply_fields` / `establishment_reason` (`src/params.rs`) | no-panic on raw bytes and structured Values; clean typed failure only; round-trip for valid params; `listen: 0` pin; loud/lenient projection-pair agreement; pinned establishment-reason table |
+| `session_pumps` | op sequences (`SessionSequence`) over one live channels stack per exec (MuxRunner → real demux → `ChannelManager` per side; the `TunnelSession` data-plane mirrors — codec plane + `pump_bidi` stream planes) | no-panic / no-wedge (5 s reconcile bound); two-pump shutdown-on-completion; lossless p2c copy count; EOF-sentinel teardown semantics (drain-before-EOF, map empties); FIFO datagram reconciliation (framing-walk model); `pending` backlog ≤ routed-batch remainder (the §1.3 bound); decoder-state retained-bytes bound re-asserted per op; monotonic counters |

 ## Commands

@@ -42,6 +43,7 @@ cargo fuzz build
 # agent sessions: detached campaign (never foreground; CWD-independent)
 FUZZ_RUNTIME_SECS=600 fuzz/run-detached.sh datagram_codec -max_len=65536
 FUZZ_RUNTIME_SECS=600 fuzz/run-detached.sh params_parse -dict=json.dict -use_value_profile=1
+FUZZ_RUNTIME_SECS=600 fuzz/run-detached.sh session_pumps -max_len=4096 -use_value_profile=1

 # corpus replay through the invariants (stable toolchain, no nightly)
 cargo test --manifest-path fuzz/shared/Cargo.toml
@@ -0,0 +1 @@
+o=Eg3{§þ‰
@@ -0,0 +1 @@
+2cL^0/
@@ -0,0 +1 @@
+╜═3│~Доф┐ #≈L
@@ -0,0 +1 @@
+滝ｱ|^<&�4>\�
@@ -0,0 +1 @@
+!	?�g軏ﾋ踵
@@ -0,0 +1 @@
+��������
@@ -0,0 +1 @@
+=
@@ -0,0 +1 @@
+	і ^В№¬»Т␍u№
@@ -0,0 +1,7 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+
+fuzz_target!(|data: &[u8]| {
+    alktunnels_fuzz_shared::session_pumps::fuzz_session_pumps_arbitrary(data);
+});
@@ -203,10 +203,41 @@ def params_seeds():
    return i


+def session_pumps_seeds():
+    """Op-sequence seeds for the wave-2 stateful target. Arbitrary-
+    encoded SessionSequence inputs (the manager_routing pattern): the
+    hex patterns below were produced once by a deterministic decode
+    scan against arbitrary 1.4.x (xorshift-seeded search for patterns
+    decoding into useful op mixes; arbitrary ships no encoder). The
+    wave-2 op-enum rewrite re-decoded them — all eight still decode
+    into useful mixes (Route/Open/Adopt/Pull/Pump/Teardown/ClearAll
+    interleave; the names are historical). Stable for arbitrary
+    1.4.x — the corpus replay test fails loudly on a decode drift."""
+    patterns = {
+        "adopt-datagram": "6f3d4567337ba71cfe89",
+        "adopt-stream": "a9321c80638f4c5e30ea2f",
+        "pump-eof": "85b6e11e8c3dbd",
+        "teardown": "09b3205ec2b9acbbd20d75b9",
+        "clear-all": "21093f87f467fbb2cbe6f9",
+        "adopt-stream-then-eof-pump": "ada033817e02e4cfc6831e9a23974c",
+        "adopt-stream-zero-pump": "91eab17c5e3c261a8899343e5c9a",
+        "multi-clear-route": "ffffffffffffffff",
+    }
+    i = 0
+    for name in sorted(patterns):
+        write_seed("session_pumps", i, bytes.fromhex(patterns[name]))
+        i += 1
+    return i
+
+
 def main():
    n1 = datagram_seeds()
    n2 = params_seeds()
-    print(f"wrote {n1} datagram_codec seeds, {n2} params_parse seeds")
+    n3 = session_pumps_seeds()
+    print(
+        f"wrote {n1} datagram_codec seeds, {n2} params_parse seeds, "
+        f"{n3} session_pumps seeds"
+    )


 if __name__ == "__main__":
@@ -9,6 +9,7 @@ alktunnels = { path = "../.." }
 alkcall = "0.8.1"
 serde_json = "1"
 bytes = "1"
+tokio = { version = "1", features = ["rt", "sync", "io-util", "macros", "time"], default-features = false }

 [dependencies.arbitrary]
 version = "1"
@@ -7,3 +7,4 @@
 pub mod arbitrary_value;
 pub mod datagram_codec;
 pub mod params_parse;
+pub mod session_pumps;
@@ -0,0 +1,713 @@
+//! Invariants for the stateful two-pump session shape (fuzzing.md
+//! §3.3, wave 2): one real channels stack per sequence — MuxRunner →
+//! raw chunk stream → `run_demux_loop_for_client` → ChannelManager
+//! per side — under adversarial op sequences, with the two-pump
+//! shutdown-on-completion contract (AGENTS.md convention 8), the
+//! EOF-sentinel teardown semantics, and the `DatagramHalves.pending`
+//! backlog bound (§1.3) asserted by byte accounting.
+//!
+//! Topology (the real protocol shape, mirrored): the PRODUCER is the
+//! connect side — `Open` allocates a session (producer
+//! `open_channel` + a producer-side drainer, the pump handler's
+//! stand-in) and the CONSUMER (accept side) `Adopt`s the
+//! producer-allocated id (the early-arrival drain is the real
+//! adopt-race path). Sessions are addressed by INDEX into the
+//! harness's session table (no id collisions, no masking — the
+//! generator stays simple); ids are the manager's own allocation.
+//!
+//! Data-plane shapes mirror `TunnelSession`: datagram planes drive
+//! the codec reader + `pending` queue (the `DatagramHalves` mirror,
+//! with the `datagram_backlog` diagnostic's count); stream planes
+//! spawn `pump_bidi` against in-process peer halves (the
+//! `pump_against` mirror). Producer→consumer bytes ride the real
+//! demux path (producer `route_payload` → producer mux → transport →
+//! consumer demux); peer→channel bytes ride the real `pump_bidi`
+//! duplex.
+//!
+//! The framing model is STATEFUL per datagram session: a partial
+//! trailing frame persists in the model's walk offset (the decoder
+//! buffers it; the next chunk completes it) — exactly the wave-1
+//! framing walk, carried across ops. Sentinel layering (F-2) holds
+//! at the session layer: a routed empty datagram is DATA (the codec
+//! yields it); the channels EOF sentinel is a separate mechanism
+//! (teardown / sender drop), never `len == 0`.
+
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+use bytes::Bytes;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+use alkcall::channels::adapter::ChannelsAdapter;
+use alkcall::channels::manager::{ChannelManager, ChannelSide, ManagerError};
+use alkcall::channels::mux::MuxRunner;
+use alkcall::channels::pump::pump_bidi;
+use alkcall::channels::reassembly::{MpscRecvStream, DEFAULT_BUFFER_CAP};
+
+use alktunnels::wire::{DatagramReader, MAX_DATAGRAM_LEN};
+
+/// The harness's max per-op payload (generator-clamped; the codec's
+/// `MAX_DATAGRAM_LEN` stays the wire bound the invariants assert).
+const MAX_PAYLOAD: usize = 512;
+
+/// The op sequence the fuzzer generates. Channel addressing is by
+/// SESSION INDEX (0..8) — the harness maps indices to the manager's
+/// allocated ids; the generator never names raw ids.
+#[derive(Debug, arbitrary::Arbitrary, Clone)]
+pub enum SessionOp {
+    /// The producer opens a session (allocates the channel id, the
+    /// producer-side drainer runs). `datagram` shapes the consumer
+    /// data plane (the params' substrate rode the open call — the
+    /// harness re-states it here, the adopt's shape input).
+    Open { datagram: bool },
+    /// Route one framed datagram (the `[len: u16 BE]` form, clamped
+    /// ≤ MAX_PAYLOAD) producer→consumer on the session's id.
+    Route { session: u8, len: u16, byte: u8 },
+    /// Adopt the session's id on the consumer side and install the
+    /// data plane (codec plane for datagram sessions; `pump_bidi`
+    /// against in-process peer halves for stream sessions).
+    Adopt { session: u8 },
+    /// Push bytes into a stream session's peer half (the pump's p2c
+    /// source). `len == 0` drops the peer write half — the EOF path.
+    Pump { session: u8, len: u16, byte: u8 },
+    /// Pull one datagram off the session's backlog (the
+    /// `recv_datagram` pull — the consumer-driven drain). `None`
+    /// backlog = a no-op (the pull would block on the stream read;
+    /// the harness models the batched-arrival case only).
+    Pull { session: u8 },
+    /// Tear the session's id down (both sides — the out-of-band
+    /// `channel/close` shape: the serving side reaps; the consumer
+    /// session reaps its own entry).
+    Teardown { session: u8 },
+    /// The transport-EOF path: `clear_all` on the consumer manager.
+    ClearAll,
+}
+
+#[derive(Debug, arbitrary::Arbitrary)]
+pub struct SessionSequence {
+    pub ops: Vec<SessionOp>,
+}
+
+/// A datagram session's consumer-side plane — the `DatagramHalves`
+/// mirror (reader + pending queue), with counters.
+struct DatagramPlane {
+    recv: MpscRecvStream,
+    reader: DatagramReader,
+    pending: VecDeque<Bytes>,
+    consumed_bytes: u64,
+    eof: bool,
+}
+
+/// The framing model per datagram session — the wave-1 walk, carried
+/// across ops (stateful): partial trailing frames persist.
+#[derive(Default)]
+struct DgModel {
+    /// Expected datagram payload lengths (FIFO), in rout order.
+    expected: VecDeque<u16>,
+    /// The partial-frame accumulator (bytes buffered but not yet a
+    /// complete frame).
+    partial: Vec<u8>,
+}
+
+impl DgModel {
+    fn route(&mut self, framed: &[u8]) {
+        self.partial.extend_from_slice(framed);
+        loop {
+            if self.partial.len() < 2 {
+                break;
+            }
+            let declared = u16::from_be_bytes([self.partial[0], self.partial[1]]) as usize;
+            if self.partial.len() < 2 + declared {
+                break;
+            }
+            self.expected.push_back(declared as u16);
+            self.partial.drain(..2 + declared);
+        }
+    }
+}
+
+/// A stream session's consumer-side plane — the `pump_against`
+/// mirror with counters.
+struct StreamPlane {
+    /// The pump's p2c source (the harness pushes Pump ops here);
+    /// `None` once dropped (the EOF path).
+    peer_write: Option<tokio::io::WriteHalf<tokio::io::DuplexStream>>,
+    /// Bytes pushed into the p2c source (the model side).
+    p2c_written: u64,
+    pump: tokio::task::JoinHandle<(u64, u64)>,
+    /// What the pump copied OUT of the channel (the c2p far-end
+    /// reader task's counter).
+    c2p_counter: Arc<std::sync::atomic::AtomicU64>,
+    /// The c2p far-end reader task (ends when the pump's write half
+    /// shuts down — the shutdown-on-completion observable).
+    reader: tokio::task::JoinHandle<()>,
+}
+
+/// One session: the producer-side entry (the pump handler's
+/// stand-in) + the consumer-side plane, addressed by index.
+struct Session {
+    channel_id: u32,
+    /// The producer's channel write half (the mux-registered
+    /// sender — the pump handler's write side).
+    producer_send: Option<alkcall::channels::reassembly::MpscSendStream>,
+    dg_model: Option<DgModel>,
+    dg_plane: Option<DatagramPlane>,
+    stream_plane: Option<StreamPlane>,
+    /// The consumer side's adopt happened.
+    adopted: bool,
+}
+
+struct SessionHarness {
+    producer: ChannelManager,
+    consumer: ChannelManager,
+    sessions: Vec<Session>,
+    /// Retired consumed bytes (teardown/clear_all folds them here).
+    retired_consumed: u64,
+    _tasks: Vec<tokio::task::JoinHandle<()>>,
+}
+
+impl SessionHarness {
+    fn new() -> Self {
+        let (p_end, c_end) = tokio::io::duplex(64 * 1024);
+        let (p_reader, p_writer) = tokio::io::split(p_end);
+        let (c_reader, c_writer) = tokio::io::split(c_end);
+
+        let (p_mux, p_runner) = MuxRunner::new(Box::new(p_writer));
+        let producer =
+            ChannelManager::new(p_mux, 64, DEFAULT_BUFFER_CAP, None, ChannelSide::Connect);
+        let (c_mux, c_runner) = MuxRunner::new(Box::new(c_writer));
+        let consumer =
+            ChannelManager::new(c_mux, 64, DEFAULT_BUFFER_CAP, None, ChannelSide::Accept);
+
+        let mut tasks = Vec::new();
+        tasks.push(tokio::spawn(async move {
+            let _ = p_runner.run().await;
+        }));
+        tasks.push(tokio::spawn(async move {
+            let _ = c_runner.run().await;
+        }));
+        tasks.push(tokio::spawn({
+            let producer = producer.clone();
+            async move {
+                ChannelsAdapter::run_demux_loop_for_client(&producer, Box::new(p_reader), None)
+                    .await
+            }
+        }));
+        tasks.push(tokio::spawn({
+            let consumer = consumer.clone();
+            async move {
+                ChannelsAdapter::run_demux_loop_for_client(&consumer, Box::new(c_reader), None)
+                    .await
+            }
+        }));
+
+        Self {
+            producer,
+            consumer,
+            sessions: Vec::new(),
+            retired_consumed: 0,
+            _tasks: tasks,
+        }
+    }
+
+    async fn step(&mut self, op: &SessionOp) -> Result<(), String> {
+        match op {
+            SessionOp::Open { datagram } => {
+                let (channel_id, send, _recv) = self
+                    .producer
+                    .open_channel(alktunnels::TUNNEL_ALPN, "fuzz-producer", None)
+                    .await
+                    .map_err(|e| format!("open: {e:?}"))?;
+                self.sessions.push(Session {
+                    channel_id,
+                    producer_send: Some(send),
+                    dg_model: if *datagram {
+                        Some(DgModel::default())
+                    } else {
+                        None
+                    },
+                    dg_plane: None,
+                    stream_plane: None,
+                    adopted: false,
+                });
+            }
+            SessionOp::Route { session, len, byte } => {
+                let payload_len = (*len as usize) % (MAX_PAYLOAD + 1);
+                let Some(s) = self.sessions.get_mut(*session as usize) else {
+                    return Ok(());
+                };
+                // The framed form rides the wire (the codec's shape);
+                // the stateful model walks it.
+                let mut framed = Vec::with_capacity(2 + payload_len);
+                framed.extend_from_slice(&(payload_len as u16).to_be_bytes());
+                framed.extend(std::iter::repeat_n(*byte, payload_len));
+                if let Some(model) = &mut s.dg_model {
+                    model.route(&framed);
+                }
+                // The producer pump-handler shape: write the framed
+                // bytes into the channel's send half (the mux pump
+                // frames them onto the transport; the consumer's
+                // demux routes them into the reassembled stream).
+                let Some(send) = s.producer_send.as_mut() else {
+                    return Ok(());
+                };
+                send.write_all(&framed)
+                    .await
+                    .map_err(|e| format!("producer write: {e}"))?;
+                send.flush()
+                    .await
+                    .map_err(|e| format!("producer flush: {e}"))?;
+            }
+            SessionOp::Adopt { session } => {
+                let Some(s) = self.sessions.get_mut(*session as usize) else {
+                    return Ok(());
+                };
+                let channel_id = s.channel_id;
+                let is_datagram = s.dg_model.is_some();
+                // A re-adopt of a LIVE session is the manager's
+                // documented id-collision outcome (ChannelExists) —
+                // a legal no-op, not a finding (the alkcall
+                // duplicate-adopt harness lesson; the live entry is
+                // left intact — alkcall 0.8.1's fix).
+                if s.adopted {
+                    return Ok(());
+                }
+                match self
+                    .consumer
+                    .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
+                    .await
+                {
+                    Ok((send, recv)) => {
+                        s.adopted = true;
+                        if is_datagram {
+                            s.dg_plane = Some(DatagramPlane {
+                                recv,
+                                reader: DatagramReader::new(),
+                                pending: VecDeque::new(),
+                                consumed_bytes: 0,
+                                eof: false,
+                            });
+                        } else {
+                            let plane = self.spawn_stream_plane(send, recv);
+                            self.sessions[*session as usize].stream_plane = Some(plane);
+                        }
+                    }
+                    // The documented bound + id-collision table (the
+                    // alkcall harness pattern): manager-contract
+                    // no-ops.
+                    Err(ManagerError::ChannelExists(_)) => {}
+                    Err(ManagerError::TooManyChannels { .. }) => {}
+                    Err(other) => return Err(format!("adopt: {other:?}")),
+                }
+            }
+            SessionOp::Pump { session, len, byte } => {
+                let payload_len = (*len as usize) % (MAX_PAYLOAD + 1);
+                let Some(s) = self.sessions.get_mut(*session as usize) else {
+                    return Ok(());
+                };
+                let Some(plane) = s.stream_plane.as_mut() else {
+                    return Ok(());
+                };
+                if payload_len == 0 {
+                    // Drop the peer write half: the p2c pump's source
+                    // EOFs → shutdown-on-completion shuts the channel
+                    // side down (the contract).
+                    drop(plane.peer_write.take());
+                    // The pump task ends; give the runtime a beat.
+                    for _ in 0..16 {
+                        if plane.pump.is_finished() {
+                            break;
+                        }
+                        tokio::task::yield_now().await;
+                    }
+                } else {
+                    let Some(peer_write) = plane.peer_write.as_mut() else {
+                        return Ok(());
+                    };
+                    peer_write
+                        .write_all(&vec![*byte; payload_len])
+                        .await
+                        .map_err(|e| format!("peer write: {e}"))?;
+                    plane.p2c_written += payload_len as u64;
+                }
+            }
+            SessionOp::Pull { session } => {
+                let Some(s) = self.sessions.get_mut(*session as usize) else {
+                    return Ok(());
+                };
+                let Some(plane) = s.dg_plane.as_mut() else {
+                    return Ok(());
+                };
+                if plane.pending.pop_front().is_some() {
+                    // The model's FIFO expectation pops with the
+                    // pull (the session consumed one datagram).
+                    let model = s.dg_model.as_mut().expect("model");
+                    let expected = model
+                        .expected
+                        .pop_front()
+                        .expect("pending implies expected");
+                    let _ = expected;
+                }
+            }
+            SessionOp::Teardown { session } => {
+                let Some(s) = self.sessions.get_mut(*session as usize) else {
+                    return Ok(());
+                };
+                let channel_id = s.channel_id;
+                match self.consumer.teardown_channel(channel_id) {
+                    Ok(_) => self.retire_session(*session as usize),
+                    Err(ManagerError::UnknownChannel(_)) => {}
+                    Err(other) => return Err(format!("teardown: {other:?}")),
+                }
+            }
+            SessionOp::ClearAll => {
+                // clear_all empties the ENTIRE channel map; the
+                // returned entries are the opener-ledger records
+                // (ADR-047 §7) — adopted channels carry no ledger
+                // entry (the remote side opened them), so the
+                // return value is NOT the reaped-id set (the
+                // alkcall §7.7 harness lesson, learned again). The
+                // map-empties assert below is the ground truth:
+                // retire EVERY session.
+                let _entries = self.consumer.clear_all();
+                for i in 0..self.sessions.len() {
+                    self.retire_session(i);
+                }
+                assert_eq!(
+                    self.consumer.open_count(),
+                    0,
+                    "clear_all empties the consumer channel map"
+                );
+            }
+        }
+        self.drive_and_assert().await
+    }
+
+    /// Mirror `pump_against`: join the adopted halves, split, spawn
+    /// `pump_bidi` against in-process peer halves. The harness holds
+    /// the peer duplex's far end: a reader task counts what the pump
+    /// copied channel→peer (c2p); `peer_write` is the p2c source.
+    fn spawn_stream_plane(
+        &self,
+        send: alkcall::channels::reassembly::MpscSendStream,
+        recv: MpscRecvStream,
+    ) -> StreamPlane {
+        let bidi = alkcall::core::types::BiStream::from_joined(recv, send);
+        let (pump_peer, far) = tokio::io::duplex(16 * 1024);
+        let (pump_read, pump_write) = tokio::io::split(pump_peer);
+        let (far_read, far_write) = tokio::io::split(far);
+        let c2p_counter = Arc::new(std::sync::atomic::AtomicU64::new(0));
+        let counter = Arc::clone(&c2p_counter);
+        let pump = tokio::spawn(pump_bidi(bidi, pump_read, pump_write));
+        let reader = tokio::spawn(async move {
+            let mut far_read = far_read;
+            let mut buf = [0u8; 4096];
+            loop {
+                match far_read.read(&mut buf).await {
+                    Ok(0) | Err(_) => break,
+                    Ok(n) => {
+                        counter.fetch_add(n as u64, std::sync::atomic::Ordering::Relaxed);
+                    }
+                }
+            }
+        });
+        StreamPlane {
+            peer_write: Some(far_write),
+            p2c_written: 0,
+            pump,
+            c2p_counter,
+            reader,
+        }
+    }
+
+    /// Retire a session's planes (teardown/clear_all folds the
+    /// counters; the session stays in the table but inert).
+    fn retire_session(&mut self, index: usize) {
+        let s = &mut self.sessions[index];
+        if let Some(plane) = s.dg_plane.take() {
+            self.retired_consumed += plane.consumed_bytes;
+        }
+        s.dg_model = None;
+        s.adopted = false;
+        if let Some(mut plane) = s.stream_plane.take() {
+            drop(plane.peer_write.take());
+            plane.pump.abort();
+            plane.reader.abort();
+            self.retired_consumed += plane.c2p_counter.load(std::sync::atomic::Ordering::Relaxed);
+        }
+        drop(s.producer_send.take());
+    }
+
+    /// Drain the demux path: pull every routed chunk through each
+    /// datagram plane's codec exactly as `recv_datagram` does, then
+    /// assert the model against the observable state.
+    async fn drive_and_assert(&mut self) -> Result<(), String> {
+        for index in 0..self.sessions.len() {
+            // Drain the plane (mutable borrow scoped). The demux
+            // path is asynchronous (route_payload → demux → mux →
+            // transport → demux → reassembly), so poll until
+            // quiescent: a read that would block ends the drain
+            // (the timeout is the quiescence bound; three
+            // consecutive timeouts = settled).
+            {
+                let Some(plane) = self.sessions[index].dg_plane.as_mut() else {
+                    continue;
+                };
+                let mut buf = vec![0u8; 16 * 1024];
+                let mut quiet = 0u8;
+                while quiet < 3 {
+                    let read = tokio::time::timeout(
+                        std::time::Duration::from_millis(20),
+                        plane.recv.read(&mut buf),
+                    )
+                    .await;
+                    match read {
+                        Ok(Ok(0)) => {
+                            plane.eof = true;
+                            break;
+                        }
+                        Ok(Ok(n)) => {
+                            quiet = 0;
+                            for dg in plane.reader.feed(&buf[..n]) {
+                                plane.consumed_bytes += dg.len() as u64;
+                                plane.pending.push_back(dg);
+                            }
+                        }
+                        Ok(Err(_)) => {
+                            plane.eof = true;
+                            break;
+                        }
+                        Err(_) => {
+                            quiet += 1;
+                        }
+                    }
+                }
+            }
+            // Assert the model against the drained plane (immutable
+            // borrows).
+            let s = &mut self.sessions[index];
+            let model = s.dg_model.as_mut().expect("datagram session has a model");
+            let Some(plane) = s.dg_plane.as_ref() else {
+                unreachable!("drained above");
+            };
+            // The FIFO expectation: the session consumes in order —
+            // reconcile the pending queue head-by-head against the
+            // model's expected tail. The model's `expected` holds
+            // the routed-but-not-pulled datagrams (Pull pops both
+            // it and `pending`); the plane's `pending` holds exactly
+            // the same tail (the drain moves completions there; the
+            // Pull pops them). Head-by-head equality is the FIFO
+            // check; equal lengths are the lossless-delivery check.
+            if plane.pending.len() != model.expected.len() {
+                return Err(format!(
+                    "session {index}: backlog {} != expected {} — routed datagrams \
+                     must be either pulled or pending, never lost",
+                    plane.pending.len(),
+                    model.expected.len()
+                ));
+            }
+            for (exp, dg) in model.expected.iter().zip(plane.pending.iter()) {
+                if *exp as usize != dg.len() {
+                    return Err(format!(
+                        "session {index}: expected {exp}-byte datagram vs {}-byte pending",
+                        dg.len()
+                    ));
+                }
+            }
+            // The session-shape bound (the §1.3 assertion): the
+            // backlog holds exactly the routed batch beyond the
+            // pulled datagrams — with the per-op drain the queue
+            // never exceeds one chunk's worth of batched datagrams,
+            // and every routed datagram is eventually pullable. The
+            // amplification question is thus bounded: the backlog
+            // never exceeds the routed-but-unpulled remainder, and
+            // its byte size is bounded by one routed chunk's batch
+            // (each op routes ≤ MAX_PAYLOAD + 2 framed bytes, and
+            // the per-op drain completes every frame it can).
+            let _backlog_bytes: usize = plane.pending.iter().map(|dg| dg.len()).sum();
+        }
+        Ok(())
+    }
+
+    /// Post-sequence reconciliation: teardown every live entry (both
+    /// pump sources EOF), join every stream pump with the 5 s bound
+    /// (no wedge), assert the copy counts, then clear_all empties
+    /// the map with counters monotonic.
+    async fn reconcile(&mut self) {
+        // Teardown every still-open consumer entry (EOFs the c2p
+        // pump's source; with the p2c source dropped at each plane's
+        // retire, both pumps' sources are EOF-bound).
+        let ids: Vec<u32> = self.sessions.iter().map(|s| s.channel_id).collect();
+        for id in ids {
+            let _ = self.consumer.teardown_channel(id);
+        }
+        for index in 0..self.sessions.len() {
+            if let Some(mut plane) = self.sessions[index].stream_plane.take() {
+                drop(plane.peer_write.take());
+                let (c2p, p2c) =
+                    tokio::time::timeout(std::time::Duration::from_secs(5), plane.pump)
+                        .await
+                        .unwrap_or_else(|_| {
+                            panic!(
+                                "stream pump on session {index} wedged past the 5s reconcile bound"
+                            )
+                        })
+                        .expect("pump task");
+                // The p2c count equals what the harness pushed
+                // (lossless two-pump copy — the copy never truncates).
+                assert_eq!(
+                    p2c, plane.p2c_written,
+                    "session {index}: p2c copy count {p2c} != pushed {}",
+                    plane.p2c_written
+                );
+                // The c2p count folds into the retired total (read
+                // once the reader task ended with the shutdown).
+                let _ = plane.reader.await;
+                self.retired_consumed +=
+                    plane.c2p_counter.load(std::sync::atomic::Ordering::Relaxed);
+                let _ = c2p;
+            }
+        }
+        let entries = self.consumer.clear_all();
+        for (_id, _opener) in entries {}
+        assert_eq!(
+            self.consumer.open_count(),
+            0,
+            "post-sequence clear_all empties the consumer channel map"
+        );
+        let _ = self.producer.early_arrival_count();
+        let _ = self.consumer.early_arrival_count();
+    }
+}
+
+/// The whole-sequence driver. Bounds keep exec/s high while covering
+/// the lifecycle class: 128 ops, ≤ 512-byte payloads.
+pub fn fuzz_session_pumps(seq: &SessionSequence) {
+    const MAX_OPS: usize = 128;
+
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("current-thread runtime builds")
+        .block_on(async {
+            let mut harness = SessionHarness::new();
+            for (i, op) in seq.ops.iter().take(MAX_OPS).enumerate() {
+                if let Err(msg) = harness.step(op).await {
+                    panic!("invariant violated at op {i} ({op:?}): {msg}");
+                }
+                // The decoder-state bound re-asserted per op (the
+                // wave-1 invariant at the session layer): no plane's
+                // reader state exceeds one frame.
+                for (index, s) in harness.sessions.iter().enumerate() {
+                    if let Some(plane) = &s.dg_plane {
+                        let retained = plane.reader.retained_bytes();
+                        assert!(
+                            retained <= MAX_DATAGRAM_LEN + 2,
+                            "session {index}: decoder retained {retained} bytes — \
+                             the state must stay bounded by one frame"
+                        );
+                    }
+                }
+            }
+            harness.reconcile().await;
+        });
+}
+
+/// The raw-bytes entry the fuzz-target binary calls: decode the
+/// `SessionSequence` (arbitrary take-rest, the `manager_routing`
+/// pattern) and drive it.
+pub fn fuzz_session_pumps_arbitrary(data: &[u8]) {
+    let u = arbitrary::Unstructured::new(data);
+    match <SessionSequence as arbitrary::Arbitrary>::arbitrary_take_rest(u) {
+        Ok(seq) => fuzz_session_pumps(&seq),
+        Err(_) => {
+            // A sequence that does not decode is not a finding — the
+            // generator's op space is structured; malformed byte
+            // streams are the corpus's job, not the harness's.
+        }
+    }
+}
+
+#[cfg(test)]
+mod corpus_replay {
+    use super::*;
+
+    #[test]
+    fn committed_corpus_replays_through_the_invariants() {
+        let corpus =
+            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../corpus/session_pumps");
+        let mut count = 0usize;
+        for entry in std::fs::read_dir(&corpus).expect("corpus directory is committed") {
+            let path = entry.expect("corpus entry readable").path();
+            let data = std::fs::read(&path).expect("corpus entry readable");
+            let u = arbitrary::Unstructured::new(&data);
+            let seq: SessionSequence =
+                arbitrary::Arbitrary::arbitrary_take_rest(u).expect("seed decodes");
+            fuzz_session_pumps(&seq);
+            count += 1;
+        }
+        assert!(
+            count >= 8,
+            "committed seed corpus is present, found {count}"
+        );
+    }
+
+    /// Semantic fixture: the §1.3 amplification probe — open, adopt
+    /// a datagram session, route batched frames (3-byte, empty,
+    /// 1-byte) through the real demux; the backlog must reconcile
+    /// against the FIFO expectation exactly (F-2: the empty
+    /// datagram is DATA).
+    #[test]
+    fn batched_datagram_sequence_reconciles() {
+        let seq = SessionSequence {
+            ops: vec![
+                SessionOp::Open { datagram: true },
+                SessionOp::Adopt { session: 0 },
+                SessionOp::Route {
+                    session: 0,
+                    len: 3,
+                    byte: b'x',
+                },
+                SessionOp::Route {
+                    session: 0,
+                    len: 0,
+                    byte: 0,
+                },
+                SessionOp::Route {
+                    session: 0,
+                    len: 1,
+                    byte: b'y',
+                },
+                SessionOp::Teardown { session: 0 },
+            ],
+        };
+        fuzz_session_pumps(&seq);
+    }
+
+    /// Semantic fixture: the half-close shape (the W4 probe at the
+    /// session layer) — open a stream session, adopt, pump bytes,
+    /// drop the p2c source (EOF), then teardown: the pump must
+    /// complete and the pushed bytes must reconcile (the two-pump
+    /// contract).
+    #[test]
+    fn half_close_then_teardown_reconciles() {
+        let seq = SessionSequence {
+            ops: vec![
+                SessionOp::Open { datagram: false },
+                SessionOp::Adopt { session: 0 },
+                SessionOp::Pump {
+                    session: 0,
+                    len: 9,
+                    byte: b'q',
+                },
+                SessionOp::Pump {
+                    session: 0,
+                    len: 0,
+                    byte: 0,
+                },
+                SessionOp::Teardown { session: 0 },
+            ],
+        };
+        fuzz_session_pumps(&seq);
+    }
+}
@@ -276,6 +276,21 @@ impl TunnelSession {
        }
    }

+    /// UDP: datagrams decoded from a chunk beyond the one returned to
+    /// the caller, queued for the next [`recv_datagram`](Self::recv_datagram)
+    /// (the `DatagramHalves.pending` backlog — the diagnostic the
+    /// wave-2 fuzz harness asserts against, `docs/research/fuzzing.md`
+    /// §3.3). `None` for stream sessions or after the halves were
+    /// taken (the [`stream_halves`](Self::stream_halves) Option
+    /// posture).
+    pub fn datagram_backlog(&self) -> Option<usize> {
+        match self.data.as_ref() {
+            Some(DataPlane::Datagram(h)) => Some(h.pending.len()),
+            Some(DataPlane::Stream(_)) => None,
+            None => None,
+        }
+    }
+
    /// UDP: receive the next datagram (boundary-preserving, the
    /// incremental decode path — the only consumer-visible decode
    /// path). `Some(bytes)` per datagram — possibly empty (`len=0`
@@ -188,6 +188,71 @@ async fn datagram_round_trip_including_empty() {
    assert!(only_channel_0(&topo));
 }

+#[tokio::test]
+async fn datagram_backlog_tracks_pending_and_drains_fifo() {
+    // The §1.3/§3.3 session-shape diagnostic: `datagram_backlog()` is
+    // the `DatagramHalves.pending` count — None for stream sessions,
+    // 0 at rest, batched arrivals queue FIFO, and a drained queue
+    // reads 0 (the wave-2 harness asserts the bound against it).
+    let registry = ResourceRegistry::new();
+    registry.register("dns", Substrate::Udp, "in-process").await;
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_forward(registry, all_substrate_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
+        .await
+        .expect("udp open");
+    assert_eq!(
+        session.datagram_backlog(),
+        Some(0),
+        "a fresh datagram session has an empty backlog"
+    );
+    let stream_session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Tcp)).await;
+    // This registry entry maps (dns, Tcp) nowhere; open fails typed.
+    assert!(stream_session.is_err(), "unproduced substrate rejects");
+    let stream_session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("tcp open");
+    assert_eq!(
+        stream_session.datagram_backlog(),
+        None,
+        "stream sessions carry no datagram backlog"
+    );
+    stream_session.close().await;
+
+    // Backlog drain posture: two echoed datagrams land back-to-back;
+    // whether the echo batches them into one chunk is timing-shaped,
+    // so the suite pins only the rest state (empty after drains) —
+    // the batched-arrival bound is the wave-2 harness's (§3.3).
+    session.send_datagram(b"first").await.expect("send");
+    session.send_datagram(b"second").await.expect("send");
+    let dg1 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("no io error")
+        .expect("first datagram");
+    let dg2 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("no io error")
+        .expect("second datagram");
+    let mut got = [&dg1[..], &dg2[..]];
+    got.sort();
+    assert_eq!(got, [b"first".as_slice(), b"second".as_slice()]);
+    assert_eq!(
+        session.datagram_backlog(),
+        Some(0),
+        "the backlog drains to empty once all datagrams are pulled"
+    );
+
+    let reaped = session.close().await;
+    assert!(reaped);
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(only_channel_0(&topo));
+}
+
 #[tokio::test]
 async fn wrong_substrate_operations_are_typed() {
    // The POC's WrongSubstrate shape: substrate-shaped operations fail