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

winx_code_agent/daemon/
protocol.rs
