test(review 009): all seven coverage findings; errata on three as-filed claims

Review 009's coverage debt in full — the paths the review-008 gates
never walk. No wire or API changes.

- C-1: pin the plain-bundle install-failure arm (un-compilable
  input_schema; the as-filed duplicate-name route does not fail
  registration) — the install task ends before the dispatch loop,
  channel 0 never dispatches. ADR-051 §5 gains the loud-install
  coverage note: relay-openable + plain-bundle arms pinned,
  generic-ops + bootstrap-discovery arms documented as
  best-effort-loud (crate-internal specs compile by construction).
- C-2: the HubLegImports filter — filtered closure path + only
  partition unit-tested (errata: only was already pinned at filing);
  the empty-stash e2e gate (generic ops + discovery only, dropped
  ops resolve NOT_FOUND).
- C-3: the batch-form reserved-reply-key rejection pinned (reason
  handler_error, teardown, ledger decrement, no pump spawn).
- C-4: open_channel_with_reply's failure path pinned e2e (the typed
  error carries the channel:open_failed code + details reason/message).
- C-5: both byte-identical claims golden-pinned — the no-fields reply
  against {"channel_id": 2} and the standard-shape wire payload
  against the full 9-key literal.
- C-6: derivation edge shapes pinned — channels//sub, channels//direct,
  channels → None; the 4-segment strict superset annotated as the
  pre-amendment behavior change (errata: actual is Some("x/sub"), the
  multi-segment-ALPN rule, not the as-filed Some("alk/x/sub")).
- C-7: builder overwrite pinned last-win (single + batch) with the
  doc sentence on with_reply_field.

Verification: 682 tests pass, clippy -D warnings clean, fmt clean,
doc clean, wasm32 check clean.

File: docs/reviews/009 (resolved; errata marked per finding)

src/channels/client.rs

@@ -1767,6 +1767,131 @@ mod tests {
         assert!(channel_id2 > 0);
     }
 
+    /// Review 009 C-4: the new pub API's failure path — a
+    /// `channel:open_failed` resolving through
+    /// `open_channel_with_reply` carries the full wire error shape
+    /// (`code` + typed `details` with `reason` and `message`), so the
+    /// caller can branch on the establishment reason through the new
+    /// API the same way `open_channel`'s shared parse path does.
+    #[tokio::test]
+    async fn open_channel_with_reply_resolves_typed_open_failed_end_to_end() {
+        use crate::channels::operations::{
+            ChannelCore, EstablishmentError, OpenEstablisher, OpenHandler,
+        };
+        use crate::channels::policy::NoCap;
+        use crate::registry::spec::ChannelOpenSpec;
+
+        let open_handler: OpenHandler =
+            Arc::new(|_input, _plan, _channel_conn, _auth| tokio::spawn(async {}));
+        let establisher: OpenEstablisher = Arc::new(|_input, _auth| {
+            Box::pin(async {
+                Err(EstablishmentError::DialFailed {
+                    message: "target refused the connection".to_string(),
+                })
+            })
+        });
+
+        let establisher_for_hook = Arc::clone(&establisher);
+        let open_handler_for_hook = Arc::clone(&open_handler);
+        let install_hook: crate::channels::adapter::InstallChannelZero =
+            Arc::new(move |manager, channel0_conn, auth| {
+                let establisher = Arc::clone(&establisher_for_hook);
+                let open_handler = Arc::clone(&open_handler_for_hook);
+                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 core = ChannelCore::new(manager, Arc::new(NoCap));
+                    let registry = crate::registry::registration::OperationRegistry::new();
+                    let spec = OperationSpec::new(
+                        "channels/tty/sub",
+                        OperationType::Sub,
+                        Visibility::External,
+                        serde_json::json!({}),
+                        serde_json::json!({
+                            "type": "object",
+                            "properties": { "channel_id": { "type": "integer" } }
+                        }),
+                        vec![],
+                        AccessControl::default(),
+                        None,
+                    )
+                    .with_channel_open(ChannelOpenSpec::new("alk/tty"));
+                    core.register_openable_with_establisher(
+                        spec,
+                        Some(establisher),
+                        open_handler,
+                        &registry,
+                        auth.clone(),
+                        None,
+                    )
+                    .expect("register_openable_with_establisher");
+                    let registry = Arc::new(registry);
+                    let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
+                    let call_connection = Arc::new(CallConnection::new_single_stream(
+                        channel0_conn,
+                        Arc::clone(&writer),
+                    ));
+                    let dp = Dispatcher::new(registry, provider);
+                    dp.run_loop_single_stream(call_connection, reader, writer)
+                        .await;
+                })
+            });
+
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let client_conn =
+            Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
+        let server_conn =
+            Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
+
+        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
+        let auth = AuthContext::anonymous(b"alk/channels");
+        let _server_handle = tokio::spawn(async move {
+            let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
+        });
+
+        let client = ChannelClient::from_connection(client_conn)
+            .await
+            .expect("channel client init");
+
+        let err = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            client.open_channel_with_reply(
+                "channels/tty/sub",
+                serde_json::json!({ "container": "abc" }),
+                "alk/tty",
+            ),
+        )
+        .await
+        .expect("open_channel_with_reply timed out");
+        let err = match err {
+            Err(e) => e,
+            Ok(_) => panic!("the establisher failure must resolve Err through the reply API"),
+        };
+
+        // The full reply shape is assertable by the caller: the wire
+        // code, the reason code, and the message inside `details`.
+        let call_error = err.call_error().expect("CallFailed carries the CallError");
+        assert_eq!(call_error.code, "channel:open_failed");
+        assert_eq!(err.establishment_reason(), Some("dial_failed"));
+        let details = call_error
+            .details
+            .as_ref()
+            .expect("details ride the typed error");
+        assert_eq!(details["reason"], serde_json::json!("dial_failed"));
+        assert_eq!(
+            details["message"],
+            serde_json::json!("target refused the connection"),
+            "the establishment-failure message survives the wire to the caller"
+        );
+        assert!(
+            client.manager().channel_ids().into_iter().all(|id| id == 0),
+            "no data channel exists for the failed open"
+        );
+    }
+
     // --- review 004 Unit 3 acceptance gates (F-04 serving half) -----------
 
     /// F-04 gate 1: hub→consumer call over an existing `ChannelClient`

src/channels/hub_leg_tests.rs

@@ -152,6 +152,78 @@ fn stash_filter_keeps_named_ops_only() {
     assert_eq!(filtered.plain()[0].spec.name, "b/plain");
 }
 
+/// Review 009 C-2: the closure-shaped filter partitions both halves.
+/// `only(&[OP])` keeps the marked direct spec and drops every other
+/// op; the plain bundle is untouched by the marked list (and vice
+/// versa) — the filter runs on the spec name in both halves.
+#[test]
+fn stash_only_keeps_the_marked_direct_spec_and_drops_the_rest() {
+    let bundle = |name: &str, marked: bool| {
+        let spec = if marked {
+            relay_spec(name, OperationType::Sub, Some(ALPN.to_string()))
+        } else {
+            relay_spec(name, OperationType::Query, None)
+        };
+        HandlerRegistration::new(
+            spec,
+            HandlerKind::Once(make_handler(|input, ctx| async move {
+                ResponseEnvelope::ok(ctx.request_id, input)
+            })),
+            OperationProvenance::FromCall,
+            None,
+            None,
+            crate::core::types::Capabilities::new(),
+        )
+    };
+    let imports = HubLegImports::from_bundles(vec![
+        bundle(OP, true),
+        bundle("other/marked", true),
+        bundle(PLAIN_OP, false),
+        bundle("other/plain", false),
+    ]);
+    let filtered = imports.only(&[OP]);
+    assert_eq!(filtered.marked().len(), 1, "only the named marked op stays");
+    assert_eq!(filtered.marked()[0].spec.name, OP);
+    assert!(
+        filtered.plain().is_empty(),
+        "the plain bundle is untouched by the marked op's name"
+    );
+}
+
+/// Review 009 C-2: the `filtered` closure path — the per-consumer
+/// filter composes by spec-name predicate over both halves.
+#[test]
+fn stash_filtered_closure_partitions_both_halves() {
+    let bundle = |name: &str, marked: bool| {
+        let spec = if marked {
+            relay_spec(name, OperationType::Sub, Some(ALPN.to_string()))
+        } else {
+            relay_spec(name, OperationType::Query, None)
+        };
+        HandlerRegistration::new(
+            spec,
+            HandlerKind::Once(make_handler(|input, ctx| async move {
+                ResponseEnvelope::ok(ctx.request_id, input)
+            })),
+            OperationProvenance::FromCall,
+            None,
+            None,
+            crate::core::types::Capabilities::new(),
+        )
+    };
+    let imports = HubLegImports::from_bundles(vec![
+        bundle("tunnel-marked", true),
+        bundle("other-marked", true),
+        bundle("tunnel-plain", false),
+        bundle("other-plain", false),
+    ]);
+    let filtered = imports.filtered(|name| name.starts_with("tunnel"));
+    assert_eq!(filtered.marked().len(), 1);
+    assert_eq!(filtered.marked()[0].spec.name, "tunnel-marked");
+    assert_eq!(filtered.plain().len(), 1);
+    assert_eq!(filtered.plain()[0].spec.name, "tunnel-plain");
+}
+
 // --- the spoke (producer leg) serving the direct + plain ops --------------
 
 fn spoke_open_handler(data_tx: tokio::sync::mpsc::Sender<String>) -> OpenHandler {
@@ -498,6 +570,99 @@ async fn template_scope_less_identity_fails_scope_gated_op() {
 
 // --- gate: the loud assembly posture (ADR-051 §6) -------------------------
 
+/// Review 009 C-2 gate: an empty stash (`only(&[])` — the filter
+/// matched nothing) installs a leg that serves only the generic channel
+/// ops + bootstrap discovery: no re-exposed op appears in
+/// `services/list` and the dropped ops resolve NOT_FOUND.
+#[tokio::test]
+async fn template_empty_only_stash_installs_generic_ops_and_discovery_only() {
+    let consumer = start_hub(None, Some(HubLegImports::default()), None, false).await;
+
+    let listing = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op("services/list", json!({})),
+    )
+    .await
+    .expect("services/list timed out");
+    let ops: Vec<String> = listing
+        .result
+        .expect("services/list ok")
+        .get("operations")
+        .and_then(|v| v.as_array())
+        .expect("operations array")
+        .iter()
+        .filter_map(|o| o.get("name").and_then(|n| n.as_str()).map(String::from))
+        .collect();
+    assert!(
+        !ops.iter().any(|n| n == OP || n == PLAIN_OP),
+        "no re-exposed op is listed on an empty stash: {ops:?}"
+    );
+    assert!(
+        ops.iter().any(|n| n == "channel/close"),
+        "the generic channel ops are still registered: {ops:?}"
+    );
+    assert!(
+        ops.iter().any(|n| n == "services/list"),
+        "the bootstrap discovery ops are still registered: {ops:?}"
+    );
+
+    // The filtered-out open op resolves NOT_FOUND (never a silent stub).
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        consumer.call_open_op(OP, json!({})),
+    )
+    .await
+    .expect("filtered-out open op timed out");
+    match response.result {
+        Err(e) => assert_eq!(e.code, "NOT_FOUND"),
+        Ok(_) => panic!("the empty stash must re-expose no op"),
+    }
+}
+
+// --- gate: the loud assembly posture (ADR-051 §6) -------------------------
+
+/// Review 009 C-1: a plain bundle whose registration fails (an
+/// un-compilable `input_schema` — the registry's fail-closed
+/// registration rule, CF-003) is the loud posture at install: the
+/// template's install task ends before `run_loop_single_stream`, so
+/// the consumer leg's channel 0 never dispatches. Never a silent stub.
+#[tokio::test]
+async fn template_plain_bundle_registration_failure_is_loud_at_install() {
+    let mut broken = relay_spec(PLAIN_OP, OperationType::Query, None);
+    broken.input_schema = json!({
+        "type": "object",
+        "required": "not-an-array"
+    });
+    let bundle = HandlerRegistration::new(
+        broken,
+        HandlerKind::Once(make_handler(|input, ctx| async move {
+            ResponseEnvelope::ok(ctx.request_id, input)
+        })),
+        OperationProvenance::FromCall,
+        None,
+        None,
+        crate::core::types::Capabilities::new(),
+    );
+    assert!(
+        OperationRegistry::new().register(bundle.clone()).is_err(),
+        "the broken bundle fails registration (the registry rejects it)"
+    );
+
+    let imports = HubLegImports::from_bundles(vec![bundle]);
+    let consumer = start_hub(None, Some(imports), None, false).await;
+    let response = tokio::time::timeout(
+        std::time::Duration::from_secs(3),
+        consumer.call_open_op("services/list", json!({})),
+    )
+    .await;
+    assert!(
+        response.is_err(),
+        "the leg's channel 0 never dispatches after the plain-bundle \
+         registration failure, got {:?}",
+        response
+    );
+}
+
 /// A Pub-typed marked spec in the stash is the loud assembly error at
 /// install: the consumer leg's channel 0 never dispatches (a client
 /// call to any op — including the bootstrap discovery ops — does not

src/channels/operations.rs

@@ -416,7 +416,9 @@ impl Establishment {
     /// success reply by the wrapper, after its `channel_id`
     /// reservation. Establisher-supplied `channel_id` is rejected —
     /// the check runs at merge time in the wrapper, not here, so the
-    /// failure carries the wire error path. Builder-style.
+    /// failure carries the wire error path. A repeated key silently
+    /// last-wins (the batch form [`Establishment::with_reply_fields`]
+    /// extends the same way). Builder-style.
     pub fn with_reply_field(mut self, key: impl Into<String>, value: Value) -> Self {
         self.reply_fields
             .get_or_insert_with(Map::new)
@@ -2602,6 +2604,41 @@ mod tests {
         }
     }
 
+    /// Review 009 C-5: the no-fields reply is golden-pinned, not just
+    /// shape-pinned — the exact literal (key set AND the value type),
+    /// not a self-referential compare. The accept-side manager
+    /// allocates even ids starting at 2 (ADR-047 §5), so the first
+    /// open on a fresh manager is exactly 2.
+    #[tokio::test]
+    async fn run_open_wrapper_no_fields_reply_golden_pin() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let open_handler: OpenHandler =
+            Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {}));
+        let auth = AuthContext::anonymous(b"alk/call");
+        let env = run_open_wrapper(
+            &manager,
+            &policy,
+            hook(ok_establisher(), None).as_ref(),
+            &open_handler,
+            &auth,
+            "alk/tty",
+            json!({}),
+            Some(identity("alice")),
+            "req-golden-no-fields".to_string(),
+            None,
+        )
+        .await;
+        match env.result {
+            Ok(v) => assert_eq!(
+                v,
+                json!({ "channel_id": 2 }),
+                "the no-fields reply is exactly the one-key literal (golden)"
+            ),
+            Err(e) => panic!("open should succeed, got: {e:?}"),
+        }
+    }
+
     #[tokio::test]
     async fn run_open_wrapper_establisher_channel_id_reply_key_fails_loudly() {
         let manager = make_manager().await;
@@ -2663,6 +2700,102 @@ mod tests {
         );
     }
 
+    /// Review 009 C-3: the batch form of the reserved-reply-key
+    /// rejection — `with_reply_fields(map)` smuggling `channel_id`
+    /// (the shape a hub's `Establishment::with_reply_fields(relay_map)`
+    /// would use) fails the open with the same loud path: reason
+    /// `handler_error`, channel torn down, ledger decremented. The
+    /// wrapper's check inspects the merged map, so the builder form
+    /// (the sibling test above) and the batch form are one check.
+    #[tokio::test]
+    async fn run_open_wrapper_batch_form_reserved_reply_key_fails_loudly() {
+        let manager = make_manager().await;
+        let concrete_policy = Arc::new(super::super::policy::PerIdentityChannelPolicy::new(8));
+        let policy: Arc<dyn ChannelLifecyclePolicy> =
+            Arc::clone(&concrete_policy) as Arc<dyn ChannelLifecyclePolicy>;
+        let spawned = Arc::new(AtomicBool::new(false));
+        let spawned_clone = Arc::clone(&spawned);
+        let open_handler: OpenHandler = Arc::new(move |_input, _plan, _conn, _auth| {
+            spawned_clone.store(true, Ordering::SeqCst);
+            tokio::spawn(async {})
+        });
+        let establishing: OpenEstablisher = Arc::new(|_input, _auth| {
+            Box::pin(async {
+                let mut map = Map::new();
+                map.insert("bound".to_string(), json!({ "port": 42113 }));
+                map.insert("channel_id".to_string(), json!(999_u64));
+                Ok(Establishment::default().with_reply_fields(map))
+            })
+        });
+        let auth = AuthContext::anonymous(b"alk/call");
+        let env = run_open_wrapper(
+            &manager,
+            &policy,
+            hook(establishing, None).as_ref(),
+            &open_handler,
+            &auth,
+            "alk/tty",
+            json!({}),
+            Some(identity("alice")),
+            "req-reserved-key-batch".to_string(),
+            None,
+        )
+        .await;
+        match env.result {
+            Err(e) => {
+                assert_eq!(e.code, "channel:open_failed");
+                let details = e.details.expect("details carry the reason");
+                assert_eq!(details["reason"], "handler_error");
+                assert!(
+                    details["message"]
+                        .as_str()
+                        .expect("message")
+                        .contains("channel_id"),
+                    "the message names the reserved key"
+                );
+            }
+            Ok(v) => panic!("batch-form reserved key must fail the open, got: {v:?}"),
+        }
+        assert!(
+            manager.channel_ids().is_empty(),
+            "the just-allocated channel was torn down"
+        );
+        assert_eq!(
+            concrete_policy.count_for(&identity("alice")),
+            0,
+            "ledger take + policy.on_close restored the cap count"
+        );
+        assert!(
+            !spawned.load(Ordering::SeqCst),
+            "pump handler must not spawn on the batch-form reserved-key rejection"
+        );
+    }
+
+    /// Review 009 C-7: the builder's overwrite semantics — a repeated
+    /// key silently last-wins (the batch form extends, also last-win).
+    /// Pub-API behavior pinned as intended; `with_reply_field`'s doc
+    /// sentence states it.
+    #[test]
+    fn establishment_builder_reply_fields_last_win_on_repeat() {
+        let e = Establishment::default()
+            .with_reply_field("k", json!("v1"))
+            .with_reply_field("k", json!("v2"));
+        assert_eq!(
+            e.reply_fields().expect("fields")["k"],
+            json!("v2"),
+            "a repeated key silently last-wins"
+        );
+        let batch: Map<String, Value> = [("k".to_string(), json!("v1"))].into_iter().collect();
+        let e = Establishment::default()
+            .with_reply_field("k", json!("v1"))
+            .with_reply_fields(batch);
+        assert_eq!(
+            e.reply_fields().expect("fields")["k"],
+            json!("v1"),
+            "the batch form's value wins when it comes later (plain extend)"
+        );
+    }
+
     #[tokio::test]
     async fn register_openable_with_establisher_replies_projected_fields_end_to_end() {
         let manager = make_manager().await;