vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

hive-agent: remove the in-container questions mirror

Browse source
This commit is contained in:
damocles 2026-08-29 21:28:37 +02:00
commit c865114743
7 changed files with 59 additions and 528 deletions
Download patch file Download diff file Expand all files Collapse all files

25
hive-agent-mcp/src/mcp/mod.rs
View file

@ -33,9 +33,9 @@ pub use args::{
pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv};
use render::{
dial_agent_socket, format_matrix_summary, local_questions, local_reminders, local_todos,
loose_end_kind_label, mark_local_todo_done, mark_local_todos_done, matrix_unread_summary,
parse_loose_end_kind, render_loose_ends, reply_err,
dial_agent_socket, format_matrix_summary, local_reminders, local_todos, loose_end_kind_label,
mark_local_todo_done, mark_local_todos_done, 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.
@ -293,16 +293,6 @@ impl AgentServer {
if is_self_query && let Some(reminders) = local_reminders().await {
loose_ends.extend(reminders);
}
// Merge local mirrored questions — same self-query-only
// restriction as todos/reminders above. c0re no longer sources
// `Question` rows for `for_agent`/`hive_wide` (see
// `hive-c0re::loose_ends` doc), so this is the only place a
// self-query sees its own questions now; a manager query for a
// child still sees that child's approvals (unaffected) but no
// longer its questions, matching the reminders precedent.
if is_self_query && let Some(questions) = local_questions().await {
loose_ends.extend(questions);
}
annotate_retries(render_loose_ends(&loose_ends), retries)
})
.await
@ -428,15 +418,6 @@ impl AgentServer {
let (resp, retries) = self
.dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id })
.await;
if resp.is_ok() && kind == hive_sh4re::inbox::CancelLooseEndKind::Question {
// Best-effort — cancel is ownership-gated to the asker on
// the c0re side, so a successful cancel here always means
// *this* agent's own `asked` mirror row for `id`. Known gap
// (documented on `Questions`): the target isn't notified,
// so their `answering` row always lingers until it goes
// stale — `answer()` no longer exists as an option.
let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await;
}
annotate_retries(
format_ack(
resp,

11
hive-agent-mcp/src/mcp/render.rs
View file

@ -408,17 +408,6 @@ pub(super) async fn local_reminders() -> Option<Vec<hive_sh4re::inbox::LooseEnd>
}
}
/// Query the harness's in-agent socket for this agent's mirrored questions
/// (both roles — asked and answering). Same best-effort
/// contract as [`local_reminders`]; c0re stays the actual `Ask`/`Answer`
/// routing, this only mirrors the durable "still owed a reply" view.
pub(super) async fn local_questions() -> Option<Vec<hive_sh4re::inbox::LooseEnd>> {
match dial_agent_socket(&hive_agent_sock::Request::ListQuestions).await? {
hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends),
_ => None,
}
}
/// Mark one of this agent's local todos (loose-ends v2) done by id, via
/// the harness's in-agent socket — reachable through `cancel_loose_end`
/// kind `"todo"` so clearing a todo never has to shell out through a

34
hive-agent-sock/src/lib.rs
View file

@ -4,9 +4,9 @@
//! ops below is a plain string, not a closed set: any user-configured MCP
//! server declared in an agent's `agent.nix` can dial this socket and push
//! its own todos the same way. Carries the loose-ends-v2 *todo* op family
//! plus the harness-local *reminder* and *question* op families; more
//! in-agent request families may be added over time (the socket is
//! deliberately named for the agent, not the todos).
//! plus the harness-local *reminder* 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
@ -123,30 +123,6 @@ pub enum Request {
/// through this variant, since a human watching the dashboard isn't
/// waiting on a wake.
Compact { wake_prompt: Option<String> },
/// Mirror an outstanding question this agent asked (`ask()` succeeded).
/// `target` is who it's waiting on (`"operator"` when asked with
/// `to: None`). Part of the questions-mirror increment — see
/// `hive-agent::questions`.
RecordAskedQuestion {
id: i64,
target: String,
question: String,
},
/// Mirror an outstanding question this agent was asked (a
/// `question_asked` system event arrived in the inbox). `asker` is who's
/// waiting on this agent for a reply.
RecordAnsweringQuestion {
id: i64,
asker: String,
question: String,
},
/// Drop the mirror row for `id` (either role) — the question resolved
/// from this agent's side (answered, or the `question_answered` event
/// for a question this agent asked arrived).
ClearQuestion { id: i64 },
/// List this agent's mirrored questions (both roles) — single-agent
/// scope, same shape as `ListReminders`.
ListQuestions,
}
/// A response on the in-agent socket. Serialised with a `kind` tag,
@ -158,8 +134,8 @@ pub enum Response {
Ok,
/// Op succeeded and touched `count` rows (clear / mark-done).
Acked { count: u64 },
/// `ListTodos` / `ListReminders` / `ListQuestions` result (each wraps
/// its rows as the matching [`LooseEnd`] variant).
/// `ListTodos` / `ListReminders` result (each wraps its rows as the
/// matching [`LooseEnd`] variant).
LooseEnds { loose_ends: Vec<LooseEnd> },
/// `CountPendingReminders` result.
PendingRemindersCount { count: u64 },

107
hive-agent/src/main.rs
View file

@ -22,7 +22,6 @@ mod otel_turn_metrics;
mod paths;
mod plugins;
mod prompt;
mod questions;
mod reminder_timer;
mod reminders;
mod serve_common;
@ -109,17 +108,7 @@ async fn main() -> Result<()> {
/// `ContainerCrash`, reparent notifications, and friends; the parse
/// and log path is identical. Quiet no-op when `from` isn't
/// `SYSTEM_SENDER`.
///
/// Also keeps the harness-local questions mirror in sync on the two
/// question-shaped events: `QuestionAsked` (a peer/manager asked *this*
/// agent — mirror an `Answering` row) and
/// `QuestionAnswered` (a question *this* agent asked got a reply —
/// clear the mirrored `Asked` row). Best-effort loopback dial of the
/// in-agent socket (see `todo_server::dial`'s docs on why a dial beats
/// threading an `Arc<Questions>` through this whole call chain) — a
/// dial failure is a missed mirror update, not a turn failure, so it's
/// logged and swallowed rather than propagated.
async fn log_system_event(bus: &Bus, from: &str, body: &str) {
fn log_system_event(bus: &Bus, from: &str, body: &str) {
if from != SYSTEM_SENDER {
return;
}
@ -129,62 +118,11 @@ async fn log_system_event(bus: &Bus, from: &str, body: &str) {
} else {
tracing::info!(%from, %body, "system message");
}
match parsed {
Some(HelperEvent::QuestionAsked {
id,
asker,
question,
..
}) => {
mirror_question(hive_agent_sock::Request::RecordAnsweringQuestion {
id,
asker,
question,
})
.await;
}
Some(HelperEvent::QuestionAnswered { id, .. }) => {
mirror_question(hive_agent_sock::Request::ClearQuestion { id }).await;
}
_ => {}
}
bus.emit(LiveEvent::Note {
text: format!("[system] {body}"),
});
}
/// Open the harness-local questions mirror against the consolidated state
/// db — same open-alongside shape as the reminders store
/// (`reminders::Reminders::open`) opened just above this call site in
/// `serve_main`, sharing the same file (distinct table, see
/// `questions::SCHEMA`). `None` on open failure disables question ops the
/// same way a failed reminders open disables reminder ops
/// (`no_questions_store` in `todo_server`). Pulled out to its own fn to
/// keep `serve_main` under the pedantic line-count lint.
fn open_question_store() -> Option<Arc<questions::Questions>> {
match questions::Questions::open(&paths::state_db()) {
Ok(store) => Some(Arc::new(store)),
Err(e) => {
tracing::error!(error = ?e, "open questions db failed — question mirror disabled");
None
}
}
}
/// Fire-and-forget helper for `log_system_event`'s questions-mirror dial:
/// logs a warning on dial failure / an `Err` response, otherwise silent.
async fn mirror_question(req: hive_agent_sock::Request) {
match todo_server::dial(&req).await {
Some(hive_agent_sock::Response::Err { message }) => {
tracing::warn!(%message, ?req, "questions mirror dial returned an error");
}
None => {
tracing::warn!(?req, "questions mirror dial failed (socket unavailable?)");
}
Some(_) => {}
}
}
/// Body string for the turn-failure notification we route to
/// `<parent>` on `TurnError::Failed`. Reads the hive-qualified
/// identity so the receiver sees `agent@hive` rather than relying on
@ -482,23 +420,22 @@ impl Surface for AgentSurface {
// ---------- generic turn loop ----------
/// Opens the todos store and spawns the in-agent todo socket (loose-ends
/// v2 + harness-local reminders + questions): the harness owns the todo +
/// reminder + question stores locally and serves the in-container
/// producers on `HIVE_AGENT_SOCKET`. A new/changed todo upsert fires the
/// returned `Notify` so the serve loop drives a turn directly — no broker
/// v2 + harness-local reminders): the harness owns the todo + reminder
/// stores locally and serves the in-container producers on
/// `HIVE_AGENT_SOCKET`. A new/changed todo upsert fires the returned
/// `Notify` so the serve loop drives a turn directly — no broker
/// round-trip, no marker files. Best-effort: if the todos store can't
/// open, the whole socket isn't served (reminder + question ops ride
/// along on the same listener, so they're gated on the same store —
/// acceptable since a from-scratch harness boot either has a writable
/// harness dir or doesn't). Split out of `serve_main` to keep it under
/// clippy's `too_many_lines` limit; kept alongside the returned `Notify`
/// so the serve loop's `LocalTodo` arm can gate a wake on `has_any()`
/// before spawning a turn — see its doc comment (the phantom-todo-wake
/// issue: a burst of same-turn upserts can arm a second `Notify` permit
/// that outlives the turn that already drained its payload).
/// open, the whole socket isn't served (reminder ops ride along on the
/// same listener, so they're gated on the same store — acceptable since a
/// from-scratch harness boot either has a writable harness dir or
/// doesn't). Split out of `serve_main` to keep it under clippy's
/// `too_many_lines` limit; kept alongside the returned `Notify` so the
/// serve loop's `LocalTodo` arm can gate a wake on `has_any()` before
/// spawning a turn — see its doc comment (the phantom-todo-wake issue: a
/// burst of same-turn upserts can arm a second `Notify` permit that
/// outlives the turn that already drained its payload).
fn spawn_todo_socket(
reminder_store: Option<Arc<reminders::Reminders>>,
question_store: Option<Arc<questions::Questions>>,
bus: &Bus,
) -> (Arc<tokio::sync::Notify>, Option<Arc<todos::Todos>>) {
let todo_wake = Arc::new(tokio::sync::Notify::new());
@ -509,14 +446,8 @@ fn spawn_todo_socket(
let bus_for_socket = bus.clone();
let store_for_socket = store.clone();
tokio::spawn(async move {
if let Err(e) = todo_server::run(
store_for_socket,
wake,
reminder_store,
question_store,
bus_for_socket,
)
.await
if let Err(e) =
todo_server::run(store_for_socket, wake, reminder_store, bus_for_socket).await
{
tracing::error!(error = %e, "in-agent todo socket exited with error");
}
@ -646,9 +577,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
}
};
tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx));
let question_store = open_question_store();
let (todo_wake, todos_store) =
spawn_todo_socket(reminder_store.clone(), question_store.clone(), &bus);
let (todo_wake, todos_store) = spawn_todo_socket(reminder_store.clone(), &bus);
if matches!(initial, LoginState::NeedsLogin) {
login::wait_for_login(
&claude_dir,
@ -973,7 +902,7 @@ async fn handle_turn<S: Surface>(
let body = first.body;
let redelivered = first.redelivered;
let msg_id = first.id;
log_system_event(bus, &from, &body).await;
log_system_event(bus, &from, &body);
tracing::info!(%from, %body, %redelivered, "inbox");
let unread = S::inbox_unread(socket).await;
bus.emit(LiveEvent::TurnStart {

28
hive-agent/src/paths.rs
View file

@ -40,19 +40,25 @@ pub fn harness_dir() -> PathBuf {
hive_agent_sock::paths::harness_dir()
}
/// Consolidated harness-local state db — todos + reminders + the questions
/// mirror, one table each — mutable per-agent state the harness owns, kept
/// out of the append-only `hyperhive-events.sqlite` sink. Per mara's call
/// ("not yet another sqlite! todos, reminders, questions should be like
/// three tiny tables in one 500kb sqlite"), this file is the shared home
/// for all loose-ends-v2 stores; each store's `open()` only applies its own
/// Consolidated harness-local state db — todos + reminders, one table
/// each — mutable per-agent state the harness owns, kept out of the
/// append-only `hyperhive-events.sqlite` sink. Per mara's call ("not yet
/// another sqlite! todos, reminders, questions should be like three tiny
/// tables in one 500kb sqlite"), this file is the shared home for the
/// loose-ends-v2 stores; each store's `open()` only applies its own
/// `CREATE TABLE IF NOT EXISTS`, so opening multiple stores against the
/// same path is safe (distinct table names, no schema collision).
/// All three stores (todos, reminders, questions) open this same path
/// directly (see their `open()` call sites) — distinct table names mean no
/// schema collision, so there's no need for per-store path wrapper fns here.
/// same path is safe (distinct table names, no schema collision). Both
/// stores (todos, reminders) open this same path directly (see their
/// `open()` call sites) — distinct table names mean no schema collision,
/// so there's no need for per-store path wrapper fns here.
///
/// Before this consolidation, todos and reminders lived in their own
/// ⚠️ A pre-existing harness may still carry a third table here, `questions`
/// — the mirror this file's own doc comment used to describe as one of
/// three. It's gone now that the ask/answer mechanism itself has been
/// removed: nothing opens or writes it any more, so any rows already in
/// it are inert, harmless leftovers, not a store to migrate or clean up.
///
/// Before the todos/reminders consolidation, they lived in their own
/// `hyperhive-todos.sqlite` / `hyperhive-reminders.sqlite` files; a
/// one-time boot migration (`db_migrate::run`) folds those into this path
/// the first time a harness boots after the upgrade.

235
hive-agent/src/questions.rs
View file

@ -1,235 +0,0 @@
//! Harness-local questions mirror — the second increment of the
//! loose-ends-v2 migration's questions phase (see the design comment on
//! the tracking issue). c0re stays the `Ask`/`Answer` routing + delivery
//! rendezvous (mara's "2a" call); this store only holds the *durable "I
//! still owe/am owed a reply" view* `get_loose_ends` renders, so that view
//! survives a hive migration the same way todos/reminders already do.
//!
//! One row per outstanding question **from this agent's point of view**,
//! keyed by the c0re-assigned question id (globally unique — an agent is
//! never both asker and target of the same question, self-asks are
//! rejected at the c0re layer):
//!
//! - `role = "asked"`: this agent called `ask()`; `peer` is the target
//! (`"operator"` when asked with `to: None`). Cleared when the
//! `question_answered` system event for `id` arrives (see `main.rs`'s
//! inbound-event hook).
//! - `role = "answering"`: this agent received a `question_asked` system
//! event for `id`; `peer` is the asker. Cleared when this agent calls
//! `answer()` for `id` (see `hive-agent-mcp`'s tool impl).
//!
//! Known gap: if the asker cancels their own question, the target is not
//! proactively notified today (`hive-c0re::questions::handle_cancel_loose_end`
//! only notifies a *different* canceller than the asker, which never
//! happens via the ownership-gated agent-socket cancel path). A target's
//! `answering` row is cleared the next time they call `answer()` — c0re's
//! "already answered"/"not found" rejection is treated as resolved-without-
//! us and clears the mirror row (see `hive-agent-mcp`'s `answer()` tool) —
//! but if they never call `answer()` at all, the row lingers with no
//! proactive nudge. Flagged on the tracking issue rather than fully fixed.
use std::path::Path;
use std::sync::Mutex;
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use rusqlite::{Connection, params};
const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS questions (
id INTEGER PRIMARY KEY,
role TEXT NOT NULL,
peer TEXT NOT NULL,
question TEXT NOT NULL,
asked_at INTEGER NOT NULL
);
";
/// Which side of the question this agent is on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Role {
/// This agent asked; `peer` is who it's waiting on.
Asked,
/// This agent was asked; `peer` is who's waiting on it.
Answering,
}
impl Role {
const fn as_str(self) -> &'static str {
match self {
Role::Asked => "asked",
Role::Answering => "answering",
}
}
fn parse(s: &str) -> Option<Self> {
match s {
"asked" => Some(Role::Asked),
"answering" => Some(Role::Answering),
_ => None,
}
}
}
/// One mirrored question row.
#[derive(Debug, Clone)]
pub struct QuestionMirror {
pub id: i64,
pub role: Role,
pub peer: String,
pub question: String,
pub asked_at: DateTime<Utc>,
}
/// The harness-local questions mirror. Same sharing/locking shape as
/// [`crate::reminders::Reminders`] — cheap behind an `Arc`, short sqlite
/// writes guarded by a `Mutex`.
pub struct Questions {
conn: Mutex<Connection>,
}
impl Questions {
/// Open (creating if needed) the questions mirror 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 questions db {}", path.display()))?;
conn.execute_batch(SCHEMA)
.context("apply questions schema")?;
Ok(Self {
conn: Mutex::new(conn),
})
}
/// Record a new outstanding row. `INSERT OR REPLACE` so a caller that
/// re-observes the same id (e.g. a redelivered `question_asked`) is a
/// harmless no-op rather than a unique-constraint error.
///
/// # Errors
///
/// Propagates the sqlite insert failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn record(&self, id: i64, role: Role, peer: &str, question: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT OR REPLACE INTO questions (id, role, peer, question, asked_at) \
VALUES (?1, ?2, ?3, ?4, ?5)",
params![id, role.as_str(), peer, question, Utc::now().timestamp()],
)?;
Ok(())
}
/// Drop the mirror row for `id` (either role) — the question resolved
/// (answered/cancelled) from this agent's side. Returns the number of
/// rows removed (0 = no local row for that id, a harmless no-op).
///
/// # Errors
///
/// Propagates the sqlite delete failure.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn clear(&self, id: i64) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute("DELETE FROM questions WHERE id = ?1", params![id])?;
Ok(n)
}
/// List every mirrored row, oldest-asked first — for `get_loose_ends`
/// rendering. A row whose `role` column doesn't parse (corruption —
/// should never happen via this module's own writes) is logged and
/// skipped rather than either propagating a hard error (which would
/// hide every other, valid row) or silently misattributing it as
/// `Asked` (which would lie about who owes whom a reply).
///
/// # Errors
///
/// Propagates the sqlite prepare / query / column-read failures.
///
/// # Panics
///
/// Panics if the connection mutex is poisoned.
pub fn list(&self) -> Result<Vec<QuestionMirror>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, role, peer, question, asked_at FROM questions ORDER BY asked_at ASC",
)?;
let mut out = Vec::new();
let mut rows = stmt.query([])?;
while let Some(row) = rows.next()? {
let id: i64 = row.get(0)?;
let role_str: String = row.get(1)?;
let Some(role) = Role::parse(&role_str) else {
tracing::warn!(%id, %role_str, "questions mirror: unknown role, skipping row");
continue;
};
let asked_at_secs: i64 = row.get(4)?;
out.push(QuestionMirror {
id,
role,
peer: row.get(2)?,
question: row.get(3)?,
asked_at: chrono::DateTime::from_timestamp(asked_at_secs, 0)
.unwrap_or_else(Utc::now),
});
}
Ok(out)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn store() -> (tempfile::TempDir, Questions) {
let dir = tempfile::tempdir().unwrap();
let db = Questions::open(&dir.path().join("questions.sqlite")).unwrap();
(dir, db)
}
#[test]
fn record_and_list_both_roles() {
let (_dir, s) = store();
s.record(1, Role::Asked, "atlas", "are we there yet")
.unwrap();
s.record(2, Role::Answering, "mara", "is this fine")
.unwrap();
let rows = s.list().unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].id, 1);
assert_eq!(rows[0].role, Role::Asked);
assert_eq!(rows[0].peer, "atlas");
assert_eq!(rows[1].id, 2);
assert_eq!(rows[1].role, Role::Answering);
assert_eq!(rows[1].peer, "mara");
}
#[test]
fn clear_removes_the_row() {
let (_dir, s) = store();
s.record(1, Role::Asked, "atlas", "q").unwrap();
assert_eq!(s.clear(1).unwrap(), 1);
assert!(s.list().unwrap().is_empty());
assert_eq!(s.clear(1).unwrap(), 0, "already-cleared id is a no-op");
assert_eq!(s.clear(999).unwrap(), 0, "unknown id is a no-op");
}
#[test]
fn record_is_idempotent_on_redelivery() {
let (_dir, s) = store();
s.record(1, Role::Answering, "mara", "q").unwrap();
s.record(1, Role::Answering, "mara", "q").unwrap();
assert_eq!(
s.list().unwrap().len(),
1,
"re-observing the same id doesn't duplicate"
);
}
}

147
hive-agent/src/todo_server.rs
View file

@ -1,24 +1,20 @@
//! In-agent socket server (loose-ends v2 + harness-local reminders +
//! questions mirror + self-service compact). Binds the harness-owned
//! `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` protocol to the
//! in-container producers — matrix / bash daemons and forge-notify are the
//! built-in ones, but any user-configured MCP server can dial the same
//! socket and push its own todos — and to
//! `hive-agent-mcp`'s `ask`/`answer`/`remind`/`get_loose_ends`/
//! self-service compact). Binds the harness-owned `HIVE_AGENT_SOCKET` and
//! serves the `hive-agent-sock` protocol to the in-container producers —
//! matrix / bash daemons and forge-notify are the built-in ones, but any
//! user-configured MCP server can dial the same socket and push its own
//! todos — and to `hive-agent-mcp`'s `remind`/`get_loose_ends`/
//! `cancel_loose_end`/`compact` tool impls. 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. Reminder ops hit the harness-local
//! [`Reminders`] store (`None` when the store failed to open — every
//! reminder op then returns `Response::Err`); a reminder *firing* is a
//! separate path (`reminder_timer`), not driven through this socket.
//! Question ops hit the harness-local [`Questions`] mirror the same way
//! (`None` when it failed to open) — c0re stays the actual `Ask`/`Answer`
//! routing + delivery rendezvous, this store only mirrors the durable
//! "still owed a reply" view for `get_loose_ends`. `Request::Compact` is
//! the odd one out — it doesn't touch any store, just the harness's
//! [`Bus`] (gate-checked context usage, then the same deferred
//! `compact_pending` flag the operator dashboard's `/compact` button
//! sets). No hive-c0re round-trip, no broker long-poll, no marker files.
//! `Request::Compact` is the odd one out — it doesn't touch any store, just
//! the harness's [`Bus`] (gate-checked context usage, then the same
//! deferred `compact_pending` flag the operator dashboard's `/compact`
//! button sets). 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
@ -36,7 +32,6 @@ use tokio::net::{UnixListener, UnixStream};
use tokio::sync::Notify;
use crate::events::Bus;
use crate::questions::{QuestionMirror, Questions, Role};
use crate::reminders::{Reminder, Reminders};
use crate::todos::{Todo, Todos};
@ -88,7 +83,6 @@ pub async fn run(
store: Arc<Todos>,
wake: Arc<Notify>,
reminders: Option<Arc<Reminders>>,
questions: Option<Arc<Questions>>,
bus: Bus,
) -> Result<()> {
let Some(path) = socket_path() else {
@ -103,18 +97,10 @@ pub async fn run(
let store = store.clone();
let wake = wake.clone();
let reminders = reminders.clone();
let questions = questions.clone();
let bus = bus.clone();
tokio::spawn(async move {
if let Err(e) = handle_conn(
stream,
&store,
&wake,
reminders.as_deref(),
questions.as_deref(),
&bus,
)
.await
if let Err(e) =
handle_conn(stream, &store, &wake, reminders.as_deref(), &bus).await
{
tracing::warn!(error = ?e, "in-agent todo connection failed");
}
@ -159,7 +145,6 @@ async fn handle_conn(
store: &Todos,
wake: &Notify,
reminders: Option<&Reminders>,
questions: Option<&Questions>,
bus: &Bus,
) -> Result<()> {
let (read, mut write) = stream.into_split();
@ -169,7 +154,7 @@ async fn handle_conn(
return Ok(());
}
let resp = match serde_json::from_str::<Request>(line.trim()) {
Ok(req) => dispatch(req, store, wake, reminders, questions, bus),
Ok(req) => dispatch(req, store, wake, reminders, bus),
Err(e) => Response::Err {
message: format!("bad request: {e}"),
},
@ -185,16 +170,15 @@ async fn handle_conn(
/// the serve loop runs a turn. Each arm calls a small named handler function
/// directly — no sub-match/`unreachable!()` indirection per family (that
/// pattern got reviewed out of the todo family in the diagnostic-logging
/// follow-up PR; kept the reminder and question families consistent with it
/// here rather than reintroducing it). `reminders`/`questions` are `None` when that store failed to open at
/// boot, in which case every op in that family returns an `Err` — each
/// follow-up PR; kept the reminder family consistent with it here rather
/// than reintroducing it). `reminders` is `None` when that store failed to
/// open at boot, in which case every reminder op returns an `Err` — each
/// handler checks for its own `None` case.
fn dispatch(
req: Request,
store: &Todos,
wake: &Notify,
reminders: Option<&Reminders>,
questions: Option<&Questions>,
bus: &Bus,
) -> Response {
match req {
@ -230,18 +214,6 @@ fn dispatch(
Request::CancelReminder { id } => cancel_reminder(reminders, id),
Request::CountPendingReminders => count_pending_reminders(reminders),
Request::ReminderRollup { since_secs } => reminder_rollup(reminders, since_secs),
Request::RecordAskedQuestion {
id,
target,
question,
} => record_asked_question(questions, id, &target, &question),
Request::RecordAnsweringQuestion {
id,
asker,
question,
} => record_answering_question(questions, id, &asker, &question),
Request::ClearQuestion { id } => clear_question(questions, id),
Request::ListQuestions => list_questions(questions),
Request::Compact { wake_prompt } => compact(bus, wake_prompt),
}
}
@ -312,65 +284,6 @@ fn reminder_rollup(reminders: Option<&Reminders>, since_secs: u64) -> Response {
}
}
/// `RecordAskedQuestion` handler: mirror a question this agent asked.
fn record_asked_question(
questions: Option<&Questions>,
id: i64,
target: &str,
question: &str,
) -> Response {
let Some(q) = questions else {
return no_questions_store();
};
match q.record(id, Role::Asked, target, question) {
Ok(()) => Response::Ok,
Err(e) => err(&e),
}
}
/// `RecordAnsweringQuestion` handler: mirror a question this agent owes a
/// reply to.
fn record_answering_question(
questions: Option<&Questions>,
id: i64,
asker: &str,
question: &str,
) -> Response {
let Some(q) = questions else {
return no_questions_store();
};
match q.record(id, Role::Answering, asker, question) {
Ok(()) => Response::Ok,
Err(e) => err(&e),
}
}
/// `ClearQuestion` handler: drop the mirror row for `id` (either role).
fn clear_question(questions: Option<&Questions>, id: i64) -> Response {
let Some(q) = questions else {
return no_questions_store();
};
match q.clear(id) {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => err(&e),
}
}
/// `ListQuestions` handler: this agent's mirrored questions (both roles).
fn list_questions(questions: Option<&Questions>) -> Response {
let Some(q) = questions else {
return no_questions_store();
};
match q.list() {
Ok(rows) => Response::LooseEnds {
loose_ends: rows.into_iter().map(question_to_loose_end).collect(),
},
Err(e) => err(&e),
}
}
/// `UpsertTodo` handler: writes/refreshes a todo row, logs the outcome, and
/// fires `wake` on a new-or-changed upsert so the serve loop runs a turn.
fn upsert_todo(
@ -511,34 +424,6 @@ fn no_reminders_store() -> Response {
}
}
/// Shared "questions mirror unavailable" response for every question op
/// when the store failed to open at boot (see `main.rs`'s best-effort open).
fn no_questions_store() -> Response {
Response::Err {
message: "questions mirror unavailable on this agent".to_owned(),
}
}
/// Map a mirrored [`QuestionMirror`] to a [`LooseEnd::Question`]. `asker`/
/// `target` are derived from `role` — this agent's own label fills whichever
/// side `role` says is us, `peer` fills the other.
fn question_to_loose_end(q: QuestionMirror) -> LooseEnd {
let now = chrono::Utc::now().timestamp();
let age = u64::try_from(now.saturating_sub(q.asked_at.timestamp())).unwrap_or(0);
let me = crate::identity::label();
let (asker, target) = match q.role {
Role::Asked => (me, Some(q.peer)),
Role::Answering => (q.peer, Some(me)),
};
LooseEnd::Question {
id: q.id,
asker,
target,
question: q.question,
age_seconds: age,
}
}
/// Map a stored [`Reminder`] to a [`LooseEnd::Reminder`], deriving
/// `age_seconds` from `created_at` (mirrors the old c0re rendering —
/// "age" is how long the reminder has been *scheduled*, not how soon