sergio/takana
> Una distribución Linux construida con un laboratorio funcional y hermético en el > sótano, y una terminal mutable, clásica y anárquica en el piso de arriba — diseñada > desde el suelo para que una IA programadora entienda, modifique y co…
Fase 6 — lenguaje de consulta del sistema (SDD 08 §6)
1//! Bus de agente: `/run/agent.sock`. Ver `docs/07-agent-bus.md`.
2//!
3//! Modelo de threading (síncrono, una conexión = N threads pequeños):
4//!
5//! ```text
6//! [accept loop] ── spawn ──▶ conn{
7//! [reader thread] ── dispatch ──┐
8//! [bus forwarder] ─────────────┼──▶ tx ──▶ [writer thread] ──▶ socket
9//! [compile worker] (puntual) ───┘
10//! }
11//! ```
12//!
13//! No usamos `tokio` para no arrastrar runtime asíncrono: la cadencia esperada del bus es
14//! humana + agente (decenas de mensajes/seg en el peor caso). Cada conexión tiene su propio
15//! canal `Sender<Event>`; la `EventBus` global re-emite Modified/Crashed a esos canales.
16
17use std::io::{BufRead, BufReader, Write};
18use std::os::fd::AsFd;
19use std::os::unix::net::{UnixListener, UnixStream};
20use std::path::{Path, PathBuf};
21use std::sync::mpsc::{self, Sender};
22use std::sync::{Arc, Mutex};
23use std::thread;
24
25use nix::sys::socket::{getsockopt, sockopt::PeerCredentials};
26
27use crate::events::EventBus;
28use hammer_core::proto::{Cap, Command, Event, Peer, RecipeInline, PROTOCOL_VERSION};
29
30/// Política de capacidades: dado el `Peer` (ya autenticado por SO_PEERCRED), devuelve qué
31/// caps tiene esa conexión. Por defecto: misma UID que el daemon ⇒ todo menos `inject-real`;
32/// otros UIDs ⇒ sólo `query`. La policy más expresiva (lectura de un toml en `/etc/hammer`)
33/// es trabajo posterior; la firma del callable basta para encajarla sin tocar este módulo.
34pub type CapsPolicy = Arc<dyn Fn(&Peer) -> Vec<Cap> + Send + Sync>;
35
36pub fn default_policy() -> CapsPolicy {
37 let my_uid = unsafe { libc::getuid() };
38 Arc::new(move |peer: &Peer| {
39 if peer.uid == my_uid {
40 vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init]
41 } else {
42 vec![Cap::Query]
43 }
44 })
45}
46
47/// Contexto que necesita el dispatcher para resolver comandos. Compartido por todas las
48/// conexiones (clones baratos: paths + Arc).
49#[derive(Clone)]
50pub struct BusContext {
51 pub store_root: PathBuf,
52 pub init_control: PathBuf,
53 pub events: EventBus,
54}
55
56#[derive(Debug, thiserror::Error)]
57pub enum Error {
58 #[error("io: {0}")]
59 Io(#[from] std::io::Error),
60 #[error("bind {0}: {1}")]
61 Bind(PathBuf, std::io::Error),
62 #[error("peer creds: {0}")]
63 Peer(String),
64}
65
66pub type Result<T> = std::result::Result<T, Error>;
67
68/// Arranca el listener bloqueante. Pensado para correr en su propio thread desde `main`.
69pub fn serve_agent_bus(
70 sock_path: &Path,
71 policy: CapsPolicy,
72 ctx: BusContext,
73) -> Result<()> {
74 // Limpia un socket olvidado de una ejecución previa. Si el path está ocupado por algo
75 // que no es un socket, dejamos que `bind` falle — no queremos rm-rf por accidente.
76 let _ = std::fs::remove_file(sock_path);
77 let listener = UnixListener::bind(sock_path)
78 .map_err(|e| Error::Bind(sock_path.to_path_buf(), e))?;
79 tracing::info!(sock = %sock_path.display(), "bus: escuchando");
80
81 for stream in listener.incoming() {
82 match stream {
83 Ok(s) => {
84 let policy = policy.clone();
85 let ctx = ctx.clone();
86 thread::Builder::new()
87 .name("bus-conn".into())
88 .spawn(move || {
89 if let Err(e) = handle_connection(s, policy, ctx) {
90 tracing::warn!(error = %e, "bus: conexión terminó con error");
91 }
92 })
93 .ok();
94 }
95 Err(e) => tracing::warn!(error = %e, "bus: accept falló"),
96 }
97 }
98 Ok(())
99}
100
101fn read_peer(stream: &UnixStream) -> Result<Peer> {
102 let cred = getsockopt(&stream.as_fd(), PeerCredentials)
103 .map_err(|e| Error::Peer(e.to_string()))?;
104 Ok(Peer {
105 uid: cred.uid(),
106 gid: cred.gid(),
107 pid: cred.pid(),
108 })
109}
110
111fn handle_connection(stream: UnixStream, policy: CapsPolicy, ctx: BusContext) -> Result<()> {
112 let peer = read_peer(&stream)?;
113 let caps = policy(&peer);
114 tracing::info!(?peer, ?caps, "bus: nueva conexión");
115
116 // Canal de eventos hacia el writer. Todos los productores (dispatcher, bus forwarder,
117 // workers de COMPILE) clonan este Sender.
118 let (tx, rx) = mpsc::channel::<Event>();
119
120 // Writer thread: serializa todo lo que recibe del canal y lo escribe al socket.
121 let stream_writer = stream.try_clone()?;
122 let _writer_handle = thread::Builder::new()
123 .name("bus-conn-writer".into())
124 .spawn(move || writer_loop(stream_writer, rx))
125 .ok();
126
127 // Bus forwarder thread: cualquier evento global (Modified/Crashed) se re-emite a este
128 // peer. Se cierra solo cuando el global drop el último Sender (en la práctica, nunca)
129 // o cuando nuestro tx local cae (peer cerró).
130 let bus_rx = ctx.events.subscribe();
131 let bus_tx = tx.clone();
132 thread::Builder::new()
133 .name("bus-conn-fwd".into())
134 .spawn(move || {
135 for ev in bus_rx {
136 if bus_tx.send(ev).is_err() {
137 break;
138 }
139 }
140 })
141 .ok();
142
143 // Handshake. La primera línea debe ser Hello.
144 let mut reader = BufReader::new(stream);
145 let mut first = String::new();
146 reader.read_line(&mut first)?;
147 match serde_json::from_str::<Command>(first.trim()) {
148 Ok(Command::Hello { ver, client }) => {
149 tracing::info!(ver, client = %client, "bus: hello");
150 let _ = tx.send(Event::Welcome {
151 ver: PROTOCOL_VERSION,
152 caps: caps.clone(),
153 peer: peer.clone(),
154 });
155 }
156 Ok(other) => {
157 let _ = tx.send(Event::Error {
158 code: "no_hello".into(),
159 msg: format!("se esperaba 'hello', llegó '{}'", t_of(&other)),
160 });
161 return Ok(());
162 }
163 Err(e) => {
164 let _ = tx.send(Event::Error {
165 code: "bad_handshake".into(),
166 msg: format!("hello inválido: {e}"),
167 });
168 return Ok(());
169 }
170 }
171
172 // Read loop. Cada línea es un Command. La conexión se cierra cuando se cierra EOF, o
173 // cuando un error de IO lo fuerza.
174 let caps_set: std::collections::BTreeSet<Cap> = caps.iter().copied().collect();
175 let caps_set = Arc::new(caps_set);
176 for line in reader.lines() {
177 let line = match line {
178 Ok(l) => l,
179 Err(_) => break,
180 };
181 if line.trim().is_empty() {
182 continue;
183 }
184 let cmd: Command = match serde_json::from_str(&line) {
185 Ok(c) => c,
186 Err(e) => {
187 let _ = tx.send(Event::Error {
188 code: "bad_command".into(),
189 msg: format!("JSON inválido: {e}"),
190 });
191 continue;
192 }
193 };
194 dispatch(cmd, &tx, &caps_set, &ctx);
195 }
196 Ok(())
197}
198
199fn writer_loop(mut stream: UnixStream, rx: mpsc::Receiver<Event>) {
200 for ev in rx {
201 let line = match serde_json::to_string(&ev) {
202 Ok(s) => s,
203 Err(e) => {
204 tracing::error!(error = %e, "bus: serializando event");
205 continue;
206 }
207 };
208 if stream.write_all(line.as_bytes()).is_err() {
209 break;
210 }
211 if stream.write_all(b"\n").is_err() {
212 break;
213 }
214 // El bus es de baja cadencia; flush por mensaje no es relevante en costes y mejora
215 // mucho la latencia percibida del cliente.
216 let _ = stream.flush();
217 }
218}
219
220fn dispatch(
221 cmd: Command,
222 tx: &Sender<Event>,
223 caps: &Arc<std::collections::BTreeSet<Cap>>,
224 ctx: &BusContext,
225) {
226 if let Some(needed) = Cap::required_for(&cmd) {
227 if !caps.contains(&needed) {
228 let _ = tx.send(Event::Error {
229 code: "no_cap".into(),
230 msg: format!("comando requiere capacidad '{needed:?}', no concedida"),
231 });
232 return;
233 }
234 }
235 match cmd {
236 Command::Hello { .. } => {
237 // Doble hello: lo ignoramos con un Error suave.
238 let _ = tx.send(Event::Error {
239 code: "duplicate_hello".into(),
240 msg: "ya hicimos hello en esta conexión".into(),
241 });
242 }
243 Command::Compile { recipe } => {
244 // Build pesado: a un worker. El read-loop sigue libre para más comandos.
245 let tx = tx.clone();
246 let store_root = ctx.store_root.clone();
247 thread::Builder::new()
248 .name("bus-compile".into())
249 .spawn(move || run_compile(recipe, store_root, tx))
250 .ok();
251 }
252 Command::Inject { artifact, target, overlay: _ } => {
253 let store = match hammer_core::Store::open(&ctx.store_root) {
254 Ok(s) => s,
255 Err(e) => {
256 let _ = tx.send(Event::Error {
257 code: "store".into(),
258 msg: e.to_string(),
259 });
260 return;
261 }
262 };
263 let artifact_dir = match store.find_by_hash(&artifact) {
264 Ok(p) => p,
265 Err(e) => {
266 let _ = tx.send(Event::Error {
267 code: "not_found".into(),
268 msg: e.to_string(),
269 });
270 return;
271 }
272 };
273 let report = match hammer_build::run_hydrate(
274 &artifact_dir,
275 Path::new(&target),
276 hammer_core::LinkMode::Static,
277 ) {
278 Ok(r) => r,
279 Err(e) => {
280 let _ = tx.send(Event::Error {
281 code: "hydrate_failed".into(),
282 msg: e.to_string(),
283 });
284 return;
285 }
286 };
287 let _ = tx.send(Event::Injected {
288 artifact,
289 target,
290 files: report.files.len(),
291 });
292 }
293 Command::Query { what, path, name, expr } => {
294 let value = match what.as_str() {
295 "file" => {
296 let Some(p) = path else {
297 let _ = tx.send(Event::Error {
298 code: "bad_query".into(),
299 msg: "query 'file' requiere 'path'".into(),
300 });
301 return;
302 };
303 query_file(&p)
304 }
305 "artifact" => {
306 let Some(h) = name.as_deref().or(path.as_deref()) else {
307 let _ = tx.send(Event::Error {
308 code: "bad_query".into(),
309 msg: "query 'artifact' requiere 'name' (el hash)".into(),
310 });
311 return;
312 };
313 query_artifact(&ctx.store_root, h)
314 }
315 "expr" => {
316 let Some(e) = expr.as_deref() else {
317 let _ = tx.send(Event::Error {
318 code: "bad_query".into(),
319 msg: "query 'expr' requiere 'expr' (la expresión)".into(),
320 });
321 return;
322 };
323 // El daemon evalúa con el `fs_root` del sistema host (None ⇒ rutas
324 // absolutas tal y como vienen). Si el .swm necesitara otra base, el
325 // cliente puede evaluar localmente con un EvalContext distinto.
326 hammer_core::query::eval_str(e, &hammer_core::query::EvalContext::new())
327 }
328 other => {
329 let _ = tx.send(Event::Error {
330 code: "unknown_query".into(),
331 msg: format!("'what'='{other}' no soportado"),
332 });
333 return;
334 }
335 };
336 let _ = tx.send(Event::QueryResult { what, value });
337 }
338 Command::Init { cmd } => {
339 match crate::control::send_to_fifo(&ctx.init_control, &cmd) {
340 Ok(()) => {
341 let _ = tx.send(Event::InitAck { cmd });
342 }
343 Err(e) => {
344 let _ = tx.send(Event::Error {
345 code: "init_send_failed".into(),
346 msg: e.to_string(),
347 });
348 }
349 }
350 }
351 }
352}
353
354fn run_compile(recipe: RecipeInline, store_root: PathBuf, tx: Sender<Event>) {
355 let name = recipe.name.clone();
356 let mutation = hammer_core::swm::Mutation::SourcePatch {
357 repo: recipe.repo,
358 commit: recipe.commit,
359 patch: recipe.patch,
360 patch_url: None,
361 build: hammer_core::swm::SwmBuild {
362 compiler: recipe.compiler,
363 target: recipe.target,
364 link: recipe.link,
365 flags: recipe.flags,
366 },
367 target_bin: format!("/usr/bin/{name}"),
368 expected_hash: None,
369 };
370 let store = match hammer_core::Store::open(&store_root) {
371 Ok(s) => s,
372 Err(e) => {
373 let _ = tx.send(Event::BuildFailed {
374 recipe: name,
375 reason: format!("store: {e}"),
376 log_tail: None,
377 });
378 return;
379 }
380 };
381 let cfg = hammer_build::BuildConfig::from_env_or_defaults(store.root());
382 match hammer_build::build_source_patch(&mutation, &cfg, &store, None) {
383 Ok(h) => {
384 let _ = tx.send(Event::BuildReady {
385 recipe: name,
386 artifact: h.as_str().to_string(),
387 });
388 }
389 Err(e) => {
390 let _ = tx.send(Event::BuildFailed {
391 recipe: name,
392 reason: e.to_string(),
393 log_tail: None,
394 });
395 }
396 }
397}
398
399fn query_file(path: &str) -> serde_json::Value {
400 let p = Path::new(path);
401 let meta = match std::fs::symlink_metadata(p) {
402 Ok(m) => m,
403 Err(e) => return serde_json::json!({"exists": false, "error": e.to_string()}),
404 };
405 use std::os::unix::fs::MetadataExt;
406 serde_json::json!({
407 "exists": true,
408 "size": meta.size(),
409 "mode": meta.mode(),
410 "uid": meta.uid(),
411 "gid": meta.gid(),
412 "is_dir": meta.is_dir(),
413 "is_symlink": meta.file_type().is_symlink(),
414 })
415}
416
417fn query_artifact(store_root: &Path, hash: &str) -> serde_json::Value {
418 let store = match hammer_core::Store::open(store_root) {
419 Ok(s) => s,
420 Err(e) => return serde_json::json!({"error": e.to_string()}),
421 };
422 match store.find_by_hash(hash) {
423 Ok(p) => serde_json::json!({"exists": true, "path": p.display().to_string()}),
424 Err(_) => serde_json::json!({"exists": false}),
425 }
426}
427
428fn t_of(cmd: &Command) -> &'static str {
429 match cmd {
430 Command::Hello { .. } => "hello",
431 Command::Compile { .. } => "compile",
432 Command::Inject { .. } => "inject",
433 Command::Query { .. } => "query",
434 Command::Init { .. } => "init",
435 }
436}
437
438// Re-export para que el caller no tenga que importar Mutex/Arc directamente.
439#[allow(dead_code)]
440pub(crate) type SharedCaps = Arc<Mutex<std::collections::BTreeSet<Cap>>>;
441
442#[cfg(test)]
443mod tests {
444 use super::*;
445
446 #[test]
447 fn default_policy_grants_query_to_others() {
448 let p = default_policy();
449 // Un peer con UID distinto al actual debe quedar como sólo-query.
450 let foreign = Peer {
451 uid: u32::MAX,
452 gid: 0,
453 pid: 1,
454 };
455 assert_eq!(p(&foreign), vec![Cap::Query]);
456 }
457
458 #[test]
459 fn t_of_matches_all_commands() {
460 // No es exhaustivo de Cargo, pero verifica que no nos olvidemos del nombre del tag.
461 assert_eq!(t_of(&Command::Hello { ver: 1, client: "".into() }), "hello");
462 assert_eq!(
463 t_of(&Command::Compile {
464 recipe: RecipeInline {
465 name: "x".into(),
466 repo: "g".into(),
467 commit: "c".into(),
468 patch: None,
469 compiler: "zig-cc".into(),
470 target: "t".into(),
471 link: "static".into(),
472 flags: vec![],
473 },
474 }),
475 "compile"
476 );
477 }
478
479 #[test]
480 fn query_file_reports_missing() {
481 let v = query_file("/no/existe/seguramente/ahora");
482 assert_eq!(v["exists"], serde_json::json!(false));
483 }
484}