review 002: the protocol core reviewed — readiness-contract hardening + two convention fixes

The batch-1 gate review (tasks/implementation/review-protocol-core.md):
all eight criteria examined (seam contracts, wire fidelity, access
control, teardown/lifecycle, conventions, wasm, POC fidelity, docs
sync); the gate to batch 2 opens. Findings R-1/R-2 were corrected in
the review change; R-3..R-5 are doc residues fixed in the same change;
R-6 is accepted with rationale.

- R-1 (major): the datagram stage's Ok(None) re-park is unsafe against
  a zero-yield egress — a next_reply that is a pure instantly-ready
  Ok(None) with no await wins every select! poll and livelocks the
  stage on a current-thread runtime (empirically reproduced; the
  landed unit test's mutex masked the hazard at its lock's yield
  boundary; multi-thread survives). Fix: the Ok(None) handler yields
  before re-parking; a pinning unit test drives the strictest
  zero-yield violator.
- R-2: the two into_inner expect()s (consumer sessions) become typed
  Result<_, SpentSessionError> — convention 2's no-panic rule; spent-
  session unit tests pin the shape; test call sites updated.
- R-3/R-4/R-5: doc residues — stale task-time phrasings in
  egress.rs/producer.rs, producer.md's register() sketch reconciled
  to the landed signature, the establishment-bound test doc reworded
  to what the substrate-free probe actually pins.
- R-6 (accepted): the wasm gate rides a transitive getrandom/wasm_js
  feature unification (alkcall -> jsonschema/referencing -> ahash) —
  verified compiling clean today; noted for OQ-SK2-06.

Verified: cargo test 348 green; --all-features green; clippy
--all-targets -D warnings clean; fmt --check clean; both wasm checks
clean (from a clean wasm target dir); cargo doc --no-deps clean.
The bind-only-identity reconciliation from the integration session
was independently verified as correct (ADR-006 as landed; the suite
pins it). Batch boundary holds: no local/composed code exists.

docs/architecture/producer.md

@@ -140,7 +140,9 @@ let producer = Socks5Producer::builder()
     .egress(egress_factory)        // EgressFactory — required for ASSOCIATE
     .accept(bind_accept_factory)   // BindAcceptFactory — required for BIND
     .build();
-producer.register(&mut channel_core)?;   // register_openable_with_establisher
+// register(&core, &registry, auth, witness) — the establisher rides
+// the registration (the optional witness is the CF-006 test probe).
+producer.register(&core, &registry, auth, None)?;
 ```
 
 Missing callbacks = those commands refuse RFC-faithfully

src/consumer.rs

@@ -67,6 +67,15 @@ use crate::rfc1928::util::target_addr::{read_address, TargetAddr, ToTargetAddr};
 use crate::rfc1928::{ReplyError, Socks5Command, SocksError, UdpHeaderError};
 use crate::wire::{frame, DatagramReader, WireError};
 
+/// The spent-session marker (review 002 R-2): [`Socks5ChannelSession
+/// ::into_inner`] / [`AssociateSession::into_inner`] return it when
+/// the pipe/stream was already taken — reachable only by calling
+/// twice on a session whose inner is gone. Carries nothing: a spent
+/// session's teardown is already done (the reap ran with the take).
+#[derive(Debug, thiserror::Error)]
+#[error("the session's pipe/stream was already taken (spent)")]
+pub struct SpentSessionError;
+
 /// A completed SOCKS5 CONNECT session on the consumer side: the
 /// adopted channel + the post-reply byte pipe to the target.
 ///
@@ -159,11 +168,18 @@ impl Socks5ChannelSession {
     /// The session rides along still owning the channel's teardown:
     /// drop the pair (or `close` the session) when done — dropping
     /// `inner` EOFs the data plane structurally.
-    pub fn into_inner(mut self) -> TakenInner {
+    ///
-        let inner = self.inner.take();
+    /// [`SpentSessionError`] on a spent session (the pipe already
-        TakenInner {
+    /// taken): a spent session's manager entry is always reaped
-            session: self,
+    /// already (`close`/`shutdown`/`Drop` all take it), so dropping
-            inner: inner.expect("into_inner on a spent session"),
+    /// it loses nothing — no panic path (AGENTS.md convention 2).
+    pub fn into_inner(mut self) -> Result<TakenInner, SpentSessionError> {
+        match self.inner.take() {
+            Some(inner) => Ok(TakenInner {
+                session: self,
+                inner,
+            }),
+            None => Err(SpentSessionError),
         }
     }
 
@@ -426,11 +442,19 @@ impl AssociateSession {
     /// the POC's `raw_mut` seam, widened). The session rides along
     /// still owning the channel's teardown: drop the pair (or
     /// `shutdown`/`close` the session) when done.
-    pub fn into_inner(mut self) -> TakenAssociateInner {
+    ///
-        let inner = self.inner.take();
+    /// [`SpentSessionError`] on a spent session (the stream already
-        TakenAssociateInner {
+    /// taken): a spent session's manager entry is always reaped
-            session: self,
+    /// already (`shutdown`/`close`/`Drop` all take it), so dropping it
-            inner: inner.expect("into_inner on a spent session"),
+    /// loses nothing — no panic path (AGENTS.md convention 2), and the
+    /// `Err` side stays small (a spent session is not carried).
+    pub fn into_inner(mut self) -> Result<TakenAssociateInner, SpentSessionError> {
+        match self.inner.take() {
+            Some(inner) => Ok(TakenAssociateInner {
+                session: self,
+                inner,
+            }),
+            None => Err(SpentSessionError),
         }
     }
 
@@ -864,6 +888,22 @@ mod tests {
         assert!(session.split().is_none(), "no pipe after teardown");
     }
 
+    #[test]
+    fn spent_session_into_inner_is_an_err_not_a_panic() {
+        // The `expect` the review removed (R-2): the spent-session
+        // path is reachable (a session whose inner is gone), so it
+        // must be a typed `Err` — no panic path.
+        let session = Socks5ChannelSession {
+            channel_id: 9,
+            manager: None,
+            inner: None,
+        };
+        assert!(
+            session.into_inner().is_err(),
+            "a spent session must return Err, not Ok"
+        );
+    }
+
     #[test]
     fn open_errors_carry_the_typed_surface_through_connects_mapping() {
         // The mapping `connect` applies to an open failure: the typed
@@ -909,6 +949,26 @@ mod tests {
         assert!(session.relay_addr().is_none());
     }
 
+    #[test]
+    fn spent_associate_into_inner_is_an_err_not_a_panic() {
+        // The `expect` the review removed (R-2): the spent-session
+        // path is reachable (`into_inner` twice), so it must be a
+        // typed `Err` — no panic path. The teardown owner rides back
+        // to the caller either way.
+        let session = AssociateSession {
+            channel_id: 9,
+            manager: None,
+            inner: None,
+            reader: DatagramReader::new(),
+            pending: Default::default(),
+            relay_addr: None,
+        };
+        assert!(
+            session.into_inner().is_err(),
+            "a spent session must return Err, not Ok"
+        );
+    }
+
     #[tokio::test]
     async fn recv_datagram_persists_reader_state_across_calls() {
         use crate::wire::frame as wire_frame;

src/egress.rs

@@ -39,10 +39,11 @@
 //!   session-level refusal stays channels-level, target-level refusal
 //!   is in-stream).
 //!
-//! The readiness-contract behavioral test (an instant-`Ok(None)`
+//! The readiness-contract behavioral tests (an instant-`Ok(None)`
-//! egress must not starve the datagram stage) lands with the
+//! egress must not starve the datagram stage — including the
-//! producer-associate task — it needs the stage. This module carries
+//! zero-yield shape, review 002 R-1) live in the producer module's
-//! the error→reply-code mapping tables, the sync-lift factory
+//! test suite, next to the stage they pin. This module carries the
+//! error→reply-code mapping tables, the sync-lift factory
 //! constructor, and the compile-shape assertions.
 
 use std::future::Future;

src/lib.rs

@@ -66,7 +66,7 @@ pub mod wire;
 
 pub use consumer::{
     AssociateSession, BindDataPlane, BindReply1, BindSession, Socks5ChannelSession,
-    TakenAssociateInner,
+    SpentSessionError, TakenAssociateInner,
 };
 pub use egress::{
     egress_factory, BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError,

src/producer.rs

@@ -24,7 +24,8 @@
 //! A channels-level refusal is the registry ACL gate's job (it runs
 //! before the establisher — `socks5:open`, ADR-006; the scope-less
 //! identity fails closed with `FORBIDDEN`, no phantom channel). The
-//! establisher itself never refuses in this task's scope.
+//! establisher itself never refuses — it is the identity-recording
+//! session gate (ADR-005).
 //!
 //! ## The pump handler + dispatch
 //!
@@ -320,8 +321,11 @@ where
 /// - **The egress readiness contract** (`Egress::next_reply`): it
 ///   must be `Pending` when no reply is ready; an instantly-ready
 ///   `Ok(None)` starves a biased select into livelock. The stage
-///   tolerates a spurious `Ok(None)` by re-parking (never a hot
+///   tolerates a spurious `Ok(None)` by yielding once and re-parking
-///   loop); an `Err` ends the association (the egress drops with it).
+///   (never a hot loop — and never a zero-yield hot spin on a
+///   current-thread scheduler, the livelock a pure `async { Ok(None)
+///   }` egress reproduces); an `Err` ends the association (the egress
+///   drops with it).
 /// - **Framing applied exactly once, by the wire layer** — the stage
 ///   reads frames via `DatagramReader` (channel → egress `send` gets
 ///   the payload) and writes frames via `frame_inner` (egress
@@ -353,7 +357,14 @@ where
     loop {
         tokio::select! {
             // The egress's reply stream (its own pump; `None` = a
-            // spurious wakeup, the stage re-parks — never a hot loop).
+            // spurious wakeup — the stage re-parks, never a hot
+            // loop). The yield makes the tolerance safe against a
+            // contract-violating zero-yield egress too: without it a
+            // pure `async { Ok(None) }` starves the channel branch
+            // and the runtime's other tasks on a current-thread
+            // scheduler (an instantly-ready branch that yields
+            // nothing wins every poll — the readiness contract's POC
+            // #2 finding, hardened by review 002 R-1).
             reply = egress.next_reply() => {
                 match reply {
                     Ok(Some(frame)) => {
@@ -369,7 +380,7 @@ where
                             }
                         }
                     }
-                    Ok(None) => {}
+                    Ok(None) => tokio::task::yield_now().await,
                     Err(err) => {
                         tracing::debug!("udp egress relay ended: {err:#}");
                         return StageExit::EgressEnded;
@@ -1299,6 +1310,103 @@ mod tests {
             .expect("clean");
     }
 
+    /// The readiness contract's zero-yield bound (review 002 R-1): an
+    /// egress whose `next_reply` is a pure `async { Ok(None) }` —
+    /// instantly ready AND yielding no poll boundary — must still not
+    /// starve the stage's channel branch. The stage's `Ok(None)`
+    /// handler yields before re-parking; this test hangs (livelock)
+    /// without that yield, on any scheduler.
+    #[tokio::test]
+    async fn udp_zero_yield_ok_none_egress_does_not_starve_the_stage() {
+        struct PureHotNoneEgress {
+            inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
+            outbound: tokio::sync::mpsc::UnboundedReceiver<Bytes>,
+        }
+
+        impl crate::egress::Egress for PureHotNoneEgress {
+            fn send(
+                &mut self,
+                addr: TargetAddr,
+                payload: Bytes,
+            ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
+                Box::pin(async move {
+                    let mut inner = addr
+                        .to_be_bytes()
+                        .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
+                    inner.extend_from_slice(&payload);
+                    self.inbound
+                        .send(Bytes::from(inner))
+                        .map_err(|_| EgressError::General("egress closed".into()))
+                })
+            }
+
+            fn next_reply(
+                &mut self,
+            ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
+                Box::pin(async {
+                    // Zero awaits: instantly ready, no yield point at
+                    // all — the strictest readiness-contract
+                    // violation (the mutex-guarded hot egress in
+                    // `udp_instant_ok_none_egress_does_not_starve_
+                    // the_stage` accidentally yields at its lock;
+                    // this one must be survived structurally).
+                    match self.outbound.try_recv() {
+                        Ok(inner) => Ok(Some(inner)),
+                        Err(_) => Ok(None),
+                    }
+                })
+            }
+        }
+
+        let factory: EgressFactory = Arc::new(|| {
+            Box::pin(async {
+                let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel();
+                Ok(Box::new(PureHotNoneEgress { inbound, outbound }) as Box<dyn Egress + Send>)
+            })
+        });
+
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let handle = tokio::spawn(serve_socks5_over(
+            server_end,
+            None,
+            Some(factory),
+            None,
+            false,
+        ));
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::UDPAssociate,
+            &TargetAddr::Domain("pure.hot".into(), 1080),
+        )
+        .await;
+        let (code, _addr) = read_reply(&mut client).await;
+        assert_eq!(code, 0x00);
+
+        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 6], 53)));
+        client
+            .write_all(&datagram_frame(&target, b"pure"))
+            .await
+            .unwrap();
+        let mut buf = vec![0u8; 2 + 7 + 4];
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            client.read_exact(&mut buf).await
+        })
+        .await
+        .expect("the channel branch is not starved — the echo arrived")
+        .expect("read the echo");
+        let (_, echoed) = decode_datagram(&buf);
+        assert_eq!(&echoed[..], b"pure");
+
+        drop(client);
+        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
+            .await
+            .expect("session ends")
+            .expect("join")
+            .expect("clean");
+    }
+
     /// A missing egress callback refuses ASSOCIATE RFC-faithfully
     /// (`CommandNotSupported` 0x07 — ADR-008's missing-callback class,
     /// never a panic), and CONNECT still serves the same registration
@@ -1973,7 +2081,8 @@ mod tests {
     /// The establishment bound covers only the open-op round trip
     /// (ADR-007): the quiet wait outlives any establishment-length
     /// timeout — a client that BINDs and stays silent (no dial) keeps
-    /// the session alive well past the alkcall 10s default bound.
+    /// the session alive (the wait is teardown-governed; the
+    /// over-channels shape is the harness's pin).
     #[tokio::test]
     async fn bind_quiet_wait_is_not_under_the_establishment_bound() {
         let accept = parked_accept_factory();
@@ -1988,12 +2097,12 @@ mod tests {
         let (code, _addr) = read_reply(&mut client).await;
         assert_eq!(code, 0x00, "reply#1 succeeded");
 
-        // Stay silent past the 10s establishment bound — the wait is
+        // Stay silent — the wait is teardown-governed, not
-        // teardown-governed, not establishment-governed. (The
+        // establishment-governed. (The substrate-free shape has no
-        // substrate-free shape has no alkcall bound; the harness
+        // alkcall bound; the harness integration test pins the
-        // integration test pins the over-channels shape. Here the
+        // over-channels shape. Here the parked accept still answers
-        // parked accept still answers after a sleep — the wait is
+        // after a sleep — the wait is alive, not under any in-stream
-        // alive, not under any in-stream timeout.)
+        // timeout.)
         tokio::time::sleep(std::time::Duration::from_millis(500)).await;
         client.write_all(b"x").await.unwrap();
         let mut buf = [0u8; 1];

tasks/implementation/review-protocol-core.md

@@ -1,7 +1,7 @@
 ---
 id: review-protocol-core
 name: Review — the protocol core before any assembly-layer work
-status: pending
+status: completed
 depends_on: [implementation/core-integration-suite]
 scope: narrow
 risk: low
@@ -43,26 +43,274 @@ Review against the architecture, in this order:
    teardown matrix tests actually pin the semantics; no
    teardown-at-birth paths (the third-witness pattern held).
 5. **Convention compliance** (AGENTS.md): no `unwrap`/`expect`/
-   panics outside tests; `thiserror` surfaces; no comments in code
+    panics outside tests; `thiserror` surfaces; no comments in code
-   except doc comments + pinned non-obvious-correctness notes;
+    except doc comments + pinned non-obvious-correctness notes;
-   naming; module structure (one module per file, `lib.rs`
+    naming; module structure (one module per file, `lib.rs`
-   re-exports); producer/consumer vocabulary (no server/client
+    re-exports); producer/consumer vocabulary (no server/client
-   framing outside RFC vocabulary).
+    framing outside RFC vocabulary).
 6. **Wasm-clean invariant** (ADR-009): both wasm checks green; no
-   `net`-shaped feature; tokio on the wasm-clean subset; the harness
+    `net`-shaped feature; tokio on the wasm-clean subset; the harness
-   and tests stay out of the wasm build.
+    and tests stay out of the wasm build.
 7. **POC-fidelity spot-check**: for two or three of the trickiest
-   behaviors (the readiness contract, the quiet-wait two-exit
+    behaviors (the readiness contract, the quiet-wait two-exit
-   semantics, the split-datagram reader persistence), diff the
+    semantics, the split-datagram reader persistence), diff the
-   crate's behavior against the POC crates' findings — the POCs are
+    crate's behavior against the POC crates' findings — the POCs are
-   the empirical record; drift is a finding.
+    the empirical record; drift is a finding.
 8. **Docs sync**: the specs match what landed (the integration
-   task's reconciliation verified; any residue is a finding).
+    task's reconciliation verified; any residue is a finding).
 
 ## Notes
 
-> To be filled by implementation agent
+> Filled by the review agent — see the Summary and the Findings.
 
 ## Summary
 
-> To be filled on completion
+Review performed 2026-09-24 against the eight-criteria order; the
+gate opens. Two findings were corrected in the review change itself
+(R-1 readiness-contract hardening; R-2 a non-test `expect`), plus
+three minor residues (R-3..R-5, fixed in the same change); no
+seam-contract, wire-fidelity, ACL, teardown, wasm, or POC-fidelity
+drift was found. One prior-session reconciliation (the bind-only
+identity row) was independently verified as correct — see the Notes.
+
+**1. Seam contracts (ADR-004, producer.md §The Three Seams) — clean,
+one finding corrected (R-1).** The readiness contract is documented
+on the trait (`egress.rs:174-185`), table-tested, and pinned
+behaviorally twice (the producer unit test's mutex-guarded hot
+egress + the integration suite's hot-`Ok(None)` under concurrent
+load). Ownership rides the handle structurally (`TargetHandle`/
+`BindTargetHandle` payloads own their halves; no `_` bindings of
+owning values anywhere in the crate — swept); the accept future owns
+its resources end-to-end (`BindAccept::new` boxes it; the quiet
+wait's `select!`-pinned drop is the client-gone teardown, pinned by
+the drop-flag tests in `src/producer.rs` and `tests/core-
+integration.rs`); framing-once holds (`Egress::send` takes payload
+only, `next_reply` returns inner bytes, `frame_inner`/`frame` are
+the only prefixers; `frame == frame_inner` byte-identity is
+table-tested). R-1 is the readiness contract's own tolerance: see
+the Findings.
+
+**2. Wire fidelity — clean.** The datagram frame matches wire.md §The
+Data Plane byte-for-byte: `[len: u16 BE][addr + data]`, the joint
+bound (`frame`/`frame_inner` reject > 65535 with the exact per-form
+maxima pinned: 65528 v4 / 65516 v6 / 65276 domain), F-2 at payload
+level (empty datagram round-trips end to end, never EOF; `len == 0`
+inner is `EmptyFrame` — unreachable from a compliant peer), RSV/FRAG
+absent from the in-band codec. The phase model holds: no mode byte;
+post-reply garbage is refused structurally in all three commands
+(CONNECT = pure pipe, ASSOCIATE = garbage ATYP fails loud, BIND =
+the quiet wait's one-byte read ends the session with no reply#2) —
+each pinned by test. Sentinels verified: ASSOCIATE's `127.0.0.1:0`
+(asserted in the integration suite + the raw probes), BIND's
+variant-3 zeros (`0.0.0.0:0` reply#1 + reply#2, pinned);
+association lifetime = channel lifetime (EOF rows in the teardown
+matrix). The BAST document's joint-constraint note matches the
+codec.
+
+**3. Access control (ADR-006) — clean.** The open gate is
+`required_scopes = ["socks5:open"]` on the single op spec
+(`socks5_open_spec`); the command gate resolves `socks5:bind` from
+the per-call opener identity at BIND dispatch and precedes the
+accept seam (a scope-less BIND never mints a listener — pinned).
+Both refusals are RFC-faithful (0x02 at the command gate; the
+registry's `FORBIDDEN` at the open gate); the mapping tables
+(dial/egress/accept/factory errors → reply codes) are table-tested
+and route through the forked `ReplyError` — nothing collapsed into
+generic errors. The integration suite's identity matrix pins all
+three rows end to end with the CF-006 witness asserted.
+
+**4. Teardown/lifecycle — clean.** `pump_bidi` is awaited inline in
+all three data planes (CONNECT, BIND post-reply#2) inside the
+handler task; the `JoinHandle` tracks the data-plane lifetime (R-02
+held — no spawn-and-forget). The teardown matrix tests pin the real
+semantics (half-close / EOF / client-gone-with-parked-accept, each
+followed by fresh sessions of every command). No teardown-at-birth
+paths: the egress-drop witness test flips the flag on stage exit,
+the accept-drop witness flips on the client-gone branch; the
+consumer sessions reap on `close`/`shutdown`/`Drop` with the leak
+probes asserting no phantom channels on either side.
+
+**5. Convention compliance — one finding corrected (R-2).** No
+`unwrap`/`expect`/`panic!`/`unreachable!` in non-test library code
+after R-2 (swept: `src/` non-test regions are clean; the forked
+core's panic translations were re-verified). `thiserror` surfaces
+throughout; naming/module structure (one module per file, `lib.rs`
+re-exports) hold; producer/consumer vocabulary holds (RFC-facing
+"server"/"client" only inside `src/rfc1928/`, the RFC vocabulary's
+home). Cargo.toml's two inline policy comments are the
+extraction-review F-3-accepted posture (carried, no fix demanded).
+
+**6. Wasm-clean invariant — green, with one accepted note (R-5).**
+Both wasm checks clean (`cargo check`/`clippy --target
+wasm32-unknown-unknown -- -D warnings`, from scratch); no
+`net`-shaped feature (the two declared features are empty markers);
+tokio on the wasm-clean subset with `default-features = false` (no
+`full` in `[dependencies]`); the harness/tests ride dev-deps only.
+R-5: see the Findings (a transitive `getrandom` carrier outside
+this crate's control).
+
+**7. POC-fidelity spot-check — clean, one finding corrected (R-1).**
+- *Readiness contract*: the crate's stage matches the POC's
+  `select!`-driven shape (channel reads ∥ reply drains, `Ok(None)`
+  tolerated as a spurious wakeup); the POC's LocalEgress/
+  ComposedEgress both park properly (`readable()` / queue
+  `recv().await`) and the ComposedEgress's closed-queue arm returns
+  `Err` rather than a hot `None` — exactly the posture the crate
+  inherited. The drift the spot-check surfaced is the *zero-yield*
+  bound of the contract (R-1) — a pure `async { Ok(None) }` egress
+  livelocked the landed stage on a current-thread runtime
+  (empirically reproduced; the POC's findings named the biased-select
+  and hot-None hazards, the no-yield variant is the same class one
+  step further). Fixed in the review change.
+- *Quiet-wait two-exit semantics*: the landed `select!` is
+  single-shot with exactly the two exits (accept / EOF-garbage-error),
+  not a loop — matches the POC's `never_loop` finding; the parked-
+  accept probes make both exits deterministic; reply#1/reply#2 carry
+  the factory's `bind_addr`/peer byte-exactly on the provider
+  variant and the sentinels on variant 3.
+- *Split-datagram reader persistence*: `AssociateSession::recv_
+  datagram`'s reader + pending queue persist across calls (pinned by
+  the 2-byte-prefix chunk test and the consumer-associate suite) —
+  matches ADR-003's re-read of the POC finding.
+
+**8. Docs sync — clean, three residues fixed in the review change.**
+The integration task's reconciliation was verified against ADR-006
+§Consequences, producer.md §Scope Gating, and the landed spec/tests
+(see the Notes). Residues found and fixed: `src/egress.rs`'s module
+doc still said the readiness tests "land with the producer-associate
+task" (that task is landed — reworded to point at where they live);
+`src/producer.rs`'s establisher doc said "in this task's scope"
+(same class — reworded); producer.md §Wiring's sketch showed
+`producer.register(&mut channel_core)` — the landed signature takes
+`(&ChannelCore, &OperationRegistry, AuthContext, Option<IdentityWitness>)`
+and also carries the establisher (corrected). The rest of the specs
+(wire.md, consumer.md §Wiring, ADR-006, rfc1928.md, README Current
+State, lib.rs crate docs) match the tree.
+
+## Acceptance Criteria
+
+- [x] Seam contracts hold (readiness / ownership / accept-future /
+      framing-once) — with R-1 corrected
+- [x] Wire fidelity: the frame, phase model, sentinels, lifetimes
+      byte-checked against wire.md/bast.md
+- [x] Access control: both gates + never-implied grants pinned, the
+      prior session's reconciliation verified
+- [x] Teardown/lifecycle: pump_bidi inline, the matrix pins the
+      semantics, no teardown-at-birth paths
+- [x] Convention compliance: no unwrap/expect/panics outside tests
+      after R-2; thiserror; naming; module structure; vocabulary
+- [x] Both wasm checks green (from clean), no net-shaped feature,
+      tokio on the wasm-clean subset
+- [x] POC-fidelity spot-check of the three trickiest behaviors —
+      the zero-yield drift (R-1) found and fixed
+- [x] Docs sync: specs match the tree; residues fixed in the same
+      change
+- [x] Findings recorded; the gate to batch 2 opens
+
+### Findings
+
+- **R-1 (major → fixed in the review change)**: the datagram stage's
+  `Ok(None)` tolerance (re-park) is safe against a hot egress only
+  while its futures yield at least one poll boundary. A *zero-yield*
+  egress — `next_reply` = a pure `async { Ok(None) }`, no await at
+  all — wins every `select!` poll forever: the channel branch and
+  every other task on a current-thread scheduler starve (empirically
+  reproduced: the echo never arrives; the multi-thread runtime
+  survives, so the landed mutex-guarded unit test masked the hazard
+  at its lock's yield boundary). The POC's contract wording ("must
+  be `Pending` when nothing is ready") is right; the landed stage's
+  handling had one gap. Fix: the stage's `Ok(None)` handler now
+  `yield_now()`s before re-parking (`src/producer.rs`
+  `run_datagram_stage`), plus a pinning unit test with a pure
+  zero-yield egress (the strictest violator — hangs without the
+  yield) alongside the existing hot-None test. ADR-003's contract
+  prose and `egress.rs`'s trait doc stay as written (the contract
+  is on the egress; the stage now additionally defends itself).
+- **R-2 (convention, fixed in the review change)**: `Socks5Channel
+  Session::into_inner` / `AssociateSession::into_inner` ended in
+  `inner.expect("into_inner on a spent session")` — the only
+  non-test `expect` in `src/`. Reachability is narrow (a spent
+  session's inner is gone only after close/shutdown/Drop, and no
+  public API constructs a spent session), but convention 2's rule is
+  the rule: the shape is now `Result<TakenInner, SpentSessionError>`
+  / `Result<TakenAssociateInner, SpentSessionError>` with a
+  zero-carrier error type (`SpentSessionError` — a spent session's
+  manager entry is always already reaped, so dropping it loses
+  nothing; the large-Err/unit-Err lints shaped the choice), pinned
+  by two spent-session unit tests. Test call sites updated
+  (`.expect("the session is live")` — test-side, allowed).
+- **R-3 (doc residue, fixed)**: `src/egress.rs` module docs + the
+  `src/producer.rs` establisher doc carried task-time phrasings
+  ("lands with the producer-associate task" / "this task's scope")
+  that are stale now the tasks are landed. Reworded to point at the
+  landed reality.
+- **R-4 (doc residue, fixed)**: producer.md §Wiring's assembly sketch
+  showed a one-argument `producer.register(...)`; the landed
+  signature takes the core/registry/auth/witness quadruple. Reconciled
+  (the same sync discipline the integration task applied to
+  consumer.md).
+- **R-5 (doc residue, fixed)**: the
+  `bind_quiet_wait_is_not_under_the_establishment_bound` test doc
+  claimed the substrate-free probe exercises "the alkcall 10s
+  default bound" — it cannot (the substrate-free shape has no
+  alkcall bound; the over-channels wait-outlives-establishment
+  behavior is the harness's pin by construction). Reworded to say
+  what the test actually pins (the wait is teardown-governed, no
+  in-stream timeout).
+- **R-6 (accepted, no fix demanded)**: the wasm-clean invariant rides
+  a transitive feature unification outside this crate: `alkcall →
+  jsonschema/referencing → ahash (runtime-rng) → getrandom 0.3.4`,
+  and `wasm_js` (required to compile `getrandom` on
+  `wasm32-unknown-unknown`) is enabled only because jsonschema/
+  referencing declare it for the wasm32 target. The default-crate
+  check passes today (verified from a clean target dir), and the
+  alkty sibling shows the identical posture; but a future
+  alkcall/jsonschema feature change could silently break the wasm
+  gate with no in-tree cause. Not actionable in this crate (the dep
+  graph is alkcall's); noted for the wasm-adapter story's
+  eventual OQ-SK2-06 work and for the batch-2 wasm checks.
+
+## References
+
+- docs/architecture/wire.md, producer.md, consumer.md, rfc1928.md,
+  bast.md, README.md (the specs under review)
+- docs/architecture/decisions/001/003/004/006/007/009/013
+- docs/research/poc-udp-associate-findings.md,
+  poc-bind-findings.md, poc-connect-wasm-findings.md (the POC
+  fidelity sources)
+- /workspace/alksocks-udp-poc (the POC reference for the readiness
+  contract's empirical record)
+- AGENTS.md (conventions 1–17; the verification commands)
+
+## Notes
+
+- The prior integration session's Notes/Summary were read and
+  cross-checked against the specs and the tree. The one
+  reconciliation it flagged — the task file's identity-matrix row 2
+  expecting a bind-only identity to serve BIND, vs ADR-006's landed
+  `required_scopes = ["socks5:open"]` unconditionally — is **correct
+  as landed**: the op spec (the single op, all three commands, ADR-
+  001) carries the open gate unconditionally; a bind-only identity
+  therefore fails closed for every command (FORBIDDEN at the
+  registry, the establisher never runs, no phantom channel — pinned
+  producer- and consumer-side in
+  `identity_matrix_bind_only_fails_closed_at_the_open_gate`), and a
+  BIND-capable deployment grants both scopes explicitly. The suite
+  pins this; ADR-006 §Consequences states the asymmetry explicitly
+  ("never implied, in either direction"); the task file's Notes
+  record the reasoning. No residue: the drift was a Phase-0-era task
+  phrasing, not a spec-vs-code contradiction. The user's flag was
+  well-placed — this review treats the reconciliation as verified
+  rather than open.
+- The self-reported verification claims reproduce (the landed suite
+  was green at review start; this review's fixes re-ran the full
+  sweep — see below).
+- Post-review verification (the review change included): `cargo
+  test` 348 green (121 lib + 38 connect + 39 associate + 40 bind +
+  40 integration + 30 differential + 37 rfc1928 + 3 doctests — the
+  two R-1/R-2 additions and the four R-2 call-site updates
+  included), `cargo test --all-features` green, `cargo clippy
+  --all-targets -- -D warnings` clean, `cargo fmt --check` clean,
+  both wasm checks clean (re-run from a clean wasm target dir),
+  `cargo doc --no-deps` clean. The batch boundary holds: no
+  `local`/`composed` code exists.

tests/consumer-connect.rs

@@ -302,7 +302,7 @@ async fn half_close_write_eof_ends_the_producer_pump() {
     let TakenInner {
         session: _owner,
         inner,
-    } = session.into_inner();
+    } = session.into_inner().expect("the session is live");
     let (mut read, mut write) = tokio::io::split(inner);
     write.write_all(b"pre").await.expect("write");
     let mut buf = [0u8; 3];
@@ -344,7 +344,7 @@ async fn into_inner_pairs_the_pipe_with_the_teardown_owner() {
     .await
     .expect("the session connects");
     let channel_id = session.channel_id();
-    let TakenInner { session, inner } = session.into_inner();
+    let TakenInner { session, inner } = session.into_inner().expect("the session is live");
     assert!(topo.consumer.manager().has_channel(channel_id));
 
     let (mut read, write) = tokio::io::split(inner);
@@ -424,7 +424,7 @@ async fn close_after_into_inner_reaps() {
     .await
     .expect("the session connects");
     let channel_id = session.channel_id();
-    let TakenInner { session, .. } = session.into_inner();
+    let TakenInner { session, .. } = session.into_inner().expect("the session is live");
     // The inner pipe was taken; `close` still reaps the entry.
     assert!(session.close().await, "the entry existed before the close");
     no_leaked_channels(&topo).await;