sim_lib_web_bridge/
fixture.rs

1//! The fixture transport: a deterministic in-memory runtime.
2//!
3//! The fixture maps the bus onto an in-memory value store, so a session can be
4//! driven end to end without a server or browser. It also models connection
5//! loss and reconnection so the UI's session-status handling is testable.
6
7use std::collections::BTreeMap;
8
9use sim_kernel::{Cx, Error, Expr, Result, Symbol};
10use sim_lib_stream_core::{
11    PushResult, StreamEnvelope, StreamInspectorSnapshot, StreamItem, StreamMetadata, StreamPacket,
12    StreamStats, StreamValue, TransportProfile, stream_inspector_route_local_symbol,
13};
14use sim_lib_view::Operation;
15
16use crate::transport::{
17    BrowserStreamStatus, ChangeEvent, SessionStatus, StreamInspectorRecord, Transport,
18    TransportKind,
19};
20
21/// An in-memory transport for deterministic sessions and replay.
22pub struct FixtureTransport {
23    store: BTreeMap<Symbol, Expr>,
24    streams: BTreeMap<Symbol, FixtureStream>,
25    events: Vec<ChangeEvent>,
26    status: SessionStatus,
27}
28
29struct FixtureStream {
30    stream: StreamValue,
31    buffered: usize,
32    status: BrowserStreamStatus,
33    diagnostics: Vec<Symbol>,
34}
35
36impl Default for FixtureTransport {
37    fn default() -> Self {
38        Self::new()
39    }
40}
41
42impl FixtureTransport {
43    /// A connected, empty fixture.
44    pub fn new() -> Self {
45        Self {
46            store: BTreeMap::new(),
47            streams: BTreeMap::new(),
48            events: Vec::new(),
49            status: SessionStatus::Connected,
50        }
51    }
52
53    /// Seed a resource value (builder form).
54    pub fn with(mut self, resource: Symbol, value: Expr) -> Self {
55        self.store.insert(resource, value);
56        self
57    }
58
59    /// Seed or replace a resource value.
60    pub fn set(&mut self, resource: Symbol, value: Expr) {
61        self.store.insert(resource, value);
62    }
63
64    /// Seed a deterministic finite stream.
65    pub fn with_finite_stream(mut self, metadata: StreamMetadata, items: Vec<StreamItem>) -> Self {
66        self.set_finite_stream(metadata, items);
67        self
68    }
69
70    /// Seed a push stream.
71    pub fn with_push_stream(mut self, metadata: StreamMetadata) -> Self {
72        self.set_push_stream(metadata);
73        self
74    }
75
76    /// Seed or replace a deterministic finite stream.
77    pub fn set_finite_stream(&mut self, metadata: StreamMetadata, items: Vec<StreamItem>) {
78        self.streams.insert(
79            metadata.id().clone(),
80            FixtureStream {
81                buffered: items.len(),
82                stream: StreamValue::pull(metadata, items),
83                status: BrowserStreamStatus::Live,
84                diagnostics: Vec::new(),
85            },
86        );
87    }
88
89    /// Seed or replace a push stream.
90    pub fn set_push_stream(&mut self, metadata: StreamMetadata) {
91        self.streams.insert(
92            metadata.id().clone(),
93            FixtureStream {
94                stream: StreamValue::push(metadata),
95                buffered: 0,
96                status: BrowserStreamStatus::Live,
97                diagnostics: Vec::new(),
98            },
99        );
100    }
101
102    /// Mark a stream as refused after a profile diagnostic.
103    pub fn mark_stream_refused(&mut self, stream_id: &Symbol, diagnostic: Symbol) -> Result<()> {
104        let stream = self.stream_mut(stream_id)?;
105        stream.status = BrowserStreamStatus::RefusedProfile;
106        stream.diagnostics.push(diagnostic);
107        Ok(())
108    }
109
110    /// Simulate connection loss.
111    pub fn disconnect(&mut self) {
112        self.status = SessionStatus::Disconnected;
113    }
114
115    /// Simulate a reconnecting transport.
116    pub fn begin_reconnect(&mut self) {
117        self.status = SessionStatus::Reconnecting;
118    }
119
120    /// Simulate a restored connection.
121    pub fn reconnect(&mut self) {
122        self.status = SessionStatus::Connected;
123    }
124
125    fn ensure_live(&self) -> Result<()> {
126        if self.status.is_live() {
127            Ok(())
128        } else {
129            Err(Error::HostError(format!(
130                "fixture session is {:?}; no traffic can flow",
131                self.status
132            )))
133        }
134    }
135
136    fn stream_ref(&self, stream_id: &Symbol) -> Result<&FixtureStream> {
137        self.streams
138            .get(stream_id)
139            .ok_or_else(|| Error::UnknownSymbol {
140                symbol: stream_id.clone(),
141            })
142    }
143
144    fn stream_mut(&mut self, stream_id: &Symbol) -> Result<&mut FixtureStream> {
145        self.streams
146            .get_mut(stream_id)
147            .ok_or_else(|| Error::UnknownSymbol {
148                symbol: stream_id.clone(),
149            })
150    }
151
152    fn visible_stream_status(&self, stream: &FixtureStream) -> BrowserStreamStatus {
153        match self.status {
154            SessionStatus::Disconnected => BrowserStreamStatus::Disconnected,
155            SessionStatus::Reconnecting => BrowserStreamStatus::Reconnecting,
156            _ => stream.status,
157        }
158    }
159
160    fn inspector(&self, stream_id: &Symbol) -> Result<StreamInspectorRecord> {
161        let stream = self.stream_ref(stream_id)?;
162        let stats = stream.stream.stats()?;
163        let status = self.visible_stream_status(stream);
164        let queue_depth = stream.stream.queue_depth()?;
165        let observed = stats
166            .accepted
167            .max(stats.yielded.saturating_add(queue_depth as u64));
168        let snapshot = StreamInspectorSnapshot::new(
169            stream.stream.metadata(),
170            stream_inspector_route_local_symbol(),
171            TransportProfile::memory_local().name().clone(),
172            status.inspector_status(),
173            queue_depth,
174            &stats,
175            observed.checked_sub(1),
176            stream.diagnostics.clone(),
177        );
178        Ok(StreamInspectorRecord {
179            stream_id: stream_id.clone(),
180            status,
181            buffered: stream.buffered,
182            stats,
183            diagnostics: stream.diagnostics.clone(),
184            snapshot,
185        })
186    }
187}
188
189impl Transport for FixtureTransport {
190    fn kind(&self) -> TransportKind {
191        TransportKind::Fixture
192    }
193
194    fn status(&self) -> SessionStatus {
195        self.status
196    }
197
198    fn read(&mut self, _cx: &mut Cx, resource: &Symbol) -> Result<Expr> {
199        self.ensure_live()?;
200        self.store
201            .get(resource)
202            .cloned()
203            .ok_or_else(|| Error::UnknownSymbol {
204                symbol: resource.clone(),
205            })
206    }
207
208    fn realize_operation(
209        &mut self,
210        _cx: &mut Cx,
211        resource: &Symbol,
212        operation: &Operation,
213    ) -> Result<Expr> {
214        self.ensure_live()?;
215        let new_value = apply_operation(self.store.get(resource), &operation.form)?;
216        self.store.insert(resource.clone(), new_value.clone());
217        self.events.push(ChangeEvent {
218            resource: resource.clone(),
219        });
220        Ok(new_value)
221    }
222
223    fn drain_events(&mut self, _cx: &mut Cx) -> Result<Vec<ChangeEvent>> {
224        Ok(std::mem::take(&mut self.events))
225    }
226
227    fn stream_subscribe(
228        &mut self,
229        _cx: &mut Cx,
230        stream_id: &Symbol,
231    ) -> Result<StreamInspectorRecord> {
232        self.ensure_live()?;
233        self.inspector(stream_id)
234    }
235
236    fn stream_read(
237        &mut self,
238        _cx: &mut Cx,
239        stream_id: &Symbol,
240        limit: usize,
241    ) -> Result<Vec<StreamItem>> {
242        self.ensure_live()?;
243        let stream = self.stream_mut(stream_id)?;
244        let items = stream.stream.take_packets(limit)?;
245        stream.buffered = stream.buffered.saturating_sub(items.len());
246        if stream.stream.stats()?.cancelled {
247            stream.status = BrowserStreamStatus::Cancelled;
248        } else if stream.stream.is_done()? {
249            stream.status = BrowserStreamStatus::Ended;
250        }
251        Ok(items)
252    }
253
254    fn stream_push(
255        &mut self,
256        _cx: &mut Cx,
257        stream_id: &Symbol,
258        envelope: StreamEnvelope,
259    ) -> Result<PushResult> {
260        self.ensure_live()?;
261        if envelope.stream_id() != stream_id {
262            return Err(Error::HostError(format!(
263                "stream push envelope id {} does not match target {}",
264                envelope.stream_id(),
265                stream_id
266            )));
267        }
268        let item = StreamItem::with_ticks(envelope.packet().clone(), envelope.ticks().to_vec())?;
269        let stream = self.stream_mut(stream_id)?;
270        let result = stream.stream.push_packet(item)?;
271        match &result {
272            PushResult::Accepted => {
273                stream.buffered = stream.buffered.saturating_add(1);
274                stream.status = BrowserStreamStatus::Live;
275            }
276            PushResult::DroppedNewest(item) | PushResult::DroppedOldest(item) => {
277                stream.status = BrowserStreamStatus::BufferOverflow;
278                if matches!(item.packet(), StreamPacket::Diagnostic(_)) {
279                    stream
280                        .diagnostics
281                        .push(Symbol::qualified("stream/browser", "buffer-overflow"));
282                }
283            }
284            PushResult::Rejected(_) => {
285                stream.status = BrowserStreamStatus::BufferOverflow;
286            }
287            PushResult::Closed(_) => {
288                stream.status = BrowserStreamStatus::Cancelled;
289            }
290        }
291        Ok(result)
292    }
293
294    fn stream_cancel(&mut self, _cx: &mut Cx, stream_id: &Symbol) -> Result<()> {
295        self.ensure_live()?;
296        let stream = self.stream_mut(stream_id)?;
297        stream.stream.cancel()?;
298        stream.buffered = 0;
299        stream.status = BrowserStreamStatus::Cancelled;
300        Ok(())
301    }
302
303    fn stream_stats(&mut self, _cx: &mut Cx, stream_id: &Symbol) -> Result<StreamStats> {
304        self.stream_ref(stream_id)?.stream.stats()
305    }
306
307    fn stream_inspector(
308        &mut self,
309        _cx: &mut Cx,
310        stream_id: &Symbol,
311    ) -> Result<StreamInspectorRecord> {
312        self.inspector(stream_id)
313    }
314}
315
316/// Interpret a checked operation against the current value. The fixture
317/// understands the universal editor's `set-value` operation; unknown operations
318/// fail closed.
319fn apply_operation(current: Option<&Expr>, operation: &Expr) -> Result<Expr> {
320    let Expr::Map(entries) = operation else {
321        return Err(Error::HostError("operation is not a map".to_owned()));
322    };
323    let op_name = entries.iter().find_map(|(key, value)| {
324        let is_op = matches!(key, Expr::Symbol(symbol) if &*symbol.name == "op");
325        match value {
326            Expr::Symbol(symbol) if is_op => Some(symbol.name.to_string()),
327            _ => None,
328        }
329    });
330    match op_name.as_deref() {
331        Some("set-value") => entries
332            .iter()
333            .find_map(|(key, value)| {
334                matches!(key, Expr::Symbol(symbol) if &*symbol.name == "value").then_some(value)
335            })
336            .cloned()
337            .ok_or_else(|| Error::HostError("set-value operation is missing a 'value'".to_owned())),
338        Some(other) => Err(Error::HostError(format!(
339            "fixture transport cannot realize operation '{other}'"
340        ))),
341        None => {
342            let _ = current;
343            Err(Error::HostError(
344                "operation is missing an 'op' tag".to_owned(),
345            ))
346        }
347    }
348}
sim_lib_web_bridge/fixture.rs

sim_lib_web_bridge/
fixture.rs
