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 trigger: PostTrigger,
192}
193
194#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
195#[serde(tag = "kind", rename_all = "snake_case")]
196pub enum PostTrigger {
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 trigger: RunTrigger,
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 RunTrigger {
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    /// How sure the reporter says it is. Absent from a client that predates
311    /// the field, and read as `medium`.
312    #[serde(default, skip_serializing_if = "Option::is_none")]
313    pub confidence: Option<ConfidenceValue>,
314}
315
316#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
317#[serde(rename_all = "snake_case")]
318pub enum VerdictValue {
319    Pass,
320    Fail,
321}
322
323/// Three named levels and nothing else. A float would decode here and then
324/// have to mean something in aggregation, which v1 deliberately does not
325/// define, so the wire type refuses one outright.
326#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
327#[serde(rename_all = "snake_case")]
328pub enum ConfidenceValue {
329    Low,
330    Medium,
331    High,
332}
333
334impl From<ConfidenceValue> for crate::flow::Confidence {
335    fn from(value: ConfidenceValue) -> Self {
336        match value {
337            ConfidenceValue::Low => Self::Low,
338            ConfidenceValue::Medium => Self::Medium,
339            ConfidenceValue::High => Self::High,
340        }
341    }
342}
343
344#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
345#[serde(rename_all = "snake_case")]
346pub enum ErrorCode {
347    InvalidArguments,
348    InvalidRequest,
349    UnsupportedVersion,
350    UnknownVerb,
351    DaemonUnavailable,
352    Unauthorized,
353    NotFound,
354    Conflict,
355    CooldownActive,
356    Internal,
357}
358
359#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
360pub struct ErrorBody {
361    pub code: ErrorCode,
362    pub message: String,
363    pub details: Value,
364}
365
366#[derive(Debug, Clone, PartialEq)]
367pub struct ProtocolError {
368    pub body: ErrorBody,
369}
370
371impl ProtocolError {
372    pub fn new(code: ErrorCode, message: impl Into<String>, details: Value) -> Self {
373        Self {
374            body: ErrorBody {
375                code,
376                message: message.into(),
377                details,
378            },
379        }
380    }
381
382    fn invalid_request(message: impl Into<String>) -> Self {
383        Self::new(ErrorCode::InvalidRequest, message, json!({}))
384    }
385}
386
387impl fmt::Display for ProtocolError {
388    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
389        formatter.write_str(&self.body.message)
390    }
391}
392
393impl std::error::Error for ProtocolError {}
394
395#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
396pub struct ResponseEnvelope {
397    pub id: Option<RequestId>,
398    pub ok: bool,
399    #[serde(skip_serializing_if = "Option::is_none")]
400    pub data: Option<Value>,
401    #[serde(skip_serializing_if = "Option::is_none")]
402    pub error: Option<ErrorBody>,
403}
404
405impl ResponseEnvelope {
406    pub fn success(id: Option<RequestId>, data: Value) -> Self {
407        Self {
408            id,
409            ok: true,
410            data: Some(data),
411            error: None,
412        }
413    }
414
415    pub fn failure(id: Option<RequestId>, error: ErrorBody) -> Self {
416        Self {
417            id,
418            ok: false,
419            data: None,
420            error: Some(error),
421        }
422    }
423}
424
425#[cfg(test)]
426mod tests {
427    use serde_json::{Value, json};
428
429    use super::{
430        EmptyArgs, ErrorBody, ErrorCode, Request, RequestEnvelope, RequestId, ResponseEnvelope,
431        RunArgs, RunTrigger,
432    };
433
434    #[test]
435    fn request_envelope_serializes_to_the_public_wire_shape() {
436        let envelope = RequestEnvelope::new(
437            RequestId::new("req-123"),
438            Request::Run(RunArgs {
439                ticket: Some("T1".into()),
440                project: None,
441                trigger: RunTrigger::Now,
442                only: Vec::new(),
443            }),
444            None,
445        );
446
447        let value: Value = serde_json::from_str(&envelope.encode().unwrap()).unwrap();
448        assert_eq!(
449            value,
450            json!({
451                "v": 1,
452                "id": "req-123",
453                "verb": "run",
454                "args": {
455                    "ticket": "T1",
456                    "trigger": {"kind": "now"},
457                    "only": []
458                },
459                "token": null
460            })
461        );
462    }
463
464    #[test]
465    fn request_envelope_round_trips() {
466        let expected = RequestEnvelope::new(
467            RequestId::new("req-1"),
468            Request::Brief(EmptyArgs::default()),
469            Some("worker-token".into()),
470        );
471
472        let decoded = RequestEnvelope::decode(&expected.encode().unwrap()).unwrap();
473        assert_eq!(decoded, expected);
474    }
475
476    #[test]
477    fn restart_is_a_public_operator_verb() {
478        let request = RequestEnvelope::decode(
479            r#"{"v":1,"id":"req-1","verb":"restart","args":{},"token":null}"#,
480        )
481        .unwrap()
482        .request;
483
484        assert!(matches!(request, Request::Restart(_)));
485        assert_eq!(request.capability(), super::Capability::Operator);
486    }
487
488    #[test]
489    fn malformed_json_is_an_invalid_request() {
490        let error = RequestEnvelope::decode("{").unwrap_err();
491        assert_eq!(error.body.code, ErrorCode::InvalidRequest);
492    }
493
494    #[test]
495    fn unsupported_versions_have_a_stable_error_code() {
496        let error = RequestEnvelope::decode(
497            r#"{"v":2,"id":"req-1","verb":"status","args":{},"token":null}"#,
498        )
499        .unwrap_err();
500
501        assert_eq!(error.body.code, ErrorCode::UnsupportedVersion);
502        assert_eq!(error.body.details["received"], 2);
503    }
504
505    #[test]
506    fn unknown_verbs_have_a_stable_error_code() {
507        let error = RequestEnvelope::decode(
508            r#"{"v":1,"id":"req-1","verb":"merge","args":{},"token":null}"#,
509        )
510        .unwrap_err();
511
512        assert_eq!(error.body.code, ErrorCode::UnknownVerb);
513        assert_eq!(error.body.details["verb"], "merge");
514    }
515
516    #[test]
517    fn known_verbs_reject_invalid_arguments() {
518        let error = RequestEnvelope::decode(
519            r#"{"v":1,"id":"req-1","verb":"show","args":{"unknown":true},"token":"token"}"#,
520        )
521        .unwrap_err();
522
523        assert_eq!(error.body.code, ErrorCode::InvalidRequest);
524    }
525
526    #[test]
527    fn show_accepts_additive_dashboard_pattern_and_limit_arguments() {
528        for args in [r#"{}"#, r#"{"ref":"log","limit":5}"#] {
529            let request = RequestEnvelope::decode(&format!(
530                r#"{{"v":1,"id":"req-1","verb":"show","args":{args},"token":null}}"#
531            ))
532            .expect("decode show request");
533            assert!(matches!(request.request, Request::Show(_)));
534        }
535    }
536
537    #[test]
538    fn response_envelopes_have_exclusive_success_and_error_payloads() {
539        let success = serde_json::to_value(ResponseEnvelope::success(
540            Some(RequestId::new("req-1")),
541            json!({"paused": false}),
542        ))
543        .unwrap();
544        assert_eq!(
545            success,
546            json!({"id": "req-1", "ok": true, "data": {"paused": false}})
547        );
548
549        let failure = serde_json::to_value(ResponseEnvelope::failure(
550            Some(RequestId::new("req-2")),
551            ErrorBody {
552                code: ErrorCode::Conflict,
553                message: "ticket is already claimed".into(),
554                details: json!({"ticket": "T1"}),
555            },
556        ))
557        .unwrap();
558        assert_eq!(
559            failure,
560            json!({
561                "id": "req-2",
562                "ok": false,
563                "error": {
564                    "code": "conflict",
565                    "message": "ticket is already claimed",
566                    "details": {"ticket": "T1"}
567                }
568            })
569        );
570    }
571}
sloop/protocol.rs

sloop/
protocol.rs
