diff --git a/Cargo.lock b/Cargo.lock index 0d5b1376..7d573164 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1523,7 +1523,6 @@ dependencies = [ "clap", "forgejo-api", "futures-util", - "hive-agent-sock", "hive-claude", "hive-core-agent-sock", "hive-sh4re", @@ -1553,7 +1552,6 @@ dependencies = [ "anyhow", "axum", "clap", - "hive-agent-sock", "hive-core-agent-sock", "hive-sh4re", "rmcp", @@ -1564,14 +1562,6 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "hive-agent-sock" -version = "0.1.0" -dependencies = [ - "hive-sh4re", - "serde", -] - [[package]] name = "hive-agent-wake" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 5ff5da9c..549abeb5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,6 @@ resolver = "3" members = [ "hive-agent", "hive-agent-mcp", - "hive-agent-sock", "hive-core-agent-sock", "hive-agent-wake", "hive-bash-mcp", @@ -52,7 +51,6 @@ clap = { version = "4", features = ["derive"] } clap_complete = "4" indicatif = "0.18" hive-sh4re = { path = "hive-sh4re" } -hive-agent-sock = { path = "hive-agent-sock" } hive-core-agent-sock = { path = "hive-core-agent-sock" } hive-claude = { path = "hive-claude" } hive-host-sock = { path = "hive-host-sock" } diff --git a/hive-agent-mcp/Cargo.toml b/hive-agent-mcp/Cargo.toml index 33576ddf..6fad2eb9 100644 --- a/hive-agent-mcp/Cargo.toml +++ b/hive-agent-mcp/Cargo.toml @@ -14,7 +14,6 @@ workspace = true anyhow.workspace = true axum.workspace = true clap.workspace = true -hive-agent-sock.workspace = true hive-core-agent-sock.workspace = true hive-sh4re.workspace = true rmcp.workspace = true diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 8acc2c6e..5ac7c35d 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -35,8 +35,8 @@ pub use args::{ pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; use render::{ - format_matrix_summary, local_todos, loose_end_kind_label, matrix_unread_summary, - parse_loose_end_kind, render_loose_ends, reply_err, + format_matrix_summary, loose_end_kind_label, matrix_unread_summary, parse_loose_end_kind, + render_loose_ends, reply_err, }; /// Write (or remove) the status file in the agent's own `state/` directory. @@ -333,12 +333,6 @@ impl AgentServer { ); } } - // Merge the harness's local todos (loose-ends v2) for self-queries. - // The harness owns the todo store in-container; another agent's - // todos aren't reachable from here (same as matrix above). - if is_self_query && let Some(todos) = local_todos().await { - loose_ends.extend(todos); - } let mut out = annotate_retries(render_loose_ends(&loose_ends), retries); // Append loose-end items published by external MCP daemons // (e.g. active bash tasks from hive-bash-mcp). Generic — no diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index 858e75ba..dddc9eca 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -297,30 +297,6 @@ pub(super) async fn matrix_unread_summary() -> Option> { serde_json::from_value(serde_json::Value::Array(arr.clone())).ok() } -/// Query the harness's in-agent socket for this agent's local todos -/// (loose-ends v2). Returns `None` when `HIVE_AGENT_SOCKET` is unset / -/// absent or the query fails — best-effort, like [`matrix_unread_summary`], -/// so an agent without the socket is not penalised. -pub(super) async fn local_todos() -> Option> { - use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; - use tokio::net::UnixStream; - let socket = std::env::var_os("HIVE_AGENT_SOCKET").map(std::path::PathBuf::from)?; - if !socket.exists() { - return None; - } - let mut stream = UnixStream::connect(&socket).await.ok()?; - let mut req = - serde_json::to_string(&hive_agent_sock::Request::ListTodos { subsystem: None }).ok()?; - req.push('\n'); - stream.write_all(req.as_bytes()).await.ok()?; - let mut lines = BufReader::new(stream).lines(); - let line = lines.next_line().await.ok()??; - match serde_json::from_str::(&line).ok()? { - hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends), - _ => None, - } -} - /// Format a `Vec` into a per-room summary string. /// Single room / single message collapses to one line; multi-room /// expands to a bulleted list. Returns an empty string for empty input. diff --git a/hive-agent-sock/Cargo.toml b/hive-agent-sock/Cargo.toml deleted file mode 100644 index 7818923d..00000000 --- a/hive-agent-sock/Cargo.toml +++ /dev/null @@ -1,11 +0,0 @@ -[package] -name = "hive-agent-sock" -edition.workspace = true -version.workspace = true - -[lints] -workspace = true - -[dependencies] -serde.workspace = true -hive-sh4re.workspace = true diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs deleted file mode 100644 index dc7bd7a8..00000000 --- a/hive-agent-sock/src/lib.rs +++ /dev/null @@ -1,76 +0,0 @@ -//! Wire types for the *in-agent* socket, served by the hive-agent harness -//! to the in-container producers (matrix / bash MCP daemons) and -//! `forge_notify`. Currently carries the loose-ends-v2 *todo* op family; -//! more in-agent request families may be added over time (the socket is -//! deliberately named for the agent, not the todos). -//! -//! Distinct from `hive-core-agent-sock`, the *host*-served core↔agent -//! protocol on `/run/hive/mcp.sock`: this socket never leaves the -//! container. The harness owns the todo store locally and signals its own -//! turn loop directly, so hive-c0re is not in the todo path — no broker -//! round-trip, no long-poll, no marker files. - -use serde::{Deserialize, Serialize}; - -use hive_sh4re::LooseEnd; - -/// In-container path of the harness-served in-agent socket. The harness -/// binds it on boot; the in-container producers dial it for todo ops. -/// (Placeholder default — the harness + producers resolve the real path -/// from config; kept here so a producer with no override has a sane one.) -pub const DEFAULT_AGENT_SOCKET: &str = "/run/hive/agent.sock"; - -/// A request on the in-agent socket. Serialised with a `cmd` tag so the -/// in-container producers can emit a plain JSON line without linking a -/// typed client (matrix/bash build the JSON by hand). -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "cmd", rename_all = "snake_case")] -pub enum Request { - /// Upsert a todo from an in-container subsystem (matrix / bash / - /// forge). `subsystem` is the producer marker; `key` the optional - /// subsystem-specific dedup key (a matrix room id, a bash task id). - /// A new-or-changed row signals the turn loop; an identical keyed - /// re-push is a silent no-op. Keyless todos always insert as one-offs. - UpsertTodo { - subsystem: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - key: Option, - summary: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - source: Option, - }, - /// Clear producer-resolved todo(s). `key = Some(k)` clears the one - /// keyed row; `key = None` clears the subsystem's keyless rows; `all - /// = true` wipes the producer's whole set (cancel-and-recreate on - /// daemon restart). - ClearTodo { - subsystem: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - key: Option, - #[serde(default)] - all: bool, - }, - /// List todos, optionally filtered to one `subsystem` (a producer - /// reconciling its own set). `None` = all. - ListTodos { - #[serde(default, skip_serializing_if = "Option::is_none")] - subsystem: Option, - }, - /// The agent marks one of its own todos done, by id. - MarkTodoDone { id: i64 }, -} - -/// A response on the in-agent socket. Serialised with a `kind` tag, -/// mirroring the core↔agent protocol's response shape. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum Response { - /// Op succeeded, no payload. - Ok, - /// Op succeeded and touched `count` rows (clear / mark-done). - Acked { count: u64 }, - /// `ListTodos` result. - LooseEnds { loose_ends: Vec }, - /// Op failed; `message` is operator-facing. - Err { message: String }, -} diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index e09c6da8..2b0c74c1 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -19,7 +19,6 @@ time.workspace = true futures-util = "0.3" clap.workspace = true hive-claude.workspace = true -hive-agent-sock.workspace = true hive-core-agent-sock.workspace = true hive-sh4re.workspace = true rmcp.workspace = true diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 4c241696..69b8b0b1 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -23,8 +23,6 @@ mod prompt; mod serve_common; mod stats; mod stream_enrich; -mod todo_server; -mod todos; mod turn; mod turn_stats; mod vacuum; @@ -171,22 +169,6 @@ fn synthetic_continue() -> hive_sh4re::DeliveredMessage { } } -/// Synthesize the message that drives a turn when an in-container producer -/// upserted a new/changed *todo* over the in-agent socket (loose-ends v2). -/// The harness owns the todo store locally and signals the serve loop -/// directly — so this wake never touches the broker (no long-poll, no -/// marker file). `id = 0` is the same non-broker sentinel as -/// [`synthetic_continue`]. -fn synthetic_todo_message() -> hive_sh4re::DeliveredMessage { - hive_sh4re::DeliveredMessage { - from: "todo".into(), - body: "you have todos — call get_loose_ends to see them".into(), - id: 0, - redelivered: false, - in_reply_to: None, - } -} - /// Synthetic message that drives the single stop-checkpoint turn when c0re /// signals a graceful stop. The agent gets one final turn to flush durable /// `/state` before the container is stopped; new inbound is already fenced. @@ -226,10 +208,6 @@ enum RecvOutcome { /// one stop-checkpoint turn (flush durable `/state`), reports /// `GracefulStopComplete`, and exits so the container can be stopped. GracefulStop, - /// An in-container producer upserted a new/changed todo over the - /// in-agent socket; the serve loop drives a `synthetic_todo_message` - /// turn. Not a broker message — the harness signalled itself directly. - LocalTodo, } /// Wire surface abstraction. `AgentSurface` is the only impl — the trait @@ -445,26 +423,6 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { tracing::error!(error = %e, "web_ui::serve exited with error"); } }); - // In-agent todo socket (loose-ends v2): the harness owns the todo store - // locally and serves the in-container producers on `HIVE_AGENT_SOCKET`. - // A new/changed upsert fires `todo_wake` so the serve loop drives a turn - // directly — no broker round-trip, no marker files. Best-effort: if the - // store can't open, the socket just isn't served. - let todo_wake = Arc::new(tokio::sync::Notify::new()); - match todos::Todos::open(&paths::todos_db()) { - Ok(store) => { - let store = Arc::new(store); - let wake = todo_wake.clone(); - tokio::spawn(async move { - if let Err(e) = todo_server::run(store, wake).await { - tracing::error!(error = %e, "in-agent todo socket exited with error"); - } - }); - } - Err(e) => { - tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled"); - } - } if matches!(initial, LoginState::NeedsLogin) { login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; } else { @@ -481,7 +439,6 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { bus, stats, &files, - todo_wake, ) .await } @@ -503,7 +460,6 @@ async fn serve_loop( bus: Bus, stats: Option, files: &turn::TurnFiles, - todo_wake: Arc, ) -> Result<()> { tracing::info!(socket = %socket.display(), "harness serve"); S::requeue_inflight(socket).await; @@ -518,20 +474,8 @@ async fn serve_loop( loop { let next = match self_continue.take() { Some(msg) => msg, - None => match { - // Idle wait: race the broker long-poll against a local - // todo signal so an in-container producer's upsert drives a - // turn without any broker round-trip. `biased` polls the - // broker recv first, so a genuinely-ready inbox message is - // never dropped in favour of the todo wake. - tokio::select! { - biased; - o = S::recv_next(socket) => o, - () = todo_wake.notified() => RecvOutcome::LocalTodo, - } - } { + None => match S::recv_next(socket).await { RecvOutcome::Message(first) => first, - RecvOutcome::LocalTodo => synthetic_todo_message(), RecvOutcome::Empty => { // Idle: no message this poll. Service a queued operator // `/compact` here so it runs even when no turn is driving diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index 2d7c7b99..9535c7a5 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -40,14 +40,6 @@ pub fn harness_dir() -> PathBuf { hive_sh4re::paths::harness_dir() } -/// Harness-local todo store (loose-ends v2). A dedicated sqlite db under -/// the harness dir — the todos are mutable per-agent state the harness -/// owns, kept out of the append-only `hyperhive-events.sqlite` sink. -#[must_use] -pub fn todos_db() -> PathBuf { - harness_dir().join("hyperhive-todos.sqlite") -} - /// Per-turn config dir for the regenerated claude-{mcp-config,settings, /// system-prompt} files the harness drops before each turn. Set by /// systemd via `RuntimeDirectory = "hive-config"`: a per-service runtime diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs deleted file mode 100644 index 7ad2382c..00000000 --- a/hive-agent/src/todo_server.rs +++ /dev/null @@ -1,168 +0,0 @@ -//! In-agent socket server (loose-ends v2). Binds the harness-owned -//! `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` protocol to the -//! in-container producers (matrix / bash daemons, forge-notify). Todo ops -//! hit the harness-local [`Todos`] store; a new-or-changed upsert fires an -//! in-process [`Notify`] so the serve loop drives a turn — no hive-c0re -//! round-trip, no broker long-poll, no marker files. -//! -//! One request/response line per connection, matching the producers' -//! existing best-effort JSON-line clients (they just change which socket -//! they dial, not the payload). - -use std::path::{Path, PathBuf}; -use std::sync::Arc; - -use anyhow::{Context, Result}; -use hive_agent_sock::{Request, Response}; -use hive_sh4re::LooseEnd; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::{UnixListener, UnixStream}; -use tokio::sync::Notify; - -use crate::todos::{Todo, Todos}; - -/// Resolve the in-agent socket path from `HIVE_AGENT_SOCKET`. `None` when -/// unset/empty (e.g. a standalone dev run) — the server then stays off. -fn socket_path() -> Option { - std::env::var_os("HIVE_AGENT_SOCKET") - .filter(|v| !v.is_empty()) - .map(PathBuf::from) -} - -/// Run the in-agent socket server: bind + accept loop, one request/response -/// line per connection. A no-op (returns `Ok`) when `HIVE_AGENT_SOCKET` is -/// unset, so a standalone harness without producers just skips it. -/// -/// # Errors -/// -/// Returns an error if the socket path is set but can't be bound. -pub async fn run(store: Arc, wake: Arc) -> Result<()> { - let Some(path) = socket_path() else { - tracing::info!("HIVE_AGENT_SOCKET unset — in-agent todo socket disabled"); - return Ok(()); - }; - let listener = bind(&path)?; - tracing::info!(socket = %path.display(), "in-agent todo socket listening"); - loop { - match listener.accept().await { - Ok((stream, _)) => { - let store = store.clone(); - let wake = wake.clone(); - tokio::spawn(async move { - if let Err(e) = handle_conn(stream, &store, &wake).await { - tracing::warn!(error = ?e, "in-agent todo connection failed"); - } - }); - } - Err(e) => tracing::warn!(error = ?e, "in-agent todo accept failed"), - } - } -} - -/// Bind a `UnixListener` at `path`, creating the parent dir and unlinking a -/// stale socket left by a prior boot (which would otherwise block `bind` -/// with `EADDRINUSE`). -fn bind(path: &Path) -> Result { - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent) - .with_context(|| format!("create in-agent socket dir {}", parent.display()))?; - } - if path.exists() { - let _ = std::fs::remove_file(path); - } - UnixListener::bind(path).with_context(|| format!("bind in-agent socket {}", path.display())) -} - -/// Handle one connection: read a single JSON request line, apply it to the -/// store, write the JSON response line back. -async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result<()> { - let (read, mut write) = stream.into_split(); - let mut reader = BufReader::new(read); - let mut line = String::new(); - if reader.read_line(&mut line).await? == 0 { - return Ok(()); - } - let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(req, store, wake), - Err(e) => Response::Err { - message: format!("bad request: {e}"), - }, - }; - let mut out = serde_json::to_string(&resp)?; - out.push('\n'); - write.write_all(out.as_bytes()).await?; - write.flush().await?; - Ok(()) -} - -/// Apply one request to the store, firing `wake` on a new/changed upsert so -/// the serve loop runs a turn. -fn dispatch(req: Request, store: &Todos, wake: &Notify) -> Response { - match req { - Request::UpsertTodo { - subsystem, - key, - summary, - source, - } => match store.upsert(&subsystem, key.as_deref(), &summary, source.as_deref()) { - Ok((_, changed)) => { - if changed { - wake.notify_one(); - } - Response::Ok - } - Err(e) => err(&e), - }, - Request::ClearTodo { - subsystem, - key, - all, - } => { - let result = if all { - store.clear_subsystem(&subsystem) - } else { - store.clear(&subsystem, key.as_deref()) - }; - match result { - Ok(count) => Response::Acked { - count: u64::try_from(count).unwrap_or(0), - }, - Err(e) => err(&e), - } - } - Request::ListTodos { subsystem } => match store.list(subsystem.as_deref()) { - Ok(todos) => Response::LooseEnds { - loose_ends: todos.into_iter().map(to_loose_end).collect(), - }, - Err(e) => err(&e), - }, - Request::MarkTodoDone { id } => match store.mark_done(id) { - Ok(count) => Response::Acked { - count: u64::try_from(count).unwrap_or(0), - }, - Err(e) => err(&e), - }, - } -} - -/// Format a store error into an `Err` response. -fn err(e: &anyhow::Error) -> Response { - Response::Err { - message: format!("{e:#}"), - } -} - -/// Map a stored [`Todo`] to a [`LooseEnd::Todo`], deriving `age_seconds` -/// from `updated_at` (saturating so a backwards clock step reads 0). -fn to_loose_end(t: Todo) -> LooseEnd { - let now = hive_sh4re::wire_time::now_unix(); - let age = u64::try_from(now.saturating_sub(t.updated_at)).unwrap_or(0); - LooseEnd::Todo { - id: t.id, - subsystem: t.subsystem, - subsystem_key: t.subsystem_key, - summary: t.summary, - source: t.source, - age_seconds: age, - } -} diff --git a/hive-agent/src/todos.rs b/hive-agent/src/todos.rs deleted file mode 100644 index 19af3142..00000000 --- a/hive-agent/src/todos.rs +++ /dev/null @@ -1,293 +0,0 @@ -//! Harness-local todo store — the persistent, DB-backed half of the -//! "todos" (loose-ends v2) system, owned by the in-container harness. -//! -//! In-container subsystems (matrix, forge-notify, bash) push *todos* to -//! the harness over the in-agent socket instead of firing wakes directly. -//! The harness owns this store locally (one sqlite db under the harness -//! dir) and signals its own turn loop on a new/changed row — hive-c0re is -//! not involved (no broker round-trip, no marker files). -//! -//! Because the store lives inside a single agent's container, todos are -//! **not** agent-scoped here (unlike the old c0re store): every row -//! belongs to this agent. A todo is tagged with a `subsystem` marker plus -//! an optional `subsystem_key` (a matrix room id, a bash task id, …), and -//! `(subsystem, subsystem_key)` is the upsert/dedup key — re-pushing the -//! same item is idempotent, and a producer can list / clear / rebuild only -//! its own set (e.g. matrix wipes + recreates its todos on daemon restart). -//! -//! Removal has two paths (mara's call): the producing subsystem `clear`s a -//! todo it has resolved (keyed by subsystem + key), or the agent itself -//! `mark_done`s one by id. Both delete the row. - -use std::path::Path; -use std::sync::Mutex; - -use anyhow::{Context, Result}; -use hive_sh4re::wire_time::now_unix; -use rusqlite::{Connection, params}; - -const SCHEMA: &str = r" -CREATE TABLE IF NOT EXISTS todos ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - subsystem TEXT NOT NULL, - subsystem_key TEXT, - summary TEXT NOT NULL, - source TEXT, - updated_at INTEGER NOT NULL -); --- (subsystem, subsystem_key) is the upsert/dedup key. A NULL key never --- conflicts (SQLite treats NULLs as distinct), so keyless todos always --- insert as one-offs; keyed todos update in place. -CREATE UNIQUE INDEX IF NOT EXISTS idx_todos_dedup - ON todos (subsystem, subsystem_key); -"; - -/// One dynamic, subsystem-pushed todo. Timestamps are unix seconds; the -/// consumer derives `age_seconds` from `updated_at`. -#[derive(Debug, Clone)] -pub struct Todo { - pub id: i64, - /// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …). - pub subsystem: String, - /// Optional subsystem-specific dedup key (matrix room id, bash task - /// id, …). `None` = a keyless one-off todo. - pub subsystem_key: Option, - /// Human-readable one-line summary shown to the agent. - pub summary: String, - /// Optional free-text provenance (e.g. the room name / task label). - pub source: Option, - pub updated_at: i64, -} - -/// The harness-local todo store. Cheap to share behind an `Arc`; the inner -/// connection is guarded by a `Mutex` (todo ops are short sqlite writes). -pub struct Todos { - conn: Mutex, -} - -impl Todos { - /// Open (creating if needed) the todo store at `path`. - /// - /// # Errors - /// - /// Propagates sqlite open / schema-apply failures. - pub fn open(path: &Path) -> Result { - let conn = - Connection::open(path).with_context(|| format!("open todos db {}", path.display()))?; - conn.execute_batch(SCHEMA).context("apply todos schema")?; - Ok(Self { - conn: Mutex::new(conn), - }) - } - - /// Insert a todo, or update the existing one for `(subsystem, key)` - /// when `key` is `Some` and already present. A `None` key never - /// conflicts, so it always inserts a fresh row. - /// - /// Returns `(id, changed)` where `changed` is `true` when the row is - /// new OR its `summary`/`source` actually differed — the caller uses - /// this to decide whether to signal the turn loop (re-pushing an - /// identical keyed todo is a no-op and must not re-wake). - /// - /// # Errors - /// - /// Propagates sqlite query / execute failures. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn upsert( - &self, - subsystem: &str, - key: Option<&str>, - summary: &str, - source: Option<&str>, - ) -> Result<(i64, bool)> { - let conn = self.conn.lock().unwrap(); - let now = now_unix(); - let existing: Option<(i64, String, Option)> = if key.is_some() { - conn.query_row( - "SELECT id, summary, source FROM todos \ - WHERE subsystem = ?1 AND subsystem_key IS ?2", - params![subsystem, key], - |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), - ) - .ok() - } else { - None - }; - if let Some((id, cur_summary, cur_source)) = existing { - let unchanged = cur_summary == summary && cur_source.as_deref() == source; - if unchanged { - return Ok((id, false)); - } - conn.execute( - "UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3 WHERE id = ?4", - params![summary, source, now, id], - )?; - return Ok((id, true)); - } - conn.execute( - "INSERT INTO todos \ - (subsystem, subsystem_key, summary, source, updated_at) \ - VALUES (?1, ?2, ?3, ?4, ?5)", - params![subsystem, key, summary, source, now], - )?; - Ok((conn.last_insert_rowid(), true)) - } - - /// Clear producer-resolved todo(s) by `(subsystem, key)`. `key = - /// Some(k)` targets the one keyed row; `key = None` matches - /// `subsystem_key IS NULL`, i.e. **all** keyless todos for that - /// subsystem (clear a specific keyless one via [`Todos::mark_done`] by - /// id instead). Returns the number of rows deleted. - /// - /// # Errors - /// - /// Propagates the sqlite delete failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn clear(&self, subsystem: &str, key: Option<&str>) -> Result { - let conn = self.conn.lock().unwrap(); - let n = conn.execute( - "DELETE FROM todos WHERE subsystem = ?1 AND subsystem_key IS ?2", - params![subsystem, key], - )?; - Ok(n) - } - - /// Clear every todo `subsystem` owns — used by a producer that rebuilds - /// its whole set on restart (cancel-and-recreate). Returns the number - /// of rows deleted. - /// - /// # Errors - /// - /// Propagates the sqlite delete failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn clear_subsystem(&self, subsystem: &str) -> Result { - let conn = self.conn.lock().unwrap(); - let n = conn.execute("DELETE FROM todos WHERE subsystem = ?1", params![subsystem])?; - Ok(n) - } - - /// The agent marks one of its todos done, by id. Returns the number of - /// rows deleted (0 when the id was unknown / already gone). - /// - /// # Errors - /// - /// Propagates the sqlite delete failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn mark_done(&self, id: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let n = conn.execute("DELETE FROM todos WHERE id = ?1", params![id])?; - Ok(n) - } - - /// List todos, newest-updated first. `subsystem = Some(..)` filters to - /// one producer's set; `None` returns all. - /// - /// # Errors - /// - /// Propagates the sqlite prepare / query failures. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn list(&self, subsystem: Option<&str>) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT id, subsystem, subsystem_key, summary, source, updated_at \ - FROM todos \ - WHERE (?1 IS NULL OR subsystem = ?1) \ - ORDER BY updated_at DESC, id DESC", - )?; - let rows = stmt - .query_map(params![subsystem], |row| { - Ok(Todo { - id: row.get(0)?, - subsystem: row.get(1)?, - subsystem_key: row.get(2)?, - summary: row.get(3)?, - source: row.get(4)?, - updated_at: row.get(5)?, - }) - })? - .collect::>>()?; - Ok(rows) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - // Return the `TempDir` alongside the store so it outlives the test — - // dropping it early deletes the dir and SQLite fails with - // `SQLITE_READONLY_DBMOVED`. - fn store() -> (tempfile::TempDir, Todos) { - let dir = tempfile::tempdir().unwrap(); - let db = Todos::open(&dir.path().join("todos.sqlite")).unwrap(); - (dir, db) - } - - #[test] - fn keyed_upsert_dedups_and_reports_changed() { - let (_dir, s) = store(); - let (id1, changed1) = s - .upsert("matrix", Some("!room:x"), "1 unread", None) - .unwrap(); - assert!(changed1, "first push is new → changed"); - // Same key + same summary → no-op, not changed (must not re-wake). - let (id2, changed2) = s - .upsert("matrix", Some("!room:x"), "1 unread", None) - .unwrap(); - assert_eq!(id1, id2, "keyed upsert updates in place, same row"); - assert!(!changed2, "identical re-push is a no-op"); - // Same key, new summary → updates, changed. - let (id3, changed3) = s - .upsert("matrix", Some("!room:x"), "3 unread", None) - .unwrap(); - assert_eq!(id1, id3); - assert!(changed3); - assert_eq!(s.list(Some("matrix")).unwrap().len(), 1); - } - - #[test] - fn keyless_todos_always_insert() { - let (_dir, s) = store(); - let (a, _) = s.upsert("bash", None, "task done", None).unwrap(); - let (b, _) = s.upsert("bash", None, "task done", None).unwrap(); - assert_ne!(a, b, "keyless pushes are distinct one-offs"); - assert_eq!(s.list(Some("bash")).unwrap().len(), 2); - } - - #[test] - fn clear_and_mark_done_remove_rows() { - let (_dir, s) = store(); - s.upsert("matrix", Some("!a:x"), "unread", None).unwrap(); - let (id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap(); - assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1); - assert_eq!(s.mark_done(id).unwrap(), 1); - assert!(s.list(None).unwrap().is_empty()); - } - - #[test] - fn clear_subsystem_wipes_only_its_own() { - let (_dir, s) = store(); - s.upsert("matrix", Some("!a:x"), "u", None).unwrap(); - s.upsert("matrix", Some("!b:x"), "u", None).unwrap(); - s.upsert("forge", Some("pr-1"), "r", None).unwrap(); - assert_eq!(s.clear_subsystem("matrix").unwrap(), 2); - let left = s.list(None).unwrap(); - assert_eq!(left.len(), 1); - assert_eq!(left[0].subsystem, "forge"); - } -} diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 9d40377d..707ac73c 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -16,6 +16,7 @@ use crate::container_view::{self, ContainerView}; use crate::dashboard_events::DashboardEvent; use crate::operator_questions::OperatorQuestions; use crate::socket_server::{self, AgentSocket}; +use crate::todos::Todos; /// Capacity of the dashboard event channel. Slow browser subscribers /// (idle tab, throttled connection) drop frames past this — that's @@ -32,6 +33,11 @@ pub struct Coordinator { pub broker: Arc, pub approvals: Arc, pub questions: Arc, + /// Dynamic, subsystem-pushed todos (loose-ends v2). In-agent + /// subsystems (matrix, forge, bash) upsert/clear todos over mcp.sock + /// instead of firing wakes directly; `get_todos` merges these with + /// the computed static loose ends. + pub todos: Arc, /// Scheduled-prompts queue. One sqlite connection, /// internal mutex; the worker drains due rows and the manager /// handlers insert / cancel through the same handle. @@ -460,6 +466,7 @@ impl Coordinator { let broker = Broker::open(db_path).context("open broker")?; let approvals = Approvals::open(db_path).context("open approvals")?; let questions = OperatorQuestions::open(db_path).context("open operator_questions")?; + let todos = Todos::open(db_path).context("open todos")?; let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path) .context("open scheduled_prompts")?; // BuildLogs wants a directory (it picks its own `build_logs.sqlite` @@ -489,6 +496,7 @@ impl Coordinator { broker: Arc::new(broker), approvals: Arc::new(approvals), questions: Arc::new(questions), + todos: Arc::new(todos), scheduled_prompts: Arc::new(scheduled_prompts), build_logs, audit_log, diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index b09228f8..72a9b0ee 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -97,9 +97,40 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { age_seconds: saturating_age(now, r.created_at.timestamp()), }); } + // Dynamic, subsystem-pushed todos (loose-ends v2). Scoped to + // this agent; the producing subsystem or the agent itself clears them. + out.extend(todos_for(coord, agent, None)?); Ok(out) } +/// This agent's dynamic todos as `LooseEnd::Todo` rows, optionally +/// filtered to one `subsystem`. Shared by [`for_agent`] and the +/// `ListTodos` handler so the row-mapping lives in one place. +/// +/// # Errors +/// +/// Propagates the todo-store query failure. +pub fn todos_for( + coord: &Coordinator, + agent: &str, + subsystem: Option<&str>, +) -> Result> { + let now = now_unix(); + Ok(coord + .todos + .list(agent, subsystem)? + .into_iter() + .map(|t| LooseEnd::Todo { + id: t.id, + subsystem: t.subsystem, + subsystem_key: t.subsystem_key, + summary: t.summary, + source: t.source, + age_seconds: saturating_age(now, t.updated_at.timestamp()), + }) + .collect()) +} + /// Hive-wide loose-ends view: EVERY pending approval + EVERY /// unanswered question + EVERY pending reminder. Manager surface /// only; sub-agents can't see each other's threads via the agent diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index a97aff7b..dd704ad4 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -43,6 +43,7 @@ pub(crate) use stats::{ }; pub(crate) use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, + todos, }; pub(crate) use workers::{ agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler, diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 8f14dfd5..99be0eb2 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -596,6 +596,29 @@ async fn dispatch(req: &Request, agent: &str, coord: &Arc) -> Respo since_secs, agent: target, } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), + // Todos (loose-ends v2): in-container subsystems push/clear + // their own; the agent lists / marks its own done. Scoped to the + // calling agent (the socket identity) — no cross-agent access. + Request::UpsertTodo { + subsystem, + key, + summary, + source, + } => handle_upsert_todo( + coord, + agent, + subsystem, + key.as_deref(), + summary, + source.as_deref(), + ), + Request::ClearTodo { + subsystem, + key, + all, + } => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all), + Request::ListTodos { subsystem } => handle_list_todos(coord, agent, subsystem.as_deref()), + Request::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id), // Orchestration / diagnostics verbs — gated per-verb on tool-group // membership or topology (see `dispatch_orchestration`). _ => dispatch_orchestration(req, agent, coord).await, @@ -790,6 +813,83 @@ fn handle_get_loose_ends(coord: &Arc, agent: &str, target: Option<& } } +/// `UpsertTodo` — a subsystem pushes/updates one of this agent's todos. +/// Coalesces a wake ONLY when the row is new or actually changed, so +/// re-pushing an identical keyed todo is a silent no-op. +fn handle_upsert_todo( + coord: &Arc, + agent: &str, + subsystem: &str, + key: Option<&str>, + summary: &str, + source: Option<&str>, +) -> Response { + match coord.todos.upsert(agent, subsystem, key, summary, source) { + Ok((_, changed)) => { + if changed { + let _ = coord.broker.send(&Message { + from: "todo".to_owned(), + to: agent.to_owned(), + body: "you have todos — call get_loose_ends to see them".to_owned(), + in_reply_to: None, + }); + } + Response::Ok + } + Err(e) => Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `ClearTodo` — a producer clears a resolved todo by `(subsystem, key)`, +/// or wipes its whole set when `all` (cancel-and-recreate on restart). +fn handle_clear_todo( + coord: &Arc, + agent: &str, + subsystem: &str, + key: Option<&str>, + all: bool, +) -> Response { + let result = if all { + coord.todos.clear_subsystem(agent, subsystem) + } else { + coord.todos.clear(agent, subsystem, key) + }; + match result { + Ok(count) => Response::Acked { + count: u64::try_from(count).unwrap_or(0), + }, + Err(e) => Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `ListTodos` — enumerate this agent's todos (optionally one subsystem's) +/// as `LooseEnd::Todo` rows, so a producer can reconcile its own set. +fn handle_list_todos(coord: &Arc, agent: &str, subsystem: Option<&str>) -> Response { + match crate::loose_ends::todos_for(coord, agent, subsystem) { + Ok(loose_ends) => Response::LooseEnds { loose_ends }, + Err(e) => Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `MarkTodoDone` — the agent clears one of its own todos by id (scoped to +/// the agent, so it can't touch another agent's). +fn handle_mark_todo_done(coord: &Arc, agent: &str, id: i64) -> Response { + match coord.todos.mark_done(agent, id) { + Ok(count) => Response::Acked { + count: u64::try_from(count).unwrap_or(0), + }, + Err(e) => Response::Err { + message: format!("{e:#}"), + }, + } +} + /// `CountPendingReminders` — resolve the target (own / subtree free, else /// `QueryAgentState`) then count its pending reminders. fn handle_count_pending_reminders( diff --git a/hive-c0re/src/stores/mod.rs b/hive-c0re/src/stores/mod.rs index 1c02820c..45e676c6 100644 --- a/hive-c0re/src/stores/mod.rs +++ b/hive-c0re/src/stores/mod.rs @@ -12,3 +12,4 @@ pub mod db; pub mod operator_questions; pub mod power; pub mod scheduled_prompts; +pub mod todos; diff --git a/hive-c0re/src/stores/todos.rs b/hive-c0re/src/stores/todos.rs new file mode 100644 index 00000000..8effdd65 --- /dev/null +++ b/hive-c0re/src/stores/todos.rs @@ -0,0 +1,341 @@ +//! Todo store — the persistent, DB-backed half of the "todos" +//! (loose-ends v2) system. +//! +//! Subsystems inside an agent's container (matrix, forge-notify, bash, +//! …) push *todos* to the agent over the mcp.sock protocol instead of +//! firing wakes directly. Todos are scoped to the owning `agent` (the +//! socket identity of the pushing container) and tagged with a +//! `subsystem` marker plus an optional `subsystem_key` (a matrix room +//! id, a bash task id, …); together `(agent, subsystem, subsystem_key)` +//! is the upsert/dedup key, so re-pushing the same item is idempotent +//! (no duplicate) and a producer can list / clear / rebuild only its own +//! set (e.g. matrix wipes + recreates its todos on daemon restart). +//! +//! Removal has two paths (mara's call): the producing subsystem +//! `clear`s a todo it has resolved (keyed by subsystem + key), or the +//! agent itself `mark_done`s one by id. Both delete the row. +//! +//! This table holds only the *dynamic* subsystem-pushed todos. The +//! static ones (pending approvals / questions / reminders / undelivered +//! messages) are still computed on demand in `loose_ends.rs`; `get_todos` +//! merges the two. Folding the static kinds into this table is a later +//! increment. + +use std::path::Path; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use hive_sh4re::wire_time::now_unix; +use rusqlite::{Connection, params}; +use serde::Serialize; + +use crate::db::Migration; + +const SCHEMA: &str = r" +CREATE TABLE IF NOT EXISTS todos ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + agent TEXT NOT NULL, + subsystem TEXT NOT NULL, + subsystem_key TEXT, + summary TEXT NOT NULL, + source TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); +-- (agent, subsystem, subsystem_key) is the upsert/dedup key. A NULL key +-- never conflicts (SQLite treats NULLs as distinct), so keyless todos +-- always insert as one-offs; keyed todos update in place. +CREATE UNIQUE INDEX IF NOT EXISTS idx_todos_dedup + ON todos (agent, subsystem, subsystem_key); +"; + +// New table — no legacy rows to migrate past the initial schema. +const MIGRATIONS: &[Migration] = &[]; + +/// One dynamic, subsystem-pushed todo. +#[derive(Debug, Clone, Serialize)] +pub struct Todo { + pub id: i64, + /// Owning agent (the container whose subsystem pushed it). + pub agent: String, + /// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …). + pub subsystem: String, + /// Optional subsystem-specific dedup key (matrix room id, bash task + /// id, …). `None` = a keyless one-off todo. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subsystem_key: Option, + /// Human-readable one-line summary shown to the agent. + pub summary: String, + /// Optional free-text provenance (e.g. the room name / task label). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +pub struct Todos { + conn: Mutex, +} + +impl Todos { + /// Open (creating if needed) the todo store at `path`. + /// + /// # Errors + /// + /// Propagates sqlite open / schema-apply / migration failures. + pub fn open(path: &Path) -> Result { + let conn = crate::db::open(path, "todos")?; + conn.execute_batch(SCHEMA).context("apply todos schema")?; + crate::db::apply_versioned_migrations(&conn, "todos", MIGRATIONS)?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// Insert a todo for `agent`, or update the existing one for + /// `(agent, subsystem, key)` when `key` is `Some` and already + /// present. A `None` key never conflicts, so it always inserts a + /// fresh row. + /// + /// Returns `(id, changed)` where `changed` is `true` when the row is + /// new OR its `summary`/`source` actually differed — the caller uses + /// this to decide whether to coalesce a wake (re-pushing an identical + /// keyed todo is a no-op and must not re-wake). + /// + /// # Errors + /// + /// Propagates sqlite query / execute failures. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn upsert( + &self, + agent: &str, + subsystem: &str, + key: Option<&str>, + summary: &str, + source: Option<&str>, + ) -> Result<(i64, bool)> { + let conn = self.conn.lock().unwrap(); + let now = now_unix(); + let existing: Option<(i64, String, Option)> = if key.is_some() { + conn.query_row( + "SELECT id, summary, source FROM todos \ + WHERE agent = ?1 AND subsystem = ?2 AND subsystem_key IS ?3", + params![agent, subsystem, key], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .ok() + } else { + None + }; + if let Some((id, cur_summary, cur_source)) = existing { + let unchanged = cur_summary == summary && cur_source.as_deref() == source; + if unchanged { + return Ok((id, false)); + } + conn.execute( + "UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3 WHERE id = ?4", + params![summary, source, now, id], + )?; + return Ok((id, true)); + } + conn.execute( + "INSERT INTO todos \ + (agent, subsystem, subsystem_key, summary, source, created_at, updated_at) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)", + params![agent, subsystem, key, summary, source, now], + )?; + Ok((conn.last_insert_rowid(), true)) + } + + /// Clear producer-resolved todo(s) by `(agent, subsystem, key)`. + /// `key = Some(k)` targets the one keyed row; `key = None` matches + /// `subsystem_key IS NULL`, i.e. **all** keyless todos for that + /// subsystem (keyless rows have no distinguishing key — clear a + /// specific one via [`Todos::mark_done`] by id instead). Returns the + /// number of rows deleted (0 when nothing matched). + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn clear(&self, agent: &str, subsystem: &str, key: Option<&str>) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM todos WHERE agent = ?1 AND subsystem = ?2 AND subsystem_key IS ?3", + params![agent, subsystem, key], + )?; + Ok(n) + } + + /// Clear every todo `agent`'s `subsystem` owns — used by a producer + /// that rebuilds its whole set on restart (cancel-and-recreate). + /// Returns the number of rows deleted. + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn clear_subsystem(&self, agent: &str, subsystem: &str) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM todos WHERE agent = ?1 AND subsystem = ?2", + params![agent, subsystem], + )?; + Ok(n) + } + + /// The agent marks one of *its own* todos done, by id. Scoped to + /// `agent` so one agent can't clear another's. Returns the number of + /// rows deleted (0 when the id was unknown / not owned / already gone). + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn mark_done(&self, agent: &str, id: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM todos WHERE id = ?1 AND agent = ?2", + params![id, agent], + )?; + Ok(n) + } + + /// List `agent`'s todos, newest-updated first. `subsystem = Some(..)` + /// filters to one producer's set (so a producer can enumerate + + /// reconcile only its own); `None` returns all of the agent's. + /// + /// # Errors + /// + /// Propagates the sqlite prepare / query failures. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn list(&self, agent: &str, subsystem: Option<&str>) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, agent, subsystem, subsystem_key, summary, source, created_at, updated_at \ + FROM todos \ + WHERE agent = ?1 AND (?2 IS NULL OR subsystem = ?2) \ + ORDER BY updated_at DESC, id DESC", + )?; + let rows = stmt + .query_map(params![agent, subsystem], |row| { + let created: i64 = row.get(6)?; + let updated: i64 = row.get(7)?; + Ok(Todo { + id: row.get(0)?, + agent: row.get(1)?, + subsystem: row.get(2)?, + subsystem_key: row.get(3)?, + summary: row.get(4)?, + source: row.get(5)?, + created_at: DateTime::from_timestamp(created, 0).unwrap_or_default(), + updated_at: DateTime::from_timestamp(updated, 0).unwrap_or_default(), + }) + })? + .collect::>>()?; + Ok(rows) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // Return the `TempDir` alongside the store so it outlives the test — + // dropping it early deletes the dir and SQLite fails with + // `SQLITE_READONLY_DBMOVED`. + fn store() -> (tempfile::TempDir, Todos) { + let dir = tempfile::tempdir().unwrap(); + let db = Todos::open(&dir.path().join("todos.sqlite")).unwrap(); + (dir, db) + } + + #[test] + fn keyed_upsert_dedups_and_reports_changed() { + let (_dir, s) = store(); + let (id1, changed1) = s + .upsert("alice", "matrix", Some("!room:x"), "1 unread", None) + .unwrap(); + assert!(changed1, "first push is new → changed"); + // Same key + same summary → no-op, not changed (must not re-wake). + let (id2, changed2) = s + .upsert("alice", "matrix", Some("!room:x"), "1 unread", None) + .unwrap(); + assert_eq!(id1, id2, "keyed upsert updates in place, same row"); + assert!(!changed2, "identical re-push is a no-op"); + // Same key, new summary → updates, changed. + let (id3, changed3) = s + .upsert("alice", "matrix", Some("!room:x"), "3 unread", None) + .unwrap(); + assert_eq!(id1, id3); + assert!(changed3); + assert_eq!(s.list("alice", Some("matrix")).unwrap().len(), 1); + } + + #[test] + fn todos_are_scoped_per_agent() { + let (_dir, s) = store(); + // Same subsystem+key for two agents → distinct rows. + s.upsert("alice", "matrix", Some("!r:x"), "u", None) + .unwrap(); + s.upsert("bob", "matrix", Some("!r:x"), "u", None).unwrap(); + assert_eq!(s.list("alice", None).unwrap().len(), 1); + assert_eq!(s.list("bob", None).unwrap().len(), 1); + // bob can't mark alice's todo done. + let alice_id = s.list("alice", None).unwrap()[0].id; + assert_eq!(s.mark_done("bob", alice_id).unwrap(), 0); + assert_eq!(s.mark_done("alice", alice_id).unwrap(), 1); + } + + #[test] + fn keyless_todos_always_insert() { + let (_dir, s) = store(); + let (a, _) = s.upsert("alice", "bash", None, "task done", None).unwrap(); + let (b, _) = s.upsert("alice", "bash", None, "task done", None).unwrap(); + assert_ne!(a, b, "keyless pushes are distinct one-offs"); + assert_eq!(s.list("alice", Some("bash")).unwrap().len(), 2); + } + + #[test] + fn clear_and_mark_done_remove_rows() { + let (_dir, s) = store(); + s.upsert("alice", "matrix", Some("!a:x"), "unread", None) + .unwrap(); + let (id, _) = s + .upsert("alice", "forge", Some("pr-1"), "review", None) + .unwrap(); + assert_eq!(s.clear("alice", "matrix", Some("!a:x")).unwrap(), 1); + assert_eq!(s.mark_done("alice", id).unwrap(), 1); + assert!(s.list("alice", None).unwrap().is_empty()); + } + + #[test] + fn clear_subsystem_wipes_only_its_own() { + let (_dir, s) = store(); + s.upsert("alice", "matrix", Some("!a:x"), "u", None) + .unwrap(); + s.upsert("alice", "matrix", Some("!b:x"), "u", None) + .unwrap(); + s.upsert("alice", "forge", Some("pr-1"), "r", None).unwrap(); + assert_eq!(s.clear_subsystem("alice", "matrix").unwrap(), 2); + let left = s.list("alice", None).unwrap(); + assert_eq!(left.len(), 1); + assert_eq!(left[0].subsystem, "forge"); + } +} diff --git a/hive-core-agent-sock/src/lib.rs b/hive-core-agent-sock/src/lib.rs index 8c4b58ba..e3ef2093 100644 --- a/hive-core-agent-sock/src/lib.rs +++ b/hive-core-agent-sock/src/lib.rs @@ -102,6 +102,41 @@ pub enum Request { #[serde(default, skip_serializing_if = "Option::is_none")] agent: Option, }, + /// Upsert a *todo* (loose-ends v2) from an in-container + /// subsystem (matrix / forge / bash). `subsystem` is the producer + /// marker; `key` is the optional subsystem-specific dedup key (a + /// matrix room id, a bash task id). Re-pushing an identical keyed + /// todo is a no-op; a new-or-changed one coalesces a wake to the + /// agent. Keyless todos always insert as one-offs. + UpsertTodo { + subsystem: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + key: Option, + summary: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + source: Option, + }, + /// Clear producer-resolved todo(s) by `(subsystem, key)`. `key = + /// Some(k)` clears the one keyed row; `key = None` clears **all** of + /// the subsystem's keyless todos (rows with no key can't be told + /// apart — clear a specific one via `MarkTodoDone` by id). `all = + /// true` wipes the producer's whole set (cancel-and-recreate on + /// daemon restart). + ClearTodo { + subsystem: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + key: Option, + #[serde(default)] + all: bool, + }, + /// List todos, optionally filtered to one `subsystem` (a producer + /// enumerating its own set). `None` = all. + ListTodos { + #[serde(default, skip_serializing_if = "Option::is_none")] + subsystem: Option, + }, + /// The agent marks one of its own todos done, by id. + MarkTodoDone { id: i64 }, /// Count of pending (un-delivered) reminders. On the agent socket: /// same target rules as `GetLooseEnds` (self/children free; /// non-children require `query_agent_state`; `"*"` rejected). diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index b4c1810d..7c4d906e 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -512,6 +512,44 @@ pub fn list_invites(client: &Client) -> DaemonResponse { DaemonResponse::ok(&invites) } +/// Rewrite `mcp-loose-ends/matrix.json` with a summary of all pending +/// room invites. The harness scans this directory generically in +/// `get_loose_ends` — no matrix-specific code needed there. +/// +/// Called after an invite arrives (from the sync handler) and after a +/// room is joined (to remove the accepted invite from loose-ends). +/// Atomic write (tmp + rename) so the harness never reads a partial file. +pub async fn refresh_invite_loose_ends(client: &Client) { + let invites = client.invited_rooms(); + let dir = crate::paths::mcp_loose_ends_dir(); + if let Err(e) = tokio::fs::create_dir_all(&dir).await { + tracing::warn!(error = ?e, "matrix: create mcp-loose-ends dir failed"); + return; + } + let items: Vec = invites + .iter() + .map(|room| { + let label = room_label(room); + format!( + "[matrix] pending invite: {label} — use list_invites to see, resolve_invite to accept or reject" + ) + }) + .collect(); + let dest = dir.join("matrix.json"); + let tmp = dest.with_extension("json.tmp"); + let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned()); + match tokio::fs::write(&tmp, &json).await { + Ok(()) => { + if let Err(e) = tokio::fs::rename(&tmp, &dest).await { + tracing::warn!(error = ?e, "matrix: rename mcp-loose-ends/matrix.json failed"); + } + } + Err(e) => { + tracing::warn!(error = ?e, "matrix: write mcp-loose-ends/matrix.json.tmp failed"); + } + } +} + pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse { let parsed: &RoomOrAliasId = match room_ref.try_into() { Ok(p) => p, @@ -522,14 +560,9 @@ pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse { let server_names: Vec = vec![]; match client.join_room_by_id_or_alias(parsed, &server_names).await { Ok(room) => { - // Clear the invite todo so the accepted invite drops out of - // `get_loose_ends` immediately (the sweep would also clear it - // on its next tick, but this makes it instant). - let _ = crate::wake::send_todo_clear( - Some(&crate::timeline::invite_key(room.room_id())), - false, - ) - .await; + // Refresh loose-ends so the accepted invite is removed from + // `get_loose_ends` output immediately after the agent joins. + refresh_invite_loose_ends(client).await; DaemonResponse::ok(&serde_json::json!({ "joined": true, "room_id": room.room_id().to_string(), @@ -562,12 +595,7 @@ pub async fn resolve_invite( }; match room.leave().await { Ok(()) => { - // Clear the invite todo immediately on reject. - let _ = crate::wake::send_todo_clear( - Some(&crate::timeline::invite_key(room.room_id())), - false, - ) - .await; + refresh_invite_loose_ends(client).await; DaemonResponse::ok(&serde_json::json!({ "rejected": true, "room_id": room.room_id().to_string(), diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index dbb4f39c..42d768dd 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -73,6 +73,7 @@ async fn main() -> Result<()> { let cfgs = accounts::configured().context("read matrix account config")?; let mcp_socket = paths::daemon_socket(); + let hyperhive_socket = paths::hyperhive_socket(); let multi = cfgs.len() > 1; let primary = cfgs[0].name.clone(); let mut registry = Registry::new(primary); @@ -80,10 +81,10 @@ async fn main() -> Result<()> { for (idx, cfg) in cfgs.into_iter().enumerate() { let is_primary = idx == 0; - // Account-tag the todos only in multi-account mode so single- - // account todo summaries stay byte-identical to the legacy format. + // Account-tag the wakes only in multi-account mode so single- + // account wake bodies stay byte-identical to the legacy format. let tag = multi.then(|| cfg.name.clone()); - match bring_up_account(&cfg, tag.clone(), is_primary).await { + match bring_up_account(&cfg, &hyperhive_socket, tag.clone(), is_primary).await { Ok(Some((client, sync_loop))) => { registry.insert(cfg.name, client); sync_loops.push(sync_loop); @@ -108,7 +109,8 @@ async fn main() -> Result<()> { // before being skipped for this daemon lifetime. Err(e) if is_primary => return Err(e.context("bring up primary matrix account")), Err(e) => { - if let Some((client, sync_loop)) = bring_up_secondary_with_retry(&cfg, tag, e).await + if let Some((client, sync_loop)) = + bring_up_secondary_with_retry(&cfg, &hyperhive_socket, tag, e).await { registry.insert(cfg.name, client); sync_loops.push(sync_loop); @@ -187,6 +189,7 @@ async fn main() -> Result<()> { /// failure, no token, or all retries exhausted). async fn bring_up_secondary_with_retry( cfg: &AccountCfg, + hyperhive_socket: &std::path::Path, tag: Option, first_error: anyhow::Error, ) -> Option<(Client, SyncLoop)> { @@ -216,7 +219,7 @@ async fn bring_up_secondary_with_retry( "retrying secondary account bring-up after backoff" ); tokio::time::sleep(std::time::Duration::from_secs(delay)).await; - match bring_up_account(cfg, tag.clone(), false).await { + match bring_up_account(cfg, hyperhive_socket, tag.clone(), false).await { Ok(Some((client, sync_loop))) => { tracing::info!(account = %cfg.name, "secondary matrix account recovered"); return Some((client, sync_loop)); @@ -257,6 +260,7 @@ async fn bring_up_secondary_with_retry( /// handling). The returned sync loop is driven by the caller. async fn bring_up_account( cfg: &AccountCfg, + hyperhive_socket: &std::path::Path, tag: Option, is_primary: bool, ) -> Result> { @@ -289,26 +293,30 @@ async fn bring_up_account( let sync_client = client.clone(); let cb_client = client.clone(); - // Separate dedup sets: one tracks invites already pushed as todos, the + let wake_socket = Arc::new(hyperhive_socket.to_path_buf()); + // Separate dedup sets: one tracks invites already woken about, the // other unread-message rooms. Both are pruned to their current state - // each sweep (see the sweep fns) so re-invites / new messages re-push. + // each sweep (see the sweep fns) so re-invites / new messages re-wake. let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); // Startup cancel-and-recreate (loose-ends v2): wipe this agent's - // matrix todos so stale ones (rooms read / invites resolved while the - // daemon was down) don't linger, then let the first sweep rebuild the - // set to match current reality. Best-effort; the sweep converges. - let _ = crate::wake::send_todo_clear(None, true).await; + // matrix todos so stale ones (rooms read while the daemon was down) don't + // linger, then let the first sweep rebuild the set to match current + // unread reality. Best-effort; the sweep converges regardless. + let _ = crate::wake::send_todo_clear(hyperhive_socket, None, true).await; let sync_loop: SyncLoop = Box::pin(async move { sync_client .sync_with_callback(SyncSettings::default(), move |_response| { let client = cb_client.clone(); + let socket = wake_socket.clone(); let invite_notified = invite_notified.clone(); let unread_notified = unread_notified.clone(); let tag = tag.clone(); async move { - timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await; - timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await; + timeline::sweep_invites(&client, &socket, &invite_notified, tag.as_deref()) + .await; + timeline::sweep_unread(&client, &socket, &unread_notified, tag.as_deref()) + .await; matrix_sdk::LoopCtrl::Continue } }) diff --git a/hive-matrix-mcp/src/paths.rs b/hive-matrix-mcp/src/paths.rs index dc9dcf17..ccc8a441 100644 --- a/hive-matrix-mcp/src/paths.rs +++ b/hive-matrix-mcp/src/paths.rs @@ -72,3 +72,22 @@ pub fn accounts_file() -> PathBuf { let state_dir = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); PathBuf::from(format!("{state_dir}/matrix-accounts.json")) } + +/// Hyperhive control socket — the daemon writes wake signals here so +/// the harness drives a new claude turn on incoming matrix events. +/// Mirrors the path `forge_notify` writes to. +#[must_use] +pub fn hyperhive_socket() -> PathBuf { + std::env::var_os("HIVE_CONTROL_SOCKET") + .map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from) +} + +/// Directory where MCP daemons write loose-end summary files for the harness. +/// Each daemon writes `.json` here; the harness scans the dir in +/// `get_loose_ends` to surface active work from all MCPs generically. +/// Resolution lives in `hive_sh4re::paths` so the harness + every MCP +/// daemon agree on the location. +#[must_use] +pub fn mcp_loose_ends_dir() -> PathBuf { + hive_sh4re::paths::mcp_loose_ends_dir() +} diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 6e46e4fa..50ebc082 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -1,18 +1,19 @@ -//! Matrix → hyperhive todo bridge (loose-ends v2). Both room messages and -//! room invites are surfaced to the agent by SWEEPING state on every -//! post-sync callback, not by one-shot `m.room.message` / -//! `StrippedRoomMemberEvent` handlers: a one-shot signal that raced a -//! socket-down window (a container rebuild) was dropped with no retry, -//! leaving the agent deaf to matrix activity until manually prompted. -//! Sweeping the reliable sync path re-checks each tick and self-heals a -//! dropped push on the next one — each todo is idempotent by room key. +//! Matrix → hyperhive wake bridge. Both room messages and room invites +//! are surfaced to the agent by SWEEPING state on every post-sync +//! callback, not by one-shot `m.room.message` / `StrippedRoomMemberEvent` +//! handlers: a one-shot wake whose `send_wake` raced a hive-c0re / socket- +//! down window (a container rebuild) was dropped with no retry, leaving +//! the agent deaf to matrix activity until manually prompted. Sweeping the +//! reliable sync path re-checks each tick and self-heals a dropped wake on +//! the next one. //! -//! Todo summaries stay short: `sweep_unread` pushes the per-room unread +//! Wake bodies stay short: `sweep_unread` sends the all-rooms unread //! summary (`wake::format_unread_summary`); the message stays unread //! server-side so the agent fetches detail via `read_room`. Self-sent //! messages never raise an unread notification, so they never wake. use std::collections::HashSet; +use std::path::Path; use matrix_sdk::{Client, ruma::OwnedRoomId}; use tokio::sync::Mutex; @@ -37,6 +38,7 @@ use crate::{handlers, wake}; /// no explicit self-filter is needed here. pub async fn sweep_unread( client: &Client, + socket: &Path, notified: &Mutex>, account_tag: Option<&str>, ) { @@ -51,7 +53,7 @@ pub async fn sweep_unread( active.difference(&unread_ids).cloned().collect() }; for id in stale { - if wake::send_todo_clear(Some(id.as_str()), false) + if wake::send_todo_clear(socket, Some(id.as_str()), false) .await .is_ok() { @@ -59,15 +61,15 @@ pub async fn sweep_unread( } } - // Upsert a todo per currently-unread room. The harness coalesces the - // wake iff the summary is new or changed, so re-upserting an unchanged - // room every sync tick is a cheap no-op (no re-wake). + // Upsert a todo per currently-unread room. hive-c0re coalesces the wake + // iff the summary is new or changed, so re-upserting an unchanged room + // every sync tick is a cheap server-side no-op (no re-wake). for (id, ru) in &unread { let summary = wake::tag_account( account_tag, wake::format_unread_summary(std::slice::from_ref(ru)), ); - match wake::send_todo_upsert(id.as_str(), &summary).await { + match wake::send_todo_upsert(socket, id.as_str(), &summary).await { Ok(()) => { notified.lock().await.insert(id.clone()); } @@ -93,6 +95,7 @@ pub async fn sweep_unread( /// decides whether to accept or reject by calling `resolve_invite`. pub async fn sweep_invites( client: &Client, + socket: &Path, notified: &Mutex>, account_tag: Option<&str>, ) { @@ -100,52 +103,38 @@ pub async fn sweep_invites( let current_ids: HashSet = current.iter().map(|r| r.room_id().to_owned()).collect(); - // Invites we previously pushed a todo for that are no longer pending - // (joined / rejected / withdrawn) → clear their todo, then drop from - // `notified` on success so a future re-invite re-upserts (retry next - // tick on failure). - let stale: Vec = { - let seen = notified.lock().await; - seen.difference(¤t_ids).cloned().collect() - }; - for id in stale { - if wake::send_todo_clear(Some(&invite_key(&id)), false) - .await - .is_ok() - { - notified.lock().await.remove(&id); + let mut fresh = Vec::new(); + { + let mut seen = notified.lock().await; + // Drop invites that are no longer pending (joined/rejected/withdrawn) + // so a future re-invite to the same room wakes the agent again. + seen.retain(|id| current_ids.contains(id)); + for room in ¤t { + if seen.insert(room.room_id().to_owned()) { + fresh.push(room.clone()); + } } } + if fresh.is_empty() { + return; + } - // Upsert a todo per pending invite. Keyed `invite:` (distinct - // from the unread sweep's `` key) so the two never collide; the - // harness coalesces the wake iff the summary is new or changed, so - // re-upserting an unchanged invite every tick is a cheap no-op. - for room in ¤t { + // Refresh loose-ends before waking so the invite is visible in + // get_loose_ends during the agent's turn. + handlers::refresh_invite_loose_ends(client).await; + for room in fresh { let room_id = room.room_id().to_owned(); let label = room.name().unwrap_or_else(|| room_id.to_string()); - let summary = wake::tag_account( + tracing::info!(%room_id, "matrix: pending invite swept, waking agent"); + let body = wake::tag_account( account_tag, format!( "[matrix] invited to {label} ({room_id}) — \ use list_invites to see pending invites, resolve_invite to accept or reject" ), ); - match wake::send_todo_upsert(&invite_key(&room_id), &summary).await { - Ok(()) => { - notified.lock().await.insert(room_id); - } - Err(e) => { - tracing::warn!(error = %e, room = %room_id, "matrix: invite todo upsert failed; will retry next sweep"); - } + if let Err(e) = wake::send_wake(socket, &body).await { + tracing::warn!(error = %e, "matrix: failed to deliver invite-wake to hyperhive"); } } } - -/// Todo dedup key for a pending invite. Namespaced with an `invite:` -/// prefix so an invited room and an unread room (which the unread sweep -/// keys by bare room id) never share a todo row. `pub(crate)` so the -/// invite-resolution handlers can clear the matching todo immediately. -pub(crate) fn invite_key(room_id: &matrix_sdk::ruma::RoomId) -> String { - format!("invite:{room_id}") -} diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 4598af2c..18c41d62 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -1,14 +1,17 @@ -//! Todo writer: pushes matrix *todos* (loose-ends v2) to the harness's -//! in-agent socket (`HIVE_AGENT_SOCKET`) when rooms have unread messages -//! or pending invites, so claude drives a turn to handle them. One JSON -//! line per op (`upsert_todo` / `clear_todo`), keyed by room id so -//! re-pushing an unchanged item is an idempotent no-op and resolving one -//! clears it. The harness owns the todo store locally and signals its own -//! turn loop — no hive-c0re round-trip. +//! Wake-signal writer: notifies the hyperhive harness when an incoming +//! matrix event arrives so claude drives a new turn. //! -//! Todo summaries stay short: a SHORT TEASER, not the full message — the -//! agent then reads the unmarked event via the `read_room` MCP tool. -//! Truncation to ~100 chars keeps the summary focused. +//! Same wire shape as `hive-agent::forge_notify`'s wake: a single JSON +//! line written to the hyperhive control socket (`/run/hive/mcp.sock` +//! by default) carrying an `Request::Wake { from, body }`. +//! The agent harness's `agent_server` parses it and treats it as a +//! `Wake` from the matrix subsystem. +//! +//! Per the operator's call (phase 3): the body is a SHORT TEASER, not +//! the full message — the agent then reads the unmarked event via +//! the `read_room` MCP tool. Truncation to ~100 chars keeps the wake +//! prompt focused (`forge_notify` embeds longer excerpts because the +//! agent doesn't have a follow-up read-the-original tool for forge). use std::path::Path; @@ -16,59 +19,65 @@ use anyhow::{Context, Result}; use tokio::io::AsyncWriteExt; use tokio::net::UnixStream; -/// The harness-served in-agent socket (`HIVE_AGENT_SOCKET`) where todo ops -/// go — distinct from the host-served control socket used by [`send_wake`]. -/// `None` when unset/empty, in which case todo sends are a best-effort -/// no-op (a standalone daemon without the harness socket). -fn agent_socket() -> Option { - std::env::var_os("HIVE_AGENT_SOCKET") - .filter(|v| !v.is_empty()) - .map(std::path::PathBuf::from) -} - -/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the harness's -/// in-agent socket — the replacement for a direct wake. `key` is the room -/// id (the dedup key); the harness signals a turn iff the todo is new or -/// its `summary` changed. Best-effort: a no-op when `HIVE_AGENT_SOCKET` -/// isn't configured. +/// Send an `Request::Wake { from: "matrix", body }` to the hyperhive +/// control socket at `socket`. Best-effort: returns Err on any plumbing +/// failure; callers log + ignore so a wake delivery hiccup doesn't tear +/// down the matrix sync loop. +/// +/// Wire format matches `hive_core_agent_sock::Request` tagged with `"cmd"` per +/// `#[serde(tag = "cmd", rename_all = "snake_case")]`. Must be `"cmd"`, +/// not `"kind"` — the harness deserialises against the hive-sh4re type +/// and silently discards requests that don't match. /// /// # Errors /// /// Returns an error on socket connect failure, serialisation failure, /// or I/O error writing to or reading from the socket. -pub async fn send_todo_upsert(key: &str, summary: impl AsRef) -> Result<()> { - let Some(socket) = agent_socket() else { - return Ok(()); - }; +pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { + let payload = serde_json::json!({ + "cmd": "wake", + "from": "matrix", + "body": body.as_ref(), + "transient": true, + }); + send_line(socket, &payload).await +} + +/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the +/// hyperhive control socket — the replacement for a direct wake. `key` is +/// the room id (the dedup key); hive-c0re coalesces a wake iff the todo is +/// new or its `summary` changed. Best-effort like [`send_wake`]. +/// +/// # Errors +/// +/// Returns an error on socket connect failure, serialisation failure, +/// or I/O error writing to or reading from the socket. +pub async fn send_todo_upsert(socket: &Path, key: &str, summary: impl AsRef) -> Result<()> { let payload = serde_json::json!({ "cmd": "upsert_todo", "subsystem": "matrix", "key": key, "summary": summary.as_ref(), }); - send_line(&socket, &payload).await + send_line(socket, &payload).await } -/// Clear matrix-subsystem todos on the harness's in-agent socket. `key = -/// Some(room)` clears one room's todo (it was read); `all = true` wipes the -/// whole matrix set (cancel-and-recreate on daemon restart). Best-effort: -/// a no-op when `HIVE_AGENT_SOCKET` isn't configured. +/// Clear matrix-subsystem todos. `key = Some(room)` clears one room's +/// todo (it was read); `all = true` wipes the whole matrix set +/// (cancel-and-recreate on daemon restart). Best-effort. /// /// # Errors /// /// Returns an error on socket connect failure, serialisation failure, /// or I/O error writing to or reading from the socket. -pub async fn send_todo_clear(key: Option<&str>, all: bool) -> Result<()> { - let Some(socket) = agent_socket() else { - return Ok(()); - }; +pub async fn send_todo_clear(socket: &Path, key: Option<&str>, all: bool) -> Result<()> { let payload = serde_json::json!({ "cmd": "clear_todo", "subsystem": "matrix", "key": key, "all": all, }); - send_line(&socket, &payload).await + send_line(socket, &payload).await } /// Write one JSON request line to the hyperhive control socket and drain diff --git a/nix/agent-modules/agent-service.nix b/nix/agent-modules/agent-service.nix index 21671fb5..2dca4049 100644 --- a/nix/agent-modules/agent-service.nix +++ b/nix/agent-modules/agent-service.nix @@ -201,13 +201,6 @@ in # `hive_c0re::agent_sockets::socket_path_for(name)` so lifecycle # bind-mounts and gateway upstream config stay in sync. HIVE_WEB_SOCKET = "/run/hive-agent/${userName}/web.sock"; - # In-agent socket (loose-ends v2): the harness binds this and - # serves the `hive-agent-sock` todo protocol to the in-container - # producers (matrix daemon) + the MCP bridge (`get_loose_ends`). - # Same per-agent runtime dir as the web socket so all agent-user - # services in this container can reach it; purely in-container - # (never bind-mounted to the host — unlike hive-c0re's mcp.sock). - HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock"; # Loopback URL of the persistent `hive-mcp-http` daemon that # `render_claude_config` points claude at for the built-in # surface (HTTP is the sole transport — no per-turn stdio child). diff --git a/nix/agent-modules/matrix.nix b/nix/agent-modules/matrix.nix index 1d13ce15..06c50331 100644 --- a/nix/agent-modules/matrix.nix +++ b/nix/agent-modules/matrix.nix @@ -219,10 +219,6 @@ in wants = [ "network-online.target" ]; environment = { HIVE_MATRIX_SOCKET = "/run/hive-matrix/socket"; - # In-agent todo socket the harness serves (loose-ends v2): the - # matrix sweep pushes unread-room + pending-invite todos here - # instead of firing wakes at hive-c0re's mcp.sock. - HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock"; RUST_LOG = "info"; } # Homeserver URL: by default the daemon inherits the host-forwarded diff --git a/nix/agent-modules/mcp.nix b/nix/agent-modules/mcp.nix index 6a6aaeca..8d039260 100644 --- a/nix/agent-modules/mcp.nix +++ b/nix/agent-modules/mcp.nix @@ -221,10 +221,6 @@ in wantedBy = [ "multi-user.target" ]; before = [ "hive-agent.service" ]; environment.RUST_LOG = "info"; - # In-agent todo socket the harness serves (loose-ends v2): the - # `get_loose_ends` handler dials it to merge this agent's local todos - # with the static loose-ends from hive-c0re. - environment.HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock"; serviceConfig = { ExecStart = "${config.hyperhive.packages.hive-agent-mcp}/bin/hive-agent-mcp --http 127.0.0.1:${toString config.hyperhive.mcp.httpPort}"; SyslogIdentifier = "hive-mcp-http";