fix(review-008 audit): adopted-entry drop guard; explicit-ALPN guards; review 009

Post-landing audit of the 0.7.1 -> 0.8.0 remediation diff: two
hardening guards, one rejection-posture fix, two log/message
corrections, and the deferred coverage debt filed as review 009.

- RelayPlan owns the producer-leg ChannelManager and reclaims the
  adopted spoke channel_id via a Drop guard (replaces the pump
  handler's post-pump_bidi explicit reclaim). Closes the leak
  windows the pump's normal path cannot reach: the wrapper's
  establishment bound expiring after the adopt, and the pump
  handler's early-return arms (plan absent, downcast failure,
  try_unwrap failure, accept_bi failure). The send-half drop still
  EOFs the spoke leg via the mux pump's implicit-EOF sentinel, so
  the spoke-side cascade is unchanged. ADR-051 §6 documents the
  closed post-adopt window (the pre-adopt §6 window and the
  inside-adopt_channel cancellation point stay as documented).
- rebuild_spec_for trims and rejects empty/whitespace
  channel_open_alpn strings — an empty explicit string previously
  overrode a sane name-derived ALPN.
- op_name_is_standard_channel_open_shape applies the same
  empty-segment guard as the derivation: channels//sub no longer
  serializes boolean-only and then reconstructs unmarked (silent
  stub for a marked op); the explicit string rides instead.
- reserved_reply_key_call_error interpolates RESERVED_REPLY_KEY;
  the establisher-bug log fires at warn! (programming error).
- Regression tests: the plan drop guard, the empty-ALPN fallback,
  the empty-segment shape check (672 tests, 3 new).
- CHANGELOG [Unreleased] entry for the audit fixes.
- docs/reviews/009 — the audit's deferred test-coverage gaps
  (template failure arms, filtered/only, batch reserved key,
  wire failure path, golden pins, derivation edge shapes, builder
  overwrite semantics), each with the test to add and gates.

Verification: cargo test 672 passed; clippy --all-targets -D
warnings clean; fmt --check clean; doc --no-deps clean; wasm32
check clean.

src/channels/operations.rs

@@ -789,12 +789,12 @@ fn establishment_timeout_call_error() -> CallError {
 fn reserved_reply_key_call_error() -> CallError {
     CallError::new(
         CHANNEL_OPEN_FAILED,
-        "establisher supplied the reserved reply key `channel_id`",
+        format!("establisher supplied the reserved reply key `{RESERVED_REPLY_KEY}`"),
         false,
     )
     .with_details(json!({
         "reason": "handler_error",
-        "message": "establisher supplied the reserved reply key `channel_id`",
+        "message": format!("establisher supplied the reserved reply key `{RESERVED_REPLY_KEY}`"),
     }))
 }
 
@@ -929,7 +929,7 @@ async fn run_open_wrapper(
                                 .as_ref()
                                 .is_some_and(|m| m.contains_key(RESERVED_REPLY_KEY))
                             {
-                                tracing::debug!(
+                                tracing::warn!(
                                     channel_id = id,
                                     "open wrapper: establisher supplied the reserved reply key; tearing down channel"
                                 );

src/channels/relay.rs

@@ -66,10 +66,27 @@ pub struct ProducerLeg {
 /// establisher and the handler agree on this concrete type (the
 /// `ChannelPlan` downcast contract, ADR-049 amendment 2); alkcall
 /// never inspects it.
+///
+/// **Adopted-entry ownership (ADR-051 §6):** the plan holds the
+/// producer-leg `ChannelManager` and reclaims the adopted
+/// `spoke_id` entry on drop — including the teardown windows the
+/// pump handler never reaches (the wrapper's establishment bound
+/// expiring after the adopt, the plan dropped before the pump
+/// spawns). Dropping the plan's `spoke_send` also EOFs the spoke
+/// side via the mux pump's implicit-EOF path (REQ-CH-01), so the
+/// spoke handler reclaims through the same cascade the normal path
+/// uses.
 struct RelayPlan {
     spoke_id: u32,
-    spoke_send: MpscSendStream,
-    spoke_recv: MpscRecvStream,
+    producer_manager: ChannelManager,
+    spoke_send: Option<MpscSendStream>,
+    spoke_recv: Option<MpscRecvStream>,
+}
+
+impl Drop for RelayPlan {
+    fn drop(&mut self) {
+        let _ = self.producer_manager.teardown_channel(self.spoke_id);
+    }
 }
 
 /// The relay component (ADR-051). Holds the producer-leg surface;
@@ -141,7 +158,6 @@ impl ChannelRelay {
         let producer_call = Arc::clone(&self.producer_leg.call);
         let producer_manager = self.producer_leg.manager.clone();
         let op_name = spec.name.clone();
-        let handler_manager = producer_manager.clone();
         let establisher: OpenEstablisher = Arc::new(move |input, per_call_auth| {
             let producer_call = Arc::clone(&producer_call);
             let producer_manager = producer_manager.clone();
@@ -159,7 +175,6 @@ impl ChannelRelay {
         });
 
         let open_handler: OpenHandler = Arc::new(move |_input, plan, channel_conn, _auth| {
-            let producer_manager = handler_manager.clone();
             tokio::spawn(async move {
                 let Some(plan) = plan else {
                     return;
@@ -168,34 +183,36 @@ impl ChannelRelay {
                 // 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.
+                // when the plan finally drops (its Drop guard reclaims
+                // the adopted entry whenever that drop happens).
                 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 {
+                let Ok(mut 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;
                 };
+                // Take the streams out of the plan; the plan itself
+                // stays alive as the adopted-entry drop guard (see
+                // `RelayPlan`) until this task's end.
+                let Some(spoke_recv) = relay_plan.spoke_recv.take() else {
+                    return;
+                };
+                let Some(spoke_send) = relay_plan.spoke_send.take() 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. The
-                // adopted spoke channel entry is reclaimed here so the
-                // producer-leg manager holds no stale routing state
-                // after the relayed channel is done (the consumer-leg
-                // wrapper's teardown cannot see it — it is on the
-                // producer leg's manager).
-                let RelayPlan {
-                    spoke_id,
-                    spoke_send,
-                    spoke_recv,
-                } = relay_plan;
+                // pump's send-half drops EOF the spoke leg (the mux
+                // pump's implicit-EOF sentinel) and the plan's Drop
+                // guard reclaims the adopted producer-leg entry; the
+                // wrapper's handler-exit teardown covers the consumer
+                // leg.
                 let _ = pump_bidi(bidi, spoke_recv, spoke_send).await;
-                let _ = producer_manager.teardown_channel(spoke_id);
             })
         });
 
@@ -268,8 +285,9 @@ async fn open_on_producer_leg(
 
     let establishment = Establishment::new(Arc::new(RelayPlan {
         spoke_id,
-        spoke_send,
-        spoke_recv,
+        producer_manager: producer_manager.clone(),
+        spoke_send: Some(spoke_send),
+        spoke_recv: Some(spoke_recv),
     }) as super::operations::ChannelPlan);
     Ok(if reply_fields.is_empty() {
         establishment

src/channels/relay_tests.rs

@@ -203,6 +203,48 @@ fn relay_payload_omits_forwarded_for_when_identity_none() {
 
 // --- helpers --------------------------------------------------------------
 
+/// The `RelayPlan` drop guard (ADR-051 §6, the adopted-entry teardown
+/// bullet): dropping the plan — via whatever path — reclaims the
+/// adopted producer-leg entry. Covers the windows the pump handler's
+/// normal-path reclaim cannot reach: the wrapper's establishment bound
+/// expiring after the adopt (the plan dropped before the pump spawns),
+/// and the pump handler's early-return arms (plan held elsewhere —
+/// `try_unwrap` failure — where the pump never runs).
+#[tokio::test]
+async fn relay_plan_drop_reclaims_the_adopted_producer_leg_entry() {
+    let producer_manager = manager_for_test();
+
+    // Adopt a channel the way the relay establisher does.
+    let (spoke_send, spoke_recv) = producer_manager
+        .adopt_channel(4, "alk/relay", None)
+        .await
+        .expect("adopt the spoke channel");
+    assert_eq!(data_channels_len(&producer_manager), 1);
+
+    // Wrap it in a plan and drop the plan without any pump ever
+    // running (the establishment-timeout-after-adopt shape).
+    let plan = RelayPlan {
+        spoke_id: 4,
+        producer_manager: producer_manager.clone(),
+        spoke_send: Some(spoke_send),
+        spoke_recv: Some(spoke_recv),
+    };
+    drop(plan);
+
+    assert!(
+        producer_manager.channel_ids().into_iter().all(|id| id == 0),
+        "the adopted producer-leg entry must reclaim on the plan's drop"
+    );
+}
+
+fn data_channels_len(manager: &ChannelManager) -> usize {
+    manager
+        .channel_ids()
+        .into_iter()
+        .filter(|id| *id != 0)
+        .count()
+}
+
 fn overlay_connection() -> Arc<CallConnection> {
     Arc::new(CallConnection::new_overlay_only(identity("spoke", &[])))
 }