heartbit_core/agent/flow/
event.rs

1//! Workflow-level events and the `phase()` / `log()` observability helpers.
2
3use std::sync::Arc;
4
5use serde::{Deserialize, Serialize};
6
7use crate::llm::types::TokenUsage;
8
9use super::ctx::WorkflowCtx;
10
11/// Events emitted by the workflow combinators, mirroring the per-agent
12/// [`AgentEvent`](crate::AgentEvent) plane at the workflow level.
13///
14/// This is a **separate** enum from `AgentEvent` (rather than new variants on
15/// it) because `AgentEvent` is not `#[non_exhaustive]` and has an exhaustive
16/// `type_name()` match plus external matchers — extending it would be a
17/// breaking change. `WorkflowEvent` *is* `#[non_exhaustive]` so later phases
18/// can add variants without breaking downstream matchers.
19#[non_exhaustive]
20#[derive(Debug, Clone, Serialize, Deserialize)]
21#[serde(tag = "type", rename_all = "snake_case")]
22pub enum WorkflowEvent {
23    /// A new phase began; subsequently-issued agents group under `title`.
24    PhaseStarted {
25        /// Phase title.
26        title: String,
27    },
28    /// An agent leaf began executing.
29    AgentStarted {
30        /// Agent label (defaults to a generated name when unset).
31        label: String,
32        /// The phase this agent was issued under, if any.
33        phase: Option<String>,
34    },
35    /// An agent leaf completed successfully.
36    AgentFinished {
37        /// Agent label.
38        label: String,
39        /// Token usage reported by the underlying runner.
40        usage: TokenUsage,
41    },
42    /// An agent leaf's result was replayed from the resume journal (no model
43    /// call). `usage` is the originally-recorded usage, for progress totals.
44    AgentReplayed {
45        /// Agent label.
46        label: String,
47        /// Token usage recorded when this output was first produced.
48        usage: TokenUsage,
49    },
50    /// An agent leaf was skipped (e.g. cancelled mid-run); the combinator
51    /// surfaces this as `Ok(None)`.
52    AgentSkipped {
53        /// Agent label.
54        label: String,
55    },
56    /// An agent-domain failure inside a combinator; the slot collapses to
57    /// `None` (the combinator never rejects on an agent-domain error).
58    AgentFailed {
59        /// Agent label.
60        label: String,
61        /// Display string of the underlying error.
62        error: String,
63    },
64    /// A free-form progress line emitted via [`log`].
65    LogLine {
66        /// Message text.
67        msg: String,
68    },
69}
70
71/// Callback type for receiving [`WorkflowEvent`]s. Kept object-safe and
72/// `Send + Sync` so it can be shared across concurrently-running agents.
73pub type OnWorkflowEvent = dyn Fn(WorkflowEvent) + Send + Sync;
74
75/// Begin a new phase. Emits [`WorkflowEvent::PhaseStarted`] and sets the
76/// default phase for subsequently-issued agents. The returned [`PhaseGuard`]
77/// restores the prior default phase when dropped, so phases nest correctly.
78///
79/// Note: the phase is the default for agents *issued after* this call; each
80/// [`AgentCall`](super::agent::AgentCall) snapshots the default at construction,
81/// so concurrently-running agents never observe a torn phase.
82#[must_use = "the phase ends when the returned guard is dropped"]
83pub fn phase(ctx: &WorkflowCtx, title: impl Into<String>) -> PhaseGuard {
84    let title = title.into();
85    ctx.emit(WorkflowEvent::PhaseStarted {
86        title: title.clone(),
87    });
88    let prior = ctx.swap_default_phase(Some(Arc::from(title.as_str())));
89    PhaseGuard {
90        ctx: ctx.clone(),
91        prior,
92    }
93}
94
95/// Emit a free-form progress line ([`WorkflowEvent::LogLine`]).
96pub fn log(ctx: &WorkflowCtx, msg: impl Into<String>) {
97    ctx.emit(WorkflowEvent::LogLine { msg: msg.into() });
98}
99
100/// RAII guard returned by [`phase`]; restores the prior default phase on drop.
101pub struct PhaseGuard {
102    ctx: WorkflowCtx,
103    prior: Option<Arc<str>>,
104}
105
106impl Drop for PhaseGuard {
107    fn drop(&mut self) {
108        self.ctx.swap_default_phase(self.prior.take());
109    }
110}
111
112#[cfg(test)]
113mod tests {
114    use super::WorkflowEvent;
115
116    #[test]
117    fn workflow_event_serializes_with_snake_case_type_tag() {
118        let event = WorkflowEvent::PhaseStarted {
119            title: "review".into(),
120        };
121        let json = serde_json::to_string(&event).expect("serialize");
122        assert!(json.contains(r#""type":"phase_started""#), "json: {json}");
123        assert!(json.contains(r#""title":"review""#), "json: {json}");
124    }
125
126    #[test]
127    fn log_line_event_roundtrips_through_json() {
128        let event = WorkflowEvent::LogLine {
129            msg: "3/10 found".into(),
130        };
131        let json = serde_json::to_string(&event).expect("serialize");
132        let back: WorkflowEvent = serde_json::from_str(&json).expect("deserialize");
133        match back {
134            WorkflowEvent::LogLine { msg } => assert_eq!(msg, "3/10 found"),
135            other => panic!("expected LogLine, got {other:?}"),
136        }
137    }
138
139    // -----------------------------------------------------------------------
140    // phase() / log() observability helpers
141    // -----------------------------------------------------------------------
142
143    use std::sync::{Arc, Mutex};
144
145    use super::super::ctx::WorkflowCtx;
146    use super::{OnWorkflowEvent, log, phase};
147    use crate::agent::test_helpers::MockProvider;
148    use crate::llm::BoxedProvider;
149
150    /// Build a ctx whose events are captured into the returned `Vec`.
151    fn ctx_capturing() -> (WorkflowCtx, Arc<Mutex<Vec<WorkflowEvent>>>) {
152        let sink: Arc<Mutex<Vec<WorkflowEvent>>> = Arc::new(Mutex::new(Vec::new()));
153        let sink2 = Arc::clone(&sink);
154        let cb: Arc<OnWorkflowEvent> = Arc::new(move |ev: WorkflowEvent| {
155            sink2.lock().expect("sink lock").push(ev);
156        });
157        let ctx = WorkflowCtx::builder(Arc::new(BoxedProvider::new(MockProvider::new(vec![]))))
158            .on_event(cb)
159            .build()
160            .expect("build ctx");
161        (ctx, sink)
162    }
163
164    #[test]
165    fn phase_emits_started_and_sets_default_phase() {
166        let (ctx, sink) = ctx_capturing();
167        assert!(ctx.current_phase().is_none());
168
169        let _guard = phase(&ctx, "review");
170        assert_eq!(ctx.current_phase().as_deref(), Some("review"));
171
172        let events = sink.lock().expect("lock");
173        assert!(
174            matches!(events.first(), Some(WorkflowEvent::PhaseStarted { title }) if title == "review"),
175            "events: {events:?}"
176        );
177    }
178
179    #[test]
180    fn phase_guard_restores_prior_default_on_drop() {
181        let (ctx, _sink) = ctx_capturing();
182        {
183            let _outer = phase(&ctx, "outer");
184            assert_eq!(ctx.current_phase().as_deref(), Some("outer"));
185            {
186                let _inner = phase(&ctx, "inner");
187                assert_eq!(ctx.current_phase().as_deref(), Some("inner"));
188            }
189            // inner dropped -> restored to outer
190            assert_eq!(ctx.current_phase().as_deref(), Some("outer"));
191        }
192        // outer dropped -> restored to None
193        assert!(ctx.current_phase().is_none());
194    }
195
196    #[test]
197    fn log_emits_log_line() {
198        let (ctx, sink) = ctx_capturing();
199        log(&ctx, "halfway there");
200        let events = sink.lock().expect("lock");
201        assert!(
202            matches!(events.last(), Some(WorkflowEvent::LogLine { msg }) if msg == "halfway there"),
203            "events: {events:?}"
204        );
205    }
206}
heartbit_core/agent/flow/event.rs

heartbit_core/agent/flow/
event.rs
