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.
This commit is contained in:
glm-5.3-flash committed 2026-09-05 16:47:45 +00:00
1 parent 7b0bac0671
commit 16f6d07038
14 files changed
+372 -105

No files matched your search

+15 -10
View File
@@ -246,16 +246,18 @@ wasm-clean" invariant — run it whenever a non-`local` module changes.
## Architecture Context ## Architecture Context
- `docs/plans/project-setup.md` — the current plan (phases 0–5). Phase 0 - `docs/plans/project-setup.md` — the current plan (phases 0–5). All
(scaffold hygiene), Phase 1 (port core types), Phase 2 (channels five phases are landed: Phase 0 (scaffold hygiene), Phase 1 (port
integration + `TtySession`), Phase 3 (`local` backend), and Phase 4 core types), Phase 2 (channels integration + `TtySession`), Phase 3
(architecture docs + BAST schema + renumbered ADRs) are landed. (`local` backend), Phase 4 (architecture docs + BAST schema +
Phase 5 (tests, including integration tests in `tests/` at the crate renumbered ADRs), and Phase 5 (tests, including the integration
root) is not yet done. tests in `tests/` at the crate root).
- `docs/architecture/` is created in Phase 4. The alknet ADRs - `docs/architecture/` is created in Phase 4. The alknet ADRs
referenced below (052, 053, 054, 055, 056, 057, 077, 093) are ported referenced below (052, 053, 054, 055, 056, 057, 077, 093) are ported
and renumbered into alktty's ADR range (001..008) in and renumbered into alktty's ADR range (001..009) in
`docs/architecture/decisions/`. The alknet originals at `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 `/workspace/@alkdev/alknet/docs/architecture/decisions/` remain the
authoritative source for any ADR not yet ported — read them before authoritative source for any ADR not yet ported — read them before
non-trivial changes to the wire format, trait, or adapter. 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) for historical context with its reversal notice pointing to 008)
- ADR-093 → 008 — TTY always uses its 5-byte format inside channels - ADR-093 → 008 — TTY always uses its 5-byte format inside channels
(reverses ADR-077; channels strips its 8-byte header transparently) (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, - 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. 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 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`, `BiStream`, `BidiStreamSource`, `AuthContext`, `Identity`,
`IdentityProvider`, `AccessControl`, `OwnershipProvider`, `IdentityProvider`, `AccessControl`, `OwnershipProvider`,
`OwnershipStore`, `InMemoryOwnershipStore`, `HandlerError`, `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 breaking changes are expected at this major-zero stage; this is the
first real consumer, so we find and fix issues upstream rather than 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.
+19 -5
View File
@@ -88,11 +88,14 @@ alknet mono-repo.
`send_stdin`/`close_stdin` (stdin chunks / EOF sentinel), `send_stdin`/`close_stdin` (stdin chunks / EOF sentinel),
`recv_stdout`/`recv_stderr` (`Stream<Item = Bytes>`), `resize`, `recv_stdout`/`recv_stderr` (`Stream<Item = Bytes>`), `resize`,
`signal`, and `wait` (awaits the `Exit` control chunk; watch-based, `signal`, and `wait` (awaits the `Exit` control chunk; watch-based,
no lost wakeup). Both constructors disambiguate the first response no lost wakeup). `recv_stdout` ends ON the zero-length drained
frame (`0x00` prefix = negotiation error frame → `NegotiationRejected` sentinel (it is not yielded as an item). Both constructors
with the server's error code; raw chunk = session proceeds). disambiguate the first response frame (`0x00` prefix = negotiation
`TtySessionError` is `#[non_exhaustive]`. Dropping the session aborts error frame → `NegotiationRejected` with the server's error code;
the read pump and closes the write half. 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 - **Control messages.** `ControlMessage` (`Resize`/`Signal`/`Eof` on
`ctrl_in`; `Exit` on `ctrl_out`) — JSON, tagged by `"type"`, unknown `ctrl_in`; `Exit` on `ctrl_out`) — JSON, tagged by `"type"`, unknown
types ignored by policy. `signal_from_name` maps the common signal 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 regression fails the test instead of hanging it, plus exact-boundary
(`MAX_CHUNK_LEN`) round-trip and server-sends-client-direction- (`MAX_CHUNK_LEN`) round-trip and server-sends-client-direction-
stream-types discard tests. 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 [0.1.0]: https://git.alk.dev/alkdev/alktty/releases/tag/v0.1.0
+1 -2
View File
@@ -9,7 +9,7 @@ repository = "https://git.alk.dev/alkdev/alktty"
readme = "README.md" readme = "README.md"
keywords = ["tty", "terminal", "pty", "channels", "alkcall"] keywords = ["tty", "terminal", "pty", "channels", "alkcall"]
categories = ["network-programming", "asynchronous"] 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] [lib]
name = "alktty" name = "alktty"
@@ -48,7 +48,6 @@ libc = "0.2"
[dev-dependencies] [dev-dependencies]
tokio = { version = "1", features = ["full", "test-util", "macros"] } tokio = { version = "1", features = ["full", "test-util", "macros"] }
tokio-stream = "0.1"
serde_json = "1" serde_json = "1"
tempfile = "3" tempfile = "3"
bytes = "1" bytes = "1"
+2 -1
View File
@@ -73,7 +73,8 @@ session.close_stdin().await?;
let mut stdout = session.recv_stdout().await; let mut stdout = session.recv_stdout().await;
while let Some(bytes) = futures::StreamExt::next(&mut 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 ... // ... render bytes ...
} }
let code = session.wait().await?; let code = session.wait().await?;
+6 -6
View File
@@ -19,7 +19,7 @@ Syntax Tree) document for the wire format, and the ADRs.
## Applicable ADRs ## Applicable ADRs
Ported from the alknet mono-repo and renumbered into alktty's ADR 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 `/workspace/@alkdev/alknet/docs/architecture/decisions/` remain the
authoritative source for any ADR not yet ported, and for the alknet authoritative source for any ADR not yet ported, and for the alknet
ADRs referenced by alknet number in the docs below (which are not 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 | | [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) | | [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) | | [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 ## Key Design Principles
@@ -135,11 +136,10 @@ tty-specific and therefore not ported into alktty's ADR range).
## References ## References
- `docs/plans/project-setup.md` — the current plan (phases 0–5). - `docs/plans/project-setup.md` — the current plan (phases 0–5). All
Phases 0–3 are landed; Phase 4 (this directory: architecture docs + five phases are landed: Phase 4 landed this directory (architecture
BAST schema + renumbered ADRs) is landed by this commit; Phase 5 docs + BAST schema + renumbered ADRs); Phase 5 landed the tests,
(tests, including integration tests in `tests/` at the crate root) is including the integration tests in `tests/` at the crate root.
not yet done.
- alknet originals of the ported ADRs (alknet ADR-052, 053, 054, 055, - alknet originals of the ported ADRs (alknet ADR-052, 053, 054, 055,
056, 057, 077, 093) at 056, 057, 077, 093) at
`/workspace/@alkdev/alknet/docs/architecture/decisions/` — the `/workspace/@alkdev/alknet/docs/architecture/decisions/` — the
+2 -2
View File
@@ -2,8 +2,8 @@
status: draft (ported from alknet 2026-08-17; alknet-tty → alktty, status: draft (ported from alknet 2026-08-17; alknet-tty → alktty,
alknet-tty-local → alktty's `local` feature module, alknet/tty → alknet-tty-local → alktty's `local` feature module, alknet/tty →
alk/tty, alknet-core → alkcall::core, alknet-call → alkcall, ADRs alk/tty, alknet-core → alkcall::core, alknet-call → alkcall, ADRs
renumbered 052..093 → 001..008) renumbered 052..093 → 001..009)
last_updated: 2026-08-17 last_updated: 2026-09-05
--- ---
# alktty — Overview # alktty — Overview
+4 -4
View File
@@ -1,8 +1,8 @@
--- ---
status: draft (ported from alknet 2026-08-17; alknet-tty → alktty, status: draft (ported from alknet 2026-08-17; alknet/tty → alk/tty,
alknet/tty → alk/tty, alknet-core → alkcall::core, alknet-call → alknet-core → alkcall::core, alknet-call → alkcall, ADRs renumbered
alkcall, ADRs renumbered 052..093 → 001..008) 052..093 → 001..009)
last_updated: 2026-08-17 last_updated: 2026-09-05
--- ---
# alktty — TtyAdapter and Session Lifecycle # alktty — TtyAdapter and Session Lifecycle
+4 -2
View File
@@ -527,8 +527,10 @@ either). `AGENTS.md` was updated to match this mapping.
`MockBackend` from `backend.rs` for the producer side so no `MockBackend` from `backend.rs` for the producer side so no
real PTY is needed. real PTY is needed.
Total: 80 lib tests + 19 integration tests = 99 passing under Total: 119 lib tests + 19 integration tests = 138 passing under
`--all-features`; 80 lib tests under default (wasm-clean) build. `--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 ## Open Questions
+114 -10
View File
@@ -1,7 +1,7 @@
--- ---
status: open status: closed
last_updated: 2026-09-05 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: reviewed_artifacts:
- src/lib.rs - src/lib.rs
- src/wire.rs - src/wire.rs
@@ -30,6 +30,7 @@ base_commit: 918af40 (review #002 resolved)
resolutions: resolutions:
- "P1 + P13 resolved 2026-09-05 (87c52e5) — see the Session-1 Resolution section" - "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" - "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) # Code Review #003 — Prepublish Review (v0.1.0)
@@ -467,20 +468,20 @@ theirs; don't replicate that nit.
| ID | Finding | Fix | Effort | Risk | Status | | ID | Finding | Fix | Effort | Risk | Status |
|----|---------|-----|--------|------|--------| |----|---------|-----|--------|------|--------|
| P1 | pump_session deadlock ≥63 chunks | concurrent drainer + regression test | small | medium (core pump) | ✅ resolved (`87c52e5`) | | 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 | | 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 | open | | 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) | | 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) | | 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) | | 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 | | P8 | poll_shutdown missing waker | inflight-future pattern | small | low | ✅ resolved (Session 3) |
| P9 | poisoned-lock `.expect()` × 5 | `into_inner()` | trivial | none | open | | P9 | poisoned-lock `.expect()` × 5 | `into_inner()` | trivial | none | ✅ resolved (Session 3) |
| P10 | thread-spawn `.expect()` × 3 | map to `TtyError` | trivial | none | open | | 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) | open | | 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) | | 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) | | 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 | | 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) | | P16 | no CHANGELOG.md | start changelog with [0.1.0] | small | none | ✅ resolved (Session 2) |
## Suggested Session Breakdown ## 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, (0 warnings), `cargo publish --dry-run --allow-dirty` (44 files,
702.8KiB) — all clean. 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 ## Notes
- Line numbers refer to the tree at `918af40` (review base). - Line numbers refer to the tree at `918af40` (review base).
+24 -14
View File
@@ -130,7 +130,11 @@ impl ProtocolHandler for TtyAdapter {
async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError> { async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError> {
if let Some(identity) = auth.identity.clone() { 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 { loop {
let stream = match connection.accept_bi().await { let stream = match connection.accept_bi().await {
@@ -143,11 +147,6 @@ impl ProtocolHandler for TtyAdapter {
let ownership = self.ownership.clone(); let ownership = self.ownership.clone();
let identity = auth.identity.clone(); let identity = auth.identity.clone();
tokio::spawn(async move { 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 (client_read, client_write) = tokio::io::split(stream);
let _ = let _ =
drive_session(client_write, client_read, backends, ownership, identity).await; 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 /// `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 /// path (the adapter is the only gate), `false` on the channels path (the
/// registry's `AccessControl` enforced the scope at open time — see /// 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)] #[allow(clippy::type_complexity)]
async fn validate_and_allocate<W>( async fn validate_and_allocate<W>(
neg_writer: NegotiationWriter<W>, neg_writer: NegotiationWriter<W>,
@@ -278,6 +282,11 @@ async fn validate_and_allocate<W>(
where where
W: AsyncWrite + Send + Unpin + 'static, 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" { if req.carriage != "raw" {
send_negotiation_error( send_negotiation_error(
neg_writer, 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); let params = crate::backend::TtyParams::from(req);
if let Some(provider) = ownership { if let Some(provider) = ownership {
@@ -359,7 +363,10 @@ where
let frame = match neg_reader.read_frame().await { let frame = match neg_reader.read_frame().await {
Ok(f) => f, Ok(f) => f,
Err(NegotiationError::ConnectionClosed) => return Ok(()), 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(_)) => { Err(NegotiationError::FrameTooLarge(_)) => {
send_negotiation_error( send_negotiation_error(
neg_writer, neg_writer,
@@ -485,7 +492,10 @@ where
} }
drop(writer_tx); 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; let _ = drainer.await;
debug!("tty: session complete"); debug!("tty: session complete");
+2 -2
View File
@@ -251,8 +251,8 @@ impl TtyControlHandle {
/// or `tokio::task::spawn_blocking` feeding tokio mpsc/oneshot channels. /// or `tokio::task::spawn_blocking` feeding tokio mpsc/oneshot channels.
/// This bridging pattern is a **documented, supported implementation /// This bridging pattern is a **documented, supported implementation
/// strategy**, not a workaround. The local backend (alktty's `local` /// strategy**, not a workaround. The local backend (alktty's `local`
/// the reference implementation: it spawns reader/writer/waiter threads /// feature module) is the reference implementation: it spawns
/// that feed `mpsc::Receiver<Bytes>` (stdout), an `AsyncWrite` adapter /// reader/writer/waiter threads that feed `mpsc::Receiver<Bytes>` (stdout), an `AsyncWrite` adapter
/// over `mpsc::Sender<StdinCmd>` (stdin), and a `oneshot::Receiver<i32>` /// over `mpsc::Sender<StdinCmd>` (stdin), and a `oneshot::Receiver<i32>`
/// wrapped in a kill-guard future (exit). The adapter consumes the bridged /// wrapped in a kill-guard future (exit). The adapter consumes the bridged
/// async-facing types and is unaware of the threading. /// async-facing types and is unaware of the threading.
+53 -10
View File
@@ -131,7 +131,7 @@ impl TtyControl for PtyControl {
pixel_width, pixel_width,
pixel_height, 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) { if let Err(e) = master.resize(size) {
warn!("pty resize failed: {e}"); warn!("pty resize failed: {e}");
} }
@@ -172,7 +172,7 @@ impl TtyControl for PtyControl {
} }
} }
// Unknown name or no pid: fall back to ChildKiller (SIGHUP). // 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() { if let Err(e) = killer.kill() {
warn!("pty fallback ChildKiller::kill failed: {e}"); warn!("pty fallback ChildKiller::kill failed: {e}");
} }
@@ -181,7 +181,7 @@ impl TtyControl for PtyControl {
#[cfg(not(unix))] #[cfg(not(unix))]
{ {
let _ = name; 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() { if let Err(e) = killer.kill() {
warn!("pty ChildKiller::kill failed: {e}"); 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 /// 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 /// future stored on the struct (so a re-poll resumes the same send rather
/// than starting a new one — `reserve()` is not `Unpin`). /// 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 { struct StdinSink {
tx: mpsc::Sender<StdinCmd>, tx: mpsc::Sender<StdinCmd>,
/// In-flight `reserve()` + send, captured as a boxed future. `None` /// 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 /// Bytes for the in-flight write (returned as the write count on
/// completion). /// completion).
inflight_len: usize, 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, close_sent: bool,
} }
@@ -277,6 +284,7 @@ impl StdinSink {
tx, tx,
inflight: None, inflight: None,
inflight_len: 0, inflight_len: 0,
inflight_close: None,
close_sent: false, close_sent: false,
} }
} }
@@ -338,17 +346,46 @@ impl tokio::io::AsyncWrite for StdinSink {
fn poll_shutdown( fn poll_shutdown(
mut self: Pin<&mut Self>, mut self: Pin<&mut Self>,
_cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> { ) -> Poll<Result<(), std::io::Error>> {
if self.close_sent { if self.close_sent {
return Poll::Ready(Ok(())); 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) { match self.tx.try_send(StdinCmd::Eof) {
Ok(()) => { Ok(()) => {
self.close_sent = true; self.close_sent = true;
Poll::Ready(Ok(())) 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(_)) => { Err(mpsc::error::TrySendError::Closed(_)) => {
self.close_sent = true; self.close_sent = true;
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
@@ -430,7 +467,7 @@ pub fn allocate_pty(
.name("pty-reader".into()) .name("pty-reader".into())
.spawn(move || { .spawn(move || {
let reader = { 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() { match m.try_clone_reader() {
Ok(r) => r, Ok(r) => r,
Err(e) => { Err(e) => {
@@ -462,7 +499,9 @@ pub fn allocate_pty(
let _ = stdout_tx.blocking_send(Bytes::new()); let _ = stdout_tx.blocking_send(Bytes::new());
debug!("pty-reader thread done"); 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 --- // --- Writer thread: drain mpsc<StdinCmd> → blocking writes ---
let writer_master = master.clone(); let writer_master = master.clone();
@@ -471,7 +510,7 @@ pub fn allocate_pty(
.name("pty-writer".into()) .name("pty-writer".into())
.spawn(move || { .spawn(move || {
let writer = { 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() { match m.take_writer() {
Ok(w) => w, Ok(w) => w,
Err(e) => { Err(e) => {
@@ -501,7 +540,9 @@ pub fn allocate_pty(
} }
debug!("pty-writer thread done"); 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> --- // --- Waiter thread: blocking Child::wait() → oneshot<i32> ---
let (exit_tx, exit_rx) = oneshot::channel::<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"); debug!(exit_code = code, "pty-waiter: child reaped");
let _ = exit_tx.send(code); let _ = exit_tx.send(code);
}) })
.expect("spawn pty-waiter"); .map_err(|e| TtyError::AllocFailed {
message: format!("spawn pty-waiter thread: {e}"),
})?;
// --- Assemble the TtyHandle --- // --- Assemble the TtyHandle ---
let stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>> = let stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>> =
+6
View File
@@ -265,6 +265,12 @@ pub fn error_response_bytes(error: &str, fields: &[(&str, &str)]) -> serde_json:
let mut map = serde_json::Map::new(); let mut map = serde_json::Map::new();
map.insert("error".to_string(), json!(error)); map.insert("error".to_string(), json!(error));
for (k, v) in fields { 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)); map.insert((*k).to_string(), json!(v));
} }
serde_json::to_vec(&map) serde_json::to_vec(&map)
+120 -37
View File
@@ -119,6 +119,12 @@ pub enum TtySessionError {
/// The `Exit` control chunk's JSON payload failed to parse. /// The `Exit` control chunk's JSON payload failed to parse.
#[error("malformed exit chunk: {0}")] #[error("malformed exit chunk: {0}")]
MalformedExitChunk(String), 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. /// A live `alk/tty` session — the typed consumer-side handle.
@@ -188,9 +194,7 @@ impl TtySession {
connection: Connection, connection: Connection,
negotiate: NegotiateRequest, negotiate: NegotiateRequest,
) -> Result<Self, TtySessionError> { ) -> Result<Self, TtySessionError> {
let stream = connection.accept_bi().await.map_err(|e| { let stream = connection.accept_bi().await?;
std::io::Error::new(std::io::ErrorKind::ConnectionReset, format!("{e}"))
})?;
Self::from_bidi_stream(stream, negotiate).await Self::from_bidi_stream(stream, negotiate).await
} }
@@ -225,7 +229,12 @@ impl TtySession {
client: &ChannelClient, client: &ChannelClient,
params: serde_json::Value, params: serde_json::Value,
) -> Result<Self, TtySessionError> { ) -> 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()))?; .map_err(|e| TtySessionError::InvalidParams(e.to_string()))?;
let (channel_id, send, recv) = client let (channel_id, send, recv) = client
.open_channel( .open_channel(
@@ -261,9 +270,7 @@ impl TtySession {
/// `Connection::from_source`). The negotiation already happened in /// `Connection::from_source`). The negotiation already happened in
/// the open op (ADR-009) — the stream starts in raw-chunk mode. /// the open op (ADR-009) — the stream starts in raw-chunk mode.
async fn from_bidi_stream_via(channel_conn: Connection) -> Result<Self, TtySessionError> { async fn from_bidi_stream_via(channel_conn: Connection) -> Result<Self, TtySessionError> {
let stream = channel_conn.accept_bi().await.map_err(|e| { let stream = channel_conn.accept_bi().await?;
std::io::Error::new(std::io::ErrorKind::ConnectionReset, format!("{e}"))
})?;
let (read, write) = tokio::io::split(stream); let (read, write) = tokio::io::split(stream);
Self::from_halves_raw(read, write).await Self::from_halves_raw(read, write).await
} }
@@ -430,8 +437,9 @@ impl TtySession {
/// Get the stdout stream. Returns a `Stream<Item = Bytes>` that /// Get the stdout stream. Returns a `Stream<Item = Bytes>` that
/// yields stdout chunks as they arrive. The stream ends when the /// yields stdout chunks as they arrive. The stream ends when the
/// server's stdout reaches EOF (a zero-length stdout sentinel /// server's stdout reaches EOF — the zero-length stdout sentinel
/// chunk, see `tty-wire.md` §"Sentinels"). /// chunk (`tty-wire.md` §"Sentinels") terminates the stream and is
/// NOT yielded as an item.
/// ///
/// This consumes the stdout receiver — calling it twice returns /// This consumes the stdout receiver — calling it twice returns
/// an empty stream the second time (the receiver is behind a /// 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; let mut guard = self.stdout_rx.lock().await;
if let Some(rx) = guard.take() { if let Some(rx) = guard.take() {
return Box::pin(futures::stream::unfold(rx, |mut rx| async move { 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()) Box::pin(futures::stream::empty())
} }
/// Get the stderr stream. `None` for PTY-mode backends /// Get the stderr stream. `None` for PTY-mode backends
/// (stdout/stderr merged into stdout by the kernel PTY), or if /// (stdout/stderr merged into stdout by the kernel PTY), or if
/// already taken. The stream ends when the server's stderr reaches /// 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>>> { pub async fn recv_stderr(&self) -> Option<Pin<Box<dyn Stream<Item = Bytes> + Send>>> {
let mut guard = self.stderr_rx.lock().await; let mut guard = self.stderr_rx.lock().await;
let rx = guard.take()?; let rx = guard.take()?;
@@ -710,25 +728,88 @@ mod tests {
assert_eq!(code, 0); assert_eq!(code, 0);
} }
#[tokio::test] /// A backend whose exit is held back until the test releases it, so
async fn send_stdin_round_trips_to_backend() { /// typed-method writes happen while the session data plane is
// Use a backend that captures stdin — but `MockBackend` doesn't /// definitively open (the exit-resolves-immediately `MockBackend`
// expose the stdin channel to the test. This test just verifies /// races them against session teardown — review #003 P15's
// `send_stdin` doesn't error; the adapter tests cover the /// `input_pump.abort()` made that race fail loudly).
// stdin-to-backend pump. struct GatedBackend {
let backend = Arc::new(MockBackend::with_exit_code(0)); 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 { let identity = Some(Identity {
id: "alice".to_string(), id: "alice".to_string(),
scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()], scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
resources: HashMap::new(), 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 session
.send_stdin(Bytes::from_static(b"hello")) .send_stdin(Bytes::from_static(b"hello"))
.await .await
.expect("send_stdin"); .expect("send_stdin");
session.close_stdin().await.expect("close_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] #[tokio::test]
@@ -736,17 +817,17 @@ mod tests {
// The session writes control chunks; whether the backend // The session writes control chunks; whether the backend
// receives them is the adapter's concern (covered by the // receives them is the adapter's concern (covered by the
// adapter tests). This test verifies the typed methods // adapter tests). This test verifies the typed methods
// serialize and write without error. // serialize and write without error, with the exit held until
let backend = Arc::new(MockBackend::with_exit_code(0)); // after the writes.
let identity = Some(Identity { let (session, release, _server) = wire_gated_session().await;
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;
session.resize(80, 24, 0, 0).await.expect("resize"); session.resize(80, 24, 0, 0).await.expect("resize");
session.signal("INT").await.expect("signal"); 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] #[tokio::test]
@@ -903,10 +984,10 @@ mod tests {
let stdout = session.recv_stdout().await; let stdout = session.recv_stdout().await;
let collected: Vec<Bytes> = stdout.collect().await; let collected: Vec<Bytes> = stdout.collect().await;
// The adapter emits a zero-length stdout sentinel after the // The adapter emits a zero-length stdout sentinel after the
// backend stream ends; filter it out to assert the data chunks. // backend stream ends; the stream terminates ON the sentinel
let data: Vec<Bytes> = collected.into_iter().filter(|b| !b.is_empty()).collect(); // (P11), so no filtering is needed to see the data chunks.
assert_eq!( assert_eq!(
data, collected,
vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")], vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")],
"stdout chunks should route to the stdout stream" "stdout chunks should route to the stdout stream"
); );
@@ -1023,9 +1104,11 @@ mod tests {
let stdout = session.recv_stdout().await; let stdout = session.recv_stdout().await;
let collected: Vec<Bytes> = stdout.collect().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!( assert_eq!(
data, collected,
vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")], vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")],
"stdout routing must be unaffected by the ignored chunks" "stdout routing must be unaffected by the ignored chunks"
); );
@@ -1220,9 +1303,9 @@ mod tests {
let stdout = session.recv_stdout().await; let stdout = session.recv_stdout().await;
let collected: Vec<Bytes> = stdout.collect().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!( assert_eq!(
data, collected,
vec![Bytes::from_static(b"ch-out")], vec![Bytes::from_static(b"ch-out")],
"stdout should route through the channels data plane" "stdout should route through the channels data plane"
); );