diff --git a/src/handler.rs b/src/handler.rs index 3f3e595..f3245b8 100644 --- a/src/handler.rs +++ b/src/handler.rs @@ -69,7 +69,30 @@ //! the store-and-take handoff (`channels.remove(&id)` → `into_stream` //! in the spawned serving task) is wave-3 mapping material; the //! routing decisions around it are already this wave's, because the -//! routing stubs decline at the seam. +//! routing stubs spawn their decline at the seam. +//! +//! ## The spawn-and-return structural constraint (ADR-006 decision 4) +//! +//! russh at this pin runs handler callbacks **inline inside the +//! session's packet loop** — a blocking handler suspends all packet +//! routing and deadlocks against the peer's follow-up bytes (POC-3 +//! proved by construction with receive-pack). Every serving path is +//! therefore spawned through [`crate::serve::spawn_serving_path`] and +//! the handler returns fast: the session-row seam stubs clone +//! `session.handle()`, spawn their decline through the clone +//! ([`crate::serve`]), and are gone from the hook's stack — no hook +//! awaits a serving path inline. The decline keeps its wire shape +//! (failure then close), riding the spawn. Exit/status reporting from +//! spawned tasks rides the same cloned handle +//! ([`crate::serve::report_exit_status_then_close`] — the mechanics; +//! the derivation rules are wave-3 `mapping.md` material), pinned +//! end-to-end by the mechanics rows in `tests/spawn_return.rs`. +//! +//! The open-level declines (unfeatured/never-feature rows, the tunnel +//! seam stub) stay inline by class: they answer the open *before* any +//! channel is owned — nothing is served yet, so nothing spawns. Wave +//! 3's accept-only-after-fabric-open flow is where the tunnel row +//! gains its spawned serving path, per the same constraint. //! //! Request-level declines: `subsystem_request` declines every name — //! `sftp` is the OQ-13 placeholder (the feature exists, the route @@ -101,7 +124,9 @@ //! usable (the POC-1 §5 session-survives-declines posture); a //! *serving-path* decline (the wave-3 seam stubs) answers failure //! and closes the channel — the vanilla-clean shape for a row that -//! will spawn its serving path next wave. +//! owns the channel. The serving-path decline now rides the spawned +//! serving task (ADR-006 decision 4, §The spawn-and-return +//! structural constraint above); the wire shape is unchanged. //! //! ## Test surface //! @@ -118,13 +143,16 @@ use std::sync::Arc; use alkcall::core::auth::Identity; use russh::keys::Algorithm; -use russh::server::{Auth, ChannelOpenHandle, Msg, PublicKeyOffer, Session, VerifiedPublicKey}; +use russh::server::{ + Auth, ChannelOpenHandle, Handle, Msg, PublicKeyOffer, Session, VerifiedPublicKey, +}; use russh::{Channel, ChannelId, Pty}; use crate::config::ServeConfig; use crate::dispatch; use crate::error::AdapterError; use crate::fingerprint::fingerprint; +use crate::serve; /// The per-connection handler (one per mounted stream): russh constructs /// nothing here — [`crate::SshAdapter::handle`] builds one per accepted @@ -403,14 +431,14 @@ impl russh::server::Handler for SessionHandler { match dispatch::classify_exec(command) { dispatch::ExecRoute::Git => { if self.serves_git_row() { - git_row_stub(channel, session).await; + git_row_stub(channel, session); } else { decline_unfeatured_request(channel, session).await; } } dispatch::ExecRoute::Pipe => { if self.serves_session_rows() { - pipe_row_stub(channel, session).await; + pipe_row_stub(channel, session); } else { decline_unfeatured_request(channel, session).await; } @@ -433,7 +461,7 @@ impl russh::server::Handler for SessionHandler { session: &mut Session, ) -> Result<(), Self::Error> { if self.serves_session_rows() { - pty_row_stub(channel, session).await; + pty_row_stub(channel, session); } else { decline_unfeatured_request(channel, session).await; } @@ -450,7 +478,7 @@ impl russh::server::Handler for SessionHandler { session: &mut Session, ) -> Result<(), Self::Error> { if self.serves_session_rows() { - shell_row_stub(channel, session).await; + shell_row_stub(channel, session); } else { decline_unfeatured_request(channel, session).await; } @@ -571,45 +599,57 @@ async fn decline_unfeatured_request(channel: ChannelId, session: &mut Session) { } /// The family-seam decline shape for a **routed-but-not-yet-served** -/// row: fail the request and close the channel — the vanilla-clean -/// shape for a serving path that would have owned the channel (wave 3 -/// replaces these stubs with the spawned serving paths; a declined -/// seam never leaks a half-served channel, since the decline IS the -/// row this wave). Decline-first-alloc-never holds structurally: -/// nothing is allocated before the failure lands. -async fn seam_stub_decline(channel: ChannelId, session: &mut Session) { - let _ = session.channel_failure(channel); - let _ = session.close(channel); +/// row — now the spawned serving task's body: fail the request and +/// close the channel, through the cloned handle the task was handed. +/// The vanilla-clean shape for a serving path that owns the channel +/// (wave 3 replaces these stubs with the real serving paths; a +/// declined seam never leaks a half-served channel, since the decline +/// IS the row this wave). Decline-first-alloc-never holds structurally: +/// nothing is allocated before the failure lands — the spawn's handle +/// clone is the only resource ever taken. +async fn seam_stub_decline(handle: Handle, channel: ChannelId) { + let _ = handle.channel_failure(channel).await; + let _ = handle.close(channel).await; +} + +/// Spawns the seam-stub decline as the row's serving path (ADR-006 +/// decision 4): the handle is cloned here, the decline runs on its own +/// task, and the hook returns — no serving path awaits inline. +fn spawn_seam_stub(channel: ChannelId, session: &Session) { + let handle = session.handle(); + serve::spawn_serving_path(seam_stub_decline(handle, channel)); } /// The git row's seam stub — wave 3 lands the alkgit seam here /// (`parse command → Service + repo → serve_duplex_resolved`), spawned -/// per ADR-006 decision 4; the skeleton declines at the seam. -async fn git_row_stub(channel: ChannelId, session: &mut Session) { - seam_stub_decline(channel, session).await; +/// per ADR-006 decision 4; the skeleton declines at the seam, on its +/// own task. +fn git_row_stub(channel: ChannelId, session: &Session) { + spawn_seam_stub(channel, session); } /// The session pipe row's seam stub — wave 3 lands the alktty pipe /// seam here (`wrapper ++ [cmd]`, the OQ-08 argv append), spawned per -/// ADR-006 decision 4; the skeleton declines at the seam. -async fn pipe_row_stub(channel: ChannelId, session: &mut Session) { - seam_stub_decline(channel, session).await; +/// ADR-006 decision 4; the skeleton declines at the seam, on its own +/// task. +fn pipe_row_stub(channel: ChannelId, session: &Session) { + spawn_seam_stub(channel, session); } /// The session pty row's seam stub — wave 3 lands the alktty /// `NegotiateRequest.tty` construction here /// (`drive_session_pre_negotiated`), spawned per ADR-006 decision 4; -/// the skeleton declines at the seam. -async fn pty_row_stub(channel: ChannelId, session: &mut Session) { - seam_stub_decline(channel, session).await; +/// the skeleton declines at the seam, on its own task. +fn pty_row_stub(channel: ChannelId, session: &Session) { + spawn_seam_stub(channel, session); } /// The session shell row's seam stub — wave 3 lands the alktty /// `NegotiateRequest.cmd` = serve-config shell argv here (OQ-08, no /// crate default), spawned per ADR-006 decision 4; the skeleton -/// declines at the seam. -async fn shell_row_stub(channel: ChannelId, session: &mut Session) { - seam_stub_decline(channel, session).await; +/// declines at the seam, on its own task. +fn shell_row_stub(channel: ChannelId, session: &Session) { + spawn_seam_stub(channel, session); } /// The tunnel row's seam stub: the explicit `ConnectFailed` reject diff --git a/src/lib.rs b/src/lib.rs index de6ea28..95ad99b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,6 +32,7 @@ mod fingerprint; mod handler; mod host_key; mod key_source; +mod serve; pub use adapter::SshAdapter; pub use config::{DialConfig, Limits, ServeConfig}; @@ -46,3 +47,9 @@ pub use fingerprint::fingerprint; pub use handler::SessionHandler; pub use host_key::{HostKeyPolicy, PresentedHostKey, RejectAll}; pub use key_source::{HostKeySource, KeySource, VaultHostKeySource, VaultKeySource}; +/// Contract-test surface (`#[doc(hidden)]`): the spawn-and-return +/// plumbing (`ADR-006 decision 4`, see `serve.rs`) — the mechanics the +/// mechanics-proof rows in `tests/spawn_return.rs` compose over the +/// real wire to pin the constraint as tests. Not documented API. +#[doc(hidden)] +pub use serve::{report_exit_status_then_close, spawn_serving_path}; diff --git a/src/serve.rs b/src/serve.rs new file mode 100644 index 0000000..d3771fa --- /dev/null +++ b/src/serve.rs @@ -0,0 +1,93 @@ +//! The spawn-and-return plumbing (`w2-spawn-return`) — ADR-006 +//! decision 4, the structural constraint of the whole spine, in code. +//! +//! russh at this pin runs server handler callbacks **inline inside the +//! session's packet loop** — a blocking handler suspends all packet +//! routing for the connection and deadlocks against the peer's +//! immediate follow-up bytes (POC-3 proved by construction with +//! receive-pack). The crate's answer is mechanical, not conventional: +//! +//! - **Every serving path is spawned** ([`spawn_serving_path`]) — the +//! handler hook clones the session handle +//! (`Session::handle()`), hands it into its serving future, spawns, +//! and returns. No serving path is ever awaited inline; the dispatch +//! seam stubs ([`crate::handler`]) hang off this helper, so the +//! wave-3 mappings inherit the pattern mechanically from the stubs. +//! - **Exit/status reporting rides the cloned handle** — the spawned +//! task's only report surface is the handle clone it was handed; +//! [`report_exit_status_then_close`] preserves the wire ordering the +//! `mapping.md` `exit_status` row pins: `exit-status` sent **before** +//! close, from the spawned task. +//! +//! The *derivation contract's rules* (which exit code → which SSH exit +//! shape, `mapping.md` §Exit chunk → SSH exit reporting rules 1–6) are +//! wave-3 session-mapping material; this module is the mechanics only. +//! +//! ## Test surface +//! +//! Both helpers are re-exported `#[doc(hidden)]` from the crate root — +//! the `handler.rs` §Test surface pattern: the mechanics-proof rows in +//! `tests/spawn_return.rs` compose them over the real wire (a minimal +//! proof handler standing in for the crate's own auth/dispatch hooks, +//! which have their own contract suites) to pin the constraint as +//! tests, not convention. + +use std::future::Future; + +use russh::ChannelId; +use russh::server::Handle; + +/// Spawns one serving path on its own task (ADR-006 decision 4): the +/// caller — the handler hook — clones the session handle, hands it +/// into the serving future, calls this, and returns immediately. +/// Nothing here is ever awaited inline: an inline serve deadlocks +/// against the peer's follow-up bytes (POC-3 proved by construction). +pub fn spawn_serving_path(serving: impl Future + Send + 'static) { + tokio::spawn(serving); +} + +/// The spawned task's exit report through the cloned handle: +/// `exit-status(N)` then close — the ordering contract +/// (`mapping.md`'s `exit_status` row: sent **before** close, from the +/// spawned task). A failed send means the session is gone; nothing is +/// reportable then. +pub async fn report_exit_status_then_close(handle: &Handle, channel: ChannelId, exit_code: u32) { + let _ = handle.exit_status_request(channel, exit_code).await; + let _ = handle.close(channel).await; +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + use super::spawn_serving_path; + + /// - *Contract stamp:* ADR-006 decision 4 (spawn-and-return: the + /// handler call returns while the serving path is still running + /// — handler-returns-fast is structural). + #[tokio::test] + async fn spawn_returns_while_the_serving_path_is_still_pending() { + let (release, waited) = tokio::sync::oneshot::channel::<()>(); + let done = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&done); + + spawn_serving_path(async move { + let _ = waited.await; + flag.store(true, Ordering::Release); + }); + + assert!( + !done.load(Ordering::Acquire), + "the spawn call returned while the serving path was still gated — \ + no serving path is awaited inline" + ); + + release + .send(()) + .expect("the gate releases exactly once, to the still-running path"); + while !done.load(Ordering::Acquire) { + tokio::task::yield_now().await; + } + } +} diff --git a/tasks/implementation/w2-spawn-return.md b/tasks/implementation/w2-spawn-return.md index c90f536..ef1152a 100644 --- a/tasks/implementation/w2-spawn-return.md +++ b/tasks/implementation/w2-spawn-return.md @@ -1,7 +1,7 @@ --- id: w2-spawn-return name: Wave 2 — spawn-and-return plumbing + exit/status reporting mechanics -status: pending +status: completed depends_on: [w2-dispatch-decline-core] scope: narrow risk: medium @@ -42,16 +42,16 @@ mechanically from the stubs this task leaves behind. ## Acceptance Criteria -- [ ] Every serving path (the dispatch stubs) runs in a +- [x] Every serving path (the dispatch stubs) runs in a `tokio::spawn`; no handler hook awaits a serving path inline -- [ ] Exit/status reporting from spawned tasks rides the cloned +- [x] Exit/status reporting from spawned tasks rides the cloned `Session::handle()`; `exit-status` is sent before channel close -- [ ] A harness test proves the mechanics end to end: the client +- [x] A harness test proves the mechanics end to end: the client observes `exit-status(N)` from a spawned stub that completed after the handler returned -- [ ] A handler-returns-fast assertion pins the structural constraint +- [x] A handler-returns-fast assertion pins the structural constraint as a test -- [ ] `cargo test`, `cargo clippy --all-targets -- -D warnings`, +- [x] `cargo test`, `cargo clippy --all-targets -- -D warnings`, `cargo fmt --check` pass ## References @@ -67,8 +67,101 @@ mechanically from the stubs this task leaves behind. ## Notes -> To be filled by implementation agent +- `src/serve.rs` (new): the spawn-and-return plumbing as *mechanics*, + not convention — `spawn_serving_path(serving future)` (the one + choke point for the ADR-006 decision-4 spawn; nothing it takes is + ever awaited inline) and `report_exit_status_then_close(&Handle, + channel, N)` (the cloned-handle report with the `mapping.md` + `exit_status` row's ordering baked in: exit-status **before** + close). Both re-exported `#[doc(hidden)]` from the crate root — the + `SessionHandler` §Test-surface pattern — as the contract-test + surface the mechanics rows compose over the real wire. A unit row + (`spawn_returns_while_the_serving_path_is_still_pending` — + latch-gated future, spawn call returns first) pins spawn-then-return + directly. The derivation contract's *rules* (which exit code → which + SSH exit shape) stay wave-3 `mapping.md` material; no rule logic + landed here. +- `src/handler.rs`: the four session-row seam stubs (`git_row_stub`, + `pipe_row_stub`, `pty_row_stub`, `shell_row_stub`) are now spawned + serving paths — `spawn_seam_stub` clones `session.handle()`, + `seam_stub_decline(handle, channel)` runs the failure-then-close + decline on the spawned task, and the hooks return without awaiting + anything. A compile-level fact coded around: the row stubs are + plain `fn`s (not `async`) — an `async fn` holding `&Session` across + the spawn makes the handler future non-`Send` (russh requires `+ + Send` on handler futures; `Session` is not `Sync`), so the clone is + taken synchronously and only the handle moves into the spawned + future. Wire shapes are unchanged (all `w2-dispatch-decline-core` + rows pass unmodified with the spawned decline); the tunnel seam stub + and every open-level/capability decline stay inline by class — they + answer the open request *before* any channel is owned, so nothing + spawns yet (documented in the new handler module-doc section §The + spawn-and-return structural constraint; wave 3's + accept-only-after-fabric-open flow is where the tunnel row gains + its spawn). +- `tests/spawn_return.rs` (new): the end-to-end mechanics proof over + the real wire — the russh server stack (duplex transport, harness + client/harness transport posture) mounted over a minimal proof + handler that is *not* the crate's `SessionHandler` (the auth and + dispatch hooks have their own suites; the proof subject is the + mechanics) and that serves execs by composing the exported + plumbing: `spawn_serving_path` of a release-gated serving future + reporting `exit-status(7)` + close through the cloned handle. Two + rows, each with a `Contract stamp:`: + - `handler_returns_fast_while_the_spawned_serving_path_is_pending` — + the structural constraint as a wire test: after the exec hook has + provably returned (a bounded wait on the handler-side timestamp; + the client's exec returns when the request is on the wire, not + when the hook ran) and with the serving path still gated, the + client's *follow-up traffic* (a second session channel + second + exec) is routed and dispatched normally — POC-3's follow-up-bytes + deadlock shape inverted — and the still-gated no-report state is + asserted before the release. + - `spawned_stub_reports_exit_status_before_close_after_the_handler_returned` + — the client observes `ExitStatus{7}` strictly before `Close` + (message-by-message, no draining that could hide ordering), and + the recorded timestamps prove the exit-status left *after* the + handler had already returned (the literal criterion). +- **The proof-handler design note:** the gate is `tokio::sync::watch`- + based (no lost-wakeup race — a late subscriber sees the release, + which a `Notify`-with-`notify_waiters` shape could miss against + multiple spawned paths); the handler-side timestamps ride + `parking_lot::Mutex`. +- Feature independence: the mechanics are feature-independent (the + helpers exist in every build and the proof rows run in every build + they were verified under — default, `git`, `session`, `tunnel`, + `sftp`, `--all-features`); no feature-gated code was added, so no + new decline-path coverage obligations arise. ## Summary -> To be filled on completion \ No newline at end of file +Landed the spawn-and-return plumbing + exit/status reporting +mechanics (`w2-spawn-return`) in three files: + +- **`src/serve.rs`** — `cargo test`-pinned mechanics: the + `spawn_serving_path` choke point (the ADR-006 decision-4 rule, in + code, never awaited inline) and `report_exit_status_then_close` + (exit-status before close through the cloned `Session::handle()`), + both `#[doc(hidden)]` contract-test re-exports; a unit row pins + spawn-returns-while-pending directly. +- **`src/handler.rs`** — the four session-row seam stubs became + spawned serving paths (handle clone taken synchronously — the + non-`Send` shape that forcing `async fn` stubs would produce is + coded around and recorded); failure-then-close wire shapes unchanged + (all dispatch-core rows pass unmodified); open-level declines and + the tunnel seam stub stay inline by class, documented; a new + module-doc section carries the constraint and the wave-3 + inherit-the-pattern note. +- **`tests/spawn_return.rs`** — the end-to-end mechanics proof: two + `Contract stamp:` rows over the real wire (handler-returns-fast + under follow-up traffic with the spawn still pending; exit-status + before close, timestamps proving the report outlived the handler's + return), driven by a proof handler composing the exported plumbing. + +Verification (all green): `cargo test` · `cargo test +--features {git,session,tunnel,sftp} --test dispatch_core` · +`cargo test --all-features` · `cargo clippy --all-targets -- -D +warnings` · `cargo clippy --all-features --all-targets -- -D +warnings` · `cargo fmt --check` · `cargo doc --no-deps` (no +warnings). 93 lib/unit rows + the per-build dispatch matrices + the +two mechanics rows across 7 test binaries. \ No newline at end of file diff --git a/tests/spawn_return.rs b/tests/spawn_return.rs new file mode 100644 index 0000000..16a114e --- /dev/null +++ b/tests/spawn_return.rs @@ -0,0 +1,371 @@ +//! The spawn-and-return mechanics proof (`w2-spawn-return`) — the +//! structural constraint (ADR-006 decision 4) pinned as tests, over +//! the real wire. +//! +//! russh at this pin runs server handler callbacks inline inside the +//! session's packet loop; a blocking handler deadlocks against the +//! peer's follow-up bytes (POC-3 proved by construction with +//! receive-pack). The crate's plumbing +//! ([`alkssh::spawn_serving_path`], +//! [`alkssh::report_exit_status_then_close`] — the `#[doc(hidden)]` +//! contract-test re-exports from `src/serve.rs`) makes the escape +//! mechanical: the handler hook clones `session.handle()`, spawns the +//! serving future, and returns. +//! +//! These rows compose exactly that plumbing end to end: a minimal +//! proof handler stands in for the crate's own auth/dispatch hooks +//! (their correctness is `w2-auth-bridge`/`w2-dispatch-decline-core` +//! contract territory; the `dispatch` seam stubs now spawn through the +//! same plumbing, so the crate's own rows exercise the spawn too — +//! `tests/dispatch_core.rs`). What is under test here is the +//! *mechanics*: the spawn-then-return, the cloned-handle reporting, +//! and the wire ordering — `exit-status` before close, arriving after +//! the handler already returned, while the session's packet loop keeps +//! routing the connection's follow-up traffic. + +#![allow(dead_code)] + +mod common; + +use std::sync::Arc; +use std::time::Instant; + +use common::fixture; +use common::harness::{ + HarnessClientHandler, authenticate_publickey, client_config, next_message, open_session_channel, +}; +use parking_lot::Mutex; +use russh::keys::Algorithm; +use russh::server::{Auth, ChannelOpenHandle, Session}; +use russh::{Channel, ChannelId, ChannelMsg, Disconnect, Error, MethodKind, MethodSet}; + +use alkssh::Limits; + +/// The exit status the proof stub reports — an arbitrary distinct +/// value the rows assert verbatim (`exit-status(7)`). +const PROOF_EXIT: u32 = 7; + +/// The command the rows exec — anything; the proof handler ignores it. +const PROOF_COMMAND: &str = "mechanics-proof-serving-path"; + +/// The async completion gate the spawned serving paths hold on — +/// released only after the rows have proven the handler returned. +/// `watch`-based (no lost-wakeup race: a late subscriber still sees +/// the release). +fn serving_gate() -> tokio::sync::watch::Sender { + tokio::sync::watch::channel(false).0 +} + +/// The spawned task's wait: blocks until (or returns if) released — +/// `borrow_and_update` marks the release seen so the wakeup is never +/// lost or replayed. +async fn gate_wait(rx: &mut tokio::sync::watch::Receiver) { + if !*rx.borrow_and_update() { + rx.changed().await.expect("the gate lives for the row"); + } +} + +/// What the proof handler records — the timestamps the rows compare. +#[derive(Default)] +struct ProofRecord { + handler_returned_at: Option, + exit_reported_at: Option, +} + +type Shared = Arc>; + +/// The proof handler (server side): mechanics-only — it authenticates +/// the fixture alpha by the crate's one fingerprint path, accepts +/// session channels, and serves every exec by *the crate's own +/// spawn-and-return plumbing*: a serving path gated on the row's +/// release, reporting `exit-status(7)` then close through the cloned +/// handle. +struct ProofServer { + gate: Arc>, + record: Shared, +} + +impl russh::server::Handler for ProofServer { + type Error = Error; + + async fn auth_publickey_offered( + &mut self, + _user: &str, + _offer: russh::server::PublicKeyOffer<'_>, + ) -> Result { + Ok(Auth::Accept) + } + + /// The mechanics rows' own fixture check — fingerprint equality + /// against fixture alpha through the crate's fingerprint path + /// (not the bridge contract; that is `w2-auth-bridge`'s suite). + async fn auth_publickey( + &mut self, + _user: &str, + key: russh::server::VerifiedPublicKey<'_>, + ) -> Result { + let accepted = *key.algorithm == Algorithm::Ed25519 + && alkssh::fingerprint(key.public_key) == fixture("alpha").fingerprint; + Ok(if accepted { + Auth::Accept + } else { + Auth::reject() + }) + } + + async fn channel_open_session( + &mut self, + _channel: Channel, + reply: ChannelOpenHandle, + _session: &mut Session, + ) -> Result<(), Self::Error> { + reply.accept().await; + Ok(()) + } + + /// The hook under proof: clone the handle, spawn the serving path + /// (gated — it stays pending past this method's return), record + /// the handler's return instant, and go. + async fn exec_request( + &mut self, + channel: ChannelId, + _data: &[u8], + session: &mut Session, + ) -> Result<(), Self::Error> { + let handle = session.handle(); + let record = Arc::clone(&self.record); + let mut gate_rx = self.gate.subscribe(); + + alkssh::spawn_serving_path(async move { + gate_wait(&mut gate_rx).await; + alkssh::report_exit_status_then_close(&handle, channel, PROOF_EXIT).await; + record.lock().exit_reported_at = Some(Instant::now()); + }); + + self.record.lock().handler_returned_at = Some(Instant::now()); + Ok(()) + } +} + +/// One mounted mechanics-proof session: the duplex pair, the spawned +/// server task (russh's own stack over the proof handler), and the +/// connected client. Transport and posture mirror the loopback +/// harness (its recorded decisions); the handler is intentionally not +/// the crate's `SessionHandler` — the crate stack under a custom +/// serving hook is the proof's subject. +struct ProofSession { + client: russh::client::Handle, + gate: Arc>, + record: Shared, + server_task: tokio::task::JoinHandle>, +} + +/// Mounts one proof session: the russh server stack over the proof +/// handler, the harness client against it, the fixture alpha +/// authenticated. +async fn mount() -> ProofSession { + let gate = Arc::new(serving_gate()); + let record: Shared = Arc::new(Mutex::new(ProofRecord::default())); + + let config = Arc::new(russh::server::Config { + methods: MethodSet::from(&[MethodKind::PublicKey][..]), + keys: vec![fixture("bravo").private_key()], + inactivity_timeout: Some(Limits::default().inactivity_timeout), + ..russh::server::Config::default() + }); + + let (client_io, server_io) = tokio::io::duplex(common::harness::DUPLEX_CAPACITY); + let server = ProofServer { + gate: Arc::clone(&gate), + record: Arc::clone(&record), + }; + let server_task = tokio::spawn(async move { + let running = russh::server::run_stream(config, server_io, server) + .await + .expect("the proof server mounts"); + running.await + }); + + let policy = Arc::new(common::MemHostKeyPolicy::with_default(true)); + let mut client = russh::client::connect_stream( + client_config(Limits::default()), + client_io, + HarnessClientHandler::new(Arc::clone(&policy)), + ) + .await + .expect("the client mounts against the proof server"); + + let result = authenticate_publickey(&mut client, "mechanics-proof", fixture("alpha")).await; + assert!( + result.success(), + "the fixture identity authenticates before every mechanics row" + ); + + ProofSession { + client, + gate, + record, + server_task, + } +} + +impl ProofSession { + /// Releases the gate — the spawned serving paths complete + /// asynchronously from here on. + fn release_gate(&self) { + let _ = self.gate.send(true); + } + + /// Ends a row cleanly: disconnect, wait for the server task, and + /// require a clean session end (the harness posture). + async fn unmount(self) { + self.client + .disconnect(Disconnect::ByApplication, "row done", "en") + .await + .expect("client disconnects cleanly"); + let outcome = self.server_task.await.expect("server task ends"); + outcome.expect("clean session end"); + } + + /// Waits (bounded) until the exec hook has provably run and + /// returned — the client's `exec` returns when the request is on + /// the wire; the loop's processing of it is observed here. + async fn exec_hook_has_returned(&self) { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while self.record.lock().handler_returned_at.is_none() { + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + } + }) + .await + .expect("the exec hook returns promptly"); + } + + /// The handler provably returned while the spawned serving path + /// was still pending — the handler-returns-fast assertion. + fn assert_handler_returned_spawn_still_pending(&self, context: &str) { + let record = self.record.lock(); + assert!( + record.handler_returned_at.is_some(), + "{context}: the exec hook must already have returned" + ); + assert!( + record.exit_reported_at.is_none(), + "{context}: the spawned serving path is still gated — nothing reported yet" + ); + } +} + +// --------------------------------------------------------------------------- +// Rows. +// --------------------------------------------------------------------------- + +/// - *Contract stamp:* ADR-006 decision 4 (the structural constraint: +/// every serving path is spawned — the handler completes while the +/// spawned task is still running) + mapping.md §The structural +/// constraint: spawn-and-return (the follow-up-bytes deadlock +/// inverted: the connection's loop keeps routing while a serving +/// path is pending). +#[tokio::test] +async fn handler_returns_fast_while_the_spawned_serving_path_is_pending() { + let session = mount().await; + + let mut channel = open_session_channel(&session.client) + .await + .expect("the session channel opens"); + channel + .exec(false, PROOF_COMMAND) + .await + .expect("the exec request rides the wire"); + session.exec_hook_has_returned().await; + session.assert_handler_returned_spawn_still_pending("after the exec"); + + // The follow-up-bytes inversion — POC-3's deadlock shape: the peer + // sends immediate follow-up traffic and the loop routes it while + // the first serving path is still pending. + let follow_up = open_session_channel(&session.client) + .await + .expect("the packet loop routes while the spawn is pending"); + follow_up + .exec(false, PROOF_COMMAND) + .await + .expect("a second exec dispatches while the first serving path is pending"); + session.assert_handler_returned_spawn_still_pending("after the follow-up traffic"); + + session.release_gate(); + + let first = next_message(&mut channel) + .await + .expect("the released serving path reports"); + assert!( + matches!( + first, + ChannelMsg::ExitStatus { + exit_status: PROOF_EXIT + } + ), + "the spawned stub reports exit-status({PROOF_EXIT}), got {first:?}" + ); + + session.unmount().await; +} + +/// - *Contract stamp:* mapping.md §exit_status row (sent **before** +/// close, from the spawned task via `Session::handle()`) + ADR-006 +/// decision 4 (exit/status reporting rides the cloned handle). +#[tokio::test] +async fn spawned_stub_reports_exit_status_before_close_after_the_handler_returned() { + let session = mount().await; + + let mut channel = open_session_channel(&session.client) + .await + .expect("the session channel opens"); + let handler_returned_at = { + channel + .exec(false, PROOF_COMMAND) + .await + .expect("the exec request rides the wire"); + session.exec_hook_has_returned().await; + let handler_returned_at = session + .record + .lock() + .handler_returned_at + .expect("the exec hook returned before this row observes anything"); + session.release_gate(); + handler_returned_at + }; + + // Strict wire order, message by message: exit-status first, close + // after — the ordering contract verbatim. + let first = next_message(&mut channel) + .await + .expect("the released serving path reports"); + assert!( + matches!( + first, + ChannelMsg::ExitStatus { + exit_status: PROOF_EXIT + } + ), + "exit-status precedes close, got {first:?}" + ); + let exit_reported_at = session.record.lock().exit_reported_at; + + let second = next_message(&mut channel) + .await + .expect("the channel closes after its exit report"); + assert!( + matches!(second, ChannelMsg::Close), + "close follows the exit report, got {second:?}" + ); + + // The literal criterion: the client observes exit-status that left + // the server *after* the handler had already returned. + let exit_reported_at = exit_reported_at.expect("the report timestamped"); + assert!( + exit_reported_at > handler_returned_at, + "the spawned task reported exit-status after the handler returned \ + (handler {handler_returned_at:?} vs report {exit_reported_at:?})" + ); + + session.unmount().await; +}