majordomus_cli/execution/
sink.rs

1//! What a handler reports with, and how it learns it should stop.
2//!
3//! A handler is written once and reached by every transport. Most of them have no
4//! execution behind them — a command line call, an MCP tool call, a benchmark sample — so
5//! reporting has to be free to call and do nothing when nobody is listening. That is what
6//! [`Progress`] is: a handle every [`crate::capability::Context`] carries, silent unless
7//! the call came through the execution engine.
8//!
9//! The consequence is the point of the design. `health.report` does not know whether it is
10//! answering `GET /api/v1/health`, an MCP tool call or a browser watching it run; it
11//! reports its steps the same way in all three, and only the execution engine turns those
12//! reports into events anybody can see.
13
14use std::sync::atomic::{AtomicBool, Ordering};
15use std::sync::Arc;
16
17use super::event::{sanitise, EventPayload};
18use super::model::{ExecutionDiagnostic, ExecutionId, LogStream, ProgressView};
19use super::store::ExecutionStore;
20
21/// The one thing a handler is given to say what it is doing.
22///
23/// Cheap to clone and safe to hold: cloning it does not create a second stream, and a
24/// silent handle costs nothing to call.
25///
26/// ```
27/// use majordomus_cli::execution::Progress;
28/// // what a handler outside an execution is given
29/// let silent = Progress::silent();
30/// silent.step("scan", "Scanning");
31/// silent.progress(1, Some(2), "half way");
32/// silent.step_done("scan", true, None);
33/// assert!(!silent.cancelled(), "nothing can cancel a call that is not an execution");
34/// assert!(!silent.is_reporting());
35/// ```
36#[derive(Clone, Default)]
37pub struct Progress(Option<Arc<Reporter>>);
38
39struct Reporter {
40    store: Arc<ExecutionStore>,
41    id: ExecutionId,
42    cancel: Arc<AtomicBool>,
43    max_log_chars: usize,
44}
45
46impl std::fmt::Debug for Progress {
47    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
48        match &self.0 {
49            Some(r) => write!(f, "Progress({})", r.id),
50            None => f.write_str("Progress(silent)"),
51        }
52    }
53}
54
55impl Progress {
56    /// A handle that reports nothing: what every call outside an execution is given.
57    pub fn silent() -> Self {
58        Progress(None)
59    }
60
61    /// A handle that publishes into a store, for one execution.
62    pub fn reporting(store: Arc<ExecutionStore>, id: ExecutionId, cancel: Arc<AtomicBool>) -> Self {
63        let max_log_chars = store.limits().max_log_chars;
64        Progress(Some(Arc::new(Reporter {
65            store,
66            id,
67            cancel,
68            max_log_chars,
69        })))
70    }
71
72    /// Is anybody listening? A handler that would do real work to produce a report — count
73    /// the files it is about to walk, say — asks first.
74    pub fn is_reporting(&self) -> bool {
75        self.0.is_some()
76    }
77
78    /// The execution this reports into, when there is one.
79    pub fn execution_id(&self) -> Option<&ExecutionId> {
80        self.0.as_ref().map(|r| &r.id)
81    }
82
83    /// Has this execution been asked to stop?
84    ///
85    /// Cancellation is cooperative and this is the whole of it: a handler that never asks
86    /// runs to completion, and the capability's own policy is what tells a client whether
87    /// asking will achieve anything.
88    pub fn cancelled(&self) -> bool {
89        self.0
90            .as_ref()
91            .is_some_and(|r| r.cancel.load(Ordering::SeqCst))
92    }
93
94    /// The error a handler returns when it stops because it was asked to.
95    pub fn cancellation(&self) -> crate::capability::CapabilityError {
96        crate::capability::CapabilityError::Refused("the execution was cancelled".into())
97    }
98
99    /// Enter a named phase.
100    pub fn step(&self, name: &str, title: &str) {
101        self.emit(EventPayload::StepStarted {
102            name: name.to_string(),
103            title: title.to_string(),
104        });
105    }
106
107    /// Leave a named phase, saying whether what it was asked to do happened.
108    pub fn step_done(&self, name: &str, ok: bool, detail: Option<String>) {
109        self.emit(EventPayload::StepCompleted {
110            name: name.to_string(),
111            ok,
112            detail: detail.map(|d| sanitise(&d, self.limit())),
113        });
114    }
115
116    /// How far along.
117    pub fn progress(&self, current: u64, total: Option<u64>, message: impl AsRef<str>) {
118        let message = message.as_ref();
119        self.emit(EventPayload::Progress(ProgressView {
120            current,
121            total,
122            message: (!message.is_empty()).then(|| sanitise(message, self.limit())),
123        }));
124    }
125
126    /// A line of output, from the handler itself.
127    pub fn log(&self, message: impl AsRef<str>) {
128        self.stream(LogStream::Handler, message);
129    }
130
131    /// A line of output, from a named stream.
132    pub fn stream(&self, stream: LogStream, message: impl AsRef<str>) {
133        let limit = self.limit();
134        self.emit(EventPayload::Log {
135            stream,
136            message: sanitise(message.as_ref(), limit),
137        });
138    }
139
140    /// A structured finding.
141    pub fn diagnostic(&self, diagnostic: ExecutionDiagnostic) {
142        self.emit(EventPayload::Diagnostic(diagnostic));
143    }
144
145    fn limit(&self) -> usize {
146        self.0.as_ref().map(|r| r.max_log_chars).unwrap_or(2_000)
147    }
148
149    fn emit(&self, payload: EventPayload) {
150        if let Some(r) = &self.0 {
151            r.store.publish(&r.id, payload);
152        }
153    }
154}
155
156#[cfg(test)]
157mod tests {
158    use super::*;
159    use crate::execution::model::{Actor, ActorKind, RepositoryRef};
160    use crate::execution::store::{Delivery, Filter, Limits};
161
162    #[test]
163    fn a_silent_handle_reports_nothing_and_costs_nothing() {
164        let p = Progress::silent();
165        assert!(!p.is_reporting());
166        assert_eq!(p.execution_id(), None);
167        assert!(!p.cancelled());
168        p.step("a", "A");
169        p.progress(1, Some(2), "x");
170        p.log("y");
171        p.stream(LogStream::Stderr, "z");
172        p.step_done("a", true, Some("done".into()));
173        p.diagnostic(ExecutionDiagnostic {
174            severity: crate::model::Severity::Info,
175            code: "c".into(),
176            summary: "s".into(),
177            detail: None,
178            suggestion: None,
179        });
180        assert_eq!(format!("{p:?}"), "Progress(silent)");
181    }
182
183    #[test]
184    fn a_reporting_handle_publishes_what_the_handler_says_and_sanitises_it() {
185        let store = Arc::new(ExecutionStore::new(Limits::default()));
186        let id = ExecutionId::fresh();
187        let cancel = store.create(
188            id.clone(),
189            "demo.x",
190            "Demo",
191            serde_json::json!({}),
192            true,
193            Actor::of(ActorKind::Internal),
194            RepositoryRef {
195                name: "r".into(),
196                id: "i".into(),
197                branch: None,
198            },
199        );
200        let (_, rx) = store.subscribe(Filter::All);
201        let p = Progress::reporting(Arc::clone(&store), id.clone(), Arc::clone(&cancel));
202        assert!(p.is_reporting());
203        assert_eq!(p.execution_id(), Some(&id));
204        assert!(format!("{p:?}").contains(id.as_str()));
205        p.step("scan", "Scanning");
206        p.stream(LogStream::Stderr, "\u{1b}[31mred\u{1b}[0m");
207        p.step_done("scan", true, None);
208        let types: Vec<String> = rx
209            .try_iter()
210            .filter_map(|d| match d {
211                Delivery::Event(e) => Some(e.type_name().to_string()),
212                Delivery::Lagged { .. } => None,
213            })
214            .collect();
215        assert_eq!(
216            types,
217            [
218                "execution.step.started",
219                "execution.log",
220                "execution.step.completed"
221            ]
222        );
223        let page = store.events(&id, 1, 10).unwrap();
224        let log = page
225            .events
226            .iter()
227            .find(|e| e.type_name() == "execution.log")
228            .expect("the log event");
229        let value = serde_json::to_value(&**log).unwrap();
230        assert_eq!(
231            value["data"]["message"], "red",
232            "escapes never reach a client"
233        );
234        assert_eq!(value["data"]["stream"], "stderr");
235
236        assert!(!p.cancelled());
237        cancel.store(true, Ordering::SeqCst);
238        assert!(p.cancelled());
239        assert!(matches!(
240            p.cancellation(),
241            crate::capability::CapabilityError::Refused(_)
242        ));
243    }
244}
majordomus_cli/execution/sink.rs

majordomus_cli/execution/
sink.rs
