feat: producer-listen — listen establisher (shape 2) + AcceptQueue contract

- AcceptFn: the injected accepted-connection source (the assembly
  layer owns the listener + accept loop; the protocol never binds —
  OQ-TN-04).
- AcceptQueue: the protocol-side queue contract (async push/pop/
  close; FIFO always-before-take ordering; close-while-waiting
  resolves None). Empty-queue posture is assembly-owned (a late
  accept is legitimate; resource_shortage is the closure's mapping;
  the wrapper's 10s establishment timeout is the backstop).
- listen_establisher: same open op, same params, same typed errors —
  registry namespace gate → accept() → Establishment::new(plan);
  the pump handler is untouched (plan-flow with a different source).
- register_tunnel_listen_openable: same spec/pump registration with
  the listen establisher (the honest shape vs a dial/accept enum:
  the establisher is the only difference; one establisher per op id
  per session registry — documented).
- Tests (tests/producer_listen.rs, 6): listen flow end-to-end, FIFO
  ordering across two opens (R-01 plan-flow, listen-flavored), empty
  queue -> resource_shortage, closed listener -> dial_failed,
  unknown resource -> unknown_resource, late-push wait-then-resolve.
- Harness: RegistrationMode enum + wire_listen.
- TargetHandle gains a structural Debug impl (test ergonomics).

Verified: cargo test green (44), clippy -D warnings (native + wasm32),
fmt clean, wasm32 check passes.

tests/harness.rs

@@ -92,9 +92,16 @@ pub struct Topology {
 /// `serving_identity` is the explicit `ServingConfig.identity`
 /// override (CF-005 remediation (a)) — `None` in the primary path.
 /// `provider` resolves payload tokens (the hub-forwarding path).
+/// The open-op registration mode: a dial establisher (shape 1) or a
+/// listen establisher over an assembly-fed [`AcceptQueue`] (shape 2).
+pub enum RegistrationMode {
+    Dial(alktunnels::producer::DialFn),
+    Listen(alktunnels::producer::AcceptFn),
+}
+
 pub async fn wire_with(
     registry: ResourceRegistry,
-    dial: alktunnels::producer::DialFn,
+    mode: RegistrationMode,
     transport_identity: Option<Identity>,
     serving_identity: Option<Identity>,
     provider: Arc<dyn IdentityProvider>,
@@ -185,15 +192,29 @@ pub async fn wire_with(
         producer_client.manager().clone(),
         alkcall::channels::policy::default_policy(),
     );
-    register_tunnel_openable(
-        &core,
-        &registry,
-        &producer_op_registry,
-        AuthContext::anonymous(b"alk/tunnel"),
-        dial,
-        Some(Arc::clone(&identity_witness)),
-    )
-    .expect("register tunnel openable");
+    match mode {
+        RegistrationMode::Dial(dial) => {
+            register_tunnel_openable(
+                &core,
+                &registry,
+                &producer_op_registry,
+                AuthContext::anonymous(b"alk/tunnel"),
+                dial,
+                Some(Arc::clone(&identity_witness)),
+            )
+            .expect("register tunnel openable");
+        }
+        RegistrationMode::Listen(accept) => {
+            alktunnels::producer::register_tunnel_listen_openable(
+                &core,
+                &registry,
+                &producer_op_registry,
+                AuthContext::anonymous(b"alk/tunnel"),
+                accept,
+            )
+            .expect("register tunnel listen openable");
+        }
+    }
     let producer_client = Arc::new(producer_client);
 
     // --- consumer pieces (captured from the hook) -------------------------
@@ -215,7 +236,24 @@ pub async fn wire_with(
 pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn) -> Topology {
     wire_with(
         registry,
-        dial,
+        RegistrationMode::Dial(dial),
+        Some(consumer_identity()),
+        None,
+        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`].
+pub async fn wire_listen(
+    registry: ResourceRegistry,
+    accept: alktunnels::producer::AcceptFn,
+) -> Topology {
+    wire_with(
+        registry,
+        RegistrationMode::Listen(accept),
         Some(consumer_identity()),
         None,
         Arc::new(alkcall::core::auth::NoopIdentityProvider),

tests/producer_listen.rs

@@ -0,0 +1,322 @@
+//! Integration tests for the listen establisher (shape 2 — producer.md
+//! §The Establisher): a producer whose resource is a LISTENER. The
+//! assembly layer owns the listener + its accept loop and feeds an
+//! [`AcceptQueue`]; the establisher pops one accepted handle during
+//! the open op's establishment phase. Same open op, same params, same
+//! typed errors — the pump handler is unchanged.
+//!
+//! Covered: the listen flow end-to-end (in-process listener queue fed
+//! by a test accept loop; consumer opens toward it; the accepted
+//! handle is the plan payload; the two-pump round-trip completes),
+//! pop-during-establishment ordering (always-before-take), empty
+//! queue → `resource_shortage`, closed listener → `dial_failed`,
+//! unknown resource → `unknown_resource`, and the out-of-band close.
+
+mod harness;
+
+use std::sync::Arc;
+
+use alktunnels::params::{Substrate, TunnelParams};
+use alktunnels::producer::{AcceptQueue, ResourceRegistry, TargetHandle, TunnelEstablishError};
+use alktunnels::{open_reverse_channel, TunnelSession};
+
+use harness::{wire_listen, Topology};
+
+fn params(resource: &str) -> TunnelParams {
+    TunnelParams {
+        resource: resource.to_string(),
+        substrate: Substrate::Tcp,
+    }
+}
+
+/// An accept closure popping from the queue (the assembly-side
+/// contract the real `local` listener helper will present).
+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()))
+        })
+    })
+}
+
+/// An in-process accepted handle: a duplex pair, the far end driven by
+/// an echo task (the same shape the dial harness echoes with).
+fn accepted_handle() -> 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),
+    }
+}
+
+fn only_channel_0(topo: &Topology) -> bool {
+    topo.consumer_manager
+        .channel_ids()
+        .iter()
+        .all(|&id| id == 0)
+        && topo
+            .producer
+            .manager()
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0)
+}
+
+#[tokio::test]
+async fn listen_flow_round_trips() {
+    // The listen topology: the assembly accept loop pre-accepts one
+    // connection; the consumer opens toward the resource; the accepted
+    // handle is the plan payload; the pump round-trips.
+    let queue = AcceptQueue::new();
+    queue.push(accepted_handle()).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 channel_id = open_reverse_channel(&topo.consumer_call, &params("listener"), None)
+        .await
+        .expect("listen open");
+    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("stream halves");
+    write.write_all(b"listen-ping").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 11];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip")
+        .expect("read");
+    assert_eq!(&buf, b"listen-ping");
+
+    session.close().await;
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn pop_during_establishment_is_always_before_take() {
+    // The always-before-take guarantee: two accepts pushed BEFORE an
+    // open are popped in FIFO order across two opens — each open's
+    // establisher sees the handle accepted before it (the plan flow:
+    // every open's result flows to ITS pump).
+    let queue = AcceptQueue::new();
+    queue.push(accepted_handle()).await.expect("push 1");
+    queue.push(accepted_handle()).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"), None)
+        .await
+        .expect("open 1");
+    let id2 = open_reverse_channel(&topo.consumer_call, &params("listener"), None)
+        .await
+        .expect("open 2");
+    assert_ne!(id1, id2);
+
+    // Each session round-trips its own marker (the R-01 plan-flow
+    // property, listen-flavored: distinct handles per open).
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    for (channel_id, marker) in [(id1, "one"), (id2, "two")] {
+        let mut session = TunnelSession::adopt(
+            &topo.consumer_manager,
+            channel_id,
+            Substrate::Tcp,
+            alktunnels::TUNNEL_ALPN,
+        )
+        .await
+        .expect("adopt");
+        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));
+}
+
+#[tokio::test]
+async fn empty_queue_is_resource_shortage() {
+    // producer.md's error mapping: an empty queue during establishment
+    // → `resource_shortage`. The QUEUE's pop waits (a late accept is
+    // legitimate — see accept_wait_resolves_when_push_arrives_late);
+    // the assembly layer decides its own posture. An assembly that
+    // knows nothing is coming (a drained listener, a bounded accept
+    // budget) closes its queue or maps empty-pop to the typed error —
+    // here, the closure maps it directly (fail-fast posture).
+    let queue = AcceptQueue::new();
+    queue.close().await;
+    let registry = ResourceRegistry::new();
+    registry
+        .register("listener", Substrate::Tcp, "assembly-owned")
+        .await;
+    let accept: alktunnels::producer::AcceptFn = Arc::new(move || {
+        let queue = queue.clone();
+        Box::pin(async move {
+            // Bounded wait (the assembly's own deadline, not the
+            // wrapper's): a closed queue is an empty listen backlog —
+            // resource_shortage per producer.md's table.
+            match tokio::time::timeout(std::time::Duration::from_millis(50), queue.pop()).await {
+                Ok(Some(handle)) => Ok(handle),
+                _ => Err(TunnelEstablishError::ResourceShortage(
+                    "no accepted connection available".to_string(),
+                )),
+            }
+        })
+    });
+    let topo = wire_listen(registry, accept).await;
+
+    match open_reverse_channel(&topo.consumer_call, &params("listener"), None).await {
+        Ok(_) => panic!("empty-queue open must fail"),
+        Err(alktunnels::ReverseOpenError::Call(call_err)) => {
+            assert_eq!(call_err.code, "channel:open_failed");
+            let reason = call_err
+                .details
+                .as_ref()
+                .and_then(|d| d.get("reason"))
+                .and_then(|r| r.as_str())
+                .expect("reason in details");
+            assert_eq!(reason, "resource_shortage");
+        }
+        Err(other) => panic!("expected Call error, got {other:?}"),
+    }
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo), "failed open leaves no channel");
+}
+#[tokio::test]
+async fn closed_listener_is_dial_failed() {
+    // producer.md's error mapping: a closed listener (the queue closed
+    // before the open) → `dial_failed`-class.
+    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;
+
+    match open_reverse_channel(&topo.consumer_call, &params("listener"), None).await {
+        Ok(_) => panic!("closed-listener open must fail"),
+        Err(alktunnels::ReverseOpenError::Call(call_err)) => {
+            assert_eq!(call_err.code, "channel:open_failed");
+            let reason = call_err
+                .details
+                .as_ref()
+                .and_then(|d| d.get("reason"))
+                .and_then(|r| r.as_str())
+                .expect("reason in details");
+            assert_eq!(reason, "dial_failed");
+        }
+        Err(other) => panic!("expected Call error, got {other:?}"),
+    }
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn unknown_resource_stays_unknown() {
+    // The registry namespace gate: a resource the listen producer does
+    // not produce is `unknown_resource` (same as the dial shape).
+    let queue = AcceptQueue::new();
+    queue.push(accepted_handle()).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;
+
+    match open_reverse_channel(&topo.consumer_call, &params("ghost"), None).await {
+        Ok(_) => panic!("unknown resource must fail"),
+        Err(alktunnels::ReverseOpenError::Call(call_err)) => {
+            assert_eq!(call_err.code, "channel:open_failed");
+            let reason = call_err
+                .details
+                .as_ref()
+                .and_then(|d| d.get("reason"))
+                .and_then(|r| r.as_str())
+                .expect("reason in details");
+            assert_eq!(reason, "unknown_resource");
+        }
+        Err(other) => panic!("expected Call error, got {other:?}"),
+    }
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn accept_wait_resolves_when_push_arrives_late() {
+    // Pop-during-establishment ordering under a LIVE accept loop: the
+    // establisher's pop waits; the accept loop pushes after a tick;
+    // the open resolves once the push lands (the wrapper's bounded
+    // wait — ADR-049 §2's deadline applies above us).
+    let queue = AcceptQueue::new();
+    let accept_loop = tokio::spawn({
+        let queue = queue.clone();
+        async move {
+            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+            queue.push(accepted_handle()).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 channel_id = open_reverse_channel(&topo.consumer_call, &params("listener"), None)
+        .await
+        .expect("open resolves after the late push");
+    accept_loop.await.expect("accept loop task");
+
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    let reaped = session.close().await;
+    assert!(reaped);
+}

tests/producer_open_op.rs

@@ -188,7 +188,7 @@ async fn open_denied_without_any_identity() {
         .await;
     let topo = wire_with(
         registry,
-        echo_dial(),
+        harness::RegistrationMode::Dial(echo_dial()),
         None,
         None,
         Arc::new(alkcall::core::auth::NoopIdentityProvider),
@@ -239,7 +239,7 @@ async fn open_token_overrides_transport_identity() {
         .await;
     let topo = wire_with(
         registry,
-        echo_dial(),
+        harness::RegistrationMode::Dial(echo_dial()),
         None,
         None,
         Arc::new(harness::TestIdProvider),