⎇ d3649bfd1a80153652c2a028315369bb4b184024
crates/modules/nakui/core/src/bin/nakui.rs
History
Who touched what
raw
feat: segundo módulo (nakui) + admin API + brahman-status
1//! `nakui` — operator CLI for inspecting, replaying, and verifying an
2//! event log produced by the kernel. The three subcommands map to the
3//! three things you need when something goes sideways in production:
4//!
5//! - `inspect` — what's in the log? (audit trail)
6//! - `replay` — what state does the log produce? (recovery dry-run)
7//! - `verify-log` — does every morphism still reproduce its ops?
8//! (determinism contract — the regression alarm)
9//!
10//! Exit codes: 0 on success, 1 on operational error, 2 on bad arguments.
11
12use std::collections::BTreeMap;
13use std::path::PathBuf;
14use std::process::ExitCode;
15
16use nakui_core::drift::{DriftDiff, check_against_socket};
17use nakui_core::event_log::{
18 EventLog, LogEntry, Snapshot, replay_with_snapshot_into, verify_log,
19};
20use nakui_core::executor::Executor;
21use nakui_core::run::run_server;
22use nakui_core::store::MemoryStore;
23
24fn main() -> ExitCode {
25 let args: Vec<String> = std::env::args().collect();
26 let prog = args.first().cloned().unwrap_or_else(|| "nakui".into());
27 let sub = match args.get(1).map(String::as_str) {
28 Some(s) => s,
29 None => {
30 print_usage(&prog);
31 return ExitCode::from(2);
32 }
33 };
34 let rest = &args[2..];
35
36 let result = match sub {
37 "inspect" => cmd_inspect(rest),
38 "replay" => cmd_replay(rest),
39 "verify-log" => cmd_verify_log(rest),
40 "run" => cmd_run(rest),
41 "drift" => cmd_drift(rest),
42 "snapshot" => cmd_snapshot(rest),
43 "compact" => cmd_compact(rest),
44 "-h" | "--help" | "help" => {
45 print_usage(&prog);
46 return ExitCode::SUCCESS;
47 }
48 other => {
49 eprintln!("nakui: unknown subcommand `{}`", other);
50 print_usage(&prog);
51 return ExitCode::from(2);
52 }
53 };
54
55 match result {
56 Ok(()) => ExitCode::SUCCESS,
57 Err(CliError::BadArgs(msg)) => {
58 eprintln!("nakui: {}", msg);
59 print_usage(&prog);
60 ExitCode::from(2)
61 }
62 Err(CliError::Op(msg)) => {
63 eprintln!("nakui: {}", msg);
64 ExitCode::from(1)
65 }
66 // Drift uses its own exit code so callers can distinguish "the
67 // tool failed" (1) from "the tool worked and detected drift" (3).
68 Err(CliError::DriftDetected) => ExitCode::from(3),
69 }
70}
71
72enum CliError {
73 BadArgs(String),
74 Op(String),
75 DriftDetected,
76}
77
78fn print_usage(prog: &str) {
79 eprintln!(
80 "usage:
81 {p} inspect --log <path>
82 {p} replay --log <path> [--snapshot <path>]
83 {p} verify-log --log <path> --module <dir>
84 {p} run --log <path> --module <dir> --socket <path>
85 [--snapshot <path>] [--store-path <dir>]
86 {p} drift --log <path> --against <socket>
87 {p} snapshot --log <path> --module <dir> --out <path>
88 {p} compact --log <path> --snapshot <path>
89
90 --store-path activates persistent SurrealStore (kv-surrealkv);
91 requires the binary to be built with `--features persistent`.",
92 p = prog
93 );
94}
95
96/// Minimal flag parser: `--name value` pairs, no `=` form, no clustering.
97/// Returns a map of name -> value. Unknown flags are an error so typos
98/// surface immediately instead of silently being ignored.
99fn parse_flags(args: &[String], allowed: &[&str]) -> Result<BTreeMap<String, String>, CliError> {
100 let mut out = BTreeMap::new();
101 let mut i = 0;
102 while i < args.len() {
103 let flag = &args[i];
104 if !flag.starts_with("--") {
105 return Err(CliError::BadArgs(format!(
106 "expected --flag, got `{}`",
107 flag
108 )));
109 }
110 let name = &flag[2..];
111 if !allowed.contains(&name) {
112 return Err(CliError::BadArgs(format!("unknown flag `--{}`", name)));
113 }
114 let val = args.get(i + 1).ok_or_else(|| {
115 CliError::BadArgs(format!("flag `--{}` requires a value", name))
116 })?;
117 out.insert(name.to_string(), val.clone());
118 i += 2;
119 }
120 Ok(out)
121}
122
123fn require<'a>(
124 flags: &'a BTreeMap<String, String>,
125 name: &str,
126) -> Result<&'a String, CliError> {
127 flags
128 .get(name)
129 .ok_or_else(|| CliError::BadArgs(format!("missing required flag `--{}`", name)))
130}
131
132fn cmd_inspect(args: &[String]) -> Result<(), CliError> {
133 let flags = parse_flags(args, &["log"])?;
134 let log_path = PathBuf::from(require(&flags, "log")?);
135 let log = EventLog::open(&log_path).map_err(|e| CliError::Op(format!("open log: {}", e)))?;
136 let entries = log
137 .entries()
138 .map_err(|e| CliError::Op(format!("read log: {}", e)))?;
139 println!("log: {}", log.path().display());
140 println!("entries: {}", entries.len());
141 if entries.is_empty() {
142 return Ok(());
143 }
144 println!("seq range: {}..={}", entries[0].seq(), entries.last().unwrap().seq());
145 println!();
146 for e in &entries {
147 match e {
148 LogEntry::Seed {
149 seq, entity, id, ..
150 } => println!(" #{:04} seed {} {}", seq, entity, id),
151 LogEntry::Morphism {
152 seq,
153 morphism,
154 ops,
155 inputs,
156 ..
157 } => {
158 let inputs_s = inputs
159 .iter()
160 .map(|(k, v)| format!("{}={}", k, v))
161 .collect::<Vec<_>>()
162 .join(", ");
163 println!(
164 " #{:04} morph {} ({} ops) [{}]",
165 seq,
166 morphism,
167 ops.len(),
168 inputs_s
169 );
170 }
171 }
172 }
173 Ok(())
174}
175
176fn cmd_replay(args: &[String]) -> Result<(), CliError> {
177 let flags = parse_flags(args, &["log", "snapshot"])?;
178 let log_path = PathBuf::from(require(&flags, "log")?);
179 let log = EventLog::open(&log_path).map_err(|e| CliError::Op(format!("open log: {}", e)))?;
180
181 let snapshot = if let Some(p) = flags.get("snapshot") {
182 let path = PathBuf::from(p);
183 Snapshot::load(&path)
184 .map_err(|e| CliError::Op(format!("load snapshot: {}", e)))?
185 .ok_or_else(|| CliError::Op(format!("snapshot not found: {}", path.display())))?
186 .into()
187 } else {
188 None::<Snapshot>
189 };
190
191 let mut store = MemoryStore::new();
192 replay_with_snapshot_into(&log, snapshot.as_ref(), &mut store)
193 .map_err(|e| CliError::Op(format!("replay: {}", e)))?;
194
195 let entries = log
196 .entries()
197 .map_err(|e| CliError::Op(format!("read log: {}", e)))?;
198 let last_seq = entries.last().map(|e| e.seq().to_string()).unwrap_or_else(|| "<empty>".into());
199 println!("replayed log: {}", log.path().display());
200 if let Some(snap) = &snapshot {
201 println!("snapshot: seq {} (covers seq <= {})", snap.seq, snap.seq);
202 }
203 println!("last seq: {}", last_seq);
204 println!("entities:");
205 let mut by_entity: Vec<(&String, usize)> = store
206 .records()
207 .iter()
208 .map(|(k, v)| (k, v.len()))
209 .collect();
210 by_entity.sort_by(|a, b| a.0.cmp(b.0));
211 if by_entity.is_empty() {
212 println!(" (none)");
213 } else {
214 for (entity, count) in by_entity {
215 println!(" {:<20} {}", entity, count);
216 }
217 }
218 Ok(())
219}
220
221fn cmd_drift(args: &[String]) -> Result<(), CliError> {
222 let flags = parse_flags(args, &["log", "against"])?;
223 let log_path = PathBuf::from(require(&flags, "log")?);
224 let socket_path = PathBuf::from(require(&flags, "against")?);
225
226 let report = check_against_socket(&log_path, &socket_path)
227 .map_err(|e| CliError::Op(format!("drift check: {}", e)))?;
228
229 let log_hex = hex_encode(&report.log_hash);
230 let server_hex = hex_encode(&report.server_hash);
231 if report.in_sync() {
232 println!(
233 "ok: in sync (hash {}, {} records)",
234 short_hash(&log_hex),
235 report.log_records
236 );
237 return Ok(());
238 }
239
240 println!("DRIFT detected");
241 println!(
242 " log replay: hash {} ({} records)",
243 log_hex, report.log_records
244 );
245 println!(
246 " server state: hash {} ({} records)",
247 server_hex, report.server_records
248 );
249 println!();
250 println!("diffs:");
251 for d in &report.diffs {
252 match d {
253 DriftDiff::OnlyOnServer { entity, id, .. } => {
254 println!(" + {} {} (only on server)", entity, id);
255 }
256 DriftDiff::OnlyInLog { entity, id, .. } => {
257 println!(" - {} {} (only in log replay)", entity, id);
258 }
259 DriftDiff::Tampered {
260 entity,
261 id,
262 log_value,
263 server_value,
264 } => {
265 println!(
266 " ~ {} {} (tampered)\n log: {}\n server: {}",
267 entity, id, log_value, server_value
268 );
269 }
270 }
271 }
272 Err(CliError::DriftDetected)
273}
274
275fn hex_encode(bytes: &[u8]) -> String {
276 const HEX: &[u8; 16] = b"0123456789abcdef";
277 let mut out = String::with_capacity(bytes.len() * 2);
278 for &b in bytes {
279 out.push(HEX[(b >> 4) as usize] as char);
280 out.push(HEX[(b & 0x0f) as usize] as char);
281 }
282 out
283}
284
285fn short_hash(hex: &str) -> String {
286 if hex.len() <= 12 {
287 hex.to_string()
288 } else {
289 format!("{}…{}", &hex[..6], &hex[hex.len() - 4..])
290 }
291}
292
293fn cmd_run(args: &[String]) -> Result<(), CliError> {
294 let flags = parse_flags(args, &["log", "module", "socket", "snapshot", "store-path"])?;
295 let log_path = PathBuf::from(require(&flags, "log")?);
296 let module_dir = PathBuf::from(require(&flags, "module")?);
297 let socket_path = PathBuf::from(require(&flags, "socket")?);
298 let snapshot_path = flags.get("snapshot").map(PathBuf::from);
299 let store_path = flags.get("store-path").map(PathBuf::from);
300
301 eprintln!(
302 "nakui run: module={} log={} socket={} snapshot={} store={}",
303 module_dir.display(),
304 log_path.display(),
305 socket_path.display(),
306 snapshot_path
307 .as_ref()
308 .map(|p| p.display().to_string())
309 .unwrap_or_else(|| "<none>".into()),
310 store_path
311 .as_ref()
312 .map(|p| p.display().to_string())
313 .unwrap_or_else(|| "<memory>".into()),
314 );
315
316 // Sidecar brahman: nakui se presenta al Init mientras el daemon vive.
317 // No bloquea; si el Init no está, el sidecar termina silenciosamente.
318 brahman_sidecar::spawn(brahman_card_for_nakui());
319
320 let executor = Executor::load_module(&module_dir)
321 .map_err(|e| CliError::Op(format!("load module {}: {}", module_dir.display(), e)))?;
322 let log = EventLog::open(&log_path).map_err(|e| CliError::Op(format!("open log: {}", e)))?;
323 let snapshot = match &snapshot_path {
324 Some(p) => Some(
325 Snapshot::load(p)
326 .map_err(|e| CliError::Op(format!("load snapshot: {}", e)))?
327 .ok_or_else(|| {
328 CliError::Op(format!("snapshot file does not exist: {}", p.display()))
329 })?,
330 ),
331 None => None,
332 };
333
334 if let Some(p) = store_path {
335 run_persistent(executor, log, snapshot, &socket_path, &p)
336 } else {
337 let store = MemoryStore::new();
338 run_server(executor, log, store, snapshot, &socket_path)
339 .map_err(|e| CliError::Op(format!("run: {}", e)))
340 }
341}
342
343#[cfg(feature = "persistent")]
344fn run_persistent(
345 executor: Executor,
346 log: EventLog,
347 snapshot: Option<Snapshot>,
348 socket_path: &std::path::Path,
349 store_path: &std::path::Path,
350) -> Result<(), CliError> {
351 use nakui_core::surreal_store::SurrealStore;
352 let store = SurrealStore::new_persistent(store_path).map_err(|e| {
353 CliError::Op(format!(
354 "open persistent store at {}: {}",
355 store_path.display(),
356 e
357 ))
358 })?;
359 run_server(executor, log, store, snapshot, socket_path)
360 .map_err(|e| CliError::Op(format!("run: {}", e)))
361}
362
363#[cfg(not(feature = "persistent"))]
364fn run_persistent(
365 _executor: Executor,
366 _log: EventLog,
367 _snapshot: Option<Snapshot>,
368 _socket_path: &std::path::Path,
369 _store_path: &std::path::Path,
370) -> Result<(), CliError> {
371 Err(CliError::Op(
372 "--store-path requires building with `--features persistent`".into(),
373 ))
374}
375
376fn cmd_snapshot(args: &[String]) -> Result<(), CliError> {
377 let flags = parse_flags(args, &["log", "module", "out"])?;
378 let log_path = PathBuf::from(require(&flags, "log")?);
379 let module_dir = PathBuf::from(require(&flags, "module")?);
380 let out_path = PathBuf::from(require(&flags, "out")?);
381
382 let exec = Executor::load_module(&module_dir)
383 .map_err(|e| CliError::Op(format!("load module {}: {}", module_dir.display(), e)))?;
384 let log = EventLog::open(&log_path).map_err(|e| CliError::Op(format!("open log: {}", e)))?;
385 let mut store = MemoryStore::new();
386 replay_with_snapshot_into(&log, None, &mut store)
387 .map_err(|e| CliError::Op(format!("replay: {}", e)))?;
388 let last_seq = log
389 .entries()
390 .map_err(|e| CliError::Op(format!("read log: {}", e)))?
391 .last()
392 .map(|e| e.seq())
393 .ok_or_else(|| CliError::Op("log is empty; nothing to snapshot".into()))?;
394 let snap = Snapshot::capture(&store, last_seq, &exec);
395 snap.write(&out_path)
396 .map_err(|e| CliError::Op(format!("write snapshot: {}", e)))?;
397
398 let entity_count: usize = store.records().values().map(|m| m.len()).sum();
399 println!(
400 "snapshot written to {} (seq {}, {} records, schema {})",
401 out_path.display(),
402 last_seq,
403 entity_count,
404 short_hash(&hex_encode(&exec.module_schema_hash())),
405 );
406 Ok(())
407}
408
409fn cmd_compact(args: &[String]) -> Result<(), CliError> {
410 let flags = parse_flags(args, &["log", "snapshot"])?;
411 let log_path = PathBuf::from(require(&flags, "log")?);
412 let snap_path = PathBuf::from(require(&flags, "snapshot")?);
413
414 let snap = Snapshot::load(&snap_path)
415 .map_err(|e| CliError::Op(format!("load snapshot: {}", e)))?
416 .ok_or_else(|| CliError::Op(format!("snapshot not found: {}", snap_path.display())))?;
417 let mut log = EventLog::open(&log_path).map_err(|e| CliError::Op(format!("open log: {}", e)))?;
418 let before = log
419 .entries()
420 .map(|es| es.len())
421 .map_err(|e| CliError::Op(format!("read log: {}", e)))?;
422 log.compact_through(snap.seq)
423 .map_err(|e| CliError::Op(format!("compact: {}", e)))?;
424 let after = log
425 .entries()
426 .map(|es| es.len())
427 .map_err(|e| CliError::Op(format!("read log: {}", e)))?;
428 println!(
429 "compacted {} through seq {} ({} → {} entries; {} dropped)",
430 log_path.display(),
431 snap.seq,
432 before,
433 after,
434 before.saturating_sub(after),
435 );
436 Ok(())
437}
438
439fn cmd_verify_log(args: &[String]) -> Result<(), CliError> {
440 let flags = parse_flags(args, &["log", "module"])?;
441 let log_path = PathBuf::from(require(&flags, "log")?);
442 let module_dir = PathBuf::from(require(&flags, "module")?);
443
444 let exec = Executor::load_module(&module_dir)
445 .map_err(|e| CliError::Op(format!("load module {}: {}", module_dir.display(), e)))?;
446 let log = EventLog::open(&log_path).map_err(|e| CliError::Op(format!("open log: {}", e)))?;
447
448 match verify_log(&log, &exec) {
449 Ok(()) => {
450 let n = log
451 .entries()
452 .map(|es| es.len())
453 .map_err(|e| CliError::Op(format!("read log: {}", e)))?;
454 println!("ok: {} entries; every morphism reproduced its ops", n);
455 Ok(())
456 }
457 Err(e) => Err(CliError::Op(format!("verify failed: {}", e))),
458 }
459}
460
461/// Card que nakui presenta al Init brahman cuando arranca como daemon.
462///
463/// Lifecycle Daemon (proceso largo). Flujos JSON: consume `command`
464/// (queries del UI), produce `report` (resultados de cómputo). Los
465/// nombres están escogidos para que el broker pueda matchearlos contra
466/// `user-intent` / `render-data` de yahweh-shell por compatibilidad de
467/// tipo (todos `json`).
468fn brahman_card_for_nakui() -> brahman_card::Card {
469 use brahman_card::{
470 Card, Flow, Flows, FsPolicy, IpcPolicy, Lifecycle, Payload, Permissions, Priority,
471 Supervision, TypeRef, CARD_SCHEMA_VERSION,
472 };
473 use std::collections::BTreeSet;
474 use std::time::Duration;
475
476 Card {
477 schema_version: CARD_SCHEMA_VERSION,
478 id: ulid::Ulid::new(),
479 lineage: None,
480 label: "brahman.nakui_erp".into(),
481 provides: BTreeSet::new(),
482 requires: BTreeSet::new(),
483 payload: Payload::Virtual,
484 supervision: Supervision::Restart {
485 initial: Duration::from_millis(200),
486 max: Duration::from_secs(30),
487 },
488 lifecycle: Lifecycle::Daemon,
489 priority: Priority::Normal,
490 permissions: Permissions {
491 filesystem: FsPolicy::ReadWrite,
492 ipc: IpcPolicy {
493 allow: vec!["wit-v1".into()],
494 },
495 ..Default::default()
496 },
497 flow: Flows {
498 input: vec![Flow {
499 name: "command".into(),
500 ty: TypeRef::Primitive {
501 name: "json".into(),
502 },
503 pin_to: None,
504 }],
505 output: vec![Flow {
506 name: "report".into(),
507 ty: TypeRef::Primitive {
508 name: "json".into(),
509 },
510 pin_to: None,
511 }],
512 },
513 ..Default::default()
514 }
515}