omni_dev/daemon/

protocol.rs

//! Wire types for the daemon's Unix-domain control socket.
//!
//! The control plane speaks newline-delimited JSON (NDJSON): one
//! [`DaemonEnvelope`] request per line, one [`DaemonReply`] response per line.
//! New optional fields use `#[serde(default, skip_serializing_if = ...)]` so
//! older peers stay byte-compatible on the wire.
//!
//! ## Subscriptions (streaming replies)
//!
//! Most ops are strictly request→one-reply. A **subscription** op is the one
//! exception (#1267): when a service recognises the op as streaming (via
//! [`DaemonService::subscribe`](super::service::DaemonService::subscribe)) the
//! server switches that connection to push mode and emits **many**
//! [`DaemonReply`] lines on the same connection — an initial snapshot, then a
//! fresh snapshot each time the service's state changes (coalesced, and diffed
//! so two identical frames are never sent in a row). Each pushed line is an
//! ordinary `DaemonReply::ok(payload)`; there is **no** new wire type, so a
//! reader distinguishes a subscription only by continuing to read lines instead
//! of stopping after one. The stream ends when the client sends any further line
//! (an explicit cancel), disconnects, or the daemon shuts down.
//!
//! The only subscription today is `worktrees` / `subscribe`, whose payload is
//! the `{ "repos": [...], "show_closed": <bool> }` tree snapshot. Back-compat is
//! total: an older client never sends a subscription op, so it only ever sees the
//! classic one-reply exchange.

use serde::{Deserialize, Serialize};
use serde_json::Value;

use super::service::ServiceStatus;
use crate::build_info::Provenance;

/// The reserved service name for the daemon's own built-in operations
/// (`ping`, `status`, `shutdown`). A `None` `service` targets the same.
pub const DAEMON_SERVICE: &str = "daemon";

/// Maximum length, in bytes, of a single NDJSON line on the control socket.
///
/// Applies to both requests and replies, capping the per-connection read buffer
/// so a peer that never sends a newline can't exhaust memory. 1 MiB is far above
/// any real envelope.
pub const MAX_LINE_BYTES: usize = 1024 * 1024;

/// A request sent to the daemon over the control socket.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DaemonEnvelope {
    /// Target service [`name`](super::service::DaemonService::name). `None`
    /// (or `"daemon"`) routes to the built-in daemon ops.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub service: Option<String>,
    /// Operation name, interpreted by the target service (or the daemon).
    pub op: String,
    /// Operation payload; `null` when the op takes no arguments.
    #[serde(default, skip_serializing_if = "Value::is_null")]
    pub payload: Value,
    /// Originating client's request-log `invocation_id`, threaded across the
    /// socket so daemon-side HTTP records correlate to the CLI/MCP invocation
    /// that triggered them rather than the daemon's own (#1198). Non-secret.
    /// Absent from older clients; the daemon simply skips the correlation then.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub origin_invocation_id: Option<String>,
}

impl DaemonEnvelope {
    /// Builds an envelope targeting a named service.
    pub fn service(name: impl Into<String>, op: impl Into<String>, payload: Value) -> Self {
        Self {
            service: Some(name.into()),
            op: op.into(),
            payload,
            origin_invocation_id: None,
        }
    }

    /// Builds an envelope targeting the built-in daemon ops.
    pub fn builtin(op: impl Into<String>) -> Self {
        Self {
            service: None,
            op: op.into(),
            payload: Value::Null,
            origin_invocation_id: None,
        }
    }

    /// Stamps the originating client's request-log `invocation_id` on the
    /// envelope so the daemon can correlate the requests it serves back to the
    /// caller's invocation (#1198).
    #[must_use]
    pub fn with_origin(mut self, invocation_id: impl Into<String>) -> Self {
        self.origin_invocation_id = Some(invocation_id.into());
        self
    }
}

/// A response returned by the daemon over the control socket.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DaemonReply {
    /// Whether the operation succeeded.
    pub ok: bool,
    /// Success payload; `null` for ops that return nothing.
    #[serde(default, skip_serializing_if = "Value::is_null")]
    pub payload: Value,
    /// Error message when [`ok`](Self::ok) is `false`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

impl DaemonReply {
    /// Builds a successful reply carrying `payload`.
    pub fn ok(payload: Value) -> Self {
        Self {
            ok: true,
            payload,
            error: None,
        }
    }

    /// Builds a failure reply carrying an error message.
    pub fn err(message: impl Into<String>) -> Self {
        Self {
            ok: false,
            payload: Value::Null,
            error: Some(message.into()),
        }
    }
}

/// The payload of a built-in `status` reply: per-service status snapshots.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct StatusReport {
    /// One entry per registered service, in registration order.
    pub services: Vec<ServiceStatus>,
    /// The daemon binary's crate version (`CARGO_PKG_VERSION`), so a client can
    /// detect it is driving a stale resident daemon after a binary upgrade
    /// (#1113). Absent from a pre-#1113 daemon; a client shows "unknown" then.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub version: Option<String>,
    /// The daemon's latest GitHub API rate-limit reading (#1375), when its
    /// monitor has polled at least once. Absent before the first poll, when no
    /// `gh` is available, or on a pre-#1375 daemon — a client shows nothing then.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub github_rate_limit: Option<crate::github_rate_limit::RateLimitSnapshot>,
    /// Git provenance (commit SHA, dirty flag, dates) of the daemon binary,
    /// flattened as sibling fields of [`version`](Self::version) so two builds
    /// of the same crate version can be told apart (#1374). Every field is
    /// omitted when absent, so a daemon built without git metadata stays
    /// byte-identical on the wire to a pre-#1374 one.
    #[serde(flatten)]
    pub provenance: Provenance,
}

impl StatusReport {
    /// Builds a status report stamped with this daemon binary's own version and
    /// git provenance (from [`crate::build_info`]). Runtime-only fields (e.g. the
    /// GitHub rate-limit reading) default to absent and are injected by the
    /// caller after construction.
    #[must_use]
    pub fn current(services: Vec<ServiceStatus>) -> Self {
        Self {
            services,
            version: Some(crate::VERSION.to_string()),
            provenance: crate::build_info::provenance(),
            ..Self::default()
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    #[test]
    fn envelope_omits_origin_when_absent() {
        // Without an origin the field is skipped, keeping the wire byte-identical
        // to what older clients/servers exchange.
        let line = serde_json::to_string(&DaemonEnvelope::service(
            "snowflake",
            "query",
            serde_json::json!({ "sql": "SELECT 1" }),
        ))
        .unwrap();
        assert!(!line.contains("origin_invocation_id"), "{line}");
    }

    #[test]
    fn envelope_round_trips_origin() {
        let env = DaemonEnvelope::service("snowflake", "query", Value::Null).with_origin("cli-42");
        let line = serde_json::to_string(&env).unwrap();
        assert!(line.contains("origin_invocation_id"), "{line}");
        let back: DaemonEnvelope = serde_json::from_str(&line).unwrap();
        assert_eq!(back.origin_invocation_id.as_deref(), Some("cli-42"));
    }

    #[test]
    fn envelope_from_older_client_defaults_origin_to_none() {
        // A line written before #1198 has no origin field; it must decode fine.
        let back: DaemonEnvelope =
            serde_json::from_str(r#"{"service":"snowflake","op":"query"}"#).unwrap();
        assert!(back.origin_invocation_id.is_none());
    }

    #[test]
    fn status_report_from_older_daemon_defaults_version_to_none() {
        // A pre-#1113 daemon's status payload has no `version`; a newer client
        // must still decode it, seeing `None` (rendered as "unknown").
        let back: StatusReport = serde_json::from_str(r#"{"services":[]}"#).unwrap();
        assert!(back.version.is_none());
    }

    #[test]
    fn status_report_round_trips_and_omits_absent_version() {
        // With no version and no provenance both are skipped, keeping the wire
        // byte-identical to what an older client/daemon exchanges.
        let line = serde_json::to_string(&StatusReport {
            services: vec![],
            version: None,
            ..StatusReport::default()
        })
        .unwrap();
        assert_eq!(line, r#"{"services":[]}"#, "{line}");

        // With a version it serializes and round-trips.
        let line = serde_json::to_string(&StatusReport {
            services: vec![],
            version: Some("1.2.3".to_string()),
            ..StatusReport::default()
        })
        .unwrap();
        assert!(line.contains("\"version\":\"1.2.3\""), "{line}");
        let back: StatusReport = serde_json::from_str(&line).unwrap();
        assert_eq!(back.version.as_deref(), Some("1.2.3"));
    }

    #[test]
    fn status_report_omits_and_round_trips_github_rate_limit() {
        use crate::github_rate_limit::{RateLimitResource, RateLimitSnapshot};

        // Absent by default: byte-identical to a pre-#1375 daemon's payload.
        let line = serde_json::to_string(&StatusReport {
            services: vec![],
            version: Some("1.2.3".to_string()),
            github_rate_limit: None,
            ..StatusReport::default()
        })
        .unwrap();
        assert!(!line.contains("github_rate_limit"), "{line}");

        // Present: serializes and decodes back, and a pre-#1375 client (which sees
        // an unknown field) ignores it — serde does that automatically.
        let snap = RateLimitSnapshot {
            graphql: Some(RateLimitResource {
                used: 4100,
                limit: 5000,
                remaining: 900,
                percent: 82.0,
                reset: 1_700_000_000,
            }),
            core: None,
            search: None,
        };
        let line = serde_json::to_string(&StatusReport {
            services: vec![],
            version: None,
            github_rate_limit: Some(snap),
            ..StatusReport::default()
        })
        .unwrap();
        assert!(line.contains("github_rate_limit"), "{line}");
        let back: StatusReport = serde_json::from_str(&line).unwrap();
        assert_eq!(back.github_rate_limit, Some(snap));
    }

    #[test]
    fn status_report_flattens_provenance_as_siblings_and_round_trips() {
        // Provenance fields render as siblings of `version`, not nested, and a
        // newer client decodes them back into the flattened struct.
        let report = StatusReport {
            services: vec![],
            version: Some("1.2.3".to_string()),
            provenance: Provenance {
                commit: Some("a6d304fd".to_string()),
                dirty: Some(true),
                ..Provenance::default()
            },
            ..StatusReport::default()
        };
        let line = serde_json::to_string(&report).unwrap();
        assert!(line.contains("\"commit\":\"a6d304fd\""), "{line}");
        assert!(line.contains("\"dirty\":true"), "{line}");
        let back: StatusReport = serde_json::from_str(&line).unwrap();
        assert_eq!(back.provenance.commit.as_deref(), Some("a6d304fd"));
        assert_eq!(back.provenance.dirty, Some(true));
        // Absent provenance fields decode to `None`, not an error.
        assert!(back.provenance.commit_long.is_none());
    }
}