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

winx_code_agent/daemon/
protocol.rs
