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 },
56 /// A run's token totals changed.
57 Tokens {
58 /// The run id.
59 run_id: String,
60 /// The agent id.
61 agent_id: String,
62 /// Cumulative prompt tokens.
63 prompt_tokens: usize,
64 /// Cumulative completion tokens.
65 completion_tokens: usize,
66 /// Cumulative cached tokens.
67 cached_tokens: usize,
68 /// Cumulative cache-write tokens.
69 cache_write_tokens: usize,
70 },
71 /// A run's context-window token usage changed.
72 Context {
73 /// The run id.
74 run_id: String,
75 /// The agent id.
76 agent_id: String,
77 /// Current context tokens.
78 total_tokens: usize,
79 /// Max context tokens.
80 max_tokens: usize,
81 },
82 /// A run raised a new interaction awaiting an answer.
83 Interaction {
84 /// The run id.
85 run_id: String,
86 /// The agent id.
87 agent_id: String,
88 /// The interaction request.
89 request: InteractionRequest,
90 },
91 /// A run reached a terminal status.
92 Completed {
93 /// The run id.
94 run_id: String,
95 /// The agent id.
96 agent_id: String,
97 /// The terminal status label.
98 status: String,
99 /// What the run handed back, when it submitted anything.
100 ///
101 /// Carried on the event rather than left for the consumer to read off
102 /// disk: this fires the moment the run goes terminal, and the persist
103 /// tick that writes `meta.json` has not necessarily run yet. A webhook
104 /// or websocket consumer reading the file would race it and report a
105 /// finished run with no answer.
106 #[serde(default, skip_serializing_if = "Option::is_none")]
107 final_output: Option<leviath_core::output::FinalOutput>,
108 },
109 /// A run moved from one stage to another. Emitted by the transition systems
110 /// at the moment the new stage is entered (the initial stage at spawn is
111 /// covered by [`WorldEvent::Spawned`], not by this).
112 StageTransition {
113 /// The run id.
114 run_id: String,
115 /// The agent id.
116 agent_id: String,
117 /// The stage being left.
118 from: String,
119 /// The stage being entered.
120 to: String,
121 /// How many times the destination stage has been entered, this entry
122 /// included.
123 iteration: usize,
124 },
125 /// A tool call was handed to the async tool lane for execution. Inline
126 /// calls (context tools, refusals, gate blocks) resolve without touching
127 /// the lane and don't produce this event.
128 ToolCallStarted {
129 /// The run id.
130 run_id: String,
131 /// The agent id.
132 agent_id: String,
133 /// The provider-assigned tool call id.
134 call_id: String,
135 /// The tool name.
136 tool: String,
137 },
138 /// A lane-executed tool call returned. Paired with
139 /// [`WorldEvent::ToolCallStarted`] by `call_id`.
140 ToolCallFinished {
141 /// The run id.
142 run_id: String,
143 /// The agent id.
144 agent_id: String,
145 /// The provider-assigned tool call id.
146 call_id: String,
147 /// The tool name.
148 tool: String,
149 /// Whether the call took effect (`false` for `[error]`/`[blocked]`/
150 /// `[unavailable]` results).
151 ok: bool,
152 /// The result, flattened to one line and truncated.
153 summary: String,
154 },
155 /// A run produced a per-agent log/output line (readable assistant output or
156 /// an operational `[Tokens: …]` / `[tool] …` / `[error] …` line).
157 Log {
158 /// The run id.
159 run_id: String,
160 /// The agent id.
161 agent_id: String,
162 /// The log line text.
163 line: String,
164 },
165}
166
167impl WorldEvent {
168 /// The run id this event belongs to. Every variant carries one; this saves
169 /// consumers an exhaustive match (which, with the enum non-exhaustive,
170 /// they could not write anyway).
171 pub fn run_id(&self) -> &str {
172 match self {
173 WorldEvent::Spawned { run_id, .. }
174 | WorldEvent::Status { run_id, .. }
175 | WorldEvent::Tokens { run_id, .. }
176 | WorldEvent::Context { run_id, .. }
177 | WorldEvent::Interaction { run_id, .. }
178 | WorldEvent::Completed { run_id, .. }
179 | WorldEvent::StageTransition { run_id, .. }
180 | WorldEvent::ToolCallStarted { run_id, .. }
181 | WorldEvent::ToolCallFinished { run_id, .. }
182 | WorldEvent::Log { run_id, .. } => run_id,
183 }
184 }
185}
186
187/// A world resource holding a clone of the host's [`WorldEvent`] broadcast
188/// sender, so ECS systems (e.g. the persistence drain) can push events - notably
189/// per-agent [`WorldEvent::Log`] lines - into the same stream the control
190/// transport serves. Absent in worlds that don't stream (test / `lev run`), where
191/// systems that depend on it become no-ops.
192// `Resource` moved from `bevy_ecs::system` to `bevy_ecs::resource` in 0.19.
193#[derive(bevy_ecs::resource::Resource, Clone)]
194pub struct WorldEventSink(pub broadcast::Sender<WorldEvent>);
195
196/// A short, stable status label for [`WorldEvent`]. Part of the daemon's wire
197/// contract (the REST WebSocket forwards it verbatim), so it comes from the one
198/// table on [`AgentStatus`] rather than a copy that could drift from it.
199pub(super) fn status_str(status: &AgentStatus) -> &'static str {
200 status.label()
201}
202
203/// The last-emitted snapshot of an agent, for change detection.
204#[derive(Clone, Hash)]
205pub(super) struct Emitted {
206 pub(super) status: &'static str,
207 pub(super) stage: String,
208 pub(super) iteration: usize,
209 pub(super) tool_calls: usize,
210 pub(super) accepts_messages: bool,
211 pub(super) prompt_tokens: usize,
212 pub(super) completion_tokens: usize,
213 pub(super) cached_tokens: usize,
214 pub(super) cache_write_tokens: usize,
215 pub(super) context_tokens: usize,
216 pub(super) terminal: bool,
217}