⎇ 72758e75ce515f355f27615671c95ecaf8c2a8bc
crates/modules/shipote/shipote-protocol/src/lib.rs
Historia
Quién tocó qué
crudo
feat(shipote): health endpoint + audit log + token-bucket real (fase R)
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 /// Health endpoint estructurado: versión + uptime + counts.
34 Health,
35
36 /// Crear un workspace nuevo.
37 WorkspaceCreate { spec: WorkspaceSpec },
38
39 /// Listar todos los workspaces vivos.
40 WorkspaceList,
41
42 /// Detener un workspace y reapear sus comandos. `grace_ms`: tiempo
43 /// que se espera tras SIGTERM antes de SIGKILL. 0 = SIGKILL inmediato.
44 WorkspaceStop {
45 id: WorkspaceId,
46 #[serde(default = "default_grace_ms")]
47 grace_ms: u64,
48 },
49
50 /// Ejecutar un comando one-shot dentro de un workspace existente.
51 Run {
52 workspace: WorkspaceId,
53 exec: String,
54 argv: Vec<String>,
55 envp: Vec<(String, String)>,
56 /// Si `true` y el comando muere con exit_status != 0, el reaper
57 /// lo relaunch con backoff exponencial.
58 #[serde(default)]
59 restart_on_failure: bool,
60 },
61
62 /// Lanzar un Pipeline completo dentro de un workspace.
63 PipelineRun {
64 spec: PipelineSpec,
65 /// Si `true`, el daemon interpone un tap entre productor y
66 /// consumidor de cada FlowEdge, sampleando los primeros bytes
67 /// y discerniendo el TypeRef.
68 tap: bool,
69 /// Variables para sustitución `${KEY}` en strings del spec
70 /// antes de spawn (templating).
71 #[serde(default)]
72 vars: std::collections::BTreeMap<String, String>,
73 },
74
75 /// Discernir un buffer ad-hoc (sin workspace). Útil para `shipote discern <file>`.
76 Discern { sample: Vec<u8>, hint_path: Option<PathBuf> },
77
78 /// Capacidades runtime del kernel/proceso del daemon.
79 Capabilities,
80
81 /// Listar comandos vivos+pasados de un workspace.
82 CommandList { workspace: shipote_card::WorkspaceId },
83
84 /// Tail del log capturado para un comando.
85 CommandLogs {
86 workspace: shipote_card::WorkspaceId,
87 command: Ulid,
88 tail_bytes: usize,
89 /// "stdout" | "stderr" | "both" (default "both" si vacío).
90 stream: String,
91 },
92
93 /// Guardar (o reemplazar) un PipelineSpec bajo un nombre.
94 PipelineSave { name: String, spec: PipelineSpec },
95
96 /// Listar nombres de pipelines guardados.
97 PipelineSavedList,
98
99 /// Eliminar un pipeline guardado.
100 PipelineDrop { name: String },
101
102 /// Ejecutar un pipeline guardado.
103 PipelineRunSaved {
104 name: String,
105 tap: bool,
106 #[serde(default)]
107 vars: std::collections::BTreeMap<String, String>,
108 },
109
110 /// Resource accounting de un workspace.
111 WorkspaceStats { workspace: shipote_card::WorkspaceId },
112
113 /// Reporte de quotas (rlimits declarados vs uso actual).
114 WorkspaceQuota { workspace: shipote_card::WorkspaceId },
115
116 /// History de samples del workspace (server-side). Sobrevive
117 /// restart del shell. `tail`: cantidad de samples desde el final
118 /// (0 = todo).
119 WorkspaceStatsHistory {
120 workspace: shipote_card::WorkspaceId,
121 tail: usize,
122 },
123
124 /// Resumen completo de un workspace: stats + quota + commands +
125 /// flow sockets en una sola roundtrip. Reduce N×4 requests del
126 /// shell a N×1.
127 WorkspaceFullSummary { workspace: shipote_card::WorkspaceId },
128
129 /// Detener selectivamente los comandos de un pipeline (no el workspace
130 /// entero). `grace_ms`: SIGTERM → wait → SIGKILL.
131 PipelineStop {
132 pipeline: Ulid,
133 #[serde(default = "default_grace_ms")]
134 grace_ms: u64,
135 },
136
137 /// Listar pipelines activos con sus flow channels (data plane).
138 FlowList,
139
140 /// Throughput por flow socket: bytes_total + bytes_per_sec.
141 FlowThroughput,
142
143 /// Cerrar el data plane de un pipeline (drop sockets + canales).
144 FlowDrop { pipeline: Ulid },
145}
146
147#[derive(Debug, Clone, Serialize, Deserialize)]
148pub enum Response {
149 Pong,
150
151 Health {
152 version: String,
153 uptime_ms: u64,
154 alive_workspaces: u32,
155 alive_commands: u32,
156 alive_pipelines: u32,
157 active_flows: u32,
158 dirty: bool,
159 },
160
161 WorkspaceCreated {
162 id: WorkspaceId,
163 warnings: Vec<String>,
164 },
165
166 WorkspaceList {
167 items: Vec<WorkspaceSummary>,
168 },
169
170 WorkspaceStopped {
171 id: WorkspaceId,
172 reaped: u32,
173 },
174
175 RunStarted {
176 workspace: WorkspaceId,
177 command_id: Ulid,
178 pid: i32,
179 },
180
181 PipelineStarted {
182 pipeline: Ulid,
183 command_pids: Vec<(String, i32)>,
184 /// Discernments por edge cuando tap=true. Vacío sin tap.
185 edges: Vec<EdgeDiscernmentInfo>,
186 },
187
188 Discernment {
189 ty: String,
190 confidence: f32,
191 mime: Option<String>,
192 lens: Option<String>,
193 },
194
195 Capabilities {
196 kernel_version: (u32, u32, u32),
197 user_ns: String,
198 cgroup_v2: String,
199 cgroup_delegated: bool,
200 has_cap_sys_admin: bool,
201 },
202
203 CommandList {
204 items: Vec<CommandInfo>,
205 },
206
207 CommandLogs {
208 bytes: Vec<u8>,
209 },
210
211 PipelineSaved {
212 name: String,
213 },
214
215 PipelineSavedList {
216 names: Vec<String>,
217 },
218
219 PipelineDropped {
220 name: String,
221 existed: bool,
222 },
223
224 PipelineStopped {
225 pipeline: Ulid,
226 reaped: u32,
227 },
228
229 WorkspaceStats {
230 info: WorkspaceStatsInfo,
231 },
232
233 WorkspaceQuota {
234 info: QuotaReportInfo,
235 },
236
237 WorkspaceStatsHistory {
238 samples: Vec<WorkspaceStatsInfo>,
239 },
240
241 WorkspaceFullSummary {
242 stats: WorkspaceStatsInfo,
243 quota: QuotaReportInfo,
244 commands: Vec<CommandInfo>,
245 flow_sockets: Vec<PathBuf>,
246 },
247
248 FlowList {
249 items: Vec<FlowInfo>,
250 },
251
252 FlowThroughput {
253 items: Vec<FlowThroughputInfo>,
254 },
255
256 FlowDropped {
257 pipeline: Ulid,
258 existed: bool,
259 },
260
261 Error {
262 message: String,
263 },
264}
265
266#[derive(Debug, Clone, Serialize, Deserialize)]
267pub struct QuotaReportInfo {
268 pub mem_limit: Option<u64>,
269 pub nproc_limit: Option<u32>,
270 pub breaches: Vec<String>,
271}
272
273#[derive(Debug, Clone, Serialize, Deserialize)]
274pub struct WorkspaceStatsInfo {
275 pub commands_alive: u32,
276 pub commands_total: u32,
277 pub rss_bytes: Option<u64>,
278 #[serde(default)]
279 pub rss_peak_bytes: Option<u64>,
280 pub cpu_usec: Option<u64>,
281 #[serde(default)]
282 pub cpu_percent: Option<f32>,
283 #[serde(default = "default_cpu_cores")]
284 pub cpu_cores: u32,
285 pub source: String,
286 pub uptime_ms: u64,
287}
288
289fn default_cpu_cores() -> u32 {
290 1
291}
292
293#[derive(Debug, Clone, Serialize, Deserialize)]
294pub struct FlowThroughputInfo {
295 pub socket: PathBuf,
296 pub bytes_total: u64,
297 pub bytes_per_sec: f64,
298}
299
300#[derive(Debug, Clone, Serialize, Deserialize)]
301pub struct FlowInfo {
302 pub pipeline: Ulid,
303 pub sockets: Vec<PathBuf>,
304}
305
306#[derive(Debug, Clone, Serialize, Deserialize)]
307pub struct CommandInfo {
308 pub id: Ulid,
309 pub label: String,
310 pub pid: i32,
311 pub alive: bool,
312 pub exit_status: Option<i32>,
313 pub log_bytes: u64,
314}
315
316#[derive(Debug, Clone, Serialize, Deserialize)]
317pub struct EdgeDiscernmentInfo {
318 pub from_label: String,
319 pub from_output: String,
320 pub to_label: String,
321 pub to_input: String,
322 /// `Some(ty)` si el discerner detectó algo. `None` si no hubo data
323 /// suficiente o no matcheó ningún discerner.
324 pub ty: Option<String>,
325 pub mime: Option<String>,
326 pub lens: Option<String>,
327 pub confidence: f32,
328 /// Path del Unix socket donde otros módulos pueden suscribirse a los
329 /// bytes replicados de este edge (data plane). `None` si tap=false.
330 pub flow_socket: Option<PathBuf>,
331}
332
333#[derive(Debug, Clone, Serialize, Deserialize)]
334pub struct WorkspaceSummary {
335 pub id: WorkspaceId,
336 pub label: String,
337 pub commands: u32,
338 pub uptime_ms: u64,
339}
340
341// =====================================================================
342// Errores
343// =====================================================================
344
345#[derive(Debug, Error)]
346pub enum ProtocolError {
347 #[error("frame oversize: {0} bytes (max {MAX_FRAME})")]
348 FrameOversize(usize),
349 #[error("io: {0}")]
350 Io(#[from] std::io::Error),
351 #[error("postcard: {0}")]
352 Postcard(#[from] postcard::Error),
353 #[error("connection closed")]
354 Closed,
355}
356
357// =====================================================================
358// Framing helpers
359// =====================================================================
360
361pub async fn write_frame<T: Serialize>(stream: &mut UnixStream, msg: &T) -> Result<(), ProtocolError> {
362 let bytes = postcard::to_allocvec(msg)?;
363 if bytes.len() > MAX_FRAME {
364 return Err(ProtocolError::FrameOversize(bytes.len()));
365 }
366 let len = (bytes.len() as u32).to_be_bytes();
367 stream.write_all(&len).await?;
368 stream.write_all(&bytes).await?;
369 stream.flush().await?;
370 Ok(())
371}
372
373pub async fn read_frame<T: for<'de> Deserialize<'de>>(
374 stream: &mut UnixStream,
375) -> Result<T, ProtocolError> {
376 let mut len_buf = [0u8; 4];
377 stream.read_exact(&mut len_buf).await.map_err(|e| {
378 if e.kind() == std::io::ErrorKind::UnexpectedEof {
379 ProtocolError::Closed
380 } else {
381 ProtocolError::Io(e)
382 }
383 })?;
384 let len = u32::from_be_bytes(len_buf) as usize;
385 if len > MAX_FRAME {
386 return Err(ProtocolError::FrameOversize(len));
387 }
388 let mut buf = vec![0u8; len];
389 stream.read_exact(&mut buf).await?;
390 Ok(postcard::from_bytes(&buf)?)
391}
392
393/// Path canónico del socket del daemon: `$XDG_RUNTIME_DIR/shipote.sock`,
394/// fallback `/run/user/$UID/shipote.sock`, fallback `/tmp/shipote-$UID.sock`.
395pub fn default_socket_path() -> PathBuf {
396 if let Ok(xdg) = std::env::var("XDG_RUNTIME_DIR") {
397 return PathBuf::from(xdg).join(DEFAULT_SOCK_NAME);
398 }
399 let uid = nix::unistd::getuid().as_raw();
400 let p = PathBuf::from(format!("/run/user/{uid}"));
401 if p.exists() {
402 return p.join(DEFAULT_SOCK_NAME);
403 }
404 PathBuf::from(format!("/tmp/shipote-{uid}.sock"))
405}
406
407#[cfg(test)]
408mod tests {
409 use super::*;
410
411 #[test]
412 fn ping_roundtrip() {
413 let bytes = postcard::to_allocvec(&Request::Ping).unwrap();
414 let back: Request = postcard::from_bytes(&bytes).unwrap();
415 assert!(matches!(back, Request::Ping));
416 }
417
418 #[test]
419 fn workspace_create_roundtrip() {
420 let req = Request::WorkspaceCreate {
421 spec: WorkspaceSpec {
422 label: "demo".into(),
423 soma: Default::default(),
424 permissions: Default::default(),
425 ttl: None,
426 flow_dirs: vec![],
427 on_exit: shipote_card::ExitPolicy::Reap,
428 quota_enforce: Default::default(),
429 },
430 };
431 let bytes = postcard::to_allocvec(&req).unwrap();
432 let back: Request = postcard::from_bytes(&bytes).unwrap();
433 match back {
434 Request::WorkspaceCreate { spec } => assert_eq!(spec.label, "demo"),
435 _ => panic!("wrong variant"),
436 }
437 }
438
439 #[test]
440 fn default_socket_path_uses_runtime_dir() {
441 let p = default_socket_path();
442 assert!(p.to_string_lossy().ends_with("shipote.sock"));
443 }
444}