fix: local-bridge robustness, scope-gate order, sentinel contract; packaging/docs closeout (P8-P11, P2/P3/P15)

Closes review #003 (prepublish review for v0.1.0). - P8: StdinSink::poll_shutdown parks an inflight reserve+send on a full channel (waker registered) — a stdin blast followed by EOF delivers the EOF instead of stranding it - P9: five poisoned-lock .expect() sites -> unwrap_or_else(into_inner) - P10: three thread-spawn .expect() sites -> TtyError::AllocFailed - P5: tty:open scope gate runs before carriage/cmd/backend-lookup checks (no backend-name enumeration differential for unscoped ids) - P11: recv_stdout terminates on the zero-length drained sentinel; the sentinel is no longer yielded as an item (doc was already the contract); stderr has no sentinel (doc noted) - P2: exclude AGENTS.md + docs/plans/, drop dead Cargo.lock and docs/research/ entries (package list: 42 files, 659.2KiB) - P3: AGENTS.md phase status (all five landed), ADR range 001..009 (+ alktty-native ADR-009 in the mapping), alkcall guidance corrected to v0.4.x / pin "0.4.0"; architecture README ADR-009 row + landed-phase status - P15: backend.rs doc typo; redundant tokio-stream dev-dep removed; NegotiationError::Io arm logs; set_identity failure logs; input_pump.abort() at session end; TtySessionError::Open carries the accept_bi StreamError (no io::Error flattening); borrowing deserialize in open_via_channels (no params.clone()); error_response_bytes guards an "error" key in fields; trivial inline comments promoted/removed; plan-doc test counts + doc front-matter refreshed; session tests that raced session teardown under the abort change use a GatedBackend (exit held until released) Verification: cargo test 104 lib / --all-features 147; clippy (all-targets + wasm32) -D warnings; fmt; wasm check; doc 0 warnings; publish dry-run OK.
2026-09-05 16:47:45 +00:00
parent 7b0bac0671
commit 3baa993edb
14 changed files with 372 additions and 105 deletions
@@ -246,16 +246,18 @@ wasm-clean" invariant — run it whenever a non-`local` module changes.

 ## Architecture Context

- `docs/plans/project-setup.md` — the current plan (phases 0–5). Phase 0
-  (scaffold hygiene), Phase 1 (port core types), Phase 2 (channels
-  integration + `TtySession`), Phase 3 (`local` backend), and Phase 4
-  (architecture docs + BAST schema + renumbered ADRs) are landed.
-  Phase 5 (tests, including integration tests in `tests/` at the crate
-  root) is not yet done.
+- `docs/plans/project-setup.md` — the current plan (phases 0–5). All
+  five phases are landed: Phase 0 (scaffold hygiene), Phase 1 (port
+  core types), Phase 2 (channels integration + `TtySession`), Phase 3
+  (`local` backend), Phase 4 (architecture docs + BAST schema +
+  renumbered ADRs), and Phase 5 (tests, including the integration
+  tests in `tests/` at the crate root).
 - `docs/architecture/` is created in Phase 4. The alknet ADRs
  referenced below (052, 053, 054, 055, 056, 057, 077, 093) are ported
-  and renumbered into alktty's ADR range (001..008) in
-  `docs/architecture/decisions/`. The alknet originals at
+  and renumbered into alktty's ADR range (001..009) in
+  `docs/architecture/decisions/`. ADR-009 (the channels open op's
+  `input` is the negotiation) is alktty-native — it resolves review
+  #001 L1 and has no alknet original. The alknet originals at
  `/workspace/@alkdev/alknet/docs/architecture/decisions/` remain the
  authoritative source for any ADR not yet ported — read them before
  non-trivial changes to the wire format, trait, or adapter.
@@ -284,6 +286,8 @@ wasm-clean" invariant — run it whenever a non-`local` module changes.
    for historical context with its reversal notice pointing to 008)
  - ADR-093 → 008 — TTY always uses its 5-byte format inside channels
    (reverses ADR-077; channels strips its 8-byte header transparently)
+  - (alktty-native) ADR-009 — the channels open op's `input` IS the
+    negotiation (no alknet original; resolves review #001 L1)
 - ADR-050 (dynamic resource ownership) is an alkcall/alknet-core ADR,
  not a tty-specific one — it is not ported into alktty's ADR range.
  The access-control work that declares against the ADR-050 model
@@ -304,7 +308,8 @@ wasm-clean" invariant — run it whenever a non-`local` module changes.
  `BiStream`, `BidiStreamSource`, `AuthContext`, `Identity`,
  `IdentityProvider`, `AccessControl`, `OwnershipProvider`,
  `OwnershipStore`, `InMemoryOwnershipStore`, `HandlerError`,
-  `StreamError`) come from `alkcall::core`. alkcall is v0.1.x —
+  `StreamError`) come from `alkcall::core`. alkcall is v0.4.x —
  breaking changes are expected at this major-zero stage; this is the
  first real consumer, so we find and fix issues upstream rather than
-  working around them. Pin `alkcall = "0.1.1"` and bump deliberately.
+  working around them. Pin `alkcall = "0.4.0"` (lockfile resolves
+  0.4.1) and bump deliberately.
@@ -88,11 +88,14 @@ alknet mono-repo.
  `send_stdin`/`close_stdin` (stdin chunks / EOF sentinel),
  `recv_stdout`/`recv_stderr` (`Stream<Item = Bytes>`), `resize`,
  `signal`, and `wait` (awaits the `Exit` control chunk; watch-based,
-  no lost wakeup). Both constructors disambiguate the first response
-  frame (`0x00` prefix = negotiation error frame → `NegotiationRejected`
-  with the server's error code; raw chunk = session proceeds).
-  `TtySessionError` is `#[non_exhaustive]`. Dropping the session aborts
-  the read pump and closes the write half.
+  no lost wakeup). `recv_stdout` ends ON the zero-length drained
+  sentinel (it is not yielded as an item). Both constructors
+  disambiguate the first response frame (`0x00` prefix = negotiation
+  error frame → `NegotiationRejected` with the server's error code;
+  raw chunk = session proceeds); a refused session open surfaces as
+  `TtySessionError::Open` carrying the upstream error type.
+  `TtySessionError` is `#[non_exhaustive]`. Dropping the session
+  aborts the read pump and closes the write half.
 - **Control messages.** `ControlMessage` (`Resize`/`Signal`/`Eof` on
  `ctrl_in`; `Exit` on `ctrl_out`) — JSON, tagged by `"type"`, unknown
  types ignored by policy. `signal_from_name` maps the common signal
@@ -120,5 +123,16 @@ alknet mono-repo.
  regression fails the test instead of hanging it, plus exact-boundary
  (`MAX_CHUNK_LEN`) round-trip and server-sends-client-direction-
  stream-types discard tests.
+- **PTY bridge robustness (`local` feature).** The stdin sink's EOF
+  (`poll_shutdown`) parks an in-flight send on a full channel so the
+  waker is registered — a stdin blast followed by EOF always delivers
+  the EOF to the child. Lock-poisoning no longer cascades (the five
+  lock sites adopt `into_inner()`), and thread-spawn failure maps to
+  `TtyError::AllocFailed` instead of panicking.
+- **Producer hardening.** The `tty:open` scope gate runs before the
+  backend lookup (an unscoped identity gets `forbidden` regardless of
+  the request body — no backend-name enumeration differential), and
+  the client→backend pump is aborted at session end instead of
+  lingering until the client disconnects.

 [0.1.0]: https://git.alk.dev/alkdev/alktty/releases/tag/v0.1.0
@@ -9,7 +9,7 @@ repository = "https://git.alk.dev/alkdev/alktty"
 readme = "README.md"
 keywords = ["tty", "terminal", "pty", "channels", "alkcall"]
 categories = ["network-programming", "asynchronous"]
-exclude = [".opencode/", "docs/reviews/", "docs/research/", "docs/sdd_process.md", "Cargo.lock"]
+exclude = [".opencode/", "AGENTS.md", "docs/reviews/", "docs/plans/", "docs/sdd_process.md"]

 [lib]
 name = "alktty"
@@ -48,7 +48,6 @@ libc = "0.2"

 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util", "macros"] }
-tokio-stream = "0.1"
 serde_json = "1"
 tempfile = "3"
 bytes = "1"
@@ -73,7 +73,8 @@ session.close_stdin().await?;

 let mut stdout = session.recv_stdout().await;
 while let Some(bytes) = futures::StreamExt::next(&mut stdout).await {
-    if bytes.is_empty() { break; } // zero-length stdout sentinel = drained
+    // The stream ends on the zero-length stdout sentinel ("drained") —
+    // it is never yielded as an item.
    // ... render bytes ...
 }
 let code = session.wait().await?;
@@ -19,7 +19,7 @@ Syntax Tree) document for the wire format, and the ADRs.
 ## Applicable ADRs

 Ported from the alknet mono-repo and renumbered into alktty's ADR
-range (001..008). The alknet originals at
+range (001..009; ADR-009 is alktty-native). The alknet originals at
 `/workspace/@alkdev/alknet/docs/architecture/decisions/` remain the
 authoritative source for any ADR not yet ported, and for the alknet
 ADRs referenced by alknet number in the docs below (which are not
@@ -35,6 +35,7 @@ tty-specific and therefore not ported into alktty's ADR range).
 | [006](decisions/006-negotiation-framing-self-contained.md) | Self-Contained Negotiation Framing (No alkcall-Internal-Wire-Types Dependency) | alknet ADR-057 | Accepted |
 | [007](decisions/007-tty-inside-channels.md) | TTY Inside Channels — Sub-Streams, Not Wire Format | alknet ADR-077 | Accepted (**reversed by ADR-008** — kept for historical context) |
 | [008](decisions/008-channels-pure-channel-multiplexing.md) | Channels Pure Channel Multiplexing (8-Byte Header, No `stream_type`) | alknet ADR-093 | Accepted (amends alknet ADR-071/074; reverses ADR-007) |
+| [009](decisions/009-channels-open-op-is-the-negotiation.md) | The Channels Open Op's `input` Is the Negotiation | alktty-native | Accepted (resolves review #001 L1; amended by review #002 R4 — parse failure is a client-visible error frame) |

 ## Key Design Principles

@@ -135,11 +136,10 @@ tty-specific and therefore not ported into alktty's ADR range).

 ## References

- `docs/plans/project-setup.md` — the current plan (phases 0–5).
-  Phases 0–3 are landed; Phase 4 (this directory: architecture docs +
-  BAST schema + renumbered ADRs) is landed by this commit; Phase 5
-  (tests, including integration tests in `tests/` at the crate root) is
-  not yet done.
+- `docs/plans/project-setup.md` — the current plan (phases 0–5). All
+  five phases are landed: Phase 4 landed this directory (architecture
+  docs + BAST schema + renumbered ADRs); Phase 5 landed the tests,
+  including the integration tests in `tests/` at the crate root.
 - alknet originals of the ported ADRs (alknet ADR-052, 053, 054, 055,
  056, 057, 077, 093) at
  `/workspace/@alkdev/alknet/docs/architecture/decisions/` — the
@@ -2,8 +2,8 @@
 status: draft (ported from alknet 2026-08-17; alknet-tty → alktty,
  alknet-tty-local → alktty's `local` feature module, alknet/tty →
  alk/tty, alknet-core → alkcall::core, alknet-call → alkcall, ADRs
-  renumbered 052..093 → 001..008)
-last_updated: 2026-08-17
+  renumbered 052..093 → 001..009)
+last_updated: 2026-09-05
 ---

 # alktty — Overview
@@ -1,8 +1,8 @@
 ---
-status: draft (ported from alknet 2026-08-17; alknet-tty → alktty,
-  alknet/tty → alk/tty, alknet-core → alkcall::core, alknet-call →
-  alkcall, ADRs renumbered 052..093 → 001..008)
-last_updated: 2026-08-17
+status: draft (ported from alknet 2026-08-17; alknet/tty → alk/tty,
+  alknet-core → alkcall::core, alknet-call → alkcall, ADRs renumbered
+  052..093 → 001..009)
+last_updated: 2026-09-05
 ---

 # alktty — TtyAdapter and Session Lifecycle
@@ -527,8 +527,10 @@ either). `AGENTS.md` was updated to match this mapping.
   `MockBackend` from `backend.rs` for the producer side so no
   real PTY is needed.

-Total: 80 lib tests + 19 integration tests = 99 passing under
-`--all-features`; 80 lib tests under default (wasm-clean) build.
+Total: 119 lib tests + 19 integration tests = 138 passing under
+`--all-features`; 104 lib tests under default (wasm-clean) build
+(counts as of the review #003 remediation; the suites have grown since
+the 80/19 the plan originally recorded).

 ## Open Questions

@@ -1,7 +1,7 @@
 ---
-status: open
+status: closed
 last_updated: 2026-09-05
-resolved: 1, 4, 6, 7, 12, 13, 16 (P14 closed)
+resolved: 1-16 (P14 closed; P13's accept-loop harness deferred by disposition)
 reviewed_artifacts:
  - src/lib.rs
  - src/wire.rs
@@ -30,6 +30,7 @@ base_commit: 918af40 (review #002 resolved)
 resolutions:
  - "P1 + P13 resolved 2026-09-05 (87c52e5) — see the Session-1 Resolution section"
  - "P12 + P6 + P7 + P4 + P16 resolved 2026-09-05 — see the Session-2 Resolution section"
+  - "P8 + P9 + P10 + P5 + P11 + P2 + P3 + P15 resolved 2026-09-05 — see the Session-3 Resolution section"
 ---

 # Code Review #003 — Prepublish Review (v0.1.0)
@@ -467,20 +468,20 @@ theirs; don't replicate that nit.
 | ID | Finding | Fix | Effort | Risk | Status |
 |----|---------|-----|--------|------|--------|
 | P1 | pump_session deadlock ≥63 chunks | concurrent drainer + regression test | small | medium (core pump) | ✅ resolved (`87c52e5`) |
-| P2 | AGENTS.md ships; dead exclude entries | fix `exclude` | trivial | none | open |
-| P3 | stale AGENTS.md + architecture README | refresh phase/ADR/alkcall text | trivial | none | open |
+| P2 | AGENTS.md ships; dead exclude entries | fix `exclude` | trivial | none | ✅ resolved (Session 3) |
+| P3 | stale AGENTS.md + architecture README | refresh phase/ADR/alkcall text | trivial | none | ✅ resolved (Session 3) |
 | P4 | no README + `readme` field | write README (after P5/P6/P12) | medium | none | ✅ resolved (Session 2) |
-| P5 | scope gate after backend lookup | reorder checks | trivial | low (test updates) | open |
+| P5 | scope gate after backend lookup | reorder checks | trivial | low (test updates) | ✅ resolved (Session 3) |
 | P6 | write paths accept invalid chunks | validate on write; unify empty-payload shape | small | low | ✅ resolved (Session 2) |
 | P7 | read-after-peek framing corruption | peeked-state guard | small | none | ✅ resolved (Session 2) |
-| P8 | poll_shutdown missing waker | inflight-future pattern | small | low | open |
-| P9 | poisoned-lock `.expect()` × 5 | `into_inner()` | trivial | none | open |
-| P10 | thread-spawn `.expect()` × 3 | map to `TtyError` | trivial | none | open |
-| P11 | recv_stdout sentinel contract | terminate stream on sentinel (or fix doc) | small | low (semver-visible) | open |
+| P8 | poll_shutdown missing waker | inflight-future pattern | small | low | ✅ resolved (Session 3) |
+| P9 | poisoned-lock `.expect()` × 5 | `into_inner()` | trivial | none | ✅ resolved (Session 3) |
+| P10 | thread-spawn `.expect()` × 3 | map to `TtyError` | trivial | none | ✅ resolved (Session 3) |
+| P11 | recv_stdout sentinel contract | terminate stream on sentinel (or fix doc) | small | low (semver-visible) | ✅ resolved (Session 3) |
 | P12 | no crate-root re-exports | `pub use` primary types | trivial | none | ✅ resolved (Session 2) |
 | P13 | test gaps | backpressure (>64 chunks) + boundary tests | small | none | ✅ resolved (`87c52e5`; accept-loop harness deferred) |
 | P14 | `Result<(), String>` on register_openable | closed — alkcall parity | none | none | ✅ closed |
-| P15 | minor polish batch | one batched commit | small | none | open |
+| P15 | minor polish batch | one batched commit | small | none | ✅ resolved (Session 3) |
 | P16 | no CHANGELOG.md | start changelog with [0.1.0] | small | none | ✅ resolved (Session 2) |

 ## Suggested Session Breakdown
@@ -603,6 +604,109 @@ clippy (all-targets + wasm32) `-D warnings`, fmt, wasm check, doc
 (0 warnings), `cargo publish --dry-run --allow-dirty` (44 files,
 702.8KiB) — all clean.

+## Session-3 Resolution (P8 + P9 + P10 + P5 + P11 + P2 + P3 + P15, 2026-09-05)
+
+The `local` robustness batch, the producer-hygiene batch, and the
+packaging/docs closeout — the review's suggested session breakdown
+items 3, 4, and 5. This closes review #003.
+
+- **P8 — `poll_shutdown` waker** (`src/local/pty.rs`): `StdinSink`
+  gained an `inflight_close` slot mirroring `poll_write`'s inflight
+  pattern — on a full stdin channel, `poll_shutdown` parks a
+  `reserve + send(StdinCmd::Eof)` future and polls it on re-poll, so
+  the poller's waker is registered and a stdin blast followed by EOF
+  delivers the EOF instead of stranding it. Distinct from the
+  byte-write slot so the two sends never share a future.
+- **P9 — poisoned-lock `.expect()` ×5** (`src/local/pty.rs`): the five
+  non-test lock sites (`PtyControl::resize`, both `PtyControl::signal`
+  arms, the reader-thread and writer-thread master-lock acquisitions)
+  use `unwrap_or_else(|e| e.into_inner())` per convention 2.
+- **P10 — thread-spawn `.expect()` ×3** (`src/local/pty.rs`): the
+  reader/writer/waiter `thread::Builder::spawn` results map to
+  `TtyError::AllocFailed` (the error path existed — `allocate_pty`
+  returns `Result`); fd/thread exhaustion surfaces as an allocation
+  error, not a panic.
+- **P5 — scope gate first** (`src/adapter.rs`):
+  `validate_and_allocate` now runs the `tty:open` scope check before
+  the carriage/cmd checks and the backend lookup. An
+  authenticated-but-unscoped identity gets `forbidden` regardless of
+  the request body — the `unknown_backend`-vs-`forbidden` name-
+  enumeration differential is gone. The existing
+  `scope_gate_forbidden_without_tty_open` test asserted the scoped
+  behavior and still passes (it uses a *registered* backend name, so
+  the reorder doesn't change its outcome); no test asserted
+  `unknown_backend` with an unscoped identity.
+- **P11 — sentinel contract** (`src/session.rs`): `recv_stdout` now
+  terminates the stream ON the zero-length stdout sentinel ("drained")
+  — the sentinel is not yielded as an item. This matches the
+  documented contract and the sentinel's meaning; consumers no longer
+  filter. `recv_stderr`'s doc now states stderr has no sentinel (the
+  adapter's stderr pump emits none; that stream ends when the read
+  pump terminates). The three session tests that filtered the
+  sentinel manually now assert the stream ends on it (no filter).
+  The README example's manual `is_empty` break was removed.
+- **P2 — packaging** (`Cargo.toml`): `exclude` is now
+  `[".opencode/", "AGENTS.md", "docs/reviews/", "docs/plans/",
+  "docs/sdd_process.md"]` — AGENTS.md no longer ships (alkcall
+  parity), `docs/plans/` (absolute `/workspace/...` paths) no longer
+  ships, and both dead entries (`Cargo.lock` — force-included anyway;
+  `docs/research/` — nonexistent) are dropped. Package list verified:
+  42 files, 659.2KiB (was 44 files, 702.8KiB).
+- **P3 — stale docs**: AGENTS.md now says all five phases are landed,
+  the ADR range is 001..009 with the alktty-native ADR-009 noted in
+  the mapping list, and the alkcall guidance is corrected to "v0.4.x —
+  Pin `alkcall = "0.4.0"` (lockfile resolves 0.4.1)" (was "v0.1.x —
+  pin 0.1.1", which would have downgraded the dependency).
+  `docs/architecture/README.md` gained the ADR-009 table row, the
+  001..009 preamble, and the landed-phase status.
+- **P15 — polish batch** (one commit, as planned):
+  - `src/backend.rs` doc typo fixed ("`local` feature module) is the
+    reference implementation").
+  - `tokio-stream` dev-dependency removed (inherited from the main
+    dep).
+  - `src/adapter.rs` — the `NegotiationError::Io(_)` arm now logs
+    (`debug!`) for parity with the catch-all arm; `set_identity`
+    failure is logged (`AlreadySet` is benign-by-construction, noted
+    in a comment); `drop(input_pump)` → `input_pump.abort()` at
+    session end (no lingering client→backend task holding the backend
+    stdin half after the exit chunk).
+  - `src/session.rs` — `connect_direct`/`from_bidi_stream_via` map the
+    `accept_bi` `StreamError` into a new `TtySessionError::Open`
+    variant (`#[from]`) instead of flattening it into an anonymous
+    `io::Error(ConnectionReset)` (additive pre-consumers, same logic
+    as review #002 R5); the fail-fast params parse uses
+    `NegotiateRequest::deserialize(&params)` (borrowing) instead of
+    `from_value(params.clone())` — no clone.
+  - `src/negotiation.rs` — `error_response_bytes` skips an `"error"`
+    key in `fields` so it cannot overwrite the error code.
+  - Two trivial inline comments promoted or removed (the split-idiom
+    comment in `adapter.rs` removed; the "Already taken" comment in
+    `session.rs` now points at the documented behavior).
+  - `docs/plans/project-setup.md` test counts refreshed (119 lib + 19
+    integration = 138 under `--all-features` at the review baseline,
+    with a note that the suites have grown); `overview.md` and
+    `tty-adapter.md` front-matter `last_updated` dates and the ADR
+    range updated for the ADR-009 edits.
+  - The `input_pump.abort()` change surfaced a latent test race: the
+    `send_stdin_round_trips_to_backend` and
+    `resize_and_signal_dont_error` tests used `MockBackend` (exit
+    resolves on allocate) and were writing while the session data
+    plane was already closing — previously masked by the lingering
+    input pump the abort now reclaims. Both use a new `GatedBackend`
+    (exit held until the test releases it), making the writes
+    deterministic and asserting the exit code.
+
+Verification on the resolution tree: `cargo test` 104 lib (unchanged) /
+`cargo test --all-features` 128 lib + 19 integration = 147
+(unchanged; the two reworked tests kept their names); clippy
+(all-targets + wasm32) `-D warnings`, fmt, wasm check, doc
+(0 warnings), `cargo publish --dry-run --allow-dirty` (42 files,
+659.2KiB) — all clean.
+
+**Review #003 is closed.** Remaining before publish: tag `v0.1.0` (the
+CHANGELOG link definition points at it) and `cargo publish` (dry-run
+clean).
+
 ## Notes

 - Line numbers refer to the tree at `918af40` (review base).
@@ -130,7 +130,11 @@ impl ProtocolHandler for TtyAdapter {

    async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError> {
        if let Some(identity) = auth.identity.clone() {
-            let _ = connection.set_identity(identity);
+            // "Already set" is benign (one identity per connection by
+            // construction); anything else is a real wiring bug.
+            if let Err(e) = connection.set_identity(identity) {
+                debug!("tty: set_identity: {e}");
+            }
        }
        loop {
            let stream = match connection.accept_bi().await {
@@ -143,11 +147,6 @@ impl ProtocolHandler for TtyAdapter {
            let ownership = self.ownership.clone();
            let identity = auth.identity.clone();
            tokio::spawn(async move {
-                // `stream` is a `BiStream` (ADR-092) — `AsyncRead + AsyncWrite
-                // + Send + Unpin`. Split into halves for `drive_session`
-                // (which takes separate `AsyncWrite` + `AsyncRead` args). The
-                // split is the stdlib idiom for `TcpStream`-style duplex
-                // streams; no per-handler wrapper.
                let (client_read, client_write) = tokio::io::split(stream);
                let _ =
                    drive_session(client_write, client_read, backends, ownership, identity).await;
@@ -265,7 +264,12 @@ pub async fn drive_session_pre_negotiated(
 /// `enforce_scope` gates the `tty:open` scope check: `true` on the direct
 /// path (the adapter is the only gate), `false` on the channels path (the
 /// registry's `AccessControl` enforced the scope at open time — see
-/// [`drive_session_pre_negotiated`]).
+/// [`drive_session_pre_negotiated`]). The scope gate runs *first* (review
+/// #003 P5): checking it after the backend lookup would let an
+/// authenticated-but-unscoped identity distinguish registered backend
+/// names (`unknown_backend`) from unregistered ones (`forbidden`) — a
+/// name-enumeration differential. Gated first, every unscoped request
+/// gets the same `forbidden` regardless of the request body.
 #[allow(clippy::type_complexity)]
 async fn validate_and_allocate<W>(
    neg_writer: NegotiationWriter<W>,
@@ -278,6 +282,11 @@ async fn validate_and_allocate<W>(
 where
    W: AsyncWrite + Send + Unpin + 'static,
 {
+    if enforce_scope && !has_scope(identity, TTY_OPEN_SCOPE) {
+        send_negotiation_error(neg_writer, "forbidden", &[]).await;
+        return Err(());
+    }
+
    if req.carriage != "raw" {
        send_negotiation_error(
            neg_writer,
@@ -306,11 +315,6 @@ where
        }
    };

-    if enforce_scope && !has_scope(identity, TTY_OPEN_SCOPE) {
-        send_negotiation_error(neg_writer, "forbidden", &[]).await;
-        return Err(());
-    }
-
    let params = crate::backend::TtyParams::from(req);

    if let Some(provider) = ownership {
@@ -359,7 +363,10 @@ where
    let frame = match neg_reader.read_frame().await {
        Ok(f) => f,
        Err(NegotiationError::ConnectionClosed) => return Ok(()),
-        Err(NegotiationError::Io(_)) => return Ok(()),
+        Err(NegotiationError::Io(e)) => {
+            debug!("tty: negotiation read io error: {e}");
+            return Ok(());
+        }
        Err(NegotiationError::FrameTooLarge(_)) => {
            send_negotiation_error(
                neg_writer,
@@ -485,7 +492,10 @@ where
    }

    drop(writer_tx);
-    drop(input_pump);
+    // The client→backend pump outlives the session data plane only until
+    // the client disconnects; aborting it at session end is tighter — no
+    // lingering task holding the backend stdin half after the exit chunk.
+    input_pump.abort();

    let _ = drainer.await;
    debug!("tty: session complete");
@@ -251,8 +251,8 @@ impl TtyControlHandle {
 /// or `tokio::task::spawn_blocking` feeding tokio mpsc/oneshot channels.
 /// This bridging pattern is a **documented, supported implementation
 /// strategy**, not a workaround. The local backend (alktty's `local`
-/// the reference implementation: it spawns reader/writer/waiter threads
-/// that feed `mpsc::Receiver<Bytes>` (stdout), an `AsyncWrite` adapter
+/// feature module) is the reference implementation: it spawns
+/// reader/writer/waiter threads that feed `mpsc::Receiver<Bytes>` (stdout), an `AsyncWrite` adapter
 /// over `mpsc::Sender<StdinCmd>` (stdin), and a `oneshot::Receiver<i32>`
 /// wrapped in a kill-guard future (exit). The adapter consumes the bridged
 /// async-facing types and is unaware of the threading.
@@ -131,7 +131,7 @@ impl TtyControl for PtyControl {
            pixel_width,
            pixel_height,
        };
-        let master = self.master.lock().expect("master mutex poisoned");
+        let master = self.master.lock().unwrap_or_else(|e| e.into_inner());
        if let Err(e) = master.resize(size) {
            warn!("pty resize failed: {e}");
        }
@@ -172,7 +172,7 @@ impl TtyControl for PtyControl {
                }
            }
            // Unknown name or no pid: fall back to ChildKiller (SIGHUP).
-            let mut killer = self.killer.lock().expect("killer mutex poisoned");
+            let mut killer = self.killer.lock().unwrap_or_else(|e| e.into_inner());
            if let Err(e) = killer.kill() {
                warn!("pty fallback ChildKiller::kill failed: {e}");
            }
@@ -181,7 +181,7 @@ impl TtyControl for PtyControl {
        #[cfg(not(unix))]
        {
            let _ = name;
-            let mut killer = self.killer.lock().expect("killer mutex poisoned");
+            let mut killer = self.killer.lock().unwrap_or_else(|e| e.into_inner());
            if let Err(e) = killer.kill() {
                warn!("pty ChildKiller::kill failed: {e}");
            }
@@ -256,6 +256,10 @@ impl Drop for LocalExitFuture {
 /// When the channel is full, `poll_write` parks in an in-flight send
 /// future stored on the struct (so a re-poll resumes the same send rather
 /// than starting a new one — `reserve()` is not `Unpin`).
+/// `poll_shutdown` parks the same way (a separate slot, so the EOF send
+/// cannot be confused with a byte send) — returning `Pending` without a
+/// registered waker would strand the EOF forever on a full channel
+/// (review #003 P8).
 struct StdinSink {
    tx: mpsc::Sender<StdinCmd>,
    /// In-flight `reserve()` + send, captured as a boxed future. `None`
@@ -264,6 +268,9 @@ struct StdinSink {
    /// Bytes for the in-flight write (returned as the write count on
    /// completion).
    inflight_len: usize,
+    /// In-flight EOF send (`poll_shutdown`), distinct from the byte-write
+    /// slot so the two never share a future.
+    inflight_close: Option<InflightSend>,
    close_sent: bool,
 }

@@ -277,6 +284,7 @@ impl StdinSink {
            tx,
            inflight: None,
            inflight_len: 0,
+            inflight_close: None,
            close_sent: false,
        }
    }
@@ -338,17 +346,46 @@ impl tokio::io::AsyncWrite for StdinSink {

    fn poll_shutdown(
        mut self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
+        cx: &mut Context<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
        if self.close_sent {
            return Poll::Ready(Ok(()));
        }
+        // Drain any in-flight EOF send first (review #003 P8: returning
+        // `Pending` without polling the parked future registers no waker
+        // and the EOF would never be delivered on a full channel).
+        if let Some(fut) = self.inflight_close.as_mut() {
+            match fut.as_mut().poll(cx) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Ok(())) => {
+                    self.inflight_close = None;
+                    self.close_sent = true;
+                    return Poll::Ready(Ok(()));
+                }
+                Poll::Ready(Err(_)) => {
+                    // Channel closed: EOF is moot (the consumer is gone).
+                    self.inflight_close = None;
+                    self.close_sent = true;
+                    return Poll::Ready(Ok(()));
+                }
+            }
+        }
        match self.tx.try_send(StdinCmd::Eof) {
            Ok(()) => {
                self.close_sent = true;
                Poll::Ready(Ok(()))
            }
-            Err(mpsc::error::TrySendError::Full(_)) => Poll::Pending,
+            Err(mpsc::error::TrySendError::Full(_)) => {
+                // Park a `reserve + send` future so the poller's waker is
+                // registered; a later poll resumes the same send.
+                let tx = self.tx.clone();
+                self.inflight_close = Some(Box::pin(async move {
+                    let permit = tx.reserve().await?;
+                    permit.send(StdinCmd::Eof);
+                    Ok(())
+                }));
+                self.poll_shutdown(cx)
+            }
            Err(mpsc::error::TrySendError::Closed(_)) => {
                self.close_sent = true;
                Poll::Ready(Ok(()))
@@ -430,7 +467,7 @@ pub fn allocate_pty(
        .name("pty-reader".into())
        .spawn(move || {
            let reader = {
-                let m = reader_master.lock().expect("master mutex poisoned");
+                let m = reader_master.lock().unwrap_or_else(|e| e.into_inner());
                match m.try_clone_reader() {
                    Ok(r) => r,
                    Err(e) => {
@@ -462,7 +499,9 @@ pub fn allocate_pty(
            let _ = stdout_tx.blocking_send(Bytes::new());
            debug!("pty-reader thread done");
        })
-        .expect("spawn pty-reader");
+        .map_err(|e| TtyError::AllocFailed {
+            message: format!("spawn pty-reader thread: {e}"),
+        })?;

    // --- Writer thread: drain mpsc<StdinCmd> → blocking writes ---
    let writer_master = master.clone();
@@ -471,7 +510,7 @@ pub fn allocate_pty(
        .name("pty-writer".into())
        .spawn(move || {
            let writer = {
-                let m = writer_master.lock().expect("master mutex poisoned");
+                let m = writer_master.lock().unwrap_or_else(|e| e.into_inner());
                match m.take_writer() {
                    Ok(w) => w,
                    Err(e) => {
@@ -501,7 +540,9 @@ pub fn allocate_pty(
            }
            debug!("pty-writer thread done");
        })
-        .expect("spawn pty-writer");
+        .map_err(|e| TtyError::AllocFailed {
+            message: format!("spawn pty-writer thread: {e}"),
+        })?;

    // --- Waiter thread: blocking Child::wait() → oneshot<i32> ---
    let (exit_tx, exit_rx) = oneshot::channel::<i32>();
@@ -520,7 +561,9 @@ pub fn allocate_pty(
            debug!(exit_code = code, "pty-waiter: child reaped");
            let _ = exit_tx.send(code);
        })
-        .expect("spawn pty-waiter");
+        .map_err(|e| TtyError::AllocFailed {
+            message: format!("spawn pty-waiter thread: {e}"),
+        })?;

    // --- Assemble the TtyHandle ---
    let stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>> =
@@ -265,6 +265,12 @@ pub fn error_response_bytes(error: &str, fields: &[(&str, &str)]) -> serde_json:
    let mut map = serde_json::Map::new();
    map.insert("error".to_string(), json!(error));
    for (k, v) in fields {
+        // An `"error"` key in `fields` would overwrite the error code
+        // above (the consumer parses `error` out of the JSON object);
+        // drop the conflicting entry (review #003 P15).
+        if *k == "error" {
+            continue;
+        }
        map.insert((*k).to_string(), json!(v));
    }
    serde_json::to_vec(&map)
@@ -119,6 +119,12 @@ pub enum TtySessionError {
    /// The `Exit` control chunk's JSON payload failed to parse.
    #[error("malformed exit chunk: {0}")]
    MalformedExitChunk(String),
+    /// The transport refused the session open (the `Connection`'s
+    /// `accept_bi` failed on either constructor path). Carries the
+    /// upstream error type rather than flattening it into an
+    /// `io::Error` (review #003 P15 — additive pre-consumers).
+    #[error("session open failed: {0}")]
+    Open(#[from] alkcall::core::StreamError),
 }

 /// A live `alk/tty` session — the typed consumer-side handle.
@@ -188,9 +194,7 @@ impl TtySession {
        connection: Connection,
        negotiate: NegotiateRequest,
    ) -> Result<Self, TtySessionError> {
-        let stream = connection.accept_bi().await.map_err(|e| {
-            std::io::Error::new(std::io::ErrorKind::ConnectionReset, format!("{e}"))
-        })?;
+        let stream = connection.accept_bi().await?;
        Self::from_bidi_stream(stream, negotiate).await
    }

@@ -225,7 +229,12 @@ impl TtySession {
        client: &ChannelClient,
        params: serde_json::Value,
    ) -> Result<Self, TtySessionError> {
-        let _: NegotiateRequest = serde_json::from_value(params.clone())
+        // Borrowing deserialize (`&Value` implements `Deserializer`):
+        // fails fast before a channel is allocated, and the owned
+        // `params` still goes to `open_channel` (review #003 P15 — no
+        // clone).
+        use serde::Deserialize as _;
+        NegotiateRequest::deserialize(&params)
            .map_err(|e| TtySessionError::InvalidParams(e.to_string()))?;
        let (channel_id, send, recv) = client
            .open_channel(
@@ -261,9 +270,7 @@ impl TtySession {
    /// `Connection::from_source`). The negotiation already happened in
    /// the open op (ADR-009) — the stream starts in raw-chunk mode.
    async fn from_bidi_stream_via(channel_conn: Connection) -> Result<Self, TtySessionError> {
-        let stream = channel_conn.accept_bi().await.map_err(|e| {
-            std::io::Error::new(std::io::ErrorKind::ConnectionReset, format!("{e}"))
-        })?;
+        let stream = channel_conn.accept_bi().await?;
        let (read, write) = tokio::io::split(stream);
        Self::from_halves_raw(read, write).await
    }
@@ -430,8 +437,9 @@ impl TtySession {

    /// Get the stdout stream. Returns a `Stream<Item = Bytes>` that
    /// yields stdout chunks as they arrive. The stream ends when the
-    /// server's stdout reaches EOF (a zero-length stdout sentinel
-    /// chunk, see `tty-wire.md` §"Sentinels").
+    /// server's stdout reaches EOF — the zero-length stdout sentinel
+    /// chunk (`tty-wire.md` §"Sentinels") terminates the stream and is
+    /// NOT yielded as an item.
    ///
    /// This consumes the stdout receiver — calling it twice returns
    /// an empty stream the second time (the receiver is behind a
@@ -440,17 +448,27 @@ impl TtySession {
        let mut guard = self.stdout_rx.lock().await;
        if let Some(rx) = guard.take() {
            return Box::pin(futures::stream::unfold(rx, |mut rx| async move {
-                rx.recv().await.map(|bytes| (bytes, rx))
+                // The zero-length stdout chunk is the server's
+                // "drained" sentinel; end the stream on it rather than
+                // delivering it as an item (review #003 P11).
+                match rx.recv().await {
+                    Some(bytes) if bytes.is_empty() => None,
+                    Some(bytes) => Some((bytes, rx)),
+                    None => None,
+                }
            }));
        }
-        // Already taken — return an empty stream.
+        // Already taken — return an empty stream (documented on
+        // `recv_stdout`).
        Box::pin(futures::stream::empty())
    }

    /// Get the stderr stream. `None` for PTY-mode backends
    /// (stdout/stderr merged into stdout by the kernel PTY), or if
    /// already taken. The stream ends when the server's stderr reaches
-    /// EOF.
+    /// EOF. Stderr has no sentinel on the wire (the adapter's stderr
+    /// pump emits none — only stdout carries the drained sentinel), so
+    /// this stream ends when the read pump terminates.
    pub async fn recv_stderr(&self) -> Option<Pin<Box<dyn Stream<Item = Bytes> + Send>>> {
        let mut guard = self.stderr_rx.lock().await;
        let rx = guard.take()?;
@@ -710,25 +728,88 @@ mod tests {
        assert_eq!(code, 0);
    }

-    #[tokio::test]
-    async fn send_stdin_round_trips_to_backend() {
-        // Use a backend that captures stdin — but `MockBackend` doesn't
-        // expose the stdin channel to the test. This test just verifies
-        // `send_stdin` doesn't error; the adapter tests cover the
-        // stdin-to-backend pump.
-        let backend = Arc::new(MockBackend::with_exit_code(0));
+    /// A backend whose exit is held back until the test releases it, so
+    /// typed-method writes happen while the session data plane is
+    /// definitively open (the exit-resolves-immediately `MockBackend`
+    /// races them against session teardown — review #003 P15's
+    /// `input_pump.abort()` made that race fail loudly).
+    struct GatedBackend {
+        release: Arc<tokio::sync::Mutex<Option<tokio::sync::oneshot::Receiver<()>>>>,
+    }
+
+    #[async_trait::async_trait]
+    impl TtyBackend for GatedBackend {
+        async fn allocate(
+            &self,
+            _params: &crate::backend::TtyParams,
+        ) -> Result<crate::backend::TtyHandle, crate::backend::TtyError> {
+            use crate::backend::{TtyControlHandle, TtyHandle};
+
+            let (_stdout_tx, stdout_rx) = mpsc::channel::<Bytes>(8);
+            let release = self.release.lock().await.take();
+
+            let stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>> =
+                Box::pin(tokio_stream::wrappers::ReceiverStream::new(stdout_rx));
+            let stdin: Box<dyn AsyncWrite + Send + Unpin> = Box::new(tokio::io::sink());
+            let control = Some(TtyControlHandle::new(Arc::new(
+                crate::backend::MockControl::default(),
+            )));
+            let exit_code: crate::backend::BoxFuture<Result<i32, crate::backend::TtyError>> =
+                Box::pin(async move {
+                    // Hold the exit until released (dropped = cancel path).
+                    if let Some(rx) = release {
+                        let _ = rx.await;
+                    }
+                    Ok(0)
+                });
+
+            Ok(TtyHandle {
+                stdin,
+                stdout,
+                stderr: None,
+                exit_code,
+                control,
+            })
+        }
+    }
+
+    /// Wire a `GatedBackend` session; returns the release sender.
+    async fn wire_gated_session() -> (
+        TtySession,
+        tokio::sync::oneshot::Sender<()>,
+        tokio::task::JoinHandle<()>,
+    ) {
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
+        let backend = Arc::new(GatedBackend {
+            release: Arc::new(tokio::sync::Mutex::new(Some(release_rx))),
+        });
        let identity = Some(Identity {
            id: "alice".to_string(),
            scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
            resources: HashMap::new(),
        });
-        let (session, _server) = wire_session_and_server(backend, identity).await;
+        let (session, server) = wire_session_and_server(backend, identity).await;
+        (session, release_tx, server)
+    }
+
+    #[tokio::test]
+    async fn send_stdin_round_trips_to_backend() {
+        // `send_stdin`/`close_stdin` serialize and write while the
+        // session is open (the adapter tests cover the stdin-to-backend
+        // pump itself). The backend's exit is held until the writes are
+        // done so the data plane cannot race them.
+        let (session, release, _server) = wire_gated_session().await;
        session
            .send_stdin(Bytes::from_static(b"hello"))
            .await
            .expect("send_stdin");
        session.close_stdin().await.expect("close_stdin");
-        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()).await;
+        release.send(()).unwrap();
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait())
+            .await
+            .expect("wait didn't time out")
+            .expect("wait returns exit code");
+        assert_eq!(code, 0);
    }

    #[tokio::test]
@@ -736,17 +817,17 @@ mod tests {
        // The session writes control chunks; whether the backend
        // receives them is the adapter's concern (covered by the
        // adapter tests). This test verifies the typed methods
-        // serialize and write without error.
-        let backend = Arc::new(MockBackend::with_exit_code(0));
-        let identity = Some(Identity {
-            id: "alice".to_string(),
-            scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
-            resources: HashMap::new(),
-        });
-        let (session, _server) = wire_session_and_server(backend, identity).await;
+        // serialize and write without error, with the exit held until
+        // after the writes.
+        let (session, release, _server) = wire_gated_session().await;
        session.resize(80, 24, 0, 0).await.expect("resize");
        session.signal("INT").await.expect("signal");
-        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()).await;
+        release.send(()).unwrap();
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait())
+            .await
+            .expect("wait didn't time out")
+            .expect("wait returns exit code");
+        assert_eq!(code, 0);
    }

    #[tokio::test]
@@ -903,10 +984,10 @@ mod tests {
        let stdout = session.recv_stdout().await;
        let collected: Vec<Bytes> = stdout.collect().await;
        // The adapter emits a zero-length stdout sentinel after the
-        // backend stream ends; filter it out to assert the data chunks.
-        let data: Vec<Bytes> = collected.into_iter().filter(|b| !b.is_empty()).collect();
+        // backend stream ends; the stream terminates ON the sentinel
+        // (P11), so no filtering is needed to see the data chunks.
        assert_eq!(
-            data,
+            collected,
            vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")],
            "stdout chunks should route to the stdout stream"
        );
@@ -1023,9 +1104,11 @@ mod tests {

        let stdout = session.recv_stdout().await;
        let collected: Vec<Bytes> = stdout.collect().await;
-        let data: Vec<Bytes> = collected.into_iter().filter(|b| !b.is_empty()).collect();
+        // The stream ends on the drained sentinel (P11): the client→server
+        // stream types (0 and 3) from the server are discarded, and the
+        // empty stdout item is consumed as the terminator, not yielded.
        assert_eq!(
-            data,
+            collected,
            vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")],
            "stdout routing must be unaffected by the ignored chunks"
        );
@@ -1220,9 +1303,9 @@ mod tests {

        let stdout = session.recv_stdout().await;
        let collected: Vec<Bytes> = stdout.collect().await;
-        let data: Vec<Bytes> = collected.into_iter().filter(|b| !b.is_empty()).collect();
+        // The stream ends on the drained sentinel (P11).
        assert_eq!(
-            data,
+            collected,
            vec![Bytes::from_static(b"ch-out")],
            "stdout should route through the channels data plane"
        );