majordomus_cli/capability/builtin/
observatory.rs

1//! The `observatory` module: what the Observatory (`/ui/observatory`) may carry to a place
2//! where no mesh runs, and the recording the repository carries there (ADR 0158, I2678).
3//!
4//! The reader of a published snapshot is a process that runs no mesh, so the three scenes of
5//! the mesh have nothing live to draw on the public site. They may draw a **recording**: the
6//! scenes of a process that did run one, captured at a moment, made fit for publication, and
7//! committed. A recording is drawn only where no mesh runs, and it is labelled a replay with
8//! the time it was recorded; it is never drawn as live.
9//!
10//! - `observatory.capture` answers the recording this process would make now: the mesh
11//!   scenes it has live, pseudonymized. It writes nothing; whoever publishes it commits it.
12//! - `observatory.replay` answers the recording the repository carries at [`RECORDING`].
13//!
14//! Fit for publication means that a recording names no machine, runtime, session or event by
15//! its identity: machines are `machine 1`, `machine 2` in the order the mesh listed them,
16//! runtimes and sessions are ordinals, and an event keeps only its kind, its logical clock and
17//! its time. What remains is the shape of the cooperation, which is what the scene draws.
18//!
19//! The lifecycle, from capture to replay: a process that runs the mesh projects its scenes,
20//! the projection is redacted with a manifest, the recording is committed, and a process that
21//! runs none reads it back.
22//!
23//! ```
24//! use majordomus_cli::capability::builtin::observatory::{redact, replay_at, RECORDING, RECORDING_SCHEMA};
25//! use serde_json::json;
26//! let live = [("mesh".to_string(), json!({
27//!     "machines": [{ "id": "641bdb94dae9f622", "label": "641bdb94", "local": true, "runtimes": [] }],
28//!     "known": 1, "refused": 0,
29//! }))].into_iter().collect();
30//! let scenes = redact(&live);
31//! let dir = tempfile::tempdir().unwrap();
32//! assert!(!replay_at(dir.path()).unwrap().present, "nothing recorded yet");
33//! let recording = json!({
34//!     "schema": RECORDING_SCHEMA, "recorded_at": "2026-10-14T05:00:00Z", "scenes": scenes,
35//!     "manifest": { "sources": ["mesh.peers"], "removed": ["identities"] },
36//! });
37//! let path = dir.path().join(RECORDING);
38//! std::fs::create_dir_all(path.parent().unwrap()).unwrap();
39//! std::fs::write(&path, recording.to_string()).unwrap();
40//! let replayed = replay_at(dir.path()).unwrap().recording.unwrap();
41//! assert_eq!(replayed.scenes["mesh"]["machines"][0]["label"], "machine 1");
42//! ```
43
44use std::collections::BTreeMap;
45use std::path::Path;
46
47use schemars::JsonSchema;
48use serde::{Deserialize, Serialize};
49use serde_json::Value;
50
51use crate::capability::benchmark::{BenchmarkCases, CaseContext, NamedCase};
52use crate::capability::handler::{CapabilityError, Context};
53use crate::capability::model::{Exposure, Stability};
54use crate::capability::module::ModuleDescriptor;
55use crate::{capability, module};
56
57use super::{get, mcp, Empty};
58
59/// Where the repository carries its recording, relative to the repository root.
60pub const RECORDING: &str = ".ai/repo/observatory/recording.json";
61
62/// The schema word of a recording.
63pub const RECORDING_SCHEMA: &str = "observatory-recording/v1";
64
65/// The scenes a recording may carry: the ones that need a mesh.
66pub const RECORDED: &[&str] = &["mesh", "activity", "coop"];
67
68/// The mesh scenes of one process at one moment, fit for publication.
69///
70/// ```
71/// use majordomus_cli::capability::builtin::observatory::ObservatoryRecording;
72/// let r: ObservatoryRecording = serde_json::from_value(serde_json::json!({
73///     "schema": "observatory-recording/v1", "recorded_at": "2026-10-14T05:00:00Z", "scenes": {},
74///     "manifest": { "sources": [], "removed": [] },
75/// })).unwrap();
76/// assert!(r.scenes.is_empty() && r.revision.is_none());
77/// ```
78#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
79pub struct ObservatoryRecording {
80    /// [`RECORDING_SCHEMA`].
81    pub schema: String,
82    /// When the scenes were read.
83    pub recorded_at: String,
84    /// The commit the recording process stood at, short; absent when it could not say.
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub revision: Option<String>,
87    /// Each recorded scene's model, by scene id; a scene the process had nothing live for is
88    /// absent, never empty.
89    pub scenes: BTreeMap<String, Value>,
90    /// What the recording is a record of, and what it leaves out.
91    pub manifest: RecordingManifest,
92}
93
94/// What a recording is a record of: the capabilities read, the window of events it holds,
95/// and what was removed from what they answered, with the reason.
96///
97/// ```
98/// use majordomus_cli::capability::builtin::observatory::RecordingManifest;
99/// let m: RecordingManifest = serde_json::from_value(serde_json::json!({
100///     "sources": ["mesh.events"], "removed": ["event bodies"],
101/// })).unwrap();
102/// assert!(m.window.is_none(), "a recording that holds no event has no window");
103/// ```
104#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
105pub struct RecordingManifest {
106    /// The capabilities whose answers the recorded scenes were projected from.
107    pub sources: Vec<String>,
108    /// The first and last event the recording holds, by time and by logical clock; absent
109    /// when it holds none.
110    #[serde(default, skip_serializing_if = "Option::is_none")]
111    pub window: Option<RecordingWindow>,
112    /// Every kind of thing removed before publication, and why.
113    pub removed: Vec<String>,
114}
115
116/// The events a recording holds, from first to last.
117///
118/// ```
119/// use majordomus_cli::capability::builtin::observatory::RecordingWindow;
120/// let w = RecordingWindow { events: 2, from_at: Some(100), to_at: Some(160), from_lamport: Some(9), to_lamport: Some(10) };
121/// assert_eq!(w.to_at.unwrap() - w.from_at.unwrap(), 60, "a minute of the journal");
122/// ```
123#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
124pub struct RecordingWindow {
125    /// How many events.
126    pub events: usize,
127    /// The earliest event's time, in seconds since the epoch.
128    pub from_at: Option<u64>,
129    /// The latest event's time, in seconds since the epoch.
130    pub to_at: Option<u64>,
131    /// The smallest logical clock.
132    pub from_lamport: Option<u64>,
133    /// The largest logical clock.
134    pub to_lamport: Option<u64>,
135}
136
137/// What every recording removes, and why: the manifest states it rather than leaving a reader
138/// to infer it from what is missing.
139pub const REMOVED: &[&str] = &[
140    "machine, runtime and session identities: replaced by ordinals, because a node id names a machine and a session key names a person's work",
141    "machine names: replaced by `machine N` in the order the mesh listed them",
142    "event bodies: an intent, a path or a message is the capability's to show to a reader of the repository, not a public page's",
143    "event identities (stream and sequence): replaced by ordinals in the recorded order",
144];
145
146/// The manifest of a set of redacted scenes.
147fn manifest(scenes: &BTreeMap<String, Value>) -> RecordingManifest {
148    let sources = ["mesh.peers", "mesh.events", "mesh.state"]
149        .into_iter()
150        .zip(RECORDED)
151        .filter(|(_, scene)| scenes.contains_key(**scene))
152        .map(|(capability, _)| capability.to_string())
153        .collect();
154    let events: Vec<&Value> = scenes
155        .get("activity")
156        .and_then(|a| a.get("events"))
157        .and_then(Value::as_array)
158        .map(|v| v.iter().collect())
159        .unwrap_or_default();
160    let of = |key: &str| {
161        events
162            .iter()
163            .filter_map(|e| e.get(key).and_then(Value::as_u64))
164            .collect::<Vec<_>>()
165    };
166    let (at, lamport) = (of("at"), of("lamport"));
167    let window = (!events.is_empty()).then(|| RecordingWindow {
168        events: events.len(),
169        from_at: at.iter().min().copied(),
170        to_at: at.iter().max().copied(),
171        from_lamport: lamport.iter().min().copied(),
172        to_lamport: lamport.iter().max().copied(),
173    });
174    RecordingManifest {
175        sources,
176        window,
177        removed: REMOVED.iter().map(|r| r.to_string()).collect(),
178    }
179}
180
181/// The recording the repository carries, or that it carries none.
182///
183/// ```
184/// use majordomus_cli::capability::builtin::observatory::{replay_at, ObservatoryReplay, RECORDING};
185/// let dir = tempfile::tempdir().unwrap();
186/// let none: ObservatoryReplay = replay_at(dir.path()).unwrap();
187/// assert!(!none.present && none.recording.is_none());
188/// assert_eq!(none.path, RECORDING);
189/// ```
190#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
191pub struct ObservatoryReplay {
192    /// Whether the repository carries a recording.
193    pub present: bool,
194    /// The file it is read from.
195    pub path: String,
196    /// The recording.
197    #[serde(default, skip_serializing_if = "Option::is_none")]
198    pub recording: Option<ObservatoryRecording>,
199}
200
201/// The scenes of a projection made fit for publication: identities replaced by ordinals,
202/// event bodies and names never carried. A scene this does not know is dropped.
203///
204/// ```
205/// use majordomus_cli::capability::builtin::observatory::redact;
206/// use serde_json::json;
207/// let scenes = [("mesh".to_string(), json!({
208///     "machines": [{ "id": "641bdb94dae9f622", "label": "641bdb94", "local": true,
209///                    "runtimes": [{ "id": "b10f19aa2cd888a6", "liveness": "live", "this": true, "sessions": 2 }] }],
210///     "known": 1, "refused": 0,
211/// }))].into_iter().collect();
212/// let out = redact(&scenes);
213/// let text = serde_json::to_string(&out).unwrap();
214/// assert!(!text.contains("641bdb94") && !text.contains("b10f19aa"));
215/// assert_eq!(out["mesh"]["machines"][0]["label"], "machine 1");
216/// assert_eq!(out["mesh"]["machines"][0]["runtimes"][0]["liveness"], "live");
217/// ```
218pub fn redact(scenes: &BTreeMap<String, Value>) -> BTreeMap<String, Value> {
219    let list = |v: &Value, key: &str| {
220        v.get(key)
221            .and_then(Value::as_array)
222            .cloned()
223            .unwrap_or_default()
224    };
225    let prefix = |id: &str| id.chars().take(12).collect::<String>();
226    // a machine is named by the order the mesh listed it, and every node prefix maps to it
227    let mut machines: BTreeMap<String, String> = BTreeMap::new();
228    let mut out = BTreeMap::new();
229    if let Some(mesh) = scenes.get("mesh") {
230        let listed = list(mesh, "machines");
231        let redacted: Vec<Value> = listed
232            .iter()
233            .enumerate()
234            .map(|(i, m)| {
235                let label = format!("machine {}", i + 1);
236                let id = m.get("id").and_then(Value::as_str).unwrap_or_default();
237                machines.insert(prefix(id), label.clone());
238                let runtimes: Vec<Value> = list(m, "runtimes")
239                    .iter()
240                    .enumerate()
241                    .map(|(j, r)| {
242                        serde_json::json!({
243                            "id": format!("m{}-r{}", i + 1, j + 1),
244                            "liveness": r.get("liveness").cloned().unwrap_or(Value::Null),
245                            "this": r.get("this").cloned().unwrap_or(Value::Bool(false)),
246                            "sessions": r.get("sessions").cloned().unwrap_or(Value::Null),
247                        })
248                    })
249                    .collect();
250                serde_json::json!({
251                    "id": format!("m{}", i + 1),
252                    "label": label,
253                    "local": m.get("local").cloned().unwrap_or(Value::Null),
254                    "runtimes": runtimes,
255                })
256            })
257            .collect();
258        out.insert(
259            "mesh".to_string(),
260            serde_json::json!({
261                "machines": redacted,
262                "known": mesh.get("known").cloned().unwrap_or(Value::Null),
263                "refused": mesh.get("refused").cloned().unwrap_or(Value::Null),
264            }),
265        );
266    }
267    let machine = |node: Option<&str>| match node.map(prefix).and_then(|p| machines.get(&p)) {
268        Some(label) => Value::String(label.clone()),
269        None => Value::String("another machine".into()),
270    };
271    if let Some(activity) = scenes.get("activity") {
272        let events: Vec<Value> = list(activity, "events")
273            .iter()
274            .enumerate()
275            .map(|(k, e)| {
276                serde_json::json!({
277                    "id": format!("e{}", k + 1),
278                    "kind": e.get("kind").cloned().unwrap_or(Value::Null),
279                    "node": machine(e.get("node").and_then(Value::as_str)),
280                    "lamport": e.get("lamport").cloned().unwrap_or(Value::Null),
281                    "at": e.get("at").cloned().unwrap_or(Value::Null),
282                })
283            })
284            .collect();
285        out.insert(
286            "activity".to_string(),
287            serde_json::json!({
288                "events": events,
289                "lamport": activity.get("lamport").cloned().unwrap_or(Value::Null),
290            }),
291        );
292    }
293    if let Some(coop) = scenes.get("coop") {
294        let sessions: Vec<Value> = list(coop, "sessions")
295            .iter()
296            .enumerate()
297            .map(|(k, s)| {
298                serde_json::json!({
299                    "id": format!("s{}", k + 1),
300                    "node": machine(s.get("node").and_then(Value::as_str)),
301                    "state": s.get("state").cloned().unwrap_or(Value::Null),
302                    "claims": s.get("claims").cloned().unwrap_or(Value::Null),
303                    "reviews": s.get("reviews").cloned().unwrap_or(Value::Null),
304                })
305            })
306            .collect();
307        let mut model = serde_json::json!({ "sessions": sessions });
308        for key in ["claims", "overlaps", "handovers", "reviews"] {
309            model[key] = coop.get(key).cloned().unwrap_or(Value::Null);
310        }
311        out.insert("coop".to_string(), model);
312    }
313    out
314}
315
316fn capture(ctx: &Context, _: Empty) -> Result<ObservatoryRecording, CapabilityError> {
317    let (projection, _) = crate::ui::observatory::project(ctx);
318    // only what this process has live: a scene drawn from a replay is never recorded again
319    let live: BTreeMap<String, Value> = projection
320        .scenes
321        .iter()
322        .filter(|s| RECORDED.contains(&s.id) && s.available && s.replay.is_none())
323        .map(|s| (s.id.to_string(), s.model.clone()))
324        .collect();
325    let scenes = redact(&live);
326    Ok(ObservatoryRecording {
327        schema: RECORDING_SCHEMA.into(),
328        recorded_at: projection.observed_at,
329        revision: projection.revision,
330        manifest: manifest(&scenes),
331        scenes,
332    })
333}
334
335/// The recording at [`RECORDING`] under `root`. Absent is `present: false`; a file that is
336/// there and cannot be read or is not a recording is an error, never an empty answer.
337///
338/// ```
339/// use majordomus_cli::capability::builtin::observatory::{replay_at, RECORDING};
340/// let dir = tempfile::tempdir().unwrap();
341/// let path = dir.path().join(RECORDING);
342/// std::fs::create_dir_all(path.parent().unwrap()).unwrap();
343/// std::fs::write(&path, "{}").unwrap();
344/// assert!(replay_at(dir.path()).is_err(), "a file that is not a recording is refused");
345/// ```
346pub fn replay_at(root: &Path) -> Result<ObservatoryReplay, CapabilityError> {
347    let path = root.join(RECORDING);
348    let text = match std::fs::read_to_string(&path) {
349        Ok(text) => text,
350        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
351            return Ok(ObservatoryReplay {
352                present: false,
353                path: RECORDING.into(),
354                recording: None,
355            })
356        }
357        Err(e) => {
358            return Err(CapabilityError::Internal(format!(
359                "{RECORDING} could not be read: {e}"
360            )))
361        }
362    };
363    let recording: ObservatoryRecording = serde_json::from_str(&text)
364        .map_err(|e| CapabilityError::Internal(format!("{RECORDING} is not a recording: {e}")))?;
365    if recording.schema != RECORDING_SCHEMA {
366        return Err(CapabilityError::Internal(format!(
367            "{RECORDING} declares schema {}, not {RECORDING_SCHEMA}",
368            recording.schema
369        )));
370    }
371    if let Some(other) = recording
372        .scenes
373        .keys()
374        .find(|k| !RECORDED.contains(&k.as_str()))
375    {
376        return Err(CapabilityError::Internal(format!(
377            "{RECORDING} carries the scene {other}, which is not one a recording may carry"
378        )));
379    }
380    // an event is drawn by its kind, clock and time; one that lacks any of them was not
381    // recorded from a source record, and the recording is refused naming it
382    let events = recording
383        .scenes
384        .get("activity")
385        .and_then(|a| a.get("events"))
386        .and_then(Value::as_array)
387        .cloned()
388        .unwrap_or_default();
389    for (i, e) in events.iter().enumerate() {
390        let whole = e.get("kind").is_some_and(Value::is_string)
391            && e.get("lamport").is_some_and(Value::is_u64)
392            && e.get("at").is_some_and(Value::is_u64);
393        if !whole {
394            let id = e
395                .get("id")
396                .and_then(Value::as_str)
397                .map(str::to_string)
398                .unwrap_or_else(|| format!("#{}", i + 1));
399            return Err(CapabilityError::Internal(format!(
400                "{RECORDING} holds the event {id} without its kind, logical clock and time: it was not recorded from a source record"
401            )));
402        }
403    }
404    Ok(ObservatoryReplay {
405        present: true,
406        path: RECORDING.into(),
407        recording: Some(recording),
408    })
409}
410
411/// The input of `observatory.events`: a cursor and a page size.
412///
413/// ```
414/// use majordomus_cli::capability::builtin::observatory::ObservatoryEventsInput;
415/// let q: ObservatoryEventsInput = serde_json::from_value(serde_json::json!({ "after": 42 })).unwrap();
416/// assert_eq!((q.after, q.limit), (Some(42), None));
417/// assert!(serde_json::from_value::<ObservatoryEventsInput>(serde_json::json!({ "since": 1 })).is_err());
418/// ```
419#[derive(Debug, Default, Serialize, Deserialize, JsonSchema)]
420#[serde(deny_unknown_fields)]
421pub struct ObservatoryEventsInput {
422    /// The cursor: only events whose logical clock is above it. Absent is the newest page.
423    #[serde(default, skip_serializing_if = "Option::is_none")]
424    pub after: Option<u64>,
425    /// At most this many (default 200, at most 1000).
426    #[serde(default, skip_serializing_if = "Option::is_none")]
427    pub limit: Option<u64>,
428}
429
430impl BenchmarkCases for ObservatoryEventsInput {
431    fn benchmark_cases(_: &CaseContext<'_>) -> Vec<NamedCase<Self>> {
432        vec![
433            NamedCase::new("newest", ObservatoryEventsInput::default()),
434            NamedCase::new(
435                "after a cursor",
436                ObservatoryEventsInput {
437                    after: Some(0),
438                    limit: Some(50),
439                },
440            ),
441        ]
442    }
443}
444
445/// The Observatory's events above a cursor, as the activity scene draws them.
446///
447/// ```
448/// use majordomus_cli::capability::builtin::observatory::ObservatoryEvents;
449/// let e: ObservatoryEvents = serde_json::from_value(serde_json::json!({
450///     "active": true, "after": 9, "cursor": 12, "expired": false, "events": [{ "id": "s:3" }],
451/// })).unwrap();
452/// assert!(e.cursor > e.after.unwrap(), "the next request resumes past what was read");
453/// ```
454#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
455pub struct ObservatoryEvents {
456    /// Whether a mesh runs in this process; without one there is no journal to follow.
457    pub active: bool,
458    /// The cursor that was asked, when one was.
459    #[serde(default, skip_serializing_if = "Option::is_none")]
460    pub after: Option<u64>,
461    /// The cursor to ask next: the logical clock of the newest event read, or the journal's
462    /// newest clock when the cursor had expired.
463    pub cursor: u64,
464    /// The cursor names a clock this journal never reached (another runtime's journal, or one
465    /// that restarted): nothing above it can be read, and the reader takes a fresh projection.
466    pub expired: bool,
467    /// The events above the cursor, newest first, as marks: identity, kind, node, logical
468    /// clock and time, never a body.
469    pub events: Vec<Value>,
470}
471
472fn events(
473    ctx: &Context,
474    input: ObservatoryEventsInput,
475) -> Result<ObservatoryEvents, CapabilityError> {
476    let limit = input.limit.unwrap_or(200).clamp(1, 1000);
477    let newest = ctx.execute(
478        "mesh.events",
479        serde_json::json!({ "newest": true, "limit": limit }),
480    )?;
481    let active = newest.get("active").and_then(Value::as_bool) == Some(true);
482    let head = newest.get("lamport").and_then(Value::as_u64).unwrap_or(0);
483    let list = |v: &Value| {
484        v.get("events")
485            .and_then(Value::as_array)
486            .cloned()
487            .unwrap_or_default()
488    };
489    let Some(after) = input.after else {
490        return Ok(ObservatoryEvents {
491            active,
492            after: None,
493            cursor: head,
494            expired: false,
495            events: crate::ui::observatory::event_marks(&list(&newest)),
496        });
497    };
498    if !active || after > head {
499        return Ok(ObservatoryEvents {
500            active,
501            after: Some(after),
502            cursor: head,
503            expired: active,
504            events: Vec::new(),
505        });
506    }
507    let page = ctx.execute(
508        "mesh.events",
509        serde_json::json!({ "after": after, "limit": limit }),
510    )?;
511    Ok(ObservatoryEvents {
512        active,
513        after: Some(after),
514        cursor: page.get("lamport").and_then(Value::as_u64).unwrap_or(after),
515        expired: false,
516        events: crate::ui::observatory::event_marks(&list(&page)),
517    })
518}
519
520fn replay(ctx: &Context, _: Empty) -> Result<ObservatoryReplay, CapabilityError> {
521    replay_at(Path::new(&ctx.index.repository.root))
522}
523
524/// The `observatory` module: three read-only capabilities, none cached, because each is read
525/// to learn what is true now: the capture of this process's mesh, the recording the
526/// repository carries, and the journal's events above a cursor.
527///
528/// ```
529/// use majordomus_cli::capability::builtin::observatory::module;
530/// use majordomus_cli::capability::Effect;
531/// let m = module();
532/// let ids: Vec<&str> = m.capabilities.iter().map(|e| e.capability.id.as_str()).collect();
533/// assert_eq!(ids, ["observatory.capture", "observatory.replay", "observatory.events"]);
534/// assert!(m.capabilities.iter().all(|e| e.capability.execution.effect == Effect::Read));
535/// ```
536pub fn module() -> ModuleDescriptor {
537    module! {
538        id: "observatory",
539        title: "Observatory",
540        description: "What the Observatory may carry to a place where no mesh runs: the mesh scenes this process has live, pseudonymized for publication, and the recording the repository carries, which the published Observatory replays and labels as a replay.",
541        stability: Stability::BehaviorallyVerified,
542        capabilities: [
543            capability! {
544                id: "observatory.capture",
545                title: "Observatory capture",
546                description: "The mesh scenes of the Observatory as this process has them live, fit for publication: machines, runtimes, sessions and events by ordinal, an event by its kind, logical clock and time only. Writes nothing; committing the answer to .ai/repo/observatory/recording.json is what publishes it.",
547                input: Empty,
548                output: ObservatoryRecording,
549                stability: Stability::BehaviorallyVerified,
550                exposure: Exposure { mcp: mcp("majordomus_observatory_capture"), http: get("/api/v1/observatory/capture"), cli: None },
551                tags: ["observatory", "mesh", "ui"],
552                handler: capture,
553            },
554            capability! {
555                id: "observatory.replay",
556                title: "Observatory replay",
557                description: "The recording of the mesh scenes the repository carries at .ai/repo/observatory/recording.json, which the Observatory draws, labelled as a replay, where no mesh runs. Absent is present: false; a file that cannot be read is an error.",
558                input: Empty,
559                output: ObservatoryReplay,
560                stability: Stability::BehaviorallyVerified,
561                exposure: Exposure { mcp: mcp("majordomus_observatory_replay"), http: get("/api/v1/observatory/replay"), cli: None },
562                tags: ["observatory", "mesh", "ui"],
563                handler: replay,
564            },
565            capability! {
566                id: "observatory.events",
567                title: "Observatory events",
568                description: "The events of the mesh journal above a cursor, as the Observatory's activity scene draws them: identity, kind, node, logical clock and time, newest first, at most a page. The answer's cursor is the next request's `after`; a cursor this journal never reached is expired, and the reader takes a fresh projection. Without a cursor, the newest page.",
569                input: ObservatoryEventsInput,
570                output: ObservatoryEvents,
571                stability: Stability::BehaviorallyVerified,
572                exposure: Exposure { mcp: mcp("majordomus_observatory_events"), http: get("/api/v1/observatory/events"), cli: None },
573                tags: ["observatory", "mesh", "ui"],
574                handler: events,
575            },
576        ],
577    }
578}
579
580#[cfg(test)]
581mod tests {
582    use super::*;
583    use serde_json::json;
584
585    fn scenes() -> BTreeMap<String, Value> {
586        [
587            (
588                "mesh".to_string(),
589                json!({ "machines": [
590                    { "id": "aaaaaaaaaaaa1111", "label": "aaaaaaaa", "local": true,
591                      "runtimes": [{ "id": "rt-secret-1", "liveness": "live", "this": true, "sessions": 1 }] },
592                    { "id": "bbbbbbbbbbbb2222", "label": "bbbbbbbb", "local": false, "runtimes": [] },
593                ], "known": 2, "refused": 1 }),
594            ),
595            (
596                "activity".to_string(),
597                json!({ "events": [
598                    { "id": "aaaaaaaaaaaa1111-x:7", "kind": "claim_acquired", "node": "bbbbbbbbbbbb", "lamport": 9, "at": 100 },
599                    { "id": "cccc:1", "kind": "session_opened", "node": "cccccccccccc", "lamport": 10, "at": 101 },
600                ], "lamport": 10 }),
601            ),
602            (
603                "coop".to_string(),
604                json!({ "sessions": [{ "id": "sess-secret", "node": "aaaaaaaaaaaa", "state": "active", "claims": 3, "reviews": 0 }],
605                        "claims": 3, "overlaps": 0, "handovers": 1, "reviews": 0 }),
606            ),
607            ("heartbeat".to_string(), json!({ "checks": [] })),
608        ]
609        .into_iter()
610        .collect()
611    }
612
613    /// A recording carries the shape of the cooperation and no identity of it: no id, label
614    /// or session key of the input survives, events keep kind, clock and time, and a node is
615    /// the machine it was listed as, or "another machine".
616    #[test]
617    fn a_recording_names_nothing_by_its_identity() {
618        let out = redact(&scenes());
619        let text = serde_json::to_string(&out).unwrap();
620        for secret in ["aaaaaaaa", "bbbbbbbb", "cccc", "rt-secret", "sess-secret"] {
621            assert!(!text.contains(secret), "{secret} survived: {text}");
622        }
623        assert_eq!(out.keys().collect::<Vec<_>>(), ["activity", "coop", "mesh"]);
624        assert_eq!(out["mesh"]["machines"][1]["label"], "machine 2");
625        assert_eq!(out["mesh"]["machines"][0]["runtimes"][0]["id"], "m1-r1");
626        assert_eq!(out["activity"]["events"][0]["node"], "machine 2");
627        assert_eq!(out["activity"]["events"][1]["node"], "another machine");
628        assert_eq!(out["activity"]["events"][0]["kind"], "claim_acquired");
629        assert_eq!(out["activity"]["events"][1]["lamport"], 10);
630        assert_eq!(out["coop"]["sessions"][0]["node"], "machine 1");
631        assert_eq!(out["coop"]["handovers"], 1);
632    }
633
634    /// No file is no recording; a file that is not a recording, or carries a scene a
635    /// recording may not, is an error and never an empty answer.
636    #[test]
637    fn the_replay_is_the_file_or_says_why_not() {
638        let dir = tempfile::tempdir().unwrap();
639        let absent = replay_at(dir.path()).unwrap();
640        assert!(!absent.present && absent.recording.is_none());
641        let path = dir.path().join(RECORDING);
642        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
643        std::fs::write(&path, "not json").unwrap();
644        assert!(replay_at(dir.path()).is_err());
645        let redacted = redact(&scenes());
646        let mut recording = ObservatoryRecording {
647            schema: RECORDING_SCHEMA.into(),
648            recorded_at: "2026-10-14T05:00:00Z".into(),
649            revision: Some("abc".into()),
650            manifest: manifest(&redacted),
651            scenes: redacted,
652        };
653        // the manifest names what was read, the window held and what was removed
654        assert_eq!(
655            recording.manifest.sources,
656            ["mesh.peers", "mesh.events", "mesh.state"]
657        );
658        let window = recording.manifest.window.clone().expect("a window");
659        assert_eq!(
660            (window.events, window.from_lamport, window.to_lamport),
661            (2, Some(9), Some(10))
662        );
663        assert_eq!((window.from_at, window.to_at), (Some(100), Some(101)));
664        assert_eq!(recording.manifest.removed.len(), REMOVED.len());
665        std::fs::write(&path, serde_json::to_string(&recording).unwrap()).unwrap();
666        let read = replay_at(dir.path()).unwrap();
667        assert!(read.present);
668        assert_eq!(read.recording.as_ref(), Some(&recording));
669        // an event without its clock is refused, by name
670        let mut partial = recording.clone();
671        partial.scenes.get_mut("activity").unwrap()["events"][1]
672            .as_object_mut()
673            .unwrap()
674            .remove("lamport");
675        std::fs::write(&path, serde_json::to_string(&partial).unwrap()).unwrap();
676        match replay_at(dir.path()) {
677            Err(CapabilityError::Internal(m)) => assert!(m.contains("the event e2 "), "{m}"),
678            other => panic!("expected a refusal naming e2, got {other:?}"),
679        }
680        recording.scenes.insert("heartbeat".into(), json!({}));
681        std::fs::write(&path, serde_json::to_string(&recording).unwrap()).unwrap();
682        assert!(
683            replay_at(dir.path()).is_err(),
684            "a recording carries mesh scenes only"
685        );
686    }
687}
majordomus_cli/capability/builtin/observatory.rs

majordomus_cli/capability/builtin/
observatory.rs
