test(tty): wire-witness the channel_close cancel path + the no-exit-chunk close shape, record the write_all no-deadline limit (w4-tty-coverage)

- wire row: a client close on a served pipe row fires the per-channel
  cancel (kill-on-drop witnessed before teardown, no exit report
  fabricated, ended lifecycle event) and releases the cap (over-cap
  open refused at 1, re-open admits + routes after the close)
- wire row: the Closed no-exit-chunk close shape, constructed through
  the real path (an oversized shim chunk dies at alktty's own
  ChunkWriter validation -> bridge EOF, scripted exit never forwarded,
  plain close, nothing fabricated); Break's frame classification
  pinned by a unit row (wire-unreachable through the validated writer
  at this pin - recorded)
- module docs: the stdout write_all no-deadline known limit recorded
  (the gate's disposition: bounded backpressure, inactivity-timeout
  liveness; a deadline would be a behavior change)

Verification: cargo test (default/session/session+git/git/tunnel/sftp/
--all-features, tty rows 5x repeat), clippy x2 -D warnings, fmt, doc,
taskgraph validate

src/tty_map.rs

@@ -102,6 +102,15 @@
 //!   nothing (the chunk is the sole fidelity input; its absence is
 //!   neither 0 nor 255 — no fabricated report): plain close, and
 //!   the lifecycle's ended event.
+//! - **Known limit (recorded disposition, wave-3 gate booked item 3):
+//!   the stdout [`tokio::io::AsyncWriteExt::write_all`] carries no
+//!   deadline.** Backpressure is bounded (the bounded bridging duplex,
+//!   russh's per-channel window and the client's own flow control bound
+//!   what can pile up) and liveness is covered by the inactivity
+//!   timeout ([`crate::Limits::inactivity_timeout`], always set) — a
+//!   deadline here would be a behavior change (a slow client may stall
+//!   its stream, vanilla sshd's posture included), not a coverage
+//!   minor. So the limit is recorded, not "fixed".
 //!
 //! ## The control rows
 //!
@@ -893,4 +902,30 @@ mod tests {
         assert_eq!(sig_control_name(&Sig::USR1), "USR1");
         assert_eq!(sig_control_name(&Sig::Custom("ABRT".to_owned())), "ABRT");
     }
+
+    /// - *Contract stamp:* the no-exit-chunk close family's frame
+    ///   classification (`w4-tty-coverage` — the wire row in
+    ///   `tests/tty_mapping.rs` lands the family's serving outcome): a
+    ///   clean read-end classifies `Closed`, an oversized claimed
+    ///   length classifies `Break` — bounds-checked both directions,
+    ///   never an oversized read. `Break`'s serving arm is one body
+    ///   with `Closed`'s; through alktty's own validated `ChunkWriter`
+    ///   a `Break` frame is unreachable on the duplex (the drainer
+    ///   rejects the oversized chunk locally and the read sees EOF —
+    ///   the wire row's construction), so the classification rows pin
+    ///   the defensive arm at the frame layer.
+    #[tokio::test]
+    async fn clean_ends_classify_closed_and_oversized_lengths_classify_break() {
+        let mut clean: &[u8] = &[];
+        assert!(matches!(
+            super::read_bridge_frame(&mut clean).await,
+            super::BridgeFrame::Closed
+        ));
+
+        let mut oversized: &[u8] = &[0x01, 0x01, 0x00, 0x00, 0x01];
+        assert!(matches!(
+            super::read_bridge_frame(&mut oversized).await,
+            super::BridgeFrame::Break
+        ));
+    }
 }

tasks/implementation/w4-tty-coverage.md

@@ -1,7 +1,7 @@
 ---
 id: w4-tty-coverage
 name: Wave 4 — tty coverage minors (wave-3 gate booked item 3)
-status: pending
+status: completed
 depends_on: [w4-channel-not-held-rows]
 scope: narrow
 risk: low
@@ -37,12 +37,12 @@ settled file).
 
 ## Acceptance Criteria
 
-- [ ] Wire row: the `channel_close`-fired per-channel cancel path
+- [x] Wire row: the `channel_close`-fired per-channel cancel path
       (cap release + state teardown observed on the wire)
-- [ ] Wire row(s): the Closed/Break no-exit-chunk close shape
-- [ ] The stdout `write_all` no-deadline limit recorded in module
+- [x] Wire row(s): the Closed/Break no-exit-chunk close shape
+- [x] The stdout `write_all` no-deadline limit recorded in module
       docs (no code change)
-- [ ] `cargo test` (+ per-feature + `--all-features`), clippy ×2
+- [x] `cargo test` (+ per-feature + `--all-features`), clippy ×2
       (`-D warnings`), fmt, `cargo doc --no-deps` pass
 
 ## References
@@ -56,8 +56,73 @@ settled file).
 
 ## Notes
 
-> To be filled by implementation agent
+- **Test + module-docs only, as planned** — no production code change;
+  the shapes held on the wire exactly as documented.
+- **`channel_close`-fired cancel path** —
+  `a_client_close_on_a_served_row_fires_the_per_channel_cancel_and_releases_the_cap`
+  (`tests/tty_mapping.rs`, cfg `session`): a served pipe row under a
+  cap of 1; a second open is refused (`AdministrativelyProhibited` —
+  the admission held), the client then closes the served channel, and
+  the row witnesses (a) the per-channel state teardown — the shim's
+  kill-on-drop witness fires (the bridge surface's cancel → the
+  serving task aborts the alktty session), polled *before* unmount so
+  it is the `channel_close` site, not the teardown; (b) no exit report
+  fabricated on the wire (plain close / channel end); (c) the serving
+  path's ended lifecycle event; (d) the **cap release** — the next
+  open admits and routes (release site 1, disjoint from the
+  bridge-failure and arrival sites).
+- **Closed/Break no-exit-chunk close shape** —
+  `a_bridge_end_without_an_exit_chunk_reports_nothing_and_closes_plain`
+  (same file): the shape is constructed deterministically through the
+  real path — the shim's stdout carries a payload beyond alktty's own
+  `MAX_CHUNK_LEN` wire bound; alktty's `ChunkWriter` rejects it
+  locally, its drainer shuts the write half down, and the bridge's
+  next read is EOF (`BridgeFrame::Closed`) with the scripted exit
+  (`Some(0)`) never forwarded. The row pins: the in-bounds chunk rode
+  first, the end is a plain close, **no exit-status/exit-signal ever
+  rides** (the absent-report posture — neither 0 nor 255), the session
+  truly served (an allocation happened, so this is the close shape,
+  not a decline), and the ended lifecycle event landed.
+- **Break's classification row** —
+  `clean_ends_classify_closed_and_oversized_lengths_classify_break`
+  (`src/tty_map.rs` test mod): Break is **unreachable on the wire**
+  through alktty's own validated `ChunkWriter` (an oversized chunk
+  fails local validation → the drainer break → the read sees EOF —
+  exactly the Closed row's construction; a non-EOF duplex read error
+  has no wire-constructible producer at this pin, and the
+  serving-path body of Closed/Break is one arm). So the family's
+  *serving* outcome is wire-witnessed by the Closed row, and the
+  frame-level `Break` classification (bounds-checked, never an
+  oversized read) is pinned at the frame layer by the unit row —
+  recorded here as the reason one wire row + one unit row cover the
+  "row(s)".
+- **The stdout `write_all` no-deadline limit** — recorded in
+  `src/tty_map.rs`'s module docs (§the translation bridge): the gate's
+  recorded disposition verbatim — backpressure bounded, liveness
+  covered by the inactivity timeout, a deadline would be a behavior
+  change, not a coverage minor. No code change.
+- A `session_with_limits` / `session_config(limits)` driver variant
+  was added to the binary (the cap row's ceiling of 1); `session_with`
+  delegates unchanged.
 
 ## Summary
 
-> To be filled on completion
+The wave-3 gate's booked item 3 is closed: the `channel_close`-fired
+per-channel cancel path is now wire-witnessed (a served pipe row under
+a cap of 1: the refused over-cap open pins the admission, the client
+close pins the cancel — the kill-on-drop witness fires before any
+teardown, no exit report is fabricated, the ended lifecycle event
+lands — and the succeeding re-open pins the cap release from the
+established-channel-close release site); the Closed/Break
+no-exit-chunk close shape landed its wire row (constructed through
+the real path via an oversized shim chunk dying at alktty's own
+`ChunkWriter` validation: EOF without an exit chunk → plain close,
+nothing fabricated, the scripted exit never forwarded) plus the
+frame-level Break classification row (Break is wire-unreachable
+through alktty's validated writer at this pin — recorded); and the
+stdout `write_all` no-deadline limit is recorded in the module docs
+per the gate's disposition. No production code change. Verified:
+`cargo test` (default, `session`, `session,git`, `git`, `tunnel`,
+`sftp`, `--all-features` — the tty rows' 5×-repeat flake check
+included), clippy ×2 (`-D warnings`), `cargo fmt --check`,
+`cargo doc --no-deps` 0 warnings, `taskgraph validate` ✓ (35 tasks).

tests/tty_mapping.rs

@@ -22,8 +22,10 @@ mod serving {
     use std::sync::Arc;
 
     use alktty::backend::TerminalParams;
+    use alktty::wire::MAX_CHUNK_LEN;
+    use russh::ChannelOpenFailure;
     use russh::client::Msg;
-    use russh::{Channel, ChannelMsg, Sig};
+    use russh::{Channel, ChannelMsg, Error, Sig};
 
     use super::common::event_capture as ec;
     use super::common::harness::{
@@ -187,8 +189,9 @@ mod serving {
     pub(super) fn session_config(
         provider: Arc<MemIdentityProvider>,
         mutate: impl FnOnce(&mut ServeConfig),
+        limits: Limits,
     ) -> ServeConfig {
-        let mut serve = harness_serve_config(provider, &[fixture("bravo")], Limits::default());
+        let mut serve = harness_serve_config(provider, &[fixture("bravo")], limits);
         serve.shell_argv = Some(vec!["/bin/none".to_owned(), "-login".to_owned()]);
         serve.exec_wrapper_argv = Some(vec!["/bin/wrap".to_owned(), "-x".to_owned()]);
         serve.session_cwd = Some(PathBuf::from("/srv/session"));
@@ -202,12 +205,21 @@ mod serving {
     /// Mounts an authenticated loopback session over the given config
     /// — the rows' driver (fixture `alpha` → `ssh-rider`).
     pub(super) async fn session_with(mutate: impl FnOnce(&mut ServeConfig)) -> LoopbackSession {
+        session_with_limits(mutate, Limits::default()).await
+    }
+
+    /// [`session_with`] over non-default [`Limits`] — the rows that
+    /// tune a budget (the channel-cap release row's ceiling of 1).
+    pub(super) async fn session_with_limits(
+        mutate: impl FnOnce(&mut ServeConfig),
+        limits: Limits,
+    ) -> LoopbackSession {
         let provider = Arc::new(MemIdentityProvider::new());
         provider.register(
             fixture("alpha").fingerprint,
             identity("ssh-rider", &["ssh"]),
         );
-        let adapter = SshAdapter::new(session_config(provider, mutate))
+        let adapter = SshAdapter::new(session_config(provider, mutate, limits))
             .await
             .expect("the session build's adapter constructs");
         let policy = Arc::new(MemHostKeyPolicy::with_default(true));
@@ -528,6 +540,176 @@ mod serving {
         session.unmount().await;
     }
 
+    // ------------------------------------------------------------------
+    // The channel_close-fired per-channel cancel path on the wire
+    // (`w4-tty-coverage`, wave-3 gate booked item 3): the cap release +
+    // the per-channel state teardown, wire-witnessed — the disjoint
+    // release sites the gate verified per row (the wire rows here
+    // witness the established-channel-close site, `channel_cap.rs`
+    // release site 1).
+    // ------------------------------------------------------------------
+
+    /// - *Contract stamp:* mapping.md §Bounded resources /
+    ///   `channel_cap.rs` release site 1 + `handler.rs`
+    ///   §the control rows' per-channel surface (a client close on a
+    ///   served row fires the `channel_close` hook: the bridge surface
+    ///   fires the per-channel cancel — the serving task aborts the
+    ///   alktty session, alktty's kill-on-drop contract fires, nothing
+    ///   fabricated on the wire — and the cap releases, disjoint from
+    ///   the bridge-failure and arrival sites).
+    #[tokio::test]
+    async fn a_client_close_on_a_served_row_fires_the_per_channel_cancel_and_releases_the_cap() {
+        ec::install();
+        let shim = Arc::new(td::TtyShim::new());
+        shim.set_script(td::ShimScript {
+            stdout: vec![],
+            stderr: vec![],
+            exit: None,
+        });
+        let session = session_with_limits(
+            shim_seam(&shim),
+            Limits {
+                max_concurrent_channels: 1,
+                ..Limits::default()
+            },
+        )
+        .await;
+
+        let mut served = open_session_channel(&session.client)
+            .await
+            .expect("the serving channel opens (the cap admits)");
+        request_exec(&served, "uptime")
+            .await
+            .expect("the exec routed");
+        expect_success(&mut served, "the served pipe exec").await;
+
+        let refused = open_session_channel(&session.client)
+            .await
+            .expect_err("beyond the cap of 1: refused at the open");
+        assert!(
+            matches!(
+                refused,
+                Error::ChannelOpenFailure(ChannelOpenFailure::AdministrativelyProhibited)
+            ),
+            "the cap-exceeded refusal is the AdministrativelyProhibited open decline, \
+             got {refused:?}"
+        );
+
+        served.close().await.expect("the client close rides");
+
+        for _ in 0..200 {
+            if !shim.observation().kills.lock().is_empty() {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        }
+        assert!(
+            !shim.observation().kills.lock().is_empty(),
+            "the close-fired per-channel cancel aborted the alktty session — \
+             alktty's kill-on-drop contract fired (witnessed before unmount, \
+             so this is the channel_close path, not the teardown)"
+        );
+        match next_observation(&mut served).await {
+            Some(Observed::Close) | None => {}
+            other => panic!("the cancel fabricates no exit report, got {other:?}"),
+        }
+        let _ = ec::await_event(&[
+            ("row", "exec_request"),
+            ("message", "serving path ended"),
+            ("exec_route", "pipe"),
+        ])
+        .await;
+
+        let mut recovered = open_session_channel(&session.client)
+            .await
+            .expect("the cap recovered after the close — the channel_close hook released");
+        request_exec(&recovered, "uptime")
+            .await
+            .expect("the recovered channel routes");
+        expect_success(&mut recovered, "the recovered channel serves").await;
+
+        session.unmount().await;
+    }
+
+    // ------------------------------------------------------------------
+    // The Closed/Break no-exit-chunk close shape on the wire
+    // (`w4-tty-coverage`): a bridge read that ends without an exit
+    // chunk reports nothing — the derivation contract's absent-report
+    // posture pinned, not just the happy paths. (The frame-level
+    // Closed/Break classification rows live in `src/tty_map.rs`'s test
+    // mod; the Break classification is unreachable through alktty's
+    // own validated `ChunkWriter` — a defensive arm.)
+    // ------------------------------------------------------------------
+
+    /// - *Contract stamp:* mapping.md §Exit chunk (the contract's
+    ///   close-shape class: the bridge read ends **without** an exit
+    ///   chunk → no report, plain close, the lifecycle's ended event —
+    ///   its absence is neither 0 nor 255). The shape is constructed
+    ///   through the real path: the shim's stdout carries a payload
+    ///   beyond alktty's own `MAX_CHUNK_LEN` wire bound — alktty's
+    ///   `ChunkWriter` rejects it locally, its drainer shuts the write
+    ///   half down, and the bridge's next read is EOF (`Closed`) with
+    ///   the scripted exit never forwarded. Nothing fabricated.
+    #[tokio::test]
+    async fn a_bridge_end_without_an_exit_chunk_reports_nothing_and_closes_plain() {
+        ec::install();
+        let shim = Arc::new(td::TtyShim::new());
+        shim.set_script(td::ShimScript {
+            stdout: vec![b"bridge-ok".to_vec(), vec![0u8; MAX_CHUNK_LEN as usize + 1]],
+            stderr: vec![],
+            exit: Some(0),
+        });
+        let session = session_with(shim_seam(&shim)).await;
+
+        let mut channel = open_session_channel(&session.client)
+            .await
+            .expect("the channel opens");
+        request_exec(&channel, "uptime")
+            .await
+            .expect("the exec routed");
+        expect_success(&mut channel, "the routed exec").await;
+
+        assert!(
+            !shim.observation().allocations.lock().is_empty(),
+            "the session truly served (the no-report close shape, not a decline)"
+        );
+        let mut saw_data = false;
+        let mut saw_close = false;
+        loop {
+            match next_observation(&mut channel).await {
+                Some(Observed::Data(data)) if data == b"bridge-ok".to_vec() => {
+                    saw_data = true;
+                }
+                Some(Observed::Close) => {
+                    saw_close = true;
+                    break;
+                }
+                Some(Observed::ExitStatus(_)) | Some(Observed::ExitSignal { .. }) => panic!(
+                    "the no-exit-chunk close shape fabricates no exit report, \
+                     the scripted exit never rode"
+                ),
+                None => break,
+                Some(Observed::Data(other)) => {
+                    panic!("unexpected data after the chunk plane died: {other:?}")
+                }
+                Some(_) => continue,
+            }
+        }
+        assert!(
+            saw_data,
+            "the in-bounds chunk rode SSH data before the chunk plane died"
+        );
+        assert!(saw_close, "the Closed shape ends with a plain close");
+        let _ = ec::await_event(&[
+            ("row", "exec_request"),
+            ("message", "serving path ended"),
+            ("exec_route", "pipe"),
+        ])
+        .await;
+
+        session.unmount().await;
+    }
+
     #[tokio::test]
     async fn stdin_bytes_and_the_eof_sentinel_reach_the_backend() {
         let shim = Arc::new(td::TtyShim::new());