1use serde::{Deserialize, Serialize};
4use serde_json::Value;
5
6pub const PROTOCOL_VERSION: u32 = 3;
7
8#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
9#[serde(rename_all = "snake_case")]
10pub enum ComponentState {
11 Disabled,
12 Starting,
13 Connected,
14 Disconnected,
15 Backoff,
16 Stopping,
17 Stopped,
18 Failed,
19}
20
21#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
22pub struct ComponentHealth {
23 pub id: String,
24 #[serde(default)]
26 pub channel: String,
27 pub account: String,
28 pub bot_id: Option<String>,
29 pub user_id: Option<String>,
30 pub enabled: bool,
31 pub state: ComponentState,
32 pub last_success_unix_seconds: Option<u64>,
33 pub error: Option<String>,
34 pub restarts: u64,
35 #[serde(default)]
37 pub remote_tools: RemoteTools,
38}
39
40#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
42#[serde(rename_all = "snake_case")]
43pub enum RemoteTools {
44 #[default]
46 None,
47 Owner,
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
52pub struct DaemonStatus {
53 pub version: String,
54 pub pid: u32,
55 pub components: Vec<ComponentHealth>,
56 #[serde(default)]
57 pub delegations: DelegationSummary,
58}
59
60#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
62pub struct DelegationSummary {
63 pub active: u64,
65 pub reaped: u64,
67 #[serde(default, skip_serializing_if = "Vec::is_empty")]
69 pub entries: Vec<DelegationInfo>,
70 #[serde(default, skip_serializing_if = "Vec::is_empty")]
72 pub killed: Vec<String>,
73}
74
75#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
76pub struct DelegationInfo {
77 pub handle: String,
78 pub agent: String,
79 pub session: String,
80 pub depth: u32,
81 pub pid: u32,
82 pub owner_pid: u32,
84 pub processes: u32,
86 pub cwd: String,
87 pub started_unix_seconds: u64,
88 pub orphaned: bool,
90 #[serde(default, skip_serializing_if = "Option::is_none")]
92 pub conversation: Option<String>,
93 #[serde(default, skip_serializing_if = "Option::is_none")]
94 pub turn: Option<u32>,
95}
96
97#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
98#[serde(tag = "action", rename_all = "snake_case")]
99pub enum DaemonCommand {
100 Status,
101 Reload,
102 ChannelSet {
105 channel: String,
106 account: String,
107 enabled: bool,
108 workspace: Option<String>,
109 #[serde(default, skip_serializing_if = "Option::is_none")]
111 remote_tools: Option<RemoteTools>,
112 },
113 ChannelLogout {
115 channel: String,
116 account: String,
117 },
118 Delegations {
120 #[serde(default)]
121 all: bool,
122 },
123 DelegationKill {
125 #[serde(default, skip_serializing_if = "Option::is_none")]
126 handle: Option<String>,
127 #[serde(default)]
128 orphans: bool,
129 },
130}
131
132#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
133pub struct QueueEntry {
134 pub queue_id: String,
135 pub revision: u64,
136 pub prompt: String,
137 pub submitter: String,
138}
139
140#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
142pub struct TurnOrigin {
143 pub kind: String,
145 #[serde(default, skip_serializing_if = "Vec::is_empty")]
147 pub jobs: Vec<String>,
148}
149
150pub const ORIGIN_BACKGROUND: &str = "background";
152
153#[derive(Debug, Default, Clone, PartialEq, Eq)]
160pub struct BackgroundJobUpdate {
161 pub started: Vec<String>,
162 pub settled: Vec<String>,
163}
164
165pub fn background_job_update(output: &str) -> BackgroundJobUpdate {
166 let mut update = BackgroundJobUpdate::default();
167 let Ok(serde_json::Value::Object(value)) = serde_json::from_str::<serde_json::Value>(output)
168 else {
169 return update;
170 };
171 let mut visit = |job: &serde_json::Map<String, serde_json::Value>| {
172 let (Some(id), Some(status)) = (
173 job.get("job").and_then(serde_json::Value::as_str),
174 job.get("status").and_then(serde_json::Value::as_str),
175 ) else {
176 return;
177 };
178 if status == "running" {
179 if job.get("background").and_then(serde_json::Value::as_bool) == Some(true) {
180 update.started.push(id.to_owned());
181 }
182 } else {
183 update.settled.push(id.to_owned());
184 }
185 };
186 visit(&value);
187 for job in value
188 .get("jobs")
189 .and_then(serde_json::Value::as_array)
190 .into_iter()
191 .flatten()
192 {
193 if let serde_json::Value::Object(job) = job {
194 visit(job);
195 }
196 }
197 update
198}
199
200#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
201pub struct PeerInfo {
202 pub name: String,
203 pub version: String,
204}
205
206#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
207pub struct Usage {
208 #[serde(skip_serializing_if = "Option::is_none")]
209 pub input_tokens: Option<u64>,
210 #[serde(skip_serializing_if = "Option::is_none")]
211 pub output_tokens: Option<u64>,
212}
213
214#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
215#[serde(tag = "type")]
216pub enum ClientMessage {
217 #[serde(rename = "daemon.control")]
218 DaemonControl {
219 request_id: String,
220 command: DaemonCommand,
221 },
222 #[serde(rename = "initialize")]
223 Initialize {
224 request_id: String,
225 protocol_version: u32,
226 client: PeerInfo,
227 },
228 #[serde(rename = "session.start")]
229 SessionStart {
230 request_id: String,
231 cwd: String,
232 #[serde(default, skip_serializing_if = "Option::is_none")]
233 provider: Option<String>,
234 #[serde(default, skip_serializing_if = "Option::is_none")]
235 model: Option<String>,
236 #[serde(default, skip_serializing_if = "Option::is_none")]
237 base_url: Option<String>,
238 #[serde(default, skip_serializing_if = "Option::is_none")]
239 no_tools: Option<bool>,
240 #[serde(default, skip_serializing_if = "Option::is_none")]
244 delegation_depth: Option<u32>,
245 },
246 #[serde(rename = "session.attach")]
247 SessionAttach {
248 request_id: String,
249 session_id: String,
250 cwd: String,
251 },
252 #[serde(rename = "turn.start")]
253 TurnStart {
254 request_id: String,
255 session_id: String,
256 prompt: String,
257 },
258 #[serde(rename = "queue.update")]
259 QueueUpdate {
260 request_id: String,
261 session_id: String,
262 queue_id: String,
263 revision: u64,
264 prompt: String,
265 },
266 #[serde(rename = "queue.move")]
267 QueueMove {
268 request_id: String,
269 session_id: String,
270 queue_id: String,
271 revision: u64,
272 before_queue_id: Option<String>,
273 },
274 #[serde(rename = "queue.remove")]
275 QueueRemove {
276 request_id: String,
277 session_id: String,
278 queue_id: String,
279 revision: u64,
280 },
281 #[serde(rename = "session.pause")]
282 SessionPause {
283 request_id: String,
284 session_id: String,
285 paused: bool,
286 },
287 #[serde(rename = "turn.cancel")]
288 TurnCancel {
289 request_id: String,
290 session_id: String,
291 turn_id: String,
292 },
293 #[serde(rename = "approval.resolve")]
294 ApprovalResolve {
295 request_id: String,
296 session_id: String,
297 approval_id: String,
298 approved: bool,
299 },
300 #[serde(rename = "session.clear")]
301 SessionClear {
302 request_id: String,
303 session_id: String,
304 },
305}
306
307impl ServerEvent {
308 pub fn turn_request_id(&self) -> Option<&str> {
311 match self {
312 Self::QueueDequeued { request_id, .. }
313 | Self::TurnStarted { request_id, .. }
314 | Self::AssistantDelta { request_id, .. }
315 | Self::AssistantCompleted { request_id, .. }
316 | Self::ToolProposed { request_id, .. }
317 | Self::ApprovalRequested { request_id, .. }
318 | Self::ToolStarted { request_id, .. }
319 | Self::ToolProgress { request_id, .. }
320 | Self::ToolCompleted { request_id, .. }
321 | Self::ContextCompacted { request_id, .. }
322 | Self::TurnCompleted { request_id, .. }
323 | Self::TurnCancelled { request_id, .. }
324 | Self::TurnFailed { request_id, .. } => Some(request_id),
325 _ => None,
326 }
327 }
328}
329
330impl ClientMessage {
331 pub fn request_id(&self) -> &str {
332 match self {
333 Self::Initialize { request_id, .. }
334 | Self::DaemonControl { request_id, .. }
335 | Self::SessionStart { request_id, .. }
336 | Self::SessionAttach { request_id, .. }
337 | Self::TurnStart { request_id, .. }
338 | Self::QueueUpdate { request_id, .. }
339 | Self::QueueMove { request_id, .. }
340 | Self::QueueRemove { request_id, .. }
341 | Self::SessionPause { request_id, .. }
342 | Self::TurnCancel { request_id, .. }
343 | Self::ApprovalResolve { request_id, .. }
344 | Self::SessionClear { request_id, .. } => request_id,
345 }
346 }
347}
348
349#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
350#[serde(tag = "type")]
351pub enum ServerEvent {
352 #[serde(rename = "daemon.status")]
353 DaemonStatus {
354 request_id: String,
355 status: DaemonStatus,
356 },
357 #[serde(rename = "initialized")]
358 Initialized {
359 request_id: String,
360 protocol_version: u32,
361 server: PeerInfo,
362 },
363 #[serde(rename = "session.started")]
364 SessionStarted {
365 request_id: String,
366 session_id: String,
367 cwd: String,
368 model: String,
369 context_max_tokens: usize,
370 max_server_frame_bytes: usize,
371 max_transcript_bytes: usize,
372 max_transcript_items: usize,
373 max_prompt_history_bytes: usize,
374 max_prompt_history_items: usize,
375 },
376 #[serde(rename = "queue.snapshot")]
377 QueueSnapshot {
378 request_id: Option<String>,
379 session_id: String,
380 seq: u64,
381 entries: Vec<QueueEntry>,
382 paused: bool,
383 },
384 #[serde(rename = "queue.enqueued")]
385 QueueEnqueued {
386 request_id: String,
387 session_id: String,
388 seq: u64,
389 entry: QueueEntry,
390 position: usize,
391 },
392 #[serde(rename = "queue.updated")]
393 QueueUpdated {
394 request_id: String,
395 session_id: String,
396 seq: u64,
397 entry: QueueEntry,
398 },
399 #[serde(rename = "queue.moved")]
400 QueueMoved {
401 request_id: String,
402 session_id: String,
403 seq: u64,
404 queue_id: String,
405 position: usize,
406 revision: u64,
407 },
408 #[serde(rename = "queue.removed")]
409 QueueRemoved {
410 request_id: String,
411 session_id: String,
412 seq: u64,
413 queue_id: String,
414 revision: u64,
415 },
416 #[serde(rename = "queue.dequeued")]
417 QueueDequeued {
418 request_id: String,
419 session_id: String,
420 seq: u64,
421 queue_id: String,
422 turn_id: String,
423 },
424 #[serde(rename = "session.paused")]
425 SessionPaused {
426 request_id: String,
427 session_id: String,
428 seq: u64,
429 paused: bool,
430 },
431 #[serde(rename = "turn.started")]
432 TurnStarted {
433 request_id: String,
434 session_id: String,
435 turn_id: String,
436 seq: u64,
437 #[serde(default, skip_serializing_if = "Option::is_none")]
440 origin: Option<TurnOrigin>,
441 },
442 #[serde(rename = "assistant.delta")]
443 AssistantDelta {
444 request_id: String,
445 session_id: String,
446 turn_id: String,
447 seq: u64,
448 content: String,
449 },
450 #[serde(rename = "assistant.completed")]
451 AssistantCompleted {
452 request_id: String,
453 session_id: String,
454 turn_id: String,
455 seq: u64,
456 content: String,
457 },
458 #[serde(rename = "tool.proposed")]
459 ToolProposed {
460 request_id: String,
461 session_id: String,
462 turn_id: String,
463 seq: u64,
464 call_id: String,
465 name: String,
466 arguments: Value,
467 },
468 #[serde(rename = "approval.requested")]
469 ApprovalRequested {
470 request_id: String,
471 session_id: String,
472 turn_id: String,
473 seq: u64,
474 approval_id: String,
475 call_id: String,
476 name: String,
477 risk: String,
478 cwd: String,
479 summary: String,
480 },
481 #[serde(rename = "tool.started")]
482 ToolStarted {
483 request_id: String,
484 session_id: String,
485 turn_id: String,
486 seq: u64,
487 call_id: String,
488 name: String,
489 },
490 #[serde(rename = "tool.progress")]
493 ToolProgress {
494 request_id: String,
495 session_id: String,
496 turn_id: String,
497 seq: u64,
498 call_id: String,
499 text: String,
500 },
501 #[serde(rename = "tool.completed")]
502 ToolCompleted {
503 request_id: String,
504 session_id: String,
505 turn_id: String,
506 seq: u64,
507 call_id: String,
508 name: String,
509 success: bool,
510 output: String,
511 truncated: bool,
512 },
513 #[serde(rename = "context.compacted")]
514 ContextCompacted {
515 request_id: String,
516 session_id: String,
517 turn_id: String,
518 seq: u64,
519 before_tokens: usize,
520 after_tokens: usize,
521 removed_messages: usize,
522 },
523 #[serde(rename = "session.trimmed")]
524 SessionTrimmed {
525 request_id: String,
526 session_id: String,
527 seq: u64,
528 removed_messages: usize,
529 history_bytes: usize,
530 },
531 #[serde(rename = "session.cleared")]
532 SessionCleared {
533 request_id: String,
534 session_id: String,
535 seq: u64,
536 },
537 #[serde(rename = "turn.completed")]
538 TurnCompleted {
539 request_id: String,
540 session_id: String,
541 turn_id: String,
542 seq: u64,
543 steps: usize,
544 usage: Usage,
545 #[serde(default, skip_serializing_if = "Option::is_none")]
548 origin: Option<TurnOrigin>,
549 },
550 #[serde(rename = "turn.cancelled")]
551 TurnCancelled {
552 request_id: String,
553 session_id: String,
554 turn_id: String,
555 seq: u64,
556 #[serde(default, skip_serializing_if = "Option::is_none")]
559 origin: Option<TurnOrigin>,
560 },
561 #[serde(rename = "turn.failed")]
562 TurnFailed {
563 request_id: String,
564 session_id: String,
565 turn_id: String,
566 seq: u64,
567 code: String,
568 message: String,
569 #[serde(default, skip_serializing_if = "Option::is_none")]
572 origin: Option<TurnOrigin>,
573 },
574 #[serde(rename = "error")]
575 Error {
576 #[serde(skip_serializing_if = "Option::is_none")]
577 request_id: Option<String>,
578 code: String,
579 message: String,
580 fatal: bool,
581 },
582}
583
584#[cfg(test)]
585mod tests {
586 use super::*;
587
588 #[test]
589 fn client_message_round_trip() {
590 let message = ClientMessage::TurnStart {
591 request_id: "3".into(),
592 session_id: "session".into(),
593 prompt: "hello".into(),
594 };
595 let json = serde_json::to_string(&message).unwrap();
596 assert!(json.contains("\"type\":\"turn.start\""));
597 assert_eq!(
598 serde_json::from_str::<ClientMessage>(&json).unwrap(),
599 message
600 );
601 }
602
603 #[test]
604 fn tool_progress_and_delegation_depth_round_trip() {
605 let event = ServerEvent::ToolProgress {
606 request_id: "r".into(),
607 session_id: "s".into(),
608 turn_id: "t".into(),
609 seq: 4,
610 call_id: "c".into(),
611 text: "$ cargo test\nupdate …/src/lib.rs".into(),
612 };
613 let wire = serde_json::to_string(&event).unwrap();
614 assert!(wire.contains(r#""type":"tool.progress""#));
615 assert_eq!(serde_json::from_str::<ServerEvent>(&wire).unwrap(), event);
616
617 let start = |depth| ClientMessage::SessionStart {
618 request_id: "1".into(),
619 cwd: "/w".into(),
620 provider: None,
621 model: None,
622 base_url: None,
623 no_tools: None,
624 delegation_depth: depth,
625 };
626 let nested = serde_json::to_string(&start(Some(2))).unwrap();
627 assert!(nested.contains(r#""delegation_depth":2"#));
628 assert_eq!(
629 serde_json::from_str::<ClientMessage>(&nested).unwrap(),
630 start(Some(2))
631 );
632 let direct = serde_json::to_string(&start(None)).unwrap();
634 assert!(!direct.contains("delegation_depth"));
635 let older = r#"{"type":"session.start","request_id":"1","cwd":"/w"}"#;
636 assert_eq!(
637 serde_json::from_str::<ClientMessage>(older).unwrap(),
638 start(None)
639 );
640 assert_eq!(PROTOCOL_VERSION, 3);
641 }
642
643 #[test]
644 fn additive_fields_are_ignored() {
645 let json = r#"{"type":"session.clear","request_id":"1","session_id":"s","future":true}"#;
646 assert!(matches!(
647 serde_json::from_str::<ClientMessage>(json).unwrap(),
648 ClientMessage::SessionClear { .. }
649 ));
650 }
651
652 #[test]
653 fn event_round_trip() {
654 let event = ServerEvent::AssistantDelta {
655 request_id: "1".into(),
656 session_id: "s".into(),
657 turn_id: "t".into(),
658 seq: 4,
659 content: "hello".into(),
660 };
661 let encoded = serde_json::to_string(&event).unwrap();
662 assert_eq!(
663 serde_json::from_str::<ServerEvent>(&encoded).unwrap(),
664 event
665 );
666 }
667
668 #[test]
669 fn queue_messages_and_events_round_trip() {
670 let message = ClientMessage::QueueMove {
671 request_id: "q1".into(),
672 session_id: "s".into(),
673 queue_id: "q".into(),
674 revision: 2,
675 before_queue_id: None,
676 };
677 let encoded = serde_json::to_string(&message).unwrap();
678 assert_eq!(
679 serde_json::from_str::<ClientMessage>(&encoded).unwrap(),
680 message
681 );
682 let event = ServerEvent::QueueSnapshot {
683 request_id: None,
684 session_id: "s".into(),
685 seq: 4,
686 entries: vec![QueueEntry {
687 queue_id: "q".into(),
688 revision: 1,
689 prompt: "hello".into(),
690 submitter: "cli".into(),
691 }],
692 paused: false,
693 };
694 let encoded = serde_json::to_string(&event).unwrap();
695 assert_eq!(
696 serde_json::from_str::<ServerEvent>(&encoded).unwrap(),
697 event
698 );
699 }
700
701 #[test]
702 fn remote_tools_fields_are_additive() {
703 let legacy: DaemonCommand = serde_json::from_str(
704 r#"{"action":"channel_set","channel":"wechat","account":"a","enabled":true,"workspace":null}"#,
705 )
706 .unwrap();
707 assert!(matches!(
708 legacy,
709 DaemonCommand::ChannelSet {
710 remote_tools: None,
711 ..
712 }
713 ));
714 let owner = DaemonCommand::ChannelSet {
715 channel: "wechat".into(),
716 account: "a".into(),
717 enabled: true,
718 workspace: None,
719 remote_tools: Some(RemoteTools::Owner),
720 };
721 let encoded = serde_json::to_string(&owner).unwrap();
722 assert!(encoded.contains(r#""remote_tools":"owner""#));
723 assert_eq!(
724 serde_json::from_str::<DaemonCommand>(&encoded).unwrap(),
725 owner
726 );
727 let health: ComponentHealth = serde_json::from_str(
728 r#"{"id":"clawbot:a","account":"a","bot_id":null,"user_id":null,"enabled":true,"state":"connected","last_success_unix_seconds":null,"error":null,"restarts":0}"#,
729 )
730 .unwrap();
731 assert_eq!(health.remote_tools, RemoteTools::None);
732 assert!(health.channel.is_empty());
734 }
735
736 #[test]
737 fn delegation_control_round_trips_and_older_status_still_parses() {
738 for (command, wire) in [
739 (
740 DaemonCommand::Delegations { all: true },
741 r#"{"action":"delegations","all":true}"#,
742 ),
743 (
744 DaemonCommand::DelegationKill {
745 handle: Some("codex-3f9a2c".into()),
746 orphans: false,
747 },
748 r#"{"action":"delegation_kill","handle":"codex-3f9a2c","orphans":false}"#,
749 ),
750 ] {
751 assert_eq!(serde_json::to_string(&command).unwrap(), wire);
752 assert_eq!(
753 serde_json::from_str::<DaemonCommand>(wire).unwrap(),
754 command
755 );
756 }
757 assert_eq!(
758 serde_json::from_str::<DaemonCommand>(r#"{"action":"delegation_kill","orphans":true}"#)
759 .unwrap(),
760 DaemonCommand::DelegationKill {
761 handle: None,
762 orphans: true
763 }
764 );
765 let status: DaemonStatus =
767 serde_json::from_str(r#"{"version":"0.1.23","pid":7,"components":[]}"#).unwrap();
768 assert_eq!(status.delegations, DelegationSummary::default());
769 }
770
771 #[test]
772 fn server_started_turns_carry_their_origin_and_client_turns_omit_it() {
773 let started = ServerEvent::TurnStarted {
774 request_id: "background:1".into(),
775 session_id: "s".into(),
776 turn_id: "t".into(),
777 seq: 4,
778 origin: Some(TurnOrigin {
779 kind: ORIGIN_BACKGROUND.into(),
780 jobs: vec!["job-1".into()],
781 }),
782 };
783 let json = serde_json::to_value(&started).unwrap();
784 assert_eq!(
785 json["origin"],
786 serde_json::json!({"kind":"background","jobs":["job-1"]})
787 );
788 assert_eq!(
789 serde_json::from_value::<ServerEvent>(json).unwrap(),
790 started
791 );
792 assert_eq!(started.turn_request_id(), Some("background:1"));
793 let own: ServerEvent = serde_json::from_str(
795 r#"{"type":"turn.completed","request_id":"r","session_id":"s","turn_id":"t","seq":9,"steps":1,"usage":{}}"#,
796 )
797 .unwrap();
798 assert!(matches!(
799 own,
800 ServerEvent::TurnCompleted { origin: None, .. }
801 ));
802 assert!(!serde_json::to_string(&own).unwrap().contains("origin"));
803 }
804
805 #[test]
806 fn background_job_updates_come_from_start_wait_and_status_outputs() {
807 let started = background_job_update(
808 r#"{"job":"job-1","tool":"agent_codex","status":"running","background":true}"#,
809 );
810 assert_eq!(started.started, vec!["job-1".to_owned()]);
811 assert!(started.settled.is_empty());
812 let listed = background_job_update(
814 r#"{"jobs":[{"job":"job-1","status":"running"},{"job":"job-2","status":"failed"}]}"#,
815 );
816 assert!(listed.started.is_empty());
817 assert_eq!(listed.settled, vec!["job-2".to_owned()]);
818 let waited = background_job_update(r#"{"job":"job-1","status":"completed"}"#);
819 assert_eq!(waited.settled, vec!["job-1".to_owned()]);
820 for other in [
822 r#"{"agent":"codex","status":"completed"}"#,
823 "plain text",
824 "[1]",
825 ] {
826 assert_eq!(background_job_update(other), BackgroundJobUpdate::default());
827 }
828 }
829}