fix(channels): duplicate adopt/open must not destroy the live channel; fuzz targets 3-5

Found by the manager_routing fuzz target (docs/research/fuzzing.md \u00a77.8):

- ChannelManager::open_channel / adopt_channel used
  HashMap::insert(...).is_some() as a collision check \u2014 insert REPLACES
  the existing entry, so a duplicate adopt/open installed the new state,
  dropped the live channel's demux_sender (spurious EOF to its readers,
  subsequently routed chunks lost) and still returned
  Err(ChannelExists). Fixed with contains_key pre-check; map untouched
  on collision. Regression tests: adopt_channel_duplicate_id_leaves_
  live_channel_intact, open_channel_duplicate_id_leaves_live_channel_
  intact.

New fuzz targets (\u00a77.4 step 3):
- manager_routing: Arbitrary op sequences over ChannelManager; exact
  counter models (parked/dropped must equal the manager's monotonic
  counters), parked-bytes bound per \u00a76.2-1, clear_all ledger-vs-map
  semantics, drainer-byte reconciliation (lossless routing)
- envelope_semantic: constructors -> serde -> write_frame/read_frame
  structural round-trip; event-type constants; call.error parse-back
- spec_parse: OpRegisterRequest::from_json -> rebuild -> registry
  registration (attacker schemas compile at register, CF-003)
- fuzz/shared/src/arbitrary_value.rs: bounded Arbitrary for
  serde_json::Value; 20 spec_parse + 4 manager_routing seeds
- corpus replay for the new targets in fuzz/shared tests

Verification: cargo test 684 passed (682 + 2 regression); clippy
-D warnings clean (main + fuzz/shared); fmt clean (main + fuzz);
cargo fuzz build clean; 20 s smoke on all three new targets clean
(manager_routing 79k, envelope_semantic 141k, spec_parse 517k runs);
crash input replays clean post-fix

fuzz/shared/Cargo.toml

@@ -7,4 +7,10 @@ edition = "2021"
 [dependencies]
 alkcall = { path = "../.." }
 serde_json = "1"
-tokio = { version = "1", features = ["rt"], default-features = false }
+futures = "0.3"
+bytes = "1"
+tokio = { version = "1", features = ["rt"], default-features = false }
+
+[dependencies.arbitrary]
+version = "1"
+features = ["derive"]

fuzz/shared/src/arbitrary_value.rs

@@ -0,0 +1,108 @@
+//! `Arbitrary` support for `serde_json::Value` — a bounded-depth
+//! recursive generator shared by the semantic targets. Depth and size
+//! bounds keep exec/s high (serde_json's own 128-depth recursion limit
+//! is the decode-side guard; this generator explores structure).
+
+use arbitrary::{Arbitrary, Unstructured};
+use serde_json::{Map, Number, Value};
+
+#[derive(Debug, arbitrary::Arbitrary, Clone)]
+#[allow(dead_code)]
+enum JsonShape {
+    Null,
+    Bool(bool),
+    Number {
+        mantissa: u32,
+        negative: bool,
+    },
+    String {
+        len: u8,
+        byte: u8,
+    },
+    Array {
+        len: u8,
+        items: Vec<JsonShape>,
+    },
+    Object {
+        len: u8,
+        keys: Vec<(u8, u8)>,
+        values: Vec<JsonShape>,
+    },
+}
+
+/// Generate a `Value` with bounded depth: arrays/objects nest at most
+/// 6 levels, each holding at most 6 items. The `u8`-bounded shapes
+/// keep the generator cheap; the semantic target's value space is
+/// structure, not size.
+impl<'a> Arbitrary<'a> for JsonValue {
+    fn arbitrary(u: &mut Unstructured<'a>) -> arbitrary::Result<Self> {
+        let shape = JsonShape::arbitrary(u)?;
+        Ok(JsonValue(shape_to_value(shape, 6)))
+    }
+}
+
+fn shape_to_value(shape: JsonShape, depth: u32) -> Value {
+    if depth == 0 {
+        return Value::Null;
+    }
+    match shape {
+        JsonShape::Null => Value::Null,
+        JsonShape::Bool(b) => Value::Bool(b),
+        JsonShape::Number { mantissa, negative } => {
+            let n = i64::from(mantissa % 1_000_000);
+            Number::from(if negative { -n } else { n }).into()
+        }
+        JsonShape::String { len, byte } => {
+            // Repeat a printable-ish char; embedded control bytes and
+            // invalid UTF-8 come from the fuzzer's byte mutations of
+            // the raw frame in `envelope_frame`, not here.
+            let n = (len % 64) as usize;
+            std::iter::repeat_n((0x41u8 + (byte % 26)) as char, n)
+                .collect::<String>()
+                .into()
+        }
+        JsonShape::Array { len, items } => {
+            let items: Vec<Value> = items
+                .into_iter()
+                .take(6)
+                .map(|s| shape_to_value(s, depth - 1))
+                .collect();
+            let _ = len;
+            Value::Array(items)
+        }
+        JsonShape::Object { keys, values, .. } => {
+            let mut map = Map::new();
+            for (i, (k_len, k_byte)) in keys.into_iter().take(6).enumerate() {
+                let key: String = std::iter::repeat_n(
+                    (0x61u8 + (k_byte % 26)) as char,
+                    (k_len % 16) as usize + 1,
+                )
+                .collect();
+                let value = values
+                    .get(i)
+                    .map(|s| shape_to_value(s.clone(), depth - 1))
+                    .unwrap_or(Value::Null);
+                map.insert(key, value);
+            }
+            Value::Object(map)
+        }
+    }
+}
+
+/// Newtype so `impl Arbitrary` applies (serde_json re-exports `Value`
+/// as `serde_json::Value`; a foreign-type impl is not allowed).
+#[derive(Debug, Clone, PartialEq)]
+pub struct JsonValue(pub Value);
+
+impl std::ops::Deref for JsonValue {
+    type Target = Value;
+    fn deref(&self) -> &Value {
+        &self.0
+    }
+}
+
+impl From<JsonValue> for Value {
+    fn from(value: JsonValue) -> Value {
+        value.0
+    }
+}

fuzz/shared/src/envelope_semantic.rs

@@ -0,0 +1,166 @@
+//! Invariants for the `EventEnvelope` constructors and `CallError`
+//! payloads (ADR-014/ADR-016): structural round-trip through
+//! `write_frame`/`read_frame` for every event kind, `CallError`
+//! serialization never panics and parses back, and the wire event-type
+//! constants survive the constructors.
+
+use crate::arbitrary_value::JsonValue;
+use alkcall::protocol::wire::{
+    CallError, EventEnvelope, FrameError, FrameFramedReader, FrameFramedWriter,
+};
+use std::io::Cursor;
+
+#[derive(Debug, arbitrary::Arbitrary)]
+pub enum EnvelopeKind {
+    Requested {
+        id_len: u8,
+        id_byte: u8,
+        payload: JsonValue,
+    },
+    Responded {
+        id_len: u8,
+        id_byte: u8,
+        output: JsonValue,
+    },
+    Completed {
+        id_len: u8,
+        id_byte: u8,
+    },
+    Published {
+        id_len: u8,
+        id_byte: u8,
+        chunk: JsonValue,
+    },
+    Aborted {
+        id_len: u8,
+        id_byte: u8,
+    },
+    Error {
+        id_len: u8,
+        id_byte: u8,
+        code_len: u8,
+        code_byte: u8,
+        message_len: u8,
+        message_byte: u8,
+        retryable: bool,
+        details: Option<JsonValue>,
+    },
+}
+
+fn repeated_string(len: u8, byte: u8) -> String {
+    // Bounded: at most 64 repetitions of one char. Long strings are a
+    // frame-size concern covered by `envelope_frame`; this target is
+    // about the constructors and structural invariants.
+    let n = (len % 64) as usize;
+    std::iter::repeat_n(byte as char, n).collect()
+}
+
+fn assert_round_trip(envelope: &EventEnvelope) {
+    let mut bytes = Vec::new();
+    {
+        let mut writer = FrameFramedWriter::new(&mut bytes);
+        tokio::runtime::Builder::new_current_thread()
+            .build()
+            .expect("runtime builds")
+            .block_on(writer.write_frame(envelope))
+            .expect("write_frame of any constructor output must succeed");
+    }
+    let mut reader = FrameFramedReader::new(Cursor::new(bytes));
+    let decoded = match tokio::runtime::Builder::new_current_thread()
+        .build()
+        .expect("runtime builds")
+        .block_on(reader.read_frame())
+    {
+        Ok(decoded) => decoded,
+        Err(FrameError::Json(e)) => {
+            panic!("a written envelope must re-decode: {e}");
+        }
+        Err(other) => panic!("a written envelope must re-decode, got {other:?}"),
+    };
+    assert_eq!(
+        &decoded, envelope,
+        "structural round-trip (serde_json key order is not preserved)"
+    );
+}
+
+pub fn fuzz_envelope_semantic(kind: &EnvelopeKind) {
+    let envelope = match kind {
+        EnvelopeKind::Requested {
+            id_len,
+            id_byte,
+            payload,
+        } => EventEnvelope::requested(repeated_string(*id_len, *id_byte), payload.0.clone()),
+        EnvelopeKind::Responded {
+            id_len,
+            id_byte,
+            output,
+        } => EventEnvelope::responded(repeated_string(*id_len, *id_byte), output.0.clone()),
+        EnvelopeKind::Completed { id_len, id_byte } => {
+            EventEnvelope::completed(repeated_string(*id_len, *id_byte))
+        }
+        EnvelopeKind::Published {
+            id_len,
+            id_byte,
+            chunk,
+        } => EventEnvelope::published(repeated_string(*id_len, *id_byte), chunk.0.clone()),
+        EnvelopeKind::Aborted { id_len, id_byte } => {
+            EventEnvelope::aborted(repeated_string(*id_len, *id_byte))
+        }
+        EnvelopeKind::Error {
+            id_len,
+            id_byte,
+            code_len,
+            code_byte,
+            message_len,
+            message_byte,
+            retryable,
+            details,
+        } => {
+            let mut error = CallError::new(
+                repeated_string(*code_len, *code_byte),
+                repeated_string(*message_len, *message_byte),
+                *retryable,
+            );
+            if let Some(details) = details {
+                error = error.with_details(details.0.clone());
+            }
+            EventEnvelope::error(repeated_string(*id_len, *id_byte), &error)
+        }
+    };
+
+    // The event-type constants are wire-stable; the constructors must
+    // emit exactly them.
+    let expected_type = match kind {
+        EnvelopeKind::Requested { .. } => alkcall::protocol::wire::EVENT_REQUESTED,
+        EnvelopeKind::Responded { .. } => alkcall::protocol::wire::EVENT_RESPONDED,
+        EnvelopeKind::Completed { .. } => alkcall::protocol::wire::EVENT_COMPLETED,
+        EnvelopeKind::Published { .. } => alkcall::protocol::wire::EVENT_PUBLISHED,
+        EnvelopeKind::Aborted { .. } => alkcall::protocol::wire::EVENT_ABORTED,
+        EnvelopeKind::Error { .. } => alkcall::protocol::wire::EVENT_ERROR,
+    };
+    assert_eq!(envelope.r#type, expected_type);
+
+    // The `payload` field is schema-free JSON: any Value must survive
+    // a serde round-trip intact (the serde_json 128-depth recursion
+    // limit is a decode-side guard, not an encode-side one).
+    let as_value = serde_json::to_value(&envelope).expect("envelope serializes");
+    let reparsed: EventEnvelope =
+        serde_json::from_value(as_value).expect("a serialized envelope re-parses");
+    assert_eq!(
+        reparsed, envelope,
+        "serde round-trip is structural identity"
+    );
+
+    // A `call.error` payload always parses back into a `CallError`
+    // (ADR-016: the error schema is closed and serde-derivable).
+    if matches!(kind, EnvelopeKind::Error { .. }) {
+        let parsed: Result<CallError, _> = serde_json::from_value(envelope.payload.clone());
+        assert!(
+            parsed.is_ok(),
+            "call.error payload must parse back as CallError, got {:?}",
+            envelope.payload
+        );
+    }
+
+    assert_round_trip(&envelope);
+}

fuzz/shared/src/lib.rs

@@ -3,5 +3,9 @@
 //! exercise the same invariant checks against every committed corpus
 //! entry (the quinn CI pattern) without a nightly toolchain.
 
+pub mod arbitrary_value;
 pub mod chunk_header;
 pub mod envelope_frame;
+pub mod envelope_semantic;
+pub mod manager_routing;
+pub mod spec_parse;

fuzz/shared/src/manager_routing.rs

@@ -0,0 +1,394 @@
+//! Invariants for `ChannelManager` routing under adversarial op
+//! sequences (ADR-039/ADR-040). The harness mirrors the manager's
+//! documented behavior exactly — parked-chunk and drop counters must
+//! match to the unit after every op, parked bytes must respect the
+//! per-channel cap bound, and every byte routed into a known channel
+//! must be readable until an EOF sentinel (lossless bounded-buffer
+//! routing, ADR-040 REQ-CH-05). The §6.2-1 probe: distinct
+//! never-adopted channel ids each park up to 64 chunks — the fuzzer
+//! explores whether the total can exceed that documented bound.
+
+use std::collections::{HashMap, VecDeque};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+
+use bytes::Bytes;
+
+use alkcall::channels::manager::{ChannelManager, ChannelSide, ManagerError};
+use alkcall::channels::mux::MuxRunner;
+use alkcall::channels::reassembly::{MpscRecvStream, DEFAULT_BUFFER_CAP};
+
+/// The documented per-channel early-arrival cap (`EARLY_ARRIVAL_CAP`,
+/// private in `manager.rs`; review 006 E-04 treats 64 parked chunks as
+/// the observable bound).
+pub const EARLY_ARRIVAL_CAP: usize = 64;
+
+/// The maximum chunk payload length (16 MiB, ADR-034).
+pub const MAX_CHUNK_LEN: u32 = 16 * 1024 * 1024;
+
+#[derive(Debug, arbitrary::Arbitrary, Clone)]
+pub enum ManagerOp {
+    Route { channel_id: u32, len: u16, byte: u8 },
+    Open { alpn_idx: u8, opener_idx: u8 },
+    Adopt { channel_id: u32 },
+    Teardown { channel_id: u32 },
+    ClearAll,
+}
+
+#[derive(Debug, arbitrary::Arbitrary)]
+pub struct ManagerSequence {
+    pub ops: Vec<ManagerOp>,
+}
+
+/// Exact model of the manager's early-arrival bookkeeping.
+#[derive(Default)]
+struct ParkModel {
+    /// Parked chunk payloads per channel id, FIFO (0-length marks the
+    /// EOF sentinel a drained queue will deliver).
+    parked: HashMap<u32, VecDeque<u16>>,
+    /// Model of the monotonic `early_arrival_count`.
+    parked_total: u64,
+    /// Model of the monotonic `dropped_unknown_chunks`.
+    dropped_total: u64,
+}
+
+impl ParkModel {
+    fn route_to_unknown(&mut self, channel_id: u32, len: u16) {
+        let queue = self.parked.entry(channel_id).or_default();
+        if queue.len() >= EARLY_ARRIVAL_CAP {
+            self.dropped_total += 1;
+        } else {
+            queue.push_back(len);
+            self.parked_total += 1;
+        }
+    }
+
+    fn parked_bytes(&self) -> u64 {
+        self.parked
+            .values()
+            .map(|q| q.iter().map(|l| u64::from(*l)).sum::<u64>())
+            .sum()
+    }
+
+    /// Bytes of a drained queue that are readable (everything before
+    /// the first zero-length sentinel — the reassembled stream latches
+    /// EOF there and drops the rest, reassembly.rs `poll_read`).
+    fn readable_bytes_until_sentinel(queue: &VecDeque<u16>) -> u64 {
+        let mut total = 0;
+        for len in queue {
+            if *len == 0 {
+                break;
+            }
+            total += u64::from(*len);
+        }
+        total
+    }
+}
+
+struct ManagerHarness {
+    manager: ChannelManager,
+    model: ParkModel,
+    /// Readable-byte model per drained channel id (what its drainer
+    /// will have read once the channel ends). Drainers accumulate:
+    /// `open_channel` then `adopt_channel` on the same id each install
+    /// a fresh stream; earlier drainers end at the sender drop but the
+    /// id's routed bytes must reconcile exactly once, so each new
+    /// drainer supersedes the id's model entry.
+    readable: HashMap<u32, u64>,
+    /// Byte totals for drainers that ended mid-sequence (sender drop
+    /// on adopt-over-adopt) — reconciled into the post-sequence total.
+    retired_drainer_bytes: u64,
+    /// Channels whose stream latched EOF (sentinel routed while known).
+    sentinel_seen: HashMap<u32, bool>,
+    /// (channel_id, bytes-read counter, drainer task).
+    drainers: Vec<(u32, Arc<AtomicU64>, tokio::task::JoinHandle<()>)>,
+    _mux_task: tokio::task::JoinHandle<()>,
+}
+
+impl ManagerHarness {
+    fn new() -> Self {
+        let (_client, server) = tokio::io::duplex(4096);
+        let (_reader, writer) = tokio::io::split(server);
+        let (mux_handle, runner) = MuxRunner::new(Box::new(writer));
+        let mux_task = tokio::spawn(async move {
+            let _ = runner.run().await;
+        });
+        let manager = ChannelManager::new(
+            mux_handle,
+            64,
+            DEFAULT_BUFFER_CAP,
+            None,
+            ChannelSide::Accept,
+        );
+        Self {
+            manager,
+            model: ParkModel::default(),
+            readable: HashMap::new(),
+            retired_drainer_bytes: 0,
+            sentinel_seen: HashMap::new(),
+            drainers: Vec::new(),
+            _mux_task: mux_task,
+        }
+    }
+
+    async fn step(&mut self, op: &ManagerOp, max_payload: usize) -> Result<(), String> {
+        match op {
+            ManagerOp::Route {
+                channel_id,
+                len,
+                byte,
+            } => {
+                let payload_len = (*len as usize) % (max_payload + 1);
+                let len = payload_len as u16;
+                let known = self.manager.has_channel(*channel_id);
+                self.manager
+                    .route_payload(*channel_id, Bytes::from(vec![*byte; payload_len]))
+                    .await;
+                if known {
+                    if !self.sentinel_seen(*channel_id) {
+                        *self.readable.entry(*channel_id).or_insert(0) += u64::from(len);
+                        if len == 0 {
+                            self.sentinel_seen.insert(*channel_id, true);
+                        }
+                    }
+                } else {
+                    self.model.route_to_unknown(*channel_id, len);
+                }
+            }
+            ManagerOp::Open {
+                alpn_idx,
+                opener_idx,
+            } => {
+                let alpn = pick(alpn_idx, ["alk/tty", "alk/sub", "alk/x"]);
+                let opener = pick(opener_idx, ["alice", "bob", "carol"]);
+                match self.manager.open_channel(alpn, opener, None).await {
+                    Ok((_id, _send, recv)) => self.spawn_drainer(_id, recv),
+                    // The per-connection cap is a documented bound.
+                    Err(ManagerError::TooManyChannels { .. }) => {}
+                    Err(other) => return Err(format!("open_channel: unexpected {other:?}")),
+                }
+            }
+            ManagerOp::Adopt { channel_id } => {
+                match self
+                    .manager
+                    .adopt_channel(*channel_id, "alk/tty", None)
+                    .await
+                {
+                    Ok((_send, recv)) => {
+                        // Adoption drains the parked queue FIFO into
+                        // the receiver — readable until a sentinel.
+                        if let Some(queue) = self.model.parked.remove(channel_id) {
+                            let drained_readable = ParkModel::readable_bytes_until_sentinel(&queue);
+                            *self.readable.entry(*channel_id).or_insert(0) += drained_readable;
+                            if queue.contains(&0) {
+                                self.sentinel_seen.insert(*channel_id, true);
+                            }
+                        }
+                        self.spawn_drainer(*channel_id, recv);
+                    }
+                    Err(ManagerError::ChannelExists(_)) => {}
+                    Err(ManagerError::TooManyChannels { .. }) => {}
+                    Err(other) => return Err(format!("adopt_channel: unexpected {other:?}")),
+                }
+            }
+            ManagerOp::Teardown { channel_id } => {
+                match self.manager.teardown_channel(*channel_id) {
+                    Ok(_task) => {}
+                    Err(ManagerError::UnknownChannel(_)) => {}
+                    Err(other) => return Err(format!("teardown_channel: unexpected {other:?}")),
+                }
+            }
+            ManagerOp::ClearAll => {
+                let dropped_before = self.manager.dropped_unknown_chunks();
+                let _ledger_entries = self.manager.clear_all();
+                // clear_all returns the opener-ledger entries (ADR-047
+                // §7 — the per-identity decrement source), not the
+                // channel map: adopted channels carry no ledger entry
+                // (the remote side owns their count) and torn-down
+                // channels keep theirs.
+                self.model.parked.clear();
+                assert_eq!(
+                    self.manager.dropped_unknown_chunks(),
+                    dropped_before,
+                    "clear_all does not reset the monotonic drop counter"
+                );
+                assert_eq!(
+                    self.manager.open_count(),
+                    0,
+                    "clear_all empties the channel map"
+                );
+            }
+        }
+        self.assert_counters()
+    }
+
+    fn sentinel_seen(&self, channel_id: u32) -> bool {
+        self.sentinel_seen
+            .get(&channel_id)
+            .copied()
+            .unwrap_or(false)
+    }
+
+    fn spawn_drainer(&mut self, channel_id: u32, recv: MpscRecvStream) {
+        // A fresh drainer for an id supersedes any earlier one: the
+        // earlier stream's sender was dropped (teardown) or routed
+        // past an EOF sentinel, so its readable-byte model is retired
+        // into the reconciled total and the id's model restarts.
+        if let Some(bytes) = self.readable.remove(&channel_id) {
+            self.retired_drainer_bytes += bytes;
+        }
+        self.sentinel_seen.remove(&channel_id);
+        let counter = Arc::new(AtomicU64::new(0));
+        let task = tokio::spawn(drain_task(recv, Arc::clone(&counter)));
+        self.drainers.push((channel_id, counter, task));
+    }
+
+    fn assert_counters(&self) -> Result<(), String> {
+        let actual_parked = self.manager.early_arrival_count();
+        let actual_dropped = self.manager.dropped_unknown_chunks();
+        if self.model.parked_total != actual_parked {
+            return Err(format!(
+                "parked-chunk model {} != manager counter {actual_parked}",
+                self.model.parked_total
+            ));
+        }
+        if self.model.dropped_total != actual_dropped {
+            return Err(format!(
+                "dropped-chunk model {} != manager counter {actual_dropped}",
+                self.model.dropped_total
+            ));
+        }
+        Ok(())
+    }
+}
+
+/// Drain a channel's reassembled stream to EOF, counting bytes read.
+/// Runs until Ok(0) — sender drop (teardown / clear_all) or EOF
+/// sentinel both end the stream after queued bytes drain.
+async fn drain_task(mut recv: MpscRecvStream, counter: Arc<AtomicU64>) {
+    use tokio::io::AsyncReadExt;
+    let mut buf = [0u8; 512];
+    loop {
+        match recv.read(&mut buf).await {
+            Ok(0) => break,
+            Ok(n) => {
+                counter.fetch_add(n as u64, Ordering::Relaxed);
+            }
+            Err(_) => break,
+        }
+    }
+}
+
+fn pick<const N: usize>(idx: &u8, options: [&'static str; N]) -> &'static str {
+    options[(*idx as usize) % N]
+}
+
+/// The whole-sequence driver. Bounds: 256 ops and 4 KiB payloads keep
+/// exec/s high while covering the DoS class (distinct-id spray: a
+/// 256-op sequence at 4 KiB parks up to ~1 MiB, and the exact counter
+/// plus cap-bound invariants are what the fuzzer must not break).
+pub fn fuzz_manager_routing(seq: &ManagerSequence) {
+    const MAX_OPS: usize = 256;
+    const MAX_PAYLOAD: usize = 4096;
+
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("current-thread runtime builds")
+        .block_on(async {
+            let mut harness = ManagerHarness::new();
+            for (i, op) in seq.ops.iter().take(MAX_OPS).enumerate() {
+                if let Err(msg) = harness.step(op, MAX_PAYLOAD).await {
+                    panic!("invariant violated at op {i} ({op:?}): {msg}");
+                }
+                // The parked-bytes total must respect the documented
+                // per-channel bound summed over distinct parked ids —
+                // the §6.2-1 property.
+                let bound = harness.model.parked.len() as u64
+                    * EARLY_ARRIVAL_CAP as u64
+                    * u64::from(MAX_CHUNK_LEN);
+                assert!(
+                    harness.model.parked_bytes() <= bound,
+                    "parked bytes {} exceed the documented bound {bound} at op {i}",
+                    harness.model.parked_bytes()
+                );
+            }
+            // Post-sequence: clear_all (the transport-EOF path) drops
+            // every sender — all drainers run to EOF and their byte
+            // totals must reconcile with the readable-byte model
+            // (lossless bounded-buffer routing, sentinel semantics
+            // included).
+            harness
+                .step(&ManagerOp::ClearAll, MAX_PAYLOAD)
+                .await
+                .expect("clear_all");
+            let mut expected_total = harness.retired_drainer_bytes;
+            for bytes in harness.readable.values() {
+                expected_total += bytes;
+            }
+            let mut actual_total = 0u64;
+            for (_id, counter, task) in harness.drainers.drain(..) {
+                let _ = task.await;
+                actual_total += counter.load(Ordering::Relaxed);
+            }
+            assert_eq!(
+                actual_total, expected_total,
+                "routed-to-known bytes must reconcile with drained bytes (lossless reassembly)"
+            );
+        });
+}
+
+#[cfg(test)]
+mod corpus_replay {
+    use super::*;
+
+    /// Replay the committed op-sequence corpus through the driver.
+    /// Note `fuzz_manager_routing` spawns a current-thread runtime per
+    /// call; each sequence runs to completion inside it.
+    #[test]
+    fn committed_corpus_replays_through_the_invariants() {
+        let corpus =
+            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../corpus/manager_routing");
+        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: ManagerSequence =
+                arbitrary::Arbitrary::arbitrary_take_rest(u).expect("seed decodes");
+            fuzz_manager_routing(&seq);
+            count += 1;
+        }
+        assert!(
+            count >= 2,
+            "committed seed corpus is present, found {count}"
+        );
+    }
+
+    /// Semantic fixtures: the op sequences that pinned the adopt/open
+    /// insert-replace bug, hand-encoded through `Unstructured` bytes
+    /// the same way libFuzzer would feed them.
+    #[test]
+    fn duplicate_adopt_sequence_survives_the_fix() {
+        // Encoded via the mr-decode tool's format (arbitrary 1.4.x):
+        // 3 adopts of one id, a route, an EOF sentinel route.
+        let seq = ManagerSequence {
+            ops: vec![
+                ManagerOp::Adopt { channel_id: 42 },
+                ManagerOp::Adopt { channel_id: 42 },
+                ManagerOp::Adopt { channel_id: 42 },
+                ManagerOp::Route {
+                    channel_id: 42,
+                    len: 2418,
+                    byte: 121,
+                },
+                ManagerOp::Route {
+                    channel_id: 0,
+                    len: 0,
+                    byte: 0,
+                },
+            ],
+        };
+        fuzz_manager_routing(&seq);
+    }
+}

fuzz/shared/src/spec_parse.rs

@@ -0,0 +1,138 @@
+//! Invariants for the `op/register` DTO and the remote-spec rebuild
+//! path (`rebuild_spec_for` via `OpRegisterRequest::from_json`): no
+//! panic on any JSON shape, registration is always a `Result`, schemas
+//! compiled at registration are attacker-shaped but never panic the
+//! registry, and `spec_to_json_pub` round-trips through the rebuild.
+
+use alkcall::protocol::wire::ResponseEnvelope;
+use alkcall::registry::op_register::{op_register_spec, OpRegisterRequest};
+use alkcall::registry::registration::{
+    make_handler, make_sink_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
+    OperationProvenance, OperationRegistry,
+};
+use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
+
+/// The registry-facing registration the fuzzer drives. Registration is
+/// where wire-announced schemas compile (`jsonschema::options().build`
+/// at `register`, CF-003 fail-closed) — a pathological-but-compilable
+/// remote schema is the §5 CPU-amplification class this target probes.
+fn register_spec(registry: &OperationRegistry, spec: &OperationSpec) -> Result<(), String> {
+    let kind = match spec.op_type {
+        OperationType::Query | OperationType::Mutation => {
+            HandlerKind::Once(make_handler(|input, _ctx| async move {
+                ResponseEnvelope::ok("fuzz", input)
+            }))
+        }
+        OperationType::Sub => HandlerKind::Stream(make_streaming_handler(|_input, _ctx| {
+            futures::stream::empty()
+        })),
+        OperationType::Pub => {
+            HandlerKind::Sink(make_sink_handler(|_input, _ctx, _stream| async move {
+                ResponseEnvelope::ok("fuzz", serde_json::Value::Null)
+            }))
+        }
+    };
+    registry
+        .register(HandlerRegistration::new(
+            spec.clone(),
+            kind,
+            OperationProvenance::FromCall,
+            None,
+            None,
+            alkcall::core::types::Capabilities::new(),
+        ))
+        .map_err(|e| format!("register failed: {e}"))
+}
+
+pub fn fuzz_spec_parse(data: &[u8]) {
+    // Layer 1: raw bytes as JSON — the wire's first parse. Any byte
+    // sequence must be a clean error, never a panic.
+    let Ok(value) = serde_json::from_slice::<serde_json::Value>(data) else {
+        return;
+    };
+
+    // Layer 2: the op/register DTO parse (`rebuild_spec_for` inside).
+    // Attacker-shaped JSON becomes a compiled `OperationSpec` here —
+    // or a clean `INVALID_INPUT`.
+    let request = match OpRegisterRequest::from_json(&value) {
+        Ok(request) => request,
+        Err(call_error) => {
+            assert_eq!(
+                call_error.code, "INVALID_INPUT",
+                "op/register parse failures are INVALID_INPUT, got {call_error:?}"
+            );
+            return;
+        }
+    };
+
+    // Layer 3: the rebuilt spec serializes back to the wire shape.
+    // (Structural equality with the input is NOT asserted — the wire
+    // shape is a projection of the spec, and rebuild normalizes; the
+    // round-trip invariant is that `spec_to_json_pub` output re-parses
+    // to the same spec.)
+    let wire = alkcall::registry::discovery::spec_to_json_pub(&request.spec);
+    let reparsed = OpRegisterRequest::from_json(&wire)
+        .expect("spec_to_json_pub output must re-parse through op/register");
+    assert_eq!(
+        reparsed.spec, request.spec,
+        "spec_to_json_pub → from_json round-trips the rebuilt spec"
+    );
+
+    // Layer 4: registration compiles the attacker-shaped schemas
+    // (fail-closed per CF-003: an uncompilable schema is an Err).
+    let registry = OperationRegistry::new();
+    match register_spec(&registry, &request.spec) {
+        Ok(()) => {}
+        Err(message) => {
+            assert!(
+                message.contains("failed to compile") || message.contains("handler kind mismatch"),
+                "registration failures are schema-compile or handler-kind errors, got: {message}"
+            );
+            return;
+        }
+    }
+
+    // Layer 5: the registered spec survives registry bookkeeping
+    // unchanged, and the compiled validator exists for the op.
+    let registered = registry
+        .registration(&request.spec.name)
+        .expect("registered spec is retrievable");
+    assert_eq!(registered.spec, request.spec);
+
+    // Layer 6: a registration with `replace` semantics — re-register
+    // the same name must succeed (HashMap insert path) and the
+    // registry must never panic on repeated attacker-shaped inserts.
+    let _ = register_spec(&registry, &request.spec);
+
+    // Layer 7: the `op/register` op's own spec round-trips (sanity —
+    // the surface this fuzz target models).
+    let surface = op_register_spec(AccessControl::default());
+    assert_eq!(surface.visibility, Visibility::External);
+    let _ = OpRegisterRequest::from_json(&serde_json::json!({
+        "spec": alkcall::registry::discovery::spec_to_json_pub(&surface),
+        "replace": true,
+    }))
+    .expect("op/register's own spec parses");
+}
+
+#[cfg(test)]
+mod corpus_replay {
+    /// Replay the new targets' corpora through their invariant
+    /// functions (quinn CI pattern — see `envelope_frame`'s twin).
+    #[test]
+    fn spec_parse_corpus_replays() {
+        let corpus =
+            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../corpus/spec_parse");
+        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");
+            crate::spec_parse::fuzz_spec_parse(&data);
+            count += 1;
+        }
+        assert!(
+            count >= 16,
+            "committed seed corpus is present, found {count}"
+        );
+    }
+}