xtask/drift/
baseline.rs

1//! The vendored snapshot of the upstream event surface.
2//!
3//! `xtask/baseline/events.json` is the only thing `drift-check` compares
4//! against in its offline mode, which is what CI runs. It is written by
5//! `drift-check --refresh` and reviewed by a human in the pull request that
6//! bumps it — that review is the point where an upstream protocol change is
7//! consciously accepted.
8
9use std::path::Path;
10
11use serde::{Deserialize, Serialize};
12
13use crate::drift::upstream::{self, Upstream};
14
15/// Bumped when the meaning of the file changes, so an old baseline is reported
16/// rather than silently misread.
17pub const FORMAT: u32 = 1;
18
19pub const UPSTREAM_REPO: &str = "ag-ui-protocol/ag-ui";
20pub const UPSTREAM_PATH: &str = "sdks/typescript/packages/core/src/events.ts";
21
22const NOTE: &str = "Generated by `cargo run -p xtask -- drift-check --refresh`. \
23                    Reviewed by a human, never hand-edited.";
24
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct Baseline {
27    #[serde(rename = "$note")]
28    pub note: String,
29    pub format: u32,
30    pub source: Source,
31    /// Fields every event inherits from `BaseEventSchema`. Kept out of the
32    /// per-event field lists, where they would be repeated 36 times, and
33    /// compared once against the Rust `BaseEvent` instead.
34    pub base_event_fields: Vec<Field>,
35    /// `EventType` values in upstream declaration order.
36    pub event_types: Vec<String>,
37    /// One entry per event type, same order.
38    pub events: Vec<Event>,
39}
40
41#[derive(Debug, Clone, Serialize, Deserialize)]
42pub struct Source {
43    pub repo: String,
44    pub path: String,
45    /// Upstream commit the snapshot was taken at.
46    pub commit: String,
47    /// Date of that commit (ISO 8601, UTC).
48    pub commit_date: String,
49    /// Date `--refresh` ran (ISO 8601, UTC).
50    pub fetched_at: String,
51}
52
53#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
54pub struct Field {
55    pub name: String,
56    pub required: bool,
57}
58
59#[derive(Debug, Clone, Serialize, Deserialize)]
60pub struct Event {
61    #[serde(rename = "type")]
62    pub event_type: String,
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub schema: Option<String>,
65    pub fields: Vec<Field>,
66    /// Set when the Zod schema could not be read confidently. Fields are then
67    /// not compared, and `drift-check` reports a warning instead of a failure.
68    #[serde(skip_serializing_if = "Option::is_none")]
69    pub unparsed: Option<String>,
70}
71
72impl Baseline {
73    /// Builds a baseline from a freshly extracted upstream surface.
74    pub fn from_upstream(up: &Upstream, source: Source) -> Self {
75        Self {
76            note: NOTE.to_string(),
77            format: FORMAT,
78            source,
79            base_event_fields: up.base_fields.iter().map(field).collect(),
80            event_types: up.event_types.clone(),
81            events: up
82                .events
83                .iter()
84                .map(|e| Event {
85                    event_type: e.event_type.clone(),
86                    schema: e.schema.clone(),
87                    fields: e.fields.iter().map(field).collect(),
88                    unparsed: e.unparsed.clone(),
89                })
90                .collect(),
91        }
92    }
93
94    pub fn load(path: &Path) -> Result<Self, String> {
95        let text = std::fs::read_to_string(path).map_err(|e| {
96            format!(
97                "cannot read the vendored baseline at {}: {e}\n\
98                 Create it with: cargo run -p xtask -- drift-check --refresh",
99                path.display()
100            )
101        })?;
102        let baseline: Self = serde_json::from_str(&text)
103            .map_err(|e| format!("{} is not a valid baseline: {e}", path.display()))?;
104        if baseline.format != FORMAT {
105            return Err(format!(
106                "{} was written in format {} but this xtask reads format {FORMAT}.\n\
107                 Re-capture it with: cargo run -p xtask -- drift-check --refresh",
108                path.display(),
109                baseline.format
110            ));
111        }
112        Ok(baseline)
113    }
114
115    /// Refuses to persist a snapshot that cannot do the job of a baseline.
116    ///
117    /// `--refresh` replaces a human-reviewed file with whatever came back over
118    /// the network, and the dangerous failure is not a loud one. A response
119    /// truncated after the `EventType` enum but before the schemas parses
120    /// perfectly well: every event type is still found, every schema is
121    /// recorded `unparsed`, and because unparsed schemas are deliberately
122    /// warnings rather than failures, the resulting baseline agrees with any
123    /// Rust source at all. The gate would go quiet instead of going red, which
124    /// is the one outcome this crate exists to prevent.
125    ///
126    /// So a baseline must name at least one event type, carry one entry per
127    /// event type, and have read a majority of the schemas. A healthy capture
128    /// reads all of them; half is slack for upstream reformatting, not for a
129    /// half-delivered file.
130    fn validate(&self) -> Result<(), String> {
131        if self.event_types.is_empty() {
132            return Err("it names no event types".to_string());
133        }
134        if self.events.len() != self.event_types.len() {
135            return Err(format!(
136                "it names {} event types but carries {} entries",
137                self.event_types.len(),
138                self.events.len()
139            ));
140        }
141        let readable = self.events.iter().filter(|e| e.unparsed.is_none()).count();
142        if readable * 2 <= self.events.len() {
143            return Err(format!(
144                "only {readable} of {} schemas could be read. A baseline of unreadable schemas \
145                 compares equal to anything, so it would silently disable the drift gate rather \
146                 than fail it",
147                self.events.len()
148            ));
149        }
150        Ok(())
151    }
152
153    pub fn save(&self, path: &Path) -> Result<(), String> {
154        self.validate().map_err(|why| {
155            format!(
156                "refusing to write {}: {why}.\n\
157                 This is what a truncated or substituted upstream response looks like. Check \
158                 https://github.com/{UPSTREAM_REPO}/blob/main/{UPSTREAM_PATH} and re-run; the \
159                 existing baseline has been left untouched.",
160                path.display()
161            )
162        })?;
163
164        if let Some(parent) = path.parent() {
165            std::fs::create_dir_all(parent).map_err(|e| format!("{}: {e}", parent.display()))?;
166        }
167        let mut json = serde_json::to_string_pretty(self)
168            .map_err(|e| format!("cannot serialise the baseline: {e}"))?;
169        json.push('\n');
170
171        // Write-then-rename, so an interrupted write cannot leave a half-written
172        // baseline where a reviewed one used to be. The temporary file sits next
173        // to the target to keep the rename on one filesystem.
174        let temp = path.with_extension("json.tmp");
175        std::fs::write(&temp, json).map_err(|e| format!("{}: {e}", temp.display()))?;
176        std::fs::rename(&temp, path).map_err(|e| {
177            let _ = std::fs::remove_file(&temp);
178            format!("cannot move {} into place: {e}", temp.display())
179        })
180    }
181
182    pub fn event(&self, event_type: &str) -> Option<&Event> {
183        self.events.iter().find(|e| e.event_type == event_type)
184    }
185}
186
187fn field(f: &upstream::Field) -> Field {
188    Field {
189        name: f.name.clone(),
190        required: f.required,
191    }
192}
193
194#[cfg(test)]
195mod tests {
196    use super::*;
197    use std::sync::atomic::{AtomicUsize, Ordering};
198
199    fn temp_path() -> std::path::PathBuf {
200        static NEXT: AtomicUsize = AtomicUsize::new(0);
201        let dir = std::env::temp_dir().join(format!(
202            "xtask-baseline-{}-{}",
203            std::process::id(),
204            NEXT.fetch_add(1, Ordering::Relaxed)
205        ));
206        std::fs::create_dir_all(&dir).unwrap();
207        dir.join("events.json")
208    }
209
210    fn baseline(events: Vec<Event>) -> Baseline {
211        Baseline {
212            note: NOTE.to_string(),
213            format: FORMAT,
214            source: Source {
215                repo: UPSTREAM_REPO.into(),
216                path: UPSTREAM_PATH.into(),
217                commit: "0123456789abcdef".into(),
218                commit_date: "2026-08-01".into(),
219                fetched_at: "2026-08-01".into(),
220            },
221            base_event_fields: vec![],
222            event_types: events.iter().map(|e| e.event_type.clone()).collect(),
223            events,
224        }
225    }
226
227    fn event(ty: &str) -> Event {
228        Event {
229            event_type: ty.into(),
230            schema: Some(format!("{ty}Schema")),
231            fields: vec![Field {
232                name: "messageId".into(),
233                required: true,
234            }],
235            unparsed: None,
236        }
237    }
238
239    fn unreadable(ty: &str) -> Event {
240        Event {
241            unparsed: Some("no Zod schema found for this event type".into()),
242            fields: vec![],
243            ..event(ty)
244        }
245    }
246
247    #[test]
248    fn a_healthy_baseline_round_trips() {
249        let path = temp_path();
250        let written = baseline(vec![event("RAW"), event("TEXT_MESSAGE_START")]);
251        written.save(&path).unwrap();
252        let read = Baseline::load(&path).unwrap();
253        assert_eq!(read.event_types, ["RAW", "TEXT_MESSAGE_START"]);
254        assert_eq!(read.events.len(), 2);
255    }
256
257    /// The truncated-fetch case: every event type still parses, every schema is
258    /// `unparsed`, and such a baseline compares equal to any Rust source. It
259    /// must never reach disk.
260    #[test]
261    fn a_baseline_of_unreadable_schemas_is_refused() {
262        let path = temp_path();
263        let error = baseline(vec![unreadable("RAW"), unreadable("TEXT_MESSAGE_START")])
264            .save(&path)
265            .unwrap_err();
266        assert!(error.contains("0 of 2 schemas"), "{error}");
267        assert!(error.contains("silently disable the drift gate"), "{error}");
268        assert!(!path.exists(), "the baseline must not have been written");
269    }
270
271    #[test]
272    fn a_baseline_that_lost_half_its_schemas_is_refused() {
273        let path = temp_path();
274        let error = baseline(vec![event("RAW"), unreadable("TEXT_MESSAGE_START")])
275            .save(&path)
276            .unwrap_err();
277        assert!(error.contains("only 1 of 2 schemas"), "{error}");
278    }
279
280    #[test]
281    fn an_empty_baseline_is_refused() {
282        let path = temp_path();
283        let error = baseline(vec![]).save(&path).unwrap_err();
284        assert!(error.contains("names no event types"), "{error}");
285        assert!(!path.exists());
286    }
287
288    /// A refused write must leave the reviewed file exactly as it was, rather
289    /// than truncating it on the way to failing.
290    #[test]
291    fn a_refused_write_leaves_the_previous_baseline_intact() {
292        let path = temp_path();
293        let good = baseline(vec![event("RAW"), event("TEXT_MESSAGE_START")]);
294        good.save(&path).unwrap();
295        let before = std::fs::read_to_string(&path).unwrap();
296
297        baseline(vec![unreadable("RAW"), unreadable("TEXT_MESSAGE_START")])
298            .save(&path)
299            .unwrap_err();
300
301        assert_eq!(std::fs::read_to_string(&path).unwrap(), before);
302        assert!(
303            !path.with_extension("json.tmp").exists(),
304            "no temporary file should be left behind"
305        );
306    }
307
308    #[test]
309    fn a_baseline_from_an_older_format_is_reported_not_misread() {
310        let path = temp_path();
311        let mut old = baseline(vec![event("RAW")]);
312        old.format = FORMAT + 1;
313        // `save` guards content, not format, so write the file directly.
314        let mut json = serde_json::to_string_pretty(&old).unwrap();
315        json.push('\n');
316        std::fs::write(&path, json).unwrap();
317
318        let error = Baseline::load(&path).unwrap_err();
319        assert!(error.contains("was written in format"), "{error}");
320    }
321
322    #[test]
323    fn a_baseline_that_is_not_json_is_reported() {
324        let path = temp_path();
325        std::fs::write(&path, "<!DOCTYPE html><title>404</title>").unwrap();
326        let error = Baseline::load(&path).unwrap_err();
327        assert!(error.contains("is not a valid baseline"), "{error}");
328    }
329
330    #[test]
331    fn a_missing_baseline_says_how_to_make_one() {
332        let error = Baseline::load(Path::new("/nonexistent/events.json")).unwrap_err();
333        assert!(error.contains("drift-check --refresh"), "{error}");
334    }
335
336    /// End to end for the failure that motivates [`Baseline::validate`]: a
337    /// response cut off after the enum parses cleanly all the way through
338    /// `from_upstream`, and is caught only at the point of writing.
339    #[test]
340    fn a_response_truncated_after_the_enum_never_reaches_disk() {
341        let truncated = r#"
342import { z } from "zod";
343
344export enum EventType {
345  TEXT_MESSAGE_START = "TEXT_MESSAGE_START",
346  TEXT_MESSAGE_END = "TEXT_MESSAGE_END",
347  RAW = "RAW",
348}
349"#;
350        // It parses: the event types are all there.
351        let extracted = upstream::extract(truncated).unwrap();
352        assert_eq!(extracted.event_types.len(), 3);
353        // ...and every schema is unreadable, which the comparison treats as a
354        // warning, so nothing downstream would have objected.
355        assert!(extracted.events.iter().all(|e| e.unparsed.is_some()));
356
357        let path = temp_path();
358        let source = Source {
359            repo: UPSTREAM_REPO.into(),
360            path: UPSTREAM_PATH.into(),
361            commit: "0123456789abcdef".into(),
362            commit_date: "2026-08-01".into(),
363            fetched_at: "2026-08-01".into(),
364        };
365        let error = Baseline::from_upstream(&extracted, source)
366            .save(&path)
367            .unwrap_err();
368        assert!(error.contains("0 of 3 schemas"), "{error}");
369        assert!(!path.exists());
370    }
371}
xtask/drift/baseline.rs

xtask/drift/
baseline.rs
