⎇ d3649bfd1a80153652c2a028315369bb4b184024
crates/modules/nouser/core/src/bin/nouser.rs
History
Who touched what
raw
feat(explorer+daemon): discovery dinamico via broker + query socket
1//! `nouser` CLI — explorador de Mónadas.
2//!
3//! Subcomandos:
4//!
5//! - `scan <dir>` recorre `dir` y muestra las Mónadas detectadas.
6//! - `show <dir> <id?>` scan + detalles de la Mónada con prefijo de ID.
7//! - `json <dir>` scan + dump JSON con los manifests.
8//!
9//! Phase A: in-memory, sin persistencia, sin brahman sidecar. La
10//! sesión termina y todo se descarta. Phase B agrega persistencia y
11//! presencia ante el Init.
12
13use std::path::PathBuf;
14use std::process::ExitCode;
15
16use nouser_core::{
17 cluster, db, embed,
18 scanner::{self, ScanConfig},
19};
20
21fn main() -> ExitCode {
22 let args: Vec<String> = std::env::args().collect();
23 let prog = args.first().cloned().unwrap_or_else(|| "nouser".into());
24 let sub = match args.get(1).map(String::as_str) {
25 Some(s) => s,
26 None => {
27 print_usage(&prog);
28 return ExitCode::from(2);
29 }
30 };
31 let rest = &args[2..];
32
33 let result = match sub {
34 "scan" => cmd_scan(rest),
35 "show" => cmd_show(rest),
36 "json" => cmd_json(rest),
37 "daemon" => cmd_daemon(rest),
38 "attract" => cmd_attract(rest),
39 "--help" | "-h" | "help" => {
40 print_usage(&prog);
41 return ExitCode::SUCCESS;
42 }
43 other => {
44 eprintln!("nouser: comando desconocido '{other}'");
45 print_usage(&prog);
46 return ExitCode::from(2);
47 }
48 };
49
50 match result {
51 Ok(()) => ExitCode::SUCCESS,
52 Err(e) => {
53 eprintln!("nouser: {e}");
54 ExitCode::from(1)
55 }
56 }
57}
58
59fn print_usage(prog: &str) {
60 eprintln!("uso: {prog} <comando> [args]");
61 eprintln!();
62 eprintln!("comandos:");
63 eprintln!(" scan <dir> recorre un directorio y lista las Mónadas detectadas");
64 eprintln!(" show <dir> <prefix> scan + detalle de la Mónada cuyo ID empieza con <prefix>");
65 eprintln!(" json <dir> scan + dump JSON de todos los manifests");
66 eprintln!(" daemon <dir> scan + sidecarea cada Mónada al Init brahman");
67 eprintln!(" attract <dir> <file> dado un archivo, qué Mónada del scan lo atrae más");
68 eprintln!();
69 eprintln!("env:");
70 eprintln!(" NOUSER_MIN_FILES mínimo de archivos por Mónada (default: 3)");
71 eprintln!(" NOUSER_DB_PATH si está set, abre sled en esa ruta (persistencia)");
72 eprintln!(" BRAHMAN_INIT_SOCKET socket del Init (heredado de brahman-handshake)");
73}
74
75type Cmd = Result<(), Box<dyn std::error::Error>>;
76
77fn min_files() -> usize {
78 std::env::var("NOUSER_MIN_FILES")
79 .ok()
80 .and_then(|s| s.parse().ok())
81 .unwrap_or(cluster::DEFAULT_MIN_FILES_PER_MONAD)
82}
83
84fn require_dir(args: &[String]) -> Result<PathBuf, Box<dyn std::error::Error>> {
85 let dir = args.first().ok_or("falta argumento <dir>")?;
86 Ok(PathBuf::from(dir))
87}
88
89fn run_scan(dir: &PathBuf) -> Result<(db::MonadDb, usize), Box<dyn std::error::Error>> {
90 let files = scanner::scan_directory(dir, &ScanConfig::default())?;
91 let n_files = files.len();
92 let monads = cluster::by_directory(&files, min_files());
93 let mut db = open_db()?;
94 db.ingest_files(files);
95 db.replace_monads(monads);
96 Ok((db, n_files))
97}
98
99/// Abre el `MonadDb`. Si `NOUSER_DB_PATH` está set, persistencia sled;
100/// si no, store en memoria.
101fn open_db() -> Result<db::MonadDb, Box<dyn std::error::Error>> {
102 if let Ok(path) = std::env::var("NOUSER_DB_PATH") {
103 Ok(db::MonadDb::open(&path)?)
104 } else {
105 Ok(db::MonadDb::new())
106 }
107}
108
109fn cmd_scan(args: &[String]) -> Cmd {
110 let dir = require_dir(args)?;
111 let (db, n_files) = run_scan(&dir)?;
112
113 println!(
114 "scan: {} archivos en {}, {} mónadas (min_files={})",
115 n_files,
116 dir.display(),
117 db.monad_count(),
118 min_files()
119 );
120 if db.monad_count() == 0 {
121 println!(" (ninguna Mónada — bajá NOUSER_MIN_FILES o apuntá a un dir con más archivos)");
122 return Ok(());
123 }
124 println!();
125 for m in db.monads() {
126 let id_short = format!("{}", m.id);
127 let id_short = &id_short[..8];
128 println!(
129 " [{}] {:30} card={} ent={:.2} lens={:?}",
130 id_short, m.label, m.cardinality, m.entropy, m.dominant_lens,
131 );
132 if !m.keywords.is_empty() {
133 println!(" keywords: {}", m.keywords.join(", "));
134 }
135 }
136 Ok(())
137}
138
139fn cmd_show(args: &[String]) -> Cmd {
140 let dir = require_dir(args)?;
141 let prefix = args.get(1).ok_or("falta argumento <prefix>")?;
142 let (db, _) = run_scan(&dir)?;
143
144 let m = db
145 .monads()
146 .find(|m| m.id.to_string().starts_with(prefix))
147 .ok_or_else(|| format!("ninguna Mónada con prefijo '{prefix}'"))?;
148
149 println!("Monad {}", m.id);
150 println!(" label: {}", m.label);
151 println!(" summary: {}", m.summary);
152 println!(" cardinality: {}", m.cardinality);
153 println!(" entropy: {:.4}", m.entropy);
154 println!(" lens: {:?}", m.dominant_lens);
155 println!(" keywords: {}", m.keywords.join(", "));
156 println!(" members ({}):", m.members.len());
157 for f in db.resolve_members(m.id) {
158 println!(
159 " {:>10} bytes {}",
160 f.size,
161 f.path.display()
162 );
163 }
164 Ok(())
165}
166
167fn cmd_json(args: &[String]) -> Cmd {
168 let dir = require_dir(args)?;
169 let (db, _) = run_scan(&dir)?;
170 let manifests: Vec<_> = db.monads().cloned().collect();
171 println!("{}", serde_json::to_string_pretty(&manifests)?);
172 Ok(())
173}
174
175fn cmd_daemon(args: &[String]) -> Cmd {
176 let dir = require_dir(args)?;
177
178 let pool = std::sync::Arc::new(
179 brahman_sidecar::SidecarPool::new().map_err(|e| format!("crear pool: {e}"))?,
180 );
181
182 // 1. Decidir el path del query socket ANTES de armar el engine
183 // Card (porque viaja como service_socket en la Card).
184 let query_socket = nouser_card::query::transport::default_socket_path();
185
186 // 2. Engine como Ente. Declara service_socket + flow.output para
187 // que el broker pueda emitir MatchEvent::Available a consumers
188 // interesados en `flow.input = monad-list:json`.
189 let engine_card = build_engine_card(query_socket.clone());
190 let engine_id = engine_card.id;
191 let engine_label = engine_card.label.clone();
192 eprintln!(
193 "nouser daemon: publicando engine '{}' (kind=Ente, id={}, socket={})",
194 engine_label,
195 engine_id,
196 query_socket.display()
197 );
198 pool.spawn(engine_card);
199
200 // 2. Hidratación: si NOUSER_DB_PATH apunta a un sled poblado,
201 // publicar lo que ya tenemos ANTES del re-scan. brahman-status
202 // ve mónadas reales en milisegundos, no en segundos.
203 let mut db = open_db()?;
204 let prior_count = db.monad_count();
205 if prior_count > 0 {
206 let mut hydrated = 0usize;
207 let mut skipped_model = 0usize;
208 for monad in db.monads() {
209 // Sólo publicamos centroides del modelo actual; los demás
210 // son data muerta hasta que el re-scan los reemplace.
211 let valid = monad
212 .centroid_model
213 .as_deref()
214 .map(|id| id == embed::MODEL_ID)
215 .unwrap_or(false);
216 if !valid {
217 skipped_model += 1;
218 continue;
219 }
220 let mut card = monad.to_brahman_card();
221 card.references.push(brahman_card::CardReference {
222 kind: brahman_card::RelationshipKind::OwnedBy,
223 target_id: engine_id,
224 target_label: engine_label.clone(),
225 });
226 pool.spawn(card);
227 hydrated += 1;
228 }
229 eprintln!(
230 "nouser daemon: hidratadas {} mónadas previas{} en O(1)",
231 hydrated,
232 if skipped_model > 0 {
233 format!(" ({} dropeadas por centroid_model distinto)", skipped_model)
234 } else {
235 String::new()
236 }
237 );
238 }
239
240 // 3. Re-scan con hidratación: las Mónadas con mismo path_hint
241 // reusan id, así que NO generamos sesiones duplicadas para los
242 // mismos directorios — el sidecar previo ya tiene esa identidad.
243 let files = scanner::scan_directory(&dir, &scanner::ScanConfig::default())?;
244 let n_files = files.len();
245 let monads = cluster::by_directory_hydrated(&files, min_files(), Some(&db));
246 let scanned_count = monads.len();
247 eprintln!(
248 "nouser daemon: re-scan {} archivos en {} → {} mónadas",
249 n_files,
250 dir.display(),
251 scanned_count
252 );
253
254 // Publicamos sólo las Mónadas NUEVAS (las que no estaban en la
255 // hidratación inicial). El criterio: si el id estaba en la DB
256 // previa, el sidecar de la hidratación ya cubre esa identidad.
257 let prior_ids: std::collections::BTreeSet<_> = db.monads().map(|m| m.id).collect();
258 let mut newly_spawned = 0usize;
259 for monad in &monads {
260 if prior_ids.contains(&monad.id) {
261 continue; // ya publicada en hidratación
262 }
263 let mut card = monad.to_brahman_card();
264 card.references.push(brahman_card::CardReference {
265 kind: brahman_card::RelationshipKind::OwnedBy,
266 target_id: engine_id,
267 target_label: engine_label.clone(),
268 });
269 pool.spawn(card);
270 newly_spawned += 1;
271 }
272
273 // Reescribimos la DB con el set actual (idempotente para los
274 // hidratados; reemplazo para los nuevos).
275 db.ingest_files(files);
276 db.replace_monads(monads);
277
278 eprintln!(
279 "nouser daemon: 1 ente + {} mónadas vivas ({} nuevas vs hidratación)",
280 scanned_count, newly_spawned
281 );
282
283 // Engine query socket: bind antes del watcher para que cualquier
284 // consumer descubierto vía broker pueda consultarnos enseguida.
285 // Si el bind falla, seguimos sin él — la UI degrada a "no
286 // alcanzable" pero el daemon sigue procesando cambios.
287 let db_shared = std::sync::Arc::new(std::sync::Mutex::new(db));
288 let _query_listener = match nouser_core::engine_socket::spawn_listener(
289 nouser_core::engine_socket::ListenerConfig {
290 socket_path: query_socket.clone(),
291 engine_id,
292 engine_label: engine_label.clone(),
293 watching: Some(dir.clone()),
294 },
295 db_shared.clone(),
296 ) {
297 Ok(h) => {
298 eprintln!(
299 "nouser daemon: query socket activo en {} (proto: nouser_card::query)",
300 query_socket.display()
301 );
302 Some(h)
303 }
304 Err(e) => {
305 eprintln!(
306 "nouser daemon: query socket NO disponible ({e}) — explorer no podrá consultar"
307 );
308 None
309 }
310 };
311
312 // Watcher: cada cambio en el árbol — coalescido con debounce de
313 // 150ms — dispara un re-scan + re-cluster del directorio y
314 // re-publica al broker las Mónadas afectadas (drop + spawn por id,
315 // gracias al replace en `SidecarPool::spawn`).
316 let _watcher = match spawn_fs_watcher(
317 dir.clone(),
318 db_shared.clone(),
319 pool.clone(),
320 engine_id,
321 engine_label.clone(),
322 ) {
323 Ok(w) => {
324 eprintln!(
325 "nouser daemon: watcher activo en {} (debounce 150ms, re-publish on) — Ctrl-C para terminar.",
326 dir.display()
327 );
328 Some(w)
329 }
330 Err(e) => {
331 eprintln!(
332 "nouser daemon: watcher deshabilitado ({e}) — Ctrl-C para terminar."
333 );
334 None
335 }
336 };
337
338 std::thread::park();
339 drop(_watcher);
340 drop(_query_listener);
341 let _ = std::fs::remove_file(&query_socket); // best-effort cleanup
342 drop(pool);
343 Ok(())
344}
345
346/// Ventana de debounce: notify dispara Create+Modify(+) por cada
347/// edición; sin coalescer veríamos N reacciones por un solo `:w`.
348/// 150ms es generoso para editores típicos (vim/code) y mantiene el
349/// feedback "vivo" para el usuario.
350const WATCHER_DEBOUNCE_MS: u64 = 150;
351
352/// Watcher de filesystem con debounce + re-publish al broker.
353///
354/// Pipeline:
355///
356/// 1. **notify** dispara eventos crudos a un canal interno.
357/// 2. **dispatcher**: filtra a Create/Modify/Remove de paths bajo
358/// `dir`, descarta el resto, reenvía al canal de debounce.
359/// 3. **coordinator**: mantiene un `HashMap<PathBuf, Instant>`.
360/// Cada vez que el canal queda en silencio durante
361/// `WATCHER_DEBOUNCE_MS`, agrupa los paths cuya última actividad
362/// superó la ventana y los procesa en **un solo batch**.
363/// 4. **process_change_batch**: re-scan + re-cluster hidratado +
364/// diff vs DB + `pool.drop_session` para Mónadas desaparecidas
365/// + `pool.spawn` para Mónadas nuevas o con composición distinta.
366/// `pool.spawn` reemplaza la sesión previa con el mismo `Card.id`,
367/// así que el broker ve el manifest fresco sin sesiones huérfanas.
368fn spawn_fs_watcher(
369 dir: std::path::PathBuf,
370 db: std::sync::Arc<std::sync::Mutex<db::MonadDb>>,
371 pool: std::sync::Arc<brahman_sidecar::SidecarPool>,
372 engine_id: brahman_card::ulid::Ulid,
373 engine_label: String,
374) -> Result<notify::RecommendedWatcher, Box<dyn std::error::Error>> {
375 use notify::{Event, EventKind, RecursiveMode, Watcher};
376
377 let (notify_tx, notify_rx) = std::sync::mpsc::channel::<notify::Result<Event>>();
378 let mut watcher = notify::recommended_watcher(move |res| {
379 let _ = notify_tx.send(res);
380 })?;
381 watcher.watch(&dir, RecursiveMode::Recursive)?;
382
383 let (path_tx, path_rx) = std::sync::mpsc::channel::<std::path::PathBuf>();
384
385 // Dispatcher: notify → filtro → canal de paths.
386 let dispatch_dir = dir.clone();
387 std::thread::Builder::new()
388 .name("nouser-watcher-dispatch".into())
389 .spawn(move || {
390 for res in notify_rx {
391 let event = match res {
392 Ok(e) => e,
393 Err(e) => {
394 eprintln!("[watcher] error: {e}");
395 continue;
396 }
397 };
398 // Create/Modify viven; Remove también nos importa
399 // (puede colapsar Mónadas).
400 let interesting = matches!(
401 event.kind,
402 EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
403 );
404 if !interesting {
405 continue;
406 }
407 for path in event.paths {
408 if !path.starts_with(&dispatch_dir) {
409 continue;
410 }
411 let _ = path_tx.send(path);
412 }
413 }
414 })?;
415
416 // Coordinator: debounce + batch dispatch.
417 let coord_dir = dir;
418 std::thread::Builder::new()
419 .name("nouser-watcher-coord".into())
420 .spawn(move || {
421 let debounce = std::time::Duration::from_millis(WATCHER_DEBOUNCE_MS);
422 let mut pending: std::collections::HashMap<
423 std::path::PathBuf,
424 std::time::Instant,
425 > = std::collections::HashMap::new();
426 loop {
427 match path_rx.recv_timeout(debounce) {
428 Ok(path) => {
429 pending.insert(path, std::time::Instant::now());
430 }
431 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
432 Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
433 }
434 let now = std::time::Instant::now();
435 let due: Vec<std::path::PathBuf> = pending
436 .iter()
437 .filter(|(_, t)| now.duration_since(**t) >= debounce)
438 .map(|(p, _)| p.clone())
439 .collect();
440 if due.is_empty() {
441 continue;
442 }
443 for p in &due {
444 pending.remove(p);
445 }
446 process_change_batch(
447 &due,
448 &coord_dir,
449 &db,
450 &pool,
451 engine_id,
452 &engine_label,
453 );
454 }
455 })?;
456
457 Ok(watcher)
458}
459
460/// Procesa un batch de paths cambiados: re-scanea el árbol, re-clusteriza
461/// con hidratación, y propaga el delta de Mónadas al broker.
462///
463/// El re-scan global es deliberado: el clustering por directorio es global
464/// por diseño, así que un cambio en `src/foo.rs` puede mover Mónadas en
465/// `src/` sin tocar `tests/`. Coste O(N archivos), aceptable para
466/// directorios típicos (<10k archivos). Optimizar a re-cluster parcial
467/// cuando duela.
468fn process_change_batch(
469 paths: &[std::path::PathBuf],
470 dir: &std::path::Path,
471 db: &std::sync::Arc<std::sync::Mutex<db::MonadDb>>,
472 pool: &std::sync::Arc<brahman_sidecar::SidecarPool>,
473 engine_id: brahman_card::ulid::Ulid,
474 engine_label: &str,
475) {
476 eprintln!(
477 "[watcher] ⚙ batch: {} path(s) coalescidos → re-scan",
478 paths.len()
479 );
480
481 let files = match scanner::scan_directory(dir, &scanner::ScanConfig::default()) {
482 Ok(f) => f,
483 Err(e) => {
484 eprintln!("[watcher] re-scan falló: {e}");
485 return;
486 }
487 };
488
489 let mut db_lock = match db.lock() {
490 Ok(g) => g,
491 Err(_) => {
492 eprintln!("[watcher] mutex envenenado — abortando batch");
493 return;
494 }
495 };
496
497 let prior_monads: Vec<nouser_card::MonadManifest> = db_lock.monads().cloned().collect();
498 let prior_ref: &db::MonadDb = &db_lock;
499 let monads = cluster::by_directory_hydrated(&files, min_files(), Some(prior_ref));
500
501 let prior_ids: std::collections::BTreeSet<_> =
502 prior_monads.iter().map(|m| m.id).collect();
503 let new_ids: std::collections::BTreeSet<_> = monads.iter().map(|m| m.id).collect();
504
505 // Mónadas que ya no existen (directorio quedó por debajo de
506 // min_files o fue removido): cerramos su sesión en el broker.
507 let mut removed = 0usize;
508 for id in prior_ids.difference(&new_ids) {
509 pool.drop_session(*id);
510 removed += 1;
511 if let Some(prev) = prior_monads.iter().find(|m| &m.id == id) {
512 eprintln!(
513 "[watcher] ✖ {} ({}) desapareció — sesión cerrada",
514 &id.to_string()[..8],
515 prev.label
516 );
517 }
518 }
519
520 // Mónadas nuevas o cuya composición cambió (members/centroid):
521 // (re)spawn — el pool reemplaza la sesión previa con el mismo id.
522 let mut respawned = 0usize;
523 let mut fresh = 0usize;
524 for monad in &monads {
525 let prev = prior_monads.iter().find(|m| m.id == monad.id);
526 let is_new = prev.is_none();
527 let changed = match prev {
528 Some(p) => p.members != monad.members || p.centroid != monad.centroid,
529 None => true,
530 };
531 if !changed {
532 continue;
533 }
534 let mut card = monad.to_brahman_card();
535 card.references.push(brahman_card::CardReference {
536 kind: brahman_card::RelationshipKind::OwnedBy,
537 target_id: engine_id,
538 target_label: engine_label.to_string(),
539 });
540 pool.spawn(card);
541 if is_new {
542 fresh += 1;
543 eprintln!(
544 "[watcher] ✦ {} nace ({} miembros, lens={:?})",
545 monad.label, monad.cardinality, monad.dominant_lens
546 );
547 } else {
548 respawned += 1;
549 let prev = prev.unwrap();
550 let delta_members = monad.members.len() as i64 - prev.members.len() as i64;
551 eprintln!(
552 "[watcher] ↻ {} refresh ({} miembros, Δ={:+})",
553 monad.label, monad.cardinality, delta_members
554 );
555 }
556 }
557
558 if removed == 0 && fresh == 0 && respawned == 0 {
559 eprintln!("[watcher] (sin cambios estructurales tras re-cluster)");
560 } else {
561 eprintln!(
562 "[watcher] ⌃ delta: {} nuevas, {} refrescadas, {} cerradas — {} sesiones vivas",
563 fresh,
564 respawned,
565 removed,
566 pool.live_sessions()
567 );
568 }
569
570 db_lock.ingest_files(files);
571 db_lock.replace_monads(monads);
572}
573
574fn cmd_attract(args: &[String]) -> Cmd {
575 let mut remote = false;
576 let mut positional: Vec<&String> = Vec::new();
577 for a in args {
578 if a == "--remote" {
579 remote = true;
580 } else {
581 positional.push(a);
582 }
583 }
584 let dir = positional
585 .first()
586 .map(|s| std::path::PathBuf::from(s.as_str()))
587 .ok_or("falta argumento <dir>")?;
588 let file_path = positional.get(1).ok_or("falta argumento <file>")?;
589 let file_path = std::path::PathBuf::from(file_path.as_str());
590 if !file_path.exists() {
591 return Err(format!("archivo no existe: {}", file_path.display()).into());
592 }
593
594 let (db, _) = run_scan(&dir)?;
595
596 // Construimos un FileEntry para el archivo objetivo.
597 let metadata = std::fs::metadata(&file_path)?;
598 let mtime_ms = metadata
599 .modified()
600 .ok()
601 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
602 .map(|d| d.as_millis() as u64)
603 .unwrap_or(0);
604 let target = nouser_card::FileEntry {
605 id: nouser_card::FileId::from(nouser_card::ulid::Ulid::new()),
606 path: file_path.clone(),
607 content_hash: None,
608 size: metadata.len(),
609 mtime_ms,
610 extension: file_path
611 .extension()
612 .and_then(|s| s.to_str())
613 .map(|s| s.to_lowercase()),
614 };
615
616 // Embedding del target + identificación del modelo que lo produjo.
617 // Local: pseudo-32d. Remote: lo que devuelva el provider electo
618 // (mock=pseudo-32d, real=fastembed-384d).
619 let (target_vec, target_model, source) = if remote {
620 let (v, model) = remote_embed(&target)?;
621 (v, model, "remote")
622 } else {
623 (
624 embed::embed(&target).to_vec(),
625 embed::MODEL_ID.to_string(),
626 "local",
627 )
628 };
629
630 // Filtramos Mónadas cuyo centroid_model NO matchee. Mezclar
631 // 32-d con 384-d daría scores sin sentido (diferente semántica
632 // y cosine no compara cross-modelo).
633 let mut ranked: Vec<(&nouser_card::MonadManifest, f32)> = db
634 .monads()
635 .filter(|m| !m.centroid.is_empty())
636 .filter(|m| match &m.centroid_model {
637 Some(id) => id == &target_model,
638 None => true, // legacy sin tag — comparamos best-effort
639 })
640 .map(|m| (m, embed::attraction_score(&target_vec, m)))
641 .collect();
642 ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
643
644 let total_monads = db.monads().filter(|m| !m.centroid.is_empty()).count();
645 let skipped = total_monads - ranked.len();
646
647 if ranked.is_empty() {
648 println!("ninguna Mónada con centroide en {}", dir.display());
649 return Ok(());
650 }
651
652 println!("archivo: {}", file_path.display());
653 println!("scan dir: {}", dir.display());
654 println!("embed: {} ({})", source, target_model);
655 if skipped > 0 {
656 println!(
657 "skipped: {} mónada(s) con centroid_model distinto (no comparables)",
658 skipped
659 );
660 }
661 println!("ranking de atracción (cosine similarity):");
662 println!();
663 for (i, (m, score)) in ranked.iter().take(5).enumerate() {
664 let marker = if *score >= embed::DEFAULT_ATTRACTION_THRESHOLD && i == 0 {
665 "🧲"
666 } else if i == 0 {
667 "·"
668 } else {
669 " "
670 };
671 let id_short = format!("{}", m.id);
672 let id_short = &id_short[..8];
673 println!(
674 " {} {:.4} [{}] {:30} ({})",
675 marker, score, id_short, m.label, m.summary
676 );
677 }
678 if ranked[0].1 < embed::DEFAULT_ATTRACTION_THRESHOLD {
679 println!();
680 println!(
681 " (mejor score {:.4} < umbral {:.4} — el archivo no se 'pega' a ninguna)",
682 ranked[0].1,
683 embed::DEFAULT_ATTRACTION_THRESHOLD
684 );
685 }
686 Ok(())
687}
688
689/// Pipeline completo del modo `--remote`:
690/// 1. Si `NOUSER_NOUS_SOCKET` está set, lo usa directo (override
691/// explícito, atajo para tests).
692/// 2. Si no, delega en `brahman_sidecar::await_provider_blocking` —
693/// el sidecar se conecta al broker, registra un consumer Card con
694/// `flow.input = embed-result:json`, espera el primer
695/// `MatchEvent::Available` y devuelve el socket. Esto activa la
696/// lógica de `priority_contexts`: bajo `BRAHMAN_BROKER_CONTEXT=test/prod`,
697/// el proveedor electo cambia sin que este código toque nada.
698/// 3. Con el socket resuelto, dispara la RPC `EmbedFile`.
699///
700/// Devuelve `(embedding, model_id)` — el caller necesita ambos para
701/// comparar contra centroides taggeados con su mismo `centroid_model`.
702fn remote_embed(
703 file: &nouser_card::FileEntry,
704) -> Result<(Vec<f32>, String), Box<dyn std::error::Error>> {
705 if let Ok(explicit) = std::env::var("NOUSER_NOUS_SOCKET") {
706 let sock = std::path::PathBuf::from(explicit);
707 return embed_via(&sock, file);
708 }
709
710 let consumer = brahman_sidecar::build_consumer_card(
711 "nouser.attract-cli",
712 nouser_nous::FLOW_EMBED_RESULT,
713 nouser_nous::FLOW_TYPE_NAME,
714 );
715 let producer_sock = brahman_sidecar::await_provider_blocking(
716 consumer,
717 std::time::Duration::from_secs(3),
718 )?;
719 embed_via(&producer_sock, file)
720}
721
722/// RPC blocking contra un socket nouser-nous concreto. Devuelve
723/// `(embedding, model_id)` — el `model_id` viaja en la response y
724/// permite al caller saber qué centroides son comparables.
725fn embed_via(
726 sock_path: &std::path::Path,
727 file: &nouser_card::FileEntry,
728) -> Result<(Vec<f32>, String), Box<dyn std::error::Error>> {
729 use std::io::{BufRead, BufReader, Write};
730 use std::os::unix::net::UnixStream;
731
732 if !sock_path.exists() {
733 return Err(format!("socket no existe: {}", sock_path.display()).into());
734 }
735
736 let mut stream = UnixStream::connect(sock_path)?;
737 let req = nouser_nous::EmbedRequest {
738 kind: nouser_nous::RequestKind::EmbedFile,
739 payload: serde_json::to_value(nouser_nous::EmbedFilePayload {
740 path: file.path.display().to_string(),
741 extension: file.extension.clone(),
742 size: file.size,
743 mtime_ms: file.mtime_ms,
744 })?,
745 };
746 let line = serde_json::to_string(&req)?;
747 stream.write_all(line.as_bytes())?;
748 stream.write_all(b"\n")?;
749 stream.flush()?;
750
751 let mut reader = BufReader::new(stream);
752 let mut response = String::new();
753 reader.read_line(&mut response)?;
754 if response.is_empty() {
755 return Err("nouser-nous cerró sin respuesta".into());
756 }
757
758 if let Ok(resp) = serde_json::from_str::<nouser_nous::EmbedResponse>(&response) {
759 return Ok((resp.embedding, resp.model));
760 }
761 let err: nouser_nous::ErrorResponse = serde_json::from_str(&response)?;
762 Err(format!("nouser-nous: {}", err.error).into())
763}
764
765/// Card del propio engine (kind=Ente). Es el "ser" que produce y
766/// administra Mónadas; aparece en brahman-status junto a sus Mónadas.
767///
768/// Declara `service_socket` y `flow.output = monad-list:json` para
769/// que un consumer (UI, CLI) pueda descubrir al daemon vía broker
770/// MatchEvent y consultarle por sus Mónadas sin pasar por
771/// brahman-admin.
772fn build_engine_card(service_socket: std::path::PathBuf) -> brahman_card::Card {
773 use brahman_card::{Card, CardKind, Flow, Flows, Lifecycle, Payload, Priority, Supervision, TypeRef};
774 use nouser_card::query::{FLOW_MONAD_LIST, FLOW_TYPE_NAME};
775
776 Card {
777 payload: Payload::Virtual,
778 supervision: Supervision::Delegate,
779 lifecycle: Lifecycle::Daemon,
780 priority: Priority::Normal,
781 kind: CardKind::Ente,
782 service_socket: Some(service_socket),
783 flow: Flows {
784 input: vec![],
785 output: vec![Flow {
786 name: FLOW_MONAD_LIST.into(),
787 ty: TypeRef::Primitive {
788 name: FLOW_TYPE_NAME.into(),
789 },
790 pin_to: None,
791 }],
792 },
793 ..Card::new("brahman.nouser_engine")
794 }
795}