feat: the forwarded op's params + the bind-first plan type (ADR-008 as amended)

- params.rs: OP_TUNNEL_FORWARDED / TUNNEL_FORWARDED_SCOPE; TunnelForwardedParams
  {listen, substrate, peer} (deny-unknown; listen positive via schema minimum 1 +
  a serde pin); tunnel_forwarded_spec() (Sub, External, alk/tunnel ALPN, scope
  tunnel:forwarded, per-substrate if/then peer sub-schema, no establisher);
  TunnelListenReply + parse_listen_reply (the additive bound/listen reply
  projection — unknown reply fields tolerated, malformed ones fail loud)
- error.rs: TunnelError::MalformedReply (the viewer's fail-loud path)
- producer.rs: BindFirstPlan {listener: Arc<dyn Any + Send + Sync>, token: u32}
  — the opaque assembly-typed listener + the correlation token; the mint
  discipline pinned in docs (non-zero-seeded AtomicU32, per-open fresh,
  per-producer unique, wraps acceptable)
- Wire-stable types pin first (the tunnels/params ordering); the establisher +
  accept dispatcher + registration land with tunnels/bindfirst-producer

Verified: cargo test (105 default; 125 --all-features), clippy --all-targets
-D warnings (native + wasm32), fmt --check, wasm32-unknown-unknown check,
cargo doc --no-deps — all clean.

CHANGELOG.md

@@ -14,6 +14,24 @@ dependency bump below is the only behavior-affecting change so far
 
 ### Added
 
+- **The forwarded op + the bind-first params surface (ADR-008 as
+  amended)** — the wire-stable types pin first (the ordering
+  `tunnels/params` used for the base op): `OP_TUNNEL_FORWARDED` /
+  `TUNNEL_FORWARDED_SCOPE` constants; `TunnelForwardedParams
+  {listen, substrate, peer}` (deny-unknown; `listen` is the
+  producer-minted correlation token, positive by wire contract —
+  schema `minimum: 1` + a serde-level pin) with a per-substrate
+  `peer` sub-schema (malformed peers fail the registry's schema gate
+  before the opener-side handler); `tunnel_forwarded_spec()`
+  (`Sub`-typed, External, `alk/tunnel` ALPN marker, no establisher —
+  the forwarded data plane starts immediately); `TunnelListenReply` +
+  `parse_listen_reply` (the typed projection of the bind-first
+  listen open's additive `bound`/`listen` reply fields — unknown
+  reply fields tolerated, malformed ones fail loud via the new
+  `TunnelError::MalformedReply`); `BindFirstPlan` (the opaque
+  assembly-typed listener + the token — the plan payload the
+  bind-first establisher/accept-dispatcher task consumes).
+
 - **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

src/error.rs

@@ -20,6 +20,13 @@ pub enum TunnelError {
     /// An I/O failure on the session's data plane.
     #[error("io: {0}")]
     Io(#[from] std::io::Error),
+    /// A reply field failed its typed projection — the wire carried a
+    /// malformed shape (e.g. [`crate::params::parse_listen_reply`]
+    /// found `bound` not substrate-shaped, a non-positive `listen`,
+    /// or a reply with no `channel_id` at all). A misbehaving
+    /// producer is never silently coerced.
+    #[error("malformed reply: {0}")]
+    MalformedReply(String),
 }
 
 /// A failed tunnel open or adopt. A failure never yields a session

src/lib.rs

@@ -30,14 +30,15 @@ pub mod local;
 pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession};
 pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError};
 pub use params::{
-    establishment_reason, tunnel_direct_spec, SubstrateAddr, TunnelDirectParams, OP_TUNNEL_DIRECT,
-    TUNNEL_DIRECT_SCOPE,
+    establishment_reason, parse_listen_reply, tunnel_direct_spec, tunnel_forwarded_spec,
+    SubstrateAddr, TunnelDirectParams, TunnelForwardedParams, TunnelListenReply, OP_TUNNEL_DIRECT,
+    OP_TUNNEL_FORWARDED, TUNNEL_DIRECT_SCOPE, TUNNEL_FORWARDED_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,
+    register_tunnel_listen_openable, AcceptFn, AcceptQueue, BindFirstPlan, DialFn,
+    ResourceRegistry, TargetHandle, TunnelEstablishError,
 };
 pub use wire::MAX_DATAGRAM_LEN;

src/params.rs

@@ -10,6 +10,16 @@
 //! `tunnel:direct` (a separate grant from `tunnel:open`, never implied
 //! by it).
 //!
+//! The forwarded op's params ([`TunnelForwardedParams`], ADR-008 §3
+//! as amended) and its spec builder live here as well:
+//! `{listen, substrate, peer}` — the accept-as-open ingress op, keyed
+//! by the producer-minted correlation token (Amendment 1), scope-gated
+//! by `tunnel:forwarded` (a separate grant again). The bind-first
+//! listen open's reply projection ([`TunnelListenReply`] +
+//! [`parse_listen_reply`]) is the typed view of the two additive
+//! reply fields (`bound`, `listen`) the bind-first establisher
+//! contributes.
+//!
 //! 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
@@ -22,7 +32,9 @@ use alkcall::registry::spec::{
     AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility,
 };
 use serde::{Deserialize, Serialize};
-use serde_json::json;
+use serde_json::{json, Value};
+
+use crate::error::TunnelError;
 
 /// The `channels/tunnel/sub` operation id (the open op).
 pub const OP_TUNNEL_OPEN: &str = "channels/tunnel/sub";
@@ -45,6 +57,18 @@ pub const OP_TUNNEL_DIRECT: &str = "channels/tunnel/direct";
 /// network" are different capabilities. Stable once published.
 pub const TUNNEL_DIRECT_SCOPE: &str = "tunnel:direct";
 
+/// The `channels/tunnel/forwarded` operation id (the forwarded op —
+/// accept-as-open ingress, ADR-008 §3; the flavor-form op-id
+/// convention, ADR-002 Amendment 1).
+pub const OP_TUNNEL_FORWARDED: &str = "channels/tunnel/forwarded";
+
+/// The scope gating the forwarded op (ADR-008 §3 / ADR-006 Amendment
+/// 1). A separate grant from [`TUNNEL_OPEN_SCOPE`] and
+/// [`TUNNEL_DIRECT_SCOPE`] — **never implied by either**: receiving
+/// producer-initiated forwarded connections (ingress) is a different
+/// capability from dialing egress. Stable once published.
+pub const TUNNEL_FORWARDED_SCOPE: &str = "tunnel:forwarded";
+
 /// 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)]
@@ -240,6 +264,218 @@ pub fn tunnel_direct_spec() -> OperationSpec {
     )
 }
 
+/// The bind-first listen open's reply projection (ADR-008 §2 as
+/// amended): the two additive fields the bind-first establisher
+/// contributes, extracted from the full success reply.
+///
+/// **The struct is the projected pair, not a deny-unknown wire gate.**
+/// The reply arrives as `{channel_id, bound?, listen?}` (the
+/// wrapper-reserved `channel_id` key plus the establisher's additive
+/// fields); unknown extra reply fields must be tolerated (the additive
+/// posture — old consumers ignore unknown fields). Parse the full
+/// reply with [`parse_listen_reply`], never with `deny_unknown_fields`
+/// deserialization of the whole reply. Both fields are `None` for
+/// accept-first/dial opens (the establisher supplied nothing).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TunnelListenReply {
+    /// The real OS-chosen bound address (port `0` resolved),
+    /// substrate-shaped — present only when the establisher supplied
+    /// it. Informational surface, NOT addressing: a consumer MUST NOT
+    /// treat `bound` as a dial target (the producer owns the backing —
+    /// ADR-001's invariant); it exists for RFC-1928-reply-class
+    /// fidelity (BND.ADDR) and observability.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub bound: Option<SubstrateAddr>,
+    /// The correlation token (ADR-008 Amendment 1): the u32 the
+    /// establisher minted — the key the forwarded op's `listen` param
+    /// carries. Present only when the establisher supplied it.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub listen: Option<u32>,
+}
+
+/// Project the bind-first listen open's reply (ADR-008 §2 as amended)
+/// out of the FULL success output [`Value`] — the shape
+/// `ChannelClient::open_channel_with_reply` returns. Extracts the
+/// optional `bound` + `listen` fields into [`TunnelListenReply`].
+///
+/// - Unknown extra reply fields are **tolerated** — the reply viewer
+///   must not deny unknown fields (the additive posture; the wire is
+///   plain JSON reply fields, and the projection is a typed view).
+/// - `channel_id` must be present and a positive integer (the
+///   wrapper-owned key — its absence means the caller is not looking
+///   at an open reply at all; fail loud).
+/// - A present-but-malformed `bound` (not substrate-shaped) or
+///   `listen` (not a positive u32) is a loud error — a misbehaving
+///   producer is never silently coerced ([`TunnelError::
+///   MalformedReply`]).
+pub fn parse_listen_reply(reply: &Value) -> Result<TunnelListenReply, TunnelError> {
+    match reply.get("channel_id").and_then(Value::as_u64) {
+        Some(id) if id > 0 => {}
+        _ => {
+            return Err(TunnelError::MalformedReply(
+                "listen reply missing channel_id".to_string(),
+            ))
+        }
+    }
+    let bound = match reply.get("bound") {
+        None | Some(Value::Null) => None,
+        Some(v) => Some(
+            serde_json::from_value::<SubstrateAddr>(v.clone()).map_err(|e| {
+                TunnelError::MalformedReply(format!("bound not substrate-shaped: {e}"))
+            })?,
+        ),
+    };
+    let listen = match reply.get("listen") {
+        None | Some(Value::Null) => None,
+        Some(v) => Some(
+            v.as_u64()
+                .filter(|n| *n > 0 && *n <= u32::MAX as u64)
+                .map(|n| n as u32)
+                .ok_or_else(|| {
+                    TunnelError::MalformedReply("listen not a positive u32".to_string())
+                })?,
+        ),
+    };
+    Ok(TunnelListenReply { bound, listen })
+}
+
+/// The forwarded-op input params (ADR-008 §3 as amended). Wire-stable:
+/// all three fields required; renames are a breaking wire change.
+/// `deny_unknown_fields` is the loud-rejection posture (ADR-001's
+/// precedent — unknown fields fail, never silently ignored).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TunnelForwardedParams {
+    /// The correlation token (ADR-008 Amendment 1): the u32 the
+    /// bind-first establisher minted and returned in the listen
+    /// open's `listen` reply field — NOT the listen channel's ID
+    /// (that key is superseded: the serving-side handler never sees
+    /// its own channel ID, and generic relays don't rewrite params).
+    /// Positive by wire contract (schema `minimum: 1`; the serde pin
+    /// is the backstop).
+    #[serde(deserialize_with = "positive_u32")]
+    pub listen: u32,
+    /// The listen tunnel's substrate; shapes the forwarded channel's
+    /// data-plane framing (ADR-003).
+    pub substrate: Substrate,
+    /// The accepted connection's remote address, keyed by the sibling
+    /// `substrate` field ([`SubstrateAddr`]). Required in v1 — a
+    /// producer that cannot determine the peer does not serve a
+    /// bind-first listener over that substrate (ADR-008 §3).
+    pub peer: SubstrateAddr,
+}
+
+/// The serde-level pin behind `listen`'s schema `minimum: 1`: the
+/// correlation token is positive by contract (the establisher seeds
+/// the mint counter non-zero), so a `listen: 0` on the wire is a
+/// malformed open — rejected at parse time, never silently accepted.
+fn positive_u32<'de, D>(deserializer: D) -> Result<u32, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let v = u32::deserialize(deserializer)?;
+    if v == 0 {
+        return Err(serde::de::Error::custom("listen must be > 0"));
+    }
+    Ok(v)
+}
+
+/// The forwarded-op spec (`channels/tunnel/forwarded`, ADR-008 §3 as
+/// amended): `Sub`-typed, external, channel-open marker on the
+/// `alk/tunnel` ALPN, scope-gated by `tunnel:forwarded` (ingress — a
+/// separate grant from `tunnel:open` and `tunnel:direct`, ADR-006
+/// Amendment 1), input/output schemas per wire.md §The Forwarded Op.
+/// All three params required; the `peer` sub-schema is per-substrate —
+/// `if`/`then` conditioned on the sibling `substrate` value (`tcp`/
+/// `udp` → `{host, port}`, `unix` → `{path}`), so a malformed peer
+/// fails the registry's schema validation (the typed
+/// `INVALID_INPUT` class) before the opener-side handler runs.
+///
+/// **No establisher** on the forwarded op — the data plane starts
+/// immediately (the opener-side serving registration is
+/// `register_openable`, handler = correlate by `listen` + accept the
+/// `BiStream` + pump; the correlation registry is assembly-side
+/// bookkeeping, ADR-008 §3).
+pub fn tunnel_forwarded_spec() -> OperationSpec {
+    OperationSpec::new(
+        OP_TUNNEL_FORWARDED,
+        OperationType::Sub,
+        Visibility::External,
+        json!({
+            "type": "object",
+            "properties": {
+                "listen": { "type": "integer", "minimum": 1 },
+                "substrate": { "type": "string", "enum": ["tcp", "udp", "unix"] }
+            },
+            "required": ["listen", "substrate", "peer"],
+            "allOf": [
+                {
+                    "if": {
+                        "properties": { "substrate": { "enum": ["tcp", "udp"] } },
+                        "required": ["substrate"]
+                    },
+                    "then": {
+                        "properties": {
+                            "peer": {
+                                "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": {
+                            "peer": {
+                                "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_FORWARDED_SCOPE.to_string()],
+            required_scopes_any: None,
+            resource_type: None,
+            resource_action: None,
+        },
+        None,
+    )
+    .with_channel_open(ChannelOpenSpec::new(TUNNEL_ALPN))
+    .with_description(
+        "Receive a producer-initiated forwarded connection (accept-as-open; \
+         params: {listen, substrate, peer}; correlate by the `listen` token)",
+    )
+}
+
 /// The establishment-failure reason code of a
 /// `channel:open_failed` error (`details.reason`): `dial_failed`,
 /// `unknown_resource`, `resource_shortage`, `handler_error`,
@@ -557,4 +793,251 @@ mod tests {
         };
         assert_eq!(unix.render_target(), "/run/postgresql/.s.PGSQL.5432");
     }
+
+    // --- forwarded op + the listen-reply projection (ADR-008, amended) --
+
+    #[test]
+    fn forwarded_params_round_trip_per_substrate() {
+        let cases: Vec<(Substrate, serde_json::Value, SubstrateAddr)> = vec![
+            (
+                Substrate::Tcp,
+                json!({"host": "198.51.100.4", "port": 55231}),
+                SubstrateAddr::HostPort {
+                    host: "198.51.100.4".to_string(),
+                    port: 55231,
+                },
+            ),
+            (
+                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, peer, addr) in cases {
+            let params = TunnelForwardedParams {
+                listen: 918273,
+                substrate,
+                peer: addr,
+            };
+            let v = serde_json::to_value(&params).unwrap();
+            assert_eq!(
+                v,
+                json!({
+                    "listen": 918273,
+                    "substrate": substrate_str(substrate),
+                    "peer": peer,
+                })
+            );
+            let back: TunnelForwardedParams = serde_json::from_value(v).unwrap();
+            assert_eq!(back, params);
+        }
+    }
+
+    #[test]
+    fn forwarded_params_unknown_fields_rejected() {
+        let v = json!({
+            "listen": 918273,
+            "substrate": "tcp",
+            "peer": {"host": "h", "port": 1},
+            "channel": 7
+        });
+        assert!(serde_json::from_value::<TunnelForwardedParams>(v).is_err());
+    }
+
+    #[test]
+    fn forwarded_params_missing_fields_rejected() {
+        let no_listen = json!({"substrate": "tcp", "peer": {"host": "h", "port": 1}});
+        let no_substrate = json!({"listen": 1, "peer": {"host": "h", "port": 1}});
+        let no_peer = json!({"listen": 1, "substrate": "tcp"});
+        assert!(serde_json::from_value::<TunnelForwardedParams>(no_listen).is_err());
+        assert!(serde_json::from_value::<TunnelForwardedParams>(no_substrate).is_err());
+        assert!(serde_json::from_value::<TunnelForwardedParams>(no_peer).is_err());
+    }
+
+    #[test]
+    fn forwarded_params_listen_zero_rejected() {
+        let v = json!({
+            "listen": 0,
+            "substrate": "tcp",
+            "peer": {"host": "h", "port": 1}
+        });
+        assert!(
+            serde_json::from_value::<TunnelForwardedParams>(v).is_err(),
+            "the correlation token is positive by contract (schema minimum 1; \
+             the serde pin is the backstop)"
+        );
+    }
+
+    #[test]
+    fn listen_reply_full_projection() {
+        let reply = json!({
+            "channel_id": 7,
+            "bound": {"host": "203.0.113.9", "port": 42113},
+            "listen": 918273
+        });
+        let projected = parse_listen_reply(&reply).unwrap();
+        assert_eq!(
+            projected,
+            TunnelListenReply {
+                bound: Some(SubstrateAddr::HostPort {
+                    host: "203.0.113.9".to_string(),
+                    port: 42113,
+                }),
+                listen: Some(918273),
+            }
+        );
+    }
+
+    #[test]
+    fn listen_reply_bare_projection() {
+        // The accept-first/dial shape: the establisher contributed
+        // nothing — both fields None.
+        let reply = json!({"channel_id": 7});
+        let projected = parse_listen_reply(&reply).unwrap();
+        assert_eq!(
+            projected,
+            TunnelListenReply {
+                bound: None,
+                listen: None,
+            }
+        );
+    }
+
+    #[test]
+    fn listen_reply_tolerates_unknown_extra_fields() {
+        // The additive posture — the reply viewer must NOT deny
+        // unknown fields (future additive reply fields keep old
+        // consumers working).
+        let reply = json!({
+            "channel_id": 7,
+            "bound": {"path": "/run/x.sock"},
+            "listen": 5,
+            "future_field": {"nested": true}
+        });
+        let projected = parse_listen_reply(&reply).unwrap();
+        assert_eq!(
+            projected,
+            TunnelListenReply {
+                bound: Some(SubstrateAddr::Path {
+                    path: "/run/x.sock".to_string()
+                }),
+                listen: Some(5),
+            }
+        );
+    }
+
+    #[test]
+    fn listen_reply_rejects_malformed_shapes() {
+        // No channel_id at all — not an open reply.
+        assert!(parse_listen_reply(&json!({"bound": {"path": "/p"}})).is_err());
+        // channel_id = 0 (the wrapper never allocates 0 — channel 0 is
+        // the call plane).
+        assert!(parse_listen_reply(&json!({"channel_id": 0})).is_err());
+        // Malformed bound (not substrate-shaped).
+        let bad_bound = json!({"channel_id": 7, "bound": {"port": 1}});
+        assert!(parse_listen_reply(&bad_bound).is_err());
+        // Malformed listen (negative-shaped / not a u32 / zero).
+        let bad_listen = json!({"channel_id": 7, "listen": 0});
+        assert!(parse_listen_reply(&bad_listen).is_err());
+        let bad_listen_type = json!({"channel_id": 7, "listen": "token"});
+        assert!(parse_listen_reply(&bad_listen_type).is_err());
+    }
+
+    #[test]
+    fn forwarded_spec_matches_wire_md() {
+        let spec = tunnel_forwarded_spec();
+        assert_eq!(spec.name, "channels/tunnel/forwarded");
+        assert_eq!(spec.op_type, OperationType::Sub);
+        assert_eq!(spec.visibility, Visibility::External);
+        assert_eq!(
+            spec.input_schema["properties"]["listen"]["minimum"],
+            json!(1)
+        );
+        assert_eq!(
+            spec.input_schema["properties"]["substrate"]["enum"],
+            json!(["tcp", "udp", "unix"])
+        );
+        assert_eq!(
+            spec.input_schema["required"],
+            json!(["listen", "substrate", "peer"])
+        );
+        assert_eq!(spec.output_schema["properties"]["channel_id"]["minimum"], 1);
+        assert_eq!(
+            spec.access_control.required_scopes,
+            vec![TUNNEL_FORWARDED_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 forwarded_spec_peer_schema_gates_shape_per_substrate() {
+        // The per-substrate peer sub-schema: the same if/then
+        // structure the direct op pins its target with — structural
+        // assertions per tunnels/params' precedent (no jsonschema dep
+        // here).
+        let schema = tunnel_forwarded_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"]["peer"];
+        let unix_then = &allof[1]["then"]["properties"]["peer"];
+        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 peer 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 peer fields fail the schema"
+        );
+        assert_eq!(unix_then["additionalProperties"], json!(false));
+        assert_eq!(allof[0]["then"]["properties"]["peer"]["type"], "object");
+        assert_eq!(allof[1]["then"]["properties"]["peer"]["type"], "object");
+    }
+
+    #[test]
+    fn forwarded_params_parse_backstops_the_schema() {
+        // The serde parse is the backstop for the shape it can use:
+        // a peer that schema-gates passed but parse-time mismatches
+        // (or a `listen: 0` riding a registry without the schema gate)
+        // fails loud at the deserializer. `deny_unknown_fields` holds:
+        // unknown top-level params reject; the untagged peer is
+        // lenient about extra peer fields (same documented asymmetry
+        // as the direct op's target — the SCHEMA is the shape-vs-
+        // substrate gate).
+        let cross_shape = json!({
+            "listen": 1,
+            "substrate": "unix",
+            "peer": {"path": "/p", "host": "h"}
+        });
+        let parsed: TunnelForwardedParams = serde_json::from_value(cross_shape).unwrap();
+        assert_eq!(
+            parsed.peer,
+            SubstrateAddr::Path {
+                path: "/p".to_string()
+            }
+        );
+    }
 }

src/producer.rs

@@ -24,7 +24,14 @@
 //! [`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.
+//!
+//! The bind-first shape's plan payload ([`BindFirstPlan`] — the opaque
+//! listener + the correlation token, ADR-008 as amended) is defined
+//! here; the establisher and accept dispatcher that consume it land
+//! with `tunnels/bindfirst-producer` (the wire-stable types pin
+//! first, same ordering `tunnels/params` used for the base op).
 
+use std::any::Any;
 use std::sync::Arc;
 
 use alkcall::channels::operations::{
@@ -40,7 +47,8 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use crate::params::{Substrate, TunnelDirectParams, TunnelParams};
 
 pub use crate::params::{
-    OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE, TUNNEL_OPEN_SCOPE,
+    OP_TUNNEL_DIRECT, OP_TUNNEL_FORWARDED, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE,
+    TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE,
 };
 
 /// The dialed target handle — the plan payload (typed-opaque
@@ -61,6 +69,49 @@ impl std::fmt::Debug for TargetHandle {
     }
 }
 
+/// The bind-first establisher's plan payload (ADR-008 §1 as amended):
+/// the assembly's listener handle + the correlation token.
+///
+/// **The listener is NOT `TargetHandle` halves** — a listener is
+/// accepted FROM, not pumped. The plan carries whatever the assembly's
+/// `BindFn` produced, typed-opaque ([`Any`]; the same
+/// injection pattern as `DialFn`/`AcceptFn` — the accept dispatcher
+/// downcasts it to the assembly's listener type via a type-erased
+/// accept closure; the ADR's "listener handle in the plan" is
+/// assembly-typed, and the protocol sees only this opaque pair).
+///
+/// **The correlation token (ADR-008 Amendment 1):** minted by the
+/// bind-first establisher — a fresh u32 per open, unique per producer
+/// (an `AtomicU32` counter seeded non-zero, e.g. 1, inside the
+/// establisher's closure state; wraps are acceptable — u32, and a wrap
+/// collision requires 2^32 concurrent opens on one producer). Carried
+/// twice: in the plan (the accept dispatcher stamps it into every
+/// forwarded open it opens for this listener) and as the additive
+/// `listen` reply field (the opener correlates its waiting session on
+/// it). It is NOT the listen channel's ID — the serving-side handler
+/// never sees its own channel ID, and generic hub relays forward
+/// params untouched (Amendment 1's rationale). Per-open fresh,
+/// per-producer unique; retired with the listener session.
+pub struct BindFirstPlan {
+    /// The assembly-typed listener handle (e.g. a `TcpListener`
+    /// wrapper the `BindFn` produced). Opaque to the protocol — the
+    /// accept dispatcher downcasts via a type-erased accept closure.
+    pub listener: Arc<dyn Any + Send + Sync>,
+    /// The correlation token (see above): minted fresh per open,
+    /// unique per producer, stamped into each forwarded open's
+    /// `listen` param.
+    pub token: u32,
+}
+
+impl std::fmt::Debug for BindFirstPlan {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("BindFirstPlan")
+            .field("listener", &"<opaque>")
+            .field("token", &self.token)
+            .finish()
+    }
+}
+
 /// The typed establishment-failure vocabulary (ADR-049 §3's reason
 /// codes as this crate's establisher errors; mapped 1:1 onto
 /// `EstablishmentError`).
@@ -487,3 +538,38 @@ pub fn register_tunnel_listen_openable(
         None,
     )
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The plan-payload contract: `BindFirstPlan` rides the
+    /// typed-opaque `ChannelPlan` (`Arc<dyn Any + Send + Sync>`) and
+    /// the accept dispatcher recovers it by downcast — the same
+    /// establisher/handler agreement `TargetHandle` pins. The opaque
+    /// listener half downcasts to the assembly's concrete type.
+    #[test]
+    fn bindfirst_plan_downcasts_through_channel_plan() {
+        struct TestListener(u32);
+        let plan: ChannelPlan = Arc::new(BindFirstPlan {
+            listener: Arc::new(TestListener(42)) as Arc<dyn Any + Send + Sync>,
+            token: 918273,
+        });
+        let unwrapped = plan
+            .downcast::<BindFirstPlan>()
+            .expect("BindFirstPlan plan");
+        assert_eq!(unwrapped.token, 918273);
+        let listener = unwrapped
+            .listener
+            .downcast_ref::<TestListener>()
+            .expect("assembly-typed listener");
+        assert_eq!(listener.0, 42);
+        // A wrong-type downcast fails (the opaque pair is not a
+        // TargetHandle).
+        let other: ChannelPlan = Arc::new(TargetHandle {
+            read: Box::new(tokio::io::empty()),
+            write: Box::new(tokio::io::sink()),
+        });
+        assert!(other.downcast_ref::<BindFirstPlan>().is_none());
+    }
+}

tasks/tunnels/bindfirst-params.md

@@ -1,7 +1,7 @@
 ---
 id: tunnels/bindfirst-params
 name: Bind-first params — ListenParams, the forwarded-op spec, the bind-first plan type (ADR-008 as amended)
-status: pending
+status: completed
 depends_on: []
 scope: narrow
 risk: medium
@@ -129,14 +129,14 @@ pub struct BindFirstPlan {
 
 ## Acceptance Criteria
 
-- [ ] `OP_TUNNEL_FORWARDED`/`TUNNEL_FORWARDED_SCOPE` constants;
+- [x] `OP_TUNNEL_FORWARDED`/`TUNNEL_FORWARDED_SCOPE` constants;
       `tunnel_forwarded_spec()` matches wire.md §The Forwarded Op
-- [ ] `TunnelForwardedParams` (deny-unknown, `listen: u32 > 0`) + the
+- [x] `TunnelForwardedParams` (deny-unknown, `listen: u32 > 0`) + the
       reply-projection viewer round-trip the pinned shapes
-- [ ] `BindFirstPlan` (opaque listener + token) defined; the token
+- [x] `BindFirstPlan` (opaque listener + token) defined; the token
       mint documented (per-producer unique, per-open fresh)
-- [ ] Unit tests green; wasm32 check passes (pure protocol types)
-- [ ] Clippy/fmt clean
+- [x] Unit tests green; wasm32 check passes (pure protocol types)
+- [x] Clippy/fmt clean
 
 ## References
 
@@ -150,8 +150,69 @@ pub struct BindFirstPlan {
 
 ## Notes
 
-> Agent fills during implementation.
+- `SubstrateAddr` had already landed (via `tunnels/direct-op`) —
+  consumed, not redefined (the "either order works" branch resolved
+  to the consuming side).
+- The reply viewer is `parse_listen_reply(&Value) ->
+  Result<TunnelListenReply, TunnelError>`: parses the FULL success
+  output (the `open_channel_with_reply` shape), tolerates unknown
+  extra reply fields (the additive posture — no deny-unknown gate on
+  the reply), requires a positive `channel_id` (its absence/zero =
+  not an open reply — loud), and fails loud on malformed `bound`
+  (not substrate-shaped) or `listen` (not a positive u32) via the
+  new `TunnelError::MalformedReply` variant (error.rs). `null`
+  fields parse as `None` (defensive; a well-formed reply omits them).
+- `TunnelForwardedParams.listen` carries a serde-level
+  `positive_u32` pin behind the schema's `minimum: 1` (the task's
+  "listen: 0 rejected (schema minimum 1 + serde-level pin)" — both
+  gates, test-pinned).
+- The `peer` sub-schema follows the direct op's if/then precedent
+  (tcp/udp → `{host, port}` with `additionalProperties: false` +
+  `maximum: 65535`; unix → `{path}`), structural assertions per the
+  no-jsonschema-dep precedent. The untagged-peer lenient-extra-fields
+  asymmetry carries over from the direct op (schema = the shape gate;
+  serde = the backstop) — documented in the test.
+- `BindFirstPlan` lives in `producer.rs` (the plan-payload family,
+  next to `TargetHandle`): `{ listener: Arc<dyn Any + Send + Sync>,
+  token: u32 }` per the honest shape — the listener is NOT
+  `TargetHandle` halves (a listener is accepted FROM, not pumped);
+  the accept dispatcher downcasts via a type-erased accept closure.
+  The token-mint discipline (AtomicU32 seeded non-zero, per-open
+  fresh, per-producer unique, wraps acceptable) is pinned in the
+  type's doc — the establisher that implements it lands with
+  `tunnels/bindfirst-producer`.
+- Custom `Debug` impls (matching `TargetHandle`'s precedent): the
+  listener renders `<opaque>`, the token renders plainly.
+- Tests: 10 new unit tests in params.rs (forwarded round-trips per
+  substrate, unknown/missing-field rejection, `listen: 0` serde
+  rejection, full/bare/extra-field reply projection, malformed-reply
+  rejection, spec conformance, peer-schema structure, the
+  parse-backstop asymmetry) + the `BindFirstPlan`
+  downcast-through-`ChannelPlan` test in producer.rs (incl. the
+  wrong-type downcast failure).
 
 ## Summary
 
-> Agent fills this on completion.
+Implemented the bind-first wire surface's params half per ADR-008 as
+amended: `params.rs` — `OP_TUNNEL_FORWARDED` / `TUNNEL_FORWARDED_
+SCOPE` constants; `TunnelForwardedParams {listen, substrate, peer}`
+(deny-unknown, `listen` positive via schema `minimum: 1` + the serde
+pin); `tunnel_forwarded_spec()` (`Sub`-typed, External,
+`ChannelOpenSpec::new(TUNNEL_ALPN)`, scope `["tunnel:forwarded"]`,
+per-substrate if/then `peer` sub-schema, output
+`{channel_id: integer > 0}`, description, no establisher);
+`TunnelListenReply` + `parse_listen_reply` (the projected reply
+viewer — unknown extra fields tolerated, `channel_id` presence
+required, malformed fields fail loud via `TunnelError::
+MalformedReply`). `producer.rs` — `BindFirstPlan {listener:
+Arc<dyn Any + Send + Sync>, token: u32}` with the mint discipline
+pinned in docs (non-zero-seeded AtomicU32, per-open fresh,
+per-producer unique, wraps acceptable). Re-exported through
+`lib.rs`; the establisher/dispatcher/registration that consume these
+types land with `tunnels/bindfirst-producer`.
+
+Verified: `cargo test` green (105 default; 125 with `--all-
+features`), clippy `--all-targets -D warnings` clean (native +
+wasm32), fmt clean, `wasm32-unknown-unknown` check passes (pure
+protocol types — the default crate stays wasm-clean), `cargo doc
+--no-deps` clean. CHANGELOG updated (Unreleased → Added).