leviath_runtime/host/events.rs
1//! What the host broadcasts as the world changes.
2//!
3//! Two sources feed one stream. The coarse per-run variants come from the
4//! host's change-detection pass, which compares each run against the [`Emitted`]
5//! snapshot it kept from the previous cycle; the fine-grained ones are pushed at
6//! the source by pipeline systems through [`WorldEventSink`]. Kept beside the
7//! snapshot type rather than in the host, because the two only make sense
8//! together: the snapshot exists to decide what is worth emitting.
9
10use serde::{Deserialize, Serialize};
11use tokio::sync::broadcast;
12
13use crate::components::AgentStatus;
14use leviath_core::interaction::InteractionRequest;
15
16/// A change in the world, broadcast to subscribers (the HTTP/WS gateway and
17/// in-process embedders) so they get pushed updates instead of polling. The
18/// coarse per-run variants (`Spawned`/`Status`/`Tokens`/`Context`/`Completed`)
19/// are emitted by the host's change-detection pass as it drives the world;
20/// `StageTransition`/`ToolCallStarted`/`ToolCallFinished`/`Log` are pushed at
21/// the source by pipeline systems through [`WorldEventSink`]. Streamed over the
22/// control transport via `ControlRequest::Subscribe`.
23///
24/// Marked non-exhaustive: new variants are additive, so consumers outside this
25/// crate must keep a catch-all arm.
26#[non_exhaustive]
27#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
28#[serde(tag = "event", rename_all = "snake_case")]
29pub enum WorldEvent {
30 /// A run first appeared in the world.
31 Spawned {
32 /// The run id.
33 run_id: String,
34 /// The agent id.
35 agent_id: String,
36 /// The blueprint / agent name.
37 blueprint: String,
38 },
39 /// A run's status, stage, iteration, or tool-call count changed.
40 Status {
41 /// The run id.
42 run_id: String,
43 /// The agent id.
44 agent_id: String,
45 /// Short status label (`active`, `waiting`, `complete`, …).
46 status: String,
47 /// The current stage name.
48 stage: String,
49 /// The current iteration.
50 iteration: usize,
51 /// Cumulative tool calls.
52 tool_calls: usize,
53 /// Whether the current stage accepts messages.
54 accepts_messages: bool,
55 /// Why the run is parked, when it is.
56 ///
57 /// A subscriber watching live otherwise sees a run turn `waiting` or
58 /// `paused` and has to fetch the run to learn whether that means "go
59 /// and answer something" or "its workers are still going" - which is
60 /// the guess this vocabulary exists to remove.
61 wait_reason: Option<leviath_core::run_meta::WaitReason>,
62 },
63 /// A run's token totals changed.
64 Tokens {
65 /// The run id.
66 run_id: String,
67 /// The agent id.
68 agent_id: String,
69 /// Cumulative prompt tokens.
70 prompt_tokens: usize,
71 /// Cumulative completion tokens.
72 completion_tokens: usize,
73 /// Cumulative cached tokens.
74 cached_tokens: usize,
75 /// Cumulative cache-write tokens.
76 cache_write_tokens: usize,
77 },
78 /// A run's context-window token usage changed.
79 Context {
80 /// The run id.
81 run_id: String,
82 /// The agent id.
83 agent_id: String,
84 /// Current context tokens.
85 total_tokens: usize,
86 /// Max context tokens.
87 max_tokens: usize,
88 },
89 /// A run raised a new interaction awaiting an answer.
90 Interaction {
91 /// The run id.
92 run_id: String,
93 /// The agent id.
94 agent_id: String,
95 /// The interaction request.
96 request: InteractionRequest,
97 },
98 /// A run reached a terminal status.
99 Completed {
100 /// The run id.
101 run_id: String,
102 /// The agent id.
103 agent_id: String,
104 /// The terminal status label.
105 status: String,
106 /// What the run handed back, when it submitted anything.
107 ///
108 /// Carried on the event rather than left for the consumer to read off
109 /// disk: this fires the moment the run goes terminal, and the persist
110 /// tick that writes `meta.json` has not necessarily run yet. A webhook
111 /// or websocket consumer reading the file would race it and report a
112 /// finished run with no answer.
113 #[serde(default, skip_serializing_if = "Option::is_none")]
114 final_output: Option<leviath_core::output::FinalOutput>,
115 },
116 /// A run moved from one stage to another. Emitted by the transition systems
117 /// at the moment the new stage is entered (the initial stage at spawn is
118 /// covered by [`WorldEvent::Spawned`], not by this).
119 StageTransition {
120 /// The run id.
121 run_id: String,
122 /// The agent id.
123 agent_id: String,
124 /// The stage being left.
125 from: String,
126 /// The stage being entered.
127 to: String,
128 /// How many times the destination stage has been entered, this entry
129 /// included.
130 iteration: usize,
131 },
132 /// A tool call was handed to the async tool lane for execution. Inline
133 /// calls (context tools, refusals, gate blocks) resolve without touching
134 /// the lane and don't produce this event.
135 ToolCallStarted {
136 /// The run id.
137 run_id: String,
138 /// The agent id.
139 agent_id: String,
140 /// The provider-assigned tool call id.
141 call_id: String,
142 /// The tool name.
143 tool: String,
144 },
145 /// A lane-executed tool call returned. Paired with
146 /// [`WorldEvent::ToolCallStarted`] by `call_id`.
147 ToolCallFinished {
148 /// The run id.
149 run_id: String,
150 /// The agent id.
151 agent_id: String,
152 /// The provider-assigned tool call id.
153 call_id: String,
154 /// The tool name.
155 tool: String,
156 /// Whether the call took effect (`false` for `[error]`/`[blocked]`/
157 /// `[unavailable]` results).
158 ok: bool,
159 /// The result, flattened to one line and truncated.
160 summary: String,
161 },
162 /// A run produced a per-agent log/output line (readable assistant output or
163 /// an operational `[Tokens: …]` / `[tool] …` / `[error] …` line).
164 Log {
165 /// The run id.
166 run_id: String,
167 /// The agent id.
168 agent_id: String,
169 /// The log line text.
170 line: String,
171 },
172}
173
174impl WorldEvent {
175 /// The run id this event belongs to. Every variant carries one; this saves
176 /// consumers an exhaustive match (which, with the enum non-exhaustive,
177 /// they could not write anyway).
178 pub fn run_id(&self) -> &str {
179 match self {
180 WorldEvent::Spawned { run_id, .. }
181 | WorldEvent::Status { run_id, .. }
182 | WorldEvent::Tokens { run_id, .. }
183 | WorldEvent::Context { run_id, .. }
184 | WorldEvent::Interaction { run_id, .. }
185 | WorldEvent::Completed { run_id, .. }
186 | WorldEvent::StageTransition { run_id, .. }
187 | WorldEvent::ToolCallStarted { run_id, .. }
188 | WorldEvent::ToolCallFinished { run_id, .. }
189 | WorldEvent::Log { run_id, .. } => run_id,
190 }
191 }
192}
193
194/// A world resource holding a clone of the host's [`WorldEvent`] broadcast
195/// sender, so ECS systems (e.g. the persistence drain) can push events - notably
196/// per-agent [`WorldEvent::Log`] lines - into the same stream the control
197/// transport serves. Absent in worlds that don't stream (test / `lev run`), where
198/// systems that depend on it become no-ops.
199// `Resource` moved from `bevy_ecs::system` to `bevy_ecs::resource` in 0.19.
200#[derive(bevy_ecs::resource::Resource, Clone)]
201pub struct WorldEventSink(pub broadcast::Sender<WorldEvent>);
202
203/// A short, stable status label for [`WorldEvent`]. Part of the daemon's wire
204/// contract (the REST WebSocket forwards it verbatim), so it comes from the one
205/// table on [`AgentStatus`] rather than a copy that could drift from it.
206pub(super) fn status_str(status: &AgentStatus) -> &'static str {
207 status.label()
208}
209
210/// The last-emitted snapshot of an agent, for change detection.
211#[derive(Clone, Hash)]
212pub(super) struct Emitted {
213 pub(super) status: &'static str,
214 pub(super) stage: String,
215 pub(super) iteration: usize,
216 pub(super) tool_calls: usize,
217 pub(super) accepts_messages: bool,
218 pub(super) prompt_tokens: usize,
219 pub(super) completion_tokens: usize,
220 pub(super) cached_tokens: usize,
221 pub(super) cache_write_tokens: usize,
222 pub(super) context_tokens: usize,
223 pub(super) terminal: bool,
224 /// Why the run is parked, so a change of reason counts as a change worth
225 /// telling subscribers about.
226 pub(super) wait_reason: Option<leviath_core::run_meta::WaitReason>,
227}