7a826f9ee2
Six places in the tree hand-rolled the same connect / write one JSON line / read one JSON line back. Two of them — the harness serve loop's client and the MCP server's — were byte-identical apart from a six-line wrapper, ~145 lines of literal copy-paste. The other four each reimplemented a subset, and the subsets had drifted: some named the socket path in their errors and some did not, one classified transient against fatal failures and the rest retried nothing at all, two drained the response and two decoded it. That duplication was defended when the daemons were split out, on the grounds that a daemon's socket etiquette should stay visible in the crate that depends on it. The etiquette genuinely does differ. The code does not, and five copies is where "each daemon documents its own etiquette" stops paying for itself. `hive-sock-client` now owns the transport once, generic over the request and response types so it is protocol-agnostic: the host-served control socket and the harness's in-agent socket both use it with their own wire-type crates. The two real differences become values instead of forks. Retry is `Retry::RideOutRestart` (2/4/8/16/30s, sized to ride out a service restart) for callers with no natural retry of their own, or `Retry::None` for callers already inside a poll loop where the poll interval is the retry — and the reason each caller picked one is a comment at the call site rather than a reimplementation. The response is either decoded (`request`) or half-closed and drained (`notify`, where the drain exists so the server's write-back doesn't land on a closed socket). Whether a failure propagates or is logged and swallowed stays at the call site, because that is the caller's choice and not a property of the transport. Errors always name the socket path now, everywhere. That detail is load-bearing: a permission problem on a socket that reads as "is the daemon running?" sends the operator to fix the wrong thing. The transient-against-fatal enum is gone rather than moved. Serialising happens before the retry loop and deserialising after it, so only connect, I/O and short-read failures can reach the loop at all — a deterministic failure is now unretryable by construction instead of by classification. It is deliberately a new crate and not part of `hive-agent-sock`. The `*-sock` crates are pure wire types by convention — `hive-agent-sock` depends on serde and nothing else — and the two largest copies talk to the host socket, whose types live in a different crate entirely. A transport in either wire-type crate would drag tokio into it and point the wrong way besides. No wire-format change: same JSON line in, same line out.
986 lines
41 KiB
Rust
986 lines
41 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 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";
|
|
|
|
/// Retry policy for every request to the host-served control socket.
|
|
/// Nothing on this side of the socket has a natural retry — the serve loop
|
|
/// and the web UI both hand a failure straight to a human or to claude —
|
|
/// so a hive-c0re restart is worth waiting out rather than surfacing.
|
|
const CONTROL_SOCKET_RETRY: Retry = Retry::RideOutRestart;
|
|
|
|
/// 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};
|
|
use hive_sock_client::Retry;
|
|
|
|
#[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 hive_sock_client::request::<_, Response>(socket, &req, CONTROL_SOCKET_RETRY).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 hive_sock_client::request::<_, Response>(
|
|
socket,
|
|
&Request::Status,
|
|
CONTROL_SOCKET_RETRY,
|
|
)
|
|
.await
|
|
{
|
|
Ok(Response::Status { unread }) => unread,
|
|
_ => 0,
|
|
}
|
|
}
|
|
|
|
async fn post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
|
|
let threads = match hive_sock_client::request::<_, Response>(
|
|
socket,
|
|
&Request::GetLooseEnds { agent: None },
|
|
CONTROL_SOCKET_RETRY,
|
|
)
|
|
.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 = hive_sock_client::request::<_, Response>(
|
|
socket,
|
|
&Request::Send {
|
|
to: hive_sh4re::PARENT_RECIPIENT.into(),
|
|
body,
|
|
in_reply_to: None,
|
|
},
|
|
CONTROL_SOCKET_RETRY,
|
|
)
|
|
.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> = hive_sock_client::request(
|
|
socket,
|
|
&Request::Recv {
|
|
wait_seconds: Some(180),
|
|
max: None,
|
|
},
|
|
CONTROL_SOCKET_RETRY,
|
|
)
|
|
.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());
|
|
}
|
|
}
|