feat(fuzz): session_opseq target - stateful op-sequence campaign against the session pump

Target 4 from the fuzzing plan (the alkcall-bug-class hunt): a
#[derive(Arbitrary)] op sequence drives the producer session pump
(drive_session_pre_negotiated) over a duplex pair against a fuzz-local
mock backend.

- fuzz/shared/src/session_opseq.rs: op vocabulary (client writes incl.
  invalid-type/oversize-length barrier probes, backend production/EOF,
  exit resolve/fail/drop, control messages, bounded reads, yields),
  kill-guard exit future (ADR-005 observability), delivery model, and
  the invariants: kill-on-Drop (kill_fired == !exit_resolved), exit
  chunk is last with first-issued code -1 on wait-failure (ADR-004),
  lossless per-stream FIFO + single stdout sentinel, stdin
  prefix-losslessness (post-shutdown chunks never delivered), control
  dispatch bounds, take-once allocation, teardown termination.
- Harness-model correction caught by a 60s smoke run before the
  campaign (the alkcall §7.7 pattern, third time): a failed client
  write can mean the session completed and dropped the server duplex
  half while the write was in flight - the harness now drains and
  requires the exit chunk on any write/shutdown error (premature close
  without it is a finding). No adapter code changed.
- 12 committed seeds via the deterministic generator, hand-encoded
  against the arbitrary 1.4.x derive layout; the encoding is pinned by
  a seed-decode test. Existing chunk_frame/negotiation_frame/
  control_json seed corpora are byte-identical to before.
- fuzz/.gitignore: the per-target seed negations never matched (corpus/*
  excludes the parent dir; git cannot re-include beneath an excluded
  dir). Fixed with the !corpus/*/ + corpus/*/* + !corpus/*/seed-*
  recipe - new seed files were silently ignored until now.
- Campaign: 10 min detached via run-detached.sh, exited 0, artifact
  dir empty - 42.3k execs across 33 fork jobs, 0 crash/oom/timeout,
  cov 3934 -> 4075 edges. Grown corpus excised per the corpus policy.

Verification: corpus replay green on stable (9 tests), cargo test 113
+ --all-features 156, clippy stable/wasm/fuzz-shared, fmt, doc,
publish dry-run, wasm check - all pass.

docs/plans/fuzzing.md

@@ -144,7 +144,7 @@ alkcall §7.4).
 | 1 | `chunk_frame` | raw `&[u8]` | `ChunkReader`/`ChunkWriter` over `Cursor` + `block_on` | no-panic; error-shape partition (`ConnectionClosed` only on truncation; `InvalidStreamType` iff `stream_type > 4`; `ChunkTooLarge` iff `length > MAX_CHUNK_LEN`; `Io` impossible on a Cursor); Ok iff header complete + payload complete; round-trip (`write_chunk` then `read_chunk` reproduces stream_type + bytes); 5-byte consumption accounting; peek interleave (`peek_stream_type` → `read_chunk_after_peek` == `read_chunk`, idempotent peek, peek state resets per chunk); never allocates for out-of-range headers |
 | 2 | `negotiation_frame` | raw `&[u8]` | `NegotiationReader`/`NegotiationWriter` + `NegotiateRequest` serde parse + `error_response_bytes` | no-panic; error-shape partition (`ConnectionClosed` only on truncation; `FrameTooLarge` iff `length > MAX_CHUNK_LEN`; `Json` only when the full body was present; `Io` impossible); consumption accounting (4 + length); negotiation JSON round-trip (`NegotiateRequest` → `to_json` → frame → `from_slice` → equal struct); **cross-codec disambiguation** (see §4.1) |
 | 3 | `control_json` | raw `&[u8]` | `ControlMessage::from_slice` / `to_json`, `signal_from_name` | no-panic on any bytes; clean serde errors (never silent); `to_json` → `from_slice` structural round-trip; `signal_from_name` is total (returns `Option`, never panics) |
-| 4 | `session_opseq` (second wave) | `#[derive(Arbitrary)]` op sequence | the producer session pump + mock backend (`src/testing.rs`, `drive_session`) | no-panic; take-once slot semantics; duplicate/replayed session ops leave the live session intact — the alkcall-bug-class target |
+| 4 | `session_opseq` | `#[derive(Arbitrary)]` op sequence | the producer session pump (`drive_session_pre_negotiated`) + a fuzz-local mock backend (`fuzz/shared/src/session_opseq.rs`; the in-crate `MockBackend` is `#[cfg(test)]`-only) | no-panic (driver, pump tasks via the session `JoinError`, harness model); **kill-on-Drop (ADR-005)** — after teardown `kill_fired == !exit_resolved`, and with no exit op issued the kill MUST have fired; **exit chunk is last (ADR-004)** — at most one ctrl_out chunk, `type == "exit"`, `code` matches the first-issued exit op (`-1` on wait-failure or dropped exit sender), nothing observed after it, server never writes stream types 0/3; lossless per-stream FIFO of produced `(len, byte)` patterns with exactly one stdout sentinel (only after the stdout-EOF op) and no stderr sentinel; stdin prefix-losslessness (post-shutdown chunks never delivered); control dispatch never over-counts and dispatches promptly pre-exit; take-once allocation; teardown terminates within the bound |
 
 ### 4.1 Cross-codec disambiguation invariant (target 2, the subtle one)
 
@@ -182,9 +182,10 @@ fuzz/
 ├── fuzz_targets/
 │   ├── chunk_frame.rs         thin fuzz_target! wrapper
 │   ├── negotiation_frame.rs   thin wrapper
-│   └── control_json.rs        thin wrapper
+│   ├── control_json.rs        thin wrapper
+│   └── session_opseq.rs       thin wrapper (typed SessionSequence input)
 ├── shared/               alktty-fuzz-shared — STABLE-toolchain library
-│   └── src/{chunk_frame,negotiation_frame,control_json}.rs
+│   └── src/{chunk_frame,negotiation_frame,control_json,session_opseq}.rs
 ├── corpus/<target>/      committed seeds
 ├── gen_fuzz_seeds.py     deterministic seed generator
 ├── run-detached.sh       §3 detached runner (CWD-independent)
@@ -200,11 +201,17 @@ Root `Cargo.toml` changes (the alkcall §7.7 footguns, both required):
 
 No `#[cfg(fuzzing)]` exposure expected — `ChunkReader`/`ChunkWriter`,
 `NegotiationReader`/`NegotiationWriter`, `NegotiateRequest`,
-`ControlMessage`, and `error_response_bytes` are all already `pub`
+`ControlMessage`, `error_response_bytes`, `drive_session_pre_negotiated`,
+and the `TtyBackend`/`TtyHandle`/`TtyParams` shapes are all already `pub`
 (alkcall needed none of its five either). The one private constant,
 `MAX_CHUNK_LEN`, is wire-stable (ADR-001); the shared crate carries its
 own copy, and the shape invariants assert the rejection boundary so a
-drift is caught.
+drift is caught. The `session_opseq` harness is fuzz-local for the same
+reason `MockBackend`/`TestBackend` are `#[cfg(test)]`-only: a test
+backend in `src/` would either leak into the public API or force the
+whole mock apparatus behind `#[cfg(any(test, fuzzing))]`; the shared
+crate is the fuzz-side home for it (mirrors alkcall, whose `manager_routing`
+harness also lives in `fuzz/shared/`).
 
 ## 6. Corpus policy
 
@@ -232,6 +239,7 @@ deterministic (no randomness) so seeds are reproducible.
 5. Detached campaigns (10 min per target via `run-detached.sh`), triage
    anything found, record results in §8.
 6. Decide on target 4 (`session_opseq`) after 1–3 are clean.
+   **Status: landed — see §8. All four targets are in scope and live.**
 
 ## 8. Progress log (append-only across sessions)
 
@@ -273,6 +281,52 @@ deterministic (no randomness) so seeds are reproducible.
     §6 policy; committed seeds unchanged.
   - §7 step 6 (target 4 `session_opseq`) remains open for a follow-up
     session.
+- **2026-09-28 (later still)** — **§7 step 6 landed: target 4
+  `session_opseq`. All four targets are now live.**
+  - `fuzz/shared/src/session_opseq.rs`: `SessionOp`/`SessionSequence`
+    (`#[derive(Arbitrary)]`, 12-op vocabulary — client writes incl.
+    barrier probes, backend production/EOF, exit resolve/fail/drop,
+    control messages, bounded reads, yields) + a fuzz-local
+    `FuzzBackend` (the in-crate `MockBackend` is `#[cfg(test)]`-only) +
+    the harness driving `drive_session_pre_negotiated` over a
+    `tokio::io::duplex` pair, with a delivery model mirroring §4's
+    invariants. 12 committed seeds via the deterministic generator
+    (hand-encoded against the `arbitrary` 1.4.x derive layout — variant
+    tag `(u32_le * 12) >> 32`, fields LE zero-filled, keep-going byte
+    gating each `Vec` element; the encoding is pinned by a decode test
+    asserting the exact ops each seed decodes to).
+  - Invariants encoded: kill-on-Drop (ADR-005), exit-chunk-is-last
+    (ADR-004), lossless per-stream FIFO + sentinel semantics, stdin
+    prefix-losslessness (post-shutdown chunks never delivered), control
+    dispatch bounds, take-once allocation, teardown termination.
+    Harness bounds: 128 ops, 4 KiB/chunk, 32 KiB produce/write caps,
+    unbounded backend stdin (control dispatch stays prompt).
+  - **Harness-model correction before the campaign** (the alkcall §7.7
+    pattern repeating, third time): a 60s smoke run crashed on
+    `ClientWrite` after the session had completed — a failed client
+    write can mean the *server half of the duplex dropped* because the
+    session finished (exit resolved → pumps join → exit chunk →
+    drainer exits → duplex halves drop) while the write was in flight.
+    The harness now treats any write/shutdown error as "the session
+    closed": it drains the read side and requires the exit chunk (a
+    close without it is a premature close — a real bug). No adapter
+    code changed; the finding was a harness-model gap.
+  - Corpus replay green on stable (9 tests, incl. the seed-decode
+    pin). Full checklist green: 113 + 156 all-features tests, clippy
+    stable/wasm + fuzz-shared clippy, fmt, doc, publish dry-run, wasm
+    check.
+  - Campaign: **10 min detached via `run-detached.sh`, exited 0 with an
+    empty artifact directory — no crashes, hangs, OOMs, or leaks** (all
+    33 fork jobs `oom/timeout/crash: 0/0/0`). 42.3k execs at ~74/s per
+    job (the op-seq target is stateful — each input runs a full tokio
+    session with bounded reads/yields, so exec/s is inherently tens, not
+    millions; the value is interleaving exploration). Cov grew
+    3934 → 4075 edges over the run, still finding coverage at budget
+    end. Grown corpus excised per the §6 policy; 12 committed seeds
+    unchanged; corpus replay green after.
+  - §7 fuzzing plan complete: all four targets live, corpus replay is
+    the standing gate, campaigns pre-release and on adapter/pump/wire
+    changes.
 
 ## 9. Relationship to existing tests
 

fuzz/.gitignore

@@ -1,7 +1,7 @@
 target
 corpus/*
-!corpus/chunk_frame/seed-*
-!corpus/negotiation_frame/seed-*
-!corpus/control_json/seed-*
+!corpus/*/
+corpus/*/*
+!corpus/*/seed-*
 artifacts
-coverage
+coverage

fuzz/Cargo.lock

@@ -78,9 +78,13 @@ name = "alktty-fuzz-shared"
 version = "0.0.0"
 dependencies = [
  "alktty",
+ "arbitrary",
+ "async-trait",
  "bytes",
+ "futures-core",
  "serde_json",
  "tokio",
+ "tokio-stream",
 ]
 
 [[package]]
@@ -94,6 +98,9 @@ name = "arbitrary"
 version = "1.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1"
+dependencies = [
+ "derive_arbitrary",
+]
 
 [[package]]
 name = "async-trait"
@@ -181,6 +188,17 @@ version = "2.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"
 
+[[package]]
+name = "derive_arbitrary"
+version = "1.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "displaydoc"
 version = "0.2.7"

fuzz/Cargo.toml

@@ -35,4 +35,11 @@ test = false
 doc = false
 bench = false
 
+[[bin]]
+name = "session_opseq"
+path = "fuzz_targets/session_opseq.rs"
+test = false
+doc = false
+bench = false
+
 [workspace]

fuzz/README.md

@@ -20,6 +20,7 @@ operating rules live in `docs/plans/fuzzing.md` (adopted from alkcall's
 | `chunk_frame` | `ChunkReader`/`ChunkWriter` (5-byte chunk codec, ADR-001) |
 | `negotiation_frame` | `NegotiationReader`/`NegotiationWriter` + `NegotiateRequest` + `error_response_bytes` + the cross-codec peek-disambiguation seam |
 | `control_json` | `ControlMessage::from_slice`/`to_json` + `signal_from_name` |
+| `session_opseq` | the producer session pump (`drive_session_pre_negotiated`) + a fuzz-local mock backend — the stateful op-sequence target (ADR-004/ADR-005 invariants; the alkcall-bug-class hunt) |
 
 ## Running a campaign — always detached
 
@@ -57,4 +58,16 @@ required by cargo-fuzz); the crate itself stays on stable at MSRV 1.88.
 toolchain file (rustup resolves per directory).
 
 Run campaigns before releases and after touching `src/wire.rs`,
-`src/negotiation.rs`, `src/control.rs`, the adapter, or the session pump.
+`src/negotiation.rs`, `src/control.rs`, the adapter, or the session pump.
+
+## session_opseq notes
+
+The op-sequence target is inherently slower than the byte-format targets
+(each input drives a full tokio session with bounded reads and yields —
+tens of exec/s per worker, not millions). That is expected: the value is
+state-space exploration, not throughput. It uses `tokio::time::timeout`
+internally, so the nightly target needs the `--features time` tokio
+surface already present via the shared crate's deps — no extra flags.
+Its invariants (kill-on-Drop, exit-chunk-is-last, sentinel/FIFO
+semantics) are the ones unit tests assert individually; the fuzzer
+explores interleavings no example test can enumerate.

fuzz/fuzz_targets/session_opseq.rs

@@ -0,0 +1,9 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+
+fuzz_target!(
+    |seq: alktty_fuzz_shared::session_opseq::SessionSequence| {
+        alktty_fuzz_shared::session_opseq::fuzz_session_opseq(&seq);
+    }
+);

fuzz/gen_fuzz_seeds.py

@@ -241,10 +241,277 @@ def control_json_seeds():
     return seeds
 
 
+# session_opseq seeds are arbitrary-encoded `SessionSequence` inputs, so
+# they are hand-encoded against the `arbitrary` 1.4.x derive layout the
+# same way alkcall's manager_routing seeds are: each element of a Vec is
+# gated by a keep-going byte (odd = another element follows; the final
+# gated read before data exhaustion yields true via zero-fill), enum
+# variant tags are 4-byte LE u32s with `(tag * 12) >> 32` selecting by
+# declaration order, and fields follow in declaration order
+# little-endian, zero-filled on short data. The encoding is pinned by
+# the seed-decode test in fuzz/shared/src/session_opseq.rs.
+SESSION_OP_VARIANTS = [
+    "ClientWrite",
+    "ClientCloseWrite",
+    "Stdout",
+    "StdoutEof",
+    "Stderr",
+    "StderrEof",
+    "Exit",
+    "ExitFail",
+    "DropExitTx",
+    "CtrlIn",
+    "ClientReadAll",
+    "Yield",
+]
+
+
+def _le32(v):
+    return struct.pack("<I", v & 0xFFFFFFFF)
+
+
+class OpSeqEncoder:
+    def __init__(self, stderr_enabled):
+        self.buf = bytearray()
+        self.buf.append(1 if stderr_enabled else 0)
+
+    def op(self, name):
+        idx = SESSION_OP_VARIANTS.index(name)
+        self.buf += _le32((idx << 32) // 12)
+
+    def u8(self, v):
+        self.buf.append((v if isinstance(v, int) else v[0]) & 0xFF)
+
+    def u16(self, v):
+        self.buf += struct.pack("<H", v & 0xFFFF)
+
+    def i16(self, v):
+        self.buf += struct.pack("<h", v)
+
+    def end(self):
+        self.buf.append(0)
+        return bytes(self.buf)
+
+
+def session_opseq_seeds():
+    seeds = []
+    n = 0
+
+    def add(blob):
+        nonlocal n
+        seeds.append((f"seed-{n:03d}", blob))
+        n += 1
+
+    add(
+        bytes(
+            [
+                0x00, 0x01, 0xAB, 0xAA, 0xAA, 0x2A, 0x10, 0x00, 0x61, 0x01,
+                0xAB, 0xAA, 0xAA, 0xEA, 0x04,
+            ]
+        )
+    )
+
+    e = OpSeqEncoder(stderr_enabled=True)
+    e.op("Stdout")
+    e.u16(32)
+    e.u8(b"a")
+    e.op("Stderr")
+    e.u16(8)
+    e.u8(b"e")
+    e.op("Yield")
+    e.u8(8)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=True)
+    e.op("ClientWrite")
+    e.u8(0)
+    e.u16(3)
+    e.u8(b"x")
+    e.op("ClientWrite")
+    e.u8(0)
+    e.u16(0)
+    e.u8(0)
+    e.op("ClientWrite")
+    e.u8(0)
+    e.u16(1)
+    e.u8(b"y")
+    e.op("CtrlIn")
+    e.u8(2)
+    e.u16(0)
+    e.u16(0)
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("StdoutEof")
+    e.op("Exit")
+    e.i16(3)
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=True)
+    e.op("Exit")
+    e.i16(7)
+    e.op("Exit")
+    e.i16(43)
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("StdoutEof")
+    e.op("Stderr")
+    e.u16(4)
+    e.u8(b"e")
+    e.op("StderrEof")
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=False)
+    e.op("ExitFail")
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("StdoutEof")
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=True)
+    e.op("ClientWrite")
+    e.u8(200)
+    e.u16(0)
+    e.u8(0)
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("ClientWrite")
+    e.u8(5)
+    e.u16(0)
+    e.u8(0)
+    e.op("StdoutEof")
+    e.op("ClientReadAll")
+    e.u8(63)
+    e.op("Exit")
+    e.i16(-1)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=False)
+    for cols, rows in [(80, 24), (0, 0), (65535, 65535), (120, 40)]:
+        e.op("CtrlIn")
+        e.u8(0)
+        e.u16(cols)
+        e.u16(rows)
+        e.op("CtrlIn")
+        e.u8(1)
+        e.u16(0)
+        e.u16(0)
+        e.op("ClientReadAll")
+        e.u8(4)
+    e.op("Exit")
+    e.i16(0)
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=False)
+    e.op("CtrlIn")
+    e.u8(3)
+    e.u16(99)
+    e.u16(0)
+    e.op("CtrlIn")
+    e.u8(4)
+    e.u16(0)
+    e.u16(0)
+    e.op("ClientWrite")
+    e.u8(4)
+    e.u16(2)
+    e.u8(b"z")
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("StdoutEof")
+    e.op("Exit")
+    e.i16(5)
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=True)
+    e.op("ClientWrite")
+    e.u8(0)
+    e.u16(2)
+    e.u8(b"w")
+    e.op("ClientCloseWrite")
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("StdoutEof")
+    e.op("Stderr")
+    e.u16(4)
+    e.u8(b"e")
+    e.op("StderrEof")
+    e.op("Exit")
+    e.i16(0)
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=False)
+    e.op("DropExitTx")
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("StdoutEof")
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=True)
+    e.op("Stderr")
+    e.u16(4)
+    e.u8(b"e")
+    e.op("StderrEof")
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("Exit")
+    e.i16(9)
+    e.op("StdoutEof")
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    e = OpSeqEncoder(stderr_enabled=False)
+    e.op("ClientWrite")
+    e.u8(0)
+    e.u16(2)
+    e.u8(b"a")
+    e.op("ClientWrite")
+    e.u8(1)
+    e.u16(2)
+    e.u8(b"b")
+    e.op("ClientWrite")
+    e.u8(0)
+    e.u16(2)
+    e.u8(b"c")
+    e.op("Exit")
+    e.i16(1)
+    e.op("Stdout")
+    e.u16(4)
+    e.u8(b"o")
+    e.op("StdoutEof")
+    e.op("ClientReadAll")
+    e.u8(63)
+    add(e.end())
+
+    return seeds
+
+
 def main():
     write_seeds("chunk_frame", chunk_frame_seeds())
     write_seeds("negotiation_frame", negotiation_frame_seeds())
     write_seeds("control_json", control_json_seeds())
+    write_seeds("session_opseq", session_opseq_seeds())
     print("seeds written")
 
 

fuzz/shared/Cargo.toml

@@ -8,7 +8,14 @@ edition = "2021"
 alktty = { path = "../.." }
 serde_json = "1"
 bytes = "1"
-tokio = { version = "1", features = ["rt"], default-features = false }
+futures-core = "0.3"
+tokio = { version = "1", features = ["rt", "macros", "io-util", "sync", "time"], default-features = false }
+tokio-stream = { version = "0.1", default-features = false }
+async-trait = "0.1"
+
+[dependencies.arbitrary]
+version = "1"
+features = ["derive"]
 
 [dev-dependencies]
 serde_json = { version = "1", features = ["preserve_order"] }

fuzz/shared/src/lib.rs

@@ -6,3 +6,4 @@
 pub mod chunk_frame;
 pub mod control_json;
 pub mod negotiation_frame;
+pub mod session_opseq;