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

winx_code_agent/daemon/
protocol.rs
