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";
23pub const MAX_FRAME_BYTES: usize = 8 * 1024 * 1024;
24
25/// Process role advertised during daemon negotiation.
26#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
27#[serde(rename_all = "snake_case")]
28pub enum DaemonProcessRole {
29    Control,
30    Guardian,
31}
32
33#[derive(Clone, Debug, Serialize, Deserialize)]
34pub(crate) struct RpcRequest {
35    pub jsonrpc: String,
36    pub id: u64,
37    pub method: String,
38    pub params: serde_json::Value,
39}
40
41#[derive(Debug, Serialize, Deserialize)]
42pub(crate) struct RpcResponse {
43    pub jsonrpc: String,
44    pub id: u64,
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub result: Option<serde_json::Value>,
47    #[serde(skip_serializing_if = "Option::is_none")]
48    pub error: Option<RpcError>,
49}
50
51#[derive(Debug, Serialize, Deserialize)]
52pub(crate) struct RpcError {
53    pub code: i32,
54    pub message: String,
55}
56
57#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
58pub struct HelloResult {
59    pub protocol_major: u16,
60    pub protocol_minor: u16,
61    pub capabilities: Vec<String>,
62    pub max_frame_bytes: usize,
63    pub daemon_epoch: String,
64    pub daemon_pid: u32,
65    /// Absent when talking to a daemon from before protocol 1.6.
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    pub process_role: Option<DaemonProcessRole>,
68    /// Exact executable identity. Optional for backwards-compatible decoding.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub build: Option<BuildIdentity>,
71}
72
73impl HelloResult {
74    pub(crate) fn current(
75        process_role: DaemonProcessRole,
76        mut capabilities: Vec<String>,
77        daemon_epoch: impl Into<String>,
78    ) -> Self {
79        if !capabilities.iter().any(|value| value == BUILD_IDENTITY_CAPABILITY) {
80            capabilities.push(BUILD_IDENTITY_CAPABILITY.to_string());
81        }
82        Self {
83            protocol_major: PROTOCOL_MAJOR,
84            protocol_minor: PROTOCOL_MINOR,
85            capabilities,
86            max_frame_bytes: MAX_FRAME_BYTES,
87            daemon_epoch: daemon_epoch.into(),
88            daemon_pid: std::process::id(),
89            process_role: Some(process_role),
90            build: Some(BuildIdentity::current()),
91        }
92    }
93
94    pub fn build_matches(&self, expected: &BuildIdentity) -> bool {
95        self.build.as_ref().is_some_and(|build| build.is_compatible_build(expected))
96    }
97}
98
99#[derive(Clone, Debug, Serialize, Deserialize)]
100pub(crate) struct RunActionParams {
101    pub snapshot: BashStateSnapshot,
102    pub command: BashCommand,
103    pub request_key: String,
104    #[serde(default)]
105    pub consumer_id: String,
106    #[serde(default, skip_serializing_if = "crate::runtime::ShellActionOptions::is_default")]
107    pub options: crate::runtime::ShellActionOptions,
108}
109
110#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
111#[serde(rename_all = "snake_case")]
112pub(crate) enum ConfigureSessionTransition {
113    FirstCall,
114    ModeChange,
115    Reset,
116    WorkspaceChange,
117}
118
119#[derive(Debug, Serialize, Deserialize)]
120pub(crate) struct ConfigureSessionParams {
121    pub snapshot: BashStateSnapshot,
122    pub transition: ConfigureSessionTransition,
123}
124
125#[derive(Clone, Debug, Serialize, Deserialize)]
126pub(crate) struct ConfigureSessionResult {
127    pub attach_hint: Option<String>,
128    /// Existing guardians return their authoritative state so a restarted adapter
129    /// can attach without resetting the PTY. Absent for protocol-1.2 guardians.
130    #[serde(default)]
131    pub snapshot: Option<BashStateSnapshot>,
132    #[serde(default)]
133    pub attached_existing: bool,
134}
135
136#[derive(Clone, Debug, Serialize, Deserialize)]
137pub(crate) struct RunActionResult {
138    pub output: Option<String>,
139    /// Trailer-free runtime output added in protocol 1.5. New adapters safely
140    /// fall back to `output` when attached to a protocol-1.4 guardian.
141    #[serde(default, skip_serializing_if = "Option::is_none")]
142    pub compact_output: Option<String>,
143    #[serde(default, skip_serializing_if = "Option::is_none")]
144    pub command_generation: Option<u64>,
145    #[serde(default, skip_serializing_if = "Option::is_none")]
146    pub execution_token: Option<crate::runtime::ShellExecutionToken>,
147    #[serde(default, skip_serializing_if = "Option::is_none")]
148    pub dropped_output_file: Option<std::path::PathBuf>,
149    #[serde(default)]
150    pub output_truncated: bool,
151    /// Runtime-owned state added in protocol 1.4. Optional only so a new adapter
152    /// can reject an older guardian with an actionable upgrade error instead of
153    /// trying to reconstruct state from attacker-controlled output text.
154    #[serde(default, skip_serializing_if = "Option::is_none")]
155    pub state: Option<BashCommandState>,
156    pub snapshot: BashStateSnapshot,
157    pub error: Option<WireShellError>,
158}
159
160#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
161pub struct JournalRead {
162    pub output: String,
163    pub next_seq: u64,
164    pub gap: bool,
165}
166
167#[derive(Debug, Serialize, Deserialize)]
168pub(crate) struct JournalReadParams {
169    pub thread_id: String,
170    pub consumer_id: String,
171}
172
173#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
174pub struct SessionInfo {
175    pub thread_id: String,
176    pub cwd: String,
177    pub shell_pid: Option<u32>,
178    pub command_id: Option<String>,
179    pub running: bool,
180    pub background_command_ids: Vec<String>,
181    /// Guardian-owned lifecycle clock. Optional for compatibility with guardians
182    /// created by releases before protocol 1.3.
183    #[serde(default)]
184    pub created_at_unix_ms: Option<u64>,
185    #[serde(default)]
186    pub last_activity_unix_ms: Option<u64>,
187    #[serde(default)]
188    pub last_command_at_unix_ms: Option<u64>,
189    #[serde(default)]
190    pub ever_ran_command: bool,
191    /// Guardian identity attached by the control plane. Direct and legacy
192    /// guardian responses omit it and remain wire-compatible.
193    #[serde(default, skip_serializing_if = "Option::is_none")]
194    pub runtime: Option<HelloResult>,
195}
196
197#[derive(Debug, Serialize, Deserialize)]
198pub(crate) struct SessionParams {
199    pub thread_id: String,
200    #[serde(default, skip_serializing_if = "Option::is_none")]
201    pub expected_generation: Option<u64>,
202    #[serde(default, skip_serializing_if = "Option::is_none")]
203    pub expected_execution: Option<crate::runtime::ShellExecutionToken>,
204    #[serde(default, skip_serializing_if = "Option::is_none")]
205    pub expected_guardian_epoch: Option<String>,
206}
207
208#[derive(Debug, Serialize, Deserialize)]
209pub(crate) struct CancelActionParams {
210    pub thread_id: String,
211    pub cancellation_key: String,
212    #[serde(default, skip_serializing_if = "Option::is_none")]
213    pub expected_guardian_epoch: Option<String>,
214}
215
216#[derive(Debug, Default, Serialize, Deserialize)]
217pub(crate) struct PruneParams {
218    #[serde(default)]
219    pub idle_seconds: Option<u64>,
220}
221
222#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
223pub struct PruneResult {
224    pub removed_thread_ids: Vec<String>,
225    pub skipped_active_thread_ids: Vec<String>,
226    pub stale_socket_count: usize,
227    pub unreachable_guardian_count: usize,
228}
229
230#[derive(Clone, Debug, Serialize, Deserialize)]
231#[serde(tag = "kind", content = "message", rename_all = "snake_case")]
232pub(crate) enum WireShellError {
233    BashStateNotInitialized,
234    ShellInitialization(String),
235    CommandExecution(String),
236    NoActiveCommand(String),
237    BackgroundSessionNotFound(String),
238    EmptyInteractiveInput(String),
239    InteractiveTargetNotRunning(String),
240    CommandAlreadyRunning { current_command: String, duration_seconds: f64 },
241    CommandNotAllowed(String),
242    ThreadIdMismatch(String),
243    InvalidInput(String),
244    Other(String),
245}
246
247pub(crate) async fn write_json_frame<W, T>(writer: &mut W, value: &T) -> std::io::Result<()>
248where
249    W: AsyncWrite + Unpin,
250    T: Serialize,
251{
252    let payload = serde_json::to_vec(value)
253        .map_err(|error| IoError::new(ErrorKind::InvalidData, error.to_string()))?;
254    if payload.len() > MAX_FRAME_BYTES {
255        return Err(IoError::new(
256            ErrorKind::InvalidData,
257            format!("JSON-RPC frame exceeds {MAX_FRAME_BYTES} bytes"),
258        ));
259    }
260    let length = u32::try_from(payload.len())
261        .map_err(|_| IoError::new(ErrorKind::InvalidData, "JSON-RPC frame is too large"))?;
262    writer.write_all(&length.to_be_bytes()).await?;
263    writer.write_all(&payload).await?;
264    writer.flush().await
265}
266
267pub(crate) async fn read_json_frame<R, T>(reader: &mut R) -> std::io::Result<T>
268where
269    R: AsyncRead + Unpin,
270    T: DeserializeOwned,
271{
272    let mut length = [0_u8; 4];
273    reader.read_exact(&mut length).await?;
274    let length = u32::from_be_bytes(length) as usize;
275    if length == 0 || length > MAX_FRAME_BYTES {
276        return Err(IoError::new(
277            ErrorKind::InvalidData,
278            format!("invalid JSON-RPC frame length: {length}"),
279        ));
280    }
281    let mut payload = vec![0_u8; length];
282    reader.read_exact(&mut payload).await?;
283    serde_json::from_slice(&payload)
284        .map_err(|error| IoError::new(ErrorKind::InvalidData, error.to_string()))
285}
286
287#[cfg(test)]
288mod tests {
289    #![allow(clippy::expect_used)]
290
291    use super::*;
292
293    #[test]
294    fn guardian_1_4_params_default_new_options_and_do_not_forward_wait_policy() {
295        let command: BashCommand = serde_json::from_value(serde_json::json!({
296            "command": "printf compatible",
297            "wait_policy": "until_complete",
298            "thread_id": "guardian14"
299        }))
300        .expect("adapter command");
301        let mut wire = serde_json::to_value(RunActionParams {
302            snapshot: crate::state::bash_state::BashState::new().snapshot(),
303            command,
304            request_key: "request14".to_string(),
305            consumer_id: "consumer14".to_string(),
306            options: crate::runtime::ShellActionOptions::default(),
307        })
308        .expect("current params");
309        let object = wire.as_object_mut().expect("params object");
310        assert!(object.get("options").is_none());
311        assert!(object["command"].get("wait_policy").is_none());
312
313        let decoded: RunActionParams = serde_json::from_value(wire).expect("protocol 1.4 params");
314        assert_eq!(decoded.options, crate::runtime::ShellActionOptions::default());
315    }
316
317    #[test]
318    fn guardian_1_4_result_defaults_compact_and_generation_extensions() {
319        let mut wire = serde_json::to_value(RunActionResult {
320            output: Some("legacy output".to_string()),
321            compact_output: Some("compact output".to_string()),
322            command_generation: Some(7),
323            execution_token: None,
324            dropped_output_file: Some("/tmp/winx-output".into()),
325            output_truncated: false,
326            state: None,
327            snapshot: crate::state::bash_state::BashState::new().snapshot(),
328            error: None,
329        })
330        .expect("current result");
331        let object = wire.as_object_mut().expect("result object");
332        object.remove("compact_output");
333        object.remove("command_generation");
334        object.remove("execution_token");
335        object.remove("dropped_output_file");
336        object.remove("output_truncated");
337
338        let decoded: RunActionResult = serde_json::from_value(wire).expect("protocol 1.4 result");
339        assert_eq!(decoded.output.as_deref(), Some("legacy output"));
340        assert_eq!(decoded.compact_output, None);
341        assert_eq!(decoded.command_generation, None);
342        assert_eq!(decoded.execution_token, None);
343        assert_eq!(decoded.dropped_output_file, None);
344        assert!(!decoded.output_truncated);
345    }
346
347    #[test]
348    fn compact_wire_result_contains_one_output_payload() {
349        let compact = "x".repeat(1024);
350        let wire = serde_json::to_value(RunActionResult {
351            output: None,
352            compact_output: Some(compact.clone()),
353            command_generation: Some(9),
354            execution_token: None,
355            dropped_output_file: None,
356            output_truncated: false,
357            state: None,
358            snapshot: crate::state::bash_state::BashState::new().snapshot(),
359            error: None,
360        })
361        .expect("compact result");
362        assert!(wire["output"].is_null());
363        assert_eq!(wire["compact_output"], compact);
364        let encoded = serde_json::to_string(&wire).expect("wire bytes");
365        assert_eq!(encoded.matches(&compact).count(), 1, "compact payload must not be duplicated");
366    }
367}
winx_code_agent/daemon/protocol.rs

winx_code_agent/daemon/
protocol.rs
