From 2bf11e0fdb938d61f09c08be189fafccd1a74d78 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 3 Aug 2026 14:18:24 +0400 Subject: [PATCH] feat: actor view panels for the `dashboard` Add read-only actor overview and per-actor dossier pages to the dashboard, fed by enriched per-actor runtime snapshots. - `dashboard/swactor/actor_view`: new `ActorPanelView`, a tolerant frame consumer over `runtime.actors`/`runtime.stats` that folds per-actor snapshots and serves `/view/swactor/actor-overview` (roster) and `/view/swactor/actor-dossier` (per-actor detail), each backed by an embedded HTML template (`actor_overview.html`, `actor_dossier.html`) - `dashboard`: register both views in `DashboardHandle` and export `actor_overview_view()`/`actor_dossier_view()` from the swactor module - `swactor` core: enrich `ActorSnapshot` with `actor_type` and `message_type` (populated from `slot.actor.metadata()` in `ActorPool`) and add `ActorAddress::to_full_hex()` for untruncated display - `myelin/orchestration`: publish actor stats to the dashboard via a `runtime.actors` channel producer (`stats_hook_on`) threaded through the distribution stack, and carry the orchestrator actor address into readiness signaling - workspace `Cargo.toml`: add `default-members` for native iteration and a centralized `[workspace.dependencies] tokio` so members share one feature set; `dashboard/Cargo.toml` switches to `tokio.workspace = true` Signed-off-by: Zachery Aaron Shores-Chmielewski --- Cargo.toml | 20 + apps/myelin/Cargo.toml | 2 +- apps/myelin/src/chat/runtime.rs | 64 +- apps/myelin/src/node/actor.rs | 38 +- apps/myelin/src/node/worker_node_runtime.rs | 76 ++- apps/myelin/src/orchestration/actor.rs | 22 +- apps/myelin/src/orchestration/app.rs | 79 ++- .../src/orchestration/distribution_stack.rs | 7 +- .../orchestration/provider_adapters/relay.rs | 6 - apps/myelin/src/orchestration/run_fsm.rs | 16 - apps/myelin/src/orchestration/run_plan.rs | 1 - apps/myelin/src/staging/control.rs | 19 - apps/myelin/src/tests/harness.rs | 50 ++ .../src/tests/local_mock/environment.rs | 8 +- apps/myelin/src/tests/local_mock/mock_node.rs | 5 +- apps/myelin/src/tests/mod.rs | 1 + .../src/tests/orchestration_guarantees.rs | 6 +- apps/myelin/src/tests/staging_guarantees.rs | 16 +- crates/dashboard/AGENTS.md | 2 +- crates/dashboard/Cargo.toml | 2 +- crates/dashboard/README.md | 6 +- crates/dashboard/src/lib.rs | 2 + .../dashboard/src/swactor/actor_dossier.html | 185 +++++ .../dashboard/src/swactor/actor_overview.html | 312 +++++++++ crates/dashboard/src/swactor/actor_view.rs | 635 ++++++++++++++++++ crates/dashboard/src/swactor/mod.rs | 12 + crates/datastream/src/endpoint.rs | 6 +- .../datastream/tests/t_datastream_endpoint.rs | 6 +- crates/iroh-driver/Cargo.toml | 3 +- rust-toolchain.toml | 2 + src/actor.rs | 12 + src/stats.rs | 2 + src/worker.rs | 3 + tools/vastai/Cargo.toml | 2 +- xtask/src/main.rs | 2 +- 35 files changed, 1490 insertions(+), 140 deletions(-) create mode 100644 apps/myelin/src/tests/harness.rs create mode 100644 crates/dashboard/src/swactor/actor_dossier.html create mode 100644 crates/dashboard/src/swactor/actor_overview.html create mode 100644 crates/dashboard/src/swactor/actor_view.rs create mode 100644 rust-toolchain.toml diff --git a/Cargo.toml b/Cargo.toml index 5a5b7ac..325c8ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,27 @@ members = [ "xtask", "tools/vastai", ] +default-members = [ + ".", + "crates/process", + "crates/provisioning", + "crates/transport", + "crates/distribution", + "crates/iroh-driver", + "crates/datastream", + "crates/data-plane", + "crates/dashboard", + "apps/myelin", + "tools/vastai", +] +# The language bindings (python, wasm) and xtask are built on demand +# (-p / --workspace / cargo-xtask), not during normal native iteration. exclude = ["crates/bindings/wasm-crypto", "examples"] +[workspace.dependencies] +# Centralized so every member resolves tokio with the SAME feature set. +# Without this, members declared tokio with different features, so building +# one member vs another (or `run` vs `test`) recompiled tokio each time. +tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "process", "io-util", "sync", "time", "net", "signal"] } [package] name = "swactor" diff --git a/apps/myelin/Cargo.toml b/apps/myelin/Cargo.toml index 6c44a1b..fa29a2f 100644 --- a/apps/myelin/Cargo.toml +++ b/apps/myelin/Cargo.toml @@ -22,7 +22,7 @@ swactor-process = { path = "../../crates/process" } distribution = { path = "../../crates/distribution" } iroh-driver = { path = "../../crates/iroh-driver" } iroh = "0.98" -tokio = { version = "1", features = ["rt-multi-thread", "macros", "process", "io-util", "sync", "time", "net", "signal"] } +tokio.workspace = true swactor-vastai = { path = "../../tools/vastai" } parking_lot = "0.12" blake3 = "1" diff --git a/apps/myelin/src/chat/runtime.rs b/apps/myelin/src/chat/runtime.rs index 9b755d7..32694a2 100644 --- a/apps/myelin/src/chat/runtime.rs +++ b/apps/myelin/src/chat/runtime.rs @@ -56,7 +56,7 @@ OPTIONS: --relay-mode Relay mode: default or disabled --relay-url Custom relay URL passed to myelin-orchestrator --endpoint-addr-mask Endpoint address mask: full or relay-only - --cached-model[=] Use discovered or explicit cached GGUF model + --cached-model[=] Use discovered or explicit cached GGUF model (default for --process) --dump-logs[=] Write datastream frame log --run-id Override run id --skip-rebuild Reuse existing Cargo artifacts @@ -793,9 +793,15 @@ impl Config { let gpu_run = args.gpu || env_flag(MYELIN_CHAT_GPU_RUN_ENV, false); let endpoint_addr_mask = Self::endpoint_addr_mask(&args, &toml)?; let (relay_mode, relay_url) = Self::relay_settings(&args, &toml, endpoint_addr_mask)?; - let cached_model = Self::cached_model_source(&args, gpu_run, &provider) - .map(CachedModelConfig::from_source) - .transpose()?; + let cached_model = match Self::cached_model_source(&args, &provider) { + None => None, + Some(CachedModelSource::Path(path)) => Some(CachedModelConfig::from_path(path)?), + Some(CachedModelSource::Discover) => match CachedModelConfig::discover() { + Ok(config) => Some(config), + Err(error) if args.cached_model.is_some() => return Err(error), + Err(_) => None, + }, + }; let model = Self::model_config(&provider, &toml, cached_model.as_ref())?; let datastream_frame_log = Self::datastream_frame_log(&args, &toml); let vastai = if provider == provider_kind::vastai() { @@ -876,16 +882,15 @@ impl Config { Ok((relay_mode, relay_url)) } - fn cached_model_source( - args: &ParsedArgs, - gpu_run: bool, - provider: &ProviderKind, - ) -> Option { + /// Resolve the cached-model source. The process provider defaults to + /// best-effort discovery of `.model-cache/` so `cargo myelin-chat` runs a + /// cached GGUF model without explicit flags. Explicit `--cached-model` is + /// always honored (and stays strict); the default degrades gracefully to + /// the normal download path when no cached model is present. + fn cached_model_source(args: &ParsedArgs, provider: &ProviderKind) -> Option { match &args.cached_model { Some(source) => Some(source.clone()), - None if gpu_run && provider == &provider_kind::process() => { - Some(CachedModelSource::Discover) - } + None if provider == &provider_kind::process() => Some(CachedModelSource::Discover), None => None, } } @@ -2122,13 +2127,6 @@ struct CachedModelConfig { } impl CachedModelConfig { - fn from_source(source: CachedModelSource) -> Result { - match source { - CachedModelSource::Discover => Self::discover(), - CachedModelSource::Path(path) => Self::from_path(path), - } - } - fn from_path(path: PathBuf) -> Result { let metadata = fs::metadata(&path) .map_err(|e| format!("stat cached model {}: {e}", path.display()))?; @@ -2517,6 +2515,34 @@ tag = " alias " ); } + #[test] + fn config_defaults_use_cached_model_for_process_when_present() { + let temp = TempDir::new("cached-model-default"); + + with_process_state(&[], Some(temp.path()), || { + let cache_dir = temp.path().join(REPO_MODEL_CACHE_DIR); + fs::create_dir_all(&cache_dir).expect("create model cache dir"); + fs::write( + cache_dir.join(DEFAULT_PIPELINE_CACHED_MODEL_FILE), + Vec::::new(), + ) + .expect("seed cached model file"); + + let defaults = Config::from_args(Vec::::new()).expect("defaults resolve"); + + assert_eq!(defaults.provider, provider_kind::process()); + let cached_model = defaults + .cached_model + .expect("process provider discovers a cached model by default"); + assert!( + cached_model + .host_path + .ends_with(DEFAULT_PIPELINE_CACHED_MODEL_FILE), + "discovered the seeded cached model" + ); + }); + } + #[test] fn config_max_tokens_drives_orchestrator_args_and_submit_prompt() { let temp = TempDir::new("config-max-tokens"); diff --git a/apps/myelin/src/node/actor.rs b/apps/myelin/src/node/actor.rs index 8f4d072..2170472 100644 --- a/apps/myelin/src/node/actor.rs +++ b/apps/myelin/src/node/actor.rs @@ -170,6 +170,19 @@ pub(crate) enum NodeAgentMsg { WorkerCrashed { reason: Option, }, + StepFailed { + step_id: u64, + }, + ObjectFailed { + edge_id: u64, + object_id: Option, + }, + OutputFault { + edge_id: u64, + }, + EdgeFault { + edge_id: u64, + }, StopRun { run_id: u64, }, @@ -236,9 +249,6 @@ pub(crate) enum StageCommandWire { layer_end_exclusive: u32, stage_shard_plan: Option, }, - RewireEdge { - edge_id: u64, - }, ExecuteStep { step_id: u64, input_edge_id: u64, @@ -531,6 +541,27 @@ impl NodeAgentActor { self.last_worker_crash = reason; self.core.observe(stage::StageEvent::WorkerCrashed) } + NodeAgentMsg::StepFailed { step_id } => { + self.core.observe(stage::StageEvent::StepFailed { + step_id: stage::StepId(step_id), + }) + } + NodeAgentMsg::ObjectFailed { edge_id, object_id } => { + self.core.observe(stage::StageEvent::ObjectFailed { + edge_id: stage::EdgeId(edge_id), + object_id: object_id.map(stage::ObjectId), + }) + } + NodeAgentMsg::OutputFault { edge_id } => { + self.core.observe(stage::StageEvent::OutputFault { + edge_id: stage::EdgeId(edge_id), + }) + } + NodeAgentMsg::EdgeFault { edge_id } => { + self.core.observe(stage::StageEvent::EdgeFault { + edge_id: stage::EdgeId(edge_id), + }) + } NodeAgentMsg::StopRun { run_id } => self.core.observe(stage::StageEvent::StopRun { run_id: stage::RunId(run_id), }), @@ -692,7 +723,6 @@ impl From<&stage::StageCommand> for StageCommandWire { layer_end_exclusive: range.end_exclusive, stage_shard_plan: shard_plan.clone(), }, - stage::StageCommand::RewireEdge { edge_id } => Self::RewireEdge { edge_id: edge_id.0 }, stage::StageCommand::ExecuteStep(step) => Self::ExecuteStep { step_id: step.step_id.0, input_edge_id: step.input.edge_id.0, diff --git a/apps/myelin/src/node/worker_node_runtime.rs b/apps/myelin/src/node/worker_node_runtime.rs index 1119494..dd5eddd 100644 --- a/apps/myelin/src/node/worker_node_runtime.rs +++ b/apps/myelin/src/node/worker_node_runtime.rs @@ -1013,7 +1013,7 @@ impl WorkerEdgeRuntime { ); let step_started = Instant::now(); let mut pump = || pump_network(driver, stack); - let committed_bytes = worker.execute_step( + let committed_bytes = match worker.execute_step( u64::from(config.stage_index) + 1, step_id, object_id, @@ -1027,7 +1027,15 @@ impl WorkerEdgeRuntime { config, datastream, &mut pump, - )?; + ) { + Ok(bytes) => bytes, + Err(e) => { + let _ = stack + .runtime + .send_to(node_actor, NodeAgentMsg::StepFailed { step_id }); + return Err(e); + } + }; let helper_execute_ms = duration_ms_u64(step_started.elapsed()); let egress_read_started = Instant::now(); let record = { @@ -1035,9 +1043,16 @@ impl WorkerEdgeRuntime { let lease = arena .lookup_lease(arena::RingId(output_ring_id)) .ok_or_else(|| format!("outbound ring {output_ring_id} lease missing"))?; - arena - .read_arena(lease.layout.data_offset, committed_bytes) - .map_err(|e| format!("read egress ring: {e}"))? + arena.read_arena(lease.layout.data_offset, committed_bytes) + }; + let record = match record { + Ok(record) => record, + Err(e) => { + let _ = stack + .runtime + .send_to(node_actor, NodeAgentMsg::OutputFault { edge_id: outbound.edge_id }); + return Err(format!("read egress ring: {e}")); + } }; let egress_read_ms = duration_ms_u64(egress_read_started.elapsed()); let record_bytes = record.len(); @@ -1067,7 +1082,12 @@ impl WorkerEdgeRuntime { .as_ref() .ok_or_else(|| "outbound edge sender missing".to_owned())?; let edge_send_started = Instant::now(); - sender.send(record)?; + if let Err(e) = sender.send(record) { + let _ = stack + .runtime + .send_to(node_actor, NodeAgentMsg::OutputFault { edge_id: outbound.edge_id }); + return Err(e); + } let edge_send_ms = duration_ms_u64(edge_send_started.elapsed()); node_stage( datastream, @@ -1131,8 +1151,17 @@ impl WorkerEdgeRuntime { buffer.extend_from_slice(&bytes); let buffered_bytes = buffer.len(); let mut records = Vec::new(); - while let Some(record) = take_complete_ingress_record(buffer, inbound.object_spec)? { - records.push(record); + loop { + match take_complete_ingress_record(buffer, inbound.object_spec) { + Ok(Some(record)) => records.push(record), + Ok(None) => break, + Err(e) => { + let _ = stack + .runtime + .send_to(node_actor, NodeAgentMsg::ObjectFailed { edge_id, object_id: None }); + return Err(e); + } + } } (records, buffered_bytes) }; @@ -1171,14 +1200,22 @@ impl WorkerEdgeRuntime { }), ); let object_load_started = Instant::now(); - let loaded = worker.ring_readable( + let loaded = match worker.ring_readable( ring_id, edge_id, inbound.object_spec, config, datastream, &mut || {}, - )?; + ) { + Ok(loaded) => loaded, + Err(e) => { + let _ = stack + .runtime + .send_to(node_actor, NodeAgentMsg::ObjectFailed { edge_id, object_id: Some(record.object_id) }); + return Err(e); + } + }; let object_load_ms = duration_ms_u64(object_load_started.elapsed()); let key = ObjectKey { edge_id, @@ -1510,11 +1547,9 @@ impl WorkerEdgeRuntime { .runtime .send_to( node_actor, - NodeAgentMsg::WorkerCrashed { - reason: Some(format!("edge {} faulted: {reason:?}", edge_id.0)), - }, + NodeAgentMsg::EdgeFault { edge_id: edge_id.0 }, ) - .map_err(|e| format!("mark worker crashed after edge fault: {e}"))?; + .map_err(|e| format!("report edge fault: {e}"))?; return Err(format!("edge {} faulted: {reason:?}", edge_id.0)); } edge::EdgeLifecycleEvent::EdgeStopped { .. } => {} @@ -1691,6 +1726,8 @@ fn run() -> Result<(), String> { )?; } + let mut datastream = NodeDatastream::new(&config); + let worker_stats_hook = datastream.producer.stats_hook(); let stack = DistributionRuntimeStack::new_with_codecs( driver.node_id(), DistributedNodeConfig::default(), @@ -1698,6 +1735,7 @@ fn run() -> Result<(), String> { register_myelin_actor_codecs(registry); datastream::wire::register_datastream_codec(registry); }, + Some(worker_stats_hook), ); boot( "distribution_stack", @@ -1750,7 +1788,6 @@ fn run() -> Result<(), String> { }; let arena_fd = arena_manager.lock().arena_fd(); - let mut datastream = NodeDatastream::new(&config); let node_boot = |ds: &mut NodeDatastream, phase: &str, status: &str, detail: Value| { emit_node_event(ds, &config, NODE_BOOTSTRAP_CHANNEL, phase, status, detail) }; @@ -3560,15 +3597,6 @@ fn handle_stage_command( ); Ok(()) } - StageCommandWire::RewireEdge { .. } => { - node_stage( - datastream, - "rewire_edge", - "skipped", - json!({"reason":"not implemented in myelin-worker image path"}), - ); - Ok(()) - } StageCommandWire::ReleaseInputHandle { handle_id, .. } => { edge_runtime.release_input_handle(handle_id, worker, config, datastream, driver, stack) } diff --git a/apps/myelin/src/orchestration/actor.rs b/apps/myelin/src/orchestration/actor.rs index 6e87216..bf6719b 100644 --- a/apps/myelin/src/orchestration/actor.rs +++ b/apps/myelin/src/orchestration/actor.rs @@ -68,6 +68,13 @@ pub(crate) enum OrchestratorMsg { stage_index: u32, }, ObserveTokenEndpointsStopped, + ObserveOperatorStop { + run_id: u64, + }, + ObserveMembershipLost { + run_id: u64, + node_id: u64, + }, AdvanceTimeMs(u64), Snapshot { reply_to: ActorAddress, @@ -131,7 +138,6 @@ pub(crate) enum RunCommandWire { #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub(crate) enum LifecycleEventWire { - RunRejected { run_id: u64 }, RunFaulted { run_id: u64 }, RunCompleted { run_id: u64 }, RunOperatorStopped { run_id: u64 }, @@ -272,6 +278,17 @@ impl OrchestratorActor { OrchestratorMsg::ObserveTokenEndpointsStopped => { self.core.observe(core::RunEvent::TokenEndpointsStopped) } + OrchestratorMsg::ObserveOperatorStop { run_id } => { + self.core.observe(core::RunEvent::OperatorStop { + run_id: core::RunId(run_id), + }); + } + OrchestratorMsg::ObserveMembershipLost { run_id, node_id } => { + self.core.observe(core::RunEvent::MembershipLost { + run_id: core::RunId(run_id), + node_id: core::NodeId(node_id), + }); + } OrchestratorMsg::AdvanceTimeMs(delta) => self.core.advance_time_ms(delta), } } @@ -449,9 +466,6 @@ impl From<&core::RunCommand> for RunCommandWire { impl From<&core::LifecycleEvent> for LifecycleEventWire { fn from(event: &core::LifecycleEvent) -> Self { match event { - core::LifecycleEvent::RunRejected { run_id, .. } => { - Self::RunRejected { run_id: run_id.0 } - } core::LifecycleEvent::RunFaulted { run_id, .. } => { Self::RunFaulted { run_id: run_id.0 } } diff --git a/apps/myelin/src/orchestration/app.rs b/apps/myelin/src/orchestration/app.rs index 96aadbd..79e6f64 100644 --- a/apps/myelin/src/orchestration/app.rs +++ b/apps/myelin/src/orchestration/app.rs @@ -19,7 +19,7 @@ use crate::node_actor::{ #[cfg(feature = "dashboard")] use crate::observability::dashboard_view::MyelinClusterDashboardView; use crate::observability::{benchmark, frame_archive::FrameArchive}; -use crate::orchestration::actor::{OrchestratorActor, OrchestratorReport}; +use crate::orchestration::actor::{OrchestratorActor, OrchestratorMsg, OrchestratorReport}; use crate::orchestration::config::{DEFAULT_CONFIG_PATH, TomlConfigOverlay}; const PROVIDER_START_MAX_ATTEMPTS: usize = 4; @@ -56,6 +56,7 @@ use datastream::{ }; use distribution::node::DistributedNodeConfig; use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; +use distribution::swim::telemetry::ObservedTransition; use distribution::types::{MemberState, NodeId as DistNodeId}; use iroh::EndpointAddr; use iroh_driver::{ @@ -283,6 +284,8 @@ where "connectivity_preflight":"ready", }), ); + let actors_channel = orch_datastream.channel_by_name("runtime.actors"); + let orch_stats_hook = orch_datastream.producer.stats_hook_on(actors_channel); let stack = DistributionRuntimeStack::new_with_codecs( driver.node_id(), DistributedNodeConfig::default(), @@ -290,6 +293,7 @@ where register_myelin_actor_codecs(registry); datastream::wire::register_datastream_codec(registry); }, + Some(orch_stats_hook), ); bootstrap( &mut orch_datastream, @@ -455,7 +459,7 @@ where tx: Mutex::new(obs_tx), })); let pipeline_coordinator_endpoint = coordinator_endpoint.clone(); - let (mut provisioned_nodes, ready) = start_and_provision_workers( + let (mut provisioned_nodes, ready, swim_to_node) = start_and_provision_workers( provisioner, &config, pipeline_plan.as_ref(), @@ -473,6 +477,7 @@ where run_id: config.run_id, orchestrator_node_id: config.node_id, provider: &config.provider, + orchestrator_actor, }, sink, coordinator_endpoint, @@ -547,6 +552,7 @@ where run_id: config.run_id, orchestrator_node_id: config.node_id, provider: &config.provider, + orchestrator_actor, }, &work_rx, &prompt_events, @@ -558,6 +564,7 @@ where tokenizer_reply_actor, pipeline_plan.as_ref(), ready.first_stage.endpoint.clone(), + &swim_to_node, ); if let Err(error) = &result { bootstrap( @@ -2024,6 +2031,7 @@ struct RuntimeReadyAckLoop<'a> { run_id: u64, orchestrator_node_id: u64, provider: &'a ProviderKind, + orchestrator_actor: ActorAddress, } fn wait_for_runtime_ready_acks( @@ -2045,6 +2053,7 @@ fn wait_for_runtime_ready_acks( run_id, orchestrator_node_id, provider, + .. } = ctx; let bootstrap = |ds: &mut OrchDatastream, phase: &str, status: &str, detail: Value| { ds.emit_bootstrap( @@ -2232,7 +2241,7 @@ fn start_and_provision_workers( coordinator: EndpointAddr, pipeline_coordinator: EndpointAddr, orchestrator_actor: ActorAddress, -) -> Result<(ProvisionedClusterGuard, PromptRuntimeReady), String> { +) -> Result<(ProvisionedClusterGuard, PromptRuntimeReady, BTreeMap), String> { let RuntimeReadyAckLoop { driver, stack, @@ -2439,6 +2448,7 @@ fn start_and_provision_workers( run_id: config.run_id, orchestrator_node_id: config.node_id, provider: &config.provider, + orchestrator_actor, }, &expected_node_ids, ) { @@ -2468,6 +2478,7 @@ fn start_and_provision_workers( run_id: config.run_id, orchestrator_node_id: config.node_id, provider: &config.provider, + orchestrator_actor, }) { Ok(ready) => ready, Err(error) => { @@ -2512,6 +2523,7 @@ fn start_and_provision_workers( run_id: config.run_id, orchestrator_node_id: config.node_id, provider: &config.provider, + orchestrator_actor, }, &ack_targets, &pipeline_coordinator, @@ -2555,6 +2567,7 @@ fn start_and_provision_workers( run_id: config.run_id, orchestrator_node_id: config.node_id, provider: &config.provider, + orchestrator_actor, }, expected_node_ids.len(), pipeline_plan.expect("pipeline mode requires plan"), @@ -2578,6 +2591,7 @@ fn start_and_provision_workers( run_id: config.run_id, orchestrator_node_id: config.node_id, provider: &config.provider, + orchestrator_actor, }, config.stage_index, ) @@ -2640,7 +2654,11 @@ fn start_and_provision_workers( final_stage: ready, } }; - Ok((provisioned_nodes, prompt_ready)) + let swim_to_node = readies + .iter() + .map(|(node_id, ready)| (ready.swim_node_id, *node_id)) + .collect::>(); + Ok((provisioned_nodes, prompt_ready, swim_to_node)) } fn stage_node_specs( @@ -2999,6 +3017,7 @@ fn wait_for_weights_loaded_count( run_id, orchestrator_node_id: node_id, provider, + .. } = ctx; let expected_stages = pipeline_plan .stages @@ -4126,6 +4145,7 @@ fn wait_for_runtime_ready(ctx: RuntimeReadyAckLoop<'_>) -> Result = None; let mut node_swim_started = false; @@ -4271,6 +4291,7 @@ fn wait_for_weights_loaded(ctx: RuntimeReadyAckLoop<'_>, stage_index: u32) -> Re run_id, orchestrator_node_id: node_id, provider, + .. } = ctx; loop { pump(driver, stack, frame_tx); @@ -4913,6 +4934,7 @@ fn serve_prompts( tokenizer_reply_to: ActorAddress, pipeline_plan: Option<&run_plan::RunPlan>, prompt_endpoint: EndpointAddr, + swim_to_node: &BTreeMap, ) -> Result<(), String> { let RuntimeReadyAckLoop { driver, @@ -4927,6 +4949,7 @@ fn serve_prompts( run_id, orchestrator_node_id: node_id, provider, + orchestrator_actor, .. } = ctx; let emit_prompt_evt = @@ -4949,6 +4972,7 @@ fn serve_prompts( let mut active: Option = None; loop { pump(driver, stack, frame_tx); + orch_datastream.flush(dashboard, "orchestrator"); if let Some(pipeline) = pipeline_runtime.as_mut() { pipeline.poll_driver(driver); pipeline.drain_tokenizer_events( @@ -4971,6 +4995,21 @@ fn serve_prompts( )?; drain_frames(frame_rx, dashboard, orch_datastream); drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id); + let swim_transitions = + emit_swim_transitions(orch_datastream, dashboard, run_id, node_id, stack); + for transition in &swim_transitions { + if transition.to == MemberState::Dead + && let Some(&lost_node_id) = swim_to_node.get(&transition.peer) + { + let _ = stack.runtime.send_to( + orchestrator_actor, + OrchestratorMsg::ObserveMembershipLost { + run_id, + node_id: lost_node_id, + }, + ); + } + } if stop_rx.try_recv().is_ok() { orch_datastream.emit_bootstrap( dashboard, @@ -4980,6 +5019,10 @@ fn serve_prompts( "started", json!({"source":"stdin"}), ); + let _ = stack + .runtime + .send_to(orchestrator_actor, OrchestratorMsg::ObserveOperatorStop { run_id }); + pump(driver, stack, frame_tx); return Ok(()); } @@ -5337,8 +5380,9 @@ fn emit_swim_transitions( run_id: u64, node_id: u64, stack: &DistributionRuntimeStack, -) { - for transition in stack.drain_swim_transitions() { +) -> Vec { + let transitions = stack.drain_swim_transitions(); + for transition in &transitions { let peer = format!("{:?}", transition.peer); let from = transition.from.map(|state| format!("{:?}", state)); let to = format!("{:?}", transition.to); @@ -5366,8 +5410,9 @@ fn emit_swim_transitions( "member_state":member_state.clone(), }), ); - orch_datastream.emit_record(dashboard, &stack.membership_transition(&transition)); + orch_datastream.emit_record(dashboard, &stack.membership_transition(transition)); } + transitions } fn emit_swim_probe_events( @@ -5418,18 +5463,24 @@ fn local_tinygrad_worker_env(provider: &str) -> Option<(String, String)> { } fn default_local_tinygrad_worker_path() -> Option { - let cwd_candidate = std::env::current_dir() - .ok() - .map(|cwd| cwd.join("apps").join("myelin-node").join("tinygrad_worker.py")); + // The tinygrad worker script ships in the node-image build context at + // `apps/myelin/node-image/tinygrad_worker.py` (see `chat/node_image.rs` and + // the node-image Dockerfile). The process provider runs it directly via + // `python3`, so resolve that path from the workspace cwd or this crate's + // manifest dir. (Previously looked in `apps/myelin-node/`, a path left stale + // by the `mvp-system` -> `myelin` app refactor and never present on disk.) + let cwd_candidate = std::env::current_dir().ok().map(|cwd| { + cwd.join("apps") + .join("myelin") + .join("node-image") + .join("tinygrad_worker.py") + }); if let Some(candidate) = cwd_candidate.filter(|path| path.is_file()) { return Some(candidate.canonicalize().unwrap_or(candidate)); } let manifest_candidate = PathBuf::from(env!("CARGO_MANIFEST_DIR")) - .join("..") - .join("..") - .join("apps") - .join("myelin-node") + .join("node-image") .join("tinygrad_worker.py"); manifest_candidate.is_file().then(|| { manifest_candidate diff --git a/apps/myelin/src/orchestration/distribution_stack.rs b/apps/myelin/src/orchestration/distribution_stack.rs index cd6f7e3..4c50cad 100644 --- a/apps/myelin/src/orchestration/distribution_stack.rs +++ b/apps/myelin/src/orchestration/distribution_stack.rs @@ -13,6 +13,7 @@ use std::time::{Duration, Instant}; use swactor::actor::{ActorAddress, ActorInterface}; use swactor::config::RuntimeConfig; use swactor::runtime::{Ctx, Runtime}; +use swactor::stats::StatsHook; use swactor::std::StdExtension; use swactor_transport::{CodecRegistry, CodecRemoteSink, NetworkMessage, TransportRouter}; @@ -44,7 +45,6 @@ pub(crate) struct DistributionActorAddrs { } pub(crate) struct DistributionRuntimeStack { - pub node_id: NodeId, pub runtime: Arc, pub codec: Arc, pub outbox: Outbox, @@ -61,6 +61,7 @@ impl DistributionRuntimeStack { node_id: NodeId, config: DistributedNodeConfig, extend_codecs: impl FnOnce(&mut CodecRegistry), + stats_hook: Option>, ) -> Self { let mut runtime = Runtime::new(RuntimeConfig::default()).with_extension(Arc::new(StdExtension::new())); @@ -72,6 +73,9 @@ impl DistributionRuntimeStack { Arc::clone(&codec), Arc::clone(&transport_router), ))); + if let Some(hook) = stats_hook { + runtime.set_stats_hook(hook); + } let runtime = Arc::new(runtime); let outbox: Outbox = Arc::new(Mutex::new(Vec::new())); @@ -146,7 +150,6 @@ impl DistributionRuntimeStack { .expect("subscribe MembershipFanout"); Self { - node_id, runtime, codec, outbox, diff --git a/apps/myelin/src/orchestration/provider_adapters/relay.rs b/apps/myelin/src/orchestration/provider_adapters/relay.rs index d33fec8..de224cc 100644 --- a/apps/myelin/src/orchestration/provider_adapters/relay.rs +++ b/apps/myelin/src/orchestration/provider_adapters/relay.rs @@ -72,12 +72,6 @@ impl StaticRelayProvider { parse_relay_url(raw).map(Self::new) } - pub(crate) fn from_env() -> Result, String> { - selected_relay_url_from_env() - .map(|url| Self::from_url_str(&url).map(Some)) - .unwrap_or(Ok(None)) - } - pub(crate) fn url(&self) -> String { self.url.to_string() } diff --git a/apps/myelin/src/orchestration/run_fsm.rs b/apps/myelin/src/orchestration/run_fsm.rs index 35e133c..125aaa0 100644 --- a/apps/myelin/src/orchestration/run_fsm.rs +++ b/apps/myelin/src/orchestration/run_fsm.rs @@ -16,16 +16,6 @@ pub(crate) struct RunPlan { pub stages: Vec, } -impl RunPlan { - pub(crate) fn test_linear(run_id: RunId, stages: Vec) -> Self { - Self { run_id, stages } - } - - pub(crate) fn stage_nodes(&self) -> Vec { - self.stages.iter().map(|stage| stage.node_id).collect() - } -} - #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct RunConfig { pub run_id: RunId, @@ -125,10 +115,6 @@ pub(crate) enum RunFaultReason { #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) enum LifecycleEvent { - RunRejected { - run_id: RunId, - reason: RunFaultReason, - }, RunFaulted { run_id: RunId, reason: RunFaultReason, @@ -175,8 +161,6 @@ pub(crate) enum RunCommand { }, } -pub type OrchestratorHarness = OrchestratorRun; - pub(crate) struct OrchestratorRun { config: RunConfig, plan: Option, diff --git a/apps/myelin/src/orchestration/run_plan.rs b/apps/myelin/src/orchestration/run_plan.rs index f3c4c0b..a7287db 100644 --- a/apps/myelin/src/orchestration/run_plan.rs +++ b/apps/myelin/src/orchestration/run_plan.rs @@ -165,7 +165,6 @@ pub(crate) enum EdgeKind { pub(crate) enum ObjectKind { Token, Activation, - Weight, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] diff --git a/apps/myelin/src/staging/control.rs b/apps/myelin/src/staging/control.rs index eb140eb..d024f0a 100644 --- a/apps/myelin/src/staging/control.rs +++ b/apps/myelin/src/staging/control.rs @@ -19,12 +19,6 @@ pub(crate) struct DeviceHandle { pub id: u64, } -impl DeviceHandle { - pub(crate) fn new_current(id: u64) -> Self { - Self { generation: 1, id } - } -} - #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) struct LayerRange { pub start: u32, @@ -50,14 +44,6 @@ impl WeightSource { tokenizer, } } - - pub(crate) fn embedded_gguf(model_id: impl Into, path: impl Into) -> Self { - Self::new( - model_id, - GgufSource::LocalPath(path.into()), - TokenizerSource::EmbeddedGguf, - ) - } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -228,9 +214,6 @@ pub(crate) enum StageCommand { range: LayerRange, shard_plan: Option, }, - RewireEdge { - edge_id: EdgeId, - }, ExecuteStep(ExecuteStep), ReleaseInputHandle { object_id: ObjectId, @@ -244,8 +227,6 @@ pub(crate) enum StageCommand { }, } -pub type StageControllerHarness = StageController; - pub(crate) struct StageController { local_node_id: NodeId, provision: Option, diff --git a/apps/myelin/src/tests/harness.rs b/apps/myelin/src/tests/harness.rs new file mode 100644 index 0000000..e68e8af --- /dev/null +++ b/apps/myelin/src/tests/harness.rs @@ -0,0 +1,50 @@ +//! Test-only constructors and harness type aliases for the `tests` tree. +//! +//! These helpers previously lived in the production modules +//! (`orchestration/run_fsm`, `staging::control`) but serve only test code. +//! Defining them here keeps them compiled under `#[cfg(test)]` only, so +//! production builds stay free of the dead-code warnings they would otherwise +//! trigger. Inherent methods resolve through their type, so existing call sites +//! such as `fsm::RunPlan::test_linear(..)` and +//! `stage::WeightSource::embedded_gguf(..)` work unchanged. + +use crate::run_fsm::{NodeId as FsmNodeId, OrchestratorRun, RunId as FsmRunId, RunPlan, StageRef}; +use crate::run_plan::{GgufSource, TokenizerSource}; +use crate::staging::{DeviceHandle, StageController, WeightSource}; + +impl RunPlan { + /// Build a linear pipeline plan (stage 0 -> stage 1 -> ... -> last stage). + /// Test-only convenience over the public `RunPlan` fields. + pub(crate) fn test_linear(run_id: FsmRunId, stages: Vec) -> Self { + Self { run_id, stages } + } + + /// Node ids in declared stage order. + pub(crate) fn stage_nodes(&self) -> Vec { + self.stages.iter().map(|stage| stage.node_id).collect() + } +} + +impl DeviceHandle { + /// A handle on the worker's current (first) generation. + pub(crate) fn new_current(id: u64) -> Self { + Self { generation: 1, id } + } +} + +impl WeightSource { + /// Weights carried by an embedded GGUF file with an embedded tokenizer. + pub(crate) fn embedded_gguf(model_id: impl Into, path: impl Into) -> Self { + Self::new( + model_id, + GgufSource::LocalPath(path.into()), + TokenizerSource::EmbeddedGguf, + ) + } +} + +/// Test alias for the orchestrator run core, retained for readable test prose. +pub(crate) type OrchestratorHarness = OrchestratorRun; + +/// Test alias for the stage controller core, retained for readable test prose. +pub(crate) type StageControllerHarness = StageController; diff --git a/apps/myelin/src/tests/local_mock/environment.rs b/apps/myelin/src/tests/local_mock/environment.rs index e24af3b..b7d2512 100644 --- a/apps/myelin/src/tests/local_mock/environment.rs +++ b/apps/myelin/src/tests/local_mock/environment.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, BTreeSet}; use crate::run_fsm as fsm; use crate::run_plan as plan; +use crate::tests::harness::OrchestratorHarness; use data_plane::edge_actor; use myelin::observability::lifecycle as obs; use myelin::orchestration::engine_builder as engine; @@ -34,7 +35,7 @@ pub struct LocalMockCluster { engine_events: Vec, plan: plan::RunPlan, nodes: BTreeMap, - orchestrator: Option, + orchestrator: Option, orchestrator_command_cursor: usize, orchestrator_event_cursor: usize, trace: Vec, @@ -303,7 +304,7 @@ impl LocalMockCluster { self.orchestrator_command_cursor = 0; self.orchestrator_event_cursor = 0; self.scenario = scenario; - self.orchestrator = Some(fsm::OrchestratorHarness::new(fsm::RunConfig { + self.orchestrator = Some(OrchestratorHarness::new(fsm::RunConfig { run_id: fsm::RunId(self.run_id.0), max_tokens: u64::from(self.max_tokens), prompt: tokenize(prompt), @@ -742,7 +743,6 @@ impl LocalMockCluster { self.push_run(obs::EventKind::RunCompleted, obs::Component::Orchestrator); } fsm::LifecycleEvent::RunFaulted { .. } - | fsm::LifecycleEvent::RunRejected { .. } | fsm::LifecycleEvent::RunOperatorStopped { .. } => { self.push_run(obs::EventKind::RunFaulted, obs::Component::Orchestrator); } @@ -852,7 +852,7 @@ impl LocalMockCluster { .count() } - fn orchestrator_mut(&mut self) -> &mut fsm::OrchestratorHarness { + fn orchestrator_mut(&mut self) -> &mut OrchestratorHarness { self.orchestrator .as_mut() .expect("run_prompt must initialize orchestrator") diff --git a/apps/myelin/src/tests/local_mock/mock_node.rs b/apps/myelin/src/tests/local_mock/mock_node.rs index 4f85c7a..0a39de8 100644 --- a/apps/myelin/src/tests/local_mock/mock_node.rs +++ b/apps/myelin/src/tests/local_mock/mock_node.rs @@ -1,6 +1,7 @@ use crate::run_plan as plan; use data_plane::edge_actor; use myelin::staging as stage; +use crate::tests::harness::StageControllerHarness; use super::mock_transport::MockObject; use super::mock_worker::MockWorker; @@ -16,7 +17,7 @@ pub struct MockNode { inbound_edge: Option, outbound_edge: Option, outbound_object_allocator: Option, - controller: stage::StageControllerHarness, + controller: StageControllerHarness, worker: MockWorker, event_cursor: usize, } @@ -34,7 +35,7 @@ impl MockNode { inbound_edge: None, outbound_edge: None, outbound_object_allocator: None, - controller: stage::StageControllerHarness::new(stage::NodeId(node_id.0)), + controller: StageControllerHarness::new(stage::NodeId(node_id.0)), worker: MockWorker::new(stage_index, eos_after_sequence), event_cursor: 0, } diff --git a/apps/myelin/src/tests/mod.rs b/apps/myelin/src/tests/mod.rs index e757758..dabed93 100644 --- a/apps/myelin/src/tests/mod.rs +++ b/apps/myelin/src/tests/mod.rs @@ -1,3 +1,4 @@ +mod harness; mod local_e2e_guarantees; mod local_mock; mod node_guarantees; diff --git a/apps/myelin/src/tests/orchestration_guarantees.rs b/apps/myelin/src/tests/orchestration_guarantees.rs index 705b5ed..a81f6ae 100644 --- a/apps/myelin/src/tests/orchestration_guarantees.rs +++ b/apps/myelin/src/tests/orchestration_guarantees.rs @@ -502,6 +502,7 @@ mod run_fsm { //! `specs/BEHAVIOR_GUARANTEES.md`. use crate::run_fsm as fsm; + use crate::tests::harness::OrchestratorHarness; // A three-stage plan proves multi-stage provisioning and readiness without // making tests depend on any placement heuristic. The plan is already valid; @@ -529,8 +530,8 @@ mod run_fsm { // The harness is the black-box public boundary for the run FSM. It accepts // observable events and records emitted commands/events; tests never inspect an // internal FSM enum or private readiness counter. - fn new_run() -> fsm::OrchestratorHarness { - fsm::OrchestratorHarness::new(fsm::RunConfig { + fn new_run() -> OrchestratorHarness { + OrchestratorHarness::new(fsm::RunConfig { run_id: fsm::RunId(7), max_tokens: 4, prompt: vec![101, 102, 103], @@ -680,7 +681,6 @@ mod run_fsm { }); assert!(invalid.events().iter().any(|event| { matches!(event, fsm::LifecycleEvent::RunFaulted { .. }) - || matches!(event, fsm::LifecycleEvent::RunRejected { .. }) })); } diff --git a/apps/myelin/src/tests/staging_guarantees.rs b/apps/myelin/src/tests/staging_guarantees.rs index f5a16bf..a7efea2 100644 --- a/apps/myelin/src/tests/staging_guarantees.rs +++ b/apps/myelin/src/tests/staging_guarantees.rs @@ -12,6 +12,7 @@ mod stage_controller { //! `specs/BEHAVIOR_GUARANTEES.md`. use myelin::staging as stage; + use crate::tests::harness::StageControllerHarness; // This provision fixture represents a single middle stage. It has one inbound // and one outbound edge so tests can prove the controller uses assigned edges @@ -37,8 +38,8 @@ mod stage_controller { // The harness exposes only public messages. Tests intentionally do not inspect // private controller states such as "Preparing" or "Executing"; they infer // controller behavior from emitted commands and lifecycle events. - fn new_controller() -> stage::StageControllerHarness { - stage::StageControllerHarness::new(stage::NodeId(11)) + fn new_controller() -> StageControllerHarness { + StageControllerHarness::new(stage::NodeId(11)) } // Preparation readiness has four independent prerequisites. Listing them as @@ -60,7 +61,7 @@ mod stage_controller { // This helper provisions and readies a stage through public events. Tests that // focus on execution use it to avoid duplicating setup while still going through // the same observable path as production. - fn ready_stage() -> stage::StageControllerHarness { + fn ready_stage() -> StageControllerHarness { let mut harness = new_controller(); harness.observe(stage::StageEvent::ProvisionStage { from: stage::NodeId(99), @@ -103,15 +104,6 @@ mod stage_controller { ) })); - // The controller must not emit any command that replaces the provisioned - // edge ids with a locally chosen edge. - assert!( - !harness - .commands() - .iter() - .any(|command| { matches!(command, stage::StageCommand::RewireEdge { .. }) }) - ); - // An unauthorized provision attempt must fault before setup can begin. let mut unauthorized = new_controller(); unauthorized.observe(stage::StageEvent::ProvisionStage { diff --git a/crates/dashboard/AGENTS.md b/crates/dashboard/AGENTS.md index 5454bc1..cf7a436 100644 --- a/crates/dashboard/AGENTS.md +++ b/crates/dashboard/AGENTS.md @@ -8,4 +8,4 @@ Keep this crate read-only with respect to observed programs. - It must not send control signals to observed runtimes. - It must not require changes outside `crates/dashboard` for dashboard-only work. -Main built-in view: `/view/swactor/workers`, backed by `runtime.stats`, `runtime.workers`, and `runtime.actors` frames when present. +Main built-in views: `/view/swactor/workers` (worker-centric) and `/view/swactor/actor-overview` + `/view/swactor/actor-dossier` (actor-centric), all backed by `runtime.stats`, `runtime.workers`, and `runtime.actors` frames when present. Actor views are pure frame consumers and tolerant of publisher shape. diff --git a/crates/dashboard/Cargo.toml b/crates/dashboard/Cargo.toml index e89c983..67803de 100644 --- a/crates/dashboard/Cargo.toml +++ b/crates/dashboard/Cargo.toml @@ -10,5 +10,5 @@ swactor = { path = "../..", features = ["serde"] } parking_lot = "0.12" serde = { version = "1", features = ["derive"] } serde_json = "1" -tokio = { version = "1", features = ["net", "rt-multi-thread", "sync"] } +tokio.workspace = true tokio-stream = "0.1" diff --git a/crates/dashboard/README.md b/crates/dashboard/README.md index 7b6c265..b2908d5 100644 --- a/crates/dashboard/README.md +++ b/crates/dashboard/README.md @@ -2,7 +2,7 @@ Read-only HTML/SSE dashboard over incoming datastream frames. -The crate owns the Axum server, bounded raw frame window, and view registry. Component crates can keep their own view implementations beside their code and register them through `DashboardHandle::register_view`. The built-in swactor worker page is hosted here because worker/actor/message processing is universal to swactor programs. +The crate owns the Axum server, bounded raw frame window, and view registry. Component crates can keep their own view implementations beside their code and register them through `DashboardHandle::register_view`. The built-in swactor views are hosted here because worker/actor/message processing is universal to swactor programs: the worker page, and the actor overview (fused roster) plus per-actor dossier. ## Routes @@ -16,5 +16,9 @@ The crate owns the Axum server, bounded raw frame window, and view registry. Com - `GET /api/view/fleet` — fleet and machine telemetry JSON snapshot - `GET /view/swactor/workers` — built-in worker page - `GET /api/view/swactor/workers` — worker page JSON snapshot +- `GET /view/swactor/actor-overview` — built-in actor overview + roster page +- `GET /api/view/swactor/actor-overview` — actor overview JSON snapshot +- `GET /view/swactor/actor-dossier` — built-in per-actor dossier page +- `GET /api/view/swactor/actor-dossier` — actor dossier JSON snapshot All state is derived from observed frames. The dashboard sends no control signals back to producers. diff --git a/crates/dashboard/src/lib.rs b/crates/dashboard/src/lib.rs index e0be424..bf3ca3a 100644 --- a/crates/dashboard/src/lib.rs +++ b/crates/dashboard/src/lib.rs @@ -97,6 +97,8 @@ impl DashboardHandle { views.register(Arc::new(live_explorer::LiveDatastreamExplorer::default())); views.register(Arc::new(hardware_view::HardwareDashboardView::default())); views.register(swactor::worker_view()); + views.register(swactor::actor_overview_view()); + views.register(swactor::actor_dossier_view()); let store = Arc::new(DashboardStore::new( config.raw_frame_history, Arc::clone(&views), diff --git a/crates/dashboard/src/swactor/actor_dossier.html b/crates/dashboard/src/swactor/actor_dossier.html new file mode 100644 index 0000000..f554ac5 --- /dev/null +++ b/crates/dashboard/src/swactor/actor_dossier.html @@ -0,0 +1,185 @@ + + + + + +Actor dossier — swactor + + + +
+

Actor dossier

+ + +
+
+
+ + + ← back to overview +
+ +
+
+ + + diff --git a/crates/dashboard/src/swactor/actor_overview.html b/crates/dashboard/src/swactor/actor_overview.html new file mode 100644 index 0000000..20db27f --- /dev/null +++ b/crates/dashboard/src/swactor/actor_overview.html @@ -0,0 +1,312 @@ + + + + + +Actor overview — swactor + + + +
+

Actor overview

+ + +
+
+ +
+ +

State distribution

+
+
+ +

Throughput

+
+ +

Worker balance

+
+ +

Hot actors — mailbox growth

+
+ +

Alerts

+
+ +

Roster

+
+ + + + + + +
+
+ + + +
+
+
+ + + diff --git a/crates/dashboard/src/swactor/actor_view.rs b/crates/dashboard/src/swactor/actor_view.rs new file mode 100644 index 0000000..54704cc --- /dev/null +++ b/crates/dashboard/src/swactor/actor_view.rs @@ -0,0 +1,635 @@ +//! Actor-centric read-only view over `runtime.actors` / `runtime.stats` frames. +//! +//! Pure frame consumer: it folds incoming per-actor snapshots into local state +//! and exposes JSON + HTML. It sends nothing back to observed runtimes. Parsing +//! is intentionally tolerant of publisher shape — a frame is a flat record that +//! may come from the per-worker `DatastreamStatsHook` envelope +//! (`{worker_id, actors:[...]}`) or from a process that publishes a merged +//! `{actors:[...]}` payload (the dashboard dummy node, app runtimes). Fields the +//! publisher includes (name, worker_id, lifecycle flags) are displayed; ones it +//! omits are left blank rather than fabricated. +//! +//! Reliably on the frame today: address, mailbox depth, throughput, last +//! message, poisoned flag, and the message-type diet. `worker_id` and `name` +//! appear when the publisher sends them. Finer lifecycle granularity +//! (new/suspended/stopping), actor Rust type, and spawn age are not on the +//! current frame and are therefore not shown — enriching the feed is a separate +//! core concern, not a dashboard one. +//! +//! Two pages share one data model: the fused overview+roster +//! (`/view/swactor/actor-overview`) and the per-actor dossier +//! (`/view/swactor/actor-dossier`). Each is a self-contained `DashboardView` +//! instance ingesting the same channels. + +use std::collections::{BTreeMap, VecDeque}; +use std::time::{Duration, Instant}; + +use datastream::frame::{Frame, StreamId}; +use parking_lot::RwLock; +use serde::Serialize; +use serde_json::{Value, json}; + +use crate::swactor::{RUNTIME_ACTORS, RUNTIME_STATS}; +use crate::view::DashboardView; +use crate::{FrameEvent, StreamEvent}; + +const CHANNELS: &[&str] = &[RUNTIME_ACTORS, RUNTIME_STATS]; +const HISTORY_CAP: usize = 512; +const HISTORY_MIN_INTERVAL: Duration = Duration::from_millis(250); +const PER_ACTOR_HISTORY_CAP: usize = 120; +const GROWTH_WINDOW: usize = 12; +const LIVE_TTL: Duration = Duration::from_secs(8); + +/// Read-only actor panel. One instance per served page so each page owns its +/// state independently; both ingest the same frames. +pub struct ActorPanelView { + id: &'static str, + title: &'static str, + path: &'static str, + html: &'static str, + state: RwLock, +} + +impl ActorPanelView { + /// Fused overview + roster at `/view/swactor/actor-overview`. + pub fn overview() -> Self { + Self::new( + "swactor-actor-overview", + "Actor overview", + "swactor/actor-overview", + include_str!("actor_overview.html"), + ) + } + + /// Per-actor dossier at `/view/swactor/actor-dossier`. + pub fn dossier() -> Self { + Self::new( + "swactor-actor-dossier", + "Actor dossier", + "swactor/actor-dossier", + include_str!("actor_dossier.html"), + ) + } + + fn new(id: &'static str, title: &'static str, path: &'static str, html: &'static str) -> Self { + Self { + id, + title, + path, + html, + state: RwLock::new(PanelState::default()), + } + } +} + +#[derive(Default)] +struct PanelState { + runtimes: BTreeMap, +} + +struct RuntimeState { + stream: StreamEvent, + last_seen: Instant, + num_workers: Option, + uptime_ms: Option, + actors: BTreeMap, + history: VecDeque, +} + +impl RuntimeState { + fn new(stream: StreamEvent, now: Instant) -> Self { + Self { + stream, + last_seen: now, + num_workers: None, + uptime_ms: None, + actors: BTreeMap::new(), + history: VecDeque::with_capacity(HISTORY_CAP), + } + } + + fn update(&mut self, channel: &str, payload: &[u8], now: Instant) { + self.last_seen = now; + let Ok(value) = serde_json::from_slice::(payload) else { + return; + }; + match channel { + RUNTIME_ACTORS => self.apply_actors(&value, now), + RUNTIME_STATS => self.apply_stats(&value, now), + _ => {} + } + self.push_history(now); + } + + fn apply_actors(&mut self, value: &Value, now: Instant) { + // Per-worker `worker_id` wrapper (DatastreamStatsHook shape) is the + // default placement for actors that do not carry one inline. + let wrapper_worker = u32_field(value, &["worker_id", "worker"]); + if let Some(actors) = value.get("actors").and_then(Value::as_array) { + for actor in actors { + self.apply_actor(actor, now, wrapper_worker); + } + return; + } + // A bare actor object per frame (no envelope). + self.apply_actor(value, now, wrapper_worker); + } + + fn apply_actor(&mut self, value: &Value, now: Instant, default_worker: Option) { + let Some(address) = string_field(value, &["address", "addr", "actor_addr"]) else { + return; + }; + let actor = self + .actors + .entry(address.clone()) + .or_insert_with(|| ActorState::new(address)); + actor.apply_json(value, now, default_worker); + } + + fn apply_stats(&mut self, value: &Value, now: Instant) { + if let Some(num_workers) = u32_field(value, &["num_workers", "workers_live"]) { + self.num_workers = Some(num_workers); + } + if let Some(uptime_ms) = u64_field(value, &["uptime_ms"]) { + self.uptime_ms = Some(uptime_ms); + } + // address -> worker_id mapping; runtime.stats `actors` is [[addr, wid]]. + if let Some(actors) = value.get("actors").and_then(Value::as_array) { + for entry in actors { + if let Some(items) = entry.as_array() + && items.len() >= 2 + && let (Some(address), Some(worker_id)) = + (value_to_string(&items[0]), value_to_u32(&items[1])) + { + self.actors + .entry(address.clone()) + .or_insert_with(|| ActorState::new(address)) + .worker_id = Some(worker_id); + } + } + } + // Some publishers carry full per-actor detail under `actor_details`. + if let Some(details) = value.get("actor_details").and_then(Value::as_array) { + for actor in details { + self.apply_actor(actor, now, None); + } + } + } + + fn push_history(&mut self, now: Instant) { + let totals = self.totals(); + if let Some(last) = self.history.back_mut() + && now.duration_since(last.at) < HISTORY_MIN_INTERVAL + { + last.mailbox_depth = totals.mailbox_depth; + last.msg_per_sec = totals.msg_per_sec; + return; + } + if self.history.len() == HISTORY_CAP { + self.history.pop_front(); + } + self.history.push_back(HistorySample { + at: now, + mailbox_depth: totals.mailbox_depth, + msg_per_sec: totals.msg_per_sec, + }); + } + + fn totals(&self) -> Totals { + let mut totals = Totals::default(); + totals.actors = self.actors.len().min(u32::MAX as usize) as u32; + for actor in self.actors.values() { + totals.mailbox_depth = totals.mailbox_depth.saturating_add(actor.mailbox_depth); + totals.msg_per_sec += actor.msg_per_sec; + if actor.poisoned { + totals.poisoned = totals.poisoned.saturating_add(1); + } + } + totals + } +} + +#[derive(Clone)] +struct ActorState { + address: String, + name: Option, + actor_type: Option, + message_type: Option, + worker_id: Option, + mailbox_depth: u32, + mailbox_growth: f64, + messages_processed: u64, + msg_per_sec: f64, + last_msg_type: Option, + poisoned: bool, + message_type_counts: Vec<(String, u64)>, + history: VecDeque, + last_update: Option, +} + +impl ActorState { + fn new(address: String) -> Self { + Self { + address, + name: None, + actor_type: None, + message_type: None, + worker_id: None, + mailbox_depth: 0, + mailbox_growth: 0.0, + messages_processed: 0, + msg_per_sec: 0.0, + last_msg_type: None, + poisoned: false, + message_type_counts: Vec::new(), + history: VecDeque::new(), + last_update: None, + } + } + + fn apply_json(&mut self, value: &Value, now: Instant, default_worker: Option) { + let elapsed = self + .last_update + .map(|then| now.duration_since(then).as_secs_f64()) + .unwrap_or(0.0); + if let Some(name) = string_field(value, &["name"]).filter(|name| !name.is_empty()) { + self.name = Some(name); + } + if let Some(actor_type) = string_field(value, &["actor_type"]).filter(|t| !t.is_empty()) { + self.actor_type = Some(actor_type); + } + if let Some(message_type) = string_field(value, &["message_type"]).filter(|t| !t.is_empty()) { + self.message_type = Some(message_type); + } + if let Some(worker_id) = u32_field(value, &["worker_id", "worker"]) { + self.worker_id = Some(worker_id); + } else if self.worker_id.is_none() { + self.worker_id = default_worker; + } + assign_u32(&mut self.mailbox_depth, value, &["mailbox_depth", "queued"]); + assign_u64_rate( + &mut self.messages_processed, + &mut self.msg_per_sec, + value, + &["messages_processed", "processed", "messages_handled"], + elapsed, + ); + if let Some(last) = string_field( + value, + &["last_msg_type", "last_message", "last_message_type"], + ) + .filter(|last| !last.is_empty()) + { + self.last_msg_type = Some(last); + } + if let Some(poisoned) = value.get("poisoned").and_then(Value::as_bool) { + self.poisoned = poisoned; + } + if let Some(counts) = parse_message_type_counts(value.get("message_type_counts")) { + self.message_type_counts = counts; + } + self.last_update = Some(now); + self.push_history(now); + self.recompute_growth(); + } + + fn push_history(&mut self, now: Instant) { + if let Some(last) = self.history.back_mut() + && now.duration_since(last.at) < HISTORY_MIN_INTERVAL + { + last.mailbox_depth = self.mailbox_depth; + last.msg_per_sec = self.msg_per_sec; + return; + } + if self.history.len() == PER_ACTOR_HISTORY_CAP { + self.history.pop_front(); + } + self.history.push_back(ActorHistorySample { + at: now, + mailbox_depth: self.mailbox_depth, + msg_per_sec: self.msg_per_sec, + }); + } + + /// Mailbox depth/s over the recent window, for trend colouring. + fn recompute_growth(&mut self) { + let n = self.history.len(); + if n < 2 { + self.mailbox_growth = 0.0; + return; + } + let start = n - n.min(GROWTH_WINDOW); + let first = &self.history[start]; + let last = &self.history[n - 1]; + let span = last.at.duration_since(first.at).as_secs_f64(); + self.mailbox_growth = if span > 0.0 { + (last.mailbox_depth as f64 - first.mailbox_depth as f64) / span + } else { + 0.0 + }; + } +} + +#[derive(Clone, Copy)] +struct HistorySample { + at: Instant, + mailbox_depth: u32, + msg_per_sec: f64, +} + +#[derive(Clone, Copy)] +struct ActorHistorySample { + at: Instant, + mailbox_depth: u32, + msg_per_sec: f64, +} + +#[derive(Default)] +struct Totals { + actors: u32, + mailbox_depth: u32, + msg_per_sec: f64, + poisoned: u32, +} + +#[derive(Serialize)] +struct PanelSnapshot { + runtimes: Vec, +} + +#[derive(Serialize)] +struct RuntimeSnapshot { + stream: StreamSnapshot, + live: bool, + last_seen_ms_ago: u64, + summary: SummarySnapshot, + actors: Vec, + history: Vec, +} + +#[derive(Serialize)] +struct StreamSnapshot { + key: String, + node: String, + life: u64, +} + +#[derive(Serialize)] +struct SummarySnapshot { + actors: u32, + msg_per_sec: f64, + mailbox_depth: u32, + poisoned: u32, + uptime_ms: Option, + num_workers: Option, +} + +#[derive(Serialize)] +struct ActorSnapshot { + address: String, + name: Option, + actor_type: Option, + message_type: Option, + worker_id: Option, + /// Single derived display state. The frame carries only `poisoned`, so the + /// granularity is poisoned | running until the feed is enriched. + state: &'static str, + mailbox_depth: u32, + mailbox_growth: f64, + messages_processed: u64, + msg_per_sec: f64, + last_msg_type: Option, + poisoned: bool, + message_type_counts: Vec, + history: Vec, + last_seen_ms_ago: u64, +} + +#[derive(Serialize)] +struct MessageTypeCountSnapshot { + message_type: String, + count: u64, +} + +#[derive(Serialize)] +struct HistorySnapshot { + ms_ago: u64, + msg_per_sec: f64, + mailbox_depth: u32, +} + +#[derive(Serialize)] +struct ActorHistorySnapshot { + ms_ago: u64, + mailbox_depth: u32, + msg_per_sec: f64, +} + +impl DashboardView for ActorPanelView { + fn id(&self) -> &'static str { + self.id + } + + fn title(&self) -> &'static str { + self.title + } + + fn path(&self) -> &'static str { + self.path + } + + fn channels(&self) -> &'static [&'static str] { + CHANNELS + } + + fn ingest(&self, _stream: &StreamId, _frame: &Frame, event: &FrameEvent) { + let now = Instant::now(); + let mut state = self.state.write(); + let key = stream_key(&event.stream); + state + .runtimes + .entry(key) + .or_insert_with(|| RuntimeState::new(event.stream.clone(), now)) + .update(&event.channel, &event.payload, now); + } + + fn snapshot_json(&self) -> Value { + let now = Instant::now(); + let snapshot = PanelSnapshot { + runtimes: self + .state + .read() + .runtimes + .values() + .map(|runtime| runtime_snapshot(runtime, now)) + .collect(), + }; + serde_json::to_value(snapshot).unwrap_or_else(|_| json!({ "runtimes": [] })) + } + + fn html(&self) -> Option<&'static str> { + Some(self.html) + } +} + +fn runtime_snapshot(runtime: &RuntimeState, now: Instant) -> RuntimeSnapshot { + let totals = runtime.totals(); + RuntimeSnapshot { + stream: StreamSnapshot { + key: stream_key(&runtime.stream), + node: runtime.stream.node.clone(), + life: runtime.stream.life, + }, + live: now.duration_since(runtime.last_seen) <= LIVE_TTL, + last_seen_ms_ago: now.duration_since(runtime.last_seen).as_millis() as u64, + summary: SummarySnapshot { + actors: totals.actors, + msg_per_sec: totals.msg_per_sec, + mailbox_depth: totals.mailbox_depth, + poisoned: totals.poisoned, + uptime_ms: runtime.uptime_ms, + num_workers: runtime.num_workers, + }, + actors: runtime + .actors + .values() + .map(|actor| ActorSnapshot { + address: actor.address.clone(), + name: actor.name.clone(), + actor_type: actor.actor_type.clone(), + message_type: actor.message_type.clone(), + worker_id: actor.worker_id, + state: if actor.poisoned { "poisoned" } else { "running" }, + mailbox_depth: actor.mailbox_depth, + mailbox_growth: actor.mailbox_growth, + messages_processed: actor.messages_processed, + msg_per_sec: actor.msg_per_sec, + last_msg_type: actor.last_msg_type.clone(), + poisoned: actor.poisoned, + message_type_counts: actor + .message_type_counts + .iter() + .map(|(message_type, count)| MessageTypeCountSnapshot { + message_type: message_type.clone(), + count: *count, + }) + .collect(), + history: actor + .history + .iter() + .map(|sample| ActorHistorySnapshot { + ms_ago: now.duration_since(sample.at).as_millis() as u64, + mailbox_depth: sample.mailbox_depth, + msg_per_sec: sample.msg_per_sec, + }) + .collect(), + last_seen_ms_ago: actor + .last_update + .map(|then| now.duration_since(then).as_millis() as u64) + .unwrap_or(0), + }) + .collect(), + history: runtime + .history + .iter() + .map(|sample| HistorySnapshot { + ms_ago: now.duration_since(sample.at).as_millis() as u64, + msg_per_sec: sample.msg_per_sec, + mailbox_depth: sample.mailbox_depth, + }) + .collect(), + } +} + +// --- tolerant JSON helpers (publisher-shape-agnostic readers) --------------- + +fn assign_u32(slot: &mut u32, value: &Value, names: &[&str]) { + if let Some(v) = u32_field(value, names) { + *slot = v; + } +} + +fn assign_u64_rate(slot: &mut u64, rate: &mut f64, value: &Value, names: &[&str], elapsed: f64) { + if let Some(next) = u64_field(value, names) { + if elapsed > 0.0 && next > *slot { + *rate = (next - *slot) as f64 / elapsed; + } else if next < *slot { + *rate = 0.0; + } + *slot = next; + } +} + +fn u32_field(value: &Value, names: &[&str]) -> Option { + u64_field(value, names).and_then(|v| u32::try_from(v).ok()) +} + +fn u64_field(value: &Value, names: &[&str]) -> Option { + names + .iter() + .find_map(|name| value.get(*name).and_then(value_to_u64)) +} + +fn string_field(value: &Value, names: &[&str]) -> Option { + names + .iter() + .find_map(|name| value.get(*name).and_then(value_to_string)) +} + +fn value_to_u32(value: &Value) -> Option { + value_to_u64(value).and_then(|v| u32::try_from(v).ok()) +} + +fn value_to_u64(value: &Value) -> Option { + value + .as_u64() + .or_else(|| value.as_str().and_then(|s| s.parse::().ok())) +} + +fn value_to_string(value: &Value) -> Option { + value.as_str().map(ToOwned::to_owned).or_else(|| { + if value.is_null() { + None + } else { + Some(value.to_string()) + } + }) +} + +/// Accepts pair form `["Type", N]`, object form `{"ty": "Type", "count": N}`, +/// and map form `{"Type": N}`. +fn parse_message_type_counts(value: Option<&Value>) -> Option> { + let value = value?; + if let Some(items) = value.as_array() { + let mut out = Vec::new(); + for item in items { + if let Some(pair) = item.as_array() + && pair.len() >= 2 + && let (Some(name), Some(count)) = + (value_to_string(&pair[0]), value_to_u64(&pair[1])) + { + out.push((name, count)); + continue; + } + if let Some(name) = string_field(item, &["ty", "message_type", "type", "name"]) + && let Some(count) = u64_field(item, &["count"]) + { + out.push((name, count)); + } + } + return Some(out); + } + if let Some(map) = value.as_object() { + let mut out: Vec<(String, u64)> = map + .iter() + .filter_map(|(name, count)| value_to_u64(count).map(|count| (name.clone(), count))) + .collect(); + out.sort_by(|a, b| b.1.cmp(&a.1)); + return Some(out); + } + None +} + +fn stream_key(stream: &StreamEvent) -> String { + format!("{}#{}", stream.node, stream.life) +} diff --git a/crates/dashboard/src/swactor/mod.rs b/crates/dashboard/src/swactor/mod.rs index fc424e0..7b1f59f 100644 --- a/crates/dashboard/src/swactor/mod.rs +++ b/crates/dashboard/src/swactor/mod.rs @@ -2,10 +2,12 @@ use std::sync::Arc; use crate::view::DashboardView; +mod actor_view; mod worker_page; mod worker_view; pub use worker_view::SwactorWorkerView; +pub use actor_view::ActorPanelView; pub const RUNTIME_STATS: &str = "runtime.stats"; pub const RUNTIME_WORKERS: &str = "runtime.workers"; @@ -14,3 +16,13 @@ pub const RUNTIME_ACTORS: &str = "runtime.actors"; pub fn worker_view() -> Arc { Arc::new(SwactorWorkerView::default()) } + +/// Built-in actor overview + roster view (`/view/swactor/actor-overview`). +pub fn actor_overview_view() -> Arc { + Arc::new(ActorPanelView::overview()) +} + +/// Built-in actor dossier view (`/view/swactor/actor-dossier`). +pub fn actor_dossier_view() -> Arc { + Arc::new(ActorPanelView::dossier()) +} diff --git a/crates/datastream/src/endpoint.rs b/crates/datastream/src/endpoint.rs index 5ca8721..31364f3 100644 --- a/crates/datastream/src/endpoint.rs +++ b/crates/datastream/src/endpoint.rs @@ -758,9 +758,11 @@ impl<'a> RuntimeActorStatsRecord<'a> { actors: snapshots .iter() .map(|snapshot| RuntimeActorSnapshotRecord { - address: snapshot.address.to_string(), + address: snapshot.address.to_full_hex(), mailbox_depth: snapshot.mailbox_depth, last_msg_type: snapshot.last_msg_type, + actor_type: snapshot.actor_type, + message_type: snapshot.message_type, messages_processed: snapshot.messages_processed, poisoned: snapshot.poisoned, message_type_counts: snapshot @@ -779,6 +781,8 @@ struct RuntimeActorSnapshotRecord<'a> { address: String, mailbox_depth: usize, last_msg_type: Option<&'static str>, + actor_type: Option<&'static str>, + message_type: Option<&'static str>, messages_processed: u64, poisoned: bool, message_type_counts: Vec>, diff --git a/crates/datastream/tests/t_datastream_endpoint.rs b/crates/datastream/tests/t_datastream_endpoint.rs index 9d5d7fc..3bafd54 100644 --- a/crates/datastream/tests/t_datastream_endpoint.rs +++ b/crates/datastream/tests/t_datastream_endpoint.rs @@ -300,6 +300,8 @@ fn stats_hook_adapter_submits_worker_snapshot_json() { address: actor, mailbox_depth: 3, last_msg_type: Some("Ping"), + actor_type: Some("TestActor"), + message_type: Some("Ping"), messages_processed: 5, poisoned: false, message_type_counts: vec![("Ping", 5)], @@ -313,11 +315,13 @@ fn stats_hook_adapter_submits_worker_snapshot_json() { assert_eq!(delivery.channel.channel, runtime); let json: Value = serde_json::from_slice(&delivery.payload).unwrap(); assert_eq!(json["worker_id"], 2); - assert_eq!(json["actors"][0]["address"], actor.to_string()); + assert_eq!(json["actors"][0]["address"], actor.to_full_hex()); assert_eq!(json["actors"][0]["mailbox_depth"], 3); assert_eq!(json["actors"][0]["last_msg_type"], "Ping"); assert_eq!(json["actors"][0]["messages_processed"], 5); assert_eq!(json["actors"][0]["message_type_counts"][0]["ty"], "Ping"); + assert_eq!(json["actors"][0]["actor_type"], "TestActor"); + assert_eq!(json["actors"][0]["message_type"], "Ping"); } fn positions(events: &[DatastreamEvent]) -> Vec { diff --git a/crates/iroh-driver/Cargo.toml b/crates/iroh-driver/Cargo.toml index 172fce6..5cfeb00 100644 --- a/crates/iroh-driver/Cargo.toml +++ b/crates/iroh-driver/Cargo.toml @@ -19,9 +19,8 @@ iroh = "0.98" # `test-utils` exposes `CaRootsConfig::insecure_skip_verify()` so clients can # trust operator-controlled custom relays with self-signed QAD certs. iroh-relay = { version = "0.98", features = ["test-utils"] } -tokio = { version = "1", features = ["rt-multi-thread", "time"] } +tokio.workspace = true parking_lot = "0.12" [dev-dependencies] iroh-relay = { version = "0.98", features = ["server", "test-utils"] } -tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util", "time"] } diff --git a/rust-toolchain.toml b/rust-toolchain.toml new file mode 100644 index 0000000..5f3bc6d --- /dev/null +++ b/rust-toolchain.toml @@ -0,0 +1,2 @@ +[toolchain] +channel = "nightly-2026-02-07" diff --git a/src/actor.rs b/src/actor.rs index 68a2adf..bec68de 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -162,6 +162,18 @@ impl ActorAddress { crate::get_random(&mut bytes); Self(bytes) } + + /// Full 64-character hex encoding of all 32 bytes, for displays that need + /// the untruncated address (dashboards). [`Display`](std::fmt::Display) + /// stays short for logs. + pub fn to_full_hex(&self) -> String { + use std::fmt::Write; + let mut s = String::with_capacity(64); + for b in &self.0 { + let _ = write!(s, "{:02x}", b); + } + s + } } // ─── Environment ───────────────────────────────────────────────────────────── diff --git a/src/stats.rs b/src/stats.rs index 491e37c..962183d 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -124,6 +124,8 @@ pub struct ActorSnapshot { pub address: ActorAddress, pub mailbox_depth: usize, pub last_msg_type: Option<&'static str>, + pub actor_type: Option<&'static str>, + pub message_type: Option<&'static str>, pub messages_processed: u64, pub poisoned: bool, /// Per-message-type counts, sorted descending by count. diff --git a/src/worker.rs b/src/worker.rs index b480d47..86e2d0e 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -998,10 +998,13 @@ impl ActorPool { let mut type_counts: Vec<(&'static str, u64)> = slot.msg_type_counts.iter().map(|(&k, &v)| (k, v)).collect(); type_counts.sort_by(|a, b| b.1.cmp(&a.1)); + let metadata = slot.actor.metadata(); ActorSnapshot { address: addr, mailbox_depth: slot.mailbox.len(), last_msg_type: slot.last_msg_type, + actor_type: Some(metadata.actor_type_name), + message_type: Some(metadata.message_type_name), messages_processed: slot.messages_processed, poisoned: slot.poisoned, message_type_counts: type_counts, diff --git a/tools/vastai/Cargo.toml b/tools/vastai/Cargo.toml index 7ef7ab2..0b88b55 100644 --- a/tools/vastai/Cargo.toml +++ b/tools/vastai/Cargo.toml @@ -8,7 +8,7 @@ publish = false reqwest = { version = "0.12", features = ["json"] } serde = { version = "1", features = ["derive"] } serde_json = "1" -tokio = { version = "1", features = ["macros", "rt", "time"] } +tokio.workspace = true urlencoding = "2" [dev-dependencies] diff --git a/xtask/src/main.rs b/xtask/src/main.rs index c38fc86..8fa5669 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -280,7 +280,7 @@ OPTIONS: Select the runtime provider --config Load config overlay --pipeline-stages Number of pipeline stages - --cached-model[=] Use discovered or explicit cached GGUF model + --cached-model[=] Use discovered or explicit cached GGUF model (default for --process) --dump-logs[=] Write datastream frame log --run-id Override run id --skip-rebuild Reuse existing Cargo artifacts