feat(#2569): wire HIVE_AGENT_SOCKET for the harness, mcp-http, and matrix services

Cargo.lock

@@ -1523,6 +1523,7 @@ dependencies = [
  "clap",
  "forgejo-api",
  "futures-util",
+ "hive-agent-sock",
  "hive-claude",
  "hive-core-agent-sock",
  "hive-sh4re",
@@ -1552,6 +1553,7 @@ dependencies = [
  "anyhow",
  "axum",
  "clap",
+ "hive-agent-sock",
  "hive-core-agent-sock",
  "hive-sh4re",
  "rmcp",
@@ -1562,6 +1564,14 @@ 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"

Cargo.toml

@@ -3,6 +3,7 @@ resolver = "3"
 members = [
   "hive-agent",
   "hive-agent-mcp",
+  "hive-agent-sock",
   "hive-core-agent-sock",
   "hive-agent-wake",
   "hive-bash-mcp",
@@ -51,6 +52,7 @@ 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" }

hive-agent-mcp/Cargo.toml

@@ -14,6 +14,7 @@ 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

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, loose_end_kind_label, matrix_unread_summary, parse_loose_end_kind,
+    format_matrix_summary, local_todos, loose_end_kind_label, matrix_unread_summary,
-    render_loose_ends, reply_err,
+    parse_loose_end_kind, render_loose_ends, reply_err,
 };
 
 /// Write (or remove) the status file in the agent's own `state/` directory.
@@ -333,6 +333,12 @@ 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

hive-agent-mcp/src/mcp/render.rs

@@ -297,6 +297,30 @@ pub(super) async fn matrix_unread_summary() -> Option<Vec<MatrixRoomUnread>> {
     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<Vec<hive_sh4re::LooseEnd>> {
+    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::<hive_agent_sock::Response>(&line).ok()? {
+        hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends),
+        _ => None,
+    }
+}
+
 /// Format a `Vec<MatrixRoomUnread>` 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.

hive-agent-sock/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "hive-agent-sock"
+edition.workspace = true
+version.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+serde.workspace = true
+hive-sh4re.workspace = true

hive-agent-sock/src/lib.rs

@@ -0,0 +1,76 @@
+//! 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<String>,
+        summary: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        source: Option<String>,
+    },
+    /// 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<String>,
+        #[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<String>,
+    },
+    /// 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<LooseEnd> },
+    /// Op failed; `message` is operator-facing.
+    Err { message: String },
+}

hive-agent/Cargo.toml

@@ -19,6 +19,7 @@ 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

hive-agent/src/main.rs

@@ -23,6 +23,8 @@ mod prompt;
 mod serve_common;
 mod stats;
 mod stream_enrich;
+mod todo_server;
+mod todos;
 mod turn;
 mod turn_stats;
 mod vacuum;
@@ -169,6 +171,22 @@ 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.
@@ -208,6 +226,10 @@ 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
@@ -423,6 +445,26 @@ async fn serve_main<S: Surface>(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 {
@@ -439,6 +481,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         bus,
         stats,
         &files,
+        todo_wake,
     )
     .await
 }
@@ -460,6 +503,7 @@ async fn serve_loop<S: Surface>(
     bus: Bus,
     stats: Option<TurnStats>,
     files: &turn::TurnFiles,
+    todo_wake: Arc<tokio::sync::Notify>,
 ) -> Result<()> {
     tracing::info!(socket = %socket.display(), "harness serve");
     S::requeue_inflight(socket).await;
@@ -474,8 +518,20 @@ async fn serve_loop<S: Surface>(
     loop {
         let next = match self_continue.take() {
             Some(msg) => msg,
-            None => match S::recv_next(socket).await {
+            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,
+                }
+            } {
                 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

hive-agent/src/paths.rs

@@ -40,6 +40,14 @@ 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

hive-agent/src/todo_server.rs

@@ -0,0 +1,168 @@
+//! 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<PathBuf> {
+    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<Todos>, wake: Arc<Notify>) -> 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<UnixListener> {
+    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::<Request>(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,
+    }
+}

hive-agent/src/todos.rs

@@ -0,0 +1,293 @@
+//! 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<String>,
+    /// 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<String>,
+    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<Connection>,
+}
+
+impl Todos {
+    /// Open (creating if needed) the todo store at `path`.
+    ///
+    /// # Errors
+    ///
+    /// Propagates sqlite open / schema-apply failures.
+    pub fn open(path: &Path) -> Result<Self> {
+        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<String>)> = 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<usize> {
+        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<usize> {
+        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<usize> {
+        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<Vec<Todo>> {
+        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::<rusqlite::Result<Vec<_>>>()?;
+        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");
+    }
+}

hive-c0re/src/coordinator.rs

@@ -16,7 +16,6 @@ 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
@@ -33,11 +32,6 @@ pub struct Coordinator {
     pub broker: Arc<Broker>,
     pub approvals: Arc<Approvals>,
     pub questions: Arc<OperatorQuestions>,
-    /// 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<Todos>,
     /// Scheduled-prompts queue. One sqlite connection,
     /// internal mutex; the worker drains due rows and the manager
     /// handlers insert / cancel through the same handle.
@@ -466,7 +460,6 @@ 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`
@@ -496,7 +489,6 @@ 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,

hive-c0re/src/loose_ends.rs

@@ -97,40 +97,9 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> {
             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<Vec<LooseEnd>> {
-    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

hive-c0re/src/main.rs

@@ -43,7 +43,6 @@ 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,

hive-c0re/src/socket_server/mod.rs

@@ -596,29 +596,6 @@ async fn dispatch(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> 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,
@@ -813,83 +790,6 @@ fn handle_get_loose_ends(coord: &Arc<Coordinator>, 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<Coordinator>,
-    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<Coordinator>,
-    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<Coordinator>, 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<Coordinator>, 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(

hive-c0re/src/stores/mod.rs

@@ -12,4 +12,3 @@ pub mod db;
 pub mod operator_questions;
 pub mod power;
 pub mod scheduled_prompts;
-pub mod todos;

hive-c0re/src/stores/todos.rs

@@ -1,341 +0,0 @@
-//! 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<String>,
-    /// 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<String>,
-    pub created_at: DateTime<Utc>,
-    pub updated_at: DateTime<Utc>,
-}
-
-pub struct Todos {
-    conn: Mutex<Connection>,
-}
-
-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<Self> {
-        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<String>)> = 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<usize> {
-        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<usize> {
-        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<usize> {
-        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<Vec<Todo>> {
-        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::<rusqlite::Result<Vec<_>>>()?;
-        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");
-    }
-}

hive-core-agent-sock/src/lib.rs

@@ -102,41 +102,6 @@ pub enum Request {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         agent: Option<String>,
     },
-    /// 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<String>,
-        summary: String,
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        source: Option<String>,
-    },
-    /// 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<String>,
-        #[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<String>,
-    },
-    /// 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).

hive-matrix-mcp/src/handlers.rs

@@ -512,44 +512,6 @@ 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<String> = 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,
@@ -560,9 +522,14 @@ pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse {
     let server_names: Vec<OwnedServerName> = vec![];
     match client.join_room_by_id_or_alias(parsed, &server_names).await {
         Ok(room) => {
-            // Refresh loose-ends so the accepted invite is removed from
+            // Clear the invite todo so the accepted invite drops out of
-            // `get_loose_ends` output immediately after the agent joins.
+            // `get_loose_ends` immediately (the sweep would also clear it
-            refresh_invite_loose_ends(client).await;
+            // 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;
             DaemonResponse::ok(&serde_json::json!({
                 "joined": true,
                 "room_id": room.room_id().to_string(),
@@ -595,7 +562,12 @@ pub async fn resolve_invite(
             };
             match room.leave().await {
                 Ok(()) => {
-                    refresh_invite_loose_ends(client).await;
+                    // Clear the invite todo immediately on reject.
+                    let _ = crate::wake::send_todo_clear(
+                        Some(&crate::timeline::invite_key(room.room_id())),
+                        false,
+                    )
+                    .await;
                     DaemonResponse::ok(&serde_json::json!({
                         "rejected": true,
                         "room_id": room.room_id().to_string(),

hive-matrix-mcp/src/main.rs

@@ -73,7 +73,6 @@ 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);
@@ -81,10 +80,10 @@ async fn main() -> Result<()> {
 
     for (idx, cfg) in cfgs.into_iter().enumerate() {
         let is_primary = idx == 0;
-        // Account-tag the wakes only in multi-account mode so single-
+        // Account-tag the todos only in multi-account mode so single-
-        // account wake bodies stay byte-identical to the legacy format.
+        // account todo summaries stay byte-identical to the legacy format.
         let tag = multi.then(|| cfg.name.clone());
-        match bring_up_account(&cfg, &hyperhive_socket, tag.clone(), is_primary).await {
+        match bring_up_account(&cfg, tag.clone(), is_primary).await {
             Ok(Some((client, sync_loop))) => {
                 registry.insert(cfg.name, client);
                 sync_loops.push(sync_loop);
@@ -109,8 +108,7 @@ 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)) =
+                if let Some((client, sync_loop)) = bring_up_secondary_with_retry(&cfg, tag, e).await
-                    bring_up_secondary_with_retry(&cfg, &hyperhive_socket, tag, e).await
                 {
                     registry.insert(cfg.name, client);
                     sync_loops.push(sync_loop);
@@ -189,7 +187,6 @@ 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<String>,
     first_error: anyhow::Error,
 ) -> Option<(Client, SyncLoop)> {
@@ -219,7 +216,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, hyperhive_socket, tag.clone(), false).await {
+        match bring_up_account(cfg, tag.clone(), false).await {
             Ok(Some((client, sync_loop))) => {
                 tracing::info!(account = %cfg.name, "secondary matrix account recovered");
                 return Some((client, sync_loop));
@@ -260,7 +257,6 @@ 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<String>,
     is_primary: bool,
 ) -> Result<Option<(Client, SyncLoop)>> {
@@ -293,30 +289,26 @@ async fn bring_up_account(
 
     let sync_client = client.clone();
     let cb_client = client.clone();
-    let wake_socket = Arc::new(hyperhive_socket.to_path_buf());
+    // Separate dedup sets: one tracks invites already pushed as todos, the
-    // 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-wake.
+    // each sweep (see the sweep fns) so re-invites / new messages re-push.
     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 while the daemon was down) don't
+    // matrix todos so stale ones (rooms read / invites resolved while the
-    // linger, then let the first sweep rebuild the set to match current
+    // daemon was down) don't linger, then let the first sweep rebuild the
-    // unread reality. Best-effort; the sweep converges regardless.
+    // set to match current reality. Best-effort; the sweep converges.
-    let _ = crate::wake::send_todo_clear(hyperhive_socket, None, true).await;
+    let _ = crate::wake::send_todo_clear(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, &socket, &invite_notified, tag.as_deref())
+                    timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await;
-                        .await;
+                    timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await;
-                    timeline::sweep_unread(&client, &socket, &unread_notified, tag.as_deref())
-                        .await;
                     matrix_sdk::LoopCtrl::Continue
                 }
             })

hive-matrix-mcp/src/paths.rs

@@ -72,22 +72,3 @@ 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 `<name>.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()
-}

hive-matrix-mcp/src/timeline.rs

@@ -1,19 +1,18 @@
-//! Matrix → hyperhive wake bridge. Both room messages and room invites
+//! Matrix → hyperhive todo bridge (loose-ends v2). Both room messages and
-//! are surfaced to the agent by SWEEPING state on every post-sync
+//! room invites are surfaced to the agent by SWEEPING state on every
-//! callback, not by one-shot `m.room.message` / `StrippedRoomMemberEvent`
+//! post-sync callback, not by one-shot `m.room.message` /
-//! handlers: a one-shot wake whose `send_wake` raced a hive-c0re / socket-
+//! `StrippedRoomMemberEvent` handlers: a one-shot signal that raced a
-//! down window (a container rebuild) was dropped with no retry, leaving
+//! socket-down window (a container rebuild) was dropped with no retry,
-//! the agent deaf to matrix activity until manually prompted. Sweeping the
+//! leaving the agent deaf to matrix activity until manually prompted.
-//! reliable sync path re-checks each tick and self-heals a dropped wake on
+//! Sweeping the reliable sync path re-checks each tick and self-heals a
-//! the next one.
+//! dropped push on the next one — each todo is idempotent by room key.
 //!
-//! Wake bodies stay short: `sweep_unread` sends the all-rooms unread
+//! Todo summaries stay short: `sweep_unread` pushes the per-room 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;
@@ -38,7 +37,6 @@ use crate::{handlers, wake};
 /// no explicit self-filter is needed here.
 pub async fn sweep_unread(
     client: &Client,
-    socket: &Path,
     notified: &Mutex<HashSet<OwnedRoomId>>,
     account_tag: Option<&str>,
 ) {
@@ -53,7 +51,7 @@ pub async fn sweep_unread(
         active.difference(&unread_ids).cloned().collect()
     };
     for id in stale {
-        if wake::send_todo_clear(socket, Some(id.as_str()), false)
+        if wake::send_todo_clear(Some(id.as_str()), false)
             .await
             .is_ok()
         {
@@ -61,15 +59,15 @@ pub async fn sweep_unread(
         }
     }
 
-    // Upsert a todo per currently-unread room. hive-c0re coalesces the wake
+    // Upsert a todo per currently-unread room. The harness coalesces the
-    // iff the summary is new or changed, so re-upserting an unchanged room
+    // wake iff the summary is new or changed, so re-upserting an unchanged
-    // every sync tick is a cheap server-side no-op (no re-wake).
+    // room every sync tick is a cheap 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(socket, id.as_str(), &summary).await {
+        match wake::send_todo_upsert(id.as_str(), &summary).await {
             Ok(()) => {
                 notified.lock().await.insert(id.clone());
             }
@@ -95,7 +93,6 @@ 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<HashSet<OwnedRoomId>>,
     account_tag: Option<&str>,
 ) {
@@ -103,38 +100,52 @@ pub async fn sweep_invites(
     let current_ids: HashSet<OwnedRoomId> =
         current.iter().map(|r| r.room_id().to_owned()).collect();
 
-    let mut fresh = Vec::new();
+    // Invites we previously pushed a todo for that are no longer pending
-    {
+    // (joined / rejected / withdrawn) → clear their todo, then drop from
-        let mut seen = notified.lock().await;
+    // `notified` on success so a future re-invite re-upserts (retry next
-        // Drop invites that are no longer pending (joined/rejected/withdrawn)
+    // tick on failure).
-        // so a future re-invite to the same room wakes the agent again.
+    let stale: Vec<OwnedRoomId> = {
-        seen.retain(|id| current_ids.contains(id));
+        let seen = notified.lock().await;
-        for room in &current {
+        seen.difference(&current_ids).cloned().collect()
-            if seen.insert(room.room_id().to_owned()) {
+    };
-                fresh.push(room.clone());
+    for id in stale {
-            }
+        if wake::send_todo_clear(Some(&invite_key(&id)), false)
+            .await
+            .is_ok()
+        {
+            notified.lock().await.remove(&id);
         }
     }
-    if fresh.is_empty() {
-        return;
-    }
 
-    // Refresh loose-ends before waking so the invite is visible in
+    // Upsert a todo per pending invite. Keyed `invite:<room>` (distinct
-    // get_loose_ends during the agent's turn.
+    // from the unread sweep's `<room>` key) so the two never collide; the
-    handlers::refresh_invite_loose_ends(client).await;
+    // harness coalesces the wake iff the summary is new or changed, so
-    for room in fresh {
+    // re-upserting an unchanged invite every tick is a cheap no-op.
+    for room in &current {
         let room_id = room.room_id().to_owned();
         let label = room.name().unwrap_or_else(|| room_id.to_string());
-        tracing::info!(%room_id, "matrix: pending invite swept, waking agent");
+        let summary = wake::tag_account(
-        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"
             ),
         );
-        if let Err(e) = wake::send_wake(socket, &body).await {
+        match wake::send_todo_upsert(&invite_key(&room_id), &summary).await {
-            tracing::warn!(error = %e, "matrix: failed to deliver invite-wake to hyperhive");
+            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");
+            }
         }
     }
 }
+
+/// 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}")
+}

hive-matrix-mcp/src/wake.rs

@@ -1,17 +1,14 @@
-//! Wake-signal writer: notifies the hyperhive harness when an incoming
+//! Todo writer: pushes matrix *todos* (loose-ends v2) to the harness's
-//! matrix event arrives so claude drives a new turn.
+//! 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.
 //!
-//! Same wire shape as `hive-agent::forge_notify`'s wake: a single JSON
+//! Todo summaries stay short: a SHORT TEASER, not the full message — the
-//! line written to the hyperhive control socket (`/run/hive/mcp.sock`
+//! agent then reads the unmarked event via the `read_room` MCP tool.
-//! by default) carrying an `Request::Wake { from, body }`.
+//! Truncation to ~100 chars keeps the summary focused.
-//! 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;
 
@@ -19,65 +16,59 @@ use anyhow::{Context, Result};
 use tokio::io::AsyncWriteExt;
 use tokio::net::UnixStream;
 
-/// Send an `Request::Wake { from: "matrix", body }` to the hyperhive
+/// The harness-served in-agent socket (`HIVE_AGENT_SOCKET`) where todo ops
-/// control socket at `socket`. Best-effort: returns Err on any plumbing
+/// go — distinct from the host-served control socket used by [`send_wake`].
-/// failure; callers log + ignore so a wake delivery hiccup doesn't tear
+/// `None` when unset/empty, in which case todo sends are a best-effort
-/// down the matrix sync loop.
+/// no-op (a standalone daemon without the harness socket).
-///
+fn agent_socket() -> Option<std::path::PathBuf> {
-/// Wire format matches `hive_core_agent_sock::Request` tagged with `"cmd"` per
+    std::env::var_os("HIVE_AGENT_SOCKET")
-/// `#[serde(tag = "cmd", rename_all = "snake_case")]`.  Must be `"cmd"`,
+        .filter(|v| !v.is_empty())
-/// not `"kind"` — the harness deserialises against the hive-sh4re type
+        .map(std::path::PathBuf::from)
-/// 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_wake(socket: &Path, body: impl AsRef<str>) -> 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
+/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the harness's
-/// hyperhive control socket — the replacement for a direct wake. `key` is
+/// in-agent socket — the replacement for a direct wake. `key` is the room
-/// the room id (the dedup key); hive-c0re coalesces a wake iff the todo is
+/// id (the dedup key); the harness signals a turn iff the todo is new or
-/// new or its `summary` changed. Best-effort like [`send_wake`].
+/// its `summary` changed. Best-effort: a no-op when `HIVE_AGENT_SOCKET`
+/// isn't configured.
 ///
 /// # 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<str>) -> Result<()> {
+pub async fn send_todo_upsert(key: &str, summary: impl AsRef<str>) -> Result<()> {
+    let Some(socket) = agent_socket() else {
+        return Ok(());
+    };
     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. `key = Some(room)` clears one room's
+/// Clear matrix-subsystem todos on the harness's in-agent socket. `key =
-/// todo (it was read); `all = true` wipes the whole matrix set
+/// Some(room)` clears one room's todo (it was read); `all = true` wipes the
-/// (cancel-and-recreate on daemon restart). Best-effort.
+/// whole matrix set (cancel-and-recreate on daemon restart). Best-effort:
+/// a no-op when `HIVE_AGENT_SOCKET` isn't configured.
 ///
 /// # 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(socket: &Path, key: Option<&str>, all: bool) -> Result<()> {
+pub async fn send_todo_clear(key: Option<&str>, all: bool) -> Result<()> {
+    let Some(socket) = agent_socket() else {
+        return Ok(());
+    };
     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

nix/agent-modules/agent-service.nix

@@ -201,6 +201,13 @@ 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).

nix/agent-modules/matrix.nix

@@ -219,6 +219,10 @@ 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

nix/agent-modules/mcp.nix

@@ -221,6 +221,10 @@ 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";