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    #[serde(default)]
220    pub cpu_percent: Option<f32>,
221    pub source: String,
222    pub uptime_ms: u64,
223}
224
225#[derive(Debug, Clone, Serialize, Deserialize)]
226pub struct FlowInfo {
227    pub pipeline: Ulid,
228    pub sockets: Vec<PathBuf>,
229}
230
231#[derive(Debug, Clone, Serialize, Deserialize)]
232pub struct CommandInfo {
233    pub id: Ulid,
234    pub label: String,
235    pub pid: i32,
236    pub alive: bool,
237    pub exit_status: Option<i32>,
238    pub log_bytes: u64,
239}
240
241#[derive(Debug, Clone, Serialize, Deserialize)]
242pub struct EdgeDiscernmentInfo {
243    pub from_label: String,
244    pub from_output: String,
245    pub to_label: String,
246    pub to_input: String,
247    /// `Some(ty)` si el discerner detectó algo. `None` si no hubo data
248    /// suficiente o no matcheó ningún discerner.
249    pub ty: Option<String>,
250    pub mime: Option<String>,
251    pub lens: Option<String>,
252    pub confidence: f32,
253    /// Path del Unix socket donde otros módulos pueden suscribirse a los
254    /// bytes replicados de este edge (data plane). `None` si tap=false.
255    pub flow_socket: Option<PathBuf>,
256}
257
258#[derive(Debug, Clone, Serialize, Deserialize)]
259pub struct WorkspaceSummary {
260    pub id: WorkspaceId,
261    pub label: String,
262    pub commands: u32,
263    pub uptime_ms: u64,
264}
265
266// =====================================================================
267// Errores
268// =====================================================================
269
270#[derive(Debug, Error)]
271pub enum ProtocolError {
272    #[error("frame oversize: {0} bytes (max {MAX_FRAME})")]
273    FrameOversize(usize),
274    #[error("io: {0}")]
275    Io(#[from] std::io::Error),
276    #[error("postcard: {0}")]
277    Postcard(#[from] postcard::Error),
278    #[error("connection closed")]
279    Closed,
280}
281
282// =====================================================================
283// Framing helpers
284// =====================================================================
285
286pub async fn write_frame<T: Serialize>(stream: &mut UnixStream, msg: &T) -> Result<(), ProtocolError> {
287    let bytes = postcard::to_allocvec(msg)?;
288    if bytes.len() > MAX_FRAME {
289        return Err(ProtocolError::FrameOversize(bytes.len()));
290    }
291    let len = (bytes.len() as u32).to_be_bytes();
292    stream.write_all(&len).await?;
293    stream.write_all(&bytes).await?;
294    stream.flush().await?;
295    Ok(())
296}
297
298pub async fn read_frame<T: for<'de> Deserialize<'de>>(
299    stream: &mut UnixStream,
300) -> Result<T, ProtocolError> {
301    let mut len_buf = [0u8; 4];
302    stream.read_exact(&mut len_buf).await.map_err(|e| {
303        if e.kind() == std::io::ErrorKind::UnexpectedEof {
304            ProtocolError::Closed
305        } else {
306            ProtocolError::Io(e)
307        }
308    })?;
309    let len = u32::from_be_bytes(len_buf) as usize;
310    if len > MAX_FRAME {
311        return Err(ProtocolError::FrameOversize(len));
312    }
313    let mut buf = vec![0u8; len];
314    stream.read_exact(&mut buf).await?;
315    Ok(postcard::from_bytes(&buf)?)
316}
317
318/// Path canónico del socket del daemon: `$XDG_RUNTIME_DIR/shipote.sock`,
319/// fallback `/run/user/$UID/shipote.sock`, fallback `/tmp/shipote-$UID.sock`.
320pub fn default_socket_path() -> PathBuf {
321    if let Ok(xdg) = std::env::var("XDG_RUNTIME_DIR") {
322        return PathBuf::from(xdg).join(DEFAULT_SOCK_NAME);
323    }
324    let uid = nix::unistd::getuid().as_raw();
325    let p = PathBuf::from(format!("/run/user/{uid}"));
326    if p.exists() {
327        return p.join(DEFAULT_SOCK_NAME);
328    }
329    PathBuf::from(format!("/tmp/shipote-{uid}.sock"))
330}
331
332#[cfg(test)]
333mod tests {
334    use super::*;
335
336    #[test]
337    fn ping_roundtrip() {
338        let bytes = postcard::to_allocvec(&Request::Ping).unwrap();
339        let back: Request = postcard::from_bytes(&bytes).unwrap();
340        assert!(matches!(back, Request::Ping));
341    }
342
343    #[test]
344    fn workspace_create_roundtrip() {
345        let req = Request::WorkspaceCreate {
346            spec: WorkspaceSpec {
347                label: "demo".into(),
348                soma: Default::default(),
349                permissions: Default::default(),
350                ttl: None,
351                flow_dirs: vec![],
352                on_exit: shipote_card::ExitPolicy::Reap,
353            },
354        };
355        let bytes = postcard::to_allocvec(&req).unwrap();
356        let back: Request = postcard::from_bytes(&bytes).unwrap();
357        match back {
358            Request::WorkspaceCreate { spec } => assert_eq!(spec.label, "demo"),
359            _ => panic!("wrong variant"),
360        }
361    }
362
363    #[test]
364    fn default_socket_path_uses_runtime_dir() {
365        let p = default_socket_path();
366        assert!(p.to_string_lossy().ends_with("shipote.sock"));
367    }
368}