onevcs/event.rs
1//! The shared NDJSON event envelope, and the filter a consumer reads it through.
2//!
3//! Every process in the stack emits this one shape. The types are duplicated in
4//! each producing crate on purpose — there is deliberately no shared util crate —
5//! so the suite holds them to the fixture in `docs/contract.md` rather than to a
6//! dependency. [`EventFilter`] follows the same pattern: the grammar is shared
7//! across the three repositories, and each of them owns its copy of it.
8
9use serde::{Deserialize, Deserializer, Serialize};
10use serde_json::{Map, Value};
11
12use crate::error::{self, Result};
13
14/// One event, as a producing process writes it to its NDJSON stream.
15///
16/// Merge order across streams is `(ts, stream, seq)`; a consumer detects loss via
17/// per-stream [`seq`](Envelope::seq) gaps. There are no cross-stream ordering
18/// promises beyond timestamps.
19#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
20// llmlint: ignore-block[boundary_inputs_validated] the envelope's semantic boundary checks
21// — that `v` is 1, that `ts` is millisecond-precision UTC RFC3339, that a payload text
22// field was truncated at 4096 bytes — are the reader's job, and this crate is
23// interface-only: it declares the shape and adds no logic behind it. The shape itself is
24// enforced here (an unknown `kind` or `source`, a `seq` that is not a u64, a missing
25// field are all rejected by serde and asserted in tests/contract.rs), and the semantic
26// checks land with the parser seam that reads a stream.
27#[serde(try_from = "StoredEnvelope")]
28pub struct Envelope {
29 /// Envelope schema version. `1` is the shape `docs/contract.md` declares.
30 pub v: u32,
31 /// RFC3339 with millisecond precision, in UTC.
32 pub ts: String,
33 /// Unique id per producing process.
34 pub stream: String,
35 /// Monotonic per [`stream`](Envelope::stream), so a consumer sees a gap when
36 /// it has lost an event.
37 pub seq: u64,
38 /// Which library produced the event.
39 pub source: Source,
40 /// What happened.
41 pub kind: EventKind,
42 /// Which part of a change's life the event belongs to, as the producer
43 /// classified it.
44 ///
45 /// Stamped by whoever emitted the event rather than derived by whoever reads
46 /// one, because one kind's phase is not a fact about the kind: a `push` of the
47 /// session's own branch is [`Phase::Development`] and a push of anything else is
48 /// [`Phase::Integrate`], and only the producer knows which branch it pushed.
49 pub phase: Phase,
50 /// What the producer knew about the run when it stamped the event.
51 pub labels: Labels,
52 /// The event's own fields. Text fields truncate at 4096 bytes and set
53 /// `"truncated": true`; anything larger is an artifact instead.
54 pub payload: Map<String, Value>,
55 /// Evidence too large for the payload, stored by the producing library and
56 /// fetched through its CLI.
57 pub artifacts: Vec<ArtifactRef>,
58}
59// llmlint: ignore-end[boundary_inputs_validated]
60
61/// The library that produced an event.
62#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(rename_all = "lowercase")]
64pub enum Source {
65 /// `oneagentgraph`.
66 Agentgraph,
67 /// `onevcs`, this crate.
68 Vcs,
69 /// `onepipeline`.
70 Pipeline,
71}
72
73/// What an event says happened.
74///
75/// These are the kinds `onevcs` produces. A consumer merging several sources
76/// reads each source's own kinds; nothing here promises to name another
77/// library's.
78///
79/// **A kind is retired by keeping it recognised and inert, never by deleting it.**
80/// A stream is a record, and this enum is what says whether a line of one can be
81/// read at all: delete a variant and every line any earlier build wrote with it
82/// stops parsing, in a build that has no way to know what it lost. That is not
83/// hypothetical — `gate-started` and `gate-verdict` went with the host-run gate in
84/// 0.11.0, and 30% of the streams on the host that consumes this crate turned into
85/// one refusal per line, most expensively in a status read that walks every stream
86/// there is. So a kind nothing emits any more keeps its variant, documented as
87/// retired and produced by nothing, and a reader goes on being able to say what
88/// that line recorded.
89///
90/// The permissive read this crate's stream readers go through is the other half of
91/// that rule rather than a substitute for it. It makes a kind this build has *never
92/// had* — one a later build wrote — cheap to pass over, which is the case no
93/// vocabulary here can cover. It cannot give back the meaning of a kind that was
94/// deleted, which is why the two retired above are a loss this crate carries rather
95/// than one it recovered.
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
97#[serde(rename_all = "kebab-case")]
98pub enum EventKind {
99 /// A session was opened over a per-run clone and worktree.
100 SessionOpened,
101 /// Objects were fetched, deliberately outside any exclusive section.
102 Fetch,
103 /// Waiting on an identity's lock; carries the identity, elapsed, and queue
104 /// position.
105 LockWait,
106 /// The identity's lock was acquired.
107 LockAcquired,
108 /// Work was committed onto a preserved branch; carries the provenance kind.
109 CommitPreserved,
110 /// A branch was pushed.
111 Push,
112 /// A change request was opened; carries its URL and the host kind.
113 ChangeOpened,
114 /// A check moved; carries its name, whether it is required, the status
115 /// transition, the conclusion, and its log as an artifact once complete.
116 ChangeCheck,
117 /// The change request merged.
118 ChangeMerged,
119 /// The change request entered the host's merge queue.
120 MergeQueued,
121 /// The merge the host had queued completed.
122 MergeCompleted,
123 /// Preserved work that carried an incomplete-step marker was verified and
124 /// attested.
125 RecoveryAttested,
126 /// The base moved under a publication and the bounded resolve-and-requeue did
127 /// not converge.
128 SyncConflict,
129 /// The session's worktree and lease were released.
130 SessionClosed,
131 /// An **automated** release target's probe was run; carries the identity, the
132 /// target, which form of probe it was, what it answered, the version where it
133 /// answered one, and how long it took. A human-step target never produces one,
134 /// and that absence is the observable proof that no probe ran for it.
135 ReleaseProbed,
136 /// Somebody recorded a release of a human-step target; carries the identity,
137 /// the target, the version, the landing commit, the actor, and the version it
138 /// superseded where it replaced one.
139 ReleaseAcknowledged,
140 /// A landing was released, the first time it was: its baseline passed for an
141 /// automated target, its acknowledgement recorded for a human-step one.
142 ReleaseObserved,
143}
144
145/// One line of a stream as a build that predates a field on the envelope wrote it.
146///
147/// The envelope is versioned, and `v: 1` is the shape both this build and every
148/// released one call version 1 — so a *field* added inside that version has to be
149/// readable in its absence or an older stream stops being readable at all. Every
150/// key here is the envelope's own; the one that may be missing is
151/// [`phase`](Envelope::phase), and what stands in for it is the mapping below,
152/// which is exact for every kind whose target does not decide it.
153#[derive(Deserialize)]
154struct StoredEnvelope {
155 v: u32,
156 ts: String,
157 stream: String,
158 seq: u64,
159 source: Source,
160 kind: StoredKind,
161 #[serde(default)]
162 phase: Option<Phase>,
163 labels: Labels,
164 payload: Map<String, Value>,
165 artifacts: Vec<ArtifactRef>,
166}
167
168/// The `kind` a line carries: this build's own word for it, or the wire spelling
169/// where this build has no word at all.
170///
171/// The fallback is only ever reached by a spelling [`EventKind`] does not name, so
172/// a kind this build knows can never arrive as one it does not. What is *not* here
173/// is a fallback for anything else on the envelope: a `source` this build cannot
174/// name, a `seq` that is not a number, a missing field are each still a line
175/// nothing can be concluded from, and each is still refused.
176#[derive(Deserialize)]
177#[serde(untagged)]
178enum StoredKind {
179 /// A kind in this build's vocabulary.
180 Known(EventKind),
181 /// A kind this build has no word for — a later build's, or one an earlier
182 /// build wrote and this one deleted.
183 Unknown(String),
184}
185
186impl StoredEnvelope {
187 /// The envelope this line is, or the wire spelling of the kind that stopped it
188 /// being one.
189 fn read(self) -> std::result::Result<Envelope, String> {
190 let kind = match self.kind {
191 StoredKind::Known(kind) => kind,
192 StoredKind::Unknown(spelling) => return Err(spelling),
193 };
194 Ok(Envelope {
195 v: self.v,
196 ts: self.ts,
197 stream: self.stream,
198 seq: self.seq,
199 source: self.source,
200 kind,
201 // A `push` an older build wrote carries no phase and named no target, so
202 // there is nothing to recover the producer's classification from. It reads
203 // as the phase a push of the session's own branch is, which is what all but
204 // one of the three producers of one emits and the one a session's stream
205 // holds — and it is a reading of a record rather than a claim about the
206 // push, which is why the phase is stamped from here on.
207 phase: self
208 .phase
209 .or_else(|| Phase::of(kind))
210 .unwrap_or(Phase::Development),
211 labels: self.labels,
212 payload: self.payload,
213 artifacts: self.artifacts,
214 })
215 }
216}
217
218/// An [`Envelope`] is the *value*, so it is exactly as strict as it has always
219/// been: a kind this build cannot name is a kind it cannot hand over, and pretending
220/// otherwise would put a payload nothing knows how to read behind a type that says
221/// it does. Tolerating one is [`Line`]'s job, one layer out, where the caller can
222/// pass the line over instead of acting on it.
223impl TryFrom<StoredEnvelope> for Envelope {
224 type Error = String;
225
226 fn try_from(stored: StoredEnvelope) -> std::result::Result<Self, Self::Error> {
227 stored
228 .read()
229 .map_err(|kind| format!("unknown variant {kind:?} for an event kind"))
230 }
231}
232
233/// One line of a stream, read as far as this build's vocabulary reaches.
234///
235/// The distinction the readers of a stream are held to. A line that is *not what a
236/// writer left* — torn, of an envelope version this build does not read, stamped in
237/// a way nothing can order — is a gap, and saying so is not negotiable: reporting
238/// "could not look" as "there is none" is how a report about half a record reads as
239/// a report about all of it. A line that is a perfectly good envelope recording
240/// something this build has no word for is not that. There is nothing to conclude
241/// from it and nothing missing from the read either, so it is passed over rather
242/// than announced — which is what a reader of streams a *later* build wrote needs,
243/// and what a reader of streams an earlier one wrote needed and did not have.
244pub(crate) enum Line {
245 /// An envelope of a kind this build knows, and can therefore act on.
246 ///
247 /// Boxed: an envelope carries three strings, a map and a vector, and the other
248 /// variant is a header — so an unboxed enum would make every line of every
249 /// stream cost the larger of the two.
250 Known(Box<Envelope>),
251 /// A well-formed envelope of a kind this build has no word for.
252 Unknown(UnknownKind),
253}
254
255/// A line whose kind this build cannot name, as much of it as such a reader can use.
256///
257/// Its header and nothing else, deliberately. Whether the line is a *gap* is
258/// decided by the envelope's own fields — the version it declares, the stamp it is
259/// ordered by, the stream it belongs to — and those mean the same thing whatever
260/// happened, so every check a known kind gets this one gets too. The payload is the
261/// part the kind is the key to, so there is nothing here to offer of it.
262pub(crate) struct UnknownKind {
263 /// The envelope schema version the line declares.
264 pub v: u32,
265 /// When it was stamped, unparsed: ordering it is the reader's own check.
266 pub ts: String,
267 /// The stream it names, so attribution is asked of it as of any other line.
268 pub stream: String,
269}
270
271impl Line {
272 /// One line of a stream, tolerant of its kind and of nothing else.
273 // llmlint: ignore[boundary_inputs_validated] the envelope's *shape* is validated here
274 // and exactly as strictly as it always was: every field but `kind` goes through the
275 // same derive, and a `kind` that is not a string is still refused. The two semantic
276 // checks — that `v` is the version this build reads, and that `ts` is a stamp it can
277 // order — are deliberately not here, for the reason `Envelope` has never made them
278 // either: what to do about one is the *reader's*, and the readers disagree. `status`
279 // reports each as a gap in its notes rather than refusing, because it is asked what
280 // became of a piece of work and must not answer "there is none" for "could not
281 // look" — and it applies both to a line whose kind has no word here exactly as to
282 // one it knows, which is what `UnknownKind` carries the header for. Deciding either
283 // here would take that choice away from the one caller that has to report rather
284 // than refuse.
285 pub(crate) fn read(line: &str) -> serde_json::Result<Self> {
286 let stored: StoredEnvelope = serde_json::from_str(line)?;
287 let header = UnknownKind {
288 v: stored.v,
289 ts: stored.ts.clone(),
290 stream: stored.stream.clone(),
291 };
292 Ok(match stored.read() {
293 Ok(envelope) => Line::Known(Box::new(envelope)),
294 Err(_) => Line::Unknown(header),
295 })
296 }
297
298 /// The stream this line names, whether or not its kind has a word here.
299 pub(crate) fn stream(&self) -> &str {
300 match self {
301 Line::Known(envelope) => &envelope.stream,
302 Line::Unknown(header) => &header.stream,
303 }
304 }
305}
306
307/// Which part of a change's life an event belongs to.
308///
309/// Four phases over one change: the work is made ([`Development`](Phase::Development)),
310/// it is brought together with the base it is going onto
311/// ([`Integrate`](Phase::Integrate)), it is proposed and ruled on
312/// ([`Review`](Phase::Review)), and what carries it is released
313/// ([`Release`](Phase::Release)). A consumer reading a session under an attention
314/// budget names a phase rather than enumerating the kinds that happen to be in it —
315/// which is what makes a kind added later arrive in the read that already wanted it.
316///
317/// Spelled `Phase` rather than `Lifecycle`, because [`crate::Lifecycle`] is already
318/// where a *session* is in its life and the two are different questions: a session
319/// that has closed still has a change in review.
320#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
321#[serde(rename_all = "kebab-case")]
322pub enum Phase {
323 /// The work is being made: the session, its clone, its lock, its commits, the
324 /// repair of a branch that carried an incomplete step, and the push of the
325 /// session's own branch.
326 Development,
327 /// The work is being brought together with the base: the host's merge queue,
328 /// the merge it completed, a base that moved out from under the publication,
329 /// and a push of any branch but the session's own.
330 Integrate,
331 /// The change request is open and being ruled on: it was opened, its checks
332 /// moved, and it merged.
333 Review,
334 /// What carries the landed change is being released: a probe was run, a person
335 /// acknowledged a release, and a landing was observed as released.
336 Release,
337}
338
339impl Phase {
340 /// The word this phase is spelled with, in a filter and in a rendering.
341 #[must_use]
342 pub fn as_str(self) -> &'static str {
343 match self {
344 Phase::Development => "development",
345 Phase::Integrate => "integrate",
346 Phase::Review => "review",
347 Phase::Release => "release",
348 }
349 }
350
351 /// Every phase, in the order a refusal lists them.
352 #[must_use]
353 pub fn every() -> [Phase; 4] {
354 [
355 Phase::Development,
356 Phase::Integrate,
357 Phase::Review,
358 Phase::Release,
359 ]
360 }
361
362 /// The phase an event of this kind belongs to, where the kind alone decides it.
363 ///
364 /// `None` for exactly one kind, and it is not an omission: a
365 /// [`Push`](EventKind::Push) of the session's own branch is
366 /// [`Development`](Phase::Development) and a push of anything else — the base a
367 /// `local-direct` squash lands on, the base a merge train advanced — is
368 /// [`Integrate`](Phase::Integrate). Which of the two it was is a fact about the
369 /// push rather than about the kind, so the producer stamps it and this answers
370 /// that it cannot.
371 #[must_use]
372 pub fn of(kind: EventKind) -> Option<Phase> {
373 Some(match kind {
374 EventKind::SessionOpened
375 | EventKind::Fetch
376 | EventKind::LockWait
377 | EventKind::LockAcquired
378 | EventKind::CommitPreserved
379 // The repair of a preserved branch, and therefore the work being made:
380 // it puts the branch back into a state its merge path can rule on, and
381 // it happens before that branch may enter one at all.
382 | EventKind::RecoveryAttested
383 | EventKind::SessionClosed => Phase::Development,
384 EventKind::MergeQueued | EventKind::MergeCompleted | EventKind::SyncConflict => {
385 Phase::Integrate
386 }
387 EventKind::ChangeOpened | EventKind::ChangeCheck | EventKind::ChangeMerged => {
388 Phase::Review
389 }
390 EventKind::ReleaseProbed
391 | EventKind::ReleaseAcknowledged
392 | EventKind::ReleaseObserved => Phase::Release,
393 EventKind::Push => return None,
394 })
395 }
396}
397
398impl std::fmt::Display for Phase {
399 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
400 f.write_str(self.as_str())
401 }
402}
403
404/// What a producer knew about the run when it stamped an event.
405///
406/// The reserved keys are the fields below; anything else a producer knows lands
407/// in [`extra`](Labels::extra). Producers stamp what they know and enrichers
408/// never rewrite what is already there.
409#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
410pub struct Labels {
411 /// The run this event belongs to.
412 #[serde(default, skip_serializing_if = "Option::is_none")]
413 pub run_id: Option<String>,
414 /// The round within the run.
415 #[serde(default, skip_serializing_if = "Option::is_none")]
416 pub round: Option<u64>,
417 /// The graph node being executed.
418 #[serde(default, skip_serializing_if = "Option::is_none")]
419 pub node: Option<String>,
420 /// The step within a node that runs several in sequence.
421 #[serde(default, skip_serializing_if = "Option::is_none")]
422 pub step: Option<String>,
423 /// Which member of a conversation produced the event.
424 #[serde(default, skip_serializing_if = "Option::is_none")]
425 pub member: Option<String>,
426 /// The persona that member is running under.
427 #[serde(default, skip_serializing_if = "Option::is_none")]
428 pub persona: Option<String>,
429 /// Free-form extras beyond the reserved keys above.
430 #[serde(flatten)]
431 pub extra: Map<String, Value>,
432}
433
434/// A reference to evidence stored beside the stream rather than inside it.
435#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
436pub struct ArtifactRef {
437 /// The id the producing library's CLI fetches this artifact by.
438 pub id: ArtifactId,
439 /// What the artifact is, e.g. `log`.
440 pub kind: String,
441 /// Its size in bytes.
442 pub bytes: u64,
443}
444
445/// The id of a stored artifact.
446#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
447#[serde(transparent)]
448pub struct ArtifactId(pub String);
449
450impl EventKind {
451 /// The kebab-case string this kind travels as, which is what a filter's
452 /// `kind` glob is matched against.
453 ///
454 /// The spelling is serde's, and `the_wire_spelling_of_every_kind_is_the_one_a_filter_matches`
455 /// in `tests/contract.rs` holds the two together — a match rather than a
456 /// serialization so that judging an event costs no allocation, and exhaustive
457 /// so that a kind added to the enum cannot reach a filter unnamed.
458 fn wire(self) -> &'static str {
459 match self {
460 EventKind::SessionOpened => "session-opened",
461 EventKind::Fetch => "fetch",
462 EventKind::LockWait => "lock-wait",
463 EventKind::LockAcquired => "lock-acquired",
464 EventKind::CommitPreserved => "commit-preserved",
465 EventKind::Push => "push",
466 EventKind::ChangeOpened => "change-opened",
467 EventKind::ChangeCheck => "change-check",
468 EventKind::ChangeMerged => "change-merged",
469 EventKind::MergeQueued => "merge-queued",
470 EventKind::MergeCompleted => "merge-completed",
471 EventKind::RecoveryAttested => "recovery-attested",
472 EventKind::SyncConflict => "sync-conflict",
473 EventKind::SessionClosed => "session-closed",
474 EventKind::ReleaseProbed => "release-probed",
475 EventKind::ReleaseAcknowledged => "release-acknowledged",
476 EventKind::ReleaseObserved => "release-observed",
477 }
478 }
479}
480
481/// The fields one matcher may name, in the order a refusal lists them.
482///
483/// `every_matcher_field_the_type_has_is_one_a_refusal_names_and_the_parser_takes`
484/// in `tests/contract.rs` holds this list, [`EventMatcher`]'s own fields, and the
485/// fields the parser below accepts to being the one vocabulary — so a field added
486/// to the type cannot reach an operator unnamed, or be named and not taken.
487const MATCHER_FIELDS: &str = "source, phase, kind, run_id, node, step, member, persona";
488
489/// Which events of a stream a consumer wants, in the grammar the three producing
490/// libraries share.
491///
492/// An envelope passes when it matches **any** `include` matcher — or `include` is
493/// absent or empty — and matches **no** `exclude` matcher. `exclude` wins over
494/// `include`, so a consumer narrows a broad `include` without restating it. The
495/// default is everything, which is what a stream read with no filter answers.
496///
497/// ```yaml
498/// include: # absent or empty: everything passes include
499/// - {source: vcs, kind: "gate-*"}
500/// exclude: # a match here always rejects
501/// - {kind: lock-wait}
502/// ```
503///
504/// Read a spec with [`parse`](EventFilter::parse), which refuses one that names a
505/// field this grammar does not have rather than reading it as match-everything or
506/// match-nothing.
507#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
508pub struct EventFilter {
509 /// Matchers an envelope passes by matching any one of them. Empty admits every
510 /// envelope, so a filter that only excludes says nothing here.
511 #[serde(skip_serializing_if = "Vec::is_empty")]
512 pub include: Vec<EventMatcher>,
513 /// Matchers that reject an envelope by matching it, whatever `include` said.
514 #[serde(skip_serializing_if = "Vec::is_empty")]
515 pub exclude: Vec<EventMatcher>,
516}
517
518/// One matcher: every field it sets must match, and a field it leaves unset is not
519/// consulted.
520///
521/// Deliberately not in the grammar: `stream`, which is a producing process's id
522/// rather than a family, and the payload, whose fields differ per kind — a filter
523/// that could reach into one would be a query language whose meaning changed with
524/// every event this crate learns to emit.
525#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
526pub struct EventMatcher {
527 /// The library that produced the event, by exact equality.
528 #[serde(skip_serializing_if = "Option::is_none")]
529 pub source: Option<Source>,
530 /// The part of a change's life the event belongs to, by exact equality against
531 /// what its producer stamped.
532 ///
533 /// A phase rather than the kinds in it, so a consumer that wants the review of a
534 /// change keeps wanting it when a kind is added to that phase. Which phases a
535 /// session *has* is decided where the stream is opened: naming one the session
536 /// cannot produce is refused by name, and a read that named none takes the ones
537 /// it can.
538 #[serde(skip_serializing_if = "Option::is_none")]
539 pub phase: Option<Phase>,
540 /// A glob over the kind's kebab-case wire string, where `*` matches any run of
541 /// characters including none: `change-*` is every change-request kind.
542 #[serde(skip_serializing_if = "Option::is_none")]
543 pub kind: Option<String>,
544 /// The `run_id` label, by exact equality. An envelope the producer did not
545 /// stamp it on does not match.
546 #[serde(skip_serializing_if = "Option::is_none")]
547 pub run_id: Option<String>,
548 /// The `node` label, by exact equality.
549 #[serde(skip_serializing_if = "Option::is_none")]
550 pub node: Option<String>,
551 /// The `step` label, by exact equality.
552 #[serde(skip_serializing_if = "Option::is_none")]
553 pub step: Option<String>,
554 /// The `member` label, by exact equality.
555 #[serde(skip_serializing_if = "Option::is_none")]
556 pub member: Option<String>,
557 /// The `persona` label, by exact equality.
558 #[serde(skip_serializing_if = "Option::is_none")]
559 pub persona: Option<String>,
560}
561
562impl EventFilter {
563 /// Read a filter from the text of a spec: JSON, or the YAML the grammar is
564 /// written in.
565 ///
566 /// A spec is refused where it is read, naming the matcher that was wrong, for
567 /// the reason the rules loader refuses an unusable bound: a filter nobody can
568 /// read means either everything or nothing, and both of those are answers a
569 /// consumer would act on without ever being told it had asked for something
570 /// else.
571 pub fn parse(spec: &str) -> Result<Self> {
572 serde_yaml_ng::from_str(spec)
573 .map_err(|failure| error::invalid(format!("the event filter is unusable: {failure}")))
574 }
575
576 /// Whether an envelope reaches a consumer reading through this filter.
577 #[must_use]
578 pub fn matches(&self, envelope: &Envelope) -> bool {
579 if self.exclude.iter().any(|matcher| matcher.matches(envelope)) {
580 return false;
581 }
582 self.include.is_empty() || self.include.iter().any(|matcher| matcher.matches(envelope))
583 }
584
585 /// The filter a document holds, or the reason it is not one.
586 ///
587 /// Every refusal names the matcher it is about — which list and which position
588 /// in it — because that is what an operator has to find in the file they wrote.
589 fn from_document(document: &Value) -> std::result::Result<Self, String> {
590 let object = document.as_object().ok_or_else(|| {
591 format!(
592 "an event filter is a mapping of `include` and `exclude`, not {}",
593 shape(document)
594 )
595 })?;
596 if let Some(stray) = object
597 .keys()
598 .find(|key| !matches!(key.as_str(), "include" | "exclude"))
599 {
600 return Err(format!(
601 "an event filter names `include` and `exclude`; {stray:?} is neither"
602 ));
603 }
604 Ok(Self {
605 include: matchers(object.get("include"), "include")?,
606 exclude: matchers(object.get("exclude"), "exclude")?,
607 })
608 }
609}
610
611/// Routed through the same validation [`EventFilter::parse`] uses rather than
612/// derived, so a filter
613/// embedded in a consumer's own configuration is refused by the same rules — and
614/// with the same message — as one this crate's CLI reads from a spec.
615impl<'de> Deserialize<'de> for EventFilter {
616 fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
617 let document = Value::deserialize(deserializer)?;
618 Self::from_document(&document).map_err(serde::de::Error::custom)
619 }
620}
621
622impl EventMatcher {
623 /// Whether every field this matcher sets matches the envelope.
624 fn matches(&self, envelope: &Envelope) -> bool {
625 let labels = &envelope.labels;
626 self.source.is_none_or(|want| want == envelope.source)
627 && self.phase.is_none_or(|want| want == envelope.phase)
628 && self
629 .kind
630 .as_ref()
631 .is_none_or(|pattern| crate::policy::glob(pattern, envelope.kind.wire()))
632 && stamped(&self.run_id, &labels.run_id)
633 && stamped(&self.node, &labels.node)
634 && stamped(&self.step, &labels.step)
635 && stamped(&self.member, &labels.member)
636 && stamped(&self.persona, &labels.persona)
637 }
638}
639
640/// Whether a label matcher is satisfied: unset asks nothing, and a label the
641/// producer did not stamp answers no rather than being read as a wildcard.
642fn stamped(want: &Option<String>, stamped: &Option<String>) -> bool {
643 match want {
644 None => true,
645 Some(want) => stamped.as_deref() == Some(want.as_str()),
646 }
647}
648
649/// The matchers one of the two lists holds, or the reason it is not a list of them.
650fn matchers(value: Option<&Value>, list: &str) -> std::result::Result<Vec<EventMatcher>, String> {
651 // Absent is the documented "everything passes include" / "nothing is excluded".
652 // Present-but-not-a-list is not: `include: ` with nothing after it means one of
653 // those two to whoever wrote it and the other to whoever reads it, which is the
654 // guess this refuses to make.
655 let Some(value) = value else {
656 return Ok(Vec::new());
657 };
658 let entries = value.as_array().ok_or_else(|| {
659 format!(
660 "an event filter's `{list}` is a list of matchers, not {}",
661 shape(value)
662 )
663 })?;
664 entries
665 .iter()
666 .enumerate()
667 .map(|(index, entry)| matcher(entry, list, index + 1))
668 .collect()
669}
670
671/// One matcher of a list, or the reason it is not one.
672fn matcher(
673 value: &Value,
674 list: &str,
675 position: usize,
676) -> std::result::Result<EventMatcher, String> {
677 let named = format!("the event filter's {list} matcher {position}");
678 let fields = value.as_object().ok_or_else(|| {
679 format!(
680 "{named} is a mapping of matcher fields, not {}",
681 shape(value)
682 )
683 })?;
684 let mut matcher = EventMatcher::default();
685 for (field, value) in fields {
686 match field.as_str() {
687 // The families are named by serde's own refusal rather than restated
688 // here: [`Source`]'s derive already spells every one it has, and a second
689 // copy is a list that a family added to the enum leaves behind.
690 "source" => {
691 matcher.source = Some(
692 serde_json::from_value(value.clone())
693 .map_err(|failure| format!("{named} names no source family: {failure}"))?,
694 );
695 }
696 // Named by serde's own refusal, for the reason `source` is: [`Phase`]'s
697 // derive already spells every phase there is, and a second list here is
698 // one a phase added to the enum leaves behind.
699 "phase" => {
700 matcher.phase = Some(
701 serde_json::from_value(value.clone())
702 .map_err(|failure| format!("{named} names no phase: {failure}"))?,
703 );
704 }
705 "kind" => matcher.kind = Some(text(value, &named, field)?),
706 "run_id" => matcher.run_id = Some(text(value, &named, field)?),
707 "node" => matcher.node = Some(text(value, &named, field)?),
708 "step" => matcher.step = Some(text(value, &named, field)?),
709 "member" => matcher.member = Some(text(value, &named, field)?),
710 "persona" => matcher.persona = Some(text(value, &named, field)?),
711 unknown => {
712 return Err(format!(
713 "{named} names {unknown:?}, which is not a matcher field ({MATCHER_FIELDS})"
714 ))
715 }
716 }
717 }
718 Ok(matcher)
719}
720
721/// One matcher field's value, which every field but `source` compares as a string.
722fn text(value: &Value, named: &str, field: &str) -> std::result::Result<String, String> {
723 value.as_str().map(str::to_owned).ok_or_else(|| {
724 format!(
725 "{named} matches {field} against {}, which is not a string",
726 shape(value)
727 )
728 })
729}
730
731/// What a value is, for a refusal that has to say what was there instead.
732fn shape(value: &Value) -> &'static str {
733 match value {
734 Value::Null => "nothing",
735 Value::Bool(_) => "a boolean",
736 Value::Number(_) => "a number",
737 Value::String(_) => "a string",
738 Value::Array(_) => "a list",
739 Value::Object(_) => "a mapping",
740 }
741}