kanade_shared/wire/obs_event.rs
1//! Issue #246 — per-PC observability events.
2//!
3//! Distinct from [`super::EventStarted`] (lifecycle: "script just
4//! spawned, watch this result_id"). `ObsEvent` is the timeline
5//! data type: sign-in / sign-out, power on / off, sleep / resume,
6//! agent self-update milestones — anything an operator wants to
7//! see on a per-PC timeline. The actual log source is open-ended;
8//! today it's Windows Event Log via a scheduled PowerShell job,
9//! tomorrow it could be agent-emitted milestones or the
10//! `kanade logs collect` diagnostic bundle (#219) carrying a
11//! pointer to an Object Store blob.
12//!
13//! Schema is intentionally narrow: a stable triplet
14//! (`pc_id`, `source`, `event_record_id`) for dedup-on-replay,
15//! plus a `kind` enum-ish string for the SPA filter chip, plus a
16//! free-form JSON `payload` for the per-kind details. Open-ended
17//! by design — adding a new event source doesn't require a wire
18//! change, just a new `kind` value.
19//!
20//! NATS subject: `obs.<pc_id>` (see [`super::super::subject::obs`]).
21//! JetStream stream `OBS_EVENTS` retains ~30d so a backend that
22//! was offline catches up on reconnect.
23//!
24//! De-dup is the BACKEND's job — agent re-sends are explicitly
25//! allowed and useful (a watermark mismatch shouldn't lose data).
26//! Backend UNIQUE constraint on
27//! `(pc_id, source, event_record_id)` makes re-sends a no-op.
28
29use chrono::{DateTime, Utc};
30use serde::{Deserialize, Serialize};
31
32/// One observability event published to `obs.<pc_id>`.
33///
34/// The `kind` field is a free-form string — vocabulary lives at
35/// the consumer (backend projector decides filtering / coloring;
36/// SPA decides chip labels). Established kinds at #246 land:
37///
38/// - `logon` / `logoff` — Security log 4624 / 4634
39/// - `boot` / `shutdown` — System log 12 / 13 (kernel-general)
40/// - `unexpected_shutdown` — System log 41
41/// - `sleep` / `resume` — System log 42 / 107
42/// - `active` / `idle` — agent idle sampler (#841); the one lane
43/// signal with no Event Log source
44/// - `agent_update` — agent self-update milestone
45/// - `agent_online` / `agent_offline` — whether the fleet was being
46/// *observed*, as opposed to what a host was doing. Emitted from
47/// two sides with different `source`s: the agent stamps
48/// `agent:startup` at boot, and the backend infers both from
49/// heartbeat gaps as `backend:heartbeat-watchdog` (#1107). They
50/// drive no lane; they bound what the other lanes may claim.
51/// - `diagnostic` — kanade logs collect bundles (#219)
52///
53/// New kinds can be added without a wire change; the backend
54/// projector stores whatever string the agent sends and the SPA
55/// surfaces it.
56#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
57pub struct ObsEvent {
58 /// PC reporting the event. Routing key on the subject side
59 /// (`obs.<pc_id>`) and primary scope of the SPA timeline view.
60 pub pc_id: String,
61
62 /// Wall-clock instant of the event as known to the SOURCE
63 /// (e.g. Windows Event Log's `TimeCreated`), NOT the moment
64 /// the agent published it. The timeline must reflect when
65 /// things happened on the box, not when the projector heard
66 /// about them — the two can differ by minutes when the agent
67 /// is catching up from outbox after a broker outage.
68 pub at: DateTime<Utc>,
69
70 /// What kind of event this is — the SPA's filter chip and
71 /// the backend projector's coloring key. See the doc comment
72 /// on this struct for the vocabulary at landing.
73 pub kind: String,
74
75 /// Where this event came from. Format `<scheme>:<detail>`
76 /// (e.g. `winlog:System`, `winlog:Security`, `agent:internal`,
77 /// `kanade:logs_collect`). Two roles:
78 ///
79 /// - Distinguishes events from different sources that might
80 /// share an `event_record_id` namespace.
81 /// - Lets the SPA filter "show me only winlog events" without
82 /// needing a separate enum.
83 pub source: String,
84
85 /// Stable per-source unique identifier — e.g. EventRecordID
86 /// from the Windows Event Log. Combined with `pc_id` and
87 /// `source` it forms the dedup key, so agent re-sends (under
88 /// watermark drift, outbox replay, etc.) are harmless.
89 ///
90 /// `None` for sources that have no natural unique id (e.g.
91 /// agent-emitted milestones where the only candidate is the
92 /// `at` timestamp + kind, which the backend can synthesize
93 /// from those fields if needed).
94 ///
95 /// `#[serde(default)]` so an agent publisher that has no id
96 /// to emit can omit the field entirely; serde fills `None`
97 /// rather than refusing the message. Without this, agent
98 /// versions that always send the field can deserialize but
99 /// future ones that drop it on `null` cases would silently
100 /// land in the warn-log → projector drop path.
101 #[serde(default)]
102 pub event_record_id: Option<String>,
103
104 /// Free-form per-kind details. The wire stays narrow
105 /// (`pc_id`, `at`, `kind`, `source`, `event_record_id`) and
106 /// the per-kind shape lives here:
107 ///
108 /// - `logon`: `{ "user": "...", "logon_type": 2 }`
109 /// - `boot`: typically `null` or `{}` — the bare presence is
110 /// the event
111 /// - `diagnostic`: `{ "bucket": "OBJECT_DIAGNOSTICS", "key":
112 /// "..." }` — pointer to the actual log blob
113 ///
114 /// Backend projector stores this as TEXT (the JSON
115 /// representation); SPA renders it kind-aware.
116 ///
117 /// `#[serde(default)]` so a publisher emitting a bare-presence
118 /// event can omit the field entirely (serde fills
119 /// `Value::Null`) rather than being forced to write
120 /// `"payload": null` on every line.
121 #[serde(default)]
122 pub payload: serde_json::Value,
123}
124
125#[cfg(test)]
126mod tests {
127 use super::*;
128 use chrono::TimeZone;
129 use serde_json::json;
130
131 #[test]
132 fn obs_event_round_trips_through_json() {
133 let t = Utc.with_ymd_and_hms(2026, 5, 28, 10, 41, 0).unwrap();
134 let e = ObsEvent {
135 pc_id: "pc-01".into(),
136 at: t,
137 kind: "logon".into(),
138 source: "winlog:Security".into(),
139 event_record_id: Some("1234567".into()),
140 payload: json!({ "user": "yukimemi", "logon_type": 2 }),
141 };
142 let s = serde_json::to_string(&e).unwrap();
143 let back: ObsEvent = serde_json::from_str(&s).unwrap();
144 assert_eq!(back, e);
145 }
146
147 #[test]
148 fn obs_event_null_payload_is_valid() {
149 // boot / shutdown / similar bare-presence events: the
150 // `at` + `kind` is the whole signal, payload carries
151 // nothing. Make sure null serialises round-trip clean
152 // so the agent's PowerShell side can emit an explicit
153 // `null` without backend deserialise rejecting.
154 let e = ObsEvent {
155 pc_id: "pc-01".into(),
156 at: Utc.with_ymd_and_hms(2026, 5, 28, 0, 0, 0).unwrap(),
157 kind: "boot".into(),
158 source: "winlog:System".into(),
159 event_record_id: Some("99".into()),
160 payload: serde_json::Value::Null,
161 };
162 let s = serde_json::to_string(&e).unwrap();
163 let back: ObsEvent = serde_json::from_str(&s).unwrap();
164 assert_eq!(back, e);
165 }
166
167 #[test]
168 fn obs_event_missing_event_record_id_deserialises() {
169 // Agent-emitted milestones (e.g. agent_started) have no
170 // natural EventRecordID equivalent. The field is
171 // `Option<String>` annotated `#[serde(default)]`, so a
172 // publisher omitting it entirely lands as `None` rather
173 // than failing the deserialise. Gemini #247 HIGH —
174 // without the explicit default, serde requires Option
175 // fields to be present (allowed to be null, not absent).
176 let s = r#"{
177 "pc_id": "pc-01",
178 "at": "2026-05-28T10:00:00Z",
179 "kind": "agent_started",
180 "source": "agent:internal",
181 "payload": null
182 }"#;
183 let e: ObsEvent = serde_json::from_str(s).unwrap();
184 assert_eq!(e.event_record_id, None);
185 assert_eq!(e.kind, "agent_started");
186 }
187
188 #[test]
189 fn obs_event_missing_payload_defaults_to_null() {
190 // Bare-presence events (boot / shutdown / agent_started)
191 // have no per-kind details. A publisher that omits the
192 // `payload` field entirely should parse — `#[serde(default)]`
193 // on the field fills `Value::Null` so the projector's
194 // `INSERT` sees the same value as if the publisher had
195 // written `"payload": null` explicitly.
196 let s = r#"{
197 "pc_id": "pc-01",
198 "at": "2026-05-28T10:00:00Z",
199 "kind": "boot",
200 "source": "winlog:System",
201 "event_record_id": "1"
202 }"#;
203 let e: ObsEvent = serde_json::from_str(s).unwrap();
204 assert_eq!(e.payload, serde_json::Value::Null);
205 }
206}