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

feat(#2635): wire harness-local questions mirror (inc2 pt2)

Browse source
This commit is contained in:
damocles 2026-07-23 21:10:51 +02:00 • committed by mara
commit e5ef5a72be
8 changed files with 465 additions and 191 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,18 +1,22 @@
//! In-agent socket server (loose-ends v2 + harness-local reminders +
//! 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, forge-notify) 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. `Request::Compact` is the odd one out — it doesn't touch either
//! 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.
//! 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, forge-notify) and to
//! `hive-agent-mcp`'s `ask`/`answer`/`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.
//!
//! One request/response line per connection, matching the producers'
//! existing best-effort JSON-line clients (they just change which socket
@ -29,6 +33,7 @@ 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};
@ -84,6 +89,7 @@ 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 {
@ -98,10 +104,18 @@ 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(), &bus).await
if let Err(e) = handle_conn(
stream,
&store,
&wake,
reminders.as_deref(),
questions.as_deref(),
&bus,
)
.await
{
tracing::warn!(error = ?e, "in-agent todo connection failed");
}
@ -133,6 +147,7 @@ async fn handle_conn(
store: &Todos,
wake: &Notify,
reminders: Option<&Reminders>,
questions: Option<&Questions>,
bus: &Bus,
) -> Result<()> {
let (read, mut write) = stream.into_split();
@ -142,7 +157,7 @@ async fn handle_conn(
return Ok(());
}
let resp = match serde_json::from_str::<Request>(line.trim()) {
Ok(req) => dispatch(req, store, wake, reminders, bus),
Ok(req) => dispatch(req, store, wake, reminders, questions, bus),
Err(e) => Response::Err {
message: format!("bad request: {e}"),
},
@ -155,13 +170,19 @@ async fn handle_conn(
}
/// Apply one request to the store, firing `wake` on a new/changed upsert so
/// the serve loop runs a turn. `reminders` is `None` when that store
/// failed to open at boot — every reminder op then returns an `Err`.
/// 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 #2679; 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
/// 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 {
@ -189,54 +210,152 @@ fn dispatch(
message,
timing,
file_path,
} => match reminders {
Some(r) => {
match crate::reminder_timer::store(r, &message, &timing, file_path.as_deref()) {
Ok(_id) => Response::Ok,
Err(message) => Response::Err { message },
}
}
None => no_reminders_store(),
},
Request::ListReminders => match reminders {
Some(r) => match r.list_pending() {
Ok(rows) => Response::LooseEnds {
loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(),
},
Err(e) => err(&e),
},
None => no_reminders_store(),
},
Request::CancelReminder { id } => match reminders {
Some(r) => match r.cancel(id) {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => err(&e),
},
None => no_reminders_store(),
},
Request::CountPendingReminders => match reminders {
Some(r) => match r.count_pending() {
Ok(count) => Response::PendingRemindersCount { count },
Err(e) => err(&e),
},
None => no_reminders_store(),
},
Request::ReminderRollup { since_secs } => match reminders {
Some(r) => {
let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX);
match r.rollup(since_secs) {
Ok(stats) => Response::ReminderRollup { stats },
Err(e) => err(&e),
}
}
None => no_reminders_store(),
},
} => store_reminder(reminders, &message, &timing, file_path.as_deref()),
Request::ListReminders => list_reminders(reminders),
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 => compact(bus),
}
}
/// `StoreReminder` handler: persists a reminder due at `timing`.
fn store_reminder(
reminders: Option<&Reminders>,
message: &str,
timing: &hive_sh4re::ReminderTiming,
file_path: Option<&str>,
) -> Response {
let Some(r) = reminders else {
return no_reminders_store();
};
match crate::reminder_timer::store(r, message, timing, file_path) {
Ok(_id) => Response::Ok,
Err(message) => Response::Err { message },
}
}
/// `ListReminders` handler: this agent's own pending reminders.
fn list_reminders(reminders: Option<&Reminders>) -> Response {
let Some(r) = reminders else {
return no_reminders_store();
};
match r.list_pending() {
Ok(rows) => Response::LooseEnds {
loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(),
},
Err(e) => err(&e),
}
}
/// `CancelReminder` handler: drops one pending reminder by id.
fn cancel_reminder(reminders: Option<&Reminders>, id: i64) -> Response {
let Some(r) = reminders else {
return no_reminders_store();
};
match r.cancel(id) {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => err(&e),
}
}
/// `CountPendingReminders` handler.
fn count_pending_reminders(reminders: Option<&Reminders>) -> Response {
let Some(r) = reminders else {
return no_reminders_store();
};
match r.count_pending() {
Ok(count) => Response::PendingRemindersCount { count },
Err(e) => err(&e),
}
}
/// `ReminderRollup` handler: scheduled/delivered/pending counts over a
/// trailing `since_secs` window (`0` = all time).
fn reminder_rollup(reminders: Option<&Reminders>, since_secs: u64) -> Response {
let Some(r) = reminders else {
return no_reminders_store();
};
let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX);
match r.rollup(since_secs) {
Ok(stats) => Response::ReminderRollup { stats },
Err(e) => err(&e),
}
}
/// `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(
@ -352,6 +471,34 @@ 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 = hive_sh4re::wire_time::now_unix();
let age = u64::try_from(now.saturating_sub(q.asked_at)).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