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::{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(&self, 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(&mut self, resource: &Symbol, operation: &Operation) -> Result<Expr> {
209        self.ensure_live()?;
210        let new_value = apply_operation(self.store.get(resource), &operation.form)?;
211        self.store.insert(resource.clone(), new_value.clone());
212        self.events.push(ChangeEvent {
213            resource: resource.clone(),
214        });
215        Ok(new_value)
216    }
217
218    fn drain_events(&mut self) -> Vec<ChangeEvent> {
219        std::mem::take(&mut self.events)
220    }
221
222    fn stream_subscribe(&mut self, stream_id: &Symbol) -> Result<StreamInspectorRecord> {
223        self.ensure_live()?;
224        self.inspector(stream_id)
225    }
226
227    fn stream_read(&mut self, stream_id: &Symbol, limit: usize) -> Result<Vec<StreamItem>> {
228        self.ensure_live()?;
229        let stream = self.stream_mut(stream_id)?;
230        let items = stream.stream.take_packets(limit)?;
231        stream.buffered = stream.buffered.saturating_sub(items.len());
232        if stream.stream.stats()?.cancelled {
233            stream.status = BrowserStreamStatus::Cancelled;
234        } else if stream.stream.is_done()? {
235            stream.status = BrowserStreamStatus::Ended;
236        }
237        Ok(items)
238    }
239
240    fn stream_push(&mut self, stream_id: &Symbol, envelope: StreamEnvelope) -> Result<PushResult> {
241        self.ensure_live()?;
242        if envelope.stream_id() != stream_id {
243            return Err(Error::HostError(format!(
244                "stream push envelope id {} does not match target {}",
245                envelope.stream_id(),
246                stream_id
247            )));
248        }
249        let item = StreamItem::with_ticks(envelope.packet().clone(), envelope.ticks().to_vec())?;
250        let stream = self.stream_mut(stream_id)?;
251        let result = stream.stream.push_packet(item)?;
252        match &result {
253            PushResult::Accepted => {
254                stream.buffered = stream.buffered.saturating_add(1);
255                stream.status = BrowserStreamStatus::Live;
256            }
257            PushResult::DroppedNewest(item) | PushResult::DroppedOldest(item) => {
258                stream.status = BrowserStreamStatus::BufferOverflow;
259                if matches!(item.packet(), StreamPacket::Diagnostic(_)) {
260                    stream
261                        .diagnostics
262                        .push(Symbol::qualified("stream/browser", "buffer-overflow"));
263                }
264            }
265            PushResult::Rejected(_) => {
266                stream.status = BrowserStreamStatus::BufferOverflow;
267            }
268            PushResult::Closed(_) => {
269                stream.status = BrowserStreamStatus::Cancelled;
270            }
271        }
272        Ok(result)
273    }
274
275    fn stream_cancel(&mut self, stream_id: &Symbol) -> Result<()> {
276        self.ensure_live()?;
277        let stream = self.stream_mut(stream_id)?;
278        stream.stream.cancel()?;
279        stream.buffered = 0;
280        stream.status = BrowserStreamStatus::Cancelled;
281        Ok(())
282    }
283
284    fn stream_stats(&self, stream_id: &Symbol) -> Result<StreamStats> {
285        self.stream_ref(stream_id)?.stream.stats()
286    }
287
288    fn stream_inspector(&self, stream_id: &Symbol) -> Result<StreamInspectorRecord> {
289        self.inspector(stream_id)
290    }
291}
292
293/// Interpret a checked operation against the current value. The fixture
294/// understands the universal editor's `set-value` operation; unknown operations
295/// fail closed.
296fn apply_operation(current: Option<&Expr>, operation: &Expr) -> Result<Expr> {
297    let Expr::Map(entries) = operation else {
298        return Err(Error::HostError("operation is not a map".to_owned()));
299    };
300    let op_name = entries.iter().find_map(|(key, value)| {
301        let is_op = matches!(key, Expr::Symbol(symbol) if &*symbol.name == "op");
302        match value {
303            Expr::Symbol(symbol) if is_op => Some(symbol.name.to_string()),
304            _ => None,
305        }
306    });
307    match op_name.as_deref() {
308        Some("set-value") => entries
309            .iter()
310            .find_map(|(key, value)| {
311                matches!(key, Expr::Symbol(symbol) if &*symbol.name == "value").then_some(value)
312            })
313            .cloned()
314            .ok_or_else(|| Error::HostError("set-value operation is missing a 'value'".to_owned())),
315        Some(other) => Err(Error::HostError(format!(
316            "fixture transport cannot realize operation '{other}'"
317        ))),
318        None => {
319            let _ = current;
320            Err(Error::HostError(
321                "operation is missing an 'op' tag".to_owned(),
322            ))
323        }
324    }
325}
sim_lib_web_bridge/fixture.rs

sim_lib_web_bridge/
fixture.rs
