nyx_agent_types/
event.rs

1// Typed event bus. Every phase publishes through `AgentEvent`; subscribers
2// fan out per variant. `EventSink` is the shared producer side; each
3// consumer holds its own `EventStream` newtype around a broadcast receiver
4// so the rest of the codebase never names tokio's concrete receiver.
5
6use serde::{Deserialize, Serialize};
7use tokio::sync::broadcast;
8use ts_rs::TS;
9
10use crate::agent::HaltReason;
11
12#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
13#[serde(tag = "kind")]
14pub enum AgentEvent {
15    Run { data: RunEvent },
16    Ai { data: AiEvent },
17    Sandbox { data: SandboxEvent },
18    Finding { data: FindingEvent },
19    Budget { data: BudgetEvent },
20    Quarantine { data: QuarantineEvent },
21    Repro { data: ReproEvent },
22}
23
24#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
25#[serde(tag = "kind")]
26pub enum RunEvent {
27    Heartbeat {
28        #[ts(type = "number")]
29        ts: i64,
30    },
31    /// Run-level lifecycle: dispatcher accepted the work and is about to
32    /// fan out per-repo jobs.
33    RunStarted {
34        run_id: String,
35        project_id: String,
36        repos: Vec<String>,
37        #[ts(type = "number")]
38        started_at_ms: i64,
39    },
40    /// Project-level lifecycle: emitted once per project before the
41    /// per-repo fan-out begins. A run scans exactly one project today;
42    /// the event carries the project's stable id and human-facing name
43    /// so subscribers can group per-repo events under the right project
44    /// without a side lookup.
45    ProjectStarted {
46        run_id: String,
47        project_id: String,
48        project_name: String,
49        #[ts(type = "number")]
50        started_at_ms: i64,
51    },
52    PhaseStarted {
53        run_id: String,
54        project_id: String,
55        phase: String,
56        #[ts(type = "number")]
57        started_at_ms: i64,
58    },
59    PhaseFinished {
60        run_id: String,
61        project_id: String,
62        phase: String,
63        status: String,
64        message: Option<String>,
65        #[ts(type = "number")]
66        finished_at_ms: i64,
67    },
68    EnvironmentStatus {
69        run_id: String,
70        project_id: String,
71        environment_run_id: String,
72        status: String,
73        message: Option<String>,
74        target_urls: Vec<String>,
75        #[ts(type = "number")]
76        ts_ms: i64,
77    },
78    AuthSessionStatus {
79        run_id: String,
80        project_id: String,
81        role: String,
82        status: String,
83        acquired_by: String,
84        message: Option<String>,
85        #[ts(type = "number")]
86        ts_ms: i64,
87    },
88    LiveVerificationCapabilities {
89        run_id: String,
90        project_id: String,
91        #[ts(type = "unknown")]
92        report: serde_json::Value,
93        #[ts(type = "number")]
94        ts_ms: i64,
95    },
96    /// Per-repo lifecycle: a rayon job picked up `repo` and the static
97    /// pass is running.
98    RepoStarted {
99        run_id: String,
100        project_id: String,
101        repo: String,
102        #[ts(type = "number")]
103        started_at_ms: i64,
104    },
105    /// Static pass returned (success or no findings).
106    RepoStaticDone {
107        run_id: String,
108        project_id: String,
109        repo: String,
110        n_diags: u32,
111        #[ts(type = "number")]
112        elapsed_ms: i64,
113    },
114    /// Dynamic / sandbox pass returned. Reserved for the sandbox
115    /// publisher; the static-pass dispatcher does not emit this yet
116    /// and the variant exists so the sandbox crate can publish into
117    /// the same bus without a `RunEvent` shape change.
118    RepoDynamicDone {
119        run_id: String,
120        project_id: String,
121        repo: String,
122        #[ts(type = "number")]
123        elapsed_ms: i64,
124    },
125    /// Per-repo failure: the static pass exited non-zero, panicked, or
126    /// the scan lane refused to start (e.g. binary missing).
127    RepoFailed {
128        run_id: String,
129        project_id: String,
130        repo: String,
131        message: String,
132        #[ts(type = "number")]
133        elapsed_ms: i64,
134    },
135    /// Ingest-time failure: the repo could not be cloned, fetched, or
136    /// snapshotted before the dispatcher saw a workspace. Emitted from
137    /// the caller (CLI / API drive-scan path) *before* `RunStarted`, so
138    /// subscribers connected at run start time can reconstruct the full
139    /// attempted-repo set from `RunStarted.repos` alone; the failing
140    /// repo is included there and `RepoIngestFailed` carries the
141    /// upstream error string for UI surfacing.
142    RepoIngestFailed { run_id: String, project_id: String, repo: String, message: String },
143    /// Per-repo terminator. Always emitted regardless of outcome so
144    /// subscribers can drop bookkeeping for the repo without diffing
145    /// the success / failure event streams.
146    RepoFinished {
147        run_id: String,
148        project_id: String,
149        repo: String,
150        outcome: RepoOutcomeTag,
151        #[ts(type = "number")]
152        elapsed_ms: i64,
153    },
154    /// Project-level terminator. Emitted once every repo in the project
155    /// has produced a `RepoFinished` but before the run-level
156    /// `RunFinished`.
157    ProjectFinished {
158        run_id: String,
159        project_id: String,
160        #[ts(type = "number")]
161        finished_at_ms: i64,
162    },
163    /// Run-level terminator. Emitted once every repo has produced a
164    /// `RepoFinished`.
165    RunFinished {
166        run_id: String,
167        project_id: String,
168        #[ts(type = "number")]
169        finished_at_ms: i64,
170        #[ts(type = "number")]
171        wall_clock_ms: i64,
172        succeeded: u32,
173        inconclusive: u32,
174        failed: u32,
175    },
176}
177
178/// Compressed flavour tag carried on `RepoFinished`. The aggregator's
179/// `RepoBundle` keeps the full typed outcome; subscribers that only
180/// need a colour for a UI badge read this.
181#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
182pub enum RepoOutcomeTag {
183    Success,
184    /// Static pass did not finish (e.g. per-repo timeout).
185    Inconclusive,
186    /// Static pass failed outright (scanner crash, refusal, etc.).
187    Failed,
188}
189
190/// AI-runtime event stream. Adapters publish one of these per token /
191/// tool call / cache event / budget tick / halt as a `one_shot` or
192/// `agent_loop` progresses. `task_id` is the caller-supplied
193/// identifier from the `Prompt` / `AgentTask`; subscribers fan out by
194/// it to multiplex concurrent calls on a single bus.
195#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
196#[serde(tag = "kind")]
197pub enum AiEvent {
198    TokenReceived {
199        task_id: String,
200        token: String,
201    },
202    ToolCallStarted {
203        task_id: String,
204        name: String,
205    },
206    ToolCallFinished {
207        task_id: String,
208        name: String,
209        ok: bool,
210    },
211    CacheHit {
212        task_id: String,
213        tokens: u32,
214    },
215    CacheMiss {
216        task_id: String,
217        tokens: u32,
218    },
219    BudgetTick {
220        task_id: String,
221        run_id: String,
222        #[ts(type = "number")]
223        spent_usd_micros: i64,
224    },
225    TaskHalted {
226        task_id: String,
227        reason: HaltReason,
228    },
229}
230
231/// Sandbox / verifier lifecycle events. Subscribers fan out by
232/// `run_id`; consumers that only care about a single finding's verifier
233/// progress key off `finding_id`. Today only the deterministic
234/// verifier publishes here; future backends (chain-lane runner,
235/// AI exploration sandbox) will add their own variants.
236#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
237#[serde(tag = "kind")]
238pub enum SandboxEvent {
239    /// Verifier pass picked up a finding and is about to launch the
240    /// vuln/benign payload pair. Emitted once per finding the pass
241    /// actually drives (skipped findings produce no event).
242    VerifierStarted {
243        run_id: String,
244        finding_id: String,
245        repo: String,
246        #[ts(type = "number")]
247        started_at_ms: i64,
248    },
249    /// Verifier pass finished a finding. `verdict` mirrors
250    /// `VerifyVerdict::as_str()` (`"Confirmed"` / `"NotConfirmed"` /
251    /// `"Errored"`). `replay_stable` stays `None` when the
252    /// `[run] replay_stable_check` knob is off; `Some(true)` when the
253    /// optional second run produced an identical verdict.
254    VerifierFinished {
255        run_id: String,
256        finding_id: String,
257        repo: String,
258        verdict: String,
259        replay_stable: Option<bool>,
260        #[ts(type = "number")]
261        elapsed_ms: i64,
262    },
263}
264
265#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, TS)]
266pub struct FindingEvent {}
267
268#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, TS)]
269pub struct BudgetEvent {}
270
271#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, TS)]
272pub struct QuarantineEvent {}
273
274/// Repro / replay lifecycle events. Mirrors the SSE frames emitted by
275/// `POST /api/v1/findings/:id/replay` so a WebSocket subscriber can
276/// keep rendering the replay log after the operator navigates away
277/// from the FindingDetail panel and the SSE connection drops.
278#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
279#[serde(tag = "kind")]
280#[allow(clippy::enum_variant_names)]
281pub enum ReproEvent {
282    /// Replay just kicked off. Stamped with the resolved bundle path
283    /// so subscribers can correlate against `repro_bundles.path`.
284    ReplayStarted {
285        finding_id: String,
286        bundle_path: String,
287        #[ts(type = "number")]
288        started_at_ms: i64,
289    },
290    /// One line of replay stdout.
291    ReplayStdout { finding_id: String, line: String },
292    /// One line of replay stderr.
293    ReplayStderr { finding_id: String, line: String },
294    /// Out-of-band error (spawn / IO / timeout). `message` mirrors the
295    /// SSE `error` frame body verbatim.
296    ReplayError { finding_id: String, message: String },
297    /// Replay finished (clean exit or non-zero). `status` mirrors the
298    /// SSE `end` frame's status field (`"Pass"` / `"Fail"`).
299    ReplayFinished {
300        finding_id: String,
301        status: String,
302        #[ts(type = "number")]
303        exit_code: i32,
304        #[ts(type = "number")]
305        started_at_ms: i64,
306        #[ts(type = "number")]
307        finished_at_ms: i64,
308        #[ts(type = "number")]
309        duration_ms: i64,
310    },
311}
312
313pub type EventSink = broadcast::Sender<AgentEvent>;
314
315#[derive(Debug)]
316pub struct EventStream(pub broadcast::Receiver<AgentEvent>);
317
318impl EventStream {
319    pub fn new(rx: broadcast::Receiver<AgentEvent>) -> Self {
320        Self(rx)
321    }
322}
323
324#[cfg(test)]
325mod tests {
326    use super::*;
327
328    #[tokio::test]
329    async fn heartbeat_roundtrips_via_broadcast() {
330        let (tx, rx) = broadcast::channel::<AgentEvent>(8);
331        let mut stream = EventStream::new(rx);
332        let original = AgentEvent::Run { data: RunEvent::Heartbeat { ts: 42 } };
333        tx.send(original.clone()).expect("send");
334        let received = stream.0.recv().await.expect("recv");
335        assert_eq!(received, original);
336    }
337
338    #[test]
339    fn heartbeat_serde_roundtrip() {
340        let original = AgentEvent::Run { data: RunEvent::Heartbeat { ts: 7 } };
341        let json = serde_json::to_string(&original).expect("serialize");
342        let back: AgentEvent = serde_json::from_str(&json).expect("deserialize");
343        assert_eq!(back, original);
344    }
345}
nyx_agent_types/event.rs

nyx_agent_types/
event.rs
