feat: the direct op (channels/tunnel/direct, ADR-007) end to end

- params.rs: OP_TUNNEL_DIRECT / TUNNEL_DIRECT_SCOPE constants;
  SubstrateAddr (untagged {host,port} | {path}; also the ADR-008 §3
  peer shape); TunnelDirectParams with render_target();
  tunnel_direct_spec() — Sub-typed, External, alk/tunnel ALPN
  marker, scope ["tunnel:direct"] (never implied by tunnel:open),
  per-substrate if/then target sub-schema (malformed targets fail
  the registry's schema gate as INVALID_INPUT before the
  establisher). 7 unit tests.
- producer.rs: direct_establisher (+ witness variant) — parse,
  render the target to the registry backing-string form, dial via
  the SAME injected DialFn; no registry lookup (unknown_resource
  can never fire); typed errors per ADR-007 §4 minus the registry
  row; CF-006 witness seam. register_tunnel_direct_openable reuses
  the base pump handler (plan-flow unchanged).
- consumer.rs: TunnelSession::open_direct — separate constructor
  (two scopes are two capabilities), plain open_channel (dial
  establishers never bind), identical session/data-planes/teardown.
- tests/producer_direct_op.rs: 15 integration tests — tcp + udp e2e
  round-trips, malformed-target rejection (no phantom session),
  dial-failure pass-through, NOT_FOUND posture, scope separation
  both ways (FORBIDDEN, raw-call + session-level), CF-006 witness
  (raw + session paths), parse backstop, target-rendering pin.
  Harness: Direct registration mode, wire_direct* topologies,
  direct_identity/both_scopes_identity.
- CHANGELOG: Unreleased → Added.

Verified: cargo test green (93 native, 94 --all-features); clippy
-D warnings clean (native + wasm32); fmt clean; wasm32 check passes
(default crate stays wasm-clean); cargo doc clean.

Task: tasks/tunnels/direct-op.md (status: completed)

CHANGELOG.md

@@ -12,6 +12,25 @@ is decomposed in `tasks/tunnels/` and lands in the next release. The
 dependency bump below is the only behavior-affecting change so far
 (no wire or API change at this level yet).
 
+### Added
+
+- **The direct op (`channels/tunnel/direct`, ADR-007)** — dynamic-
+  target egress (the ssh `direct-tcpip` / SOCKS5 CONNECT shape),
+  riding the same `alk/tunnel` ALPN: `SubstrateAddr` (the untagged
+  substrate-shaped address object, also the forwarded op's `peer`
+  shape) + `TunnelDirectParams {substrate, target}` with a
+  per-substrate `input_schema` (malformed targets fail the registry's
+  schema gate — `INVALID_INPUT` — before the establisher),
+  `tunnel_direct_spec()`, `direct_establisher` (+ witness variant;
+  no registry lookup — `unknown_resource` can never fire; targets
+  render to the same backing-string form the registry backings use
+  and ride the SAME injected `DialFn`), `register_tunnel_direct_
+  openable` (same pump handler + plan-flow as the base op), and
+  `TunnelSession::open_direct` (separate constructor — `tunnel:direct`
+  is a separate grant from `tunnel:open`, never implied by it; the
+  session type, data planes, and teardown API are identical). An old
+  producer rejects the unknown op loudly (`NOT_FOUND`).
+
 ### Changed
 
 - **alkcall 0.7.1 → 0.8.0** (crates.io). The two upstream asks the

src/consumer.rs

@@ -22,7 +22,10 @@ use bytes::Bytes;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf};
 
 use crate::error::{ReverseOpenError, TunnelIoError, TunnelOpenError};
-use crate::params::{Substrate, TunnelParams, OP_TUNNEL_OPEN, TUNNEL_ALPN};
+use crate::params::{
+    Substrate, SubstrateAddr, TunnelDirectParams, TunnelParams, OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN,
+    TUNNEL_ALPN,
+};
 use crate::wire::{frame_datagram, DatagramReader};
 
 /// The typed consumer session (ADR-005): one session = one channel =
@@ -104,6 +107,46 @@ impl TunnelSession {
         ))
     }
 
+    /// Open a tunnel channel toward a CALLER-NAMED target (the direct
+    /// op — dynamic-target egress, ADR-007; the ssh `direct-tcpip` /
+    /// SOCKS5 CONNECT shape). A separate constructor from [`open`](Self::open),
+    /// not a params enum: `tunnel:direct` and `tunnel:open` are two
+    /// scopes — two capabilities — and the caller states which one it
+    /// holds.
+    ///
+    /// Serializes `{substrate, target}`, calls
+    /// `open_channel(OP_TUNNEL_DIRECT, …, TUNNEL_ALPN)` (plain
+    /// `open_channel` — dial establishers never bind, so no reply
+    /// fields are expected), and presents the identical session type,
+    /// data planes, and teardown API as [`open`](Self::open).
+    ///
+    /// Error surface: the base open errors minus `unknown_resource`
+    /// (no registry lookup); a malformed target is the schema /
+    /// `invalid_input`-class failure (registry-side); an old producer
+    /// rejects the unknown op loudly (`NOT_FOUND` — the SSH posture).
+    pub async fn open_direct(
+        client: &ChannelClient,
+        target: SubstrateAddr,
+        substrate: Substrate,
+    ) -> Result<Self, TunnelOpenError> {
+        let input =
+            serde_json::to_value(&TunnelDirectParams { substrate, target }).map_err(|e| {
+                TunnelOpenError::Open(ChannelOpenError::CallFailed {
+                    error: CallError::internal(format!("params serialize: {e}")),
+                })
+            })?;
+        let (channel_id, send, recv) = client
+            .open_channel(OP_TUNNEL_DIRECT, input, TUNNEL_ALPN)
+            .await?;
+        Ok(Self::from_halves(
+            channel_id,
+            substrate,
+            client.manager().clone(),
+            send,
+            recv,
+        ))
+    }
+
     /// Adopt a producer-allocated channel ID (the reverse path, `-R`):
     /// install the data plane from the adopted halves. Early arrivals
     /// are parked by the manager and drained on adopt (the adoption

src/lib.rs

@@ -29,9 +29,14 @@ pub mod local;
 
 pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession};
 pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError};
-pub use params::{establishment_reason, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE};
+pub use params::{
+    establishment_reason, tunnel_direct_spec, SubstrateAddr, TunnelDirectParams, OP_TUNNEL_DIRECT,
+    TUNNEL_DIRECT_SCOPE,
+};
 pub use params::{tunnel_open_spec, ChannelOpenError, Substrate, TunnelParams, OP_TUNNEL_OPEN};
+pub use params::{TUNNEL_ALPN, TUNNEL_OPEN_SCOPE};
 pub use producer::{
+    direct_establisher, direct_establisher_with_witness, register_tunnel_direct_openable,
     register_tunnel_listen_openable, AcceptFn, AcceptQueue, DialFn, ResourceRegistry, TargetHandle,
     TunnelEstablishError,
 };

src/params.rs

@@ -3,6 +3,13 @@
 //! `OperationSpec` builder, the scope/ALPN/op-id constants, and the
 //! typed establishment-reason helper (ADR-049 §4 surface).
 //!
+//! The direct op's params ([`TunnelDirectParams`] + [`SubstrateAddr`],
+//! ADR-007) and its spec builder live here too: `{substrate, target}`
+//! — the caller names the dial target at session time (the ssh
+//! `direct-tcpip` / SOCKS5 CONNECT shape), scope-gated by
+//! `tunnel:direct` (a separate grant from `tunnel:open`, never implied
+//! by it).
+//!
 //! The params layout is a one-way door once a consumer exists
 //! (ADR-001): `{resource, substrate}` only — the producer owns the
 //! backing; no URL-style addressing. `deny_unknown_fields` is the
@@ -27,6 +34,17 @@ pub const TUNNEL_ALPN: &str = "alk/tunnel";
 /// The scope gating tunnel opens (ADR-006). Stable once published.
 pub const TUNNEL_OPEN_SCOPE: &str = "tunnel:open";
 
+/// The `channels/tunnel/direct` operation id (the direct op —
+/// dynamic-target egress, ADR-007; the flavor-form op-id convention,
+/// ADR-002 Amendment 1).
+pub const OP_TUNNEL_DIRECT: &str = "channels/tunnel/direct";
+
+/// The scope gating the direct op (ADR-007 §3 / ADR-006 Amendment 1).
+/// A separate grant from [`TUNNEL_OPEN_SCOPE`] — **never implied by
+/// it**: "dial `postgres-primary`" and "dial anything from my
+/// network" are different capabilities. Stable once published.
+pub const TUNNEL_DIRECT_SCOPE: &str = "tunnel:direct";
+
 /// The open-op input params (ADR-001). Wire-stable: both fields
 /// required; renames are a breaking wire change.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -52,6 +70,50 @@ pub enum Substrate {
     Unix,
 }
 
+/// The substrate-shaped address object (ADR-007 §2): keyed by the
+/// sibling `substrate` field of the params — `tcp`/`udp` →
+/// [`SubstrateAddr::HostPort`] (domain-form hosts stay unresolved on
+/// the wire — producer-side resolution, the ssh `-D` semantic),
+/// `unix` → [`SubstrateAddr::Path`]. Untagged: the shapes are
+/// mutually exclusive in practice and self-describing. Also the
+/// `peer` value shape of the forwarded op (ADR-008 §3) — defined
+/// once here.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum SubstrateAddr {
+    HostPort { host: String, port: u16 },
+    Path { path: String },
+}
+
+/// The direct-op input params (ADR-007 §2). Wire-stable: both fields
+/// required; renames are a breaking wire change. The target's
+/// shape-vs-substrate correspondence is pinned by the spec's
+/// `input_schema` (per-substrate `oneOf`, validated registry-side
+/// before the establisher) and re-checked by serde here (the
+/// backstop — a mismatch parses as the wrong variant or fails).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TunnelDirectParams {
+    /// The substrate discriminator; selects the data-plane framing
+    /// (ADR-003) and the target's expected shape.
+    pub substrate: Substrate,
+    /// The dial target, keyed by `substrate` ([`SubstrateAddr`]).
+    pub target: SubstrateAddr,
+}
+
+impl TunnelDirectParams {
+    /// Render the target to the backing-string form the `DialFn` /
+    /// registry backings use: `"host:port"` for tcp/udp, the path for
+    /// unix. The direct establisher calls this before dialing — the
+    /// dial closure stays substrate-agnostic (ADR-007 §4).
+    pub fn render_target(&self) -> String {
+        match &self.target {
+            SubstrateAddr::HostPort { host, port } => format!("{host}:{port}"),
+            SubstrateAddr::Path { path } => path.clone(),
+        }
+    }
+}
+
 /// The open-op spec (`channels/tunnel/sub`): `Sub`-typed, external,
 /// channel-open marker on the `alk/tunnel` ALPN, scope-gated by
 /// `tunnel:open` (ADR-006), input/output schemas per wire.md §The
@@ -91,6 +153,93 @@ pub fn tunnel_open_spec() -> OperationSpec {
     )
 }
 
+/// The direct-op spec (`channels/tunnel/direct`, ADR-007): `Sub`-typed,
+/// external, channel-open marker on the `alk/tunnel` ALPN, scope-gated
+/// by `tunnel:direct` (a separate grant from `tunnel:open` — never
+/// implied by it, ADR-006 Amendment 1), input/output schemas per
+/// wire.md §The Direct Op. The target sub-schema is per-substrate —
+/// `if`/`then` conditioned on the sibling `substrate` value
+/// (`tcp`/`udp` → `{host, port}`, `unix` → `{path}`) — so a malformed
+/// target fails the registry's schema validation (the typed
+/// `INVALID_INPUT` class) before the establisher runs.
+pub fn tunnel_direct_spec() -> OperationSpec {
+    OperationSpec::new(
+        OP_TUNNEL_DIRECT,
+        OperationType::Sub,
+        Visibility::External,
+        json!({
+            "type": "object",
+            "properties": {
+                "substrate": { "type": "string", "enum": ["tcp", "udp", "unix"] }
+            },
+            "required": ["substrate", "target"],
+            "allOf": [
+                {
+                    "if": {
+                        "properties": { "substrate": { "enum": ["tcp", "udp"] } },
+                        "required": ["substrate"]
+                    },
+                    "then": {
+                        "properties": {
+                            "target": {
+                                "type": "object",
+                                "properties": {
+                                    "host": { "type": "string" },
+                                    "port": {
+                                        "type": "integer",
+                                        "minimum": 0,
+                                        "maximum": 65535
+                                    }
+                                },
+                                "required": ["host", "port"],
+                                "additionalProperties": false
+                            }
+                        }
+                    }
+                },
+                {
+                    "if": {
+                        "properties": { "substrate": { "const": "unix" } },
+                        "required": ["substrate"]
+                    },
+                    "then": {
+                        "properties": {
+                            "target": {
+                                "type": "object",
+                                "properties": {
+                                    "path": { "type": "string" }
+                                },
+                                "required": ["path"],
+                                "additionalProperties": false
+                            }
+                        }
+                    }
+                }
+            ]
+        }),
+        json!({
+            "type": "object",
+            "properties": {
+                "channel_id": { "type": "integer", "minimum": 1 }
+            },
+            "required": ["channel_id"]
+        }),
+        vec![],
+        AccessControl {
+            required_scopes: vec![TUNNEL_DIRECT_SCOPE.to_string()],
+            required_scopes_any: None,
+            resource_type: None,
+            resource_action: None,
+        },
+        None,
+    )
+    .with_channel_open(ChannelOpenSpec::new(TUNNEL_ALPN))
+    .with_description(
+        "Open a tunnel channel to a caller-named target (params: {substrate, target}; \
+         dynamic-target egress)",
+    )
+}
+
 /// The establishment-failure reason code of a
 /// `channel:open_failed` error (`details.reason`): `dial_failed`,
 /// `unknown_resource`, `resource_shortage`, `handler_error`,
@@ -216,4 +365,196 @@ mod tests {
             "tcp" | "udp" | "unix" if allowed.contains(&substrate_str(params.substrate))
         )
     }
+
+    // --- direct op (ADR-007) -------------------------------------------
+
+    #[test]
+    fn direct_params_round_trip_per_substrate() {
+        let cases: Vec<(Substrate, serde_json::Value, SubstrateAddr)> = vec![
+            (
+                Substrate::Tcp,
+                json!({"host": "db.internal", "port": 5432}),
+                SubstrateAddr::HostPort {
+                    host: "db.internal".to_string(),
+                    port: 5432,
+                },
+            ),
+            (
+                Substrate::Udp,
+                json!({"host": "10.0.0.7", "port": 53}),
+                SubstrateAddr::HostPort {
+                    host: "10.0.0.7".to_string(),
+                    port: 53,
+                },
+            ),
+            (
+                Substrate::Unix,
+                json!({"path": "/run/postgresql/.s.PGSQL.5432"}),
+                SubstrateAddr::Path {
+                    path: "/run/postgresql/.s.PGSQL.5432".to_string(),
+                },
+            ),
+        ];
+        for (substrate, target, addr) in cases {
+            let params = TunnelDirectParams {
+                substrate,
+                target: addr.clone(),
+            };
+            let v = serde_json::to_value(&params).unwrap();
+            assert_eq!(
+                v,
+                json!({"substrate": substrate_str(substrate), "target": target})
+            );
+            let back: TunnelDirectParams = serde_json::from_value(v).unwrap();
+            assert_eq!(back, params);
+        }
+    }
+
+    #[test]
+    fn substrate_addr_untagged_disambiguation() {
+        let hostport: SubstrateAddr =
+            serde_json::from_value(json!({"host": "h", "port": 1})).unwrap();
+        let path: SubstrateAddr = serde_json::from_value(json!({"path": "/p"})).unwrap();
+        assert_eq!(
+            hostport,
+            SubstrateAddr::HostPort {
+                host: "h".to_string(),
+                port: 1
+            }
+        );
+        assert_eq!(
+            path,
+            SubstrateAddr::Path {
+                path: "/p".to_string()
+            }
+        );
+        // Host/port without the host field fails; an array fails.
+        assert!(serde_json::from_value::<SubstrateAddr>(json!({"port": 1})).is_err());
+        assert!(serde_json::from_value::<SubstrateAddr>(json!([1, 2])).is_err());
+    }
+
+    #[test]
+    fn direct_params_unknown_fields_rejected() {
+        let v = json!({
+            "substrate": "tcp",
+            "target": {"host": "h", "port": 1},
+            "resource": "sneaky"
+        });
+        assert!(serde_json::from_value::<TunnelDirectParams>(v).is_err());
+        // The untagged target is lenient about extra target fields
+        // (untagged variants carry no deny_unknown_fields) — a
+        // cross-shape target parses as whichever variant fits. The
+        // SCHEMA is the gate that rejects shape-vs-substrate
+        // mismatches registry-side (additionalProperties: false per
+        // branch; direct_target_schema_gates_shape_per_substrate).
+        // The establisher's serde parse is the backstop for the
+        // shape it can actually use (render_target keys off the
+        // sibling substrate).
+        let cross_shape = json!({
+            "substrate": "unix",
+            "target": {"path": "/p", "host": "h"}
+        });
+        let parsed: TunnelDirectParams = serde_json::from_value(cross_shape).unwrap();
+        assert_eq!(
+            parsed.target,
+            SubstrateAddr::Path {
+                path: "/p".to_string()
+            }
+        );
+    }
+
+    #[test]
+    fn direct_params_missing_fields_rejected() {
+        let no_target = json!({"substrate": "tcp"});
+        let no_substrate = json!({"target": {"host": "h", "port": 1}});
+        assert!(serde_json::from_value::<TunnelDirectParams>(no_target).is_err());
+        assert!(serde_json::from_value::<TunnelDirectParams>(no_substrate).is_err());
+    }
+
+    #[test]
+    fn direct_spec_matches_wire_md() {
+        let spec = tunnel_direct_spec();
+        assert_eq!(spec.name, "channels/tunnel/direct");
+        assert_eq!(spec.op_type, OperationType::Sub);
+        assert_eq!(spec.visibility, Visibility::External);
+        assert_eq!(
+            spec.input_schema["properties"]["substrate"]["enum"],
+            json!(["tcp", "udp", "unix"])
+        );
+        assert_eq!(
+            spec.input_schema["required"],
+            json!(["substrate", "target"])
+        );
+        assert_eq!(spec.output_schema["properties"]["channel_id"]["minimum"], 1);
+        assert_eq!(
+            spec.access_control.required_scopes,
+            vec![TUNNEL_DIRECT_SCOPE.to_string()]
+        );
+        let channel_open = spec.channel_open.as_ref().expect("channel-open marker");
+        assert_eq!(channel_open.alpn, TUNNEL_ALPN);
+        assert!(spec.description.is_some());
+    }
+
+    #[test]
+    fn direct_target_schema_gates_shape_per_substrate() {
+        // The oneOf/if-then structure: per-substrate target shapes are
+        // enforced by the schema itself (registry-side, before the
+        // establisher). Structural assertions per tunnels/params'
+        // precedent (no jsonschema dep here).
+        let schema = tunnel_direct_spec().input_schema;
+        let allof = schema["allOf"].as_array().expect("allOf branches");
+        assert_eq!(allof.len(), 2, "tcp/udp branch + unix branch");
+        let tcpudp_then = &allof[0]["then"]["properties"]["target"];
+        let unix_then = &allof[1]["then"]["properties"]["target"];
+        assert_eq!(
+            allof[0]["if"]["properties"]["substrate"]["enum"],
+            json!(["tcp", "udp"])
+        );
+        assert_eq!(
+            allof[1]["if"]["properties"]["substrate"]["const"],
+            json!("unix")
+        );
+        assert_eq!(
+            tcpudp_then["required"],
+            json!(["host", "port"]),
+            "tcp/udp target requires host+port"
+        );
+        assert_eq!(tcpudp_then["properties"]["port"]["maximum"], json!(65535));
+        assert_eq!(unix_then["required"], json!(["path"]));
+        assert_eq!(
+            tcpudp_then["additionalProperties"],
+            json!(false),
+            "cross-shape fields fail the schema"
+        );
+        assert_eq!(unix_then["additionalProperties"], json!(false));
+        assert_eq!(allof[0]["then"]["properties"]["target"]["type"], "object");
+        assert_eq!(allof[1]["then"]["properties"]["target"]["type"], "object");
+    }
+
+    #[test]
+    fn render_target_matches_registry_backing_form() {
+        let tcp = TunnelDirectParams {
+            substrate: Substrate::Tcp,
+            target: SubstrateAddr::HostPort {
+                host: "db.internal".to_string(),
+                port: 5432,
+            },
+        };
+        assert_eq!(tcp.render_target(), "db.internal:5432");
+        let udp = TunnelDirectParams {
+            substrate: Substrate::Udp,
+            target: SubstrateAddr::HostPort {
+                host: "10.0.0.7".to_string(),
+                port: 53,
+            },
+        };
+        assert_eq!(udp.render_target(), "10.0.0.7:53");
+        let unix = TunnelDirectParams {
+            substrate: Substrate::Unix,
+            target: SubstrateAddr::Path {
+                path: "/run/postgresql/.s.PGSQL.5432".to_string(),
+            },
+        };
+        assert_eq!(unix.render_target(), "/run/postgresql/.s.PGSQL.5432");
+    }
 }

src/producer.rs

@@ -21,7 +21,9 @@
 //!
 //! The listen establisher variant (the producer-side listener — the
 //! `-R` far-side listener shape) is [`listen_establisher`] +
-//! [`AcceptQueue`] below (the `tunnels/producer-listen` shape).
+//! [`AcceptQueue`] below (the `tunnels/producer-listen` shape). The
+//! direct op's establisher ([`direct_establisher`] — dynamic-target
+//! egress, ADR-007) is the dial shape minus the registry lookup.
 
 use std::sync::Arc;
 
@@ -35,9 +37,11 @@ use alkcall::registry::registration::OperationRegistry;
 use serde_json::Value;
 use tokio::io::{AsyncRead, AsyncWrite};
 
-use crate::params::{Substrate, TunnelParams};
+use crate::params::{Substrate, TunnelDirectParams, TunnelParams};
 
-pub use crate::params::{OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE};
+pub use crate::params::{
+    OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE, TUNNEL_OPEN_SCOPE,
+};
 
 /// The dialed target handle — the plan payload (typed-opaque
 /// `ChannelPlan`; establisher and handler agree on the concrete
@@ -189,6 +193,57 @@ fn parse_params(input: &Value) -> Result<TunnelParams, TunnelEstablishError> {
         .map_err(|e| TunnelEstablishError::HandlerError(format!("invalid params: {e}")))
 }
 
+/// The direct establisher (ADR-007 §4 — the dial shape minus the
+/// registry lookup): parse `TunnelDirectParams`, render the target to
+/// the same backing-string form the registry backings use
+/// (`"host:port"` for tcp/udp, the path for unix), and call the
+/// injected [`DialFn`] — the SAME dial function the base op uses, so
+/// a producer that registered `dial_tcp` for the base op registers
+/// the direct op with zero new substrate code (the UDP framed-adapter
+/// wrap, ADR-003, happens inside the dial closure — the establisher
+/// stays substrate-agnostic). No registry lookup — `unknown_resource`
+/// can never fire.
+///
+/// The per-call `auth` carries the opener identity (CF-006); record it
+/// into `identity_witness` when provided (the direct op is the
+/// arbitrary-egress capability — identity visibility matters more,
+/// not less).
+///
+/// Error mapping (ADR-007 §4's table minus the registry row): the
+/// serde-parse failure of a schema-valid-shaped target maps
+/// [`TunnelEstablishError::HandlerError`] (the schema gate should
+/// have caught it — the parse is the backstop); dial failures pass
+/// through the `DialFn`'s own typed errors unchanged (`dial_failed` /
+/// `resource_shortage` / `handler_error`).
+pub fn direct_establisher(dial: DialFn) -> OpenEstablisher {
+    direct_establisher_with_witness(dial, None)
+}
+
+/// [`direct_establisher`] with an optional identity witness (the test
+/// seam for the CF-006 contract, same as the base op's).
+pub fn direct_establisher_with_witness(
+    dial: DialFn,
+    identity_witness: Option<IdentityWitness>,
+) -> OpenEstablisher {
+    Arc::new(move |input: Value, auth| {
+        let dial = Arc::clone(&dial);
+        let witness = identity_witness.clone();
+        Box::pin(async move {
+            if let Some(w) = &witness {
+                *w.lock().await = auth.identity.as_ref().map(|i| i.id.clone());
+            }
+            let params: TunnelDirectParams =
+                serde_json::from_value(input.clone()).map_err(|e| {
+                    TunnelEstablishError::HandlerError(format!("invalid direct params: {e}"))
+                })?;
+            let rendered = params.render_target();
+            let substrate = params.substrate;
+            let dialed: TargetHandle = dial(substrate, &rendered).await?;
+            Ok(Establishment::new(Arc::new(dialed) as ChannelPlan))
+        })
+    })
+}
+
 /// The listen establisher (shape 2 — producer.md §The Establisher):
 /// the same open op, same params, same typed errors; the establisher
 /// pops the next accepted connection from the injected [`AcceptFn`]
@@ -381,6 +436,33 @@ pub fn register_tunnel_openable(
     )
 }
 
+/// Register the direct op (`channels/tunnel/direct`, ADR-007) on a
+/// `ChannelCore` + the session's dispatch registry: the direct spec +
+/// [`direct_establisher_with_witness`] + the SAME pump handler as the
+/// base op (the direct op's plan payload is a [`TargetHandle`] exactly
+/// like the base op's; plan-flow is unchanged), via
+/// `register_openable_with_establisher` (`timeout: None` = the 10s
+/// default bound). `dial` is the SAME [`DialFn`] the base op uses —
+/// no new substrate code. `identity_witness` is the optional CF-006
+/// probe seam (the establisher records the per-call opener identity
+/// it saw).
+pub fn register_tunnel_direct_openable(
+    core: &ChannelCore,
+    on_registry: &Arc<OperationRegistry>,
+    auth: AuthContext,
+    dial: DialFn,
+    identity_witness: Option<IdentityWitness>,
+) -> Result<(), String> {
+    core.register_openable_with_establisher(
+        crate::params::tunnel_direct_spec(),
+        Some(direct_establisher_with_witness(dial, identity_witness)),
+        make_tunnel_pump_handler(),
+        on_registry,
+        auth,
+        None,
+    )
+}
+
 /// Register the tunnel open op with a LISTEN establisher (shape 2):
 /// the same spec + pump handler as [`register_tunnel_openable`]; the
 /// establisher pops accepted handles from the injected [`AcceptFn`]

tasks/tunnels/direct-op.md

@@ -1,7 +1,7 @@
 ---
 id: tunnels/direct-op
 name: Direct op — params, spec, establisher, consumer open_direct (ADR-007)
-status: pending
+status: completed
 depends_on: []
 scope: broad
 risk: medium
@@ -146,20 +146,20 @@ Integration (extend the duplex harness from `tunnels/producer-open-op`):
 
 ## Acceptance Criteria
 
-- [ ] `TunnelDirectParams`/`SubstrateAddr` round-trip; malformed
+- [x] `TunnelDirectParams`/`SubstrateAddr` round-trip; malformed
       targets rejected (serde + schema shape pinned)
-- [ ] `tunnel_direct_spec()` matches wire.md §The Direct Op exactly
-- [ ] `direct_establisher`: no registry lookup, target rendering into
+- [x] `tunnel_direct_spec()` matches wire.md §The Direct Op exactly
+- [x] `direct_establisher`: no registry lookup, target rendering into
       the `DialFn` backing string, typed errors per ADR-007 §4
-- [ ] `register_tunnel_direct_openable` reuses the base pump handler +
+- [x] `register_tunnel_direct_openable` reuses the base pump handler +
       dial injection; witness seam present
-- [ ] `TunnelSession::open_direct` (separate ctor; identical
+- [x] `TunnelSession::open_direct` (separate ctor; identical
       session/teardown semantics)
-- [ ] Integration tests: e2e round-trips (tcp + udp), malformed-target
+- [x] Integration tests: e2e round-trips (tcp + udp), malformed-target
       rejection, scope separation (FORBIDDEN), NOT_FOUND posture,
       per-call identity witness
-- [ ] Clippy/fmt clean; wasm32 check passes (no substrate code added)
-- [ ] `cargo test` green
+- [x] Clippy/fmt clean; wasm32 check passes (no substrate code added)
+- [x] `cargo test` green
 
 ## References
 
@@ -172,8 +172,77 @@ Integration (extend the duplex harness from `tunnels/producer-open-op`):
 
 ## Notes
 
-> Agent fills during implementation.
+- The `input_schema` pins the per-substrate target sub-schema as two
+  `if`/`then` branches under `allOf`, keyed on the sibling
+  `substrate` value (tcp/udp → `{host, port}` with
+  `additionalProperties: false`; unix → `{path}`, same posture) —
+  JSON Schema `if/then` composes through alkcall's jsonschema 0.46
+  compile, and the integration tests prove the gate fires
+  (`INVALID_INPUT` before the establisher). Structural assertions
+  pin the branch shape per `tunnels/params`'s no-jsonschema-dep
+  precedent.
+- Untagged `SubstrateAddr` is deliberately lenient about extra target
+  fields (untagged variants carry no `deny_unknown_fields`): a
+  cross-shape target parses as whichever variant fits. The schema is
+  the shape-vs-substrate gate; the establisher's serde parse is the
+  backstop only (unknown substrate value → `handler_error`,
+  `direct_establisher` maps `TunnelEstablishError::HandlerError`).
+  This asymmetry is documented in both the type docs and the test.
+- `TunnelDirectParams::render_target()` centralizes the backing-string
+  form (`"host:port"` / path) — the establisher injects the SAME
+  `DialFn` with the rendered string; the render is pinned by test to
+  the registry-backing form.
+- `TunnelSession::open_direct` uses plain `open_channel` (no reply
+  fields — dial establishers never bind); `SubstrateAddr` re-exports
+  as the future forwarded-op `peer` shape (ADR-008 §3, defined once).
+- Harness: `RegistrationMode::Direct`, `wire_direct` /
+  `wire_direct_with_identity` (the raw-call topology),
+  `wire_direct_forward` / `wire_direct_forward_with_identity` (the
+  session-level forward topology; identity parameterized so the
+  session-level FORBIDDEN probe runs through the real client API),
+  `direct_identity` / `both_scopes_identity`.
 
 ## Summary
 
-> Agent fills this on completion.
+Implemented the direct op end to end per ADR-007:
+
+- **params.rs** — `OP_TUNNEL_DIRECT` / `TUNNEL_DIRECT_SCOPE`
+  constants; `SubstrateAddr` (untagged `{host, port}` | `{path}`,
+  the ADR-008 §3 `peer` shape too); `TunnelDirectParams`
+  (`deny_unknown_fields`, `render_target()`); `tunnel_direct_spec()`
+  (`Sub`-typed, External, `ChannelOpenSpec::new(TUNNEL_ALPN)`, scope
+  `["tunnel:direct"]`, output `{channel_id: integer > 0}`,
+  per-substrate `if/then` target sub-schema). 7 unit tests: round-
+  trips per substrate, untagged disambiguation, loud unknown-field
+  rejection, spec-shape conformance, schema structure.
+- **producer.rs** — `direct_establisher` /
+  `direct_establisher_with_witness` (parse → render → injected
+  `DialFn`; no registry lookup; typed-error mapping per ADR-007 §4
+  minus the registry row; CF-006 witness seam) +
+  `register_tunnel_direct_openable` (direct spec + establisher + the
+  SAME `make_tunnel_pump_handler`, plan-flow unchanged).
+- **consumer.rs** — `TunnelSession::open_direct(client, target,
+  substrate)`: separate constructor, serializes `{substrate, target}`,
+  plain `open_channel(OP_TUNNEL_DIRECT, …, TUNNEL_ALPN)`,
+  `from_halves` — identical session/data-planes/teardown as `open`.
+- **lib.rs** — the new surface re-exported.
+
+Tests: `tests/producer_direct_op.rs` — 15 integration tests over the
+duplex harness: tcp + udp e2e round-trips through `pump_bidi`
+(datagram boundary preserved via the framed dial closure), malformed
+targets (`unix` without `path`, `tcp` without `port`, missing target)
+→ `INVALID_INPUT` with no phantom channel/session, dial-failure
+pass-through (`dial_failed`), NOT_FOUND on a producer without the op,
+scope separation both ways (`tunnel:open` ↛ `tunnel:direct` and
+`tunnel:direct` ↛ `tunnel:open` — FORBIDDEN, raw-call AND session-
+level), CF-006 witness on the direct establisher (raw + session
+paths), the establisher parse backstop (`handler_error`), and the
+target-rendering pin (the dial closure observes the exact
+backing-string form). Harness: `Direct` registration mode +
+`wire_direct*` topologies + `direct_identity`/`both_scopes_identity`.
+
+Verified: `cargo test` green (93 native; 94 with `--all-features`),
+clippy `-D warnings` clean (native + wasm32), fmt clean, wasm32
+check passes (default crate stays wasm-clean — no substrate code
+added), `cargo doc --no-deps` clean. CHANGELOG updated (Unreleased →
+Added).

tests/harness.rs

@@ -34,8 +34,10 @@ use alkcall::protocol::connection::{split_single_stream, CallConnection};
 use alkcall::protocol::dispatch::Dispatcher;
 use alkcall::registry::registration::OperationRegistry;
 
-use alktunnels::params::{Substrate, TUNNEL_OPEN_SCOPE};
-use alktunnels::producer::{register_tunnel_openable, ResourceRegistry};
+use alktunnels::params::{Substrate, TUNNEL_DIRECT_SCOPE, TUNNEL_OPEN_SCOPE};
+use alktunnels::producer::{
+    register_tunnel_direct_openable, register_tunnel_openable, ResourceRegistry,
+};
 
 /// The consumer identity the transport carries (the mTLS/QUIC
 /// analogue — key-based identity resolved out-of-band, attached to
@@ -48,6 +50,29 @@ pub fn consumer_identity() -> Identity {
     }
 }
 
+/// The direct-egress identity: `tunnel:direct` ONLY (no
+/// `tunnel:open`) — the scope-separation probe's identity.
+pub fn direct_identity() -> Identity {
+    Identity {
+        id: "consumer-direct".to_string(),
+        scopes: vec![TUNNEL_DIRECT_SCOPE.to_string()],
+        resources: std::collections::HashMap::new(),
+    }
+}
+
+/// Both scopes: `tunnel:open` AND `tunnel:direct` (a deployment
+/// granting both grants both explicitly — ADR-006 Amendment 1).
+pub fn both_scopes_identity() -> Identity {
+    Identity {
+        id: "consumer-both".to_string(),
+        scopes: vec![
+            TUNNEL_OPEN_SCOPE.to_string(),
+            TUNNEL_DIRECT_SCOPE.to_string(),
+        ],
+        resources: std::collections::HashMap::new(),
+    }
+}
+
 /// The scoped token (the hub-forwarding path) — resolves to
 /// [`consumer_identity`] via [`TestIdProvider`].
 pub const TEST_AUTH_TOKEN: &str = "test-token-with-tunnel-open-scope";
@@ -105,6 +130,12 @@ pub enum RegistrationMode {
     /// The no-witness dial establisher (`tunnel_establisher` — the
     /// public default shape, no CF-006 probe seam).
     DialNoWitness(alktunnels::producer::DialFn),
+    /// The direct op (`channels/tunnel/direct`, ADR-007): the direct
+    /// spec + the direct establisher (no registry lookup) over the
+    /// SAME injected dial, with the CF-006 witness seam. The witness
+    /// rides the topology's `identity_witness` (the direct op's
+    /// witness and the base op's never mix in one test).
+    Direct(alktunnels::producer::DialFn),
 }
 
 pub async fn wire_with(
@@ -212,6 +243,16 @@ pub async fn wire_with(
             )
             .expect("register tunnel openable");
         }
+        RegistrationMode::Direct(dial) => {
+            register_tunnel_direct_openable(
+                &core,
+                &producer_op_registry,
+                AuthContext::anonymous(b"alk/tunnel"),
+                dial,
+                Some(Arc::clone(&identity_witness)),
+            )
+            .expect("register tunnel direct openable");
+        }
         RegistrationMode::Listen(accept) => {
             alktunnels::producer::register_tunnel_listen_openable(
                 &core,
@@ -308,6 +349,33 @@ pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn
     .await
 }
 
+/// The direct-op topology (ADR-007): the direct op registered with
+/// the CF-006 witness; the consumer's transport identity is
+/// [`both_scopes_identity`] (both grants — the opens authorize).
+pub async fn wire_direct(dial: alktunnels::producer::DialFn) -> Topology {
+    wire_with(
+        ResourceRegistry::new(),
+        RegistrationMode::Direct(dial),
+        Some(both_scopes_identity()),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await
+}
+
+/// [`wire_direct`] with an explicit transport identity (the
+/// scope-separation probes construct their own identities).
+pub async fn wire_direct_with_identity(identity: Identity) -> Topology {
+    wire_with(
+        ResourceRegistry::new(),
+        RegistrationMode::Direct(failing_dial("unused")),
+        Some(identity),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await
+}
+
 /// A topology with the consumer's transport identity REPLACED by the
 /// serving-side override (`ServingConfig.identity` — CF-005 (a) probe
 /// shape) or stripped entirely (the fail-closed probe). The transport
@@ -472,6 +540,107 @@ pub async fn wire_forward(
     }
 }
 
+/// The DIRECT-op forward topology (ADR-007): the producer (accept
+/// side, serving) registers the direct op — [`register_tunnel_direct_openable`]
+/// over the injected dial, CF-006 witness attached — and the consumer
+/// (connect side, pure client) holds the identity the session tests
+/// authorize with. This is the topology `TunnelSession::open_direct`
+/// is normative for.
+pub async fn wire_direct_forward(dial: alktunnels::producer::DialFn) -> ForwardTopology {
+    wire_direct_forward_with_identity(dial, both_scopes_identity()).await
+}
+
+/// [`wire_direct_forward`] with an explicit transport identity (the
+/// session-level scope-separation probes construct their own).
+pub async fn wire_direct_forward_with_identity(
+    dial: alktunnels::producer::DialFn,
+    identity: Identity,
+) -> ForwardTopology {
+    let producer_op_registry = Arc::new(OperationRegistry::new());
+    let (producer_manager_tx, mut producer_manager_rx) =
+        tokio::sync::mpsc::channel::<ChannelManager>(1);
+    let identity_witness = Arc::new(tokio::sync::Mutex::new(None::<String>));
+    let dial_witness = Arc::clone(&identity_witness);
+    let expected_caller = identity.clone();
+    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
+        let registry_arc = Arc::clone(&producer_op_registry);
+        let dial = Arc::clone(&dial);
+        let witness = Arc::clone(&dial_witness);
+        let manager_tx = producer_manager_tx.clone();
+        let caller = expected_caller.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);
+            // The accept-side transport-identity posture: the transport
+            // authenticated the dialer; attach the RESOLVED identity
+            // (scopes included — resolved out-of-band) to channel 0.
+            let _ = channel0_conn.set_identity(caller);
+            let call_connection = Arc::new(CallConnection::new_single_stream(
+                channel0_conn,
+                Arc::clone(&writer),
+            ));
+            let core = alkcall::channels::operations::ChannelCore::new(
+                manager.clone(),
+                alkcall::channels::policy::default_policy(),
+            );
+            let ops = ChannelOperations::with_default_policy(manager.clone());
+            ops.register_on(&registry_arc)
+                .expect("register channel ops on producer");
+            register_tunnel_direct_openable(
+                &core,
+                &registry_arc,
+                AuthContext::anonymous(b"alk/tunnel"),
+                dial,
+                Some(witness),
+            )
+            .expect("register tunnel direct openable");
+            let _ = manager_tx.send(manager).await;
+            let dp = Dispatcher::new(
+                registry_arc,
+                Arc::new(alkcall::core::auth::NoopIdentityProvider),
+            );
+            dp.serve_single_stream(call_connection, reader, writer)
+                .await;
+        })
+    });
+
+    let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
+    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
+    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
+    consumer_conn
+        .set_identity(identity)
+        .expect("transport identity set once");
+
+    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
+    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
+    let _adapter_task = tokio::spawn(async move {
+        let _ = alkcall::core::types::ProtocolHandler::handle(
+            &adapter,
+            producer_conn,
+            &producer_transport_auth,
+        )
+        .await;
+    });
+
+    let consumer_client = ChannelClient::from_connection(consumer_conn)
+        .await
+        .expect("consumer channel client init");
+    let consumer_client = Arc::new(consumer_client);
+    let producer_manager = producer_manager_rx
+        .recv()
+        .await
+        .expect("producer manager captured");
+
+    ForwardTopology {
+        consumer: consumer_client,
+        producer_manager,
+        identity_witness,
+    }
+}
+
 /// An echo dial closure (the ADR-004 injection point under test): an
 /// in-process pipe per dial — the "target" echoes bytes until EOF.
 /// The consumer side gets split `DuplexStream` halves; a spawned task

tests/producer_direct_op.rs

@@ -0,0 +1,558 @@
+//! Integration tests for the direct op (`channels/tunnel/direct`,
+//! ADR-007) end to end over the duplex harness: the params → spec →
+//! establisher → registration → `TunnelSession::open_direct` chain.
+//!
+//! Covered: e2e round-trips through `pump_bidi` (tcp + udp — the
+//! framed adapter wraps at the dial closure, datagram boundary
+//! preserved), the schema-class failure for malformed targets (no
+//! phantom session), the `NOT_FOUND` posture on a producer without
+//! the direct op registered, the structural scope separation
+//! (`tunnel:open` does NOT imply `tunnel:direct` — FORBIDDEN, and
+//! vice versa), and the CF-006 per-call opener identity on the
+//! direct establisher (witness).
+
+mod harness;
+
+use std::sync::Arc;
+
+use alkcall::protocol::wire::CallError;
+use alktunnels::params::{Substrate, SubstrateAddr};
+use alktunnels::{TunnelEstablishError, TunnelSession};
+
+use harness::{
+    all_substrate_dial, consumer_identity, echo_dial, failing_dial, framed_udp_echo_dial, wire,
+    wire_direct, wire_direct_forward, wire_direct_with_identity, Topology,
+};
+
+fn target_hostport(host: &str, port: u16) -> SubstrateAddr {
+    SubstrateAddr::HostPort {
+        host: host.to_string(),
+        port,
+    }
+}
+
+/// Call the producer's direct op on channel 0 (the raw call surface),
+/// returning the wire `CallError` on failure.
+async fn call_direct(
+    topo: &Topology,
+    input: serde_json::Value,
+) -> Result<serde_json::Value, CallError> {
+    let payload = serde_json::json!({
+        "operationId": "channels/tunnel/direct",
+        "input": input,
+    });
+    let response = topo.consumer_call.call_with_payload(payload).await;
+    response.result
+}
+
+/// Adopt the producer-allocated channel and pump it against in-process
+/// halves (the consumer-side data plane, driven manually).
+async fn adopt_and_pump(
+    topo: &Topology,
+    channel_id: u32,
+) -> (
+    tokio::io::ReadHalf<tokio::io::DuplexStream>,
+    tokio::io::WriteHalf<tokio::io::DuplexStream>,
+    tokio::task::JoinHandle<(u64, u64)>,
+) {
+    let (send, recv) = topo
+        .consumer_manager
+        .adopt_channel(channel_id, "alk/tunnel", None)
+        .await
+        .expect("adopt producer-allocated channel");
+    let bidi = alkcall::core::types::BiStream::from_joined(recv, send);
+    let (channel_end, local_end) = tokio::io::duplex(64 * 1024);
+    let (c_read, c_write) = tokio::io::split(channel_end);
+    let pump = tokio::spawn(alkcall::channels::pump::pump_bidi(bidi, c_read, c_write));
+    let (l_read, l_write) = tokio::io::split(local_end);
+    (l_read, l_write, pump)
+}
+
+/// Only channel 0 may remain — a failed open leaves no data channel
+/// on either side (the phantom-channel property).
+fn no_data_channels(topo: &Topology) -> bool {
+    let consumer = topo.consumer_manager.channel_ids();
+    let producer = topo.producer.manager().channel_ids();
+    consumer.iter().all(|&id| id == 0) && producer.iter().all(|&id| id == 0)
+}
+
+fn establishment_reason_of(err: &CallError) -> Option<&str> {
+    err.details
+        .as_ref()
+        .and_then(|d| d.get("reason"))
+        .and_then(|r| r.as_str())
+}
+
+#[tokio::test]
+async fn direct_tcp_round_trips_through_pump_bidi() {
+    let topo = wire_direct(echo_dial()).await;
+
+    let out = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "tcp",
+            "target": {"host": "db.internal", "port": 5432}
+        }),
+    )
+    .await
+    .expect("direct open must succeed");
+    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
+    assert!(topo.producer.manager().has_channel(channel_id));
+
+    let (mut read, mut write, pump) = adopt_and_pump(&topo, channel_id).await;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    write.write_all(b"ping-direct").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 11];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip timed out")
+        .expect("read");
+    assert_eq!(&buf, b"ping-direct");
+
+    // EOF propagates and the wrapper reaps the producer-side channel.
+    drop(write);
+    drop(read);
+    let (c2p, p2c) = tokio::time::timeout(std::time::Duration::from_secs(5), pump)
+        .await
+        .expect("pump completion timed out")
+        .expect("pump task");
+    assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}");
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(
+        !topo.producer.manager().has_channel(channel_id),
+        "wrapper reaped the producer-side channel after pump completion"
+    );
+}
+
+#[tokio::test]
+async fn direct_udp_round_trips_framed() {
+    // The UDP dial closure wraps the framed adapter (ADR-003) at the
+    // boundary — the establisher stays substrate-agnostic; the
+    // session's datagram plane sees boundary-preserved frames.
+    let topo = wire_direct(all_substrate_dial()).await;
+
+    let out = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "udp",
+            "target": {"host": "10.0.0.7", "port": 53}
+        }),
+    )
+    .await
+    .expect("udp direct open must succeed");
+    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
+
+    let mut session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Udp,
+        "alk/tunnel",
+    )
+    .await
+    .expect("adopt into a udp session");
+    session
+        .send_datagram(b"datagram-direct")
+        .await
+        .expect("send");
+    let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("round trip timed out")
+        .expect("recv")
+        .expect("datagram");
+    assert_eq!(dg.as_ref(), b"datagram-direct");
+    session.close().await;
+}
+
+#[tokio::test]
+async fn malformed_target_is_schema_class_and_leaves_no_channel() {
+    // The registry's input-schema gate runs before the establisher: a
+    // unix target without `path` (and a tcp target without `port`)
+    // fails as `INVALID_INPUT` — never a phantom session, never a
+    // establisher dial.
+    let topo = wire_direct(echo_dial()).await;
+
+    let err = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "unix",
+            "target": {"host": "nope", "port": 1}
+        }),
+    )
+    .await
+    .expect_err("unix target without path must fail the schema gate");
+    assert_eq!(err.code, "INVALID_INPUT");
+    assert!(establishment_reason_of(&err).is_none());
+    assert!(no_data_channels(&topo));
+
+    let err = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "tcp",
+            "target": {"host": "nope"}
+        }),
+    )
+    .await
+    .expect_err("tcp target without port must fail the schema gate");
+    assert_eq!(err.code, "INVALID_INPUT");
+    assert!(no_data_channels(&topo));
+
+    // Missing target entirely.
+    let err = call_direct(&topo, serde_json::json!({"substrate": "tcp"}))
+        .await
+        .expect_err("missing target must fail the schema gate");
+    assert_eq!(err.code, "INVALID_INPUT");
+    assert!(no_data_channels(&topo));
+}
+
+#[tokio::test]
+async fn dial_failure_passes_through_typed() {
+    // The dial closure's own typed errors pass through unchanged
+    // (ADR-007 §4's table minus the registry row).
+    let topo = wire_direct(failing_dial("no route to target")).await;
+
+    let err = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "tcp",
+            "target": {"host": "blackhole.internal", "port": 9}
+        }),
+    )
+    .await
+    .expect_err("failed dial must fail the open");
+    assert_eq!(err.code, "channel:open_failed");
+    assert_eq!(establishment_reason_of(&err), Some("dial_failed"));
+    assert!(no_data_channels(&topo));
+}
+
+#[tokio::test]
+async fn producer_without_direct_op_rejects_loudly() {
+    // An old producer (only the base op registered) rejects the
+    // unknown op loudly — NOT_FOUND, the SSH "unknown channel type"
+    // posture.
+    let registry = alktunnels::producer::ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+    assert!(topo
+        .producer_registry
+        .registration("channels/tunnel/direct")
+        .is_none());
+
+    let err = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "tcp",
+            "target": {"host": "db.internal", "port": 5432}
+        }),
+    )
+    .await
+    .expect_err("unknown op must be rejected loudly");
+    assert_eq!(err.code, "NOT_FOUND");
+    assert!(no_data_channels(&topo));
+}
+
+#[tokio::test]
+async fn tunnel_open_scope_does_not_imply_tunnel_direct() {
+    // ADR-006 Amendment 1: the scopes are structurally separate
+    // grants — an identity holding `tunnel:open` (but not
+    // `tunnel:direct`) is FORBIDDEN on the direct op.
+    let topo = wire_direct_with_identity(consumer_identity()).await;
+
+    let err = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "tcp",
+            "target": {"host": "db.internal", "port": 5432}
+        }),
+    )
+    .await
+    .expect_err("tunnel:open must NOT authorize the direct op");
+    assert_eq!(err.code, "FORBIDDEN");
+    assert!(no_data_channels(&topo));
+}
+
+#[tokio::test]
+async fn direct_scope_does_not_imply_tunnel_open() {
+    // The separation is symmetric: `tunnel:direct` does not authorize
+    // the base open op either (two capabilities; a deployment
+    // granting both grants both explicitly). A tunnel:direct-only
+    // identity is FORBIDDEN on `channels/tunnel/sub` even though the
+    // producer registered it.
+    let registry = alktunnels::producer::ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = harness::wire_with(
+        registry,
+        harness::RegistrationMode::Dial(echo_dial()),
+        Some(harness::direct_identity()),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await;
+    let payload = serde_json::json!({
+        "operationId": "channels/tunnel/sub",
+        "input": {"resource": "echo", "substrate": "tcp"},
+    });
+    let response = topo.consumer_call.call_with_payload(payload).await;
+    let err = response
+        .result
+        .expect_err("tunnel:direct must NOT authorize the base op");
+    assert_eq!(err.code, "FORBIDDEN");
+    assert!(no_data_channels(&topo));
+}
+
+#[tokio::test]
+async fn direct_establisher_witnesses_per_call_identity() {
+    // CF-006: the direct establisher's per-call auth carried the END
+    // CALLER's identity (the direct op is the arbitrary-egress
+    // capability — identity visibility matters more, not less).
+    let topo = wire_direct(echo_dial()).await;
+
+    let out = call_direct(
+        &topo,
+        serde_json::json!({
+            "substrate": "tcp",
+            "target": {"host": "db.internal", "port": 5432}
+        }),
+    )
+    .await
+    .expect("direct open authorized (both-scopes identity)");
+    assert!(out["channel_id"].is_u64());
+
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(
+        witness.as_deref(),
+        Some("consumer-both"),
+        "direct establisher saw the per-call opener identity (CF-006)"
+    );
+}
+
+#[tokio::test]
+async fn establisher_parse_backstop_maps_handler_error() {
+    // The establisher's serde parse is the backstop behind the schema
+    // gate: feed an input the schema would have rejected but a direct
+    // establisher call never schema-checks (the raw-call shape —
+    // assembly variants, or a registry whose schema gate is looser)
+    // and observe the typed `handler_error` mapping (ADR-007 §4).
+    // An unknown substrate value fails the enum parse.
+    let establisher = alktunnels::producer::direct_establisher(echo_dial());
+    let err = establisher(
+        serde_json::json!({
+            "substrate": "sctp",
+            "target": {"host": "h", "port": 1}
+        }),
+        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
+    )
+    .await;
+    match err {
+        Err(alkcall::channels::operations::EstablishmentError::HandlerError { message }) => {
+            assert!(message.contains("invalid direct params"), "{message}");
+        }
+        other => panic!("expected HandlerError backstop, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn direct_establisher_renders_target_to_backing_string() {
+    // The design pin: the establisher renders the target to the SAME
+    // backing-string form the registry backings use and calls the
+    // injected DialFn with it — the dial closure observes the exact
+    // rendered form ("host:port" / path), per ADR-007 §4.
+    let observed: Arc<tokio::sync::Mutex<Vec<(Substrate, String)>>> =
+        Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let obs = Arc::clone(&observed);
+    let dial: alktunnels::producer::DialFn =
+        Arc::new(move |substrate: Substrate, backing: &str| {
+            let obs = Arc::clone(&obs);
+            let backing = backing.to_string();
+            Box::pin(async move {
+                obs.lock().await.push((substrate, backing));
+                Err(TunnelEstablishError::DialFailed("stop here".to_string()))
+            })
+        });
+    let establisher = alktunnels::producer::direct_establisher(dial);
+    let auth = alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel");
+
+    let _ = establisher(
+        serde_json::json!({"substrate": "tcp", "target": {"host": "db.internal", "port": 5432}}),
+        auth.clone(),
+    )
+    .await;
+    let _ = establisher(
+        serde_json::json!({"substrate": "unix", "target": {"path": "/run/pg/.s.PGSQL.5432"}}),
+        auth,
+    )
+    .await;
+
+    let seen = observed.lock().await.clone();
+    assert_eq!(
+        seen,
+        vec![
+            (Substrate::Tcp, "db.internal:5432".to_string()),
+            (Substrate::Unix, "/run/pg/.s.PGSQL.5432".to_string()),
+        ],
+        "targets rendered to the registry backing-string form"
+    );
+}
+
+// --- consumer session level (TunnelSession::open_direct) ---------------
+
+#[tokio::test]
+async fn open_direct_session_round_trip_and_teardown() {
+    let topo = wire_direct_forward(echo_dial()).await;
+
+    let mut session = TunnelSession::open_direct(
+        &topo.consumer,
+        target_hostport("db.internal", 5432),
+        Substrate::Tcp,
+    )
+    .await
+    .expect("open_direct");
+    assert!(topo.producer_manager.has_channel(session.channel_id));
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (read, write) = session.stream_halves().expect("stream session has halves");
+    write.write_all(b"ping-direct").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 11];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip")
+        .expect("read");
+    assert_eq!(&buf, b"ping-direct");
+
+    // Teardown semantics identical to `open` (ADR-007 §5).
+    let channel_id = session.channel_id;
+    let reaped = session.close().await;
+    assert!(reaped, "close reaped the adopted entry");
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(
+        !topo.producer_manager.has_channel(channel_id),
+        "producer side reaped after pump completion"
+    );
+}
+
+#[tokio::test]
+async fn open_direct_session_datagram_boundary_preserved() {
+    let topo = wire_direct_forward(framed_udp_echo_dial()).await;
+
+    let mut session = TunnelSession::open_direct(
+        &topo.consumer,
+        target_hostport("10.0.0.7", 53),
+        Substrate::Udp,
+    )
+    .await
+    .expect("udp open_direct");
+    session.send_datagram(b"one").await.expect("send");
+    session.send_datagram(b"").await.expect("send empty");
+    let first = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("round trip")
+        .expect("recv")
+        .expect("datagram");
+    assert_eq!(first.as_ref(), b"one");
+    let second = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("round trip")
+        .expect("recv")
+        .expect("datagram");
+    assert!(
+        second.is_empty(),
+        "the empty datagram crosses framed (F-2 layering)"
+    );
+    session.close().await;
+}
+
+#[tokio::test]
+async fn open_direct_malformed_target_no_phantom_session() {
+    // The session-level failure surface: a malformed target is the
+    // schema/invalid_input-class failure (registry-side); a failed
+    // open never yields a session.
+    let topo = wire_direct_forward(echo_dial()).await;
+
+    let err = match TunnelSession::open_direct(
+        &topo.consumer,
+        SubstrateAddr::Path {
+            path: "/nope".to_string(),
+        },
+        Substrate::Tcp,
+    )
+    .await
+    {
+        Err(e) => e,
+        Ok(_) => panic!("path-shaped target on tcp must fail"),
+    };
+    let call = err.open_ref().call_error().expect("wire error");
+    assert_eq!(call.code, "INVALID_INPUT");
+
+    // Both managers hold only channel 0 (the phantom-channel property).
+    assert!(
+        topo.consumer
+            .manager()
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0)
+            && topo
+                .producer_manager
+                .channel_ids()
+                .iter()
+                .all(|&id| id == 0),
+        "no phantom session"
+    );
+}
+
+#[tokio::test]
+async fn open_direct_scope_separated_session_level() {
+    // Session-level scope separation (ADR-006 Amendment 1): a
+    // tunnel:open-only consumer client is FORBIDDEN on open_direct —
+    // the scopes are separate grants, and the session constructor
+    // states which capability it invokes.
+    let topo = harness::wire_direct_forward_with_identity(echo_dial(), consumer_identity()).await;
+
+    let err = match TunnelSession::open_direct(
+        &topo.consumer,
+        target_hostport("db.internal", 5432),
+        Substrate::Tcp,
+    )
+    .await
+    {
+        Err(e) => e,
+        Ok(_) => panic!("tunnel:open-only identity must NOT open_direct"),
+    };
+    let call = err.open_ref().call_error().expect("wire error");
+    assert_eq!(call.code, "FORBIDDEN");
+    assert!(
+        topo.consumer
+            .manager()
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0)
+            && topo
+                .producer_manager
+                .channel_ids()
+                .iter()
+                .all(|&id| id == 0),
+        "no phantom session"
+    );
+}
+
+#[tokio::test]
+async fn open_direct_witness_carries_session_identity() {
+    let topo = wire_direct_forward(echo_dial()).await;
+    let _session = TunnelSession::open_direct(
+        &topo.consumer,
+        target_hostport("db.internal", 5432),
+        Substrate::Tcp,
+    )
+    .await
+    .expect("open_direct");
+    let witness = topo.identity_witness.lock().await.clone();
+    assert_eq!(
+        witness.as_deref(),
+        Some("consumer-both"),
+        "the session path's establisher saw the per-call opener identity"
+    );
+}