sim_lib_web_bridge/
transport.rs

1//! The transport trait and session status.
2//!
3//! The UI never speaks a transport-specific API. It targets the Intent/Scene
4//! bus, which is expressed here as the [`Transport`] trait: reading a resource
5//! value, realizing a checked operation against it (the `realize_final`
6//! surface), and draining change events (the `realize_events` surface). Four
7//! interchangeable transports implement it -- a deterministic fixture, plus
8//! in-browser wasm, local server, and remote server -- so wasm, local, remote,
9//! and fixture sessions are interchangeable behind the session bridge.
10
11use sim_kernel::{CapabilityName, Cx, Expr, Result, Symbol};
12use sim_lib_stream_core::{
13    PushResult, StreamEnvelope, StreamInspectorSnapshot, StreamInspectorStatus, StreamItem,
14    StreamStats, stream_cancel_capability, stream_open_capability, stream_push_capability,
15    stream_read_capability, stream_stats_capability,
16};
17use sim_lib_stream_fabric::{
18    stream_control_cancel_symbol, stream_control_next_symbol, stream_control_open_symbol,
19    stream_control_push_symbol, stream_control_stats_symbol,
20};
21use sim_lib_view::Operation;
22
23/// The visible state of a session's connection.
24#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25pub enum SessionStatus {
26    /// Establishing the connection.
27    Connecting,
28    /// Connected and live.
29    Connected,
30    /// Lost the connection; the UI surfaces this rather than crashing.
31    Disconnected,
32    /// Attempting to restore a lost connection.
33    Reconnecting,
34    /// Deliberately closed.
35    Closed,
36}
37
38impl SessionStatus {
39    /// Whether reads and operations can flow right now.
40    pub fn is_live(self) -> bool {
41        matches!(self, SessionStatus::Connected)
42    }
43}
44
45/// A change notification: a resource whose value was updated.
46#[derive(Clone, Debug, PartialEq, Eq)]
47pub struct ChangeEvent {
48    /// The resource that changed.
49    pub resource: Symbol,
50}
51
52/// Browser-visible stream status for inspectors and transport badges.
53#[derive(Clone, Copy, Debug, PartialEq, Eq)]
54pub enum BrowserStreamStatus {
55    /// Packets can flow.
56    Live,
57    /// The transport is disconnected.
58    Disconnected,
59    /// The transport is reconnecting.
60    Reconnecting,
61    /// A stream profile was refused by the bridge.
62    RefusedProfile,
63    /// Backpressure dropped or rejected packets.
64    BufferOverflow,
65    /// The stream was cancelled.
66    Cancelled,
67    /// A finite stream ended normally.
68    Ended,
69}
70
71impl BrowserStreamStatus {
72    /// Stable label for browser/UI data.
73    pub fn wire_label(self) -> &'static str {
74        match self {
75            Self::Live => "live",
76            Self::Disconnected => "disconnected",
77            Self::Reconnecting => "reconnecting",
78            Self::RefusedProfile => "refused-profile",
79            Self::BufferOverflow => "buffer-overflow",
80            Self::Cancelled => "cancelled",
81            Self::Ended => "ended",
82        }
83    }
84
85    /// Stable symbol for inspector data.
86    pub fn symbol(self) -> Symbol {
87        Symbol::qualified("stream/browser-status", self.wire_label())
88    }
89
90    /// Maps this browser status to the inspector status enum.
91    pub fn inspector_status(self) -> StreamInspectorStatus {
92        match self {
93            Self::Live => StreamInspectorStatus::Live,
94            Self::Disconnected => StreamInspectorStatus::Disconnected,
95            Self::Reconnecting => StreamInspectorStatus::Reconnecting,
96            Self::RefusedProfile => StreamInspectorStatus::RefusedProfile,
97            Self::BufferOverflow => StreamInspectorStatus::BufferOverflow,
98            Self::Cancelled => StreamInspectorStatus::Cancelled,
99            Self::Ended => StreamInspectorStatus::Ended,
100        }
101    }
102}
103
104/// Web bridge stream operations and their corresponding fabric controls.
105#[derive(Clone, Copy, Debug, PartialEq, Eq)]
106pub enum WebStreamOperation {
107    /// Read up to `limit` packets.
108    Read,
109    /// Subscribe to stream metadata and status.
110    Subscribe,
111    /// Push one envelope into a stream.
112    Push,
113    /// Cancel the stream.
114    Cancel,
115    /// Inspect stream statistics.
116    Stats,
117}
118
119impl WebStreamOperation {
120    /// Returns the stable wire label for this operation.
121    pub fn wire_label(self) -> &'static str {
122        match self {
123            Self::Read => "read",
124            Self::Subscribe => "subscribe",
125            Self::Push => "push",
126            Self::Cancel => "cancel",
127            Self::Stats => "stats",
128        }
129    }
130
131    /// Returns the stable symbol naming this operation.
132    pub fn symbol(self) -> Symbol {
133        Symbol::qualified("stream/web", self.wire_label())
134    }
135
136    /// Returns the fabric control symbol this operation maps to.
137    pub fn fabric_symbol(self) -> Symbol {
138        match self {
139            Self::Read => stream_control_next_symbol(),
140            Self::Subscribe => stream_control_open_symbol(),
141            Self::Push => stream_control_push_symbol(),
142            Self::Cancel => stream_control_cancel_symbol(),
143            Self::Stats => stream_control_stats_symbol(),
144        }
145    }
146
147    /// Returns the capability required to invoke this operation.
148    pub fn capability(self) -> CapabilityName {
149        match self {
150            Self::Read => stream_read_capability(),
151            Self::Subscribe => stream_open_capability(),
152            Self::Push => stream_push_capability(),
153            Self::Cancel => stream_cancel_capability(),
154            Self::Stats => stream_stats_capability(),
155        }
156    }
157}
158
159/// Operation names exposed by the web bridge.
160pub fn web_stream_operation_symbols() -> [Symbol; 5] {
161    [
162        WebStreamOperation::Read.symbol(),
163        WebStreamOperation::Subscribe.symbol(),
164        WebStreamOperation::Push.symbol(),
165        WebStreamOperation::Cancel.symbol(),
166        WebStreamOperation::Stats.symbol(),
167    ]
168}
169
170/// Capability names required by browser-visible stream operations.
171pub fn web_stream_operation_capability_names() -> Vec<CapabilityName> {
172    [
173        WebStreamOperation::Read,
174        WebStreamOperation::Subscribe,
175        WebStreamOperation::Push,
176        WebStreamOperation::Cancel,
177        WebStreamOperation::Stats,
178    ]
179    .into_iter()
180    .map(WebStreamOperation::capability)
181    .collect()
182}
183
184/// Inspector data shown by browser stream tools.
185#[derive(Clone, Debug, PartialEq, Eq)]
186pub struct StreamInspectorRecord {
187    /// Id of the inspected stream.
188    pub stream_id: Symbol,
189    /// Current browser-side status of the stream.
190    pub status: BrowserStreamStatus,
191    /// Number of buffered packets.
192    pub buffered: usize,
193    /// Accumulated stream statistics.
194    pub stats: StreamStats,
195    /// Diagnostics reported for the stream.
196    pub diagnostics: Vec<Symbol>,
197    /// Inspector snapshot of the stream.
198    pub snapshot: StreamInspectorSnapshot,
199}
200
201/// Which kind of runtime a transport connects to.
202#[derive(Clone, Copy, Debug, PartialEq, Eq)]
203pub enum TransportKind {
204    /// A deterministic in-memory fixture (tests, replay).
205    Fixture,
206    /// An in-browser wasm runtime.
207    Wasm,
208    /// A local server (HTTP bootstrap + WebSocket live).
209    LocalServer,
210    /// A remote server (HTTP bootstrap + WebSocket live).
211    RemoteServer,
212    /// A kernel [`EvalFabric`](sim_kernel::EvalFabric) target: commits are
213    /// delegated to `realize`, proving a session is a realize target.
214    Fabric,
215}
216
217/// The location-transparent bus the UI targets. Implementors map these calls to
218/// `realize`/`EvalFabric` (`realize_final` for [`Transport::realize`] and
219/// `realize_events` for [`Transport::drain_events`]); the fixture maps them to
220/// an in-memory store.
221pub trait Transport {
222    /// Which kind of runtime this transport connects to.
223    fn kind(&self) -> TransportKind;
224
225    /// The current connection status.
226    fn status(&self) -> SessionStatus;
227
228    /// Read the current value of a resource.
229    fn read(&mut self, cx: &mut Cx, resource: &Symbol) -> Result<Expr>;
230
231    /// Realize a checked operation expression against a resource.
232    ///
233    /// This compatibility path wraps `operation` without authority metadata.
234    /// Session commits should call [`Transport::realize_operation`] so required
235    /// capabilities and result shapes are preserved.
236    fn realize(&mut self, cx: &mut Cx, resource: &Symbol, operation: &Expr) -> Result<Expr> {
237        self.realize_operation(cx, resource, &Operation::new(operation.clone()))
238    }
239
240    /// Realize a checked operation against a resource, returning the new value
241    /// (the `realize_final` surface). Implementations also record a
242    /// [`ChangeEvent`] for the resource.
243    fn realize_operation(
244        &mut self,
245        cx: &mut Cx,
246        resource: &Symbol,
247        operation: &Operation,
248    ) -> Result<Expr> {
249        self.commit_operation(cx, resource, operation, None)
250    }
251
252    /// Commit an operation, optionally requiring the resource to still match
253    /// `expected_current` on the server side.
254    fn commit_operation(
255        &mut self,
256        cx: &mut Cx,
257        resource: &Symbol,
258        operation: &Operation,
259        expected_current: Option<&Expr>,
260    ) -> Result<Expr> {
261        if let Some(expected) = expected_current {
262            let current = self.read(cx, resource)?;
263            if &current != expected {
264                return Err(sim_kernel::Error::HostError(format!(
265                    "resource '{resource}' is stale; refresh before committing"
266                )));
267            }
268        }
269        self.realize_operation(cx, resource, operation)
270    }
271
272    /// Drain the pending change events (the `realize_events` surface).
273    fn drain_events(&mut self, cx: &mut Cx) -> Result<Vec<ChangeEvent>>;
274
275    /// Subscribe to a stream and return browser-visible inspector data.
276    fn stream_subscribe(
277        &mut self,
278        cx: &mut Cx,
279        stream_id: &Symbol,
280    ) -> Result<StreamInspectorRecord>;
281
282    /// Read at most `limit` packets from a stream.
283    fn stream_read(
284        &mut self,
285        cx: &mut Cx,
286        stream_id: &Symbol,
287        limit: usize,
288    ) -> Result<Vec<StreamItem>>;
289
290    /// Push one stream envelope.
291    fn stream_push(
292        &mut self,
293        cx: &mut Cx,
294        stream_id: &Symbol,
295        envelope: StreamEnvelope,
296    ) -> Result<PushResult>;
297
298    /// Cancel a stream.
299    fn stream_cancel(&mut self, cx: &mut Cx, stream_id: &Symbol) -> Result<()>;
300
301    /// Return current stream stats.
302    fn stream_stats(&mut self, cx: &mut Cx, stream_id: &Symbol) -> Result<StreamStats>;
303
304    /// Return browser-visible inspector data without changing stream state.
305    fn stream_inspector(
306        &mut self,
307        cx: &mut Cx,
308        stream_id: &Symbol,
309    ) -> Result<StreamInspectorRecord>;
310}
sim_lib_web_bridge/transport.rs

sim_lib_web_bridge/
transport.rs
