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

winx_code_agent/daemon/
protocol.rs
