tawasuyu/brahman

1//! `shipote-protocol` — wire daemon ↔ cliente (cli/gui).
2//!
3//! Framing: u32 BE length-prefix + payload postcard. Mismo patrón que
4//! `ente-bus`/`brahman-handshake` para que clientes existentes compartan
5//! reader/writer helpers si quieren.
6
7#![forbid(unsafe_code)]
8
9use serde::{Deserialize, Serialize};
10use shipote_card::{PipelineSpec, WorkspaceId, WorkspaceSpec};
11use std::path::PathBuf;
12use thiserror::Error;
13use tokio::io::{AsyncReadExt, AsyncWriteExt};
14use tokio::net::UnixStream;
15use ulid::Ulid;
16
17pub const DEFAULT_SOCK_NAME: &str = "shipote.sock";
18pub const MAX_FRAME: usize = 1 << 20;
19
20fn default_grace_ms() -> u64 {
21    1000
22}
23
24// =====================================================================
25// Mensajes
26// =====================================================================
27
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub enum Request {
30    /// Health-check.
31    Ping,
32
33    /// Crear un workspace nuevo.
34    WorkspaceCreate { spec: WorkspaceSpec },
35
36    /// Listar todos los workspaces vivos.
37    WorkspaceList,
38
39    /// Detener un workspace y reapear sus comandos. `grace_ms`: tiempo
40    /// que se espera tras SIGTERM antes de SIGKILL. 0 = SIGKILL inmediato.
41    WorkspaceStop {
42        id: WorkspaceId,
43        #[serde(default = "default_grace_ms")]
44        grace_ms: u64,
45    },
46
47    /// Ejecutar un comando one-shot dentro de un workspace existente.
48    Run {
49        workspace: WorkspaceId,
50        exec: String,
51        argv: Vec<String>,
52        envp: Vec<(String, String)>,
53    },
54
55    /// Lanzar un Pipeline completo dentro de un workspace.
56    PipelineRun {
57        spec: PipelineSpec,
58        /// Si `true`, el daemon interpone un tap entre productor y
59        /// consumidor de cada FlowEdge, sampleando los primeros bytes
60        /// y discerniendo el TypeRef.
61        tap: bool,
62        /// Variables para sustitución `${KEY}` en strings del spec
63        /// antes de spawn (templating).
64        #[serde(default)]
65        vars: std::collections::BTreeMap<String, String>,
66    },
67
68    /// Discernir un buffer ad-hoc (sin workspace). Útil para `shipote discern <file>`.
69    Discern { sample: Vec<u8>, hint_path: Option<PathBuf> },
70
71    /// Capacidades runtime del kernel/proceso del daemon.
72    Capabilities,
73
74    /// Listar comandos vivos+pasados de un workspace.
75    CommandList { workspace: shipote_card::WorkspaceId },
76
77    /// Tail del log capturado para un comando.
78    CommandLogs {
79        workspace: shipote_card::WorkspaceId,
80        command: Ulid,
81        tail_bytes: usize,
82        /// "stdout" | "stderr" | "both" (default "both" si vacío).
83        stream: String,
84    },
85
86    /// Guardar (o reemplazar) un PipelineSpec bajo un nombre.
87    PipelineSave { name: String, spec: PipelineSpec },
88
89    /// Listar nombres de pipelines guardados.
90    PipelineSavedList,
91
92    /// Eliminar un pipeline guardado.
93    PipelineDrop { name: String },
94
95    /// Ejecutar un pipeline guardado.
96    PipelineRunSaved {
97        name: String,
98        tap: bool,
99        #[serde(default)]
100        vars: std::collections::BTreeMap<String, String>,
101    },
102
103    /// Resource accounting de un workspace.
104    WorkspaceStats { workspace: shipote_card::WorkspaceId },
105
106    /// Detener selectivamente los comandos de un pipeline (no el workspace
107    /// entero). `grace_ms`: SIGTERM → wait → SIGKILL.
108    PipelineStop {
109        pipeline: Ulid,
110        #[serde(default = "default_grace_ms")]
111        grace_ms: u64,
112    },
113
114    /// Listar pipelines activos con sus flow channels (data plane).
115    FlowList,
116
117    /// Cerrar el data plane de un pipeline (drop sockets + canales).
118    FlowDrop { pipeline: Ulid },
119}
120
121#[derive(Debug, Clone, Serialize, Deserialize)]
122pub enum Response {
123    Pong,
124
125    WorkspaceCreated {
126        id: WorkspaceId,
127        warnings: Vec<String>,
128    },
129
130    WorkspaceList {
131        items: Vec<WorkspaceSummary>,
132    },
133
134    WorkspaceStopped {
135        id: WorkspaceId,
136        reaped: u32,
137    },
138
139    RunStarted {
140        workspace: WorkspaceId,
141        command_id: Ulid,
142        pid: i32,
143    },
144
145    PipelineStarted {
146        pipeline: Ulid,
147        command_pids: Vec<(String, i32)>,
148        /// Discernments por edge cuando tap=true. Vacío sin tap.
149        edges: Vec<EdgeDiscernmentInfo>,
150    },
151
152    Discernment {
153        ty: String,
154        confidence: f32,
155        mime: Option<String>,
156        lens: Option<String>,
157    },
158
159    Capabilities {
160        kernel_version: (u32, u32, u32),
161        user_ns: String,
162        cgroup_v2: String,
163        cgroup_delegated: bool,
164        has_cap_sys_admin: bool,
165    },
166
167    CommandList {
168        items: Vec<CommandInfo>,
169    },
170
171    CommandLogs {
172        bytes: Vec<u8>,
173    },
174
175    PipelineSaved {
176        name: String,
177    },
178
179    PipelineSavedList {
180        names: Vec<String>,
181    },
182
183    PipelineDropped {
184        name: String,
185        existed: bool,
186    },
187
188    PipelineStopped {
189        pipeline: Ulid,
190        reaped: u32,
191    },
192
193    WorkspaceStats {
194        info: WorkspaceStatsInfo,
195    },
196
197    FlowList {
198        items: Vec<FlowInfo>,
199    },
200
201    FlowDropped {
202        pipeline: Ulid,
203        existed: bool,
204    },
205
206    Error {
207        message: String,
208    },
209}
210
211#[derive(Debug, Clone, Serialize, Deserialize)]
212pub struct WorkspaceStatsInfo {
213    pub commands_alive: u32,
214    pub commands_total: u32,
215    pub rss_bytes: Option<u64>,
216    #[serde(default)]
217    pub rss_peak_bytes: Option<u64>,
218    pub cpu_usec: Option<u64>,
219    pub source: String,
220    pub uptime_ms: u64,
221}
222
223#[derive(Debug, Clone, Serialize, Deserialize)]
224pub struct FlowInfo {
225    pub pipeline: Ulid,
226    pub sockets: Vec<PathBuf>,
227}
228
229#[derive(Debug, Clone, Serialize, Deserialize)]
230pub struct CommandInfo {
231    pub id: Ulid,
232    pub label: String,
233    pub pid: i32,
234    pub alive: bool,
235    pub exit_status: Option<i32>,
236    pub log_bytes: u64,
237}
238
239#[derive(Debug, Clone, Serialize, Deserialize)]
240pub struct EdgeDiscernmentInfo {
241    pub from_label: String,
242    pub from_output: String,
243    pub to_label: String,
244    pub to_input: String,
245    /// `Some(ty)` si el discerner detectó algo. `None` si no hubo data
246    /// suficiente o no matcheó ningún discerner.
247    pub ty: Option<String>,
248    pub mime: Option<String>,
249    pub lens: Option<String>,
250    pub confidence: f32,
251    /// Path del Unix socket donde otros módulos pueden suscribirse a los
252    /// bytes replicados de este edge (data plane). `None` si tap=false.
253    pub flow_socket: Option<PathBuf>,
254}
255
256#[derive(Debug, Clone, Serialize, Deserialize)]
257pub struct WorkspaceSummary {
258    pub id: WorkspaceId,
259    pub label: String,
260    pub commands: u32,
261    pub uptime_ms: u64,
262}
263
264// =====================================================================
265// Errores
266// =====================================================================
267
268#[derive(Debug, Error)]
269pub enum ProtocolError {
270    #[error("frame oversize: {0} bytes (max {MAX_FRAME})")]
271    FrameOversize(usize),
272    #[error("io: {0}")]
273    Io(#[from] std::io::Error),
274    #[error("postcard: {0}")]
275    Postcard(#[from] postcard::Error),
276    #[error("connection closed")]
277    Closed,
278}
279
280// =====================================================================
281// Framing helpers
282// =====================================================================
283
284pub async fn write_frame<T: Serialize>(stream: &mut UnixStream, msg: &T) -> Result<(), ProtocolError> {
285    let bytes = postcard::to_allocvec(msg)?;
286    if bytes.len() > MAX_FRAME {
287        return Err(ProtocolError::FrameOversize(bytes.len()));
288    }
289    let len = (bytes.len() as u32).to_be_bytes();
290    stream.write_all(&len).await?;
291    stream.write_all(&bytes).await?;
292    stream.flush().await?;
293    Ok(())
294}
295
296pub async fn read_frame<T: for<'de> Deserialize<'de>>(
297    stream: &mut UnixStream,
298) -> Result<T, ProtocolError> {
299    let mut len_buf = [0u8; 4];
300    stream.read_exact(&mut len_buf).await.map_err(|e| {
301        if e.kind() == std::io::ErrorKind::UnexpectedEof {
302            ProtocolError::Closed
303        } else {
304            ProtocolError::Io(e)
305        }
306    })?;
307    let len = u32::from_be_bytes(len_buf) as usize;
308    if len > MAX_FRAME {
309        return Err(ProtocolError::FrameOversize(len));
310    }
311    let mut buf = vec![0u8; len];
312    stream.read_exact(&mut buf).await?;
313    Ok(postcard::from_bytes(&buf)?)
314}
315
316/// Path canónico del socket del daemon: `$XDG_RUNTIME_DIR/shipote.sock`,
317/// fallback `/run/user/$UID/shipote.sock`, fallback `/tmp/shipote-$UID.sock`.
318pub fn default_socket_path() -> PathBuf {
319    if let Ok(xdg) = std::env::var("XDG_RUNTIME_DIR") {
320        return PathBuf::from(xdg).join(DEFAULT_SOCK_NAME);
321    }
322    let uid = nix::unistd::getuid().as_raw();
323    let p = PathBuf::from(format!("/run/user/{uid}"));
324    if p.exists() {
325        return p.join(DEFAULT_SOCK_NAME);
326    }
327    PathBuf::from(format!("/tmp/shipote-{uid}.sock"))
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333
334    #[test]
335    fn ping_roundtrip() {
336        let bytes = postcard::to_allocvec(&Request::Ping).unwrap();
337        let back: Request = postcard::from_bytes(&bytes).unwrap();
338        assert!(matches!(back, Request::Ping));
339    }
340
341    #[test]
342    fn workspace_create_roundtrip() {
343        let req = Request::WorkspaceCreate {
344            spec: WorkspaceSpec {
345                label: "demo".into(),
346                soma: Default::default(),
347                permissions: Default::default(),
348                ttl: None,
349                flow_dirs: vec![],
350                on_exit: shipote_card::ExitPolicy::Reap,
351            },
352        };
353        let bytes = postcard::to_allocvec(&req).unwrap();
354        let back: Request = postcard::from_bytes(&bytes).unwrap();
355        match back {
356            Request::WorkspaceCreate { spec } => assert_eq!(spec.label, "demo"),
357            _ => panic!("wrong variant"),
358        }
359    }
360
361    #[test]
362    fn default_socket_path_uses_runtime_dir() {
363        let p = default_socket_path();
364        assert!(p.to_string_lossy().ends_with("shipote.sock"));
365    }
366}