[Rust] Add PreparedSession for loss-free startup event subscription

[jmoseley]

The problem

Session::subscribe() can only be called once the session handle exists. Session::subscribe is backed by a tokio::sync::broadcast channel, and a broadcast send with zero receivers drops the value. So every event the runtime emitted while session.create / session.resume was still in flight was broadcast into a channel nobody was listening to and silently discarded.

Both startup paths have the hole, from opposite directions:

• create_session allocated the broadcast sender before the RPC but only spawned the event loop after the response — the events sat in the router's unbounded queue and were then fanned out to nobody.
• resume_session allocated and spawned the event loop before awaiting session.resume, so events were broadcast to nobody in real time.

getMessages can't paper over it: ephemeral events such as session.idle are never written to the session log. A consumer that needs to know the agent went idle during a resume with continuePendingWork has no way to recover that. Returning a (Session, EventSubscription) tuple after the await doesn't fix it either — the events are already gone by then.

The change

Client::prepare_session / Client::prepare_resume_session return a PreparedSession that owns the session's broadcast channel up front. Subscribe first, then start:

let prepared = client.prepare_session(
    SessionConfig::default().with_event_buffer_capacity(2048),
)?;
let mut events = prepared.subscribe();   // installed before anything hits the wire
let session = prepared.start().await?;

prepare_* is synchronous and inert. It validates the event buffer capacity, allocates a local channel and cancellation token, and does nothing else — no router registration, no task spawn, no bytes on the wire until start() is first polled. start(self) consumes the handle and PreparedSession is deliberately not Clone, so a prepared session can never produce two event loops.

create_session / resume_session are now wrappers over prepare_*(config)?.start().await. Their bodies moved into private start paths that take the sender and cancellation token by injection instead of allocating their own, so there's one implementation rather than two.

Both configs gain a runtime-only event_buffer_capacity (default 512, Some(0) rejected as InvalidConfig rather than clamped). The buffer is finite by design: slow subscribers observe Lagged with a skipped count instead of applying backpressure to the event loop.

Cancellation and cleanup

This is the other half of the change. resume_session already had a PendingSessionRegistration RAII guard; create_session had none, so dropping a create future mid-RPC leaked the router registration outright.

PendingSessionRegistration now carries either a known session ID or a deferred one that it resolves from the inline-response stash, which covers the cloud server-assigned-ID path where registration happens inside the JSON-RPC response callback. Registration and stashing happen under a single lock hold, closing the window where a concurrent guard drop would observe an empty stash and miss a session that was just registered. The mcp-auth-interest error path on both create and resume now cancels and awaits the event loop instead of returning through ?.

Net semantics:

• Dropping an unstarted PreparedSession is fully inert and closes its subscriptions.
• Dropping a polled start() future cancels the token, unregisters the session, and closes early subscriptions — a retry with the same session ID succeeds.
• Startup errors do the same and keep the exact ErrorKinds these calls have always returned.

Drop is synchronous and can't await, so the event loop terminates promptly rather than synchronously. The docs say that rather than claiming otherwise.

Known limitation, documented precisely

For cloud sessions where the server assigns the session ID, the SDK can't route notifications until the response arrives and the ID is known — pre-registration notifications aren't routable to any session. The guarantee is narrower and stated as such: routed events are never dropped for lack of an installed receiver. Pinning session_id gets you registration before the RPC and full pre-response coverage.

Tests

New rust/tests/prepared_session_test.rs, 16 tests on the existing in-memory duplex harness with a hand-rolled JSON-RPC peer. No correctness sleeps — timeouts are failure backstops only.

Covered: a 600-event pre-response burst plus an ephemeral session.idle delivered exactly once and in order on create (both concurrent-drain and deferred-consumer variants) and on resume with continuePendingWork; an undersized buffer surfacing Lagged rather than silent loss with the live tail still consumable; prepare inertness (no wire traffic, no registration, no spawned task, verified against num_alive_tasks); dropping an unstarted handle; cancelling a polled create and resume with same-ID retry; RPC error and session-ID-mismatch cleanup preserving error kinds; one early plus one late subscriber sharing exactly one event loop; wrapper RPC sequences; and a compile-time assertion that PreparedSession is Send + 'static and not Clone.

Verification

• just lint-rust (nightly fmt check + clippy with the repo's full deny set) — clean
• just test-rust — 791 tests, all targets green, including the 390 replay-proxy E2E tests
• RUSTDOCFLAGS="-D warnings" cargo doc --no-deps --all-features — clean

Docs: rustdoc on the new surface, a "Subscribing before a session starts" section in docs/features/streaming-events.md, the Rust README streaming section and Rust-only API list, and a CHANGELOG.md Unreleased entry. No protocol or generated-type changes.

[Copilot]

Pull request overview

Adds Rust PreparedSession APIs to subscribe before create/resume startup events occur.

Changes:

• Adds configurable event buffering and cancellation-safe startup paths.
• Adds comprehensive prepared-session tests.
• Documents the API and startup-event semantics.

Show a summary per file

File	Description
rust/src/session.rs	Implements prepared sessions and cleanup.
rust/src/types.rs	Adds event-buffer configuration.
rust/src/lib.rs	Adds test-only router inspection.
rust/tests/prepared_session_test.rs	Tests delivery, lag, cancellation, and wrappers.
rust/Cargo.toml	Registers the new test target.
rust/README.md	Documents Rust usage.
docs/features/streaming-events.md	Adds early-subscription guidance.
CHANGELOG.md	Announces the feature.

Review details

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

• Files reviewed: 8/8 changed files
• Comments generated: 4
• Review effort level: Balanced

[jmoseley]

Pushed a7471a05 → d8b48c79. All four unresolved review threads are addressed and resolved:

• de2ef259 — removed the lone <details> wrapper around the Rust example in docs/features/streaming-events.md.
• 69da7703 — narrowed the prepare_session rustdoc and changelog to routed events, with pinning session_id called out as the requirement for complete pre-response coverage.
• d8b48c79 — keeps session IDs out of the two prepared_session_test.rs polling helpers' failure diagnostics (a downstream consumer that vendors this crate flags identifiers reaching formatted output). Predicates and deadlines are unchanged, so no assertion is weakened.

The cancellation-race thread needed no code change: bb8ca432 already closes both interleavings, and I verified it against the final code and the six tests that cover the two orderings — details in the thread.

No API, behavior, or wire change on this head. Local gates green: nightly cargo fmt --check, cargo clippy --all-targets --features test-support,bundled-in-process -D warnings, RUSTDOCFLAGS=-D warnings cargo doc --no-deps --all-features, and the full non-E2E test matrix under both --no-default-features --features test-support and --all-features (prepared_session 21/21, session 117/117, lib 216/232, doctests 21). E2E is unrunnable locally (no CLI install) and is untouched by this delta.

@copilot-pull-request-reviewer ready for re-review.

[Copilot]

Review details

Suppressed comments (4)

rust/src/session.rs:1826

• This promises delivery of every event to every subscriber, but the finite Tokio broadcast buffer explicitly allows slow subscribers to receive Lagged and skip events. Qualify the statement so it does not contradict the buffering contract immediately above.

    /// May be called any number of times; every subscriber receives every
    /// event. Subscriptions taken here close if the prepared session is
    /// dropped without starting, or if startup fails.

rust/src/session.rs:1421

• The new create cleanup branch for a failed session.eventLog.registerInterest call is not exercised: the MCP-auth tests in rust/tests/session_test.rs:292-430 only return successful interest responses, while the prepared-session failure tests cover only create RPC errors and ID mismatches. Add a failing interest response test that verifies the original error kind, router unregistration, and closure of an early subscription.

        if has_mcp_auth_handler
            && let Err(error) = register_mcp_auth_interest(self, &session_id).await
        {
            registration.cleanup(event_loop).await;
            return Err(error);

rust/src/session.rs:1659

• The equivalent resume cleanup branch also lacks failure coverage: existing MCP-auth resume tests return a successful interest response. Add a failing interest response test for prepared resume and assert that the event loop/subscription closes and the router registration is removed before the error is returned.

        if has_mcp_auth_handler
            && let Err(error) = register_mcp_auth_interest(self, &session_id).await
        {
            registration.cleanup(event_loop).await;
            return Err(error);

docs/features/streaming-events.md:258

• For a consumer that requires lossless startup delivery, one of these actions is required. The docs style guide avoids “should” for required actions; use “must” to make the constraint definitive.

* Consumers that need a lossless view of a large startup burst should either configure a capacity that covers it or drain the subscription concurrently with `start()`.

• Files reviewed: 9/9 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[jmoseley]

Follow-up on the re-review's four suppressed comments — all addressed. d8b48c79 → c98cbee6.

session.rs:1826 (subscribe overclaim) — fixed in c98cbee6. "every subscriber receives every event" did contradict the buffering paragraph directly above it. It now states that each subscriber gets its own copy of the stream subject to the buffering contract, and that a subscriber falling behind the configured capacity observes Lagged and skips events rather than stalling the event loop, with an intra-doc link to Lagged.

docs/features/streaming-events.md:258 (should → must) — fixed in c98cbee6. Also applied the same wording to the equivalent changelog sentence, since it is the identical claim and rust/README.md already said "must".

session.rs:1421 and :1659 (uncovered interest-failure branches) — fixed in dd4e56f4 with one test per path: create_mcp_auth_interest_error_preserves_kind_and_cleans_up and resume_mcp_auth_interest_error_preserves_kind_and_cleans_up. Each installs an MCP-auth handler, answers session.create/session.resume successfully, fails session.eventLog.registerInterest, then asserts the three things you asked for: the original ErrorKind::Rpc code reaches the caller, the router registration is gone, and a subscription installed before start() is closed. The resume test additionally asserts no session.skills.reload is issued, since interest registration runs ahead of it.

One honest note on those two: I verified by mutation that both tests execute the branch, and removing the registration.cleanup(event_loop) call does not turn them red. PendingSessionRegistration::drop cancels the shutdown token and releases the same registration synchronously, so on this path the explicit cleanup is defense in depth rather than the sole mechanism. I tried a stricter "already torn down the instant the error surfaces" assertion to separate the two, confirmed by mutation that it still could not distinguish them, and dropped it rather than ship a test coupled to scheduling timing. The tests therefore assert the observable contract, consistent with the sibling failure tests.

Test-only and doc-only: git diff over rust/src/session.rs for all five new commits contains zero non-/// lines, so there is still no API, behavior, or wire change for the consumer that vendors this head.

Gates re-run green on c98cbee6: nightly cargo fmt --check, cargo clippy --all-targets --features test-support,bundled-in-process -D warnings, RUSTDOCFLAGS=-D warnings cargo doc --no-deps --all-features, and the full non-E2E matrix under --no-default-features --features test-support (prepared_session now 23/23, session 117/117, lib 216, doctests 21) plus --all-features.

[jmoseley]

Adds awaited CLI process termination — c98cbee6 → 036bd233.

Problem. force_stop() sent a kill and returned; nothing waited on the child, so on Unix it stayed a zombie for the lifetime of the parent. An embedded host that force-stops on shutdown timeout accumulates them. The deeper defect was ownership: stop() took the child out of its slot and awaited the kill inline, so an outer timeout cancelling it in that window dropped the handle with the future — neither reaped nor recoverable by a later force_stop.

Fix. New rust/src/child.rs ChildLifecycle owns the child instead of lending it out. The kill is delivered synchronously under the slot lock (start_kill needs no reactor, so signalling never depends on a task being polled), and the claimed child is reaped on a dedicated thread with its own runtime. A tokio::spawned reaper is cancelled when its runtime drops — precisely when an embedded host needs it — so the thread is what makes the guarantee real. Outcomes are published on a watch channel, so concurrent and repeat callers observe the same terminal result, and stop() claims through the same path.

Public API (additive; force_stop keeps its signature and semantics):

Client::force_stop_and_wait() -> Result<Option<ExitStatus>>   // killed AND reaped
Client::start_force_stop() -> ForcedShutdown                  // owned handle
ForcedShutdown::wait() / ::pid()

None means the client never spawned a child; failure is always Err. Cancelling any of it, or dropping the handle, cannot strand the process.

Three mutation-verified guards — each fails when the property it protects is removed:

• reverting detached ownership → cancel-after-claim recovery hangs (Elapsed)
• moving the kill back into the reaper → process left in ps state S
• dropping enable_all() from the reaper runtime → still-running-child reap hangs

15 lifecycle tests: cross-runtime survival, no-runtime-context, cancelled stop(), already-exited, still-running-at-first-poll, childless stream/in-process, repeat + concurrent, drop-everything, Send + 'static.

Reviewed across three rounds by an independent reviewer, who caught a critical regression in my first attempt (I had wrongly believed start_kill required the reactor, so the kill had moved into the detached task — that would have traded a zombie for a live orphan). All findings fixed; the final pass reported no correctness defects.

Gates green in both feature configs: nightly fmt, clippy -D warnings, cargo doc -D warnings, lib 231 / 247, prepared_session 23, session 117, doctests 22. PreparedSession work and diagnostic redaction untouched.

@copilot-pull-request-reviewer ready for re-review.

[Copilot]

Review details

Suppressed comments (2)

CHANGELOG.md:10

• This PR is described as the PreparedSession startup-subscription change, but this entry exposes a second, unrelated public feature (ForcedShutdown, start_force_stop, and force_stop_and_wait) backed by a new 677-line process-lifecycle implementation and additional tests. That materially expands the API and risk beyond the stated scope. Please split the confirmed-process-termination work into its own PR (or, at minimum, retitle and fully describe the combined change) so it can be reviewed and released independently.

### Feature: confirmed CLI process termination (Rust)

rust/README.md:82

• This example first awaits client.stop() unconditionally and then calls stop() a second time inside the timeout, so it does not demonstrate a forced fallback when the original graceful shutdown hangs. Replace the standalone shutdown call with a single timeout-wrapped call and force-stop only that call's timeout path.

// Forced shutdown that confirms the OS reaped the CLI process, for hosts
// that must not leak a zombie when graceful shutdown times out.
if tokio::time::timeout(Duration::from_secs(5), client.stop()).await.is_err() {
    let exit_status = client.force_stop_and_wait().await?;

• Files reviewed: 10/10 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

Review details

Suppressed comments (3)

rust/src/session.rs:1752

• Dropping start() cannot guarantee that a same-ID create retry succeeds once the RPC has been sent. Client::call commits the frame before yielding and explicitly notes that a cancelled session.create can still take effect (rust/src/lib.rs:1897-1907); this guard only removes the local router registration. The abandoned create may therefore complete in the CLI and collide with the retry. Either implement server-side cancellation/late-response cleanup, or narrow this contract and its tests to guarantee only local registration cleanup.

/// * Dropping the [`start`](Self::start) future mid-flight cancels the
///   session token, unregisters the session from the router if it was
///   registered, and closes early subscriptions. A retry with the same
///   session ID succeeds — cleanup removes only the exact registration that
///   startup owned, so a retry started before the abandoned attempt has

rust/README.md:82

• This example performs a successful stop() immediately before attempting the timeout fallback, so the second stop() has no live child and the forced-shutdown branch cannot demonstrate the stated scenario. Replace the first shutdown call with the timeout/fallback sequence rather than running both sequentially.

// Forced shutdown that confirms the OS reaped the CLI process, for hosts
// that must not leak a zombie when graceful shutdown times out.
if tokio::time::timeout(Duration::from_secs(5), client.stop()).await.is_err() {
    let exit_status = client.force_stop_and_wait().await?;
}

CHANGELOG.md:10

• The PR title and description cover only loss-free session startup, but this change also introduces a separate public forced-shutdown API and replaces child-process ownership/reaping across rust/src/child.rs, rust/src/lib.rs, and rust/README.md. This independent lifecycle redesign materially expands the API and risk without being described by the PR. Split it into its own PR, or update the PR scope and verification so it can be reviewed as an intentional second feature.

### Feature: confirmed CLI process termination (Rust)

• Files reviewed: 10/10 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

Review details

Suppressed comments (5)

rust/src/child.rs:357

• This error contract is false for the new signal-refusal path: start_kill can fail on a live child, and lines 161–165 publish Failed precisely because no signal was delivered. Document refusal separately so callers do not assume the process was terminated after every Err.

    /// Returns [`ErrorKind::Io`] if waiting on the child failed, or if the
    /// reaper could not run to completion — its thread failed to start,
    /// its runtime could not be built, or it panicked. The child has
    /// always been signalled by then; only confirmation is lost.

rust/src/child.rs:198

• When start_kill is refused, the code publishes Failed immediately, so an early waiter returns Err. This unconditional replacement can later publish Reaped if the child exits naturally, causing later/concurrent handles to return Ok for the same termination. That violates the documented single terminal outcome. Keep signal refusal sticky in the externally observed result, or model signal and reap outcomes separately so every waiter resolves identically.

This issue also appears on line 354 of the same file.

                        let outcome = runtime.block_on(reap(child));
                        guard.completed = true;
                        guard.state.send_replace(outcome);

rust/README.md:82

• This example shuts the client down successfully on line 76 and then invokes stop() a second time inside the timeout, so it never demonstrates the intended graceful-then-forced fallback. Use a single timed stop() call and force-stop only when that call times out.

// Forced shutdown that confirms the OS reaped the CLI process, for hosts
// that must not leak a zombie when graceful shutdown times out.
if tokio::time::timeout(Duration::from_secs(5), client.stop()).await.is_err() {
    let exit_status = client.force_stop_and_wait().await?;

CHANGELOG.md:10

• The PR title and description scope this change to PreparedSession, but this adds a second, unrelated public feature (ForcedShutdown, start_force_stop, force_stop_and_wait) plus a substantial child-process lifecycle refactor. This makes the advertised scope and verification incomplete for the actual API changes; split the process-termination feature into its own PR (or update the PR metadata and review it independently).

### Feature: confirmed CLI process termination (Rust)

rust/src/lib.rs:2622

• The signal-refusal branch returns ErrorKind::Io without signalling the live child, so “the child is signalled in every case” is not true. Include kill refusal in the error list and state that the process may still be running in that case.

    /// Returns [`ErrorKind::Io`] if waiting on the child failed, or if the
    /// reaper could not run to completion. The child is signalled in every
    /// case; only confirmation of the reap is lost. No outcome is ever

• Files reviewed: 10/10 changed files
• Comments generated: 0 new
• Review effort level: Balanced