Compare commits

..
42 changed files with 1518 additions and 2153 deletions

View file

@ -8,12 +8,6 @@ jobs:
check:
name: nix flake check
runs-on: [hive-ci]
# Bound the job so a wedged build fails in minutes instead of
# hanging until the runner's 3h cap (or, when the runner itself
# deadlocks, never). 30 min is well above a cold-cache rebuild
# (~15 min observed) and well under the 3h hard cap — tune if a
# legit cold build ever trips it.
timeout-minutes: 30
steps:
- uses: actions/checkout@v3
- name: check
@ -21,19 +15,3 @@ jobs:
# cargo clippy, and module evaluation. No --no-build: the checks
# derivations are the canonical source of truth.
run: nix flake check
tracker-tags:
name: tracker-tag lint
runs-on: [hive-ci]
# Pure git+grep — seconds normally; a few minutes is already a hang.
timeout-minutes: 5
steps:
- uses: actions/checkout@v3
- name: lint
# Flags hash-number tracker tags in source (hive convention is
# prose, not tags — /knowledge/hive-rules.md). Runs as its own
# job, kept out of the required checks while the legacy backlog
# is cleaned up: a hit fails this check (red) without blocking
# merge. Promote to a required check once the tree is clean.
# See scripts/check-issue-refs.sh.
run: sh scripts/check-issue-refs.sh

View file

@ -82,7 +82,7 @@ zero-sized impl (`AgentSurface`) wrapping:
- One async method per wire op: `ack_turn`, `requeue_inflight`,
`inbox_unread`, `post_turn_counts`, `send_to_parent`,
`recv_next`, `wake_external`.
`self_wake`, `recv_next`, `wake_external`.
`main()` calls `serve_main::<AgentSurface>` for all roles. The turn
loop (`serve_loop` / `handle_turn` / `wake`) has no per-role branches.
@ -117,19 +117,9 @@ and the manager fall through to operator).
After the outcome handler, the stats sink records a row and the
`hyperhive-continue` sentinel (dropped by the `request_next_turn`
MCP tool) is consumed if present. `handle_turn` reports the result
to `serve_loop` via `TurnControl { auth_failed, continue_requested,
pending }`. When a continue was requested, the turn did not
auth-fail, and the inbox is empty (`pending == 0`), `serve_loop`
drives the next turn in-process with a synthetic
`{ from: "self", body: "continue" }` message (`synthetic_continue`)
— it never goes through the broker, so the self-continue doesn't
persist to sqlite or show up as a recv'able inbox message. If real
messages are already pending the continue is dropped: those messages
drive the next turn(s) via `recv_next`, so an explicit self-wake
isn't needed (this is the `request_next_turn` contract — "no effect
if a new inbox message arrives before this turn ends"). The
`should_self_continue` predicate encodes exactly that decision.
MCP tool) is consumed if present, firing `self_wake` so the next
turn starts with `{ from: "self", body: "continue" }` even if the
inbox is empty.
## The claude invocation

View file

@ -4,7 +4,7 @@
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-26.05";
nixpkgs-unstable.url = "github:NixOS/nixpkgs/nixpkgs-unstable";
# Crane (replaces the former naersk-based build). Stateless — no nixpkgs input to
# Crane (replaces naersk #538). Stateless — no nixpkgs input to
# follow; `crane.mkLib pkgs` returns the lib at whatever pkgs we
# pass it (we use the project's pinned nixpkgs).
crane.url = "github:ipetkov/crane";
@ -51,7 +51,7 @@
# cares about" filter (Cargo.toml/Cargo.lock + *.rs). All
# non-rust runtime assets — branding + the claude system
# prompt template + claude-settings.json — live in the
# separate `hyperhive-assets` derivation and are
# separate `hyperhive-assets` derivation (#555) and are
# loaded by the binaries at runtime from `$HIVE_ASSETS_DIR`.
# The unit tests in `hive-ag3nt::prompt` read the same
# `prompts/system.md` directly from the workspace tree at
@ -91,10 +91,10 @@
# sandbox.
# `librsvg` used to live here for `hive-c0re/build.rs`'s
# rsvg-convert call — that whole codepath moved into the
# `hyperhive-assets` derivation, so the rust
# `hyperhive-assets` derivation in #555, so the rust
# derivation no longer needs the dependency.
# `sqlite` required by matrix-sdk's `sqlite` feature
# (`hive-matrix-mcp` workspace member) — the
# (`hive-matrix-mcp` workspace member, #548 phase 3) — the
# matrix-sdk-sqlite + rusqlite stack links against system
# libsqlite3 by default.
nativeBuildInputs = [
@ -129,7 +129,7 @@
# prompt::tests` reads the production prompt template at test
# runtime through `$HIVE_ASSETS_DIR`, so wiring the env var
# into the build phase here would make the prompt's hash a
# build input of `default` (defeats the asset-split cache goal: a
# build input of `default` (defeats #555's cache goal: a
# prompt edit would still bust the binary derivation, even
# though no .rs file changed). Keeping tests in a separate
# check derivation localises the asset-rebuild blast radius
@ -144,12 +144,12 @@
};
# Bundled browser assets — see ./nix/frontend.nix. Output is
# $out/{dashboard,agent}/ which the Rust binaries serve via
# tower_http::ServeDir.
# tower_http::ServeDir (wired up in Phase 4 of #273).
frontend = pkgs.callPackage ./nix/frontend.nix {
branding-svg = ./branding/hyperhive.svg;
};
# Static runtime assets the rust binaries read via
# `hive_sh4re::assets::*`: branding/* + prompts/*,
# `hive_sh4re::assets::*` (#555): branding/* + prompts/*,
# plus the rendered agent-configs.png. Split out of the
# rust derivation so a tweak to e.g. system.md doesn't bust
# the cargo cache. Build input of the `cargo-test` check but
@ -164,6 +164,7 @@
# spawn dramatically because the heavy lifting (nixpkgs +
# claude-code + hive-ag3nt binary) is already in the store
# when the meta evaluator goes to build the container.
# Closes #97.
#
# nixosConfigurations are pinned to x86_64-linux (nixos-
# containers only run native arch), so these toplevels are
@ -174,7 +175,7 @@
agent-base-toplevel = self.nixosConfigurations.agent-base.config.system.build.toplevel;
ruth-toplevel = self.nixosConfigurations.ruth.config.system.build.toplevel;
# Auto-generated nix options reference for hyperhive.
# Auto-generated nix options reference for hyperhive (#616).
# `docs` bundles host + agent pages into one tree; the split
# outputs are useful when consumers only want one surface.
# All three are pure markdown — no rust or frontend deps in
@ -195,7 +196,7 @@
# is applied (manager + agent containers both apply it via
# `mkContainer` further down).
hyperhive-frontend = self.packages.${prev.stdenv.hostPlatform.system}.frontend;
# Static runtime assets. Exposed alongside the binary
# Static runtime assets (#555). Exposed alongside the binary
# so the harness module can wire $HIVE_ASSETS_DIR straight
# to `${pkgs.hyperhive-assets}/share/hyperhive`.
hyperhive-assets = self.packages.${prev.stdenv.hostPlatform.system}.assets;
@ -242,7 +243,7 @@
hyperhiveNixpkgsUnstable = "path:${nixpkgs-unstable}";
# Per-container toplevels — wired into `system.extraDependencies`
# when `services.hyperhive.c0re.preBuildAgentTemplates` is on so the
# host system closure pre-fetches the heavy build inputs.
# host system closure pre-fetches the heavy build inputs (#97).
# Defined only for x86_64-linux because nixosConfigurations are
# hardcoded to that system; the option's default keeps the
# extra deps gated so aarch64 hosts don't accidentally pull
@ -291,7 +292,7 @@
packages = with pkgs; [
cargo
clippy
librsvg # rsvg-convert — hive-c0re/build.rs invokes it
librsvg # rsvg-convert — hive-c0re/build.rs invokes it (#424)
pkg-config
rust-analyzer
rustc
@ -330,7 +331,7 @@
# group from that gate: pedantic is the "extra, opinionated"
# group the clippy team grows freely, so denying it means
# every toolchain bump that adds a new pedantic lint breaks CI
# with zero code changes. The `pedantic = warn`
# with zero code changes (#1368). The `pedantic = warn`
# workspace lint (Cargo.toml) keeps it as advisory signal in
# local `cargo clippy` — it just no longer blocks the build.
# (`-A` rather than `-W` here: `-W clippy::pedantic` would
@ -360,7 +361,7 @@
cargoTestExtraArgs = "--workspace";
HIVE_ASSETS_DIR = "${self.packages.${system}.assets}/share/hyperhive";
};
# Nix options docs evaluation. Cheap: pulls in
# Nix options docs evaluation (#616). Cheap: pulls in
# `nixosOptionsDoc` + the host module's stub eval, no rust or
# frontend deps. CI fails fast if a module change breaks
# option declarations or the doc rendering. Reuses the

View file

@ -3,7 +3,6 @@
@import "@hive/shared/base.css";
@import "@hive/shared/terminal.css";
@import "@hive/shared/tabs.css";
@import "@hive/shared/chrome.css";
/* global typography
Element-level rules shared across all three pages (index, flow,
@ -221,9 +220,38 @@ code {
color: var(--fg);
}
/* page chrome (.page-header / .page-back / .page-title the standalone
pages' sticky back-link bar) moved to @hive/shared/chrome.css, imported
at the top of this file. */
/* page chrome: header + back link (flow + logs pages)
.logs-header / .logs-back / .logs-title are used by both
/flow.html and /logs.html the same frosted sticky bar pattern. */
.logs-header {
position: sticky;
top: 0;
z-index: 25;
background: color-mix(in srgb, var(--bg) 92%, transparent);
-webkit-backdrop-filter: blur(8px) saturate(120%);
backdrop-filter: blur(8px) saturate(120%);
border-bottom: 1px solid var(--purple-dim);
display: flex;
align-items: center;
gap: 1.5em;
padding: 0.5em 1.5em;
}
.logs-back {
color: var(--purple);
text-decoration: none;
font-size: 0.88em;
white-space: nowrap;
flex: none;
}
.logs-back:hover { text-decoration: underline; }
.logs-title {
color: var(--subtext0);
font-size: 0.85em;
letter-spacing: 0.05em;
flex: none;
}
/* operator inbox + message rows
.inbox and .msg-* are rendered by common.js (renderInbox) and
@ -525,17 +553,14 @@ body.side-panel-resizing * { cursor: ew-resize !important; }
font-weight: bold;
letter-spacing: 0.02em;
}
/* Tint composited over the opaque elevated-surface grey (not transparent):
the bar is `position: sticky`, so a see-through background would let
page content scroll through behind it. */
.server-warn-warn {
color: var(--amber);
background: color-mix(in srgb, var(--amber) 16%, var(--bg-elev));
background: color-mix(in srgb, var(--amber) 16%, transparent);
border-bottom: 1px solid var(--amber);
}
.server-warn-crit {
color: var(--red);
background: color-mix(in srgb, var(--red) 18%, var(--bg-elev));
background: color-mix(in srgb, var(--red) 18%, transparent);
border-bottom: 1px solid var(--red);
}

View file

@ -46,11 +46,10 @@ export const form = (action, btnClass, btnLabel, confirmMsg, extra = {}, opts =
return f;
};
// `truncate`, `fmtAgo`, `fmtElapsed`, `fmtDuration` + the `paintAtomic`
// render helper live in the dashboard-internal `./util.js`, not here —
// their phrasing ("X running", "X ago") is dashboard-specific, so they
// stay out of the cross-page `common.js` but are shared across the
// dashboard's own tab modules.
// `truncate`, `fmtAgo`, `fmtElapsed`, `fmtDuration` stay in tabs.js
// for now — each has display-specific phrasing ("X running", "X ago")
// tied to its caller, so they don't generalise cleanly. We can lift
// them when a second consumer needs the same shape.
// ─── shared-worker SSE pipe ─────────────────────────────────────────────
// Returns an EventSource-shaped facade backed by a SharedWorker that

View file

@ -21,12 +21,6 @@ body.dashboard-shell {
}
.dashboard-chrome {
/* Home back-link + tab strip share one row (flex). The chrome owns the
horizontal gutter and the single full-width bottom divider; its
children drop their own side padding / border so they sit inline. */
display: flex;
align-items: center;
gap: 0.75em;
position: sticky;
top: 0;
z-index: 25;
@ -34,19 +28,19 @@ body.dashboard-shell {
-webkit-backdrop-filter: blur(8px) saturate(120%);
backdrop-filter: blur(8px) saturate(120%);
border-bottom: 1px solid var(--purple-dim);
padding: 0.3em 1em;
padding: 0.4em 0 0;
margin: 0 0 1em;
}
/* home back-link to the H0M3 hub. Small purple link top-left of the
chrome, mirroring the .page-back treatment the standalone pages use
(now shared from @hive/shared/chrome.css) so the navigation reads
consistently. The dashboard keeps this richer sticky chrome of its
own rather than folding into the simple page header. */
chrome, mirroring the .logs-back treatment the sub-pages use so the
navigation reads consistently. Full chrome unification (a shared,
reusable tab/back-link component) is a later step. */
.dash-home-back {
flex: none;
display: inline-block;
color: var(--purple);
text-decoration: none;
font-size: 0.82em;
padding: 0.1em 1em 0;
white-space: nowrap;
}
.dash-home-back:hover { text-decoration: underline; }
@ -67,36 +61,38 @@ body.dashboard-shell {
flex-wrap: nowrap;
align-items: center;
gap: 0.2em;
/* On one row with the home link now; the chrome owns the gutter +
the bottom divider, so no outer padding / border here. */
padding: 0.5em 1em 0;
border-bottom: 1px solid var(--purple-dim);
}
/* Shares the flat-pill aesthetic of the shared `.hive-tab` (sub-tabs on
the standalone pages) so the chrome reads as one family. Kept a touch
weightier than bare `.hive-tab` larger font, bold labels, count
pills since this is the dashboard's primary nav, not a sub-strip. */
.tabbar .tab {
display: inline-flex;
align-items: center;
gap: 0.5em;
flex-shrink: 0;
padding: 0.4em 0.95em;
padding: 0.55em 1em 0.45em;
margin-bottom: -1px; /* overlap the tabbar bottom border */
color: var(--muted);
font-family: inherit;
font-size: 0.92em;
letter-spacing: 0.06em;
letter-spacing: 0.08em;
text-decoration: none;
border: 0;
border-radius: 4px;
border: 1px solid transparent;
border-bottom: 0;
border-radius: 4px 4px 0 0;
cursor: pointer;
transition: color 0.15s ease, background 0.15s ease;
transition: color 0.15s ease, background 0.15s ease, border-color 0.15s ease;
}
.tabbar .tab:hover {
color: var(--fg);
background: var(--border);
color: var(--purple);
background: color-mix(in srgb, var(--purple) 6%, transparent);
}
.tabbar .tab.hive-tab--active {
color: var(--purple);
background: var(--border);
border-color: var(--purple-dim);
background: var(--bg);
/* Lift the active tab visually the bottom border of the tabbar
yields under it via the -1px margin above. */
box-shadow: 0 -2px 12px -4px color-mix(in srgb, var(--purple) 40%, transparent);
}
.tab-label { font-weight: bold; white-space: nowrap; }
.tab-count {

View file

@ -16,9 +16,9 @@
No full tabbar — the flow page is a dedicated full-viewport
terminal surface; navigating back to the menu is the only chrome
needed. Pages link back to H0M3, not the dashboard. -->
<header class="page-header">
<a class="page-back" href="/">← home</a>
<span class="page-title">FL0W</span>
<header class="logs-header">
<a class="logs-back" href="/">← home</a>
<span class="logs-title">FL0W</span>
<select id="flow-agent-filter" class="flow-filter" title="filter timeline by agent" aria-label="filter timeline by agent">
<option value="">all agents</option>
</select>

View file

@ -51,19 +51,6 @@ body.home-shell {
box-shadow: 0 -2px 14px -6px color-mix(in srgb, var(--purple) 50%, transparent);
}
/* Icon + label share the top row of each tile; the description sits
below. The icon is decorative (aria-hidden) the label carries the
accessible name. */
.home-tile-head {
display: flex;
align-items: center;
gap: 0.55em;
}
.home-tile-icon {
font-size: 1.25em;
line-height: 1;
flex: none;
}
.home-tile-label {
font-weight: bold;
letter-spacing: 0.08em;

View file

@ -29,42 +29,27 @@
<nav class="home-menu" aria-label="hyperhive surfaces">
<a class="home-tile" href="/dashboard.html">
<span class="home-tile-head">
<span class="home-tile-icon" aria-hidden="true">🖥</span>
<span class="home-tile-label">Dashboard</span>
</span>
<span class="home-tile-label">Dashboard</span>
<span class="home-tile-desc">containers · approvals · permissions · schedules · system</span>
</a>
<a class="home-tile" href="/flow.html">
<span class="home-tile-head">
<span class="home-tile-icon" aria-hidden="true">📡</span>
<span class="home-tile-label">Flow</span>
</span>
<span class="home-tile-label">Flow</span>
<span class="home-tile-desc">live all-agents message firehose</span>
</a>
<a class="home-tile" href="/logs.html">
<span class="home-tile-head">
<span class="home-tile-icon" aria-hidden="true">📜</span>
<span class="home-tile-label">Logs</span>
</span>
<span class="home-tile-label">Logs</span>
<span class="home-tile-desc">build · agent · system logs</span>
</a>
<a class="home-tile" href="/stats.html">
<span class="home-tile-head">
<span class="home-tile-icon" aria-hidden="true">📊</span>
<span class="home-tile-label">Stats</span>
</span>
<span class="home-tile-label">Stats</span>
<span class="home-tile-desc">hive-wide turn stats · cost · model mix</span>
</a>
<a class="home-tile" href="/settings.html">
<span class="home-tile-head">
<span class="home-tile-icon" aria-hidden="true"></span>
<span class="home-tile-label">Settings</span>
</span>
<span class="home-tile-label">Settings</span>
<span class="home-tile-desc">operator-local prefs · browser notifications</span>
</a>
@ -72,10 +57,7 @@
enabled (state.matrix_gui_enabled), mirroring the dashboard
tab gating so operators without it don't see a dead link. -->
<a class="home-tile" id="home-tile-matrix" href="/matrix/" hidden>
<span class="home-tile-head">
<span class="home-tile-icon" aria-hidden="true">💬</span>
<span class="home-tile-label">Matrix</span>
</span>
<span class="home-tile-label">Matrix</span>
<span class="home-tile-desc">matrix chat client</span>
</a>

View file

@ -15,8 +15,8 @@
<!-- Minimal chrome: back link + sub-tab strip.
Same pattern as flow.html — no full dashboard tabbar. Back link
points to the H0M3 hub (served at /), not the dashboard. -->
<header class="page-header">
<a class="page-back" href="/">← home</a>
<header class="logs-header">
<a class="logs-back" href="/">← home</a>
<nav class="hive-tabbar logs-tabbar" id="logs-tabbar" role="tablist">
<a class="hive-tab" id="logs-tab-build" href="#build" role="tab"
aria-controls="logs-pane-build" data-tab="build">

View file

@ -1,270 +0,0 @@
// Dashboard P3RM1SS10NS tab — the per-agent capabilities + tool-groups
// matrices.
//
// Both tables are fetched on tab activation (`GET /api/capabilities`,
// `GET /api/tool-groups`) and after each save. Columns (caps / groups)
// come from the backend so the UI needs no change when a new one is
// added. Live updates arrive via the `capabilities_changed` /
// `tool_groups_changed` dashboard events (fired after the rebuild-queue
// worker commits the perm JSON file), wired into the entry's mutation
// dispatch table.
//
// Stateless at module scope: each render builds fresh from the fetched
// payload. The agent roster (`containersState`) is the only shared state
// it reads — to union live containers with agents already named in the
// assignments map.
import { $, el } from './common.js';
import { containersState } from './state.js';
export function applyCapabilitiesChanged(ev) {
const root = $('capabilities-section');
if (!root) return;
// Skip re-render while operator has a checkbox focused in this
// section — the tab-activation re-fetch is the recovery path.
if (root.contains(document.activeElement)) return;
renderCapabilities(root, ev);
}
export function applyToolGroupsChanged(ev) {
const root = $('tool-groups-section');
if (!root) return;
if (root.contains(document.activeElement)) return;
renderToolGroups(root, ev);
}
export async function fetchAndRenderCapabilities() {
const root = $('capabilities-section');
if (!root) return;
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'loading…'));
try {
const resp = await fetch('/api/capabilities');
if (!resp.ok) throw new Error('http ' + resp.status);
const data = await resp.json();
renderCapabilities(root, data);
} catch (err) {
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'fetch failed: ' + err));
}
}
function renderCapabilities(root, data) {
root.replaceChildren();
const { caps, descriptions = {}, assignments } = data;
if (!caps || !caps.length) {
root.append(el('p', { class: 'meta' }, '(no capabilities defined)'));
return;
}
// Agent names: union of live containers + keys already in assignments.
const agentNames = [...new Set([
...Array.from(containersState.keys()),
...Object.keys(assignments),
])].sort();
if (!agentNames.length) {
root.append(el('p', { class: 'meta' }, '(no agents)'));
return;
}
const wrap = el('div', { class: 'cap-table-wrap' });
const table = el('table', { class: 'cap-table' });
// Header row.
const thead = el('thead');
const hrow = el('tr');
hrow.append(el('th', { class: 'cap-agent-col' }, 'agent'));
for (const c of caps) {
hrow.append(el('th', { class: 'cap-col', title: descriptions[c] || c }, c));
}
hrow.append(el('th', { class: 'cap-save-col' }, ''));
thead.append(hrow);
table.append(thead);
const tbody = el('tbody');
for (const name of agentNames) {
const assigned = assignments[name] || [];
const tr = el('tr', { class: 'cap-row' });
// Agent name cell.
tr.append(el('td', { class: 'cap-agent-col' },
el('span', { class: 'cap-agent-name' }, name)));
// One checkbox per capability.
const checkboxes = [];
for (const c of caps) {
const checked = assigned.includes(c);
const td = el('td', { class: 'cap-col' });
const cb = el('input', {
type: 'checkbox',
class: 'cap-cb',
'data-cap': c,
'aria-label': c,
});
cb.checked = checked;
td.append(cb);
tr.append(td);
checkboxes.push(cb);
}
// Save button cell.
const saveTd = el('td', { class: 'cap-save-col' });
const saveBtn = el('button', { type: 'button', class: 'btn cap-save-btn' }, 'save');
saveBtn.addEventListener('click', async () => {
const selectedCaps = checkboxes
.filter((cb) => cb.checked)
.map((cb) => cb.dataset.cap);
saveBtn.disabled = true;
saveBtn.textContent = '…';
try {
const r = await fetch('/api/capabilities/' + encodeURIComponent(name), {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ caps: selectedCaps }),
});
if (!r.ok) {
const txt = await r.text();
saveBtn.textContent = 'err';
saveBtn.title = txt;
} else {
saveBtn.textContent = '✓';
setTimeout(fetchAndRenderCapabilities, 800);
}
} catch (err) {
saveBtn.textContent = 'err';
saveBtn.title = String(err);
} finally {
saveBtn.disabled = false;
}
});
saveTd.append(saveBtn);
tr.append(saveTd);
tbody.append(tr);
}
table.append(tbody);
wrap.append(table);
root.append(wrap);
}
export async function fetchAndRenderToolGroups() {
const root = $('tool-groups-section');
if (!root) return;
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'loading…'));
try {
const resp = await fetch('/api/tool-groups');
if (!resp.ok) throw new Error('http ' + resp.status);
const data = await resp.json();
renderToolGroups(root, data);
} catch (err) {
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'fetch failed: ' + err));
}
}
function renderToolGroups(root, data) {
root.replaceChildren();
const { groups, descriptions = {}, assignments } = data;
if (!groups || !groups.length) {
root.append(el('p', { class: 'meta' }, '(no tool groups defined)'));
return;
}
// Agent names: union of live containers + keys already in assignments,
// sorted alphabetically.
const agentNames = [...new Set([
...Array.from(containersState.keys()),
...Object.keys(assignments),
])].sort();
if (!agentNames.length) {
root.append(el('p', { class: 'meta' }, '(no agents)'));
return;
}
const wrap = el('div', { class: 'tg-table-wrap' });
const table = el('table', { class: 'tg-table' });
// Header row.
const thead = el('thead');
const hrow = el('tr');
hrow.append(el('th', { class: 'tg-agent-col' }, 'agent'));
for (const g of groups) {
hrow.append(el('th', { class: 'tg-group-col', title: descriptions[g] || g }, g));
}
hrow.append(el('th', { class: 'tg-save-col' }, ''));
thead.append(hrow);
table.append(thead);
const tbody = el('tbody');
for (const name of agentNames) {
// Explicit assignment or empty = using role default.
const assigned = assignments[name] || [];
const hasExplicit = Object.prototype.hasOwnProperty.call(assignments, name);
const tr = el('tr', { class: 'tg-row' });
// Agent name cell.
const nameTd = el('td', { class: 'tg-agent-col' });
nameTd.append(el('span', { class: 'tg-agent-name' }, name));
if (!hasExplicit) {
nameTd.append(el('span', { class: 'meta tg-default-label' }, '(default)'));
}
tr.append(nameTd);
// One checkbox per group.
const checkboxes = [];
for (const g of groups) {
const checked = assigned.includes(g);
const td = el('td', { class: 'tg-group-col' });
const cb = el('input', {
type: 'checkbox',
class: 'tg-cb',
'data-group': g,
'aria-label': g,
});
cb.checked = checked;
td.append(cb);
tr.append(td);
checkboxes.push(cb);
}
// Save button cell.
const saveTd = el('td', { class: 'tg-save-col' });
const saveBtn = el('button', { type: 'button', class: 'btn tg-save-btn' }, 'save');
saveBtn.addEventListener('click', async () => {
const selectedGroups = checkboxes
.filter((cb) => cb.checked)
.map((cb) => cb.dataset.group);
saveBtn.disabled = true;
saveBtn.textContent = '…';
try {
const r = await fetch('/api/tool-groups/' + encodeURIComponent(name), {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ groups: selectedGroups }),
});
if (!r.ok) {
const txt = await r.text();
saveBtn.textContent = 'err';
saveBtn.title = txt;
} else {
saveBtn.textContent = '✓';
setTimeout(fetchAndRenderToolGroups, 800);
}
} catch (err) {
saveBtn.textContent = 'err';
saveBtn.title = String(err);
} finally {
saveBtn.disabled = false;
}
});
saveTd.append(saveBtn);
tr.append(saveTd);
tbody.append(tr);
}
table.append(tbody);
wrap.append(table);
root.append(wrap);
}

View file

@ -1,9 +1,9 @@
/* /settings.html operator-local preferences
Extracted from the dashboard S3TT1NGS tab. Same
minimal-chrome pattern as /flow.html and /logs.html: the back-link
header + title (.page-header / .page-back / .page-title) live in
@hive/shared/chrome.css; only the page body padding + the notif-toggle
row are page-specific and live here. */
header + title (.logs-header / .logs-back / .logs-title) live in
common.css; only the page body padding + the notif-toggle row are
page-specific and live here. */
body.settings-shell {
margin: 0;

View file

@ -15,9 +15,9 @@
<!-- Minimal chrome: back link + title. Same pattern as flow.html /
logs.html — no full dashboard tabbar. Back link points to the
H0M3 hub (served at /), not the dashboard. -->
<header class="page-header">
<a class="page-back" href="/">← home</a>
<span class="page-title">S3TT1NGS</span>
<header class="logs-header">
<a class="logs-back" href="/">← home</a>
<span class="logs-title">S3TT1NGS</span>
</header>
<main class="settings-main">

View file

@ -1,25 +0,0 @@
// Shared dashboard state — the one piece of cross-domain state in the
// dashboard SPA: the live agent roster.
//
// The roster is read by nearly every tab (the container tree, the
// capabilities + tool-group matrices, the schedule target-chips, the
// container-load poll, the selection bar, and the operator inbox), so
// it lives here as a single source of truth that each domain module
// imports rather than threading through call signatures. Everything
// else in the dashboard is domain-local and lives with its own module.
//
// `containersState` is keyed by `ContainerView.name` so a lifecycle
// form's POST → 200 → matching SSE event can flip a single row without
// a full snapshot refetch. Mutated in place (`set` / `delete` / `clear`)
// by the container apply-handlers; consumers read it by reference, so
// the imported binding always reflects the latest roster.
export const containersState = new Map();
// Replace the whole roster from a fresh `/api/state` snapshot. Called by
// the entry's `refreshState` on cold load and after async-form submits;
// live single-row updates go through the container apply-handlers.
export function syncContainersFromSnapshot(s) {
containersState.clear();
for (const c of s.containers || []) containersState.set(c.name, c);
}

View file

@ -1,8 +1,7 @@
/* /stats.html hive-wide turn-stats rollup
Extracted from the dashboard ST4TS tab. Same minimal-
chrome pattern as /flow.html and /logs.html: the back-link header +
title (.page-header / .page-back / .page-title) live in
@hive/shared/chrome.css.
title (.logs-header / .logs-back / .logs-title) live in common.css.
`.hive-stats-table` also lives in common.css it's shared with the
dashboard SYST3M C0NT41N3R L04D table. Only the ST4TS-specific
window selector, summary chips, and bars are page-specific. */

View file

@ -15,9 +15,9 @@
<!-- Minimal chrome: back link + title. Same pattern as flow.html /
logs.html — no full dashboard tabbar. Back link points to the
H0M3 hub (served at /), not the dashboard. -->
<header class="page-header">
<a class="page-back" href="/">← home</a>
<span class="page-title">ST4TS</span>
<header class="logs-header">
<a class="logs-back" href="/">← home</a>
<span class="logs-title">ST4TS</span>
</header>
<main class="stats-main">

View file

@ -19,12 +19,6 @@ import {
openStream, renderServerWarnings,
} from './common.js';
import { createTabStrip } from '@hive/shared/tabs.js';
import { containersState, syncContainersFromSnapshot } from './state.js';
import { paintAtomic, fmtAgo, truncate, fmtElapsed, fmtDuration } from './util.js';
import {
applyCapabilitiesChanged, applyToolGroupsChanged,
fetchAndRenderCapabilities, fetchAndRenderToolGroups,
} from './permissions.js';
// mdNode (in common.js) reads `window.marked` for the markdown side
// panel preview path. Set it here on the dashboard entry so file
@ -44,6 +38,16 @@ window.marked = marked;
const CTX_WARN_TOKENS = 150_000; // fallback red threshold (≈ 75% of 200k)
const CTX_CAUTION_TOKENS = 100_000; // fallback yellow threshold (≈ 50% of 200k)
// Atomic-swap render helper: build into a DocumentFragment off-DOM,
// commit with one `replaceChildren`. See docs/web-ui.md::Atomic
// section repaint for why (no intermediate empty-state flash on
// poll cycles even when the builder allocates a lot of `el()`).
function paintAtomic(liveRoot, build) {
const buf = document.createDocumentFragment();
build(buf);
liveRoot.replaceChildren(buf);
}
// Track which items we've already notified about so a re-render
// doesn't re-fire for the same row. Keyed by stable ids; reset only
// when the page reloads.
@ -145,18 +149,23 @@ window.marked = marked;
}
});
// The live agent roster (`containersState`) + its snapshot-sync now
// live in `./state.js` — it's the one piece of cross-domain state, read
// by nearly every tab. It's still mutated live by `container_state_changed`
// (upsert by name) and `container_removed` (drop by name) via the apply
// handlers just below; the imported binding reflects those in place.
//
// Derived container state — cold-loaded from /api/state.containers,
// then mutated live by `container_state_changed` (upsert by name)
// and `container_removed` (drop by name). The coordinator's rescan
// helper fires these after every mutation site + on a periodic poll
// in crash_watch. Keyed by ContainerView.name so the lifecycle
// forms' POST → 200 → matching event flips the row without a
// snapshot refetch.
const containersState = new Map();
// Keyed container row cache. Maps agent name → { el: <li>, fingerprint }.
// Allows renderContainers to skip rebuilding rows whose displayed state
// hasn't changed — prevents full-wipe flicker + avoids redundant async
// dashboard-state fetches on every SSE event. Container-domain only,
// so it stays here (not in state.js).
// dashboard-state fetches on every SSE event.
const containerRowCache = new Map();
function syncContainersFromSnapshot(s) {
containersState.clear();
for (const c of s.containers || []) containersState.set(c.name, c);
}
function applyContainerStateChanged(ev) {
if (!ev.container || !ev.container.name) return;
containersState.set(ev.container.name, ev.container);
@ -1342,6 +1351,264 @@ window.marked = marked;
root.append(ul);
}
// ── tool-groups (permissions) table ─────────────────────────────────────
// Fetched from GET /api/tool-groups on system tab activation and after
// each save. Groups (columns) come from the backend so the UI doesn't
// need updating when a new group is added. Live updates via
// `capabilities_changed` / `tool_groups_changed` SSE events fired
// after the rebuild-queue worker commits the perm JSON file.
function applyCapabilitiesChanged(ev) {
const root = $('capabilities-section');
if (!root) return;
// Skip re-render while operator has a checkbox focused in this
// section — the tab-activation re-fetch is the recovery path.
if (root.contains(document.activeElement)) return;
renderCapabilities(root, ev);
}
function applyToolGroupsChanged(ev) {
const root = $('tool-groups-section');
if (!root) return;
if (root.contains(document.activeElement)) return;
renderToolGroups(root, ev);
}
async function fetchAndRenderCapabilities() {
const root = $('capabilities-section');
if (!root) return;
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'loading…'));
try {
const resp = await fetch('/api/capabilities');
if (!resp.ok) throw new Error('http ' + resp.status);
const data = await resp.json();
renderCapabilities(root, data);
} catch (err) {
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'fetch failed: ' + err));
}
}
function renderCapabilities(root, data) {
root.replaceChildren();
const { caps, descriptions = {}, assignments } = data;
if (!caps || !caps.length) {
root.append(el('p', { class: 'meta' }, '(no capabilities defined)'));
return;
}
// Agent names: union of live containers + keys already in assignments.
const agentNames = [...new Set([
...Array.from(containersState.keys()),
...Object.keys(assignments),
])].sort();
if (!agentNames.length) {
root.append(el('p', { class: 'meta' }, '(no agents)'));
return;
}
const wrap = el('div', { class: 'cap-table-wrap' });
const table = el('table', { class: 'cap-table' });
// Header row.
const thead = el('thead');
const hrow = el('tr');
hrow.append(el('th', { class: 'cap-agent-col' }, 'agent'));
for (const c of caps) {
hrow.append(el('th', { class: 'cap-col', title: descriptions[c] || c }, c));
}
hrow.append(el('th', { class: 'cap-save-col' }, ''));
thead.append(hrow);
table.append(thead);
const tbody = el('tbody');
for (const name of agentNames) {
const assigned = assignments[name] || [];
const tr = el('tr', { class: 'cap-row' });
// Agent name cell.
tr.append(el('td', { class: 'cap-agent-col' },
el('span', { class: 'cap-agent-name' }, name)));
// One checkbox per capability.
const checkboxes = [];
for (const c of caps) {
const checked = assigned.includes(c);
const td = el('td', { class: 'cap-col' });
const cb = el('input', {
type: 'checkbox',
class: 'cap-cb',
'data-cap': c,
'aria-label': c,
});
cb.checked = checked;
td.append(cb);
tr.append(td);
checkboxes.push(cb);
}
// Save button cell.
const saveTd = el('td', { class: 'cap-save-col' });
const saveBtn = el('button', { type: 'button', class: 'btn cap-save-btn' }, 'save');
saveBtn.addEventListener('click', async () => {
const selectedCaps = checkboxes
.filter((cb) => cb.checked)
.map((cb) => cb.dataset.cap);
saveBtn.disabled = true;
saveBtn.textContent = '…';
try {
const r = await fetch('/api/capabilities/' + encodeURIComponent(name), {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ caps: selectedCaps }),
});
if (!r.ok) {
const txt = await r.text();
saveBtn.textContent = 'err';
saveBtn.title = txt;
} else {
saveBtn.textContent = '✓';
setTimeout(fetchAndRenderCapabilities, 800);
}
} catch (err) {
saveBtn.textContent = 'err';
saveBtn.title = String(err);
} finally {
saveBtn.disabled = false;
}
});
saveTd.append(saveBtn);
tr.append(saveTd);
tbody.append(tr);
}
table.append(tbody);
wrap.append(table);
root.append(wrap);
}
async function fetchAndRenderToolGroups() {
const root = $('tool-groups-section');
if (!root) return;
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'loading…'));
try {
const resp = await fetch('/api/tool-groups');
if (!resp.ok) throw new Error('http ' + resp.status);
const data = await resp.json();
renderToolGroups(root, data);
} catch (err) {
root.replaceChildren();
root.append(el('p', { class: 'meta' }, 'fetch failed: ' + err));
}
}
function renderToolGroups(root, data) {
root.replaceChildren();
const { groups, descriptions = {}, assignments } = data;
if (!groups || !groups.length) {
root.append(el('p', { class: 'meta' }, '(no tool groups defined)'));
return;
}
// Agent names: union of live containers + keys already in assignments,
// sorted alphabetically.
const agentNames = [...new Set([
...Array.from(containersState.keys()),
...Object.keys(assignments),
])].sort();
if (!agentNames.length) {
root.append(el('p', { class: 'meta' }, '(no agents)'));
return;
}
const wrap = el('div', { class: 'tg-table-wrap' });
const table = el('table', { class: 'tg-table' });
// Header row.
const thead = el('thead');
const hrow = el('tr');
hrow.append(el('th', { class: 'tg-agent-col' }, 'agent'));
for (const g of groups) {
hrow.append(el('th', { class: 'tg-group-col', title: descriptions[g] || g }, g));
}
hrow.append(el('th', { class: 'tg-save-col' }, ''));
thead.append(hrow);
table.append(thead);
const tbody = el('tbody');
for (const name of agentNames) {
// Explicit assignment or empty = using role default.
const assigned = assignments[name] || [];
const hasExplicit = Object.prototype.hasOwnProperty.call(assignments, name);
const tr = el('tr', { class: 'tg-row' });
// Agent name cell.
const nameTd = el('td', { class: 'tg-agent-col' });
nameTd.append(el('span', { class: 'tg-agent-name' }, name));
if (!hasExplicit) {
nameTd.append(el('span', { class: 'meta tg-default-label' }, '(default)'));
}
tr.append(nameTd);
// One checkbox per group.
const checkboxes = [];
for (const g of groups) {
const checked = assigned.includes(g);
const td = el('td', { class: 'tg-group-col' });
const cb = el('input', {
type: 'checkbox',
class: 'tg-cb',
'data-group': g,
'aria-label': g,
});
cb.checked = checked;
td.append(cb);
tr.append(td);
checkboxes.push(cb);
}
// Save button cell.
const saveTd = el('td', { class: 'tg-save-col' });
const saveBtn = el('button', { type: 'button', class: 'btn tg-save-btn' }, 'save');
saveBtn.addEventListener('click', async () => {
const selectedGroups = checkboxes
.filter((cb) => cb.checked)
.map((cb) => cb.dataset.group);
saveBtn.disabled = true;
saveBtn.textContent = '…';
try {
const r = await fetch('/api/tool-groups/' + encodeURIComponent(name), {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ groups: selectedGroups }),
});
if (!r.ok) {
const txt = await r.text();
saveBtn.textContent = 'err';
saveBtn.title = txt;
} else {
saveBtn.textContent = '✓';
setTimeout(fetchAndRenderToolGroups, 800);
}
} catch (err) {
saveBtn.textContent = 'err';
saveBtn.title = String(err);
} finally {
saveBtn.disabled = false;
}
});
saveTd.append(saveBtn);
tr.append(saveTd);
tbody.append(tr);
}
table.append(tbody);
wrap.append(table);
root.append(wrap);
}
// Derived question state — cold-loaded from /api/state, then mutated
// live by `question_added` / `question_resolved` dashboard events.
const QUESTION_HISTORY_LIMIT = 20;
@ -2101,6 +2368,17 @@ window.marked = marked;
root.append(ul);
}
// Relative time, anchored to now. resolved_at is unix seconds (server-
// authored), so we don't have to worry about client/server clock skew
// for sub-minute precision.
function fmtAgo(unixSecs) {
const ageSec = Math.max(0, Math.floor(Date.now() / 1000 - unixSecs));
if (ageSec < 60) return ageSec + 's ago';
if (ageSec < 3600) return Math.floor(ageSec / 60) + 'm ago';
if (ageSec < 86400) return Math.floor(ageSec / 3600) + 'h ago';
return Math.floor(ageSec / 86400) + 'd ago';
}
function renderMetaInputs(s) {
const root = $('meta-inputs-section');
if (!root) return;
@ -2210,6 +2488,10 @@ window.marked = marked;
root.append(form);
}
function truncate(s, n) {
return s.length <= n ? s : s.slice(0, n - 1) + '…';
}
// ─── rebuild queue ──────────────────────────────────────────────────────
// Keyed row cache for the rebuild-queue list. Maps entry.id → { el, fingerprint }.
// Same pattern as containerRowCache: reuse <li> nodes whose state hasn't
@ -2437,6 +2719,12 @@ window.marked = marked;
return li;
}
function fmtElapsed(secs) {
if (secs < 60) return secs + 's running';
if (secs < 3600) return Math.floor(secs / 60) + 'm ' + (secs % 60) + 's running';
return Math.floor(secs / 3600) + 'h ' + Math.floor((secs % 3600) / 60) + 'm running';
}
// Tick once per second to refresh "running Xs" badges in place
// (mirrors the question-TTL ticker pattern above).
// Tick rebuild-queue elapsed-time badges once per second.
@ -2575,6 +2863,13 @@ window.marked = marked;
root.append(ul);
});
}
function fmtDuration(secs) {
if (secs < 60) return secs + 's';
if (secs < 3600) return Math.floor(secs / 60) + 'm ' + (secs % 60) + 's';
if (secs < 86400) return Math.floor(secs / 3600) + 'h ' + Math.floor((secs % 3600) / 60) + 'm';
return Math.floor(secs / 86400) + 'd ' + Math.floor((secs % 86400) / 3600) + 'h';
}
// ─── scheduled prompts ─────────────────────────────────────────────────
// Backend exposes `/api/schedules` (snapshot), `/api/schedules`
// (POST, operator-direct submit), `/api/schedules/{id}/cancel`

View file

@ -1,49 +0,0 @@
// Shared dashboard render + format helpers.
//
// Small pure utilities used across most dashboard tabs (the container
// tree, the rebuild queue, questions / approvals, schedules + reminders):
// an atomic-swap render helper and a handful of relative-time / duration
// formatters. They live here — a dashboard-internal module — rather than
// in the cross-page `common.js`, because they're specific to the
// dashboard's render style and not needed by the stand-alone pages.
//
// All pure: no DOM/module state captured, so any tab module can import
// them freely without ordering concerns.
// Atomic-swap render: build into an off-DOM DocumentFragment, then commit
// with a single `replaceChildren`. Avoids an intermediate empty-state
// flash on poll cycles even when the builder allocates a lot of nodes.
export function paintAtomic(liveRoot, build) {
const buf = document.createDocumentFragment();
build(buf);
liveRoot.replaceChildren(buf);
}
// Relative age of a unix timestamp, coarsened to one unit ("5m ago").
export function fmtAgo(unixSecs) {
const ageSec = Math.max(0, Math.floor(Date.now() / 1000 - unixSecs));
if (ageSec < 60) return ageSec + 's ago';
if (ageSec < 3600) return Math.floor(ageSec / 60) + 'm ago';
if (ageSec < 86400) return Math.floor(ageSec / 3600) + 'h ago';
return Math.floor(ageSec / 86400) + 'd ago';
}
// Truncate a string to `n` chars, appending an ellipsis when clipped.
export function truncate(s, n) {
return s.length <= n ? s : s.slice(0, n - 1) + '…';
}
// Running-duration label for in-flight items ("3m 12s running").
export function fmtElapsed(secs) {
if (secs < 60) return secs + 's running';
if (secs < 3600) return Math.floor(secs / 60) + 'm ' + (secs % 60) + 's running';
return Math.floor(secs / 3600) + 'h ' + Math.floor((secs % 3600) / 60) + 'm running';
}
// Compact duration label, two units deep ("1h 5m", "2d 3h").
export function fmtDuration(secs) {
if (secs < 60) return secs + 's';
if (secs < 3600) return Math.floor(secs / 60) + 'm ' + (secs % 60) + 's';
if (secs < 86400) return Math.floor(secs / 3600) + 'h ' + Math.floor((secs % 3600) / 60) + 'm';
return Math.floor(secs / 86400) + 'd ' + Math.floor((secs % 86400) / 3600) + 'h';
}

View file

@ -13,8 +13,7 @@
"./colors.css": "./src/colors.css",
"./theme.css": "./src/theme.css",
"./base.css": "./src/base.css",
"./terminal.css": "./src/terminal.css",
"./chrome.css": "./src/chrome.css"
"./terminal.css": "./src/terminal.css"
},
"files": [
"src/"

View file

@ -1,45 +0,0 @@
/* page chrome: back-link header
The minimal sticky "frosted bar" header used by every standalone page
reached from the H0M3 hub /flow, /logs, /stats, /settings. A back
link to H0M3 on the left, then either a page title or (on /logs) a
createTabStrip sub-tab nav. The operator dashboard uses its own richer
sticky chrome (`.dashboard-chrome`); these classes cover the simple
pages.
Shared across packages via @hive/shared so a new page only links the
stylesheet + writes the three-element header markup:
<header class="page-header">
<a class="page-back" href="/"> home</a>
<span class="page-title">TITLE</span>
</header>
*/
.page-header {
position: sticky;
top: 0;
z-index: 25;
background: color-mix(in srgb, var(--bg) 92%, transparent);
-webkit-backdrop-filter: blur(8px) saturate(120%);
backdrop-filter: blur(8px) saturate(120%);
border-bottom: 1px solid var(--purple-dim);
display: flex;
align-items: center;
gap: 1.5em;
padding: 0.5em 1.5em;
}
.page-back {
color: var(--purple);
text-decoration: none;
font-size: 0.88em;
white-space: nowrap;
flex: none;
}
.page-back:hover { text-decoration: underline; }
.page-title {
color: var(--subtext0);
font-size: 0.85em;
letter-spacing: 0.05em;
flex: none;
}

View file

@ -29,7 +29,7 @@
a swap. Each is pixel-identical to its prior literal under the
default palette: --crust is a darkened bg; --muted / --subtext0 are
foregroundbackground blends (two levels of dimmed text). */
--crust: color-mix(in srgb, var(--base00) 58%, #000000); /* terminal / code bg, below --bg */
--crust: color-mix(in srgb, var(--base00) 58%, #000); /* terminal / code bg, below --bg */
--muted: color-mix(in srgb, var(--base05) 55.5%, var(--base00)); /* secondary / dimmed text */
--subtext0: color-mix(in srgb, var(--base05) 77.7%, var(--base00)); /* toolbar/status text; dimmer than --fg, lighter than --muted */
}

View file

@ -130,49 +130,6 @@ fn consume_continue_sentinel() -> bool {
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,
}
}
// ---------- surface trait ----------
/// What a `Recv` long-poll returned. Decoupled from the per-role
@ -216,6 +173,10 @@ trait Surface {
/// agents/manager fall through to operator).
fn send_to_parent(socket: &Path, body: String) -> impl Future<Output = ()>;
/// Fire a `Wake { from: "self", body: "continue" }` at our own
/// inbox — the `request_next_turn` sentinel pickup.
fn self_wake(socket: &Path) -> 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
@ -303,6 +264,30 @@ impl Surface for AgentSurface {
}
}
async fn self_wake(socket: &Path) {
let res = client::request::<_, AgentResponse>(
socket,
&AgentRequest::Wake {
from: "self".into(),
body: "continue".into(),
transient: false,
},
)
.await;
match res {
Ok(AgentResponse::Ok) => {
tracing::info!("request_next_turn: injected self-continue wake");
}
Ok(AgentResponse::Err { message }) => {
tracing::warn!(%message, "check_and_inject_continue: wake rejected");
}
Err(e) => {
tracing::warn!(error = ?e, "check_and_inject_continue: wake transport error");
}
_ => {}
}
}
async fn recv_next(socket: &Path) -> RecvOutcome {
let recv: Result<AgentResponse> = client::request(
socket,
@ -454,40 +439,29 @@ async fn serve_loop<S: Surface>(
) -> Result<()> {
tracing::info!(socket = %socket.display(), "harness serve");
S::requeue_inflight(socket).await;
// 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;
loop {
let next = match self_continue.take() {
Some(msg) => msg,
None => match S::recv_next(socket).await {
RecvOutcome::Message(first) => first,
RecvOutcome::Empty => {
tokio::time::sleep(interval).await;
continue;
match S::recv_next(socket).await {
RecvOutcome::Message(first) => {
let auth_failed =
handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, first).await;
if auth_failed {
*login_state.lock().unwrap() = LoginState::NeedsLogin;
turn::wait_for_login(
&claude_dir,
login_state.clone(),
&bus,
u64::try_from(interval.as_millis()).unwrap_or(2000),
)
.await;
}
RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry.
// No backoff: the long-poll wait is itself the throttle.
continue;
}
},
};
let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, next).await;
if ctrl.auth_failed {
*login_state.lock().unwrap() = LoginState::NeedsLogin;
turn::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());
}
RecvOutcome::Empty => {
tokio::time::sleep(interval).await;
}
RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry.
// No backoff: the long-poll wait is itself the throttle.
}
}
}
}
@ -495,9 +469,8 @@ async fn serve_loop<S: Surface>(
/// 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.
/// been dropped during the turn. Returns true iff the outcome was
/// `AuthFailed` — the caller flips the harness to needs-login.
async fn handle_turn<S: Surface>(
socket: &Path,
bus: &Bus,
@ -505,7 +478,7 @@ async fn handle_turn<S: Surface>(
files: &turn::TurnFiles,
turn_lock: &TurnLock,
first: hive_sh4re::DeliveredMessage,
) -> TurnControl {
) -> bool {
let from = first.from;
let body = first.body;
let redelivered = first.redelivered;
@ -560,28 +533,27 @@ async fn handle_turn<S: Surface>(
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 {
let row = serve_common::build_row(
started_at,
ended_at,
duration_ms,
model: model_at_start,
wake_from: from.clone(),
outcome: &outcome,
model_at_start,
from.clone(),
&outcome,
bus,
open_threads_count: open_threads,
open_reminders_count: open_reminders,
});
open_threads,
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, turn::TurnOutcome::AuthFailed),
continue_requested: consume_continue_sentinel(),
pending,
if consume_continue_sentinel() {
S::self_wake(socket).await;
}
matches!(outcome, turn::TurnOutcome::AuthFailed)
}
/// External `hive wake` subcommand — push a message into our own
@ -597,50 +569,3 @@ async fn wake<S: Surface>(socket: &Path, from: String, body: String) -> Result<(
};
S::wake_external(socket, from, body).await
}
#[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());
}
}

View file

@ -36,36 +36,22 @@ pub fn now_unix() -> i64 {
.unwrap_or(0)
}
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
/// columns; `outcome` and `bus` borrow for the duration of the call.
pub struct TurnRowArgs<'a> {
pub started_at: i64,
pub ended_at: i64,
pub duration_ms: i64,
pub model: String,
pub wake_from: String,
pub outcome: &'a TurnOutcome,
pub bus: &'a Bus,
pub open_threads_count: Option<u64>,
pub open_reminders_count: Option<u64>,
}
/// Assemble a `TurnStatRow` from the harness's per-turn state. Used by both
/// the agent and manager serve loops — the shape is identical, only the
/// post-turn count fetch helpers differ (and those stay in each binary).
#[must_use]
pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow {
let TurnRowArgs {
started_at,
ended_at,
duration_ms,
model,
wake_from,
outcome,
bus,
open_threads_count,
open_reminders_count,
} = args;
#[allow(clippy::too_many_arguments)]
pub fn build_row(
started_at: i64,
ended_at: i64,
duration_ms: i64,
model: String,
wake_from: String,
outcome: &TurnOutcome,
bus: &Bus,
open_threads_count: Option<u64>,
open_reminders_count: Option<u64>,
) -> TurnStatRow {
// Prefer the API-resolved model id (e.g. `claude-opus-4-8`) captured
// from this turn's assistant events over the requested `--model`
// name/alias, so the model-mix + cost rollup label the concrete

View file

@ -637,12 +637,7 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome {
outcome
}
#[allow(
clippy::too_many_lines,
reason = "one linear subprocess driver: spawn claude, stream + classify \
stdout/stderr, then assemble the outcome; splitting it would \
fragment the streaming state across helpers"
)]
#[allow(clippy::too_many_lines)]
async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, bool, bool)> {
// Keep the last STDERR_TAIL_LINES of stderr so a non-zero exit can
// include real context in the bail message (and downstream in the

View file

@ -46,19 +46,17 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await
}
ApprovalKind::ApplyCommit => {
coord
.rebuild_queue
.enqueue_full(crate::rebuild_queue::FullEnqueue {
kind: crate::rebuild_queue::QueueKind::Rebuild,
agent: approval.agent.clone(),
source: crate::rebuild_queue::QueueSource::Approval,
reason: format!("approval #{id} apply commit"),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(id),
perm_payload: None,
depends_on: Vec::new(),
});
coord.rebuild_queue.enqueue_full(
crate::rebuild_queue::QueueKind::Rebuild,
approval.agent.clone(),
crate::rebuild_queue::QueueSource::Approval,
format!("approval #{id} apply commit"),
None,
Vec::new(),
Some(id),
None,
Vec::new(),
);
coord.emit_rebuild_queue_snapshot();
Ok(())
}
@ -68,19 +66,17 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// dashboard can show *which* inputs are about to bump.
let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let parent_id = coord
.rebuild_queue
.enqueue_full(crate::rebuild_queue::FullEnqueue {
kind: crate::rebuild_queue::QueueKind::MetaUpdate,
agent: approval.agent.clone(),
source: crate::rebuild_queue::QueueSource::Approval,
reason: format!("approval #{id} meta input update"),
parent_id: None,
inputs: inputs.clone(),
approval_id: Some(id),
perm_payload: None,
depends_on: Vec::new(),
});
let parent_id = coord.rebuild_queue.enqueue_full(
crate::rebuild_queue::QueueKind::MetaUpdate,
approval.agent.clone(),
crate::rebuild_queue::QueueSource::Approval,
format!("approval #{id} meta input update"),
None,
inputs.clone(),
Some(id),
None,
Vec::new(),
);
// Pre-enqueue cascade rebuilds in topological order so
// agents depending on updated inputs are rebuilt after the
// lock bump, matching the dashboard post_meta_update path.
@ -99,19 +95,17 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
Ok(())
}
ApprovalKind::Spawn => {
coord
.rebuild_queue
.enqueue_full(crate::rebuild_queue::FullEnqueue {
kind: crate::rebuild_queue::QueueKind::Spawn,
agent: approval.agent.clone(),
source: crate::rebuild_queue::QueueSource::Approval,
reason: format!("approval #{id} spawn"),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(id),
perm_payload: None,
depends_on: Vec::new(),
});
coord.rebuild_queue.enqueue_full(
crate::rebuild_queue::QueueKind::Spawn,
approval.agent.clone(),
crate::rebuild_queue::QueueSource::Approval,
format!("approval #{id} spawn"),
None,
Vec::new(),
Some(id),
None,
Vec::new(),
);
coord.emit_rebuild_queue_snapshot();
Ok(())
}
@ -369,15 +363,15 @@ fn finish_approval(
.as_deref()
.map(|s| s[..s.len().min(12)].to_owned());
let status_str = if ok { "approved" } else { "failed" };
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id: approval.id,
agent: &approval.agent,
coord.emit_approval_resolved(
approval.id,
&approval.agent,
approval_kind,
sha_short,
status: status_str,
note: note.clone(),
description: approval.description.clone(),
});
status_str,
note.clone(),
approval.description.clone(),
);
// For spawn/rebuild/init_config approvals, also surface the underlying
// action so the manager knows whether the lifecycle step succeeded.
// The ApprovalResolved event already carries the same `ok` signal but
@ -429,11 +423,7 @@ fn finish_approval(
/// and reset the working tree back to the last known-good main. main
/// never advances on a failed build, so a crash-and-recover doesn't
/// leave the agent pointing at a tree it can't evaluate.
#[allow(
clippy::too_many_lines,
reason = "one sequential build/tag/notify pipeline; splitting the steps \
across helpers would obscure the linear flow without shrinking it"
)]
#[allow(clippy::too_many_lines)] // sequential build/tag/notify pipeline; splitting would obscure the flow
async fn run_apply_commit(
coord: &Arc<Coordinator>,
approval: &hive_sh4re::Approval,
@ -762,15 +752,15 @@ pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()
sha,
tag,
});
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
coord.emit_approval_resolved(
id,
agent: &agent_owned,
&agent_owned,
approval_kind,
sha_short,
status: "denied",
note: note.map(String::from),
"denied",
note.map(String::from),
description,
});
);
}
Ok(())
}

View file

@ -113,6 +113,7 @@ pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
}
}
#[allow(clippy::too_many_lines)]
/// Handle the subset of `Request` variants that are identical on both
/// the agent socket and the manager socket. Returns `Some(response)` for
/// every variant it handles; returns `None` for variants with socket-specific
@ -126,6 +127,7 @@ pub(crate) async fn dispatch_shared(
agent: &str,
coord: &Arc<Coordinator>,
) -> Option<hive_sh4re::Response> {
let broker = &coord.broker;
Some(match req {
hive_sh4re::Request::Send {
to,
@ -133,16 +135,76 @@ pub(crate) async fn dispatch_shared(
in_reply_to,
} => handle_send(coord, agent, to, body, *in_reply_to),
hive_sh4re::Request::Recv { wait_seconds, max } => {
handle_recv(coord, agent, *wait_seconds, *max).await
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
match broker
.recv_blocking_batch(agent, recv_timeout(*wait_seconds), cap)
.await
{
Ok(deliveries) => hive_sh4re::Response::Messages {
messages: deliveries
.into_iter()
.map(|d| hive_sh4re::DeliveredMessage {
from: d.message.from,
body: d.message.body,
id: d.id,
redelivered: d.redelivered,
in_reply_to: d.message.in_reply_to,
})
.collect(),
},
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
hive_sh4re::Request::Status => handle_status(coord, agent),
hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body),
hive_sh4re::Request::Status => match broker.count_pending(agent) {
Ok(unread) => hive_sh4re::Response::Status { unread },
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
hive_sh4re::Request::OperatorMsg { body } => match broker.send(&Message {
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
to: agent.to_owned(),
body: body.clone(),
in_reply_to: None,
}) {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
hive_sh4re::Request::Wake {
from,
body,
transient,
} => handle_wake(coord, agent, from, body, *transient),
hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit),
} => {
if *transient {
// Transient wakes bypass sqlite — they fire the broadcast
// channel only. No redelivery on restart; no message history
// entry. Used by bash task completions.
broker.ping(agent, from, body);
hive_sh4re::Response::Ok
} else {
match broker.send(&Message {
from: from.clone(),
to: agent.to_owned(),
body: body.clone(),
in_reply_to: None,
}) {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
}
hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) {
Ok(rows) => hive_sh4re::Response::Recent { rows },
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
hive_sh4re::Request::Ask {
question,
options,
@ -173,9 +235,32 @@ pub(crate) async fn dispatch_shared(
timing,
file_path,
} => handle_remind(coord, agent, message, timing, file_path.as_deref()),
hive_sh4re::Request::SetStatus { text } => handle_set_status(coord, text),
hive_sh4re::Request::SetStatus { text } => {
if let Err(message) = crate::limits::check_status_text(text) {
return Some(hive_sh4re::Response::Err { message });
}
// The harness writes the status file to its own `state/` dir
// before sending this request (it runs as the agent user, so
// it has write access). We just trigger a dashboard rescan so
// the new value is reflected immediately.
let coord2 = Arc::clone(coord);
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
hive_sh4re::Response::Ok
}
hive_sh4re::Request::GetAgentMeta { name } => {
handle_get_agent_meta(coord, agent, name.as_deref()).await
let target = name.as_deref().unwrap_or(agent);
let (status_text, status_set_at, running) =
crate::container_view::read_agent_status_live(target).await;
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
hive_sh4re::Response::AgentMeta {
name: target.to_owned(),
running,
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
status_text,
status_set_at,
hive_name,
swarm_name,
}
}
hive_sh4re::Request::CancelLooseEnd { kind, id } => {
crate::questions::handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
@ -183,8 +268,23 @@ pub(crate) async fn dispatch_shared(
|()| hive_sh4re::Response::Ok,
)
}
hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent),
hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
hive_sh4re::Request::AckTurn => match broker.ack_turn(agent) {
Ok(_n) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
hive_sh4re::Request::RequeueInflight => match broker.requeue_inflight(agent) {
Ok(n) => {
if n > 0 {
tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
}
hive_sh4re::Response::Ok
}
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
hive_sh4re::Request::GetHostJournal {
unit,
container,
@ -194,223 +294,250 @@ pub(crate) async fn dispatch_shared(
since,
until,
} => {
dispatch_host_journal(
agent,
HostJournalArgs {
unit,
container,
lines,
priority,
grep,
since,
until,
},
)
.await
dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await
}
// Not a shared variant.
_ => return None,
})
}
/// `Recv` — long-poll the broker for up to `max` messages (capped at
/// `RECV_BATCH_MAX`), mapping deliveries onto the wire response.
async fn handle_recv(
coord: &Arc<Coordinator>,
agent: &str,
wait_seconds: Option<u64>,
max: Option<u32>,
) -> hive_sh4re::Response {
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
match coord
.broker
.recv_blocking_batch(agent, recv_timeout(wait_seconds), cap)
.await
{
Ok(deliveries) => hive_sh4re::Response::Messages {
messages: deliveries
.into_iter()
.map(|d| hive_sh4re::DeliveredMessage {
from: d.message.from,
body: d.message.body,
id: d.id,
redelivered: d.redelivered,
in_reply_to: d.message.in_reply_to,
})
.collect(),
},
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `Wake` — inject a wake into `agent`'s own inbox. Transient wakes
/// fire the broadcast channel only (no sqlite row, no redelivery on
/// restart — used by bash-task completions); durable wakes persist
/// through the broker like any other message.
fn handle_wake(
coord: &Arc<Coordinator>,
agent: &str,
from: &str,
body: &str,
transient: bool,
) -> hive_sh4re::Response {
let broker = &coord.broker;
if transient {
broker.ping(agent, from, body);
hive_sh4re::Response::Ok
} else {
match broker.send(&Message {
from: from.to_owned(),
to: agent.to_owned(),
body: body.to_owned(),
in_reply_to: None,
}) {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
}
/// `SetStatus` — validate the status text, then trigger a dashboard
/// rescan. The harness has already written the status file to its own
/// `state/` dir (it runs as the agent user), so this only refreshes the
/// dashboard's view.
fn handle_set_status(coord: &Arc<Coordinator>, text: &str) -> hive_sh4re::Response {
if let Err(message) = crate::limits::check_status_text(text) {
return hive_sh4re::Response::Err { message };
}
let coord2 = Arc::clone(coord);
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
hive_sh4re::Response::Ok
}
/// `GetAgentMeta` — identity + live status for `name` (defaults to the
/// caller). Reads the live container-view status and the hive/swarm
/// display names.
async fn handle_get_agent_meta(
coord: &Arc<Coordinator>,
agent: &str,
name: Option<&str>,
) -> hive_sh4re::Response {
let target = name.unwrap_or(agent);
let (status_text, status_set_at, running) =
crate::container_view::read_agent_status_live(target).await;
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
hive_sh4re::Response::AgentMeta {
name: target.to_owned(),
running,
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
status_text,
status_set_at,
hive_name,
swarm_name,
}
}
/// `Status` — count of pending (unread) inbox messages for `agent`.
fn handle_status(coord: &Arc<Coordinator>, agent: &str) -> hive_sh4re::Response {
match coord.broker.count_pending(agent) {
Ok(unread) => hive_sh4re::Response::Status { unread },
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `OperatorMsg` — deliver an operator-authored message into `agent`'s
/// inbox (from the `operator` recipient).
fn handle_operator_msg(coord: &Arc<Coordinator>, agent: &str, body: &str) -> hive_sh4re::Response {
match coord.broker.send(&Message {
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
to: agent.to_owned(),
body: body.to_owned(),
in_reply_to: None,
}) {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `Recent` — the last `limit` inbox rows for `agent` (read-only,
/// doesn't consume).
fn handle_recent(coord: &Arc<Coordinator>, agent: &str, limit: u64) -> hive_sh4re::Response {
match coord.broker.recent_for(agent, limit) {
Ok(rows) => hive_sh4re::Response::Recent { rows },
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `AckTurn` — mark `agent`'s in-flight delivered messages acked so
/// they don't redeliver on the next turn.
fn handle_ack_turn(coord: &Arc<Coordinator>, agent: &str) -> hive_sh4re::Response {
match coord.broker.ack_turn(agent) {
Ok(_n) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `RequeueInflight` — resurface `agent`'s unacked in-flight messages
/// (crash recovery on harness boot).
fn handle_requeue_inflight(coord: &Arc<Coordinator>, agent: &str) -> hive_sh4re::Response {
match coord.broker.requeue_inflight(agent) {
Ok(n) => {
if n > 0 {
tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
}
hive_sh4re::Response::Ok
}
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
#[allow(clippy::too_many_lines)]
async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) -> AgentResponse {
if let Some(resp) = dispatch_shared(req, agent, coord).await {
return resp;
}
match req {
AgentRequest::GetLooseEnds { agent: target } => {
handle_get_loose_ends(coord, agent, target.as_deref())
let name = resolve_agent_state_target(agent, target.as_deref());
match name {
Ok(name) => match crate::loose_ends::for_agent(coord, name) {
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
AgentRequest::CountPendingReminders { agent: target } => {
handle_count_pending_reminders(coord, agent, target.as_deref())
let name = resolve_agent_state_target(agent, target.as_deref());
match name {
Ok(name) => match coord.broker.count_pending_reminders_for(name) {
Ok(count) => AgentResponse::PendingRemindersCount { count },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
AgentRequest::ReminderRollup {
since_secs,
agent: target,
} => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
AgentRequest::Start { name } => handle_start_child(coord, agent, name).await,
AgentRequest::Restart { name } => handle_restart_child(coord, agent, name),
AgentRequest::Kill { name } => handle_kill_child(coord, agent, name).await,
AgentRequest::Update { name } => handle_update_child(coord, agent, name),
AgentRequest::ListDescendants => handle_list_descendants(agent).await,
} => {
let name = resolve_agent_state_target(agent, target.as_deref());
match name {
Ok(name) => match coord.broker.reminder_rollup_for(name, *since_secs) {
Ok(stats) => AgentResponse::ReminderRollup(stats),
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
AgentRequest::Start { name } => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot start `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: start child");
match crate::lifecycle::start(name).await {
Ok(()) => {
coord.kick_agent(name, "container started");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
AgentRequest::Restart { name } => {
// Topology check: the caller must be the direct parent of the
// target. This is the only authorisation criterion — no
// capability flag needed; parenthood is sufficient privilege.
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot restart `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: enqueue restart for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Restart,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` restart tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
AgentRequest::Kill { name } => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot kill `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: kill child");
let result: anyhow::Result<()> = async {
crate::lifecycle::kill(name).await?;
coord.unregister_agent(name);
Ok(())
}
.await;
match result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.clone(),
});
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
AgentRequest::Update { name } => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot rebuild `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: enqueue rebuild for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Rebuild,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` update tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
AgentRequest::ListDescendants => {
tracing::debug!(%agent, "agent: list descendants");
// All containers known to nixos-container (running only).
let running_set: std::collections::HashSet<String> =
match crate::lifecycle::list().await {
Ok(names) => names
.into_iter()
.filter_map(|c| {
c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
.map(str::to_owned)
})
.collect(),
Err(e) => {
return AgentResponse::Err {
message: format!("list containers failed: {e:#}"),
};
}
};
// Walk the full topology and collect every descendant.
let topo = crate::topology::read();
let mut names: Vec<String> = topo
.keys()
.filter(|name| crate::topology::is_descendant_of(name, agent))
.cloned()
.collect();
// Parents before children, then alpha within each tier.
crate::auto_update::topology_sort(&mut names, &topo);
let containers = names
.into_iter()
.map(|name| {
let running = running_set.contains(&name);
hive_sh4re::ContainerInfo { name, running }
})
.collect();
AgentResponse::Containers { containers }
}
AgentRequest::RequestInitConfig { name, description } => {
handle_request_init_config(coord, agent, name, description.clone())
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot request_init_config for `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: request_init_config for child");
match crate::manager_server::submit_init_config(coord, name, description.clone()) {
Ok(_id) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
AgentRequest::RequestApplyCommit {
agent: target_agent,
commit_ref,
description,
} => {
handle_request_apply_commit(
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == target_agent)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot request_apply_commit for `{target_agent}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %target_agent, %commit_ref, "agent: request_apply_commit for child");
match crate::manager_server::submit_apply_commit(
coord,
agent,
target_agent,
commit_ref,
description.as_deref(),
)
.await
{
Ok((id, sha)) => {
tracing::info!(%id, %target_agent, %sha, "agent: apply_commit approval queued");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
// Manager-only variants are not valid on the agent socket.
_ => AgentResponse::Err {
@ -419,257 +546,6 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
}
}
/// Topology guard for the agent-socket lifecycle/config tools: the
/// caller must be the direct parent of `target`. Returns `Some(Err)`
/// to short-circuit the dispatch arm when it isn't, `None` when the
/// call is authorised. `action` is the verb phrase for the message
/// (e.g. `"start"`, `"request_apply_commit for"`).
fn require_child(agent: &str, target: &str, action: &str) -> Option<AgentResponse> {
if crate::topology::children_of(agent)
.iter()
.any(|c| c == target)
{
None
} else {
Some(AgentResponse::Err {
message: format!(
"agent `{agent}` cannot {action} `{target}`: \
not a direct child in the topology tree"
),
})
}
}
/// `GetLooseEnds` — resolve the (optionally cross-agent) target then
/// read its loose ends.
fn handle_get_loose_ends(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
) -> AgentResponse {
match resolve_agent_state_target(agent, target) {
Ok(name) => match crate::loose_ends::for_agent(coord, name) {
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
/// `CountPendingReminders` — resolve the target then count its pending
/// reminders.
fn handle_count_pending_reminders(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
) -> AgentResponse {
match resolve_agent_state_target(agent, target) {
Ok(name) => match coord.broker.count_pending_reminders_for(name) {
Ok(count) => AgentResponse::PendingRemindersCount { count },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
/// `ReminderRollup` — resolve the target then roll up its reminders
/// fired in the last `since_secs`.
fn handle_reminder_rollup(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
since_secs: u64,
) -> AgentResponse {
match resolve_agent_state_target(agent, target) {
Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) {
Ok(stats) => AgentResponse::ReminderRollup(stats),
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
/// `Start` — start a direct-child container, kicking its next turn.
async fn handle_start_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "start") {
return err;
}
tracing::info!(%agent, %name, "agent: start child");
match crate::lifecycle::start(name).await {
Ok(()) => {
coord.kick_agent(name, "container started");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `Restart` — enqueue a restart for a direct-child container.
/// Topology parenthood is the only authorisation criterion — no
/// capability flag needed.
fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "restart") {
return err;
}
tracing::info!(%agent, %name, "agent: enqueue restart for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Restart,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` restart tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
/// `Kill` — kill a direct-child container, unregister it, notify the
/// manager.
async fn handle_kill_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "kill") {
return err;
}
tracing::info!(%agent, %name, "agent: kill child");
let result: anyhow::Result<()> = async {
crate::lifecycle::kill(name).await?;
coord.unregister_agent(name);
Ok(())
}
.await;
match result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.to_owned(),
});
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `Update` — enqueue a rebuild for a direct-child container.
fn handle_update_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "rebuild") {
return err;
}
tracing::info!(%agent, %name, "agent: enqueue rebuild for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Rebuild,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` update tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
/// `ListDescendants` — every topological descendant of `agent` with
/// its running/stopped state, parents before children.
async fn handle_list_descendants(agent: &str) -> AgentResponse {
tracing::debug!(%agent, "agent: list descendants");
// All containers known to nixos-container (running only).
let running_set: std::collections::HashSet<String> = match crate::lifecycle::list().await {
Ok(names) => names
.into_iter()
.filter_map(|c| {
c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
.map(str::to_owned)
})
.collect(),
Err(e) => {
return AgentResponse::Err {
message: format!("list containers failed: {e:#}"),
};
}
};
// Walk the full topology and collect every descendant.
let topo = crate::topology::read();
let mut names: Vec<String> = topo
.keys()
.filter(|name| crate::topology::is_descendant_of(name, agent))
.cloned()
.collect();
// Parents before children, then alpha within each tier.
crate::auto_update::topology_sort(&mut names, &topo);
let containers = names
.into_iter()
.map(|name| {
let running = running_set.contains(&name);
hive_sh4re::ContainerInfo { name, running }
})
.collect();
AgentResponse::Containers { containers }
}
/// `RequestInitConfig` — queue an `InitConfig` approval for a
/// direct-child agent.
fn handle_request_init_config(
coord: &Arc<Coordinator>,
agent: &str,
name: &str,
description: Option<String>,
) -> AgentResponse {
if let Some(err) = require_child(agent, name, "request_init_config for") {
return err;
}
tracing::info!(%agent, %name, "agent: request_init_config for child");
match crate::manager_server::submit_init_config(coord, name, description) {
Ok(_id) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `RequestApplyCommit` — queue an apply-commit approval for a
/// direct-child agent.
async fn handle_request_apply_commit(
coord: &Arc<Coordinator>,
agent: &str,
target_agent: &str,
commit_ref: &str,
description: Option<&str>,
) -> AgentResponse {
if let Some(err) = require_child(agent, target_agent, "request_apply_commit for") {
return err;
}
tracing::info!(%agent, %target_agent, %commit_ref, "agent: request_apply_commit for child");
match crate::manager_server::submit_apply_commit(coord, target_agent, commit_ref, description)
.await
{
Ok((id, sha)) => {
tracing::info!(%id, %target_agent, %sha, "agent: apply_commit approval queued");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// Field-named journal-query knobs for [`dispatch_host_journal`].
/// Borrows straight from the matched `GetHostJournal` request variant.
pub struct HostJournalArgs<'a> {
pub unit: &'a Option<String>,
pub container: &'a Option<String>,
pub lines: &'a Option<u32>,
pub priority: &'a Option<hive_sh4re::JournalPriority>,
pub grep: &'a Option<String>,
pub since: &'a Option<String>,
pub until: &'a Option<String>,
}
/// Handle `GetHostJournal` from both the agent and manager sockets.
/// Capability-gated: the calling agent must hold `read_host_journal` in
/// `meta/capabilities.json`. Runs `journalctl` host-side and returns
@ -677,16 +553,17 @@ pub struct HostJournalArgs<'a> {
///
/// The manager is not exempt - grant `read_host_journal` in
/// `meta/capabilities.json` to enable it for any agent including the manager.
pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> AgentResponse {
let HostJournalArgs {
unit,
container,
lines,
priority,
grep,
since,
until,
} = args;
#[allow(clippy::too_many_arguments)]
pub async fn dispatch_host_journal(
agent: &str,
unit: &Option<String>,
container: &Option<String>,
lines: &Option<u32>,
priority: &Option<hive_sh4re::JournalPriority>,
grep: &Option<String>,
since: &Option<String>,
until: &Option<String>,
) -> AgentResponse {
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::ReadHostJournal) {
return AgentResponse::Err {
message: "agent does not have the read_host_journal capability".to_owned(),

View file

@ -349,33 +349,6 @@ impl TransientKind {
}
}
/// Field-named payload for [`Coordinator::emit_approval_resolved`].
/// Mirrors the `ApprovalResolved` dashboard-event fields. `agent`
/// borrows from the caller; `approval_kind` / `status` are
/// compile-time constants.
pub struct ApprovalResolved<'a> {
pub id: i64,
pub agent: &'a str,
pub approval_kind: &'static str,
pub sha_short: Option<String>,
pub status: &'static str,
pub note: Option<String>,
pub description: Option<String>,
}
/// Field-named payload for [`Coordinator::emit_question_added`].
/// Mirrors the `QuestionAdded` dashboard-event fields; all references
/// share the caller's lifetime.
pub struct QuestionAdded<'a> {
pub id: i64,
pub asker: &'a str,
pub question: &'a str,
pub options: &'a [String],
pub multi: bool,
pub deadline_at: Option<i64>,
pub target: Option<&'a str>,
}
impl Coordinator {
pub fn open(
db_path: &Path,
@ -679,19 +652,17 @@ impl Coordinator {
/// already have an authoritative timestamp from the db update,
/// the tiny skew between "row updated" and "event emitted" is
/// presentation-only and doesn't matter to clients.
///
/// Takes [`ApprovalResolved`] rather than a positional arg list so
/// the seven fields are named at every call site.
pub fn emit_approval_resolved(&self, ev: ApprovalResolved<'_>) {
let ApprovalResolved {
id,
agent,
approval_kind,
sha_short,
status,
note,
description,
} = ev;
#[allow(clippy::too_many_arguments)]
pub fn emit_approval_resolved(
&self,
id: i64,
agent: &str,
approval_kind: &'static str,
sha_short: Option<String>,
status: &'static str,
note: Option<String>,
description: Option<String>,
) {
let resolved_at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
@ -714,16 +685,17 @@ impl Coordinator {
/// both operator-targeted (`target = None`) and peer-to-peer
/// (`target = Some(agent)`) threads — the dashboard surfaces
/// both, distinguishing visually + offering operator override.
pub fn emit_question_added(&self, ev: &QuestionAdded<'_>) {
let &QuestionAdded {
id,
asker,
question,
options,
multi,
deadline_at,
target,
} = ev;
#[allow(clippy::too_many_arguments)]
pub fn emit_question_added(
&self,
id: i64,
asker: &str,
question: &str,
options: &[String],
multi: bool,
deadline_at: Option<i64>,
target: Option<&str>,
) {
let asked_at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()

View file

@ -82,15 +82,15 @@ pub(super) fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<A
.fetched_sha
.as_deref()
.map(|s| s[..s.len().min(12)].to_owned());
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id: a.id,
agent: &a.agent,
approval_kind: "apply_commit",
coord.emit_approval_resolved(
a.id,
&a.agent,
"apply_commit",
sha_short,
status: "failed",
note: Some(note.to_owned()),
description: a.description.clone(),
});
"failed",
Some(note.to_owned()),
a.description.clone(),
);
false
}
})

View file

@ -1128,11 +1128,7 @@ fn bind_child_agent_dirs(child: &str, binds: &mut Vec<BindMount>) {
});
}
#[allow(
clippy::too_many_lines,
reason = "one contiguous nspawn-flag assembly block; the length is the flag \
surface itself, splitting it would just hide the shape"
)]
#[allow(clippy::too_many_lines)]
async fn set_nspawn_flags(
container: &str,
runtime_dir: &Path,

View file

@ -74,6 +74,7 @@ async fn serve(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
}
}
#[allow(clippy::too_many_lines)]
async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResponse {
// Delegate all variants shared with the agent socket to the common handler.
if let Some(resp) = crate::agent_server::dispatch_shared(req, MANAGER_AGENT, coord).await {
@ -81,16 +82,112 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
}
match req {
ManagerRequest::RequestInitConfig { name, description } => {
handle_manager_init_config(coord, name, description.clone())
tracing::info!(%name, "manager: request_init_config");
match submit_init_config(coord, name, description.clone()) {
Ok(_id) => ManagerResponse::Ok,
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
ManagerRequest::Kill { name } => {
tracing::info!(%name, "manager: kill");
let result: Result<()> = async {
lifecycle::kill(name).await?;
coord.unregister_agent(name);
Ok(())
}
.await;
match result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.clone(),
});
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
ManagerRequest::Start { name } => {
tracing::info!(%name, "manager: start");
match lifecycle::start(name).await {
Ok(()) => {
coord.kick_agent(name, "container started");
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
ManagerRequest::Restart { name } => {
tracing::info!(%name, "manager: enqueue restart");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Restart,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
"manager `restart` tool".to_owned(),
None,
);
coord.emit_rebuild_queue_snapshot();
ManagerResponse::Ok
}
ManagerRequest::Update { name } => {
tracing::info!(%name, "manager: enqueue update");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Rebuild,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
"manager `update` tool".to_owned(),
None,
);
coord.emit_rebuild_queue_snapshot();
ManagerResponse::Ok
}
ManagerRequest::Kill { name } => handle_manager_kill(coord, name).await,
ManagerRequest::Start { name } => handle_manager_start(coord, name).await,
ManagerRequest::Restart { name } => handle_manager_restart(coord, name),
ManagerRequest::Update { name } => handle_manager_update(coord, name),
ManagerRequest::RequestUpdateMetaInputs {
inputs,
description,
} => handle_request_update_meta_inputs(coord, inputs, description.as_deref()),
} => {
let label = if inputs.is_empty() {
"all inputs".to_string()
} else {
inputs.join(", ")
};
tracing::info!(%label, "manager: request_update_meta_inputs");
// Encode the inputs list as JSON and store it in commit_ref
// (there's no git commit involved; the field carries the
// payload for the approval handler to decode at run time).
let commit_ref = serde_json::to_string(inputs).unwrap_or_default();
let id = match coord
.approvals
.submit_kind(
MANAGER_AGENT,
hive_sh4re::ApprovalKind::UpdateMetaInputs,
&commit_ref,
description.as_deref(),
)
.map_err(|e| anyhow::anyhow!("{e:#}"))
{
Ok(id) => id,
Err(e) => {
return ManagerResponse::Err {
message: format!("queue update_meta_inputs approval: {e:#}"),
};
}
};
tracing::info!(%id, %label, "update_meta_inputs approval queued");
coord.emit_approval_added(
id,
MANAGER_AGENT,
"update_meta_inputs",
None,
None,
description.clone(),
);
ManagerResponse::Ok
}
ManagerRequest::RequestSchedulePrompt(payload) => {
handle_request_schedule_prompt(coord, MANAGER_AGENT, payload)
}
@ -109,33 +206,108 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
coord,
MANAGER_AGENT,
*id,
EditSchedulePatch {
body: body.clone(),
description: description.clone(),
interval_seconds: *interval_seconds,
next_fire_at_unix: *next_fire_at_unix,
targets_add: targets_add.clone(),
targets_remove: targets_remove.clone(),
},
body.clone(),
description.clone(),
*interval_seconds,
*next_fire_at_unix,
targets_add.clone(),
targets_remove.clone(),
),
ManagerRequest::ListSchedules => handle_list_schedules(coord),
ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() {
Ok(schedules) => ManagerResponse::Schedules {
schedules: schedules.into_iter().map(schedule_to_wire).collect(),
},
Err(e) => ManagerResponse::Err {
message: format!("list scheduled prompts: {e:#}"),
},
},
ManagerRequest::FireScheduleNow { id } => {
handle_fire_schedule_now(coord, MANAGER_AGENT, *id).await
}
ManagerRequest::GetLogs { agent, lines } => handle_get_logs(agent, *lines).await,
ManagerRequest::GetLogs { agent, lines } => {
let n = lines.unwrap_or(50);
// `journalctl -M` wants the container name (`h-<name>`),
// not the logical agent name. `container_name` adds the prefix.
// The `-M` read needs root, so it goes through hive-priv.
let machine = crate::lifecycle::container_name(agent);
tracing::info!(%agent, %machine, %n, "manager: get_logs");
match crate::priv_client::read_container_journal(
&machine,
hive_sh4re::priv_proto::JournalQuery {
lines: n,
..Default::default()
},
)
.await
{
Ok((stdout, stderr)) => {
let content = if stdout.is_empty() { stderr } else { stdout };
ManagerResponse::Logs { content }
}
Err(e) => ManagerResponse::Err {
message: format!("get_logs: {e:#}"),
},
}
}
ManagerRequest::RequestApplyCommit {
agent,
commit_ref,
description,
} => handle_manager_apply_commit(coord, agent, commit_ref, description.as_deref()).await,
} => {
tracing::info!(%agent, %commit_ref, "manager: request_apply_commit");
match submit_apply_commit(coord, agent, commit_ref, description.as_deref()).await {
Ok((id, sha)) => {
tracing::info!(%id, %agent, manager_ref = %commit_ref, %sha, "approval queued + proposal tag planted");
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
ManagerRequest::GetLooseEnds { agent } => {
handle_manager_loose_ends(coord, agent.as_deref())
let result = match agent.as_deref() {
Some("*") => {
// Hive-wide query requires query_agent_state capability.
if !crate::capabilities::has_cap(
MANAGER_AGENT,
hive_sh4re::Capability::QueryAgentState,
) {
return ManagerResponse::Err {
message:
"query_agent_state capability required for hive-wide loose ends"
.into(),
};
}
crate::loose_ends::hive_wide(coord)
}
Some(name) => crate::loose_ends::for_agent(coord, name),
None => crate::loose_ends::for_agent(coord, MANAGER_AGENT),
};
match result {
Ok(loose_ends) => ManagerResponse::LooseEnds { loose_ends },
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
ManagerRequest::CountPendingReminders { agent } => {
handle_manager_count_pending_reminders(coord, agent.as_deref())
let target = agent.as_deref().unwrap_or(MANAGER_AGENT);
match coord.broker.count_pending_reminders_for(target) {
Ok(count) => ManagerResponse::PendingRemindersCount { count },
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
ManagerRequest::ReminderRollup { since_secs, agent } => {
handle_manager_reminder_rollup(coord, agent.as_deref(), *since_secs)
let target = agent.as_deref().unwrap_or(MANAGER_AGENT);
match coord.broker.reminder_rollup_for(target, *since_secs) {
Ok(stats) => ManagerResponse::ReminderRollup(stats),
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
_ => ManagerResponse::Err {
message: "request not handled on manager socket".to_owned(),
@ -143,245 +315,6 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
}
}
/// `RequestInitConfig` (manager socket) — queue an `InitConfig`
/// approval. No topology check: the manager can act on any agent.
fn handle_manager_init_config(
coord: &Arc<Coordinator>,
name: &str,
description: Option<String>,
) -> ManagerResponse {
tracing::info!(%name, "manager: request_init_config");
match submit_init_config(coord, name, description) {
Ok(_id) => ManagerResponse::Ok,
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `Kill` (manager socket) — kill the container, unregister it, notify.
async fn handle_manager_kill(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
tracing::info!(%name, "manager: kill");
let result: Result<()> = async {
lifecycle::kill(name).await?;
coord.unregister_agent(name);
Ok(())
}
.await;
match result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.to_owned(),
});
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `Start` (manager socket) — start the container, kick its next turn.
async fn handle_manager_start(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
tracing::info!(%name, "manager: start");
match lifecycle::start(name).await {
Ok(()) => {
coord.kick_agent(name, "container started");
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `Restart` (manager socket) — enqueue a restart.
fn handle_manager_restart(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
tracing::info!(%name, "manager: enqueue restart");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Restart,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
"manager `restart` tool".to_owned(),
None,
);
coord.emit_rebuild_queue_snapshot();
ManagerResponse::Ok
}
/// `Update` (manager socket) — enqueue a rebuild.
fn handle_manager_update(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
tracing::info!(%name, "manager: enqueue update");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Rebuild,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
"manager `update` tool".to_owned(),
None,
);
coord.emit_rebuild_queue_snapshot();
ManagerResponse::Ok
}
/// `RequestUpdateMetaInputs` — queue an `UpdateMetaInputs` approval
/// carrying the JSON-encoded input list in `commit_ref` (no git commit
/// is involved; the field is the payload the approval handler decodes).
fn handle_request_update_meta_inputs(
coord: &Arc<Coordinator>,
inputs: &[String],
description: Option<&str>,
) -> ManagerResponse {
let label = if inputs.is_empty() {
"all inputs".to_string()
} else {
inputs.join(", ")
};
tracing::info!(%label, "manager: request_update_meta_inputs");
let commit_ref = serde_json::to_string(inputs).unwrap_or_default();
let id = match coord
.approvals
.submit_kind(
MANAGER_AGENT,
hive_sh4re::ApprovalKind::UpdateMetaInputs,
&commit_ref,
description,
)
.map_err(|e| anyhow::anyhow!("{e:#}"))
{
Ok(id) => id,
Err(e) => {
return ManagerResponse::Err {
message: format!("queue update_meta_inputs approval: {e:#}"),
};
}
};
tracing::info!(%id, %label, "update_meta_inputs approval queued");
coord.emit_approval_added(
id,
MANAGER_AGENT,
"update_meta_inputs",
None,
None,
description.map(str::to_owned),
);
ManagerResponse::Ok
}
/// `ListSchedules` — snapshot every scheduled prompt onto the wire.
fn handle_list_schedules(coord: &Arc<Coordinator>) -> ManagerResponse {
match coord.scheduled_prompts.list() {
Ok(schedules) => ManagerResponse::Schedules {
schedules: schedules.into_iter().map(schedule_to_wire).collect(),
},
Err(e) => ManagerResponse::Err {
message: format!("list scheduled prompts: {e:#}"),
},
}
}
/// `GetLogs` — read a child container's journal via hive-priv (the
/// `-M` read needs root). `journalctl -M` wants the `h-<name>` machine
/// name, which `container_name` derives.
async fn handle_get_logs(agent: &str, lines: Option<u32>) -> ManagerResponse {
let n = lines.unwrap_or(50);
let machine = crate::lifecycle::container_name(agent);
tracing::info!(%agent, %machine, %n, "manager: get_logs");
match crate::priv_client::read_container_journal(
&machine,
hive_sh4re::priv_proto::JournalQuery {
lines: n,
..Default::default()
},
)
.await
{
Ok((stdout, stderr)) => {
let content = if stdout.is_empty() { stderr } else { stdout };
ManagerResponse::Logs { content }
}
Err(e) => ManagerResponse::Err {
message: format!("get_logs: {e:#}"),
},
}
}
/// `RequestApplyCommit` (manager socket) — queue an apply-commit
/// approval + plant the `proposal/<id>` tag.
async fn handle_manager_apply_commit(
coord: &Arc<Coordinator>,
agent: &str,
commit_ref: &str,
description: Option<&str>,
) -> ManagerResponse {
tracing::info!(%agent, %commit_ref, "manager: request_apply_commit");
match submit_apply_commit(coord, agent, commit_ref, description).await {
Ok((id, sha)) => {
tracing::info!(%id, %agent, manager_ref = %commit_ref, %sha, "approval queued + proposal tag planted");
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `GetLooseEnds` (manager socket) — manager's own threads by default,
/// a named agent's when given, or hive-wide for `"*"` (which requires
/// the `query_agent_state` capability).
fn handle_manager_loose_ends(coord: &Arc<Coordinator>, agent: Option<&str>) -> ManagerResponse {
let result = match agent {
Some("*") => {
if !crate::capabilities::has_cap(MANAGER_AGENT, hive_sh4re::Capability::QueryAgentState)
{
return ManagerResponse::Err {
message: "query_agent_state capability required for hive-wide loose ends"
.into(),
};
}
crate::loose_ends::hive_wide(coord)
}
Some(name) => crate::loose_ends::for_agent(coord, name),
None => crate::loose_ends::for_agent(coord, MANAGER_AGENT),
};
match result {
Ok(loose_ends) => ManagerResponse::LooseEnds { loose_ends },
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `CountPendingReminders` (manager socket) — count pending reminders
/// for the target (defaults to the manager itself).
fn handle_manager_count_pending_reminders(
coord: &Arc<Coordinator>,
agent: Option<&str>,
) -> ManagerResponse {
let target = agent.unwrap_or(MANAGER_AGENT);
match coord.broker.count_pending_reminders_for(target) {
Ok(count) => ManagerResponse::PendingRemindersCount { count },
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `ReminderRollup` (manager socket) — roll up reminders fired in the
/// last `since_secs` for the target (defaults to the manager itself).
fn handle_manager_reminder_rollup(
coord: &Arc<Coordinator>,
agent: Option<&str>,
since_secs: u64,
) -> ManagerResponse {
let target = agent.unwrap_or(MANAGER_AGENT);
match coord.broker.reminder_rollup_for(target, since_secs) {
Ok(stats) => ManagerResponse::ReminderRollup(stats),
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `request_apply_commit` takes a commit SHA only — not a branch or
/// tag name. A branch is mutable; pinning the proposal to a concrete
/// sha keeps "what the manager asked to deploy" unambiguous and means
@ -488,15 +421,15 @@ pub(crate) async fn submit_apply_commit(
// explanation of why the approval can't be approved.
let note = format!("{e:#}");
let _ = coord.approvals.mark_failed(id, &note);
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
coord.emit_approval_resolved(
id,
agent,
approval_kind: "apply_commit",
sha_short: None,
status: "failed",
note: Some(note),
description: description.map(str::to_owned),
});
"apply_commit",
None,
"failed",
Some(note),
description.map(str::to_owned),
);
return Err(anyhow::anyhow!("git_fetch_to_tag: {e:#}"));
}
};
@ -518,29 +451,29 @@ pub(crate) async fn submit_apply_commit(
if let Err(e) = crate::flake_check::check_lock_in_sync(&applied_dir, &tag, id).await {
let note = format!("{e:#}");
let _ = coord.approvals.mark_failed(id, &note);
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
coord.emit_approval_resolved(
id,
agent,
approval_kind: "apply_commit",
sha_short: Some(sha_short.clone()),
status: "failed",
note: Some(note),
description: description.map(str::to_owned),
});
"apply_commit",
Some(sha_short.clone()),
"failed",
Some(note),
description.map(str::to_owned),
);
return Err(anyhow::anyhow!("flake lock-sync check: {e:#}"));
}
if let Err(e) = crate::flake_check::check_no_duplicate_inputs(&applied_dir, &tag).await {
let note = format!("{e:#}");
let _ = coord.approvals.mark_failed(id, &note);
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
coord.emit_approval_resolved(
id,
agent,
approval_kind: "apply_commit",
sha_short: Some(sha_short.clone()),
status: "failed",
note: Some(note),
description: description.map(str::to_owned),
});
"apply_commit",
Some(sha_short.clone()),
"failed",
Some(note),
description.map(str::to_owned),
);
return Err(anyhow::anyhow!("flake dedup check: {e:#}"));
}
// Mirror the freshly-planted proposal/<id> tag to the forge.
@ -723,24 +656,6 @@ async fn handle_fire_schedule_now(
}
}
/// Field-named PATCH payload for [`handle_edit_schedule`]. Every
/// field is "leave alone" when `None`; the double-`Option` fields
/// additionally distinguish clear (`Some(None)`) from set
/// (`Some(Some(v))`).
#[allow(
clippy::option_option,
reason = "double-Option carries three-state PATCH semantics: outer None = \
leave alone, Some(None) = clear, Some(Some(v)) = set"
)]
struct EditSchedulePatch {
body: Option<String>,
description: Option<Option<String>>,
interval_seconds: Option<Option<u64>>,
next_fire_at_unix: Option<i64>,
targets_add: Option<Vec<String>>,
targets_remove: Option<Vec<String>>,
}
/// Authorize + dispatch a `EditSchedule` patch. Same ownership
/// rules as `CancelSchedule` — the manager can edit
/// schedules it owns + any owned by an agent in its subtree.
@ -749,20 +664,23 @@ struct EditSchedulePatch {
/// zero-interval validation. Returns `Ok` on a clean update;
/// `Err` with the underlying message on any auth / validation
/// failure so the dashboard can surface it verbatim.
#[allow(clippy::too_many_arguments)]
#[allow(
clippy::option_option,
reason = "double-Option carries three-state PATCH semantics: outer None = \
leave alone, Some(None) = clear, Some(Some(v)) = set"
)]
fn handle_edit_schedule(
coord: &Arc<Coordinator>,
requester: &str,
schedule_id: i64,
patch: EditSchedulePatch,
body: Option<String>,
description: Option<Option<String>>,
interval_seconds: Option<Option<u64>>,
next_fire_at_unix: Option<i64>,
targets_add: Option<Vec<String>>,
targets_remove: Option<Vec<String>>,
) -> ManagerResponse {
let EditSchedulePatch {
body,
description,
interval_seconds,
next_fire_at_unix,
targets_add,
targets_remove,
} = patch;
let schedule = match coord.scheduled_prompts.get(schedule_id) {
Ok(Some(s)) => s,
Ok(None) => {

View file

@ -90,15 +90,7 @@ pub fn handle_ask(
}
// Always fire on the dashboard channel — both operator-targeted
// and peer threads now surface in the dashboard's questions pane.
coord.emit_question_added(&crate::coordinator::QuestionAdded {
id,
asker,
question,
options,
multi,
deadline_at,
target,
});
coord.emit_question_added(id, asker, question, options, multi, deadline_at, target);
if let Some(t) = ttl {
spawn_question_watchdog(coord, id, t);
}
@ -203,15 +195,15 @@ pub fn handle_cancel_loose_end(
.fetched_sha
.as_deref()
.map(|s| s[..s.len().min(12)].to_owned());
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id: approval.id,
agent: &approval.agent,
approval_kind: kind_to_str(approval.kind),
coord.emit_approval_resolved(
approval.id,
&approval.agent,
kind_to_str(approval.kind),
sha_short,
status: "cancelled",
note: approval.note,
description: approval.description,
});
"cancelled",
approval.note,
approval.description,
);
Ok(())
}
}

View file

@ -259,22 +259,6 @@ impl Default for RebuildQueue {
}
}
/// Full-shape submit spec for [`RebuildQueue::enqueue_full`] — every
/// `QueueEntry` field settable at submit time. The thinner `enqueue`
/// / `enqueue_with_inputs` / `enqueue_with_perm` wrappers build this
/// for the common cases.
pub struct FullEnqueue {
pub kind: QueueKind,
pub agent: String,
pub source: QueueSource,
pub reason: String,
pub parent_id: Option<u64>,
pub inputs: Vec<String>,
pub approval_id: Option<i64>,
pub perm_payload: Option<PermPayload>,
pub depends_on: Vec<u64>,
}
impl RebuildQueue {
pub fn new() -> Self {
Self::default()
@ -310,17 +294,17 @@ impl RebuildQueue {
reason: String,
parent_id: Option<u64>,
) -> u64 {
self.enqueue_full(FullEnqueue {
self.enqueue_full(
kind,
agent,
source,
reason,
parent_id,
inputs: Vec::new(),
approval_id: None,
perm_payload: None,
depends_on: Vec::new(),
})
Vec::new(),
None,
None,
Vec::new(),
)
}
/// Same as `enqueue` but carries an `inputs` payload — used by
@ -337,17 +321,17 @@ impl RebuildQueue {
parent_id: Option<u64>,
inputs: Vec<String>,
) -> u64 {
self.enqueue_full(FullEnqueue {
self.enqueue_full(
kind,
agent,
source,
reason,
parent_id,
inputs,
approval_id: None,
perm_payload: None,
depends_on: Vec::new(),
})
None,
None,
Vec::new(),
)
}
/// Enqueue a `PermChange` entry for `agent`. The worker applies the
@ -360,17 +344,17 @@ impl RebuildQueue {
reason: String,
payload: PermPayload,
) -> u64 {
self.enqueue_full(FullEnqueue {
kind: QueueKind::PermChange,
self.enqueue_full(
QueueKind::PermChange,
agent,
source,
reason,
parent_id: None,
inputs: Vec::new(),
approval_id: None,
perm_payload: Some(payload),
depends_on: Vec::new(),
})
None,
Vec::new(),
None,
Some(payload),
Vec::new(),
)
}
/// Full-shape enqueue — every `QueueEntry` field that's settable
@ -378,18 +362,23 @@ impl RebuildQueue {
/// `enqueue_with_perm` delegate to this; the approval-driven POST
/// handlers call it directly with the source row's id so the
/// worker can re-fetch the kind-specific payload.
pub fn enqueue_full(&self, spec: FullEnqueue) -> u64 {
let FullEnqueue {
kind,
agent,
source,
reason,
parent_id,
inputs,
approval_id,
perm_payload,
depends_on,
} = spec;
// 10 args: the queue entry has 6 independent submit-time fields plus
// four kind-specific payload fields (inputs, approval_id, perm_payload,
// depends_on). A builder struct would obscure the call sites; the
// shorter wrappers already cover all common cases.
#[allow(clippy::too_many_arguments)]
pub fn enqueue_full(
&self,
kind: QueueKind,
agent: String,
source: QueueSource,
reason: String,
parent_id: Option<u64>,
inputs: Vec<String>,
approval_id: Option<i64>,
perm_payload: Option<PermPayload>,
depends_on: Vec<u64>,
) -> u64 {
let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
// Dedup against a pending entry with the same (kind, agent) —
// and, for MetaUpdate, the same `inputs` list (see method
@ -1220,17 +1209,17 @@ mod tests {
#[test]
fn approval_entries_keep_approval_id() {
let q = RebuildQueue::new();
let id = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "agent-a".to_owned(),
source: QueueSource::Approval,
reason: "approval #42 apply commit".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(42),
perm_payload: None,
depends_on: Vec::new(),
});
let id = q.enqueue_full(
QueueKind::Rebuild,
"agent-a".to_owned(),
QueueSource::Approval,
"approval #42 apply commit".to_owned(),
None,
Vec::new(),
Some(42),
None,
Vec::new(),
);
let snap = q.snapshot();
let entry = snap.iter().find(|e| e.id == id).expect("entry present");
assert_eq!(entry.approval_id, Some(42));
@ -1244,43 +1233,43 @@ mod tests {
// approve click is a separate piece of work even when the
// (kind, agent) pair matches.
let q = RebuildQueue::new();
let a = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "agent-a".to_owned(),
source: QueueSource::Approval,
reason: "approval #1".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(1),
perm_payload: None,
depends_on: Vec::new(),
});
let b = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "agent-a".to_owned(),
source: QueueSource::Approval,
reason: "approval #2".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(2),
perm_payload: None,
depends_on: Vec::new(),
});
let a = q.enqueue_full(
QueueKind::Rebuild,
"agent-a".to_owned(),
QueueSource::Approval,
"approval #1".to_owned(),
None,
Vec::new(),
Some(1),
None,
Vec::new(),
);
let b = q.enqueue_full(
QueueKind::Rebuild,
"agent-a".to_owned(),
QueueSource::Approval,
"approval #2".to_owned(),
None,
Vec::new(),
Some(2),
None,
Vec::new(),
);
assert_ne!(a, b);
assert_eq!(q.snapshot().len(), 2);
// Same approval_id submitted twice DOES dedup (rapid double-
// click on the dashboard's approve button is a single op).
let c = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "agent-a".to_owned(),
source: QueueSource::Approval,
reason: "approval #1 (duplicate)".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: Some(1),
perm_payload: None,
depends_on: Vec::new(),
});
let c = q.enqueue_full(
QueueKind::Rebuild,
"agent-a".to_owned(),
QueueSource::Approval,
"approval #1 (duplicate)".to_owned(),
None,
Vec::new(),
Some(1),
None,
Vec::new(),
);
assert_eq!(a, c);
assert_eq!(q.snapshot().len(), 2);
}
@ -1466,17 +1455,17 @@ mod tests {
"first".to_owned(),
None,
);
let b = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "agent-b".to_owned(),
source: QueueSource::Manual,
reason: "second (blocked on a)".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: None,
perm_payload: None,
depends_on: vec![a],
});
let b = q.enqueue_full(
QueueKind::Rebuild,
"agent-b".to_owned(),
QueueSource::Manual,
"second (blocked on a)".to_owned(),
None,
Vec::new(),
None,
None,
vec![a],
);
// B depends on A — take_next should give A first.
let first = q.take_next().expect("a is ready");
assert_eq!(first.id, a);
@ -1536,17 +1525,17 @@ mod tests {
"dep must be evicted from history"
);
// An entry that depends on the (evicted) dep must be immediately runnable.
let downstream = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "downstream".to_owned(),
source: QueueSource::Manual,
reason: "downstream (dep evicted = resolved)".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: None,
perm_payload: None,
depends_on: vec![dep],
});
let downstream = q.enqueue_full(
QueueKind::Rebuild,
"downstream".to_owned(),
QueueSource::Manual,
"downstream (dep evicted = resolved)".to_owned(),
None,
Vec::new(),
None,
None,
vec![dep],
);
let got = q.take_next().expect("downstream runnable when dep evicted");
assert_eq!(got.id, downstream);
}
@ -1570,42 +1559,42 @@ mod tests {
"d2".to_owned(),
None,
);
let a = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "target".to_owned(),
source: QueueSource::Manual,
reason: "r".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: None,
perm_payload: None,
depends_on: vec![dep1],
});
let a = q.enqueue_full(
QueueKind::Rebuild,
"target".to_owned(),
QueueSource::Manual,
"r".to_owned(),
None,
Vec::new(),
None,
None,
vec![dep1],
);
// Same kind+agent but different depends_on — must NOT dedup.
let b = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "target".to_owned(),
source: QueueSource::Manual,
reason: "r".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: None,
perm_payload: None,
depends_on: vec![dep2],
});
let b = q.enqueue_full(
QueueKind::Rebuild,
"target".to_owned(),
QueueSource::Manual,
"r".to_owned(),
None,
Vec::new(),
None,
None,
vec![dep2],
);
assert_ne!(a, b, "different depends_on must produce distinct entries");
// Same depends_on as a — must dedup.
let c = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "target".to_owned(),
source: QueueSource::Manual,
reason: "r again".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: None,
perm_payload: None,
depends_on: vec![dep1],
});
let c = q.enqueue_full(
QueueKind::Rebuild,
"target".to_owned(),
QueueSource::Manual,
"r again".to_owned(),
None,
Vec::new(),
None,
None,
vec![dep1],
);
assert_eq!(a, c, "identical depends_on must dedup");
}
@ -1622,17 +1611,17 @@ mod tests {
"a".to_owned(),
None,
);
let b = q.enqueue_full(FullEnqueue {
kind: QueueKind::Rebuild,
agent: "b".to_owned(),
source: QueueSource::Manual,
reason: "b (blocked on a)".to_owned(),
parent_id: None,
inputs: Vec::new(),
approval_id: None,
perm_payload: None,
depends_on: vec![a],
});
let b = q.enqueue_full(
QueueKind::Rebuild,
"b".to_owned(),
QueueSource::Manual,
"b (blocked on a)".to_owned(),
None,
Vec::new(),
None,
None,
vec![a],
);
q.take_next(); // pop a, mark Running
q.finish(a, QueueState::Failed, Some("nix build exploded".to_owned()));
let got = q.take_next().expect("b runnable after a failed");

View file

@ -73,10 +73,38 @@ async fn handle(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
}
}
#[allow(clippy::too_many_lines)]
async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
let result: anyhow::Result<HostResponse> = async {
Ok(match req {
HostRequest::Spawn { name } => handle_spawn(&coord, name).await?,
HostRequest::Spawn { name } => {
tracing::info!(%name, "spawn");
let agent_dir = coord.ensure_runtime(name)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
match lifecycle::spawn(name, &hive, &paths).await {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
agent: name.clone(),
ok: true,
note: None,
sha: None,
});
}
Err(e) => {
// Roll back socket registration if container creation failed.
coord.unregister_agent(name);
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
agent: name.clone(),
ok: false,
note: Some(format!("{e:#}")),
sha: None,
});
return Err(e);
}
}
HostResponse::success()
}
HostRequest::RequestSpawn { name } => {
tracing::info!(%name, "request_spawn");
let id =
@ -86,18 +114,86 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
tracing::info!(%id, %name, "spawn approval queued");
HostResponse::success()
}
HostRequest::Kill { name } => handle_kill(&coord, name).await?,
HostRequest::Kill { name } => {
tracing::info!(%name, "kill");
lifecycle::kill(name).await?;
coord.unregister_agent(name);
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.clone(),
});
HostResponse::success()
}
HostRequest::Restart { name } => {
tracing::info!(%name, "restart");
lifecycle::restart(name).await?;
HostResponse::success()
}
HostRequest::RestartAll => handle_restart_all().await?,
HostRequest::RestartAll => {
tracing::info!("restart-all");
let agents = lifecycle::list().await?;
let mut ok_agents: Vec<String> = Vec::new();
let mut errors: Vec<String> = Vec::new();
for agent in &agents {
if let Err(e) = lifecycle::restart(agent).await {
tracing::warn!(%agent, error = ?e, "restart-all: failed to restart agent");
errors.push(format!("{agent}: {e:#}"));
} else {
ok_agents.push(agent.clone());
}
}
if errors.is_empty() {
HostResponse::list(ok_agents)
} else {
HostResponse {
ok: false,
error: Some(errors.join("; ")),
agents: Some(ok_agents),
approvals: None,
}
}
}
HostRequest::Destroy { name, purge } => {
actions::destroy(&coord, name, *purge).await?;
HostResponse::success()
}
HostRequest::Rebuild { name } => handle_rebuild(&coord, name).await?,
HostRequest::Rebuild { name } => {
tracing::info!(%name, "rebuild");
let agent_dir = coord.ensure_runtime(name)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result = lifecycle::rebuild(name, &hive, &paths, &|_| (), &|_| ()).await;
// Mirror auto_update::rebuild_agent — the manager wants
// to know about every rebuild attempt regardless of
// which surface triggered it, especially failures
// (build error → manager can adjust the agent's
// agent.nix). Without this the admin-socket CLI was
// a notify-gap.
match &result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent: name.clone(),
ok: true,
note: None,
sha: None,
tag: None,
});
// Wake the agent's next turn with the
// "you were rebuilt" hint. Same pattern as
// auto_update::rebuild_agent and the dashboard
// rebuild path — this is the CLI's equivalent.
coord.kick_agent(name, "container rebuilt");
}
Err(e) => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent: name.clone(),
ok: false,
note: Some(format!("{e:#}")),
sha: None,
tag: None,
}),
}
result?;
HostResponse::success()
}
HostRequest::List => HostResponse::list(lifecycle::list().await?),
HostRequest::Pending => HostResponse::pending(coord.approvals.pending()?),
HostRequest::Approve { id } => {
@ -129,111 +225,3 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
Err(e) => HostResponse::error(format!("{e:#}")),
}
}
/// Create + start the container for `name`, rolling back socket
/// registration and notifying the manager on failure.
async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
tracing::info!(%name, "spawn");
let agent_dir = coord.ensure_runtime(name)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
match lifecycle::spawn(name, &hive, &paths).await {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
agent: name.to_owned(),
ok: true,
note: None,
sha: None,
});
}
Err(e) => {
// Roll back socket registration if container creation failed.
coord.unregister_agent(name);
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
agent: name.to_owned(),
ok: false,
note: Some(format!("{e:#}")),
sha: None,
});
return Err(e);
}
}
Ok(HostResponse::success())
}
/// Kill `name`'s container, unregister its socket, notify the manager.
async fn handle_kill(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
tracing::info!(%name, "kill");
lifecycle::kill(name).await?;
coord.unregister_agent(name);
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.to_owned(),
});
Ok(HostResponse::success())
}
/// Restart every container, aggregating per-agent failures into one
/// response rather than aborting on the first error.
async fn handle_restart_all() -> Result<HostResponse> {
tracing::info!("restart-all");
let agents = lifecycle::list().await?;
let mut ok_agents: Vec<String> = Vec::new();
let mut errors: Vec<String> = Vec::new();
for agent in &agents {
if let Err(e) = lifecycle::restart(agent).await {
tracing::warn!(%agent, error = ?e, "restart-all: failed to restart agent");
errors.push(format!("{agent}: {e:#}"));
} else {
ok_agents.push(agent.clone());
}
}
if errors.is_empty() {
Ok(HostResponse::list(ok_agents))
} else {
Ok(HostResponse {
ok: false,
error: Some(errors.join("; ")),
agents: Some(ok_agents),
approvals: None,
})
}
}
/// Rebuild `name`'s container, notifying the manager of the outcome
/// (success or failure) and kicking the agent's next turn on success.
async fn handle_rebuild(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
tracing::info!(%name, "rebuild");
let agent_dir = coord.ensure_runtime(name)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result = lifecycle::rebuild(name, &hive, &paths, &|_| (), &|_| ()).await;
// Mirror auto_update::rebuild_agent — the manager wants to know
// about every rebuild attempt regardless of which surface triggered
// it, especially failures (build error → manager can adjust the
// agent's agent.nix). Without this the admin-socket CLI was a
// notify-gap.
match &result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent: name.to_owned(),
ok: true,
note: None,
sha: None,
tag: None,
});
// Wake the agent's next turn with the "you were rebuilt"
// hint. Same pattern as auto_update::rebuild_agent and the
// dashboard rebuild path — this is the CLI's equivalent.
coord.kick_agent(name, "container rebuilt");
}
Err(e) => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent: name.to_owned(),
ok: false,
note: Some(format!("{e:#}")),
sha: None,
tag: None,
}),
}
result?;
Ok(HostResponse::success())
}

View file

@ -62,11 +62,7 @@ pub fn run(client: &Client, args: Args) -> Result<()> {
/// output for every supported event type without re-implementing the
/// per-arm dispatch. `print_event` is the only caller that adds the
/// terminating newline.
#[allow(
clippy::too_many_lines,
reason = "flat per-event-type dispatch: one arm per timeline event kind; \
splitting it would scatter the formatting without shrinking it"
)]
#[allow(clippy::too_many_lines)]
fn format_event(ev: &Value) -> String {
let event_type = ev.get("type").and_then(Value::as_str).unwrap_or("?");
let user = ev

View file

@ -147,11 +147,7 @@ struct InviteUserArgs {
}
struct MatrixBridge {
#[allow(
dead_code,
reason = "populated by the #[tool_router] macro; the generated \
ServerHandler wiring consumes it, the field is never read directly"
)]
#[allow(dead_code)]
tool_router: rmcp::handler::server::router::tool::ToolRouter<Self>,
}

View file

@ -155,6 +155,7 @@ async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data:
/// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`)
/// output lines are forwarded to `writer` as `PrivEvent::Line` messages and
/// the returned strings are empty.
#[allow(clippy::too_many_lines)]
async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
match req {
PrivRequest::StartContainer { ref name } => {
@ -173,11 +174,35 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
}
PrivRequest::UpdateContainer { ref name, stream } => {
container_flake_action("update", name, stream, writer).await
validate_container_name(name)?;
let flake_ref = agent_flake_ref(name);
let args = [
"update",
&container_system_name(name),
"--flake",
&flake_ref,
];
if stream {
container_run_streaming(&args, writer).await
} else {
container_run(&args).await
}
}
PrivRequest::CreateContainer { ref name, stream } => {
container_flake_action("create", name, stream, writer).await
validate_container_name(name)?;
let flake_ref = agent_flake_ref(name);
let args = [
"create",
&container_system_name(name),
"--flake",
&flake_ref,
];
if stream {
container_run_streaming(&args, writer).await
} else {
container_run(&args).await
}
}
PrivRequest::DestroyContainer { ref name } => {
@ -199,17 +224,54 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
ref container,
ref binds,
ref isolation,
} => handle_write_nspawn_flags(container, binds, isolation.as_ref()),
} => {
validate_container_system_name(container)?;
for bind in binds {
validate_bind_path(&bind.host_path)?;
validate_bind_path(&bind.container_path)?;
}
write_nspawn_flags(container, binds, isolation.as_ref())?;
Ok((String::new(), String::new()))
}
PrivRequest::WriteResourceLimits {
ref container,
ref memory_max,
ref cpu_quota,
} => write_resource_limits(container, memory_max, cpu_quota),
} => {
validate_container_system_name(container)?;
let dir = format!("/run/systemd/system/container@{container}.service.d");
std::fs::create_dir_all(&dir).with_context(|| format!("create {dir}"))?;
let path = format!("{dir}/hyperhive-limits.conf");
let content = format!("[Service]\nMemoryMax={memory_max}\nCPUQuota={cpu_quota}\n");
std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
Ok((String::new(), String::new()))
}
PrivRequest::RemoveServiceDropin { ref container } => remove_service_dropin(container),
PrivRequest::RemoveServiceDropin { ref container } => {
validate_container_system_name(container)?;
let dir = format!("/run/systemd/system/container@{container}.service.d");
if Path::new(&dir).exists() {
std::fs::remove_dir_all(&dir).with_context(|| format!("remove {dir}"))?;
}
Ok((String::new(), String::new()))
}
PrivRequest::DaemonReload => daemon_reload().await,
PrivRequest::DaemonReload => {
let out = Command::new("systemctl")
.arg("daemon-reload")
.output()
.await
.context("invoke systemctl daemon-reload")?;
if !out.status.success() {
bail!(
"systemctl daemon-reload failed ({}): {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok((String::new(), String::new()))
}
PrivRequest::ReloadGatewayNginx => sync_gateway_nginx().await,
@ -217,12 +279,25 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
ref agent_name,
uid,
gid,
} => chown_socket_dir(agent_name, uid, gid),
} => {
validate_agent_name(agent_name)?;
let path = socket_dir_path(agent_name);
std::os::unix::fs::chown(&path, Some(uid), Some(gid))
.with_context(|| format!("chown {} to {uid}:{gid}", path.display()))?;
Ok((String::new(), String::new()))
}
PrivRequest::ChmodSocketDir {
ref agent_name,
mode,
} => chmod_socket_dir(agent_name, mode),
} => {
use std::os::unix::fs::PermissionsExt as _;
validate_agent_name(agent_name)?;
let path = socket_dir_path(agent_name);
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode))
.with_context(|| format!("chmod {:o} {}", mode, path.display()))?;
Ok((String::new(), String::new()))
}
PrivRequest::RunForgeAdmin { ref args } => {
for arg in args {
@ -248,134 +323,29 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
}
PrivRequest::RestartMatrixDaemon { ref agent_name } => {
restart_matrix_daemon(agent_name).await
validate_agent_name(agent_name)?;
let machine = format!("--machine=h-{agent_name}");
let unit = "hive-matrix-daemon.service";
let out = Command::new("systemctl")
.args([&machine, "restart", unit])
.output()
.await
.with_context(|| format!("systemctl restart {unit} in container h-{agent_name}"))?;
if !out.status.success() {
bail!(
"systemctl restart {unit} in h-{agent_name} exited {}: {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok((
String::from_utf8_lossy(&out.stdout).into_owned(),
String::from_utf8_lossy(&out.stderr).into_owned(),
))
}
}
}
/// Shared body for `CreateContainer` / `UpdateContainer`: validate the
/// name, build the `nixos-container <verb> … --flake <ref>` argv, and
/// run it (streaming line events to `writer` when `stream` is set).
async fn container_flake_action(
verb: &str,
name: &str,
stream: bool,
writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
validate_container_name(name)?;
let flake_ref = agent_flake_ref(name);
let args = [verb, &container_system_name(name), "--flake", &flake_ref];
if stream {
container_run_streaming(&args, writer).await
} else {
container_run(&args).await
}
}
/// `WriteNspawnFlags` — validate the container + every bind path, then
/// write the container's nspawn flag overrides.
fn handle_write_nspawn_flags(
container: &str,
binds: &[BindMount],
isolation: Option<&NetworkIsolation>,
) -> Result<(String, String)> {
validate_container_system_name(container)?;
for bind in binds {
validate_bind_path(&bind.host_path)?;
validate_bind_path(&bind.container_path)?;
}
write_nspawn_flags(container, binds, isolation)?;
Ok((String::new(), String::new()))
}
/// `RemoveServiceDropin` — remove the container service's drop-in dir
/// if present (idempotent).
fn remove_service_dropin(container: &str) -> Result<(String, String)> {
validate_container_system_name(container)?;
let dir = format!("/run/systemd/system/container@{container}.service.d");
if Path::new(&dir).exists() {
std::fs::remove_dir_all(&dir).with_context(|| format!("remove {dir}"))?;
}
Ok((String::new(), String::new()))
}
/// `ChownSocketDir` — chown the agent's host socket dir to its
/// container uid/gid.
fn chown_socket_dir(agent_name: &str, uid: u32, gid: u32) -> Result<(String, String)> {
validate_agent_name(agent_name)?;
let path = socket_dir_path(agent_name);
std::os::unix::fs::chown(&path, Some(uid), Some(gid))
.with_context(|| format!("chown {} to {uid}:{gid}", path.display()))?;
Ok((String::new(), String::new()))
}
/// `ChmodSocketDir` — set the mode on the agent's host socket dir.
fn chmod_socket_dir(agent_name: &str, mode: u32) -> Result<(String, String)> {
use std::os::unix::fs::PermissionsExt as _;
validate_agent_name(agent_name)?;
let path = socket_dir_path(agent_name);
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode))
.with_context(|| format!("chmod {:o} {}", mode, path.display()))?;
Ok((String::new(), String::new()))
}
/// `WriteResourceLimits` — drop a systemd `MemoryMax`/`CPUQuota`
/// override into the container service's drop-in dir.
fn write_resource_limits(
container: &str,
memory_max: &str,
cpu_quota: &str,
) -> Result<(String, String)> {
validate_container_system_name(container)?;
let dir = format!("/run/systemd/system/container@{container}.service.d");
std::fs::create_dir_all(&dir).with_context(|| format!("create {dir}"))?;
let path = format!("{dir}/hyperhive-limits.conf");
let content = format!("[Service]\nMemoryMax={memory_max}\nCPUQuota={cpu_quota}\n");
std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
Ok((String::new(), String::new()))
}
/// `DaemonReload` — `systemctl daemon-reload` on the host.
async fn daemon_reload() -> Result<(String, String)> {
let out = Command::new("systemctl")
.arg("daemon-reload")
.output()
.await
.context("invoke systemctl daemon-reload")?;
if !out.status.success() {
bail!(
"systemctl daemon-reload failed ({}): {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok((String::new(), String::new()))
}
/// `RestartMatrixDaemon` — restart the matrix daemon unit inside the
/// agent's container.
async fn restart_matrix_daemon(agent_name: &str) -> Result<(String, String)> {
validate_agent_name(agent_name)?;
let machine = format!("--machine=h-{agent_name}");
let unit = "hive-matrix-daemon.service";
let out = Command::new("systemctl")
.args([&machine, "restart", unit])
.output()
.await
.with_context(|| format!("systemctl restart {unit} in container h-{agent_name}"))?;
if !out.status.success() {
bail!(
"systemctl restart {unit} in h-{agent_name} exited {}: {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok((
String::from_utf8_lossy(&out.stdout).into_owned(),
String::from_utf8_lossy(&out.stderr).into_owned(),
))
}
/// Shared helper for `WriteAgentForgeToken` and `WriteAgentMatrixToken`.
/// Writes `content` to `AGENT_STATE_ROOT/<agent_name>/state/<filename>`,
/// chowns to the agent user (derived from the state dir's existing owner),

View file

@ -458,7 +458,7 @@ in
Pre-fetch the per-container system closures (agent-base +
manager toplevels) into the host's /nix/store as part of this
host's NixOS build, instead of letting the first agent spawn
do all the work.
do all the work. Closes #97.
Enabling this adds roughly the full nixpkgs runtime closure +
claude-code + the harness binary to your system closure size
@ -612,8 +612,8 @@ in
pkgs.git
];
# Pull the per-container toplevels into the host system closure.
# `system.extraDependencies` adds paths to the system build
# Pull the per-container toplevels into the host system closure
# (#97). `system.extraDependencies` adds paths to the system build
# without referencing them at runtime — nixos-rebuild fetches /
# builds them, they end up in /nix/store, and the first
# nixos-container update + start for an agent has nothing left to

View file

@ -118,7 +118,7 @@ let
# attempt for the full 60s loop, then exits with a misleading
# "core token absent or forge unreachable" — masking the real cause.
# Fail fast + loudly on 401/403 so the failure mode is legible and
# the operator/hive-c0re knows to re-mint forge-core-token.
# the operator/hive-c0re knows to re-mint forge-core-token (#1475).
RESP=$(${pkgs.curl}/bin/curl -s -w $'\n%{http_code}' \
"$FORGE_URL/api/v1/admin/runners/registration-token" \
-H "Authorization: token $CORE_TOKEN" || printf '\n000')
@ -231,23 +231,6 @@ in
defaultText = lib.literalExpression "pkgs.gitea-actions-runner";
description = "gitea-actions-runner package.";
};
jobTimeout = lib.mkOption {
type = lib.types.str;
default = "1h";
example = "3h";
description = ''
Per-job wall-clock timeout the runner enforces (act_runner's
`runner.timeout`). A job that exceeds it is killed, so a hung or
runaway build is bounded instead of holding the runner's single
slot indefinitely. Default `1h` comfortably covers a cold-cache
nix build while still bounding a stuck job; raise it (e.g.
`"3h"`) if you legitimately run jobs longer than that. Accepts a
Go duration string (`30m`, `1h`, `2h30m`). Note: this is
enforced by the runner process, so it only fires while that
process is itself healthy.
'';
};
};
config = lib.mkIf cfg.enable {
@ -304,9 +287,9 @@ in
# `nixos-container@…`. The earlier `nixos-container@hive-ci.service`
# matched no real unit, so before/wantedBy/partOf were silent
# no-ops — the partOf never bound, the oneshot stayed
# `active (exited)`, and the token was never refreshed on restart.
# Confirmed against the live `container@hive-matrix.service` unit
# during the matrix-outage incident.
# `active (exited)`, and the token was never refreshed on restart
# (a contributor to #1475). Confirmed against the live
# `container@hive-matrix.service` unit during the #1465 incident.
partOf = [ "container@hive-ci.service" ];
serviceConfig = {
Type = "oneshot";
@ -360,8 +343,8 @@ in
labels = cfg.labels;
settings = {
runner.capacity = cfg.concurrency;
# Per-job wall-clock cap — see the `jobTimeout` option.
runner.timeout = cfg.jobTimeout;
# Generous timeout for cold-cache nix builds.
runner.timeout = "3h";
};
};

View file

@ -290,12 +290,9 @@ in
THEMES = "catppuccin-vibec0re,forgejo-auto,forgejo-light,forgejo-dark,gitea-auto,gitea-light,gitea-dark";
};
# Point forgejo at the GPG key generated by the
# forgejo-gpg-init service below. SIGNING_KEY = "default"
# resolves via the forgejo process's git config
# (`user.signingkey`) — which forgejo-gpg-init sets to the
# generated key — not by scanning GNUPGHOME. GNUPGHOME is
# the keyring forgejo signs from; must be absolute +
# writeable by the forgejo user.
# forgejo-gpg-init oneshot below. "default" resolves to
# the first secret key found in GNUPGHOME. GNUPGHOME
# must be absolute and writeable by the forgejo user.
"repository.signing" = {
SIGNING_KEY = "default";
GNUPGHOME = "/var/lib/forgejo/.gnupg";
@ -324,67 +321,42 @@ in
pkgs.gnupg
];
# Ensure Forgejo has a usable GPG signing key so UI merges / CRUD
# commits are signed instead of erroring "does not have a signing
# key". This service (a) generates a key in forgejo's persistent
# keyring iff one isn't already present — keyed on the actual
# secret key, NOT a stamp file, so a partial state wipe that loses
# the key still regenerates it — and (b) points the forgejo user's
# git config at it (`user.signingkey` + commit/tag gpgsign), which
# is how `SIGNING_KEY = "default"` actually resolves. Runs as the
# forgejo user before forgejo on each start; idempotent (the keygen
# is guarded, the git-config is a cheap re-set).
# Generate a GPG signing key for Forgejo on first boot so UI
# merges produce signed commits instead of erroring "no key to
# sign with". The key lives in forgejo's persistent state dir
# (/var/lib/forgejo/.gnupg) and survives container restarts.
# The stamp file prevents re-generation on subsequent boots.
# Service runs as the forgejo user so file ownership is correct.
systemd.services.forgejo-gpg-init = {
description = "ensure Forgejo's GPG signing key + git signing config";
# Start before forgejo so the key + signing config are ready when
# forgejo reads repository.signing on startup.
description = "generate GPG signing key for Forgejo (once)";
# Start before forgejo so the key is ready when forgejo reads
# repository.signing config on startup.
wantedBy = [ "forgejo.service" ];
before = [ "forgejo.service" ];
unitConfig.ConditionPathExists = "!/var/lib/forgejo/.gnupg/hive-key-init.stamp";
serviceConfig = {
Type = "oneshot";
RemainAfterExit = true;
User = "forgejo";
Group = "forgejo";
};
# GNUPGHOME = the keyring forgejo signs from; HOME so
# `git config --global` lands where the forgejo process reads it.
environment = {
GNUPGHOME = "/var/lib/forgejo/.gnupg";
HOME = "/var/lib/forgejo";
};
environment.GNUPGHOME = "/var/lib/forgejo/.gnupg";
path = [
pkgs.gnupg
pkgs.git
pkgs.gnugrep
pkgs.gawk
pkgs.coreutils
];
script = ''
set -euo pipefail
mkdir -p "$GNUPGHOME"
chmod 700 "$GNUPGHOME"
# Generate only if no secret key is present (key-based guard,
# not a stamp — a stamp can outlive the key after a state wipe
# and wrongly suppress regeneration).
if ! gpg --list-secret-keys --with-colons 2>/dev/null | grep -q '^sec:'; then
printf '%s\n' \
'%no-protection' \
'Key-Type: RSA' \
'Key-Length: 4096' \
'Name-Real: HyperHive Forge' \
'Name-Email: forgejo@hive' \
'Expire-Date: 0' \
| gpg --batch --gen-key
fi
# Point git (hence Forgejo's SIGNING_KEY="default") at the key.
KEYID=$(gpg --list-secret-keys --keyid-format long --with-colons \
| awk -F: '/^sec:/ { print $5; exit }')
if [ -n "$KEYID" ]; then
git config --global user.signingkey "$KEYID"
git config --global commit.gpgsign true
git config --global tag.gpgsign true
fi
mkdir -p "$GNUPGHOME"
chmod 700 "$GNUPGHOME"
gpg --batch --gen-key <<'EOF'
%no-protection
Key-Type: RSA
Key-Length: 4096
Name-Real: HyperHive Forge
Name-Email: forgejo@hive
Expire-Date: 0
EOF
touch "$GNUPGHOME/hive-key-init.stamp"
'';
};
};

View file

@ -973,8 +973,8 @@ in
# Hive authoritative records — answer queries for the
# hive domain + its sub-domains with the bridge IP
# (where nginx is reachable from container netns once
# per-agent netns isolation lands; today it's the host
# loopback alias and works in either shape).
# #14 lands; today it's the host loopback alias and
# works in either shape).
#
# The forge / matrix entries are redundant in the
# common case where `forge.domain` /

View file

@ -97,8 +97,8 @@ in
default = false;
example = true;
description = ''
Flip agent containers from shared host netns to private netns.
When true, each agent container gets a dedicated veth
Flip agent containers from shared host netns to private netns
(#14). When true, each agent container gets a dedicated veth
pair attached to `bridgeName` and a deterministic IP from
the bridge subnet. The bridge (already up when `enable = true`)
becomes the sole routed path between the host and agent

View file

@ -1,42 +0,0 @@
#!/bin/sh
# CI lint: flags tracker tags (a hash followed by an issue number) in
# source comments. The hive convention is prose, not tracker tags, in
# code (see /knowledge/hive-rules.md) — tags rot, they point at moving
# targets and leak tracker coupling into the source tree.
#
# Emits a CI error annotation per hit and exits 1 if any tag is found,
# 0 otherwise. It runs as its own CI job, deliberately kept OUT of the
# required checks while the legacy backlog is cleaned up: a hit turns
# the job red (a visible, non-blocking signal on the PR) without
# blocking merge. Promote it to a required check once the tree is clean
# to make it a hard gate — no code change, just branch-protection.
#
# Scope: tracked *.rs *.nix *.js *.ts *.css *.html. Markdown is exempt
# (prose docs may legitimately cite the tracker). The pattern matches a
# hash, 2-5 digits, then a non-alphanumeric char or end-of-line. A real
# tracker tag is never glued to a letter, so the trailing class skips
# both letter-bearing / 6-8-digit hex colours (the digit run breaks or
# overruns) and digit-runs followed by a letter — e.g. hash-route
# fragments like #24h. Residual: a pure-numeric short hex (e.g. three
# identical digits) trips it — write the six-digit form to dodge.
set -eu
pattern='#[0-9]{2,5}([^0-9a-zA-Z]|$)'
# `/dev/null` forces grep to always print a filename prefix, even when
# xargs hands it a single file. `-r`/`-0` keep it robust to odd paths
# and an empty file list.
hits="$(
git ls-files -z '*.rs' '*.nix' '*.js' '*.ts' '*.css' '*.html' \
| xargs -0 -r grep -nE "$pattern" /dev/null 2>/dev/null || true
)"
if [ -n "$hits" ]; then
echo "$hits" | while IFS=: read -r file lineno _; do
printf '::error file=%s,line=%s::tracker tag in source — write prose, not a hash-number tag (see /knowledge/hive-rules.md)\n' "$file" "$lineno"
done
count="$(printf '%s\n' "$hits" | wc -l | tr -d ' ')"
printf 'check-issue-refs: %s tracker tag(s) found in source\n' "$count" >&2
exit 1
fi
exit 0