telltale_vm/vm/
runtime_and_execution.rs

1/// Approximate retained state for the live VM runtime.
2#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
3pub struct VmMemoryUsage {
4    /// Session-store retained state.
5    pub session_store: SessionStoreMemoryUsage,
6    /// Number of coroutine records still retained by the VM.
7    pub coroutine_records: usize,
8    /// Number of terminal coroutine records retained by the VM.
9    pub terminal_coroutines: usize,
10    /// Number of loaded immutable program records.
11    pub program_count: usize,
12    /// Total instruction count across loaded programs.
13    pub program_instruction_count: usize,
14    /// Number of retained observable events.
15    pub obs_events: usize,
16    /// Number of retained effect-trace entries.
17    pub effect_trace_entries: usize,
18    /// Number of retained replay-consumption artifacts.
19    pub communication_artifacts: usize,
20    /// Number of retained output-condition checks.
21    pub output_condition_checks: usize,
22    /// Estimated retained bytes by VM subsystem.
23    pub retained_bytes: VmRetainedBytes,
24}
25
26/// Estimated retained bytes for VM subsystems.
27#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
28pub struct VmRetainedBytes {
29    /// Session-store retained bytes.
30    pub session_store: usize,
31    /// Coroutine state.
32    pub coroutines: usize,
33    /// Immutable program storage.
34    pub programs: usize,
35    /// Resource-state storage.
36    pub resource_states: usize,
37    /// Observable/effect trace storage.
38    pub traces: usize,
39    /// Replay-state and replay-artifact storage.
40    pub replay: usize,
41    /// Output-condition diagnostics.
42    pub output_condition_checks: usize,
43    /// Scheduler and control-state bookkeeping.
44    pub scheduler_and_control: usize,
45    /// Symbol interning tables.
46    pub symbols: usize,
47    /// Guard-layer resources.
48    pub guard_layer: usize,
49    /// Session monitor metadata.
50    pub monitor: usize,
51    /// Arena slot storage.
52    pub arena: usize,
53    /// Aggregate retained bytes across VM subsystems.
54    pub total: usize,
55}
56
57fn vm_serialized_bytes<T: Serialize>(value: &T) -> usize {
58    bincode::serialized_size(value)
59        .ok()
60        .and_then(|bytes| usize::try_from(bytes).ok())
61        .unwrap_or(0)
62}
63
64impl VM {
65    fn communication_replay_enabled(&self) -> bool {
66        !matches!(
67            self.config.communication_replay_mode,
68            CommunicationReplayMode::Off
69        )
70    }
71
72    fn intern_load_plan_symbols(&mut self, plan: &crate::session::SessionOpenPlan, sid: SessionId) {
73        for role in plan.roles() {
74            let _: StringId = self.role_symbols.intern(role);
75        }
76        let _: StringId = self.handler_symbols.intern(crate::session::DEFAULT_HANDLER_ID);
77        let edge_handlers: Vec<_> = self
78            .sessions
79            .get(sid)
80            .map(|session| session.edge_handlers.keys().cloned().collect())
81            .unwrap_or_default();
82        for edge in edge_handlers {
83            let _: EdgeId = self.intern_edge(&edge);
84        }
85    }
86
87    /// Create a VM instance from configuration.
88    #[must_use]
89    pub fn new(config: VMConfig) -> Self {
90        Self::new_with_models(config)
91    }
92
93    fn bind_default_handlers_for_session(&mut self, sid: SessionId) {
94        self.sessions.set_default_handler_for_session(
95            sid,
96            crate::session::DEFAULT_HANDLER_ID.to_string(),
97        );
98        self.handler_symbols.intern(crate::session::DEFAULT_HANDLER_ID);
99    }
100
101    fn ensure_session_capacity(&self) -> Result<(), VMError> {
102        if self.sessions.active_count() >= self.config.max_sessions {
103            return Err(VMError::TooManySessions {
104                max: self.config.max_sessions,
105            });
106        }
107        Ok(())
108    }
109
110    fn coroutine_runtime_eligible(&self, coro_id: usize) -> bool {
111        let Some(idx) = self.coro_index(coro_id) else {
112            return false;
113        };
114        let role = &self.coroutines[idx].role;
115        !(self.paused_coro_ids.contains(&coro_id)
116            || self.paused_roles.contains(role)
117            || self.crashed_sites.contains(role)
118            || self.timed_out_coro_ids.contains(&coro_id)
119            || self.timed_out_sites.contains_key(role))
120    }
121
122    fn mark_eligibility_dirty(&mut self) {
123        self.eligibility_dirty = true;
124    }
125
126    fn sync_ready_eligibility_for(&mut self, coro_id: usize) {
127        let eligible = self.sched.is_ready(coro_id) && self.coroutine_runtime_eligible(coro_id);
128        let eligibility = if eligible {
129            crate::scheduler::ReadyEligibility::Eligible
130        } else {
131            crate::scheduler::ReadyEligibility::Ineligible
132        };
133        self.sched.set_ready_eligibility(coro_id, eligibility);
134        #[cfg(debug_assertions)]
135        {
136            if eligible {
137                self.eligible_ready.insert(coro_id);
138            } else {
139                self.eligible_ready.remove(&coro_id);
140            }
141        }
142    }
143
144    fn refresh_ready_eligibility(&mut self) {
145        self.sched.clear_ready_eligibility();
146        #[cfg(debug_assertions)]
147        self.eligible_ready.clear();
148        for coro_id in self.sched.ready_set_snapshot() {
149            let eligible = self.coroutine_runtime_eligible(coro_id);
150            let eligibility = if eligible {
151                crate::scheduler::ReadyEligibility::Eligible
152            } else {
153                crate::scheduler::ReadyEligibility::Ineligible
154            };
155            self.sched.set_ready_eligibility(coro_id, eligibility);
156            #[cfg(debug_assertions)]
157            if eligible {
158                self.eligible_ready.insert(coro_id);
159            }
160        }
161        self.eligibility_dirty = false;
162    }
163
164    fn ensure_ready_eligibility(&mut self) {
165        if self.eligibility_dirty {
166            self.refresh_ready_eligibility();
167        }
168    }
169
170    #[cfg(debug_assertions)]
171    fn debug_assert_ready_eligibility_consistent(&self) {
172        for coro_id in &self.eligible_ready {
173            debug_assert!(self.sched.is_ready(*coro_id));
174            debug_assert!(self.coroutine_runtime_eligible(*coro_id));
175        }
176    }
177
178    fn sync_communication_consumption_mode(&mut self) {
179        self.communication_consumption
180            .set_mode(self.config.communication_replay_mode);
181    }
182
183    fn allocate_send_sequence(&mut self, edge: &Edge) -> u64 {
184        if !self.communication_replay_enabled() {
185            // Off mode preserves legacy behavior and avoids replay bookkeeping.
186            return 0;
187        }
188        self.sync_communication_consumption_mode();
189        self.communication_consumption.allocate_send_sequence(edge)
190    }
191
192    fn consume_receive_identity(
193        &mut self,
194        identity: CommunicationIdentity,
195    ) -> Result<CommunicationConsumeResult, CommunicationReplayError> {
196        if !self.communication_replay_enabled() {
197            // Off mode intentionally skips replay-consumption state and artifacts.
198            return Ok(CommunicationConsumeResult {
199                mode: CommunicationReplayMode::Off,
200                pre_root: self.communication_consumption.root(),
201                post_root: self.communication_consumption.root(),
202                consumed_nullifier: None,
203            });
204        }
205        self.sync_communication_consumption_mode();
206        let result = self.communication_consumption.consume_receive(&identity)?;
207        self.communication_consumption_artifacts.push(
208            CommunicationConsumptionArtifact {
209                tick: self.clock.tick,
210                identity,
211                mode: result.mode,
212                pre_root: result.pre_root,
213                post_root: result.post_root,
214            },
215            &self.config.observability_retention,
216        );
217        Ok(result)
218    }
219
220    fn session_open_plan(
221        &mut self,
222        image: &CodeImage,
223    ) -> &crate::session::SessionOpenPlan {
224        let key = format!("{image:p}");
225        self.session_open_plans
226            .entry(key)
227            .or_insert_with(|| crate::session::SessionOpenPlan::new(&image.roles(), &image.local_types))
228    }
229
230    fn open_choreography_session(
231        &mut self,
232        plan: &crate::session::SessionOpenPlan,
233    ) -> (SessionId, Vec<String>) {
234        let sid = self.sessions.next_session_id();
235        let roles = plan.roles().to_vec();
236        self.sessions
237            .open_with_sid_from_plan(sid, plan, &self.config.buffer_config);
238        (sid, roles)
239    }
240
241    fn finalize_open_choreography_session(
242        &mut self,
243        sid: SessionId,
244        roles: &[String],
245        plan: &crate::session::SessionOpenPlan,
246    ) -> Result<(), VMError> {
247        self.next_session_id = self.sessions.next_session_id();
248        self.bind_default_handlers_for_session(sid);
249        self.intern_load_plan_symbols(plan, sid);
250        self.monitor.set_kind(sid, SessionKind::Peer);
251        self.resource_states
252            .entry(sid)
253            .or_default();
254        self.apply_open_delta(sid)
255            .map_err(VMError::PersistenceError)?;
256        self.obs_trace.push(
257            ObsEvent::Opened {
258                tick: self.clock.tick,
259                session: sid,
260                roles: roles.to_vec(),
261            },
262            &self.config.observability_retention,
263        );
264        Ok(())
265    }
266
267    fn spawn_coroutine_for_role(
268        &mut self,
269        image: &CodeImage,
270        sid: SessionId,
271        role: &str,
272    ) -> Result<(), VMError> {
273        if self.coroutines.len() >= self.config.max_coroutines {
274            return Err(VMError::TooManyCoroutines {
275                max: self.config.max_coroutines,
276            });
277        }
278
279        let program_id = self
280            .programs
281            .intern(image.programs.get(role).cloned().unwrap_or_default());
282        if self.code.is_none() {
283            let program = self
284                .programs
285                .get(program_id)
286                .expect("interned program must exist")
287                .clone();
288            self.code = Some(program);
289        }
290
291        let coro_id = self.next_coro_id;
292        self.next_coro_id += 1;
293
294        let endpoint = Endpoint {
295            sid,
296            role: role.to_string(),
297        };
298        self.role_coroutines
299            .entry(role.to_string())
300            .or_default()
301            .push(coro_id);
302        if self.paused_roles.contains(role) {
303            self.paused_coro_ids.insert(coro_id);
304        }
305        if self.timed_out_sites.contains_key(role) {
306            self.timed_out_coro_ids.insert(coro_id);
307        }
308        let mut coro = Coroutine::new(
309            coro_id,
310            program_id,
311            sid,
312            role.to_string(),
313            self.config.num_registers,
314            self.config.initial_cost_budget,
315        );
316        coro.owned_endpoints.push(endpoint.clone());
317        if !coro.regs.is_empty() {
318            coro.regs[0] = Value::Endpoint(endpoint);
319        }
320        self.sched.add_ready(coro_id);
321        self.coroutines.push(coro);
322        self.coro_slots.insert(coro_id, self.coroutines.len() - 1);
323        self.sync_ready_eligibility_for(coro_id);
324        Ok(())
325    }
326
327    fn spawn_session_coroutines(
328        &mut self,
329        image: &CodeImage,
330        sid: SessionId,
331        roles: &[String],
332    ) -> Result<(), VMError> {
333        for role in roles {
334            self.spawn_coroutine_for_role(image, sid, role)?;
335        }
336        Ok(())
337    }
338
339    /// Load a choreography from a verified code image.
340    ///
341    /// Creates a session (with local types), spawns coroutines per role,
342    /// and returns the session ID. Type state is initialized in the
343    /// session store — no separate monitor needed.
344    ///
345    /// # Errors
346    ///
347    /// Returns an error if session or coroutine limits are exceeded.
348    pub fn load_choreography(&mut self, image: &CodeImage) -> Result<SessionId, VMError> {
349        self.ensure_session_capacity()?;
350        image.validate_runtime_shape().map_err(|reason| VMError::InvalidCodeImage { reason })?;
351        let plan = self.session_open_plan(image).clone();
352        let (sid, roles) = self.open_choreography_session(&plan);
353        self.finalize_open_choreography_session(sid, &roles, &plan)?;
354        self.programs.reserve(image.programs.len());
355        self.coroutines.reserve(roles.len());
356        self.spawn_session_coroutines(image, sid, &roles)?;
357        Ok(sid)
358    }
359
360    /// Execute one scheduler round: advance at most one ready coroutine.
361    ///
362    /// # Errors
363    ///
364    /// Returns a `VMError` if a coroutine faults.
365    #[allow(clippy::too_many_lines)]
366    pub(crate) fn kernel_step_round(
367        &mut self,
368        handler: &dyn EffectHandler,
369        n: usize,
370    ) -> Result<StepResult, VMError> {
371        #[cfg(debug_assertions)]
372        debug_assert!(self.wf_vm_state().is_ok());
373        if n == 0 {
374            return Err(VMError::InvalidConcurrency { n });
375        }
376        self.last_sched_step = None;
377        self.clock.advance();
378        if self.all_done() {
379            return Ok(StepResult::AllDone);
380        }
381
382        // Event ordering contract: topology effects ingress first each round,
383        // before unblocking and scheduler selection.
384        self.ingest_topology_events(handler)?;
385        self.prune_expired_timeouts();
386        self.try_unblock_receivers();
387        self.ensure_ready_eligibility();
388        #[cfg(debug_assertions)]
389        self.debug_assert_ready_eligibility_consistent();
390        if !self.sched.has_eligible_ready() {
391            return Ok(StepResult::Stuck);
392        }
393        let coroutines = &self.coroutines;
394        let coro_slots = &self.coro_slots;
395        let Some(coro_id) = self.sched.pick_eligible_runnable(|id| {
396            coro_slots
397                .get(&id)
398                .and_then(|idx| coroutines.get(*idx))
399                .or_else(|| coroutines.get(id).filter(|coro| coro.id == id))
400                .or_else(|| coroutines.iter().find(|coro| coro.id == id))
401                .is_some_and(|coro| !coro.progress_tokens.is_empty())
402        }) else {
403            return Ok(StepResult::Stuck);
404        };
405        #[cfg(debug_assertions)]
406        self.eligible_ready.remove(&coro_id);
407
408        let result = self.exec_instr(coro_id, handler);
409
410        match result {
411            Ok(ExecOutcome::Continue) => {
412                self.last_sched_step = Some(SchedStepDebug {
413                    selected_coro: coro_id,
414                    exec_status: SchedExecStatus::Continue,
415                });
416                self.sched.reschedule(coro_id);
417                self.sync_ready_eligibility_for(coro_id);
418            }
419            Ok(ExecOutcome::Blocked(reason)) => {
420                let yielded = matches!(reason, BlockReason::Spawn);
421                self.last_sched_step = Some(SchedStepDebug {
422                    selected_coro: coro_id,
423                    exec_status: if yielded {
424                        SchedExecStatus::Yielded
425                    } else {
426                        SchedExecStatus::Blocked
427                    },
428                });
429                if yielded {
430                    self.sched.reschedule(coro_id);
431                    self.sync_ready_eligibility_for(coro_id);
432                } else {
433                    self.sched.mark_blocked(coro_id, reason);
434                    #[cfg(debug_assertions)]
435                    self.eligible_ready.remove(&coro_id);
436                }
437            }
438            Ok(ExecOutcome::Halted) => {
439                self.last_sched_step = Some(SchedStepDebug {
440                    selected_coro: coro_id,
441                    exec_status: SchedExecStatus::Halted,
442                });
443                self.sched.mark_done(coro_id);
444                #[cfg(debug_assertions)]
445                self.eligible_ready.remove(&coro_id);
446                self.obs_trace.push(
447                    ObsEvent::Halted {
448                        tick: self.clock.tick,
449                        coro_id,
450                    },
451                    &self.config.observability_retention,
452                );
453            }
454            Err(fault) => {
455                self.last_sched_step = Some(SchedStepDebug {
456                    selected_coro: coro_id,
457                    exec_status: SchedExecStatus::Faulted,
458                });
459                self.obs_trace.push(
460                    ObsEvent::Faulted {
461                        tick: self.clock.tick,
462                        coro_id,
463                        fault: fault.clone(),
464                    },
465                    &self.config.observability_retention,
466                );
467                let Some(idx) = self.coro_index(coro_id) else {
468                    return Err(VMError::Fault { coro_id, fault });
469                };
470                self.coroutines[idx].status = CoroStatus::Faulted(fault.clone());
471                self.sched.mark_done(coro_id);
472                #[cfg(debug_assertions)]
473                self.eligible_ready.remove(&coro_id);
474                return Err(VMError::Fault { coro_id, fault });
475            }
476        }
477
478        if self.all_done() {
479            #[cfg(debug_assertions)]
480            self.debug_assert_ready_eligibility_consistent();
481            #[cfg(debug_assertions)]
482            debug_assert!(self.wf_vm_state().is_ok());
483            Ok(StepResult::AllDone)
484        } else {
485            #[cfg(debug_assertions)]
486            self.debug_assert_ready_eligibility_consistent();
487            #[cfg(debug_assertions)]
488            debug_assert!(self.wf_vm_state().is_ok());
489            Ok(StepResult::Continue)
490        }
491    }
492
493    /// Execute one scheduler step: pick a coroutine, run one instruction.
494    ///
495    /// # Errors
496    ///
497    /// Returns a `VMError` if a coroutine faults.
498    pub fn step(&mut self, handler: &dyn EffectHandler) -> Result<StepResult, VMError> {
499        self.step_round(handler, 1)
500    }
501
502    /// Execute one scheduler round through the canonical kernel API.
503    ///
504    /// # Errors
505    ///
506    /// Returns a `VMError` if a coroutine faults.
507    pub fn step_round(
508        &mut self,
509        handler: &dyn EffectHandler,
510        n: usize,
511    ) -> Result<StepResult, VMError> {
512        VMKernel::step_round(self, handler, n)
513    }
514
515    /// Run the VM until all coroutines complete or an error occurs, with concurrency N.
516    ///
517    /// `max_rounds` prevents infinite loops.
518    ///
519    /// # Errors
520    ///
521    /// Returns a `VMError` if any coroutine faults.
522    pub fn run_concurrent(
523        &mut self,
524        handler: &dyn EffectHandler,
525        max_rounds: usize,
526        concurrency: usize,
527    ) -> Result<RunStatus, VMError> {
528        VMKernel::run_concurrent(self, handler, max_rounds, concurrency)
529    }
530
531    /// Run the VM until all coroutines complete or an error occurs.
532    ///
533    /// `max_steps` prevents infinite loops.
534    ///
535    /// # Errors
536    ///
537    /// Returns a `VMError` if any coroutine faults.
538    pub fn run(
539        &mut self,
540        handler: &dyn EffectHandler,
541        max_steps: usize,
542    ) -> Result<RunStatus, VMError> {
543        VMKernel::run(self, handler, max_steps)
544    }
545
546    /// Run with replayed effect outcomes captured from a prior execution.
547    ///
548    /// The `fallback` handler is only used for optional hooks not encoded in
549    /// replay entries.
550    ///
551    /// # Errors
552    ///
553    /// Returns a `VMError` if replay data is exhausted/mismatched or a coroutine faults.
554    pub fn run_replay(
555        &mut self,
556        fallback: &dyn EffectHandler,
557        replay_trace: &[EffectTraceEntry],
558        max_steps: usize,
559    ) -> Result<RunStatus, VMError> {
560        self.run_replay_shared(
561            fallback,
562            Arc::<[EffectTraceEntry]>::from(replay_trace),
563            max_steps,
564        )
565    }
566
567    /// Run with replayed effect outcomes using shared trace storage.
568    ///
569    /// Accepts an `Arc`-backed trace to avoid cloning when callers already hold
570    /// shared replay buffers.
571    ///
572    /// # Errors
573    ///
574    /// Returns a `VMError` if replay data is exhausted/mismatched or a coroutine faults.
575    pub fn run_replay_shared(
576        &mut self,
577        fallback: &dyn EffectHandler,
578        replay_trace: Arc<[EffectTraceEntry]>,
579        max_steps: usize,
580    ) -> Result<RunStatus, VMError> {
581        let replay = ReplayEffectHandler::with_fallback(replay_trace, fallback);
582        self.run(&replay, max_steps)
583    }
584
585    /// Run concurrently with replayed effect outcomes.
586    ///
587    /// # Errors
588    ///
589    /// Returns a `VMError` if replay data is exhausted/mismatched or a coroutine faults.
590    pub fn run_concurrent_replay(
591        &mut self,
592        fallback: &dyn EffectHandler,
593        replay_trace: &[EffectTraceEntry],
594        max_rounds: usize,
595        concurrency: usize,
596    ) -> Result<RunStatus, VMError> {
597        self.run_concurrent_replay_shared(
598            fallback,
599            Arc::<[EffectTraceEntry]>::from(replay_trace),
600            max_rounds,
601            concurrency,
602        )
603    }
604
605    /// Run concurrently with replayed outcomes using shared trace storage.
606    ///
607    /// # Errors
608    ///
609    /// Returns a `VMError` if replay data is exhausted/mismatched or a coroutine faults.
610    pub fn run_concurrent_replay_shared(
611        &mut self,
612        fallback: &dyn EffectHandler,
613        replay_trace: Arc<[EffectTraceEntry]>,
614        max_rounds: usize,
615        concurrency: usize,
616    ) -> Result<RunStatus, VMError> {
617        let replay = ReplayEffectHandler::with_fallback(replay_trace, fallback);
618        self.run_concurrent(&replay, max_rounds, concurrency)
619    }
620
621    /// Get the observable trace.
622    #[must_use]
623    pub fn trace(&self) -> &[ObsEvent] {
624        self.obs_trace.as_slice()
625    }
626
627    /// Reap closed sessions once all associated coroutines are terminal.
628    pub fn reap_closed_sessions(&mut self) -> Vec<ClosedSessionSummary> {
629        let eligible: Vec<SessionId> = self
630            .sessions
631            .closed_session_ids()
632            .into_iter()
633            .filter(|sid| {
634                self.coroutines
635                    .iter()
636                    .filter(|coro| coro.session_id == *sid)
637                    .all(Coroutine::is_terminal)
638            })
639            .collect();
640        if eligible.is_empty() {
641            return Vec::new();
642        }
643
644        for sid in &eligible {
645            self.monitor.remove_kind(*sid);
646            self.resource_states.remove(sid);
647            self.communication_consumption.prune_session(*sid);
648        }
649        self.coroutines.retain(|coro| {
650            !(eligible.contains(&coro.session_id) && coro.is_terminal())
651        });
652        self.rebuild_coroutine_indexes();
653        self.sessions.reap_sessions(&eligible)
654    }
655
656    /// Lean-aligned observable trace accessor.
657    #[must_use]
658    pub fn obs_trace(&self) -> &[ObsEvent] {
659        self.obs_trace.as_slice()
660    }
661
662    /// Number of interned role symbols.
663    #[must_use]
664    pub fn role_symbol_count(&self) -> usize {
665        self.role_symbols.len()
666    }
667
668    /// Number of interned label symbols.
669    #[must_use]
670    pub fn label_symbol_count(&self) -> usize {
671        self.label_symbols.len()
672    }
673
674    /// Number of interned handler symbols.
675    #[must_use]
676    pub fn handler_symbol_count(&self) -> usize {
677        self.handler_symbols.len()
678    }
679
680    /// Number of interned edge symbols.
681    #[must_use]
682    pub fn edge_symbol_count(&self) -> usize {
683        self.edge_symbols.len()
684    }
685
686    /// Access VM configuration.
687    #[must_use]
688    pub fn config(&self) -> &VMConfig {
689        &self.config
690    }
691
692    /// Last scheduler-dispatched step metadata, if any coroutine ran.
693    #[must_use]
694    pub fn last_sched_step(&self) -> Option<&SchedStepDebug> {
695        self.last_sched_step.as_ref()
696    }
697
698    /// Scheduler-dispatched step counter.
699    #[must_use]
700    pub fn scheduler_step_count(&self) -> usize {
701        self.sched.step_count()
702    }
703
704    /// Number of coroutine records in the VM.
705    #[must_use]
706    pub fn coroutine_count(&self) -> usize {
707        self.coroutines.len()
708    }
709
710    /// Next session identifier reserved for allocation.
711    #[must_use]
712    pub fn next_session_id(&self) -> SessionId {
713        self.sessions.next_session_id()
714    }
715
716    /// Number of active sessions in the VM.
717    #[must_use]
718    pub fn session_count(&self) -> usize {
719        self.sessions.active_count()
720    }
721
722    /// Number of sessions still resident in the VM, including closed ones.
723    #[must_use]
724    pub fn live_session_count(&self) -> usize {
725        self.sessions.live_count()
726    }
727
728    /// Approximate retained state for the VM runtime.
729    #[must_use]
730    pub fn memory_usage(&self) -> VmMemoryUsage {
731        let session_store = self.sessions.memory_usage();
732        let retained_bytes = self.retained_bytes(session_store.retained_bytes.total);
733        VmMemoryUsage {
734            session_store,
735            coroutine_records: self.coroutines.len(),
736            terminal_coroutines: self.coroutines.iter().filter(|coro| coro.is_terminal()).count(),
737            program_count: self.programs.len(),
738            program_instruction_count: self.programs.instruction_count(),
739            obs_events: self.obs_trace.len(),
740            effect_trace_entries: self.effect_trace.len(),
741            communication_artifacts: self.communication_consumption_artifacts.len(),
742            output_condition_checks: self.output_condition_checks.len(),
743            retained_bytes,
744        }
745    }
746
747    fn retained_bytes(&self, session_store_bytes: usize) -> VmRetainedBytes {
748        let mut retained_bytes = VmRetainedBytes {
749            session_store: session_store_bytes,
750            coroutines: self.coroutines.iter().map(vm_serialized_bytes).sum(),
751            programs: vm_serialized_bytes(&self.programs)
752                .saturating_add(vm_serialized_bytes(&self.code)),
753            resource_states: vm_serialized_bytes(&self.resource_states),
754            traces: vm_serialized_bytes(&self.obs_trace)
755                .saturating_add(vm_serialized_bytes(&self.effect_trace)),
756            replay: vm_serialized_bytes(&self.communication_consumption)
757                .saturating_add(vm_serialized_bytes(&self.communication_consumption_artifacts)),
758            output_condition_checks: vm_serialized_bytes(&self.output_condition_checks),
759            scheduler_and_control: vm_serialized_bytes(&self.sched)
760                .saturating_add(vm_serialized_bytes(&self.eligible_ready))
761                .saturating_add(vm_serialized_bytes(&self.paused_roles))
762                .saturating_add(vm_serialized_bytes(&self.crashed_sites))
763                .saturating_add(vm_serialized_bytes(&self.partitioned_edges))
764                .saturating_add(vm_serialized_bytes(&self.corrupted_edges))
765                .saturating_add(vm_serialized_bytes(&self.timed_out_sites))
766                .saturating_add(vm_serialized_bytes(&self.clock))
767                .saturating_add(vm_serialized_bytes(&self.last_sched_step))
768                .saturating_add(vm_serialized_bytes(&self.handler_identity_anchor))
769                .saturating_add(vm_serialized_bytes(&self.next_coro_id))
770                .saturating_add(vm_serialized_bytes(&self.next_session_id)),
771            symbols: vm_serialized_bytes(&self.role_symbols)
772                .saturating_add(vm_serialized_bytes(&self.label_symbols))
773                .saturating_add(vm_serialized_bytes(&self.handler_symbols))
774                .saturating_add(vm_serialized_bytes(&self.edge_symbols)),
775            guard_layer: vm_serialized_bytes(&self.guard_layer),
776            monitor: vm_serialized_bytes(&self.monitor),
777            arena: vm_serialized_bytes(&self.arena),
778            total: 0,
779        };
780        retained_bytes.total = Self::retained_bytes_total(&retained_bytes);
781        retained_bytes
782    }
783
784    fn retained_bytes_total(retained_bytes: &VmRetainedBytes) -> usize {
785        retained_bytes
786            .session_store
787            .saturating_add(retained_bytes.coroutines)
788            .saturating_add(retained_bytes.programs)
789            .saturating_add(retained_bytes.resource_states)
790            .saturating_add(retained_bytes.traces)
791            .saturating_add(retained_bytes.replay)
792            .saturating_add(retained_bytes.output_condition_checks)
793            .saturating_add(retained_bytes.scheduler_and_control)
794            .saturating_add(retained_bytes.symbols)
795            .saturating_add(retained_bytes.guard_layer)
796            .saturating_add(retained_bytes.monitor)
797            .saturating_add(retained_bytes.arena)
798    }
799
800    /// Get recorded output-condition verification checks.
801    #[must_use]
802    pub fn output_condition_checks(&self) -> &[OutputConditionCheck] {
803        self.output_condition_checks.as_slice()
804    }
805
806    /// Get recorded effect-trace entries.
807    #[must_use]
808    pub fn effect_trace(&self) -> &[EffectTraceEntry] {
809        self.effect_trace.as_slice()
810    }
811
812    /// Deterministic communication replay-state root.
813    #[must_use]
814    pub fn communication_replay_root(&self) -> crate::verification::Hash {
815        self.communication_consumption.root()
816    }
817
818    /// Receive-boundary replay-consumption artifacts.
819    #[must_use]
820    pub fn communication_consumption_artifacts(&self) -> &[CommunicationConsumptionArtifact] {
821        self.communication_consumption_artifacts.as_slice()
822    }
823
824    /// Drain retained observable events in canonical insertion order.
825    pub fn drain_obs_trace(&mut self) -> Vec<ObsEvent> {
826        self.obs_trace.drain()
827    }
828
829    /// Drain retained effect-trace entries in canonical insertion order.
830    pub fn drain_effect_trace(&mut self) -> Vec<EffectTraceEntry> {
831        self.effect_trace.drain()
832    }
833
834    /// Drain retained output-condition diagnostics in canonical insertion order.
835    pub fn drain_output_condition_checks(&mut self) -> Vec<OutputConditionCheck> {
836        self.output_condition_checks.drain()
837    }
838
839    /// Drain retained communication replay-consumption artifacts in canonical insertion order.
840    pub fn drain_communication_consumption_artifacts(
841        &mut self,
842    ) -> Vec<CommunicationConsumptionArtifact> {
843        self.communication_consumption_artifacts.drain()
844    }
845
846    /// Canonical replay/state fragment for deterministic diffing and snapshots.
847    #[must_use]
848    pub fn canonical_replay_fragment(&self) -> CanonicalReplayFragmentV1 {
849        let partitioned_edges = self.partitioned_edges.iter().cloned().collect();
850        let corrupted_edges = self
851            .corrupted_edges
852            .iter()
853            .map(|(edge, corruption)| (edge.clone(), *corruption))
854            .collect();
855        let timed_out_sites = self
856            .timed_out_sites
857            .iter()
858            .map(|(site, until_tick)| (site.clone(), *until_tick))
859            .collect();
860        canonical_replay_fragment_v1(
861            self.obs_trace.as_slice(),
862            self.effect_trace.as_slice(),
863            self.crashed_sites.iter().cloned().collect(),
864            partitioned_edges,
865            corrupted_edges,
866            timed_out_sites,
867            self.config.effect_determinism_tier,
868            self.config.communication_replay_mode,
869            Some(self.communication_consumption.root()),
870            self.communication_consumption_artifacts.as_slice().to_vec(),
871        )
872    }
873
874    /// Crashed sites currently active in topology state.
875    #[must_use]
876    pub fn crashed_sites(&self) -> &BTreeSet<SiteId> {
877        &self.crashed_sites
878    }
879
880    /// Partitioned site-links currently active in topology state.
881    #[must_use]
882    pub fn partitioned_edges(&self) -> &BTreeSet<(SiteId, SiteId)> {
883        &self.partitioned_edges
884    }
885
886    /// Corrupted directed edges currently active in topology state.
887    #[must_use]
888    pub fn corrupted_edges(&self) -> &BTreeMap<(SiteId, SiteId), CorruptionType> {
889        &self.corrupted_edges
890    }
891
892    /// Active site timeouts.
893    #[must_use]
894    pub fn timed_out_sites(&self) -> &BTreeMap<SiteId, u64> {
895        &self.timed_out_sites
896    }
897}
telltale_vm/vm/runtime_and_execution.rs

telltale_vm/vm/
runtime_and_execution.rs
