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
20// =====================================================================
21// Mensajes
22// =====================================================================
23
24#[derive(Debug, Clone, Serialize, Deserialize)]
25pub enum Request {
26    /// Health-check.
27    Ping,
28
29    /// Crear un workspace nuevo.
30    WorkspaceCreate { spec: WorkspaceSpec },
31
32    /// Listar todos los workspaces vivos.
33    WorkspaceList,
34
35    /// Detener un workspace y reapear sus comandos.
36    WorkspaceStop { id: WorkspaceId },
37
38    /// Ejecutar un comando one-shot dentro de un workspace existente.
39    Run {
40        workspace: WorkspaceId,
41        exec: String,
42        argv: Vec<String>,
43        envp: Vec<(String, String)>,
44    },
45
46    /// Lanzar un Pipeline completo dentro de un workspace.
47    PipelineRun {
48        spec: PipelineSpec,
49        /// Si `true`, el daemon interpone un tap entre productor y
50        /// consumidor de cada FlowEdge, sampleando los primeros bytes
51        /// y discerniendo el TypeRef.
52        tap: bool,
53    },
54
55    /// Discernir un buffer ad-hoc (sin workspace). Útil para `shipote discern <file>`.
56    Discern { sample: Vec<u8>, hint_path: Option<PathBuf> },
57
58    /// Capacidades runtime del kernel/proceso del daemon.
59    Capabilities,
60
61    /// Listar comandos vivos+pasados de un workspace.
62    CommandList { workspace: shipote_card::WorkspaceId },
63
64    /// Tail del log capturado para un comando.
65    CommandLogs {
66        workspace: shipote_card::WorkspaceId,
67        command: Ulid,
68        tail_bytes: usize,
69    },
70
71    /// Guardar (o reemplazar) un PipelineSpec bajo un nombre.
72    PipelineSave { name: String, spec: PipelineSpec },
73
74    /// Listar nombres de pipelines guardados.
75    PipelineSavedList,
76
77    /// Eliminar un pipeline guardado.
78    PipelineDrop { name: String },
79
80    /// Ejecutar un pipeline guardado.
81    PipelineRunSaved { name: String, tap: bool },
82}
83
84#[derive(Debug, Clone, Serialize, Deserialize)]
85pub enum Response {
86    Pong,
87
88    WorkspaceCreated {
89        id: WorkspaceId,
90        warnings: Vec<String>,
91    },
92
93    WorkspaceList {
94        items: Vec<WorkspaceSummary>,
95    },
96
97    WorkspaceStopped {
98        id: WorkspaceId,
99        reaped: u32,
100    },
101
102    RunStarted {
103        workspace: WorkspaceId,
104        command_id: Ulid,
105        pid: i32,
106    },
107
108    PipelineStarted {
109        pipeline: Ulid,
110        command_pids: Vec<(String, i32)>,
111        /// Discernments por edge cuando tap=true. Vacío sin tap.
112        edges: Vec<EdgeDiscernmentInfo>,
113    },
114
115    Discernment {
116        ty: String,
117        confidence: f32,
118        mime: Option<String>,
119        lens: Option<String>,
120    },
121
122    Capabilities {
123        kernel_version: (u32, u32, u32),
124        user_ns: String,
125        cgroup_v2: String,
126        cgroup_delegated: bool,
127        has_cap_sys_admin: bool,
128    },
129
130    CommandList {
131        items: Vec<CommandInfo>,
132    },
133
134    CommandLogs {
135        bytes: Vec<u8>,
136    },
137
138    PipelineSaved {
139        name: String,
140    },
141
142    PipelineSavedList {
143        names: Vec<String>,
144    },
145
146    PipelineDropped {
147        name: String,
148        existed: bool,
149    },
150
151    Error {
152        message: String,
153    },
154}
155
156#[derive(Debug, Clone, Serialize, Deserialize)]
157pub struct CommandInfo {
158    pub id: Ulid,
159    pub label: String,
160    pub pid: i32,
161    pub alive: bool,
162    pub exit_status: Option<i32>,
163    pub log_bytes: u64,
164}
165
166#[derive(Debug, Clone, Serialize, Deserialize)]
167pub struct EdgeDiscernmentInfo {
168    pub from_label: String,
169    pub from_output: String,
170    pub to_label: String,
171    pub to_input: String,
172    /// `Some(ty)` si el discerner detectó algo. `None` si no hubo data
173    /// suficiente o no matcheó ningún discerner.
174    pub ty: Option<String>,
175    pub mime: Option<String>,
176    pub lens: Option<String>,
177    pub confidence: f32,
178}
179
180#[derive(Debug, Clone, Serialize, Deserialize)]
181pub struct WorkspaceSummary {
182    pub id: WorkspaceId,
183    pub label: String,
184    pub commands: u32,
185    pub uptime_ms: u64,
186}
187
188// =====================================================================
189// Errores
190// =====================================================================
191
192#[derive(Debug, Error)]
193pub enum ProtocolError {
194    #[error("frame oversize: {0} bytes (max {MAX_FRAME})")]
195    FrameOversize(usize),
196    #[error("io: {0}")]
197    Io(#[from] std::io::Error),
198    #[error("postcard: {0}")]
199    Postcard(#[from] postcard::Error),
200    #[error("connection closed")]
201    Closed,
202}
203
204// =====================================================================
205// Framing helpers
206// =====================================================================
207
208pub async fn write_frame<T: Serialize>(stream: &mut UnixStream, msg: &T) -> Result<(), ProtocolError> {
209    let bytes = postcard::to_allocvec(msg)?;
210    if bytes.len() > MAX_FRAME {
211        return Err(ProtocolError::FrameOversize(bytes.len()));
212    }
213    let len = (bytes.len() as u32).to_be_bytes();
214    stream.write_all(&len).await?;
215    stream.write_all(&bytes).await?;
216    stream.flush().await?;
217    Ok(())
218}
219
220pub async fn read_frame<T: for<'de> Deserialize<'de>>(
221    stream: &mut UnixStream,
222) -> Result<T, ProtocolError> {
223    let mut len_buf = [0u8; 4];
224    stream.read_exact(&mut len_buf).await.map_err(|e| {
225        if e.kind() == std::io::ErrorKind::UnexpectedEof {
226            ProtocolError::Closed
227        } else {
228            ProtocolError::Io(e)
229        }
230    })?;
231    let len = u32::from_be_bytes(len_buf) as usize;
232    if len > MAX_FRAME {
233        return Err(ProtocolError::FrameOversize(len));
234    }
235    let mut buf = vec![0u8; len];
236    stream.read_exact(&mut buf).await?;
237    Ok(postcard::from_bytes(&buf)?)
238}
239
240/// Path canónico del socket del daemon: `$XDG_RUNTIME_DIR/shipote.sock`,
241/// fallback `/run/user/$UID/shipote.sock`, fallback `/tmp/shipote-$UID.sock`.
242pub fn default_socket_path() -> PathBuf {
243    if let Ok(xdg) = std::env::var("XDG_RUNTIME_DIR") {
244        return PathBuf::from(xdg).join(DEFAULT_SOCK_NAME);
245    }
246    let uid = nix::unistd::getuid().as_raw();
247    let p = PathBuf::from(format!("/run/user/{uid}"));
248    if p.exists() {
249        return p.join(DEFAULT_SOCK_NAME);
250    }
251    PathBuf::from(format!("/tmp/shipote-{uid}.sock"))
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257
258    #[test]
259    fn ping_roundtrip() {
260        let bytes = postcard::to_allocvec(&Request::Ping).unwrap();
261        let back: Request = postcard::from_bytes(&bytes).unwrap();
262        assert!(matches!(back, Request::Ping));
263    }
264
265    #[test]
266    fn workspace_create_roundtrip() {
267        let req = Request::WorkspaceCreate {
268            spec: WorkspaceSpec {
269                label: "demo".into(),
270                soma: Default::default(),
271                permissions: Default::default(),
272                ttl: None,
273                flow_dirs: vec![],
274                on_exit: shipote_card::ExitPolicy::Reap,
275            },
276        };
277        let bytes = postcard::to_allocvec(&req).unwrap();
278        let back: Request = postcard::from_bytes(&bytes).unwrap();
279        match back {
280            Request::WorkspaceCreate { spec } => assert_eq!(spec.label, "demo"),
281            _ => panic!("wrong variant"),
282        }
283    }
284
285    #[test]
286    fn default_socket_path_uses_runtime_dir() {
287        let p = default_socket_path();
288        assert!(p.to_string_lossy().ends_with("shipote.sock"));
289    }
290}