Compare commits

...
Author SHA1 Message Date
iris
9ed58ab96d clippy: fix lints that crane's cargoClippy properly enforces (#538)
The naersk → crane swap in the parent commit flips clippy from
silently passing to actually failing on `-D warnings` (naersk's
`mode = "clippy"` mangled the `--` separator so the deny never took
effect). This commit clears the surfaced lints so the workspace
builds clean under the new enforcement — every fix is mechanical and
preserves behaviour. Tests still pass (160 across the workspace).

Auto-fixes via `cargo clippy --fix`:
- `doc_markdown` (19 sites): bare identifiers in doc comments
  wrapped in backticks
- `format_in_format_args`, `explicit_into_iter_loop`,
  `redundant_closure_for_method_calls`, `useless_conversion`, and
  a few more — mechanical rewrites of the kind cargo can apply
  safely.

Hand-fixed:
- `match_same_arms` (forge_notify::is_atx_heading): two arms returning
  `true` collapsed into a single `matches!` pattern.
- `cast_sign_loss` + `format_push_string` (mcp.rs status formatter):
  guarded `i64 → u64` through `u64::try_from(…).unwrap_or(0)` (status
  timestamps are always positive in practice; clamp the skew edge to
  0) and swapped `out.push_str(&format!(…))` for `write!` into the
  buffer with an infallible-writer `let _ =`.
- `doc_lazy_continuation` in turn.rs + manager_server.rs + sh4re/lib.rs:
  doc paragraphs that the markdown parser was treating as list-item
  continuations got either a separating blank line or a `/`-for-`+`
  word swap so the parser stops seeing a list.
- `unused_async` (manager_server::handle_request_schedule_prompt):
  function has no `.await`; dropped the `async` and its `.await` call
  site.
- `needless_pass_by_value` (scheduled_prompts::submit): take
  `&NewSchedule` instead of moving the struct in; updated two prod
  callers and eight test sites to pass references.
- `type_complexity` (approvals::mark_cancelled): hoisted the
  7-tuple SELECT row shape into a `type CancelLookupRow = (…);` alias.

Allow-with-reason for intentional patterns:
- `option_option` (6 sites across dashboard / scheduled_prompts /
  manager_server): `Option<Option<T>>` carries three-state PATCH
  semantics (missing key = leave alone, `Some(None)` = clear,
  `Some(Some(v))` = set). Collapsing to `Option<T>` loses the
  "clear" state.
- `dead_code` (rebuild_queue::QueueKind::Destroy /
  QueueSource::CrashRecover; topology::parent_of / default_seed):
  wire-shape variants + API surfaces kept for the upcoming features
  (#361 follow-ups, future `Destroy` queue routing, crash-recovery
  path). Allowed at the variant / function level with the rationale
  in `reason = "…"`.
- `too_many_lines` on three specific call-sites: a 117-line
  exhaustive-variant test (dashboard_events::kind_tag_matches_…),
  the meta-flake string template renderer
  (meta::render_flake_with_lookup), and the notification poll loop
  (forge_notify::poll_once) — splitting any of them would just hide
  the contiguous shape they exist to keep visible.

`nix flake check` formatting target is still broken on main itself
(pre-existing nixfmt drift across ~28 files unrelated to this PR);
left alone here so the scope stays "crane port + lints the port
exposed" and the operator's review doesn't have to triage drive-by
nixfmt churn.
2026-05-29 01:45:48 +02:00
iris
4b6c733afb flake: replace naersk with crane (#538)
Framework swap, no public API change.

- naersk input → crane (`github:ipetkov/crane`); crane is stateless, no
  nixpkgs.follows needed.
- `forAllSystems` exposes `craneLib = crane.mkLib pkgs`,
  `cargoArtifacts = craneLib.buildDepsOnly` (built once, reused), and
  a shared `nativeBuildInputs = [ pkgs.librsvg pkgs.git ]` consumed by
  buildDepsOnly + buildPackage + cargoClippy so the three derivations
  see the same toolchain shape.
- `packages.default = craneLib.buildPackage` (was naersk-lib.buildPackage)
  with explicit `pname = "hyperhive-workspace"; version = "0.1.0";` —
  the virtual workspace Cargo.toml has no [package].name so crane
  needs the hint.
- `checks.clippy = craneLib.cargoClippy` (was naersk + overrideAttrs
  hack). Crane parses `cargoClippyExtraArgs = "--workspace --all-targets
  -- -D warnings"` correctly; naersk's `mode = "clippy"` used to mangle
  the `--` separator which is why the old wiring went through
  overrideAttrs. The whole hack — including `doCheck = false`,
  `copyTarget = false`, and the swapped buildPhase/installPhase — is
  now gone.
- librsvg native dep (#424) preserved on all three derivations. Added
  pkgs.git too — naersk auto-included it; crane is more minimal, so
  hive-c0re's `lifecycle::tests::setup_proposed_*` (which shell out to
  `git init`+commit) need it explicit to pass under `cargo test` in
  the sandbox.
- build.rs + hive-c0re/Cargo.toml comments updated from "naersk
  derivation" to "crane derivation".
- 3 doc-list-indentation lints in hive-sh4re/src/lib.rs cleaned up
  (replaced `+`-at-line-start with `and`/`/` so doc continuations
  don't trigger `clippy::doc_lazy_continuation`).

Validated locally: `nix build .#default --fallback` succeeds, all
117 tests pass, all four bins in `result/bin/`.
2026-05-29 01:45:48 +02:00
damocles
b652e6a6b0 hive-matrix: always use matrix.<domain> subdomain (mara on #552) 2026-05-29 01:31:40 +02:00
damocles
5a4eb3e053 hive-matrix: stateVersion 26.05 + drop premature federationPort firewall (argus nits #552) 2026-05-29 01:28:46 +02:00
damocles
2b1c1b54ac nix: add hive-matrix module + hyperhive.domain option (#548 part 1) 2026-05-29 01:25:29 +02:00
damocles
b43438d1a6 hive-forge: collapse autogenerated diffs to a single one-line summary (#222) 2026-05-29 01:16:51 +02:00
lexis
c9f537c2e3 docs: forge_notify.rs file map + wait_for_login mtime resumption (follow-up to #544 #545) 2026-05-29 00:34:11 +02:00
lexis
4a2b2fca4f docs(web-ui): logs flyout fills side-panel height (follow-up to #543) 2026-05-29 00:22:19 +02:00
damocles
994166b20c turn: snapshot file_count alongside mtime to survive unreadable meta.modified() 2026-05-28 20:08:27 +02:00
damocles
6e833b22d6 turn: wait_for_login resumes only on credentials mtime bump (closes #542) 2026-05-28 20:08:27 +02:00
damocles
b5be139fb7 forge_notify: diff mentions against raw excerpt to avoid heading-mention duplication 2026-05-28 20:08:05 +02:00
damocles
4b30f20dcf forge_notify: embed issue/pr body excerpt + surface truncated mention lines (closes #539) 2026-05-28 20:08:05 +02:00
iris
8bbb3e4905 tabs: logs flyout fills the side-panel height (closes #541)
The journald viewer opens in the side panel, but the <pre> only takes
the height of its content — so a short log fetch leaves the bottom 80%
of the panel empty, and a long one pushes the controls past the
viewport. Wrap the body in a column-flex container that fills the
side-panel-body, keep controls fixed at the top, and let the <pre>
flex-grow into the rest of the panel with its own overflow:

  .journal-body { display: flex; flex-direction: column; height: 100%; }
  .journal-controls { flex: 0 0 auto; }
  .journal-output { flex: 1 1 0; min-height: 0; overflow: auto; }

`min-height: 0` is the canonical "let me actually flex-shrink for
overflow" escape hatch on flex children. Also moved the auto-scroll
target from side-panel-body onto the <pre> itself — the panel-body no
longer overflows now that journal-body fills it; the <pre> is the new
scroll surface.

Pure CSS + one JS line; no DOM shape change.
2026-05-28 18:54:02 +02:00
30 changed files with 1449 additions and 644 deletions

View file

@ -150,6 +150,14 @@ hive-ag3nt/ in-container harness crate; produces TWO binaries
src/turn.rs claude --print + stream-json pump; --compact retry;
proactive compaction + auto session-reset
src/mcp.rs embedded MCP server (rmcp): AgentServer + ManagerServer
src/forge_notify.rs Forgejo webhook subscriber: formats new-issue /
new-PR / comment / review notifications as
broker messages; embeds a body excerpt
(up to BODY_TRUNCATE bytes, ATX headings
escaped) and surfaces any @mention lines that
fell outside the truncation window in a
trailing `mentions (truncated from body):`
block (#539 / #544).
src/login.rs probe /root/.claude/ for a valid session
src/login_session.rs drives `claude auth login` over stdio pipes
src/prompt.rs system prompt renderer: filters the unified
@ -189,10 +197,18 @@ nix/
modules/hive-c0re.nix systemd service + firewall + git wiring;
`contextWindowTokens` attrset (per-model,
injected as env vars into all containers);
imports hive-forge.nix
top-level `hyperhive.domain` option
(nullable; required when matrix.enable);
imports hive-forge.nix + hive-matrix.nix
modules/hive-forge.nix optional in-container Forgejo
(`hyperhive.forge.enable`, default on);
Catppuccin Mocha theme via tmpfiles C+ copy
modules/hive-matrix.nix optional in-container matrix-tuwunel
homeserver (`hyperhive.matrix.enable`,
default off); server_name derives from
`hyperhive.domain` (subdomain by default);
federation on (empty trusted_servers),
registration off, e2ee deferred (#551)
templates/harness-base.nix shared scaffolding for sub-agents + manager;
`hyperhive.model` option (HIVE_DEFAULT_MODEL)
templates/agent-base.nix sub-agent nixosConfiguration

View file

@ -38,6 +38,15 @@ Each agent harness (`hive-ag3nt serve` or `hive-m1nd serve`) runs:
the same path used at boot. The operator re-authenticates via
the per-agent web UI login flow; on success the sentinel is
cleared and the queued message drives the next turn normally.
**Mtime-snapshot resumption** (closes #542): `wait_for_login`
snapshots the `~/.claude/` dir (newest file mtime + file count)
at entry and only resumes when that snapshot advances — not
just when credentials exist on disk. This prevents a silent
infinite-401 loop: stale credentials already on disk at the
time of the 401 no longer cause an immediate false-resume. The
`DirSnapshot` struct tracks both axes; either a mtime advance
OR a file-count change triggers resume (the count axis handles
filesystems where `modified()` errors on every file).
7. Emit `LiveEvent::TurnEnd { ok, note }`. Sleep `poll_ms` to avoid
tight loops on transient failures.

View file

@ -383,7 +383,10 @@ body sits to the right with three stacked lines
fetches journald via `GET /api/journal/{name}?unit=&lines=`
(`journalctl -M <container> -b --no-pager --output=short-iso`).
A unit dropdown (harness service / full machine journal) and
a refresh button live in the panel.
a refresh button live in the panel. The panel uses a column-flex
layout so the `<pre>` log surface fills the full remaining panel
height (#541); scroll happens inside the `<pre>`, not the side
panel body.
- Plain navigation links (config repo, forge profile,
`dashboardLinks` extras) now live in the icon-only nav strip
on Line 1 — see above (issue #262). The agent's `config` link

63
flake.lock generated
View file

@ -1,45 +1,17 @@
{
"nodes": {
"fenix": {
"inputs": {
"nixpkgs": [
"naersk",
"nixpkgs"
],
"rust-analyzer-src": "rust-analyzer-src"
},
"crane": {
"locked": {
"lastModified": 1752475459,
"narHash": "sha256-z6QEu4ZFuHiqdOPbYss4/Q8B0BFhacR8ts6jO/F/aOU=",
"owner": "nix-community",
"repo": "fenix",
"rev": "bf0d6f70f4c9a9cf8845f992105652173f4b617f",
"lastModified": 1779130139,
"narHash": "sha256-BLrtr42azquO7MdGFU5a7KiMl3YpFlTeIXqy1fT5GlQ=",
"owner": "ipetkov",
"repo": "crane",
"rev": "edb38893982a3338972bb4a2ec7ce7c29ba10fd9",
"type": "github"
},
"original": {
"owner": "nix-community",
"repo": "fenix",
"type": "github"
}
},
"naersk": {
"inputs": {
"fenix": "fenix",
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1778151388,
"narHash": "sha256-lldMJPUeouEjO8/7aLuwhcsIw29vVihm2ZALzjiqfec=",
"owner": "nix-community",
"repo": "naersk",
"rev": "efdddff9ff4d8e7d0056d57ec67dac50f75ab8f6",
"type": "github"
},
"original": {
"owner": "nix-community",
"repo": "naersk",
"owner": "ipetkov",
"repo": "crane",
"type": "github"
}
},
@ -77,29 +49,12 @@
},
"root": {
"inputs": {
"naersk": "naersk",
"crane": "crane",
"nixpkgs": "nixpkgs",
"nixpkgs-unstable": "nixpkgs-unstable",
"treefmt-nix": "treefmt-nix"
}
},
"rust-analyzer-src": {
"flake": false,
"locked": {
"lastModified": 1752428706,
"narHash": "sha256-EJcdxw3aXfP8Ex1Nm3s0awyH9egQvB2Gu+QEnJn2Sfg=",
"owner": "rust-lang",
"repo": "rust-analyzer",
"rev": "591e3b7624be97e4443ea7b5542c191311aa141d",
"type": "github"
},
"original": {
"owner": "rust-lang",
"ref": "nightly",
"repo": "rust-analyzer",
"type": "github"
}
},
"treefmt-nix": {
"inputs": {
"nixpkgs": [

116
flake.nix
View file

@ -4,10 +4,10 @@
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-25.11";
nixpkgs-unstable.url = "github:NixOS/nixpkgs/nixpkgs-unstable";
naersk = {
url = "github:nix-community/naersk";
inputs.nixpkgs.follows = "nixpkgs";
};
# 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";
treefmt-nix = {
url = "github:numtide/treefmt-nix";
inputs.nixpkgs.follows = "nixpkgs";
@ -19,7 +19,7 @@
self,
nixpkgs,
nixpkgs-unstable,
naersk,
crane,
treefmt-nix,
}:
let
@ -45,22 +45,59 @@
inherit system;
pkgs = nixpkgs.legacyPackages.${system};
treefmt-eval = treefmt-nix.lib.evalModule pkgs treefmt-config;
naersk-lib = pkgs.callPackage naersk { };
craneLib = crane.mkLib pkgs;
# Build the workspace's dependency tree once, cached as
# its own derivation. `buildPackage` and `cargoClippy`
# both reuse this via `inherit cargoArtifacts;` so a
# workspace-only edit doesn't rebuild deps. Same
# `nativeBuildInputs` as the workspace build itself —
# build.rs runs during dep-build too (any deps with a
# build.rs need rsvg too if they transitively pull it
# in; harmless if they don't).
cargoArtifacts = craneLib.buildDepsOnly {
src = ./.;
# Workspace Cargo.toml is virtual (no `[package].name`),
# so crane can't auto-derive a name. Spell it out
# explicitly here and below — keeps the derivation name
# stable across crane bumps + silences the placeholder
# warning. Same `pname` everywhere so dep + workspace +
# clippy share a clean naming family.
pname = "hyperhive-workspace";
version = "0.1.0";
inherit nativeBuildInputs;
};
# Shared between buildDepsOnly + buildPackage + cargoClippy
# so the three derivations see the same toolchain shape.
# librsvg: hive-c0re/build.rs invokes `rsvg-convert` to
# render branding/agent-configs.svg → PNG that the daemon
# `include_bytes!`s (#424); SVG stays source-of-truth.
# git: naersk used to auto-include it; crane is more
# minimal, so we add it explicitly so hive-c0re's
# `lifecycle::tests::setup_proposed_*` (which shell out to
# `git init` + commit) pass under `cargo test` in the
# sandbox.
nativeBuildInputs = [
pkgs.librsvg
pkgs.git
];
}
);
in
{
packages = forAllSystems (
{ pkgs, naersk-lib, ... }:
{
default = naersk-lib.buildPackage {
pkgs,
craneLib,
cargoArtifacts,
nativeBuildInputs,
...
}:
{
default = craneLib.buildPackage {
src = ./.;
# librsvg ships `rsvg-convert`, which hive-c0re/build.rs
# invokes to render branding/agent-configs.svg into the
# PNG it embeds via `include_bytes!` (#424). Keeps the
# raster out of git — SVG stays source-of-truth, PNG is
# a build artifact in $OUT_DIR.
nativeBuildInputs = [ pkgs.librsvg ];
inherit cargoArtifacts nativeBuildInputs;
pname = "hyperhive-workspace";
version = "0.1.0";
meta.description = "hyperhive workspace (hive-c0re, hive-ag3nt, hive-m1nd)";
};
# Bundled browser assets — see ./nix/frontend.nix. Output is
@ -201,43 +238,28 @@
checks = forAllSystems (
{
treefmt-eval,
pkgs,
naersk-lib,
craneLib,
cargoArtifacts,
nativeBuildInputs,
...
}:
{
formatting = treefmt-eval.config.build.check self;
# Clippy as a check: reuse naersk's vendored-deps environment but
# replace the build phase with `cargo clippy --workspace --all-targets
# -- -D warnings`. Naersk's own `mode = "clippy"` mangles the `--`
# separator, so we go through overrideAttrs instead.
clippy =
(naersk-lib.buildPackage {
src = ./.;
# Skip the actual build; we only care about the clippy lint.
doCheck = false;
copyTarget = false;
# hive-c0re/build.rs needs rsvg-convert on PATH (#424);
# mirror the runtime derivation's nativeBuildInputs so
# clippy's vendored-deps build phase doesn't break on
# the missing tool.
nativeBuildInputs = [ pkgs.librsvg ];
}).overrideAttrs
(old: {
name = "${old.name}-clippy";
nativeBuildInputs = (old.nativeBuildInputs or [ ]) ++ [ pkgs.clippy ];
buildPhase = ''
runHook preBuild
cargo clippy --workspace --all-targets -- -D warnings
runHook postBuild
'';
installPhase = ''
runHook preInstall
mkdir -p $out
touch $out/.clippy-passed
runHook postInstall
'';
});
# Clippy as a check via crane's first-class `cargoClippy`
# builder. Reuses the shared `cargoArtifacts` (deps already
# built) and runs `cargo clippy --workspace --all-targets
# -- -D warnings` directly — no `overrideAttrs` hack needed,
# because crane parses `cargoClippyExtraArgs` correctly
# (naersk's `mode = "clippy"` used to mangle the `--`
# separator, which is why the old wiring went through
# overrideAttrs).
clippy = craneLib.cargoClippy {
src = ./.;
inherit cargoArtifacts nativeBuildInputs;
pname = "hyperhive-workspace";
version = "0.1.0";
cargoClippyExtraArgs = "--workspace --all-targets -- -D warnings";
};
}
);
};

View file

@ -450,11 +450,21 @@ a:hover {
.container-row.tombstone .name { color: var(--muted); }
/* Per-container journald viewer + applied-config viewer. Both open
in the side panel and lazy-fetch on open; output is monospace
inside a bordered <pre>, controls (unit select + refresh) above. */
inside a bordered <pre>, controls (unit select + refresh) above.
#541: the panel-body wrapper is a column flex container that fills
the side-panel-body so the <pre> can flex-grow into a single tall
scrollable surface instead of a short box at the top with the rest
of the panel empty. */
.journal-body {
display: flex;
flex-direction: column;
height: 100%;
gap: 0.4em;
}
.journal-controls {
flex: 0 0 auto;
display: flex;
gap: 0.5em;
margin-bottom: 0.4em;
align-items: center;
}
.journal-unit {
@ -472,7 +482,13 @@ a:hover {
color: var(--fg);
border: 1px solid var(--purple-dim);
padding: 0.5em 0.7em;
overflow-x: auto;
/* #541: take all leftover panel height + scroll inside the pre so
long log fetches don't push the controls off-screen. `min-height:
0` is the canonical "let me actually flex-shrink for overflow"
escape hatch on flex children. */
flex: 1 1 0;
min-height: 0;
overflow: auto;
font-size: 0.85em;
line-height: 1.4;
white-space: pre;

View file

@ -946,9 +946,12 @@ window.marked = marked;
pre.textContent = 'error: ' + resp.status + '\n' + text;
} else {
pre.textContent = text || '(empty)';
// Auto-scroll the panel to the newest lines on fresh fetch.
const sb = $('side-panel-body');
if (sb) sb.scrollTop = sb.scrollHeight;
// Auto-scroll to the newest lines on fresh fetch. #541
// moved the scroll surface from side-panel-body onto the
// <pre> itself (the panel-body now fills the viewport and
// the <pre> is the inner overflow container), so scroll
// the <pre> instead of the side-panel-body.
pre.scrollTop = pre.scrollHeight;
}
} catch (err) {
pre.textContent = 'fetch failed: ' + err;

View file

@ -24,6 +24,9 @@ tower-http.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true
[dev-dependencies]
tempfile = "3"
[[bin]]
name = "hive-ag3nt"
path = "src/bin/hive-ag3nt.rs"

View file

@ -143,10 +143,7 @@ pub enum LiveEvent {
/// compacting). `since_unix` matches `Bus::state_snapshot().1`
/// so the client's elapsed-time ticker keeps progressing across
/// SSE reconnects without drift.
TurnStateChanged {
state: TurnState,
since_unix: i64,
},
TurnStateChanged { state: TurnState, since_unix: i64 },
}
/// sqlite-backed event log. Wraps a `Connection` behind a `Mutex` so the
@ -284,10 +281,12 @@ impl TokenUsage {
let model_usage = v.get("modelUsage")?;
let map = model_usage.as_object()?;
for (_model, stats) in map {
if let Some(w) = stats.get("contextWindow").and_then(serde_json::Value::as_u64) {
if w > 0 {
return Some(w);
}
if let Some(w) = stats
.get("contextWindow")
.and_then(serde_json::Value::as_u64)
&& w > 0
{
return Some(w);
}
}
None
@ -353,18 +352,21 @@ pub fn context_window_tokens(model: &str) -> u64 {
// Per-model env vars set by `hyperhive.contextWindowTokens` in Nix.
for (key, val) in std::env::vars() {
if let Some(suffix) = key.strip_prefix("HIVE_CONTEXT_WINDOW_TOKENS_")
&& !suffix.is_empty() && m.contains(&suffix.to_ascii_lowercase())
&& let Ok(v) = val.trim().parse::<u64>()
&& v > 0 {
return v;
}
&& !suffix.is_empty()
&& m.contains(&suffix.to_ascii_lowercase())
&& let Ok(v) = val.trim().parse::<u64>()
&& v > 0
{
return v;
}
}
// Global override (single value, any model).
if let Ok(s) = std::env::var("HIVE_CONTEXT_WINDOW_TOKENS")
&& let Ok(v) = s.trim().parse::<u64>()
&& v > 0 {
return v;
}
&& v > 0
{
return v;
}
// Hard fallback for dev/test outside NixOS where env vars aren't set.
200_000
}
@ -450,9 +452,10 @@ impl Bus {
// persisted runtime override > compiled-in DEFAULT_MODEL.
// The nix config always wins on rebuild; the persisted file is kept
// for within-session tracking only (see persist_model / set_model).
let initial_model = configured_model()
.map(str::to_owned)
.unwrap_or_else(|| load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()));
let initial_model = configured_model().map_or_else(
|| load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()),
str::to_owned,
);
// Restore rate_limited from the sentinel file — if the harness
// crashed while parked, we should still show the right status on
// cold load until the next turn clears it.
@ -560,7 +563,8 @@ impl Bus {
pub fn record_turn_usage(&self, ctx: TokenUsage, cost: TokenUsage) {
*self.last_ctx_usage.lock().unwrap() = Some(ctx);
*self.last_cost_usage.lock().unwrap() = Some(cost);
self.last_turn_ended_unix.store(now_unix(), Ordering::Relaxed);
self.last_turn_ended_unix
.store(now_unix(), Ordering::Relaxed);
self.emit(LiveEvent::TokenUsageChanged { ctx, cost });
}

View file

@ -162,11 +162,7 @@ pub async fn run(socket: PathBuf, is_manager: bool) {
/// Fetch a JSON value from a URL using the agent's forge token. Returns
/// `None` on any HTTP or parse error (best-effort enrichment).
async fn fetch_json(
client: &reqwest::Client,
url: &str,
token: &str,
) -> Option<serde_json::Value> {
async fn fetch_json(client: &reqwest::Client, url: &str, token: &str) -> Option<serde_json::Value> {
let resp = client
.get(url)
.header("Authorization", format!("token {token}"))
@ -207,7 +203,7 @@ fn notif_type_label(t: &str) -> &str {
/// (`heading\n===`) are not handled here — rarer in practice and
/// would need multi-line lookahead; revisit if it actually shows up.
///
/// **ATX shape strictly:** CommonMark requires a space (or end-of-line)
/// **ATX shape strictly:** `CommonMark` requires a space (or end-of-line)
/// after the 1-6 leading `#`s to count as an ATX heading. Lines like
/// `#tag`, `#123`, `#!/bin/bash` are NOT headings — passing them through
/// untouched avoids the cosmetic noise argus flagged on PR #518 (`\#tag`
@ -236,7 +232,7 @@ fn escape_md_headings(body: &str) -> String {
out
}
/// Strict CommonMark ATX-heading detector: 1-6 leading `#`s followed
/// Strict `CommonMark` ATX-heading detector: 1-6 leading `#`s followed
/// by either a space, tab, or end-of-line. Anything tighter (`#tag`,
/// `#123`) is a non-heading line that the renderer will not promote.
fn is_atx_heading(line: &str) -> bool {
@ -244,11 +240,10 @@ fn is_atx_heading(line: &str) -> bool {
if !(1..=6).contains(&hashes) {
return false;
}
match line.as_bytes().get(hashes) {
None => true, // bare `#` / `##` / ... on its own line
Some(b' ') | Some(b'\t') => true, // proper ATX with space/tab after #s
_ => false, // `#tag` / `#123` — not a heading
}
// Bare `#` / `##` / ... on its own line, or proper ATX with a
// space/tab after the run of `#`s; anything else (`#tag` / `#123`)
// is not a heading.
matches!(line.as_bytes().get(hashes), None | Some(b' ' | b'\t'))
}
fn truncate(s: &str, max: usize) -> String {
@ -264,6 +259,69 @@ fn truncate(s: &str, max: usize) -> String {
format!("{}", &s[..end])
}
/// Detect `@username` mentions on a line. A mention is `@` followed by
/// at least one username char (alphanumeric / `_` / `-`) where the `@`
/// is at line start or follows a non-username char — so email-style
/// `foo@bar.com` does NOT count as a mention.
fn contains_mention(line: &str) -> bool {
let bytes = line.as_bytes();
for (i, &b) in bytes.iter().enumerate() {
if b != b'@' {
continue;
}
// Boundary: preceding byte must NOT be a username char.
let boundary_ok = match i.checked_sub(1).map(|j| bytes[j]) {
None => true,
Some(prev) => !is_username_byte(prev),
};
if !boundary_ok {
continue;
}
// Following byte must be at least one username char.
if bytes.get(i + 1).is_some_and(|&c| is_username_byte(c)) {
return true;
}
}
false
}
fn is_username_byte(b: u8) -> bool {
b.is_ascii_alphanumeric() || b == b'_' || b == b'-'
}
/// Walk `full_body` line-by-line; return lines that contain an
/// `@username` mention AND aren't already present (as a substring) in
/// `included_excerpt`. Used to surface mentions that fell outside the
/// truncation window so addressed agents see they were tagged even
/// when the body is long (closes #539).
fn extract_truncated_mention_lines<'a>(
full_body: &'a str,
included_excerpt: &str,
) -> Vec<&'a str> {
full_body
.lines()
.filter(|line| {
let trimmed = line.trim();
!trimmed.is_empty() && contains_mention(trimmed)
})
.filter(|line| !included_excerpt.contains(line.trim()))
.collect()
}
/// Build the trailing `mentions (truncated from body):\n > …` block.
/// Empty string when there's nothing to surface. Caller embeds it
/// directly before the meta suffix.
fn render_truncated_mentions(lines: &[&str]) -> String {
if lines.is_empty() {
return String::new();
}
let mut out = String::from("\n\nmentions (truncated from body):");
for line in lines {
write!(out, "\n > {}", line.trim()).ok();
}
out
}
/// Map a Forgejo review state to a readable action label.
/// Returns `None` for non-review states (regular comments have no `state` field;
/// `PENDING` means the review was saved but not submitted yet).
@ -327,7 +385,9 @@ async fn format_notification(
// API URLs for fetching content
let subject_api_url = notif["subject"]["url"].as_str().unwrap_or("");
let comment_api_url = notif["subject"]["latest_comment_url"].as_str().unwrap_or("");
let comment_api_url = notif["subject"]["latest_comment_url"]
.as_str()
.unwrap_or("");
let comment_html_url = notif["subject"]["latest_comment_html_url"]
.as_str()
.unwrap_or("");
@ -348,9 +408,27 @@ async fn format_notification(
// by creation/state-change of the subject itself.
let has_comment = !comment_api_url.is_empty() && comment_api_url != subject_api_url;
let meta = NotifMeta { title, notif_type, html_url, num, repo, meta_suffix, reason, subject, is_pr };
let meta = NotifMeta {
title,
notif_type,
html_url,
num,
repo,
meta_suffix,
reason,
subject,
is_pr,
};
if has_comment {
format_comment_notification(client, token, &meta, comment_api_url, comment_html_url, own_login).await
format_comment_notification(
client,
token,
&meta,
comment_api_url,
comment_html_url,
own_login,
)
.await
} else {
format_state_change_notification(notif, &meta, own_login)
}
@ -391,16 +469,28 @@ fn build_meta_suffix(subject: Option<&serde_json::Value>, is_pr: bool, reason: &
.and_then(|s| s["requested_reviewers"].as_array())
.map(|arr| arr.iter().filter_map(|r| r["login"].as_str()).collect())
.unwrap_or_default();
if reviewers.is_empty() { None } else { Some(format!("reviewer: {}", reviewers.join(", "))) }
if reviewers.is_empty() {
None
} else {
Some(format!("reviewer: {}", reviewers.join(", ")))
}
} else {
None
};
// Always include reason so multiple notifications for the same event
// (each with a different Forgejo reason) are distinguishable (closes #110).
let reason_line = if reason.is_empty() { None } else { Some(format!("reason: {reason}")) };
let reason_line = if reason.is_empty() {
None
} else {
Some(format!("reason: {reason}"))
};
let mut out = format!("\n{assignee_line}");
if let Some(r) = reviewer_line { write!(out, "\n{r}").ok(); }
if let Some(r) = reason_line { write!(out, "\n{r}").ok(); }
if let Some(r) = reviewer_line {
write!(out, "\n{r}").ok();
}
if let Some(r) = reason_line {
write!(out, "\n{r}").ok();
}
out
}
@ -441,17 +531,42 @@ async fn format_comment_notification(
.and_then(|c| c["state"].as_str())
.and_then(review_state_label);
let url = if comment_html_url.is_empty() { meta.html_url } else { comment_html_url };
let author = if actor_login.is_empty() { "?" } else { actor_login };
let NotifMeta { title, notif_type, num, repo, meta_suffix, .. } = meta;
let url = if comment_html_url.is_empty() {
meta.html_url
} else {
comment_html_url
};
let author = if actor_login.is_empty() {
"?"
} else {
actor_login
};
let NotifMeta {
title,
notif_type,
num,
repo,
meta_suffix,
..
} = meta;
// Truncate the raw body first so the mention-overflow diff compares
// like-for-like (escape_md_headings rewrites `# foo` to `\# foo`, so
// doing it before the diff would re-surface heading-prefixed mention
// lines as fake overflow). Escape happens after for display only.
let raw_excerpt = truncate(body_text, BODY_TRUNCATE);
// Surface @mentions that fell outside the truncation window so an
// addressed agent never silently misses a tag on a long comment
// (closes #539). Skipped when the embed wasn't actually truncated.
let truncated_mentions = if body_text.len() > BODY_TRUNCATE {
render_truncated_mentions(&extract_truncated_mention_lines(body_text, &raw_excerpt))
} else {
String::new()
};
// Escape ATX headings in the user-authored body so the embedded
// text doesn't blow into top-level h1/h2 in the wrapper message
// when the dashboard renders it (closes #455). Done once here
// because both code paths fall through the same truncate+embed
// pattern.
let escaped = escape_md_headings(body_text);
let body_for_embed = truncate(&escaped, BODY_TRUNCATE);
// when the dashboard renders it (closes #455).
let body_for_embed = escape_md_headings(&raw_excerpt);
if let Some(review_label) = review_state {
// Review submission on a PR.
let kind = format!("PR {review_label}{num}{repo}");
@ -459,14 +574,16 @@ async fn format_comment_notification(
if body_text.is_empty() {
write!(out, "\n\nreviewer: {author}").ok();
} else {
write!(out, "\n\n{author}: {body_for_embed}").ok();
write!(out, "\n\n{author}: {body_for_embed}{truncated_mentions}").ok();
}
out.push_str(meta_suffix);
Some(out)
} else {
// Regular comment.
let kind = format!("comment on {}{num}{repo}", notif_type_label(notif_type));
let mut out = format!("[{kind}] {title}\nurl: {url}\n\n{author}: {body_for_embed}");
let mut out = format!(
"[{kind}] {title}\nurl: {url}\n\n{author}: {body_for_embed}{truncated_mentions}"
);
if out.ends_with('\n') {
out.pop();
}
@ -499,7 +616,17 @@ fn format_state_change_notification(
return None;
}
let NotifMeta { title, notif_type, html_url, num, repo, meta_suffix, reason: _, subject, is_pr } = meta;
let NotifMeta {
title,
notif_type,
html_url,
num,
repo,
meta_suffix,
reason: _,
subject,
is_pr,
} = meta;
let label = notif_type_label(notif_type);
let kind = match notif_state {
"merged" => format!("{label} merged{num}{repo}"),
@ -526,12 +653,41 @@ fn format_state_change_notification(
kind
};
let mut out = format!("[{kind}] {title}\nurl: {html_url}");
// Include the start of the issue/PR description so the agent gets
// context without a follow-up fetch (closes #539). Same escape +
// truncate pipeline as comment bodies. Mentions that fell outside
// the truncation window are surfaced separately so an addressed
// agent never silently misses a long-body @tag.
let body_block = subject
.as_ref()
.and_then(|s| s["body"].as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|raw| {
// Truncate raw first so mention diff sees the same heading
// markers as full_body (see comment in
// format_comment_notification). Escape happens after for
// display only.
let raw_excerpt = truncate(raw, BODY_TRUNCATE);
let truncated = extract_truncated_mention_lines(raw, &raw_excerpt);
let mentions = render_truncated_mentions(&truncated);
let excerpt = escape_md_headings(&raw_excerpt);
format!("\n\n{excerpt}{mentions}")
})
.unwrap_or_default();
let mut out = format!("[{kind}] {title}\nurl: {html_url}{body_block}");
out.push_str(meta_suffix);
Some(out)
}
#[allow(clippy::too_many_arguments)]
#[allow(
clippy::too_many_lines,
reason = "single-pass notification poll loop — split would obscure the \
sequential 'fetch / classify / dispatch' rhythm and add helper \
functions for state shared across all three phases"
)]
async fn poll_once(
client: &reqwest::Client,
forge_url: &str,
@ -574,10 +730,15 @@ async fn poll_once(
return;
}
debug!(count = notifications.len(), "forge_notify: delivering notifications");
debug!(
count = notifications.len(),
"forge_notify: delivering notifications"
);
for notif in &notifications {
let Some(id) = notif["id"].as_u64() else { continue };
let Some(id) = notif["id"].as_u64() else {
continue;
};
// Reason drop-list: suppress noisy reasons (subscribed/participating).
// null/unknown reasons pass through — directed signals are never
@ -635,28 +796,30 @@ async fn poll_once(
// when HIVE_FORGE_KEEP_SUBSCRIPTIONS=1 — triage and other firehose
// consumers set this to retain broad repo visibility.
let reason = notif["reason"].as_str().unwrap_or("");
if !keep_subscriptions && reason == "subscribed"
if !keep_subscriptions
&& reason == "subscribed"
&& let Some(repo) = notif["repository"]["full_name"].as_str()
&& !unsubbed_repos.contains(repo) {
let unsub_url = format!("{forge_url}/api/v1/repos/{repo}/subscription");
match client
.delete(&unsub_url)
.header("Authorization", format!("token {token}"))
.send()
.await
{
Ok(r) if r.status().is_success() || r.status().as_u16() == 404 => {
debug!(%repo, "forge_notify: unsubscribed from repo watch");
unsubbed_repos.insert(repo.to_owned());
}
Ok(r) => {
debug!(%repo, status = %r.status(), "forge_notify: unsub non-2xx (ignored)");
}
Err(e) => {
debug!(%repo, error = ?e, "forge_notify: unsub request failed (ignored)");
}
}
&& !unsubbed_repos.contains(repo)
{
let unsub_url = format!("{forge_url}/api/v1/repos/{repo}/subscription");
match client
.delete(&unsub_url)
.header("Authorization", format!("token {token}"))
.send()
.await
{
Ok(r) if r.status().is_success() || r.status().as_u16() == 404 => {
debug!(%repo, "forge_notify: unsubscribed from repo watch");
unsubbed_repos.insert(repo.to_owned());
}
Ok(r) => {
debug!(%repo, status = %r.status(), "forge_notify: unsub non-2xx (ignored)");
}
Err(e) => {
debug!(%repo, error = ?e, "forge_notify: unsub request failed (ignored)");
}
}
}
}
}
@ -772,4 +935,89 @@ mod tests {
assert_eq!(escape_md_headings("body\n"), "body\n");
assert_eq!(escape_md_headings("no trailing"), "no trailing");
}
#[test]
fn contains_mention_matches_at_line_start_and_mid_line() {
assert!(contains_mention("@damocles take a look"));
assert!(contains_mention("cc @argus please"));
assert!(contains_mention("see (@mara) for context"));
// Hyphens / underscores / digits are valid username chars.
assert!(contains_mention("ping @h-m1nd-2"));
}
#[test]
fn contains_mention_rejects_email_and_bare_at() {
// Email addresses (`foo@bar.com`) and `@` followed by
// whitespace or punctuation are not mentions — boundary check
// requires the preceding byte to NOT be a username char.
assert!(!contains_mention("foo@bar.com"));
assert!(!contains_mention("send to user@example.org"));
assert!(!contains_mention("just an @"));
assert!(!contains_mention("@ space"));
assert!(!contains_mention("plain text"));
assert!(!contains_mention(""));
}
#[test]
fn extract_truncated_keeps_mention_lines_outside_excerpt() {
// Long body where the @mention sits AFTER the excerpt's cutoff
// — the truncated extractor must surface it.
let full = "first line\nsecond line\n@damocles tagged here\n";
let excerpt = "first line\nsecond line\n"; // mention not present
let lines = extract_truncated_mention_lines(full, excerpt);
assert_eq!(lines, vec!["@damocles tagged here"]);
}
#[test]
fn extract_truncated_drops_mentions_already_in_excerpt() {
// Mention is inside the embed window already — no need to
// re-surface, would be noise.
let full = "@damocles read this\nmore body\n";
let excerpt = "@damocles read this\nmore body\n";
let lines = extract_truncated_mention_lines(full, excerpt);
assert!(lines.is_empty());
}
#[test]
fn extract_truncated_skips_blank_and_no_mention_lines() {
// Only lines with an actual mention survive — random body
// text past the cutoff stays dropped.
let full = "first\n\nsecond paragraph\n@argus reviewer\nfinal\n";
let excerpt = "first";
let lines = extract_truncated_mention_lines(full, excerpt);
assert_eq!(lines, vec!["@argus reviewer"]);
}
#[test]
fn extract_truncated_does_not_resurface_heading_mention_inside_window() {
// Regression for argus's nit on PR #544: the diff used to
// compare full_body against the *escaped* excerpt. Lines like
// `# @argus check this` survived as-is in the body but became
// `\# @argus check this` in the excerpt, so the `contains`
// check failed and the mention was re-surfaced as if it had
// fallen outside the window. Pass the unescaped excerpt and
// the duplicate disappears.
let full = "# @argus check this\nmore body\n";
let raw_excerpt = full; // fits entirely
let lines = extract_truncated_mention_lines(full, raw_excerpt);
assert!(lines.is_empty(), "heading+mention inside window must not be re-surfaced, got {lines:?}");
}
#[test]
fn render_truncated_mentions_empty_is_empty_string() {
// Zero overhead on the healthy short-body path: caller
// concatenates this directly so an empty input must produce
// no spacing.
assert_eq!(render_truncated_mentions(&[]), "");
}
#[test]
fn render_truncated_mentions_formats_block() {
let lines = ["cc @damocles", " @argus second mention"];
let rendered = render_truncated_mentions(&lines);
assert_eq!(
rendered,
"\n\nmentions (truncated from body):\n > cc @damocles\n > @argus second mention",
);
}
}

View file

@ -47,7 +47,7 @@ pub enum SocketReply {
Recent(Vec<hive_sh4re::InboxRow>),
Logs(String),
/// `list_schedules` result — used by the manager surface only;
/// AgentResponse has no equivalent variant.
/// `AgentResponse` has no equivalent variant.
Schedules(Vec<hive_sh4re::WireSchedule>),
LooseEnds(Vec<hive_sh4re::LooseEnd>),
PendingRemindersCount(u64),
@ -185,8 +185,7 @@ pub fn format_recv(resp: Result<SocketReply, anyhow::Error>) -> String {
/// resurfaced by `RequeueInflight` on this session's boot. Same
/// string surfaces in the wake prompt (see the bin loops) and the
/// in-turn `recv` tool result so claude sees the warning either way.
pub const REDELIVERY_HINT: &str =
"[redelivered after harness restart — may already be handled]\n";
pub const REDELIVERY_HINT: &str = "[redelivered after harness restart — may already be handled]\n";
/// Format helper for `get_loose_ends`: renders a short bulleted list
/// of pending approvals + questions + reminders. Empty list collapses
@ -307,17 +306,32 @@ pub fn format_agent_meta(resp: Result<SocketReply, anyhow::Error>) -> String {
match status_text {
None => out.push_str("\nstatus: <none>"),
Some(s) => {
use std::fmt::Write as _;
let age = status_set_at.and_then(|ts| {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()?
.as_secs();
let secs = now.saturating_sub(ts as u64);
// `ts` is a unix epoch second the agent itself
// sourced from `SystemTime` — always positive
// in normal operation. Clamp the negative
// (clock-skew) edge to 0 before the unsigned
// cast so the cast loses no real precision.
let ts_secs = u64::try_from(ts).unwrap_or(0);
let secs = now.saturating_sub(ts_secs);
Some(format_age_secs(secs))
});
// `write!` into the buffer instead of `push_str(&format!(…))` —
// avoids the intermediate allocation clippy::format_push_string
// flags. The infallible `String` writer makes this safe to
// `let _ =`-ignore.
match age {
Some(a) => out.push_str(&format!("\nstatus: {s} (set {a} ago)")),
None => out.push_str(&format!("\nstatus: {s}")),
Some(a) => {
let _ = write!(out, "\nstatus: {s} (set {a} ago)");
}
None => {
let _ = write!(out, "\nstatus: {s}");
}
}
}
}
@ -626,9 +640,13 @@ impl AgentServer {
)]
async fn set_status(&self, Parameters(args): Parameters<SetStatusArgs>) -> String {
run_tool_envelope("set_status", args.text.clone(), async move {
let (resp, retries) =
self.dispatch(hive_sh4re::AgentRequest::SetStatus { text: args.text }).await;
annotate_retries(format_ack(resp, "set_status", "status updated".to_owned()), retries)
let (resp, retries) = self
.dispatch(hive_sh4re::AgentRequest::SetStatus { text: args.text })
.await;
annotate_retries(
format_ack(resp, "set_status", "status updated".to_owned()),
retries,
)
})
.await
}
@ -644,10 +662,7 @@ impl AgentServer {
where the system-prompt label could be stale. Status reads `<none>` when the \
target has never called `set_status` or has cleared it."
)]
async fn get_agent_meta(
&self,
Parameters(args): Parameters<GetAgentMetaArgs>,
) -> String {
async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
let log = args.name.clone().unwrap_or_else(|| "<self>".to_owned());
run_tool_envelope("get_agent_meta", log, async move {
let (resp, retries) = self
@ -679,7 +694,11 @@ impl AgentServer {
.dispatch(hive_sh4re::AgentRequest::CancelLooseEnd { kind, id })
.await;
annotate_retries(
format_ack(resp, "cancel_loose_end", format!("cancelled {kind_label} {id}")),
format_ack(
resp,
"cancel_loose_end",
format!("cancelled {kind_label} {id}"),
),
retries,
)
})
@ -721,7 +740,10 @@ impl AgentServer {
file_path: args.file_path,
})
.await;
annotate_retries(format_ack(resp, "remind", "reminder scheduled".to_string()), retries)
annotate_retries(
format_ack(resp, "remind", "reminder scheduled".to_string()),
retries,
)
})
.await
}
@ -949,7 +971,7 @@ pub struct RequestSchedulePromptArgs {
/// `None` / absent = one-shot. `Some(n > 0)` = recurring every
/// `n` seconds. The worker clamps catch-up so a long downtime
/// fires ONCE on resume (skipped-cycle count surfaces in the
/// per-target last_result), not N delayed pulses in a row.
/// per-target `last_result`), not N delayed pulses in a row.
#[serde(default)]
pub interval_seconds: Option<u64>,
/// Optional description shown on the dashboard approval card +
@ -1301,10 +1323,7 @@ impl ManagerServer {
Authorization mirrors `cancel_schedule`: you can fire your own schedules + any \
owned by a sub-agent in your subtree per topology.json."
)]
async fn fire_schedule_now(
&self,
Parameters(args): Parameters<FireScheduleNowArgs>,
) -> String {
async fn fire_schedule_now(&self, Parameters(args): Parameters<FireScheduleNowArgs>) -> String {
let log = format!("{args:?}");
run_tool_envelope("fire_schedule_now", log, async move {
let id = args.id;
@ -1337,7 +1356,10 @@ impl ManagerServer {
targets: args.targets,
})
.await;
annotate_retries(format_ack(resp, "cancel_schedule", format!("cancelled #{id}")), retries)
annotate_retries(
format_ack(resp, "cancel_schedule", format!("cancelled #{id}")),
retries,
)
})
.await
}
@ -1394,7 +1416,9 @@ impl ManagerServer {
)]
async fn list_schedules(&self) -> String {
run_tool_envelope("list_schedules", String::new(), async move {
let (resp, retries) = self.dispatch(hive_sh4re::ManagerRequest::ListSchedules).await;
let (resp, retries) = self
.dispatch(hive_sh4re::ManagerRequest::ListSchedules)
.await;
let body = match resp {
Ok(SocketReply::Schedules(schedules)) => serde_json::to_string(&schedules)
.unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")),
@ -1541,7 +1565,10 @@ impl ManagerServer {
file_path: args.file_path,
})
.await;
annotate_retries(format_ack(resp, "remind", "reminder scheduled".to_string()), retries)
annotate_retries(
format_ack(resp, "remind", "reminder scheduled".to_string()),
retries,
)
})
.await
}
@ -1573,9 +1600,13 @@ impl ManagerServer {
)]
async fn set_status(&self, Parameters(args): Parameters<SetStatusArgs>) -> String {
run_tool_envelope("set_status", args.text.clone(), async move {
let (resp, retries) =
self.dispatch(hive_sh4re::ManagerRequest::SetStatus { text: args.text }).await;
annotate_retries(format_ack(resp, "set_status", "status updated".to_owned()), retries)
let (resp, retries) = self
.dispatch(hive_sh4re::ManagerRequest::SetStatus { text: args.text })
.await;
annotate_retries(
format_ack(resp, "set_status", "status updated".to_owned()),
retries,
)
})
.await
}
@ -1590,10 +1621,7 @@ impl ManagerServer {
drift across renames. Status reads `<none>` when the target has never called \
`set_status` or has cleared it."
)]
async fn get_agent_meta(
&self,
Parameters(args): Parameters<GetAgentMetaArgs>,
) -> String {
async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
let log = args.name.clone().unwrap_or_else(|| "<self>".to_owned());
run_tool_envelope("get_agent_meta", log, async move {
let (resp, retries) = self
@ -1628,7 +1656,11 @@ impl ManagerServer {
.dispatch(hive_sh4re::ManagerRequest::CancelLooseEnd { kind, id })
.await;
annotate_retries(
format_ack(resp, "cancel_loose_end", format!("cancelled {kind_label} {id}")),
format_ack(
resp,
"cancel_loose_end",
format!("cancelled {kind_label} {id}"),
),
retries,
)
})
@ -1800,10 +1832,7 @@ pub fn allowed_tools_arg(flavor: Flavor) -> String {
if patterns.is_empty() {
vec!["Bash".to_owned()]
} else {
patterns
.into_iter()
.map(|p| format!("Bash({p})"))
.collect()
patterns.into_iter().map(|p| format!("Bash({p})")).collect()
}
} else {
vec![(*s).to_owned()]

View file

@ -15,7 +15,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command;
use crate::events::{Bus, LiveEvent};
use crate::login::{self, LoginState};
use crate::login::LoginState;
use crate::mcp;
/// `--settings` JSON applied to every claude invocation. Lives as a
@ -235,6 +235,7 @@ fn effective_context_window(bus: &Bus) -> u64 {
/// Resolve the auto-reset watermark. Priority order:
/// 1. `HIVE_AUTO_RESET_WATERMARK_TOKENS` env var (explicit override).
/// 2. 50% of `effective_context_window(bus)`.
///
/// `0` disables auto-reset entirely.
fn auto_reset_watermark_tokens(bus: &Bus) -> u64 {
if let Some(v) = std::env::var("HIVE_AUTO_RESET_WATERMARK_TOKENS")
@ -259,6 +260,7 @@ fn cache_ttl_secs() -> u64 {
/// Resolve the proactive-compaction watermark. Priority order:
/// 1. `HIVE_COMPACT_WATERMARK_TOKENS` env var (explicit override).
/// 2. 75% of `effective_context_window(bus)`.
///
/// `0` disables proactive compaction (reactive path still applies).
fn compact_watermark_tokens(bus: &Bus) -> u64 {
if let Some(v) = std::env::var("HIVE_COMPACT_WATERMARK_TOKENS")
@ -299,9 +301,7 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco
// /compact itself would be absurd recursion; bubble it up as
// a normal failure path.
match compact_session(files, bus).await {
TurnOutcome::Ok | TurnOutcome::Compacted => {
run_turn(prompt, files, bus).await
}
TurnOutcome::Ok | TurnOutcome::Compacted => run_turn(prompt, files, bus).await,
other => return other,
}
}
@ -315,10 +315,9 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco
// turn overflows into the reactive path. Best-effort — never changes
// the outcome of the turn that already succeeded, but records it as
// `Compacted` so turn stats can distinguish it from a plain `Ok`.
if matches!(outcome, TurnOutcome::Ok)
&& maybe_checkpoint_and_compact(files, bus).await {
return TurnOutcome::Compacted;
}
if matches!(outcome, TurnOutcome::Ok) && maybe_checkpoint_and_compact(files, bus).await {
return TurnOutcome::Compacted;
}
outcome
}
@ -464,9 +463,20 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) {
}
}
/// Block until the bound `~/.claude/` dir contains a session, polling
/// `claude_dir` on a `poll_ms` interval (min 2s). Flips `state` to
/// `Online` when login lands; caller resumes its serve loop.
/// Block until the bound `~/.claude/` dir contains a session that
/// post-dates this call, polling on a `poll_ms` interval (min 2s).
/// Flips `state` to `Online` when login lands; caller resumes its
/// serve loop.
///
/// **Mtime-progress, not bare existence (closes #542)**: an
/// existence-only check (the pre-#542 behaviour) immediately returns
/// after a 401 because the stale `credentials.json` is still on disk
/// — the next turn then 401s on the same tokens and the harness
/// loops forever. We snapshot the newest file mtime in `claude_dir`
/// at entry and only resume when something has been written since
/// that snapshot (the operator's `/login/code` flow lands a refreshed
/// credentials file, bumping its mtime). First-time login (empty
/// dir → `None` snapshot) still flips on the first file appearing.
///
/// # Panics
///
@ -481,11 +491,12 @@ pub async fn wait_for_login(
claude_dir = %claude_dir.display(),
"no claude session — staying in partial-run mode (web UI only)"
);
let snapshot = snapshot_dir(claude_dir);
let probe = Duration::from_millis(poll_ms.max(2000));
loop {
tokio::time::sleep(probe).await;
if login::has_session(claude_dir) {
tracing::info!("claude session detected — entering turn loop");
if session_refreshed(snapshot, snapshot_dir(claude_dir)) {
tracing::info!("claude session refreshed — entering turn loop");
*state.lock().unwrap() = LoginState::Online;
bus.emit_status("online");
return;
@ -493,6 +504,61 @@ pub async fn wait_for_login(
}
}
/// Snapshot of the credentials dir at a point in time: number of
/// regular files + newest `mtime` across them. The two axes are both
/// load-bearing for `wait_for_login`'s refresh check (`session_refreshed`):
/// mtime catches the common case (re-login overwrites an existing
/// credentials file in-place), file_count catches the pathological case
/// where `meta.modified()` errors on every file (exotic fs, NFS quirks)
/// so the mtime axis stays `None` forever but new files still trigger a
/// resume. Defaults to `{0, None}` on read_dir failure (missing or
/// unreadable dir) — `wait_for_login` then resumes when files first
/// appear.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct DirSnapshot {
file_count: usize,
newest_mtime: Option<std::time::SystemTime>,
}
fn snapshot_dir(dir: &Path) -> DirSnapshot {
let Ok(entries) = std::fs::read_dir(dir) else {
return DirSnapshot::default();
};
let mut snap = DirSnapshot::default();
for entry in entries.flatten() {
if !entry.file_type().is_ok_and(|t| t.is_file()) {
continue;
}
snap.file_count += 1;
let Ok(meta) = entry.metadata() else { continue };
let Ok(mtime) = meta.modified() else { continue };
if snap.newest_mtime.is_none_or(|cur| mtime > cur) {
snap.newest_mtime = Some(mtime);
}
}
snap
}
/// Has the credentials dir been written since `prev`? Used as the
/// exit condition for `wait_for_login`:
///
/// - file_count changed → something was added or removed, treat as
/// refresh (covers the "all files have unreadable mtime" edge case).
/// - newest_mtime advanced → existing file was rewritten in place
/// (the common claude re-login path).
/// - prev had no mtime (empty or all-unreadable) and now has one →
/// first useful signal we've seen, treat as refresh.
fn session_refreshed(prev: DirSnapshot, now: DirSnapshot) -> bool {
if now.file_count != prev.file_count {
return true;
}
match (prev.newest_mtime, now.newest_mtime) {
(None, Some(_)) => true,
(Some(p), Some(n)) => n > p,
_ => false,
}
}
/// Spawn `claude` for one turn and pump `stream-json` stdout into the
/// live event bus. Prompt goes over stdin (variadic
/// `--allowedTools`/`--tools` would otherwise eat a trailing positional
@ -640,53 +706,48 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
auth_out.store(true, Ordering::Relaxed);
}
match serde_json::from_str::<serde_json::Value>(&line) {
Ok(v) => {
// Rate-limit detection: only fire on JSON `error` events,
// not on arbitrary text content. An agent discussing a past
// rate limit in its response would otherwise trigger a false
// positive (the full conversation flows through stdout as
// stream-json, so any text the model outputs is visible here).
if v.get("type").and_then(|t| t.as_str()) == Some("error") {
let raw = v.to_string();
if RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m)) {
rate_out.store(true, Ordering::Relaxed);
}
}
if let Some(u) = crate::events::TokenUsage::from_assistant_event(&v) {
last_inference = Some(u);
}
if let Some(cost) = crate::events::TokenUsage::from_stream_event(&v) {
// Fallback to `cost` if the turn somehow produced
// a result without any assistant event — keeps the
// ctx badge from going stale on a degenerate turn.
let ctx = last_inference.unwrap_or(cost);
bus_out.record_turn_usage(ctx, cost);
}
// Seed the API-reported context-window from the result
// event's `modelUsage.*.contextWindow` field. This is
// the authoritative per-inference active window used for
// compaction watermarks — it reflects what the model
// actually enforces, which may differ from the Nix
// config (e.g. 200k active window on a 1M cache model).
if let Some(w) =
crate::events::TokenUsage::context_window_from_result_event(&v)
{
bus_out.set_api_context_window(w);
}
bus_out.observe_stream(&v);
bus_out.emit(LiveEvent::Stream(v));
}
Err(_) => {
// Non-JSON stdout: raw text check is fine here since these
// are claude CLI messages, not conversation content.
if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) {
// Rate-limit detection: only fire on JSON `error` events,
// not on arbitrary text content. An agent discussing a past
// rate limit in its response would otherwise trigger a false
// positive (the full conversation flows through stdout as
// stream-json, so any text the model outputs is visible here).
if v.get("type").and_then(|t| t.as_str()) == Some("error") {
let raw = v.to_string();
if RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m)) {
rate_out.store(true, Ordering::Relaxed);
}
bus_out.emit(LiveEvent::Note {
text: format!("(non-json) {line}"),
});
}
if let Some(u) = crate::events::TokenUsage::from_assistant_event(&v) {
last_inference = Some(u);
}
if let Some(cost) = crate::events::TokenUsage::from_stream_event(&v) {
// Fallback to `cost` if the turn somehow produced
// a result without any assistant event — keeps the
// ctx badge from going stale on a degenerate turn.
let ctx = last_inference.unwrap_or(cost);
bus_out.record_turn_usage(ctx, cost);
}
// Seed the API-reported context-window from the result
// event's `modelUsage.*.contextWindow` field. This is
// the authoritative per-inference active window used for
// compaction watermarks — it reflects what the model
// actually enforces, which may differ from the Nix
// config (e.g. 200k active window on a 1M cache model).
if let Some(w) = crate::events::TokenUsage::context_window_from_result_event(&v) {
bus_out.set_api_context_window(w);
}
bus_out.observe_stream(&v);
bus_out.emit(LiveEvent::Stream(v));
} else {
// Non-JSON stdout: raw text check is fine here since these
// are claude CLI messages, not conversation content.
if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) {
rate_out.store(true, Ordering::Relaxed);
}
bus_out.emit(LiveEvent::Note {
text: format!("(non-json) {line}"),
});
}
}
});
@ -737,3 +798,114 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
}
Ok((too_long, is_rate_limited, is_auth_failed))
}
#[cfg(test)]
mod tests {
use std::fs;
use std::time::{Duration, SystemTime};
use super::{DirSnapshot, session_refreshed, snapshot_dir};
#[test]
fn snapshot_dir_empty_dir_is_default() {
let dir = tempfile::tempdir().unwrap();
let snap = snapshot_dir(dir.path());
assert_eq!(snap.file_count, 0);
assert!(snap.newest_mtime.is_none());
}
#[test]
fn snapshot_dir_missing_dir_is_default() {
// Defensive: a nonexistent dir must NOT panic. Bind mounts that
// disappear mid-poll (host purge during operator intervention)
// would otherwise crash the harness.
let missing = tempfile::tempdir()
.unwrap()
.path()
.join("never-created-subdir");
let snap = snapshot_dir(&missing);
assert_eq!(snap, DirSnapshot::default());
}
#[test]
fn snapshot_dir_picks_latest_mtime_and_counts_files() {
let dir = tempfile::tempdir().unwrap();
fs::write(dir.path().join("old.json"), b"{}").unwrap();
// Sleep so the second file's mtime is strictly greater than
// the first on filesystems with low timestamp resolution.
std::thread::sleep(Duration::from_millis(20));
let newer_path = dir.path().join("newer.json");
fs::write(&newer_path, b"{}").unwrap();
let snap = snapshot_dir(dir.path());
assert_eq!(snap.file_count, 2);
let newer_meta = fs::metadata(&newer_path).unwrap().modified().unwrap();
assert_eq!(snap.newest_mtime, Some(newer_meta));
}
#[test]
fn session_refreshed_first_login_flips_on_any_file() {
// Empty-dir snapshot → any file appearing means a fresh
// login landed. Pre-#542 semantics for first-time login.
let dir = tempfile::tempdir().unwrap();
let snapshot = snapshot_dir(dir.path());
assert!(!session_refreshed(snapshot, snapshot_dir(dir.path())));
fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
assert!(session_refreshed(snapshot, snapshot_dir(dir.path())));
}
#[test]
fn session_refreshed_stale_creds_dont_flip_immediately() {
// The #542 repro: stale credentials.json already exists at
// entry; wait_for_login must NOT immediately return — it
// would loop straight into another 401-failing turn.
let dir = tempfile::tempdir().unwrap();
fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
let snapshot = snapshot_dir(dir.path());
assert_eq!(snapshot.file_count, 1);
// No change to the file → loop must NOT exit.
assert!(!session_refreshed(snapshot, snapshot_dir(dir.path())));
}
#[test]
fn session_refreshed_after_creds_rewrite_flips() {
// After the stale-creds snapshot, the operator's `/login/code`
// flow lands a refreshed credentials file — its mtime bumps
// strictly past the snapshot and wait_for_login resumes.
let dir = tempfile::tempdir().unwrap();
fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
let snapshot = snapshot_dir(dir.path());
std::thread::sleep(Duration::from_millis(20));
fs::write(dir.path().join("credentials.json"), b"{\"v\":2}").unwrap();
assert!(session_refreshed(snapshot, snapshot_dir(dir.path())));
}
#[test]
fn session_refreshed_snapshot_with_future_mtime_doesnt_flip() {
// Defensive: a snapshot set to a future timestamp (e.g. clock
// skew between snapshot and probe) must keep waiting until a
// file's mtime actually exceeds it, not return on first poll.
let dir = tempfile::tempdir().unwrap();
fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
let snapshot = DirSnapshot {
file_count: 1,
newest_mtime: Some(SystemTime::now() + Duration::from_secs(3600)),
};
assert!(!session_refreshed(snapshot, snapshot_dir(dir.path())));
}
#[test]
fn session_refreshed_count_change_flips_when_mtime_unreadable() {
// Defensive (argus #545 nit): if all files have unreadable
// `meta.modified()` (exotic fs / NFS), newest_mtime stays
// `None` forever — but file_count axis still catches new
// files appearing. Simulated here by forging a snapshot with
// file_count=1 + no mtime, then writing a second file.
let dir = tempfile::tempdir().unwrap();
fs::write(dir.path().join("a"), b"{}").unwrap();
let forged = DirSnapshot { file_count: 1, newest_mtime: None };
fs::write(dir.path().join("b"), b"{}").unwrap();
// Real snapshot has file_count=2, so refresh fires even
// though the mtime axis would be inconclusive.
assert!(session_refreshed(forged, snapshot_dir(dir.path())));
}
}

View file

@ -4,7 +4,7 @@ edition.workspace = true
version.workspace = true
# Render branding/agent-configs.svg → $OUT_DIR/agent-configs.png at
# compile time (#424). build.rs shells out to `rsvg-convert`
# (librsvg, pulled in via flake.nix' naersk nativeBuildInputs); the
# (librsvg, pulled in via flake.nix' crane nativeBuildInputs); the
# baked PNG is included via include_bytes! from forge.rs so no
# raster gets checked into git.
build = "build.rs"

View file

@ -5,7 +5,7 @@
//! artifact.
//!
//! Uses `rsvg-convert` from PATH (librsvg, already available in
//! nixpkgs and added to the naersk derivation's `nativeBuildInputs`
//! nixpkgs and added to the crane derivation's `nativeBuildInputs`
//! in `flake.nix`). For dev builds outside Nix, install librsvg via
//! your system package manager (Debian/Ubuntu: `librsvg2-bin`,
//! macOS: `brew install librsvg`).
@ -43,7 +43,7 @@ fn main() {
"failed to invoke rsvg-convert: {e}\n\
install librsvg (Debian/Ubuntu: librsvg2-bin, macOS: brew install librsvg, \
NixOS: pkgs.librsvg). The Nix derivation already pulls it in via \
flake.nix naersk-lib.buildPackage.nativeBuildInputs.",
flake.nix craneLib.buildPackage.nativeBuildInputs.",
),
}
}

View file

@ -151,21 +151,18 @@ async fn run_approval_schedule_prompt(
approval: hive_sh4re::Approval,
) -> Result<()> {
let result: Result<()> = async {
let payload: hive_sh4re::SchedulePromptPayload =
serde_json::from_str(&approval.commit_ref)
.context("decode SchedulePromptPayload from approval.commit_ref")?;
let payload: hive_sh4re::SchedulePromptPayload = serde_json::from_str(&approval.commit_ref)
.context("decode SchedulePromptPayload from approval.commit_ref")?;
coord
.scheduled_prompts
.submit(crate::scheduled_prompts::NewSchedule {
.submit(&crate::scheduled_prompts::NewSchedule {
owner: approval.agent.clone(),
targets: payload.targets,
body: payload.body,
first_fire_at_unix: payload.first_fire_at_unix,
interval_seconds: payload.interval_seconds,
description: payload.description,
source: crate::scheduled_prompts::ScheduleSource::Approval {
id: approval.id,
},
source: crate::scheduled_prompts::ScheduleSource::Approval { id: approval.id },
})
.map(|_| ())
.context("insert scheduled prompt")
@ -290,9 +287,10 @@ async fn forge_after_first_spawn(coord: &Arc<Coordinator>, agent: &str) {
tracing::warn!(%agent, error = ?e, "forge: ensure_config_repo after first spawn failed");
}
if let Some(core_token) = crate::forge::core_token()
&& let Err(e) = crate::forge::meta_read_access(agent, &core_token).await {
tracing::warn!(%agent, error = ?e, "forge: meta_read_access after first spawn failed");
}
&& let Err(e) = crate::forge::meta_read_access(agent, &core_token).await
{
tracing::warn!(%agent, error = ?e, "forge: meta_read_access after first spawn failed");
}
if let Err(e) = crate::forge::ensure_meta_remote(agent).await {
tracing::warn!(%agent, error = ?e, "forge: ensure_meta_remote after first spawn failed");
}
@ -466,7 +464,7 @@ async fn run_apply_commit(
Err(anyhow::anyhow!("read applied/main: {e:#}")),
None,
is_first_spawn,
)
);
}
};
@ -521,8 +519,7 @@ async fn run_apply_commit(
Ok(a) => a,
Err(e) => {
let _ =
lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha)
.await;
lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await;
let _ = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await;
return (
Err(anyhow::anyhow!("agents_for_meta_listing_with: {e:#}")),
@ -540,8 +537,7 @@ async fn run_apply_commit(
)
.await
{
let _ =
lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await;
let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await;
let _ = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await;
return (
Err(anyhow::anyhow!("meta sync_agents for first spawn: {e:#}")),

View file

@ -251,9 +251,11 @@ impl Approvals {
/// kind / agent / sha. Errors if the approval isn't pending — once
/// it's approved/denied/failed/cancelled, the resolution is final.
pub fn mark_cancelled(&self, id: i64, canceller: &str) -> Result<Approval> {
let mut conn = self.conn.lock().unwrap();
let tx = conn.transaction()?;
let row: Option<(
// Row-shape alias for the SELECT below so we don't trip
// clippy::type_complexity. Order matches the SELECT projection:
// agent, kind, commit_ref, requested_at, status, fetched_sha,
// description.
type CancelLookupRow = (
String,
String,
String,
@ -261,7 +263,10 @@ impl Approvals {
String,
Option<String>,
Option<String>,
)> = tx
);
let mut conn = self.conn.lock().unwrap();
let tx = conn.transaction()?;
let row: Option<CancelLookupRow> = tx
.query_row(
"SELECT agent, kind, commit_ref, requested_at, status, fetched_sha, description
FROM approvals WHERE id = ?1",
@ -326,9 +331,7 @@ impl Approvals {
/// bad row used to make `pending()` / `recent_resolved()` error out
/// wholesale — the dashboard then rendered an empty approvals queue
/// (issue #160, an unhandled `init_config` kind poisoning every read).
fn collect_lenient(
rows: impl Iterator<Item = rusqlite::Result<Approval>>,
) -> Vec<Approval> {
fn collect_lenient(rows: impl Iterator<Item = rusqlite::Result<Approval>>) -> Vec<Approval> {
rows.filter_map(|r| match r {
Ok(a) => Some(a),
Err(e) => {
@ -467,7 +470,12 @@ mod tests {
// status + a "cancelled by <who>" note.
let (_dir, _path, db) = open_temp();
let id = db
.submit_kind("bitburner", ApprovalKind::ApplyCommit, "cafef00d", Some("test"))
.submit_kind(
"bitburner",
ApprovalKind::ApplyCommit,
"cafef00d",
Some("test"),
)
.unwrap();
let row = db.mark_cancelled(id, "manager").expect("cancel");
assert_eq!(row.id, id);

View file

@ -65,7 +65,7 @@ pub fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> bool {
/// can't diverge.
///
/// `queue_entry_id` is `Some(id)` when the rebuild was dispatched from
/// the rebuild_queue worker (lets the function annotate its phase via
/// the `rebuild_queue` worker (lets the function annotate its phase via
/// `coord.set_queue_step`) and `None` when called directly (e.g. the
/// manager-migration nudge in `ensure_manager`).
pub async fn rebuild_agent(
@ -220,7 +220,10 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
let _current_rev = current_flake_rev(&coord.hyperhive_flake).unwrap_or_default();
tracing::info!(agents = containers.len(), "auto-update: queueing all on startup");
tracing::info!(
agents = containers.len(),
"auto-update: queueing all on startup"
);
for container in containers {
let logical = if container == MANAGER_NAME {
Some(MANAGER_NAME.to_owned())

View file

@ -139,6 +139,46 @@ fn is_deliberate_stop(
active.is_some_and(is_op_kind) || recently_cleared.is_some_and(is_op_kind)
}
fn emit_login_transitions(
coord: &Coordinator,
prev: &HashSet<String>,
current: &HashSet<String>,
sub_agents: &[String],
prev_sub_agents: &HashSet<String>,
) {
for agent in current.difference(prev) {
tracing::info!(%agent, "agent logged in");
coord.notify_manager(&hive_sh4re::HelperEvent::LoggedIn {
agent: agent.clone(),
});
}
// Detect transitions into "needs login": an agent that was previously
// logged-in goes unsigned (credentials deleted), OR a brand-new agent
// appears without a session.
//
// prev_needs uses prev_sub_agents (the agent set from the last tick) so
// that a newly-spawned agent — which does not appear in prev_sub_agents —
// is absent from prev_needs even though it's not in prev_logged_in.
// Without this, new agents land in both prev_needs and current_needs and
// the set difference is empty, silently dropping the event.
let prev_needs: HashSet<&str> = prev_sub_agents
.iter()
.map(String::as_str)
.filter(|n| !prev.contains(*n))
.collect();
let current_needs: HashSet<&str> = sub_agents
.iter()
.map(String::as_str)
.filter(|n| !current.contains(*n))
.collect();
for agent in current_needs.difference(&prev_needs) {
tracing::info!(%agent, "agent needs login");
coord.notify_manager(&hive_sh4re::HelperEvent::NeedsLogin {
agent: (*agent).to_owned(),
});
}
}
#[cfg(test)]
mod tests {
use super::*;
@ -186,44 +226,3 @@ mod tests {
}
}
}
fn emit_login_transitions(
coord: &Coordinator,
prev: &HashSet<String>,
current: &HashSet<String>,
sub_agents: &[String],
prev_sub_agents: &HashSet<String>,
) {
for agent in current.difference(prev) {
tracing::info!(%agent, "agent logged in");
coord.notify_manager(&hive_sh4re::HelperEvent::LoggedIn {
agent: agent.clone(),
});
}
// Detect transitions into "needs login": an agent that was previously
// logged-in goes unsigned (credentials deleted), OR a brand-new agent
// appears without a session.
//
// prev_needs uses prev_sub_agents (the agent set from the last tick) so
// that a newly-spawned agent — which does not appear in prev_sub_agents —
// is absent from prev_needs even though it's not in prev_logged_in.
// Without this, new agents land in both prev_needs and current_needs and
// the set difference is empty, silently dropping the event.
let prev_needs: HashSet<&str> = prev_sub_agents
.iter()
.map(String::as_str)
.filter(|n| !prev.contains(*n))
.collect();
let current_needs: HashSet<&str> = sub_agents
.iter()
.map(String::as_str)
.filter(|n| !current.contains(*n))
.collect();
for agent in current_needs.difference(&prev_needs) {
tracing::info!(%agent, "agent needs login");
coord.notify_manager(&hive_sh4re::HelperEvent::NeedsLogin {
agent: (*agent).to_owned(),
});
}
}

View file

@ -76,13 +76,13 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
.route("/op-send", post(post_op_send))
.route("/meta-update", post(post_meta_update))
.route("/api/schedules", get(api_schedules).post(post_schedule_new))
.route(
"/api/schedules/{id}",
axum::routing::patch(patch_schedule),
)
.route("/api/schedules/{id}", axum::routing::patch(patch_schedule))
.route("/api/schedules/{id}/cancel", post(post_schedule_cancel))
.route("/api/schedules/{id}/fire-now", post(post_schedule_fire_now))
.route("/api/rebuild-queue/{id}/cancel", post(post_rebuild_queue_cancel))
.route(
"/api/rebuild-queue/{id}/cancel",
post(post_rebuild_queue_cancel),
)
.route("/dashboard/stream", get(dashboard_stream))
.route("/dashboard/history", get(dashboard_history))
// Anything not matched by the dynamic routes above falls
@ -765,7 +765,14 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
let events: Vec<crate::dashboard_events::DashboardEvent> = messages
.into_iter()
.map(|m| match m {
crate::broker::MessageEvent::Sent { id, from, to, body, at, in_reply_to } => {
crate::broker::MessageEvent::Sent {
id,
from,
to,
body,
at,
in_reply_to,
} => {
let file_refs = scan_validated_paths(&body);
crate::dashboard_events::DashboardEvent::Sent {
seq: 0,
@ -778,7 +785,14 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
file_refs,
}
}
crate::broker::MessageEvent::Delivered { id, from, to, body, at, in_reply_to } => {
crate::broker::MessageEvent::Delivered {
id,
from,
to,
body,
at,
in_reply_to,
} => {
let file_refs = scan_validated_paths(&body);
crate::dashboard_events::DashboardEvent::Delivered {
seq: 0,
@ -1155,8 +1169,8 @@ fn resolve_state_path(
return Err(format!("path not in allow-list: {raw}"));
};
reject_symlinks_below(std::path::Path::new(root), &mapped)?;
let canonical = std::fs::canonicalize(&mapped)
.map_err(|e| format!("{}: {e}", mapped.display()))?;
let canonical =
std::fs::canonicalize(&mapped).map_err(|e| format!("{}: {e}", mapped.display()))?;
if !(canonical.starts_with(AGENTS_ROOT) || canonical.starts_with(SHARED_ROOT)) {
return Err(format!(
"resolved path escapes allow-list: {}",
@ -1174,8 +1188,8 @@ fn resolve_state_path(
));
}
}
let meta = std::fs::metadata(&canonical)
.map_err(|e| format!("stat {}: {e}", canonical.display()))?;
let meta =
std::fs::metadata(&canonical).map_err(|e| format!("stat {}: {e}", canonical.display()))?;
if meta.is_file() {
let mode = meta.permissions().mode();
if mode & 0o004 == 0 {
@ -1313,12 +1327,10 @@ pub(crate) async fn emit_tombstones_snapshot(coord: &Arc<Coordinator>) {
let containers = coord.containers_snapshot().await;
let transient_snapshot = coord.transient_snapshot();
let tombstones = build_tombstone_views(coord, &containers, &transient_snapshot);
coord.emit_dashboard_event(
crate::dashboard_events::DashboardEvent::TombstonesChanged {
seq: coord.next_seq(),
tombstones,
},
);
coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::TombstonesChanged {
seq: coord.next_seq(),
tombstones,
});
}
/// Snapshot meta/flake.lock's root inputs + emit
@ -1326,12 +1338,10 @@ pub(crate) async fn emit_tombstones_snapshot(coord: &Arc<Coordinator>) {
/// (`run_meta_update`, `auto_update::rebuild_agent`).
pub(crate) fn emit_meta_inputs_snapshot(coord: &Coordinator) {
let inputs = read_meta_inputs();
coord.emit_dashboard_event(
crate::dashboard_events::DashboardEvent::MetaInputsChanged {
seq: coord.next_seq(),
inputs,
},
);
coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::MetaInputsChanged {
seq: coord.next_seq(),
inputs,
});
}
/// Scan `body` for path-shaped tokens, validate each against the
@ -1381,9 +1391,7 @@ pub(crate) fn scan_validated_paths(body: &str) -> Vec<String> {
out
}
async fn get_state_file(
axum::extract::Query(q): axum::extract::Query<StateFileQuery>,
) -> Response {
async fn get_state_file(axum::extract::Query(q): axum::extract::Query<StateFileQuery>) -> Response {
const MAX_BYTES: usize = 1 << 20; // 1 MiB
let (canonical, meta) = match resolve_state_path(&q.path) {
Ok(pair) => pair,
@ -1415,11 +1423,18 @@ async fn get_state_file(
return ([("content-type", ct)], bytes).into_response();
}
let truncated = bytes.len() > MAX_BYTES;
let body_bytes = if truncated { &bytes[..MAX_BYTES] } else { &bytes[..] };
let body_bytes = if truncated {
&bytes[..MAX_BYTES]
} else {
&bytes[..]
};
let mut body = String::from_utf8_lossy(body_bytes).into_owned();
if truncated {
use std::fmt::Write as _;
let _ = write!(body, "\n\n--- truncated at {MAX_BYTES} of {size} bytes ---\n");
let _ = write!(
body,
"\n\n--- truncated at {MAX_BYTES} of {size} bytes ---\n"
);
}
([("content-type", "text/plain; charset=utf-8")], body).into_response()
}
@ -1492,7 +1507,7 @@ async fn post_schedule_new(
description: payload.description,
source: crate::scheduled_prompts::ScheduleSource::Operator,
};
match state.coord.scheduled_prompts.submit(new) {
match state.coord.scheduled_prompts.submit(&new) {
Ok(id) => axum::Json(serde_json::json!({"id": id})).into_response(),
Err(e) => error_response(&format!("schedule submit: {e:#}")),
}
@ -1546,6 +1561,12 @@ struct CancelScheduleForm {
}
#[derive(serde::Deserialize, Default)]
#[allow(
clippy::option_option,
reason = "double-Option carries three-state PATCH semantics on the wire \
(missing key = leave alone, JSON null = clear, value = set); \
collapsing to a single Option would lose the 'clear' state"
)]
struct EditScheduleForm {
#[serde(default)]
body: Option<String>,
@ -1671,7 +1692,10 @@ async fn get_agent_links(AxumPath(name): AxumPath<String>) -> Response {
match client.get(&url).send().await {
Ok(resp) if resp.status().is_success() => match resp.json::<serde_json::Value>().await {
Ok(body) => {
let links = body.get("links").cloned().unwrap_or_else(|| serde_json::json!([]));
let links = body
.get("links")
.cloned()
.unwrap_or_else(|| serde_json::json!([]));
axum::Json(links).into_response()
}
Err(e) => {
@ -1852,7 +1876,10 @@ async fn post_op_send(State(state): State<AppState>, Form(form): Form<OpSendForm
.coord
.broadcast_send(hive_sh4re::OPERATOR_RECIPIENT, &body);
if !errors.is_empty() {
return error_response(&format!("op-send broadcast partial fail: {}", errors.join("; ")));
return error_response(&format!(
"op-send broadcast partial fail: {}",
errors.join("; ")
));
}
} else if let Err(e) = state.coord.broker.send(&hive_sh4re::Message {
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
@ -2151,7 +2178,6 @@ fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<Approval> {
.collect()
}
/// Multi-file unified diff between the currently-deployed tree and
/// the proposal for this approval. Runs against the applied repo
/// since the canonical proposal commit lives there (manager-side
@ -2282,4 +2308,3 @@ async fn get_approval_diff(
fn plain_text(body: String) -> Response {
(StatusCode::OK, body).into_response()
}

View file

@ -165,10 +165,7 @@ pub enum DashboardEvent {
/// last one cached on the coordinator. Mutation sites (lifecycle
/// endpoints, `actions::destroy` / approve, `crash_watch`'s poll loop)
/// call the rescan after their work lands.
ContainerStateChanged {
seq: u64,
container: ContainerView,
},
ContainerStateChanged { seq: u64, container: ContainerView },
/// A container that was in the previous snapshot is gone. Clients
/// drop the row by name. Fired alongside any
/// `nixos-container destroy` (operator-driven or otherwise) on the
@ -211,12 +208,9 @@ pub enum DashboardEvent {
/// snapshot-shape rationale as `TombstonesChanged` /
/// `MetaInputsChanged`: the list is small, snapshot semantics avoid
/// the add/remove races a per-row event would have, and the
/// dashboard's grouping (parent_id) is most naturally re-derived
/// dashboard's grouping (`parent_id`) is most naturally re-derived
/// from the full list.
RebuildQueueChanged {
seq: u64,
queue: Vec<QueueEntry>,
},
RebuildQueueChanged { seq: u64, queue: Vec<QueueEntry> },
}
impl DashboardEvent {
@ -259,13 +253,18 @@ mod tests {
/// the `kind` JSON field matches `kind_tag()`. The exhaustive
/// `match` in `kind_tag` already provides compile-time variant
/// coverage — this test is the value-side guard against
/// typos in the snake_case strings vs serde's `rename_all`
/// typos in the `snake_case` strings vs serde's `rename_all`
/// output. `ContainerStateChanged` is omitted from the sample
/// list only because `ContainerView` has no `Default` impl and
/// constructing one inline here is more boilerplate than the
/// test is worth; the variant is still covered by the
/// `kind_tag` match arm.
#[test]
#[allow(
clippy::too_many_lines,
reason = "exhaustive coverage of every DashboardEvent variant — the \
length is the point"
)]
fn kind_tag_matches_serde_kind_field() {
let samples: Vec<DashboardEvent> = vec![
DashboardEvent::Sent {
@ -367,11 +366,7 @@ mod tests {
.get("kind")
.and_then(|k| k.as_str())
.expect("kind field present");
assert_eq!(
ev.kind_tag(),
serde_kind,
"kind_tag() drift on {ev:?}",
);
assert_eq!(ev.kind_tag(), serde_kind, "kind_tag() drift on {ev:?}",);
}
}
}

View file

@ -90,7 +90,11 @@ fn manager_recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
#[allow(clippy::too_many_lines)]
async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResponse {
match req {
ManagerRequest::Send { to, body, in_reply_to } => {
ManagerRequest::Send {
to,
body,
in_reply_to,
} => {
if let Err(message) = crate::limits::check_size("send", body) {
return ManagerResponse::Err { message };
}
@ -195,7 +199,14 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
) {
Ok(id) => {
tracing::info!(%id, %name, "init_config approval queued");
coord.emit_approval_added(id, name, "init_config", None, None, description.clone());
coord.emit_approval_added(
id,
name,
"init_config",
None,
None,
description.clone(),
);
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
@ -302,7 +313,7 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
Err(e) => {
return ManagerResponse::Err {
message: format!("queue update_meta_inputs approval: {e:#}"),
}
};
}
};
tracing::info!(%id, %label, "update_meta_inputs approval queued");
@ -317,7 +328,7 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
ManagerResponse::Ok
}
ManagerRequest::RequestSchedulePrompt(payload) => {
handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload).await
handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload)
}
ManagerRequest::CancelSchedule { id, targets } => {
handle_cancel_schedule(coord, hive_sh4re::MANAGER_AGENT, *id, targets.as_deref())
@ -482,12 +493,13 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
ManagerRequest::SetStatus { text } => {
let path = Coordinator::agent_notes_dir(MANAGER_AGENT).join("hyperhive-status");
let result = if text.trim().is_empty() {
std::fs::remove_file(&path)
.or_else(|e| if e.kind() == std::io::ErrorKind::NotFound {
std::fs::remove_file(&path).or_else(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
Ok(())
} else {
Err(e)
})
}
})
} else {
std::fs::write(&path, format!("{}\n", text.trim()))
};
@ -497,7 +509,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err { message: format!("set_status write failed: {e}") },
Err(e) => ManagerResponse::Err {
message: format!("set_status write failed: {e}"),
},
}
}
ManagerRequest::GetAgentMeta { name } => {
@ -508,7 +522,12 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
// tell (e.g. "iris is down" vs "iris has no status set").
let (status_text, status_set_at, running) =
crate::container_view::read_agent_status_live(target).await;
let role = if target == MANAGER_AGENT { "manager" } else { "agent" }.to_owned();
let role = if target == MANAGER_AGENT {
"manager"
} else {
"agent"
}
.to_owned();
ManagerResponse::AgentMeta {
name: target.to_owned(),
role,
@ -518,16 +537,12 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
status_set_at,
}
}
ManagerRequest::CancelLooseEnd { kind, id } => crate::questions::handle_cancel_loose_end(
coord,
MANAGER_AGENT,
*kind,
*id,
)
.map_or_else(
|message| ManagerResponse::Err { message },
|()| ManagerResponse::Ok,
),
ManagerRequest::CancelLooseEnd { kind, id } => {
crate::questions::handle_cancel_loose_end(coord, MANAGER_AGENT, *kind, *id).map_or_else(
|message| ManagerResponse::Err { message },
|()| ManagerResponse::Ok,
)
}
ManagerRequest::AckTurn => match coord.broker.ack_turn(MANAGER_AGENT) {
Ok(_n) => ManagerResponse::Ok,
Err(e) => ManagerResponse::Err {
@ -706,7 +721,7 @@ async fn submit_apply_commit(
/// inputs (non-empty targets, non-empty body, sane interval) at
/// submit time — the operator should never see a malformed schedule
/// pending approval.
async fn handle_request_schedule_prompt(
fn handle_request_schedule_prompt(
coord: &Arc<Coordinator>,
requester: &str,
payload: &hive_sh4re::SchedulePromptPayload,
@ -731,7 +746,7 @@ async fn handle_request_schedule_prompt(
Err(e) => {
return ManagerResponse::Err {
message: format!("encode SchedulePromptPayload: {e:#}"),
}
};
}
};
let id = match coord.approvals.submit_kind(
@ -744,7 +759,7 @@ async fn handle_request_schedule_prompt(
Err(e) => {
return ManagerResponse::Err {
message: format!("queue schedule_prompt approval: {e:#}"),
}
};
}
};
tracing::info!(
@ -783,12 +798,12 @@ fn handle_cancel_schedule(
Ok(None) => {
return ManagerResponse::Err {
message: format!("schedule {schedule_id} not found"),
}
};
}
Err(e) => {
return ManagerResponse::Err {
message: format!("read schedule {schedule_id}: {e:#}"),
}
};
}
};
if !cancel_authorized(requester, &schedule.owner) {
@ -830,12 +845,12 @@ async fn handle_fire_schedule_now(
Ok(None) => {
return ManagerResponse::Err {
message: format!("schedule {schedule_id} not found"),
}
};
}
Err(e) => {
return ManagerResponse::Err {
message: format!("read schedule {schedule_id}: {e:#}"),
}
};
}
};
if !cancel_authorized(requester, &schedule.owner) {
@ -858,11 +873,16 @@ async fn handle_fire_schedule_now(
/// ownership rules as `CancelSchedule` — the manager can edit
/// schedules it owns + any owned by an agent in its subtree.
/// Forwards the partial payload to
/// `ScheduledPrompts::update` which enforces the cancelled-row
/// + zero-interval validation. Returns `Ok` on a clean update;
/// `ScheduledPrompts::update` which enforces the cancelled-row /
/// 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,
@ -879,12 +899,12 @@ fn handle_edit_schedule(
Ok(None) => {
return ManagerResponse::Err {
message: format!("schedule {schedule_id} not found"),
}
};
}
Err(e) => {
return ManagerResponse::Err {
message: format!("read schedule {schedule_id}: {e:#}"),
}
};
}
};
if !cancel_authorized(requester, &schedule.owner) {

View file

@ -73,7 +73,13 @@ pub async fn sync_agents(
let dir = meta_dir();
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
let new_flake = render_flake(hyperhive_flake, dashboard_port, operator_pronouns, context_window_tokens, agents);
let new_flake = render_flake(
hyperhive_flake,
dashboard_port,
operator_pronouns,
context_window_tokens,
agents,
);
let flake_path = dir.join("flake.nix");
let on_disk = std::fs::read_to_string(&flake_path).unwrap_or_default();
let initial = !dir.join(".git").exists();
@ -308,6 +314,11 @@ fn agent_canonical_inputs(name: &str) -> Vec<&'static str> {
/// Inner render helper accepting a lookup fn so tests can stub the
/// agent flake-lock introspection.
#[allow(
clippy::too_many_lines,
reason = "templated string-builder for the meta flake — the length is one \
contiguous fmt block, splitting it would just hide the shape"
)]
fn render_flake_with_lookup<F>(
hyperhive_flake: &str,
dashboard_port: u16,
@ -404,16 +415,20 @@ where
sorted_tokens.sort_by_key(|(k, _)| k.as_str());
for (key, val) in &sorted_tokens {
let upper_key = key.to_ascii_uppercase();
let _ = writeln!(out, " HIVE_CONTEXT_WINDOW_TOKENS_{upper_key} = \"{val}\";");
let _ = writeln!(
out,
" HIVE_CONTEXT_WINDOW_TOKENS_{upper_key} = \"{val}\";"
);
}
// Forge URL — injected when hive-c0re itself has HIVE_FORGE_URL set
// (the NixOS module derives it from hyperhive.forge.{domain,httpPort}).
// Agents use it in forge_notify to poll Forgejo for PR/review events.
if let Ok(forge_url) = std::env::var("HIVE_FORGE_URL")
&& !forge_url.is_empty() {
let escaped = forge_url.replace('\\', "\\\\").replace('"', "\\\"");
let _ = writeln!(out, " HIVE_FORGE_URL = \"{escaped}\";");
}
&& !forge_url.is_empty()
{
let escaped = forge_url.replace('\\', "\\\\").replace('"', "\\\"");
let _ = writeln!(out, " HIVE_FORGE_URL = \"{escaped}\";");
}
out.push_str(
r#" HYPERHIVE_STATE_DIR = "/agents/${name}/state";
};
@ -450,6 +465,82 @@ where
out
}
async fn git_is_clean(dir: &Path) -> Result<bool> {
let out = lifecycle::git_command()
.current_dir(dir)
.args(["status", "--porcelain"])
.output()
.await
.with_context(|| format!("git status in {}", dir.display()))?;
Ok(out.stdout.iter().all(u8::is_ascii_whitespace))
}
async fn git(dir: &Path, args: &[&str]) -> Result<()> {
let out = lifecycle::git_command()
.current_dir(dir)
.args(args)
.output()
.await
.with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
if !out.status.success() {
bail!(
"git {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
async fn git_commit(dir: &Path, message: &str) -> Result<()> {
git(
dir,
&[
"-c",
&format!("user.name={GIT_NAME}"),
"-c",
&format!("user.email={GIT_EMAIL}"),
"commit",
"-m",
message,
],
)
.await?;
// Best-effort mirror to the bundled forge. No-op when the forge
// isn't seeded (no core token on disk); push failures log a warn
// but don't bubble up — a missing mirror shouldn't fail an
// otherwise successful deploy.
if let Err(e) = crate::forge::push_meta(dir).await {
tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
}
Ok(())
}
async fn nix(dir: &Path, args: &[&str]) -> Result<()> {
// `--extra-experimental-features` belt-and-suspenders for hosts
// that haven't set this in nix.conf. The hyperhive module's
// deploy guide assumes flakes are already enabled, but the cost
// of being defensive is one extra argv each call.
let mut all = vec!["--extra-experimental-features", "nix-command flakes"];
all.extend(args);
let out = Command::new("nix")
.current_dir(dir)
.args(&all)
.output()
.await
.with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))?;
if !out.status.success() {
bail!(
"nix {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
@ -557,79 +648,3 @@ mod tests {
);
}
}
async fn git_is_clean(dir: &Path) -> Result<bool> {
let out = lifecycle::git_command()
.current_dir(dir)
.args(["status", "--porcelain"])
.output()
.await
.with_context(|| format!("git status in {}", dir.display()))?;
Ok(out.stdout.iter().all(u8::is_ascii_whitespace))
}
async fn git(dir: &Path, args: &[&str]) -> Result<()> {
let out = lifecycle::git_command()
.current_dir(dir)
.args(args)
.output()
.await
.with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
if !out.status.success() {
bail!(
"git {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
async fn git_commit(dir: &Path, message: &str) -> Result<()> {
git(
dir,
&[
"-c",
&format!("user.name={GIT_NAME}"),
"-c",
&format!("user.email={GIT_EMAIL}"),
"commit",
"-m",
message,
],
)
.await?;
// Best-effort mirror to the bundled forge. No-op when the forge
// isn't seeded (no core token on disk); push failures log a warn
// but don't bubble up — a missing mirror shouldn't fail an
// otherwise successful deploy.
if let Err(e) = crate::forge::push_meta(dir).await {
tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
}
Ok(())
}
async fn nix(dir: &Path, args: &[&str]) -> Result<()> {
// `--extra-experimental-features` belt-and-suspenders for hosts
// that haven't set this in nix.conf. The hyperhive module's
// deploy guide assumes flakes are already enabled, but the cost
// of being defensive is one extra argv each call.
let mut all = vec!["--extra-experimental-features", "nix-command flakes"];
all.extend(args);
let out = Command::new("nix")
.current_dir(dir)
.args(&all)
.output()
.await
.with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))?;
if !out.status.success() {
bail!(
"nix {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}

View file

@ -70,6 +70,7 @@ pub enum QueueKind {
Spawn,
/// Destroy with `--purge` (real fs work). Not yet routed here; the
/// variant exists so the wire shape doesn't need to change later.
#[allow(dead_code, reason = "wire shape — routed by a future PR")]
Destroy,
}
@ -102,10 +103,11 @@ pub enum QueueSource {
AutoUpdate,
/// Crash recovery path (future use — currently no auto-rebuild on
/// crash, but the variant exists for the imminent feature).
#[allow(dead_code, reason = "wire shape — used by a future feature")]
CrashRecover,
/// Operator approved a pending `Approval` row on the dashboard.
/// `QueueEntry.approval_id` points back at the source row so the
/// worker can fetch the kind-specific payload (commit_ref, inputs,
/// worker can fetch the kind-specific payload (`commit_ref`, inputs,
/// description) before dispatching.
Approval,
}
@ -137,7 +139,10 @@ pub enum QueueState {
impl QueueState {
pub fn is_terminal(self) -> bool {
matches!(self, QueueState::Done | QueueState::Failed | QueueState::Cancelled)
matches!(
self,
QueueState::Done | QueueState::Failed | QueueState::Cancelled
)
}
}
@ -150,7 +155,7 @@ pub struct QueueEntry {
/// so SSE upserts land in place rather than churning the list.
pub id: u64,
/// Target agent name, or the literal `"hyperhive"` for entries
/// (MetaUpdate) that affect the meta flake rather than a single
/// (`MetaUpdate`) that affect the meta flake rather than a single
/// agent.
pub agent: String,
pub kind: QueueKind,
@ -183,8 +188,8 @@ pub struct QueueEntry {
pub inputs: Vec<String>,
/// Source approval row id when this entry was created by an
/// operator-approve POST (`source == Approval`). The worker uses
/// it to re-fetch the kind-specific payload (commit_ref / inputs /
/// description / fetched_sha) and to fire `ApprovalResolved` on
/// it to re-fetch the kind-specific payload (`commit_ref` / inputs /
/// description / `fetched_sha`) and to fire `ApprovalResolved` on
/// completion. `None` for non-approval entries — preserved on
/// the wire that way too.
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -312,7 +317,7 @@ impl RebuildQueue {
// docstring + #365 for why). Approval-driven entries also
// require the approval_id to match so two distinct approvals
// for the same agent never collapse into one queue slot.
for entry in inner.entries.iter_mut() {
for entry in &mut inner.entries {
if entry.state == QueueState::Queued
&& entry.kind == kind
&& entry.agent == agent
@ -320,7 +325,8 @@ impl RebuildQueue {
&& entry.approval_id == approval_id
{
if !entry.reason.contains(&reason) {
entry.reason.push_str(&format!("\nalso requested by: {reason}"));
use std::fmt::Write as _;
let _ = write!(entry.reason, "\nalso requested by: {reason}");
}
return entry.id;
}
@ -372,7 +378,10 @@ impl RebuildQueue {
/// and leaving a stale "in flight" label after a terminal
/// transition would mislead the dashboard render.
pub fn finish(&self, id: u64, state: QueueState, error: Option<String>) {
debug_assert!(state.is_terminal(), "finish() called with non-terminal {state:?}");
debug_assert!(
state.is_terminal(),
"finish() called with non-terminal {state:?}"
);
let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) {
entry.state = state;
@ -422,7 +431,7 @@ impl RebuildQueue {
pub fn cancel_children(&self, parent: u64) -> usize {
let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
let mut count = 0;
for entry in inner.entries.iter_mut() {
for entry in &mut inner.entries {
if entry.parent_id == Some(parent) && entry.state == QueueState::Queued {
entry.state = QueueState::Cancelled;
entry.finished_at = Some(now_unix());
@ -440,13 +449,13 @@ impl RebuildQueue {
/// safely interrupted). Returns true when an entry was cancelled.
pub fn cancel(&self, id: u64) -> bool {
let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) {
if entry.state == QueueState::Queued {
entry.state = QueueState::Cancelled;
entry.finished_at = Some(now_unix());
Self::trim_history(&mut inner);
return true;
}
if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id)
&& entry.state == QueueState::Queued
{
entry.state = QueueState::Cancelled;
entry.finished_at = Some(now_unix());
Self::trim_history(&mut inner);
return true;
}
false
}
@ -533,7 +542,7 @@ pub async fn run_worker(coord: std::sync::Arc<crate::coordinator::Coordinator>)
return;
}
}
_ = coord.rebuild_queue.notify.notified() => {
() = coord.rebuild_queue.notify.notified() => {
// New entry — back to the drain loop.
}
}
@ -556,12 +565,14 @@ async fn dispatch(
crate::actions::run_approval_apply_commit(coord, Some(entry.id), approval_id).await
}
(QueueKind::Rebuild, None) => {
let current_rev = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
.unwrap_or_default();
crate::auto_update::rebuild_agent(coord, &entry.agent, &current_rev, Some(entry.id)).await
let current_rev =
crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default();
crate::auto_update::rebuild_agent(coord, &entry.agent, &current_rev, Some(entry.id))
.await
}
(QueueKind::MetaUpdate, Some(approval_id)) => {
crate::actions::run_approval_update_meta_inputs(coord, Some(entry.id), approval_id).await
crate::actions::run_approval_update_meta_inputs(coord, Some(entry.id), approval_id)
.await
}
(QueueKind::MetaUpdate, None) => run_meta_update(coord, entry).await,
(QueueKind::Spawn, Some(approval_id)) => {
@ -601,7 +612,11 @@ async fn run_meta_update(
) -> anyhow::Result<()> {
let _progress = coord.meta_update_guard();
let inputs = entry.inputs.clone();
tracing::info!(?inputs, parent = entry.id, "rebuild_queue: meta-update starting");
tracing::info!(
?inputs,
parent = entry.id,
"rebuild_queue: meta-update starting"
);
coord.set_queue_step(Some(entry.id), "nix flake update");
let result = if inputs.is_empty() {
crate::meta::lock_update(&[]).await
@ -633,7 +648,7 @@ async fn run_meta_update(
/// Compute which agents a `nix flake update <inputs>` on the meta
/// flake would affect. Used by callers that pre-enqueue cascade
/// `Rebuild` entries at MetaUpdate submission time (issue #347) so the
/// `Rebuild` entries at `MetaUpdate` submission time (issue #347) so the
/// dashboard can render the dependent work alongside its parent before
/// the lock bump actually runs.
///
@ -658,7 +673,8 @@ pub async fn meta_update_cascade_agents(inputs: &[String]) -> Vec<String> {
if c == crate::lifecycle::MANAGER_NAME {
Some(crate::lifecycle::MANAGER_NAME.to_owned())
} else {
c.strip_prefix(crate::lifecycle::AGENT_PREFIX).map(str::to_owned)
c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
.map(str::to_owned)
}
})
.collect()
@ -782,9 +798,11 @@ mod tests {
// Both inputs lists are preserved.
let inputs: Vec<&[String]> = snap.iter().map(|e| e.inputs.as_slice()).collect();
assert!(inputs.iter().any(|i| *i == ["nixpkgs"]));
assert!(inputs
.iter()
.any(|i| *i == ["agent-bitburner/bitburner-agent"]));
assert!(
inputs
.iter()
.any(|i| *i == ["agent-bitburner/bitburner-agent"])
);
}
#[test]

View file

@ -164,6 +164,12 @@ pub struct NewSchedule {
/// "this target is active again"; prior history was already visible
/// at cancel time).
#[derive(Debug, Clone, Default)]
#[allow(
clippy::option_option,
reason = "double-Option carries three-state PATCH semantics: outer None = \
leave alone, Some(None) = clear, Some(Some(v)) = set. \
collapsing to a single Option would lose the 'clear' state"
)]
pub struct UpdateSchedule {
pub body: Option<String>,
pub description: Option<Option<String>>,
@ -200,7 +206,7 @@ impl ScheduledPrompts {
/// Insert a new schedule. Returns the new id. Empty `targets` is
/// rejected — a schedule with no recipients would silently
/// never fan out, masking caller bugs.
pub fn submit(&self, new: NewSchedule) -> Result<i64> {
pub fn submit(&self, new: &NewSchedule) -> Result<i64> {
if new.targets.is_empty() {
bail!("schedule must have at least one target");
}
@ -212,13 +218,13 @@ impl ScheduledPrompts {
created_at_unix, source, description)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
new.owner,
new.body,
&new.owner,
&new.body,
new.interval_seconds.map(i64::try_from).and_then(Result::ok),
new.first_fire_at_unix,
now_unix(),
new.source.to_db_string(),
new.description,
&new.description,
],
)?;
let id = tx.last_insert_rowid();
@ -255,7 +261,7 @@ impl ScheduledPrompts {
/// Every active (non-globally-cancelled) schedule in insert
/// order. Used by the dashboard list view + the cancel-auth
/// check (the latter only needs the header but list() is the
/// check (the latter only needs the header but `list()` is the
/// shared hot path).
pub fn list(&self) -> Result<Vec<Schedule>> {
let conn = self.conn.lock().unwrap();
@ -322,7 +328,7 @@ impl ScheduledPrompts {
/// Advance a recurring schedule's `next_fire_at` to the smallest
/// multiple-of-interval > `from`. Returns the count of skipped
/// cycles (≥ 0); the worker stamps that into the per-row
/// last_result so operators see "caught up from N missed".
/// `last_result` so operators see "caught up from N missed".
///
/// For one-shots (`interval_seconds IS NULL`) this is a no-op
/// at the SQL level; callers should `delete` them after fan-out
@ -488,10 +494,7 @@ impl ScheduledPrompts {
/// id.
pub fn delete(&self, id: i64) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"DELETE FROM scheduled_prompts WHERE id = ?1",
params![id],
)?;
conn.execute("DELETE FROM scheduled_prompts WHERE id = ?1", params![id])?;
Ok(())
}
@ -630,7 +633,7 @@ mod tests {
}
fn submit_one_shot(db: &ScheduledPrompts, fire_at: i64, targets: &[&str]) -> i64 {
db.submit(NewSchedule {
db.submit(&NewSchedule {
owner: "operator".into(),
targets: targets.iter().map(|t| (*t).to_owned()).collect(),
body: "wake".into(),
@ -658,7 +661,7 @@ mod tests {
fn submit_rejects_empty_targets() {
let (_dir, db) = open();
let err = db
.submit(NewSchedule {
.submit(&NewSchedule {
owner: "operator".into(),
targets: Vec::new(),
body: "wake".into(),
@ -688,7 +691,7 @@ mod tests {
let (_dir, db) = open();
// Recurring every 60s, last fire at t=100.
let id = db
.submit(NewSchedule {
.submit(&NewSchedule {
owner: "operator".into(),
targets: vec!["alice".into()],
body: "wake".into(),
@ -710,7 +713,7 @@ mod tests {
fn rearm_advances_one_step_when_caught_up() {
let (_dir, db) = open();
let id = db
.submit(NewSchedule {
.submit(&NewSchedule {
owner: "operator".into(),
targets: vec!["alice".into()],
body: "wake".into(),
@ -742,7 +745,8 @@ mod tests {
fn cancel_targets_auto_cancels_parent_when_last_drops() {
let (_dir, db) = open();
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
db.cancel_targets(id, &["alice".to_owned()]).expect("cancel");
db.cancel_targets(id, &["alice".to_owned()])
.expect("cancel");
let s = db.get(id).expect("get").expect("present");
// Parent still active (bob remains).
assert!(s.cancelled_at_unix.is_none());
@ -775,7 +779,8 @@ mod tests {
fn record_target_result_skips_cancelled_targets() {
let (_dir, db) = open();
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
db.cancel_targets(id, &["alice".to_owned()]).expect("cancel");
db.cancel_targets(id, &["alice".to_owned()])
.expect("cancel");
db.record_target_result(id, "alice", 200, "ok")
.expect("record alice");
db.record_target_result(id, "bob", 200, "ok")
@ -794,7 +799,7 @@ mod tests {
fn update_partial_only_touches_set_fields() {
let (_dir, db) = open();
let id = db
.submit(NewSchedule {
.submit(&NewSchedule {
owner: "operator".into(),
targets: vec!["alice".into()],
body: "old body".into(),
@ -824,7 +829,7 @@ mod tests {
fn update_interval_toggle_recurring_to_one_shot() {
let (_dir, db) = open();
let id = db
.submit(NewSchedule {
.submit(&NewSchedule {
owner: "operator".into(),
targets: vec!["alice".into()],
body: "x".into(),
@ -899,7 +904,7 @@ mod tests {
fn update_clears_description() {
let (_dir, db) = open();
let id = db
.submit(NewSchedule {
.submit(&NewSchedule {
owner: "operator".into(),
targets: vec!["alice".into()],
body: "x".into(),
@ -982,7 +987,8 @@ mod tests {
let (_dir, db) = open();
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
// Record some history on alice, then cancel her.
db.record_target_result(id, "alice", 50, "ok").expect("record");
db.record_target_result(id, "alice", 50, "ok")
.expect("record");
db.update(
id,
UpdateSchedule {
@ -1034,7 +1040,7 @@ mod tests {
fn approval_source_round_trips() {
let (_dir, db) = open();
let id = db
.submit(NewSchedule {
.submit(&NewSchedule {
owner: "manager".into(),
targets: vec!["alice".into()],
body: "wake".into(),

View file

@ -32,7 +32,7 @@
//! the broker send again, so transient errors self-heal.
//! - **one-shots** delete unconditionally after their single
//! fan-out pass; a broker failure on a one-shot is NOT
//! retried (the operator advisory + last_result are the only
//! retried (the operator advisory + `last_result` are the only
//! audit trail).
use std::sync::Arc;
@ -104,7 +104,7 @@ fn tick(coord: &Arc<Coordinator>) {
}
/// Fan out one schedule's body to every active target. Records
/// per-target last_result; advances or reaps the parent row at
/// per-target `last_result`; advances or reaps the parent row at
/// the end depending on whether `interval_seconds` is set.
fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
let known: std::collections::HashSet<String> = known_agents(coord);
@ -118,12 +118,11 @@ fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
// mirrors `to == operator` into its own pane.
if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
let reason = format!("no such agent: {target}");
if let Err(e) = coord.scheduled_prompts.record_target_result(
schedule.id,
target,
now,
&reason,
) {
if let Err(e) =
coord
.scheduled_prompts
.record_target_result(schedule.id, target, now, &reason)
{
tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed");
}
notify_operator_missing_target(coord, schedule, target);
@ -292,7 +291,11 @@ pub async fn fire_now(
if schedule.cancelled_at_unix.is_some() {
anyhow::bail!("schedule {schedule_id} is already cancelled");
}
if !schedule.targets.iter().any(|t| t.cancelled_at_unix.is_none()) {
if !schedule
.targets
.iter()
.any(|t| t.cancelled_at_unix.is_none())
{
anyhow::bail!("schedule {schedule_id} has no active targets");
}
let known = known_agents_async().await;
@ -309,12 +312,11 @@ pub async fn fire_now(
let target = &target_row.target;
if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
let reason = format!("manual fire: no such agent: {target}");
if let Err(e) = coord.scheduled_prompts.record_target_result(
schedule_id,
target,
now,
&reason,
) {
if let Err(e) =
coord
.scheduled_prompts
.record_target_result(schedule_id, target, now, &reason)
{
tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
}
notify_operator_missing_target(coord, &schedule, target);

View file

@ -58,6 +58,11 @@ pub fn read() -> BTreeMap<String, Option<String>> {
/// or absent from the file. Cheap convenience over `read()` for
/// callers that want a single entry.
#[must_use]
#[allow(
dead_code,
reason = "convenience API; callers go through `read()` today, kept for the \
dashboard/manager-server surfaces landing in #361 follow-ups"
)]
pub fn parent_of(name: &str) -> Option<String> {
read().get(name).cloned().flatten()
}
@ -88,10 +93,10 @@ pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool {
false
}
/// Persist the topology map. Sorted JSON output (BTreeMap is sorted by
/// Persist the topology map. Sorted JSON output (`BTreeMap` is sorted by
/// key) keeps git diffs minimal across re-writes. Best-effort —
/// returns an `io::Error` so callers can decide whether a failure
/// should abort their op (sync_agents, RequestSetParent) or just log.
/// should abort their op (`sync_agents`, `RequestSetParent`) or just log.
pub fn write(topology: &BTreeMap<String, Option<String>>) -> std::io::Result<()> {
let path = topology_path();
if let Some(parent) = path.parent() {
@ -111,13 +116,21 @@ pub fn write(topology: &BTreeMap<String, Option<String>>) -> std::io::Result<()>
/// entries — `sync_agents` only adds rows for newly-spawned agents
/// against whatever the operator has configured.
#[must_use]
#[allow(
dead_code,
reason = "kept for the dashboard / RequestSetParent write API landing in \
#361 follow-ups; `sync_agents` does its own seeding today"
)]
pub fn default_seed(agent_names: &[String]) -> BTreeMap<String, Option<String>> {
let mut out = BTreeMap::new();
for name in agent_names {
if name == crate::lifecycle::MANAGER_NAME {
out.insert(name.clone(), None);
} else {
out.insert(name.clone(), Some(crate::lifecycle::MANAGER_NAME.to_owned()));
out.insert(
name.clone(),
Some(crate::lifecycle::MANAGER_NAME.to_owned()),
);
}
}
out
@ -285,8 +298,8 @@ mod tests {
#[test]
fn apply_set_parent_refuses_manager_move() {
let err =
apply_set_parent(&topo_three_level(), crate::lifecycle::MANAGER_NAME, None).unwrap_err();
let err = apply_set_parent(&topo_three_level(), crate::lifecycle::MANAGER_NAME, None)
.unwrap_err();
assert!(err.contains("manager"), "err = {err}");
}

View file

@ -1,14 +1,16 @@
//! `diff <pr> [repo]` — print the unified diff for a PR.
//!
//! By default the hunks for known autogenerated lockfiles
//! (`flake.lock`, `Cargo.lock`, `package-lock.json`, …) are
//! collapsed to a `[<path>: +N -M (autogenerated; pass --full for
//! content)]` placeholder so a `flake.lock` rev bump doesn't drown
//! the human-authored changes in 5 000 lines of lock churn
//! (#222). The header (`diff --git`, `index`, `---`, `+++`,
//! rename / mode metadata) is preserved so the reader can still
//! see WHICH lockfiles changed; the +/- counts give a `diff
//! --stat`-style magnitude (excluding the `@@` hunk header).
//! By default any per-file section whose target path matches a
//! known autogenerated lockfile (`flake.lock`, `Cargo.lock`,
//! `package-lock.json`, …) is collapsed to a single
//! `[<path>: contents changed (+N -M, --full for content)]`
//! line so a `flake.lock` rev bump doesn't drown the human-
//! authored changes in 5 000 lines of lock churn (#222). The
//! per-file git headers (`diff --git`, `index`, `---`, `+++`,
//! and any rename / mode metadata) are suppressed alongside the
//! hunks since the placeholder already carries the file path and
//! the +/- magnitude — the headers add four lines of noise per
//! lockfile without any information the reader can act on.
//! Pass `--full` to dump the unfiltered diff.
use anyhow::Result;
@ -40,13 +42,12 @@ pub fn run(client: &Client, args: Args) -> Result<()> {
Ok(())
}
/// Walk a unified diff line-by-line. For each per-file section
/// whose target path matches a known autogenerated file
/// (`is_autogenerated`), drop every line from the first hunk
/// header (`@@`) onward and emit a single `diff --stat`-style
/// `[file.lock: +N -M (autogenerated, --full for content)]`
/// placeholder before the next file. Non-autogenerated files
/// pass through unchanged.
/// Walk a unified diff line-by-line. When a `diff --git` line
/// names a known autogenerated file (`is_autogenerated`), suppress
/// every line of that per-file section (the git headers AND the
/// hunks) and emit a single `[file.lock: contents changed (+N -M,
/// --full for content)]` line in their place. Non-autogenerated
/// files pass through unchanged.
fn collapse_autogenerated(diff: &str) -> String {
let mut out = String::with_capacity(diff.len());
let mut state: Option<CollapseState> = None;
@ -61,23 +62,23 @@ fn collapse_autogenerated(diff: &str) -> String {
if let Some(p) = path.as_deref()
&& is_autogenerated(p)
{
// Suppress the entire per-file block (headers and
// hunks alike). The placeholder we emit on flush
// carries the path, so the git headers are pure
// noise.
state = Some(CollapseState::new(p.to_owned()));
continue;
}
out.push_str(line);
out.push('\n');
continue;
}
if let Some(s) = state.as_mut() {
// First `@@` marks the boundary between file header and
// hunk content; everything from this point is
// suppressed (and tallied) while `state` is Some.
if !s.in_body && line.starts_with("@@") {
// Inside an autogenerated section. Tally `+`/`-` body
// lines for the placeholder; drop everything else
// (headers, `@@` markers, context lines).
if s.in_body || line.starts_with("@@") {
s.in_body = true;
}
if s.in_body {
// The hunk header `@@` itself counts as a body line
// for the +/ tally only via its descendant content
// lines; skip it for the counters.
if !line.starts_with("@@") {
match line.as_bytes().first() {
Some(b'+') => s.added += 1,
@ -85,8 +86,8 @@ fn collapse_autogenerated(diff: &str) -> String {
_ => {}
}
}
continue;
}
continue;
}
out.push_str(line);
out.push('\n');
@ -97,11 +98,11 @@ fn collapse_autogenerated(diff: &str) -> String {
out
}
/// Per-file accumulator for the collapsed-hunk placeholder. Tracks
/// the file's display name plus `+`/`` line counts (excluding the
/// `@@` hunk header), so the placeholder shows operator-meaningful
/// magnitude instead of "N lines omitted" (which mixed context +
/// added + removed indistinguishably).
/// Per-file accumulator for the collapsed placeholder. Tracks
/// the file's display name plus `+`/`-` line counts (excluding
/// the `@@` hunk header), so the placeholder shows operator-
/// meaningful magnitude instead of "N lines omitted" (which
/// mixed context + added + removed indistinguishably).
struct CollapseState {
path: String,
in_body: bool,
@ -121,7 +122,7 @@ impl CollapseState {
fn placeholder(&self) -> String {
format!(
"[{}: +{} -{} (autogenerated; pass --full for content)]\n",
"[{}: contents changed (+{} -{}, --full for content)]\n",
self.path, self.added, self.removed
)
}
@ -247,18 +248,23 @@ index 3333..4444 100644
+fn main() { println!(\"hi\"); }
";
let out = collapse_autogenerated(diff);
assert!(out.contains("diff --git a/Cargo.lock"));
assert!(out.contains("--- a/Cargo.lock"));
assert!(out.contains("+++ b/Cargo.lock"));
// Per-file git headers for the lockfile are suppressed —
// the placeholder already carries the path, so the four
// header lines are pure noise (4 lines per lockfile, stacks
// fast on a multi-lockfile PR).
assert!(!out.contains("diff --git a/Cargo.lock"));
assert!(!out.contains("--- a/Cargo.lock"));
assert!(!out.contains("+++ b/Cargo.lock"));
assert!(!out.contains("[[package]]"));
// Stat-style placeholder: 1 added (`+name = "new"`), 1
// removed (`-name = "old"`); context line and `@@` header
// don't count.
assert!(
out.contains("[Cargo.lock: +1 -1"),
"expected stat placeholder, got: {out}"
out.contains("[Cargo.lock: contents changed (+1 -1, --full for content)]"),
"expected one-line placeholder, got: {out}"
);
// Non-lockfile file passes through untouched.
assert!(out.contains("diff --git a/src/main.rs"));
assert!(out.contains("fn main() {}"));
assert!(out.contains("println!(\"hi\")"));
}
@ -300,13 +306,17 @@ index 1111..2222 100644
+lock-line-two-bumped
";
let out = collapse_autogenerated(diff);
// README.md (non-lockfile) passes through with its full
// header + hunks intact.
assert!(out.contains("diff --git a/README.md"));
assert!(out.contains("-old"));
assert!(out.contains("+new"));
assert!(out.contains("diff --git a/flake.lock"));
// Lockfile is fully suppressed — no headers, no content.
assert!(!out.contains("diff --git a/flake.lock"));
assert!(!out.contains("lock-line-one"));
assert!(
out.contains("[flake.lock: +1 -1"),
"expected stat placeholder, got: {out}"
out.contains("[flake.lock: contents changed (+1 -1, --full for content)]"),
"expected one-line placeholder, got: {out}"
);
}
@ -331,7 +341,7 @@ index 1111..2222 100644
";
let out = collapse_autogenerated(diff);
assert!(
out.contains("[flake.lock: +4 -1"),
out.contains("[flake.lock: contents changed (+4 -1, --full for content)]"),
"expected +4 -1, got: {out}"
);
}
@ -401,13 +411,14 @@ index 1111..2222 100644
+new
";
let out = collapse_autogenerated(diff);
// Content must be suppressed (lockfile detected).
// Entire per-file block suppressed (lockfile detected).
assert!(!out.contains("diff --git"), "headers leaked: {out}");
assert!(!out.contains("old\n"), "lock content leaked: {out}");
assert!(!out.contains("+new"), "lock content leaked: {out}");
// Placeholder uses the parsed b-side path (without quotes).
assert!(
out.contains("[odd dir/Cargo.lock: +1 -1"),
"expected stat placeholder for quoted-path lockfile, got: {out}"
out.contains("[odd dir/Cargo.lock: contents changed (+1 -1, --full for content)]"),
"expected one-line placeholder for quoted-path lockfile, got: {out}"
);
}
}

View file

@ -23,8 +23,8 @@ pub enum HostRequest {
/// `hive-c0re request-spawn` CLI both go through this. The
/// previously-mirrored manager-side `RequestSpawn` was removed
/// (#442) — managers now go through the two-step `request_init_config`
/// + `request_apply_commit` flow so the spawn captures the manager's
/// customised config.
/// and `request_apply_commit` flow so the spawn captures the
/// manager's customised config.
RequestSpawn { name: String },
/// Stop a managed container (graceful).
Kill { name: String },
@ -752,14 +752,10 @@ pub enum ManagerRequest {
Status,
/// Operator-injected message TO the manager (from the manager's own web
/// UI). Same shape as `AgentRequest::OperatorMsg`.
OperatorMsg {
body: String,
},
OperatorMsg { body: String },
/// Last `limit` messages addressed to the manager, newest-first.
/// Non-mutating; mirror of `AgentRequest::Recent`.
Recent {
limit: u64,
},
Recent { limit: u64 },
/// Initialise a brand-new agent's proposed config repo and queue an
/// approval for the operator to review. On approval hive-c0re seeds
/// `/agents/<name>/config/` with the default `agent.nix` template,
@ -776,23 +772,15 @@ pub enum ManagerRequest {
description: Option<String>,
},
/// Stop a sub-agent (graceful).
Kill {
name: String,
},
Kill { name: String },
/// Start a previously-stopped sub-agent container.
Start {
name: String,
},
Start { name: String },
/// Restart a sub-agent container (stop + start).
Restart {
name: String,
},
Restart { name: String },
/// Rebuild a sub-agent: re-applies the current hyperhive flake +
/// agent.nix, restarts the container. No approval required —
/// it's idempotent and the manager owns its own update cadence.
Update {
name: String,
},
Update { name: String },
/// Submit a config commit for the user to approve. `commit_ref` must
/// be a commit sha (7-40 hex chars, short or full) in the agent's
/// proposed config repo — a branch or tag name is rejected so the
@ -957,8 +945,8 @@ pub enum ManagerRequest {
/// per-target fan-out once immediately without touching
/// `next_fire_at_unix` on recurring schedules; one-shots are
/// consumed by the manual fire. Authorization mirrors
/// `CancelSchedule`: the manager can fire its own schedules
/// + any owned by a sub-agent in its subtree per topology.json;
/// `CancelSchedule`: the manager can fire its own schedules and
/// any owned by a sub-agent in its subtree per topology.json;
/// the operator surface bypasses the check.
FireScheduleNow { id: i64 },
/// Edit an existing schedule's mutable fields (#474). Partial
@ -1013,9 +1001,9 @@ pub struct SchedulePromptPayload {
}
/// Schedule row shape on the wire — mirror of
/// `scheduled_prompts::Schedule` but in the public crate so dashboard
/// + agent surfaces can deserialize without depending on
/// hive-c0re-internal types. Kept structurally identical to the
/// `scheduled_prompts::Schedule` but in the public crate so the
/// dashboard and agent surfaces can deserialize without depending
/// on hive-c0re-internal types. Kept structurally identical to the
/// in-process type; the conversion is field-by-field in
/// `manager_server` / `dashboard`.
#[derive(Debug, Clone, Serialize, Deserialize)]

View file

@ -17,8 +17,32 @@ in
{
# The forge is part of the standard install — hive-c0re mirrors
# every agent's applied config repo into it. On by default; opt out
# with `hyperhive.forge.enable = false`.
imports = [ ./hive-forge.nix ];
# with `hyperhive.forge.enable = false`. hive-matrix is opt-in (off
# by default) and asserts that `hyperhive.domain` is set before it
# can be enabled.
imports = [
./hive-forge.nix
./hive-matrix.nix
];
# Top-level option shared by any hyperhive subsystem that needs a
# stable hostname (matrix server_name today, forge ROOT_URL likely
# next). Type is nullable + default null so existing operator
# configs that don't set it still evaluate; subsystems that
# actually need it (matrix) assert non-null in their own config
# block with a helpful message.
options.hyperhive.domain = lib.mkOption {
type = lib.types.nullOr lib.types.str;
default = null;
example = "darkest.space";
description = ''
Canonical host domain for hyperhive subsystems that need a
stable name (currently: `hyperhive.matrix.serverName` derives
from this when `useSubdomain = true`). No default subsystems
that opt to require it assert non-null in their own config and
fail eval with a helpful message if it's missing.
'';
};
options.services.hive-c0re = {
enable = lib.mkEnableOption "hive-c0re hyperhive coordinator daemon";
@ -171,7 +195,8 @@ in
# serves this via `tower_http::ServeDir` for any path it doesn't
# match against an API/action route.
HIVE_STATIC_DIR = "${cfg.frontend}/dashboard";
} // lib.optionalAttrs config.hyperhive.forge.enable {
}
// lib.optionalAttrs config.hyperhive.forge.enable {
# Agents poll this URL for Forgejo notifications. Derived from
# hyperhive.forge.{domain,httpPort} so it tracks forge config changes.
HIVE_FORGE_URL = "http://${config.hyperhive.forge.domain}:${toString config.hyperhive.forge.httpPort}";

201
nix/modules/hive-matrix.nix Normal file
View file

@ -0,0 +1,201 @@
{
pkgs,
lib,
config,
...
}:
let
cfg = config.hyperhive.matrix;
hyperhiveDomain = config.hyperhive.domain;
effectiveServerName =
if cfg.serverName != null then cfg.serverName else "matrix.${hyperhiveDomain}";
in
{
# Private Matrix homeserver (matrix-tuwunel — the official conduwuit
# successor) for hyperhive agents, wrapped in a nixos-container so it
# doesn't fight any existing `services.matrix-*` the operator may
# already run on the host. Same shape as `nix/modules/hive-forge.nix`:
# shared host netns (`privateNetwork = false`) so agents reach it at
# `http://localhost:<httpPort>` (or via the configured server_name
# for federation), nixos-container only here for state + systemd-unit
# isolation.
#
# Container name `hive-matrix` (not `h-*`) so the lifecycle scanner
# ignores it; operator manages via the standard `nixos-container` CLI.
#
# Persistent state at `/var/lib/nixos-containers/hive-matrix/var/lib/
# matrix-tuwunel/` (survives container restart / host reboot). To
# wipe, destroy the container.
#
# Initial rollout (#548): federation enabled (needed for multi-hive
# swarms; trusted_servers starts empty so no actual federation traffic
# leaves until peers are explicitly listed), registration via admin
# API only, e2ee disabled per operator call (tracked for follow-up at
# #551).
options.hyperhive.matrix = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
description = ''
Run hive-matrix a private matrix-tuwunel homeserver (in a
nixos-container) for hyperhive agents. Off by default while
the integration phases in; flip to `true` once the operator
has set `hyperhive.domain` and is ready to onboard agents.
'';
};
package = lib.mkOption {
type = lib.types.package;
default = pkgs.matrix-tuwunel;
defaultText = lib.literalExpression "pkgs.matrix-tuwunel";
description = ''
matrix-tuwunel package to run inside the container. Defaults
to nixpkgs's `pkgs.matrix-tuwunel`. Override to pin a
specific upstream if you need an unreleased feature.
'';
};
serverName = lib.mkOption {
type = lib.types.nullOr lib.types.str;
default = null;
example = "chat.example.org";
description = ''
Matrix `server_name` the host part of every user ID
(`@argus:<server_name>`) and room ID minted on this
homeserver. CRITICAL: must be stable from day one because
it's embedded irrevocably in the identifiers. Defaults to
`matrix.''${hyperhive.domain}` (always a subdomain keeps
the root domain free for the dashboard or forge). Override
here only if you need a name that doesn't follow the
`matrix.<domain>` shape.
'';
};
httpPort = lib.mkOption {
type = lib.types.port;
default = 8008;
description = ''
TCP port tuwunel serves the matrix client-server API on.
Default 8008 is the matrix-spec well-known port. Sits
outside hyperhive's claimed ranges (dashboard 7000, manager
8000, sub-agents 8100..8999). Federation listens on
`federationPort` separately.
'';
};
openFirewall = lib.mkOption {
type = lib.types.bool;
default = true;
description = ''
Open `httpPort` in the host firewall. Off when the
homeserver should only be reachable from inside the host
(e.g. while bringing the integration up before announcing
it to other hives).
Note: federation (the matrix-spec well-known port 8448) is
intentionally not opened here. tuwunel serves the federation
API on the same `httpPort` as the client-server API by
default; reaching it on 8448 requires either binding tuwunel
to that port explicitly OR a reverse-proxy + `.well-known/
matrix/server` delegation, neither of which lives in this
module. Add that proxy config alongside whatever serves your
dashboard or forge on 443.
'';
};
trustedServers = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [ ];
example = [ "matrix.org" ];
description = ''
List of trusted matrix servers (homeservers whose signing
keys this server will fetch identity-server-style). Empty
by default federation is enabled at the protocol level
but no peer is trusted until listed here, so the homeserver
is effectively closed until the operator declares hive
peers explicitly.
'';
};
maxRequestSize = lib.mkOption {
type = lib.types.str;
default = "20000000";
description = ''
Maximum size in bytes of a single matrix client request body.
Default 20 MB matches the matrix-spec recommendation for
media uploads. String-typed to match tuwunel's TOML config
shape (it accepts both integer + size-with-suffix strings).
'';
};
};
config = lib.mkIf cfg.enable {
# mara on #548: "there is no default, but it is required. add
# assertion." — fail eval with a helpful message rather than
# spawning a homeserver with a bogus server_name we can never
# change later. `hyperhive.domain` is host-wide; matrix derives
# the server_name from it (or from `cfg.serverName` if the
# operator wants to override).
assertions = [
{
assertion = hyperhiveDomain != null || cfg.serverName != null;
message = ''
hyperhive.matrix.enable = true requires either:
- hyperhive.domain set to your host's canonical domain
(recommended; shared with forge / dashboard), or
- hyperhive.matrix.serverName set explicitly.
The matrix server_name is embedded into every user ID and
room ID on this homeserver it cannot be changed later
without losing every account and chat history. Pick a
stable hostname before enabling.
'';
}
];
containers.hive-matrix = {
autoStart = true;
ephemeral = false;
# Share host netns — tuwunel's listeners look exactly like
# host-side services, no port-forward plumbing, and agent
# containers (also host netns) reach it via plain `localhost`.
privateNetwork = false;
config =
{ ... }:
{
system.stateVersion = "26.05";
services.matrix-tuwunel = {
enable = true;
package = cfg.package;
settings.global = {
server_name = effectiveServerName;
address = "0.0.0.0";
port = cfg.httpPort;
max_request_size = cfg.maxRequestSize;
# Federation enabled at the protocol level so swarms
# can be wired up later by extending `trustedServers`
# without a homeserver restart. Empty trusted_servers
# keeps it effectively closed until peers are listed.
allow_federation = true;
trusted_servers = cfg.trustedServers;
# Registration off — operator seeds agent accounts via
# the tuwunel admin API (mirrors the forge pattern;
# see `hive-c0re/src/matrix.rs` once #548 PR 2 lands).
allow_registration = false;
# E2EE disabled in initial rollout per operator call
# (#548) — re-enabling tracked at #551.
allow_encryption = false;
};
};
environment.systemPackages = [ cfg.package ];
};
};
networking.firewall = lib.mkIf cfg.openFirewall {
allowedTCPPorts = [
cfg.httpPort
];
};
};
}