solti_model/domain/
output.rs

1//! Output streaming types for live tail of task stdout/stderr.
2
3use std::time::SystemTime;
4
5use bytes::Bytes;
6use serde::{Deserialize, Serialize};
7
8/// Which standard stream a chunk came from.
9#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
10#[serde(rename_all = "lowercase")]
11pub enum StreamKind {
12    /// Standard output (`stdout`).
13    Stdout,
14    /// Standard error (`stderr`).
15    Stderr,
16}
17
18/// One event in the live-tail stream of a task.
19///
20/// Carries either an output line, a run boundary marker, or a backpressure signal.
21/// Wire format is JSON-tagged on `type`:
22///
23/// ```text
24/// {"type":"chunk","attempt":1,"stream":"stdout","seq":0,"ts":1700,"line":"..."}
25/// {"type":"runStarted","attempt":1,"startedAt":1700}
26/// {"type":"runFinished","attempt":1,"exitCode":0,"finishedAt":1701}
27/// {"type":"lagged","skipped":42}
28/// ```
29#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
30#[serde(tag = "type", rename_all = "camelCase")]
31pub enum OutputEvent {
32    /// One line of stdout/stderr from the currently active run.
33    Chunk(OutputChunk),
34
35    /// A new run attempt has started; sequence numbers reset from this point on.
36    #[serde(rename_all = "camelCase")]
37    RunStarted {
38        attempt: u32,
39        #[serde(with = "crate::resource::metadata::time_serde")]
40        started_at: SystemTime,
41    },
42
43    /// The current run finished. Consumers can stop accumulating chunks for this attempt.
44    #[serde(rename_all = "camelCase")]
45    RunFinished {
46        attempt: u32,
47        #[serde(skip_serializing_if = "Option::is_none")]
48        exit_code: Option<i32>,
49        #[serde(with = "crate::resource::metadata::time_serde")]
50        finished_at: SystemTime,
51    },
52
53    /// Subscriber fell behind the broadcast ring window.
54    Lagged { skipped: u64 },
55}
56
57/// One line of output from a single task-run attempt.
58///
59/// Carried through `tokio::sync::broadcast` channels in-process;
60/// sent to clients via SSE / gRPC server-stream.
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(rename_all = "camelCase")]
63pub struct OutputChunk {
64    /// Which attempt of the task this chunk belongs to (matches [`TaskRun::attempt`]).
65    ///
66    /// [`TaskRun::attempt`]: crate::TaskRun::attempt
67    pub attempt: u32,
68    /// stdout or stderr.
69    pub stream: StreamKind,
70    /// Monotonic sequence number within this attempt; resets on next run.
71    pub seq: u64,
72    /// Wall-clock time the line was read by the agent (unix milliseconds on the wire).
73    #[serde(with = "crate::resource::metadata::time_serde")]
74    pub ts: SystemTime,
75    /// One line, already truncated/cleaned by the runner.
76    #[serde(with = "bytes_as_utf8_string")]
77    pub line: Bytes,
78}
79
80/// Serde adapter: serialize `Bytes` as a UTF-8 string in JSON, deserialize from a JSON string back into `Bytes`.
81mod bytes_as_utf8_string {
82    use bytes::Bytes;
83    use serde::{Deserialize, Deserializer, Serializer};
84
85    pub(super) fn serialize<S>(b: &Bytes, s: S) -> Result<S::Ok, S::Error>
86    where
87        S: Serializer,
88    {
89        let txt = std::str::from_utf8(b).map_err(serde::ser::Error::custom)?;
90        s.serialize_str(txt)
91    }
92
93    pub(super) fn deserialize<'de, D>(d: D) -> Result<Bytes, D::Error>
94    where
95        D: Deserializer<'de>,
96    {
97        let s = String::deserialize(d)?;
98        Ok(Bytes::from(s))
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105
106    use std::time::{Duration, UNIX_EPOCH};
107
108    #[test]
109    fn stream_kind_stdout_serializes_to_lowercase() {
110        let json = serde_json::to_string(&StreamKind::Stdout).unwrap();
111        assert_eq!(json, "\"stdout\"");
112    }
113
114    #[test]
115    fn output_chunk_roundtrips_through_json() {
116        let chunk = OutputChunk {
117            attempt: 7,
118            stream: StreamKind::Stderr,
119            seq: 42,
120            ts: UNIX_EPOCH + Duration::from_millis(1_700_000_000_000),
121            line: Bytes::from_static(b"compiling foo..."),
122        };
123
124        let json = serde_json::to_string(&chunk).unwrap();
125        let back: OutputChunk = serde_json::from_str(&json).unwrap();
126
127        assert_eq!(back, chunk);
128    }
129
130    #[test]
131    fn output_chunk_serializes_ts_as_unix_milliseconds() {
132        let chunk = OutputChunk {
133            attempt: 1,
134            stream: StreamKind::Stdout,
135            seq: 0,
136            ts: UNIX_EPOCH + Duration::from_millis(1234),
137            line: Bytes::from_static(b"x"),
138        };
139
140        let json = serde_json::to_string(&chunk).unwrap();
141        assert!(
142            json.contains(r#""ts":1234"#),
143            "ts must serialize as unix milliseconds; got {json}"
144        );
145    }
146
147    #[test]
148    fn output_chunk_serializes_line_as_utf8_string_not_array() {
149        let chunk = OutputChunk {
150            attempt: 1,
151            stream: StreamKind::Stdout,
152            seq: 0,
153            ts: UNIX_EPOCH,
154            line: Bytes::from_static(b"hello"),
155        };
156        let json = serde_json::to_string(&chunk).unwrap();
157        assert!(
158            json.contains(r#""line":"hello""#),
159            "line must serialize as JSON string, not byte array; got {json}"
160        );
161    }
162
163    #[test]
164    fn output_event_chunk_inlines_chunk_fields() {
165        let event = OutputEvent::Chunk(OutputChunk {
166            attempt: 3,
167            stream: StreamKind::Stdout,
168            seq: 5,
169            ts: UNIX_EPOCH + Duration::from_millis(1_700_000_000_000),
170            line: Bytes::from_static(b"hello"),
171        });
172        let json = serde_json::to_string(&event).unwrap();
173
174        assert!(json.contains(r#""type":"chunk""#), "missing tag in {json}");
175        assert!(json.contains(r#""attempt":3"#), "{json}");
176        assert!(json.contains(r#""stream":"stdout""#), "{json}");
177        assert!(json.contains(r#""line":"hello""#), "{json}");
178    }
179
180    #[test]
181    fn output_event_run_started_carries_attempt_and_ts() {
182        let event = OutputEvent::RunStarted {
183            attempt: 2,
184            started_at: UNIX_EPOCH + Duration::from_millis(1234),
185        };
186        let json = serde_json::to_string(&event).unwrap();
187
188        assert!(json.contains(r#""type":"runStarted""#), "{json}");
189        assert!(json.contains(r#""attempt":2"#), "{json}");
190        assert!(json.contains(r#""startedAt":1234"#), "{json}");
191    }
192
193    #[test]
194    fn output_event_run_finished_carries_exit_code() {
195        let event = OutputEvent::RunFinished {
196            attempt: 2,
197            exit_code: Some(0),
198            finished_at: UNIX_EPOCH + Duration::from_millis(2222),
199        };
200        let json = serde_json::to_string(&event).unwrap();
201
202        assert!(json.contains(r#""type":"runFinished""#), "{json}");
203        assert!(json.contains(r#""exitCode":0"#), "{json}");
204        assert!(json.contains(r#""finishedAt":2222"#), "{json}");
205    }
206
207    #[test]
208    fn output_event_lagged_carries_skipped_count() {
209        let event = OutputEvent::Lagged { skipped: 1500 };
210        let json = serde_json::to_string(&event).unwrap();
211
212        assert!(json.contains(r#""type":"lagged""#), "{json}");
213        assert!(json.contains(r#""skipped":1500"#), "{json}");
214    }
215
216    #[test]
217    fn output_event_roundtrips_through_json() {
218        let cases = [
219            OutputEvent::Chunk(OutputChunk {
220                attempt: 1,
221                stream: StreamKind::Stderr,
222                seq: 0,
223                ts: UNIX_EPOCH + Duration::from_millis(1_700_000_000_000),
224                line: Bytes::from_static(b"warning"),
225            }),
226            OutputEvent::RunStarted {
227                attempt: 1,
228                started_at: UNIX_EPOCH + Duration::from_millis(1_700_000_000_000),
229            },
230            OutputEvent::RunFinished {
231                attempt: 1,
232                exit_code: Some(42),
233                finished_at: UNIX_EPOCH + Duration::from_millis(1_700_000_001_000),
234            },
235            OutputEvent::Lagged { skipped: 7 },
236        ];
237
238        for original in cases {
239            let json = serde_json::to_string(&original).unwrap();
240            let back: OutputEvent = serde_json::from_str(&json).unwrap();
241            assert_eq!(back, original, "roundtrip failed for {json}");
242        }
243    }
244
245    #[test]
246    fn output_chunk_uses_camel_case_keys() {
247        let chunk = OutputChunk {
248            attempt: 2,
249            stream: StreamKind::Stdout,
250            seq: 9,
251            ts: UNIX_EPOCH,
252            line: Bytes::from_static(b"hi"),
253        };
254
255        let json = serde_json::to_string(&chunk).unwrap();
256        for key in [
257            r#""attempt":"#,
258            r#""stream":"#,
259            r#""seq":"#,
260            r#""ts":"#,
261            r#""line":"#,
262        ] {
263            assert!(json.contains(key), "missing key {key} in {json}");
264        }
265    }
266
267    #[test]
268    fn output_chunk_clone_is_refcount_bump() {
269        let original = OutputChunk {
270            attempt: 1,
271            stream: StreamKind::Stdout,
272            seq: 0,
273            ts: UNIX_EPOCH,
274            line: Bytes::from_static(b"shared-line"),
275        };
276        let cloned = original.clone();
277        assert_eq!(original.line.as_ptr(), cloned.line.as_ptr());
278    }
279}
solti_model/domain/output.rs

solti_model/domain/
output.rs
