feat(cf-006/cf-007): per-call opener identity for open-op hooks; ADR-016 code-list sweep

Batch-fixes every remaining open alkcall-side finding from downstream
consumers so 0.7.0 is the only release they need to absorb.

- CF-006 (the CF-005 corollary): run_open_wrapper derives a per-call
  AuthContext — the opener's dispatch-resolved identity (the same
  identity the ACL gate and cap check saw) overlaid onto the
  install-time context — and passes it to both the establisher and
  the pump handler. Identity-less calls keep the install-time
  identity (no synthetic-anonymous rewrite); transport-truthful
  fields are never rewritten. Signatures unchanged — behavior-only;
  identical on per-connection registries, hub-forwarded opens now
  show the end client. Gates:
  open_wrapper_overlays_per_call_identity_on_install_time_auth +
  open_wrapper_keeps_install_time_identity_when_call_identityless.
- CF-007 (alkhttp review 006 Part C doc drift): ADR-016 amended to
  the eight-code list — ALREADY_EXISTS + CONNECTION_CLOSED in the
  Context, §3 table, and from_openapi collision rule; new §2a
  documents the undelivered-vs-ambiguous write-failure distinction.
- ChannelPlan type doc now states the Send + Sync payload constraint
  (alktunnels POC F-1 re-derived it by compiler error).
- Ledger: CF-006 and CF-007 filed + resolved; CF-005's corollary
  note points at CF-006; the Open section is empty. ADR-049 gains
  the per-call-identity note. Changelog 0.7.0 covers the batch.

Sweep result (all downstream reviews): no other open alkcall items —
alktunnels W3/F-2 are downstream-by-design, alktty R3/P14 are closed
constraints, alknet has none; OQ-24/37/39/40/41 stay
deferred-by-design (no consumer pull yet).

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

src/channels/operations.rs

@@ -362,6 +362,15 @@ pub type OpenHandler = Arc<
 /// The plan is **process-local**: establisher → wrapper → handler on
 /// the producing side. Nothing crosses the transport that isn't
 /// already the open op's input — the wire surface is unchanged.
+///
+/// **Payload constraint (alktunnels POC F-1):** `Arc<dyn Any + Send +
+/// Sync>` requires the concrete plan payload to be `Send + Sync`. A
+/// dialed-socket handle with boxed non-`Sync` halves (e.g.
+/// `Box<dyn AsyncRead>` without a `Sync` bound) does not satisfy it —
+/// wrap such handles in a `Sync` adapter or hold them unboxed. The
+/// bound is correct (the plan crosses the wrapper task boundary);
+/// this note exists so the next consumer doesn't re-derive it by
+/// compiler error.
 pub type ChannelPlan = Arc<dyn std::any::Any + Send + Sync>;
 
 /// The establishment-phase result (ADR-049 §1, as filled by ADR-049
@@ -441,8 +450,13 @@ impl EstablishmentError {
 /// is spawned. On failure the wrapper tears down the just-allocated
 /// channel and replies `channel:open_failed` with `details.reason`.
 ///
-/// The establisher takes the open op's `input` (params) and the peer's
-/// `AuthContext`. It deliberately does **not** receive the channel's
+/// The establisher takes the open op's `input` (params) and the
+/// **per-call** `AuthContext`: the opener's resolved identity (the
+/// same identity the ACL gate and the per-identity cap check saw —
+/// CF-005 corollary, 2026-09-07) overlaid onto the install-time
+/// context; the transport-truthful fields (`alpn`, `remote_addr`,
+/// `tls_client_fingerprint`) are install-time. It deliberately does
+/// **not** receive the channel's
 /// `Connection`: the channel `BiStream` is yield-once
 /// (`ChannelBidiStreamSource`) and belongs exclusively to the pump
 /// handler (the `OpenHandler`); the establisher is pre-data-plane (its
@@ -524,9 +538,13 @@ impl ChannelCore {
     /// overlay registry, Layer 2 per ADR-019 — this is the
     /// per-connection registration the §4 amendment blesses). The
     /// `auth` is the peer's `AuthContext`, captured at
-    /// `install_channel_zero` time and closed over by the wrapper so
-    /// the ALPN handler receives it without the wrapper having to
-    /// reach into `OperationContext` for it.
+    /// `install_channel_zero` time and closed over by the wrapper: it
+    /// provides the transport-truthful fields (`alpn`,
+    /// `remote_addr`, `tls_client_fingerprint`), while the identity
+    /// field is overridden **per call** — the opener's resolved
+    /// identity (the same identity the ACL gate and the cap check
+    /// saw; CF-005 corollary, 2026-09-07), falling back to the
+    /// install-time identity when the call has none.
     ///
     /// **Op type:** the spec's `op_type` determines the
     /// `HandlerKind`. For `Query`/`Mutation` the wrapper is a `Once`
@@ -639,15 +657,15 @@ impl ChannelCore {
     }
 }
 
-/// The opener identity for the per-identity cap check. The direct
-/// caller (the peer that opened this channels connection) is the
-/// identity whose cap is consulted; `forwarded_for` is metadata and
-/// is NOT consulted (ADR-026, ADR-047 §7). Falls back to a synthetic
-/// "anonymous" identity when the call has no resolved identity (the
-/// cap still applies — anonymous opens count against the anonymous
-/// identity's cap, which is 256 by default).
-fn opener_identity_from_context(ctx: &OperationContext) -> Identity {
-    ctx.identity.clone().unwrap_or_else(|| Identity {
+/// The opener identity for the per-identity cap check (CF-005
+/// corollary): `Some` is the dispatch-resolved caller identity; `None`
+/// falls back to a synthetic "anonymous" identity. The direct caller
+/// (the peer that opened this channels connection) is the identity
+/// whose cap is consulted; `forwarded_for` is metadata and is NOT
+/// consulted (ADR-026, ADR-047 §7). Anonymous opens count against the
+/// anonymous identity's cap, which is 256 by default.
+fn opener_identity_or_anonymous(call_identity: Option<Identity>) -> Identity {
+    call_identity.unwrap_or_else(|| Identity {
         id: "anonymous".to_string(),
         scopes: vec![],
         resources: Default::default(),
@@ -743,9 +761,24 @@ fn establishment_timeout_call_error() -> CallError {
 ///
 /// `input` is the open op's input (params), passed through to the
 /// establisher (when registered) and the ALPN's `OpenHandler`.
-/// `opener_id` is the `PeerId` recorded in the opener ledger (the
-/// direct caller's identity id). `opener_identity` is the identity the
-/// cap was reserved against (un-reserved on failure paths).
+/// `call_identity` is the dispatch-resolved caller identity — `Some`
+/// when the call carried one (the same identity the ACL gate saw),
+/// `None` when identity-less. The cap is checked against it (or the
+/// synthetic `anonymous` when `None`) and it is overlaid onto the
+/// per-call auth when present.
+///
+/// The `auth` handed to the establisher and the pump handler is the
+/// **per-call** context (CF-005 corollary): the opener's resolved
+/// identity overlaid onto the install-time context. When the call has
+/// no resolved identity, the install-time identity is kept (the
+/// anonymous-keep case). Transport-truthful fields (`alpn`,
+/// `remote_addr`, `tls_client_fingerprint`) are never rewritten. In
+/// the per-connection-registry deployment (ADR-047 §4) the
+/// install-time identity IS the caller's, so this is
+/// behavior-identical there; the difference shows on hub-forwarded
+/// opens (the establisher sees the end client, not the hub) and on
+/// the connect-side serving path (token- or
+/// `ServingConfig.identity`-resolved callers).
 #[allow(clippy::too_many_arguments)]
 async fn run_open_wrapper(
     manager: &ChannelManager,
@@ -755,16 +788,24 @@ async fn run_open_wrapper(
     auth: &AuthContext,
     alpn: &str,
     input: Value,
-    opener_id: String,
-    opener_identity: Identity,
+    call_identity: Option<Identity>,
     request_id: String,
     deadline: Option<Instant>,
 ) -> ResponseEnvelope {
+    let mut per_call_auth = auth.clone();
+    if let Some(identity) = &call_identity {
+        per_call_auth.identity = Some(identity.clone());
+    }
+    let opener_identity = opener_identity_or_anonymous(call_identity);
+
     if let Err(channel_err) = policy.check_open(&opener_identity) {
         return ResponseEnvelope::error(request_id, map_channel_error_to_call_error(&channel_err));
     }
 
-    let channel_id = match manager.open_channel(alpn, opener_id, None).await {
+    let channel_id = match manager
+        .open_channel(alpn, opener_identity.id.clone(), None)
+        .await
+    {
         Ok((id, send, recv)) => {
             // The establishment phase (ADR-049 §1): awaited bounded,
             // before the reply and before the pump handler is spawned.
@@ -775,7 +816,7 @@ async fn run_open_wrapper(
                     let bound = establishment_bound(hook.timeout, deadline);
                     let establishment = tokio::time::timeout(
                         bound,
-                        (hook.establisher)(input.clone(), auth.clone()),
+                        (hook.establisher)(input.clone(), per_call_auth.clone()),
                     )
                     .await;
                     match establishment {
@@ -817,7 +858,7 @@ async fn run_open_wrapper(
                 Arc::clone(&accepted),
             );
             let channel_conn = Connection::from_source(source, alpn.as_bytes().to_vec());
-            let raw_task = open_handler(input, plan, channel_conn, auth.clone());
+            let raw_task = open_handler(input, plan, channel_conn, per_call_auth);
 
             let teardown_manager = manager.clone();
             let teardown_policy = Arc::clone(policy);
@@ -948,8 +989,6 @@ fn make_open_handler_once(
         let alpn = alpn.clone();
         Box::pin(async move {
             let request_id = ctx.request_id.clone();
-            let opener_identity = opener_identity_from_context(&ctx);
-            let opener_id = opener_identity.id.clone();
             run_open_wrapper(
                 &manager,
                 &policy,
@@ -958,8 +997,7 @@ fn make_open_handler_once(
                 &auth,
                 &alpn,
                 input,
-                opener_id,
-                opener_identity,
+                ctx.identity.clone(),
                 request_id,
                 ctx.deadline,
             )
@@ -993,8 +1031,6 @@ fn make_open_handler_stream(
         let alpn = alpn.clone();
         Box::pin(futures::stream::once(async move {
             let request_id = ctx.request_id.clone();
-            let opener_identity = opener_identity_from_context(&ctx);
-            let opener_id = opener_identity.id.clone();
             run_open_wrapper(
                 &manager,
                 &policy,
@@ -1003,8 +1039,7 @@ fn make_open_handler_stream(
                 &auth,
                 &alpn,
                 input,
-                opener_id,
-                opener_identity,
+                ctx.identity.clone(),
                 request_id,
                 ctx.deadline,
             )
@@ -1357,17 +1392,15 @@ mod tests {
     }
 
     #[test]
-    fn opener_identity_from_context_returns_identity_when_present() {
+    fn opener_identity_or_anonymous_returns_identity_when_present() {
         let ctx = test_context("opener-1");
-        let id = opener_identity_from_context(&ctx);
+        let id = opener_identity_or_anonymous(ctx.identity.clone());
         assert_eq!(id.id, "alice");
     }
 
     #[test]
-    fn opener_identity_from_context_falls_back_to_anonymous() {
-        let mut ctx = test_context("opener-2");
-        ctx.identity = None;
-        let id = opener_identity_from_context(&ctx);
+    fn opener_identity_or_anonymous_falls_back_to_anonymous() {
+        let id = opener_identity_or_anonymous(None);
         assert_eq!(id.id, "anonymous");
     }
 
@@ -1415,6 +1448,106 @@ mod tests {
         assert!(spawned.load(Ordering::SeqCst));
     }
 
+    /// CF-005 corollary — the establisher and the pump handler see the
+    /// **per-call** opener identity (the identity the ACL gate and cap
+    /// check saw), not the install-time snapshot. The install-time
+    /// context here carries a stale identity (`stale-install`); the
+    /// dispatch context resolves `alice`. Both hooks must report
+    /// `alice` while `alpn` stays the install-time transport value.
+    #[tokio::test]
+    async fn open_wrapper_overlays_per_call_identity_on_install_time_auth() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let seen_establisher = Arc::new(parking_lot::Mutex::new(None::<String>));
+        let seen_handler = Arc::new(parking_lot::Mutex::new(None::<String>));
+        let seen_establisher_cl = Arc::clone(&seen_establisher);
+        let seen_handler_cl = Arc::clone(&seen_handler);
+
+        let establisher: OpenEstablisher = Arc::new(move |_input, auth| {
+            let seen = Arc::clone(&seen_establisher_cl);
+            Box::pin(async move {
+                *seen.lock() = auth.identity.as_ref().map(|i| i.id.clone());
+                Ok::<_, EstablishmentError>(Establishment::default())
+            })
+        });
+        let open_handler: OpenHandler = Arc::new(move |_input, _plan, _conn, auth| {
+            *seen_handler_cl.lock() = auth.identity.as_ref().map(|i| i.id.clone());
+            tokio::spawn(async {})
+        });
+
+        let mut install_auth = AuthContext::anonymous(b"alk/tty");
+        install_auth.identity = Some(Identity {
+            id: "stale-install".to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        });
+        let handler = make_open_handler_once(
+            manager.clone(),
+            policy,
+            Some(EstablisherHook {
+                establisher,
+                timeout: None,
+            }),
+            open_handler,
+            install_auth,
+            "alk/tty".to_string(),
+        );
+        let env = handler(json!({}), test_context("open-per-call-1")).await;
+        env.result.expect("open should succeed");
+
+        assert_eq!(
+            seen_establisher.lock().as_deref(),
+            Some("alice"),
+            "establisher saw the per-call opener identity"
+        );
+        assert_eq!(
+            seen_handler.lock().as_deref(),
+            Some("alice"),
+            "pump handler saw the per-call opener identity"
+        );
+    }
+
+    /// CF-005 corollary (fallback leg) — when the call has no resolved
+    /// identity, the anonymous opener keeps the install-time identity
+    /// (no overlay). The install-time identity here is `bob`; the
+    /// handler must see `bob`, not the synthetic `anonymous`.
+    #[tokio::test]
+    async fn open_wrapper_keeps_install_time_identity_when_call_identityless() {
+        let manager = make_manager().await;
+        let policy = super::super::policy::default_policy();
+        let seen_handler = Arc::new(parking_lot::Mutex::new(None::<String>));
+        let seen_handler_cl = Arc::clone(&seen_handler);
+        let open_handler: OpenHandler = Arc::new(move |_input, _plan, _conn, auth| {
+            *seen_handler_cl.lock() = auth.identity.as_ref().map(|i| i.id.clone());
+            tokio::spawn(async {})
+        });
+
+        let mut install_auth = AuthContext::anonymous(b"alk/tty");
+        install_auth.identity = Some(Identity {
+            id: "bob".to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        });
+        let handler = make_open_handler_once(
+            manager.clone(),
+            policy,
+            None,
+            open_handler,
+            install_auth,
+            "alk/tty".to_string(),
+        );
+        let mut ctx = test_context("open-per-call-2");
+        ctx.identity = None;
+        let env = handler(json!({}), ctx).await;
+        env.result.expect("open should succeed");
+
+        assert_eq!(
+            seen_handler.lock().as_deref(),
+            Some("bob"),
+            "identity-less call keeps the install-time identity"
+        );
+    }
+
     #[tokio::test]
     async fn make_open_handler_stream_success_path() {
         let manager = make_manager().await;
@@ -1628,11 +1761,11 @@ mod tests {
         let open_handler: OpenHandler =
             Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {}));
         let auth = AuthContext::anonymous(b"alk/call");
-        let opener_id = Identity {
+        let opener = Some(Identity {
             id: "alice".to_string(),
             scopes: vec![],
             resources: HashMap::new(),
-        };
+        });
         let env = run_open_wrapper(
             &manager,
             &policy,
@@ -1641,8 +1774,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            opener_id.id.clone(),
-            opener_id,
+            opener,
             "req-cap-deny".to_string(),
             None,
         )
@@ -1705,8 +1837,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-est-fail".to_string(),
             None,
         )
@@ -1764,8 +1895,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-est-timeout".to_string(),
             None,
         )
@@ -1809,8 +1939,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-est-deadline".to_string(),
             Some(Instant::now() + Duration::from_millis(50)),
         )
@@ -1844,8 +1973,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-est-ok".to_string(),
             None,
         )
@@ -1884,8 +2012,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-accepted".to_string(),
             None,
         )
@@ -1916,8 +2043,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-not-accepted".to_string(),
             None,
         )
@@ -2129,8 +2255,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-plan-flow".to_string(),
             None,
         )
@@ -2183,8 +2308,7 @@ mod tests {
                 &auth,
                 "alk/tty",
                 json!({ "resource": "same" }),
-                "alice".to_string(),
-                identity("alice"),
+                Some(identity("alice")),
                 "req-race-a".to_string(),
                 None,
             ),
@@ -2196,8 +2320,7 @@ mod tests {
                 &auth,
                 "alk/tty",
                 json!({ "resource": "same" }),
-                "alice".to_string(),
-                identity("alice"),
+                Some(identity("alice")),
                 "req-race-b".to_string(),
                 None,
             )
@@ -2233,8 +2356,7 @@ mod tests {
             &auth,
             "alk/tty",
             json!({}),
-            "alice".to_string(),
-            identity("alice"),
+            Some(identity("alice")),
             "req-no-plan".to_string(),
             None,
         )