logicaffeine_compile/
observe.rs

1//! The observation channel's emission buffer. When an observation statement (`Note`/`Warn`/`Record`) executes, it
2//! emits a structured [`ObsEvent`] to a per-run thread-local buffer, drained by
3//! sinks (the in-memory test sink today; console and wire in Wave 16). This is
4//! the SEPARATE observation channel (D0.1) — it never touches program output.
5//!
6//! Emission is wait-free from the program's perspective (a buffer push) and
7//! never draws from the scheduler's entropy, so a run's `SchedTrace` is
8//! identical with observation on or off.
9
10use serde::{Deserialize, Serialize};
11use std::cell::RefCell;
12
13/// The speech-act category of a message event (mirrors the language AST's
14/// `ObsCategory`, kept local so sinks need no AST dependency).
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "snake_case")]
17pub enum ObsCat {
18    Note,
19    Warn,
20    Record,
21}
22
23/// What an emitted event IS: a speech-act message, or a trace-span boundary.
24/// Spans are lexical (`During`),
25/// so every `SpanEnter` is matched by exactly one `SpanExit`.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(rename_all = "snake_case")]
28pub enum EventKind {
29    Message(ObsCat),
30    SpanEnter,
31    SpanExit,
32}
33
34/// One emitted observation event.
35#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36pub struct ObsEvent {
37    pub kind: EventKind,
38    /// For a message, the that-clause; for a span, the span label.
39    pub message: String,
40    /// Captured payloads: (name, display value). Empty for spans.
41    pub payloads: Vec<(String, String)>,
42    /// Per-run monotone sequence number (the emission order).
43    pub seq: u64,
44}
45
46thread_local! {
47    /// The active run's event buffer. `None` = no capture installed (emission
48    /// is a cheap no-op — the zero-cost-when-nobody-listens path).
49    static BUFFER: RefCell<Option<Vec<ObsEvent>>> = const { RefCell::new(None) };
50    /// Per-statement sampling counters, keyed by a statement id, so
51    /// `once in every N times` is deterministic and per-site.
52    static SAMPLE_COUNTERS: RefCell<std::collections::HashMap<u64, u64>> =
53        RefCell::new(std::collections::HashMap::new());
54}
55
56/// Begin capturing events for a run (the `largo test` runner, per test).
57pub fn install() {
58    BUFFER.with(|b| *b.borrow_mut() = Some(Vec::new()));
59    SAMPLE_COUNTERS.with(|c| c.borrow_mut().clear());
60}
61
62/// Take the captured events and stop capturing.
63pub fn drain() -> Vec<ObsEvent> {
64    let events = BUFFER.with(|b| b.borrow_mut().take()).unwrap_or_default();
65    SAMPLE_COUNTERS.with(|c| c.borrow_mut().clear());
66    events
67}
68
69/// Emit an event. Honors per-site sampling: with `sample = Some(n)` and a
70/// stable `site` id, only 1 in every `n` reaching this site is buffered (the
71/// first, then every nth). Sampling is a rendering decision — replay with
72/// capture-everything regenerates the full stream.
73pub fn emit(
74    category: ObsCat,
75    message: String,
76    payloads: Vec<(String, String)>,
77    sample: Option<u64>,
78    site: u64,
79) {
80    if let Some(n) = sample {
81        let keep = SAMPLE_COUNTERS.with(|c| {
82            let mut m = c.borrow_mut();
83            let count = m.entry(site).or_insert(0);
84            let keep = *count % n.max(1) == 0;
85            *count += 1;
86            keep
87        });
88        if !keep {
89            return;
90        }
91    }
92    push(EventKind::Message(category), message, payloads);
93}
94
95/// Emit a `SpanEnter` for a `During` block (Wave 8).
96pub fn enter_span(label: String) {
97    push(EventKind::SpanEnter, label, Vec::new());
98}
99
100/// Emit a `SpanExit` — always paired with its `enter_span` (the `During` arm
101/// emits it on every exit path, so spans stay balanced).
102pub fn exit_span(label: String) {
103    push(EventKind::SpanExit, label, Vec::new());
104}
105
106fn push(kind: EventKind, message: String, payloads: Vec<(String, String)>) {
107    BUFFER.with(|b| {
108        if let Some(buf) = b.borrow_mut().as_mut() {
109            let seq = buf.len() as u64;
110            buf.push(ObsEvent {
111                kind,
112                message,
113                payloads,
114                seq,
115            });
116        }
117    });
118}
119
120/// True when a capture buffer is installed (lets emitters skip work otherwise).
121pub fn capturing() -> bool {
122    BUFFER.with(|b| b.borrow().is_some())
123}
124
125/// Serialize an event stream to the `.logosobs` wire form — one JSON object
126/// per line, so it appends cheaply and greps line-by-line
127/// (columnar batching deferred).
128pub fn to_logosobs(events: &[ObsEvent]) -> String {
129    events
130        .iter()
131        .map(|e| serde_json::to_string(e).expect("an ObsEvent always serializes"))
132        .collect::<Vec<_>>()
133        .join("\n")
134}
135
136/// Parse a `.logosobs` stream back to events (the `largo trace` reader).
137pub fn from_logosobs(text: &str) -> Result<Vec<ObsEvent>, String> {
138    text.lines()
139        .filter(|l| !l.trim().is_empty())
140        .map(|l| serde_json::from_str(l).map_err(|e| format!("bad .logosobs line: {e}")))
141        .collect()
142}
logicaffeine_compile/observe.rs

logicaffeine_compile/
observe.rs
