hara_native/instrumentation/
interpreter.rs

1use std::collections::{BTreeSet, HashMap};
2
3use crate::core::{EvalFiber, EvalFiberState, PromiseState, Value};
4
5use super::{
6    Capability, ControlLease, DispatchReport, EventAccess, EventKind, EventLocation,
7    InstrumentDirective, InstrumentationError, InstrumentationHub, PortableProjection,
8    ProducerEvent, ProjectionLimits, TargetHandle, TargetKind,
9};
10
11const SEMANTIC_EVENTS: [EventKind; 6] = [
12    EventKind::SemanticBoundary,
13    EventKind::CallEnter,
14    EventKind::CallReturn,
15    EventKind::ExceptionRaise,
16    EventKind::VarSet,
17    EventKind::FieldSet,
18];
19
20/// Exact Rust interpreter capabilities backed by the production CPS fiber.
21pub fn interpreter_capabilities() -> BTreeSet<Capability> {
22    BTreeSet::from([
23        Capability::EventSemanticBoundary,
24        Capability::EventCall,
25        Capability::EventException,
26        Capability::EventEffect,
27        Capability::EventSuspension,
28        Capability::EventLifecycle,
29        Capability::InspectSourceLocation,
30        Capability::InspectCurrentFrame,
31        Capability::InspectFrames,
32        Capability::InspectLocals,
33        Capability::InspectValuePreview,
34        Capability::InspectSnapshot,
35        Capability::ControlPause,
36        Capability::ControlSingleStep,
37        Capability::ControlResume,
38        Capability::ControlSettle,
39        Capability::ControlTerminate,
40    ])
41}
42
43#[derive(Debug, Clone, PartialEq)]
44pub struct InterpreterBoundary {
45    pub state: EvalFiberState,
46    pub paused: bool,
47    pub reports: Vec<DispatchReport>,
48}
49
50/// One instrumentation target around the actual retained CPS continuation.
51/// It neither replays source nor owns a second evaluator.
52pub struct InterpreterTarget {
53    target: TargetHandle,
54    source_id: String,
55    fiber: EvalFiber,
56    semantic_sequence: usize,
57    paused: bool,
58    terminal_emitted: bool,
59}
60
61impl InterpreterTarget {
62    pub fn start(
63        hub: &InstrumentationHub,
64        target: TargetHandle,
65        source_id: impl Into<String>,
66        source: &str,
67        environment: HashMap<String, Value>,
68    ) -> Result<Self, InstrumentationError> {
69        let descriptor = hub.target_descriptor(&target)?;
70        if descriptor.kind != TargetKind::Interpreter {
71            return Err(InstrumentationError::InvalidTarget(
72                "interpreter probe requires an interpreter target".into(),
73            ));
74        }
75        let source_id = source_id.into();
76        if source_id.trim().is_empty() {
77            return Err(InstrumentationError::InvalidTarget(
78                "interpreter source id must be non-empty".into(),
79            ));
80        }
81        let fiber = EvalFiber::start_observed(source, environment)
82            .map_err(InstrumentationError::Execution)?;
83        fiber.configure_instrumentation_capture(false, false);
84        Ok(Self {
85            target,
86            source_id,
87            fiber,
88            semantic_sequence: 0,
89            paused: false,
90            terminal_emitted: false,
91        })
92    }
93
94    pub fn target(&self) -> &TargetHandle {
95        &self.target
96    }
97
98    pub fn state(&self) -> EvalFiberState {
99        self.fiber.state()
100    }
101
102    pub fn pending(&self) -> Option<crate::core::Promise> {
103        self.fiber.pending()
104    }
105
106    pub fn environment(&self) -> HashMap<String, Value> {
107        self.fiber.environment()
108    }
109
110    pub fn environment_clone_count(&self) -> u64 {
111        self.fiber.instrumentation_environment_clone_count()
112    }
113
114    pub fn paused(&self) -> bool {
115        self.paused
116    }
117
118    /// Executes at most one genuine evaluator or queued semantic boundary.
119    pub fn step(
120        &mut self,
121        hub: &mut InstrumentationHub,
122    ) -> Result<InterpreterBoundary, InstrumentationError> {
123        let directive = hub.take_directive(&self.target)?;
124        if self.paused && directive == InstrumentDirective::Continue {
125            return Ok(self.boundary(Vec::new()));
126        }
127        let pause_after = directive == InstrumentDirective::StepNext;
128        match directive {
129            InstrumentDirective::Suspend => {
130                self.paused = true;
131                return Ok(self.boundary(Vec::new()));
132            }
133            InstrumentDirective::Terminate => {
134                self.fiber.cancel();
135                self.paused = true;
136                let mut reports = Vec::new();
137                self.emit_terminal(hub, &mut reports)?;
138                return Ok(self.boundary(reports));
139            }
140            InstrumentDirective::Continue | InstrumentDirective::StepNext => {
141                self.paused = false;
142            }
143        }
144
145        self.refresh_capture(hub)?;
146        let before = self.fiber.state();
147        self.fiber.step_observed();
148        let after = self.fiber.state();
149        let mut reports = Vec::new();
150        self.emit_current_semantic(hub, &mut reports)?;
151        if !matches!(before, EvalFiberState::Suspended)
152            && matches!(after, EvalFiberState::Suspended)
153        {
154            self.emit(
155                hub,
156                ProducerEvent::live(EventKind::PromiseSuspend),
157                &mut reports,
158            )?;
159        }
160        self.emit_terminal(hub, &mut reports)?;
161        if pause_after {
162            self.paused = true;
163        }
164        Ok(self.boundary(reports))
165    }
166
167    /// Runs genuine boundaries until completion, suspension, an explicit pause,
168    /// or the caller's bound. This is a driver over the retained fiber, not a
169    /// replay loop.
170    pub fn run(
171        &mut self,
172        hub: &mut InstrumentationHub,
173        boundary_limit: usize,
174    ) -> Result<Vec<InterpreterBoundary>, InstrumentationError> {
175        let mut boundaries = Vec::new();
176        for _ in 0..boundary_limit {
177            if self.paused || self.finished() {
178                break;
179            }
180            let boundary = self.step(hub)?;
181            let suspended = matches!(boundary.state, EvalFiberState::Suspended);
182            boundaries.push(boundary);
183            if self.paused || suspended || self.finished() {
184                break;
185            }
186        }
187        Ok(boundaries)
188    }
189
190    pub fn continue_execution(
191        &mut self,
192        hub: &InstrumentationHub,
193        lease: &ControlLease,
194    ) -> Result<(), InstrumentationError> {
195        hub.authorize_control(lease, Capability::ControlResume)?;
196        if lease.target() != &self.target {
197            return Err(InstrumentationError::InvalidControlLease {
198                target_id: self.target.target_id().into(),
199                instrument_id: lease.instrument().instrument_id().into(),
200            });
201        }
202        self.paused = false;
203        Ok(())
204    }
205
206    /// Settles and resumes the exact Promise/continuation retained by EvalFiber.
207    pub fn settle(
208        &mut self,
209        hub: &mut InstrumentationHub,
210        lease: &ControlLease,
211        state: PromiseState,
212    ) -> Result<InterpreterBoundary, InstrumentationError> {
213        hub.authorize_control(lease, Capability::ControlSettle)?;
214        if lease.target() != &self.target {
215            return Err(InstrumentationError::InvalidControlLease {
216                target_id: self.target.target_id().into(),
217                instrument_id: lease.instrument().instrument_id().into(),
218            });
219        }
220        if !matches!(self.fiber.state(), EvalFiberState::Suspended) {
221            return Err(InstrumentationError::Execution(
222                "interpreter target is not suspended".into(),
223            ));
224        }
225        if matches!(state, PromiseState::Pending) {
226            return Err(InstrumentationError::Execution(
227                "interpreter settlement cannot remain pending".into(),
228            ));
229        }
230        self.refresh_capture(hub)?;
231        self.paused = false;
232        self.fiber.resume_observed(state);
233        let mut reports = Vec::new();
234        self.emit(
235            hub,
236            ProducerEvent::live(EventKind::PromiseResume),
237            &mut reports,
238        )?;
239        self.emit_current_semantic(hub, &mut reports)?;
240        self.emit_terminal(hub, &mut reports)?;
241        Ok(self.boundary(reports))
242    }
243
244    fn refresh_capture(&self, hub: &InstrumentationHub) -> Result<(), InstrumentationError> {
245        let mut capture_events = false;
246        let mut capture_environment = false;
247        for event in SEMANTIC_EVENTS {
248            if hub.enabled_for_target(&self.target, event)? {
249                capture_events = true;
250                capture_environment |= hub
251                    .requested_projection(&self.target, event)?
252                    .needs_interpreter_environment();
253            }
254        }
255        self.fiber
256            .configure_instrumentation_capture(capture_events, capture_environment);
257        Ok(())
258    }
259
260    fn emit_current_semantic(
261        &mut self,
262        hub: &mut InstrumentationHub,
263        reports: &mut Vec<DispatchReport>,
264    ) -> Result<(), InstrumentationError> {
265        let Some((sequence, event)) = self.fiber.instrumentation_event() else {
266            return Ok(());
267        };
268        if sequence <= self.semantic_sequence {
269            return Ok(());
270        }
271        self.semantic_sequence = sequence;
272        self.emit(hub, event, reports)
273    }
274
275    fn emit_terminal(
276        &mut self,
277        hub: &mut InstrumentationHub,
278        reports: &mut Vec<DispatchReport>,
279    ) -> Result<(), InstrumentationError> {
280        if self.terminal_emitted || self.fiber.observed_pending_boundaries() > 0 {
281            return Ok(());
282        }
283        let state = self.fiber.state();
284        let status = match &state {
285            EvalFiberState::Completed(_) => "returned",
286            EvalFiberState::Failed(_) => "failed",
287            EvalFiberState::Cancelled => "cancelled",
288            EvalFiberState::Running | EvalFiberState::Suspended => return Ok(()),
289        };
290        let mut event =
291            ProducerEvent::live(EventKind::ExecutionTerminal).with_data("status", status);
292        match state {
293            EvalFiberState::Completed(value) => {
294                event = event.with_data("result/type", crate::core::portable_type_name(&value));
295            }
296            EvalFiberState::Failed(error) => {
297                event = event.with_data("error", bounded_text(&error, 1_024));
298            }
299            EvalFiberState::Cancelled | EvalFiberState::Running | EvalFiberState::Suspended => {}
300        }
301        self.terminal_emitted = true;
302        self.emit(hub, event, reports)
303    }
304
305    fn emit(
306        &self,
307        hub: &mut InstrumentationHub,
308        event: ProducerEvent,
309        reports: &mut Vec<DispatchReport>,
310    ) -> Result<(), InstrumentationError> {
311        let mut access = InterpreterAccess {
312            fiber: &self.fiber,
313            source_id: &self.source_id,
314        };
315        reports.push(hub.emit(&self.target, event, &mut access)?);
316        Ok(())
317    }
318
319    fn finished(&self) -> bool {
320        matches!(
321            self.fiber.state(),
322            EvalFiberState::Completed(_) | EvalFiberState::Failed(_) | EvalFiberState::Cancelled
323        ) && self.fiber.observed_pending_boundaries() == 0
324    }
325
326    fn boundary(&self, reports: Vec<DispatchReport>) -> InterpreterBoundary {
327        InterpreterBoundary {
328            state: self.fiber.state(),
329            paused: self.paused,
330            reports,
331        }
332    }
333}
334
335struct InterpreterAccess<'a> {
336    fiber: &'a EvalFiber,
337    source_id: &'a str,
338}
339
340impl EventAccess for InterpreterAccess<'_> {
341    fn source_location(&mut self) -> Option<EventLocation> {
342        self.fiber.instrumentation_source_location(self.source_id)
343    }
344
345    fn current_frame(&mut self, limits: ProjectionLimits) -> Option<PortableProjection> {
346        self.fiber.instrumentation_current_frame(limits)
347    }
348
349    fn frames(&mut self, limits: ProjectionLimits) -> Option<PortableProjection> {
350        self.fiber.instrumentation_frames(limits)
351    }
352
353    fn locals(&mut self, limits: ProjectionLimits) -> Option<PortableProjection> {
354        self.fiber.instrumentation_locals(limits)
355    }
356
357    fn value_preview(&mut self, limits: ProjectionLimits) -> Option<PortableProjection> {
358        self.fiber.instrumentation_value_preview(limits)
359    }
360
361    fn machine_snapshot(&mut self, limits: ProjectionLimits) -> Option<PortableProjection> {
362        self.fiber.instrumentation_snapshot(limits)
363    }
364}
365
366fn bounded_text(value: &str, limit: usize) -> String {
367    if value.chars().count() <= limit {
368        return value.into();
369    }
370    let mut output = value.chars().take(limit).collect::<String>();
371    output.push('…');
372    output
373}
374
375#[cfg(test)]
376#[path = "interpreter/tests.rs"]
377mod tests;
hara_native/instrumentation/interpreter.rs

hara_native/instrumentation/
interpreter.rs
