test(producer): the forwarded gate — gate miss/pass, residual race, scope-gate ordering (plan 003 W1)

- New tests/producer_forwarded_gate.rs (the establisher variant's pins;
  the base shape's tests/producer_forwarded.rs untouched):
  - gate miss: typed channel:open_failed with details.reason ==
    unknown_resource, no phantom channel, the pump handler never runs
    (the gate's invocation witness IS set — the miss is at the gate)
  - gate pass: the base serving flow unchanged — parsed params
    witnessed verbatim + full byte round-trip through the
    establisher-registered variant
  - residual race: gate pass + route gone at handler time → the
    handler's error path reaps via teardown-at-birth (the base
    correlation-miss semantics re-pinned under the established shape)
  - scope-gate ordering: tunnel:forwarded-less identity is FORBIDDEN
    and the gate closure is never invoked (ACL precedes the
    establishment phase)
  - the establisher malformed-params delta pinned: schema gates
    unchanged (INVALID_INPUT); an unknown top-level field becomes a
    typed channel:open_failed (reason handler_error) and the parse
    precedes the gate
- No harness changes (wire_with + Topology.consumer_registry as-is)
- Verification: cargo test green (14 suites, 0 failures); clippy
  --all-targets -D warnings native + wasm-target clippy clean (wasm
  --all-targets cannot compile on this baseline — pre-existing,
  mio/socket2 dev-deps do not target wasm; see the task Notes);
  cargo fmt --check clean; cargo check --target
  wasm32-unknown-unknown clean; --all-features test run green

Tasks: tunnels/corrmiss-gate-tests (status → completed)

tasks/tunnels/corrmiss-gate-tests.md

@@ -1,7 +1,7 @@
 ---
 id: tunnels/corrmiss-gate-tests
 name: "Tests: the forwarded gate — typed miss at the open, gate-pass flow, residual race, scope-gate ordering"
-status: pending
+status: completed
 depends_on: [tunnels/corrmiss-forwarded-gate]
 scope: narrow
 risk: low
@@ -90,8 +90,100 @@ explicitly (do not silently assume).
 
 ## Notes
 
-> To be filled by implementation agent
+- New test file `tests/producer_forwarded_gate.rs` — the
+  establisher-variant probes live in their own file (the base shape's
+  `tests/producer_forwarded.rs` stays untouched), reusing the
+  harness's `wire_with` + `Topology.consumer_registry` shape with no
+  harness changes, exactly as the `forwarded-receipt` task's notes
+  predicted.
+- The gate under test closes over a `std::sync::Mutex<HashSet<u32>>`
+  of waiting tokens (the gate is a sync closure — no awaits) and sets
+  an `Arc<AtomicBool>` witness on entry; the witness is what the
+  miss/ordering pins assert against. Poisoned-lock handling follows
+  convention 2 (`unwrap_or_else(|e| e.into_inner())`).
+- The five tests:
+  1. `gate_miss_fails_typed_no_phantom_no_handler` — AC 1: open with
+     an absent token → `Err(CallError)` with
+     `code == "channel:open_failed"` and
+     `details.reason == "unknown_resource"` (the existing
+     `From<TunnelEstablishError>` table, 1:1); after a settle sleep
+     the opener-side manager has no channels (the wrapper's transient
+     allocation was torn down — ADR-049 §5, no phantom); the gate's
+     witness IS set (the miss is at the gate, not before it) and the
+     pump handler's witness is unmoved (the handler never ran).
+  2. `gate_pass_serves_the_unchanged_serving_flow` — AC 2: token in
+     the map + correlated halves → the open resolves with the
+     wrapper's `channel_id`; the injected handler is witnessed with
+     the parsed params verbatim; the full dispatcher-pump dance
+     (adopt + driving echo) round-trips bytes end to end — the
+     `producer_forwarded.rs` e2e shape driven through the
+     establisher-registered variant.
+  3. `residual_race_still_reaps_cleanly` — AC 3: the token is in the
+     map (the gate passes, the open RESOLVES at the call plane) but
+     the halves vec is empty at handler time → the handler's error
+     path drops the stream and returns → teardown-at-birth reaps.
+     Re-pins the base shape's `forwarded_correlation_miss_
+     tears_channel_down` semantics under the establisher variant —
+     the establisher-pass→handler-run window is the residual
+     `correlation-miss` class.
+  4. `scope_gate_precedes_the_establisher` — AC 4: an identity
+     holding only `tunnel:open` → `FORBIDDEN` naming the missing
+     scope, and the gate's witness stays unmoved even though the
+     token WOULD have passed the gate (the tie-break is the order —
+     schema → ACL → allocate → establisher → handler — pinning that
+     the registry runs the ACL before the establishment phase).
+  5. `malformed_params_fail_typed_at_the_establisher` — AC 5: the
+     establisher-shape malformed-params delta IS separately
+     testable through the public surface, so it is pinned: the
+     schema gates (`listen: 0`, missing `peer`) fire first exactly
+     as in the base shape (`INVALID_INPUT`, the test explicitly
+     re-checks both), but an unknown top-level field — passes the
+     schema's additive posture, fails the establisher's serde parse
+     — becomes a TYPED `channel:open_failed` with
+     `details.reason == "handler_error"` (vs the base shape's
+     success-with-channel_id + silent birth-teardown: the improvement
+     design decision 1 pins). The establisher's parse precedes the
+     gate (the gate's witness stays unmoved), the wrapper tears the
+     transient allocation down, and the handler never runs.
+- Verification deviation, discovered and characterized, pre-existing:
+  `cargo clippy --target wasm32-unknown-unknown --all-targets` cannot
+  compile on this repo's baseline (the wasm clippy's `--all-targets`
+  pulls the test/dev targets whose `mio`/`socket2` deps do not
+  compile for wasm) — verified by removing the new test file and
+  re-running: identical failure without the change. The repo's
+  AGENTS.md verification list specifies wasm clippy WITHOUT
+  `--all-targets`
+  (`cargo clippy --target wasm32-unknown-unknown -- -D warnings`),
+  which is green; the AC's "(native + wasm32)" is read through the
+  AGENTS.md command list as `--all-targets` native + the no-targets
+  wasm clippy (same posture the API task's summary pinned: clippy
+  `--all-targets` native, wasm clippy clean).
 
 ## Summary
 
-> To be filled on completion
+Landed W1's test item: five tests in the new
+`tests/producer_forwarded_gate.rs` pinning the establisher variant's
+behaviors — (1) gate miss → typed `channel:open_failed` /
+`details.reason == "unknown_resource"` / no phantom channel / handler
+never runs; (2) gate pass → the base serving flow unchanged (params
+witnessed verbatim + full byte round-trip through the
+establisher-registered variant); (3) the residual race (gate pass,
+route gone at handler time) reaps cleanly via teardown-at-birth,
+re-pinning the base shape's correlation-miss semantics under the
+establisher variant; (4) scope-gate ordering — a scope-less identity
+is FORBIDDEN and the gate closure is never invoked (witness unmoved);
+(5) the establisher-variant malformed-params delta pinned explicitly:
+schema gates unchanged (`INVALID_INPUT`) but an unknown top-level
+field becomes a typed `channel:open_failed` with
+`details.reason == "handler_error"` and the parse precedes the gate.
+Verification green: `cargo test` (14 suites, 0 failures),
+`cargo clippy --all-targets -- -D warnings` (native) clean,
+wasm clippy `cargo clippy --target wasm32-unknown-unknown -- -D
+warnings` clean (see Notes for the `--all-targets`-wasm deviation —
+pre-existing, dev-deps do not target wasm),
+`cargo fmt --check` clean,
+`cargo check --target wasm32-unknown-unknown` clean (default crate
+stays wasm-clean), `cargo test --all-features` green, `cargo doc
+--no-deps` clean. Harness unchanged. Follow-ups: the sibling-repo
+adoptions (`corrmiss-alkssh-adopt`, `corrmiss-alksocks-adopt`) — the
+W1 wave's tests are green.

tests/producer_forwarded_gate.rs

@@ -0,0 +1,586 @@
+//! Integration tests for the forwarded op's establisher variant
+//! (ADR-008 Amendment 2 —
+//! `register_tunnel_forwarded_openable_with_establisher` + the
+//! injected [`ForwardedGate`]): the four pins plan 003 §W1's
+//! "Tests:" bullet asks for, mirroring `tests/producer_forwarded.rs`
+//! (the base serving shape is that file's; these probes cover the
+//! establisher-registered delta).
+//!
+//! Covered:
+//! - the gate MISS: a `listen` token with no waiting session → the
+//!   open fails as a typed `channel:open_failed` with
+//!   `details.reason == "unknown_resource"`; the transiently
+//!   allocated channel is torn down by the wrapper (ADR-049 §5 — the
+//!   reply is the typed error, not a phantom channel); the pump
+//!   handler NEVER runs.
+//! - the gate PASS: the existing serving flow is unchanged — the
+//!   injected handler receives the parsed params + the `BiStream`,
+//!   the byte round-trip works end to end.
+//! - the residual race: gate passes, route gone at handler time →
+//!   the handler's error path drops the stream → teardown-at-birth
+//!   reaps the forwarded channel (re-pinned under the establisher
+//!   variant; the establisher-pass→handler-run window is the residual
+//!   `correlation-miss` class).
+//! - the ordering: an identity WITHOUT `tunnel:forwarded` → FORBIDDEN
+//!   and the gate closure is never invoked (the registry runs the ACL
+//!   before the open op's establishment phase — schema → ACL →
+//!   allocate → establisher → handler).
+//! - the establisher-variant malformed-params delta: an unknown
+//!   top-level field (passes the schema's additive posture, fails the
+//!   establisher's serde parse) → a typed `channel:open_failed` with
+//!   `details.reason == "handler_error"`, NOT the base shape's
+//!   success-then-birth-teardown; the parse precedes the gate (the
+//!   gate never sees malformed input).
+
+mod harness;
+
+use std::collections::HashSet;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+
+use alkcall::channels::manager::ChannelManager;
+use alkcall::core::auth::Identity;
+use alkcall::protocol::wire::CallError;
+use alktunnels::params::{TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE};
+use alktunnels::producer::{
+    register_tunnel_forwarded_openable_with_establisher, ForwardedGate, TargetHandle,
+    TunnelEstablishError,
+};
+
+use harness::{wire_with, RegistrationMode};
+
+/// The opener-side identity holding ONLY `tunnel:forwarded` (the
+/// ingress grant; `tests/producer_forwarded.rs`'s shape).
+fn forwarded_only_identity() -> Identity {
+    Identity {
+        id: "opener-forwarded-only".to_string(),
+        scopes: vec![TUNNEL_FORWARDED_SCOPE.to_string()],
+        resources: std::collections::HashMap::new(),
+    }
+}
+
+/// A plain dial-egress identity (`tunnel:open` only — no
+/// `tunnel:forwarded`; the ordering probe's identity).
+fn open_only_identity() -> Identity {
+    Identity {
+        id: "opener-open-only".to_string(),
+        scopes: vec![TUNNEL_OPEN_SCOPE.to_string()],
+        resources: std::collections::HashMap::new(),
+    }
+}
+
+/// What the assembly's test handler received per forwarded open (the
+/// invocation probe — unmoved = the handler never ran).
+#[derive(Default)]
+struct Received {
+    params: Option<alktunnels::params::TunnelForwardedParams>,
+}
+
+type ReceivedWitness = Arc<tokio::sync::Mutex<Received>>;
+
+/// The gate-invocation witness: the gate sets it on entry; an
+/// ordering assertion (`assert !moved`) pins "the gate never saw the
+/// open" (the scope-gate precedence pin, task AC 4).
+type GateWitness = Arc<AtomicBool>;
+
+/// The waiting-session map: `listen` token → present. The gate's
+/// sync in-memory lookup surface (convention 11's "the gate is an
+/// in-memory map lookup"): a token present at gate time = a waiting
+/// session; absent = the miss (`TunnelEstablishError::UnknownResource`
+/// — the canonical stale-token reason ADR-008 Amendment 2 pins).
+type WaitingTokens = Arc<Mutex<HashSet<u32>>>;
+
+/// The gate closure under test: set the witness on entry (even for
+/// opens it will reject — the ordering probe needs the untouched
+/// signal), then the token lookup.
+fn gate_with_tokens(waiting: WaitingTokens, invoked: GateWitness) -> ForwardedGate {
+    Arc::new(move |params: &alktunnels::params::TunnelForwardedParams| {
+        invoked.store(true, Ordering::SeqCst);
+        if waiting
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .contains(&params.listen)
+        {
+            Ok(())
+        } else {
+            Err(TunnelEstablishError::UnknownResource(format!(
+                "no waiting session for listen token {}",
+                params.listen
+            )))
+        }
+    })
+}
+
+/// Serve the forwarded op on the opener side through the establisher
+/// variant: the same injected handler shape as
+/// `tests/producer_forwarded.rs`'s `serve_forwarded_on` (record the
+/// parsed params, pop the correlated halves for the token, `pump_bidi`
+/// against them; a handler-time miss drops the stream and returns —
+/// teardown-at-birth) with the gate establisher wrapped around it.
+fn serve_forwarded_establisher_on(
+    consumer_registry: &Arc<alkcall::registry::registration::OperationRegistry>,
+    manager: ChannelManager,
+    received: ReceivedWitness,
+    local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>>,
+    gate: ForwardedGate,
+) {
+    let handler: alktunnels::producer::ForwardedHandler = Arc::new(move |forwarded, bidi| {
+        let received = Arc::clone(&received);
+        let local_halves = Arc::clone(&local_halves);
+        tokio::spawn(async move {
+            *received.lock().await = Received {
+                params: Some(forwarded),
+            };
+            let target = local_halves.lock().await.pop();
+            if let Some(target) = target {
+                alkcall::channels::pump::pump_bidi(bidi, target.read, target.write).await;
+            }
+        })
+    });
+    let core = alkcall::channels::operations::ChannelCore::new(
+        manager,
+        alkcall::channels::policy::default_policy(),
+    );
+    register_tunnel_forwarded_openable_with_establisher(
+        &core,
+        consumer_registry,
+        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
+        gate,
+        handler,
+    )
+    .expect("register forwarded openable with establisher");
+}
+
+/// A topology for the establisher-variant probes (the
+/// `wire_forwarded_serving` shape): the producer leg is the connect
+/// side (the base op fails-dial — not what's under test); the opener
+/// leg is the accept side whose serving registry carries the
+/// establisher-registered forwarded op. The test plays the producer,
+/// calling the forwarded op on the producer's call plane; the
+/// opener-allocated `channel_id` in the reply is what the producer
+/// adopts.
+async fn wire_forwarded_establisher(identity: Identity) -> harness::Topology {
+    wire_with(
+        alktunnels::producer::ResourceRegistry::new(),
+        RegistrationMode::Dial(harness::failing_dial(
+            "base op unused in forwarded-gate probes",
+        )),
+        Some(identity),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await
+}
+
+fn forwarded_input(listen: u32, peer: (u32, u16)) -> serde_json::Value {
+    serde_json::json!({
+        "listen": listen,
+        "substrate": "tcp",
+        "peer": {"host": peer.0.to_string(), "port": peer.1},
+    })
+}
+
+/// Call the forwarded op on the producer's per-accept call plane.
+async fn call_forwarded(
+    producer: &alkcall::channels::client::ChannelClient,
+    input: serde_json::Value,
+) -> Result<serde_json::Value, CallError> {
+    producer
+        .call_open_op(alktunnels::params::OP_TUNNEL_FORWARDED, input)
+        .await
+        .result
+}
+
+/// The accepted-side echo pair (the producer-side pump target).
+fn driving_halves() -> TargetHandle {
+    let (drive_end, target_end) = tokio::io::duplex(64 * 1024);
+    let (mut t_read, mut t_write) = tokio::io::split(target_end);
+    tokio::spawn(async move {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        let mut buf = vec![0u8; 8192];
+        loop {
+            match t_read.read(&mut buf).await {
+                Ok(0) | Err(_) => return,
+                Ok(n) => {
+                    if t_write.write_all(&buf[..n]).await.is_err() {
+                        return;
+                    }
+                }
+            }
+        }
+    });
+    let (d_read, d_write) = tokio::io::split(drive_end);
+    TargetHandle {
+        read: Box::new(d_read),
+        write: Box::new(d_write),
+    }
+}
+
+/// The "local BIND client" pair (the opener-side pump target).
+fn bind_client_pair() -> (
+    tokio::io::ReadHalf<tokio::io::DuplexStream>,
+    tokio::io::WriteHalf<tokio::io::DuplexStream>,
+    TargetHandle,
+) {
+    let (a, b) = tokio::io::duplex(64 * 1024);
+    let (a_read, a_write) = tokio::io::split(a);
+    let (b_read, b_write) = tokio::io::split(b);
+    (
+        b_read,
+        b_write,
+        TargetHandle {
+            read: Box::new(a_read),
+            write: Box::new(a_write),
+        },
+    )
+}
+
+/// The gate MISS: a `listen` token with no waiting session → the open
+/// fails as a typed `channel:open_failed` with
+/// `details.reason == "unknown_resource"`; the transiently allocated
+/// channel is torn down by the wrapper (no phantom channel — ADR-049
+/// §5); the pump handler NEVER runs. The gate DID run (its witness is
+/// set — the miss is AT the gate, not before it).
+#[tokio::test]
+async fn gate_miss_fails_typed_no_phantom_no_handler() {
+    let topo = wire_forwarded_establisher(forwarded_only_identity()).await;
+    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
+    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
+        Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let waiting: WaitingTokens = Arc::new(Mutex::new(HashSet::new()));
+    let gate_invoked: GateWitness = Arc::new(AtomicBool::new(false));
+    serve_forwarded_establisher_on(
+        &topo.consumer_registry,
+        topo.consumer_manager.clone(),
+        Arc::clone(&received),
+        local_halves,
+        gate_with_tokens(waiting, Arc::clone(&gate_invoked)),
+    );
+
+    // The open with a token the gate rejects: typed failure, not a
+    // success + post-open decline.
+    let err = call_forwarded(&topo.producer, forwarded_input(42113, (198, 55231)))
+        .await
+        .expect_err("a gate miss must fail the open");
+    assert_eq!(
+        err.code, "channel:open_failed",
+        "the establisher's typed failure shape"
+    );
+    let details = err.details.expect("typed error details");
+    assert_eq!(
+        details["reason"], "unknown_resource",
+        "a stale/missing token maps through the existing table"
+    );
+    assert!(
+        details["message"]
+            .as_str()
+            .is_some_and(|m| m.contains("42113")),
+        "the gate's message names the missed token: {}",
+        details["message"]
+    );
+
+    // No phantom channel: the wrapper's transient allocation is torn
+    // down with the establisher's failure (ADR-049 §5).
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+    assert!(
+        topo.consumer_manager
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0),
+        "a gate-miss open leaves no channel on the opener side"
+    );
+
+    // The gate ran (its witness is set); the pump handler NEVER did.
+    assert!(
+        gate_invoked.load(Ordering::SeqCst),
+        "the miss happened at the gate"
+    );
+    assert!(
+        received.lock().await.params.is_none(),
+        "a gate-miss open must not reach the pump handler"
+    );
+}
+
+/// The gate PASS: the registration serves exactly like the base shape
+/// (`tests/producer_forwarded.rs`'s e2e) — the injected handler
+/// receives the parsed params + the `BiStream`, and the byte
+/// round-trip works end to end through the establisher-registered
+/// variant.
+#[tokio::test]
+async fn gate_pass_serves_the_unchanged_serving_flow() {
+    let topo = wire_forwarded_establisher(forwarded_only_identity()).await;
+    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
+    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
+        Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let waiting: WaitingTokens = Arc::new(Mutex::new(HashSet::new()));
+    serve_forwarded_establisher_on(
+        &topo.consumer_registry,
+        topo.consumer_manager.clone(),
+        Arc::clone(&received),
+        Arc::clone(&local_halves),
+        gate_with_tokens(Arc::clone(&waiting), Arc::new(AtomicBool::new(false))),
+    );
+
+    // The waiting session's bookkeeping: token in the gate map + the
+    // correlated halves pre-registered (the correlate-first posture).
+    const TOKEN: u32 = 918273;
+    waiting
+        .lock()
+        .unwrap_or_else(|e| e.into_inner())
+        .insert(TOKEN);
+    let (mut b_read, mut b_write, local) = bind_client_pair();
+    local_halves.lock().await.push(local);
+
+    // The gate-pass open resolves with the wrapper's channel_id.
+    let out = call_forwarded(&topo.producer, forwarded_input(TOKEN, (198, 55231)))
+        .await
+        .expect("the gate-pass open");
+    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
+    assert!(channel_id > 0);
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(
+        topo.consumer_manager.has_channel(channel_id),
+        "the forwarded channel is live on the opener side"
+    );
+
+    // The injected handler saw the parsed params verbatim.
+    let got = received.lock().await.params.clone().expect("invoked");
+    assert_eq!(got.listen, TOKEN, "the token rides `listen`");
+    assert_eq!(got.substrate, alktunnels::params::Substrate::Tcp);
+    assert_eq!(
+        got.peer,
+        alktunnels::params::SubstrateAddr::HostPort {
+            host: "198".to_string(),
+            port: 55231,
+        }
+    );
+
+    // The producer adopts the opener-allocated ID and drives the
+    // accepted side's echo.
+    let (send, recv) = topo
+        .producer
+        .manager()
+        .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
+        .await
+        .expect("producer adopts the opener-allocated ID");
+    let driving = driving_halves();
+    let _producer_pump = tokio::spawn(alkcall::channels::pump::pump_bidi(
+        alkcall::core::types::BiStream::from_joined(recv, send),
+        driving.read,
+        driving.write,
+    ));
+
+    // THE round-trip: the test writes on the BIND client's end → the
+    // assembly's pump → the forwarded channel → the producer's pump →
+    // the driving echo → back.
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    b_write.write_all(b"gate-pass-ping").await.expect("write");
+    b_write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 14];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        b_read.read_exact(&mut buf),
+    )
+    .await
+    .expect("round trip")
+    .expect("read");
+    assert_eq!(&buf, b"gate-pass-ping");
+}
+
+/// The residual race (the base shape's pinned semantics, re-pinned
+/// under the establisher variant): the gate passes (waiting token
+/// present) but the route is GONE at handler time (no correlated
+/// halves) → the injected handler's error path drops the stream and
+/// returns → teardown-at-birth reaps the forwarded channel. The
+/// establisher-pass→handler-run window is the residual
+/// `correlation-miss` class — the handler-side fallback is what reaps
+/// it, under both shapes.
+#[tokio::test]
+async fn residual_race_still_reaps_cleanly() {
+    let topo = wire_forwarded_establisher(forwarded_only_identity()).await;
+    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
+    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
+        Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let waiting: WaitingTokens = Arc::new(Mutex::new(HashSet::new()));
+    serve_forwarded_establisher_on(
+        &topo.consumer_registry,
+        topo.consumer_manager.clone(),
+        Arc::clone(&received),
+        local_halves,
+        gate_with_tokens(Arc::clone(&waiting), Arc::new(AtomicBool::new(false))),
+    );
+
+    // The gate passes (the waiting token is present at establisher
+    // time), but nothing was correlated for the handler — the race.
+    const TOKEN: u32 = 42114;
+    waiting
+        .lock()
+        .unwrap_or_else(|e| e.into_inner())
+        .insert(TOKEN);
+
+    // The open itself STILL RESOLVES (the gate passed; the call-plane
+    // success is what makes this the residual — post-open — race).
+    let out = call_forwarded(&topo.producer, forwarded_input(TOKEN, (198, 55232)))
+        .await
+        .expect("the gate-pass open");
+    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
+
+    // The handler ran (the invocation was witnessed — the rejection is
+    // the handler's, the stream dropped) and its miss reaps the
+    // channel (teardown-at-birth).
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+    let witnessed = received.lock().await.params.clone();
+    assert_eq!(witnessed.expect("invoked").listen, TOKEN);
+    assert!(
+        !topo.consumer_manager.has_channel(channel_id),
+        "a residual-race handler-miss reaps the forwarded channel"
+    );
+}
+
+/// The ordering: an identity WITHOUT `tunnel:forwarded` → FORBIDDEN
+/// and the gate closure is NEVER invoked (a witness the gate sets on
+/// entry stays unmoved) — the registry runs the ACL before the open
+/// op's establishment phase, so the gate never sees an unauthorized
+/// open.
+#[tokio::test]
+async fn scope_gate_precedes_the_establisher() {
+    let topo = wire_forwarded_establisher(open_only_identity()).await;
+    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
+    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
+        Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let waiting: WaitingTokens = Arc::new(Mutex::new(HashSet::new()));
+    waiting.lock().unwrap_or_else(|e| e.into_inner()).insert(77);
+    let gate_invoked: GateWitness = Arc::new(AtomicBool::new(false));
+    serve_forwarded_establisher_on(
+        &topo.consumer_registry,
+        topo.consumer_manager.clone(),
+        Arc::clone(&received),
+        local_halves,
+        gate_with_tokens(waiting, Arc::clone(&gate_invoked)),
+    );
+
+    // The scope-less open is FORBIDDEN (even though the token WOULD
+    // have passed if the gate had run first — the tie-break is the
+    // order, not the gate's answer).
+    match call_forwarded(&topo.producer, forwarded_input(77, (10, 1))).await {
+        Ok(_) => panic!("a scope-less opener must not reach forwarded establishes"),
+        Err(e) => {
+            assert_eq!(
+                e.code, "FORBIDDEN",
+                "missing `tunnel:forwarded` → FORBIDDEN"
+            );
+            assert!(
+                e.message.contains("tunnel:forwarded"),
+                "the denial names the missing scope: {}",
+                e.message
+            );
+        }
+    }
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    // The establishment phase never ran: no gate invocation, no
+    // channel allocation, no handler execution.
+    assert!(
+        !gate_invoked.load(Ordering::SeqCst),
+        "the ACL precedes the establisher — the gate never saw the open"
+    );
+    assert!(
+        topo.consumer_manager
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0),
+        "a FORBIDDEN open leaves no channel on the opener side"
+    );
+    assert!(received.lock().await.params.is_none());
+}
+
+/// The establisher-variant malformed-params delta (the base shape's
+/// `forwarded_malformed_params_tear_down_loud` covers the base
+/// posture; this pins the delta): the schema gates (`listen: 0`,
+/// missing `peer`) fire first exactly as before (`INVALID_INPUT`),
+/// but an unknown top-level field — passes the schema (additive
+/// posture), fails the establisher's serde parse — becomes a TYPED
+/// `channel:open_failed` with `details.reason == "handler_error"`.
+/// The parse precedes the gate (the gate's witness is never set), the
+/// wrapper tears the transient allocation down, and the pump handler
+/// never runs.
+#[tokio::test]
+async fn malformed_params_fail_typed_at_the_establisher() {
+    let topo = wire_forwarded_establisher(forwarded_only_identity()).await;
+    let received: ReceivedWitness = Arc::new(tokio::sync::Mutex::new(Received::default()));
+    let local_halves: Arc<tokio::sync::Mutex<Vec<TargetHandle>>> =
+        Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let gate_invoked: GateWitness = Arc::new(AtomicBool::new(false));
+    serve_forwarded_establisher_on(
+        &topo.consumer_registry,
+        topo.consumer_manager.clone(),
+        Arc::clone(&received),
+        local_halves,
+        gate_with_tokens(
+            Arc::new(Mutex::new(HashSet::new())),
+            Arc::clone(&gate_invoked),
+        ),
+    );
+
+    // The schema gates are unchanged by the establisher (they precede
+    // it — the wrapper order's schema step).
+    match call_forwarded(
+        &topo.producer,
+        serde_json::json!({
+            "listen": 0,
+            "substrate": "tcp",
+            "peer": {"host": "10", "port": 53},
+        }),
+    )
+    .await
+    {
+        Ok(_) => panic!("`listen: 0` must fail the open"),
+        Err(e) => assert_eq!(e.code, "INVALID_INPUT", "schema gate unchanged"),
+    }
+    match call_forwarded(
+        &topo.producer,
+        serde_json::json!({"listen": 7, "substrate": "tcp"}),
+    )
+    .await
+    {
+        Ok(_) => panic!("missing `peer` must fail the open"),
+        Err(e) => assert_eq!(e.code, "INVALID_INPUT", "schema gate unchanged"),
+    }
+
+    // THE DELTA: an unknown top-level field fails the establisher's
+    // serde parse → a typed open failure (vs the base shape's
+    // success-with-channel_id + birth-teardown — the typed posture is
+    // the improvement ADR-008 Amendment 2 pins).
+    let err = call_forwarded(
+        &topo.producer,
+        serde_json::json!({
+            "listen": 918273,
+            "substrate": "tcp",
+            "peer": {"host": "198", "port": 55233},
+            "sneaky": true
+        }),
+    )
+    .await
+    .expect_err("the malformed parse must fail the open as a typed error");
+    assert_eq!(err.code, "channel:open_failed");
+    let details = err.details.expect("typed error details");
+    assert_eq!(
+        details["reason"], "handler_error",
+        "an establisher parse failure maps through the existing table"
+    );
+
+    // The establisher's parse precedes the gate; the wrapper tears
+    // the transient allocation down; the pump handler never runs.
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+    assert!(
+        !gate_invoked.load(Ordering::SeqCst),
+        "the establisher parses BEFORE the gate — the gate never ran"
+    );
+    assert!(
+        topo.consumer_manager
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0),
+        "a parse-failed open leaves no channel on the opener side"
+    );
+    assert!(received.lock().await.params.is_none());
+}