sloop/
protocol.rs

1use std::fmt;
2
3use serde::{Deserialize, Serialize};
4use serde_json::{Value, json};
5
6pub const PROTOCOL_VERSION: u32 = 1;
7
8#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(transparent)]
10pub struct RequestId(String);
11
12impl RequestId {
13    pub fn new(value: impl Into<String>) -> Self {
14        Self(value.into())
15    }
16
17    pub fn as_str(&self) -> &str {
18        &self.0
19    }
20}
21
22#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
23pub struct RequestEnvelope {
24    pub v: u32,
25    pub id: RequestId,
26    #[serde(flatten)]
27    pub request: Request,
28    pub token: Option<String>,
29}
30
31impl RequestEnvelope {
32    pub fn new(id: RequestId, request: Request, token: Option<String>) -> Self {
33        Self {
34            v: PROTOCOL_VERSION,
35            id,
36            request,
37            token,
38        }
39    }
40
41    pub fn decode(line: &str) -> Result<Self, ProtocolError> {
42        let value: Value = serde_json::from_str(line)
43            .map_err(|error| ProtocolError::invalid_request(format!("malformed JSON: {error}")))?;
44        let object = value
45            .as_object()
46            .ok_or_else(|| ProtocolError::invalid_request("request must be a JSON object"))?;
47
48        let version = object.get("v").and_then(Value::as_u64).ok_or_else(|| {
49            ProtocolError::invalid_request("request field `v` must be an integer")
50        })?;
51        if version != u64::from(PROTOCOL_VERSION) {
52            return Err(ProtocolError::new(
53                ErrorCode::UnsupportedVersion,
54                format!("unsupported protocol version {version}"),
55                json!({"supported": [PROTOCOL_VERSION], "received": version}),
56            ));
57        }
58
59        let verb = object.get("verb").and_then(Value::as_str).ok_or_else(|| {
60            ProtocolError::invalid_request("request field `verb` must be a string")
61        })?;
62        if !Request::is_known_verb(verb) {
63            return Err(ProtocolError::new(
64                ErrorCode::UnknownVerb,
65                format!("unknown verb `{verb}`"),
66                json!({"verb": verb}),
67            ));
68        }
69
70        serde_json::from_value(value).map_err(|error| {
71            ProtocolError::invalid_request(format!("invalid request envelope: {error}"))
72        })
73    }
74
75    pub fn encode(&self) -> Result<String, serde_json::Error> {
76        serde_json::to_string(self)
77    }
78}
79
80#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
81#[serde(tag = "verb", content = "args", rename_all = "snake_case")]
82pub enum Request {
83    Init(EmptyArgs),
84    Daemon(EmptyArgs),
85    Restart(EmptyArgs),
86    Post(PostArgs),
87    Run(RunArgs),
88    Retry(TicketReferenceArgs),
89    Hold(TicketReferenceArgs),
90    Ready(TicketReferenceArgs),
91    List(ListArgs),
92    Status(EmptyArgs),
93    Pause(EmptyArgs),
94    Resume(EmptyArgs),
95    Stop(StopArgs),
96    Cancel(RunReferenceArgs),
97    Logs(LogsArgs),
98    Wait(RunReferenceArgs),
99    Events(EventsArgs),
100    Reindex(EmptyArgs),
101    Brief(EmptyArgs),
102    Show(ShowArgs),
103    Note(NoteArgs),
104    Verdict(VerdictArgs),
105}
106
107impl Request {
108    pub fn verb(&self) -> &'static str {
109        match self {
110            Self::Init(_) => "init",
111            Self::Daemon(_) => "daemon",
112            Self::Restart(_) => "restart",
113            Self::Post(_) => "post",
114            Self::Run(_) => "run",
115            Self::Retry(_) => "retry",
116            Self::Hold(_) => "hold",
117            Self::Ready(_) => "ready",
118            Self::List(_) => "list",
119            Self::Status(_) => "status",
120            Self::Pause(_) => "pause",
121            Self::Resume(_) => "resume",
122            Self::Stop(_) => "stop",
123            Self::Cancel(_) => "cancel",
124            Self::Logs(_) => "logs",
125            Self::Wait(_) => "wait",
126            Self::Events(_) => "events",
127            Self::Reindex(_) => "reindex",
128            Self::Brief(_) => "brief",
129            Self::Show(_) => "show",
130            Self::Note(_) => "note",
131            Self::Verdict(_) => "verdict",
132        }
133    }
134
135    pub fn capability(&self) -> Capability {
136        match self {
137            Self::Brief(_) | Self::Note(_) | Self::Verdict(_) => Capability::Worker,
138            Self::Show(_) => Capability::Both,
139            _ => Capability::Operator,
140        }
141    }
142
143    fn is_known_verb(verb: &str) -> bool {
144        matches!(
145            verb,
146            "init"
147                | "daemon"
148                | "restart"
149                | "post"
150                | "run"
151                | "retry"
152                | "hold"
153                | "ready"
154                | "list"
155                | "status"
156                | "pause"
157                | "resume"
158                | "stop"
159                | "cancel"
160                | "logs"
161                | "wait"
162                | "events"
163                | "reindex"
164                | "brief"
165                | "show"
166                | "note"
167                | "verdict"
168        )
169    }
170}
171
172#[derive(Debug, Clone, Copy, PartialEq, Eq)]
173pub enum Capability {
174    Operator,
175    Worker,
176    Both,
177}
178
179#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
180#[serde(deny_unknown_fields)]
181pub struct EmptyArgs {}
182
183#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
184#[serde(deny_unknown_fields)]
185pub struct PostArgs {
186    pub file: String,
187    #[serde(default, skip_serializing_if = "Option::is_none")]
188    pub project: Option<String>,
189    #[serde(default, skip_serializing_if = "Option::is_none")]
190    pub flow: Option<String>,
191    pub activation: PostActivation,
192}
193
194#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
195#[serde(tag = "kind", rename_all = "snake_case")]
196pub enum PostActivation {
197    Auto,
198    At { time: String },
199    Manual,
200    Hold,
201}
202
203#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
204#[serde(deny_unknown_fields)]
205pub struct RunArgs {
206    #[serde(default, skip_serializing_if = "Option::is_none")]
207    pub ticket: Option<String>,
208    #[serde(default, skip_serializing_if = "Option::is_none")]
209    pub project: Option<String>,
210    pub activation: RunActivation,
211    pub only: Vec<String>,
212}
213
214#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
215#[serde(tag = "kind", rename_all = "snake_case")]
216pub enum RunActivation {
217    Now,
218    At { local_time: String },
219    Every { interval_ms: u64 },
220    Overnight,
221}
222
223#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
224#[serde(deny_unknown_fields)]
225pub struct StopArgs {
226    #[serde(default)]
227    pub force: bool,
228}
229
230#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
231#[serde(deny_unknown_fields)]
232pub struct RunReferenceArgs {
233    pub run: String,
234}
235
236/// A cursor-paginated read of one run's captured output. `stage` narrows the
237/// page to a single flow stage, `tail` keeps the last N matching entries
238/// instead of the first N, and `after` resumes from a previously returned
239/// cursor so a follower streams without replaying what it has seen.
240#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
241#[serde(deny_unknown_fields)]
242pub struct LogsArgs {
243    pub run: String,
244    #[serde(default, skip_serializing_if = "Option::is_none")]
245    pub stage: Option<String>,
246    #[serde(default, skip_serializing_if = "Option::is_none")]
247    pub tail: Option<u32>,
248    #[serde(default, skip_serializing_if = "Option::is_none")]
249    pub after: Option<u64>,
250}
251
252/// A cursor-paginated read of the activity feed. `after` resumes from a
253/// previously returned cursor; `tail` starts that many events before the
254/// newest one and wins when both are given. One page per request — clients
255/// stream by polling with the returned cursor.
256///
257/// `scope` narrows the feed to one reference, resolved by the daemon exactly
258/// as `show` resolves it, so thin clients never reimplement that ladder. The
259/// returned `next_cursor` still advances across filtered-out rows, so a scoped
260/// watcher does not rescan the feed when its scope matches nothing.
261#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
262#[serde(deny_unknown_fields)]
263pub struct EventsArgs {
264    #[serde(default, skip_serializing_if = "Option::is_none")]
265    pub after: Option<i64>,
266    #[serde(default, skip_serializing_if = "Option::is_none")]
267    pub tail: Option<u32>,
268    #[serde(default, skip_serializing_if = "Option::is_none")]
269    pub limit: Option<u32>,
270    #[serde(default, skip_serializing_if = "Option::is_none")]
271    pub scope: Option<String>,
272}
273
274/// `limit` keeps only that many of the newest tickets. Absent means all of
275/// them, which is what a client that predates the field sends.
276#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
277#[serde(deny_unknown_fields)]
278pub struct ListArgs {
279    #[serde(default, skip_serializing_if = "Option::is_none")]
280    pub limit: Option<u32>,
281}
282
283#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
284#[serde(deny_unknown_fields)]
285pub struct TicketReferenceArgs {
286    pub ticket: String,
287}
288
289#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
290#[serde(deny_unknown_fields)]
291pub struct ShowArgs {
292    #[serde(default, rename = "ref", skip_serializing_if = "Option::is_none")]
293    pub reference: Option<String>,
294    #[serde(default, skip_serializing_if = "Option::is_none")]
295    pub limit: Option<u32>,
296}
297
298#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
299#[serde(deny_unknown_fields)]
300pub struct NoteArgs {
301    pub text: String,
302}
303
304#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
305#[serde(deny_unknown_fields)]
306pub struct VerdictArgs {
307    pub verdict: VerdictValue,
308    #[serde(default, skip_serializing_if = "Option::is_none")]
309    pub reason: Option<String>,
310}
311
312#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
313#[serde(rename_all = "snake_case")]
314pub enum VerdictValue {
315    Pass,
316    Fail,
317}
318
319#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
320#[serde(rename_all = "snake_case")]
321pub enum ErrorCode {
322    InvalidArguments,
323    InvalidRequest,
324    UnsupportedVersion,
325    UnknownVerb,
326    DaemonUnavailable,
327    Unauthorized,
328    NotFound,
329    Conflict,
330    CooldownActive,
331    Internal,
332}
333
334#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
335pub struct ErrorBody {
336    pub code: ErrorCode,
337    pub message: String,
338    pub details: Value,
339}
340
341#[derive(Debug, Clone, PartialEq)]
342pub struct ProtocolError {
343    pub body: ErrorBody,
344}
345
346impl ProtocolError {
347    pub fn new(code: ErrorCode, message: impl Into<String>, details: Value) -> Self {
348        Self {
349            body: ErrorBody {
350                code,
351                message: message.into(),
352                details,
353            },
354        }
355    }
356
357    fn invalid_request(message: impl Into<String>) -> Self {
358        Self::new(ErrorCode::InvalidRequest, message, json!({}))
359    }
360}
361
362impl fmt::Display for ProtocolError {
363    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
364        formatter.write_str(&self.body.message)
365    }
366}
367
368impl std::error::Error for ProtocolError {}
369
370#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
371pub struct ResponseEnvelope {
372    pub id: Option<RequestId>,
373    pub ok: bool,
374    #[serde(skip_serializing_if = "Option::is_none")]
375    pub data: Option<Value>,
376    #[serde(skip_serializing_if = "Option::is_none")]
377    pub error: Option<ErrorBody>,
378}
379
380impl ResponseEnvelope {
381    pub fn success(id: Option<RequestId>, data: Value) -> Self {
382        Self {
383            id,
384            ok: true,
385            data: Some(data),
386            error: None,
387        }
388    }
389
390    pub fn failure(id: Option<RequestId>, error: ErrorBody) -> Self {
391        Self {
392            id,
393            ok: false,
394            data: None,
395            error: Some(error),
396        }
397    }
398}
399
400#[cfg(test)]
401mod tests {
402    use serde_json::{Value, json};
403
404    use super::{
405        EmptyArgs, ErrorBody, ErrorCode, Request, RequestEnvelope, RequestId, ResponseEnvelope,
406        RunActivation, RunArgs,
407    };
408
409    #[test]
410    fn request_envelope_serializes_to_the_public_wire_shape() {
411        let envelope = RequestEnvelope::new(
412            RequestId::new("req-123"),
413            Request::Run(RunArgs {
414                ticket: Some("T1".into()),
415                project: None,
416                activation: RunActivation::Now,
417                only: Vec::new(),
418            }),
419            None,
420        );
421
422        let value: Value = serde_json::from_str(&envelope.encode().unwrap()).unwrap();
423        assert_eq!(
424            value,
425            json!({
426                "v": 1,
427                "id": "req-123",
428                "verb": "run",
429                "args": {
430                    "ticket": "T1",
431                    "activation": {"kind": "now"},
432                    "only": []
433                },
434                "token": null
435            })
436        );
437    }
438
439    #[test]
440    fn request_envelope_round_trips() {
441        let expected = RequestEnvelope::new(
442            RequestId::new("req-1"),
443            Request::Brief(EmptyArgs::default()),
444            Some("worker-token".into()),
445        );
446
447        let decoded = RequestEnvelope::decode(&expected.encode().unwrap()).unwrap();
448        assert_eq!(decoded, expected);
449    }
450
451    #[test]
452    fn restart_is_a_public_operator_verb() {
453        let request = RequestEnvelope::decode(
454            r#"{"v":1,"id":"req-1","verb":"restart","args":{},"token":null}"#,
455        )
456        .unwrap()
457        .request;
458
459        assert!(matches!(request, Request::Restart(_)));
460        assert_eq!(request.capability(), super::Capability::Operator);
461    }
462
463    #[test]
464    fn malformed_json_is_an_invalid_request() {
465        let error = RequestEnvelope::decode("{").unwrap_err();
466        assert_eq!(error.body.code, ErrorCode::InvalidRequest);
467    }
468
469    #[test]
470    fn unsupported_versions_have_a_stable_error_code() {
471        let error = RequestEnvelope::decode(
472            r#"{"v":2,"id":"req-1","verb":"status","args":{},"token":null}"#,
473        )
474        .unwrap_err();
475
476        assert_eq!(error.body.code, ErrorCode::UnsupportedVersion);
477        assert_eq!(error.body.details["received"], 2);
478    }
479
480    #[test]
481    fn unknown_verbs_have_a_stable_error_code() {
482        let error = RequestEnvelope::decode(
483            r#"{"v":1,"id":"req-1","verb":"merge","args":{},"token":null}"#,
484        )
485        .unwrap_err();
486
487        assert_eq!(error.body.code, ErrorCode::UnknownVerb);
488        assert_eq!(error.body.details["verb"], "merge");
489    }
490
491    #[test]
492    fn known_verbs_reject_invalid_arguments() {
493        let error = RequestEnvelope::decode(
494            r#"{"v":1,"id":"req-1","verb":"show","args":{"unknown":true},"token":"token"}"#,
495        )
496        .unwrap_err();
497
498        assert_eq!(error.body.code, ErrorCode::InvalidRequest);
499    }
500
501    #[test]
502    fn show_accepts_additive_dashboard_pattern_and_limit_arguments() {
503        for args in [r#"{}"#, r#"{"ref":"log","limit":5}"#] {
504            let request = RequestEnvelope::decode(&format!(
505                r#"{{"v":1,"id":"req-1","verb":"show","args":{args},"token":null}}"#
506            ))
507            .expect("decode show request");
508            assert!(matches!(request.request, Request::Show(_)));
509        }
510    }
511
512    #[test]
513    fn response_envelopes_have_exclusive_success_and_error_payloads() {
514        let success = serde_json::to_value(ResponseEnvelope::success(
515            Some(RequestId::new("req-1")),
516            json!({"paused": false}),
517        ))
518        .unwrap();
519        assert_eq!(
520            success,
521            json!({"id": "req-1", "ok": true, "data": {"paused": false}})
522        );
523
524        let failure = serde_json::to_value(ResponseEnvelope::failure(
525            Some(RequestId::new("req-2")),
526            ErrorBody {
527                code: ErrorCode::Conflict,
528                message: "ticket is already claimed".into(),
529                details: json!({"ticket": "T1"}),
530            },
531        ))
532        .unwrap();
533        assert_eq!(
534            failure,
535            json!({
536                "id": "req-2",
537                "ok": false,
538                "error": {
539                    "code": "conflict",
540                    "message": "ticket is already claimed",
541                    "details": {"ticket": "T1"}
542                }
543            })
544        );
545    }
546}
sloop/protocol.rs

sloop/
protocol.rs
