feat(#2569): DB-backed per-agent todo store + mcp.sock upsert/clear/list/mark-done ops

hive-c0re/src/socket_server/mod.rs

@@ -575,6 +575,31 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
             since_secs,
             agent: target,
         } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
+        // Todos (loose-ends v2, #2569): 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.
+        AgentRequest::UpsertTodo {
+            subsystem,
+            key,
+            summary,
+            source,
+        } => handle_upsert_todo(
+            coord,
+            agent,
+            subsystem,
+            key.as_deref(),
+            summary,
+            source.as_deref(),
+        ),
+        AgentRequest::ClearTodo {
+            subsystem,
+            key,
+            all,
+        } => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all),
+        AgentRequest::ListTodos { subsystem } => {
+            handle_list_todos(coord, agent, subsystem.as_deref())
+        }
+        AgentRequest::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,
@@ -777,6 +802,87 @@ fn handle_get_loose_ends(
     }
 }
 
+/// `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>,
+) -> AgentResponse {
+    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_todos".to_owned(),
+                    in_reply_to: None,
+                });
+            }
+            AgentResponse::Ok
+        }
+        Err(e) => AgentResponse::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,
+) -> AgentResponse {
+    let result = if all {
+        coord.todos.clear_subsystem(agent, subsystem)
+    } else {
+        coord.todos.clear(agent, subsystem, key)
+    };
+    match result {
+        Ok(count) => AgentResponse::Acked {
+            count: u64::try_from(count).unwrap_or(0),
+        },
+        Err(e) => AgentResponse::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>,
+) -> AgentResponse {
+    match crate::loose_ends::todos_for(coord, agent, subsystem) {
+        Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
+        Err(e) => AgentResponse::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) -> AgentResponse {
+    match coord.todos.mark_done(agent, id) {
+        Ok(count) => AgentResponse::Acked {
+            count: u64::try_from(count).unwrap_or(0),
+        },
+        Err(e) => AgentResponse::Err {
+            message: format!("{e:#}"),
+        },
+    }
+}
+
 /// `CountPendingReminders` — resolve the target (own / subtree free, else
 /// `QueryAgentState`) then count its pending reminders.
 fn handle_count_pending_reminders(