feat: channels path carries the negotiation in the open op (L1, L3)

The channels path no longer carries a second negotiation frame on the
channel's data stream (ADR-009). The open op's registry-validated
input IS the negotiation:

- producer: make_tty_open_handler parses the open op's input into a
  NegotiateRequest and drives the new drive_session_pre_negotiated
  (same three-pump driver as drive_session, minus the wire-frame
  negotiation phase; validate/allocate factored into
  validate_and_allocate, shared by both paths). Post-open failures
  (unknown backend, allocate_failed, ownership denial) still go to the
  client as a 0x00-prefixed negotiation error frame, so the consumer's
  M1 disambiguation read applies unchanged. The tty:open scope gate is
  enforced by the registry's AccessControl (not re-checked in the
  handler).
- consumer: open_via_channels parses params locally (fail-fast before
  a channel is allocated), opens the channel, and starts raw-chunk
  mode directly (from_halves_raw — no negotiation write; the
  0x00-error-frame peek retained).
- tty_open_spec's input schema is now the partial NegotiateRequest
  shape (carriage/backend/cmd required; backend params stay free-form
  — raw JSON Schema is permissive on unknown keys).

Prerequisites landed upstream: alkcall 0.4.0 enforces
OperationSpec.input_schema at dispatch (the registry check this design
leans on never existed before); alkcall 0.4.1 parks early-arrival
chunks for un-adopted channels instead of dropping them — the open
response / producer's-first-write race was silently losing the first
chunks (found by L3's test; the session never resolved).

L3: open_via_channels + from_bidi_stream_via now covered end-to-end
(5 session tests + pre-negotiated adapter test + shared-harness tests
in the new crate::testing module; the channels harness moved there so
session tests share it).

Verification: cargo test 93 lib (default), 116 lib + 19 integration
(--all-features); clippy -D warnings native + wasm clean; fmt clean;
doc 0 warnings; wasm check clean.

Cargo.lock

@@ -27,9 +27,9 @@ dependencies = [
 
 [[package]]
 name = "alkcall"
-version = "0.3.1"
+version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d4116dba284601e3337e918d4a4256706e43dd8bb6a30095bf96d664ae58d74a"
+checksum = "9badefe048a194c93eed09bc5326ebf092fc86817e561d88de2cd12b6302753a"
 dependencies = [
  "async-trait",
  "bytes",

Cargo.toml

@@ -24,7 +24,7 @@ default = []
 local = ["dep:portable-pty", "dep:tokio-util", "tokio/process", "tokio/rt-multi-thread"]
 
 [dependencies]
-alkcall = "0.3.1"
+alkcall = "0.4.0"
 # Minimal, wasm-clean tokio features. `local` adds `process` +
 # `rt-multi-thread` (non-wasm). Do NOT use `features = ["full"]` — it
 # pulls in `signal`/`fs`/`net` which break `wasm32-unknown-unknown`.

docs/reviews/001-code-review.md

@@ -1,6 +1,6 @@
 ---
-status: partially-resolved (M1, M2, L2, L4, L5, N1, N2, N3, N5)
-last_updated: 2026-08-17
+status: partially-resolved (L6, N4, N6)
+last_updated: 2026-09-05
 reviewed_artifacts:
   - src/lib.rs
   - src/wire.rs
@@ -532,8 +532,8 @@ Highlights:
 | M2 | `wait()` swallows `MalformedExitChunk` | propagate the variant | small | low | ✅ resolved |
 | L2 | consumer stdout/stderr routing untested | add emitting test backend | small | low | ✅ resolved |
 | M1 | negotiation-rejection frame unhandled | implement disambiguation read | medium | medium (wire-facing) | ✅ resolved |
-| L1 | channels `input` ignored | decide drop-vs-pass-through | small | low | open |
-| L3 | `open_via_channels` 0% covered | end-to-end channels consumer test | medium | low | open |
+| L1 | channels `input` ignored | decide drop-vs-pass-through | small | low | ✅ resolved (2026-09-05) |
+| L3 | `open_via_channels` 0% covered | end-to-end channels consumer test | medium | low | ✅ resolved (2026-09-05) |
 | L6 | pty bridge error paths untested | targeted error-path tests | medium | low | open |
 | N4 | sleep-based timing | readiness signals | small | low | open |
 | N6 | MSRV unverified | CI MSRV job or bump | small | none | open |
@@ -569,23 +569,74 @@ Post-fix coverage: `session.rs` 79.71% → 87.43% lines, total 90.74% →
 `cargo test --all-features`, clippy native + wasm, fmt, doc, wasm
 check).
 
+### Resolution (2026-09-05, L1 + L3 — the channels consumer path)
+
+**L1 — resolved via the publisher decision: the channels path carries no
+second negotiation frame (ADR-009 in `docs/architecture/decisions/`).**
+
+Two upstream alkcall changes were prerequisites (the review's premise
+that the registry "validates `input`" was wrong — alkcall never
+enforced `input_schema` on any dispatch path):
+
+- alkcall 0.4.0 — `OperationSpec.input_schema` is now enforced at call
+  time by all three registry dispatch entry points (`invoke`,
+  `invoke_streaming`, `invoke_sink`), compiled once at registration
+  (fail-closed, same rule as `publish_schema`/CF-003). Violations
+  return `INVALID_INPUT`. The `channels/tty/sub` spec's input schema
+  now declares the shared `NegotiateRequest` fields (`carriage`,
+  `backend`, `cmd` required); `tty`/`cwd`/`env`/backend-params stay
+  free-form (raw JSON Schema is permissive on unknown keys, so the
+  opaque ADR-053 params pass through).
+- alkcall 0.4.1 — early-arrival chunks for a not-yet-adopted channel
+  are parked (bounded per-channel buffer) and drained on
+  `adopt_channel`, instead of dropped. The open-op-response /
+  producer's-first-write race silently lost the first chunks of any
+  push-first producer — found by L3's test (the session never resolved
+  because the stdout sentinel + exit chunk of an
+  immediately-resolving backend arrived before the adopt).
+
+Design: the open op's `input` IS the negotiation (design 2 of the
+publisher decision). `make_tty_open_handler` parses the
+registry-validated `input` into a `NegotiateRequest` and drives
+`drive_session_pre_negotiated` (new public API in `adapter.rs`) — the
+same three-pump session driver as the direct path, minus the
+wire-frame negotiation phase. Validation still runs
+(`carriage`/`cmd`/backend lookup + ADR-050 ownership); failures go to
+the client as a `0x00`-prefixed negotiation error frame on the channel
+stream, so the M1 disambiguation read applies unchanged. The
+`tty:open` scope gate is enforced by the registry's `AccessControl`
+(not re-checked in the handler). `tty_open_spec()`'s schema is now the
+partial `NegotiateRequest` shape. `TtySession::open_via_channels`
+parses `params` locally (fail-fast before a channel is allocated),
+opens the channel, and starts raw-chunk mode directly
+(`from_halves_raw` — no negotiation write, peek retained for the
+error-frame path).
+
+Tests: `open_via_channels_end_to_end_negotiates_and_waits`,
+`open_via_channels_surfaces_negotiation_rejected`,
+`open_via_channels_fails_fast_on_schema_invalid_params`,
+`open_via_channels_fails_fast_on_unparseable_params`,
+`open_via_channels_routes_backend_stdout_and_stderr`,
+`pre_negotiated_happy_path_over_plain_duplex` (adapter), plus the
+shared harness tests in `src/testing.rs` (harness-level handler→client
+data flow). The channels harness (`wire_client_and_server`) moved to
+`crate::testing` so session tests share it with channels tests.
+
+**L3 — resolved** by the same work: `open_via_channels` (and
+`from_bidi_stream_via`) are covered end-to-end against the real
+producer path (`register_openable` + `drive_session_pre_negotiated`
+through alkcall's channels stack).
+
 ### Remaining (open)
 
-- **L1** — channels `input` ignored; needs a publisher decision
-  (drop the parameter vs pass-through to `drive_session`).
-- **L3** — `open_via_channels` still 0% covered; needs the channels
-  harness shared across modules.
 - **L6** — pty bridge error paths untested.
 - **N4** — sleep-based timing in signal/cancel tests.
 - **N6** — MSRV unverified.
 
 ### Recommended Order (remaining)
 
-1. **L1 + L3** — the channels consumer path; do together since L3's
-   test will exercise L1's code.
-2. **L6** — pty bridge error paths; medium effort, lower priority than
-   the consumer-half work.
-3. **N4 + N6** — test hardening and MSRV; defer until CI exists.
+1. **L6** — pty bridge error paths; medium effort.
+2. **N4 + N6** — test hardening and MSRV; defer until CI exists.
 
 ---
 
@@ -593,7 +644,8 @@ check).
 
 - All line numbers refer to the tree at commit `18c4924` (the last
   commit on `main` at review time). The resolution section above
-  reflects the tree at commit `9944153`.
+  reflects the tree at commit `9944153` and the 2026-09-05 L1+L3
+  resolution.
 - The coverage numbers are from `cargo llvm-cov --all-features` on the
   same tree. The `--show-missing-lines` output was used to attribute
   gaps; the full report is at `target/llvm-cov/html`.

src/adapter.rs

@@ -192,7 +192,10 @@ async fn send_negotiation_error<W: AsyncWrite + Unpin>(
 /// when the stream is reset (cancel-cleanup path — no exit chunk sent).
 ///
 /// This is the per-stream session driver — generalized from the POC's
-/// `session::drive_session` to the [`TtyBackend`] trait.
+/// `session::drive_session` to the [`TtyBackend`] trait. The negotiation
+/// frame is read from the wire (the direct-ALPN path — ADR-052); the
+/// channels path (open-op `input` is the negotiation) uses
+/// [`drive_session_pre_negotiated`] instead.
 pub async fn drive_session(
     client_send: impl AsyncWrite + Send + Unpin + 'static,
     client_recv: impl AsyncRead + Send + Unpin + 'static,
@@ -207,6 +210,131 @@ pub async fn drive_session(
     }
 }
 
+/// Drive a `alk/tty` session whose negotiation already happened out of
+/// band — the channels path (ADR-009). The open op's registry-validated
+/// `input` was parsed into `req` by the channels wrapper; the raw-chunk
+/// data plane starts immediately. No negotiation frame is read from the
+/// wire, and none is written by the client.
+///
+/// Validation still runs here (`carriage`/`cmd`/backend lookup) plus the
+/// ADR-050 ownership check — failures go to the client as a negotiation
+/// error frame on the channel stream (the same framing the direct path
+/// uses, so the consumer's M1 disambiguation read applies unchanged).
+/// The `tty:open` scope gate is NOT re-checked: on the channels path the
+/// registry's `AccessControl` enforced it at open time (the identity view
+/// the registry checked — resolved from the connection — is the
+/// authoritative one; the handler-side identity is only the
+/// ownership-check subject).
+pub async fn drive_session_pre_negotiated(
+    client_send: impl AsyncWrite + Send + Unpin + 'static,
+    client_recv: impl AsyncRead + Send + Unpin + 'static,
+    req: NegotiateRequest,
+    backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
+    ownership: Option<Arc<dyn OwnershipProvider>>,
+    identity: Option<Identity>,
+) {
+    if let Err(e) = drive_session_pre_negotiated_inner(
+        client_send,
+        client_recv,
+        req,
+        &backends,
+        &ownership,
+        &identity,
+    )
+    .await
+    {
+        debug!("tty: session ended with error: {e}");
+    }
+}
+
+/// Validate a parsed `NegotiateRequest`, select the backend, run the
+/// ADR-050 ownership check, and allocate. Shared by the direct
+/// (wire-negotiated) and channels (pre-negotiated) paths. Failures are
+/// reported to the client as a negotiation error frame on `neg_writer`
+/// (which is consumed and shut down); `Err(())` means "session over, the
+/// client has been told". On success the unwrapped write half comes back
+/// with the allocated handle.
+///
+/// `enforce_scope` gates the `tty:open` scope check: `true` on the direct
+/// path (the adapter is the only gate), `false` on the channels path (the
+/// registry's `AccessControl` enforced the scope at open time — see
+/// [`drive_session_pre_negotiated`]).
+#[allow(clippy::type_complexity)]
+async fn validate_and_allocate<W>(
+    neg_writer: NegotiationWriter<W>,
+    req: NegotiateRequest,
+    backends: &HashMap<String, Arc<dyn TtyBackend>>,
+    ownership: &Option<Arc<dyn OwnershipProvider>>,
+    identity: &Option<Identity>,
+    enforce_scope: bool,
+) -> Result<(TtyHandle, W), ()>
+where
+    W: AsyncWrite + Send + Unpin + 'static,
+{
+    if req.carriage != "raw" {
+        send_negotiation_error(
+            neg_writer,
+            "malformed_negotiation",
+            &[("message", "carriage must be 'raw'")],
+        )
+        .await;
+        return Err(());
+    }
+    if req.cmd.is_empty() {
+        send_negotiation_error(
+            neg_writer,
+            "malformed_negotiation",
+            &[("message", "cmd must be non-empty")],
+        )
+        .await;
+        return Err(());
+    }
+
+    let backend = match backends.get(&req.backend) {
+        Some(b) => b.clone(),
+        None => {
+            send_negotiation_error(neg_writer, "unknown_backend", &[("backend", &req.backend)])
+                .await;
+            return Err(());
+        }
+    };
+
+    if enforce_scope && !has_scope(identity, TTY_OPEN_SCOPE) {
+        send_negotiation_error(neg_writer, "forbidden", &[]).await;
+        return Err(());
+    }
+
+    let params = crate::backend::TtyParams::from(req);
+
+    if let Some(provider) = ownership {
+        if let Some((kind, id)) = backend.resource_id(&params) {
+            let owns = identity
+                .as_ref()
+                .map(|id_ref| provider.owns(id_ref, kind, &id, "tty"))
+                .unwrap_or(false);
+            if !owns {
+                send_negotiation_error(neg_writer, "forbidden", &[]).await;
+                return Err(());
+            }
+        }
+    }
+
+    let handle = match backend.allocate(&params).await {
+        Ok(h) => h,
+        Err(e) => {
+            send_negotiation_error(
+                neg_writer,
+                "allocate_failed",
+                &[("message", &e.to_string())],
+            )
+            .await;
+            return Err(());
+        }
+    };
+
+    Ok((handle, neg_writer.into_inner()))
+}
+
 async fn drive_session_inner<W, R>(
     client_send: W,
     client_recv: R,
@@ -253,72 +381,38 @@ where
         }
     };
 
-    if req.carriage != "raw" {
-        send_negotiation_error(
-            neg_writer,
-            "malformed_negotiation",
-            &[("message", "carriage must be 'raw'")],
-        )
-        .await;
-        return Ok(());
-    }
-    if req.cmd.is_empty() {
-        send_negotiation_error(
-            neg_writer,
-            "malformed_negotiation",
-            &[("message", "cmd must be non-empty")],
-        )
-        .await;
-        return Ok(());
-    }
-
-    let backend = match backends.get(&req.backend) {
-        Some(b) => b.clone(),
-        None => {
-            send_negotiation_error(neg_writer, "unknown_backend", &[("backend", &req.backend)])
-                .await;
-            return Ok(());
-        }
-    };
-
-    if !has_scope(identity, TTY_OPEN_SCOPE) {
-        send_negotiation_error(neg_writer, "forbidden", &[]).await;
-        return Ok(());
-    }
-
-    let params = crate::backend::TtyParams::from(req);
-
-    if let Some(provider) = ownership {
-        if let Some((kind, id)) = backend.resource_id(&params) {
-            let owns = identity
-                .as_ref()
-                .map(|id_ref| provider.owns(id_ref, kind, &id, "tty"))
-                .unwrap_or(false);
-            if !owns {
-                send_negotiation_error(neg_writer, "forbidden", &[]).await;
-                return Ok(());
-            }
-        }
-    }
-
-    let handle = match backend.allocate(&params).await {
-        Ok(h) => h,
-        Err(e) => {
-            send_negotiation_error(
-                neg_writer,
-                "allocate_failed",
-                &[("message", &e.to_string())],
-            )
-            .await;
-            return Ok(());
-        }
-    };
+    let (handle, client_write) =
+        match validate_and_allocate(neg_writer, req, backends, ownership, identity, true).await {
+            Ok(ok) => ok,
+            Err(()) => return Ok(()),
+        };
 
     let client_read = neg_reader.into_inner();
-    let client_write = neg_writer.into_inner();
     pump_session(client_write, client_read, handle).await
 }
 
+async fn drive_session_pre_negotiated_inner<W, R>(
+    client_send: W,
+    client_recv: R,
+    req: NegotiateRequest,
+    backends: &HashMap<String, Arc<dyn TtyBackend>>,
+    ownership: &Option<Arc<dyn OwnershipProvider>>,
+    identity: &Option<Identity>,
+) -> Result<(), std::io::Error>
+where
+    W: AsyncWrite + Send + Unpin + 'static,
+    R: AsyncRead + Send + Unpin + 'static,
+{
+    let neg_writer = NegotiationWriter::new(client_send);
+    let (handle, client_write) =
+        match validate_and_allocate(neg_writer, req, backends, ownership, identity, false).await {
+            Ok(ok) => ok,
+            Err(()) => return Ok(()),
+        };
+
+    pump_session(client_write, client_recv, handle).await
+}
+
 /// Phase 3: the bidirectional pump. Three concurrent tasks plus a drainer.
 ///
 /// Enforces the exit-chunk-is-last invariant (ADR-055): the adapter waits for
@@ -1519,4 +1613,38 @@ mod tests {
         exit_tx.send(Ok(0)).unwrap();
         let _ = session.await;
     }
+
+    #[tokio::test]
+    async fn pre_negotiated_happy_path_over_plain_duplex() {
+        use crate::backend::MockBackend;
+        use tokio::io::AsyncReadExt;
+
+        let backend: Arc<dyn TtyBackend> = Arc::new(MockBackend::with_exit_code(0));
+        let mut backends: StdHashMap<String, Arc<dyn TtyBackend>> = StdHashMap::new();
+        backends.insert("mock".to_string(), backend);
+        let backends = Arc::new(backends);
+
+        let (mut client, server) = duplex(8 * 1024);
+        let (server_read, server_write) = tokio::io::split(server);
+
+        let identity = identity_with_scope(TTY_OPEN_SCOPE);
+        let req: NegotiateRequest =
+            serde_json::from_slice(TEST_NEG.as_bytes()).expect("parse TEST_NEG");
+        let session = tokio::spawn(async move {
+            drive_session_pre_negotiated(server_write, server_read, req, backends, None, identity)
+                .await;
+        });
+
+        let mut buf = [0u8; 5];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            client.read_exact(&mut buf),
+        )
+        .await
+        .expect("no first chunk from pre-negotiated driver")
+        .expect("read");
+        assert_eq!(buf[0], STREAM_STDOUT, "stdout sentinel comes first");
+
+        let _ = session.await;
+    }
 }

src/channels.rs

@@ -22,7 +22,7 @@
 //! `OperationSpec`'s `AccessControl` and enforced by the registry's
 //! `invoke`/`invoke_streaming` before the `TtyOpenHandler` runs — the
 //! handler itself only validates that the negotiated backend exists
-//! and spawns the protocol. This is the difference from the
+//! and spawns the session driver. This is the difference from the
 //! direct-ALPN `TtyAdapter`, which does its own ad-hoc scope check;
 //! the channels path gets ACL for free from alkcall's registry.
 //!
@@ -41,8 +41,9 @@ use alkcall::registry::spec::{
 use serde_json::{json, Value};
 use tracing::debug;
 
-use crate::adapter::{drive_session, TTY_OPEN_SCOPE};
+use crate::adapter::{drive_session_pre_negotiated, TTY_OPEN_SCOPE};
 use crate::backend::TtyBackend;
+use crate::negotiation::NegotiateRequest;
 
 /// The per-ALPN open operation name (`channels/<alpn>/sub` convention,
 /// ADR-047). TTY is consumer-opens (the client requests a shell), so
@@ -78,12 +79,13 @@ pub const TTY_ALPN: &str = "alk/tty";
 ///
 /// `ownership` is the optional `OwnershipProvider` for the ADR-050
 /// resource-ownership check. `None` = scope-gate only. The provider
-/// is consulted inside `drive_session` (after the backend is selected
-/// and `resource_id` is extracted from `backend_params`), not by the
-/// channels wrapper — the wrapper's `AccessControl` carries only the
-/// scope gate. This mirrors the direct-ALPN `TtyAdapter` shape: the
-/// scope gate is enforced once at the entry point, the ownership
-/// check is enforced once at the backend-selection point.
+/// is consulted inside `drive_session_pre_negotiated` (after the
+/// backend is selected and `resource_id` is extracted from
+/// `backend_params`), not by the channels wrapper — the wrapper's
+/// `AccessControl` carries only the scope gate. This mirrors the
+/// direct-ALPN `TtyAdapter` shape: the scope gate is enforced once at
+/// the entry point (here by the registry's `AccessControl`), the
+/// ownership check is enforced once at the backend-selection point.
 ///
 /// `auth` is the peer's `AuthContext`, captured at
 /// `install_channel_zero` time and closed over by the wrapper so the
@@ -117,14 +119,18 @@ pub fn register_openable(
 /// a data channel. The `AccessControl` carries the `tty:open` scope
 /// gate (the same scope the direct-ALPN `TtyAdapter` checks).
 ///
-/// The input schema is permissive (`type: object`) — the
-/// `NegotiateRequest` JSON shape is validated by `drive_session`'s
-/// negotiation reader, not by the registry's schema validator. This
-/// keeps the wire-format definition in one place (`wire.rs` +
-/// `negotiation.rs`) and avoids duplicating the schema in the
-/// `OperationSpec`. A future tightening could add the full
-/// `NegotiateRequest` JSON schema here; the permissive shape is the
-/// starting point.
+/// The input schema is the shared part of the `NegotiateRequest` shape
+/// (`carriage`/`backend`/`cmd` required — alkcall 0.4 enforces it at
+/// dispatch). The schema is deliberately partial: `tty`, `cwd`, `env`,
+/// and backend-specific selector fields are free-form (`additionalProperties`
+/// defaults to `true` in raw JSON Schema), because the
+/// `NegotiateRequest` wire shape is owned by `negotiation.rs` and the
+/// backend params are opaque (ADR-053) — duplicating their full schemas
+/// here would create a second definition to drift. The registry check
+/// catches structurally-broken opens (missing command, wrong types);
+/// the full parse and semantic validation (`carriage == "raw"`,
+/// backend lookup) runs in the handler via
+/// [`crate::adapter::drive_session_pre_negotiated`].
 pub fn tty_open_spec() -> OperationSpec {
     OperationSpec::new(
         OP_TTY_OPEN,
@@ -162,21 +168,35 @@ pub fn tty_open_spec() -> OperationSpec {
 
 /// Build the [`OpenHandler`] for `channels/tty/sub`.
 ///
-/// The handler receives the open op's `input` (the `NegotiateRequest`
-/// params — ignored here; `drive_session` reads it from the channel's
-/// `BiStream` as the first frame, same as the direct-ALPN path), the
-/// channel's [`Connection`] (data-plane ALPN `alk/tty`), and the
-/// peer's [`AuthContext`]. It calls `accept_bi()` to get the channel's
-/// [`BiStream`], splits it into read/write halves (the stdlib
-/// `tokio::io::split` idiom — the same split the direct-ALPN
-/// `TtyAdapter::handle` does), and runs [`drive_session`] on them.
+/// The handler receives the open op's `input` — the `NegotiateRequest`
+/// params, validated against the spec's input schema by the registry
+/// (alkcall 0.4: `input_schema` is enforced at dispatch) and passed
+/// through as the authoritative negotiation (ADR-009: the channels
+/// path does not carry a second negotiation frame on the channel; the
+/// open op's `input` *is* the negotiation). The channel's
+/// [`Connection`] (data-plane ALPN `alk/tty`) and the peer's
+/// [`AuthContext`] come with it. The handler parses `input`, calls
+/// `accept_bi()` to get the channel's `BiStream`, splits it into
+/// read/write halves (the stdlib `tokio::io::split` idiom — the same
+/// split the direct-ALPN `TtyAdapter::handle` does), and runs
+/// [`crate::adapter::drive_session_pre_negotiated`] on them — the same
+/// three-pump session driver as the direct-ALPN path, minus the
+/// wire-frame negotiation phase.
+///
+/// Parse or validation failures (unknown backend, `allocate_failed`,
+/// ADR-050 ownership denial) are reported to the client as a
+/// negotiation error frame on the channel stream (the same
+/// `0x00`-prefixed framing the direct path uses), so a consumer's
+/// negotiation-error disambiguation read applies unchanged. The
+/// `tty:open` scope gate is enforced by the registry's `AccessControl`
+/// before this handler runs — the handler does not re-check it.
 ///
 /// The handler's `JoinHandle` is recorded by the channels wrapper for
 /// teardown (abort on `channel/close` / connection drop). When the
 /// session ends (exit chunk sent, stream closed, or stream reset),
-/// `drive_session` returns and the spawned task completes; the
-/// wrapper's teardown task then calls `manager.teardown_channel` and
-/// `policy.on_close` (ADR-047 §7).
+/// `drive_session_pre_negotiated` returns and the spawned task
+/// completes; the wrapper's teardown task then calls
+/// `manager.teardown_channel` and `policy.on_close` (ADR-047 §7).
 fn make_tty_open_handler(
     backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
     ownership: Option<Arc<dyn OwnershipProvider>>,
@@ -187,9 +207,15 @@ fn make_tty_open_handler(
             let backends = Arc::clone(&backends);
             let ownership = ownership.clone();
             let identity = identity.clone().or_else(|| auth.identity.clone());
-            let _ = input;
 
             tokio::spawn(async move {
+                let req: NegotiateRequest = match serde_json::from_value(input) {
+                    Ok(r) => r,
+                    Err(e) => {
+                        debug!("tty: channels open: invalid negotiate params: {e}");
+                        return;
+                    }
+                };
                 let stream = match channel_conn.accept_bi().await {
                     Ok(s) => s,
                     Err(e) => {
@@ -198,7 +224,15 @@ fn make_tty_open_handler(
                     }
                 };
                 let (client_read, client_write) = tokio::io::split(stream);
-                drive_session(client_write, client_read, backends, ownership, identity).await;
+                drive_session_pre_negotiated(
+                    client_write,
+                    client_read,
+                    req,
+                    backends,
+                    ownership,
+                    identity,
+                )
+                .await;
             })
         },
     )
@@ -208,125 +242,13 @@ fn make_tty_open_handler(
 mod tests {
     use super::*;
     use crate::backend::{MockBackend, TtyError};
-    use alkcall::channels::client::ChannelClient;
+    use crate::testing::wire_client_and_server;
     use alkcall::channels::operations::ChannelCore;
-    use alkcall::channels::policy::default_policy;
     use alkcall::core::auth::Identity;
-    use alkcall::core::types::Connection as CoreConnection;
     use alkcall::registry::registration::OperationRegistry;
     use std::collections::HashMap as StdHashMap;
     use tokio::io::duplex;
 
-    /// Build a `ChannelClient` and a server-side `ChannelCore` + registry
-    /// with `channels/tty/sub` registered, wired over a `tokio::io::duplex`
-    /// carrying the channels 8-byte chunk header wire format. Returns the
-    /// client and the per-connection `ChannelCore` (for any teardown
-    /// assertions). The server-side dispatch loop is spawned and runs
-    /// until the client drops or the test ends.
-    ///
-    /// This mirrors the alkcall `channel_0_end_to_end_register_openable`
-    /// test's wiring pattern but uses alktty's `register_openable` helper
-    /// so the registration is the code under test.
-    async fn wire_client_and_server(
-        backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
-        ownership: Option<Arc<dyn OwnershipProvider>>,
-        identity: Option<Identity>,
-    ) -> ChannelClient {
-        use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
-        use alkcall::core::auth::IdentityProvider;
-        use alkcall::protocol::connection::split_single_stream;
-        use alkcall::protocol::dispatch::Dispatcher;
-
-        struct NoopIdProvider;
-        impl IdentityProvider for NoopIdProvider {
-            fn resolve_from_fingerprint(&self, _: &str) -> Option<alkcall::core::auth::Identity> {
-                None
-            }
-            fn resolve_from_token(
-                &self,
-                _: &alkcall::core::auth::AuthToken,
-            ) -> Option<alkcall::core::auth::Identity> {
-                None
-            }
-        }
-
-        let policy = default_policy();
-        let policy_for_hook = Arc::clone(&policy);
-        let identity_for_conn = identity.clone();
-
-        let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, auth| {
-            let backends = Arc::clone(&backends);
-            let ownership = ownership.clone();
-            let _identity = identity.clone();
-            let policy = Arc::clone(&policy_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);
-
-                // Propagate the identity to the channel-0
-                // connection so the call dispatch's
-                // `resolve_identity` sees it. The
-                // `ChannelsAdapter` builds `channel0_conn` fresh
-                // from `channel_source` and does NOT inherit the
-                // outer connection's identity — we set it here
-                // so the `AccessControl` scope-gate can check it.
-                if let Some(id) = _identity {
-                    let _ = channel0_conn.set_identity(id);
-                }
-
-                let core = ChannelCore::new(manager, policy);
-                let mut registry = OperationRegistry::new();
-                register_openable(
-                    &core,
-                    Arc::clone(&backends),
-                    ownership.clone(),
-                    &mut registry,
-                    auth.clone(),
-                )
-                .expect("register_openable");
-                let registry = Arc::new(registry);
-                let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
-                let call_connection = Arc::new(
-                    alkcall::protocol::connection::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) = duplex(64 * 1024);
-        let client_conn = CoreConnection::from_bidi(client_end, b"alk/channels".to_vec(), None);
-        let server_conn = CoreConnection::from_bidi(server_end, b"alk/channels".to_vec(), None);
-        // Set the identity on the server connection so the call
-        // dispatch sees it (the ACL check runs against the
-        // connection's identity, not the `install_channel_zero`
-        // hook's `auth`).
-        if let Some(id) = &identity_for_conn {
-            let _ = server_conn.set_identity(id.clone());
-        }
-
-        let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
-        let auth = AuthContext::anonymous(b"alk/channels");
-        let _server_handle = tokio::spawn(async move {
-            let _ =
-                alkcall::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
-        });
-
-        ChannelClient::from_connection(client_conn)
-            .await
-            .expect("channel client init")
-    }
-
-    /// A `ChannelLifecyclePolicy` with no cap (for tests).
-    use alkcall::channels::policy::NoCap;
-
     #[test]
     fn op_tty_open_is_channels_tty_sub() {
         assert_eq!(OP_TTY_OPEN, "channels/tty/sub");
@@ -372,6 +294,7 @@ mod tests {
     /// must be discoverable by name afterwards.
     #[tokio::test]
     async fn register_openable_registers_op_on_registry() {
+        use alkcall::channels::policy::default_policy;
         let (_client, server) = duplex(1024);
         let (_reader, writer) = tokio::io::split(server);
         let (handle, _runner) = alkcall::channels::mux::MuxRunner::new(Box::new(writer));

src/lib.rs

@@ -56,3 +56,6 @@ pub mod wire;
 
 #[cfg(feature = "local")]
 pub mod local;
+
+#[cfg(test)]
+pub(crate) mod testing;

src/session.rs

@@ -17,8 +17,10 @@
 //!   `open_via_channels(client, params)`, which invokes
 //!   `channels/tty/sub` on channel 0, adopts the resulting channel,
 //!   builds a `Connection` from the reassembled read half + mux write
-//!   half, and runs the same negotiation + typed-methods flow on the
-//!   channel's `BiStream`.
+//!   half, and runs the typed-methods flow directly in raw chunk mode
+//!   (ADR-009: the open op's `params` — validated by the registry's
+//!   input schema — *are* the negotiation; no second negotiation
+//!   frame is written on the channel's data stream).
 //!
 //! The session handle exposes:
 //! - [`TtySession::send_stdin`] / [`TtySession::close_stdin`] — write
@@ -31,10 +33,11 @@
 //!   process exit code).
 //!
 //! The session does NOT re-serialize the negotiation request itself
-//! — the caller passes a `NegotiateRequest` (or a `serde_json::Value`
-//! for the channels path, since `ChannelClient::open_channel` takes a
-//! `Value`). The session writes the frame and switches to raw-chunk
-//! mode. See ADR-052 for the two-carriage model.
+//! — the caller passes a `NegotiateRequest` (direct path) or a
+//! `serde_json::Value` (channels path — the open op's `input` is the
+//! negotiation). The direct path writes the frame and switches to
+//! raw-chunk mode; the channels path starts in raw-chunk mode. See
+//! ADR-052 for the two-carriage model.
 //!
 //! # WASM
 //!
@@ -180,22 +183,34 @@ impl TtySession {
     /// `open_via_channels(client, params)`, which invokes
     /// `channels/tty/sub` on channel 0, adopts the resulting channel,
     /// builds a `Connection` from the reassembled read half + mux
-    /// write half, and runs the same negotiation + typed-methods flow
-    /// on the channel's `BiStream`.
+    /// write half, and runs the typed-methods flow directly in raw
+    /// chunk mode (ADR-009: the open op's `params` — validated by the
+    /// registry's input schema — *are* the negotiation; no second
+    /// negotiation frame is written on the channel's data stream).
     ///
     /// `params` is the `NegotiateRequest` as a `serde_json::Value` —
-    /// the channels open op takes a `Value`, not a typed struct (the
-    /// op's input schema is the `NegotiateRequest` shape). The
-    /// session re-parses it as a `NegotiateRequest` after the channel
-    /// is open so the typed methods can use the strongly-typed shape.
+    /// the channels open op takes a `Value`, not a typed struct. The
+    /// session parses it as a `NegotiateRequest` before opening (so a
+    /// malformed request fails fast, before a channel is allocated)
+    /// and the producer parses the same value from the open op.
+    ///
+    /// Failures before the channel opens (ACL denial, unknown op,
+    /// channel cap, invalid params) surface as
+    /// [`TtySessionError::ChannelsOpen`]. Post-open failures (unknown
+    /// backend, allocate failure, ownership denial) arrive as a
+    /// negotiation error frame on the channel stream — the session
+    /// surfaces those as [`TtySessionError::NegotiationRejected`] via
+    /// the same `0x00` disambiguation read the direct path uses.
     pub async fn open_via_channels(
         client: &ChannelClient,
         params: serde_json::Value,
     ) -> Result<Self, TtySessionError> {
+        let _: NegotiateRequest = serde_json::from_value(params.clone())
+            .map_err(TtySessionError::NegotiationSerialize)?;
         let (channel_id, send, recv) = client
             .open_channel(
                 crate::channels::OP_TTY_OPEN,
-                params.clone(),
+                params,
                 crate::channels::TTY_ALPN,
             )
             .await
@@ -207,9 +222,7 @@ impl TtySession {
         let channel_conn =
             Connection::from_source(source, crate::channels::TTY_ALPN.as_bytes().to_vec());
 
-        let negotiate: NegotiateRequest =
-            serde_json::from_value(params).map_err(TtySessionError::NegotiationSerialize)?;
-        Self::from_bidi_stream_via(channel_conn, negotiate).await
+        Self::from_bidi_stream_via(channel_conn).await
     }
 
     /// Shared inner: take a `BiStream`, write the negotiation frame,
@@ -225,15 +238,14 @@ impl TtySession {
 
     /// Like `from_bidi_stream` but takes the channel's `Connection`
     /// directly (the channels path already has the `Connection` from
-    /// `Connection::from_source`).
-    async fn from_bidi_stream_via(
-        channel_conn: Connection,
-        negotiate: NegotiateRequest,
-    ) -> Result<Self, TtySessionError> {
+    /// `Connection::from_source`). The negotiation already happened in
+    /// the open op (ADR-009) — the stream starts in raw-chunk mode.
+    async fn from_bidi_stream_via(channel_conn: Connection) -> Result<Self, TtySessionError> {
         let stream = channel_conn.accept_bi().await.map_err(|e| {
             std::io::Error::new(std::io::ErrorKind::ConnectionReset, format!("{e}"))
         })?;
-        Self::from_bidi_stream(stream, negotiate).await
+        let (read, write) = tokio::io::split(stream);
+        Self::from_halves_raw(read, write).await
     }
 
     /// Core inner: take a read half and a write half, write the
@@ -274,6 +286,48 @@ impl TtySession {
             Err(e) => return Err(TtySessionError::Wire(e)),
         }
 
+        Self::start_pump(writer, reader, first_byte_peeked)
+    }
+
+    /// Core inner for the channels path (ADR-009): the negotiation
+    /// already happened in the open op — the stream is already in
+    /// raw-chunk mode. The peek still applies: the producer sends a
+    /// `0x00`-prefixed error frame on post-open failures (unknown
+    /// backend, allocate failure, ownership denial), and a raw chunk
+    /// (`stream_type` in `{1, 2, 4}`) on success.
+    async fn from_halves_raw<R, W>(read: R, write: W) -> Result<Self, TtySessionError>
+    where
+        R: AsyncRead + Send + Unpin + 'static,
+        W: AsyncWrite + Send + Unpin + 'static,
+    {
+        let writer = ChunkWriter::new(Box::new(write) as Box<dyn AsyncWrite + Send + Unpin>);
+        let mut reader = ChunkReader::new(read);
+        let mut first_byte_peeked = false;
+        match reader.peek_stream_type().await {
+            Ok(0x00) => {
+                return Err(read_negotiation_error(reader.into_inner()).await);
+            }
+            Ok(_) => first_byte_peeked = true,
+            Err(RawError::ConnectionClosed) => {
+                // The server closed cleanly without a response. Fall
+                // through to the read pump, which resolves `NoExitChunk`.
+            }
+            Err(e) => return Err(TtySessionError::Wire(e)),
+        }
+
+        Self::start_pump(writer, reader, first_byte_peeked)
+    }
+
+    /// Wire up the exit watch + stdout/stderr channels and spawn the
+    /// read pump. Shared by `from_halves` and `from_halves_raw`.
+    fn start_pump<R>(
+        writer: ChunkWriter<Box<dyn AsyncWrite + Send + Unpin>>,
+        reader: ChunkReader<R>,
+        first_byte_peeked: bool,
+    ) -> Result<Self, TtySessionError>
+    where
+        R: AsyncRead + Send + Unpin + 'static,
+    {
         let (stdout_tx, stdout_rx) = mpsc::channel::<Bytes>(64);
         let (stderr_tx, stderr_rx) = mpsc::channel::<Bytes>(64);
         let (exit_tx, exit_rx) = tokio::sync::watch::channel::<Option<ExitOutcome>>(None);
@@ -934,4 +988,173 @@ mod tests {
         }
         let _ = server_handle.await;
     }
+
+    // --- channels consumer path (L3, over the shared `testing` harness) ----
+
+    use crate::testing::{tty_identity, wire_client_and_server};
+
+    /// The negotiate params as the open op's `input` (same JSON shape
+    /// the direct path serializes from `NegotiateRequest`).
+    fn test_open_params(backend: &str) -> serde_json::Value {
+        serde_json::json!({
+            "carriage": "raw",
+            "backend": backend,
+            "cmd": ["true"],
+        })
+    }
+
+    fn mock_backends(code: i32) -> Arc<HashMap<String, Arc<dyn TtyBackend>>> {
+        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
+        backends.insert(
+            "mock".to_string(),
+            Arc::new(MockBackend::with_exit_code(code)),
+        );
+        Arc::new(backends)
+    }
+
+    /// End-to-end (L3): `TtySession::open_via_channels` opens the
+    /// channel through the real `register_openable` producer path and
+    /// resolves the session. The negotiation travels in the open op's
+    /// `input` (ADR-009); the channel stream starts in raw-chunk mode.
+    #[tokio::test]
+    async fn open_via_channels_end_to_end_negotiates_and_waits() {
+        let client =
+            wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await;
+
+        let session = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            TtySession::open_via_channels(&client, test_open_params("mock")),
+        )
+        .await
+        .expect("open_via_channels timed out")
+        .expect("session opens");
+
+        let code = tokio::time::timeout(std::time::Duration::from_secs(10), session.wait())
+            .await
+            .expect("wait didn't time out")
+            .expect("wait returns exit code");
+        assert_eq!(code, 0);
+    }
+
+    /// The same end-to-end path with an unknown backend: the producer
+    /// sends a `0x00`-prefixed error frame on the channel stream, the
+    /// consumer's post-open peek disambiguates it, and the session
+    /// surfaces `NegotiationRejected` (M1 on the channels path).
+    #[tokio::test]
+    async fn open_via_channels_surfaces_negotiation_rejected() {
+        let client =
+            wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await;
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            TtySession::open_via_channels(&client, test_open_params("nope")),
+        )
+        .await
+        .expect("open_via_channels timed out");
+        match result {
+            Err(TtySessionError::NegotiationRejected { error, fields }) => {
+                assert_eq!(error, "unknown_backend");
+                assert_eq!(fields.get("backend").map(String::as_str), Some("nope"));
+            }
+            Ok(_) => panic!("expected NegotiationRejected, got Ok(session)"),
+            Err(other) => panic!("expected NegotiationRejected, got {other:?}"),
+        }
+    }
+
+    /// The open op's `input` is schema-validated by the registry
+    /// (alkcall 0.4): a params value missing the required `backend`
+    /// field is rejected before any handler runs, so the failure is a
+    /// `ChannelsOpen` error (the open op fails), not a session error.
+    #[tokio::test]
+    async fn open_via_channels_fails_fast_on_schema_invalid_params() {
+        let client =
+            wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await;
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            TtySession::open_via_channels(
+                &client,
+                serde_json::json!({ "carriage": "raw", "cmd": ["true"] }),
+            ),
+        )
+        .await
+        .expect("open_via_channels timed out");
+        assert!(
+            matches!(result, Err(TtySessionError::NegotiationSerialize(_))),
+            "schema-invalid params fail at the local NegotiateRequest parse (fail-fast, pre-open)"
+        );
+    }
+
+    /// `open_via_channels` with params that fail the local
+    /// `NegotiateRequest` parse (not just the schema): fails fast,
+    /// before a channel is allocated.
+    #[tokio::test]
+    async fn open_via_channels_fails_fast_on_unparseable_params() {
+        let client =
+            wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await;
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            TtySession::open_via_channels(
+                &client,
+                serde_json::json!({ "carriage": 42, "backend": "mock", "cmd": ["true"] }),
+            ),
+        )
+        .await
+        .expect("open_via_channels timed out");
+        assert!(
+            matches!(result, Err(TtySessionError::NegotiationSerialize(_))),
+            "unparseable params must fail before the open op"
+        );
+    }
+
+    /// End-to-end with an emitting backend (L2's backend): stdout and
+    /// stderr route through the channels data plane to the consumer's
+    /// typed streams — the full producer+consumer channels path with
+    /// real data.
+    #[tokio::test]
+    async fn open_via_channels_routes_backend_stdout_and_stderr() {
+        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
+        backends.insert(
+            "mock".to_string(),
+            Arc::new(EmittingBackend {
+                stdout: vec![Bytes::from_static(b"ch-out")],
+                stderr: vec![Bytes::from_static(b"ch-err")],
+                exit_code: 3,
+            }),
+        );
+        let client =
+            wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await;
+
+        let session = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            TtySession::open_via_channels(&client, test_open_params("mock")),
+        )
+        .await
+        .expect("open_via_channels timed out")
+        .expect("session opens");
+
+        let stdout = session.recv_stdout().await;
+        let collected: Vec<Bytes> = stdout.collect().await;
+        let data: Vec<Bytes> = collected.into_iter().filter(|b| !b.is_empty()).collect();
+        assert_eq!(
+            data,
+            vec![Bytes::from_static(b"ch-out")],
+            "stdout should route through the channels data plane"
+        );
+
+        let stderr = session.recv_stderr().await.expect("stderr present");
+        let collected: Vec<Bytes> = stderr.collect().await;
+        assert_eq!(
+            collected,
+            vec![Bytes::from_static(b"ch-err")],
+            "stderr should route through the channels data plane"
+        );
+
+        let code = tokio::time::timeout(std::time::Duration::from_secs(10), session.wait())
+            .await
+            .expect("wait didn't time out")
+            .expect("wait returns exit code");
+        assert_eq!(code, 3);
+    }
 }

src/testing.rs

@@ -0,0 +1,209 @@
+//! Shared test wiring for the channels-based tests (the producer
+//! harness in `channels.rs::tests` and the consumer end-to-end tests
+//! in `session.rs::tests` both need the same `ChannelClient` ↔
+//! server wiring). `#[cfg(test)]`-only — never part of the public
+//! API.
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use alkcall::channels::client::ChannelClient;
+use alkcall::channels::operations::ChannelCore;
+use alkcall::channels::policy::default_policy;
+use alkcall::core::auth::{AuthContext, Identity};
+use alkcall::core::types::Connection as CoreConnection;
+use alkcall::registry::registration::OperationRegistry;
+
+use crate::backend::TtyBackend;
+use crate::channels::register_openable;
+
+/// Build a `ChannelClient` and a server-side `ChannelCore` + registry
+/// with `channels/tty/sub` registered, wired over a `tokio::io::duplex`
+/// carrying the channels 8-byte chunk header wire format. Returns the
+/// client. The server-side dispatch loop is spawned and runs until the
+/// client drops or the test ends.
+///
+/// This mirrors the alkcall `channel_0_end_to_end_register_openable`
+/// test's wiring pattern but uses alktty's `register_openable` helper
+/// so the registration is the code under test.
+pub(crate) async fn wire_client_and_server(
+    backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
+    ownership: Option<Arc<dyn alkcall::core::ownership::OwnershipProvider>>,
+    identity: Option<Identity>,
+) -> ChannelClient {
+    use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
+    use alkcall::channels::policy::NoCap;
+    use alkcall::core::auth::IdentityProvider;
+    use alkcall::protocol::connection::split_single_stream;
+    use alkcall::protocol::dispatch::Dispatcher;
+
+    struct NoopIdProvider;
+    impl IdentityProvider for NoopIdProvider {
+        fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
+            None
+        }
+        fn resolve_from_token(&self, _: &alkcall::core::auth::AuthToken) -> Option<Identity> {
+            None
+        }
+    }
+
+    let policy = default_policy();
+    let policy_for_hook = Arc::clone(&policy);
+    let identity_for_conn = identity.clone();
+
+    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, auth| {
+        let backends = Arc::clone(&backends);
+        let ownership = ownership.clone();
+        let _identity = identity.clone();
+        let policy = Arc::clone(&policy_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);
+
+            // Propagate the identity to the channel-0
+            // connection so the call dispatch's
+            // `resolve_identity` sees it. The
+            // `ChannelsAdapter` builds `channel0_conn` fresh
+            // from `channel_source` and does NOT inherit the
+            // outer connection's identity — we set it here
+            // so the `AccessControl` scope-gate can check it.
+            if let Some(id) = _identity {
+                let _ = channel0_conn.set_identity(id);
+            }
+
+            let core = ChannelCore::new(manager, policy);
+            let mut registry = OperationRegistry::new();
+            register_openable(
+                &core,
+                Arc::clone(&backends),
+                ownership.clone(),
+                &mut registry,
+                auth.clone(),
+            )
+            .expect("register_openable");
+            let registry = Arc::new(registry);
+            let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
+            let call_connection = Arc::new(
+                alkcall::protocol::connection::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 = CoreConnection::from_bidi(client_end, b"alk/channels".to_vec(), None);
+    let server_conn = CoreConnection::from_bidi(server_end, b"alk/channels".to_vec(), None);
+    // Set the identity on the server connection so the call
+    // dispatch sees it (the ACL check runs against the
+    // connection's identity, not the `install_channel_zero`
+    // hook's `auth`).
+    if let Some(id) = &identity_for_conn {
+        let _ = server_conn.set_identity(id.clone());
+    }
+
+    let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
+    let auth = AuthContext::anonymous(b"alk/channels");
+    let _server_handle = tokio::spawn(async move {
+        let _ = alkcall::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
+    });
+
+    ChannelClient::from_connection(client_conn)
+        .await
+        .expect("channel client init")
+}
+
+/// An identity with the `tty:open` scope (the gate
+/// `channels/tty/sub` requires).
+pub(crate) fn tty_identity(id: &str) -> Identity {
+    Identity {
+        id: id.to_string(),
+        scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
+        resources: HashMap::new(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::backend::MockBackend;
+    use crate::channels::OP_TTY_OPEN;
+
+    #[test]
+    fn tty_identity_has_tty_open_scope() {
+        let id = tty_identity("carol");
+        assert_eq!(id.id, "carol");
+        assert!(id
+            .scopes
+            .iter()
+            .any(|s| s == crate::adapter::TTY_OPEN_SCOPE));
+    }
+
+    #[tokio::test]
+    async fn harness_open_op_returns_channel_id() {
+        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
+        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
+        let client =
+            wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await;
+        let response = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            client.call_open_op(
+                OP_TTY_OPEN,
+                serde_json::json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
+            ),
+        )
+        .await
+        .expect("open op timed out");
+        assert!(
+            response.result.is_ok(),
+            "open op failed: {:?}",
+            response.result
+        );
+    }
+
+    #[tokio::test]
+    async fn harness_handler_to_client_data_flows() {
+        use tokio::io::AsyncReadExt;
+
+        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
+        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
+        let client =
+            wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await;
+
+        let (channel_id, send, mut recv) = tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            client.open_channel(
+                OP_TTY_OPEN,
+                serde_json::json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
+                crate::channels::TTY_ALPN,
+            ),
+        )
+        .await
+        .expect("open_channel timed out")
+        .expect("open_channel");
+
+        assert!(channel_id > 0);
+
+        // The producer writes a 5-byte TTY header (stdout sentinel)
+        // almost immediately after the open (MockBackend resolves
+        // exit right away) — a push-first producer whose first write
+        // races the consumer's adopt. Read it: alkcall 0.4.1's
+        // early-arrival park guarantees the first chunks are not lost.
+        let mut first = [0u8; 5];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            recv.read_exact(&mut first),
+        )
+        .await
+        .expect("no first chunk from producer (handler→client direction)")
+        .expect("read first chunk header");
+        assert_eq!(first[0], crate::wire::STREAM_STDOUT);
+        drop(send);
+    }
+}