[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.