contextvm_sdk/transport/open_stream/
mod.rs

1//! CEP-41 open-ended streaming — transport-agnostic engine.
2//!
3//! A server tool can emit an ordered, unbounded sequence of `chunk` fragments
4//! back to a client *while a request is in flight*; the client consumes them
5//! incrementally as an async [`Stream`](futures::Stream). Unlike CEP-22, the
6//! stream does **not** replace the final JSON-RPC response — one `tools/call`
7//! produces two outputs: a live `notifications/progress` stream and the normal
8//! final response that still concludes the request.
9//!
10//! Frames ride inside `notifications/progress` notifications on the existing
11//! `CTXVM_MESSAGES_KIND`, discriminated by `params.cvm.type == "open-stream"`
12//! ([`frame`]). The stream id is the request `progressToken`. See the CEP-41
13//! spec and the TypeScript reference at `sdk/src/transport/open-stream/`.
14//!
15//! This module is the **pure engine**: framing ([`frame`]), the reader
16//! [`OpenStreamSession`] (incl. the pure keepalive [`tick`](OpenStreamSession::tick)),
17//! the producer [`OpenStreamWriter`], and the per-peer [`OpenStreamRegistry`].
18//! It carries no transport or live timers — the client and server transports
19//! drive it (the keepalive sweep calls `tick`; inbound frames feed
20//! `process_frame`).
21
22pub mod constants;
23pub mod errors;
24pub mod frame;
25pub mod receiver;
26pub mod registry;
27pub mod session;
28pub mod writer;
29
30pub use constants::*;
31pub use errors::OpenStreamError;
32pub use frame::{open_stream_frame_from_notification, OpenStreamFrame};
33pub use receiver::OpenStreamReceiver;
34pub use registry::{
35    OpenStreamRegistry, OpenStreamRegistryPolicy, OpenStreamSessionInit, RegistryAbortHook,
36    RegistryCloseHook,
37};
38pub use session::{
39    FrameOutcome, KeepaliveAction, OpenStreamSession, OpenStreamSessionOptions, PublishFrame,
40};
41pub use writer::{OnAbortHook, OnCloseHook, OpenStreamWriter, OpenStreamWriterOptions};
42
43/// CEP-41 open-stream configuration shared by both transports.
44///
45/// Bundles the capability gate plus the reader buffering/keepalive knobs.
46/// Attached to
47/// [`NostrServerTransportConfig`](crate::transport::NostrServerTransportConfig)
48/// and [`NostrClientTransportConfig`](crate::transport::NostrClientTransportConfig)
49/// via their `with_open_stream` builders.
50///
51/// **Disabled by default** (opt-in): open-stream is neither advertised nor
52/// activated until enabled (matching the TS SDK's default). Opt in with
53/// [`OpenStreamConfig::enabled`] or `with_enabled(true)`. Once on, it is safe for
54/// non-CEP-41 peers — the server activates only for advertising clients, and a
55/// writer is injected only when a request carries a `progressToken`.
56#[derive(Debug, Clone)]
57#[non_exhaustive]
58pub struct OpenStreamConfig {
59    /// Master gate, `false` by default. When `false` the capability is neither
60    /// advertised nor activated, and the server does not learn a client's flag.
61    pub enabled: bool,
62    /// Upper bound on concurrently active streams (per peer/registry).
63    pub max_concurrent_streams: usize,
64    /// Upper bound on buffered + queued chunks held for a single stream.
65    pub max_buffered_chunks_per_stream: usize,
66    /// Upper bound on buffered + queued payload bytes held for a single stream.
67    pub max_buffered_bytes_per_stream: usize,
68    /// Idle interval after which a reader probes the peer with a `ping` (ms).
69    pub idle_timeout_ms: u64,
70    /// Time a reader waits for a `pong` after probing before aborting (ms).
71    pub probe_timeout_ms: u64,
72    /// Grace period after a `close` with unresolved gaps before aborting (ms).
73    pub close_grace_period_ms: u64,
74    /// Optional hard cap on total stream lifetime (ms).
75    ///
76    /// `None` (the default) means no lifetime cap. When set, only
77    /// `call_tool_stream` reads it (the registry and keepalive sweep never do);
78    /// it is **not** the CEP-22 `DEFAULT_OVERSIZED_MAX_TOTAL_TIMEOUT`.
79    pub max_total_timeout_ms: Option<u64>,
80}
81
82impl Default for OpenStreamConfig {
83    fn default() -> Self {
84        Self {
85            // Open-stream is opt-in (disabled by default), matching the TS SDK.
86            // Enable it with `OpenStreamConfig::enabled()` / `with_enabled(true)`.
87            enabled: false,
88            max_concurrent_streams: constants::DEFAULT_MAX_CONCURRENT_OPEN_STREAMS,
89            max_buffered_chunks_per_stream: constants::DEFAULT_MAX_BUFFERED_CHUNKS_PER_STREAM,
90            max_buffered_bytes_per_stream: constants::DEFAULT_MAX_BUFFERED_BYTES_PER_STREAM,
91            idle_timeout_ms: constants::DEFAULT_OPEN_STREAM_IDLE_TIMEOUT_MS,
92            probe_timeout_ms: constants::DEFAULT_OPEN_STREAM_PROBE_TIMEOUT_MS,
93            close_grace_period_ms: constants::DEFAULT_OPEN_STREAM_CLOSE_GRACE_PERIOD_MS,
94            max_total_timeout_ms: None,
95        }
96    }
97}
98
99impl OpenStreamConfig {
100    /// An explicitly enabled config with all other knobs at their defaults.
101    pub fn enabled() -> Self {
102        Self {
103            enabled: true,
104            ..Self::default()
105        }
106    }
107
108    /// Enable or disable open-stream support.
109    pub fn with_enabled(mut self, enabled: bool) -> Self {
110        self.enabled = enabled;
111        self
112    }
113
114    /// Set the upper bound on concurrently active streams.
115    pub fn with_max_concurrent_streams(mut self, max: usize) -> Self {
116        self.max_concurrent_streams = max;
117        self
118    }
119
120    /// Set the upper bound on buffered + queued chunks per stream.
121    pub fn with_max_buffered_chunks_per_stream(mut self, max: usize) -> Self {
122        self.max_buffered_chunks_per_stream = max;
123        self
124    }
125
126    /// Set the upper bound on buffered + queued payload bytes per stream.
127    pub fn with_max_buffered_bytes_per_stream(mut self, max: usize) -> Self {
128        self.max_buffered_bytes_per_stream = max;
129        self
130    }
131
132    /// Set the idle interval before a reader probes with a `ping` (ms).
133    pub fn with_idle_timeout_ms(mut self, ms: u64) -> Self {
134        self.idle_timeout_ms = ms;
135        self
136    }
137
138    /// Set the time a reader waits for a `pong` before aborting (ms).
139    pub fn with_probe_timeout_ms(mut self, ms: u64) -> Self {
140        self.probe_timeout_ms = ms;
141        self
142    }
143
144    /// Set the close grace period before aborting on unresolved gaps (ms).
145    pub fn with_close_grace_period_ms(mut self, ms: u64) -> Self {
146        self.close_grace_period_ms = ms;
147        self
148    }
149
150    /// Set the optional hard cap on total stream lifetime (ms).
151    pub fn with_max_total_timeout_ms(mut self, ms: Option<u64>) -> Self {
152        self.max_total_timeout_ms = ms;
153        self
154    }
155}
156
157impl From<&OpenStreamConfig> for OpenStreamRegistryPolicy {
158    /// Project the registry-relevant knobs of an [`OpenStreamConfig`] into an
159    /// [`OpenStreamRegistryPolicy`] (the reader admission/buffering policy).
160    fn from(config: &OpenStreamConfig) -> Self {
161        OpenStreamRegistryPolicy {
162            max_concurrent_streams: config.max_concurrent_streams,
163            max_buffered_chunks_per_stream: config.max_buffered_chunks_per_stream,
164            max_buffered_bytes_per_stream: config.max_buffered_bytes_per_stream,
165            idle_timeout_ms: config.idle_timeout_ms,
166            probe_timeout_ms: config.probe_timeout_ms,
167            close_grace_period_ms: config.close_grace_period_ms,
168        }
169    }
170}
171
172#[cfg(test)]
173mod config_tests {
174    use super::*;
175
176    #[test]
177    fn default_is_disabled_with_ts_parity_knobs() {
178        let config = OpenStreamConfig::default();
179        // Open-stream is opt-in (disabled by default), matching the TS SDK.
180        assert!(!config.enabled);
181        // Opting in is one call.
182        assert!(OpenStreamConfig::default().with_enabled(true).enabled);
183        assert!(OpenStreamConfig::enabled().enabled);
184        assert_eq!(config.max_concurrent_streams, 64);
185        assert_eq!(config.max_buffered_chunks_per_stream, 64);
186        assert_eq!(config.max_buffered_bytes_per_stream, 512 * 1024);
187        assert_eq!(config.idle_timeout_ms, 30_000);
188        assert_eq!(config.probe_timeout_ms, 20_000);
189        assert_eq!(config.close_grace_period_ms, 5_000);
190        // No hard lifetime cap by default.
191        assert_eq!(config.max_total_timeout_ms, None);
192    }
193
194    #[test]
195    fn builders_opt_in_and_override() {
196        let config = OpenStreamConfig::default()
197            .with_enabled(true)
198            .with_max_concurrent_streams(8)
199            .with_max_buffered_bytes_per_stream(1024)
200            .with_max_total_timeout_ms(Some(60_000));
201        assert!(config.enabled);
202        assert_eq!(config.max_concurrent_streams, 8);
203        assert_eq!(config.max_buffered_bytes_per_stream, 1024);
204        assert_eq!(config.max_total_timeout_ms, Some(60_000));
205
206        assert!(OpenStreamConfig::enabled().enabled);
207    }
208
209    #[test]
210    fn projects_into_registry_policy() {
211        let config = OpenStreamConfig::default()
212            .with_max_concurrent_streams(3)
213            .with_max_buffered_chunks_per_stream(5)
214            .with_max_buffered_bytes_per_stream(7)
215            .with_idle_timeout_ms(11)
216            .with_probe_timeout_ms(13)
217            .with_close_grace_period_ms(17);
218        let policy: OpenStreamRegistryPolicy = (&config).into();
219        assert_eq!(policy.max_concurrent_streams, 3);
220        assert_eq!(policy.max_buffered_chunks_per_stream, 5);
221        assert_eq!(policy.max_buffered_bytes_per_stream, 7);
222        assert_eq!(policy.idle_timeout_ms, 11);
223        assert_eq!(policy.probe_timeout_ms, 13);
224        assert_eq!(policy.close_grace_period_ms, 17);
225    }
226}
contextvm_sdk/transport/open_stream/mod.rs

contextvm_sdk/transport/open_stream/
mod.rs
