diff --git a/tasks/tunnels/end-to-end-suite.md b/tasks/tunnels/end-to-end-suite.md
index 71e813b..84cb90a 100644
--- a/tasks/tunnels/end-to-end-suite.md
+++ b/tasks/tunnels/end-to-end-suite.md
@@ -1,7 +1,7 @@
 ---
 id: tunnels/end-to-end-suite
 name: End-to-end suite — forward, reverse, listen, datagram, teardown matrix
-status: pending
+status: completed
 depends_on: [tunnels/consumer-session, tunnels/producer-listen, tunnels/local-socket-halves]
 scope: broad
 risk: medium
@@ -75,6 +75,80 @@ topologies to unify.
 
 > Agent fills during implementation.
 
+### Implementation notes (2026-09-08)
+
+The pre-existing per-task suites (36 integration tests across
+`producer_open_op`, `consumer_session`, `producer_listen`,
+`local_halves`) already carried much of the per-suite coverage; this
+task consolidated the remaining POC-validated behaviors they didn't
+cover into `tests/end_to_end.rs` (18 tests, the suites as enumerated)
+and extended `tests/harness.rs`:
+
+- `RegistrationMode::Timeout(Duration)` — a hanging establisher with
+  an explicit per-registration establishment timeout (the ADR-049 §2
+  deadline-expiry probe; the wrapper maps expiry to reason `timeout`
+  with no channel surviving).
+- `wire_serving_identity` — the `ServingConfig.identity` override
+  probe shape (CF-005 (a)).
+
+Gaps the suite filled (behaviors the per-task suites did not cover):
+1 MiB backpressure round-trip (both POCs' sizing probe); the
+deadline→`timeout` wire mapping; odd-ID allocation asserts (ADR-047 §5)
+plus distinctness; the FULL identity precedence chain — the
+`ServingConfig.identity` override and the token-over-transport
+precedence with a DISTINCT resolved id (the existing token test
+resolved to the same id as the transport identity, so it proved
+authorization but not precedence — the new `DistinctTokenProvider`
+probe proves precedence); reverse datagram sessions via the codec
+(pump-less `(0, 0, reaped)` join shape); 40-datagram bursts with
+batching (the pending-queue cover, both directions); the 7-byte
+chunk-split codec probe (100 datagrams); TCP+UDP interleaved on one
+connection; teardown matrix arms: close-with-pump (abort + reap both
+sides), join copy counts, Drop-with-pump (abort AND reap), and the
+failed-adopt no-leak arm; R-02's JoinHandle-tracks-the-data-plane
+assertion (alive-and-pumpable long after the open, completion only at
+both-direction EOF).
+
+### Test-side findings (not product bugs — recorded for the record)
+
+- **Parked ≠ adopted.** `teardown_channel` on a reverse-open'd but
+  never-adopted channel id is `UnknownChannel` — the manager parks
+  early arrivals for unseen ids (the adoption-race cover, ADR-047 §5)
+  but does not install a channel entry until adopt. Cleanup for such
+  ids is adopt-then-close (what the tests do). Worth knowing when an
+  assembly layer errors between `open_reverse_channel` and `adopt`.
+- **`read(&mut Vec)` vs `read(buf)`** — an unpopulated `Vec` passed to
+  `AsyncReadExt::read` has len 0 and reads nothing (instant EOF); the
+  1 MiB loop needed a size-`expected` buffer with `&mut received[got..]`.
+- The empty-queue-vs-error posture from `tunnels/producer-listen`
+  carries over verbatim: the queue's pop waits; mapping empty-pop to
+  `resource_shortage` is the assembly closure's posture (producer.md's
+  table), asserted here with a fail-fast closure.
+
 ## Summary
 
-> Agent fills this on completion.
\ No newline at end of file
+> Agent fills this on completion.
+
+### Summary
+
+`tests/end_to_end.rs` (18 tests) consolidates the POC-validated
+behaviors into the crate's end-to-end suite over the shared duplex
+harness: forward (1 MiB backpressure, deadline→`timeout`, CF-006
+witness on transport + token paths), reverse (odd-ID asserts +
+distinctness, the full CF-005 precedence chain incl. the
+ServingConfig override and a distinct-id token probe, R-01 same-
+resource concurrency ×3), listen (FIFO always-before-take, the typed
+error table), datagram (batching survival, 7-byte chunk-split codec
+probe, reverse codec session, TCP+UDP concurrent), teardown matrix
+(close-with-pump / join copy counts / Drop-with-pump / failed-adopt),
+and the spec-conformance assertions (no phantom channel, no leaked
+session, R-02's data-plane-tracking pump handle).
+
+Harness grew `RegistrationMode::Timeout` + `wire_serving_identity`.
+Existing suites untouched. Verification: `cargo test` green (68 tests
+incl. unit + doc), `cargo test --features local` green (75), 3×
+repeat-run clean (both feature configs), clippy `-D warnings` clean
+(all-targets, `--features local`, wasm32), `cargo fmt --check` clean,
+wasm32 check passes. No product-code changes were needed — the
+failures during development were all test-harness bugs (zero-len read
+buffer, teardown-on-parked-id), recorded above.
\ No newline at end of file
diff --git a/tests/end_to_end.rs b/tests/end_to_end.rs
new file mode 100644
index 0000000..53114e2
--- /dev/null
+++ b/tests/end_to_end.rs
@@ -0,0 +1,1140 @@
+//! The end-to-end suite — every validated POC behavior, re-expressed
+//! against the crate's public API (tests/common/harness.rs, the two
+//! POC harnesses' topologies unified). The spec's executable form: the
+//! quality gate before the implementation review.
+//!
+//! Suites:
+//! 1. **Forward (`-L`)** — open → establisher dials → pump → round-
+//!    trip; 1 MiB backpressure; typed errors; per-call opener identity
+//!    (CF-006); establishment deadline → `timeout` (ADR-049 §2).
+//! 2. **Reverse (`-R`)** — `open_reverse_channel` + `adopt` +
+//!    `pump_against`; odd-ID allocation (ADR-047 §5); the identity
+//!    precedence chain (CF-005); half-close (W4); out-of-band close +
+//!    self-reaping (W3); same-resource concurrency (R-01).
+//! 3. **Listen producer** — the accept-queue flow + typed errors.
+//! 4. **Datagram (`udp`)** — forward + reverse datagram sessions via
+//!    the codec; round-trip, empty datagram (F-2), chunk-split
+//!    survival, TCP+UDP concurrent on one connection.
+//! 5. **Teardown matrix** (ADR-005) — close/join/Drop; `channel_ids()`
+//!    asserts no leaks on either side.
+//! 6. Spec-conformance assertions — no phantom channel on a failed
+//!    open, no leaked session on a failed adopt, R-02's
+//!    JoinHandle-tracks-the-data-plane shape.
+//!
+//! (`local`-feature real-socket suites live in tests/local_halves.rs
+//! — the duplex suites here need no sockets.)
+
+mod harness;
+
+use std::sync::Arc;
+
+use alkcall::channels::client::ChannelOpenError;
+use alktunnels::params::{establishment_reason, Substrate, TunnelParams, TUNNEL_OPEN_SCOPE};
+use alktunnels::producer::{AcceptQueue, ResourceRegistry, TargetHandle, TunnelEstablishError};
+use alktunnels::{open_reverse_channel, TunnelSession};
+
+use harness::{
+    echo_dial, failing_dial, framed_udp_echo_dial, wire, wire_forward, wire_listen,
+    wire_serving_identity, wire_with, ForwardTopology, RegistrationMode, Topology, TEST_AUTH_TOKEN,
+};
+
+fn params(resource: &str, substrate: Substrate) -> TunnelParams {
+    TunnelParams {
+        resource: resource.to_string(),
+        substrate,
+    }
+}
+
+/// Only channel 0 (the pre-negotiated call channel) may remain —
+/// asserted on both sides of both topologies after every teardown.
+trait LeakCheck {
+    fn no_data_channels(&self) -> bool;
+}
+
+impl LeakCheck for Topology {
+    fn no_data_channels(&self) -> bool {
+        self.consumer_manager
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0)
+            && self
+                .producer
+                .manager()
+                .channel_ids()
+                .iter()
+                .all(|&id| id == 0)
+    }
+}
+
+impl LeakCheck for ForwardTopology {
+    fn no_data_channels(&self) -> bool {
+        self.consumer
+            .manager()
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0)
+            && self
+                .producer_manager
+                .channel_ids()
+                .iter()
+                .all(|&id| id == 0)
+    }
+}
+
+fn only_channel_0<T: LeakCheck>(topo: &T) -> bool {
+    topo.no_data_channels()
+}
+
+fn reason_of_open_failure(err: &TunnelSessionOpenError) -> Option<String> {
+    establishment_reason(err.open_ref()).map(str::to_string)
+}
+
+type TunnelSessionOpenError = alktunnels::TunnelOpenError;
+
+fn call_error_reason(err: &alktunnels::ReverseOpenError) -> Option<String> {
+    match err {
+        alktunnels::ReverseOpenError::Call(call_err) => call_err
+            .details
+            .as_ref()
+            .and_then(|d| d.get("reason"))
+            .and_then(|r| r.as_str())
+            .map(str::to_string),
+        _ => None,
+    }
+}
+
+// =====================================================================
+// Suite 1 — Forward (`-L`): open → establisher → pump → round-trip
+// =====================================================================
+
+/// A 1 MiB patterned payload round-trips through the two-pump shape
+/// (the bounded-buffer backpressure path; the POCs' sizing probe).
+/// The consumer writes via taken halves and reads the echo back on
+/// the same halves.
+#[tokio::test]
+async fn forward_large_payload_backpressure_round_trip() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_forward(registry, echo_dial()).await;
+
+    let session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("open");
+    let Taken {
+        mut read,
+        mut write,
+        session,
+    } = split_taken(session.take_halves());
+
+    let payload: Vec<u8> = (0..1024 * 1024u32).map(|i| (i % 251) as u8).collect();
+    let expected = payload.clone();
+    let writer = tokio::spawn(async move {
+        use tokio::io::AsyncWriteExt;
+        write.write_all(&payload).await.expect("write all");
+        write.flush().await.expect("flush");
+        write
+    });
+
+    use tokio::io::AsyncReadExt;
+    let mut received = vec![0u8; expected.len()];
+    let mut got = 0;
+    while got < received.len() {
+        let n = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            read.read(&mut received[got..]),
+        )
+        .await
+        .expect("read timed out (backpressure deadlock?)")
+        .expect("read");
+        assert!(n > 0, "EOF before the full echo");
+        got += n;
+    }
+    writer.await.expect("writer task");
+    assert_eq!(got, expected.len(), "1 MiB round trip intact");
+    assert_eq!(&received[..], &expected[..]);
+    drop(read);
+
+    let (_, _, reaped) = session.join().await;
+    assert!(reaped);
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// A hanging establisher is bounded by the per-registration timeout
+/// (ADR-049 §2's override): the open fails `channel:open_failed` with
+/// reason `timeout`, and no channel survives on either side (the
+/// no-phantom property, deadline-flavored).
+#[tokio::test]
+async fn forward_establishment_deadline_maps_to_timeout_typed() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let producer_op_registry = Arc::new(alkcall::registry::registration::OperationRegistry::new());
+
+    // The timeout topology: the default wire() shape with the hanging
+    // establisher replacing the dial establisher — wired by hand here
+    // via wire_with's Timeout mode on the reverse topology, then the
+    // forward direction asserted through the session API on a fresh
+    // forward wiring below.
+    let topo = wire_with(
+        registry.clone(),
+        RegistrationMode::Dial(echo_dial()),
+        Some(harness::consumer_identity()),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await;
+
+    // Reverse-side open against a hanging establisher with a 300ms
+    // per-registration bound (the deadline-expiry probe).
+    let queue = AcceptQueue::new();
+    queue.close().await;
+    let timeout_topo = harness::wire_with(
+        ResourceRegistry::new(),
+        RegistrationMode::Timeout(std::time::Duration::from_millis(300)),
+        Some(harness::consumer_identity()),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await;
+    let started = std::time::Instant::now();
+    match open_reverse_channel(
+        &timeout_topo.consumer_call,
+        &params("echo", Substrate::Tcp),
+        None,
+    )
+    .await
+    {
+        Ok(_) => panic!("hanging establisher must time out"),
+        Err(e) => {
+            let err = match e {
+                alktunnels::ReverseOpenError::Call(c) => c,
+                other => panic!("expected a call error, got {other:?}"),
+            };
+            assert_eq!(err.code, "channel:open_failed");
+            let reason = err
+                .details
+                .as_ref()
+                .and_then(|d| d.get("reason"))
+                .and_then(|r| r.as_str())
+                .expect("reason in details");
+            assert_eq!(reason, "timeout");
+        }
+    }
+    assert!(
+        started.elapsed() < std::time::Duration::from_secs(5),
+        "the bound fired, not the harness backstop: {:?}",
+        started.elapsed()
+    );
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&timeout_topo), "no phantom channel");
+
+    // The generic channel ops on the same registry still work (the
+    // timeout is per-registration, ADR-049 §2).
+    let response = topo
+        .producer
+        .call_open_op("consumer/serves/nothing", serde_json::json!({}))
+        .await;
+    assert!(
+        response.result.is_err(),
+        "the point is the reply RESOLVED (per-op isolation)"
+    );
+    let _ = queue;
+    let _ = producer_op_registry;
+}
+
+/// The identity-witness end-to-end (CF-006): the establisher's
+/// per-call auth carried the END CALLER's identity (the consumer), on
+/// both the transport-identity path and the token path.
+#[tokio::test]
+async fn forward_opener_identity_witnessed_on_both_paths() {
+    // Transport identity path (the mTLS posture — no token).
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_forward(registry, echo_dial()).await;
+    let session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("transport-identity open");
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(
+        witness.as_deref(),
+        Some("consumer"),
+        "establisher saw the per-call opener identity (CF-006)"
+    );
+    session.close().await;
+
+    // Token path (the hub-forwarding posture): the token resolves to
+    // the consumer identity and the establisher sees it end-to-end.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+    let channel_id = open_reverse_channel(
+        &topo.consumer_call,
+        &params("echo", Substrate::Tcp),
+        Some(TEST_AUTH_TOKEN),
+    )
+    .await
+    .expect("token open authorized");
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(witness.as_deref(), Some("consumer"));
+    assert_eq!(channel_id % 2, 1, "the connect side allocated an odd id");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    session.close().await;
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+}
+
+// =====================================================================
+// Suite 2 — Reverse (`-R`): open_reverse_channel + adopt + pump_against
+// =====================================================================
+
+/// Odd-ID allocation under ADR-047 §5: the serving side (producer,
+/// `ChannelSide::Connect` here) allocates odd ids; the initiator
+/// adopts. Asserted across several opens on one connection.
+#[tokio::test]
+async fn reverse_channel_ids_are_odd_and_distinct() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let mut ids = Vec::new();
+    for _ in 0..3 {
+        let id = open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open");
+        assert_eq!(id % 2, 1, "ADR-047 §5: the serving side allocates odd ids");
+        ids.push(id);
+    }
+    assert_eq!(
+        ids.len(),
+        ids.iter().collect::<std::collections::HashSet<_>>().len()
+    );
+    for id in ids {
+        // The reverse-open'd channel was never adopted: the manager
+        // parked its early arrivals. `adopt_channel` drains the park
+        // and installs the entry — adopt, then close (the session's
+        // teardown; teardown_channel alone is UnknownChannel for a
+        // parked-but-never-adopted id).
+        let session = TunnelSession::adopt(
+            &topo.consumer_manager,
+            id,
+            Substrate::Tcp,
+            alktunnels::TUNNEL_ALPN,
+        )
+        .await
+        .expect("adopt parked channel");
+        assert!(session.close().await, "close reaps the entry");
+    }
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// The identity precedence chain (CF-005), asserted through the
+/// establisher's per-call identity: token > ServingConfig.identity >
+/// transport; identity-less everywhere → fail-closed FORBIDDEN.
+#[tokio::test]
+async fn reverse_identity_precedence_chain() {
+    // (a) ServingConfig.identity overrides the transport identity.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let override_identity = alkcall::core::auth::Identity {
+        id: "worker-effective".to_string(),
+        scopes: vec![TUNNEL_OPEN_SCOPE.to_string()],
+        resources: Default::default(),
+    };
+    let topo = wire_serving_identity(registry, echo_dial(), Some(override_identity)).await;
+    let id = open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+        .await
+        .expect("override identity authorizes");
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(
+        witness.as_deref(),
+        Some("worker-effective"),
+        "ServingConfig.identity won over the transport identity"
+    );
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt parked channel");
+    session.close().await;
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+    // (b) The token wins over the transport identity: the provider
+    // resolves the scoped token to a DISTINCT id — the witness proves
+    // which one the establisher saw.
+    struct DistinctTokenProvider;
+    impl alkcall::core::auth::IdentityProvider for DistinctTokenProvider {
+        fn resolve_from_fingerprint(&self, _: &str) -> Option<alkcall::core::auth::Identity> {
+            None
+        }
+        fn resolve_from_token(
+            &self,
+            token: &alkcall::core::auth::AuthToken,
+        ) -> Option<alkcall::core::auth::Identity> {
+            if token.raw == TEST_AUTH_TOKEN.as_bytes() {
+                Some(alkcall::core::auth::Identity {
+                    id: "token-caller".to_string(),
+                    scopes: vec![TUNNEL_OPEN_SCOPE.to_string()],
+                    resources: Default::default(),
+                })
+            } else {
+                None
+            }
+        }
+    }
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = harness::wire_with(
+        registry,
+        RegistrationMode::Dial(echo_dial()),
+        Some(harness::consumer_identity()),
+        None,
+        Arc::new(DistinctTokenProvider),
+    )
+    .await;
+    let id = open_reverse_channel(
+        &topo.consumer_call,
+        &params("echo", Substrate::Tcp),
+        Some(TEST_AUTH_TOKEN),
+    )
+    .await
+    .expect("token open authorized");
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(
+        witness.as_deref(),
+        Some("token-caller"),
+        "the payload token won over the transport identity (CF-005 precedence)"
+    );
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt parked channel");
+    session.close().await;
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+
+    // (c) Identity-less everywhere → fail-closed.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_with(
+        registry,
+        RegistrationMode::Dial(echo_dial()),
+        None,
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await;
+    let err = open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+        .await
+        .expect_err("identity-less open must be denied");
+    match err {
+        alktunnels::ReverseOpenError::Call(call_err) => {
+            assert_eq!(call_err.code, "FORBIDDEN");
+        }
+        other => panic!("expected a call error, got {other:?}"),
+    }
+    assert!(only_channel_0(&topo));
+}
+
+/// Concurrent opens of the SAME resource (R-01): each open's plan
+/// carries its own dialed handle; each round-trips its own marker; no
+/// handoff race.
+#[tokio::test]
+async fn reverse_concurrent_same_resource_no_handoff_race() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let mut sessions = Vec::new();
+    for i in 0..3 {
+        let channel_id =
+            open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+                .await
+                .expect("reverse open");
+        assert_eq!(channel_id % 2, 1);
+        let session = TunnelSession::adopt(
+            &topo.consumer_manager,
+            channel_id,
+            Substrate::Tcp,
+            alktunnels::TUNNEL_ALPN,
+        )
+        .await
+        .expect("adopt");
+        let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
+        let session = session.pump_against(accepted_end).await;
+        let (mut l_read, mut l_write) = tokio::io::split(local_end);
+        let msg = format!("racer-{i}-{}", channel_id);
+        l_write.write_all(msg.as_bytes()).await.expect("write");
+        l_write.flush().await.expect("flush");
+        let mut buf = vec![0u8; msg.len()];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            l_read.read_exact(&mut buf),
+        )
+        .await
+        .expect("round trip")
+        .expect("read");
+        assert_eq!(buf, msg.as_bytes());
+        sessions.push(session);
+    }
+    let ids: std::collections::HashSet<u32> = sessions.iter().map(|s| s.channel_id).collect();
+    assert_eq!(ids.len(), sessions.len(), "distinct odd ids");
+    assert!(ids.iter().all(|&id| id % 2 == 1), "all odd (ADR-047 §5)");
+
+    for session in sessions {
+        let (_, _, reaped) = session.join().await;
+        assert!(reaped);
+    }
+    assert!(only_channel_0(&topo));
+}
+
+// =====================================================================
+// Suite 3 — Listen producer (shape 2, the accept-queue flow)
+// =====================================================================
+
+/// The listen-queue flow: FIFO always-before-take across two opens;
+/// the pump handler is the same one the dial shape registers.
+#[tokio::test]
+async fn listen_fifo_ordering_two_opens_two_handles() {
+    let queue = AcceptQueue::new();
+    queue.push(echo_handle("first")).await.expect("push 1");
+    queue.push(echo_handle("second")).await.expect("push 2");
+
+    let registry = ResourceRegistry::new();
+    registry
+        .register("listener", Substrate::Tcp, "assembly-owned")
+        .await;
+    let topo = wire_listen(registry, accept_from_queue(queue.clone())).await;
+
+    let id1 = open_reverse_channel(
+        &topo.consumer_call,
+        &params("listener", Substrate::Tcp),
+        None,
+    )
+    .await
+    .expect("open 1");
+    let id2 = open_reverse_channel(
+        &topo.consumer_call,
+        &params("listener", Substrate::Tcp),
+        None,
+    )
+    .await
+    .expect("open 2");
+    assert_ne!(id1, id2);
+    assert_eq!(id1 % 2, 1);
+    assert_eq!(id2 % 2, 1);
+
+    for (channel_id, marker) in [(id1, "first"), (id2, "second")] {
+        let mut session = TunnelSession::adopt(
+            &topo.consumer_manager,
+            channel_id,
+            Substrate::Tcp,
+            alktunnels::TUNNEL_ALPN,
+        )
+        .await
+        .expect("adopt");
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        let (read, write) = session.stream_halves().expect("halves");
+        write.write_all(marker.as_bytes()).await.expect("write");
+        write.flush().await.expect("flush");
+        let mut buf = vec![0u8; marker.len()];
+        tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+            .await
+            .expect("round trip")
+            .expect("read");
+        assert_eq!(buf, marker.as_bytes());
+        session.close().await;
+    }
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// The listen-shape typed-error table (producer.md's mapping):
+/// unknown resource, empty-queue → resource_shortage, closed listener
+/// → dial_failed — each leaves no channel (the no-phantom property).
+#[tokio::test]
+async fn listen_typed_error_table_leaves_no_channel() {
+    // Empty queue (the assembly's fail-fast posture: a bounded accept
+    // budget maps empty-pop to the typed error).
+    let empty: alktunnels::producer::AcceptFn = Arc::new(|| {
+        Box::pin(async { Err(TunnelEstablishError::ResourceShortage("drained".into())) })
+    });
+    let registry = ResourceRegistry::new();
+    registry
+        .register("listener", Substrate::Tcp, "assembly-owned")
+        .await;
+    let topo = wire_listen(registry, empty).await;
+    let err = open_reverse_channel(
+        &topo.consumer_call,
+        &params("listener", Substrate::Tcp),
+        None,
+    )
+    .await
+    .expect_err("empty-queue open must fail");
+    assert_eq!(
+        call_error_reason(&err).as_deref(),
+        Some("resource_shortage")
+    );
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+
+    // Closed listener → dial_failed.
+    let queue = AcceptQueue::new();
+    queue.close().await;
+    let registry = ResourceRegistry::new();
+    registry
+        .register("listener", Substrate::Tcp, "assembly-owned")
+        .await;
+    let topo = wire_listen(registry, accept_from_queue(queue.clone())).await;
+    let err = open_reverse_channel(
+        &topo.consumer_call,
+        &params("listener", Substrate::Tcp),
+        None,
+    )
+    .await
+    .expect_err("closed-listener open must fail");
+    assert_eq!(call_error_reason(&err).as_deref(), Some("dial_failed"));
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+
+    // Unknown resource → unknown_resource (the namespace gate).
+    let queue = AcceptQueue::new();
+    queue.push(echo_handle("unused")).await.expect("push");
+    let registry = ResourceRegistry::new();
+    registry
+        .register("listener", Substrate::Tcp, "assembly-owned")
+        .await;
+    let topo = wire_listen(registry, accept_from_queue(queue.clone())).await;
+    let err = open_reverse_channel(&topo.consumer_call, &params("ghost", Substrate::Tcp), None)
+        .await
+        .expect_err("unknown resource must fail");
+    assert_eq!(call_error_reason(&err).as_deref(), Some("unknown_resource"));
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+}
+
+// =====================================================================
+// Suite 4 — Datagram (`udp`): forward + reverse via the codec
+// =====================================================================
+
+/// Many datagrams in each direction, boundaries preserved, batched
+/// frames survive (one chunk may carry several datagrams — the
+/// pending-queue cover).
+#[tokio::test]
+async fn datagram_boundaries_preserved_both_directions() {
+    let registry = ResourceRegistry::new();
+    registry.register("dns", Substrate::Udp, "in-process").await;
+    let topo = wire_forward(registry, framed_udp_echo_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
+        .await
+        .expect("udp open");
+
+    // Burst 40 datagrams (varied sizes) — they may batch into shared
+    // chunks; each must come back as exactly one datagram.
+    let mut payloads = Vec::new();
+    for i in 0..40u16 {
+        payloads.push(vec![i as u8; (i as usize * 137 % 500) + 1]);
+    }
+    for p in &payloads {
+        session.send_datagram(p).await.expect("send");
+    }
+    let mut received = Vec::new();
+    while received.len() < payloads.len() {
+        let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("io ok")
+            .expect("not eof");
+        received.push(dg.to_vec());
+    }
+    assert_eq!(received, payloads, "boundaries preserved under batching");
+
+    // The reverse direction: the target echo only repeats — drive the
+    // other way by sending again after draining; order holds.
+    session.send_datagram(b"final").await.expect("send");
+    let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("io ok")
+        .expect("not eof");
+    assert_eq!(&dg[..], b"final");
+
+    let reaped = session.close().await;
+    assert!(reaped);
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// The codec survives maximal chunk splitting (the forward POC's
+/// chunk-split test, driven through a live session): datagrams
+/// chopped into awkward chunks still decode exactly.
+#[tokio::test]
+async fn datagrams_survive_chunk_splitting_codec_level() {
+    // Codec-level probe (the mux may split anywhere): 100 datagrams
+    // framed, chopped into 7-byte chunks, decoded incrementally.
+    let mut stream = Vec::new();
+    for i in 0..100u16 {
+        let payload = vec![i as u8; (i as usize % 300) + 1];
+        stream.extend_from_slice(&alktunnels::wire::frame_datagram(&payload).expect("frame"));
+    }
+    let mut reader = alktunnels::wire::DatagramReader::new();
+    let mut recovered = Vec::new();
+    for chunk in stream.chunks(7) {
+        for dg in reader.feed(chunk).expect("decode") {
+            recovered.push(dg.to_vec());
+        }
+    }
+    assert_eq!(recovered.len(), 100);
+    for (i, dg) in recovered.iter().enumerate() {
+        assert_eq!(dg.len(), (i % 300) + 1, "size preserved");
+        assert!(dg.iter().all(|&b| b == i as u8), "content preserved");
+    }
+}
+
+/// A reverse datagram session: the consumer pumps a UDP "accepted"
+/// socket against the codec-framed channel; datagrams round-trip via
+/// the pump (the reverse POC's UDP shape, codec-flavored).
+#[tokio::test]
+async fn reverse_udp_datagram_session_round_trips() {
+    let registry = ResourceRegistry::new();
+    registry.register("dns", Substrate::Udp, "in-process").await;
+    let topo = wire(registry, framed_udp_echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("dns", Substrate::Udp), None)
+            .await
+            .expect("reverse udp open");
+    assert_eq!(channel_id % 2, 1);
+    let mut session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Udp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    // No pump_against: the session's codec data plane is driven
+    // directly (the pump-less datagram shape — pump_bidi copies raw
+    // bytes, so the pump is the consumer's job here only when it
+    // holds local halves; the direct drive is the session API's).
+    session.send_datagram(b"query").await.expect("send");
+    let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("io ok")
+        .expect("datagram");
+    assert_eq!(&dg[..], b"query");
+    let (_, _, reaped) = session.join().await;
+    assert!(reaped, "pump-less join reaps (0, 0, reaped)");
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// TCP + UDP sessions concurrently on ONE connection — the mux is
+/// transparent across substrates (both POCs' concurrent shape).
+#[tokio::test]
+async fn tcp_and_udp_concurrent_on_one_connection() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    registry.register("dns", Substrate::Udp, "in-process").await;
+    let topo = wire_forward(registry, harness::all_substrate_dial()).await;
+
+    let mut tcp_session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("tcp open");
+    let mut udp_session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
+        .await
+        .expect("udp open");
+    assert_ne!(tcp_session.channel_id, udp_session.channel_id);
+
+    // Interleaved traffic on both planes.
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (tcp_read, tcp_write) = tcp_session.stream_halves().expect("tcp halves");
+    tcp_write.write_all(b"tcp-mixed").await.expect("tcp write");
+    tcp_write.flush().await.expect("tcp flush");
+    udp_session
+        .send_datagram(b"udp-mixed")
+        .await
+        .expect("udp send");
+    let mut tbuf = vec![0u8; 9];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        tcp_read.read_exact(&mut tbuf),
+    )
+    .await
+    .expect("tcp round trip")
+    .expect("tcp read");
+    assert_eq!(&tbuf, b"tcp-mixed");
+    let dg = tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        udp_session.recv_datagram(),
+    )
+    .await
+    .expect("udp recv timed out")
+    .expect("udp io ok")
+    .expect("udp datagram");
+    assert_eq!(&dg[..], b"udp-mixed");
+
+    let reaped = tcp_session.close().await;
+    assert!(reaped);
+    let reaped = udp_session.close().await;
+    assert!(reaped);
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(only_channel_0(&topo));
+}
+
+// =====================================================================
+// Suite 5 — Teardown matrix (ADR-005): close / join / Drop
+// =====================================================================
+
+/// close on a session WITH a spawned pump: aborts the pump, reaps the
+/// entry on the consumer side; the producer side reaps via the
+/// wrapper. Both managers end on channel 0 only.
+#[tokio::test]
+async fn teardown_close_with_pump_reaps_both_sides() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    let session = session.pump_against(tokio::io::duplex(64 * 1024).0).await;
+    assert!(topo.consumer_manager.has_channel(channel_id));
+    let reaped = session.close().await;
+    assert!(reaped, "close reaps the adopted entry");
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// join on a session WITH a pump whose local ends drop: both pumps
+/// complete, copy counts are reported, both sides reaped (the
+/// two-pump contract observed from the session).
+#[tokio::test]
+async fn teardown_join_reports_copy_counts_and_reaps() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
+    let session = session.pump_against(accepted_end).await;
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (mut l_read, mut l_write) = tokio::io::split(local_end);
+    l_write.write_all(b"joinme").await.expect("write");
+    l_write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 6];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        l_read.read_exact(&mut buf),
+    )
+    .await
+    .expect("round trip")
+    .expect("read");
+    assert_eq!(&buf, b"joinme");
+
+    drop(l_write);
+    drop(l_read);
+    let (c2p, p2c, reaped) =
+        tokio::time::timeout(std::time::Duration::from_secs(5), session.join())
+            .await
+            .expect("join timed out");
+    assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}");
+    assert!(reaped);
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// Drop after pump_against: the pump aborts AND the entry reaps —
+/// the full teardown, not just the pump (the drop matrix's
+/// pump-carrying arm).
+#[tokio::test]
+async fn teardown_drop_with_pump_reaps_entry() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    let session = session.pump_against(tokio::io::duplex(64 * 1024).0).await;
+    drop(session);
+    assert!(
+        !topo.consumer_manager.has_channel(channel_id),
+        "Drop reaped the entry alongside the aborted pump"
+    );
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+/// A failed adopt leaks nothing: the manager's parked entry is
+/// released (the failed-adopt arm of the matrix) — a bogus-ID adopt
+/// SUCCEEDS by design (the manager parks any unseen id), so the
+/// collision path is the honest probe; after the probe the queue is
+/// clean on both sides.
+#[tokio::test]
+async fn teardown_failed_adopt_leaves_no_leak() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open");
+    let first = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("first adopt");
+    let second = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await;
+    match second
+        .err()
+        .expect("adopting a live id twice must fail")
+        .open_ref()
+    {
+        ChannelOpenError::AdoptFailed(_) => {}
+        other => panic!("expected AdoptFailed, got {other:?}"),
+    }
+    first.close().await;
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+// =====================================================================
+// Suite 6 — Spec-conformance assertions (kept visible)
+// =====================================================================
+
+/// A failed open NEVER returns a channel id (the no-phantom-channel
+/// property, dial shape): unknown_resource + dial_failed both leave
+/// both sides' managers at channel 0 only, and the consumer's typed
+/// error exposes the establishment reason.
+#[tokio::test]
+async fn failed_open_never_yields_channel_id() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_forward(registry, echo_dial()).await;
+
+    let err = match TunnelSession::open(&topo.consumer, params("ghost", Substrate::Tcp)).await {
+        Ok(_) => panic!("unknown resource must not open"),
+        Err(e) => e,
+    };
+    assert_eq!(
+        reason_of_open_failure(&err).as_deref(),
+        Some("unknown_resource")
+    );
+    assert!(only_channel_0(&topo));
+
+    let registry = ResourceRegistry::new();
+    registry
+        .register("dead", Substrate::Tcp, "unreachable")
+        .await;
+    let topo = wire_forward(registry, failing_dial("no route")).await;
+    let err = match TunnelSession::open(&topo.consumer, params("dead", Substrate::Tcp)).await {
+        Ok(_) => panic!("failing dial must not open"),
+        Err(e) => e,
+    };
+    assert_eq!(reason_of_open_failure(&err).as_deref(), Some("dial_failed"));
+    assert!(only_channel_0(&topo));
+}
+
+/// The pump handler's JoinHandle tracks the data plane (R-02): the
+/// session-side pump completes when the local ends close (not
+/// teardown-at-birth), and copy counts prove bytes moved AFTER the
+/// open resolved.
+#[tokio::test]
+async fn pump_handle_tracks_the_data_plane_r02() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
+    let session = session.pump_against(accepted_end).await;
+
+    // The channel is alive and pumpable long after the open resolved
+    // (the early-return shape would have torn it down at birth).
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (mut l_read, mut l_write) = tokio::io::split(local_end);
+    for round in 0..3 {
+        let msg = format!("alive-{round}");
+        l_write.write_all(msg.as_bytes()).await.expect("write");
+        l_write.flush().await.expect("flush");
+        let mut buf = vec![0u8; msg.len()];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            l_read.read_exact(&mut buf),
+        )
+        .await
+        .expect("round trip")
+        .expect("read");
+        assert_eq!(buf, msg.as_bytes());
+    }
+
+    // Pump completion arrives only when BOTH directions end.
+    drop(l_write);
+    drop(l_read);
+    let (c2p, p2c, reaped) =
+        tokio::time::timeout(std::time::Duration::from_secs(5), session.join())
+            .await
+            .expect("join timed out (teardown-at-birth would show here)");
+    assert!(c2p > 0 && p2c > 0 && reaped);
+    assert!(only_channel_0(&topo));
+}
+
+// =====================================================================
+// Harness helpers (the accepted-handle stand-ins reused by suite 3)
+// =====================================================================
+
+fn split_taken(taken: alktunnels::TakenHalves) -> Taken {
+    Taken {
+        read: taken.read,
+        write: taken.write,
+        session: taken.session,
+    }
+}
+
+struct Taken {
+    read: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
+    write: Box<dyn tokio::io::AsyncWrite + Send + Unpin>,
+    session: TunnelSession,
+}
+
+/// An in-process accepted handle: a duplex pair whose far end echoes
+/// (the same shape the dial harness echoes with — a listen producer's
+/// "accepted connection").
+fn echo_handle(_label: &str) -> TargetHandle {
+    let (consumer_side, target_side) = tokio::io::duplex(64 * 1024);
+    let (mut t_read, mut t_write) = tokio::io::split(target_side);
+    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 (c_read, c_write) = tokio::io::split(consumer_side);
+    TargetHandle {
+        read: Box::new(c_read),
+        write: Box::new(c_write),
+    }
+}
+
+/// An accept closure popping from the queue (the assembly-side
+/// contract; a closed queue maps to `dial_failed`).
+fn accept_from_queue(queue: AcceptQueue) -> alktunnels::producer::AcceptFn {
+    Arc::new(move || {
+        let queue = queue.clone();
+        Box::pin(async move {
+            queue
+                .pop()
+                .await
+                .ok_or_else(|| TunnelEstablishError::DialFailed("listener closed".to_string()))
+        })
+    })
+}
diff --git a/tests/harness.rs b/tests/harness.rs
index 30aaf6f..852c90c 100644
--- a/tests/harness.rs
+++ b/tests/harness.rs
@@ -97,6 +97,11 @@ pub struct Topology {
 pub enum RegistrationMode {
     Dial(alktunnels::producer::DialFn),
     Listen(alktunnels::producer::AcceptFn),
+    /// A hanging establisher registered with an explicit
+    /// per-registration establishment timeout (the deadline-expiry
+    /// probe — ADR-049 §2's bound override; the open fails with reason
+    /// `timeout` and no channel survives).
+    Timeout(std::time::Duration),
 }
 
 pub async fn wire_with(
@@ -214,6 +219,35 @@ pub async fn wire_with(
             )
             .expect("register tunnel listen openable");
         }
+        RegistrationMode::Timeout(timeout) => {
+            // The deadline-expiry probe's registration shape: a hanging
+            // establisher (never resolves) with an explicit per-
+            // registration timeout. The generic channel ops registered
+            // above on the SAME registry are unaffected (per-op
+            // timeout, ADR-049 §2).
+            let core = alkcall::channels::operations::ChannelCore::new(
+                producer_client.manager().clone(),
+                alkcall::channels::policy::default_policy(),
+            );
+            core.register_openable_with_establisher(
+                alktunnels::params::tunnel_open_spec(),
+                Some(Arc::new(|_input: serde_json::Value, _auth| {
+                    Box::pin(async {
+                        tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
+                        unreachable!("hanging establisher must be timed out, not resolve")
+                            as Result<
+                                alkcall::channels::operations::Establishment,
+                                alkcall::channels::operations::EstablishmentError,
+                            >
+                    })
+                })),
+                alktunnels::producer::make_tunnel_pump_handler(),
+                &producer_op_registry,
+                AuthContext::anonymous(b"alk/tunnel"),
+                Some(timeout),
+            )
+            .expect("register hanging establisher");
+        }
     }
     let producer_client = Arc::new(producer_client);
 
@@ -244,6 +278,26 @@ pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn
     .await
 }
 
+/// A topology with the consumer's transport identity REPLACED by the
+/// serving-side override (`ServingConfig.identity` — CF-005 (a) probe
+/// shape) or stripped entirely (the fail-closed probe). The transport
+/// identity still rides the dialing connection; the override is what
+/// the serving dispatch resolves (the witness proves which one won).
+pub async fn wire_serving_identity(
+    registry: ResourceRegistry,
+    dial: alktunnels::producer::DialFn,
+    override_identity: Option<Identity>,
+) -> Topology {
+    wire_with(
+        registry,
+        RegistrationMode::Dial(dial),
+        Some(consumer_identity()),
+        override_identity,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await
+}
+
 /// The listen topology (shape 2): the producer's establisher pops
 /// accepted handles from the assembly-fed [`AcceptQueue`] instead of
 /// dialing. Same identity posture as [`wire`].