diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index f97db0ad0..76a424a82 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -138,31 +138,50 @@ pub fn claude_dir() -> PathBuf { /// are serialised on a lock. #[cfg(test)] pub fn with_harness_dir(f: impl FnOnce() -> T) -> T { - use std::ffi::OsString; + let _guard = harness_dir_guard(); + f() +} + +/// [`with_harness_dir`] for a test body that cannot be a closure: the harness +/// dir stays set, and the lock stays held, until the guard drops. +#[cfg(test)] +pub fn harness_dir_guard() -> HarnessDirGuard { use std::sync::Mutex; static LOCK: Mutex<()> = Mutex::new(()); - struct Restore(Option); - impl Drop for Restore { - fn drop(&mut self) { - // SAFETY: still holding `LOCK`, the only writer of this variable. - unsafe { - match self.0.take() { - Some(v) => std::env::set_var("HYPERHIVE_HARNESS_DIR", v), - None => std::env::remove_var("HYPERHIVE_HARNESS_DIR"), - } - } - } - } - - let _lock = LOCK + let lock = LOCK .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let dir = tempfile::tempdir().unwrap(); - // Declared after `_lock` so the variable is restored before the lock drops. - let _restore = Restore(std::env::var_os("HYPERHIVE_HARNESS_DIR")); + let previous = std::env::var_os("HYPERHIVE_HARNESS_DIR"); // SAFETY: serialised by `LOCK`. unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", dir.path()) }; - f() + HarnessDirGuard { + previous, + _dir: dir, + _lock: lock, + } +} + +/// Holds `HYPERHIVE_HARNESS_DIR` at a fresh temp dir; see [`harness_dir_guard`]. +#[cfg(test)] +pub struct HarnessDirGuard { + previous: Option, + _dir: tempfile::TempDir, + // Declared last so the variable is restored (in `drop`) before the lock goes. + _lock: std::sync::MutexGuard<'static, ()>, +} + +#[cfg(test)] +impl Drop for HarnessDirGuard { + fn drop(&mut self) { + // SAFETY: still holding the lock, the only writer of this variable. + unsafe { + match self.previous.take() { + Some(v) => std::env::set_var("HYPERHIVE_HARNESS_DIR", v), + None => std::env::remove_var("HYPERHIVE_HARNESS_DIR"), + } + } + } } diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 4746f5f02..a1e6cfdf8 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -710,6 +710,293 @@ mod tests { (resp, handler.await.expect("join")) } + /// What keeps a test's stores and `Bus` alive: the scratch dir, and the + /// harness dir a `Bus` opens its events db in (unset in the build sandbox). + type Scratch = (tempfile::TempDir, crate::paths::HarnessDirGuard); + + /// Todo and reminder stores over a scratch dir, which must outlive them. + fn stores() -> (Scratch, Todos, Reminders) { + let harness = crate::paths::harness_dir_guard(); + let dir = tempfile::tempdir().expect("tempdir"); + let todos = Todos::open(&dir.path().join("todos.sqlite")).expect("todos"); + let reminders = Reminders::open(&dir.path().join("reminders.sqlite")).expect("reminders"); + ((dir, harness), todos, reminders) + } + + fn upsert(subsystem: &str, key: Option<&str>, summary: &str) -> Request { + Request::UpsertTodo { + subsystem: subsystem.to_owned(), + key: key.map(str::to_owned), + summary: summary.to_owned(), + source: None, + reopen_if_acked: false, + } + } + + fn list(todos: &Todos, subsystem: Option<&str>) -> Vec { + let req = Request::ListTodos { + subsystem: subsystem.map(str::to_owned), + }; + match dispatch(req, todos, &Notify::new(), None, &Bus::new()) { + Response::LooseEnds { loose_ends } => loose_ends, + other => panic!("expected loose ends, got {other:?}"), + } + } + + /// Whether a wake is pending, consuming it. `notify_one` leaves a permit + /// behind, so one poll is enough: nothing here needs a clock. + fn woke(wake: &Notify) -> bool { + use futures_util::FutureExt as _; + wake.notified().now_or_never().is_some() + } + + fn acked(resp: &Response) -> u64 { + match resp { + Response::Acked { count } => *count, + other => panic!("expected Acked, got {other:?}"), + } + } + + #[test] + fn an_upsert_wakes_the_loop_only_when_it_is_new_or_changed() { + let (_dir, todos, _) = stores(); + let (wake, bus) = (Notify::new(), Bus::new()); + let up = |summary| { + dispatch( + upsert("matrix", Some("!r"), summary), + &todos, + &wake, + None, + &bus, + ) + }; + + assert!(matches!(up("one unread"), Response::Ok)); + assert!(woke(&wake), "a new row wakes the loop"); + assert!(matches!(up("one unread"), Response::Ok)); + assert!(!woke(&wake), "an identical keyed re-push stays quiet"); + assert!(matches!(up("two unread"), Response::Ok)); + assert!(woke(&wake), "a changed summary wakes it again"); + assert_eq!(list(&todos, None).len(), 1, "one keyed row, not three"); + } + + #[test] + fn listing_filters_by_subsystem_and_carries_the_row() { + let (_dir, todos, _) = stores(); + let (wake, bus) = (Notify::new(), Bus::new()); + dispatch(upsert("matrix", Some("!r"), "m"), &todos, &wake, None, &bus); + dispatch(upsert("bash", Some("t1"), "b"), &todos, &wake, None, &bus); + + assert_eq!(list(&todos, None).len(), 2); + let bash = list(&todos, Some("bash")); + let [ + LooseEnd::Todo { + subsystem, + subsystem_key, + summary, + .. + }, + ] = bash.as_slice() + else { + panic!("one bash todo"); + }; + assert_eq!( + ( + subsystem.as_str(), + subsystem_key.as_deref(), + summary.as_str() + ), + ("bash", Some("t1"), "b") + ); + } + + #[test] + fn clearing_drops_one_keyed_row_or_the_whole_subsystem() { + let (_dir, todos, _) = stores(); + let (wake, bus) = (Notify::new(), Bus::new()); + for (sub, key) in [("matrix", "!a"), ("matrix", "!b"), ("bash", "t1")] { + dispatch(upsert(sub, Some(key), "x"), &todos, &wake, None, &bus); + } + let clear = |key: Option<&str>, all| Request::ClearTodo { + subsystem: "matrix".to_owned(), + key: key.map(str::to_owned), + all, + }; + + let one = dispatch(clear(Some("!a"), false), &todos, &wake, None, &bus); + assert_eq!(acked(&one), 1); + assert_eq!(list(&todos, Some("matrix")).len(), 1); + let all = dispatch(clear(None, true), &todos, &wake, None, &bus); + assert_eq!(acked(&all), 1); + assert!(list(&todos, Some("matrix")).is_empty()); + assert_eq!(list(&todos, Some("bash")).len(), 1, "other subsystems stay"); + } + + #[test] + fn marking_done_acks_the_given_ids_and_hides_them() { + let (_dir, todos, _) = stores(); + let (wake, bus) = (Notify::new(), Bus::new()); + for key in ["a", "b", "c"] { + dispatch(upsert("s", Some(key), key), &todos, &wake, None, &bus); + } + let ids: Vec = list(&todos, None) + .into_iter() + .map(|t| match t { + LooseEnd::Todo { id, .. } => id, + other => panic!("expected a Todo, got {other:?}"), + }) + .collect(); + + let one = dispatch( + Request::MarkTodoDone { id: ids[0] }, + &todos, + &wake, + None, + &bus, + ); + assert_eq!(acked(&one), 1); + let many = dispatch( + Request::MarkTodosDone { + ids: ids[1..].to_vec(), + }, + &todos, + &wake, + None, + &bus, + ); + assert_eq!(acked(&many), 2); + assert!(list(&todos, None).is_empty()); + let again = dispatch( + Request::MarkTodoDone { id: ids[0] }, + &todos, + &wake, + None, + &bus, + ); + assert_eq!(acked(&again), 0, "an already-done id acks nothing"); + } + + #[test] + fn every_reminder_op_errors_when_the_store_did_not_open() { + let (_dir, todos, _) = stores(); + let (wake, bus) = (Notify::new(), Bus::new()); + let reqs = [ + Request::StoreReminder { + message: "m".to_owned(), + timing: hive_sh4re::inbox::ReminderTiming::InSeconds { seconds: 60 }, + file_path: None, + }, + Request::ListReminders, + Request::CancelReminder { id: 1 }, + Request::CountPendingReminders, + Request::ReminderRollup { since_secs: 0 }, + ]; + for req in reqs { + let resp = dispatch(req, &todos, &wake, None, &bus); + assert!( + matches!(&resp, Response::Err { message } if message.contains("unavailable")), + "{resp:?}" + ); + } + } + + #[test] + fn a_reminder_is_stored_listed_counted_rolled_up_and_cancelled() { + let (_dir, todos, reminders) = stores(); + let (wake, bus) = (Notify::new(), Bus::new()); + let run = |req| dispatch(req, &todos, &wake, Some(&reminders), &bus); + + let stored = run(Request::StoreReminder { + message: "check the deploy".to_owned(), + timing: hive_sh4re::inbox::ReminderTiming::InSeconds { seconds: 3600 }, + file_path: None, + }); + assert!(matches!(stored, Response::Ok), "{stored:?}"); + assert!(matches!( + run(Request::CountPendingReminders), + Response::PendingRemindersCount { count: 1 } + )); + let Response::LooseEnds { loose_ends } = run(Request::ListReminders) else { + panic!("expected loose ends"); + }; + let [LooseEnd::Reminder { id, message, .. }] = loose_ends.as_slice() else { + panic!("one reminder, got {loose_ends:?}"); + }; + assert_eq!(message, "check the deploy"); + assert!(matches!( + run(Request::ReminderRollup { since_secs: 0 }), + Response::ReminderRollup { .. } + )); + + assert_eq!(acked(&run(Request::CancelReminder { id: *id })), 1); + assert_eq!(acked(&run(Request::CancelReminder { id: *id })), 0); + assert!(matches!( + run(Request::CountPendingReminders), + Response::PendingRemindersCount { count: 0 } + )); + } + + /// A bus whose context window is 100 tokens and whose last turn used + /// `input` of them. + fn bus_with_context(input: u64) -> Bus { + let bus = Bus::new(); + bus.set_api_context_window(100); + let usage: hive_claude::TokenUsage = serde_json::from_value(serde_json::json!({ + "input_tokens": input, + "output_tokens": 0, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + })) + .expect("a usage block"); + bus.record_turn_usage(usage, usage); + bus + } + + fn compact_req(prompt: Option<&str>) -> Request { + Request::Compact { + wake_prompt: prompt.map(str::to_owned), + } + } + + #[test] + fn compact_is_refused_before_any_turn_and_below_the_usage_floor() { + let (_dir, todos, _) = stores(); + let wake = Notify::new(); + + let fresh = Bus::new(); + let resp = dispatch(compact_req(None), &todos, &wake, None, &fresh); + assert!( + matches!(&resp, Response::Err { message } if message.contains("no completed turn")), + "{resp:?}" + ); + assert!(fresh.take_compact().is_none()); + + let low = bus_with_context(10); + let resp = dispatch(compact_req(None), &todos, &wake, None, &low); + assert!( + matches!(&resp, Response::Err { message } if message.contains("below")), + "{resp:?}" + ); + assert!(low.take_compact().is_none(), "a refusal queues nothing"); + } + + #[test] + fn compact_above_the_floor_queues_the_request_with_its_wake_prompt() { + let (_dir, todos, _) = stores(); + let bus = bus_with_context(90); + let resp = dispatch( + compact_req(Some("carry on")), + &todos, + &Notify::new(), + None, + &bus, + ); + assert!(matches!(resp, Response::Ok), "{resp:?}"); + let queued = bus.take_compact().expect("a queued compact"); + assert_eq!(queued.wake_prompt.as_deref(), Some("carry on")); + assert!(bus.take_compact().is_none(), "taken once"); + } + #[tokio::test] async fn a_valid_request_is_applied() { let (resp, done) = exchange(b"{\"cmd\":\"compact\"}\n".to_vec(), REQUEST_DEADLINES).await;