246c9471b1
The poller was a `tokio::spawn` inside the `hive-agent` serve loop. It never needed anything from that loop except a socket path, so being in-process bought nothing and cost two things: a harness restart took forge notifications down with it, and the whole forge/HTTP dependency tree was linked into the serve-loop binary. It is now `hive-forge-notify`, a per-agent daemon with its own systemd unit, a sibling of `hive-bash-daemon` and `hive-matrix-daemon`. Same contract as those two: it reaches the harness only by upserting todos on the in-agent socket, and nowhere else. The module moves verbatim (`notify.rs`) — the formatters, the activation gates, the dedupe map and all 33 tests are unchanged. Only the socket call sites are rewritten, onto a small local `todo_client` rather than the harness's. That mirrors what both sibling daemons already do, and the etiquette differs on purpose: the harness's client carries a 60s backoff schedule sized to ride out a hive-c0re restart, which its callers need because they have no retry of their own. This poller's two call sites both sit inside the 30s poll loop and both treat a failure as "leave the thread unread, try next tick", so the poll interval already is the retry; a second backoff would only stack sleeps and delay the rest of the batch. The unit is `Restart=on-failure`, not `always`. An agent with no forge account is a supported configuration and the poller reports it by logging why and exiting 0 — under `always` that clean exit would be a restart loop on every forge-less agent. `forgejo-api`, `url` and `time` drop out of `hive-agent`'s dependencies with the module. Also corrects docs that outlived the code they described: the persisted `forge_cursor` field is long gone (forge's own read-state is the durable record of what has been delivered), but `docs/persistence.md` and the `harness_state` module docs still documented it as live.
969 lines
40 KiB
Rust
969 lines
40 KiB
Rust
//! Harness serve-loop binary. Long-polls the broker inbox and drives one
|
|
//! claude turn per message. There is one role: agent. The `Surface`
|
|
//! trait + `AgentSurface` zero-sized type tag keeps the turn loop
|
|
//! generic and testable. Sibling: `hive-agent-mcp` (the MCP server this
|
|
//! loop points claude at).
|
|
//! Architecture lives in
|
|
//! [`docs/turn-loop.md::Harness binary shape`](../../../docs/turn-loop.md).
|
|
//!
|
|
//! Single bin crate: the module tree below (formerly this crate's `lib.rs`,
|
|
//! before lib + bin were collapsed into one) plus the serve loop.
|
|
|
|
mod client;
|
|
mod db_migrate;
|
|
mod disk_watch;
|
|
mod events;
|
|
mod harness_state;
|
|
mod identity;
|
|
mod login;
|
|
mod login_session;
|
|
mod mcp_config;
|
|
mod paths;
|
|
mod plugins;
|
|
mod prompt;
|
|
mod questions;
|
|
mod reminder_timer;
|
|
mod reminders;
|
|
mod serve_common;
|
|
mod stats;
|
|
mod stream_enrich;
|
|
mod todo_server;
|
|
mod todos;
|
|
mod turn;
|
|
mod turn_stats;
|
|
mod vacuum;
|
|
mod web_ui;
|
|
|
|
/// Default socket path inside the container — bind-mounted by `hive-c0re`.
|
|
const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock";
|
|
|
|
/// Default web UI port — used when `HIVE_PORT` env is unset.
|
|
const DEFAULT_WEB_PORT: u16 = 8042;
|
|
|
|
/// How often the serve loop re-stats the pause marker while parked.
|
|
/// Only paid while an agent is actually paused, and only against the
|
|
/// local harness dir, so a tight-ish interval is cheap and keeps
|
|
/// `hivectl resume` feeling immediate.
|
|
const PAUSE_POLL: Duration = Duration::from_secs(5);
|
|
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::Duration;
|
|
|
|
use crate::events::{Bus, LiveEvent, TurnState};
|
|
use crate::login::LoginState;
|
|
use crate::turn_stats::TurnStats;
|
|
use anyhow::Result;
|
|
use clap::Parser;
|
|
use hive_core_agent_sock::{Request, Response};
|
|
use hive_sh4re::{HelperEvent, SYSTEM_SENDER};
|
|
|
|
#[derive(Parser)]
|
|
#[command(name = "hive-agent", about = "hyperhive harness serve loop")]
|
|
struct Cli {
|
|
/// Path to the per-agent MCP socket (bind-mounted from the host).
|
|
#[arg(long, default_value = DEFAULT_SOCKET)]
|
|
socket: PathBuf,
|
|
|
|
/// Inbox poll interval in milliseconds.
|
|
#[arg(long, default_value_t = 1000)]
|
|
poll_ms: u64,
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
tracing_subscriber::fmt()
|
|
.with_env_filter(
|
|
tracing_subscriber::EnvFilter::try_from_default_env()
|
|
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
|
|
)
|
|
.init();
|
|
|
|
let cli = Cli::parse();
|
|
serve_main::<AgentSurface>(&cli.socket, cli.poll_ms).await
|
|
}
|
|
|
|
// ---------- shared turn helpers ----------
|
|
|
|
/// Surface a `SYSTEM_SENDER` message in the live event bus + tracing
|
|
/// log. Both agents and the manager receive `QuestionAnswered`,
|
|
/// `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) {
|
|
if from != SYSTEM_SENDER {
|
|
return;
|
|
}
|
|
let parsed = serde_json::from_str::<HelperEvent>(body).ok();
|
|
if let Some(event) = &parsed {
|
|
tracing::info!(?event, "helper event");
|
|
} 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
|
|
/// the caller threading a `label` through every turn-handling layer.
|
|
/// Falls back to `<unknown>` when `HIVE_LABEL` is missing so a
|
|
/// misconfigured harness still produces a parseable line.
|
|
fn format_turn_failure(err: &anyhow::Error) -> String {
|
|
let who = crate::identity::qualified_label();
|
|
let who = if who.is_empty() {
|
|
"<unknown>".to_owned()
|
|
} else {
|
|
who
|
|
};
|
|
format!("[system] `{who}` claude turn failed:\n{err:#}")
|
|
}
|
|
|
|
/// Check for the `hyperhive-continue` sentinel under the state dir
|
|
/// (dropped by the `request_next_turn` MCP tool). Returns true and
|
|
/// consumes the file when present; false otherwise. Caller fires
|
|
/// the role-specific `Wake` request — the sentinel itself is wire-
|
|
/// agnostic so this helper lives outside both surfaces.
|
|
fn consume_continue_sentinel() -> bool {
|
|
let sentinel = crate::paths::state_dir().join("hyperhive-continue");
|
|
if !sentinel.exists() {
|
|
return false;
|
|
}
|
|
if let Err(e) = std::fs::remove_file(&sentinel) {
|
|
tracing::warn!(error = %e, "consume_continue_sentinel: remove sentinel failed");
|
|
return false;
|
|
}
|
|
true
|
|
}
|
|
|
|
/// What a finished turn tells the serve loop to do next. Replaces the
|
|
/// bare `auth_failed` bool so the loop can also act on a pending
|
|
/// `request_next_turn` without round-tripping a synthetic message
|
|
/// through the broker.
|
|
struct TurnControl {
|
|
/// The turn ended in `AuthFailed` — caller parks on login.
|
|
auth_failed: bool,
|
|
/// `request_next_turn` was called during the turn (the
|
|
/// `hyperhive-continue` sentinel was dropped + consumed).
|
|
continue_requested: bool,
|
|
/// Inbox unread count observed right after the turn. Used to
|
|
/// decide whether a self-continue is actually needed.
|
|
pending: u64,
|
|
}
|
|
|
|
/// Decide whether the serve loop should drive a self-continue turn
|
|
/// in-process. A continue is only "needed" when nothing else will
|
|
/// wake the agent: if real messages are already pending they drive
|
|
/// the next turn(s) and the continue is dropped (matches the
|
|
/// `request_next_turn` contract — "no effect if a new inbox message
|
|
/// arrives before this turn ends"). Auth-failed parks the loop on
|
|
/// login, so it suppresses the continue too.
|
|
fn should_self_continue(ctrl: &TurnControl) -> bool {
|
|
ctrl.continue_requested && !ctrl.auth_failed && ctrl.pending == 0
|
|
}
|
|
|
|
/// Synthesize the `from: "self"` / `body: "continue"` message that a
|
|
/// `request_next_turn` self-continue drives. Built in-process rather
|
|
/// than fetched from the broker — it never touches the send/recv
|
|
/// path, so it doesn't persist to sqlite or pollute the inbox.
|
|
/// `id = 0` is a non-broker sentinel: the synthetic message
|
|
/// has no DB row, and `AckTurn` keys off the recipient's in-flight
|
|
/// list (which is empty here) rather than this id.
|
|
fn synthetic_continue() -> hive_sh4re::DeliveredMessage {
|
|
hive_sh4re::DeliveredMessage {
|
|
from: "self".into(),
|
|
body: "continue".into(),
|
|
id: 0,
|
|
redelivered: false,
|
|
in_reply_to: None,
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
fn graceful_stop_message() -> hive_sh4re::DeliveredMessage {
|
|
hive_sh4re::DeliveredMessage {
|
|
from: "graceful-stop".into(),
|
|
body: "You are being gracefully stopped — the container will shut down after this turn, \
|
|
and new inbound messages are already fenced. This is your one checkpoint turn: \
|
|
flush anything worth keeping into your durable /state files now — update your \
|
|
notes / CLAUDE.md / TODO.md with in-flight task state, decisions made, important \
|
|
file paths, and whatever you'd need to resume cleanly later with only a summary \
|
|
of this conversation to go on. Do not start new work or reply to anyone; just \
|
|
write your notes and end your turn. The session may be compacted after this turn \
|
|
so a later cold start resumes cheaply."
|
|
.into(),
|
|
id: 0,
|
|
redelivered: false,
|
|
in_reply_to: None,
|
|
}
|
|
}
|
|
|
|
// ---------- surface trait ----------
|
|
|
|
/// What a `Recv` long-poll returned. Decoupled from the `Response`
|
|
/// enum so `serve_loop` can pattern-match without seeing it directly.
|
|
enum RecvOutcome {
|
|
/// Long-poll returned at least one message; first one is detached.
|
|
/// Also reused for a fired reminder: `reminder_timer` pushes a *real*
|
|
/// `DeliveredMessage` (unlike `LocalTodo`'s synthetic hint) since the
|
|
/// body/id is per-row data the producer already resolved, so the
|
|
/// select arm wraps it straight into this variant — no dedicated
|
|
/// `LocalReminder` variant needed.
|
|
Message(hive_sh4re::DeliveredMessage),
|
|
/// Long-poll timed out cleanly (empty `Messages` response). Caller
|
|
/// sleeps then retries.
|
|
Empty,
|
|
/// Wire returned an error / unexpected variant. Caller logs +
|
|
/// retries; the surface impl is responsible for tracing the
|
|
/// detail before returning this.
|
|
TransportError,
|
|
/// c0re signalled a graceful stop for this agent. The serve loop runs
|
|
/// 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
|
|
/// exists to keep the turn loop generic and testable. Every function that
|
|
/// talks to the broker goes through this so there are zero hard-coded
|
|
/// `Request` / `Response` references in the turn loop itself.
|
|
trait Surface {
|
|
/// Ack the in-flight turn. Logs warnings on transport/broker
|
|
/// errors but never propagates — turn loop continues either way.
|
|
fn ack_turn(socket: &Path) -> impl Future<Output = ()>;
|
|
|
|
/// Requeue any messages that were "in-flight" (delivered but not
|
|
/// ack'd) — fires on harness boot to recover from a crash mid-turn.
|
|
fn requeue_inflight(socket: &Path) -> impl Future<Output = ()>;
|
|
|
|
/// Current inbox unread count via `Status`. Returns 0 on any
|
|
/// transport/wire error so the caller falls through cleanly.
|
|
fn inbox_unread(socket: &Path) -> impl Future<Output = u64>;
|
|
|
|
/// `(open_threads, open_reminders)` for the post-turn stats row.
|
|
/// Either field is `None` when the underlying request errors.
|
|
fn post_turn_counts(socket: &Path) -> impl Future<Output = (Option<u64>, Option<u64>)>;
|
|
|
|
/// Tell c0re the graceful-stop checkpoint is done and the harness is
|
|
/// exiting its serve loop (fire-and-forget; logs on error). Lets the
|
|
/// `GracefulStop` orchestration stop the container without waiting out
|
|
/// its timeout fallback.
|
|
fn graceful_stop_complete(socket: &Path) -> impl Future<Output = ()>;
|
|
|
|
/// Send a message addressed to `<parent>` (broker resolves the
|
|
/// sentinel via `topology::parent_of` at delivery time; root
|
|
/// agents/manager fall through to operator).
|
|
fn send_to_parent(socket: &Path, body: String) -> impl Future<Output = ()>;
|
|
|
|
/// Long-poll the broker for the next message. Wraps the
|
|
/// `Messages`/empty/error trichotomy in `RecvOutcome` so the
|
|
/// generic `serve_loop` doesn't need the per-role Response enum
|
|
/// at all.
|
|
fn recv_next(socket: &Path) -> impl Future<Output = RecvOutcome>;
|
|
}
|
|
|
|
// ---------- AgentSurface ----------
|
|
|
|
/// Zero-sized type tag for the agent wire surface.
|
|
/// Talks `Request` / `Response`.
|
|
struct AgentSurface;
|
|
|
|
/// Issue an `Ok`-expecting fire-and-forget broker request, logging any
|
|
/// rejection / unexpected response / transport error under `label`. Shared by
|
|
/// the `Surface` methods that don't need the reply (`ack_turn`,
|
|
/// `requeue_inflight`, `graceful_stop_complete`).
|
|
async fn fire_and_forget(socket: &Path, req: Request, label: &str) {
|
|
match client::request::<_, Response>(socket, &req).await {
|
|
Ok(Response::Ok) => {}
|
|
Ok(Response::Err { message }) => {
|
|
tracing::warn!(%message, "{label} rejected by broker");
|
|
}
|
|
Ok(other) => tracing::warn!(?other, "{label} unexpected response"),
|
|
Err(e) => tracing::warn!(error = ?e, "{label} transport error"),
|
|
}
|
|
}
|
|
|
|
impl Surface for AgentSurface {
|
|
async fn ack_turn(socket: &Path) {
|
|
fire_and_forget(socket, Request::AckTurn, "ack_turn").await;
|
|
}
|
|
|
|
async fn requeue_inflight(socket: &Path) {
|
|
fire_and_forget(socket, Request::RequeueInflight, "requeue_inflight").await;
|
|
}
|
|
|
|
async fn graceful_stop_complete(socket: &Path) {
|
|
fire_and_forget(
|
|
socket,
|
|
Request::GracefulStopComplete,
|
|
"graceful_stop_complete",
|
|
)
|
|
.await;
|
|
}
|
|
|
|
async fn inbox_unread(socket: &Path) -> u64 {
|
|
match client::request::<_, Response>(socket, &Request::Status).await {
|
|
Ok(Response::Status { unread }) => unread,
|
|
_ => 0,
|
|
}
|
|
}
|
|
|
|
async fn post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
|
|
let threads =
|
|
match client::request::<_, Response>(socket, &Request::GetLooseEnds { agent: None })
|
|
.await
|
|
{
|
|
Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
|
|
_ => None,
|
|
};
|
|
// Reminders are harness-local — dial the in-agent socket directly
|
|
// instead of the broker.
|
|
let reminders =
|
|
match todo_server::dial(&hive_agent_sock::Request::CountPendingReminders).await {
|
|
Some(hive_agent_sock::Response::PendingRemindersCount { count }) => Some(count),
|
|
_ => None,
|
|
};
|
|
(threads, reminders)
|
|
}
|
|
|
|
async fn send_to_parent(socket: &Path, body: String) {
|
|
let res = client::request::<_, Response>(
|
|
socket,
|
|
&Request::Send {
|
|
to: hive_sh4re::PARENT_RECIPIENT.into(),
|
|
body,
|
|
in_reply_to: None,
|
|
},
|
|
)
|
|
.await;
|
|
if let Err(e) = res {
|
|
tracing::warn!(error = ?e, "failed to notify parent of turn failure");
|
|
}
|
|
}
|
|
|
|
async fn recv_next(socket: &Path) -> RecvOutcome {
|
|
let recv: Result<Response> = client::request(
|
|
socket,
|
|
&Request::Recv {
|
|
wait_seconds: Some(180),
|
|
max: None,
|
|
},
|
|
)
|
|
.await;
|
|
match recv {
|
|
Ok(Response::Messages { messages, .. }) if !messages.is_empty() => {
|
|
let first = messages.into_iter().next().expect("checked non-empty");
|
|
RecvOutcome::Message(first)
|
|
}
|
|
Ok(Response::Messages { .. }) => RecvOutcome::Empty,
|
|
Ok(Response::GracefulStop) => RecvOutcome::GracefulStop,
|
|
Ok(Response::Err { message }) => {
|
|
tracing::warn!(%message, "recv error");
|
|
RecvOutcome::TransportError
|
|
}
|
|
Ok(other) => {
|
|
tracing::warn!(?other, "recv produced unexpected response kind");
|
|
RecvOutcome::TransportError
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "recv failed; retrying");
|
|
RecvOutcome::TransportError
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------- 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
|
|
/// 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).
|
|
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());
|
|
let todos_store: Option<Arc<todos::Todos>> = match todos::Todos::open(&paths::state_db()) {
|
|
Ok(store) => {
|
|
let store = Arc::new(store);
|
|
let wake = todo_wake.clone();
|
|
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
|
|
{
|
|
tracing::error!(error = %e, "in-agent todo socket exited with error");
|
|
}
|
|
});
|
|
// Disk-pressure watch: an in-process todo producer, so it
|
|
// shares this store + wake directly instead of dialling the
|
|
// socket the out-of-process producers use.
|
|
tokio::spawn(disk_watch::run(store.clone(), todo_wake.clone()));
|
|
Some(store)
|
|
}
|
|
Err(e) => {
|
|
tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled");
|
|
None
|
|
}
|
|
};
|
|
(todo_wake, todos_store)
|
|
}
|
|
|
|
/// Boot — wires up the web UI, login state, stats, plugins, forge
|
|
/// notifier, and either drops into `serve_loop` directly (`Online`) or
|
|
/// parks on the login flow first (`NeedsLogin`). See
|
|
/// `docs/turn-loop.md::Boot wiring`.
|
|
async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
|
|
let port = std::env::var("HIVE_PORT")
|
|
.ok()
|
|
.and_then(|s| s.parse::<u16>().ok())
|
|
.unwrap_or(DEFAULT_WEB_PORT);
|
|
// `HIVE_LABEL` is set unconditionally by the meta-flake envelope
|
|
// for any container-deployed agent; the `"hive"` fallback here
|
|
// covers standalone `nix run .#hive` invocations and pre-meta
|
|
// dev shells. Role-independent: no semantic reason for the
|
|
// fallback to differ when the env var is missing.
|
|
let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hive".into());
|
|
let claude_dir = login::default_dir();
|
|
let initial = LoginState::from_dir(&claude_dir);
|
|
tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot");
|
|
let login_state = Arc::new(Mutex::new(initial));
|
|
let bus = Bus::new();
|
|
let stats = TurnStats::open_default();
|
|
if let Some(s) = &stats {
|
|
let (ctx, cost) = s.last_usage();
|
|
if ctx.is_some() || cost.is_some() {
|
|
bus.seed_usage(ctx, cost);
|
|
}
|
|
}
|
|
let files = turn::TurnFiles::prepare(socket, &label).await?;
|
|
// Plugin install failures come back as a Vec<String> — route each
|
|
// through `<parent>` via the `send_to_parent` failure-notify path.
|
|
// The broker resolves `<parent>` per `topology::parent_of`;
|
|
// root agents fall through to operator.
|
|
for failure in plugins::install_configured().await {
|
|
S::send_to_parent(socket, failure).await;
|
|
}
|
|
// The forge notification poller used to be spawned here. It is its own
|
|
// process now (`hive-forge-notify`, its own systemd unit) so a harness
|
|
// restart doesn't take forge notifications down with it; it reaches the
|
|
// harness the same way the bash and matrix daemons do, by upserting
|
|
// todos on the in-agent socket.
|
|
//
|
|
// Agent-side cleanup of this agent's own harness artifacts (completed
|
|
// bash-task files + verbose event rows). Runs here, not host-side in
|
|
// hive-c0re, because the files are agent-owned — see `vacuum` module docs.
|
|
tokio::spawn(crate::vacuum::run());
|
|
// Log web_ui::serve's error instead of dropping it. A bare
|
|
// `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so
|
|
// any Err (e.g. EACCES from `bind_unix` when HIVE_WEB_SOCKET points
|
|
// at a dir the agent user can't write) vanishes — leaving an
|
|
// operator with no log line and no socket, debuggable only by
|
|
// staring at lifecycle.rs.
|
|
let web_ui_args = (
|
|
label.clone(),
|
|
port,
|
|
login_state.clone(),
|
|
bus.clone(),
|
|
socket.to_path_buf(),
|
|
);
|
|
tokio::spawn(async move {
|
|
let (label, port, login_state, bus, socket) = web_ui_args;
|
|
if let Err(e) = web_ui::serve(label, port, login_state, bus, socket).await {
|
|
tracing::error!(error = %e, "web_ui::serve exited with error");
|
|
}
|
|
});
|
|
// Harness-local reminders: a due reminder fires
|
|
// straight into an mpsc channel the serve loop races against the
|
|
// broker long-poll (unlike todos, a fire carries real per-row data,
|
|
// so a bare `Notify` doesn't fit — see `reminder_timer` docs).
|
|
// Best-effort: `reminder_timer::run` parks forever (never sends)
|
|
// instead of exiting when the store can't open, so the channel
|
|
// never observes a "closed" state. Opened before the todo socket
|
|
// below so the same `Arc` can be handed to its request dispatch
|
|
// (reminder ops share the todo socket/listener).
|
|
// Fold the pre-consolidation per-concern db files into the single
|
|
// `hyperhive-state.sqlite`, if this is the first boot since the
|
|
// upgrade — see `db_migrate` module docs. Best-effort: a failure here
|
|
// just leaves the legacy file(s) in place to retry next boot, so the
|
|
// subsystems below still start (`Todos`/`Reminders::open` create the
|
|
// schema fresh if nothing migrated across).
|
|
if let Err(e) = db_migrate::run(
|
|
&paths::state_db(),
|
|
&paths::legacy_todos_db(),
|
|
&paths::legacy_reminders_db(),
|
|
) {
|
|
tracing::warn!(error = ?e, "legacy db_migrate failed — continuing with fresh/partial state db");
|
|
}
|
|
let (reminder_tx, reminder_rx) = tokio::sync::mpsc::unbounded_channel();
|
|
let reminder_store = match reminders::Reminders::open(&paths::state_db()) {
|
|
Ok(store) => Some(Arc::new(store)),
|
|
Err(e) => {
|
|
tracing::error!(error = ?e, "open reminders db failed — reminder delivery disabled");
|
|
None
|
|
}
|
|
};
|
|
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);
|
|
if matches!(initial, LoginState::NeedsLogin) {
|
|
login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
|
|
} else {
|
|
// Clear any stale `hyperhive-needs-login` sentinel left over
|
|
// from a prior boot — `online` status writes the sentinel
|
|
// cleanup in `Bus::emit_status`.
|
|
bus.emit_status("online");
|
|
}
|
|
serve_loop::<S>(
|
|
socket,
|
|
Duration::from_millis(poll_ms),
|
|
login_state,
|
|
claude_dir,
|
|
bus,
|
|
stats,
|
|
&files,
|
|
todo_wake,
|
|
todos_store,
|
|
reminder_rx,
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// The long-running message loop. Long-polls the broker via
|
|
/// `S::recv_next`, drives a turn per message, parks on auth-failed,
|
|
/// otherwise retries.
|
|
#[allow(
|
|
clippy::too_many_arguments,
|
|
reason = "the harness's long-lived deps threaded into one serve loop, \
|
|
wired once from main; bundling into a struct would just move the \
|
|
same fields one level out (cf. Coordinator::open)"
|
|
)]
|
|
async fn serve_loop<S: Surface>(
|
|
socket: &Path,
|
|
interval: Duration,
|
|
login_state: Arc<Mutex<LoginState>>,
|
|
claude_dir: std::path::PathBuf,
|
|
bus: Bus,
|
|
stats: Option<TurnStats>,
|
|
files: &turn::TurnFiles,
|
|
todo_wake: Arc<tokio::sync::Notify>,
|
|
todos_store: Option<Arc<todos::Todos>>,
|
|
mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::DeliveredMessage>,
|
|
) -> Result<()> {
|
|
tracing::info!(socket = %socket.display(), "harness serve");
|
|
S::requeue_inflight(socket).await;
|
|
// The durable claude session, built once and reused for every turn +
|
|
// idle compaction below (it's effectively stateless).
|
|
let session = turn::make_session(&bus);
|
|
// Set when a turn calls `request_next_turn` and no real work is
|
|
// pending — the next iteration drives this synthetic message
|
|
// in-process instead of long-polling the broker. Never
|
|
// persisted: it lives entirely in this loop's stack.
|
|
let mut self_continue: Option<hive_sh4re::DeliveredMessage> = None;
|
|
// Tracks the last observed pause state so the transitions get logged
|
|
// once each instead of twelve lines a minute while parked.
|
|
let mut was_paused = false;
|
|
loop {
|
|
// Pause gate. While the marker is present this loop drives no
|
|
// turns at all — deliberately *before* the `self_continue.take()`
|
|
// below, so a `request_next_turn` that raced the pause is still
|
|
// waiting when the agent resumes rather than being consumed by
|
|
// a turn that never runs.
|
|
//
|
|
// Nothing here touches the broker: not calling `S::recv_next` is
|
|
// exactly the "messages queue unacked, resume drains the
|
|
// backlog" semantic, with no fencing and nothing to requeue.
|
|
// Reminders (unbounded channel) and todo wakes (a `Notify`
|
|
// permit) buffer on their own. The web UI and MCP daemons run as
|
|
// separate tasks, so the agent stays inspectable while parked.
|
|
//
|
|
// A `GracefulStop` can't be observed while parked, and doesn't
|
|
// need to be: hive-c0re skips the stop-checkpoint handshake for
|
|
// a paused agent, because this check sits at the top of the loop
|
|
// and so a paused agent provably has no turn in flight.
|
|
if hive_sh4re::paths::paused_marker().exists() {
|
|
if !was_paused {
|
|
tracing::info!("pause marker present — parking the turn loop");
|
|
bus.emit(LiveEvent::Note {
|
|
text: "paused: turn loop parked, messages will queue".into(),
|
|
});
|
|
was_paused = true;
|
|
}
|
|
tokio::time::sleep(PAUSE_POLL).await;
|
|
continue;
|
|
}
|
|
if was_paused {
|
|
tracing::info!("pause marker cleared — resuming the turn loop");
|
|
bus.emit(LiveEvent::Note {
|
|
text: "resumed: draining whatever queued while paused".into(),
|
|
});
|
|
was_paused = false;
|
|
}
|
|
let next = match self_continue.take() {
|
|
Some(msg) => msg,
|
|
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,
|
|
Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm),
|
|
}
|
|
} {
|
|
RecvOutcome::Message(first) => first,
|
|
RecvOutcome::LocalTodo => {
|
|
// Gate on `has_any()` before spawning a turn: a burst of
|
|
// same-turn upserts can arm a second `Notify` permit that
|
|
// outlives the turn which already drained its payload
|
|
// (the phantom-todo-wake issue) — `notify_one` doesn't
|
|
// coalesce once the first permit's been consumed, so the surplus wake
|
|
// fires the instant the loop is back here even though
|
|
// there's nothing left to show. Fail open (drive a turn
|
|
// anyway) on a `has_any` error so a flaky sqlite read
|
|
// never silently swallows a real wake.
|
|
let has_any = todos_store
|
|
.as_ref()
|
|
.is_none_or(|store| store.has_any().unwrap_or(true));
|
|
if !has_any {
|
|
tracing::debug!("todo wake fired against an empty store — stale, skipping");
|
|
continue;
|
|
}
|
|
tracing::debug!("todo wake consumed, sending synthetic todo message");
|
|
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
|
|
// (the in-flight case is handled at the end of drive_turn).
|
|
let compacted = turn::run_pending_compact(files, &bus, &session).await;
|
|
if !compacted {
|
|
tokio::time::sleep(interval).await;
|
|
}
|
|
continue;
|
|
}
|
|
RecvOutcome::TransportError => {
|
|
// `recv_next` already logged the detail; just retry.
|
|
// No backoff: the long-poll wait is itself the throttle.
|
|
continue;
|
|
}
|
|
RecvOutcome::GracefulStop => {
|
|
// c0re fenced our inbox and wants a clean stop. Run one
|
|
// checkpoint turn so the agent flushes durable /state,
|
|
// report completion, then exit the loop → the harness
|
|
// process ends and the container can be stopped.
|
|
tracing::info!(
|
|
"graceful stop signalled — running stop-checkpoint turn, then exiting"
|
|
);
|
|
let _ = handle_turn::<S>(
|
|
socket,
|
|
&bus,
|
|
stats.as_ref(),
|
|
files,
|
|
&session,
|
|
graceful_stop_message(),
|
|
)
|
|
.await;
|
|
S::graceful_stop_complete(socket).await;
|
|
return Ok(());
|
|
}
|
|
},
|
|
};
|
|
let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &session, next).await;
|
|
if ctrl.auth_failed {
|
|
*login_state.lock().unwrap() = LoginState::NeedsLogin;
|
|
login::wait_for_login(
|
|
&claude_dir,
|
|
login_state.clone(),
|
|
&bus,
|
|
u64::try_from(interval.as_millis()).unwrap_or(2000),
|
|
)
|
|
.await;
|
|
} else if should_self_continue(&ctrl) {
|
|
tracing::info!("request_next_turn: driving self-continue turn in-process");
|
|
self_continue = Some(synthetic_continue());
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Drive a single turn: emit boot-of-turn events, run claude, ack on
|
|
/// success / requeue on rate-limit-or-401 / notify parent on failure,
|
|
/// record stats, then pick up the `request_next_turn` sentinel if it's
|
|
/// been dropped during the turn. Returns a `TurnControl` carrying the
|
|
/// auth-failed flag, whether a self-continue was requested, and the
|
|
/// post-turn inbox count — the serve loop decides what to do next.
|
|
async fn handle_turn<S: Surface>(
|
|
socket: &Path,
|
|
bus: &Bus,
|
|
stats: Option<&TurnStats>,
|
|
files: &turn::TurnFiles,
|
|
session: &turn::AgentSession,
|
|
first: hive_sh4re::DeliveredMessage,
|
|
) -> TurnControl {
|
|
let from = first.from;
|
|
let body = first.body;
|
|
let redelivered = first.redelivered;
|
|
let msg_id = first.id;
|
|
log_system_event(bus, &from, &body).await;
|
|
tracing::info!(%from, %body, %redelivered, "inbox");
|
|
let unread = S::inbox_unread(socket).await;
|
|
bus.emit(LiveEvent::TurnStart {
|
|
from: from.clone(),
|
|
body: body.clone(),
|
|
unread,
|
|
});
|
|
bus.set_state(TurnState::Thinking);
|
|
let started_at = serve_common::now_unix();
|
|
let started_instant = std::time::Instant::now();
|
|
let model_at_start = bus.model();
|
|
let prompt = serve_common::format_wake_prompt(msg_id, &from, &body, unread, redelivered);
|
|
let outcome = turn::drive_turn(&prompt, files, bus, session).await;
|
|
turn::emit_turn_end(bus, &outcome);
|
|
bus.set_state(TurnState::Idle);
|
|
if outcome.is_ok() {
|
|
S::ack_turn(socket).await;
|
|
}
|
|
if matches!(outcome, Err(turn::TurnError::RateLimited)) {
|
|
let secs = turn::rate_limit_sleep_secs();
|
|
bus.emit_status("rate_limited");
|
|
bus.emit(LiveEvent::Note {
|
|
text: format!("API rate-limited — sleeping {secs}s before retry"),
|
|
});
|
|
tracing::warn!(sleep_secs = secs, "rate-limited; parking");
|
|
tokio::time::sleep(Duration::from_secs(secs)).await;
|
|
S::requeue_inflight(socket).await;
|
|
bus.emit_status("online");
|
|
}
|
|
if matches!(outcome, Err(turn::TurnError::ApiStall)) {
|
|
// Idle watchdog killed claude on a suspected API stall. Park briefly to
|
|
// let the API recover, then requeue — same shape as the rate-limit path.
|
|
let secs = turn::stall_sleep_secs();
|
|
bus.emit_status("api_stall");
|
|
bus.emit(LiveEvent::Note {
|
|
text: format!(
|
|
"API stall timeout — sleeping {secs}s before retry \
|
|
(tune HIVE_STALL_SLEEP_SECS; disable the watchdog with HIVE_TURN_IDLE_SECS=0)"
|
|
),
|
|
});
|
|
tracing::warn!(sleep_secs = secs, "API stall; parking before retry");
|
|
tokio::time::sleep(Duration::from_secs(secs)).await;
|
|
S::requeue_inflight(socket).await;
|
|
bus.emit_status("online");
|
|
}
|
|
if matches!(outcome, Err(turn::TurnError::AuthFailed)) {
|
|
bus.emit_status("needs_login_idle");
|
|
bus.emit(LiveEvent::Note {
|
|
text: "API 401 — waiting for re-login via web UI".into(),
|
|
});
|
|
tracing::warn!("auth-failed; parking until re-login");
|
|
S::requeue_inflight(socket).await;
|
|
}
|
|
if matches!(outcome, Err(turn::TurnError::PromptTooLong)) {
|
|
// `drive_turn` already archived the session; requeue the message so it
|
|
// redelivers into the fresh session (which fits — the wake prompt is
|
|
// tiny, the overflow was the now-cleared context). No status park: the
|
|
// agent is healthy, it just needs one more delivery.
|
|
tracing::warn!("prompt-too-long; session archived, requeueing message for a fresh turn");
|
|
S::requeue_inflight(socket).await;
|
|
}
|
|
if matches!(outcome, Err(turn::TurnError::SessionNotFound)) {
|
|
// "Shouldn't happen": resume missed and the lib's create self-heal
|
|
// didn't resolve it. Requeue rather than ack-and-drop so the wake
|
|
// message isn't silently lost; the next turn creates the session fresh.
|
|
tracing::warn!("session-not-found; requeueing message for a fresh turn");
|
|
S::requeue_inflight(socket).await;
|
|
}
|
|
if let Err(turn::TurnError::Failed(e)) = &outcome {
|
|
S::send_to_parent(socket, format_turn_failure(e)).await;
|
|
}
|
|
if let Some(stats) = stats {
|
|
// Fresh session this turn → mint a `sessions` row and set its id on
|
|
// the bus so this turn (and subsequent ones until the next fresh
|
|
// start) stamp `turn_stats.session_id`. Takes the one-shot flag
|
|
// `run_claude` set when it suppressed `--continue`.
|
|
if bus.take_fresh_session() {
|
|
let sid = stats.start_session(started_at, &model_at_start);
|
|
bus.set_session_id(sid);
|
|
}
|
|
let ended_at = serve_common::now_unix();
|
|
let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
|
|
let (open_threads, open_reminders) = S::post_turn_counts(socket).await;
|
|
let row = serve_common::build_row(serve_common::TurnRowArgs {
|
|
started_at,
|
|
ended_at,
|
|
duration_ms,
|
|
model: model_at_start,
|
|
wake_from: from.clone(),
|
|
outcome: &outcome,
|
|
bus,
|
|
open_threads_count: open_threads,
|
|
open_reminders_count: open_reminders,
|
|
});
|
|
stats.record(&row);
|
|
}
|
|
let pending = S::inbox_unread(socket).await;
|
|
if pending > 0 {
|
|
tracing::info!(%pending, "pending messages after turn; fetching next");
|
|
}
|
|
TurnControl {
|
|
auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)),
|
|
continue_requested: consume_continue_sentinel(),
|
|
pending,
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod continue_tests {
|
|
use super::{TurnControl, should_self_continue, synthetic_continue};
|
|
|
|
fn ctrl(auth_failed: bool, continue_requested: bool, pending: u64) -> TurnControl {
|
|
TurnControl {
|
|
auth_failed,
|
|
continue_requested,
|
|
pending,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn self_continue_when_requested_and_inbox_empty() {
|
|
assert!(should_self_continue(&ctrl(false, true, 0)));
|
|
}
|
|
|
|
#[test]
|
|
fn no_self_continue_when_not_requested() {
|
|
assert!(!should_self_continue(&ctrl(false, false, 0)));
|
|
}
|
|
|
|
#[test]
|
|
fn no_self_continue_when_real_messages_pending() {
|
|
// A real message will drive the next turn via recv — the
|
|
// continue is superseded, not needed (request_next_turn contract).
|
|
assert!(!should_self_continue(&ctrl(false, true, 3)));
|
|
}
|
|
|
|
#[test]
|
|
fn no_self_continue_when_auth_failed() {
|
|
// Auth-failed parks the loop on login; a queued continue must
|
|
// not jump the gate.
|
|
assert!(!should_self_continue(&ctrl(true, true, 0)));
|
|
}
|
|
|
|
#[test]
|
|
fn synthetic_continue_shape() {
|
|
let m = synthetic_continue();
|
|
assert_eq!(m.from, "self");
|
|
assert_eq!(m.body, "continue");
|
|
assert_eq!(m.id, 0);
|
|
assert!(!m.redelivered);
|
|
assert!(m.in_reply_to.is_none());
|
|
}
|
|
}
|