Skip to main content

ironflow_engine/notify/
event.rs

1//! Domain events emitted throughout the ironflow lifecycle.
2
3use std::collections::HashMap;
4
5use chrono::{DateTime, Utc};
6use rust_decimal::Decimal;
7use serde::{Deserialize, Serialize};
8use uuid::Uuid;
9
10pub use ironflow_store::entities::LogStream;
11use ironflow_store::models::{
12    ApprovalRequirement, Assignee, ProviderAccountWindow, RunStatus, ScheduleMissReason, StepKind,
13};
14
15/// Vote counts assumed for approval events serialized before multi-approver
16/// gates existed: one approval was always enough.
17fn default_approval_count() -> u32 {
18    1
19}
20
21/// Payload of the `Event::RunCreated` event.
22///
23/// # Examples
24///
25/// ```
26/// use chrono::Utc;
27/// use ironflow_engine::notify::RunCreatedEvent;
28/// use uuid::Uuid;
29///
30/// let payload = RunCreatedEvent {
31///     run_id: Uuid::now_v7(),
32///     workflow_name: "deploy".to_string(),
33///     at: Utc::now(),
34/// };
35/// assert_eq!(payload.workflow_name, "deploy");
36/// ```
37#[derive(Debug, Clone, Serialize, Deserialize)]
38#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
39pub struct RunCreatedEvent {
40    /// Run identifier.
41    pub run_id: Uuid,
42    /// Workflow name.
43    pub workflow_name: String,
44    /// When the run was created.
45    pub at: DateTime<Utc>,
46}
47
48/// Payload of the `Event::RunStatusChanged` event.
49///
50/// # Examples
51///
52/// ```
53/// use std::collections::HashMap;
54///
55/// use chrono::Utc;
56/// use ironflow_engine::notify::RunStatusChangedEvent;
57/// use ironflow_store::models::RunStatus;
58/// use rust_decimal::Decimal;
59/// use uuid::Uuid;
60///
61/// let payload = RunStatusChangedEvent {
62///     run_id: Uuid::now_v7(),
63///     workflow_name: "deploy".to_string(),
64///     from: RunStatus::Running,
65///     to: RunStatus::Completed,
66///     error: None,
67///     cost_usd: Decimal::ZERO,
68///     duration_ms: 5000,
69///     labels: HashMap::new(),
70///     at: Utc::now(),
71/// };
72/// assert_eq!(payload.to, RunStatus::Completed);
73/// ```
74#[derive(Debug, Clone, Serialize, Deserialize)]
75#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
76pub struct RunStatusChangedEvent {
77    /// Run identifier.
78    pub run_id: Uuid,
79    /// Workflow name.
80    pub workflow_name: String,
81    /// Previous status.
82    pub from: RunStatus,
83    /// New status.
84    pub to: RunStatus,
85    /// Error message (when transitioning to Failed).
86    pub error: Option<String>,
87    /// Aggregated cost in USD at the time of transition.
88    pub cost_usd: Decimal,
89    /// Aggregated duration in milliseconds at the time of transition.
90    pub duration_ms: u64,
91    /// Labels of the run at the time of the transition.
92    #[serde(default)]
93    pub labels: HashMap<String, String>,
94    /// When the transition occurred.
95    pub at: DateTime<Utc>,
96}
97
98/// Payload of the `Event::RunFailed` event.
99///
100/// # Examples
101///
102/// ```
103/// use std::collections::HashMap;
104///
105/// use chrono::Utc;
106/// use ironflow_engine::notify::RunFailedEvent;
107/// use rust_decimal::Decimal;
108/// use uuid::Uuid;
109///
110/// let payload = RunFailedEvent {
111///     run_id: Uuid::now_v7(),
112///     workflow_name: "deploy".to_string(),
113///     error: Some("step crashed".to_string()),
114///     cost_usd: Decimal::ZERO,
115///     duration_ms: 3000,
116///     labels: HashMap::new(),
117///     at: Utc::now(),
118/// };
119/// assert_eq!(payload.error.as_deref(), Some("step crashed"));
120/// ```
121#[derive(Debug, Clone, Serialize, Deserialize)]
122#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
123pub struct RunFailedEvent {
124    /// Run identifier.
125    pub run_id: Uuid,
126    /// Workflow name.
127    pub workflow_name: String,
128    /// Error message.
129    pub error: Option<String>,
130    /// Aggregated cost in USD at the time of failure.
131    pub cost_usd: Decimal,
132    /// Aggregated duration in milliseconds at the time of failure.
133    pub duration_ms: u64,
134    /// Labels of the run at the time of the failure.
135    #[serde(default)]
136    pub labels: HashMap<String, String>,
137    /// When the failure occurred.
138    pub at: DateTime<Utc>,
139}
140
141/// Payload of the `Event::RunBudgetExceeded` event.
142///
143/// # Examples
144///
145/// ```
146/// use chrono::Utc;
147/// use ironflow_engine::notify::RunBudgetExceededEvent;
148/// use rust_decimal::Decimal;
149/// use uuid::Uuid;
150///
151/// let payload = RunBudgetExceededEvent {
152///     run_id: Uuid::now_v7(),
153///     workflow_name: "deploy".to_string(),
154///     limit_usd: Decimal::new(200, 2),
155///     spent_usd: Decimal::new(180, 2),
156///     step_budget_usd: Decimal::new(50, 2),
157///     at: Utc::now(),
158/// };
159/// assert_eq!(payload.limit_usd, Decimal::new(200, 2));
160/// ```
161#[derive(Debug, Clone, Serialize, Deserialize)]
162#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
163pub struct RunBudgetExceededEvent {
164    /// Run identifier.
165    pub run_id: Uuid,
166    /// Workflow name.
167    pub workflow_name: String,
168    /// The configured cost cap in USD.
169    pub limit_usd: Decimal,
170    /// Cost already consumed when the cap was reached, in USD.
171    pub spent_usd: Decimal,
172    /// Declared budget of the refused step, in USD.
173    pub step_budget_usd: Decimal,
174    /// When the refusal occurred.
175    pub at: DateTime<Utc>,
176}
177
178/// Payload of the `Event::RetryForced` event.
179///
180/// # Examples
181///
182/// ```
183/// use chrono::Utc;
184/// use ironflow_engine::notify::RetryForcedEvent;
185/// use uuid::Uuid;
186///
187/// let payload = RetryForcedEvent {
188///     run_id: Uuid::now_v7(),
189///     workflow_name: "deploy".to_string(),
190///     original_version: "1".to_string(),
191///     current_version: "2".to_string(),
192///     at: Utc::now(),
193/// };
194/// assert_eq!(payload.current_version, "2");
195/// ```
196#[derive(Debug, Clone, Serialize, Deserialize)]
197#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
198pub struct RetryForcedEvent {
199    /// The new run created by the forced retry.
200    pub run_id: Uuid,
201    /// Workflow name.
202    pub workflow_name: String,
203    /// Version stored on the original run.
204    pub original_version: String,
205    /// Current version of the handler.
206    pub current_version: String,
207    /// When the forced retry occurred.
208    pub at: DateTime<Utc>,
209}
210
211/// Payload of the `Event::StepCompleted` event.
212///
213/// # Examples
214///
215/// ```
216/// use chrono::Utc;
217/// use ironflow_engine::notify::StepCompletedEvent;
218/// use ironflow_store::models::StepKind;
219/// use rust_decimal::Decimal;
220/// use uuid::Uuid;
221///
222/// let payload = StepCompletedEvent {
223///     run_id: Uuid::now_v7(),
224///     step_id: Uuid::now_v7(),
225///     step_name: "build".to_string(),
226///     kind: StepKind::Shell,
227///     duration_ms: 1200,
228///     cost_usd: Decimal::ZERO,
229///     at: Utc::now(),
230/// };
231/// assert_eq!(payload.step_name, "build");
232/// ```
233#[derive(Debug, Clone, Serialize, Deserialize)]
234#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
235pub struct StepCompletedEvent {
236    /// Run identifier.
237    pub run_id: Uuid,
238    /// Step identifier.
239    pub step_id: Uuid,
240    /// Human-readable step name.
241    pub step_name: String,
242    /// Step operation kind.
243    #[cfg_attr(feature = "openapi", schema(value_type = String))]
244    pub kind: StepKind,
245    /// Step duration in milliseconds.
246    pub duration_ms: u64,
247    /// Step cost in USD.
248    pub cost_usd: Decimal,
249    /// When the step completed.
250    pub at: DateTime<Utc>,
251}
252
253/// Payload of the `Event::StepFailed` event.
254///
255/// # Examples
256///
257/// ```
258/// use chrono::Utc;
259/// use ironflow_engine::notify::StepFailedEvent;
260/// use ironflow_store::models::StepKind;
261/// use uuid::Uuid;
262///
263/// let payload = StepFailedEvent {
264///     run_id: Uuid::now_v7(),
265///     step_id: Uuid::now_v7(),
266///     step_name: "build".to_string(),
267///     kind: StepKind::Shell,
268///     error: "exit code 1".to_string(),
269///     at: Utc::now(),
270/// };
271/// assert_eq!(payload.error, "exit code 1");
272/// ```
273#[derive(Debug, Clone, Serialize, Deserialize)]
274#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
275pub struct StepFailedEvent {
276    /// Run identifier.
277    pub run_id: Uuid,
278    /// Step identifier.
279    pub step_id: Uuid,
280    /// Human-readable step name.
281    pub step_name: String,
282    /// Step operation kind.
283    #[cfg_attr(feature = "openapi", schema(value_type = String))]
284    pub kind: StepKind,
285    /// Error message.
286    pub error: String,
287    /// When the step failed.
288    pub at: DateTime<Utc>,
289}
290
291/// Payload of the `Event::ApprovalRequested` event.
292///
293/// Published when an approval gate opens. Carries the approvers the handler
294/// required, if it set any.
295///
296/// # Examples
297///
298/// ```
299/// use chrono::Utc;
300/// use ironflow_engine::notify::ApprovalRequestedEvent;
301/// use uuid::Uuid;
302///
303/// let payload = ApprovalRequestedEvent {
304///     run_id: Uuid::now_v7(),
305///     step_id: Uuid::now_v7(),
306///     message: "Deploy to prod?".to_string(),
307///     requirement: None,
308///     at: Utc::now(),
309/// };
310/// assert_eq!(payload.message, "Deploy to prod?");
311/// ```
312#[derive(Debug, Clone, Serialize, Deserialize)]
313#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
314pub struct ApprovalRequestedEvent {
315    /// Run identifier.
316    pub run_id: Uuid,
317    /// Approval step identifier.
318    pub step_id: Uuid,
319    /// Message displayed to reviewers.
320    pub message: String,
321    /// Approvers the handler required. `None` for a gate opened without
322    /// approvers: one approval resolves it.
323    #[serde(default)]
324    pub requirement: Option<ApprovalRequirement>,
325    /// When the approval was requested.
326    pub at: DateTime<Utc>,
327}
328
329/// Payload of the `Event::ApprovalGranted` event.
330///
331/// Published for every vote cast on a gate. A vote with
332/// `approvals_received < approvals_required` is recorded but does not resolve
333/// the gate: the run stays `AwaitingApproval` until enough distinct approvers
334/// voted.
335///
336/// # Examples
337///
338/// ```
339/// use chrono::Utc;
340/// use ironflow_engine::notify::ApprovalGrantedEvent;
341/// use uuid::Uuid;
342///
343/// let payload = ApprovalGrantedEvent {
344///     run_id: Uuid::now_v7(),
345///     step_id: Some(Uuid::now_v7()),
346///     approved_by: "alice".to_string(),
347///     user_id: Some(Uuid::now_v7()),
348///     approvals_received: 1,
349///     approvals_required: 2,
350///     requirement: None,
351///     at: Utc::now(),
352/// };
353/// assert!(payload.approvals_received < payload.approvals_required);
354/// ```
355#[derive(Debug, Clone, Serialize, Deserialize)]
356#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
357pub struct ApprovalGrantedEvent {
358    /// Run identifier.
359    pub run_id: Uuid,
360    /// Approval step identifier. `None` in events recorded before it existed.
361    #[serde(default)]
362    pub step_id: Option<Uuid>,
363    /// User who approved (ID or username).
364    pub approved_by: String,
365    /// ID of the user who approved. `None` for a system decision (timeout
366    /// escalation) and in events recorded before it existed. `approved_by`
367    /// stays the display label.
368    #[serde(default)]
369    pub user_id: Option<Uuid>,
370    /// Distinct approvals recorded on the gate, this one included.
371    #[serde(default = "default_approval_count")]
372    pub approvals_received: u32,
373    /// Distinct approvals needed to resolve the gate.
374    #[serde(default = "default_approval_count")]
375    pub approvals_required: u32,
376    /// Approvers the handler required, if any.
377    #[serde(default)]
378    pub requirement: Option<ApprovalRequirement>,
379    /// When the approval was granted.
380    pub at: DateTime<Utc>,
381}
382
383/// Payload of the `Event::ApprovalRejected` event.
384///
385/// # Examples
386///
387/// ```
388/// use chrono::Utc;
389/// use ironflow_engine::notify::ApprovalRejectedEvent;
390/// use uuid::Uuid;
391///
392/// let payload = ApprovalRejectedEvent {
393///     run_id: Uuid::now_v7(),
394///     step_id: Some(Uuid::now_v7()),
395///     rejected_by: "bob".to_string(),
396///     user_id: Some(Uuid::now_v7()),
397///     reason: None,
398///     requirement: None,
399///     at: Utc::now(),
400/// };
401/// assert_eq!(payload.rejected_by, "bob");
402/// ```
403#[derive(Debug, Clone, Serialize, Deserialize)]
404#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
405pub struct ApprovalRejectedEvent {
406    /// Run identifier.
407    pub run_id: Uuid,
408    /// Approval step identifier. `None` in events recorded before it existed.
409    #[serde(default)]
410    pub step_id: Option<Uuid>,
411    /// User who rejected (ID or username).
412    pub rejected_by: String,
413    /// ID of the user who rejected. `None` for a system decision (timeout
414    /// escalation) and in events recorded before it existed. `rejected_by`
415    /// stays the display label.
416    #[serde(default)]
417    pub user_id: Option<Uuid>,
418    /// Why the decision was refused, when the rejection carries one (human
419    /// input rejection). `None` for a gate rejection.
420    #[serde(default)]
421    pub reason: Option<String>,
422    /// Approvers the handler required, if any.
423    #[serde(default)]
424    pub requirement: Option<ApprovalRequirement>,
425    /// When the rejection occurred.
426    pub at: DateTime<Utc>,
427}
428
429/// Payload of the `Event::ApprovalEscalated` event.
430///
431/// Emitted every time an approval gate misses its SLA deadline, including the
432/// repeated firings of a bare `Notify`/`Escalate` policy and the final
433/// "chain exhausted" notice. The audit log persists it verbatim, so the whole
434/// escalation history of a gate is reconstructable from it.
435///
436/// # Examples
437///
438/// ```
439/// use chrono::Utc;
440/// use ironflow_engine::notify::ApprovalEscalatedEvent;
441/// use uuid::Uuid;
442///
443/// let payload = ApprovalEscalatedEvent {
444///     run_id: Uuid::now_v7(),
445///     step_id: Uuid::now_v7(),
446///     step_name: "prod-gate".to_string(),
447///     stage: 0,
448///     policy: "auto_reject".to_string(),
449///     action: "rejected".to_string(),
450///     reason: "approval deadline of 3600s expired".to_string(),
451///     assignee: None,
452///     at: Utc::now(),
453/// };
454/// assert_eq!(payload.policy, "auto_reject");
455/// ```
456#[derive(Debug, Clone, Serialize, Deserialize)]
457#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
458pub struct ApprovalEscalatedEvent {
459    /// Run identifier.
460    pub run_id: Uuid,
461    /// Approval step identifier.
462    pub step_id: Uuid,
463    /// Human-readable step name.
464    pub step_name: String,
465    /// Escalation stage that fired (0-based index into the policy chain).
466    pub stage: u32,
467    /// Policy applied, e.g. `"auto_reject"`, `"notify"`, `"escalate"`.
468    pub policy: String,
469    /// What the escalation did, for the audit log.
470    pub action: String,
471    /// Why it fired, e.g. `"approval deadline of 3600s expired"`.
472    pub reason: String,
473    /// Assignee after the escalation, when it reassigned the gate.
474    #[cfg_attr(feature = "openapi", schema(value_type = Option<String>))]
475    pub assignee: Option<Assignee>,
476    /// When the escalation ran.
477    pub at: DateTime<Utc>,
478}
479
480/// Payload of the `Event::LogLine` event.
481///
482/// # Examples
483///
484/// ```
485/// use chrono::Utc;
486/// use ironflow_engine::notify::{LogLineEvent, LogStream};
487/// use uuid::Uuid;
488///
489/// let payload = LogLineEvent {
490///     id: Uuid::now_v7(),
491///     run_id: Uuid::now_v7(),
492///     step_id: Uuid::now_v7(),
493///     step_name: "build".to_string(),
494///     stream: LogStream::Stdout,
495///     line: "Compiling ironflow v0.1.0".to_string(),
496///     at: Utc::now(),
497/// };
498/// assert_eq!(payload.line, "Compiling ironflow v0.1.0");
499/// ```
500#[derive(Debug, Clone, Serialize, Deserialize)]
501#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
502pub struct LogLineEvent {
503    /// Persisted entry identifier (UUID v7, time-ordered).
504    ///
505    /// Matches the `id` of the log entry later stored for this line, so clients
506    /// can de-duplicate the live SSE stream against the persisted history
507    /// fetched from `GET /runs/:id/logs`.
508    ///
509    /// Defaults to the nil UUID when absent, so a payload emitted by an older
510    /// producer still deserializes instead of dropping the whole event.
511    #[serde(default)]
512    pub id: Uuid,
513    /// Run identifier.
514    pub run_id: Uuid,
515    /// Step identifier.
516    pub step_id: Uuid,
517    /// Human-readable step name.
518    pub step_name: String,
519    /// Output stream.
520    pub stream: LogStream,
521    /// The log line content.
522    pub line: String,
523    /// When the line was emitted.
524    pub at: DateTime<Utc>,
525}
526
527/// Payload of the `Event::UserSignedIn` event.
528///
529/// # Examples
530///
531/// ```
532/// use chrono::Utc;
533/// use ironflow_engine::notify::UserSignedInEvent;
534/// use uuid::Uuid;
535///
536/// let payload = UserSignedInEvent {
537///     user_id: Uuid::now_v7(),
538///     username: "alice".to_string(),
539///     at: Utc::now(),
540/// };
541/// assert_eq!(payload.username, "alice");
542/// ```
543#[derive(Debug, Clone, Serialize, Deserialize)]
544#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
545pub struct UserSignedInEvent {
546    /// User identifier.
547    pub user_id: Uuid,
548    /// Username.
549    pub username: String,
550    /// When the sign-in occurred.
551    pub at: DateTime<Utc>,
552}
553
554/// Payload of the `Event::UserSignedUp` event.
555///
556/// # Examples
557///
558/// ```
559/// use chrono::Utc;
560/// use ironflow_engine::notify::UserSignedUpEvent;
561/// use uuid::Uuid;
562///
563/// let payload = UserSignedUpEvent {
564///     user_id: Uuid::now_v7(),
565///     username: "alice".to_string(),
566///     at: Utc::now(),
567/// };
568/// assert_eq!(payload.username, "alice");
569/// ```
570#[derive(Debug, Clone, Serialize, Deserialize)]
571#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
572pub struct UserSignedUpEvent {
573    /// User identifier.
574    pub user_id: Uuid,
575    /// Username.
576    pub username: String,
577    /// When the sign-up occurred.
578    pub at: DateTime<Utc>,
579}
580
581/// Payload of the `Event::UserSignedOut` event.
582///
583/// # Examples
584///
585/// ```
586/// use chrono::Utc;
587/// use ironflow_engine::notify::UserSignedOutEvent;
588/// use uuid::Uuid;
589///
590/// let user_id = Uuid::now_v7();
591/// let payload = UserSignedOutEvent {
592///     user_id,
593///     at: Utc::now(),
594/// };
595/// assert_eq!(payload.user_id, user_id);
596/// ```
597#[derive(Debug, Clone, Serialize, Deserialize)]
598#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
599pub struct UserSignedOutEvent {
600    /// User identifier.
601    pub user_id: Uuid,
602    /// When the sign-out occurred.
603    pub at: DateTime<Utc>,
604}
605
606/// What happened to a Provider Account in a
607/// [`ProviderAccountUpdatedEvent`].
608///
609/// # Examples
610///
611/// ```
612/// use ironflow_engine::notify::ProviderAccountChange;
613///
614/// assert_eq!(ProviderAccountChange::TokenReplaced.as_str(), "token_replaced");
615/// ```
616#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
617#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
618#[serde(rename_all = "snake_case")]
619pub enum ProviderAccountChange {
620    /// The account was created.
621    Created,
622    /// Settings of the account changed.
623    Updated,
624    /// The account was deleted.
625    Deleted,
626    /// The credential was replaced.
627    TokenReplaced,
628}
629
630impl ProviderAccountChange {
631    /// Wire name of the change, as stored in audit payloads.
632    pub fn as_str(&self) -> &'static str {
633        match self {
634            Self::Created => "created",
635            Self::Updated => "updated",
636            Self::Deleted => "deleted",
637            Self::TokenReplaced => "token_replaced",
638        }
639    }
640}
641
642/// Payload of the `Event::ProviderAccountUpdated` event.
643///
644/// Never carries the credential.
645///
646/// # Examples
647///
648/// ```
649/// use chrono::Utc;
650/// use ironflow_engine::notify::{ProviderAccountChange, ProviderAccountUpdatedEvent};
651/// use uuid::Uuid;
652///
653/// let payload = ProviderAccountUpdatedEvent {
654///     account_id: Uuid::now_v7(),
655///     name: "perso".to_string(),
656///     change: ProviderAccountChange::Created,
657///     at: Utc::now(),
658/// };
659/// assert_eq!(payload.name, "perso");
660/// ```
661#[derive(Debug, Clone, Serialize, Deserialize)]
662#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
663pub struct ProviderAccountUpdatedEvent {
664    /// Account identifier.
665    pub account_id: Uuid,
666    /// Account name.
667    pub name: String,
668    /// What changed.
669    pub change: ProviderAccountChange,
670    /// When the change happened.
671    pub at: DateTime<Utc>,
672}
673
674/// Payload of the `Event::ProviderAccountUsageUpdated` event.
675///
676/// # Examples
677///
678/// ```
679/// use chrono::Utc;
680/// use ironflow_engine::notify::ProviderAccountUsageUpdatedEvent;
681/// use uuid::Uuid;
682///
683/// let payload = ProviderAccountUsageUpdatedEvent {
684///     account_id: Uuid::now_v7(),
685///     name: "perso".to_string(),
686///     windows: Vec::new(),
687///     at: Utc::now(),
688/// };
689/// assert!(payload.windows.is_empty());
690/// ```
691#[derive(Debug, Clone, Serialize, Deserialize)]
692#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
693pub struct ProviderAccountUsageUpdatedEvent {
694    /// Account identifier.
695    pub account_id: Uuid,
696    /// Account name.
697    pub name: String,
698    /// Current windows of the account.
699    pub windows: Vec<ProviderAccountWindow>,
700    /// When the windows were recorded.
701    pub at: DateTime<Utc>,
702}
703
704/// Payload of the `Event::SignalAwaited` event.
705///
706/// Emitted when a run suspends on `ctx.wait_for_signal` and goes `Sleeping`.
707///
708/// # Examples
709///
710/// ```
711/// use chrono::{TimeDelta, Utc};
712/// use ironflow_engine::notify::SignalAwaitedEvent;
713/// use uuid::Uuid;
714///
715/// let now = Utc::now();
716/// let payload = SignalAwaitedEvent {
717///     run_id: Uuid::now_v7(),
718///     step_id: Uuid::now_v7(),
719///     step_name: "wait-ci".to_string(),
720///     name: "ci.pipeline_finished".to_string(),
721///     key: "4f2a9c1".to_string(),
722///     deadline_at: now + TimeDelta::hours(1),
723///     at: now,
724/// };
725/// assert_eq!(payload.key, "4f2a9c1");
726/// ```
727#[derive(Debug, Clone, Serialize, Deserialize)]
728#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
729pub struct SignalAwaitedEvent {
730    /// Run identifier.
731    pub run_id: Uuid,
732    /// The signal step the run waits on.
733    pub step_id: Uuid,
734    /// Name of the signal step.
735    pub step_name: String,
736    /// Awaited signal name.
737    pub name: String,
738    /// Awaited occurrence key.
739    pub key: String,
740    /// When the wait times out.
741    pub deadline_at: DateTime<Utc>,
742    /// When the run started waiting.
743    pub at: DateTime<Utc>,
744}
745
746/// Payload of the `Event::SignalReceived` event.
747///
748/// Emitted once per stored signal; a duplicate (same idempotency ID) emits
749/// nothing.
750///
751/// # Examples
752///
753/// ```
754/// use chrono::Utc;
755/// use ironflow_engine::notify::SignalReceivedEvent;
756/// use uuid::Uuid;
757///
758/// let payload = SignalReceivedEvent {
759///     signal_id: Uuid::now_v7(),
760///     name: "ci.pipeline_finished".to_string(),
761///     key: "4f2a9c1".to_string(),
762///     resumed_runs: vec![Uuid::now_v7()],
763///     sender_run_id: None,
764///     sender_step_id: None,
765///     at: Utc::now(),
766/// };
767/// assert_eq!(payload.resumed_runs.len(), 1);
768/// ```
769#[derive(Debug, Clone, Serialize, Deserialize)]
770#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
771pub struct SignalReceivedEvent {
772    /// Stored signal identifier.
773    pub signal_id: Uuid,
774    /// Signal name.
775    pub name: String,
776    /// Occurrence key.
777    pub key: String,
778    /// Runs whose waiting step the signal resolved.
779    pub resumed_runs: Vec<Uuid>,
780    /// The run that sent the signal with `ctx.send_signal`. `None` for a
781    /// signal sent through the API or the engine directly.
782    #[serde(default, skip_serializing_if = "Option::is_none")]
783    pub sender_run_id: Option<Uuid>,
784    /// The send step of the sending run. `None` when `sender_run_id` is.
785    #[serde(default, skip_serializing_if = "Option::is_none")]
786    pub sender_step_id: Option<Uuid>,
787    /// When the signal was received.
788    pub at: DateTime<Utc>,
789}
790
791/// Payload of the `Event::ScheduleOccurrencesMissed` event.
792///
793/// A cron schedule skipped some of its occurrences: the server was down
794/// longer than the catch-up window allows, the catch-up policy dropped them,
795/// or the previous run was still active under an overlap `skip` policy.
796///
797/// # Examples
798///
799/// ```
800/// use chrono::{TimeDelta, Utc};
801/// use ironflow_engine::notify::ScheduleOccurrencesMissedEvent;
802/// use ironflow_store::models::ScheduleMissReason;
803/// use uuid::Uuid;
804///
805/// let now = Utc::now();
806/// let payload = ScheduleOccurrencesMissedEvent {
807///     schedule_id: Uuid::now_v7(),
808///     workflow_name: "nightly-report".to_string(),
809///     reason: ScheduleMissReason::OutsideWindow,
810///     count: 3,
811///     first: now - TimeDelta::hours(3),
812///     last: now - TimeDelta::hours(1),
813///     at: now,
814/// };
815/// assert_eq!(payload.count, 3);
816/// ```
817#[derive(Debug, Clone, Serialize, Deserialize)]
818#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
819pub struct ScheduleOccurrencesMissedEvent {
820    /// Schedule that missed the occurrences.
821    pub schedule_id: Uuid,
822    /// Workflow the schedule starts.
823    pub workflow_name: String,
824    /// Why the occurrences were not run.
825    pub reason: ScheduleMissReason,
826    /// Number of missed occurrences.
827    pub count: u64,
828    /// Earliest missed occurrence.
829    pub first: DateTime<Utc>,
830    /// Latest missed occurrence.
831    pub last: DateTime<Utc>,
832    /// When the ticker noticed the miss.
833    pub at: DateTime<Utc>,
834}
835
836/// Payload of the `Event::WorkflowPaused` event.
837///
838/// Emitted on every pause request, including the pause of a workflow that
839/// was already paused: the request is a decision of its actor.
840///
841/// # Examples
842///
843/// ```
844/// use chrono::Utc;
845/// use ironflow_engine::notify::WorkflowPausedEvent;
846/// use uuid::Uuid;
847///
848/// let payload = WorkflowPausedEvent {
849///     workflow_name: "deploy".to_string(),
850///     user_id: Some(Uuid::now_v7()),
851///     at: Utc::now(),
852/// };
853/// assert_eq!(payload.workflow_name, "deploy");
854/// ```
855#[derive(Debug, Clone, Serialize, Deserialize)]
856#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
857pub struct WorkflowPausedEvent {
858    /// Paused workflow.
859    pub workflow_name: String,
860    /// ID of the user who paused the workflow. `None` when no user is behind
861    /// the call.
862    #[serde(default)]
863    pub user_id: Option<Uuid>,
864    /// When the workflow was paused.
865    pub at: DateTime<Utc>,
866}
867
868/// Payload of the `Event::WorkflowResumed` event.
869///
870/// Emitted only when the workflow was paused: resuming a workflow that is
871/// not paused changes nothing and emits nothing.
872///
873/// # Examples
874///
875/// ```
876/// use chrono::Utc;
877/// use ironflow_engine::notify::WorkflowResumedEvent;
878/// use uuid::Uuid;
879///
880/// let payload = WorkflowResumedEvent {
881///     workflow_name: "deploy".to_string(),
882///     user_id: Some(Uuid::now_v7()),
883///     at: Utc::now(),
884/// };
885/// assert_eq!(payload.workflow_name, "deploy");
886/// ```
887#[derive(Debug, Clone, Serialize, Deserialize)]
888#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
889pub struct WorkflowResumedEvent {
890    /// Resumed workflow.
891    pub workflow_name: String,
892    /// ID of the user who resumed the workflow. `None` when no user is behind
893    /// the call.
894    #[serde(default)]
895    pub user_id: Option<Uuid>,
896    /// When the workflow was resumed.
897    pub at: DateTime<Utc>,
898}
899
900/// A domain event emitted by the ironflow system.
901///
902/// Covers the full lifecycle: runs, steps, approvals, and authentication.
903/// Subscribers receive these via [`EventPublisher`](super::EventPublisher)
904/// and pattern-match on the variants they care about.
905///
906/// Each variant wraps a dedicated payload struct. The serialized form stays
907/// flat: the `type` discriminant sits next to the payload fields, so
908/// `{"type":"run_created","run_id":...}` round-trips unchanged.
909///
910/// # Examples
911///
912/// ```
913/// use std::collections::HashMap;
914/// use ironflow_engine::notify::{Event, RunStatusChangedEvent};
915/// use ironflow_store::models::RunStatus;
916/// use uuid::Uuid;
917///
918/// let event = Event::RunStatusChanged(RunStatusChangedEvent {
919///     run_id: Uuid::now_v7(),
920///     workflow_name: "deploy".to_string(),
921///     from: RunStatus::Running,
922///     to: RunStatus::Completed,
923///     error: None,
924///     cost_usd: rust_decimal::Decimal::ZERO,
925///     duration_ms: 5000,
926///     labels: HashMap::new(),
927///     at: chrono::Utc::now(),
928/// });
929/// assert_eq!(event.event_type(), "run_status_changed");
930/// ```
931#[derive(Debug, Clone, Serialize, Deserialize)]
932#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
933#[serde(tag = "type", rename_all = "snake_case")]
934pub enum Event {
935    // -- Run lifecycle --
936    /// A new run was created (status: Pending).
937    RunCreated(RunCreatedEvent),
938
939    /// A run changed status.
940    RunStatusChanged(RunStatusChangedEvent),
941
942    /// A run transitioned to [`Failed`](ironflow_store::models::RunStatus::Failed).
943    ///
944    /// This is a convenience event emitted alongside [`RunStatusChanged`](Event::RunStatusChanged)
945    /// when the target status is `Failed`. Subscribe to this instead of
946    /// `RUN_STATUS_CHANGED` when you only care about failures.
947    RunFailed(RunFailedEvent),
948
949    /// A run was stopped because it reached its cumulative cost cap.
950    ///
951    /// Emitted when the engine refuses an agent step that would cross the run's
952    /// `max_cost_usd`. The run transitions to
953    /// [`Cancelled`](ironflow_store::models::RunStatus::Cancelled) and the step
954    /// is never launched, so the reported spend is what the run had already
955    /// consumed.
956    RunBudgetExceeded(RunBudgetExceededEvent),
957
958    /// A manual retry was forced despite a handler version mismatch.
959    ///
960    /// Emitted when a caller passes `force=true` on a retry where the
961    /// handler version differs from the run's recorded version. This is
962    /// an audit event: it means the new code will execute on the old
963    /// payload without the handler explicitly declaring compatibility.
964    RetryForced(RetryForcedEvent),
965
966    // -- Step lifecycle --
967    /// A step completed successfully.
968    StepCompleted(StepCompletedEvent),
969
970    /// A step failed.
971    StepFailed(StepFailedEvent),
972
973    // -- Approval --
974    /// A run is waiting for human approval.
975    ApprovalRequested(ApprovalRequestedEvent),
976
977    /// A run was approved by a human.
978    ApprovalGranted(ApprovalGrantedEvent),
979
980    /// A run was rejected by a human.
981    ApprovalRejected(ApprovalRejectedEvent),
982
983    /// An approval gate missed its SLA deadline and an escalation policy ran.
984    ApprovalEscalated(ApprovalEscalatedEvent),
985
986    // -- Log streaming --
987    /// A log line emitted during step execution.
988    ///
989    /// Pushed by the worker in real time so that SSE clients can stream
990    /// step output as it happens, without waiting for step completion.
991    LogLine(LogLineEvent),
992
993    // -- Authentication --
994    /// A user signed in.
995    UserSignedIn(UserSignedInEvent),
996
997    /// A new user signed up.
998    UserSignedUp(UserSignedUpEvent),
999
1000    /// A user signed out.
1001    UserSignedOut(UserSignedOutEvent),
1002
1003    // -- Provider Accounts --
1004    /// A Provider Account was created, updated, deleted or had its token replaced.
1005    #[serde(rename = "provider_account.updated")]
1006    ProviderAccountUpdated(ProviderAccountUpdatedEvent),
1007
1008    /// New usage windows were recorded for a Provider Account.
1009    #[serde(rename = "provider_account.usage_updated")]
1010    ProviderAccountUsageUpdated(ProviderAccountUsageUpdatedEvent),
1011
1012    // -- Signals --
1013    /// A run started waiting for a signal.
1014    SignalAwaited(SignalAwaitedEvent),
1015
1016    /// A signal was received.
1017    SignalReceived(SignalReceivedEvent),
1018
1019    // -- Schedules --
1020    /// A cron schedule dropped occurrences instead of starting a run for them.
1021    ///
1022    /// Emitted by the schedule ticker once per schedule, tick and reason.
1023    ScheduleOccurrencesMissed(ScheduleOccurrencesMissedEvent),
1024
1025    // -- Workflows --
1026    /// A workflow was paused: its new runs are held until it is resumed.
1027    WorkflowPaused(WorkflowPausedEvent),
1028
1029    /// A paused workflow was resumed.
1030    WorkflowResumed(WorkflowResumedEvent),
1031}
1032
1033impl Event {
1034    /// Event type constant for [`RunCreated`](Event::RunCreated).
1035    pub const RUN_CREATED: &'static str = "run_created";
1036    /// Event type constant for [`RunStatusChanged`](Event::RunStatusChanged).
1037    pub const RUN_STATUS_CHANGED: &'static str = "run_status_changed";
1038    /// Event type constant for [`RunFailed`](Event::RunFailed).
1039    pub const RUN_FAILED: &'static str = "run_failed";
1040    /// Event type constant for [`RunBudgetExceeded`](Event::RunBudgetExceeded).
1041    pub const RUN_BUDGET_EXCEEDED: &'static str = "run_budget_exceeded";
1042    /// Event type constant for [`RetryForced`](Event::RetryForced).
1043    pub const RETRY_FORCED: &'static str = "retry_forced";
1044    /// Event type constant for [`StepCompleted`](Event::StepCompleted).
1045    pub const STEP_COMPLETED: &'static str = "step_completed";
1046    /// Event type constant for [`StepFailed`](Event::StepFailed).
1047    pub const STEP_FAILED: &'static str = "step_failed";
1048    /// Event type constant for [`ApprovalRequested`](Event::ApprovalRequested).
1049    pub const APPROVAL_REQUESTED: &'static str = "approval_requested";
1050    /// Event type constant for [`ApprovalGranted`](Event::ApprovalGranted).
1051    pub const APPROVAL_GRANTED: &'static str = "approval_granted";
1052    /// Event type constant for [`ApprovalRejected`](Event::ApprovalRejected).
1053    pub const APPROVAL_REJECTED: &'static str = "approval_rejected";
1054    /// Event type constant for [`ApprovalEscalated`](Event::ApprovalEscalated).
1055    pub const APPROVAL_ESCALATED: &'static str = "approval_escalated";
1056    /// Event type constant for [`LogLine`](Event::LogLine).
1057    pub const LOG_LINE: &'static str = "log_line";
1058    /// Event type constant for [`UserSignedIn`](Event::UserSignedIn).
1059    pub const USER_SIGNED_IN: &'static str = "user_signed_in";
1060    /// Event type constant for [`UserSignedUp`](Event::UserSignedUp).
1061    pub const USER_SIGNED_UP: &'static str = "user_signed_up";
1062    /// Event type constant for [`UserSignedOut`](Event::UserSignedOut).
1063    pub const USER_SIGNED_OUT: &'static str = "user_signed_out";
1064    /// Event type constant for [`ProviderAccountUpdated`](Event::ProviderAccountUpdated).
1065    pub const PROVIDER_ACCOUNT_UPDATED: &'static str = "provider_account.updated";
1066    /// Event type constant for
1067    /// [`ProviderAccountUsageUpdated`](Event::ProviderAccountUsageUpdated).
1068    pub const PROVIDER_ACCOUNT_USAGE_UPDATED: &'static str = "provider_account.usage_updated";
1069    /// Event type constant for [`SignalAwaited`](Event::SignalAwaited).
1070    pub const SIGNAL_AWAITED: &'static str = "signal_awaited";
1071    /// Event type constant for [`SignalReceived`](Event::SignalReceived).
1072    pub const SIGNAL_RECEIVED: &'static str = "signal_received";
1073    /// Event type constant for
1074    /// [`ScheduleOccurrencesMissed`](Event::ScheduleOccurrencesMissed).
1075    pub const SCHEDULE_OCCURRENCES_MISSED: &'static str = "schedule_occurrences_missed";
1076    /// Event type constant for [`WorkflowPaused`](Event::WorkflowPaused).
1077    pub const WORKFLOW_PAUSED: &'static str = "workflow_paused";
1078    /// Event type constant for [`WorkflowResumed`](Event::WorkflowResumed).
1079    pub const WORKFLOW_RESUMED: &'static str = "workflow_resumed";
1080
1081    /// All event types. Pass this to
1082    /// [`EventPublisher::subscribe`](super::EventPublisher::subscribe) to
1083    /// receive every event.
1084    ///
1085    /// # Examples
1086    ///
1087    /// ```no_run
1088    /// use ironflow_engine::notify::{Event, EventPublisher, WebhookSubscriber};
1089    ///
1090    /// let mut publisher = EventPublisher::new();
1091    /// publisher.subscribe(
1092    ///     WebhookSubscriber::new("https://example.com/all"),
1093    ///     Event::ALL,
1094    /// );
1095    /// ```
1096    pub const ALL: &'static [&'static str] = &[
1097        Self::RUN_CREATED,
1098        Self::RUN_STATUS_CHANGED,
1099        Self::RUN_FAILED,
1100        Self::RUN_BUDGET_EXCEEDED,
1101        Self::STEP_COMPLETED,
1102        Self::STEP_FAILED,
1103        Self::APPROVAL_REQUESTED,
1104        Self::APPROVAL_GRANTED,
1105        Self::APPROVAL_REJECTED,
1106        Self::APPROVAL_ESCALATED,
1107        Self::LOG_LINE,
1108        Self::USER_SIGNED_IN,
1109        Self::USER_SIGNED_UP,
1110        Self::USER_SIGNED_OUT,
1111        Self::RETRY_FORCED,
1112        Self::PROVIDER_ACCOUNT_UPDATED,
1113        Self::PROVIDER_ACCOUNT_USAGE_UPDATED,
1114        Self::SIGNAL_AWAITED,
1115        Self::SIGNAL_RECEIVED,
1116        Self::SCHEDULE_OCCURRENCES_MISSED,
1117        Self::WORKFLOW_PAUSED,
1118        Self::WORKFLOW_RESUMED,
1119    ];
1120
1121    /// Returns the event type as a static string (e.g. `"run_status_changed"`).
1122    ///
1123    /// Useful for filtering and logging without deserializing.
1124    ///
1125    /// # Examples
1126    ///
1127    /// ```
1128    /// use ironflow_engine::notify::{Event, UserSignedInEvent};
1129    /// use uuid::Uuid;
1130    /// use chrono::Utc;
1131    ///
1132    /// let event = Event::UserSignedIn(UserSignedInEvent {
1133    ///     user_id: Uuid::now_v7(),
1134    ///     username: "alice".to_string(),
1135    ///     at: Utc::now(),
1136    /// });
1137    /// assert_eq!(event.event_type(), "user_signed_in");
1138    /// ```
1139    #[deny(unreachable_patterns)]
1140    pub fn event_type(&self) -> &'static str {
1141        match self {
1142            Event::RunCreated(_) => Self::RUN_CREATED,
1143            Event::RunStatusChanged(_) => Self::RUN_STATUS_CHANGED,
1144            Event::RunFailed(_) => Self::RUN_FAILED,
1145            Event::RunBudgetExceeded(_) => Self::RUN_BUDGET_EXCEEDED,
1146            Event::RetryForced(_) => Self::RETRY_FORCED,
1147            Event::StepCompleted(_) => Self::STEP_COMPLETED,
1148            Event::StepFailed(_) => Self::STEP_FAILED,
1149            Event::ApprovalRequested(_) => Self::APPROVAL_REQUESTED,
1150            Event::ApprovalGranted(_) => Self::APPROVAL_GRANTED,
1151            Event::ApprovalRejected(_) => Self::APPROVAL_REJECTED,
1152            Event::ApprovalEscalated(_) => Self::APPROVAL_ESCALATED,
1153            Event::LogLine(_) => Self::LOG_LINE,
1154            Event::UserSignedIn(_) => Self::USER_SIGNED_IN,
1155            Event::UserSignedUp(_) => Self::USER_SIGNED_UP,
1156            Event::UserSignedOut(_) => Self::USER_SIGNED_OUT,
1157            Event::ProviderAccountUpdated(_) => Self::PROVIDER_ACCOUNT_UPDATED,
1158            Event::ProviderAccountUsageUpdated(_) => Self::PROVIDER_ACCOUNT_USAGE_UPDATED,
1159            Event::SignalAwaited(_) => Self::SIGNAL_AWAITED,
1160            Event::SignalReceived(_) => Self::SIGNAL_RECEIVED,
1161            Event::ScheduleOccurrencesMissed(_) => Self::SCHEDULE_OCCURRENCES_MISSED,
1162            Event::WorkflowPaused(_) => Self::WORKFLOW_PAUSED,
1163            Event::WorkflowResumed(_) => Self::WORKFLOW_RESUMED,
1164        }
1165    }
1166
1167    /// Returns the run this event belongs to, if any.
1168    ///
1169    /// Auth events ([`UserSignedIn`](Event::UserSignedIn),
1170    /// [`UserSignedUp`](Event::UserSignedUp),
1171    /// [`UserSignedOut`](Event::UserSignedOut)) are not tied to a run and
1172    /// return `None`. [`SignalReceived`](Event::SignalReceived) returns the
1173    /// run that sent it, `None` when it came from outside a workflow.
1174    ///
1175    /// # Examples
1176    ///
1177    /// ```
1178    /// use ironflow_engine::notify::{Event, RunCreatedEvent};
1179    /// use uuid::Uuid;
1180    /// use chrono::Utc;
1181    ///
1182    /// let run_id = Uuid::now_v7();
1183    /// let event = Event::RunCreated(RunCreatedEvent {
1184    ///     run_id,
1185    ///     workflow_name: "deploy".to_string(),
1186    ///     at: Utc::now(),
1187    /// });
1188    /// assert_eq!(event.run_id(), Some(run_id));
1189    /// ```
1190    #[deny(unreachable_patterns)]
1191    pub fn run_id(&self) -> Option<Uuid> {
1192        match self {
1193            Event::RunCreated(e) => Some(e.run_id),
1194            Event::RunStatusChanged(e) => Some(e.run_id),
1195            Event::RunFailed(e) => Some(e.run_id),
1196            Event::RunBudgetExceeded(e) => Some(e.run_id),
1197            Event::RetryForced(e) => Some(e.run_id),
1198            Event::StepCompleted(e) => Some(e.run_id),
1199            Event::StepFailed(e) => Some(e.run_id),
1200            Event::ApprovalRequested(e) => Some(e.run_id),
1201            Event::ApprovalGranted(e) => Some(e.run_id),
1202            Event::ApprovalRejected(e) => Some(e.run_id),
1203            Event::ApprovalEscalated(e) => Some(e.run_id),
1204            Event::LogLine(e) => Some(e.run_id),
1205            Event::SignalAwaited(e) => Some(e.run_id),
1206            Event::SignalReceived(e) => e.sender_run_id,
1207            Event::UserSignedIn(_)
1208            | Event::UserSignedUp(_)
1209            | Event::UserSignedOut(_)
1210            | Event::ProviderAccountUpdated(_)
1211            | Event::ProviderAccountUsageUpdated(_)
1212            | Event::ScheduleOccurrencesMissed(_)
1213            | Event::WorkflowPaused(_)
1214            | Event::WorkflowResumed(_) => None,
1215        }
1216    }
1217
1218    /// Returns the step this event belongs to, if any.
1219    ///
1220    /// Only [`StepCompleted`](Event::StepCompleted),
1221    /// [`StepFailed`](Event::StepFailed),
1222    /// [`ApprovalRequested`](Event::ApprovalRequested) and
1223    /// [`ApprovalEscalated`](Event::ApprovalEscalated) always carry a step
1224    /// identifier; [`ApprovalGranted`](Event::ApprovalGranted) and
1225    /// [`ApprovalRejected`](Event::ApprovalRejected) carry one when it was
1226    /// recorded, [`SignalReceived`](Event::SignalReceived) the step that sent
1227    /// it; every other variant returns `None`.
1228    ///
1229    /// # Examples
1230    ///
1231    /// ```
1232    /// use ironflow_engine::notify::{Event, StepFailedEvent};
1233    /// use ironflow_store::models::StepKind;
1234    /// use uuid::Uuid;
1235    /// use chrono::Utc;
1236    ///
1237    /// let step_id = Uuid::now_v7();
1238    /// let event = Event::StepFailed(StepFailedEvent {
1239    ///     run_id: Uuid::now_v7(),
1240    ///     step_id,
1241    ///     step_name: "build".to_string(),
1242    ///     kind: StepKind::Shell,
1243    ///     error: "exit code 1".to_string(),
1244    ///     at: Utc::now(),
1245    /// });
1246    /// assert_eq!(event.step_id(), Some(step_id));
1247    /// ```
1248    #[deny(unreachable_patterns)]
1249    pub fn step_id(&self) -> Option<Uuid> {
1250        match self {
1251            Event::StepCompleted(e) => Some(e.step_id),
1252            Event::StepFailed(e) => Some(e.step_id),
1253            Event::ApprovalRequested(e) => Some(e.step_id),
1254            Event::ApprovalEscalated(e) => Some(e.step_id),
1255            Event::ApprovalGranted(e) => e.step_id,
1256            Event::ApprovalRejected(e) => e.step_id,
1257            Event::SignalAwaited(e) => Some(e.step_id),
1258            Event::SignalReceived(e) => e.sender_step_id,
1259            Event::RunCreated(_)
1260            | Event::RunStatusChanged(_)
1261            | Event::RunFailed(_)
1262            | Event::RunBudgetExceeded(_)
1263            | Event::RetryForced(_)
1264            | Event::LogLine(_)
1265            | Event::UserSignedIn(_)
1266            | Event::UserSignedUp(_)
1267            | Event::UserSignedOut(_)
1268            | Event::ProviderAccountUpdated(_)
1269            | Event::ProviderAccountUsageUpdated(_)
1270            | Event::ScheduleOccurrencesMissed(_)
1271            | Event::WorkflowPaused(_)
1272            | Event::WorkflowResumed(_) => None,
1273        }
1274    }
1275
1276    /// Returns the user this event belongs to, if any.
1277    ///
1278    /// Only the auth events ([`UserSignedIn`](Event::UserSignedIn),
1279    /// [`UserSignedUp`](Event::UserSignedUp),
1280    /// [`UserSignedOut`](Event::UserSignedOut)) and the decisions and pauses
1281    /// ([`ApprovalGranted`](Event::ApprovalGranted),
1282    /// [`ApprovalRejected`](Event::ApprovalRejected),
1283    /// [`WorkflowPaused`](Event::WorkflowPaused),
1284    /// [`WorkflowResumed`](Event::WorkflowResumed)) carry a user identifier.
1285    /// A decision without a human actor (timeout escalation) returns `None`,
1286    /// like every other variant.
1287    ///
1288    /// # Examples
1289    ///
1290    /// ```
1291    /// use ironflow_engine::notify::{Event, UserSignedInEvent};
1292    /// use uuid::Uuid;
1293    /// use chrono::Utc;
1294    ///
1295    /// let user_id = Uuid::now_v7();
1296    /// let event = Event::UserSignedIn(UserSignedInEvent {
1297    ///     user_id,
1298    ///     username: "alice".to_string(),
1299    ///     at: Utc::now(),
1300    /// });
1301    /// assert_eq!(event.user_id(), Some(user_id));
1302    /// ```
1303    #[deny(unreachable_patterns)]
1304    pub fn user_id(&self) -> Option<Uuid> {
1305        match self {
1306            Event::UserSignedIn(e) => Some(e.user_id),
1307            Event::UserSignedUp(e) => Some(e.user_id),
1308            Event::UserSignedOut(e) => Some(e.user_id),
1309            Event::ApprovalGranted(e) => e.user_id,
1310            Event::ApprovalRejected(e) => e.user_id,
1311            Event::WorkflowPaused(e) => e.user_id,
1312            Event::WorkflowResumed(e) => e.user_id,
1313            Event::RunCreated(_)
1314            | Event::RunStatusChanged(_)
1315            | Event::RunFailed(_)
1316            | Event::RunBudgetExceeded(_)
1317            | Event::RetryForced(_)
1318            | Event::StepCompleted(_)
1319            | Event::StepFailed(_)
1320            | Event::ApprovalRequested(_)
1321            | Event::ApprovalEscalated(_)
1322            | Event::LogLine(_)
1323            | Event::ProviderAccountUpdated(_)
1324            | Event::ProviderAccountUsageUpdated(_)
1325            | Event::SignalAwaited(_)
1326            | Event::SignalReceived(_)
1327            | Event::ScheduleOccurrencesMissed(_) => None,
1328        }
1329    }
1330}
1331
1332#[cfg(test)]
1333mod tests {
1334    use serde_json::{from_value, json};
1335
1336    use super::*;
1337
1338    #[test]
1339    fn run_status_changed_serde_roundtrip() {
1340        let event = Event::RunStatusChanged(RunStatusChangedEvent {
1341            run_id: Uuid::now_v7(),
1342            workflow_name: "deploy".to_string(),
1343            from: RunStatus::Running,
1344            to: RunStatus::Completed,
1345            error: None,
1346            cost_usd: Decimal::new(42, 2),
1347            duration_ms: 5000,
1348            labels: HashMap::new(),
1349            at: Utc::now(),
1350        });
1351
1352        let json = serde_json::to_string(&event).expect("serialize");
1353        let back: Event = serde_json::from_str(&json).expect("deserialize");
1354
1355        assert_eq!(back.event_type(), "run_status_changed");
1356        assert!(json.contains("\"type\":\"run_status_changed\""));
1357    }
1358
1359    #[test]
1360    fn run_failed_serde_roundtrip() {
1361        let event = Event::RunFailed(RunFailedEvent {
1362            run_id: Uuid::now_v7(),
1363            workflow_name: "deploy".to_string(),
1364            error: Some("step crashed".to_string()),
1365            cost_usd: Decimal::new(10, 2),
1366            duration_ms: 3000,
1367            labels: HashMap::new(),
1368            at: Utc::now(),
1369        });
1370
1371        let json = serde_json::to_string(&event).expect("serialize");
1372        let back: Event = serde_json::from_str(&json).expect("deserialize");
1373
1374        assert_eq!(back.event_type(), "run_failed");
1375        assert!(json.contains("\"type\":\"run_failed\""));
1376        assert!(json.contains("step crashed"));
1377    }
1378
1379    #[test]
1380    fn run_budget_exceeded_serde_roundtrip() {
1381        let event = Event::RunBudgetExceeded(RunBudgetExceededEvent {
1382            run_id: Uuid::now_v7(),
1383            workflow_name: "deploy".to_string(),
1384            limit_usd: Decimal::new(200, 2),
1385            spent_usd: Decimal::new(180, 2),
1386            step_budget_usd: Decimal::new(50, 2),
1387            at: Utc::now(),
1388        });
1389
1390        let json = serde_json::to_string(&event).expect("serialize");
1391        let back: Event = serde_json::from_str(&json).expect("deserialize");
1392
1393        assert_eq!(back.event_type(), "run_budget_exceeded");
1394        assert!(json.contains("\"type\":\"run_budget_exceeded\""));
1395        assert!(json.contains("limit_usd"));
1396        assert!(json.contains("step_budget_usd"));
1397    }
1398
1399    #[test]
1400    fn all_contains_run_budget_exceeded() {
1401        assert!(Event::ALL.contains(&Event::RUN_BUDGET_EXCEEDED));
1402    }
1403
1404    #[test]
1405    fn user_signed_in_serde_roundtrip() {
1406        let event = Event::UserSignedIn(UserSignedInEvent {
1407            user_id: Uuid::now_v7(),
1408            username: "alice".to_string(),
1409            at: Utc::now(),
1410        });
1411
1412        let json = serde_json::to_string(&event).expect("serialize");
1413        let back: Event = serde_json::from_str(&json).expect("deserialize");
1414
1415        assert_eq!(back.event_type(), "user_signed_in");
1416        assert!(json.contains("alice"));
1417    }
1418
1419    #[test]
1420    fn step_failed_serde_roundtrip() {
1421        let event = Event::StepFailed(StepFailedEvent {
1422            run_id: Uuid::now_v7(),
1423            step_id: Uuid::now_v7(),
1424            step_name: "build".to_string(),
1425            kind: StepKind::Shell,
1426            error: "exit code 1".to_string(),
1427            at: Utc::now(),
1428        });
1429
1430        let json = serde_json::to_string(&event).expect("serialize");
1431        let back: Event = serde_json::from_str(&json).expect("deserialize");
1432
1433        assert_eq!(back.event_type(), "step_failed");
1434    }
1435
1436    #[test]
1437    fn legacy_approval_granted_defaults_to_a_single_vote() {
1438        let raw = r#"{"type":"approval_granted","run_id":"01890000-0000-7000-8000-000000000000","approved_by":"alice","at":"2026-01-01T00:00:00Z"}"#;
1439        let event: Event = serde_json::from_str(raw).expect("deserialize");
1440        let Event::ApprovalGranted(event) = event else {
1441            panic!("expected approval_granted");
1442        };
1443
1444        assert_eq!(event.step_id, None);
1445        assert_eq!(event.approvals_received, 1);
1446        assert_eq!(event.approvals_required, 1);
1447        assert!(event.requirement.is_none());
1448    }
1449
1450    #[test]
1451    fn legacy_approval_requested_and_rejected_have_no_requirement() {
1452        let raw = r#"{"type":"approval_requested","run_id":"01890000-0000-7000-8000-000000000000","step_id":"01890000-0000-7000-8000-000000000001","message":"ok?","at":"2026-01-01T00:00:00Z"}"#;
1453        let requested: Event = serde_json::from_str(raw).expect("deserialize");
1454        let Event::ApprovalRequested(requested) = requested else {
1455            panic!("expected approval_requested");
1456        };
1457        assert!(requested.requirement.is_none());
1458
1459        let raw = r#"{"type":"approval_rejected","run_id":"01890000-0000-7000-8000-000000000000","rejected_by":"bob","at":"2026-01-01T00:00:00Z"}"#;
1460        let rejected: Event = serde_json::from_str(raw).expect("deserialize");
1461        let Event::ApprovalRejected(rejected) = rejected else {
1462            panic!("expected approval_rejected");
1463        };
1464        assert_eq!(rejected.step_id, None);
1465        assert!(rejected.requirement.is_none());
1466    }
1467
1468    #[test]
1469    fn approval_granted_roundtrips_the_vote_counts() {
1470        let requirement = ApprovalRequirement {
1471            reason: Some("amount > 10k".to_string()),
1472            required_approvers: 2,
1473            approver_groups: vec!["finance".to_string()],
1474        };
1475        let event = Event::ApprovalGranted(ApprovalGrantedEvent {
1476            run_id: Uuid::now_v7(),
1477            step_id: Some(Uuid::now_v7()),
1478            approved_by: "alice".to_string(),
1479            user_id: None,
1480            approvals_received: 1,
1481            approvals_required: 2,
1482            requirement: Some(requirement.clone()),
1483            at: Utc::now(),
1484        });
1485
1486        let json = serde_json::to_string(&event).expect("serialize");
1487        let back: Event = serde_json::from_str(&json).expect("deserialize");
1488        let Event::ApprovalGranted(back) = back else {
1489            panic!("expected approval_granted");
1490        };
1491        assert_eq!(back.approvals_received, 1);
1492        assert_eq!(back.approvals_required, 2);
1493        assert_eq!(back.requirement, Some(requirement));
1494    }
1495
1496    #[test]
1497    fn approval_requested_serde_roundtrip() {
1498        let event = Event::ApprovalRequested(ApprovalRequestedEvent {
1499            run_id: Uuid::now_v7(),
1500            step_id: Uuid::now_v7(),
1501            message: "Deploy to prod?".to_string(),
1502            requirement: None,
1503            at: Utc::now(),
1504        });
1505
1506        let json = serde_json::to_string(&event).expect("serialize");
1507        assert!(json.contains("approval_requested"));
1508    }
1509
1510    #[test]
1511    fn log_line_serde_roundtrip() {
1512        let event = Event::LogLine(LogLineEvent {
1513            id: Uuid::now_v7(),
1514            run_id: Uuid::now_v7(),
1515            step_id: Uuid::now_v7(),
1516            step_name: "build".to_string(),
1517            stream: LogStream::Stdout,
1518            line: "Compiling ironflow v0.1.0".to_string(),
1519            at: Utc::now(),
1520        });
1521
1522        let json = serde_json::to_string(&event).expect("serialize");
1523        let back: Event = serde_json::from_str(&json).expect("deserialize");
1524
1525        assert_eq!(back.event_type(), "log_line");
1526        assert!(json.contains("\"type\":\"log_line\""));
1527        assert!(json.contains("Compiling ironflow"));
1528    }
1529
1530    /// The pre-refactor wire format used flat inline-struct variants. Newtype
1531    /// variants over named-field payloads produce and accept the exact same
1532    /// JSON, so audit rows and in-flight payloads written before the refactor
1533    /// still deserialize. No data migration is required.
1534    #[test]
1535    fn legacy_flat_json_deserializes_into_typed_payload() {
1536        let run_id: Uuid = "01890000-0000-7000-8000-000000000000"
1537            .parse()
1538            .expect("valid uuid");
1539
1540        let raw = r#"{"type":"run_created","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","at":"2026-01-01T00:00:00Z"}"#;
1541        let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1542        match event {
1543            Event::RunCreated(e) => {
1544                assert_eq!(e.run_id, run_id);
1545                assert_eq!(e.workflow_name, "deploy");
1546            }
1547            other => panic!("expected RunCreated, got {other:?}"),
1548        }
1549
1550        let raw = r#"{"type":"run_status_changed","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","from":"running","to":"completed","error":null,"cost_usd":0.5,"duration_ms":5000,"labels":{"env":"prod"},"at":"2026-01-01T00:00:00Z"}"#;
1551        let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1552        match event {
1553            Event::RunStatusChanged(e) => {
1554                assert_eq!(e.from, RunStatus::Running);
1555                assert_eq!(e.to, RunStatus::Completed);
1556                assert_eq!(e.cost_usd, Decimal::new(5, 1));
1557                assert_eq!(e.duration_ms, 5000);
1558                assert_eq!(e.labels.get("env").map(String::as_str), Some("prod"));
1559            }
1560            other => panic!("expected RunStatusChanged, got {other:?}"),
1561        }
1562
1563        // `labels` predates no payload: omitting it must still work via `#[serde(default)]`.
1564        let raw = r#"{"type":"run_status_changed","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","from":"running","to":"failed","error":"boom","cost_usd":0,"duration_ms":0,"at":"2026-01-01T00:00:00Z"}"#;
1565        let event: Event = serde_json::from_str(raw).expect("missing labels must default");
1566        match event {
1567            Event::RunStatusChanged(e) => {
1568                assert!(e.labels.is_empty());
1569                assert_eq!(e.error.as_deref(), Some("boom"));
1570            }
1571            other => panic!("expected RunStatusChanged, got {other:?}"),
1572        }
1573
1574        let raw = r#"{"type":"run_failed","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","error":"boom","cost_usd":0.25,"duration_ms":3000,"at":"2026-01-01T00:00:00Z"}"#;
1575        let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1576        match event {
1577            Event::RunFailed(e) => {
1578                assert_eq!(e.error.as_deref(), Some("boom"));
1579                assert!(e.labels.is_empty());
1580            }
1581            other => panic!("expected RunFailed, got {other:?}"),
1582        }
1583
1584        let raw = r#"{"type":"step_failed","run_id":"01890000-0000-7000-8000-000000000000","step_id":"01890000-0000-7000-8000-000000000001","step_name":"build","kind":"shell","error":"exit code 1","at":"2026-01-01T00:00:00Z"}"#;
1585        let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1586        match event {
1587            Event::StepFailed(e) => {
1588                assert_eq!(e.kind, StepKind::Shell);
1589                assert_eq!(e.error, "exit code 1");
1590            }
1591            other => panic!("expected StepFailed, got {other:?}"),
1592        }
1593
1594        let raw = r#"{"type":"log_line","run_id":"01890000-0000-7000-8000-000000000000","step_id":"01890000-0000-7000-8000-000000000001","step_name":"build","stream":"stdout","line":"hello","at":"2026-01-01T00:00:00Z"}"#;
1595        let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1596        match event {
1597            Event::LogLine(e) => {
1598                assert_eq!(e.stream, LogStream::Stdout);
1599                assert_eq!(e.line, "hello");
1600                // A payload predating the `id` field degrades to the nil UUID
1601                // rather than failing to deserialize.
1602                assert_eq!(e.id, Uuid::nil());
1603            }
1604            other => panic!("expected LogLine, got {other:?}"),
1605        }
1606
1607        let raw = r#"{"type":"user_signed_in","user_id":"01890000-0000-7000-8000-000000000000","username":"alice","at":"2026-01-01T00:00:00Z"}"#;
1608        let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1609        match event {
1610            Event::UserSignedIn(e) => assert_eq!(e.username, "alice"),
1611            other => panic!("expected UserSignedIn, got {other:?}"),
1612        }
1613    }
1614
1615    /// Guards the internally-tagged representation: payload fields must stay
1616    /// siblings of `type`, never nested under a variant key.
1617    #[test]
1618    fn serialized_event_is_flat_with_type_tag() {
1619        let run_id = Uuid::now_v7();
1620        let event = Event::RunCreated(RunCreatedEvent {
1621            run_id,
1622            workflow_name: "deploy".to_string(),
1623            at: Utc::now(),
1624        });
1625
1626        let value: serde_json::Value = serde_json::to_value(&event).expect("serialize");
1627        let object = value.as_object().expect("event serializes to an object");
1628
1629        assert_eq!(
1630            object.get("type").and_then(|v| v.as_str()),
1631            Some("run_created")
1632        );
1633        assert_eq!(
1634            object.get("workflow_name").and_then(|v| v.as_str()),
1635            Some("deploy")
1636        );
1637        assert_eq!(
1638            object.get("run_id").and_then(|v| v.as_str()),
1639            Some(run_id.to_string().as_str())
1640        );
1641        assert!(object.contains_key("at"));
1642        assert_eq!(object.len(), 4, "no nesting: {object:?}");
1643        assert!(!object.contains_key("RunCreated"));
1644    }
1645
1646    #[test]
1647    fn run_id_returns_some_for_run_events() {
1648        let run_id = Uuid::now_v7();
1649        let now = Utc::now();
1650
1651        let events = vec![
1652            Event::RunCreated(RunCreatedEvent {
1653                run_id,
1654                workflow_name: "w".to_string(),
1655                at: now,
1656            }),
1657            Event::RunStatusChanged(RunStatusChangedEvent {
1658                run_id,
1659                workflow_name: "w".to_string(),
1660                from: RunStatus::Pending,
1661                to: RunStatus::Running,
1662                error: None,
1663                cost_usd: Decimal::ZERO,
1664                duration_ms: 0,
1665                labels: HashMap::new(),
1666                at: now,
1667            }),
1668            Event::RunFailed(RunFailedEvent {
1669                run_id,
1670                workflow_name: "w".to_string(),
1671                error: None,
1672                cost_usd: Decimal::ZERO,
1673                duration_ms: 0,
1674                labels: HashMap::new(),
1675                at: now,
1676            }),
1677            Event::RunBudgetExceeded(RunBudgetExceededEvent {
1678                run_id,
1679                workflow_name: "w".to_string(),
1680                limit_usd: Decimal::ZERO,
1681                spent_usd: Decimal::ZERO,
1682                step_budget_usd: Decimal::ZERO,
1683                at: now,
1684            }),
1685            Event::RetryForced(RetryForcedEvent {
1686                run_id,
1687                workflow_name: "w".to_string(),
1688                original_version: "1".to_string(),
1689                current_version: "2".to_string(),
1690                at: now,
1691            }),
1692            Event::StepCompleted(StepCompletedEvent {
1693                run_id,
1694                step_id: Uuid::now_v7(),
1695                step_name: "s".to_string(),
1696                kind: StepKind::Shell,
1697                duration_ms: 0,
1698                cost_usd: Decimal::ZERO,
1699                at: now,
1700            }),
1701            Event::StepFailed(StepFailedEvent {
1702                run_id,
1703                step_id: Uuid::now_v7(),
1704                step_name: "s".to_string(),
1705                kind: StepKind::Shell,
1706                error: "e".to_string(),
1707                at: now,
1708            }),
1709            Event::ApprovalRequested(ApprovalRequestedEvent {
1710                run_id,
1711                step_id: Uuid::now_v7(),
1712                message: "ok?".to_string(),
1713                requirement: None,
1714                at: now,
1715            }),
1716            Event::ApprovalGranted(ApprovalGrantedEvent {
1717                run_id,
1718                step_id: None,
1719                approved_by: "alice".to_string(),
1720                user_id: None,
1721                approvals_received: 1,
1722                approvals_required: 1,
1723                requirement: None,
1724                at: now,
1725            }),
1726            Event::ApprovalRejected(ApprovalRejectedEvent {
1727                run_id,
1728                step_id: None,
1729                rejected_by: "bob".to_string(),
1730                user_id: None,
1731                reason: None,
1732                requirement: None,
1733                at: now,
1734            }),
1735            Event::LogLine(LogLineEvent {
1736                id: Uuid::now_v7(),
1737                run_id,
1738                step_id: Uuid::now_v7(),
1739                step_name: "s".to_string(),
1740                stream: LogStream::Stdout,
1741                line: "l".to_string(),
1742                at: now,
1743            }),
1744        ];
1745
1746        for event in &events {
1747            assert_eq!(
1748                event.run_id(),
1749                Some(run_id),
1750                "{} should carry a run_id",
1751                event.event_type()
1752            );
1753        }
1754    }
1755
1756    #[test]
1757    fn run_id_returns_none_for_auth_events() {
1758        let user_id = Uuid::now_v7();
1759        let now = Utc::now();
1760
1761        let events = vec![
1762            Event::UserSignedIn(UserSignedInEvent {
1763                user_id,
1764                username: "alice".to_string(),
1765                at: now,
1766            }),
1767            Event::UserSignedUp(UserSignedUpEvent {
1768                user_id,
1769                username: "alice".to_string(),
1770                at: now,
1771            }),
1772            Event::UserSignedOut(UserSignedOutEvent { user_id, at: now }),
1773        ];
1774
1775        for event in &events {
1776            assert_eq!(event.run_id(), None, "{} has no run", event.event_type());
1777        }
1778    }
1779
1780    #[test]
1781    fn step_id_returns_some_only_for_step_events() {
1782        let step_id = Uuid::now_v7();
1783        let run_id = Uuid::now_v7();
1784        let now = Utc::now();
1785
1786        let with_step = vec![
1787            Event::StepCompleted(StepCompletedEvent {
1788                run_id,
1789                step_id,
1790                step_name: "s".to_string(),
1791                kind: StepKind::Shell,
1792                duration_ms: 0,
1793                cost_usd: Decimal::ZERO,
1794                at: now,
1795            }),
1796            Event::StepFailed(StepFailedEvent {
1797                run_id,
1798                step_id,
1799                step_name: "s".to_string(),
1800                kind: StepKind::Shell,
1801                error: "e".to_string(),
1802                at: now,
1803            }),
1804            Event::ApprovalRequested(ApprovalRequestedEvent {
1805                run_id,
1806                step_id,
1807                message: "ok?".to_string(),
1808                requirement: None,
1809                at: now,
1810            }),
1811            Event::ApprovalGranted(ApprovalGrantedEvent {
1812                run_id,
1813                step_id: Some(step_id),
1814                approved_by: "alice".to_string(),
1815                user_id: None,
1816                approvals_received: 1,
1817                approvals_required: 2,
1818                requirement: None,
1819                at: now,
1820            }),
1821            Event::ApprovalRejected(ApprovalRejectedEvent {
1822                run_id,
1823                step_id: Some(step_id),
1824                rejected_by: "bob".to_string(),
1825                user_id: None,
1826                reason: None,
1827                requirement: None,
1828                at: now,
1829            }),
1830        ];
1831
1832        for event in &with_step {
1833            assert_eq!(
1834                event.step_id(),
1835                Some(step_id),
1836                "{} should carry a step_id",
1837                event.event_type()
1838            );
1839        }
1840
1841        let without_step = vec![
1842            Event::RunCreated(RunCreatedEvent {
1843                run_id,
1844                workflow_name: "w".to_string(),
1845                at: now,
1846            }),
1847            // A granted event recorded before step ids were carried.
1848            Event::ApprovalGranted(ApprovalGrantedEvent {
1849                run_id,
1850                step_id: None,
1851                approved_by: "alice".to_string(),
1852                user_id: None,
1853                approvals_received: 1,
1854                approvals_required: 1,
1855                requirement: None,
1856                at: now,
1857            }),
1858            // LogLine carries a step_id field but is reported as a run-level
1859            // stream event, matching the pre-refactor behaviour.
1860            Event::LogLine(LogLineEvent {
1861                id: Uuid::now_v7(),
1862                run_id,
1863                step_id,
1864                step_name: "s".to_string(),
1865                stream: LogStream::Stdout,
1866                line: "l".to_string(),
1867                at: now,
1868            }),
1869            Event::UserSignedOut(UserSignedOutEvent {
1870                user_id: Uuid::now_v7(),
1871                at: now,
1872            }),
1873        ];
1874
1875        for event in &without_step {
1876            assert_eq!(
1877                event.step_id(),
1878                None,
1879                "{} should not carry a step_id",
1880                event.event_type()
1881            );
1882        }
1883    }
1884
1885    #[test]
1886    fn user_id_returns_the_actor_of_auth_events_decisions_and_pauses() {
1887        let user_id = Uuid::now_v7();
1888        let run_id = Uuid::now_v7();
1889        let now = Utc::now();
1890
1891        let auth = vec![
1892            Event::UserSignedIn(UserSignedInEvent {
1893                user_id,
1894                username: "alice".to_string(),
1895                at: now,
1896            }),
1897            Event::UserSignedUp(UserSignedUpEvent {
1898                user_id,
1899                username: "alice".to_string(),
1900                at: now,
1901            }),
1902            Event::UserSignedOut(UserSignedOutEvent { user_id, at: now }),
1903            Event::ApprovalGranted(ApprovalGrantedEvent {
1904                run_id,
1905                step_id: None,
1906                approved_by: "alice".to_string(),
1907                user_id: Some(user_id),
1908                approvals_received: 1,
1909                approvals_required: 1,
1910                requirement: None,
1911                at: now,
1912            }),
1913            Event::ApprovalRejected(ApprovalRejectedEvent {
1914                run_id,
1915                step_id: None,
1916                rejected_by: "bob".to_string(),
1917                user_id: Some(user_id),
1918                reason: None,
1919                requirement: None,
1920                at: now,
1921            }),
1922            Event::WorkflowPaused(WorkflowPausedEvent {
1923                workflow_name: "w".to_string(),
1924                user_id: Some(user_id),
1925                at: now,
1926            }),
1927            Event::WorkflowResumed(WorkflowResumedEvent {
1928                workflow_name: "w".to_string(),
1929                user_id: Some(user_id),
1930                at: now,
1931            }),
1932        ];
1933
1934        for event in &auth {
1935            assert_eq!(
1936                event.user_id(),
1937                Some(user_id),
1938                "{} should carry a user_id",
1939                event.event_type()
1940            );
1941        }
1942
1943        let non_auth = vec![
1944            Event::RunCreated(RunCreatedEvent {
1945                run_id,
1946                workflow_name: "w".to_string(),
1947                at: now,
1948            }),
1949            Event::StepFailed(StepFailedEvent {
1950                run_id,
1951                step_id: Uuid::now_v7(),
1952                step_name: "s".to_string(),
1953                kind: StepKind::Shell,
1954                error: "e".to_string(),
1955                at: now,
1956            }),
1957            // A system decision (timeout escalation) has no human actor.
1958            Event::ApprovalRejected(ApprovalRejectedEvent {
1959                run_id,
1960                step_id: None,
1961                rejected_by: "system:timeout".to_string(),
1962                user_id: None,
1963                reason: None,
1964                requirement: None,
1965                at: now,
1966            }),
1967            Event::WorkflowPaused(WorkflowPausedEvent {
1968                workflow_name: "w".to_string(),
1969                user_id: None,
1970                at: now,
1971            }),
1972        ];
1973
1974        for event in &non_auth {
1975            assert_eq!(
1976                event.user_id(),
1977                None,
1978                "{} should not carry a user_id",
1979                event.event_type()
1980            );
1981        }
1982    }
1983
1984    #[test]
1985    fn event_type_all_variants() {
1986        let id = Uuid::now_v7();
1987        let now = Utc::now();
1988
1989        let cases: Vec<(Event, &str)> = vec![
1990            (
1991                Event::RunCreated(RunCreatedEvent {
1992                    run_id: id,
1993                    workflow_name: "w".to_string(),
1994                    at: now,
1995                }),
1996                "run_created",
1997            ),
1998            (
1999                Event::RunStatusChanged(RunStatusChangedEvent {
2000                    run_id: id,
2001                    workflow_name: "w".to_string(),
2002                    from: RunStatus::Pending,
2003                    to: RunStatus::Running,
2004                    error: None,
2005                    cost_usd: Decimal::ZERO,
2006                    duration_ms: 0,
2007                    labels: HashMap::new(),
2008                    at: now,
2009                }),
2010                "run_status_changed",
2011            ),
2012            (
2013                Event::RunFailed(RunFailedEvent {
2014                    run_id: id,
2015                    workflow_name: "w".to_string(),
2016                    error: Some("boom".to_string()),
2017                    cost_usd: Decimal::ZERO,
2018                    duration_ms: 0,
2019                    labels: HashMap::new(),
2020                    at: now,
2021                }),
2022                "run_failed",
2023            ),
2024            (
2025                Event::RunBudgetExceeded(RunBudgetExceededEvent {
2026                    run_id: id,
2027                    workflow_name: "w".to_string(),
2028                    limit_usd: Decimal::new(200, 2),
2029                    spent_usd: Decimal::new(180, 2),
2030                    step_budget_usd: Decimal::new(50, 2),
2031                    at: now,
2032                }),
2033                "run_budget_exceeded",
2034            ),
2035            (
2036                Event::RetryForced(RetryForcedEvent {
2037                    run_id: id,
2038                    workflow_name: "w".to_string(),
2039                    original_version: "1".to_string(),
2040                    current_version: "2".to_string(),
2041                    at: now,
2042                }),
2043                "retry_forced",
2044            ),
2045            (
2046                Event::StepCompleted(StepCompletedEvent {
2047                    run_id: id,
2048                    step_id: id,
2049                    step_name: "s".to_string(),
2050                    kind: StepKind::Shell,
2051                    duration_ms: 0,
2052                    cost_usd: Decimal::ZERO,
2053                    at: now,
2054                }),
2055                "step_completed",
2056            ),
2057            (
2058                Event::StepFailed(StepFailedEvent {
2059                    run_id: id,
2060                    step_id: id,
2061                    step_name: "s".to_string(),
2062                    kind: StepKind::Shell,
2063                    error: "err".to_string(),
2064                    at: now,
2065                }),
2066                "step_failed",
2067            ),
2068            (
2069                Event::ApprovalRequested(ApprovalRequestedEvent {
2070                    run_id: id,
2071                    step_id: id,
2072                    message: "ok?".to_string(),
2073                    requirement: None,
2074                    at: now,
2075                }),
2076                "approval_requested",
2077            ),
2078            (
2079                Event::ApprovalGranted(ApprovalGrantedEvent {
2080                    run_id: id,
2081                    step_id: Some(id),
2082                    approved_by: "alice".to_string(),
2083                    user_id: None,
2084                    approvals_received: 1,
2085                    approvals_required: 1,
2086                    requirement: None,
2087                    at: now,
2088                }),
2089                "approval_granted",
2090            ),
2091            (
2092                Event::ApprovalRejected(ApprovalRejectedEvent {
2093                    run_id: id,
2094                    step_id: Some(id),
2095                    rejected_by: "bob".to_string(),
2096                    user_id: None,
2097                    reason: None,
2098                    requirement: None,
2099                    at: now,
2100                }),
2101                "approval_rejected",
2102            ),
2103            (
2104                Event::ApprovalEscalated(ApprovalEscalatedEvent {
2105                    run_id: id,
2106                    step_id: id,
2107                    step_name: "prod-gate".to_string(),
2108                    stage: 0,
2109                    policy: "auto_reject".to_string(),
2110                    action: "rejected".to_string(),
2111                    reason: "approval deadline of 3600s expired".to_string(),
2112                    assignee: None,
2113                    at: now,
2114                }),
2115                "approval_escalated",
2116            ),
2117            (
2118                Event::LogLine(LogLineEvent {
2119                    id,
2120                    run_id: id,
2121                    step_id: id,
2122                    step_name: "build".to_string(),
2123                    stream: LogStream::Stdout,
2124                    line: "Compiling ironflow v0.1.0".to_string(),
2125                    at: now,
2126                }),
2127                "log_line",
2128            ),
2129            (
2130                Event::UserSignedIn(UserSignedInEvent {
2131                    user_id: id,
2132                    username: "u".to_string(),
2133                    at: now,
2134                }),
2135                "user_signed_in",
2136            ),
2137            (
2138                Event::UserSignedUp(UserSignedUpEvent {
2139                    user_id: id,
2140                    username: "u".to_string(),
2141                    at: now,
2142                }),
2143                "user_signed_up",
2144            ),
2145            (
2146                Event::UserSignedOut(UserSignedOutEvent {
2147                    user_id: id,
2148                    at: now,
2149                }),
2150                "user_signed_out",
2151            ),
2152            (
2153                Event::ProviderAccountUpdated(ProviderAccountUpdatedEvent {
2154                    account_id: id,
2155                    name: "perso".to_string(),
2156                    change: ProviderAccountChange::TokenReplaced,
2157                    at: now,
2158                }),
2159                "provider_account.updated",
2160            ),
2161            (
2162                Event::ProviderAccountUsageUpdated(ProviderAccountUsageUpdatedEvent {
2163                    account_id: id,
2164                    name: "perso".to_string(),
2165                    windows: Vec::new(),
2166                    at: now,
2167                }),
2168                "provider_account.usage_updated",
2169            ),
2170            (
2171                Event::SignalAwaited(SignalAwaitedEvent {
2172                    run_id: id,
2173                    step_id: id,
2174                    step_name: "wait-ci".to_string(),
2175                    name: "ci.done".to_string(),
2176                    key: "abc".to_string(),
2177                    deadline_at: now,
2178                    at: now,
2179                }),
2180                "signal_awaited",
2181            ),
2182            (
2183                Event::SignalReceived(SignalReceivedEvent {
2184                    signal_id: id,
2185                    name: "ci.done".to_string(),
2186                    key: "abc".to_string(),
2187                    resumed_runs: vec![id],
2188                    sender_run_id: None,
2189                    sender_step_id: None,
2190                    at: now,
2191                }),
2192                "signal_received",
2193            ),
2194            (
2195                Event::ScheduleOccurrencesMissed(ScheduleOccurrencesMissedEvent {
2196                    schedule_id: id,
2197                    workflow_name: "nightly".to_string(),
2198                    reason: ScheduleMissReason::Overlap,
2199                    count: 1,
2200                    first: now,
2201                    last: now,
2202                    at: now,
2203                }),
2204                "schedule_occurrences_missed",
2205            ),
2206            (
2207                Event::WorkflowPaused(WorkflowPausedEvent {
2208                    workflow_name: "w".to_string(),
2209                    user_id: Some(id),
2210                    at: now,
2211                }),
2212                "workflow_paused",
2213            ),
2214            (
2215                Event::WorkflowResumed(WorkflowResumedEvent {
2216                    workflow_name: "w".to_string(),
2217                    user_id: Some(id),
2218                    at: now,
2219                }),
2220                "workflow_resumed",
2221            ),
2222        ];
2223
2224        assert_eq!(
2225            cases.len(),
2226            Event::ALL.len(),
2227            "every variant must be covered"
2228        );
2229
2230        for (event, expected_type) in cases {
2231            assert_eq!(event.event_type(), expected_type);
2232        }
2233    }
2234
2235    #[test]
2236    fn approval_escalated_serde_roundtrip() {
2237        let run_id = Uuid::now_v7();
2238        let step_id = Uuid::now_v7();
2239        let event = Event::ApprovalEscalated(ApprovalEscalatedEvent {
2240            run_id,
2241            step_id,
2242            step_name: "prod-gate".to_string(),
2243            stage: 1,
2244            policy: "escalate".to_string(),
2245            action: "reassigned to sre-oncall".to_string(),
2246            reason: "approval deadline of 3600s expired".to_string(),
2247            assignee: Some(Assignee::group("sre-oncall")),
2248            at: Utc::now(),
2249        });
2250
2251        let json = serde_json::to_string(&event).expect("serialize");
2252        assert!(
2253            json.contains("\"type\":\"approval_escalated\""),
2254            "got {json}"
2255        );
2256
2257        let back: Event = serde_json::from_str(&json).expect("deserialize");
2258        let Event::ApprovalEscalated(payload) = back else {
2259            panic!("expected an approval_escalated event");
2260        };
2261        assert_eq!(payload.run_id, run_id);
2262        assert_eq!(payload.stage, 1);
2263        assert_eq!(payload.assignee, Some(Assignee::group("sre-oncall")));
2264    }
2265
2266    #[test]
2267    fn approval_escalated_carries_run_and_step_ids() {
2268        let run_id = Uuid::now_v7();
2269        let step_id = Uuid::now_v7();
2270        let event = Event::ApprovalEscalated(ApprovalEscalatedEvent {
2271            run_id,
2272            step_id,
2273            step_name: "prod-gate".to_string(),
2274            stage: 0,
2275            policy: "notify".to_string(),
2276            action: "notified 1 target".to_string(),
2277            reason: "approval deadline of 60s expired".to_string(),
2278            assignee: None,
2279            at: Utc::now(),
2280        });
2281
2282        assert_eq!(event.run_id(), Some(run_id));
2283        assert_eq!(event.step_id(), Some(step_id));
2284        assert_eq!(event.user_id(), None);
2285    }
2286
2287    #[test]
2288    fn signal_awaited_serde_roundtrip() {
2289        let run_id = Uuid::now_v7();
2290        let step_id = Uuid::now_v7();
2291        let event = Event::SignalAwaited(SignalAwaitedEvent {
2292            run_id,
2293            step_id,
2294            step_name: "wait-ci".to_string(),
2295            name: "ci.done".to_string(),
2296            key: "abc".to_string(),
2297            deadline_at: Utc::now(),
2298            at: Utc::now(),
2299        });
2300
2301        let json = serde_json::to_string(&event).expect("serialize");
2302        assert!(json.contains("\"type\":\"signal_awaited\""), "got {json}");
2303        let back: Event = serde_json::from_str(&json).expect("deserialize");
2304        assert_eq!(back.event_type(), Event::SIGNAL_AWAITED);
2305        assert_eq!(back.run_id(), Some(run_id));
2306        assert_eq!(back.step_id(), Some(step_id));
2307        assert_eq!(back.user_id(), None);
2308    }
2309
2310    #[test]
2311    fn signal_received_serde_roundtrip() {
2312        let resumed = Uuid::now_v7();
2313        let event = Event::SignalReceived(SignalReceivedEvent {
2314            signal_id: Uuid::now_v7(),
2315            name: "ci.done".to_string(),
2316            key: "abc".to_string(),
2317            resumed_runs: vec![resumed],
2318            sender_run_id: None,
2319            sender_step_id: None,
2320            at: Utc::now(),
2321        });
2322
2323        let json = serde_json::to_string(&event).expect("serialize");
2324        assert!(json.contains("\"type\":\"signal_received\""), "got {json}");
2325        let back: Event = serde_json::from_str(&json).expect("deserialize");
2326        let Event::SignalReceived(payload) = &back else {
2327            panic!("expected a signal_received event, got {back:?}");
2328        };
2329        assert_eq!(payload.resumed_runs, vec![resumed]);
2330        assert_eq!(back.run_id(), None);
2331        assert_eq!(back.step_id(), None);
2332        assert_eq!(back.user_id(), None);
2333        assert!(!json.contains("sender_run_id"), "got {json}");
2334    }
2335
2336    #[test]
2337    fn signal_received_from_a_workflow_belongs_to_the_sender() {
2338        let (run_id, step_id) = (Uuid::now_v7(), Uuid::now_v7());
2339        let json = json!({
2340            "type": "signal_received", "signal_id": Uuid::now_v7(), "name": "ci.done",
2341            "key": "abc", "resumed_runs": [], "at": Utc::now(),
2342            "sender_run_id": run_id, "sender_step_id": step_id,
2343        });
2344        let event: Event = from_value(json).expect("deserialize");
2345        assert_eq!(event.run_id(), Some(run_id));
2346        assert_eq!(event.step_id(), Some(step_id));
2347    }
2348
2349    #[test]
2350    fn schedule_occurrences_missed_serde_roundtrip() {
2351        let schedule_id = Uuid::now_v7();
2352        let first = Utc::now();
2353        let event = Event::ScheduleOccurrencesMissed(ScheduleOccurrencesMissedEvent {
2354            schedule_id,
2355            workflow_name: "nightly".to_string(),
2356            reason: ScheduleMissReason::OutsideWindow,
2357            count: 4,
2358            first,
2359            last: first,
2360            at: first,
2361        });
2362
2363        let json = serde_json::to_string(&event).expect("serialize");
2364        assert!(
2365            json.contains("\"type\":\"schedule_occurrences_missed\""),
2366            "got {json}"
2367        );
2368        assert!(json.contains("\"reason\":\"outside_window\""), "got {json}");
2369        let back: Event = serde_json::from_str(&json).expect("deserialize");
2370        let Event::ScheduleOccurrencesMissed(payload) = &back else {
2371            panic!("expected a schedule_occurrences_missed event, got {back:?}");
2372        };
2373        assert_eq!(payload.schedule_id, schedule_id);
2374        assert_eq!(payload.reason, ScheduleMissReason::OutsideWindow);
2375        assert_eq!(payload.count, 4);
2376        assert_eq!(payload.first, first);
2377        assert!(Event::ALL.contains(&Event::SCHEDULE_OCCURRENCES_MISSED));
2378        assert_eq!(back.run_id(), None);
2379        assert_eq!(back.step_id(), None);
2380        assert_eq!(back.user_id(), None);
2381    }
2382
2383    #[test]
2384    fn legacy_approval_decisions_have_no_user_id_nor_reason() {
2385        let raw = r#"{"type":"approval_granted","run_id":"01890000-0000-7000-8000-000000000000","approved_by":"alice","at":"2026-01-01T00:00:00Z"}"#;
2386        let granted: Event = serde_json::from_str(raw).expect("deserialize");
2387        assert_eq!(granted.user_id(), None);
2388
2389        let raw = r#"{"type":"approval_rejected","run_id":"01890000-0000-7000-8000-000000000000","rejected_by":"bob","at":"2026-01-01T00:00:00Z"}"#;
2390        let rejected: Event = serde_json::from_str(raw).expect("deserialize");
2391        let Event::ApprovalRejected(rejected) = rejected else {
2392            panic!("expected approval_rejected");
2393        };
2394        assert_eq!(rejected.user_id, None);
2395        assert_eq!(rejected.reason, None);
2396    }
2397
2398    #[test]
2399    fn approval_rejected_roundtrips_the_actor_and_reason() {
2400        let user_id = Uuid::now_v7();
2401        let event = Event::ApprovalRejected(ApprovalRejectedEvent {
2402            run_id: Uuid::now_v7(),
2403            step_id: Some(Uuid::now_v7()),
2404            rejected_by: "bob".to_string(),
2405            user_id: Some(user_id),
2406            reason: Some("wrong environment".to_string()),
2407            requirement: None,
2408            at: Utc::now(),
2409        });
2410
2411        let json = serde_json::to_string(&event).expect("serialize");
2412        let back: Event = serde_json::from_str(&json).expect("deserialize");
2413        assert_eq!(back.user_id(), Some(user_id));
2414        let Event::ApprovalRejected(back) = back else {
2415            panic!("expected approval_rejected");
2416        };
2417        assert_eq!(back.reason.as_deref(), Some("wrong environment"));
2418    }
2419
2420    #[test]
2421    fn workflow_paused_and_resumed_serde_roundtrip() {
2422        let user_id = Uuid::now_v7();
2423        let now = Utc::now();
2424        let events = vec![
2425            Event::WorkflowPaused(WorkflowPausedEvent {
2426                workflow_name: "deploy".to_string(),
2427                user_id: Some(user_id),
2428                at: now,
2429            }),
2430            Event::WorkflowResumed(WorkflowResumedEvent {
2431                workflow_name: "deploy".to_string(),
2432                user_id: Some(user_id),
2433                at: now,
2434            }),
2435        ];
2436
2437        for event in events {
2438            let json = serde_json::to_string(&event).expect("serialize");
2439            assert!(
2440                json.contains(&format!("\"type\":\"{}\"", event.event_type())),
2441                "got {json}"
2442            );
2443            assert!(json.contains("\"workflow_name\":\"deploy\""), "got {json}");
2444            let back: Event = serde_json::from_str(&json).expect("deserialize");
2445            assert_eq!(back.event_type(), event.event_type());
2446            assert_eq!(back.user_id(), Some(user_id));
2447            assert_eq!(back.run_id(), None);
2448            assert_eq!(back.step_id(), None);
2449            assert!(Event::ALL.contains(&back.event_type()));
2450        }
2451    }
2452
2453    #[test]
2454    fn workflow_paused_without_user_id_deserializes() {
2455        let raw =
2456            r#"{"type":"workflow_paused","workflow_name":"deploy","at":"2026-01-01T00:00:00Z"}"#;
2457        let event: Event = serde_json::from_str(raw).expect("deserialize");
2458        assert_eq!(event.event_type(), Event::WORKFLOW_PAUSED);
2459        assert_eq!(event.user_id(), None);
2460    }
2461}