Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 41 additions & 42 deletions src/app.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ use crate::theme::Theme;
/// needs to hand back to the main thread for status / conflict / budget
/// post-processing. All fields are owned values so the struct is `Send`
/// and can be returned across `tokio::task::spawn_blocking`.
#[derive(Default)]
pub struct RefreshIoOutput {
pub sessions: Vec<AgentSession>,
pub new_pids: Vec<u32>,
Expand Down Expand Up @@ -900,6 +901,10 @@ pub struct App {
/// duplicates each tick — the next tick reuses the existing
/// worker's eventual result. Cleared once the result is recv'd.
pub refresh_in_flight: bool,
/// The I/O-heavy pass a refresh runs. Always [`do_refresh_io`] in
/// production; a seam so tests can exercise the scheduling and channel
/// plumbing without scanning the host's real `~/.claude`.
pub(crate) refresh_worker: fn(Vec<AgentSession>) -> RefreshIoOutput,
pub table_state: TableState,
pub should_quit: bool,
pub status_msg: String,
Expand Down Expand Up @@ -1061,13 +1066,16 @@ impl App {
self.replace_data(new_data);
}

/// No session scan, no ledger read, no background thread — only the
/// parked-session file is read. Use [`App::with_host_state`] for a real run.
pub fn new() -> Self {
let (refresh_tx, refresh_rx) = tokio::sync::mpsc::unbounded_channel();
let mut app = Self {
Self {
data: Arc::new(RwLock::new(Arc::new(AppData::default()))),
refresh_tx,
refresh_rx,
refresh_in_flight: false,
refresh_worker: do_refresh_io,
table_state: TableState::default(),
should_quit: false,
status_msg: String::new(),
Expand Down Expand Up @@ -1133,7 +1141,13 @@ impl App {
idle_mode_active: false,
idle_tasks_launched: Vec::new(),
idle_report: Vec::new(),
};
}
}

/// Construct and populate from the host: the session scan, ledger
/// rollups, and background usage scan a real run needs.
pub fn with_host_state() -> Self {
let mut app = Self::new();
app.refresh();
// Seed rollups from any rows already in the CSV — instant via
// the in-memory ledger cache. Don't block startup on
Expand Down Expand Up @@ -1169,7 +1183,7 @@ impl App {
// `do_refresh_io` on a worker and routes the result through a
// channel, then calls `apply_refresh_output` on the main thread.
let prev_sessions = self.data_snapshot().sessions.clone();
let out = do_refresh_io(prev_sessions);
let out = (self.refresh_worker)(prev_sessions);
self.apply_refresh_output(out, tick_start);
}

Expand Down Expand Up @@ -1211,17 +1225,18 @@ impl App {
// in a runtime context (tests, one-shot CLI), fall back to a
// synchronous refresh on the calling thread.
let prev_sessions = self.data_snapshot().sessions.clone();
let worker = self.refresh_worker;
if let Ok(handle) = tokio::runtime::Handle::try_current() {
let tx = self.refresh_tx.clone();
handle.spawn_blocking(move || {
let out = do_refresh_io(prev_sessions);
let out = worker(prev_sessions);
let _ = tx.send(out);
});
self.refresh_in_flight = true;
} else if !applied_any {
// No runtime AND we didn't already apply a result this tick:
// run synchronously so the caller still gets fresh data.
let out = do_refresh_io(prev_sessions);
let out = worker(prev_sessions);
self.apply_refresh_output(out, std::time::Instant::now());
applied_any = true;
}
Expand Down Expand Up @@ -4148,19 +4163,9 @@ mod tests {
app
}

/// Build an App with empty AppData, bypassing the host-side
/// session discovery that `App::new()` does in its constructor.
/// The data-helper tests below assert on the post-replace state
/// directly, so we don't want real sessions leaking in.
fn app_with_empty_data() -> App {
let app = App::new();
app.replace_data(AppData::default());
app
}

#[test]
fn data_snapshot_returns_empty_appdata_after_clear() {
let app = app_with_empty_data();
let app = App::new();
let snap = app.data_snapshot();
assert!(snap.sessions.is_empty());
assert_eq!(snap.ledger_today.msg_count, 0);
Expand All @@ -4170,7 +4175,7 @@ mod tests {

#[test]
fn replace_data_swaps_atomically_and_takes_effect_for_next_snapshot() {
let app = app_with_empty_data();
let app = App::new();
let snap_before = app.data_snapshot();
assert!(snap_before.sessions.is_empty());

Expand Down Expand Up @@ -4202,7 +4207,7 @@ mod tests {

#[test]
fn modify_data_mutates_in_place_and_visible_to_subsequent_snapshot() {
let app = app_with_empty_data();
let app = App::new();
app.modify_data(|d| {
d.ledger_week.fresh_input = 12345;
d.sessions.push(make_session(
Expand Down Expand Up @@ -4253,51 +4258,45 @@ mod tests {

#[test]
fn refresh_nonblocking_falls_back_to_sync_when_no_runtime() {
// No tokio runtime is active in unit tests by default — the
// function must therefore run `do_refresh_io` synchronously
// and return `true` (work was applied) with refresh_in_flight
// cleared, rather than dispatching to a worker that never runs.
let mut app = app_with_empty_data();
// With no runtime the work must run on the calling thread and be
// applied before the call returns, not dispatched to a worker that
// never runs.
let mut app = App::new();
app.refresh_worker = |_| RefreshIoOutput::default();
assert!(app.refresh_nonblocking());
assert!(!app.refresh_in_flight);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn refresh_nonblocking_kicks_worker_under_runtime() {
// With a multi-threaded tokio runtime the first call kicks
// a `do_refresh_io` worker onto the blocking pool. Subsequent
// calls return false until the worker sends its result back
// through the channel; once the result is applied,
// refresh_nonblocking returns true. We poll on the return
// value rather than `refresh_in_flight` because each call
// also schedules the NEXT worker, so the flag is true on
// We poll the return value rather than `refresh_in_flight` because
// each call also schedules the NEXT worker, so the flag is true on
// exit even after a successful drain.
//
// Note: `do_refresh_io` reads the real host filesystem
// (~/.claude/sessions, ~/.claude/projects). On a heavy box
// the cold pass can take seconds; the 60 s deadline is
// generous to keep this test reliable on slow CI/sandboxes.
let mut app = app_with_empty_data();
let mut app = App::new();
app.refresh_worker = |_| RefreshIoOutput::default();

let kicked = app.refresh_nonblocking();
assert!(!kicked, "first call only schedules; nothing applied yet");
assert!(app.refresh_in_flight, "worker must be scheduled");

let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
// A hang guard, not a performance bound: the worker is in-memory.
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
loop {
tokio::task::yield_now().await;
if app.refresh_nonblocking() {
break;
}
if std::time::Instant::now() >= deadline {
panic!("refresh worker did not complete within 60 s");
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
assert!(
std::time::Instant::now() < deadline,
"worker result never arrived through the channel"
);
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
}

#[test]
fn with_sessions_preserves_ledger_fields() {
let app = app_with_empty_data();
let app = App::new();
app.modify_data(|d| {
d.ledger_month.cost_usd = 999.99;
});
Expand Down
40 changes: 21 additions & 19 deletions src/commands.rs
Original file line number Diff line number Diff line change
Expand Up @@ -914,10 +914,22 @@ pub(crate) fn run_clean(
.map(std::path::PathBuf::from)
.unwrap_or_else(|| std::path::PathBuf::from("/tmp"));

// Collect active PIDs to avoid deleting live sessions
let active_pids: std::collections::HashSet<u32> = {
let app = App::new();
app.data_snapshot().sessions.iter().map(|s| s.pid).collect()
// One host scan for both liveness questions. The `--finished` filter below
// runs per JSONL file, so asking it to build its own App cost a full scan
// and a `ps` fork for every file in `~/.claude/projects`.
let (active_pids, active_jsonl): (
std::collections::HashSet<u32>,
std::collections::HashSet<std::path::PathBuf>,
) = {
let app = App::with_host_state();
let snap = app.data_snapshot();
(
snap.sessions.iter().map(|s| s.pid).collect(),
snap.sessions
.iter()
.filter_map(|s| s.jsonl_path.clone())
.collect(),
)
};

let mut removed_sessions = 0u64;
Expand Down Expand Up @@ -999,18 +1011,8 @@ pub(crate) fn run_clean(
}

// If --finished only, skip JSONL files whose corresponding session is still active
if finished_only {
// Check if any active session is using this JSONL
let app = App::new();
let is_active = app.data_snapshot().sessions.iter().any(|s| {
s.jsonl_path
.as_ref()
.map(|p| p == &file_path)
.unwrap_or(false)
});
if is_active {
continue;
}
if finished_only && active_jsonl.contains(&file_path) {
continue;
}

let size = metadata.len();
Expand Down Expand Up @@ -1055,7 +1057,7 @@ pub(crate) fn run_clean(

pub(crate) fn print_summary(since: &str) -> io::Result<()> {
let since_duration = parse_duration_str(since);
let app = App::new();
let app = App::with_host_state();
let snap = app.data_snapshot();

if snap.sessions.is_empty() {
Expand Down Expand Up @@ -1193,7 +1195,7 @@ fn make_app(demo: bool, filters: &ViewFilters) -> App {
});
app
} else {
App::new()
App::with_host_state()
};
apply_filters(&mut app, filters);
app
Expand Down Expand Up @@ -1273,7 +1275,7 @@ pub(crate) fn run_watch(
use crate::session::SessionStatus;
use std::collections::HashMap;

let mut app = App::new();
let mut app = App::with_host_state();
apply_filters(&mut app, filters);
let mut prev_statuses: HashMap<u32, SessionStatus> = app
.data_snapshot()
Expand Down
4 changes: 2 additions & 2 deletions src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -766,7 +766,7 @@ fn run_tui<W: io::Write>(
max_duration: Option<Duration>,
auto_init_summary: Option<String>,
) -> io::Result<()> {
let mut app = App::new();
let mut app = App::with_host_state();
app.notify = cfg.notify;
app.debug = cfg.debug;
app.webhook_url = cfg.webhook.clone();
Expand Down Expand Up @@ -820,7 +820,7 @@ fn run_tui<W: io::Write>(
config::BrainConfig::default(),
));
}
// Re-refresh to replace real sessions discovered during App::new()
// Replace the real sessions the host scan found with demo ones.
app.refresh();
}

Expand Down
Loading