Fix/open streams

CHANGELOG.md

@@ -4,6 +4,17 @@
 
 ### Added
 
+- `ClientOpenStreamHandle::cancel(token, reason)`: a `Send`-safe way to cancel a
+  CEP-41 open-stream reader session by its progress token from any task —
+  including a `tokio::spawn` that drives the stream. `ToolStreamCall` is `!Sync`
+  (its `result: BoxFuture` field), so `abort(&self)` can't be awaited from a
+  `Send` task, forcing consumers that drive a stream in a spawned task to drop the
+  call without cleaning up. `cancel` mirrors `abort` (publishes the `abort` frame
+  and frees the reader-registry slot via the idempotent `abort`/`consumer_abort`
+  pair) but operates on the `Sync` handle, so cloning the handle into a task and
+  calling `cancel(token, reason)` works from any thread. Without it, a dropped
+  `ToolStreamCall` leaves the reader session lingering in the registry until the
+  keepalive sweep probe-times it out (`Probe timeout`).
 - A general-purpose, public inbound-middleware seam on `NostrServerTransport`:
   `add_inbound_middleware` with the `InboundMiddleware` trait, an `InboundContext`
   describing the request, and an owned `Next`, letting callers observe, gate, or
@@ -23,6 +34,35 @@
     authorizations, with atomic single-use `claim` and check-and-set `try_set_pending`
     semantics that are double-spend-safe under multi-threaded tokio.
 
+### Fixed
+
+- CEP-41 open-stream (server): the writer `progress_token → event_id` index is
+  now scoped by `(client_pubkey, token)`, not the bare token. The progress token
+  is only unique *within* a peer — rmcp mints it from a per-peer counter, so every
+  client's first stream carries token `"0"`. The old global key let two concurrent
+  clients clobber each other's entry, and either one's cleanup then deleted the
+  shared key, orphaning the other's still-live writer. The orphaned client's
+  keepalive pings found no writer → no `pong` → `Probe timeout` on an
+  otherwise-alive stream, reproducing only with ≥2 concurrent clients (hence unseen
+  by single-client test suites). The fix mirrors the TS `getProgressTokenKey`
+  composite and the existing per-peer reader registry; `slots` (keyed by
+  `event_id`) is unaffected. The oversized-transfer reassembly was verified
+  already per-peer scoped (`LruCache<sender_pubkey, …>`) and needs no change.
+- CEP-41 open-stream: a reader session whose `start` frame had not yet arrived no
+  longer emits a keepalive `ping`. The reader `SessionState::tick` now gates its
+  idle→ping transition on `started`, matching the writer's `tick` and restoring parity
+  with the TS SDK (whose idle timer arms only on `start`). Previously, a session
+  registered at request-publish time would ping after `idle_timeout` without ever
+  receiving `start`; on the server, a ping for a token whose writer had been disposed
+  (e.g. a tool that returned without streaming) was raised as a fatal
+  `Received ping frame before start` sequence error and logged as a WARN, and the
+  unanswered probe then aborted the client stream. (`session.rs`)
+- CEP-41 open-stream: a control frame (`ping`/`pong`/`abort`) for a token with no
+  reader session is now dropped at debug level instead of raising a fatal sequence
+  error. Such a frame is a teardown linger or pre-start desync, not a data-plane
+  violation; data frames (`chunk`/`close`) for unknown tokens still error.
+  (`registry.rs`)
+
 ## [0.2.1] - 2026-07-10
 
 ### Added

src/transport/client/mod.rs

@@ -248,6 +248,38 @@ pub struct ClientOpenStreamHandle {
 
 #[cfg_attr(not(feature = "rmcp"), allow(dead_code))]
 impl ClientOpenStreamHandle {
+    /// Cancel an active open-stream reader session by its progress token.
+    ///
+    /// Send-safe equivalent of `ToolStreamCall::abort`: publishes the `abort`
+    /// frame to the server and frees the reader-registry slot. Use this when the
+    /// caller can't hold `&ToolStreamCall` across an `.await` — e.g. a stream
+    /// driven inside `tokio::spawn`, where `ToolStreamCall`'s `!Sync` (its
+    /// `result: BoxFuture` field) makes `abort(&self)` unusable from a `Send`
+    /// task. This handle is `Sync`, so cloning it into a task and calling
+    /// `cancel(token, reason)` works from any thread.
+    ///
+    /// Mirrors the `abort` path: `OpenStreamSession::abort` publishes the frame
+    /// and finalizes the local stream, then `OpenStreamRegistry::consumer_abort`
+    /// removes the entry and runs the `on_abort` hook. Both are idempotent, so
+    /// canceling an unknown or already-terminated token is a harmless no-op.
+    /// Without it, dropping a `ToolStreamCall` without aborting leaves the reader
+    /// session lingering in the registry until the keepalive sweep probe-times it
+    /// out (`Probe timeout`).
+    pub async fn cancel(&self, progress_token: &str, reason: Option<String>) {
+        let registry = self.registry.clone();
+        // Clone the session out and release the lock before the publish await, so
+        // the frame send doesn't block other open-stream operations.
+        let session = registry.lock().await.get_session(progress_token);
+        if let Some(session) = session {
+            session.abort(reason.clone()).await;
+        }
+        registry
+            .lock()
+            .await
+            .consumer_abort(progress_token, reason)
+            .await;
+    }
+
     /// Register a placeholder for the next outbound `call_tool_stream` session
     /// (resolved by the served transport's `send`).
     pub(crate) fn prepare_outbound(

src/transport/open_stream/registry.rs

@@ -212,13 +212,37 @@ impl OpenStreamRegistry {
         let (progress_token, progress, frame) = parse_frame(notification)?;
 
         if !self.sessions.contains_key(&progress_token) {
-            if frame.frame_type() != "start" {
-                return Err(OpenStreamError::Sequence(format!(
-                    "Received {} frame before start for {progress_token}",
-                    frame.frame_type()
-                )));
+            match frame.frame_type() {
+                "start" => {
+                    self.create_session_with(
+                        progress_token.clone(),
+                        OpenStreamSessionInit::default(),
+                    )?;
+                }
+                // A control frame (`ping`/`pong`/`abort`) for a token with no
+                // session is a teardown linger — the writer was already disposed
+                // — or a pre-start desync. It carries no data, so drop it at
+                // debug instead of raising a fatal sequence error that logs a
+                // misleading "Received ping frame before start" WARN. Data
+                // frames (`chunk`/`close`) still error: those are a genuine
+                // protocol problem.
+                "ping" | "pong" | "abort" => {
+                    tracing::debug!(
+                        target: LOG_TARGET,
+                        frame = %frame.frame_type(),
+                        %progress_token,
+                        "open-stream control frame for unknown token; dropping \
+                         (teardown linger or pre-start desync)"
+                    );
+                    return Ok(FrameOutcome::None);
+                }
+                _ => {
+                    return Err(OpenStreamError::Sequence(format!(
+                        "Received {} frame before start for {progress_token}",
+                        frame.frame_type()
+                    )));
+                }
             }
-            self.create_session_with(progress_token.clone(), OpenStreamSessionInit::default())?;
         }
 
         // Clone the Arc-backed handle out so the map is not borrowed across the
@@ -509,6 +533,44 @@ mod tests {
         assert_eq!(registry.size(), 0);
     }
 
+    #[tokio::test]
+    async fn drops_control_frames_for_unknown_tokens() {
+        // A `ping`/`pong`/`abort` for a token with no session is a teardown
+        // linger (the writer was already disposed) or a pre-start desync — not a
+        // data-plane violation. Drop it at debug instead of raising a fatal
+        // sequence error, and create no session. Data frames still error (see
+        // `rejects_non_start_frames_for_unknown_tokens`).
+        let mut registry = OpenStreamRegistry::with_policy(small_policy(2));
+        for (label, frame) in [
+            (
+                "ping",
+                OpenStreamFrame::Ping {
+                    nonce: "n".to_string(),
+                },
+            ),
+            (
+                "pong",
+                OpenStreamFrame::Pong {
+                    nonce: "n".to_string(),
+                },
+            ),
+            (
+                "abort",
+                OpenStreamFrame::Abort {
+                    reason: Some("bye".to_string()),
+                },
+            ),
+        ] {
+            let outcome = registry
+                .process_frame(now(), &notif("token-ctrl", 1, frame))
+                .await
+                .unwrap_or_else(|e| panic!("{label} for unknown token should drop, got {e}"));
+            assert_eq!(outcome, FrameOutcome::None, "{label} outcome");
+            assert!(registry.get_session("token-ctrl").is_none());
+            assert_eq!(registry.size(), 0);
+        }
+    }
+
     #[tokio::test]
     async fn terminates_session_when_frame_processing_fails_after_creation() {
         let mut registry = OpenStreamRegistry::with_policy(OpenStreamRegistryPolicy {

src/transport/open_stream/session.rs

@@ -286,8 +286,17 @@ impl SessionState {
             }
         }
 
-        // 3. Idle past the threshold (and not closed / not already probing) → ping.
-        if !self.closed_remotely
+        // 3. Idle past the threshold (and started / not closed / not already
+        //    probing) → ping. A session that hasn't received `start` has nothing
+        //    to probe — and the writer-side `tick` already gates on `started`.
+        //    This also restores parity with the TS port, whose idle timer arms
+        //    only on `start`; the sweep-driven `tick` matches by gating here.
+        //    Without it, a reader session registered at request-publish time
+        //    (before `start` lands) pings after `idle_timeout` and the server —
+        //    which may have reaped an un-started writer — raises a fatal
+        //    "Received ping frame before start" sequence error.
+        if self.started
+            && !self.closed_remotely
             && self.pending_probe_nonce.is_none()
             && now.saturating_duration_since(self.last_activity) >= self.idle_timeout
         {
@@ -1039,6 +1048,23 @@ mod tests {
 
     // ── keepalive (pure `tick`, injected clock) ─────────────────────
 
+    #[tokio::test]
+    async fn tick_returns_none_before_start() {
+        // Parity with the writer's `tick_returns_none_before_start`: a reader
+        // session that hasn't received `start` must not probe (nothing to keep
+        // alive yet). The TS port arms its idle timer only on `start`; the
+        // sweep-driven `tick` matches by gating on `started`. Regression guard
+        // for the pre-start ping that surfaced server-side as "Received ping
+        // frame before start for <token>".
+        let t0 = Instant::now();
+        let s = make_session_timers("token-pre-start", 10, 10, 100);
+        assert_eq!(
+            s.tick(t0 + Duration::from_millis(100)),
+            KeepaliveAction::None
+        );
+        assert!(!s.has_started());
+    }
+
     #[tokio::test]
     async fn ping_frame_requests_a_pong() {
         let now = Instant::now();