Watch
0
0
Fork
You've already forked hyperhive
0

Compare commits

...
Author SHA1 Message Date
atlas
40403fc6d5 docs(#2916): drop two more doc-comments left dangling by the deletion
argus caught these on review. `recent_transient`'s doc still pointed at
NO_NODE_LABEL, and `running_transients`' still described destroy as
having no queue node behind it and linked suppress_crash_watch -- all
three deleted in the parent commit, so the first two were broken
intra-doc links and the third was simply false.

Migration is now the only operation in that sentence.
2026-08-14 00:24:45 +02:00
atlas
6338939657 refactor(#2916): destroy submits a DAG instead of an imperative teardown
Destroy was a straight-line async fn with no queue node behind it, so
nothing in the graph could answer "is this container going down on
purpose?". That gap is why an imperative crash-watch suppression guard
existed: an RAII handle held for the operation's duration, a second way
to say what every other lifecycle op already says through its node.

Reuse the existing Stop node rather than teaching a new node to stop
things:

    Stop -> DestroyContainer -> (PurgeState) -> DestroyBookkeeping

Stop already declares takes_container_down honestly, so the suppression
is now derived from the graph like every other op's. It also turns the
precondition into an edge: DestroyContainer runs only under a completed
Stop, so it operates on an already-stopped container and carries
takes_container_down = false permanently. A container still alive at
that point is a real bug and stays loud instead of being absorbed by a
flag -- which matters because a wrong true silently swallows a crash
while a wrong false only costs a spurious event.

Removes suppress_crash_watch, CrashWatchSuppression, crash_suppressed,
crash_watch_suppressed and NO_NODE_LABEL. The migration call sites went
with the obsolete startup migrations, so destroy was the last caller and
intent now has exactly one home.

destroy() becomes a submit-and-return, matching every sibling endpoint
(rebuild, kill, restart, start, pause, resume) -- it was the only
lifecycle op that awaited its work. The container rescan moves into the
bookkeeping tail, so ContainerRemoved now arrives after the 200 rather
than before it.

Also drops an orphaned doc-comment in coordinator.rs: two stacked blocks
where only the second described crash_suppressed, the first documenting
a field that no longer exists. Removing the field would have re-pointed
it at recent_transient.
2026-08-14 00:24:45 +02:00
damocles
2c7872841a hive-agent: guarantee a wake after a self-requested /compact 2026-08-13 23:17:10 +02:00
damocles
fa658567db hive-priv: replace json! with typed structs for account sidecar files 2026-08-13 23:16:17 +02:00
atlas
9f26c416c0 hide .git on the agent config mounts, like /knowledge already does
An agent's config mount is a git repo, so it could read every branch and
the full history of a config whose currently deployed value is the only
thing it may act on -- and an abandoned branch looks no different from a
live one.

The knowledge bind already solved this with an empty tmpfs overlaid on
its .git. Same rule, extended: an agent is handed a working tree, never
a repository. Folds both cases into git_overlay_flags so the reason is
stated once instead of hardcoded per mount.

Config mounts are matched by shape rather than a name list because the
set grows at runtime with each child bound into a parent.
2026-08-13 21:01:08 +02:00
atlas
c32a9367e4 gateway: reject unmatched Host instead of serving the dashboard
The `_` vhost was serving the hive's own surface, so every dashboard and
agent-UI request matched the default server rather than a named vhost --
and so did a request for any name at all, including a raw IP.

Split it: `_` keeps only `return 444`, and the hive surface moves to a
vhost named for the hive domain. `_` is `mkDefault` so an operator can
claim default_server themselves, plus an assertion for the case where
they add one without turning ours off -- nginx refuses to start on a
duplicate default_server and nixpkgs asserts nothing, so that would
otherwise surface as a gateway outage at rebuild time.
2026-08-13 19:57:27 +02:00
damocles
786e4610f0 hive-agent: replace json! with typed structs in web_ui handlers 2026-08-13 19:43:58 +02:00
atlas
127846ef1b mount the applied config repo, not the proposed one
/agents/<name>/config bound the working clone a config change is staged
in, so an agent could see a proposal that was never approved -- a config
that does not govern its container. Both objects already exist; this
repoints the bind at the deployed one.

Both mounts (own + child) now resolve through config_bind_source() so
they cannot drift, and agent_proposed_dir's doc-comment is corrected:
it claimed to be manager-editable and bind-mounted, and neither is true.
2026-08-13 19:31:17 +02:00
damocles
351341e87c hive-c0re: replace json! with typed structs in dashboard handlers 2026-08-13 18:12:35 +02:00
atlas
fbffccbbb2 feat(nix): a flake check that actually covers nix
Every other check in nix/checks.nix is a Rust derivation, so a
.nix-only diff moves no hash, the whole set is cache hits, and
`nix flake check` reports green without evaluating what changed.

`checks.module-eval` is one derivation holding a table of cases, each
named by the PROPERTY it defends. Its builder text embeds the evaluated
results, so the derivation's hash is a function of them: a nix change
that flips a property rebuilds the check and fails in the builder,
naming that property.

PROVEN, not assumed — the mechanism was executed before the cases were
written. Same expression with one property true vs false:

  drvPath true  -> 5v11mnbv…-module-eval.drv
  drvPath false -> ivm3dvv8…-module-eval.drv        (differs)
  build false   -> FAILS, stderr names the property

and the table itself was mutation-tested: inverting one case's
expectation gives `FAILED: a hive that has not opted into all-local
runs no swarm controller / module-eval: 1 of 5 properties broke`. A
check that cannot go red on a broken tree is not evidence.

Cases are named by property and never by ticket: a case named after the
ticket that prompted it has that ticket's lifetime; one named after the
property lives as long as the property does.

⚠️ It evaluates, it does not execute. Where the artifact is a command
line, a request or a certificate, a value assertion cannot stand in —
that is written into the file's header, because the gap is exactly what
made two earlier outages evaluable-but-broken.
2026-08-13 17:26:58 +02:00
atlas
f80facbbe0 refactor(3202): all-local asserts the host's own /etc/hosts entries
Clause 2 of #3202, reading 1 (mara: "the all local stuff and swarm
services auto conf belong in those mods, not spread all over").

`gateway.localHostsEntry` is the gateway's only local-deployment knob —
`openFirewall` is about EXTERNAL exposure, `tls.acme` needs a public DNS
name, `hsts` is a hardening choice. It is now asserted by the mode in
local-defaults.nix, beside the three swarm toggles, instead of being the
one all-local implication an operator still had to know about.

`mkDefault`, so "all local except this" still needs no new option.

⚠️ The non-obvious half: this does NOT change what CONTAINERS resolve.
dnsmasq sets `no-hosts = true` unconditionally, so agents keep getting
the bridge IP from the authoritative `address=` rules rather than the
host's 127.0.0.1 — which would point every agent at its own netns. That
guard already existing is what makes this safe to default on; without it
this one line would break every agent's access to the forge.
2026-08-13 17:26:08 +02:00
atlas
030eef0948 refactor(3202): the swarm UI declares its own vhost and dns name
Last of the four. The vhost, its `auth_request` block and the swarm
apex's dns record move into swarm-ui.nix; vhosts.nix drops `uiCfg`,
`controllerCfg` and `autheliaCfg` and is now 259 lines of hive surface
with no swarm service in it.

Also collapses a THIRD copy of the per-service list. `networking.hosts`
restated every service's name with its own copy of that service's guard,
after the vhosts and the dnsmasq records had each done the same. It asks
the same question — which names does this host answer for — so it now
reads the same answer: a service added later lands in /etc/hosts with no
edit, and cannot land there under a different condition than it used for
DNS.

The `forceSSL`-not-`addSSL` comment travels intact: it records that
authelia answers an http auth subrequest with 400 and nginx's
auth_request only understands 2xx/401/403, so the scheme is load-bearing
for this vhost and no other.
2026-08-13 16:19:08 +02:00
atlas
6caf177416 refactor(3202): matrix declares its own vhost, dns name and SPA map
Moves the matrix sub-domain vhost out of the gateway's vhosts.nix, its
`address=` rule out of dnsmasq.nix, and the Accept-header
`$matrix_spa_target` map out of the gateway's appendHttpConfig — all
three into hive-matrix.nix.

The map is the one that had no business being where it was: it exists
solely for the SPA fallback in the vhost's `/` location, and
`appendHttpConfig` is a `lines` option, so a module can contribute to
it without the gateway assembling it.

The `.well-known/matrix/*` delegation deliberately stays on the hive's
own vhost. The spec requires it at the SERVER NAME, which is the hive
domain: that is the hive answering "where is my homeserver", not the
homeserver answering for itself. Moving it would have been the obvious
symmetric thing and it would have been wrong.
2026-08-13 16:14:37 +02:00
atlas
56ab6d26c1 refactor(3202): authelia declares its own vhost and dns name
Moves the authelia vhost out of the gateway's vhosts.nix and its
`address=` rule out of dnsmasq.nix, into swarm-authelia.nix.

Both land inside that module's existing `cfg.enable` guard, which is
the load-bearing part: every hive in a swarm knows `authelia.url`, but
only the host that RUNS the container may claim the name. A client hive
declaring the vhost would answer for a service it does not run, and
publishing the DNS record would point every agent on its bridge at that
wrong answer.

The kit grows a fourth member, `errorPages`, because the vhost aims its
502/503/504 at the gateway's styled sso-unavailable page. Republished
rather than imported per module: a service rendering its own would drift
from the rest of the gateway the first time the theme changed.
2026-08-13 16:14:37 +02:00
atlas
d60a0585d6 refactor(3202): the forge declares its own vhost and dns name
Moves `forgeVhost` out of the gateway's vhosts.nix and the forge's
`address=` rule out of dnsmasq.nix, into nix/host-modules/hive-forge —
the module that already owns everything else about the forge.

The gateway keeps what is gateway knowledge (the listen set, which
issuer covers a name, the header block) and loses the last reason it
had to read `swarm.forge` at all: `forgeCfg` is gone from both files
and from the module's `let`.

Both halves stay gated on `behindGateway` — with it off the operator
fronts forgejo themselves, so this hive must neither claim the vhost nor
answer DNS for the name.
2026-08-13 16:14:36 +02:00
atlas
991cd24fc8 refactor(3202): the gateway takes contributed dns names
Adds `services.hyperhive.gateway.localNames` (internal): hostnames the
hive resolver answers with the bridge IP, contributed by the modules
that own them. The service says which name, the gateway says where it
points — the same split `lib.tlsFor` already makes.

No behaviour change yet: the list is empty until the service modules
contribute in the following commits.

The assertion is not defensive padding. Duplicate `address=` rules do
not make dnsmasq complain; it resolves them by precedence, so a name
claimed twice silently stops being served by one of its claimants. That
failure mode only becomes reachable because contribution is now open, so
it gets closed in the same commit that opens it.
2026-08-13 16:14:36 +02:00
41 changed files with 1469 additions and 644 deletions

1
Cargo.lock generated
View file

@ -1795,6 +1795,7 @@ dependencies = [
"anyhow",
"hive-priv-sock",
"libc",
"serde",
"serde_json",
"tokio",
"tracing",

View file

@ -204,6 +204,7 @@
system
treefmt-eval
;
inherit (nixpkgs.lib) nixosSystem;
}
);
};

View file

@ -80,6 +80,19 @@ pub struct RemindArgs {
pub file_path: Option<String>,
}
/// MCP tool args for `compact`.
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct CompactArgs {
/// Optional wake-up prompt. When set and the compact actually runs
/// (gated on context usage — see the tool description), the harness
/// drives one synthetic follow-up turn with this string as its body
/// as soon as compaction finishes, so you don't have to wait for the
/// next external event to continue. Omit for a fire-and-forget compact
/// with no follow-up.
#[serde(default)]
pub wake_prompt: Option<String>,
}
// -----------------------------------------------------------------------------
// Privileged tool arg types (lifecycle, approvals, scheduling, diagnostics)
// -----------------------------------------------------------------------------

View file

@ -25,10 +25,10 @@ mod render;
pub use args::{
AckUntilArgs, AgentGetLooseEndsArgs, AnswerArgs, AskArgs, CancelLooseEndArgs,
CancelScheduleArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, GetAgentMetaArgs,
GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, RemindArgs,
RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs, SetStatusArgs,
StartArgs, UpdateArgs, UpdateMetaInputsArgs,
CancelScheduleArgs, CompactArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs,
GetAgentMetaArgs, GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs,
RemindArgs, RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs,
SetStatusArgs, StartArgs, UpdateArgs, UpdateMetaInputsArgs,
};
pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv};
@ -657,11 +657,19 @@ impl AgentServer {
that the call is refused with an explanation and has no effect. On a pass, \
queues compaction for the end of the current turn (same deferred mechanism \
the dashboard button uses, so it never races a live claude process); the \
usual pre-compaction notes-checkpoint turn still fires first. No args."
usual pre-compaction notes-checkpoint turn still fires first. Pass \
`wake_prompt` to have the harness drive one synthetic follow-up turn with \
that body as soon as compaction finishes — without it you just go idle \
waiting for the next external event, same as ending a turn normally."
)]
async fn compact(&self) -> String {
run_tool_envelope("compact", String::new(), async move {
match dial_agent_socket(&hive_agent_sock::Request::Compact).await {
async fn compact(&self, Parameters(args): Parameters<CompactArgs>) -> String {
let log = format!("{args:?}");
run_tool_envelope("compact", log, async move {
match dial_agent_socket(&hive_agent_sock::Request::Compact {
wake_prompt: args.wake_prompt,
})
.await
{
Some(hive_agent_sock::Response::Ok) => {
"compact queued — will run at the end of the current turn".to_owned()
}

View file

@ -117,7 +117,12 @@ pub enum Request {
/// explains why and takes no action. On a pass, queues the same
/// deferred `compact_pending` flag the operator's button sets (consumed
/// at the next turn boundary), so it never races a live claude process.
Compact,
/// `wake_prompt`, when set, is driven as a synthetic follow-up turn once
/// the compaction actually finishes — the dashboard button's own
/// requests go through `Bus::request_compact` directly with `None`, not
/// through this variant, since a human watching the dashboard isn't
/// waiting on a wake.
Compact { wake_prompt: Option<String> },
/// Mirror an outstanding question this agent asked (`ask()` succeeded).
/// `target` is who it's waiting on (`"operator"` when asked with
/// `to: None`). Part of the questions-mirror increment — see

View file

@ -260,6 +260,14 @@ pub enum TurnState {
Compacting,
}
/// One pending `/compact` request (see `Bus::request_compact`).
/// `wake_prompt` is what to drive as a synthetic follow-up turn once the
/// compaction actually finishes, if anything.
#[derive(Debug, Clone)]
pub struct CompactRequest {
pub wake_prompt: Option<String>,
}
#[derive(Clone)]
pub struct Bus {
tx: Arc<broadcast::Sender<BusEvent>>,
@ -307,7 +315,23 @@ pub struct Bus {
/// One-shot: run `/compact` after the next turn ends. Consumed at the end
/// of the current/next turn by `turn::drive_turn`. Deferring to the turn
/// boundary keeps compaction from racing a live claude process mid-turn.
compact_pending: Arc<AtomicBool>,
/// `Some(request)` when a compact is pending; `request.wake_prompt` is
/// what to drive as a synthetic follow-up turn once the compaction
/// actually completes (`None` = pending but no follow-up wake wanted,
/// e.g. the operator dashboard's `/compact` button). `None` = no compact
/// pending. Wrapped in [`CompactRequest`] rather than
/// `Option<Option<String>>` (clippy pedantic's `option_option` lint,
/// and the named field reads clearer at call sites than a bare nested
/// `Option`) so "pending" and "what to wake with" can never desync.
compact_pending: Arc<Mutex<Option<CompactRequest>>>,
/// One-shot, written by `turn::drive_turn`/`turn::run_pending_compact`
/// right after a compaction they served finishes, when that compact's
/// request carried a `wake_prompt`. Read once by the `hive-agent` serve
/// loop after either call site to decide whether to drive a synthetic
/// follow-up turn. Separate from `compact_pending`: by the time this is
/// set, the compact has already run and that flag has already been
/// cleared by `take_compact`.
post_compact_wake: Arc<Mutex<Option<String>>>,
/// Current fresh-claude-session id (FK to `sessions.id`). Set by the
/// bin loop after minting a session row on a fresh start; stamped onto
/// every `turn_stats` row until the next fresh session. `None` before
@ -397,7 +421,8 @@ impl Bus {
last_cost_usage: Arc::new(Mutex::new(None)),
rate_limited: Arc::new(AtomicBool::new(was_rate_limited)),
session_reset_pending: Arc::new(AtomicBool::new(false)),
compact_pending: Arc::new(AtomicBool::new(false)),
compact_pending: Arc::new(Mutex::new(None)),
post_compact_wake: Arc::new(Mutex::new(None)),
session_id: Arc::new(Mutex::new(None)),
fresh_session: Arc::new(AtomicBool::new(false)),
tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())),
@ -438,16 +463,40 @@ impl Bus {
}
/// Request a compaction after the next turn ends (deferred to the turn
/// boundary). Idempotent.
pub fn request_compact(&self) {
self.compact_pending.store(true, Ordering::SeqCst);
/// boundary). Idempotent — a second request before the first is
/// serviced just overwrites `wake_prompt` with the latest ask. `Some
/// (wake_prompt)` schedules a synthetic follow-up turn (driven with
/// `wake_prompt` as its body) once the compaction actually completes;
/// `None` requests a plain compact with no follow-up wake (the operator
/// dashboard's `/compact` button).
pub fn request_compact(&self, wake_prompt: Option<String>) {
*self.compact_pending.lock().unwrap() = Some(CompactRequest { wake_prompt });
}
/// Take + clear the compact one-shot. Returns true iff `drive_turn` should
/// compact at the end of this turn.
/// Take + clear the compact one-shot. `Some(request)` means
/// `drive_turn`/`run_pending_compact` should compact now —
/// `request.wake_prompt` is what to pass to `set_post_compact_wake` once
/// that compaction finishes. `None` means no compact is pending.
#[must_use]
pub fn take_compact(&self) -> bool {
self.compact_pending.swap(false, Ordering::SeqCst)
pub fn take_compact(&self) -> Option<CompactRequest> {
self.compact_pending.lock().unwrap().take()
}
/// Record that a just-finished compaction should drive a synthetic
/// follow-up turn with `prompt` as its body. Called by
/// `turn::drive_turn`/`turn::run_pending_compact` right after the
/// compaction they served (whose `take_compact()` returned a request
/// with `wake_prompt: Some(prompt)`) completes.
pub fn set_post_compact_wake(&self, prompt: String) {
*self.post_compact_wake.lock().unwrap() = Some(prompt);
}
/// Take + clear the post-compact wake one-shot. The serve loop calls
/// this after either compact call site to decide whether to
/// synthesize a follow-up turn.
#[must_use]
pub fn take_post_compact_wake(&self) -> Option<String> {
self.post_compact_wake.lock().unwrap().take()
}
/// Mark that the current turn started a fresh claude session.

View file

@ -240,6 +240,21 @@ fn synthetic_todo_message(stern: bool) -> hive_sh4re::inbox::DeliveredMessage {
}
}
/// Synthetic message that drives the follow-up turn when a `/compact` call
/// (self-requested via the `compact` MCP tool's `wake_prompt` arg) finishes
/// and asked to be woken. Mirrors `synthetic_todo_message`'s "no broker row"
/// sentinel shape (`id = 0`) — the compact tool call itself is the durable
/// record that a wake was requested, not a broker message.
fn post_compact_wake_message(prompt: String) -> hive_sh4re::inbox::DeliveredMessage {
hive_sh4re::inbox::DeliveredMessage {
from: "compact".into(),
body: prompt,
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.
@ -772,8 +787,18 @@ async fn serve_loop<S: Surface>(
let compacted = turn::run_pending_compact(files, &bus, &session).await;
if !compacted {
tokio::time::sleep(interval).await;
continue;
}
continue;
// The compact that just ran may have carried a wake prompt
// (the agent's own `compact` tool, not the operator button —
// see `Bus::request_compact`). If so, drive it as a synthetic
// turn right now instead of looping back to `recv_next` and
// waiting for the next external event.
let Some(prompt) = bus.take_post_compact_wake() else {
continue;
};
tracing::debug!("post-compact wake queued, driving synthetic follow-up turn");
post_compact_wake_message(prompt)
}
RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry.
@ -802,32 +827,83 @@ async fn serve_loop<S: Surface>(
return Ok(());
}
};
let ctrl = handle_turn::<S>(
let turn_ctx = TurnCtx {
socket,
&bus,
stats.as_ref(),
bus: &bus,
stats: stats.as_ref(),
files,
&session,
session: &session,
interrupted: &interrupted,
login_state: &login_state,
claude_dir: &claude_dir,
interval,
};
drive_turn_and_wake_chain::<S>(&turn_ctx, next, &mut todo_miss_streak).await;
}
}
/// Loop-invariant turn-driving context for `drive_turn_and_wake_chain`,
/// threaded as one bundle instead of double-digit positional args
/// (clippy's `too_many_arguments`). Everything here is constant for the
/// lifetime of one `serve_loop` call; only the message to drive and the
/// todo-miss streak vary per turn and stay as separate params.
struct TurnCtx<'a> {
socket: &'a Path,
bus: &'a Bus,
stats: Option<&'a TurnStats>,
files: &'a turn::TurnFiles,
session: &'a turn::AgentSession,
interrupted: &'a Arc<std::sync::atomic::AtomicBool>,
login_state: &'a Arc<Mutex<LoginState>>,
claude_dir: &'a Path,
interval: Duration,
}
/// Drive `next`, then keep driving synthetic follow-up turns for as long as
/// a compact that just ran carries a wake prompt
/// (`Bus::take_post_compact_wake`) — see `serve_loop`'s comment at the call
/// site for why this loops in place instead of returning to the outer
/// `select!`/`recv_next`. Ordinarily runs exactly one iteration; only chains
/// further if the follow-up turn itself requests another woken compact.
/// Split out of `serve_loop` purely to keep that function under clippy's
/// line limit.
async fn drive_turn_and_wake_chain<S: Surface>(
ctx: &TurnCtx<'_>,
mut next: hive_sh4re::inbox::DeliveredMessage,
todo_miss_streak: &mut u32,
) {
loop {
let ctrl = handle_turn::<S>(
ctx.socket,
ctx.bus,
ctx.stats,
ctx.files,
ctx.session,
next,
&interrupted,
ctx.interrupted,
)
.await;
apply_todo_wake_checked(ctrl.todo_wake_checked, &mut todo_miss_streak, &bus);
apply_todo_wake_checked(ctrl.todo_wake_checked, todo_miss_streak, ctx.bus);
if ctrl.auth_failed {
*login_state.lock().unwrap() = LoginState::NeedsLogin;
*ctx.login_state.lock().unwrap() = LoginState::NeedsLogin;
// Baseline the resume check on *this instant*, not on a
// directory snapshot taken after `wait_for_login` starts
// polling — closes the race where a login lands between the
// 401 and the first poll. See `wait_for_login`'s doc comment.
login::wait_for_login(
&claude_dir,
login_state.clone(),
&bus,
u64::try_from(interval.as_millis()).unwrap_or(2000),
ctx.claude_dir,
ctx.login_state.clone(),
ctx.bus,
u64::try_from(ctx.interval.as_millis()).unwrap_or(2000),
std::time::SystemTime::now(),
)
.await;
}
let Some(prompt) = ctx.bus.take_post_compact_wake() else {
break;
};
tracing::debug!("post-compact wake queued mid-turn-flow, driving synthetic follow-up turn");
next = post_compact_wake_message(prompt);
}
}

View file

@ -242,7 +242,7 @@ fn dispatch(
} => record_answering_question(questions, id, &asker, &question),
Request::ClearQuestion { id } => clear_question(questions, id),
Request::ListQuestions => list_questions(questions),
Request::Compact => compact(bus),
Request::Compact { wake_prompt } => compact(bus, wake_prompt),
}
}
@ -458,8 +458,10 @@ fn mark_todos_done(store: &Todos, ids: &[i64]) -> Response {
/// 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.
fn compact(bus: &Bus) -> Response {
/// 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(),
@ -487,9 +489,16 @@ fn compact(bus: &Bus) -> Response {
),
};
}
bus.request_compact();
let will_wake = wake_prompt.is_some();
bus.request_compact(wake_prompt);
bus.emit(crate::events::LiveEvent::Note {
text: "agent: self-requested /compact — running at the end of the current turn".into(),
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
}

View file

@ -374,15 +374,18 @@ pub async fn drive_turn(
archive_session(bus);
return Err(TurnError::PromptTooLong);
}
// Operator `/compact` (`POST /api/compact`) deferred to the turn boundary:
// run it now that the turn is done, so it works mid-turn rather than only
// when the agent is idle. Only on a healthy turn — no point spawning a
// compaction after a rate-limited / auth-failed / crashed one.
// `is_ok()` first: `take_compact()` clears the flag, so it must only fire
// when the compaction will actually run. On an unhealthy turn
// (rate-limited / auth-failed / failed) the flag is left set for the next
// turn or the idle `run_pending_compact` to service — not silently eaten.
if outcome.is_ok() && bus.take_compact() {
// Operator `/compact` (`POST /api/compact`) or an agent's own `compact`
// MCP tool call, deferred to the turn boundary: run it now that the turn
// is done, so it works mid-turn rather than only when the agent is idle.
// Only on a healthy turn — no point spawning a compaction after a
// rate-limited / auth-failed / crashed one. `is_ok()` first: `take_compact()`
// clears the flag, so it must only fire when the compaction will actually
// run. On an unhealthy turn (rate-limited / auth-failed / failed) the flag
// is left set for the next turn or the idle `run_pending_compact` to
// service — not silently eaten.
if outcome.is_ok()
&& let Some(request) = bus.take_compact()
{
bus.emit(LiveEvent::Note {
text: "operator: /compact — running at turn end".into(),
});
@ -390,6 +393,14 @@ pub async fn drive_turn(
// does; the serve loop resets to `Idle` once this turn returns.
bus.set_state(crate::events::TurnState::Compacting);
let _ = session.compact(&config, &sink).await;
// If the compact call asked to be woken (the agent's own `compact`
// tool with a `wake_prompt`), stash it — the serve loop reads it
// back after this turn returns and drives a synthetic follow-up
// turn, so a self-requested compact provably doesn't strand the
// agent idle waiting for the next external event.
if let Some(prompt) = request.wake_prompt {
bus.set_post_compact_wake(prompt);
}
return Ok(true);
}
outcome
@ -500,11 +511,13 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) {
/// so a queued `/compact` runs even when no turn is driving. (The in-flight
/// case is handled at the end of [`drive_turn`].) Resume-only via
/// [`InfiniteSession::compact`]: a missing session is a harmless no-op. Returns
/// `true` if a compaction ran.
/// `true` if a compaction ran; the serve loop follows up with
/// `Bus::take_post_compact_wake` to see whether a synthetic follow-up turn
/// should run (set below when the compact request carried a `wake_prompt`).
pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSession) -> bool {
if !bus.take_compact() {
let Some(request) = bus.take_compact() else {
return false;
}
};
bus.emit(LiveEvent::Note {
text: "operator: /compact — running on idle session".into(),
});
@ -520,6 +533,9 @@ pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSe
}),
}
bus.set_state(crate::events::TurnState::Idle);
if let Some(prompt) = request.wake_prompt {
bus.set_post_compact_wake(prompt);
}
true
}

View file

@ -7,7 +7,7 @@ use axum::{
http::StatusCode,
response::{IntoResponse, Response},
};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use super::{AppState, SigintOutcome, error_response};
@ -80,7 +80,10 @@ pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response
/// claude process) rather than only when the agent is idle. Returns 200
/// immediately; the compaction stream lands in the live panel when it runs.
pub(super) async fn post_compact(State(state): State<AppState>) -> Response {
state.bus.request_compact();
// No wake prompt: the operator is watching the dashboard, not waiting on
// an inbox message — `request_compact`'s wake-prompt arg exists for the
// agent's own `compact` MCP tool (`todo_server.rs::compact`).
state.bus.request_compact(None);
state.bus.emit(crate::events::LiveEvent::Note {
text: "operator: /compact queued — runs at the end of the current turn".into(),
});
@ -202,5 +205,10 @@ pub(super) async fn post_mark_todos_done(Form(form): Form<MarkTodosDoneForm>) ->
acked += count;
}
}
axum::Json(serde_json::json!({ "acked": acked })).into_response()
axum::Json(MarkTodosDoneBody { acked }).into_response()
}
#[derive(Serialize)]
struct MarkTodosDoneBody {
acked: u64,
}

View file

@ -2,7 +2,7 @@
use axum::extract::State;
use axum::response::{IntoResponse, Response};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use super::AppState;
@ -40,6 +40,11 @@ async fn fetch_reminder_stats(window_secs: u64) -> Option<hive_sh4re::approvals:
}
}
#[derive(Serialize)]
struct TodosBody {
todos: Vec<hive_sh4re::inbox::LooseEnd>,
}
/// `GET /api/todos` — snapshot of this agent's local todos (loose-ends v2).
///
/// Connects to the in-agent harness socket (`HIVE_AGENT_SOCKET`) and calls
@ -54,5 +59,5 @@ pub(super) async fn api_todos() -> Response {
Some(hive_agent_sock::Response::LooseEnds { loose_ends }) => loose_ends,
_ => Vec::new(),
};
axum::Json(serde_json::json!({ "todos": todos })).into_response()
axum::Json(TodosBody { todos }).into_response()
}

View file

@ -5,11 +5,23 @@ use std::convert::Infallible;
use axum::Json;
use axum::extract::{Query, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};
use super::AppState;
/// Response body for `GET /api/events/history`. `seq` is omitted from the
/// wire entirely on a paginated (non-initial) load — matches the old
/// `json!` shape, which only ever set the `"seq"` key when `Some`.
#[derive(Serialize)]
pub(super) struct EventsHistoryBody {
events: Vec<crate::events::StoredEvent>,
min_id: Option<i64>,
has_more: bool,
#[serde(skip_serializing_if = "Option::is_none")]
seq: Option<u64>,
}
/// Query params for the paginated history endpoint.
#[derive(Debug, Deserialize)]
pub(super) struct HistoryParams {
@ -23,7 +35,7 @@ pub(super) struct HistoryParams {
pub(super) async fn events_history(
State(state): State<AppState>,
Query(params): Query<HistoryParams>,
) -> Json<serde_json::Value> {
) -> Json<EventsHistoryBody> {
use crate::events::HISTORY_CAPACITY;
let limit = params.limit.unwrap_or(100).min(HISTORY_CAPACITY);
let before = params.before;
@ -51,15 +63,12 @@ pub(super) async fn events_history(
se
})
.collect();
let mut resp = serde_json::json!({
"events": events,
"min_id": min_id,
"has_more": has_more,
});
if let Some(s) = seq {
resp["seq"] = serde_json::json!(s);
}
Json(resp)
Json(EventsHistoryBody {
events,
min_id,
has_more,
seq,
})
}
pub(super) async fn events_stream(

View file

@ -899,99 +899,20 @@ pub async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) ->
/// imperative infra that `auto_update::ensure_root_agent` recreates on the
/// next hive-c0re startup if absent, so destroying it is transient rather
/// than something to refuse at the API.
pub async fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) -> Result<()> {
///
/// Submits the teardown DAG and returns — it does not wait for the container to
/// go away. Same contract as every other lifecycle op (`rebuild`, `kill`,
/// `restart`, `start`): the queue owns the work, the caller gets an
/// acknowledgement. Progress is visible as real nodes on the dashboard.
pub fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) {
tracing::info!(%name, purge, "destroy");
// Guard auto-clears on the success path's final scope exit and on
// every early-return / cancellation along the way.
// Destroy has no queue node behind it, so nothing in the graph says this
// container is going away on purpose — without this the crash watcher
// reports every destroy as a crash and the manager tries to recover it.
let guard = coord.suppress_crash_watch(name);
lifecycle::destroy(name).await?;
coord.unregister_agent(name);
let runtime = crate::paths::agent_runtime_dir(name);
if runtime.exists() {
let _ = std::fs::remove_dir_all(&runtime);
if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::destroy(b, name, purge);
Vec::new()
}) {
tracing::error!(agent = %name, error = ?e, "destroy: insert failed");
}
if purge {
// The state root may be a btrfs subvolume: a subvolume root
// can't be removed with rmdir/`remove_dir_all`, so delete it via
// hive-priv (root) first. No-op for plain-dir agents — the loop below
// then handles the plain-dir state root plus the applied dir.
if let Err(e) = crate::priv_client::delete_agent_subvolume(name).await {
tracing::warn!(error = ?e, %name, "purge: delete state subvolume failed");
}
// A malformed name can't have a persistent state tree (the state dir
// is only ever created under a validated Ident), so its removal is a
// no-op — skip the state-dir sweep and just clear the applied dir.
let state_dir = hive_types::Ident::parse(name)
.ok()
.map(|id| crate::paths::agent_state_dir(&id));
for dir in state_dir
.into_iter()
.chain([crate::paths::applied_dir(name)])
{
if dir.exists()
&& let Err(e) = std::fs::remove_dir_all(&dir)
{
tracing::warn!(error = ?e, dir = %dir.display(), "purge: remove failed");
}
}
}
// Meta flake: drop the agent's input + nixosConfiguration so a
// future spawn under the same name re-seeds cleanly, and so the
// meta lock doesn't reference a vanished applied repo. Log + keep
// going on failure — destroy already succeeded at the
// nixos-container level, the meta repo is just bookkeeping.
if let Err(e) = sync_meta_after_lifecycle(coord).await {
tracing::warn!(error = ?e, %name, "meta sync after destroy failed");
}
let _ = coord.approvals.fail_pending_for_agent(
name,
if purge {
"agent purged"
} else {
"agent destroyed"
},
);
// Drop the durable power intent — a future agent of the same name
// seeds fresh from its observed state.
if let Err(e) = coord.power.remove(name) {
tracing::warn!(%name, error = ?e, "agent_power: remove on destroy failed");
}
drop(guard);
let _ = coord
.push_todo(
hive_sh4re::manager::MANAGER_AGENT,
"core",
Some(format!("destroyed:{name}")),
format!("agent '{name}' destroyed"),
None,
false,
)
.await;
// Container row disappeared — rescan so the dashboard fires
// `ContainerRemoved` for the gone row, then emit the
// tombstones snapshot (gained one on destroy, lost one on
// purge — recompute either way).
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(coord).await;
// Re-emit the schedules snapshot: the rescan above refreshed the live
// roster, so any schedule that still targets the just-destroyed agent
// now drops that ghost column live (no page reload needed).
coord.emit_schedules_snapshot();
// Update tmpfiles.d to remove the destroyed agent's dirs from the
// boot-time pre-creation list. Best-effort: failure is logged only.
tokio::spawn(lifecycle::sync_tmpfiles());
Ok(())
}
/// Rerender the meta flake from whatever containers still exist on
/// disk. Called after lifecycle ops that change the agent set (today:
/// destroy). Idempotent — a no-op when nothing changed.
async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> {
let agents = lifecycle::agents_for_meta_listing().await?;
crate::meta::sync_agents(&coord.hive_env(), &agents).await
coord.emit_rebuild_queue_snapshot();
}
pub fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()> {

View file

@ -116,13 +116,6 @@ pub struct Coordinator {
/// is never injected into containers.
pub model_prices: crate::hive_stats::PriceTable,
agents: Mutex<HashMap<String, AgentSocket>>,
/// Agents whose lifecycle action (currently just spawn) is in flight.
/// Read by the dashboard to render a spinner; cleared when the action
/// resolves (success or failure).
/// Agents whose container is being taken down by work with **no queue node
/// behind it** (destroy, migration), so the crash watcher must not report
/// the disappearance as a crash. Not a pill — see [`CrashWatchSuppression`].
crash_suppressed: Mutex<HashSet<String>>,
/// Tombstone for transients that have JUST been cleared. The
/// crash watcher polls every 10s and would race the
/// drop-clears-immediately path of `TransientGuard`: an operator
@ -144,9 +137,7 @@ pub struct Coordinator {
/// live and both clear. Keyed by agent, the last clear *overwrites* the
/// others: a `Prebuild` (`deliberate_stop = false`) landing after a
/// `StopForUpdate` (`true`) leaves the tombstone reading `false`, and the
/// crash watcher then reports an intentional stop as a **crash**. The
/// out-of-band suppression guard, which has no node behind it, uses
/// [`NO_NODE_LABEL`].
/// crash watcher then reports an intentional stop as a **crash**.
recent_transient: Mutex<HashMap<(String, String), (bool, std::time::Instant)>>,
/// Timestamps of recent unexpected container crashes, keyed by agent.
/// Fed by `crash_watch` each time it classifies a stop as a crash (so
@ -400,57 +391,6 @@ fn fold_tombstones_by_agent<'a>(
out
}
/// Tombstone label for work with **no queue node behind it** — the
/// out-of-band operations (destroy, migration) that hold a
/// [`CrashWatchGuard`] instead of appearing in the derived transient set.
///
/// [`Coordinator::recent_transient`] is keyed by `(agent, label)` so concurrent
/// pills can't overwrite each other's `deliberate_stop`; a guard has no node and
/// therefore no node label, so it needs one of its own. Angle-bracketed to keep
/// it out of the `NodeKind::as_str` namespace — no node can ever render this.
const NO_NODE_LABEL: &str = "<no-node>";
/// RAII handle returned by [`Coordinator::suppress_crash_watch`]. While held,
/// the crash watcher treats this container disappearing as **expected**.
///
/// This is *not* a dashboard pill. Transients are derived from running queue
/// nodes and nothing stores them. But destroy and migration take a container
/// down without a node behind them, so nothing in the graph says the
/// disappearance was intended — and without that, `crash_watch` fires a
/// `ContainerCrash` for every destroy and every migrated agent, and the manager
/// tries to "recover" containers that were removed on purpose.
///
/// It is held rather than stamped once because
/// [`crate::workers::crash_watch`]'s grace window is finite and these
/// operations are not: a long destroy would outlive a single tombstone. The
/// tombstone is stamped on drop, covering the poll that lands just after.
///
/// Goes away entirely once destroy + migration are real queue nodes.
#[must_use = "suppression lasts as long as the guard; bind it for the operation's duration \
(`let _guard = coord.suppress_crash_watch(...)`). An unbound call drops it \
immediately and the very next poll can report a deliberate stop as a crash."]
pub struct CrashWatchSuppression {
coord: Arc<Coordinator>,
name: String,
}
impl Drop for CrashWatchSuppression {
fn drop(&mut self) {
self.coord
.crash_suppressed
.lock()
.unwrap()
.remove(&self.name);
// Tombstone the release so the next poll — which may land in the
// window between the container going away and this guard dropping —
// still reads the stop as deliberate.
self.coord.recent_transient.lock().unwrap().insert(
(self.name.clone(), NO_NODE_LABEL.to_owned()),
(true, std::time::Instant::now()),
);
}
}
/// RAII guard for the `meta-update` in-progress flag, held for the
/// duration of a `run_meta_update` background task. Created by
/// `Coordinator::meta_update_guard`. Drop decrements the active-run
@ -586,7 +526,6 @@ impl Coordinator {
agent_io_weight,
model_prices,
agents: Mutex::new(HashMap::new()),
crash_suppressed: Mutex::new(HashSet::new()),
recent_transient: Mutex::new(HashMap::new()),
recent_crashes: Mutex::new(HashMap::new()),
graceful_stop_pending: Mutex::new(HashSet::new()),
@ -1300,46 +1239,16 @@ impl Coordinator {
map.iter().map(|(k, v)| (k.clone(), v.len())).collect()
}
/// Tell the crash watcher that `name`'s container is going down **on
/// purpose**, for the lifetime of the returned guard. See
/// [`CrashWatchSuppression`] for why this exists at all.
///
/// Only for the operations with no queue node behind them. Anything the
/// job queue runs answers this from the node itself
/// ([`crate::job_queue::NodeKind::takes_container_down`]) and must not come
/// through here.
///
/// The guard's `Drop` runs even on task cancellation, so an aborted HTTP
/// request or a panic mid-destroy can't leave a container permanently
/// exempt from crash reporting.
pub fn suppress_crash_watch(self: &Arc<Self>, name: &str) -> CrashWatchSuppression {
self.crash_suppressed
.lock()
.unwrap()
.insert(name.to_owned());
CrashWatchSuppression {
coord: self.clone(),
name: name.to_owned(),
}
}
/// Whether a no-node operation is currently taking this container down.
#[must_use]
pub fn crash_watch_suppressed(&self, name: &str) -> bool {
self.crash_suppressed.lock().unwrap().contains(name)
}
/// Every live transient, keyed by agent.
///
/// **Derived on read, stored nowhere.** Straight off the running graph, so
/// there is no cached copy to go stale, leak, or disagree with what is
/// actually running.
///
/// Work with no queue node behind it (destroy, migration) therefore shows
/// **no pill** — there is nothing in the graph to derive one from. Its
/// crash-watch suppression is a separate, narrower thing
/// ([`Coordinator::suppress_crash_watch`]); the pill comes back for free
/// once those become real nodes.
/// Migration is the last operation with no queue node behind it, so it
/// shows **no pill** — there is nothing in the graph to derive one from.
/// It gets one for free once it becomes real nodes, the way destroy did.
///
/// ⚠️ **A `Vec` per agent, not one entry.** `running_transients` tests
/// status alone, so a lease-exempt `Prebuild` for `a` and a lease-holding
/// `StopForUpdate` for `a` are both live pills. Collapsing them to one
@ -1567,8 +1476,13 @@ impl Coordinator {
crate::paths::agent_runtime_dir(name).join("mcp.sock")
}
/// Manager-editable proposed config repo. Bind-mounted into the manager
/// container as `/agents/<name>/config/`.
/// The *proposed* config repo: where a config change lands before it is
/// applied, and what an approved deploy promotes into `applied_dir`.
///
/// **Not bind-mounted into any container.** An agent that edits a config
/// clones it from the forge itself; `/agents/<name>/config` shows the
/// applied (deployed) tree instead — see `config_bind_source` in
/// `lifecycle/host_config.rs`.
pub fn agent_proposed_dir(name: &hive_types::Ident) -> PathBuf {
crate::paths::agent_state_dir(name).join("config")
}

View file

@ -28,19 +28,20 @@ use utoipa::ToSchema;
use crate::host_stats::ServerWarning;
#[derive(Serialize, ToSchema)]
struct LiveBody {
status: &'static str,
}
/// Liveness. Always `200`; no further checks.
#[utoipa::path(
get,
path = "/health/live",
responses((status = 200, description = "process is up", body = serde_json::Value)),
responses((status = 200, description = "process is up", body = LiveBody)),
tag = "health"
)]
pub(super) async fn get_health_live() -> Response {
(
StatusCode::OK,
axum::Json(serde_json::json!({ "status": "ok" })),
)
.into_response()
(StatusCode::OK, axum::Json(LiveBody { status: "ok" })).into_response()
}
#[derive(Serialize, ToSchema)]

View file

@ -436,10 +436,9 @@ pub(super) struct DestroyForm {
params(("name" = String, Path, description = "agent name")),
request_body(content = DestroyForm, content_type = "application/x-www-form-urlencoded"),
responses(
(status = 200, description = "destroyed", body = String),
(status = 200, description = "destroy queued", body = String),
(status = 400, description = "bad agent name"),
(status = 404, description = "no such agent"),
(status = 500, description = "destroy failed"),
),
tag = "lifecycle_ops"
)]
@ -453,11 +452,10 @@ pub(super) async fn post_destroy(
}
// Checkbox semantics: any non-empty value (axum sends "on") = purge.
let purge = form.purge.as_deref().is_some_and(|v| !v.is_empty());
// `actions::destroy` rescans the container list on success, so the
// `ContainerRemoved` event lands before we return 200. The matching
// form carries `data-no-refresh`.
match actions::destroy(&state.coord, &name, purge).await {
Ok(()) => (StatusCode::OK, "ok").into_response(),
Err(e) => error_response(&format!("destroy {name} failed: {e:#}")),
}
// Submit-and-return, like every other lifecycle endpoint here. The
// container rescan now runs in the DAG's bookkeeping tail, so
// `ContainerRemoved` arrives *after* this 200 rather than before it — the
// row disappears when the event lands, same as a rebuild's does.
actions::destroy(&state.coord, &name, purge);
(StatusCode::OK, "ok").into_response()
}

View file

@ -8,26 +8,41 @@ use axum::{
http::StatusCode,
response::{IntoResponse, Response},
};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use utoipa::{IntoParams, ToSchema};
use super::{AppState, Ident, error_response, scan_validated_paths};
use crate::audit_log::AuditEntry;
use crate::container_stats::ContainerResource;
use crate::hive_stats::HiveStats;
#[derive(Serialize, ToSchema)]
pub(super) struct OperatorInboxItem {
id: i64,
from: String,
body: String,
at: chrono::DateTime<chrono::Utc>,
in_reply_to: Option<i64>,
file_refs: Vec<String>,
}
#[derive(Serialize, ToSchema)]
pub(super) struct OperatorInboxBody {
messages: Vec<OperatorInboxItem>,
}
/// Unread operator-directed messages for the dashboard's Y3R C4LL inbox.
///
/// Returns messages addressed to `"operator"` that haven't been
/// acked yet (the operator clears them via the existing
/// `POST /api/agent/operator/mark-all-read`). Newest-first; path-shaped
/// tokens are validated so the client renders file links like the
/// terminal does. Shape: `{ "messages": [{ id, from, body, at,
/// in_reply_to, file_refs }] }`.
/// terminal does.
#[utoipa::path(
get,
path = "/api/operator-inbox",
responses(
(status = 200, description = "unread operator-directed messages", body = serde_json::Value),
(status = 200, description = "unread operator-directed messages", body = OperatorInboxBody),
(status = 500, description = "broker read failed"),
),
tag = "misc_api"
@ -40,7 +55,7 @@ pub(super) async fn api_operator_inbox(State(state): State<AppState>) -> Respons
.unread_for_recipient("operator", INBOX_LIMIT)
{
Ok(messages) => {
let items: Vec<serde_json::Value> = messages
let messages: Vec<OperatorInboxItem> = messages
.into_iter()
.filter_map(|m| {
let crate::broker::MessageEvent::Sent {
@ -55,17 +70,17 @@ pub(super) async fn api_operator_inbox(State(state): State<AppState>) -> Respons
return None;
};
let file_refs = scan_validated_paths(&body);
Some(serde_json::json!({
"id": id,
"from": from,
"body": body,
"at": hive_sh4re::wire_time::from_secs(at),
"in_reply_to": in_reply_to,
"file_refs": file_refs,
}))
Some(OperatorInboxItem {
id,
from,
at: hive_sh4re::wire_time::from_secs(at),
body,
in_reply_to,
file_refs,
})
})
.collect();
axum::Json(serde_json::json!({ "messages": items })).into_response()
axum::Json(OperatorInboxBody { messages }).into_response()
}
Err(e) => error_response(&format!("operator-inbox failed: {e:#}")),
}
@ -114,18 +129,23 @@ pub(super) async fn api_container_resources() -> Response {
axum::Json(crate::container_stats::gather().await).into_response()
}
#[derive(Serialize, ToSchema)]
pub(super) struct AuditLogBody {
entries: Vec<AuditEntry>,
total: i64,
}
/// Most-recent agent-initiated privileged-action
/// audit entries, newest first (server-clamped to 500).
///
/// Backs the operator dashboard's audit view. Returns
/// `{ "entries": [AuditEntry…], "total": N }` so the UI can show
/// Backs the operator dashboard's audit view. `total` lets the UI show
/// "latest 500 of N" rather than silently capping. `ts_unix` is in
/// **seconds**.
#[utoipa::path(
get,
path = "/api/audit-log",
responses(
(status = 200, description = "recent audit entries + total count", body = serde_json::Value),
(status = 200, description = "recent audit entries + total count", body = AuditLogBody),
(status = 500, description = "sqlite read failed"),
),
tag = "misc_api"
@ -140,7 +160,12 @@ pub(super) async fn api_audit_log(State(state): State<AppState>) -> Response {
Ok(n) => n,
Err(e) => return error_response(&format!("audit-log count: {e:#}")),
};
axum::Json(serde_json::json!({ "entries": entries, "total": total })).into_response()
axum::Json(AuditLogBody { entries, total }).into_response()
}
#[derive(Serialize, ToSchema)]
pub(super) struct MarkAllReadBody {
marked: u64,
}
/// Operator-driven "clear this agent's inbox" — backs the side-panel
@ -148,14 +173,14 @@ pub(super) async fn api_audit_log(State(state): State<AppState>) -> Response {
///
/// Marks every message addressed to the agent as acked (backfilling
/// `delivered_at` for any still-pending rows so vacuum can collect
/// them). Returns `{ "marked": N }` so the frontend can show "cleared
/// N messages" feedback without an extra fetch.
/// them). `marked` lets the frontend show "cleared N messages"
/// feedback without an extra fetch.
#[utoipa::path(
post,
path = "/api/agent/{name}/mark-all-read",
params(("name" = String, Path, description = "agent name")),
responses(
(status = 200, description = "count of messages marked read", body = serde_json::Value),
(status = 200, description = "count of messages marked read", body = MarkAllReadBody),
(status = 400, description = "bad agent name"),
(status = 500, description = "broker write failed"),
),
@ -172,9 +197,9 @@ pub(super) async fn post_mark_all_read(
}
};
match state.coord.broker.mark_all_read(name.as_str()) {
Ok(n) => {
tracing::info!(%name, marked = n, "operator marked all messages read");
axum::Json(serde_json::json!({ "marked": n })).into_response()
Ok(marked) => {
tracing::info!(%name, marked, "operator marked all messages read");
axum::Json(MarkAllReadBody { marked }).into_response()
}
Err(e) => error_response(&format!("mark-all-read {name} failed: {e:#}")),
}

View file

@ -18,6 +18,16 @@ use crate::scheduled_prompts_worker::FireNowReport;
use super::{AppState, error_problem, error_response};
#[derive(serde::Serialize, utoipa::ToSchema)]
pub(super) struct NewScheduleBody {
id: i64,
}
#[derive(serde::Serialize, utoipa::ToSchema)]
pub(super) struct CancelResultBody {
cancelled: bool,
}
/// Snapshot of every schedule for the
/// scheduled-prompts tab.
///
@ -74,7 +84,7 @@ pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
// `api_schedules` above.
request_body(content = serde_json::Value, description = "SchedulePromptPayload wire shape"),
responses(
(status = 200, description = "created; body carries the new row id", body = serde_json::Value),
(status = 200, description = "created; body carries the new row id", body = NewScheduleBody),
(status = 400, description = "no targets, empty body, or interval_seconds == 0"),
(status = 500, description = "submit failed"),
),
@ -108,7 +118,7 @@ pub(super) async fn post_schedule_new(
match state.coord.scheduled_prompts.submit(&new) {
Ok(id) => {
state.coord.emit_schedules_snapshot();
Ok(axum::Json(serde_json::json!({"id": id})).into_response())
Ok(axum::Json(NewScheduleBody { id }).into_response())
}
Err(e) => Err(error_problem(&format!("schedule submit: {e:#}"))),
}
@ -177,7 +187,7 @@ pub(super) async fn post_schedule_fire_now(
post,
path = "/api/rebuild-queue/{id}/cancel",
params(("id" = u64, Path, description = "job-queue node id (a DAG's root cancels the group)")),
responses((status = 200, description = "whether the DAG was cancelled", body = serde_json::Value)),
responses((status = 200, description = "whether the DAG was cancelled", body = CancelResultBody)),
tag = "schedules"
)]
pub(super) async fn post_rebuild_queue_cancel(
@ -188,9 +198,9 @@ pub(super) async fn post_rebuild_queue_cancel(
// Any terminal side effect is the DAG's own spared tail node, which the
// scheduler picks up on its next pass — nothing to fire from here.
state.coord.emit_rebuild_queue_snapshot();
axum::Json(serde_json::json!({"cancelled": true})).into_response()
axum::Json(CancelResultBody { cancelled: true }).into_response()
} else {
axum::Json(serde_json::json!({"cancelled": false})).into_response()
axum::Json(CancelResultBody { cancelled: false }).into_response()
}
}

View file

@ -726,6 +726,18 @@ pub(super) async fn jobq_rollup(
axum::Json(state.coord.job_queue.state_rollup())
}
/// Response body for `/api/dashboard/history`. No `ToSchema` — its
/// `events` field wraps [`crate::dashboard_events::DashboardEvent`],
/// which doesn't derive `ToSchema` either (a large enum with many
/// variants; see that type's doc comment for why annotating it is
/// out of scope here). The `responses(...)` doc below spells out the
/// shape in prose instead of a `body = ...` reference.
#[derive(Serialize)]
struct DashboardHistoryBody {
seq: u64,
events: Vec<crate::dashboard_events::DashboardEvent>,
}
#[utoipa::path(
get,
path = "/api/dashboard/history",
@ -798,7 +810,7 @@ pub(super) async fn dashboard_history(State(state): State<AppState>) -> Response
}
})
.collect();
axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response()
axum::Json(DashboardHistoryBody { seq, events }).into_response()
}
Err(e) => error_response(&format!("dashboard/history failed: {e:#}")),
}

View file

@ -73,6 +73,15 @@ pub(super) async fn run_node(
NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord, agent).await,
NodeKind::Provision { .. } => run_provision(coord, agent).await,
NodeKind::Create { .. } => run_create(agent).await,
NodeKind::DestroyContainer { .. } => run_destroy_container(coord, agent).await,
NodeKind::PurgeState { .. } => {
run_purge_state(agent).await;
Ok(())
}
NodeKind::DestroyBookkeeping { purge, .. } => {
run_destroy_bookkeeping(coord, agent, *purge).await;
Ok(())
}
NodeKind::MetaLock {
sweep,
fanout,
@ -163,6 +172,112 @@ async fn run_resolve_approval(
Ok(())
}
/// Rerender the meta flake from whatever containers still exist on disk.
/// Idempotent — a no-op when nothing changed. Lives here because the destroy
/// tail is its only caller; it moved with `destroy` when that became a DAG.
async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> {
let agents = crate::lifecycle::agents_for_meta_listing().await?;
crate::meta::sync_agents(&coord.hive_env(), &agents).await
}
/// `nixos-container destroy`, then drop the agent from the roster and clear
/// its ephemeral runtime dir (the mcp socket, which does not survive a restart
/// anyway).
///
/// The only fallible step is the destroy itself: once the container is gone the
/// un-registration cannot meaningfully fail, and returning early would strand
/// the roster claiming an agent that no longer exists.
async fn run_destroy_container(coord: &Arc<Coordinator>, agent: &str) -> Result<()> {
crate::lifecycle::destroy(agent).await?;
coord.unregister_agent(agent);
let runtime = crate::paths::agent_runtime_dir(agent);
if runtime.exists() {
let _ = std::fs::remove_dir_all(&runtime);
}
Ok(())
}
/// The `purge = true` half: wipe the agent's persistent trees.
///
/// Every step is best-effort-with-a-warning rather than fatal, and that is
/// deliberate — the container is already destroyed by the time this runs, so
/// failing the node would leave the operator with a half-purged agent and a red
/// DAG, when what they can actually act on is the log line naming the path.
async fn run_purge_state(agent: &str) {
// The state root may be a btrfs subvolume: a subvolume root can't be
// removed with rmdir/`remove_dir_all`, so delete it via hive-priv (root)
// first. No-op for plain-dir agents — the loop below then handles the
// plain-dir state root plus the applied dir.
if let Err(e) = crate::priv_client::delete_agent_subvolume(agent).await {
tracing::warn!(error = ?e, %agent, "purge: delete state subvolume failed");
}
// A malformed name can't have a persistent state tree (the state dir is
// only ever created under a validated Ident), so its removal is a no-op —
// skip the state-dir sweep and just clear the applied dir.
let state_dir = hive_types::Ident::parse(agent)
.ok()
.map(|id| crate::paths::agent_state_dir(&id));
for dir in state_dir
.into_iter()
.chain([crate::paths::applied_dir(agent)])
{
if dir.exists()
&& let Err(e) = std::fs::remove_dir_all(&dir)
{
tracing::warn!(error = ?e, dir = %dir.display(), "purge: remove failed");
}
}
}
/// Post-destroy bookkeeping. Infallible by construction: every step is
/// warn-and-continue, because the destroy it follows has already succeeded and
/// none of this is undoable — a failed meta sync or power-store write is a
/// bookkeeping drift to log, not a reason to red a DAG whose container is
/// already gone.
async fn run_destroy_bookkeeping(coord: &Arc<Coordinator>, agent: &str, purge: bool) {
// Meta flake: drop the agent's input + nixosConfiguration so a future spawn
// under the same name re-seeds cleanly, and so the meta lock doesn't
// reference a vanished applied repo.
if let Err(e) = sync_meta_after_lifecycle(coord).await {
tracing::warn!(error = ?e, %agent, "meta sync after destroy failed");
}
let _ = coord.approvals.fail_pending_for_agent(
agent,
if purge {
"agent purged"
} else {
"agent destroyed"
},
);
// Drop the durable power intent — a future agent of the same name seeds
// fresh from its observed state.
if let Err(e) = coord.power.remove(agent) {
tracing::warn!(%agent, error = ?e, "agent_power: remove on destroy failed");
}
let _ = coord
.push_todo(
hive_sh4re::manager::MANAGER_AGENT,
"core",
Some(format!("destroyed:{agent}")),
format!("agent '{agent}' destroyed"),
None,
false,
)
.await;
// Container row disappeared — rescan so the dashboard fires
// `ContainerRemoved` for the gone row, then emit the tombstones snapshot
// (gained one on destroy, lost one on purge — recompute either way).
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(coord).await;
// Re-emit the schedules snapshot: the rescan above refreshed the live
// roster, so any schedule that still targets the just-destroyed agent now
// drops that ghost column live (no page reload needed).
coord.emit_schedules_snapshot();
// Update tmpfiles.d to remove the destroyed agent's dirs from the boot-time
// pre-creation list. Best-effort: failure is logged only.
tokio::spawn(crate::lifecycle::sync_tmpfiles());
}
/// Emit this agent's rebuild-complete todo. `ok` is not computed — it is which
/// of the tail pair the graph let run. The failure note comes from the DAG's
/// first failing node, since the branch knows *that* it failed but not *why*.

Some files were not shown because too many files have changed in this diff Show more