diff --git a/tasks/tunnels/corrmiss-gate-tests.md b/tasks/tunnels/corrmiss-gate-tests.md index 4243c35..c9fb5ba 100644 --- a/tasks/tunnels/corrmiss-gate-tests.md +++ b/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>` + of waiting tokens (the gate is a sync closure — no awaits) and sets + an `Arc` 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` 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 \ No newline at end of file +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. \ No newline at end of file diff --git a/tests/producer_forwarded_gate.rs b/tests/producer_forwarded_gate.rs new file mode 100644 index 0000000..264ccd5 --- /dev/null +++ b/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, +} + +type ReceivedWitness = Arc>; + +/// 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; + +/// 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>>; + +/// 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(¶ms.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, + manager: ChannelManager, + received: ReceivedWitness, + local_halves: Arc>>, + 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 { + 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::WriteHalf, + 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>> = + 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>> = + 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>> = + 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>> = + 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>> = + 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()); +}