hive-agent: test the in-agent socket's request dispatch

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<T>(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<OsString>);
-    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<std::ffi::OsString>,
+    _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"),
+            }
+        }
+    }
 }

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<LooseEnd> {
+        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<i64> = 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;