omni_dev/daemon/
protocol.rs

1//! Wire types for the daemon's Unix-domain control socket.
2//!
3//! The control plane speaks newline-delimited JSON (NDJSON): one
4//! [`DaemonEnvelope`] request per line, one [`DaemonReply`] response per line.
5//! New optional fields use `#[serde(default, skip_serializing_if = ...)]` so
6//! older peers stay byte-compatible on the wire.
7
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10
11use super::service::ServiceStatus;
12
13/// The reserved service name for the daemon's own built-in operations
14/// (`ping`, `status`, `shutdown`). A `None` `service` targets the same.
15pub const DAEMON_SERVICE: &str = "daemon";
16
17/// Maximum length, in bytes, of a single NDJSON line on the control socket.
18///
19/// Applies to both requests and replies, capping the per-connection read buffer
20/// so a peer that never sends a newline can't exhaust memory. 1 MiB is far above
21/// any real envelope.
22pub const MAX_LINE_BYTES: usize = 1024 * 1024;
23
24/// A request sent to the daemon over the control socket.
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct DaemonEnvelope {
27    /// Target service [`name`](super::service::DaemonService::name). `None`
28    /// (or `"daemon"`) routes to the built-in daemon ops.
29    #[serde(default, skip_serializing_if = "Option::is_none")]
30    pub service: Option<String>,
31    /// Operation name, interpreted by the target service (or the daemon).
32    pub op: String,
33    /// Operation payload; `null` when the op takes no arguments.
34    #[serde(default, skip_serializing_if = "Value::is_null")]
35    pub payload: Value,
36    /// Originating client's request-log `invocation_id`, threaded across the
37    /// socket so daemon-side HTTP records correlate to the CLI/MCP invocation
38    /// that triggered them rather than the daemon's own (#1198). Non-secret.
39    /// Absent from older clients; the daemon simply skips the correlation then.
40    #[serde(default, skip_serializing_if = "Option::is_none")]
41    pub origin_invocation_id: Option<String>,
42}
43
44impl DaemonEnvelope {
45    /// Builds an envelope targeting a named service.
46    pub fn service(name: impl Into<String>, op: impl Into<String>, payload: Value) -> Self {
47        Self {
48            service: Some(name.into()),
49            op: op.into(),
50            payload,
51            origin_invocation_id: None,
52        }
53    }
54
55    /// Builds an envelope targeting the built-in daemon ops.
56    pub fn builtin(op: impl Into<String>) -> Self {
57        Self {
58            service: None,
59            op: op.into(),
60            payload: Value::Null,
61            origin_invocation_id: None,
62        }
63    }
64
65    /// Stamps the originating client's request-log `invocation_id` on the
66    /// envelope so the daemon can correlate the requests it serves back to the
67    /// caller's invocation (#1198).
68    #[must_use]
69    pub fn with_origin(mut self, invocation_id: impl Into<String>) -> Self {
70        self.origin_invocation_id = Some(invocation_id.into());
71        self
72    }
73}
74
75/// A response returned by the daemon over the control socket.
76#[derive(Debug, Clone, Serialize, Deserialize)]
77pub struct DaemonReply {
78    /// Whether the operation succeeded.
79    pub ok: bool,
80    /// Success payload; `null` for ops that return nothing.
81    #[serde(default, skip_serializing_if = "Value::is_null")]
82    pub payload: Value,
83    /// Error message when [`ok`](Self::ok) is `false`.
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub error: Option<String>,
86}
87
88impl DaemonReply {
89    /// Builds a successful reply carrying `payload`.
90    pub fn ok(payload: Value) -> Self {
91        Self {
92            ok: true,
93            payload,
94            error: None,
95        }
96    }
97
98    /// Builds a failure reply carrying an error message.
99    pub fn err(message: impl Into<String>) -> Self {
100        Self {
101            ok: false,
102            payload: Value::Null,
103            error: Some(message.into()),
104        }
105    }
106}
107
108/// The payload of a built-in `status` reply: per-service status snapshots.
109#[derive(Debug, Clone, Serialize, Deserialize)]
110pub struct StatusReport {
111    /// One entry per registered service, in registration order.
112    pub services: Vec<ServiceStatus>,
113}
114
115#[cfg(test)]
116#[allow(clippy::unwrap_used, clippy::expect_used)]
117mod tests {
118    use super::*;
119
120    #[test]
121    fn envelope_omits_origin_when_absent() {
122        // Without an origin the field is skipped, keeping the wire byte-identical
123        // to what older clients/servers exchange.
124        let line = serde_json::to_string(&DaemonEnvelope::service(
125            "snowflake",
126            "query",
127            serde_json::json!({ "sql": "SELECT 1" }),
128        ))
129        .unwrap();
130        assert!(!line.contains("origin_invocation_id"), "{line}");
131    }
132
133    #[test]
134    fn envelope_round_trips_origin() {
135        let env = DaemonEnvelope::service("snowflake", "query", Value::Null).with_origin("cli-42");
136        let line = serde_json::to_string(&env).unwrap();
137        assert!(line.contains("origin_invocation_id"), "{line}");
138        let back: DaemonEnvelope = serde_json::from_str(&line).unwrap();
139        assert_eq!(back.origin_invocation_id.as_deref(), Some("cli-42"));
140    }
141
142    #[test]
143    fn envelope_from_older_client_defaults_origin_to_none() {
144        // A line written before #1198 has no origin field; it must decode fine.
145        let back: DaemonEnvelope =
146            serde_json::from_str(r#"{"service":"snowflake","op":"query"}"#).unwrap();
147        assert!(back.origin_invocation_id.is_none());
148    }
149}
omni_dev/daemon/protocol.rs

omni_dev/daemon/
protocol.rs
