From bf79d63be49d7b71cd7828f4738b8d4da402e007 Mon Sep 17 00:00:00 2001 From: Developer Date: Thu, 12 Feb 2026 18:03:30 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20command=20interface=20=E2=80=94=20front?= =?UTF-8?q?end-agnostic=20dispatch=20crate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add swactor-command crate that decouples command dispatch from the dashboard, enabling the same commands from REPL, REST, TUI, or any future frontend. - CommandRouter with dispatch(), CommandHandler trait, CommandMeta - CommandRequest/CommandResponse (serde-serializable JSON protocol) - CommandContext with Arc + optional StatsEnricher trait - 9 built-in commands extracted from investigate.rs: help, overview, workers, worker, actors, actor, hot, phases, diff, shutdown - REPL line parser (parse_line) with positional arg mapping - REST adapter (from_query_params) for HTTP query parameters - Dashboard integration: StatsEnricher impl for StatsCollector, investigate.rs refactored to thin CommandRouter wrapper, server.rs updated to share CommandRouter across handler threads - 18 behavioral tests in crates/command/tests/command_api.rs Co-Authored-By: Claude Opus 4.6 --- CLAUDE/notes/progress.md | 43 +- Cargo.lock | 38 +- Cargo.toml | 2 +- crates/command/Cargo.toml | 9 + crates/command/src/builtins.rs | 561 ++++++++++++++++++++ crates/command/src/lib.rs | 242 +++++++++ crates/command/src/parse.rs | 90 ++++ crates/command/tests/command_api.rs | 367 +++++++++++++ crates/runtime-dashboard/Cargo.toml | 1 + crates/runtime-dashboard/src/collector.rs | 6 + crates/runtime-dashboard/src/investigate.rs | 518 +----------------- crates/runtime-dashboard/src/server.rs | 32 +- 12 files changed, 1368 insertions(+), 541 deletions(-) create mode 100644 crates/command/Cargo.toml create mode 100644 crates/command/src/builtins.rs create mode 100644 crates/command/src/lib.rs create mode 100644 crates/command/src/parse.rs create mode 100644 crates/command/tests/command_api.rs diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index 12bc493..3e56c14 100644 --- a/CLAUDE/notes/progress.md +++ b/CLAUDE/notes/progress.md @@ -13,24 +13,39 @@ - Implemented `watch`/`unwatch` on `Runtime`'s `ContextInner` impl - Added `watch_registry` field to `TickContext` in `src/delivery.rs` - 10 behavioral tests in `tests/watch_api.rs` — all passing -- All 53 tests pass (42 existing + 10 new + 1 doctest) -- All 134 distribution crate tests pass + +### Feature 2: Command Interface — `docs/os-design/04-command-interface.md` +- Created `crates/command/` crate (`swactor-command`) with core types: + - `CommandRequest`, `CommandResponse` (serde-serializable) + - `CommandHandler` trait + `CommandMeta` + - `CommandRouter` with `dispatch()` and `with_builtins()` + - `CommandContext` with `Arc` + optional `StatsEnricher` + - `StatsEnricher` trait (decouples command crate from dashboard) +- Built-in read commands: overview, workers, worker, actors, actor, hot, phases, diff +- Built-in write command: shutdown +- REPL line parser (`parse_line`) with positional arg mapping and `--flag value` support +- REST adapter (`from_query_params`) for HTTP query parameters +- Refactored `investigate.rs` to delegate to CommandRouter (thin wrapper) +- Updated `server.rs` to use CommandRouter for `/api/investigate` endpoint +- Implemented `StatsEnricher for StatsCollector` in dashboard crate +- 18 behavioral tests in `crates/command/tests/command_api.rs` — all passing +- All 53 swactor core tests pass, all 18 command tests pass ## Next Steps -1. **Command Interface** — `docs/os-design/04-command-interface.md` - - Create `crates/command/` crate with `CommandHandler` trait, `CommandRouter`, `CommandRequest`/`CommandResponse` types - - Extract existing `cmd_*` functions from `crates/runtime-dashboard/src/investigate.rs` into CommandHandler impls - - Add built-in read commands (help, overview, nodes, workers, actors) - - Add built-in write commands (spawn, stop, send, drain, shutdown) - - Write behavioral tests -2. **Cluster Registry** — `docs/os-design/02-cluster-registry.md` -3. **Node Capabilities** — `docs/os-design/03-node-capabilities.md` -4. **Remote Watching** — extends 01-actor-watching with wire protocol -5. **Supervision** — `docs/os-design/05-supervision.md` +1. **Cluster Registry** — `docs/os-design/02-cluster-registry.md` + - LWW-Register CRDT per name binding + - Propagation via SWIM piggyback + - `ClusterRegistry` struct in `crates/distribution/src/registry.rs` + - API: register_name, unregister_name, resolve_name +2. **Node Capabilities** — `docs/os-design/03-node-capabilities.md` + - New `crates/capabilities/` crate with auto-detection +3. **Remote Watching** — extends actor watching with wire protocol +4. **Supervision** — `docs/os-design/05-supervision.md` ## Open Questions -- Command Interface: Should custom actor commands (via `Ctx::register_command()`) be deferred to a later PR? -- Should the dashboard migration happen in the same PR as the command crate creation? +- Custom actor commands (via `Ctx::register_command()`) deferred to a later PR +- Distribution-aware commands (nodes, registry, resolve) deferred until cluster registry is implemented +- Write commands (spawn, stop, drain) deferred — need factory registry and actor stop mechanism ## Blockers - None diff --git a/Cargo.lock b/Cargo.lock index 06fe0c8..8ff07c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1657,6 +1657,7 @@ dependencies = [ "serde", "serde_json", "swactor", + "swactor-command", "tiny_http", "tracing", "tracing-subscriber", @@ -1986,6 +1987,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "swactor-command" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "swactor", +] + [[package]] name = "swactor-std" version = "0.1.0" @@ -2005,9 +2015,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.114" +version = "2.0.115" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4d107df263a3013ef9b1879b0df87d706ff80f65a86ea879bd9c31f9b307c2a" +checksum = "6e614ed320ac28113fa64972c4262d5dbc89deacdfd00c34a3e4cea073243c12" dependencies = [ "proc-macro2", "quote", @@ -2408,12 +2418,12 @@ dependencies = [ [[package]] name = "wasm-encoder" -version = "0.245.0" +version = "0.245.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "95d568e113f706ee7a7df9b33547bb80721f55abffc79b3dc4d09c368690e662" +checksum = "3f9dca005e69bf015e45577e415b9af8c67e8ee3c0e38b5b0add5aa92581ed5c" dependencies = [ "leb128fmt", - "wasmparser 0.245.0", + "wasmparser 0.245.1", ] [[package]] @@ -2431,9 +2441,9 @@ dependencies = [ [[package]] name = "wasmparser" -version = "0.245.0" +version = "0.245.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48a767a48974f0c8b66f211b96e01aa77feed58b8ccce4e7f0cff0ae55b174d4" +checksum = "4f08c9adee0428b7bddf3890fc27e015ac4b761cc608c822667102b8bfd6995e" dependencies = [ "bitflags", "indexmap", @@ -2706,22 +2716,22 @@ dependencies = [ [[package]] name = "wast" -version = "245.0.0" +version = "245.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75ffc7471e16a6f3c7a3c3a230314915b5dcd158e5ef13ccda2f43358a9df00c" +checksum = "28cf1149285569120b8ce39db8b465e8a2b55c34cbb586bd977e43e2bc7300bf" dependencies = [ "bumpalo", "leb128fmt", "memchr", "unicode-width 0.2.0", - "wasm-encoder 0.245.0", + "wasm-encoder 0.245.1", ] [[package]] name = "wat" -version = "1.245.0" +version = "1.245.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6bcac6f915e2a84a4c0d9df9d41ad7518d99cda13f3bb83e3b8c22bf8726ab6" +checksum = "cd48d1679b6858988cb96b154dda0ec5bbb09275b71db46057be37332d5477be" dependencies = [ "wast", ] @@ -2934,9 +2944,9 @@ checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" [[package]] name = "zmij" -version = "1.0.20" +version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4de98dfa5d5b7fef4ee834d0073d560c9ca7b6c46a71d058c48db7960f8cfaf7" +checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" [[package]] name = "zstd" diff --git a/Cargo.toml b/Cargo.toml index f582294..9afb0f6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [".", "crates/python", "crates/wasm", "crates/wasm-actor", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/simulation-dashboard", "crates/std"] +members = [".", "crates/python", "crates/wasm", "crates/wasm-actor", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/simulation-dashboard", "crates/std", "crates/command"] exclude = ["tools/depgraph"] [package] diff --git a/crates/command/Cargo.toml b/crates/command/Cargo.toml new file mode 100644 index 0000000..d4f80fa --- /dev/null +++ b/crates/command/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "swactor-command" +version = "0.1.0" +edition = "2024" + +[dependencies] +swactor = { path = "../..", features = ["serde"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" diff --git a/crates/command/src/builtins.rs b/crates/command/src/builtins.rs new file mode 100644 index 0000000..42a06ac --- /dev/null +++ b/crates/command/src/builtins.rs @@ -0,0 +1,561 @@ +//! Built-in command handlers for runtime inspection and management. +//! +//! Extracted from `crates/runtime-dashboard/src/investigate.rs`. + +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +use swactor::actor::ActorAddress; +use swactor::stats::TickTiming; + +use crate::{CommandContext, CommandHandler, CommandMeta, CommandResponse}; + +// ─── Arg helpers ───────────────────────────────────────────────────────────── + +fn arg_str<'a>(args: &'a HashMap, key: &str) -> Option<&'a str> { + args.get(key).and_then(|v| v.as_str()) +} + +fn arg_usize(args: &HashMap, key: &str) -> Option { + args.get(key).and_then(|v| { + v.as_u64() + .map(|n| n as usize) + .or_else(|| v.as_str().and_then(|s| s.parse().ok())) + }) +} + +fn arg_f64(args: &HashMap, key: &str) -> Option { + args.get(key).and_then(|v| { + v.as_f64() + .or_else(|| v.as_str().and_then(|s| s.parse().ok())) + }) +} + +// ─── Display helpers ───────────────────────────────────────────────────────── + +fn format_addr(addr: &ActorAddress) -> String { + format!("{addr}") +} + +fn full_hex(addr: &ActorAddress) -> String { + addr.0.iter().map(|b| format!("{b:02x}")).collect() +} + +// ─── Phase breakdown helper ────────────────────────────────────────────────── + +fn compute_phase_breakdown(timings: &[TickTiming]) -> serde_json::Value { + if timings.is_empty() { + return serde_json::json!({ + "ticks": 0, + "active_pct": 0.0, + "avg_tick_us": 0.0, + "phases_us": [0, 0, 0, 0, 0, 0], + "phases_pct": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0], + }); + } + + let n = timings.len(); + let active = timings.iter().filter(|t| t.did_work).count(); + let active_pct = (active as f64 / n as f64) * 100.0; + + let mut phase_sums = [0u64; 6]; + for t in timings { + for (i, &us) in t.phase_us.iter().enumerate() { + phase_sums[i] += us; + } + } + let total_us: u64 = phase_sums.iter().sum(); + let avg_tick_us = total_us as f64 / n as f64; + + let phases_pct: Vec = if total_us == 0 { + vec![0.0; 6] + } else { + phase_sums + .iter() + .map(|&s| (s as f64 / total_us as f64) * 100.0) + .collect() + }; + + serde_json::json!({ + "ticks": n, + "active_pct": active_pct, + "avg_tick_us": avg_tick_us, + "phases_us": phase_sums, + "phases_pct": phases_pct, + }) +} + +// ─── Read Commands ─────────────────────────────────────────────────────────── + +pub struct OverviewCommand; + +impl CommandHandler for OverviewCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "overview", + description: "Summary: worker count, actor count, total messages, mailbox depth, panics", + usage: "overview", + is_write: false, + } + } + + fn handle( + &self, + _args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let stats = ctx.enriched_stats(); + let total_msgs: u64 = stats.workers.iter().map(|w| w.messages_processed).sum(); + let total_mailbox: usize = stats.workers.iter().map(|w| w.mailbox_depth).sum(); + let total_panics: u64 = stats.workers.iter().map(|w| w.panics).sum(); + let total_type_mismatches: u64 = stats.workers.iter().map(|w| w.type_mismatches).sum(); + let total_local: u64 = stats.workers.iter().map(|w| w.local_sends).sum(); + let total_cross: u64 = stats.workers.iter().map(|w| w.cross_sends).sum(); + let total_inbox: u64 = stats.workers.iter().map(|w| w.inbox_sends).sum(); + + CommandResponse::ok( + "overview", + serde_json::json!({ + "workers": stats.num_workers, + "actors": stats.actor_details.len(), + "total_messages_processed": total_msgs, + "total_mailbox_depth": total_mailbox, + "total_panics": total_panics, + "total_type_mismatches": total_type_mismatches, + "sends": { + "local": total_local, + "cross_worker": total_cross, + "inbox": total_inbox, + }, + }), + ) + } +} + +pub struct WorkersCommand; + +impl CommandHandler for WorkersCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "workers", + description: "Per-worker stats: actors, mailbox depth, messages, sends, panics", + usage: "workers", + is_write: false, + } + } + + fn handle( + &self, + _args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let stats = ctx.stats(); + let workers: Vec<_> = stats + .workers + .iter() + .map(|w| { + serde_json::json!({ + "id": w.id, + "actors": w.num_actors, + "mailbox_depth": w.mailbox_depth, + "messages_processed": w.messages_processed, + "local_sends": w.local_sends, + "cross_sends": w.cross_sends, + "inbox_sends": w.inbox_sends, + "type_mismatches": w.type_mismatches, + "panics": w.panics, + }) + }) + .collect(); + CommandResponse::ok("workers", workers) + } +} + +pub struct WorkerCommand; + +impl CommandHandler for WorkerCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "worker", + description: "Single worker detail with tick-phase timing breakdown", + usage: "worker ", + is_write: false, + } + } + + fn handle( + &self, + args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let id = match arg_usize(args, "id") { + Some(id) => id, + None => return CommandResponse::err("worker", "usage: worker "), + }; + + let stats = ctx.enriched_stats(); + let w = match stats.workers.iter().find(|w| w.id == id) { + Some(w) => w, + None => { + return CommandResponse::err( + "worker", + format!("worker {id} not found (have 0..{})", stats.num_workers), + ) + } + }; + + let timings = stats.tick_timings.get(id).cloned().unwrap_or_default(); + let phase_breakdown = compute_phase_breakdown(&timings); + + let actors_on_worker: Vec<_> = stats + .actor_details + .iter() + .filter(|a| a.worker_id == id) + .map(|a| { + serde_json::json!({ + "address": format_addr(&a.address), + "mailbox_depth": a.mailbox_depth, + "last_msg_type": a.last_msg_type, + "messages_processed": a.messages_processed, + "poisoned": a.poisoned, + }) + }) + .collect(); + + CommandResponse::ok( + "worker", + serde_json::json!({ + "id": w.id, + "actors": w.num_actors, + "mailbox_depth": w.mailbox_depth, + "messages_processed": w.messages_processed, + "local_sends": w.local_sends, + "cross_sends": w.cross_sends, + "inbox_sends": w.inbox_sends, + "type_mismatches": w.type_mismatches, + "panics": w.panics, + "tick_phases": phase_breakdown, + "actor_details": actors_on_worker, + }), + ) + } +} + +pub struct ActorsCommand; + +impl CommandHandler for ActorsCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "actors", + description: "List actors with optional sorting, limit, and worker filter", + usage: "actors [--sort mailbox|worker|address] [--limit N] [--worker W]", + is_write: false, + } + } + + fn handle( + &self, + args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let stats = ctx.enriched_stats(); + let mut actors = stats.actor_details.clone(); + + let sort_by = arg_str(args, "sort").unwrap_or("mailbox"); + let limit = arg_usize(args, "limit").unwrap_or(usize::MAX); + let worker_filter = arg_usize(args, "worker"); + + if let Some(wid) = worker_filter { + actors.retain(|a| a.worker_id == wid); + } + + match sort_by { + "mailbox" => actors.sort_by(|a, b| b.mailbox_depth.cmp(&a.mailbox_depth)), + "worker" => actors.sort_by_key(|a| a.worker_id), + "address" => actors.sort_by(|a, b| a.address.0.cmp(&b.address.0)), + other => { + return CommandResponse::err( + "actors", + format!("unknown sort field `{other}` — use mailbox|worker|address"), + ) + } + } + + actors.truncate(limit); + + let rows: Vec<_> = actors + .iter() + .map(|a| { + serde_json::json!({ + "address": format_addr(&a.address), + "address_full": full_hex(&a.address), + "worker_id": a.worker_id, + "mailbox_depth": a.mailbox_depth, + "last_msg_type": a.last_msg_type, + "messages_processed": a.messages_processed, + "poisoned": a.poisoned, + }) + }) + .collect(); + + CommandResponse::ok( + "actors", + serde_json::json!({ + "total": stats.actor_details.len(), + "returned": rows.len(), + "sort": sort_by, + "actors": rows, + }), + ) + } +} + +pub struct ActorCommand; + +impl CommandHandler for ActorCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "actor", + description: "Find actor(s) whose address starts with the given hex prefix", + usage: "actor ", + is_write: false, + } + } + + fn handle( + &self, + args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let prefix = match arg_str(args, "prefix") { + Some(p) => p, + None => return CommandResponse::err("actor", "usage: actor "), + }; + + let stats = ctx.enriched_stats(); + let matches: Vec<_> = stats + .actor_details + .iter() + .filter(|a| full_hex(&a.address).starts_with(prefix)) + .map(|a| { + serde_json::json!({ + "address": format_addr(&a.address), + "address_full": full_hex(&a.address), + "worker_id": a.worker_id, + "mailbox_depth": a.mailbox_depth, + "last_msg_type": a.last_msg_type, + "messages_processed": a.messages_processed, + "poisoned": a.poisoned, + }) + }) + .collect(); + + CommandResponse::ok( + "actor", + serde_json::json!({ + "prefix": prefix, + "matches": matches.len(), + "actors": matches, + }), + ) + } +} + +pub struct HotCommand; + +impl CommandHandler for HotCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "hot", + description: "Top N actors by mailbox depth (default 10)", + usage: "hot [N]", + is_write: false, + } + } + + fn handle( + &self, + args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let n = arg_usize(args, "n").unwrap_or(10); + let stats = ctx.enriched_stats(); + + let mut actors = stats.actor_details.clone(); + actors.sort_by(|a, b| b.mailbox_depth.cmp(&a.mailbox_depth)); + actors.truncate(n); + + let rows: Vec<_> = actors + .iter() + .map(|a| { + serde_json::json!({ + "address": format_addr(&a.address), + "address_full": full_hex(&a.address), + "worker_id": a.worker_id, + "mailbox_depth": a.mailbox_depth, + "last_msg_type": a.last_msg_type, + "messages_processed": a.messages_processed, + "poisoned": a.poisoned, + }) + }) + .collect(); + + CommandResponse::ok("hot", rows) + } +} + +pub struct PhasesCommand; + +impl CommandHandler for PhasesCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "phases", + description: "Tick-phase time breakdown (all workers or one)", + usage: "phases [worker_id]", + is_write: false, + } + } + + fn handle( + &self, + args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let stats = ctx.stats(); + let worker_filter = arg_usize(args, "worker"); + + let phase_names = [ + "spawn_drain", + "transfer_drain", + "tick_all", + "spawn_drain_2", + "pending_local", + "stats_publish", + ]; + + let mut results = Vec::new(); + for (i, timings) in stats.tick_timings.iter().enumerate() { + if let Some(wid) = worker_filter { + if i != wid { + continue; + } + } + let breakdown = compute_phase_breakdown(timings); + results.push(serde_json::json!({ + "worker_id": i, + "ticks_sampled": timings.len(), + "phases": breakdown, + "phase_names": phase_names, + })); + } + + CommandResponse::ok("phases", results) + } +} + +pub struct DiffCommand; + +impl CommandHandler for DiffCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "diff", + description: "Collect two snapshots N seconds apart, report deltas and rates", + usage: "diff ", + is_write: false, + } + } + + fn handle( + &self, + args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + let secs = match arg_f64(args, "seconds") { + Some(s) if s > 0.0 && s <= 30.0 => s, + Some(_) => return CommandResponse::err("diff", "seconds must be between 0 and 30"), + None => return CommandResponse::err("diff", "usage: diff "), + }; + + let before = ctx.enriched_stats(); + let t0 = Instant::now(); + std::thread::sleep(Duration::from_secs_f64(secs)); + let after = ctx.enriched_stats(); + let elapsed = t0.elapsed().as_secs_f64(); + + let msgs_before: u64 = before.workers.iter().map(|w| w.messages_processed).sum(); + let msgs_after: u64 = after.workers.iter().map(|w| w.messages_processed).sum(); + let delta_msgs = msgs_after.saturating_sub(msgs_before); + + let local_before: u64 = before.workers.iter().map(|w| w.local_sends).sum(); + let local_after: u64 = after.workers.iter().map(|w| w.local_sends).sum(); + let cross_before: u64 = before.workers.iter().map(|w| w.cross_sends).sum(); + let cross_after: u64 = after.workers.iter().map(|w| w.cross_sends).sum(); + + let mailbox_before: usize = before.workers.iter().map(|w| w.mailbox_depth).sum(); + let mailbox_after: usize = after.workers.iter().map(|w| w.mailbox_depth).sum(); + + let per_worker: Vec<_> = after + .workers + .iter() + .enumerate() + .map(|(i, w)| { + let prev = before.workers.get(i); + let d = prev + .map(|p| w.messages_processed.saturating_sub(p.messages_processed)) + .unwrap_or(0); + serde_json::json!({ + "worker_id": i, + "delta_messages": d, + "msg_per_sec": d as f64 / elapsed, + "actors_before": prev.map(|p| p.num_actors).unwrap_or(0), + "actors_after": w.num_actors, + "mailbox_before": prev.map(|p| p.mailbox_depth).unwrap_or(0), + "mailbox_after": w.mailbox_depth, + }) + }) + .collect(); + + CommandResponse::ok( + "diff", + serde_json::json!({ + "elapsed_s": elapsed, + "actors_before": before.actor_details.len(), + "actors_after": after.actor_details.len(), + "delta_messages": delta_msgs, + "msg_per_sec": delta_msgs as f64 / elapsed, + "delta_local_sends": local_after.saturating_sub(local_before), + "delta_cross_sends": cross_after.saturating_sub(cross_before), + "mailbox_before": mailbox_before, + "mailbox_after": mailbox_after, + "per_worker": per_worker, + }), + ) + } +} + +// ─── Write Commands ────────────────────────────────────────────────────────── + +pub struct ShutdownCommand; + +impl CommandHandler for ShutdownCommand { + fn meta(&self) -> CommandMeta { + CommandMeta { + name: "shutdown", + description: "Signal the runtime to shut down gracefully", + usage: "shutdown", + is_write: true, + } + } + + fn handle( + &self, + _args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse { + ctx.runtime.shutdown(); + CommandResponse::ok( + "shutdown", + serde_json::json!({"status": "shutdown signaled"}), + ) + } +} diff --git a/crates/command/src/lib.rs b/crates/command/src/lib.rs new file mode 100644 index 0000000..95fa287 --- /dev/null +++ b/crates/command/src/lib.rs @@ -0,0 +1,242 @@ +//! Frontend-agnostic command dispatch for swactor runtimes. +//! +//! Provides [`CommandRouter`] that maps command names to [`CommandHandler`] +//! implementations, with built-in commands for runtime inspection and management. +//! +//! # Architecture +//! +//! ```text +//! Frontend (REPL, REST, TUI, WebSocket) +//! │ +//! ▼ +//! CommandRouter::dispatch(CommandRequest, CommandContext) +//! │ +//! ├── built-in handlers (overview, workers, actors, …) +//! └── custom handlers (user-registered) +//! ``` + +pub mod builtins; +mod parse; + +pub use parse::{from_query_params, parse_line}; + +use std::collections::HashMap; +use std::sync::Arc; + +use serde::{Deserialize, Serialize}; +use swactor::runtime::Runtime; +use swactor::stats::RuntimeStats; + +// ─── Core Types ────────────────────────────────────────────────────────────── + +/// A command request. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CommandRequest { + pub command: String, + pub args: HashMap, +} + +/// A command response. Always JSON-serializable. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CommandResponse { + pub ok: bool, + pub command: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +impl CommandResponse { + pub fn ok(command: &str, data: impl Serialize) -> Self { + Self { + ok: true, + command: command.to_string(), + data: Some(serde_json::to_value(data).unwrap_or(serde_json::Value::Null)), + error: None, + } + } + + pub fn err(command: &str, msg: impl Into) -> Self { + Self { + ok: false, + command: command.to_string(), + data: None, + error: Some(msg.into()), + } + } + + /// Serialize to a single JSON line (for REPL/wire protocol). + pub fn to_json_line(&self) -> String { + serde_json::to_string(self).unwrap_or_else(|e| { + format!(r#"{{"ok":false,"command":"","error":"serialization: {e}"}}"#) + }) + } +} + +// ─── Handler Trait ─────────────────────────────────────────────────────────── + +/// Metadata about a command, used for help text and validation. +pub struct CommandMeta { + pub name: &'static str, + pub description: &'static str, + pub usage: &'static str, + pub is_write: bool, +} + +/// A command handler. Implementations are stateless — all state +/// comes through [`CommandContext`]. +pub trait CommandHandler: Send + Sync { + fn meta(&self) -> CommandMeta; + fn handle( + &self, + args: &HashMap, + ctx: &CommandContext, + ) -> CommandResponse; +} + +// ─── Stats Enrichment ──────────────────────────────────────────────────────── + +/// Enriches [`RuntimeStats`] with per-actor detail. +/// +/// Implement this on your stats collector so command handlers can access +/// enriched data without depending on the dashboard crate. +pub trait StatsEnricher: Send + Sync { + fn enrich(&self, stats: &mut RuntimeStats); +} + +// ─── Context ───────────────────────────────────────────────────────────────── + +/// Context available to command handlers. +pub struct CommandContext { + pub runtime: Arc, + pub enricher: Option>, +} + +impl CommandContext { + pub fn new(runtime: Arc) -> Self { + Self { + runtime, + enricher: None, + } + } + + pub fn with_enricher( + runtime: Arc, + enricher: Arc, + ) -> Self { + Self { + runtime, + enricher: Some(enricher), + } + } + + /// Raw runtime stats (no enrichment). + pub fn stats(&self) -> RuntimeStats { + self.runtime.stats() + } + + /// Runtime stats enriched with per-actor detail (if an enricher is set). + pub fn enriched_stats(&self) -> RuntimeStats { + let mut s = self.runtime.stats(); + if let Some(e) = &self.enricher { + e.enrich(&mut s); + } + s + } +} + +// ─── Router ────────────────────────────────────────────────────────────────── + +/// Central command dispatch. +pub struct CommandRouter { + handlers: HashMap>, +} + +impl CommandRouter { + pub fn new() -> Self { + Self { + handlers: HashMap::new(), + } + } + + /// Create a router with all built-in commands registered. + pub fn with_builtins() -> Self { + let mut router = Self::new(); + router.register(Box::new(builtins::OverviewCommand)); + router.register(Box::new(builtins::WorkersCommand)); + router.register(Box::new(builtins::WorkerCommand)); + router.register(Box::new(builtins::ActorsCommand)); + router.register(Box::new(builtins::ActorCommand)); + router.register(Box::new(builtins::HotCommand)); + router.register(Box::new(builtins::PhasesCommand)); + router.register(Box::new(builtins::DiffCommand)); + router.register(Box::new(builtins::ShutdownCommand)); + router + } + + /// Register a custom command handler. + pub fn register(&mut self, handler: Box) { + let name = handler.meta().name.to_string(); + self.handlers.insert(name, handler); + } + + /// Dispatch a command request. + /// + /// The `help` command is handled directly by the router (it needs + /// access to all registered handlers). + pub fn dispatch( + &self, + req: &CommandRequest, + ctx: &CommandContext, + ) -> CommandResponse { + if req.command == "help" { + return self.cmd_help(); + } + match self.handlers.get(&req.command) { + Some(handler) => handler.handle(&req.args, ctx), + None => CommandResponse::err( + &req.command, + format!("unknown command `{}` — try `help`", req.command), + ), + } + } + + fn cmd_help(&self) -> CommandResponse { + let mut commands: Vec = self + .handlers + .values() + .map(|h| { + let m = h.meta(); + serde_json::json!({ + "name": m.name, + "usage": m.usage, + "description": m.description, + "is_write": m.is_write, + }) + }) + .collect(); + // Add help itself + commands.push(serde_json::json!({ + "name": "help", + "usage": "help", + "description": "List all available commands", + "is_write": false, + })); + commands.sort_by(|a, b| { + a["name"] + .as_str() + .unwrap_or("") + .cmp(b["name"].as_str().unwrap_or("")) + }); + CommandResponse::ok("help", serde_json::json!({ "commands": commands })) + } + + /// List names of all registered commands (sorted). + pub fn command_names(&self) -> Vec<&str> { + let mut names: Vec<_> = self.handlers.keys().map(|s| s.as_str()).collect(); + names.push("help"); + names.sort(); + names + } +} diff --git a/crates/command/src/parse.rs b/crates/command/src/parse.rs new file mode 100644 index 0000000..7703da5 --- /dev/null +++ b/crates/command/src/parse.rs @@ -0,0 +1,90 @@ +//! Input parsers for REPL lines and HTTP query parameters. + +use std::collections::HashMap; + +use crate::CommandRequest; + +/// Parse a REPL text line into a [`CommandRequest`]. +/// +/// Handles `--flag value` pairs and maps positional arguments to +/// command-specific named parameters. +/// +/// # Examples +/// +/// ```text +/// "overview" → { command: "overview", args: {} } +/// "worker 3" → { command: "worker", args: { "id": "3" } } +/// "actors --sort mailbox" → { command: "actors", args: { "sort": "mailbox" } } +/// "hot 5" → { command: "hot", args: { "n": "5" } } +/// ``` +pub fn parse_line(line: &str) -> CommandRequest { + let parts: Vec<&str> = line.split_whitespace().collect(); + if parts.is_empty() { + return CommandRequest { + command: "help".to_string(), + args: HashMap::new(), + }; + } + let command = parts[0].to_string(); + let rest = &parts[1..]; + + let mut args = HashMap::new(); + let mut i = 0; + let mut positional = 0; + + while i < rest.len() { + if let Some(key) = rest[i].strip_prefix("--") { + if i + 1 < rest.len() && !rest[i + 1].starts_with("--") { + args.insert( + key.to_string(), + serde_json::Value::String(rest[i + 1].to_string()), + ); + i += 2; + } else { + args.insert(key.to_string(), serde_json::Value::Bool(true)); + i += 1; + } + } else { + let name = positional_arg_name(&command, positional); + if !name.is_empty() { + args.insert( + name.to_string(), + serde_json::Value::String(rest[i].to_string()), + ); + } + positional += 1; + i += 1; + } + } + + CommandRequest { command, args } +} + +/// Convert HTTP query parameters to a [`CommandRequest`]. +/// +/// The `cmd` parameter becomes the command name; all other parameters +/// become string-valued arguments. +pub fn from_query_params(params: &HashMap) -> CommandRequest { + let command = params + .get("cmd") + .cloned() + .unwrap_or_else(|| "help".into()); + let args: HashMap = params + .iter() + .filter(|(k, _)| *k != "cmd") + .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone()))) + .collect(); + CommandRequest { command, args } +} + +/// Map positional argument index to the named parameter for each command. +fn positional_arg_name(command: &str, position: usize) -> &'static str { + match (command, position) { + ("worker", 0) => "id", + ("actor", 0) => "prefix", + ("hot", 0) => "n", + ("phases", 0) => "worker", + ("diff", 0) => "seconds", + _ => "", + } +} diff --git a/crates/command/tests/command_api.rs b/crates/command/tests/command_api.rs new file mode 100644 index 0000000..4e6e33d --- /dev/null +++ b/crates/command/tests/command_api.rs @@ -0,0 +1,367 @@ +//! Behavioral tests for the swactor-command crate. +//! +//! Tests exercise the full dispatch path: parse → route → handle → response. + +use std::collections::HashMap; +use std::sync::Arc; + +use swactor::actor::ActorInterface; +use swactor::runtime::{Ctx, Runtime, RuntimeConfig}; +use swactor_command::{ + from_query_params, parse_line, CommandContext, CommandRequest, CommandResponse, CommandRouter, +}; + +// ── Test Helpers ───────────────────────────────────────────────────────────── + +fn single_thread_config() -> RuntimeConfig { + RuntimeConfig { + num_threads: 1, + ..RuntimeConfig::default() + } +} + +fn make_router_and_ctx() -> (CommandRouter, CommandContext) { + let rt = Arc::new(Runtime::new(single_thread_config())); + let router = CommandRouter::with_builtins(); + let ctx = CommandContext::new(rt); + (router, ctx) +} + +fn dispatch_text(router: &CommandRouter, ctx: &CommandContext, line: &str) -> CommandResponse { + let req = parse_line(line); + let resp = router.dispatch(&req, ctx); + // Verify JSON round-trip works + let json = resp.to_json_line(); + serde_json::from_str::(&json) + .expect("response should be valid JSON") +} + +/// A no-op actor for spawning into the runtime. +struct DummyActor; +#[derive(Clone)] +struct DummyMsg; +impl ActorInterface for DummyActor { + type Incoming = DummyMsg; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: DummyMsg) {} +} + +// ── Tests ──────────────────────────────────────────────────────────────────── + +/// Given a router with builtins, +/// when "help" is dispatched, +/// then the response lists all registered commands. +#[test] +fn help_lists_all_registered_commands() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "help"); + + assert!(resp.ok, "help should succeed"); + assert_eq!(resp.command, "help"); + + let data = resp.data.unwrap(); + let commands = data["commands"].as_array().unwrap(); + + // Should have all builtins + help itself + let names: Vec<&str> = commands + .iter() + .map(|c| c["name"].as_str().unwrap()) + .collect(); + assert!(names.contains(&"overview"), "should list overview"); + assert!(names.contains(&"workers"), "should list workers"); + assert!(names.contains(&"worker"), "should list worker"); + assert!(names.contains(&"actors"), "should list actors"); + assert!(names.contains(&"hot"), "should list hot"); + assert!(names.contains(&"phases"), "should list phases"); + assert!(names.contains(&"diff"), "should list diff"); + assert!(names.contains(&"shutdown"), "should list shutdown"); + assert!(names.contains(&"help"), "should list help itself"); + + // Should be sorted + let mut sorted = names.clone(); + sorted.sort(); + assert_eq!(names, sorted, "commands should be sorted alphabetically"); +} + +/// Given a router, +/// when an unknown command is dispatched, +/// then the response indicates failure with a helpful message. +#[test] +fn unknown_command_returns_error() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "nonexistent"); + + assert!(!resp.ok, "unknown command should fail"); + assert_eq!(resp.command, "nonexistent"); + let err = resp.error.unwrap(); + assert!( + err.contains("unknown command") && err.contains("help"), + "error should mention 'unknown command' and suggest 'help', got: {err}" + ); +} + +/// Given a runtime with no actors, +/// when "overview" is dispatched, +/// then the response contains expected summary fields with zero counts. +#[test] +fn overview_returns_summary_fields() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "overview"); + + assert!(resp.ok); + assert_eq!(resp.command, "overview"); + + let data = resp.data.unwrap(); + assert_eq!(data["workers"], 1, "single-threaded = 1 worker"); + assert_eq!(data["actors"], 0, "no actors spawned"); + assert_eq!(data["total_messages_processed"], 0); + assert_eq!(data["total_panics"], 0); + assert!(data["sends"].is_object(), "sends should be an object"); +} + +/// Given a runtime with spawned actors, +/// when "workers" is dispatched, +/// then the response contains per-worker stats. +#[test] +fn workers_returns_per_worker_info() { + let rt = Arc::new(Runtime::new(single_thread_config())); + // Spawn some actors + rt.spawn(DummyActor).unwrap(); + rt.spawn(DummyActor).unwrap(); + rt.tick(); + + let router = CommandRouter::with_builtins(); + let ctx = CommandContext::new(rt); + let resp = dispatch_text(&router, &ctx, "workers"); + + assert!(resp.ok); + let data = resp.data.unwrap(); + let workers = data.as_array().unwrap(); + assert_eq!(workers.len(), 1, "single-threaded has 1 worker"); + assert_eq!(workers[0]["id"], 0); + assert_eq!(workers[0]["actors"], 2, "2 actors spawned on worker 0"); +} + +/// Given "worker 0" with a valid ID, +/// when dispatched, +/// then the response includes worker detail and tick phase info. +#[test] +fn worker_command_with_valid_id() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "worker 0"); + + assert!(resp.ok); + assert_eq!(resp.command, "worker"); + let data = resp.data.unwrap(); + assert_eq!(data["id"], 0); + assert!(data["tick_phases"].is_object(), "should include phase breakdown"); +} + +/// Given "worker 99", +/// when dispatched, +/// then the response is an error (worker not found). +#[test] +fn worker_command_invalid_id_returns_error() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "worker 99"); + + assert!(!resp.ok); + assert!(resp.error.unwrap().contains("not found")); +} + +/// Given "worker" with no ID, +/// when dispatched, +/// then the response is a usage error. +#[test] +fn worker_command_missing_id_returns_usage() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "worker"); + + assert!(!resp.ok); + assert!(resp.error.unwrap().contains("usage")); +} + +/// Given "phases", +/// when dispatched, +/// then the response includes phase breakdown per worker. +#[test] +fn phases_command_returns_breakdown() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "phases"); + + assert!(resp.ok); + let data = resp.data.unwrap(); + let phases = data.as_array().unwrap(); + assert_eq!(phases.len(), 1, "single-threaded = 1 worker"); + assert_eq!(phases[0]["worker_id"], 0); +} + +/// Given a runtime, when "shutdown" is dispatched, +/// then the response indicates success. +#[test] +fn shutdown_command_signals_runtime() { + let (router, ctx) = make_router_and_ctx(); + let resp = dispatch_text(&router, &ctx, "shutdown"); + + assert!(resp.ok); + assert_eq!(resp.command, "shutdown"); + let data = resp.data.unwrap(); + assert_eq!(data["status"], "shutdown signaled"); +} + +// ── REPL Parser Tests ──────────────────────────────────────────────────────── + +/// Given a simple command with no args, +/// when parsed, +/// then the command name is extracted correctly. +#[test] +fn parse_line_simple_command() { + let req = parse_line("overview"); + assert_eq!(req.command, "overview"); + assert!(req.args.is_empty()); +} + +/// Given a command with positional args, +/// when parsed, +/// then positional args are mapped to named parameters. +#[test] +fn parse_line_positional_args() { + let req = parse_line("worker 3"); + assert_eq!(req.command, "worker"); + assert_eq!(req.args["id"], "3"); + + let req = parse_line("hot 5"); + assert_eq!(req.command, "hot"); + assert_eq!(req.args["n"], "5"); + + let req = parse_line("diff 2.5"); + assert_eq!(req.command, "diff"); + assert_eq!(req.args["seconds"], "2.5"); + + let req = parse_line("actor a1b2"); + assert_eq!(req.command, "actor"); + assert_eq!(req.args["prefix"], "a1b2"); +} + +/// Given a command with --flag value pairs, +/// when parsed, +/// then flags are mapped to named args. +#[test] +fn parse_line_flags() { + let req = parse_line("actors --sort mailbox --limit 5"); + assert_eq!(req.command, "actors"); + assert_eq!(req.args["sort"], "mailbox"); + assert_eq!(req.args["limit"], "5"); +} + +/// Given a command with mixed positional and flag args, +/// when parsed, +/// then both are captured correctly. +#[test] +fn parse_line_mixed_args() { + let req = parse_line("actors --sort worker --worker 2 --limit 10"); + assert_eq!(req.command, "actors"); + assert_eq!(req.args["sort"], "worker"); + assert_eq!(req.args["worker"], "2"); + assert_eq!(req.args["limit"], "10"); +} + +/// Given empty input, +/// when parsed, +/// then default to "help". +#[test] +fn parse_line_empty_defaults_to_help() { + let req = parse_line(""); + assert_eq!(req.command, "help"); +} + +// ── REST Adapter Tests ─────────────────────────────────────────────────────── + +/// Given query params with cmd and other params, +/// when converted, +/// then cmd becomes the command and others become args. +#[test] +fn from_query_params_extracts_cmd() { + let mut params = HashMap::new(); + params.insert("cmd".to_string(), "actor".to_string()); + params.insert("prefix".to_string(), "a1b2".to_string()); + + let req = from_query_params(¶ms); + assert_eq!(req.command, "actor"); + assert_eq!(req.args["prefix"], "a1b2"); + assert!(!req.args.contains_key("cmd"), "cmd should not be in args"); +} + +/// Given query params with no cmd, +/// when converted, +/// then default to "help". +#[test] +fn from_query_params_defaults_to_help() { + let params = HashMap::new(); + let req = from_query_params(¶ms); + assert_eq!(req.command, "help"); +} + +// ── Custom Handler Test ────────────────────────────────────────────────────── + +/// Given a custom command handler registered on the router, +/// when that command is dispatched, +/// then the custom handler runs and returns its response. +#[test] +fn custom_command_handler() { + struct PingCommand; + impl swactor_command::CommandHandler for PingCommand { + fn meta(&self) -> swactor_command::CommandMeta { + swactor_command::CommandMeta { + name: "ping", + description: "Respond with pong", + usage: "ping", + is_write: false, + } + } + fn handle( + &self, + _args: &HashMap, + _ctx: &CommandContext, + ) -> CommandResponse { + CommandResponse::ok("ping", serde_json::json!({"reply": "pong"})) + } + } + + let rt = Arc::new(Runtime::new(single_thread_config())); + let mut router = CommandRouter::with_builtins(); + router.register(Box::new(PingCommand)); + let ctx = CommandContext::new(rt); + + let resp = dispatch_text(&router, &ctx, "ping"); + assert!(resp.ok); + assert_eq!(resp.data.unwrap()["reply"], "pong"); + + // Should also appear in help + let help = dispatch_text(&router, &ctx, "help"); + let commands = help.data.unwrap()["commands"].as_array().unwrap().clone(); + let names: Vec<&str> = commands.iter().map(|c| c["name"].as_str().unwrap()).collect(); + assert!(names.contains(&"ping"), "custom command should appear in help"); +} + +/// Given a JSON-serialized CommandResponse, +/// when deserialized, +/// then ok/false fields, data, and error are preserved. +#[test] +fn response_json_roundtrip() { + let ok_resp = CommandResponse::ok("test", serde_json::json!({"key": "value"})); + let json = ok_resp.to_json_line(); + let parsed: CommandResponse = serde_json::from_str(&json).unwrap(); + assert!(parsed.ok); + assert_eq!(parsed.command, "test"); + assert_eq!(parsed.data.unwrap()["key"], "value"); + assert!(parsed.error.is_none()); + + let err_resp = CommandResponse::err("bad", "something went wrong"); + let json = err_resp.to_json_line(); + let parsed: CommandResponse = serde_json::from_str(&json).unwrap(); + assert!(!parsed.ok); + assert_eq!(parsed.command, "bad"); + assert!(parsed.data.is_none()); + assert_eq!(parsed.error.unwrap(), "something went wrong"); +} diff --git a/crates/runtime-dashboard/Cargo.toml b/crates/runtime-dashboard/Cargo.toml index 3e24adb..1894aa7 100644 --- a/crates/runtime-dashboard/Cargo.toml +++ b/crates/runtime-dashboard/Cargo.toml @@ -14,6 +14,7 @@ crossbeam-queue = "0.3.12" ratatui = { version = "0.29", optional = true, default-features = false, features = ["crossterm"] } crossterm = { version = "0.28", optional = true } distribution = { path = "../distribution", optional = true } +swactor-command = { path = "../command" } [dependencies.ctrlc] version = "3" diff --git a/crates/runtime-dashboard/src/collector.rs b/crates/runtime-dashboard/src/collector.rs index 59df4d0..ab1dcd7 100644 --- a/crates/runtime-dashboard/src/collector.rs +++ b/crates/runtime-dashboard/src/collector.rs @@ -42,6 +42,12 @@ impl StatsCollector { } } +impl swactor_command::StatsEnricher for StatsCollector { + fn enrich(&self, stats: &mut swactor::stats::RuntimeStats) { + stats.actor_details = self.actor_details(); + } +} + impl StatsHook for StatsCollector { fn on_tick(&self, worker_id: usize, snapshots: &[ActorSnapshot]) { if let Some(slot) = self.slots.get(worker_id) { diff --git a/crates/runtime-dashboard/src/investigate.rs b/crates/runtime-dashboard/src/investigate.rs index 32bf9b8..cd81995 100644 --- a/crates/runtime-dashboard/src/investigate.rs +++ b/crates/runtime-dashboard/src/investigate.rs @@ -1,5 +1,7 @@ //! Line-oriented diagnostic protocol for LLM-driven runtime investigation. //! +//! Delegates all command logic to the `swactor-command` crate. +//! //! Send text commands on stdin, receive JSON responses on stdout (one per line). //! All human-readable diagnostics go to stderr. //! @@ -19,16 +21,17 @@ use std::collections::HashMap; use std::io::{self, BufRead, Write}; use std::sync::Arc; -use std::time::{Duration, Instant}; -use serde::Serialize; use swactor::runtime::Runtime; -use swactor::stats::RuntimeStats; +use swactor_command::{CommandContext, CommandRouter}; use crate::collector::StatsCollector; /// Run the investigate REPL. Blocks until stdin is closed or `quit` is received. pub fn run_investigate(runtime: Arc, collector: Arc) -> io::Result<()> { + let router = CommandRouter::with_builtins(); + let ctx = CommandContext::with_enricher(runtime, collector); + let stdin = io::stdin(); let mut stdout = io::stdout(); @@ -40,18 +43,14 @@ pub fn run_investigate(runtime: Arc, collector: Arc) -> if line.is_empty() { continue; } - - let parts: Vec<&str> = line.split_whitespace().collect(); - let cmd = parts[0]; - let args = &parts[1..]; - - if cmd == "quit" || cmd == "exit" { + if line == "quit" || line == "exit" { break; } - let response = dispatch_repl(cmd, args, &runtime, &collector); + let req = swactor_command::parse_line(line); + let resp = router.dispatch(&req, &ctx); - stdout.write_all(response.as_bytes())?; + stdout.write_all(resp.to_json_line().as_bytes())?; stdout.write_all(b"\n")?; stdout.flush()?; } @@ -59,501 +58,14 @@ pub fn run_investigate(runtime: Arc, collector: Arc) -> Ok(()) } -fn dispatch_repl(cmd: &str, args: &[&str], runtime: &Runtime, collector: &StatsCollector) -> String { - match cmd { - "help" => cmd_help(), - "overview" => cmd_overview(runtime, collector), - "workers" => cmd_workers(runtime), - "worker" => cmd_worker(runtime, collector, args), - "actors" => cmd_actors(runtime, collector, args), - "actor" => cmd_actor(runtime, collector, args), - "hot" => cmd_hot(runtime, collector, args), - "phases" => cmd_phases(runtime, args), - "diff" => cmd_diff(runtime, collector, args), - _ => err_response(cmd, &format!("unknown command `{cmd}` — try `help`")), - } -} - /// Dispatch an investigate command from HTTP query parameters. /// -/// Maps `?cmd=overview`, `?cmd=hot&n=10`, etc. to the appropriate command function. +/// Maps `?cmd=overview`, `?cmd=hot&n=10`, etc. to the appropriate command. pub fn dispatch_command( - cmd: &str, params: &HashMap, - runtime: &Runtime, - collector: &StatsCollector, + router: &CommandRouter, + ctx: &CommandContext, ) -> String { - match cmd { - "help" => cmd_help(), - "overview" => cmd_overview(runtime, collector), - "workers" => cmd_workers(runtime), - "worker" => { - let id = params.get("id").map(|s| s.as_str()).unwrap_or(""); - cmd_worker(runtime, collector, &[id]) - } - "actors" => { - let mut args = Vec::new(); - if let Some(sort) = params.get("sort") { - args.push("--sort"); - args.push(sort.as_str()); - } - if let Some(limit) = params.get("limit") { - args.push("--limit"); - args.push(limit.as_str()); - } - if let Some(worker) = params.get("worker") { - args.push("--worker"); - args.push(worker.as_str()); - } - cmd_actors(runtime, collector, &args) - } - "actor" => { - let prefix = params.get("prefix").map(|s| s.as_str()).unwrap_or(""); - cmd_actor(runtime, collector, &[prefix]) - } - "hot" => { - let n = params.get("n").map(|s| s.as_str()).unwrap_or("10"); - cmd_hot(runtime, collector, &[n]) - } - "phases" => { - match params.get("worker") { - Some(w) => cmd_phases(runtime, &[w.as_str()]), - None => cmd_phases(runtime, &[]), - } - } - "diff" => { - let secs = params.get("seconds").map(|s| s.as_str()).unwrap_or(""); - cmd_diff(runtime, collector, &[secs]) - } - _ => err_response(cmd, &format!("unknown command `{cmd}` — try `help`")), - } -} - -// ── Helpers ───────────────────────────────────────────────────────────── - -fn ok_response(cmd: &str, data: impl Serialize) -> String { - serde_json::to_string(&serde_json::json!({ - "ok": true, - "command": cmd, - "data": data, - })) - .unwrap_or_else(|e| err_response(cmd, &format!("serialization error: {e}"))) -} - -fn err_response(cmd: &str, msg: &str) -> String { - serde_json::to_string(&serde_json::json!({ - "ok": false, - "command": cmd, - "error": msg, - })) - .unwrap() -} - -fn format_addr(addr: &swactor::actor::ActorAddress) -> String { - format!("{addr}") -} - -fn full_hex(addr: &swactor::actor::ActorAddress) -> String { - addr.0.iter().map(|b| format!("{b:02x}")).collect() -} - -fn enriched_stats(rt: &Runtime, col: &StatsCollector) -> RuntimeStats { - let mut s = rt.stats(); - col.enrich(&mut s); - s -} - -// ── Commands ──────────────────────────────────────────────────────────── - -pub fn cmd_help() -> String { - ok_response( - "help", - serde_json::json!({ - "commands": [ - {"name": "overview", "usage": "overview", "description": "Summary: worker count, actor count, total messages, mailbox depth, panics"}, - {"name": "workers", "usage": "workers", "description": "Per-worker stats: actors, mailbox depth, messages, sends (local/cross/inbox), panics"}, - {"name": "worker", "usage": "worker ", "description": "Single worker detail with tick-phase timing breakdown"}, - {"name": "actors", "usage": "actors [--sort mailbox|worker|address] [--limit N] [--worker W]", "description": "List actors with optional sorting, limit, and worker filter"}, - {"name": "actor", "usage": "actor ", "description": "Find actor(s) whose address starts with the given hex prefix"}, - {"name": "hot", "usage": "hot [N]", "description": "Top N actors by mailbox depth (default 10)"}, - {"name": "phases", "usage": "phases [worker_id]", "description": "Tick-phase time breakdown (all workers or one)"}, - {"name": "diff", "usage": "diff ", "description": "Collect two snapshots N seconds apart, report deltas and rates"}, - {"name": "quit", "usage": "quit", "description": "Exit the investigate session"}, - ] - }), - ) -} - -pub fn cmd_overview(rt: &Runtime, col: &StatsCollector) -> String { - let stats = enriched_stats(rt, col); - let total_msgs: u64 = stats.workers.iter().map(|w| w.messages_processed).sum(); - let total_mailbox: usize = stats.workers.iter().map(|w| w.mailbox_depth).sum(); - let total_panics: u64 = stats.workers.iter().map(|w| w.panics).sum(); - let total_type_mismatches: u64 = stats.workers.iter().map(|w| w.type_mismatches).sum(); - let total_local: u64 = stats.workers.iter().map(|w| w.local_sends).sum(); - let total_cross: u64 = stats.workers.iter().map(|w| w.cross_sends).sum(); - let total_inbox: u64 = stats.workers.iter().map(|w| w.inbox_sends).sum(); - - ok_response( - "overview", - serde_json::json!({ - "workers": stats.num_workers, - "actors": stats.actor_details.len(), - "total_messages_processed": total_msgs, - "total_mailbox_depth": total_mailbox, - "total_panics": total_panics, - "total_type_mismatches": total_type_mismatches, - "sends": { - "local": total_local, - "cross_worker": total_cross, - "inbox": total_inbox, - }, - }), - ) -} - -pub fn cmd_workers(rt: &Runtime) -> String { - let stats = rt.stats(); - let workers: Vec<_> = stats - .workers - .iter() - .map(|w| { - serde_json::json!({ - "id": w.id, - "actors": w.num_actors, - "mailbox_depth": w.mailbox_depth, - "messages_processed": w.messages_processed, - "local_sends": w.local_sends, - "cross_sends": w.cross_sends, - "inbox_sends": w.inbox_sends, - "type_mismatches": w.type_mismatches, - "panics": w.panics, - }) - }) - .collect(); - ok_response("workers", workers) -} - -pub fn cmd_worker(rt: &Runtime, col: &StatsCollector, args: &[&str]) -> String { - let id: usize = match args.first().and_then(|s| s.parse().ok()) { - Some(id) => id, - None => return err_response("worker", "usage: worker "), - }; - - let stats = enriched_stats(rt, col); - let w = match stats.workers.iter().find(|w| w.id == id) { - Some(w) => w, - None => { - return err_response( - "worker", - &format!("worker {id} not found (have 0..{})", stats.num_workers), - ) - } - }; - - // Tick phase breakdown for this worker - let timings = stats.tick_timings.get(id).cloned().unwrap_or_default(); - let phase_breakdown = compute_phase_breakdown(&timings); - - let actors_on_worker: Vec<_> = stats - .actor_details - .iter() - .filter(|a| a.worker_id == id) - .map(|a| { - serde_json::json!({ - "address": format_addr(&a.address), - "mailbox_depth": a.mailbox_depth, - "last_msg_type": a.last_msg_type, - "messages_processed": a.messages_processed, - "poisoned": a.poisoned, - }) - }) - .collect(); - - ok_response( - "worker", - serde_json::json!({ - "id": w.id, - "actors": w.num_actors, - "mailbox_depth": w.mailbox_depth, - "messages_processed": w.messages_processed, - "local_sends": w.local_sends, - "cross_sends": w.cross_sends, - "inbox_sends": w.inbox_sends, - "type_mismatches": w.type_mismatches, - "panics": w.panics, - "tick_phases": phase_breakdown, - "actor_details": actors_on_worker, - }), - ) -} - -pub fn cmd_actors(rt: &Runtime, col: &StatsCollector, args: &[&str]) -> String { - let stats = enriched_stats(rt, col); - let mut actors = stats.actor_details.clone(); - - // Parse flags - let mut sort_by = "mailbox"; - let mut limit: usize = usize::MAX; - let mut worker_filter: Option = None; - let mut i = 0; - while i < args.len() { - match args[i] { - "--sort" if i + 1 < args.len() => { - sort_by = args[i + 1]; - i += 2; - } - "--limit" if i + 1 < args.len() => { - limit = args[i + 1].parse().unwrap_or(usize::MAX); - i += 2; - } - "--worker" if i + 1 < args.len() => { - worker_filter = args[i + 1].parse().ok(); - i += 2; - } - _ => { - i += 1; - } - } - } - - if let Some(wid) = worker_filter { - actors.retain(|a| a.worker_id == wid); - } - - match sort_by { - "mailbox" => actors.sort_by(|a, b| b.mailbox_depth.cmp(&a.mailbox_depth)), - "worker" => actors.sort_by_key(|a| a.worker_id), - "address" => actors.sort_by(|a, b| a.address.0.cmp(&b.address.0)), - other => return err_response("actors", &format!("unknown sort field `{other}` — use mailbox|worker|address")), - } - - actors.truncate(limit); - - let rows: Vec<_> = actors - .iter() - .map(|a| { - serde_json::json!({ - "address": format_addr(&a.address), - "address_full": full_hex(&a.address), - "worker_id": a.worker_id, - "mailbox_depth": a.mailbox_depth, - "last_msg_type": a.last_msg_type, - "messages_processed": a.messages_processed, - "poisoned": a.poisoned, - }) - }) - .collect(); - - ok_response( - "actors", - serde_json::json!({ - "total": stats.actor_details.len(), - "returned": rows.len(), - "sort": sort_by, - "actors": rows, - }), - ) -} - -pub fn cmd_actor(rt: &Runtime, col: &StatsCollector, args: &[&str]) -> String { - let prefix = match args.first() { - Some(p) => *p, - None => return err_response("actor", "usage: actor "), - }; - - let stats = enriched_stats(rt, col); - let matches: Vec<_> = stats - .actor_details - .iter() - .filter(|a| full_hex(&a.address).starts_with(prefix)) - .map(|a| { - serde_json::json!({ - "address": format_addr(&a.address), - "address_full": full_hex(&a.address), - "worker_id": a.worker_id, - "mailbox_depth": a.mailbox_depth, - "last_msg_type": a.last_msg_type, - "messages_processed": a.messages_processed, - "poisoned": a.poisoned, - }) - }) - .collect(); - - ok_response( - "actor", - serde_json::json!({ - "prefix": prefix, - "matches": matches.len(), - "actors": matches, - }), - ) -} - -pub fn cmd_hot(rt: &Runtime, col: &StatsCollector, args: &[&str]) -> String { - let n: usize = args.first().and_then(|s| s.parse().ok()).unwrap_or(10); - let stats = enriched_stats(rt, col); - - let mut actors = stats.actor_details.clone(); - actors.sort_by(|a, b| b.mailbox_depth.cmp(&a.mailbox_depth)); - actors.truncate(n); - - let rows: Vec<_> = actors - .iter() - .map(|a| { - serde_json::json!({ - "address": format_addr(&a.address), - "address_full": full_hex(&a.address), - "worker_id": a.worker_id, - "mailbox_depth": a.mailbox_depth, - "last_msg_type": a.last_msg_type, - "messages_processed": a.messages_processed, - "poisoned": a.poisoned, - }) - }) - .collect(); - - ok_response("hot", rows) -} - -pub fn cmd_phases(rt: &Runtime, args: &[&str]) -> String { - let stats = rt.stats(); - - let worker_filter: Option = args.first().and_then(|s| s.parse().ok()); - - let phase_names = [ - "spawn_drain", - "transfer_drain", - "tick_all", - "spawn_drain_2", - "pending_local", - "stats_publish", - ]; - - let mut results = Vec::new(); - for (i, timings) in stats.tick_timings.iter().enumerate() { - if let Some(wid) = worker_filter { - if i != wid { - continue; - } - } - let breakdown = compute_phase_breakdown(timings); - results.push(serde_json::json!({ - "worker_id": i, - "ticks_sampled": timings.len(), - "phases": breakdown, - "phase_names": phase_names, - })); - } - - ok_response("phases", results) -} - -pub fn cmd_diff(rt: &Runtime, col: &StatsCollector, args: &[&str]) -> String { - let secs: f64 = match args.first().and_then(|s| s.parse().ok()) { - Some(s) if s > 0.0 && s <= 30.0 => s, - Some(_) => return err_response("diff", "seconds must be between 0 and 30"), - None => return err_response("diff", "usage: diff "), - }; - - let before = enriched_stats(rt, col); - let t0 = Instant::now(); - std::thread::sleep(Duration::from_secs_f64(secs)); - let after = enriched_stats(rt, col); - let elapsed = t0.elapsed().as_secs_f64(); - - let msgs_before: u64 = before.workers.iter().map(|w| w.messages_processed).sum(); - let msgs_after: u64 = after.workers.iter().map(|w| w.messages_processed).sum(); - let delta_msgs = msgs_after.saturating_sub(msgs_before); - - let local_before: u64 = before.workers.iter().map(|w| w.local_sends).sum(); - let local_after: u64 = after.workers.iter().map(|w| w.local_sends).sum(); - let cross_before: u64 = before.workers.iter().map(|w| w.cross_sends).sum(); - let cross_after: u64 = after.workers.iter().map(|w| w.cross_sends).sum(); - - let mailbox_before: usize = before.workers.iter().map(|w| w.mailbox_depth).sum(); - let mailbox_after: usize = after.workers.iter().map(|w| w.mailbox_depth).sum(); - - let per_worker: Vec<_> = after - .workers - .iter() - .enumerate() - .map(|(i, w)| { - let prev = before.workers.get(i); - let d = prev - .map(|p| w.messages_processed.saturating_sub(p.messages_processed)) - .unwrap_or(0); - serde_json::json!({ - "worker_id": i, - "delta_messages": d, - "msg_per_sec": d as f64 / elapsed, - "actors_before": prev.map(|p| p.num_actors).unwrap_or(0), - "actors_after": w.num_actors, - "mailbox_before": prev.map(|p| p.mailbox_depth).unwrap_or(0), - "mailbox_after": w.mailbox_depth, - }) - }) - .collect(); - - ok_response( - "diff", - serde_json::json!({ - "elapsed_s": elapsed, - "actors_before": before.actor_details.len(), - "actors_after": after.actor_details.len(), - "delta_messages": delta_msgs, - "msg_per_sec": delta_msgs as f64 / elapsed, - "delta_local_sends": local_after.saturating_sub(local_before), - "delta_cross_sends": cross_after.saturating_sub(cross_before), - "mailbox_before": mailbox_before, - "mailbox_after": mailbox_after, - "per_worker": per_worker, - }), - ) -} - -// ── Phase breakdown helper ────────────────────────────────────────────── - -fn compute_phase_breakdown( - timings: &[swactor::stats::TickTiming], -) -> serde_json::Value { - if timings.is_empty() { - return serde_json::json!({ - "ticks": 0, - "active_pct": 0.0, - "avg_tick_us": 0.0, - "phases_us": [0, 0, 0, 0, 0, 0], - "phases_pct": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0], - }); - } - - let n = timings.len(); - let active = timings.iter().filter(|t| t.did_work).count(); - let active_pct = (active as f64 / n as f64) * 100.0; - - let mut phase_sums = [0u64; 6]; - for t in timings { - for (i, &us) in t.phase_us.iter().enumerate() { - phase_sums[i] += us; - } - } - let total_us: u64 = phase_sums.iter().sum(); - let avg_tick_us = total_us as f64 / n as f64; - - let phases_pct: Vec = if total_us == 0 { - vec![0.0; 6] - } else { - phase_sums - .iter() - .map(|&s| (s as f64 / total_us as f64) * 100.0) - .collect() - }; - - serde_json::json!({ - "ticks": n, - "active_pct": active_pct, - "avg_tick_us": avg_tick_us, - "phases_us": phase_sums, - "phases_pct": phases_pct, - }) + let req = swactor_command::from_query_params(params); + router.dispatch(&req, ctx).to_json_line() } diff --git a/crates/runtime-dashboard/src/server.rs b/crates/runtime-dashboard/src/server.rs index 34f1c5e..703ab68 100644 --- a/crates/runtime-dashboard/src/server.rs +++ b/crates/runtime-dashboard/src/server.rs @@ -11,7 +11,6 @@ use swactor::runtime::Runtime; use crate::actors_html::ACTORS_HTML; use crate::collector::StatsCollector; use crate::dashboard_html::DASHBOARD_HTML; -use crate::investigate; use crate::layer::EventStore; use crate::trace::RuntimeTrace; @@ -127,6 +126,7 @@ pub(crate) fn spawn_http_server( let addr = format!("0.0.0.0:{port}"); let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server"); let server = Arc::new(server); + let cmd_router = Arc::new(swactor_command::CommandRouter::with_builtins()); for _ in 0..4 { let server = Arc::clone(&server); @@ -134,6 +134,7 @@ pub(crate) fn spawn_http_server( let runtime = Arc::clone(&runtime); let collector = Arc::clone(&collector); let shutdown = Arc::clone(&shutdown); + let cmd_router = Arc::clone(&cmd_router); #[cfg(feature = "distribution")] let distribution = Arc::clone(&distribution); thread::spawn(move || { @@ -174,6 +175,7 @@ pub(crate) fn spawn_http_server( &url, Arc::clone(&runtime), Arc::clone(&collector), + Arc::clone(&cmd_router), ); } _ => respond_404(request), @@ -283,21 +285,33 @@ fn handle_investigate_api( url: &str, runtime: Arc>>>, collector: Arc>>>, + cmd_router: Arc, ) { let params = parse_query_string(url); - let cmd = params.get("cmd").map(|s| s.as_str()).unwrap_or("help"); let maybe_rt = runtime.lock().unwrap().clone(); let maybe_col = collector.lock().unwrap().clone(); let json = match (maybe_rt, maybe_col) { - (Some(rt), Some(col)) => investigate::dispatch_command(cmd, ¶ms, &rt, &col), - _ => serde_json::json!({ - "ok": false, - "command": cmd, - "error": "runtime not attached yet" - }) - .to_string(), + (Some(rt), Some(col)) => { + let ctx = swactor_command::CommandContext::with_enricher(rt, col); + let req = swactor_command::from_query_params(¶ms); + cmd_router.dispatch(&req, &ctx).to_json_line() + } + (Some(rt), None) => { + let ctx = swactor_command::CommandContext::new(rt); + let req = swactor_command::from_query_params(¶ms); + cmd_router.dispatch(&req, &ctx).to_json_line() + } + _ => { + let cmd = params.get("cmd").map(|s| s.as_str()).unwrap_or("help"); + serde_json::json!({ + "ok": false, + "command": cmd, + "error": "runtime not attached yet" + }) + .to_string() + } }; let response = tiny_http::Response::from_string(json).with_header(