feat(cf-005): connect-side serving identity — ServingConfig.identity + transport-identity propagation

Verifies and fixes CF-005 (alktunnels reverse-flow POC W1): the
connect-side serving path built channel 0 internally and never set an
identity, so a scope-gated serving op could only be satisfied via the
payload auth_token. Token is now the fallback (hub-forwarding /
browser path); transport/key-based identity is the primary path.

- ServingConfig gains identity: Option<Identity> — the explicit
  override (remediation a). Semver-relevant struct-literal change →
  0.7.0 (minor bump at 0.x, wire surface unchanged).
- from_connection_with_serving propagates the transport
  Connection::identity() to the channel-0 connection via set_identity
  before the serving loop starts (remediation b) — mirrors the accept
  side's install-hook set_identity; process-local, nothing new on the
  wire.
- Dispatch identity precedence on the serving loop: payload
  auth_token → identity_provider, then ServingConfig.identity, then
  transport identity; identity-less dispatch still fails closed
  (FORBIDDEN).
- Public core::auth::NoopIdentityProvider (resolves nothing; the
  ServingConfig::default() provider — three private test copies
  existed).
- Regression gates: four cf005_* e2e tests (transport propagation,
  override precedence, identity-less denial, token fallback +
  precedence).
- Ledger CF-005 → resolved; ADR-022 §connect-side-serving amended;
  README example updated; changelog 0.7.0.

Verification: cargo test (629) + --all-features (646), clippy
(all-targets, all-features, -D warnings), fmt --check, doc
--no-deps, wasm32 check, semver-checks (no update required at
0.7.0), publish --dry-run.

src/channels/client.rs

@@ -111,9 +111,35 @@ pub struct ChannelClient {
 /// loop (`Dispatcher::serve_single_stream`), so inbound
 /// `call.requested` frames from the peer dispatch and resolve instead
 /// of being dropped.
+///
+/// The caller identity for dispatch (CF-005): when the serving loop
+/// dispatches an inbound `call.requested`, `Dispatcher::resolve_identity`
+/// consults, in precedence order — (1) the payload `auth_token`
+/// resolved through `identity_provider` (the hub-forwarding /
+/// browser-token path), (2) `identity` if set (an explicit override),
+/// (3) the transport connection's identity, if the consumer called
+/// `Connection::set_identity` after dialing (the mTLS/QUIC key-based
+/// path). When none is available the dispatch runs identity-less and
+/// `AccessControl::check` fails closed (`FORBIDDEN`).
 pub struct ServingConfig {
     pub registry: Arc<OperationRegistry>,
     pub identity_provider: Arc<dyn crate::core::auth::IdentityProvider>,
+    /// An explicit caller identity for the serving dispatch (the
+    /// connect-side seam, CF-005 remediation (a)). `None` (the
+    /// default) falls through to the transport connection's identity
+    /// (CF-005 remediation (b) — propagated automatically when the
+    /// consumer set one before calling `from_connection_with_serving`).
+    pub identity: Option<crate::core::auth::Identity>,
+}
+
+impl Default for ServingConfig {
+    fn default() -> Self {
+        Self {
+            registry: Arc::new(OperationRegistry::new()),
+            identity_provider: Arc::new(crate::core::auth::NoopIdentityProvider),
+            identity: None,
+        }
+    }
 }
 
 impl ChannelClient {
@@ -141,6 +167,16 @@ impl ChannelClient {
     /// and outbound pendings still resolve. This is the opt-in that
     /// makes the connect side a serving half on channel 0 (ADR-022 §2
     /// — both sides can be both).
+    ///
+    /// Caller identity for the serving dispatch (CF-005): the
+    /// `ServingConfig.identity` override wins; otherwise the transport
+    /// connection's identity (`Connection::identity`, set by the
+    /// consumer after dialing from the transport-authenticated peer)
+    /// propagates to channel 0. The payload `auth_token` →
+    /// `ServingConfig.identity_provider` path still takes precedence
+    /// over both (ADR-017 §7). With no identity from any path the
+    /// dispatch runs identity-less and scope-gated ops fail closed
+    /// (`FORBIDDEN`).
     pub async fn from_connection_with_serving(
         connection: Connection,
         serving: Option<ServingConfig>,
@@ -178,6 +214,22 @@ impl ChannelClient {
         let channel0_source =
             super::source::channel_source(channel0_recv, channel0_send, remote_addr);
         let channel0_conn = Connection::from_source(channel0_source, b"alk/call".to_vec());
+        // The caller-identity seam (CF-005): the serving loop's dispatch
+        // resolves the peer identity from this connection
+        // (`dispatch_start` → `connection.identity()`), so it must carry
+        // the identity the serving ops should authenticate by. The
+        // explicit `ServingConfig.identity` wins; otherwise the
+        // transport connection's identity propagates (the
+        // mTLS/QUIC key-based path — the peer the transport already
+        // authenticated). `set_identity` is once-only; silently
+        // skip if somehow already set.
+        let identity_for_serving = serving
+            .as_ref()
+            .and_then(|config| config.identity.clone())
+            .or_else(|| connection.identity().cloned());
+        if let Some(identity) = identity_for_serving {
+            let _ = channel0_conn.set_identity(identity);
+        }
         let channel0_bidi = channel0_conn.accept_bi().await?;
         let (single_stream_writer, single_stream_reader) =
             crate::protocol::connection::split_single_stream(channel0_bidi);
@@ -1675,6 +1727,7 @@ mod tests {
             Some(ServingConfig {
                 registry: Arc::clone(&consumer_registry),
                 identity_provider: Arc::new(NoopIdProvider),
+                identity: None,
             }),
         )
         .await
@@ -1809,6 +1862,7 @@ mod tests {
             Some(ServingConfig {
                 registry: Arc::clone(&consumer_registry),
                 identity_provider: Arc::new(NoopIdProvider),
+                identity: None,
             }),
         )
         .await
@@ -2023,6 +2077,7 @@ mod tests {
             Some(ServingConfig {
                 registry: Arc::clone(&consumer_registry),
                 identity_provider: Arc::new(NoopIdProvider),
+                identity: None,
             }),
         )
         .await
@@ -2240,6 +2295,7 @@ mod tests {
             Some(ServingConfig {
                 registry: Arc::clone(&consumer_registry),
                 identity_provider: Arc::new(NoopIdProvider),
+                identity: None,
             }),
         )
         .await
@@ -2495,4 +2551,360 @@ mod tests {
             schema.result
         );
     }
+
+    // --- CF-005 regression gates -----------------------------------------
+
+    /// The scope-gated open op the connect side serves — the
+    /// alktunnels reverse-flow POC shape (`worker/tunnel/open` with
+    /// `required_scopes`). The handler reports the identity the
+    /// dispatch resolved, so the tests can assert which path won.
+    fn scope_gated_open_registry() -> crate::registry::registration::OperationRegistry {
+        let registry = crate::registry::registration::OperationRegistry::new();
+        let acl = AccessControl {
+            required_scopes: vec!["tunnel:open".to_string()],
+            ..Default::default()
+        };
+        registry
+            .register(HandlerRegistration::new(
+                OperationSpec::new(
+                    "worker/tunnel/open",
+                    OperationType::Mutation,
+                    Visibility::External,
+                    serde_json::json!({}),
+                    serde_json::json!({}),
+                    vec![],
+                    acl,
+                    None,
+                ),
+                HandlerKind::Once(make_handler(|_input, ctx| async move {
+                    ResponseEnvelope::ok(
+                        ctx.request_id,
+                        serde_json::json!({
+                            "opened": true,
+                            "caller": ctx.identity.map(|i| i.id),
+                        }),
+                    )
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                crate::core::types::Capabilities::new(),
+            ))
+            .unwrap();
+        registry
+    }
+
+    /// The accept-side (hub) install hook: serves nothing but echoes
+    /// nothing — it hands the channel-0 `CallConnection` back to the
+    /// test and runs the resolution-only serving loop, so the test can
+    /// initiate hub→worker calls.
+    fn cf005_accept_side_hook(
+        accept_conn_tx: tokio::sync::mpsc::Sender<Arc<CallConnection>>,
+    ) -> crate::channels::adapter::InstallChannelZero {
+        Arc::new(move |_manager, channel0_conn, _auth| {
+            let accept_conn_tx = accept_conn_tx.clone();
+            tokio::spawn(async move {
+                let channel0_bidi = match channel0_conn.accept_bi().await {
+                    Ok(s) => s,
+                    Err(_) => return,
+                };
+                let (writer, reader) = split_single_stream(channel0_bidi);
+                let call_connection = Arc::new(CallConnection::new_single_stream(
+                    channel0_conn,
+                    Arc::clone(&writer),
+                ));
+                let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;
+                let reader = reader;
+                crate::protocol::connection::read_single_stream_until_closed(
+                    reader,
+                    call_connection.pending(),
+                )
+                .await;
+            })
+        })
+    }
+
+    fn hub_identity() -> crate::core::auth::Identity {
+        crate::core::auth::Identity {
+            id: "hub".to_string(),
+            scopes: vec!["tunnel:open".to_string()],
+            resources: Default::default(),
+        }
+    }
+
+    fn effective_override_identity() -> crate::core::auth::Identity {
+        crate::core::auth::Identity {
+            id: "worker-effective".to_string(),
+            scopes: vec!["tunnel:open".to_string()],
+            resources: Default::default(),
+        }
+    }
+
+    /// CF-005 remediation (b) — the transport connection's identity
+    /// propagates to channel 0, so a scope-gated op served by the
+    /// connect side authenticates the peer by transport identity (the
+    /// mTLS/QUIC key-based path). The consumer dials, sets the
+    /// transport identity (the assembly layer resolves it from the
+    /// transport handshake), then `from_connection_with_serving`; the
+    /// hub calls the scope-gated op with no token and is authorized.
+    #[tokio::test]
+    async fn cf005_transport_identity_propagates_to_connect_side_serving() {
+        let worker_registry = Arc::new(scope_gated_open_registry());
+        let (accept_conn_tx, mut accept_conn_rx) =
+            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
+        let install_hook = cf005_accept_side_hook(accept_conn_tx);
+
+        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));
+        client_conn
+            .set_identity(hub_identity())
+            .expect("transport identity set once");
+        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_with_serving(
+            client_conn,
+            Some(ServingConfig {
+                registry: Arc::clone(&worker_registry),
+                identity_provider: Arc::new(NoopIdProvider),
+                identity: None,
+            }),
+        )
+        .await
+        .expect("channel client init");
+
+        let accept_conn = accept_conn_rx
+            .recv()
+            .await
+            .expect("accept side channel-0 connection handle");
+
+        let response = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
+        )
+        .await
+        .expect("hub→worker scope-gated call timed out");
+
+        let out = response
+            .result
+            .expect("scope-gated op authorized by the propagated transport identity");
+        assert_eq!(out["opened"], serde_json::json!(true));
+        assert_eq!(
+            out["caller"], "hub",
+            "the handler saw the transport-derived identity"
+        );
+    }
+
+    /// CF-005 remediation (a) — `ServingConfig.identity` explicitly
+    /// overrides, winning over the transport connection's identity.
+    /// Both are set here; the handler must see the override.
+    #[tokio::test]
+    async fn cf005_serving_config_identity_overrides_transport_identity() {
+        let worker_registry = Arc::new(scope_gated_open_registry());
+        let (accept_conn_tx, mut accept_conn_rx) =
+            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
+        let install_hook = cf005_accept_side_hook(accept_conn_tx);
+
+        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));
+        client_conn
+            .set_identity(hub_identity())
+            .expect("transport identity set once");
+        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_with_serving(
+            client_conn,
+            Some(ServingConfig {
+                registry: Arc::clone(&worker_registry),
+                identity_provider: Arc::new(NoopIdProvider),
+                identity: Some(effective_override_identity()),
+            }),
+        )
+        .await
+        .expect("channel client init");
+
+        let accept_conn = accept_conn_rx
+            .recv()
+            .await
+            .expect("accept side channel-0 connection handle");
+
+        let response = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
+        )
+        .await
+        .expect("hub→worker scope-gated call timed out");
+
+        let out = response
+            .result
+            .expect("the override identity satisfies the scope gate");
+        assert_eq!(out["caller"], "worker-effective", "the override won");
+    }
+
+    /// CF-005 negative case — no identity anywhere (no
+    /// `ServingConfig.identity`, no transport identity, no token): the
+    /// scope-gated op is denied `FORBIDDEN` ("authentication
+    /// required"). Fails closed, unchanged.
+    #[tokio::test]
+    async fn cf005_scope_gated_op_denied_without_any_identity() {
+        let worker_registry = Arc::new(scope_gated_open_registry());
+        let (accept_conn_tx, mut accept_conn_rx) =
+            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
+        let install_hook = cf005_accept_side_hook(accept_conn_tx);
+
+        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_with_serving(
+            client_conn,
+            Some(ServingConfig {
+                registry: Arc::clone(&worker_registry),
+                identity_provider: Arc::new(NoopIdProvider),
+                identity: None,
+            }),
+        )
+        .await
+        .expect("channel client init");
+
+        let accept_conn = accept_conn_rx
+            .recv()
+            .await
+            .expect("accept side channel-0 connection handle");
+
+        let response = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
+        )
+        .await
+        .expect("hub→worker scope-gated call timed out");
+
+        let err = response
+            .result
+            .expect_err("identity-less call must be denied");
+        assert_eq!(err.code, "FORBIDDEN", "no identity on the connect side");
+        assert!(
+            err.message.contains("authentication required"),
+            "AccessControl::check(None) message, got: {}",
+            err.message
+        );
+    }
+
+    /// CF-005 token fallback — the payload `auth_token` →
+    /// `ServingConfig.identity_provider` path still works (the
+    /// hub-forwarding / browser-token path, ADR-017 §7), with
+    /// precedence over the propagated transport identity.
+    #[tokio::test]
+    async fn cf005_auth_token_fallback_and_precedence() {
+        struct TokenIdProvider;
+        impl IdentityProvider for TokenIdProvider {
+            fn resolve_from_fingerprint(&self, _: &str) -> Option<crate::core::auth::Identity> {
+                None
+            }
+            fn resolve_from_token(
+                &self,
+                token: &crate::core::auth::AuthToken,
+            ) -> Option<crate::core::auth::Identity> {
+                if token.raw == b"alk_worker_token" {
+                    Some(crate::core::auth::Identity {
+                        id: "token-resolved".to_string(),
+                        scopes: vec!["tunnel:open".to_string()],
+                        resources: Default::default(),
+                    })
+                } else {
+                    None
+                }
+            }
+        }
+
+        let worker_registry = Arc::new(scope_gated_open_registry());
+        let (accept_conn_tx, mut accept_conn_rx) =
+            tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
+        let install_hook = cf005_accept_side_hook(accept_conn_tx);
+
+        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));
+        client_conn
+            .set_identity(hub_identity())
+            .expect("transport identity set once");
+        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_with_serving(
+            client_conn,
+            Some(ServingConfig {
+                registry: Arc::clone(&worker_registry),
+                identity_provider: Arc::new(TokenIdProvider),
+                identity: None,
+            }),
+        )
+        .await
+        .expect("channel client init");
+
+        let accept_conn = accept_conn_rx
+            .recv()
+            .await
+            .expect("accept side channel-0 connection handle");
+
+        // Tokenless: the propagated transport identity authorizes.
+        let transport = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            accept_conn.call("worker/tunnel/open", serde_json::json!({ "c": 1 })),
+        )
+        .await
+        .expect("transport-identity call timed out");
+        let out = transport
+            .result
+            .expect("transport identity authorizes the scope gate");
+        assert_eq!(out["caller"], "hub");
+
+        // With a token: the token resolution wins.
+        let tokened = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            accept_conn.call_with_payload(serde_json::json!({
+                "operationId": "worker/tunnel/open",
+                "input": { "c": 2 },
+                "auth_token": "alk_worker_token",
+            })),
+        )
+        .await
+        .expect("token call timed out");
+        let out = tokened
+            .result
+            .expect("the token resolves an identity satisfying the scope gate");
+        assert_eq!(
+            out["caller"], "token-resolved",
+            "the token resolution won over the transport identity"
+        );
+    }
 }