winx_code_agent/daemon/
protocol.rs

1use std::io::{Error as IoError, ErrorKind};
2
3use serde::de::DeserializeOwned;
4use serde::{Deserialize, Serialize};
5use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
6
7use crate::build_info::BuildIdentity;
8use crate::state::persistence::BashStateSnapshot;
9use crate::tools::bash_command::BashCommandState;
10use crate::types::BashCommand;
11
12pub const PROTOCOL_MAJOR: u16 = 1;
13pub const PROTOCOL_MINOR: u16 = 6;
14pub const TYPED_ACTION_RESULT_CAPABILITY: &str = "typed_action_result";
15pub const COMPACT_ACTION_OUTPUT_CAPABILITY: &str = "compact_action_output";
16pub const GENERATION_BOUND_ACTIONS_CAPABILITY: &str = "generation_bound_actions";
17pub const CANCELLABLE_ACTION_RESERVATIONS_CAPABILITY: &str = "cancellable_action_reservations";
18pub const ATTACH_OR_CREATE_CAPABILITY: &str = "attach_or_create";
19pub const BUILD_IDENTITY_CAPABILITY: &str = "build_identity";
20/// The process exits cleanly on a `winx.shutdown` request; Windows has no
21/// SIGTERM, so this is the only graceful stop there.
22pub const PROCESS_SHUTDOWN_CAPABILITY: &str = "process_shutdown";
23/// `session.kill` honors `only_if_idle` and refuses a session that is running
24/// a foreground command or owns background shells.
25pub const CONDITIONAL_SESSION_KILL_CAPABILITY: &str = "conditional_session_kill";
26pub const MAX_FRAME_BYTES: usize = 8 * 1024 * 1024;
27
28/// Process role advertised during daemon negotiation.
29#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
30#[serde(rename_all = "snake_case")]
31pub enum DaemonProcessRole {
32    Control,
33    Guardian,
34}
35
36#[derive(Clone, Debug, Serialize, Deserialize)]
37pub(crate) struct RpcRequest {
38    pub jsonrpc: String,
39    pub id: u64,
40    pub method: String,
41    pub params: serde_json::Value,
42}
43
44#[derive(Debug, Serialize, Deserialize)]
45pub(crate) struct RpcResponse {
46    pub jsonrpc: String,
47    pub id: u64,
48    #[serde(skip_serializing_if = "Option::is_none")]
49    pub result: Option<serde_json::Value>,
50    #[serde(skip_serializing_if = "Option::is_none")]
51    pub error: Option<RpcError>,
52}
53
54#[derive(Debug, Serialize, Deserialize)]
55pub(crate) struct RpcError {
56    pub code: i32,
57    pub message: String,
58}
59
60#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
61pub struct HelloResult {
62    pub protocol_major: u16,
63    pub protocol_minor: u16,
64    pub capabilities: Vec<String>,
65    pub max_frame_bytes: usize,
66    pub daemon_epoch: String,
67    pub daemon_pid: u32,
68    /// Absent when talking to a daemon from before protocol 1.6.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub process_role: Option<DaemonProcessRole>,
71    /// Exact executable identity. Optional for backwards-compatible decoding.
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub build: Option<BuildIdentity>,
74}
75
76impl HelloResult {
77    pub(crate) fn current(
78        process_role: DaemonProcessRole,
79        mut capabilities: Vec<String>,
80        daemon_epoch: impl Into<String>,
81    ) -> Self {
82        if !capabilities.iter().any(|value| value == BUILD_IDENTITY_CAPABILITY) {
83            capabilities.push(BUILD_IDENTITY_CAPABILITY.to_string());
84        }
85        Self {
86            protocol_major: PROTOCOL_MAJOR,
87            protocol_minor: PROTOCOL_MINOR,
88            capabilities,
89            max_frame_bytes: MAX_FRAME_BYTES,
90            daemon_epoch: daemon_epoch.into(),
91            daemon_pid: std::process::id(),
92            process_role: Some(process_role),
93            build: Some(BuildIdentity::current()),
94        }
95    }
96
97    pub fn build_matches(&self, expected: &BuildIdentity) -> bool {
98        self.build.as_ref().is_some_and(|build| build.is_compatible_build(expected))
99    }
100}
101
102#[derive(Clone, Debug, Serialize, Deserialize)]
103pub(crate) struct RunActionParams {
104    pub snapshot: BashStateSnapshot,
105    pub command: BashCommand,
106    pub request_key: String,
107    #[serde(default)]
108    pub consumer_id: String,
109    #[serde(default, skip_serializing_if = "crate::runtime::ShellActionOptions::is_default")]
110    pub options: crate::runtime::ShellActionOptions,
111}
112
113#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
114#[serde(rename_all = "snake_case")]
115pub(crate) enum ConfigureSessionTransition {
116    FirstCall,
117    ModeChange,
118    Reset,
119    WorkspaceChange,
120}
121
122#[derive(Debug, Serialize, Deserialize)]
123pub(crate) struct ConfigureSessionParams {
124    pub snapshot: BashStateSnapshot,
125    pub transition: ConfigureSessionTransition,
126}
127
128#[derive(Clone, Debug, Serialize, Deserialize)]
129pub(crate) struct ConfigureSessionResult {
130    pub attach_hint: Option<String>,
131    /// Existing guardians return their authoritative state so a restarted adapter
132    /// can attach without resetting the PTY. Absent for protocol-1.2 guardians.
133    #[serde(default)]
134    pub snapshot: Option<BashStateSnapshot>,
135    #[serde(default)]
136    pub attached_existing: bool,
137}
138
139#[derive(Clone, Debug, Serialize, Deserialize)]
140pub(crate) struct RunActionResult {
141    pub output: Option<String>,
142    /// Trailer-free runtime output added in protocol 1.5. New adapters safely
143    /// fall back to `output` when attached to a protocol-1.4 guardian.
144    #[serde(default, skip_serializing_if = "Option::is_none")]
145    pub compact_output: Option<String>,
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub command_generation: Option<u64>,
148    #[serde(default, skip_serializing_if = "Option::is_none")]
149    pub execution_token: Option<crate::runtime::ShellExecutionToken>,
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    pub dropped_output_file: Option<std::path::PathBuf>,
152    #[serde(default)]
153    pub output_truncated: bool,
154    /// Runtime-owned state added in protocol 1.4. Optional only so a new adapter
155    /// can reject an older guardian with an actionable upgrade error instead of
156    /// trying to reconstruct state from attacker-controlled output text.
157    #[serde(default, skip_serializing_if = "Option::is_none")]
158    pub state: Option<BashCommandState>,
159    pub snapshot: BashStateSnapshot,
160    pub error: Option<WireShellError>,
161}
162
163#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
164pub struct JournalRead {
165    pub output: String,
166    pub next_seq: u64,
167    pub gap: bool,
168}
169
170#[derive(Debug, Serialize, Deserialize)]
171pub(crate) struct JournalReadParams {
172    pub thread_id: String,
173    pub consumer_id: String,
174}
175
176#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
177pub struct SessionInfo {
178    pub thread_id: String,
179    pub cwd: String,
180    pub shell_pid: Option<u32>,
181    pub command_id: Option<String>,
182    pub running: bool,
183    pub background_command_ids: Vec<String>,
184    /// Guardian-owned lifecycle clock. Optional for compatibility with guardians
185    /// created by releases before protocol 1.3.
186    #[serde(default)]
187    pub created_at_unix_ms: Option<u64>,
188    #[serde(default)]
189    pub last_activity_unix_ms: Option<u64>,
190    #[serde(default)]
191    pub last_command_at_unix_ms: Option<u64>,
192    #[serde(default)]
193    pub ever_ran_command: bool,
194    /// Guardian identity attached by the control plane. Direct and legacy
195    /// guardian responses omit it and remain wire-compatible.
196    #[serde(default, skip_serializing_if = "Option::is_none")]
197    pub runtime: Option<HelloResult>,
198}
199
200#[derive(Debug, Serialize, Deserialize)]
201pub(crate) struct SessionParams {
202    pub thread_id: String,
203    /// `session.kill` only: refuse (return `false`) when the session is busy.
204    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
205    pub only_if_idle: bool,
206    #[serde(default, skip_serializing_if = "Option::is_none")]
207    pub expected_generation: Option<u64>,
208    #[serde(default, skip_serializing_if = "Option::is_none")]
209    pub expected_execution: Option<crate::runtime::ShellExecutionToken>,
210    #[serde(default, skip_serializing_if = "Option::is_none")]
211    pub expected_guardian_epoch: Option<String>,
212}
213
214#[derive(Debug, Serialize, Deserialize)]
215pub(crate) struct CancelActionParams {
216    pub thread_id: String,
217    pub cancellation_key: String,
218    #[serde(default, skip_serializing_if = "Option::is_none")]
219    pub expected_guardian_epoch: Option<String>,
220}
221
222#[derive(Debug, Default, Serialize, Deserialize)]
223pub(crate) struct PruneParams {
224    #[serde(default)]
225    pub idle_seconds: Option<u64>,
226}
227
228#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
229pub struct PruneResult {
230    pub removed_thread_ids: Vec<String>,
231    pub skipped_active_thread_ids: Vec<String>,
232    pub stale_socket_count: usize,
233    pub unreachable_guardian_count: usize,
234}
235
236#[derive(Clone, Debug, Serialize, Deserialize)]
237#[serde(tag = "kind", content = "message", rename_all = "snake_case")]
238pub(crate) enum WireShellError {
239    BashStateNotInitialized,
240    ShellInitialization(String),
241    CommandExecution(String),
242    NoActiveCommand(String),
243    BackgroundSessionNotFound(String),
244    EmptyInteractiveInput(String),
245    InteractiveTargetNotRunning(String),
246    CommandAlreadyRunning { current_command: String, duration_seconds: f64 },
247    CommandNotAllowed(String),
248    ThreadIdMismatch(String),
249    InvalidInput(String),
250    Other(String),
251}
252
253pub(crate) async fn write_json_frame<W, T>(writer: &mut W, value: &T) -> std::io::Result<()>
254where
255    W: AsyncWrite + Unpin,
256    T: Serialize,
257{
258    let payload = serde_json::to_vec(value)
259        .map_err(|error| IoError::new(ErrorKind::InvalidData, error.to_string()))?;
260    if payload.len() > MAX_FRAME_BYTES {
261        return Err(IoError::new(
262            ErrorKind::InvalidData,
263            format!("JSON-RPC frame exceeds {MAX_FRAME_BYTES} bytes"),
264        ));
265    }
266    let length = u32::try_from(payload.len())
267        .map_err(|_| IoError::new(ErrorKind::InvalidData, "JSON-RPC frame is too large"))?;
268    writer.write_all(&length.to_be_bytes()).await?;
269    writer.write_all(&payload).await?;
270    writer.flush().await
271}
272
273pub(crate) async fn read_json_frame<R, T>(reader: &mut R) -> std::io::Result<T>
274where
275    R: AsyncRead + Unpin,
276    T: DeserializeOwned,
277{
278    let mut length = [0_u8; 4];
279    reader.read_exact(&mut length).await?;
280    let length = u32::from_be_bytes(length) as usize;
281    if length == 0 || length > MAX_FRAME_BYTES {
282        return Err(IoError::new(
283            ErrorKind::InvalidData,
284            format!("invalid JSON-RPC frame length: {length}"),
285        ));
286    }
287    let mut payload = vec![0_u8; length];
288    reader.read_exact(&mut payload).await?;
289    serde_json::from_slice(&payload)
290        .map_err(|error| IoError::new(ErrorKind::InvalidData, error.to_string()))
291}
292
293#[cfg(test)]
294mod tests {
295    #![allow(clippy::expect_used)]
296
297    use super::*;
298
299    #[test]
300    fn guardian_1_4_params_default_new_options_and_do_not_forward_wait_policy() {
301        let command: BashCommand = serde_json::from_value(serde_json::json!({
302            "command": "printf compatible",
303            "wait_policy": "until_complete",
304            "thread_id": "guardian14"
305        }))
306        .expect("adapter command");
307        let mut wire = serde_json::to_value(RunActionParams {
308            snapshot: crate::state::bash_state::BashState::new().snapshot(),
309            command,
310            request_key: "request14".to_string(),
311            consumer_id: "consumer14".to_string(),
312            options: crate::runtime::ShellActionOptions::default(),
313        })
314        .expect("current params");
315        let object = wire.as_object_mut().expect("params object");
316        assert!(object.get("options").is_none());
317        assert!(object["command"].get("wait_policy").is_none());
318
319        let decoded: RunActionParams = serde_json::from_value(wire).expect("protocol 1.4 params");
320        assert_eq!(decoded.options, crate::runtime::ShellActionOptions::default());
321    }
322
323    #[test]
324    fn guardian_1_4_result_defaults_compact_and_generation_extensions() {
325        let mut wire = serde_json::to_value(RunActionResult {
326            output: Some("legacy output".to_string()),
327            compact_output: Some("compact output".to_string()),
328            command_generation: Some(7),
329            execution_token: None,
330            dropped_output_file: Some("/tmp/winx-output".into()),
331            output_truncated: false,
332            state: None,
333            snapshot: crate::state::bash_state::BashState::new().snapshot(),
334            error: None,
335        })
336        .expect("current result");
337        let object = wire.as_object_mut().expect("result object");
338        object.remove("compact_output");
339        object.remove("command_generation");
340        object.remove("execution_token");
341        object.remove("dropped_output_file");
342        object.remove("output_truncated");
343
344        let decoded: RunActionResult = serde_json::from_value(wire).expect("protocol 1.4 result");
345        assert_eq!(decoded.output.as_deref(), Some("legacy output"));
346        assert_eq!(decoded.compact_output, None);
347        assert_eq!(decoded.command_generation, None);
348        assert_eq!(decoded.execution_token, None);
349        assert_eq!(decoded.dropped_output_file, None);
350        assert!(!decoded.output_truncated);
351    }
352
353    #[test]
354    fn compact_wire_result_contains_one_output_payload() {
355        let compact = "x".repeat(1024);
356        let wire = serde_json::to_value(RunActionResult {
357            output: None,
358            compact_output: Some(compact.clone()),
359            command_generation: Some(9),
360            execution_token: None,
361            dropped_output_file: None,
362            output_truncated: false,
363            state: None,
364            snapshot: crate::state::bash_state::BashState::new().snapshot(),
365            error: None,
366        })
367        .expect("compact result");
368        assert!(wire["output"].is_null());
369        assert_eq!(wire["compact_output"], compact);
370        let encoded = serde_json::to_string(&wire).expect("wire bytes");
371        assert_eq!(encoded.matches(&compact).count(), 1, "compact payload must not be duplicated");
372    }
373}
374
375#[cfg(test)]
376mod wire_contract_tests {
377    //! Frozen wire contract for protocol major 1 (`tests/fixtures/daemon-wire-v1.json`).
378    //!
379    //! An adapter and a daemon from different releases exchange these shapes,
380    //! so every fixture sample must keep decoding, legacy samples must re-encode
381    //! as a superset of themselves, and current samples must re-encode exactly.
382    //! A failure here means a wire change: add a capability, a method, or an
383    //! optional field with `skip_serializing_if`, and append a new sample
384    //! instead of editing an old one.
385    #![allow(clippy::expect_used, clippy::panic)]
386
387    use serde::de::DeserializeOwned;
388    use serde::Serialize;
389    use serde_json::Value;
390
391    use super::*;
392
393    const FIXTURE: &str = include_str!("../../tests/fixtures/daemon-wire-v1.json");
394    const CONTROL_SOURCE: &str = include_str!("control.rs");
395    const GUARDIAN_SOURCE: &str = include_str!("server.rs");
396    const PROTOCOL_SOURCE: &str = include_str!("protocol.rs");
397
398    fn fixture() -> Value {
399        serde_json::from_str(FIXTURE).expect("fixture is valid JSON")
400    }
401
402    fn strings(value: &Value) -> Vec<String> {
403        value
404            .as_array()
405            .expect("array")
406            .iter()
407            .map(|item| item.as_str().expect("string").to_string())
408            .collect()
409    }
410
411    /// Every quoted `"prefix.name"` literal in `source`, for the given prefixes.
412    fn quoted_methods(source: &str) -> Vec<String> {
413        let mut found = Vec::new();
414        for (index, _) in source.match_indices('"') {
415            let rest = &source[index + 1..];
416            let Some(end) = rest.find('"') else { continue };
417            let literal = &rest[..end];
418            let is_method =
419                ["winx.", "session.", "shell."].iter().any(|prefix| literal.starts_with(prefix))
420                    && literal.len() > 6
421                    && literal.chars().all(|c| c.is_ascii_lowercase() || c == '.' || c == '_');
422            if is_method && !found.contains(&literal.to_string()) {
423                found.push(literal.to_string());
424            }
425        }
426        found
427    }
428
429    fn assert_subset(expected: &Value, actual: &Value, path: &str) {
430        match (expected, actual) {
431            (Value::Object(expected), Value::Object(actual)) => {
432                for (key, value) in expected {
433                    let child = actual.get(key).unwrap_or_else(|| {
434                        panic!("{path}.{key} disappeared from the re-encoded message")
435                    });
436                    assert_subset(value, child, &format!("{path}.{key}"));
437                }
438            }
439            (expected, actual) => assert_eq!(expected, actual, "{path} changed meaning"),
440        }
441    }
442
443    fn round_trip<T: DeserializeOwned + Serialize>(sample: &Value) -> Value {
444        let decoded: T = serde_json::from_value(sample.clone())
445            .unwrap_or_else(|error| panic!("sample must decode: {error}\n{sample}"));
446        serde_json::to_value(&decoded).expect("re-encode")
447    }
448
449    fn legacy<T: DeserializeOwned + Serialize>(name: &str) {
450        let sample = fixture()["samples"][name].clone();
451        assert!(sample.is_object(), "missing sample {name}");
452        assert_subset(&sample, &round_trip::<T>(&sample), name);
453    }
454
455    fn current<T: DeserializeOwned + Serialize>(name: &str) {
456        let sample = fixture()["samples"][name].clone();
457        assert!(sample.is_object(), "missing sample {name}");
458        assert_eq!(round_trip::<T>(&sample), sample, "{name} must re-encode exactly");
459    }
460
461    #[test]
462    fn protocol_version_stays_within_the_frozen_major() {
463        let fixture = fixture();
464        assert_eq!(u64::from(PROTOCOL_MAJOR), fixture["protocol_major"].as_u64().expect("major"));
465        assert!(
466            u64::from(PROTOCOL_MINOR) >= fixture["min_protocol_minor"].as_u64().expect("minor"),
467            "protocol minor must never go backwards"
468        );
469    }
470
471    #[test]
472    fn every_method_the_daemons_dispatch_is_frozen_in_the_fixture() {
473        let frozen = strings(&fixture()["methods"]);
474        let mut in_code = quoted_methods(CONTROL_SOURCE);
475        for method in quoted_methods(GUARDIAN_SOURCE) {
476            if !in_code.contains(&method) {
477                in_code.push(method);
478            }
479        }
480        for method in &frozen {
481            assert!(in_code.contains(method), "frozen method {method} is no longer dispatched");
482        }
483        for method in &in_code {
484            assert!(frozen.contains(method), "new method {method}: append it to the fixture");
485        }
486    }
487
488    #[test]
489    fn every_capability_constant_is_frozen_in_the_fixture() {
490        let frozen = strings(&fixture()["capabilities"]);
491        let in_code: Vec<String> = PROTOCOL_SOURCE
492            .lines()
493            .filter(|line| line.contains("_CAPABILITY: &str = \""))
494            .map(|line| {
495                let start = line.find('"').expect("open quote") + 1;
496                let end = line[start..].find('"').expect("close quote") + start;
497                line[start..end].to_string()
498            })
499            .collect();
500        for capability in &frozen {
501            assert!(in_code.contains(capability), "frozen capability {capability} was removed");
502        }
503        for capability in &in_code {
504            assert!(
505                frozen.contains(capability),
506                "new capability {capability}: append it to the fixture"
507            );
508        }
509    }
510
511    #[test]
512    fn frames_and_results_keep_their_frozen_shapes() {
513        current::<RpcRequest>("rpc_request");
514        current::<RpcResponse>("rpc_response_result");
515        current::<RpcResponse>("rpc_response_error");
516        legacy::<HelloResult>("hello_result_legacy_1_5");
517        current::<HelloResult>("hello_result_1_6");
518        legacy::<SessionInfo>("session_info_legacy_1_2");
519        current::<SessionInfo>("session_info_1_6");
520        current::<JournalRead>("journal_read");
521        current::<PruneResult>("prune_result");
522        let shutdown = fixture()["samples"]["shutdown_result"].clone();
523        assert_eq!(shutdown["shutting_down"], Value::Bool(true));
524    }
525
526    #[test]
527    fn legacy_hello_without_role_or_build_is_still_accepted() {
528        let sample = fixture()["samples"]["hello_result_legacy_1_5"].clone();
529        let hello: HelloResult = serde_json::from_value(sample).expect("decode");
530        assert_eq!(hello.process_role, None);
531        assert_eq!(hello.build, None);
532        assert!(!hello.build_matches(&crate::build_info::BuildIdentity::current()));
533    }
534}
winx_code_agent/daemon/protocol.rs

winx_code_agent/daemon/
protocol.rs
