Skip to main content

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}