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)]
102    pub output_truncated: bool,
103    /// Runtime-owned state added in protocol 1.4. Optional only so a new adapter
104    /// can reject an older guardian with an actionable upgrade error instead of
105    /// trying to reconstruct state from attacker-controlled output text.
106    #[serde(default, skip_serializing_if = "Option::is_none")]
107    pub state: Option<BashCommandState>,
108    pub snapshot: BashStateSnapshot,
109    pub error: Option<WireShellError>,
110}
111
112#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
113pub struct JournalRead {
114    pub output: String,
115    pub next_seq: u64,
116    pub gap: bool,
117}
118
119#[derive(Debug, Serialize, Deserialize)]
120pub(crate) struct JournalReadParams {
121    pub thread_id: String,
122    pub consumer_id: String,
123}
124
125#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
126pub struct SessionInfo {
127    pub thread_id: String,
128    pub cwd: String,
129    pub shell_pid: Option<u32>,
130    pub command_id: Option<String>,
131    pub running: bool,
132    pub background_command_ids: Vec<String>,
133    /// Guardian-owned lifecycle clock. Optional for compatibility with guardians
134    /// created by releases before protocol 1.3.
135    #[serde(default)]
136    pub created_at_unix_ms: Option<u64>,
137    #[serde(default)]
138    pub last_activity_unix_ms: Option<u64>,
139    #[serde(default)]
140    pub last_command_at_unix_ms: Option<u64>,
141    #[serde(default)]
142    pub ever_ran_command: bool,
143}
144
145#[derive(Debug, Serialize, Deserialize)]
146pub(crate) struct SessionParams {
147    pub thread_id: String,
148    #[serde(default, skip_serializing_if = "Option::is_none")]
149    pub expected_generation: Option<u64>,
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    pub expected_execution: Option<crate::runtime::ShellExecutionToken>,
152    #[serde(default, skip_serializing_if = "Option::is_none")]
153    pub expected_guardian_epoch: Option<String>,
154}
155
156#[derive(Debug, Serialize, Deserialize)]
157pub(crate) struct CancelActionParams {
158    pub thread_id: String,
159    pub cancellation_key: String,
160    #[serde(default, skip_serializing_if = "Option::is_none")]
161    pub expected_guardian_epoch: Option<String>,
162}
163
164#[derive(Debug, Default, Serialize, Deserialize)]
165pub(crate) struct PruneParams {
166    #[serde(default)]
167    pub idle_seconds: Option<u64>,
168}
169
170#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
171pub struct PruneResult {
172    pub removed_thread_ids: Vec<String>,
173    pub skipped_active_thread_ids: Vec<String>,
174    pub stale_socket_count: usize,
175    pub unreachable_guardian_count: usize,
176}
177
178#[derive(Clone, Debug, Serialize, Deserialize)]
179#[serde(tag = "kind", content = "message", rename_all = "snake_case")]
180pub(crate) enum WireShellError {
181    BashStateNotInitialized,
182    ShellInitialization(String),
183    CommandExecution(String),
184    NoActiveCommand(String),
185    BackgroundSessionNotFound(String),
186    EmptyInteractiveInput(String),
187    InteractiveTargetNotRunning(String),
188    CommandAlreadyRunning { current_command: String, duration_seconds: f64 },
189    CommandNotAllowed(String),
190    ThreadIdMismatch(String),
191    InvalidInput(String),
192    Other(String),
193}
194
195pub(crate) async fn write_json_frame<W, T>(writer: &mut W, value: &T) -> std::io::Result<()>
196where
197    W: AsyncWrite + Unpin,
198    T: Serialize,
199{
200    let payload = serde_json::to_vec(value)
201        .map_err(|error| IoError::new(ErrorKind::InvalidData, error.to_string()))?;
202    if payload.len() > MAX_FRAME_BYTES {
203        return Err(IoError::new(
204            ErrorKind::InvalidData,
205            format!("JSON-RPC frame exceeds {MAX_FRAME_BYTES} bytes"),
206        ));
207    }
208    let length = u32::try_from(payload.len())
209        .map_err(|_| IoError::new(ErrorKind::InvalidData, "JSON-RPC frame is too large"))?;
210    writer.write_all(&length.to_be_bytes()).await?;
211    writer.write_all(&payload).await?;
212    writer.flush().await
213}
214
215pub(crate) async fn read_json_frame<R, T>(reader: &mut R) -> std::io::Result<T>
216where
217    R: AsyncRead + Unpin,
218    T: DeserializeOwned,
219{
220    let mut length = [0_u8; 4];
221    reader.read_exact(&mut length).await?;
222    let length = u32::from_be_bytes(length) as usize;
223    if length == 0 || length > MAX_FRAME_BYTES {
224        return Err(IoError::new(
225            ErrorKind::InvalidData,
226            format!("invalid JSON-RPC frame length: {length}"),
227        ));
228    }
229    let mut payload = vec![0_u8; length];
230    reader.read_exact(&mut payload).await?;
231    serde_json::from_slice(&payload)
232        .map_err(|error| IoError::new(ErrorKind::InvalidData, error.to_string()))
233}
234
235#[cfg(test)]
236mod tests {
237    #![allow(clippy::expect_used)]
238
239    use super::*;
240
241    #[test]
242    fn guardian_1_4_params_default_new_options_and_do_not_forward_wait_policy() {
243        let command: BashCommand = serde_json::from_value(serde_json::json!({
244            "command": "printf compatible",
245            "wait_policy": "until_complete",
246            "thread_id": "guardian14"
247        }))
248        .expect("adapter command");
249        let mut wire = serde_json::to_value(RunActionParams {
250            snapshot: crate::state::bash_state::BashState::new().snapshot(),
251            command,
252            request_key: "request14".to_string(),
253            consumer_id: "consumer14".to_string(),
254            options: crate::runtime::ShellActionOptions::default(),
255        })
256        .expect("current params");
257        let object = wire.as_object_mut().expect("params object");
258        assert!(object.get("options").is_none());
259        assert!(object["command"].get("wait_policy").is_none());
260
261        let decoded: RunActionParams = serde_json::from_value(wire).expect("protocol 1.4 params");
262        assert_eq!(decoded.options, crate::runtime::ShellActionOptions::default());
263    }
264
265    #[test]
266    fn guardian_1_4_result_defaults_compact_and_generation_extensions() {
267        let mut wire = serde_json::to_value(RunActionResult {
268            output: Some("legacy output".to_string()),
269            compact_output: Some("compact output".to_string()),
270            command_generation: Some(7),
271            execution_token: None,
272            output_truncated: false,
273            state: None,
274            snapshot: crate::state::bash_state::BashState::new().snapshot(),
275            error: None,
276        })
277        .expect("current result");
278        let object = wire.as_object_mut().expect("result object");
279        object.remove("compact_output");
280        object.remove("command_generation");
281        object.remove("execution_token");
282        object.remove("output_truncated");
283
284        let decoded: RunActionResult = serde_json::from_value(wire).expect("protocol 1.4 result");
285        assert_eq!(decoded.output.as_deref(), Some("legacy output"));
286        assert_eq!(decoded.compact_output, None);
287        assert_eq!(decoded.command_generation, None);
288        assert_eq!(decoded.execution_token, None);
289        assert!(!decoded.output_truncated);
290    }
291
292    #[test]
293    fn compact_wire_result_contains_one_output_payload() {
294        let compact = "x".repeat(1024);
295        let wire = serde_json::to_value(RunActionResult {
296            output: None,
297            compact_output: Some(compact.clone()),
298            command_generation: Some(9),
299            execution_token: None,
300            output_truncated: false,
301            state: None,
302            snapshot: crate::state::bash_state::BashState::new().snapshot(),
303            error: None,
304        })
305        .expect("compact result");
306        assert!(wire["output"].is_null());
307        assert_eq!(wire["compact_output"], compact);
308        let encoded = serde_json::to_string(&wire).expect("wire bytes");
309        assert_eq!(encoded.matches(&compact).count(), 1, "compact payload must not be duplicated");
310    }
311}
winx_code_agent/daemon/protocol.rs

winx_code_agent/daemon/
protocol.rs
