hyperhive/hive-agent/src/todo_server.rs

578 lines
21 KiB
Rust

//! 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`/
//! `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
//! they dial, not the payload).
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use hive_agent_sock::{Request, Response};
use hive_sh4re::inbox::LooseEnd;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
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};
/// Context-usage floor for `Request::Compact`: below this fraction of the
/// effective context window, an agent-initiated compact is refused (nothing
/// meaningful to reclaim yet, and it'd just burn a compaction pass on a
/// near-empty session). Same number a human operator would eyeball off the
/// dashboard's `ctx·Nk` badge before hitting the `/compact` button.
const COMPACT_MIN_USAGE_FRACTION: f64 = 0.66;
/// 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)
}
/// Dial this same in-agent socket from elsewhere IN THIS PROCESS — the
/// harness's own `web_ui` endpoints (`api_todos`, `/api/stats` reminder
/// rollup) and `Surface::post_turn_counts` all need read access to the
/// `Todos`/`Reminders` stores this module owns behind an `Arc` on a
/// different spawned task, and a loopback dial is simpler than threading
/// those `Arc`s through every caller. One-shot, best-effort (no retry,
/// unlike the broker client): a connect failure means the socket server
/// task itself isn't up, which a retry within one call wouldn't fix.
pub(crate) async fn dial(req: &Request) -> Option<Response> {
let path = socket_path()?;
if !path.exists() {
return None;
}
tokio::time::timeout(
std::time::Duration::from_secs(3),
hive_sock_client::request::<_, Response>(&path, req, hive_sock_client::Retry::None),
)
.await
.ok()?
.ok()
}
/// 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>,
reminders: Option<Arc<Reminders>>,
questions: Option<Arc<Questions>>,
bus: Bus,
) -> 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();
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
{
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`).
///
/// The mode is set **here, at creation** — not corrected afterwards by
/// whoever notices. `connect(2)` on a unix socket requires *write*
/// permission, and `bind` leaves `0777 & ~umask` (typically `0755`), which
/// silently excludes every uid but the harness's own. `hive-c0re` pushes
/// todos in over this socket from the host, so that default locks it out.
/// Same two lines the sibling `web.sock` has carried all along
/// (`web_ui::serve`); access control is the containing directory's job, not
/// the socket's.
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);
}
let listener = UnixListener::bind(path)
.with_context(|| format!("bind in-agent socket {}", path.display()))?;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))
.with_context(|| format!("set perms on in-agent socket {}", path.display()))?;
Ok(listener)
}
/// 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,
reminders: Option<&Reminders>,
questions: Option<&Questions>,
bus: &Bus,
) -> 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, reminders, questions, bus),
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. 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
/// 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 {
Request::UpsertTodo {
subsystem,
key,
summary,
source,
reopen_if_acked,
} => upsert_todo(
store,
wake,
&subsystem,
key.as_deref(),
&summary,
source.as_deref(),
reopen_if_acked,
),
Request::ClearTodo {
subsystem,
key,
all,
} => clear_todo(store, &subsystem, key.as_deref(), all),
Request::ListTodos { subsystem } => list_todos(store, subsystem.as_deref()),
Request::MarkTodoDone { id } => mark_todo_done(store, id),
Request::MarkTodosDone { ids } => mark_todos_done(store, &ids),
Request::StoreReminder {
message,
timing,
file_path,
} => 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 { wake_prompt } => compact(bus, wake_prompt),
}
}
/// `StoreReminder` handler: persists a reminder due at `timing`.
fn store_reminder(
reminders: Option<&Reminders>,
message: &str,
timing: &hive_sh4re::inbox::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(
store: &Todos,
wake: &Notify,
subsystem: &str,
key: Option<&str>,
summary: &str,
source: Option<&str>,
reopen_if_acked: bool,
) -> Response {
match store.upsert(subsystem, key, summary, source, reopen_if_acked) {
Ok((id, changed)) => {
tracing::debug!(%subsystem, ?key, id, changed, "todo upsert");
if changed {
wake.notify_one();
}
Response::Ok
}
Err(e) => err(&e),
}
}
/// `ClearTodo` handler: drops one keyed todo, or every todo in `subsystem`
/// when `all` is set.
fn clear_todo(store: &Todos, subsystem: &str, key: Option<&str>, all: bool) -> Response {
let result = if all {
store.clear_subsystem(subsystem)
} else {
store.clear(subsystem, key)
};
match result {
Ok(count) => {
tracing::debug!(%subsystem, ?key, all, count, "todo clear");
Response::Acked {
count: u64::try_from(count).unwrap_or(0),
}
}
Err(e) => err(&e),
}
}
/// `ListTodos` handler: read-only, so no debug logging — not relevant to
/// diagnosing wake behaviour.
fn list_todos(store: &Todos, subsystem: Option<&str>) -> Response {
match store.list(subsystem) {
Ok(todos) => Response::LooseEnds {
loose_ends: todos.into_iter().map(to_loose_end).collect(),
},
Err(e) => err(&e),
}
}
/// `MarkTodoDone` handler: marks a single todo done by id.
fn mark_todo_done(store: &Todos, id: i64) -> Response {
match store.mark_done(id) {
Ok(count) => {
tracing::debug!(id, count, "todo mark-done");
Response::Acked {
count: u64::try_from(count).unwrap_or(0),
}
}
Err(e) => err(&e),
}
}
/// `MarkTodosDone` handler: bulk-acks an explicit list of todo ids — the
/// escape hatch for a backlog too large to triage one-by-one (see
/// `Todos::mark_done_many`'s doc for why it's ids, not a threshold).
fn mark_todos_done(store: &Todos, ids: &[i64]) -> Response {
match store.mark_done_many(ids) {
Ok(count) => {
tracing::debug!(n_ids = ids.len(), count, "todo bulk mark-done");
Response::Acked {
count: u64::try_from(count).unwrap_or(0),
}
}
Err(e) => err(&e),
}
}
/// `Request::Compact` handler: gate on context usage, then queue the same
/// deferred `compact_pending` flag the operator's `/compact` button sets.
/// Mirrors `hive-agent::web_ui::actions::post_compact` but reachable from
/// the agent's own MCP tool instead of the dashboard, and refuses below
/// [`COMPACT_MIN_USAGE_FRACTION`] instead of always honouring the request —
/// an agent can call this speculatively, a human clicking the dashboard
/// button already made the judgment call. `wake_prompt`, when set, is
/// forwarded to `Bus::request_compact` so the turn loop drives a synthetic
/// follow-up turn once the compaction actually finishes.
fn compact(bus: &Bus, wake_prompt: Option<String>) -> Response {
let Some(usage) = bus.last_ctx_usage() else {
return Response::Err {
message: "compact refused: no completed turn yet — nothing to compact".to_owned(),
};
};
let model = bus.model();
let window = bus.effective_context_window(&model);
if window == 0 {
return Response::Err {
message: "compact refused: effective context window is unknown (0)".to_owned(),
};
}
// Token counts stay well under 2^53 in practice, so the f64 conversion
// is exact; this is a threshold ratio, not a byte-count display, but the
// same "cosmetic precision loss" reasoning as `hivectl::quota::human_bytes` applies.
#[allow(clippy::cast_precision_loss)]
let fraction = usage.context_tokens() as f64 / window as f64;
if fraction < COMPACT_MIN_USAGE_FRACTION {
return Response::Err {
message: format!(
"compact refused: context usage is {:.0}% of the {window}-token window, \
below the {:.0}% floor — not worth compacting yet",
fraction * 100.0,
COMPACT_MIN_USAGE_FRACTION * 100.0
),
};
}
let will_wake = wake_prompt.is_some();
bus.request_compact(wake_prompt);
bus.emit(crate::events::LiveEvent::Note {
text: if will_wake {
"agent: self-requested /compact (with wake prompt) — running at the end of the \
current turn"
.into()
} else {
"agent: self-requested /compact — running at the end of the current turn".into()
},
});
Response::Ok
}
/// Shared "reminders db unavailable" response for every reminder op when
/// the store failed to open at boot (see `main.rs`'s best-effort open).
fn no_reminders_store() -> Response {
Response::Err {
message: "reminders store unavailable on this agent".to_owned(),
}
}
/// 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
/// it's due).
fn reminder_to_loose_end(r: Reminder) -> LooseEnd {
let now = chrono::Utc::now().timestamp();
let age = u64::try_from(now.saturating_sub(r.created_at.timestamp())).unwrap_or(0);
LooseEnd::Reminder {
id: r.id,
owner: crate::identity::label(),
message: r.message,
due_at: r.due_at,
age_seconds: age,
}
}
/// 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 = chrono::Utc::now().timestamp();
let age = u64::try_from(now.saturating_sub(t.updated_at.timestamp())).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,
}
}