feat: producer-open-op — establisher + pump handler + registration

- tunnel_establisher: registry lookup -> injected DialFn (ADR-004) ->
  Establishment::new(plan) (R-01 plan flow); typed reason table
  (unknown_resource/dial_failed/resource_shortage/handler_error)
- make_tunnel_pump_handler: accept_bi + plan downcast + pump_bidi
  awaited inline (R-02; no spawn-and-forget, no substrate types)
- register_tunnel_openable: spec + establisher + pump + dial + CF-006
  identity witness seam
- TargetHandle +Sync halves (F-1 bound); Substrate + Hash for registry
  keys; tunnel_establisher_with_witness for the per-call identity probe
- tests: duplex harness (reverse-POC topology) + 11 integration tests —
  round-trip, typed errors, FORBIDDEN fail-closed, CF-005 precedence +
  CF-006 witness, plan-flow concurrency (no race), W2 late
  registration, INVALID_INPUT schema rejection, both-sides sanity

Verified: cargo test (14 unit + 11 integration), clippy -D warnings
(native + wasm32), fmt --check, wasm32 check — all clean

src/params.rs

@@ -44,7 +44,7 @@ pub struct TunnelParams {
 /// The substrate discriminator (ADR-001). `Unix` ships in v1 (the
 /// `local` feature implements it — OQ-TN-14); a new substrate is a
 /// new value, not a format change.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum Substrate {
     Tcp,

src/producer.rs

@@ -1,10 +1,238 @@
-//! The producer half: the `channels/tunnel/sub` open op — spec
+//! The producer half: the `channels/tunnel/sub` open op — its
-//! builder, establisher shapes (dial + listen), the pump handler, and
+//! establisher (dial shape), its substrate-agnostic pump handler, and
-//! registration (`register_tunnel_openable`) per `producer.md` /
+//! its registration (`register_tunnel_openable`) per `producer.md` /
 //! ADR-004.
 //!
-//! Skeleton module — filled by `tunnels/producer-open-op` (establisher
+//! The establisher is the open op's awaited, bounded establishment
-//! + pump handler + registration) and `tunnels/producer-listen` (the
+//! phase (alkcall ADR-049): parse params semantically, look the
-//!   listen establisher variant).
+//! resource up in the producer's registry, dial via the injected
+//! `DialFn` (ADR-004 — "produce boxed halves for a resource" is a
+//! function, not a trait), and return the handle in
+//! `Establishment::new(plan)` (R-01 — typed-opaque `ChannelPlan`; its
+//! `Send + Sync` bound is why `TargetHandle`'s halves carry `+ Sync`).
+//! The dial closure closes over substrate code — the protocol crate
+//! never sees a socket type.
+//!
+//! The pump handler downcasts the plan and awaits
+//! `alkcall::channels::pump_bidi` **inline** inside its task (R-02:
+//! the returned `JoinHandle` tracks the data-plane lifetime; early
+//! return = teardown-at-birth). It cannot know what the halves came
+//! from and must not (AGENTS.md convention 7).
+//!
+//! The listen establisher variant (the producer-side listener — the
+//! `-R` far-side listener shape) lands with `tunnels/producer-listen`.
 
-pub use crate::params::{establishment_reason, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE};
+use std::sync::Arc;
+
+use alkcall::channels::operations::{
+    ChannelCore, ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler,
+};
+use alkcall::channels::pump::pump_bidi;
+use alkcall::core::auth::AuthContext;
+use alkcall::core::types::Connection;
+use alkcall::registry::registration::OperationRegistry;
+use serde_json::Value;
+use tokio::io::{AsyncRead, AsyncWrite};
+
+use crate::params::{Substrate, TunnelParams};
+
+pub use crate::params::{OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE};
+
+/// The dialed target handle — the plan payload (typed-opaque
+/// `ChannelPlan`; establisher and handler agree on the concrete
+/// type). The boxed halves carry `+ Sync` to satisfy the plan's
+/// `Send + Sync` bound (POC F-1, documented upstream).
+pub struct TargetHandle {
+    pub read: Box<dyn AsyncRead + Send + Sync + Unpin>,
+    pub write: Box<dyn AsyncWrite + Send + Sync + Unpin>,
+}
+
+/// The typed establishment-failure vocabulary (ADR-049 §3's reason
+/// codes as this crate's establisher errors; mapped 1:1 onto
+/// `EstablishmentError`).
+#[derive(Debug, thiserror::Error)]
+pub enum TunnelEstablishError {
+    #[error("unknown resource: {0}")]
+    UnknownResource(String),
+    #[error("dial failed: {0}")]
+    DialFailed(String),
+    #[error("resource shortage: {0}")]
+    ResourceShortage(String),
+    #[error("handler error: {0}")]
+    HandlerError(String),
+}
+
+impl From<TunnelEstablishError> for EstablishmentError {
+    fn from(e: TunnelEstablishError) -> Self {
+        match e {
+            TunnelEstablishError::UnknownResource(m) => {
+                EstablishmentError::UnknownResource { message: m }
+            }
+            TunnelEstablishError::DialFailed(m) => EstablishmentError::DialFailed { message: m },
+            TunnelEstablishError::ResourceShortage(m) => {
+                EstablishmentError::ResourceShortage { message: m }
+            }
+            TunnelEstablishError::HandlerError(m) => {
+                EstablishmentError::HandlerError { message: m }
+            }
+        }
+    }
+}
+
+/// The producer's registry of produced resources:
+/// `(resource, substrate)` → backing. An assembly-owned construct
+/// (OQ-TN-11 — the collision domain is the producer's registry; two
+/// producers on one connection may expose the same resource name
+/// independently). The backing is opaque to the protocol — the
+/// `DialFn` closes over whatever shape the assembly layer chose.
+#[derive(Clone, Default)]
+pub struct ResourceRegistry {
+    targets: Arc<tokio::sync::Mutex<std::collections::HashMap<(String, Substrate), String>>>,
+}
+
+impl ResourceRegistry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Produce a resource: map `(resource, substrate)` to its backing
+    /// (a local port, a socket path, an in-process address — the
+    /// dial closure's business, not the wire's).
+    pub async fn register(&self, resource: &str, substrate: Substrate, backing: &str) {
+        self.targets
+            .lock()
+            .await
+            .insert((resource.to_string(), substrate), backing.to_string());
+    }
+
+    /// Resolve a resource to its backing; `None` = the producer does
+    /// not produce it (`unknown_resource` at the establisher).
+    pub async fn lookup(&self, substrate: Substrate, resource: &str) -> Option<String> {
+        self.targets
+            .lock()
+            .await
+            .get(&(resource.to_string(), substrate))
+            .cloned()
+    }
+}
+
+/// The substrate dial function the assembly layer injects (ADR-004):
+/// given a substrate + a registry backing, produce the target's
+/// boxed halves. Behind the `local` feature for real sockets, or
+/// assembly-constructed (in-process pipes, test doubles).
+pub type DialFn = Arc<
+    dyn Fn(
+            Substrate,
+            &str,
+        )
+            -> futures::future::BoxFuture<'static, Result<TargetHandle, TunnelEstablishError>>
+        + Send
+        + Sync,
+>;
+
+/// The witness capturing the establisher's per-call opener identity
+/// (CF-006 probe — the establisher sees the END CALLER overlaid onto
+/// the install-time context, not a synthetic install-time value).
+pub type IdentityWitness = Arc<tokio::sync::Mutex<Option<String>>>;
+
+/// The establisher (dial shape): registry lookup → dial →
+/// `Establishment::new(plan)`. The per-call `auth` is the dispatch-
+/// resolved identity overlaid onto the install-time context (CF-006);
+/// record it into `identity_witness` when provided.
+pub fn tunnel_establisher(registry: ResourceRegistry, dial: DialFn) -> OpenEstablisher {
+    tunnel_establisher_with_witness(registry, dial, None)
+}
+
+/// [`tunnel_establisher`] with an optional identity witness (the test
+/// seam for the CF-006 contract).
+pub fn tunnel_establisher_with_witness(
+    registry: ResourceRegistry,
+    dial: DialFn,
+    identity_witness: Option<IdentityWitness>,
+) -> OpenEstablisher {
+    Arc::new(move |input: Value, auth| {
+        let registry = registry.clone();
+        let dial = Arc::clone(&dial);
+        let witness = identity_witness.clone();
+        Box::pin(async move {
+            if let Some(w) = &witness {
+                *w.lock().await = auth.identity.as_ref().map(|i| i.id.clone());
+            }
+            let params: TunnelParams = parse_params(&input)?;
+            let backing = registry
+                .lookup(params.substrate, &params.resource)
+                .await
+                .ok_or_else(|| {
+                    TunnelEstablishError::UnknownResource(format!(
+                        "resource `{}` ({:?}) not produced by this producer",
+                        params.resource, params.substrate
+                    ))
+                })?;
+            let dialed: TargetHandle = dial(params.substrate, &backing).await?;
+            Ok(Establishment::new(Arc::new(dialed) as ChannelPlan))
+        })
+    })
+}
+
+fn parse_params(input: &Value) -> Result<TunnelParams, TunnelEstablishError> {
+    serde_json::from_value(input.clone())
+        .map_err(|e| TunnelEstablishError::HandlerError(format!("invalid params: {e}")))
+}
+
+/// The pump handler: accept the channel's yield-once `BiStream`,
+/// downcast the plan to `TargetHandle`, and await `pump_bidi` inline
+/// (R-02 — the returned `JoinHandle` tracks the data-plane lifetime;
+/// an early return is teardown-at-birth, and the birth-teardown
+/// telemetry diagnoses the spawn-and-forget shape at runtime). The
+/// handler is substrate-agnostic by construction — it sees boxed
+/// halves, never a socket type.
+pub fn make_tunnel_pump_handler() -> OpenHandler {
+    Arc::new(
+        move |_input: Value, plan: Option<ChannelPlan>, conn: Connection, _auth: AuthContext| {
+            tokio::spawn(async move {
+                let Some(bidi) = conn.accept_bi().await.ok() else {
+                    tracing::error!("tunnel pump: accept_bi failed");
+                    return;
+                };
+                let Some(handle) = plan.and_then(|p| p.downcast::<TargetHandle>().ok()) else {
+                    tracing::error!("tunnel pump: no TargetHandle in plan");
+                    return;
+                };
+                let Ok(handle) = Arc::try_unwrap(handle) else {
+                    tracing::error!("tunnel pump: TargetHandle still shared; cannot unwrap");
+                    return;
+                };
+                let _counts = pump_bidi(bidi, handle.read, handle.write).await;
+            })
+        },
+    )
+}
+
+/// Register the tunnel open op on a `ChannelCore` + the session's
+/// dispatch registry: spec + establisher + pump + injected dial, via
+/// `register_openable_with_establisher` (`timeout: None` = the 10s
+/// default bound). Post-hoc registration is supported (W2): the
+/// dispatcher reads through the shared `Arc<OperationRegistry>` per
+/// dispatch. `identity_witness` is the optional CF-006 probe seam
+/// (the establisher records the per-call opener identity it saw).
+pub fn register_tunnel_openable(
+    core: &ChannelCore,
+    registry: &ResourceRegistry,
+    on_registry: &Arc<OperationRegistry>,
+    auth: AuthContext,
+    dial: DialFn,
+    identity_witness: Option<IdentityWitness>,
+) -> Result<(), String> {
+    core.register_openable_with_establisher(
+        crate::params::tunnel_open_spec(),
+        Some(tunnel_establisher_with_witness(
+            registry.clone(),
+            dial,
+            identity_witness,
+        )),
+        make_tunnel_pump_handler(),
+        on_registry,
+        auth,
+        None,
+    )
+}

tasks/tunnels/producer-open-op.md

@@ -1,7 +1,7 @@
 ---
 id: tunnels/producer-open-op
 name: Producer half — establisher (dial shape) + pump handler + register_tunnel_openable
-status: pending
+status: completed
 depends_on: [tunnels/params, tunnels/wire-codec]
 scope: broad
 risk: high
@@ -97,16 +97,16 @@ late-registration visibility, pump round-trip through `pump_bidi`.
 
 ## Acceptance Criteria
 
-- [ ] Establisher: typed reason mapping exact (ADR-001's table)
+- [x] Establisher: typed reason mapping exact (ADR-001's table)
-- [ ] Pump handler: `pump_bidi` inline; no spawn-and-forget; no
+- [x] Pump handler: `pump_bidi` inline; no spawn-and-forget; no
       substrate types leak into the handler
-- [ ] `register_tunnel_openable` wires spec + establisher + pump +
+- [x] `register_tunnel_openable` wires spec + establisher + pump +
       dial injection
-- [ ] Integration tests: forward AND reverse topology (the harness
+- [x] Integration tests: forward AND reverse topology (the harness
       shapes from both POCs), ≥8 tests covering the above
-- [ ] Clippy/fmt clean; wasm32 check passes (dial closures are
+- [x] Clippy/fmt clean; wasm32 check passes (dial closures are
       runtime-injected — the crate compiles without `local`)
-- [ ] `cargo test` green
+- [x] `cargo test` green
 
 ## References
 
@@ -118,8 +118,49 @@ late-registration visibility, pump round-trip through `pump_bidi`.
 
 ## Notes
 
-> Agent fills during implementation.
+- Ported from the reverse POC's producer.rs (the 0.7.0-idiomatic plan
+  flow, R-01) with the crate generalizations:
+  - The dial closure is INJECTED (`DialFn`) — the POC had real socket
+    code inline (tokio TCP/UDP); the crate's establisher is
+    substrate-agnostic and the `local` feature (or the assembly layer)
+    supplies the closure (ADR-004's inversion point, tested via
+    in-process duplex echo + failing dials).
+  - `register_tunnel_openable` takes `dial` + `identity_witness` (the
+    CF-006 probe seam) — the POC's registration closure had no dial
+    injection.
+  - `ResourceRegistry` keys on `(String, Substrate)` (the enum, not a
+    &'static str — `Hash` added to `Substrate`).
+  - `TargetHandle` carries `+ Sync` halves (the F-1 plan-payload
+    bound).
+- Harness (tests/harness.rs): the reverse POC's topology (producer =
+  `from_connection_with_serving` on the connect side, consumer =
+  ChannelsAdapter + capturing install hook) with the dial closure
+  standing in for sockets. The consumer drives the open op via
+  `call_with_payload` on channel 0 and adopts the producer-allocated
+  ID.
+- 11 integration tests: round-trip through `pump_bidi` (wrapper reaps
+  the producer channel on pump completion — R-02), unknown_resource +
+  dial_failed typed errors (no phantom channel either side), FORBIDDEN
+  identity-less (fails closed), transport-identity open (CF-005 (b)),
+  token precedence (CF-005 (a) + the CF-006 witness), same-resource
+  concurrency (no plan race), late registration (W2), unknown
+  substrate schema rejection (INVALID_INPUT — the registry's
+  schema-validation runs before the establisher), outbound-calls-
+  resolve-while-serving (ADR-022 §2), substrate-keyed registry lookups.
+- alkcall 0.7.0 note: the establisher's per-call `auth` carries the
+  dispatch-resolved identity (CF-005 corollary) — the witness proves
+  the overlay; `INVALID_INPUT` is the wire code for a schema failure
+  (distinct from `channel:open_failed` establishment failures).
+- `futures` crate used for `BoxFuture` in `DialFn` (already a dep).
 
 ## Summary
 
-> Agent fills this on completion.
+Producer half complete: dial-shape establisher (registry lookup →
+injected dial → `Establishment::new(plan)`), the substrate-agnostic
+pump handler (`pump_bidi` awaited inline, R-02), and
+`register_tunnel_openable` (spec + establisher + pump + dial
+injection + witness). 14 unit tests (params + codec) + 11 integration
+tests over the duplex harness pass. Verified: cargo test, clippy
+--all-targets -D warnings (native + wasm32), fmt --check, wasm32
+check — all clean. The listen establisher variant lands with
+`tunnels/producer-listen`.

tests/harness.rs

@@ -0,0 +1,275 @@
+//! Test harness — the producer↔consumer wiring over a
+//! `tokio::io::duplex` carrying the channels 8-byte chunk format.
+//!
+//! Topology (the reverse POC's shape, mirrored for both directions):
+//! - **producer** (connect side of the channels connection — it runs
+//!   `ChannelClient::from_connection_with_serving`): dials the
+//!   duplex, serves the tunnel open op + generic channel ops on its
+//!   serving registry. Its manager allocates the tunnel channel IDs
+//!   (`ChannelSide::Connect` — ADR-047 §5: the serving side
+//!   allocates; the initiator adopts).
+//! - **consumer** (accept side): the `ChannelsAdapter` + capturing
+//!   install hook. The hook captures the consumer's channel-0
+//!   `CallConnection` (for calling the producer's open op) and the
+//!   consumer's `ChannelManager` (for adopting the producer-
+//!   allocated IDs and tearing channels down). The consumer serves
+//!   no ops.
+//!
+//! Identity on the serving path follows alkcall 0.7.0's CF-005
+//! precedence: payload `auth_token` → `ServingConfig.identity` → the
+//! transport connection's identity (`Connection::set_identity` before
+//! dialing). CF-006 (the per-call opener identity on the establisher)
+//! is probed via the `identity_witness` the establisher records.
+
+use std::sync::Arc;
+
+use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
+use alkcall::channels::client::{ChannelClient, ServingConfig};
+use alkcall::channels::manager::ChannelManager;
+use alkcall::channels::operations::ChannelOperations;
+use alkcall::core::auth::{AuthContext, AuthToken, Identity, IdentityProvider};
+use alkcall::core::types::Connection as CoreConnection;
+use alkcall::protocol::connection::{split_single_stream, CallConnection};
+use alkcall::protocol::dispatch::Dispatcher;
+use alkcall::registry::registration::OperationRegistry;
+
+use alktunnels::params::{Substrate, TUNNEL_OPEN_SCOPE};
+use alktunnels::producer::{register_tunnel_openable, ResourceRegistry};
+
+/// The consumer identity the transport carries (the mTLS/QUIC
+/// analogue — key-based identity resolved out-of-band, attached to
+/// the dialing connection before `from_connection_with_serving`).
+pub fn consumer_identity() -> Identity {
+    Identity {
+        id: "consumer".to_string(),
+        scopes: vec![TUNNEL_OPEN_SCOPE.to_string()],
+        resources: std::collections::HashMap::new(),
+    }
+}
+
+/// The scoped token (the hub-forwarding path) — resolves to
+/// [`consumer_identity`] via [`TestIdProvider`].
+pub const TEST_AUTH_TOKEN: &str = "test-token-with-tunnel-open-scope";
+
+/// Resolve the scoped token to the consumer identity; anything else
+/// resolves to none (falling through to the transport identity per
+/// the 0.7.0 precedence order).
+pub struct TestIdProvider;
+impl IdentityProvider for TestIdProvider {
+    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
+        None
+    }
+    fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity> {
+        if token.raw == TEST_AUTH_TOKEN.as_bytes() {
+            Some(consumer_identity())
+        } else {
+            None
+        }
+    }
+}
+
+/// The full producer↔consumer topology over one duplex transport.
+pub struct Topology {
+    /// The producer's client (connect side, serving). Its manager
+    /// allocated the tunnel channels; its pumps are wrapper-managed.
+    pub producer: Arc<ChannelClient>,
+    /// The consumer's manager (accept side). `adopt` installs the
+    /// producer-allocated IDs here; reaping calls `teardown_channel`.
+    pub consumer_manager: ChannelManager,
+    /// The consumer's channel-0 call connection — its surface for
+    /// calling the producer's open op (and channel/close).
+    pub consumer_call: Arc<CallConnection>,
+    /// The producer's serving registry (probes).
+    pub producer_registry: Arc<OperationRegistry>,
+    /// What the establisher's per-call auth carried as the opener
+    /// identity id (the CF-006 probe — `None` until an open runs).
+    pub identity_witness: Arc<tokio::sync::Mutex<Option<String>>>,
+}
+
+/// Wire the topology. `transport_identity` is the key-based identity
+/// attached to the dialing connection (CF-005 remediation (b));
+/// `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).
+pub async fn wire_with(
+    registry: ResourceRegistry,
+    dial: alktunnels::producer::DialFn,
+    transport_identity: Option<Identity>,
+    serving_identity: Option<Identity>,
+    provider: Arc<dyn IdentityProvider>,
+) -> Topology {
+    let producer_op_registry = Arc::new(OperationRegistry::new());
+    let (consumer_call_tx, mut consumer_call_rx) =
+        tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
+    let (consumer_manager_tx, mut consumer_manager_rx) =
+        tokio::sync::mpsc::channel::<ChannelManager>(1);
+
+    // --- consumer (accept side): ChannelsAdapter + capturing hook ------
+    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
+        let call_tx = consumer_call_tx.clone();
+        let manager_tx = consumer_manager_tx.clone();
+        tokio::spawn(async move {
+            let channel0_bidi = match channel0_conn.accept_bi().await {
+                Ok(s) => s,
+                Err(_) => return,
+            };
+            let (writer, reader) = split_single_stream(channel0_bidi);
+            let call_connection = Arc::new(CallConnection::new_single_stream(
+                channel0_conn,
+                Arc::clone(&writer),
+            ));
+            let _ = call_tx.send(Arc::clone(&call_connection)).await;
+            let _ = manager_tx.send(manager).await;
+            // The consumer serves no ops — an empty registry. The
+            // dispatcher keeps the call loop alive (its read loop
+            // resolves the producer's responses to the consumer's
+            // pending entries).
+            let dp = Dispatcher::new(
+                Arc::new(OperationRegistry::new()),
+                Arc::new(alkcall::core::auth::NoopIdentityProvider),
+            );
+            dp.serve_single_stream(call_connection, reader, writer)
+                .await;
+        })
+    });
+
+    // --- transport ------------------------------------------------------
+    let (producer_end, consumer_end) = tokio::io::duplex(64 * 1024);
+    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
+    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
+
+    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
+    let consumer_transport_auth = AuthContext::anonymous(b"alk/channels");
+    let _adapter_task = tokio::spawn(async move {
+        let _ = alkcall::core::types::ProtocolHandler::handle(
+            &adapter,
+            consumer_conn,
+            &consumer_transport_auth,
+        )
+        .await;
+    });
+
+    // The transport identity attaches to the dialing connection
+    // BEFORE from_connection_with_serving (the mTLS/QUIC posture).
+    if let Some(id) = transport_identity {
+        producer_conn
+            .set_identity(id)
+            .expect("transport identity set once");
+    }
+
+    // --- producer (connect side, serving) --------------------------------
+    let producer_client = ChannelClient::from_connection_with_serving(
+        producer_conn,
+        Some(ServingConfig {
+            registry: Arc::clone(&producer_op_registry),
+            identity_provider: provider,
+            identity: serving_identity,
+        }),
+    )
+    .await
+    .expect("producer channel client init");
+
+    // The generic channel ops ride the producer's serving registry
+    // (channel/close tears a channel down out-of-band — the review
+    // 007 recipe).
+    let ops = ChannelOperations::with_default_policy(producer_client.manager().clone());
+    ops.register_on(&producer_op_registry)
+        .expect("register channel ops on producer");
+
+    // Post-hoc openable registration on the shared registry (W2). The
+    // identity witness captures what the establisher's per-call auth
+    // carried (the CF-006 probe).
+    let identity_witness = Arc::new(tokio::sync::Mutex::new(None::<String>));
+    let core = alkcall::channels::operations::ChannelCore::new(
+        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");
+    let producer_client = Arc::new(producer_client);
+
+    // --- consumer pieces (captured from the hook) -------------------------
+    let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn");
+    let consumer_manager = consumer_manager_rx.recv().await.expect("consumer manager");
+
+    Topology {
+        producer: producer_client,
+        consumer_manager,
+        consumer_call,
+        producer_registry: producer_op_registry,
+        identity_witness,
+    }
+}
+
+/// The default topology: the consumer's transport identity is the
+/// primary caller-identity path (no token). The provider is
+/// noop-shaped — only the identity chain authorizes.
+pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn) -> Topology {
+    wire_with(
+        registry,
+        dial,
+        Some(consumer_identity()),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await
+}
+
+/// An echo dial closure (the ADR-004 injection point under test): an
+/// in-process pipe per dial — the "target" echoes bytes until EOF.
+/// The consumer side gets split `DuplexStream` halves; a spawned task
+/// echoes the other end. No `local` feature, no sockets.
+pub fn echo_dial() -> alktunnels::producer::DialFn {
+    Arc::new(move |substrate: Substrate, backing: &str| {
+        let backing = backing.to_string();
+        Box::pin(async move {
+            match substrate {
+                Substrate::Tcp | Substrate::Unix => {
+                    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 _ = &backing;
+                    let (c_read, c_write) = tokio::io::split(consumer_side);
+                    Ok(alktunnels::producer::TargetHandle {
+                        read: Box::new(c_read),
+                        write: Box::new(c_write),
+                    })
+                }
+                Substrate::Udp => Err(alktunnels::producer::TunnelEstablishError::DialFailed(
+                    format!("udp dial not wired in this harness: {backing}"),
+                )),
+            }
+        })
+    })
+}
+
+/// A failing dial closure (the `dial_failed` stand-in).
+pub fn failing_dial(message: &'static str) -> alktunnels::producer::DialFn {
+    Arc::new(move |_substrate: Substrate, _backing: &str| {
+        Box::pin(async move {
+            Err(alktunnels::producer::TunnelEstablishError::DialFailed(
+                message.to_string(),
+            ))
+        })
+    })
+}

tests/producer_open_op.rs

@@ -0,0 +1,372 @@
+//! Integration tests for the producer half — the open-op path
+//! (establisher + pump handler + registration) over the duplex
+//! harness, ported from the reverse POC's suite (16 tests rode the
+//! topology) with the dial closure standing in for real sockets.
+//!
+//! Covered: establishment success + the typed-error table
+//! (`unknown_resource` / `dial_failed` / `FORBIDDEN`), the plan flow
+//! under same-resource concurrency (R-01 — no handoff race), the
+//! per-call opener identity (CF-006 witness), late-registration
+//! visibility (W2), pump round-trip through `pump_bidi` (R-02 — the
+//! wrapper-managed pump, no spawn-and-forget), and out-of-band
+//! `channel/close`.
+
+mod harness;
+
+use std::sync::Arc;
+
+use alkcall::protocol::wire::CallError;
+use alktunnels::params::{Substrate, TUNNEL_OPEN_SCOPE};
+use alktunnels::producer::ResourceRegistry;
+
+use harness::{echo_dial, failing_dial, wire, wire_with, Topology, TEST_AUTH_TOKEN};
+
+fn params(resource: &str) -> alktunnels::params::TunnelParams {
+    alktunnels::params::TunnelParams {
+        resource: resource.to_string(),
+        substrate: Substrate::Tcp,
+    }
+}
+
+/// Call the producer's open op on channel 0 (the consumer's surface),
+/// returning the wire `CallError` on failure.
+async fn call_open(
+    topo: &Topology,
+    input: serde_json::Value,
+    auth_token: Option<&str>,
+) -> Result<serde_json::Value, CallError> {
+    let mut payload = serde_json::json!({
+        "operationId": "channels/tunnel/sub",
+        "input": input,
+    });
+    if let Some(token) = auth_token {
+        payload["auth_token"] = serde_json::Value::String(token.to_string());
+    }
+    let response = topo.consumer_call.call_with_payload(payload).await;
+    response.result
+}
+
+async fn call_open_params(
+    topo: &Topology,
+    p: &alktunnels::params::TunnelParams,
+    auth_token: Option<&str>,
+) -> Result<serde_json::Value, CallError> {
+    call_open(
+        topo,
+        serde_json::json!({
+            "resource": p.resource,
+            "substrate": match p.substrate {
+                Substrate::Tcp => "tcp",
+                Substrate::Udp => "udp",
+                Substrate::Unix => "unix",
+            },
+        }),
+        auth_token,
+    )
+    .await
+}
+
+/// Adopt the producer-allocated channel and pump it against in-process
+/// halves (the consumer-side data plane — the session's take_halves
+/// shape, driven manually until the consumer session task lands).
+async fn adopt_and_pump(
+    topo: &Topology,
+    channel_id: u32,
+) -> (
+    tokio::io::ReadHalf<tokio::io::DuplexStream>,
+    tokio::io::WriteHalf<tokio::io::DuplexStream>,
+    tokio::task::JoinHandle<(u64, u64)>,
+) {
+    let (send, recv) = topo
+        .consumer_manager
+        .adopt_channel(channel_id, "alk/tunnel", None)
+        .await
+        .expect("adopt producer-allocated channel");
+    let bidi = alkcall::core::types::BiStream::from_joined(recv, send);
+    let (channel_end, local_end) = tokio::io::duplex(64 * 1024);
+    let (c_read, c_write) = tokio::io::split(channel_end);
+    let pump = tokio::spawn(alkcall::channels::pump::pump_bidi(bidi, c_read, c_write));
+    let (l_read, l_write) = tokio::io::split(local_end);
+    (l_read, l_write, pump)
+}
+
+/// Only channel 0 (the pre-negotiated call channel) may remain — a
+/// failed open leaves no data channel on either side (the
+/// phantom-channel property).
+fn no_data_channels(topo: &Topology) -> bool {
+    let consumer = topo.consumer_manager.channel_ids();
+    let producer = topo.producer.manager().channel_ids();
+    consumer.iter().all(|&id| id == 0) && producer.iter().all(|&id| id == 0)
+}
+
+#[tokio::test]
+async fn open_round_trips_through_pump_bidi() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let out = call_open_params(&topo, &params("echo"), None)
+        .await
+        .expect("open must succeed");
+    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
+    assert!(topo.producer.manager().has_channel(channel_id));
+
+    let (mut read, mut write, pump) = adopt_and_pump(&topo, channel_id).await;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    write.write_all(b"ping").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 4];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip timed out")
+        .expect("read");
+    assert_eq!(&buf, b"ping");
+
+    // EOF propagates: closing the local write half shuts the loop
+    // down through both pumps (the two-pump contract), and the
+    // wrapper reaps the producer-side channel on completion (R-02).
+    drop(write);
+    drop(read);
+    let (c2p, p2c) = tokio::time::timeout(std::time::Duration::from_secs(5), pump)
+        .await
+        .expect("pump completion timed out")
+        .expect("pump task");
+    assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}");
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(
+        !topo.producer.manager().has_channel(channel_id),
+        "wrapper reaped the producer-side channel after pump completion"
+    );
+}
+
+#[tokio::test]
+async fn unknown_resource_is_typed_and_leaves_no_channel() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let err = call_open_params(&topo, &params("ghost"), None)
+        .await
+        .expect_err("unknown resource must fail the open");
+    assert_eq!(err.code, "channel:open_failed");
+    assert_eq!(establishment_reason_of(&err), Some("unknown_resource"));
+    assert!(no_data_channels(&topo));
+}
+
+fn establishment_reason_of(err: &CallError) -> Option<&str> {
+    err.details
+        .as_ref()
+        .and_then(|d| d.get("reason"))
+        .and_then(|r| r.as_str())
+}
+
+#[tokio::test]
+async fn dial_failure_is_typed_and_leaves_no_channel() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("dead", Substrate::Tcp, "unreachable")
+        .await;
+    let topo = wire(registry, failing_dial("no route to target")).await;
+
+    let err = call_open_params(&topo, &params("dead"), None)
+        .await
+        .expect_err("failed dial must fail the open");
+    assert_eq!(err.code, "channel:open_failed");
+    assert_eq!(establishment_reason_of(&err), Some("dial_failed"));
+    assert!(no_data_channels(&topo));
+}
+
+#[tokio::test]
+async fn open_denied_without_any_identity() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_with(
+        registry,
+        echo_dial(),
+        None,
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await;
+
+    let err = call_open_params(&topo, &params("echo"), None)
+        .await
+        .expect_err("identity-less open must be denied");
+    assert_eq!(err.code, "FORBIDDEN");
+    assert!(no_data_channels(&topo));
+}
+
+#[tokio::test]
+async fn open_on_transport_identity_alone() {
+    // CF-005 (b): the transport identity (set before dialing)
+    // propagates to the serving dispatch and authorizes the scope
+    // gate. No token payload at all.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let out = call_open_params(&topo, &params("echo"), None)
+        .await
+        .expect("transport-identity open must be authorized");
+    assert!(out["channel_id"].is_u64());
+
+    // CF-006: the establisher's per-call auth carried the END CALLER's
+    // identity (the consumer), not a synthetic install-time value.
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(
+        witness.as_deref(),
+        Some("consumer"),
+        "establisher saw the per-call opener identity (CF-006)"
+    );
+}
+
+#[tokio::test]
+async fn open_token_overrides_transport_identity() {
+    // CF-005 precedence: the payload token wins. The provider resolves
+    // the scoped token to the consumer identity — the witness proves
+    // the establisher saw it end-to-end.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_with(
+        registry,
+        echo_dial(),
+        None,
+        None,
+        Arc::new(harness::TestIdProvider),
+    )
+    .await;
+
+    let out = call_open_params(&topo, &params("echo"), Some(TEST_AUTH_TOKEN))
+        .await
+        .expect("token open authorized");
+    assert!(out["channel_id"].is_u64());
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(witness.as_deref(), Some("consumer"));
+}
+
+#[tokio::test]
+async fn concurrent_opens_same_resource_no_plan_race() {
+    // The R-01 plan flow under concurrency: two opens of the SAME
+    // resource — the establisher returns each dialed handle via ITS
+    // plan; no handoff map to race. Each round-trips its own marker.
+    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..2 {
+        let out = call_open_params(&topo, &params("echo"), None)
+            .await
+            .expect("open must succeed");
+        let channel_id = out["channel_id"].as_u64().unwrap() as u32;
+        let (mut read, mut write, _pump) = adopt_and_pump(&topo, channel_id).await;
+        let msg = format!("marker {i}");
+        write.write_all(msg.as_bytes()).await.expect("write");
+        write.flush().await.expect("flush");
+        let mut buf = vec![0u8; msg.len()];
+        tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+            .await
+            .expect("round trip")
+            .expect("read");
+        assert_eq!(buf, msg.as_bytes());
+        sessions.push(channel_id);
+    }
+    assert_ne!(sessions[0], sessions[1]);
+}
+
+#[tokio::test]
+async fn late_registration_is_visible() {
+    // W2: the harness registers the openable AFTER
+    // from_connection_with_serving. If the dispatcher cached
+    // registrations, opens would fail with unknown-operation — the
+    // successful opens elsewhere prove it; assert registry state too.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("late", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+    assert!(topo
+        .producer_registry
+        .registration("channels/tunnel/sub")
+        .is_some());
+}
+
+#[tokio::test]
+async fn schema_rejects_unknown_substrate_loudly() {
+    // The registry schema-validates input before the establisher
+    // (ADR-001's loud posture): an unknown substrate value is a schema
+    // failure, not a silent pass-through.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let err = call_open(
+        &topo,
+        serde_json::json!({"resource": "echo", "substrate": "sctp"}),
+        None,
+    )
+    .await
+    .expect_err("unknown substrate must fail the open");
+    assert_eq!(err.code, "INVALID_INPUT");
+}
+
+#[tokio::test]
+async fn producer_outbound_calls_still_resolve_while_serving() {
+    // ADR-022 §2 both-sides sanity: the producer is the connect side
+    // AND the serving side; its outbound calling still resolves while
+    // it serves tunnel opens.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let response = topo
+        .producer
+        .call_open_op("consumer/serves/nothing", serde_json::json!({}))
+        .await;
+    assert!(
+        response.result.is_err(),
+        "consumer serves nothing; the point is the reply RESOLVED"
+    );
+}
+
+#[tokio::test]
+async fn registry_lookup_respects_substrate_key() {
+    // The registry is keyed by (resource, substrate): a resource
+    // produced for TCP is unknown for UDP — the same name on a
+    // different substrate is a different resource (ADR-001).
+    let registry = ResourceRegistry::new();
+    registry.register("svc", Substrate::Tcp, "in-process").await;
+    let topo = wire(registry, failing_dial("unreachable")).await;
+
+    let mut p = params("svc");
+    p.substrate = Substrate::Udp;
+    let err = call_open_params(&topo, &p, None)
+        .await
+        .expect_err("udp lookup of a tcp resource must miss");
+    assert_eq!(establishment_reason_of(&err), Some("unknown_resource"));
+}
+
+/// Type-shape assertions kept for harness parity.
+#[allow(dead_code)]
+fn type_assertions() {
+    let _dial = echo_dial();
+    let _scope: &str = TUNNEL_OPEN_SCOPE;
+}