feat(review 008 Unit 3a): in-tree ChannelRelay (ADR-051, ADR-042 amendment)

- src/channels/relay.rs: ChannelRelay::register_relay_openable —
  registers a from_call-imported marked spec on a consumer-leg
  channel-0 registry via the full open-op wrapper (translate hop =
  the ADR-049 establisher calling the producer leg with the
  forwarded payload; byte-forward hop = pump_bidi, awaited inline).
  The relay map dissolves (implicit per-channel mapping; ids are
  per-connection), the spoke reply's channel_id is stripped, the
  reply's other fields ride with_reply_fields (bound flows
  end-to-end), reason mapping per ADR-051 §3 (spoke code+message
  preserved; timeout → dial_failed).
- Rejection postures (ADR-051 §6): unmarked and Pub-typed marked
  specs are loud assembly errors.
- Gate tests (9): registration rejections, reason mapping,
  forwarded_for payload shape, the real-wire consumer → hub → spoke
  round trip (bound survives, data both directions, the hub never
  parses the data plane), spoke-failure teardown (ledger
  decremented, no spoke channel leak), and a channels/tty/sub
  standard-shape run pinning no regression.

Verification: cargo test (658 passed), clippy --all-targets
-D warnings, fmt --check, doc --no-deps — all clean.

src/channels/relay.rs

@@ -0,0 +1,346 @@
+//! `ChannelRelay` — the in-tree channel relay component (ADR-042 as
+//! amended by ADR-051; review 008 Unit 3a).
+//!
+//! A relay hub imports a producer leg's channel-open op (via
+//! `from_call` discovery — the spec arrives reconstructed WITH the
+//! `channel_open` marker, ADR-047 amendment 3) and re-exposes it on
+//! its own consumer-leg channel-0 fork registry. The open op resolves
+//! on the hub with a hub-allocated `channel_id`; the relay's
+//! establisher (the translate hop, ADR-042 layer 1) opens the channel
+//! on the producer leg and adopts the spoke-allocated id into the
+//! producer-leg `ChannelManager`; the relay's `OpenHandler` (the
+//! byte-forward hop, layer 2) pumps bytes between the two legs' data
+//! planes with `pump_bidi` — the hub never parses chunk framing
+//! (ADR-034/ADR-035: ids are per-connection, no rewrite exists).
+//!
+//! The consumer leg keeps the full open-op wrapper machinery (ACL,
+//! per-identity cap/ledger, establishment bound, teardown-on-failure)
+//! — the relay registers through
+//! [`ChannelCore::register_openable_with_establisher`], never a
+//! parallel authorization path.
+//!
+//! See ADR-042 (the relay contract), ADR-051 (the in-tree design:
+//! §2 what the relay holds, §3 the reason mapping, §4 the
+//! registration seam, §6 the rejection postures).
+
+use std::sync::Arc;
+
+use serde_json::{Map, Value};
+
+use crate::core::auth::AuthContext;
+use crate::protocol::connection::CallConnection;
+use crate::protocol::wire::CallError;
+use crate::registry::registration::OperationRegistry;
+use crate::registry::spec::{OperationSpec, OperationType};
+
+use super::manager::ChannelManager;
+use super::operations::{
+    ChannelCore, Establishment, EstablishmentError, OpenEstablisher, OpenHandler,
+    CHANNEL_OPEN_FAILED, RESERVED_REPLY_KEY,
+};
+use super::pump::pump_bidi;
+use super::reassembly::{MpscRecvStream, MpscSendStream};
+
+/// The producer-leg surface the relay holds (ADR-051 §2): the
+/// `CallConnection` for the translate hop's open call plus the
+/// producer-leg `ChannelManager` for `adopt_channel` and teardown
+/// visibility. Not a `ChannelClient` — the hub's `CallConnection` is
+/// shared by import, relay establisher, and hub ops (three claimants),
+/// so `take_call_connection`'s detach is wrong for a hub. A hub that
+/// dials the spoke with `ChannelClient::from_connection`, imports via
+/// `from_call`, then tears the client down keeps the relay alive on
+/// these `Arc`s.
+#[derive(Clone)]
+pub struct ProducerLeg {
+    /// The producer leg's call connection — the hub's channel-0
+    /// connection to the spoke, the one `from_call` imported the open
+    /// op over.
+    pub call: Arc<CallConnection>,
+    /// The producer leg's channel manager — the relay's establisher
+    /// adopts the spoke-allocated channel id here.
+    pub manager: ChannelManager,
+}
+
+/// The plan payload the relay's establisher hands its pump handler:
+/// the spoke-leg streams. The establisher and the handler agree on
+/// this concrete type (the `ChannelPlan` downcast contract, ADR-049
+/// amendment 2); alkcall never inspects it.
+struct RelayPlan {
+    spoke_send: MpscSendStream,
+    spoke_recv: MpscRecvStream,
+}
+
+/// The relay component (ADR-051). Holds the producer-leg surface;
+/// [`ChannelRelay::register_relay_openable`] registers a
+/// from_call-imported marked spec on a consumer-leg channel-0 fork
+/// registry with the translate + byte-forward machinery.
+#[derive(Clone)]
+pub struct ChannelRelay {
+    producer_leg: ProducerLeg,
+}
+
+impl ChannelRelay {
+    /// Construct a relay over a producer leg.
+    pub fn new(producer_leg: ProducerLeg) -> Self {
+        Self { producer_leg }
+    }
+
+    /// The producer-leg surface this relay translates onto.
+    pub fn producer_leg(&self) -> &ProducerLeg {
+        &self.producer_leg
+    }
+
+    /// Register a from_call-imported marked spec as a relay openable
+    /// on the consumer-leg channel-0 registry (ADR-051 §4 phase 2 —
+    /// the fork is the `consumer_registry` argument).
+    ///
+    /// - `consumer_core` is the consumer leg's [`ChannelCore`] (its
+    ///   channel-0 manager + lifecycle policy) — the full open-op
+    ///   wrapper machinery (ACL, cap/ledger, establishment bound,
+    ///   teardown) runs on the consumer leg exactly as for a locally
+    ///   produced open op.
+    /// - `spec` must carry the `channel_open` marker (reconstructed
+    ///   WITH the marker by discovery, ADR-047 amendment 3). A
+    ///   `Pub`-typed marked spec is a loud assembly error
+    ///   ([`RelayError::PubTypedOpen`] — the C-08 blocker, ADR-051
+    ///   §6); an unmarked spec is likewise an error — branch on the
+    ///   marker before calling.
+    /// - The translate hop calls the producer leg with the spec's own
+    ///   op name; a hub that re-names the consumer-visible op composes
+    ///   its own establisher via
+    ///   [`ChannelCore::register_openable_with_establisher`] instead.
+    pub fn register_relay_openable(
+        &self,
+        consumer_core: &ChannelCore,
+        consumer_registry: &OperationRegistry,
+        spec: OperationSpec,
+        consumer_auth: AuthContext,
+    ) -> Result<(), RelayError> {
+        if spec.channel_open.is_none() {
+            return Err(RelayError::UnmarkedSpec {
+                message: format!(
+                    "spec `{}` has no channel_open marker — relay registration is \
+                     for from_call-imported marked specs (ADR-051 §6)",
+                    spec.name
+                ),
+            });
+        }
+        if spec.op_type == OperationType::Pub {
+            return Err(RelayError::PubTypedOpen {
+                message: format!(
+                    "spec `{}` is a Pub-typed channel-open op — the Pub open path is \
+                     not implemented (C-08 blocker, `channel:pub_open_not_implemented` \
+                     class; ADR-051 §6)",
+                    spec.name
+                ),
+            });
+        }
+
+        let producer_call = Arc::clone(&self.producer_leg.call);
+        let producer_manager = self.producer_leg.manager.clone();
+        let op_name = spec.name.clone();
+        let establisher: OpenEstablisher = Arc::new(move |input, per_call_auth| {
+            let producer_call = Arc::clone(&producer_call);
+            let producer_manager = producer_manager.clone();
+            let op_name = op_name.clone();
+            Box::pin(async move {
+                open_on_producer_leg(
+                    &producer_call,
+                    &producer_manager,
+                    &op_name,
+                    input,
+                    &per_call_auth,
+                )
+                .await
+            })
+        });
+
+        let open_handler: OpenHandler = Arc::new(move |_input, plan, channel_conn, _auth| {
+            tokio::spawn(async move {
+                let Some(plan) = plan else {
+                    return;
+                };
+                // The plan is uniquely held here (establisher →
+                // wrapper → handler, moved each hop), so `try_unwrap`
+                // succeeds; on the phantom-holder case the wrapper's
+                // no-accept telemetry fires and the spoke leg EOFs
+                // when the plan finally drops.
+                let Ok(relay_plan) = plan.downcast::<RelayPlan>() else {
+                    tracing::debug!("relay: establisher plan of unexpected type");
+                    return;
+                };
+                let Ok(relay_plan) = Arc::try_unwrap(relay_plan) else {
+                    tracing::debug!("relay: establisher plan held elsewhere; skipping pump");
+                    return;
+                };
+                let Ok(bidi) = channel_conn.accept_bi().await else {
+                    return;
+                };
+                // The byte-forward hop (ADR-042 layer 2, ADR-050):
+                // two pumps joined inline (R-02). On completion the
+                // pump's drops EOF the spoke leg; the wrapper's
+                // handler-exit teardown covers the consumer leg.
+                let _ = pump_bidi(bidi, relay_plan.spoke_recv, relay_plan.spoke_send).await;
+            })
+        });
+
+        consumer_core
+            .register_openable_with_establisher(
+                spec,
+                Some(establisher),
+                open_handler,
+                consumer_registry,
+                consumer_auth,
+                None,
+            )
+            .map_err(|e| RelayError::Registration { message: e })
+    }
+}
+
+/// The translate hop (ADR-051 §1 layer 1, §2): call the producer leg's
+/// open op with the forwarded payload, adopt the returned spoke
+/// `channel_id` into the producer-leg manager, and return
+/// `Establishment::new(plan)` whose plan carries the spoke-leg
+/// streams. The spoke reply's `channel_id` is stripped (the consumer
+/// reply carries the hub-allocated id); the reply's other fields ride
+/// `with_reply_fields` (`bound` flows end-to-end, per-hop truthful).
+async fn open_on_producer_leg(
+    producer_call: &Arc<CallConnection>,
+    producer_manager: &ChannelManager,
+    op_name: &str,
+    input: Value,
+    per_call_auth: &AuthContext,
+) -> Result<Establishment, EstablishmentError> {
+    let payload = build_relay_payload(op_name, input, per_call_auth);
+    let response = producer_call.call_with_payload(payload).await;
+    let out = match response.result {
+        Ok(out) => out,
+        Err(spoke_error) => return Err(map_spoke_error(&spoke_error)),
+    };
+
+    let spoke_id = out
+        .get("channel_id")
+        .and_then(|v| v.as_u64())
+        .ok_or_else(|| EstablishmentError::HandlerError {
+            message: format!(
+                "spoke open reply for `{op_name}` missing `channel_id` — malformed responder"
+            ),
+        })? as u32;
+
+    // Adopt the spoke-allocated id into the producer-leg manager so
+    // the spoke leg's mux/demux route this channel — the same shape
+    // `ChannelClient::open_channel` performs, minus the client. The
+    // producer leg is the connect side of the hub↔spoke channels
+    // connection, so the spoke's ids are even and disjoint from the
+    // hub's odd connect-side allocations.
+    let (spoke_send, spoke_recv) = producer_manager
+        .adopt_channel(spoke_id, "alk/relay", None)
+        .await
+        .map_err(|e| EstablishmentError::ResourceShortage {
+            message: format!(
+                "spoke channel {spoke_id} adopt failed on the hub's producer leg: {e}"
+            ),
+        })?;
+
+    let mut reply_fields = Map::new();
+    if let Some(obj) = out.as_object() {
+        for (key, value) in obj {
+            if key != RESERVED_REPLY_KEY {
+                reply_fields.insert(key.clone(), value.clone());
+            }
+        }
+    }
+
+    let establishment = Establishment::new(Arc::new(RelayPlan {
+        spoke_send,
+        spoke_recv,
+    }) as super::operations::ChannelPlan);
+    Ok(if reply_fields.is_empty() {
+        establishment
+    } else {
+        establishment.with_reply_fields(reply_fields)
+    })
+}
+
+/// Map a spoke failure into the establisher vocabulary (ADR-051 §3 —
+/// the reason table; the spoke's code and message are preserved in the
+/// mapped variant's `message`, never discarded — a consumer debugging
+/// a FORBIDDEN-through-relay sees the spoke's grant failure).
+///
+/// The `timeout` row is the one non-1:1 case: the consumer-visible
+/// reason is `dial_failed` because establishment timeouts are
+/// wrapper-generated (ADR-049 §3) and not an establisher vocabulary
+/// word; the spoke's message carries the truth.
+fn map_spoke_error(spoke_error: &CallError) -> EstablishmentError {
+    if spoke_error.code == CHANNEL_OPEN_FAILED {
+        let reason = spoke_error
+            .details
+            .as_ref()
+            .and_then(|d| d.get("reason"))
+            .and_then(|r| r.as_str())
+            .unwrap_or("");
+        let detail = spoke_error
+            .details
+            .as_ref()
+            .and_then(|d| d.get("message"))
+            .and_then(|m| m.as_str())
+            .unwrap_or(spoke_error.message.as_str());
+        let message = format!("spoke leg: {reason}: {detail}");
+        return match reason {
+            "dial_failed" | "timeout" => EstablishmentError::DialFailed { message },
+            "unknown_resource" => EstablishmentError::UnknownResource { message },
+            "resource_shortage" => EstablishmentError::ResourceShortage { message },
+            _ => EstablishmentError::HandlerError { message },
+        };
+    }
+    // `NOT_FOUND` (spoke predates the op), `FORBIDDEN` (hub lacks the
+    // spoke grant), and every other call-error class.
+    EstablishmentError::HandlerError {
+        message: format!("spoke leg: {} ({})", spoke_error.code, spoke_error.message),
+    }
+}
+
+/// Build the `call.requested` payload for the translate hop: the hub
+/// as caller, the consumer as `forwarded_for` (ADR-026 §3, ADR-032 §3
+/// — the same shape `build_forwarded_payload` produces for the plain
+/// forwarding stubs). `forwarded_for` is metadata, never consulted by
+/// any `AccessControl::check`; it is omitted when the consumer's call
+/// carried no resolved identity.
+fn build_relay_payload(op_name: &str, input: Value, per_call_auth: &AuthContext) -> Value {
+    let mut payload = Map::new();
+    payload.insert(
+        "operationId".to_string(),
+        Value::String(op_name.to_string()),
+    );
+    payload.insert("input".to_string(), input);
+    if let Some(originator) = &per_call_auth.identity {
+        if let Ok(value) = serde_json::to_value(originator) {
+            payload.insert("forwarded_for".to_string(), value);
+        }
+    }
+    Value::Object(payload)
+}
+
+/// The assembly-time errors `ChannelRelay::register_relay_openable`
+/// produces (ADR-051 §6 — the rejection postures are loud, never
+/// silent stubs).
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum RelayError {
+    /// The spec carries no `channel_open` marker — the relay only
+    /// registers marked specs.
+    #[error("unmarked spec: {message}")]
+    UnmarkedSpec { message: String },
+    /// A `Pub`-typed marked spec — the Pub open path is not
+    /// implemented (C-08 blocker; `channel:pub_open_not_implemented`
+    /// class).
+    #[error("pub-typed open op: {message}")]
+    PubTypedOpen { message: String },
+    /// The consumer-leg registration itself failed (schema compile,
+    /// handler-kind mismatch).
+    #[error("registration failed: {message}")]
+    Registration { message: String },
+}
+
+#[cfg(test)]
+#[path = "relay_tests.rs"]
+mod tests;