//! Test harness: the producer↔consumer wiring over a
//! `tokio::io::duplex` carrying the channels 8-byte chunk format.
#![allow(dead_code)]
//!
//! 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 serde_json::Value;
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_DIRECT_SCOPE, TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE,
};
use alktunnels::producer::{
    register_tunnel_direct_openable, 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 direct-egress identity: `tunnel:direct` ONLY (no
/// `tunnel:open`); this is the scope-separation probe's identity.
pub fn direct_identity() -> Identity {
    Identity {
        id: "consumer-direct".to_string(),
        scopes: vec![TUNNEL_DIRECT_SCOPE.to_string()],
        resources: std::collections::HashMap::new(),
    }
}

/// Both scopes: `tunnel:open` AND `tunnel:direct` (a deployment
/// granting both grants both explicitly; ADR-006 Amendment 1).
pub fn both_scopes_identity() -> Identity {
    Identity {
        id: "consumer-both".to_string(),
        scopes: vec![
            TUNNEL_OPEN_SCOPE.to_string(),
            TUNNEL_DIRECT_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, the 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>,
    /// The consumer's serving registry (the opener-side op stubs,
    /// e.g. the forwarded-op serving stub, pre-register here).
    pub consumer_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).
/// 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),
    /// 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),
    /// The no-witness dial establisher (`tunnel_establisher`, the
    /// public default shape, no CF-006 probe seam).
    DialNoWitness(alktunnels::producer::DialFn),
    /// The direct op (`channels/tunnel/direct`, ADR-007): the direct
    /// spec + the direct establisher (no registry lookup) over the
    /// SAME injected dial, with the CF-006 witness seam. The witness
    /// rides the topology's `identity_witness` (the direct op's
    /// witness and the base op's never mix in one test).
    Direct(alktunnels::producer::DialFn),
    /// The bind-first establisher (shape 3, ADR-008): the bind-first
    /// establisher + the accept dispatcher over the injected bind +
    /// accept loop + forwarded-open caller. The forwarded-op caller
    /// rides the producer-side call plane (`producer_client`'s call
    /// connection, captured in the topology) and adopts on the
    /// producer-side manager.
    BindFirst(
        alktunnels::producer::BindFn,
        alktunnels::producer::BindFirstAcceptLoop,
    ),
}

pub async fn wire_with(
    registry: ResourceRegistry,
    mode: RegistrationMode,
    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);

    // The consumer's serving registry: empty by default; a test that
    // installs opener-side ops (the forwarded-op serving stub, the
    // ADR-008 §3 ingress shape) pre-registers them here so the
    // dispatch loop serves them from the first inbound call.
    let consumer_serving_registry = Arc::new(OperationRegistry::new());
    let registry_for_hook = Arc::clone(&consumer_serving_registry);
    let identity_for_hook = transport_identity.clone();

    // --- 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();
        let serving_registry = Arc::clone(&registry_for_hook);
        let peer_identity = identity_for_hook.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);
            // The accept-side transport-identity posture (the same
            // propagation `wire_forward` models): the transport
            // authenticated the dialer; attach the resolved identity
            // to channel 0 so the serving dispatch sees the caller
            // (CF-005 (b)'s accept-side analogue).
            if let Some(identity) = peer_identity {
                let _ = channel0_conn.set_identity(identity);
            }
            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 the ops the test pre-registered on
            // its serving registry (empty in the default topology;
            // the dispatcher keeps the call loop alive either way, and its
            // read loop resolves the producer's responses to the
            // consumer's pending entries).
            let dp = Dispatcher::new(
                serving_registry,
                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(),
    );
    let producer_client = Arc::new(producer_client);
    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::Direct(dial) => {
            register_tunnel_direct_openable(
                &core,
                &producer_op_registry,
                AuthContext::anonymous(b"alk/tunnel"),
                dial,
                Some(Arc::clone(&identity_witness)),
            )
            .expect("register tunnel direct 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");
        }
        RegistrationMode::BindFirst(bind, accept_loop) => {
            // The bind-first shape (ADR-008). The forwarded-op caller
            // closes over the producer-side call plane + manager (the
            // design pin: the per-channel `Connection` the handler
            // receives is stream-less and carries no call surface, so
            // the registration takes the call-plane handle, here the
            // serving session's `CallConnection`, which the client
            // holds; in production the assembly layer has the same
            // handle because it constructed the serving session). The
            // harness reaches it via `producer_client.call_open_op` +
            // `adopt_channel`, the reverse POC's
            // `open_reverse_channel` shape, inline (the reply's extra
            // fields are not needed for the forwarded op; the channel
            // ID extraction + adopt is the whole dance).
            let manager = producer_client.manager().clone();
            let client_for_opener = Arc::clone(&producer_client);
            let open_forwarded: alktunnels::producer::ForwardedOpener = Arc::new(
                move |forwarded: alktunnels::params::TunnelForwardedParams| {
                    let client = Arc::clone(&client_for_opener);
                    let manager = manager.clone();
                    Box::pin(async move {
                        let input = serde_json::to_value(&forwarded).map_err(|e| {
                            alktunnels::producer::TunnelEstablishError::HandlerError(format!(
                                "forwarded params serialize: {e}"
                            ))
                        })?;
                        let response = client
                            .call_open_op(alktunnels::params::OP_TUNNEL_FORWARDED, input)
                            .await;
                        let out = response.result.map_err(|e| {
                            alktunnels::producer::TunnelEstablishError::HandlerError(format!(
                                "forwarded open: {}",
                                e.message
                            ))
                        })?;
                        let channel_id =
                            out.get("channel_id")
                                .and_then(|v| v.as_u64())
                                .ok_or_else(|| {
                                    alktunnels::producer::TunnelEstablishError::HandlerError(
                                        "forwarded open reply missing channel_id".to_string(),
                                    )
                                })? as u32;
                        let (send, recv) = manager
                            .adopt_channel(channel_id, alktunnels::params::TUNNEL_ALPN, None)
                            .await
                            .map_err(|e| {
                                alktunnels::producer::TunnelEstablishError::HandlerError(format!(
                                    "forwarded adopt: {e}"
                                ))
                            })?;
                        Ok((send, recv))
                    })
                },
            );
            alktunnels::producer::register_tunnel_bindfirst_openable(
                &core,
                &registry,
                &producer_op_registry,
                AuthContext::anonymous(b"alk/tunnel"),
                alktunnels::producer::BindFirstInjections {
                    bind,
                    accept_loop,
                    open_forwarded,
                },
                Some(Arc::clone(&identity_witness)),
            )
            .expect("register tunnel bindfirst openable");
        }
        RegistrationMode::DialNoWitness(dial) => {
            // The public no-witness wrapper (tunnel_establisher):
            // registration + establisher body identical to the
            // witnessed variant minus the CF-006 seam.
            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(alktunnels::producer::tunnel_establisher(
                    registry.clone(),
                    dial,
                )),
                alktunnels::producer::make_tunnel_pump_handler(),
                &producer_op_registry,
                AuthContext::anonymous(b"alk/tunnel"),
                None,
            )
            .expect("register the no-witness wrapper establisher");
        }
        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).
            // A fresh ChannelCore over the SAME manager + policy (both
            // cores are cheap facades over the same pair; the second
            // registration rides the same ledger/cap state; this arm
            // is local to the probe so the other arms' `core` binding
            // stays untouched).
            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;
                        Err(
                            alkcall::channels::operations::EstablishmentError::HandlerError {
                                message: "hanging establisher resolved; must be timed out"
                                    .to_string(),
                            },
                        )
                    })
                })),
                alktunnels::producer::make_tunnel_pump_handler(),
                &producer_op_registry,
                AuthContext::anonymous(b"alk/tunnel"),
                Some(timeout),
            )
            .expect("register hanging establisher");
        }
    }

    // --- 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,
        consumer_registry: consumer_serving_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,
        RegistrationMode::Dial(dial),
        Some(consumer_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await
}

/// The direct-op topology (ADR-007): the direct op registered with
/// the CF-006 witness; the consumer's transport identity is
/// [`both_scopes_identity`] (both grants; the opens authorize).
pub async fn wire_direct(dial: alktunnels::producer::DialFn) -> Topology {
    wire_with(
        ResourceRegistry::new(),
        RegistrationMode::Direct(dial),
        Some(both_scopes_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await
}

/// [`wire_direct`] with an explicit transport identity (the
/// scope-separation probes construct their own identities).
pub async fn wire_direct_with_identity(identity: Identity) -> Topology {
    wire_with(
        ResourceRegistry::new(),
        RegistrationMode::Direct(failing_dial("unused")),
        Some(identity),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .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`].
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),
    )
    .await
}

/// The bind-first topology (shape 3, ADR-008): the producer's
/// establisher binds via the injected [`BindFn`]; the accept
/// dispatcher runs the injected accept loop and opens forwarded
/// channels back toward the opener (whose serving stub the test
/// installs on the consumer side; see
/// [`serve_forwarded_on_consumer`]). Same identity posture as
/// [`wire`], plus the forwarded scope (the opener-side ingress
/// grant: the consumer identity must hold `tunnel:forwarded` to
/// receive producer-initiated forwarded opens).
pub async fn wire_bindfirst(
    registry: ResourceRegistry,
    bind: alktunnels::producer::BindFn,
    accept_loop: alktunnels::producer::BindFirstAcceptLoop,
) -> Topology {
    wire_with(
        registry,
        RegistrationMode::BindFirst(bind, accept_loop),
        Some(forwarded_opener_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await
}

/// The forwarded-op opener identity: `tunnel:forwarded` AND
/// `tunnel:open` (the opener receives producer-initiated forwarded
/// opens, the ingress grant, and opens the listen tunnel itself,
/// the egress grant; ADR-008 §3 / ADR-006 Amendment 1).
pub fn forwarded_opener_identity() -> Identity {
    Identity {
        id: "opener".to_string(),
        scopes: vec![
            TUNNEL_OPEN_SCOPE.to_string(),
            TUNNEL_FORWARDED_SCOPE.to_string(),
        ],
        resources: std::collections::HashMap::new(),
    }
}

/// [`forwarded_opener_identity`] minus the forwarded grant (the
/// per-accept isolation probe's identity; the forwarded open fails
/// FORBIDDEN).
pub fn forwarded_scopeless_identity() -> Identity {
    Identity {
        id: "opener-scopeless".to_string(),
        scopes: vec![TUNNEL_OPEN_SCOPE.to_string()],
        resources: std::collections::HashMap::new(),
    }
}

/// Serve the forwarded op (`channels/tunnel/forwarded`, ADR-008 §3) on
/// the CONSUMER side (the listen opener's serving stub, the test
/// stand-in for the opener-side registration that lands with
/// `tunnels/forwarded-receipt`): a `register_openable` (no
/// establisher; the data plane starts immediately) whose handler
/// correlates the open by the `listen` token against the test's
/// correlation map, accepts the `BiStream`, and pumps against the
/// "local BIND client" halves the test registered for that token.
///
/// `correlations` maps `listen` token → the pump side to pump the
/// forwarded channel against (the assembly-side bookkeeping posture:
/// the protocol provides the op + params, the test provides the map).
/// The registration is post-hoc on the consumer's captured registry,
/// but the CONSUMER serves no ops in the default harness (its
/// dispatch registry is empty and the dispatcher runs on it), so this
/// helper must be called BEFORE the topology is wired; hence it is a
/// registry-level helper the caller wires via a custom `wire_with`
/// variant when needed.
pub fn register_forwarded_stub_on_consumer(
    consumer_registry: &Arc<OperationRegistry>,
    manager: ChannelManager,
    correlations: CorrelationMap,
) -> Result<(), String> {
    let core = alkcall::channels::operations::ChannelCore::new(
        manager,
        alkcall::channels::policy::default_policy(),
    );
    core.register_openable(
        alktunnels::params::tunnel_forwarded_spec(),
        make_forwarded_stub_handler(correlations),
        consumer_registry,
        AuthContext::anonymous(b"alk/tunnel"),
    )
}

/// The correlation map: `listen` token → the TargetHandle the
/// forwarded channel pumps against (the "local BIND client" halves).
pub type CorrelationMap =
    Arc<tokio::sync::Mutex<std::collections::HashMap<u32, alktunnels::producer::TargetHandle>>>;

/// The forwarded-op handler (the opener-side serving stub's handler):
/// correlate by `listen` → accept the `BiStream` → pump against the
/// correlated halves (`pump_bidi`, convention 8). A correlation miss
/// is a loud `handler_error`-class open failure (the typed posture:
/// the accept dies at birth; the loop keeps serving).
pub fn make_forwarded_stub_handler(
    correlations: CorrelationMap,
) -> alkcall::channels::operations::OpenHandler {
    Arc::new(
        move |input: Value,
              _plan: Option<alkcall::channels::operations::ChannelPlan>,
              conn: alkcall::core::types::Connection,
              _auth: AuthContext| {
            let correlations = Arc::clone(&correlations);
            tokio::spawn(async move {
                let forwarded: alktunnels::params::TunnelForwardedParams =
                    match serde_json::from_value(input) {
                        Ok(p) => p,
                        Err(e) => {
                            tracing::error!(error = %e, "forwarded stub: params parse failed");
                            return;
                        }
                    };
                let target = correlations.lock().await.remove(&forwarded.listen);
                let Some(target) = target else {
                    tracing::error!(
                        listen = forwarded.listen,
                        "forwarded stub: correlation miss"
                    );
                    return;
                };
                let Some(bidi) = conn.accept_bi().await.ok() else {
                    tracing::error!("forwarded stub: accept_bi failed");
                    return;
                };
                alkcall::channels::pump::pump_bidi(bidi, target.read, target.write).await;
            })
        },
    )
}

/// The forward topology (the `-L` direction): the consumer dials the
/// transport (connect side, `ChannelClient::from_connection`, a pure
/// consumer serving nothing) and the producer accepts (the adapter +
/// an install hook that registers the channel ops + the tunnel
/// openable on the producer's serving registry). This is the topology
/// `TunnelSession::open(&channel_client, params)` is normative for:
/// the consumer's own client calls the open op on the producer's
/// channel 0.
///
/// Identity on the accept-side serving path (the mTLS posture): the
/// transport authenticated the dialer; the hook attaches the resolved
/// identity to channel 0 so the serving dispatch sees the caller
/// (the accept-side analogue of CF-005 (b)'s propagation).
pub struct ForwardTopology {
    /// The consumer's own client (connect side, pure consumer). Its
    /// manager adopted the producer-allocated channel IDs; the
    /// session reaps via this manager.
    pub consumer: Arc<ChannelClient>,
    /// The producer's manager (accept side, serving), for asserting
    /// wrapper-managed reaping.
    pub producer_manager: ChannelManager,
    /// What the establisher's per-call auth carried (the CF-006 probe).
    pub identity_witness: Arc<tokio::sync::Mutex<Option<String>>>,
}

pub async fn wire_forward(
    registry: ResourceRegistry,
    dial: alktunnels::producer::DialFn,
) -> ForwardTopology {
    let producer_op_registry = Arc::new(OperationRegistry::new());
    let (producer_manager_tx, mut producer_manager_rx) =
        tokio::sync::mpsc::channel::<ChannelManager>(1);
    let identity_witness = Arc::new(tokio::sync::Mutex::new(None::<String>));
    let dial_witness = Arc::clone(&identity_witness);

    // --- producer (accept side, serving): adapter + install hook ------
    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
        let registry_arc = Arc::clone(&producer_op_registry);
        let resource_registry = registry.clone();
        let dial = Arc::clone(&dial);
        let witness = Arc::clone(&dial_witness);
        let manager_tx = producer_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);
            // The accept-side transport-identity posture: the transport
            // authenticated the dialer; attach the resolved identity to
            // channel 0 (the dispatch reads it as the caller identity).
            let _ = channel0_conn.set_identity(consumer_identity());
            let call_connection = Arc::new(CallConnection::new_single_stream(
                channel0_conn,
                Arc::clone(&writer),
            ));

            // Register the generic channel ops + the tunnel openable
            // (the producer.md recipe) before the dispatch loop runs.
            let core = alkcall::channels::operations::ChannelCore::new(
                manager.clone(),
                alkcall::channels::policy::default_policy(),
            );
            let ops = ChannelOperations::with_default_policy(manager.clone());
            ops.register_on(&registry_arc)
                .expect("register channel ops on producer");
            register_tunnel_openable(
                &core,
                &resource_registry,
                &registry_arc,
                AuthContext::anonymous(b"alk/tunnel"),
                dial,
                Some(witness),
            )
            .expect("register tunnel openable");
            let _ = manager_tx.send(manager).await;

            let dp = Dispatcher::new(
                registry_arc,
                Arc::new(alkcall::core::auth::NoopIdentityProvider),
            );
            dp.serve_single_stream(call_connection, reader, writer)
                .await;
        })
    });

    // --- transport ------------------------------------------------------
    let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);

    // The dialer's transport identity (the mTLS/QUIC posture), set
    // before from_connection. The accept side resolves it out-of-band
    // (modeled in the hook above).
    consumer_conn
        .set_identity(consumer_identity())
        .expect("transport identity set once");

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
    let _adapter_task = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(
            &adapter,
            producer_conn,
            &producer_transport_auth,
        )
        .await;
    });

    // --- consumer (connect side): pure client ---------------------------
    let consumer_client = ChannelClient::from_connection(consumer_conn)
        .await
        .expect("consumer channel client init");
    let consumer_client = Arc::new(consumer_client);

    let producer_manager = producer_manager_rx
        .recv()
        .await
        .expect("producer manager captured");

    ForwardTopology {
        consumer: consumer_client,
        producer_manager,
        identity_witness,
    }
}

/// The DIRECT-op forward topology (ADR-007): the producer (accept
/// side, serving) registers the direct op
/// ([`register_tunnel_direct_openable`] over the injected dial, CF-006
/// witness attached) and the consumer
/// (connect side, pure client) holds the identity the session tests
/// authorize with. This is the topology `TunnelSession::open_direct`
/// is normative for.
pub async fn wire_direct_forward(dial: alktunnels::producer::DialFn) -> ForwardTopology {
    wire_direct_forward_with_identity(dial, both_scopes_identity()).await
}

/// [`wire_direct_forward`] with an explicit transport identity (the
/// session-level scope-separation probes construct their own).
pub async fn wire_direct_forward_with_identity(
    dial: alktunnels::producer::DialFn,
    identity: Identity,
) -> ForwardTopology {
    let producer_op_registry = Arc::new(OperationRegistry::new());
    let (producer_manager_tx, mut producer_manager_rx) =
        tokio::sync::mpsc::channel::<ChannelManager>(1);
    let identity_witness = Arc::new(tokio::sync::Mutex::new(None::<String>));
    let dial_witness = Arc::clone(&identity_witness);
    let expected_caller = identity.clone();
    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
        let registry_arc = Arc::clone(&producer_op_registry);
        let dial = Arc::clone(&dial);
        let witness = Arc::clone(&dial_witness);
        let manager_tx = producer_manager_tx.clone();
        let caller = expected_caller.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);
            // The accept-side transport-identity posture: the transport
            // authenticated the dialer; attach the RESOLVED identity
            // (scopes included, resolved out-of-band) to channel 0.
            let _ = channel0_conn.set_identity(caller);
            let call_connection = Arc::new(CallConnection::new_single_stream(
                channel0_conn,
                Arc::clone(&writer),
            ));
            let core = alkcall::channels::operations::ChannelCore::new(
                manager.clone(),
                alkcall::channels::policy::default_policy(),
            );
            let ops = ChannelOperations::with_default_policy(manager.clone());
            ops.register_on(&registry_arc)
                .expect("register channel ops on producer");
            register_tunnel_direct_openable(
                &core,
                &registry_arc,
                AuthContext::anonymous(b"alk/tunnel"),
                dial,
                Some(witness),
            )
            .expect("register tunnel direct openable");
            let _ = manager_tx.send(manager).await;
            let dp = Dispatcher::new(
                registry_arc,
                Arc::new(alkcall::core::auth::NoopIdentityProvider),
            );
            dp.serve_single_stream(call_connection, reader, writer)
                .await;
        })
    });

    let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
    consumer_conn
        .set_identity(identity)
        .expect("transport identity set once");

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
    let _adapter_task = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(
            &adapter,
            producer_conn,
            &producer_transport_auth,
        )
        .await;
    });

    let consumer_client = ChannelClient::from_connection(consumer_conn)
        .await
        .expect("consumer channel client init");
    let consumer_client = Arc::new(consumer_client);
    let producer_manager = producer_manager_rx
        .recv()
        .await
        .expect("producer manager captured");

    ForwardTopology {
        consumer: consumer_client,
        producer_manager,
        identity_witness,
    }
}

/// The listen-flow topology (ADR-008 §4, the `TunnelListener` shape):
/// the OPENER dials the transport (connect side,
/// `ChannelClient::from_connection_with_serving`, so it can both open
/// the listen tunnel and serve the forwarded op), and the producer
/// accepts (the adapter + an install hook registering the bind-first
/// openable over the injected bind + accept loop). The producer-side
/// forwarded-open caller rides the accept-side call plane (the
/// serving session's `CallConnection`, the same surface the serving
/// dispatch loop runs on; both-sides sanity, ADR-022 §2) and adopts on
/// the producer-side manager. The forwarded op registers post-hoc (W2)
/// on the opener's captured `opener_registry`.
pub struct ListenerTopology {
    /// The opener's own client (connect side, serving). Its manager
    /// adopted the producer-allocated listen channel ID (via
    /// `open_channel_with_reply`); the listener reaps via this
    /// manager.
    pub opener: Arc<ChannelClient>,
    /// The opener's serving registry (the forwarded op registers here
    /// post-hoc).
    pub opener_registry: Arc<OperationRegistry>,
    /// The producer's manager (accept side, serving), for asserting
    /// wrapper-managed reaping + for the test's adopt dance.
    pub producer_manager: ChannelManager,
    /// The producer-side call plane (`Arc<CallConnection>`), the
    /// forwarded-op caller's surface (and the test's direct-call
    /// surface for the adopt-pin dance).
    pub producer_call: Arc<CallConnection>,
    /// What the establisher's per-call auth carried (the CF-006 probe).
    pub identity_witness: Arc<tokio::sync::Mutex<Option<String>>>,
}

pub async fn wire_listener_bindfirst(
    registry: ResourceRegistry,
    bind: alktunnels::producer::BindFn,
    accept_loop: alktunnels::producer::BindFirstAcceptLoop,
    opener_identity: Identity,
) -> ListenerTopology {
    use alktunnels::producer::{
        register_tunnel_bindfirst_openable, ForwardedOpener, TunnelEstablishError,
    };

    let producer_op_registry = Arc::new(OperationRegistry::new());
    let opener_op_registry = Arc::new(OperationRegistry::new());
    let (producer_manager_tx, mut producer_manager_rx) =
        tokio::sync::mpsc::channel::<ChannelManager>(1);
    let (producer_call_tx, mut producer_call_rx) =
        tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
    let identity_witness = Arc::new(tokio::sync::Mutex::new(None::<String>));
    let bind_witness = Arc::clone(&identity_witness);
    let opener_identity_for_hook = opener_identity.clone();

    // --- producer (accept side, serving): adapter + install hook ------
    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
        let registry_arc = Arc::clone(&producer_op_registry);
        let resource_registry = registry.clone();
        let bind = Arc::clone(&bind);
        let accept_loop = Arc::clone(&accept_loop);
        let witness = Arc::clone(&bind_witness);
        let manager_tx = producer_manager_tx.clone();
        let call_tx = producer_call_tx.clone();
        let caller = opener_identity_for_hook.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);
            // The accept-side transport-identity posture (modeled): the
            // transport authenticated the dialer (the opener); attach
            // the resolved identity to channel 0 so the serving
            // dispatch sees the caller (CF-005 (b)'s accept-side
            // analogue).
            let _ = channel0_conn.set_identity(caller);
            let call_connection = Arc::new(CallConnection::new_single_stream(
                channel0_conn,
                Arc::clone(&writer),
            ));

            // Register the generic channel ops + the bind-first
            // openable (the producer.md recipe) before the dispatch
            // loop runs.
            let core = alkcall::channels::operations::ChannelCore::new(
                manager.clone(),
                alkcall::channels::policy::default_policy(),
            );
            let ops = ChannelOperations::with_default_policy(manager.clone());
            ops.register_on(&registry_arc)
                .expect("register channel ops on producer");

            // The forwarded-open caller (the call plane): call + adopt
            // (the reverse POC's `open_reverse_channel` shape: the
            // serving session's `CallConnection` is the per-accept
            // call plane, and the producer-side manager adopts the
            // opener-allocated ID).
            let call_for_opener = Arc::clone(&call_connection);
            let manager_for_opener = manager.clone();
            let open_forwarded: ForwardedOpener = Arc::new(
                move |forwarded: alktunnels::params::TunnelForwardedParams| {
                    let call = Arc::clone(&call_for_opener);
                    let manager = manager_for_opener.clone();
                    Box::pin(async move {
                        let input = serde_json::to_value(&forwarded).map_err(|e| {
                            TunnelEstablishError::HandlerError(format!(
                                "forwarded params serialize: {e}"
                            ))
                        })?;
                        let response = call
                            .call(alktunnels::params::OP_TUNNEL_FORWARDED, input)
                            .await;
                        let out = response.result.map_err(|e| {
                            TunnelEstablishError::HandlerError(format!(
                                "forwarded open: {}",
                                e.message
                            ))
                        })?;
                        let channel_id =
                            out.get("channel_id")
                                .and_then(|v| v.as_u64())
                                .ok_or_else(|| {
                                    TunnelEstablishError::HandlerError(
                                        "forwarded open reply missing channel_id".to_string(),
                                    )
                                })? as u32;
                        let (send, recv) = manager
                            .adopt_channel(channel_id, alktunnels::params::TUNNEL_ALPN, None)
                            .await
                            .map_err(|e| {
                                TunnelEstablishError::HandlerError(format!("forwarded adopt: {e}"))
                            })?;
                        Ok((send, recv))
                    })
                },
            );

            register_tunnel_bindfirst_openable(
                &core,
                &resource_registry,
                &registry_arc,
                AuthContext::anonymous(b"alk/tunnel"),
                alktunnels::producer::BindFirstInjections {
                    bind,
                    accept_loop,
                    open_forwarded,
                },
                Some(witness),
            )
            .expect("register tunnel bindfirst openable");
            let _ = manager_tx.send(manager).await;
            let _ = call_tx.send(Arc::clone(&call_connection)).await;

            let dp = Dispatcher::new(
                registry_arc,
                Arc::new(alkcall::core::auth::NoopIdentityProvider),
            );
            dp.serve_single_stream(call_connection, reader, writer)
                .await;
        })
    });

    // --- transport ------------------------------------------------------
    let (opener_end, producer_end) = tokio::io::duplex(64 * 1024);
    let opener_conn = CoreConnection::from_bidi(opener_end, b"alk/channels".to_vec(), None);
    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
    // The dialer's transport identity (the mTLS/QUIC posture), set
    // before from_connection_with_serving. The accept side resolves it
    // out-of-band (modeled in the hook above).
    opener_conn
        .set_identity(opener_identity.clone())
        .expect("transport identity set once");

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
    let _adapter_task = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(
            &adapter,
            producer_conn,
            &producer_transport_auth,
        )
        .await;
    });

    // --- opener (connect side): client + serving (the forwarded op) -----
    let opener_client = ChannelClient::from_connection_with_serving(
        opener_conn,
        Some(ServingConfig {
            registry: Arc::clone(&opener_op_registry),
            identity_provider: Arc::new(alkcall::core::auth::NoopIdentityProvider),
            identity: None,
        }),
    )
    .await
    .expect("opener channel client init");
    let opener_client = Arc::new(opener_client);

    let producer_manager = producer_manager_rx
        .recv()
        .await
        .expect("producer manager captured");
    let producer_call = producer_call_rx
        .recv()
        .await
        .expect("producer call plane captured");

    ListenerTopology {
        opener: opener_client,
        opener_registry: opener_op_registry,
        producer_manager,
        producer_call,
        identity_witness,
    }
}

/// 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 both-substrates dial: stream substrates echo via pipes, UDP
/// echoes via the framed codec (the datagram tests' single dial).
pub fn all_substrate_dial() -> alktunnels::producer::DialFn {
    let echo = echo_dial();
    let udp = framed_udp_echo_dial();
    Arc::new(move |substrate: Substrate, backing: &str| match substrate {
        Substrate::Udp => udp(substrate, backing),
        Substrate::Tcp | Substrate::Unix => echo(substrate, backing),
    })
}

/// A framed UDP echo dial whose target end CLOSES after the first
/// echoed datagram, the EOF-path harness. `payload_completes`
/// controls the truncation shape: `true` forwards the whole frame
/// (echo + payload intact) then drops → the clean-EOF path;
/// `false` forwards only the 2-byte length prefix then drops → the
/// mid-datagram EOF path (`TruncatedDatagram`).
pub fn closing_udp_echo_dial(payload_completes: bool) -> alktunnels::producer::DialFn {
    Arc::new(move |substrate: Substrate, backing: &str| {
        let backing = backing.to_string();
        Box::pin(async move {
            match substrate {
                Substrate::Udp => {
                    use alktunnels::wire::DatagramReader;
                    use tokio::io::{AsyncReadExt, AsyncWriteExt};
                    let (consumer_side, target_side) = tokio::io::duplex(64 * 1024);
                    tokio::spawn(async move {
                        let (mut t_read, mut t_write) = tokio::io::split(target_side);
                        let mut reader = DatagramReader::new();
                        let mut buf = vec![0u8; 16 * 1024];
                        // Read until the first complete frame (the
                        // echo-closes-after-first shape): forward it
                        // whole (clean-EOF shape) or forward only its
                        // length prefix (truncated shape), then drop
                        // the far end; the channel read half EOFs.
                        let n = match t_read.read(&mut buf).await {
                            Ok(0) | Err(_) => return,
                            Ok(n) => n,
                        };
                        let dgs = reader.feed(&buf[..n]);
                        if let Some(dg) = dgs.into_iter().next() {
                            let framed = match alktunnels::wire::frame_datagram(&dg) {
                                Ok(f) => f,
                                Err(_) => return,
                            };
                            if payload_completes {
                                if t_write.write_all(&framed).await.is_err() {
                                    return;
                                }
                                let _ = t_write.flush().await;
                            } else {
                                // The truncated shape: forward ONLY
                                // the 2-byte length prefix and drop;
                                // the declared payload never arrives.
                                if t_write.write_all(&framed[..2]).await.is_err() {
                                    return;
                                }
                                let _ = t_write.flush().await;
                            }
                        }
                    });
                    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::Tcp | Substrate::Unix => {
                    Err(alktunnels::producer::TunnelEstablishError::DialFailed(
                        format!("stream dial not wired in the udp 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(),
            ))
        })
    })
}

/// A framed UDP echo dial (the ADR-003 codec at the target boundary,
/// the in-process stand-in for the real `local` UDP adapter): the
/// dial's halves carry the `[len: u16 BE]` wire framing; a spawned
/// echo task decodes frames from the channel side and re-frames the
/// echoed payloads back. The producer pump copies raw bytes; the
/// codec lives at this boundary (the pump never sees UDP specifics).
pub fn framed_udp_echo_dial() -> alktunnels::producer::DialFn {
    Arc::new(move |substrate: Substrate, backing: &str| {
        let backing = backing.to_string();
        Box::pin(async move {
            match substrate {
                Substrate::Udp => {
                    let (consumer_side, target_side) = tokio::io::duplex(64 * 1024);
                    tokio::spawn(async move {
                        use alktunnels::wire::DatagramReader;
                        use tokio::io::{AsyncReadExt, AsyncWriteExt};
                        let (mut t_read, mut t_write) = tokio::io::split(target_side);
                        let mut reader = DatagramReader::new();
                        let mut buf = vec![0u8; 16 * 1024];
                        loop {
                            let n = match t_read.read(&mut buf).await {
                                Ok(0) | Err(_) => return,
                                Ok(n) => n,
                            };
                            let dgs = reader.feed(&buf[..n]);
                            for dg in dgs {
                                let framed = match alktunnels::wire::frame_datagram(&dg) {
                                    Ok(f) => f,
                                    Err(_) => return,
                                };
                                if t_write.write_all(&framed).await.is_err() {
                                    return;
                                }
                                let _ = t_write.flush().await;
                            }
                        }
                    });
                    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::Tcp | Substrate::Unix => {
                    Err(alktunnels::producer::TunnelEstablishError::DialFailed(
                        format!("stream dial not wired in the udp harness: {backing}"),
                    ))
                }
            }
        })
    })
}