mod construct;
mod context_tree;
mod graph;
mod graph_layout;
mod helpers;
mod history;
mod input;
mod mcp;
mod new_run;
mod render;
mod selection;
mod state;
#[cfg(test)]
mod test_support;
mod types;
use crate::tui::theme;
pub use helpers::yank_to_clipboard_via;
pub use types::{AgentDisplayStatus, DashboardAgent, DashboardArgs};
pub use crate::tui::{CrosstermEventSource, EventSource, TerminalSetup};
use crossterm::event::{Event, KeyEventKind};
use leviath_runtime::control_socket::{ControlClient, ControlRequest, ControlResponse};
use ratatui::Terminal;
use std::time::Duration;
use tokio::sync::mpsc;
use state::Dashboard;
use types::DaemonCommand;
async fn daemon_background_loop(
control: ControlClient,
mut cmd_rx: mpsc::UnboundedReceiver<DaemonCommand>,
outcomes: mpsc::UnboundedSender<types::DaemonOutcome>,
) {
while let Some(cmd) = cmd_rx.recv().await {
let (run_id, request, what) = match cmd {
DaemonCommand::Cancel { run_id } => {
(run_id.clone(), ControlRequest::Cancel { run_id }, "cancel")
}
DaemonCommand::Pause { run_id } => {
(run_id.clone(), ControlRequest::Pause { run_id }, "pause")
}
DaemonCommand::Resume { run_id } => {
(run_id.clone(), ControlRequest::Resume { run_id }, "resume")
}
DaemonCommand::Answer { response } => (
response.request_id.clone(),
ControlRequest::AnswerInteraction { response },
"answer",
),
DaemonCommand::Message { agent_id, content } => (
agent_id.clone(),
ControlRequest::Message {
agent_id,
content,
target_region: None,
},
"message",
),
};
let outcome = match control.request(&request).await {
Ok(ControlResponse::Ok { ok: true }) => types::DaemonOutcome {
run_id,
message: String::new(),
ok: true,
},
Ok(ControlResponse::Ok { ok: false }) => types::DaemonOutcome {
run_id,
message: format!("the daemon has no such run to {what}"),
ok: false,
},
Ok(other) => types::DaemonOutcome {
run_id,
message: format!("unexpected daemon response to {what}: {other:?}"),
ok: false,
},
Err(e) => types::DaemonOutcome {
run_id,
message: format!("{what} failed: {e}"),
ok: false,
},
};
if outcomes.send(outcome).is_err() {
return;
}
}
}
async fn execute_core<S: TerminalSetup, E: EventSource>(
dashboard: &mut Dashboard,
control: &ControlClient,
setup: &mut S,
events: &mut E,
) -> anyhow::Result<()> {
setup.enable()?;
let mut terminal = setup.create_terminal()?;
let tick_rate = Duration::from_millis(100);
run_dashboard_loop(dashboard, control, &mut terminal, events, tick_rate).await?;
setup.disable();
setup.print_done();
Ok(())
}
async fn run_dashboard_loop<B: ratatui::backend::Backend>(
dashboard: &mut Dashboard,
control: &ControlClient,
terminal: &mut Terminal<B>,
events: &mut impl EventSource,
tick_rate: Duration,
) -> anyhow::Result<()> {
loop {
dashboard.tick_count += 1;
dashboard.tick_toasts();
dashboard.sync_interactions(control).await;
dashboard.sync_daemon_runs(control).await;
dashboard.sync_from_run_state();
dashboard.drain_mcp_outcomes();
dashboard.drain_spawn_outcomes();
dashboard.open_pending_run();
dashboard.drain_daemon_outcomes();
terminal
.draw(|frame| dashboard.draw(frame))
.map_err(|e| anyhow::anyhow!("terminal draw failed: {e}"))?;
if let Some(event) = events.poll_event(tick_rate)? {
match event {
Event::Key(key) if key.kind == KeyEventKind::Press => {
dashboard.handle_key(key);
}
Event::Mouse(m) => dashboard.handle_mouse(m),
Event::Resize(_, _) => {
}
_ => {}
}
}
if dashboard.should_quit {
return Ok(());
}
}
}
fn init_dashboard(control: ControlClient, yank_fn: fn(&str) -> bool) -> Dashboard {
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
let mcp_ctx = types::McpContext {
config_path: crate::config::Config::config_path(),
store_path: leviath_mcp::AuthStore::default_path().unwrap_or_default(),
opener: std::sync::Arc::new(leviath_sys::open_url),
clock: mcp_system_now,
};
let mut dashboard = Dashboard::new_with_log_path(
cmd_tx,
crate::runstate::dashboard_log_path(),
yank_fn,
mcp_ctx,
new_run::production_new_run_context(),
);
let daemon_outcome_tx = dashboard
.take_daemon_outcome_tx()
.expect("a fresh dashboard has its daemon outcome sender");
tokio::spawn(daemon_background_loop(
control.clone(),
cmd_rx,
daemon_outcome_tx,
));
let (mcp_cmd_rx, mcp_outcome_tx) = dashboard
.take_mcp_bg_ends()
.expect("a fresh dashboard has its MCP background channel ends");
tokio::spawn(mcp::mcp_background_loop(
dashboard.mcp_context(),
mcp_cmd_rx,
mcp_outcome_tx,
));
let (spawn_cmd_rx, spawn_outcome_tx) = dashboard
.take_spawn_bg_ends()
.expect("a fresh dashboard has its spawn background channel ends");
tokio::spawn(new_run::spawn_background_loop(
control,
spawn_cmd_rx,
spawn_outcome_tx,
));
dashboard.add_log("Dashboard started. Press `n` to start an agent run.".to_string());
dashboard
}
fn mcp_system_now() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
pub async fn execute_with<S: TerminalSetup, E: EventSource>(
control: ControlClient,
setup: &mut S,
events: &mut E,
yank_fn: fn(&str) -> bool,
) -> anyhow::Result<()> {
let mut dashboard = init_dashboard(control.clone(), yank_fn);
execute_core(&mut dashboard, &control, setup, events).await
}
#[cfg(test)]
mod tests {
use super::*;
use crate::commands::dashboard::test_support::make_test_dashboard;
#[test]
fn dashboard_args_can_be_constructed() {
let _args = DashboardArgs {};
}
#[test]
fn mcp_system_now_advances_past_the_epoch() {
assert!(mcp_system_now() > 1_600_000_000);
}
#[test]
fn agent_display_status_variants_display() {
let statuses = vec![
AgentDisplayStatus::Active,
AgentDisplayStatus::Waiting,
AgentDisplayStatus::Complete,
AgentDisplayStatus::CompleteInteractive,
AgentDisplayStatus::Error("test error".to_string()),
AgentDisplayStatus::Idle,
AgentDisplayStatus::Cancelled,
];
for status in statuses {
let display = format!("{}", status);
assert!(!display.is_empty());
}
}
fn no_daemon_control() -> ControlClient {
let dir = std::env::temp_dir().join("leviath-dash-no-daemon");
ControlClient::new(leviath_runtime::control_socket::control_id(&dir))
}
fn recording_daemon(dir: &std::path::Path) -> (ControlClient, tokio::task::JoinHandle<String>) {
use leviath_runtime::control_socket::{bind_control_listener, control_id};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
let id = control_id(dir);
let mut listener = bind_control_listener(&id).unwrap();
let handle = tokio::spawn(async move {
let stream = listener
.accept()
.await
.expect("accept succeeds")
.expect("our own connection is admitted");
let (read_half, mut write_half) = tokio::io::split(stream);
let mut lines = BufReader::new(read_half).lines();
let req = lines.next_line().await.unwrap().unwrap_or_default();
write_half
.write_all(b"{\"result\":\"ok\",\"ok\":true}\n")
.await
.unwrap();
req
});
(ControlClient::new(id), handle)
}
fn replying_daemon(
dir: &std::path::Path,
reply: Option<&'static str>,
) -> (ControlClient, tokio::task::JoinHandle<()>) {
use leviath_runtime::control_socket::{bind_control_listener, control_id};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
let id = control_id(dir);
let mut listener = bind_control_listener(&id).unwrap();
let handle = tokio::spawn(async move {
let stream = listener
.accept()
.await
.expect("accept succeeds")
.expect("our own connection is admitted");
let (read_half, mut write_half) = tokio::io::split(stream);
let mut lines = BufReader::new(read_half).lines();
let _ = lines.next_line().await;
if let Some(reply) = reply {
let _ = write_half.write_all(format!("{reply}\n").as_bytes()).await;
}
});
(ControlClient::new(id), handle)
}
async fn cancel_outcome(reply: Option<&'static str>) -> types::DaemonOutcome {
let dir = tempfile::tempdir().unwrap();
let (control, server) = replying_daemon(dir.path(), reply);
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<DaemonCommand>();
let (out_tx, mut out_rx) = mpsc::unbounded_channel();
tokio::spawn(daemon_background_loop(control, cmd_rx, out_tx));
cmd_tx
.send(DaemonCommand::Cancel {
run_id: "run-1".to_string(),
})
.unwrap();
let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), out_rx.recv())
.await
.expect("an outcome was reported")
.expect("the loop is alive");
let _ = server.await;
outcome
}
#[tokio::test]
async fn daemon_background_loop_reports_each_outcome() {
let ok = cancel_outcome(Some(r#"{"result":"ok","ok":true}"#)).await;
assert!(ok.ok, "an applied cancel is reported as success");
assert_eq!(ok.run_id, "run-1");
let missing = cancel_outcome(Some(r#"{"result":"ok","ok":false}"#)).await;
assert!(!missing.ok);
assert!(missing.message.contains("no such run to cancel"));
let odd = cancel_outcome(Some(r#"{"result":"spawned","run_id":"x"}"#)).await;
assert!(!odd.ok);
assert!(odd.message.contains("unexpected daemon response"));
let broken = cancel_outcome(None).await;
assert!(!broken.ok);
assert!(
broken.message.contains("cancel failed"),
"got: {}",
broken.message
);
}
async fn command_outcome(
cmd: DaemonCommand,
reply: Option<&'static str>,
) -> types::DaemonOutcome {
let dir = tempfile::tempdir().unwrap();
let (control, server) = replying_daemon(dir.path(), reply);
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<DaemonCommand>();
let (out_tx, mut out_rx) = mpsc::unbounded_channel();
tokio::spawn(daemon_background_loop(control, cmd_rx, out_tx));
cmd_tx.send(cmd).unwrap();
let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), out_rx.recv())
.await
.expect("an outcome was reported")
.expect("the loop is alive");
let _ = server.await;
outcome
}
#[tokio::test]
async fn daemon_background_loop_forwards_pause_and_resume() {
let ok = command_outcome(
DaemonCommand::Pause {
run_id: "run-1".to_string(),
},
Some(r#"{"result":"ok","ok":true}"#),
)
.await;
assert!(ok.ok);
assert_eq!(ok.run_id, "run-1");
let refused = command_outcome(
DaemonCommand::Pause {
run_id: "run-1".to_string(),
},
Some(r#"{"result":"ok","ok":false}"#),
)
.await;
assert!(!refused.ok);
assert!(refused.message.contains("no such run to pause"));
let ok = command_outcome(
DaemonCommand::Resume {
run_id: "run-1".to_string(),
},
Some(r#"{"result":"ok","ok":true}"#),
)
.await;
assert!(ok.ok);
let refused = command_outcome(
DaemonCommand::Resume {
run_id: "run-1".to_string(),
},
Some(r#"{"result":"ok","ok":false}"#),
)
.await;
assert!(!refused.ok);
assert!(refused.message.contains("no such run to resume"));
}
#[tokio::test]
async fn daemon_background_loop_exits_when_the_dashboard_is_gone() {
let dir = tempfile::tempdir().unwrap();
let (control, _server) = replying_daemon(dir.path(), Some(r#"{"result":"ok","ok":true}"#));
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<DaemonCommand>();
let (out_tx, out_rx) = mpsc::unbounded_channel();
drop(out_rx); let handle = tokio::spawn(daemon_background_loop(control, cmd_rx, out_tx));
cmd_tx
.send(DaemonCommand::Cancel {
run_id: "run-1".to_string(),
})
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), handle)
.await
.expect("the loop returned")
.unwrap();
}
#[tokio::test]
async fn init_dashboard_seeds_startup_log_and_forwards_commands() {
crate::runstate::with_isolated_runs_dir_async(
"init_dashboard_seeds_startup_log",
|_d| async move {
let dashboard = init_dashboard(no_daemon_control(), |_| false);
assert!(
dashboard
.log
.iter()
.any(|entry| entry.message.contains("Dashboard started"))
);
dashboard
.cmd_tx
.send(DaemonCommand::Cancel {
run_id: "nope".to_string(),
})
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
},
)
.await;
}
#[tokio::test]
async fn daemon_background_loop_forwards_cancel() {
let dir = tempfile::tempdir().unwrap();
let (control, server) = recording_daemon(dir.path());
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<DaemonCommand>();
let (out_tx, _out_rx) = mpsc::unbounded_channel();
tokio::spawn(daemon_background_loop(control, cmd_rx, out_tx));
cmd_tx
.send(DaemonCommand::Cancel {
run_id: "run-1".to_string(),
})
.unwrap();
let req = server.await.unwrap();
assert!(req.contains("cancel"));
assert!(req.contains("run-1"));
}
#[tokio::test]
async fn daemon_background_loop_forwards_answer() {
let dir = tempfile::tempdir().unwrap();
let (control, server) = recording_daemon(dir.path());
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<DaemonCommand>();
let (out_tx, _out_rx) = mpsc::unbounded_channel();
tokio::spawn(daemon_background_loop(control, cmd_rx, out_tx));
cmd_tx
.send(DaemonCommand::Answer {
response: leviath_core::interaction::InteractionResponse::text("q1", "yes"),
})
.unwrap();
let req = server.await.unwrap();
assert!(req.contains("answer_interaction"));
assert!(req.contains("q1"));
}
#[tokio::test]
async fn daemon_background_loop_forwards_message() {
let dir = tempfile::tempdir().unwrap();
let (control, server) = recording_daemon(dir.path());
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<DaemonCommand>();
let (out_tx, _out_rx) = mpsc::unbounded_channel();
tokio::spawn(daemon_background_loop(control, cmd_rx, out_tx));
cmd_tx
.send(DaemonCommand::Message {
agent_id: "a1".to_string(),
content: "hi there".to_string(),
})
.unwrap();
let req = server.await.unwrap();
assert!(req.contains("message"));
assert!(req.contains("hi there"));
}
#[tokio::test]
async fn daemon_background_loop_exits_when_channel_dropped() {
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<DaemonCommand>();
let (out_tx, _out_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(daemon_background_loop(no_daemon_control(), cmd_rx, out_tx));
drop(cmd_tx);
let result = tokio::time::timeout(std::time::Duration::from_millis(500), handle).await;
assert!(result.is_ok());
}
#[test]
fn dashboard_new_and_initial_state() {
let dash = make_test_dashboard();
assert!(!dash.should_quit);
assert!(!dash.detail_view);
assert!(!dash.show_help);
}
#[test]
fn dashboard_draw_renders_without_panic() {
use ratatui::Terminal;
use ratatui::backend::TestBackend;
let backend = TestBackend::new(120, 40);
let mut terminal = Terminal::new(backend).unwrap();
let mut dash = make_test_dashboard();
terminal.draw(|f| dash.draw(f)).unwrap();
let buf = crate::commands::dashboard::test_support::rendered_buffer(&terminal);
assert!(buf.contains("Agent Runs"), "{buf}");
}
#[test]
fn dashboard_agent_struct_fields_from_mod() {
let agent = DashboardAgent {
id: "run-test".to_string(),
blueprint_name: "tester".to_string(),
stage: "init".to_string(),
stage_index: 0,
num_stages: 1,
status: AgentDisplayStatus::Idle,
tokens_in: 0,
tokens_out: 0,
cached_tokens: 0,
iteration: 0,
waiting_prompt: None,
pending_request: None,
last_answered_request_id: None,
context_snapshot: None,
stages: vec![],
workdir: "/tmp".to_string(),
task: "test task".to_string(),
title: None,
model: None,
parent_id: None,
depth: 0,
started_at: 0,
last_progress_at: None,
active_until: None,
waiting_secs: 0,
graph_info: None,
accepts_messages: false,
taint_summary: vec![],
};
assert_eq!(agent.id, "run-test");
assert_eq!(agent.blueprint_name, "tester");
assert_eq!(agent.stage, "init");
}
use crate::tui::{TestBackendHarness, TestEventSource, TestSetup, key, test_terminal};
use crossterm::event::KeyCode;
#[tokio::test]
async fn run_dashboard_loop_quits_on_q_from_main_list() {
let mut dashboard = make_test_dashboard();
let control = no_daemon_control();
let mut terminal = test_terminal();
let mouse = |kind, column, row| {
Event::Mouse(crossterm::event::MouseEvent {
kind,
column,
row,
modifiers: crossterm::event::KeyModifiers::NONE,
})
};
use crossterm::event::{MouseButton, MouseEventKind};
let mut events = TestEventSource::new(vec![
Event::Resize(80, 24),
mouse(MouseEventKind::ScrollUp, 0, 0),
mouse(MouseEventKind::ScrollDown, 0, 0),
mouse(MouseEventKind::Moved, 0, 0),
mouse(MouseEventKind::Down(MouseButton::Left), 3, 15),
mouse(MouseEventKind::Drag(MouseButton::Left), 20, 16),
mouse(MouseEventKind::Up(MouseButton::Left), 20, 16),
key(KeyCode::Char('q')),
]);
let result = run_dashboard_loop(
&mut dashboard,
&control,
&mut terminal,
&mut events,
Duration::from_millis(1),
)
.await;
assert!(result.is_ok());
assert!(dashboard.should_quit);
}
#[tokio::test]
async fn run_dashboard_loop_no_event_tick_then_quits() {
let mut dashboard = make_test_dashboard();
let control = no_daemon_control();
let mut terminal = test_terminal();
let mut events = TestEventSource::new_with_nones(vec![None, Some(key(KeyCode::Char('q')))]);
let result = run_dashboard_loop(
&mut dashboard,
&control,
&mut terminal,
&mut events,
Duration::from_millis(1),
)
.await;
assert!(result.is_ok());
assert!(dashboard.should_quit);
}
#[tokio::test]
async fn run_dashboard_loop_ignores_non_press_and_other_events() {
let mut dashboard = make_test_dashboard();
let control = no_daemon_control();
let mut terminal = test_terminal();
let release = Event::Key(crossterm::event::KeyEvent::new_with_kind(
KeyCode::Char('x'),
crossterm::event::KeyModifiers::empty(),
crossterm::event::KeyEventKind::Release,
));
let mut events =
TestEventSource::new(vec![release, Event::FocusGained, key(KeyCode::Char('q'))]);
let result = run_dashboard_loop(
&mut dashboard,
&control,
&mut terminal,
&mut events,
Duration::from_millis(1),
)
.await;
assert!(result.is_ok());
assert!(dashboard.should_quit);
}
#[tokio::test]
async fn run_dashboard_loop_propagates_event_source_error() {
let mut dashboard = make_test_dashboard();
let control = no_daemon_control();
let mut terminal = test_terminal();
let mut events = TestEventSource::failing();
let result = run_dashboard_loop(
&mut dashboard,
&control,
&mut terminal,
&mut events,
Duration::from_millis(1),
)
.await;
assert!(result.is_err());
}
#[tokio::test]
async fn run_dashboard_loop_propagates_draw_error() {
let mut dashboard = make_test_dashboard();
let control = no_daemon_control();
let mut terminal = Terminal::new(TestBackendHarness::failing(120, 40)).unwrap();
let mut events = TestEventSource::new(vec![]);
let result = run_dashboard_loop(
&mut dashboard,
&control,
&mut terminal,
&mut events,
Duration::from_millis(1),
)
.await;
assert!(result.is_err());
}
#[tokio::test]
async fn execute_core_happy_path_quits_on_esc() {
crate::runstate::with_isolated_runs_dir_async(
"execute_core_happy_path_quits_on_esc",
|_d| async move {
let control = no_daemon_control();
let mut dashboard = init_dashboard(control.clone(), |_| false);
let mut setup = TestSetup::new();
let mut events = TestEventSource::new(vec![key(KeyCode::Char('q'))]);
let result = execute_core(&mut dashboard, &control, &mut setup, &mut events).await;
assert!(result.is_ok());
assert!(dashboard.should_quit);
},
)
.await;
}
#[tokio::test]
async fn execute_with_loads_config_inits_and_runs_the_loop() {
crate::config::with_isolated_config_path_async(
"execute_with_dashboard",
|_fake_dir| async move {
crate::runstate::with_isolated_runs_dir_async(
"execute_with_dashboard",
|_d| async move {
let mut setup = TestSetup::new();
let mut events = TestEventSource::new(vec![key(KeyCode::Char('q'))]);
let result =
execute_with(no_daemon_control(), &mut setup, &mut events, |_| false)
.await;
assert!(result.is_ok());
},
)
.await;
},
)
.await;
}
#[tokio::test]
async fn execute_core_enable_error_propagates() {
crate::runstate::with_isolated_runs_dir_async(
"execute_core_enable_error_propagates",
|_d| async move {
let control = no_daemon_control();
let mut dashboard = init_dashboard(control.clone(), |_| false);
let mut setup = TestSetup {
enable_should_fail: true,
create_should_fail: false,
draw_should_fail: false,
};
let mut events = TestEventSource::new(vec![]);
let result = execute_core(&mut dashboard, &control, &mut setup, &mut events).await;
assert!(result.is_err());
},
)
.await;
}
#[tokio::test]
async fn execute_core_create_terminal_error_propagates() {
crate::runstate::with_isolated_runs_dir_async(
"execute_core_create_terminal_error_propagates",
|_d| async move {
let control = no_daemon_control();
let mut dashboard = init_dashboard(control.clone(), |_| false);
let mut setup = TestSetup {
enable_should_fail: false,
create_should_fail: true,
draw_should_fail: false,
};
let mut events = TestEventSource::new(vec![]);
let result = execute_core(&mut dashboard, &control, &mut setup, &mut events).await;
assert!(result.is_err());
},
)
.await;
}
#[tokio::test]
async fn execute_core_loop_error_propagates() {
crate::runstate::with_isolated_runs_dir_async(
"execute_core_loop_error_propagates",
|_d| async move {
let control = no_daemon_control();
let mut dashboard = init_dashboard(control.clone(), |_| false);
let mut setup = TestSetup::new();
let mut events = TestEventSource::failing();
let result = execute_core(&mut dashboard, &control, &mut setup, &mut events).await;
assert!(result.is_err());
},
)
.await;
}
}