harn_vm/run_events.rs
1//! Run-event sink: an execution-scoped bus the CLI attaches to capture every
2//! observable side effect of a `harn run` invocation as a single
3//! ordered stream.
4//!
5//! Concrete sinks live in `harn-cli` (`harn run --json` writes them as
6//! NDJSON). The VM only knows it should call [`emit`] from a handful of
7//! observability checkpoints; sinks fan-out from there.
8//!
9//! Variants intentionally mirror the surface area of the run command
10//! rather than the on-disk event log. Stdout/stderr writes are captured
11//! here because they never enter the event log; transcript / persona /
12//! hook / tool events are forwarded here in addition to their existing
13//! persistent topics so a single subscriber can see the whole run
14//! without joining across topics.
15
16use std::cell::RefCell;
17use std::future::Future;
18use std::sync::Arc;
19
20use serde::Serialize;
21
22/// One observable event from a running pipeline. Variants are
23/// `#[serde(tag = "event_type")]` so wire consumers (notably the CLI
24/// `--json` NDJSON stream) can discriminate without inspecting the
25/// payload shape.
26#[derive(Clone, Debug, Serialize)]
27#[serde(tag = "event_type", rename_all = "snake_case")]
28pub enum RunEvent {
29 /// Bytes written to stdout (raw, including any trailing newlines).
30 Stdout { payload: String },
31 /// Bytes written to stderr (raw, including any trailing newlines).
32 Stderr { payload: String },
33 /// One append on a transcript stream (`agent.transcript.llm` topic).
34 /// `kind` mirrors the transcript entry's `type` field.
35 Transcript {
36 #[serde(skip_serializing_if = "Option::is_none")]
37 agent_id: Option<String>,
38 kind: String,
39 payload: serde_json::Value,
40 },
41 /// A model-issued tool call. `call_id` matches the transcript
42 /// `call_id`; agents reconcile [`Self::ToolCall`] /
43 /// [`Self::ToolResult`] pairs by it.
44 ToolCall {
45 call_id: String,
46 name: String,
47 args: serde_json::Value,
48 /// RFC 3339 timestamp captured at emission.
49 started_at: String,
50 },
51 /// Outcome of a tool call.
52 ToolResult {
53 call_id: String,
54 ok: bool,
55 result: serde_json::Value,
56 },
57 /// A workflow hook fired during the run.
58 Hook {
59 name: String,
60 phase: String,
61 #[serde(skip_serializing_if = "serde_json::Value::is_null")]
62 payload: serde_json::Value,
63 },
64 /// A persona-stage transition. Mirrors the `persona.runtime.events`
65 /// topic; the `transition` field captures the stage state change
66 /// (`"started"`, `"completed"`, `"handoff"`, ...).
67 PersonaStage {
68 persona: String,
69 stage: String,
70 transition: String,
71 },
72 /// `harn run <bundle.harnpack>` resolved a pack to execute. Carries
73 /// the verified bundle hash, whether the embedded Ed25519 signature
74 /// verified end-to-end, the signing key fingerprint (when signed),
75 /// and whether the unpacked archive came from the content-addressed
76 /// cache or was extracted fresh on this run.
77 PackRun {
78 bundle_hash: String,
79 signature_verified: bool,
80 #[serde(skip_serializing_if = "Option::is_none")]
81 key_id: Option<String>,
82 cache_hit: bool,
83 dry_run_verify: bool,
84 },
85}
86
87/// Receiver of [`RunEvent`]s. Implementations must be cheap (the VM
88/// calls [`emit`] on hot paths like every `println`).
89pub trait RunEventSink: Send + Sync {
90 fn emit(&self, event: RunEvent);
91}
92
93thread_local! {
94 /// The sink for the execution currently being polled on this thread.
95 ///
96 /// `AmbientExecutionScope` swaps this slot at every task poll, so the
97 /// thread-local is execution-owned even when unrelated futures interleave
98 /// on one runtime thread.
99 static RUN_EVENT_SINK_CONTEXT: RefCell<Option<Arc<dyn RunEventSink>>> =
100 RefCell::new(None);
101}
102
103pub(crate) fn swap_run_event_sink(
104 sink: Option<Arc<dyn RunEventSink>>,
105) -> Option<Arc<dyn RunEventSink>> {
106 RUN_EVENT_SINK_CONTEXT.with(|slot| std::mem::replace(&mut *slot.borrow_mut(), sink))
107}
108
109/// Run `inner` with `sink` attached to its ambient execution scope.
110///
111/// The scope is installed only while `inner` is being polled. Inline and
112/// spawned VM work that captures the ambient execution inherits the sink;
113/// unrelated or sibling executions do not.
114pub fn scope<F: Future>(sink: Arc<dyn RunEventSink>, inner: F) -> impl Future<Output = F::Output> {
115 crate::orchestration::scope_run_event_sink(sink, inner)
116}
117
118/// Whether the current execution has a sink. Useful as a fast-path gate
119/// for callers that would otherwise build a payload speculatively.
120pub fn sink_active() -> bool {
121 RUN_EVENT_SINK_CONTEXT.with(|slot| slot.borrow().is_some())
122}
123
124/// Emit `event` to the current execution's sink. No-op when no sink is active,
125/// so it is safe to call from every hook point unconditionally.
126pub fn emit(event: RunEvent) {
127 let sink = RUN_EVENT_SINK_CONTEXT.with(|slot| slot.borrow().clone());
128 if let Some(sink) = sink {
129 sink.emit(redact_run_event(event));
130 }
131}
132
133/// Scrub the secret-bearing JSON field of a `RunEvent` once, centrally, before
134/// it reaches any sink — so emitters can't forget (the `Hook` payload did, while
135/// the transcript/tool variants were only clean because `agent_observe`
136/// pre-scrubbed them). A second idempotent pass over a pre-scrubbed payload is
137/// cheap and makes the bus correct-by-construction for every future variant.
138/// `Stdout`/`Stderr` are the program's own raw output and pass through
139/// unredacted; they also take the fast path (no policy lookup) since `emit` runs
140/// on every `println`.
141fn redact_run_event(mut event: RunEvent) -> RunEvent {
142 let payload = match &mut event {
143 RunEvent::Transcript { payload, .. } => payload,
144 RunEvent::ToolCall { args, .. } => args,
145 RunEvent::ToolResult { result, .. } => result,
146 RunEvent::Hook { payload, .. } => payload,
147 RunEvent::Stdout { .. }
148 | RunEvent::Stderr { .. }
149 | RunEvent::PersonaStage { .. }
150 | RunEvent::PackRun { .. } => return event,
151 };
152 crate::redact::current_policy().redact_json_in_place(payload);
153 event
154}
155
156#[cfg(test)]
157mod tests {
158 use super::*;
159 use std::sync::Mutex;
160
161 struct CapturingSink {
162 events: Mutex<Vec<RunEvent>>,
163 }
164
165 impl RunEventSink for CapturingSink {
166 fn emit(&self, event: RunEvent) {
167 self.events.lock().unwrap().push(event);
168 }
169 }
170
171 #[tokio::test]
172 async fn nested_scopes_restore_the_outer_sink() {
173 let outer = Arc::new(CapturingSink {
174 events: Mutex::new(Vec::new()),
175 });
176 let inner = Arc::new(CapturingSink {
177 events: Mutex::new(Vec::new()),
178 });
179
180 scope(outer.clone(), async {
181 emit(RunEvent::Stdout {
182 payload: "outer-before\n".into(),
183 });
184 crate::orchestration::scope_inline_subtask(async {
185 emit(RunEvent::Stdout {
186 payload: "outer-inline\n".into(),
187 });
188 })
189 .await;
190 let inherited = crate::orchestration::AmbientExecutionScope::capture_inherited();
191 crate::orchestration::scope_ambient(inherited, async {
192 emit(RunEvent::Stdout {
193 payload: "outer-worker\n".into(),
194 });
195 })
196 .await;
197 scope(inner.clone(), async {
198 emit(RunEvent::Stdout {
199 payload: "inner\n".into(),
200 });
201 })
202 .await;
203 emit(RunEvent::Stdout {
204 payload: "outer-after\n".into(),
205 });
206 })
207 .await;
208
209 assert!(!sink_active());
210 emit(RunEvent::Stdout {
211 payload: "unscoped\n".into(),
212 });
213
214 let payloads = |sink: &CapturingSink| {
215 sink.events
216 .lock()
217 .unwrap()
218 .iter()
219 .map(|event| match event {
220 RunEvent::Stdout { payload } => payload.clone(),
221 other => panic!("unexpected event {other:?}"),
222 })
223 .collect::<Vec<_>>()
224 };
225 assert_eq!(
226 payloads(&outer),
227 [
228 "outer-before\n",
229 "outer-inline\n",
230 "outer-worker\n",
231 "outer-after\n"
232 ]
233 );
234 assert_eq!(payloads(&inner), ["inner\n"]);
235 }
236}