Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c9115bdbf5 | ||
|
|
cc962d0685 | ||
|
|
b20dd32189 | ||
|
|
141764c6eb | ||
|
|
42823a0b22 | ||
|
|
dbd4b7a15c | ||
|
|
2bfa5bc1a8 | ||
|
|
3fedc102cc | ||
|
|
512e9ff09f |
12 changed files with 574 additions and 124 deletions
|
|
@ -68,6 +68,8 @@ hand-maintained per-file tree drifts out of sync with the code.
|
||||||
Pick the doc that matches your task. None depend on the others —
|
Pick the doc that matches your task. None depend on the others —
|
||||||
read them à la carte.
|
read them à la carte.
|
||||||
|
|
||||||
|
- **"How do I bring a fresh hive online (first-run hivectl
|
||||||
|
bootstrap)?"** → [`docs/setup.md`](docs/setup.md).
|
||||||
- **"What does the dashboard look like?"** →
|
- **"What does the dashboard look like?"** →
|
||||||
[`docs/web-ui.md`](docs/web-ui.md) (index; sub-pages:
|
[`docs/web-ui.md`](docs/web-ui.md) (index; sub-pages:
|
||||||
[`shape`](docs/web-ui/shape.md),
|
[`shape`](docs/web-ui/shape.md),
|
||||||
|
|
|
||||||
105
docs/setup.md
Normal file
105
docs/setup.md
Normal file
|
|
@ -0,0 +1,105 @@
|
||||||
|
# First-run setup (fresh-deploy bootstrap)
|
||||||
|
|
||||||
|
How to bring a fresh hyperhive hive online: provision accounts, open
|
||||||
|
the gateway, make matrix reachable, and spawn the first sub-agents.
|
||||||
|
|
||||||
|
Aimed at `ruth` (the root/manager agent) on a fresh deploy, but it's a
|
||||||
|
plain reference doc — read it whenever you need the bootstrap command
|
||||||
|
sequence. All `hivectl` commands below run as **root on the host** (not
|
||||||
|
inside an agent container); the `request_*` steps run from ruth's own
|
||||||
|
turn via the MCP tools.
|
||||||
|
|
||||||
|
## Step-by-step
|
||||||
|
|
||||||
|
### 1 · Forge
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Provision (or refresh) ruth's own forge account — do this first
|
||||||
|
hivectl forge create-user ruth
|
||||||
|
|
||||||
|
# Create a human operator account (prints the token to stdout)
|
||||||
|
hivectl forge create-user mara --password hunter2
|
||||||
|
|
||||||
|
# Provision forge accounts for any sub-agents spawned later
|
||||||
|
hivectl forge create-user <agent>
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2 · Gateway (HTTP Basic auth)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Add an operator login to the gateway (reads password from stdin)
|
||||||
|
echo "hunter2" | hivectl gateway create-user mara --password-stdin
|
||||||
|
|
||||||
|
# List existing users
|
||||||
|
hivectl gateway list-users
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3 · Matrix
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 3a. Ensure the hive-internal admin account exists first
|
||||||
|
hivectl matrix sync-admin
|
||||||
|
|
||||||
|
# 3b. Provision ruth's own matrix account
|
||||||
|
hivectl matrix create-user ruth
|
||||||
|
|
||||||
|
# 3c. Create a human matrix account
|
||||||
|
hivectl matrix create-user mara --password hunter2
|
||||||
|
|
||||||
|
# 3d. Invite the operator to the hive Space (and optionally to rooms)
|
||||||
|
hivectl matrix invite mara
|
||||||
|
hivectl matrix invite @mara:yourserver --room '#hive-chat:yourserver'
|
||||||
|
|
||||||
|
# 3e. Promote the operator to homeserver admin if needed
|
||||||
|
hivectl matrix promote-user mara
|
||||||
|
```
|
||||||
|
|
||||||
|
### 4 · Spawn sub-agents
|
||||||
|
|
||||||
|
Sub-agent creation goes through the approval queue — ruth proposes, the
|
||||||
|
operator approves, the container builds. From ruth's own turn (inside
|
||||||
|
the container, via MCP tools):
|
||||||
|
|
||||||
|
```
|
||||||
|
# Step 1: initialise a new agent's config repo
|
||||||
|
request_init_config(name: "iris")
|
||||||
|
# → operator approves → config_ready event lands in the inbox
|
||||||
|
|
||||||
|
# Step 2: edit /agents/iris/config/agent.nix, commit it, then:
|
||||||
|
request_apply_commit(agent: "iris", commit_ref: "<sha>")
|
||||||
|
# → operator approves → container built + started
|
||||||
|
```
|
||||||
|
|
||||||
|
See [`approvals.md`](approvals.md) for the full two-step flow.
|
||||||
|
|
||||||
|
### 5 · Useful host commands
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Roster: all agents, status, rev, parent, pending reminders
|
||||||
|
hivectl agents list
|
||||||
|
|
||||||
|
# Restart a stuck container (no rebuild)
|
||||||
|
hivectl agents restart <agent>
|
||||||
|
|
||||||
|
# Open a Claude session inside an agent's container
|
||||||
|
hivectl choom <agent>
|
||||||
|
|
||||||
|
# Open hive web surfaces in a browser (or just print the URLs)
|
||||||
|
hivectl open # operator dashboard
|
||||||
|
hivectl open forge # Forgejo
|
||||||
|
hivectl open matrix # Matrix GUI (fluffychat)
|
||||||
|
```
|
||||||
|
|
||||||
|
See [`tools/hivectl.md`](tools/hivectl.md) for every `hivectl` verb.
|
||||||
|
|
||||||
|
## Security notes
|
||||||
|
|
||||||
|
- **No forge admin token is stored in any agent state dir.** Agents
|
||||||
|
hold a regular agent token in their `forge-token` file; sensitive
|
||||||
|
creds (the core token, the matrix admin token) live on the host.
|
||||||
|
- All config changes (`request_apply_commit`) go through operator
|
||||||
|
approval — agents can't unilaterally rebuild containers, by design.
|
||||||
|
See [`boundary.md`](boundary.md) and [`security.md`](security.md).
|
||||||
|
|
||||||
|
Once the hive is running, ruth records anything it needs to remember
|
||||||
|
across restarts in `/agents/ruth/state/notes.md`.
|
||||||
|
|
@ -1100,6 +1100,10 @@ footer .banner-thin {
|
||||||
opacity: 0.75;
|
opacity: 0.75;
|
||||||
}
|
}
|
||||||
.schedules-table-row-cancelled td { opacity: 0.55; }
|
.schedules-table-row-cancelled td { opacity: 0.55; }
|
||||||
|
.schedules-table-row-paused td { opacity: 0.75; }
|
||||||
|
.sched-paused-label { color: var(--yellow); font-size: 0.9em; }
|
||||||
|
.btn-pause-schedule { color: var(--teal); }
|
||||||
|
.btn-resume-schedule { color: var(--green); }
|
||||||
.schedules-table-body-cell {
|
.schedules-table-body-cell {
|
||||||
max-width: 30em;
|
max-width: 30em;
|
||||||
overflow: hidden;
|
overflow: hidden;
|
||||||
|
|
|
||||||
|
|
@ -390,9 +390,10 @@ function renderSchedulesList() {
|
||||||
table.append(renderSchedulesTableHead(agents));
|
table.append(renderSchedulesTableHead(agents));
|
||||||
const tbody = el('tbody', {});
|
const tbody = el('tbody', {});
|
||||||
const sorted = schedulesState.slice().sort((a, b) => {
|
const sorted = schedulesState.slice().sort((a, b) => {
|
||||||
const aDone = a.cancelled_at_unix ? 1 : 0;
|
// Cancelled → bucket 2, paused → bucket 1, active → bucket 0.
|
||||||
const bDone = b.cancelled_at_unix ? 1 : 0;
|
const aOrd = a.cancelled_at_unix ? 2 : a.paused_at_unix ? 1 : 0;
|
||||||
if (aDone !== bDone) return aDone - bDone;
|
const bOrd = b.cancelled_at_unix ? 2 : b.paused_at_unix ? 1 : 0;
|
||||||
|
if (aOrd !== bOrd) return aOrd - bOrd;
|
||||||
return a.next_fire_at_unix - b.next_fire_at_unix;
|
return a.next_fire_at_unix - b.next_fire_at_unix;
|
||||||
});
|
});
|
||||||
for (const s of sorted) {
|
for (const s of sorted) {
|
||||||
|
|
@ -637,8 +638,11 @@ function renderSchedulesTableHead(agents) {
|
||||||
}
|
}
|
||||||
function renderScheduleRow(s, agents) {
|
function renderScheduleRow(s, agents) {
|
||||||
const cancelled = !!s.cancelled_at_unix;
|
const cancelled = !!s.cancelled_at_unix;
|
||||||
|
const paused = !cancelled && !!s.paused_at_unix;
|
||||||
const tr = el('tr', {
|
const tr = el('tr', {
|
||||||
class: 'schedules-table-row' + (cancelled ? ' schedules-table-row-cancelled' : ''),
|
class: 'schedules-table-row'
|
||||||
|
+ (cancelled ? ' schedules-table-row-cancelled' : '')
|
||||||
|
+ (paused ? ' schedules-table-row-paused' : ''),
|
||||||
});
|
});
|
||||||
|
|
||||||
tr.append(el('td', { class: 'meta schedules-table-id' }, '#' + s.id));
|
tr.append(el('td', { class: 'meta schedules-table-id' }, '#' + s.id));
|
||||||
|
|
@ -649,11 +653,20 @@ function renderScheduleRow(s, agents) {
|
||||||
srcKind === 'approval' ? 'approval' : 'operator')));
|
srcKind === 'approval' ? 'approval' : 'operator')));
|
||||||
|
|
||||||
// "next" cell — relative due-in for active schedules, "cancelled"
|
// "next" cell — relative due-in for active schedules, "cancelled"
|
||||||
// for cancelled ones. Both carry the absolute ISO in the title.
|
// for cancelled ones, "paused" for paused ones. Both carry the
|
||||||
|
// absolute ISO in the title.
|
||||||
const nextCell = el('td', { class: 'meta schedules-table-next-col' });
|
const nextCell = el('td', { class: 'meta schedules-table-next-col' });
|
||||||
if (cancelled) {
|
if (cancelled) {
|
||||||
nextCell.title = 'cancelled ' + new Date(s.cancelled_at_unix * 1000).toISOString();
|
nextCell.title = 'cancelled ' + new Date(s.cancelled_at_unix * 1000).toISOString();
|
||||||
nextCell.textContent = 'cancelled';
|
nextCell.textContent = 'cancelled';
|
||||||
|
} else if (paused) {
|
||||||
|
nextCell.title = 'paused since '
|
||||||
|
+ new Date(s.paused_at_unix * 1000).toISOString()
|
||||||
|
+ '\nwould fire at '
|
||||||
|
+ new Date(s.next_fire_at_unix * 1000).toISOString();
|
||||||
|
nextCell.append(
|
||||||
|
el('span', { class: 'sched-paused-label' }, '⏸ paused'),
|
||||||
|
);
|
||||||
} else {
|
} else {
|
||||||
const dueIn = s.next_fire_at_unix - Math.floor(Date.now() / 1000);
|
const dueIn = s.next_fire_at_unix - Math.floor(Date.now() / 1000);
|
||||||
nextCell.title = new Date(s.next_fire_at_unix * 1000).toISOString();
|
nextCell.title = new Date(s.next_fire_at_unix * 1000).toISOString();
|
||||||
|
|
@ -723,10 +736,10 @@ function renderScheduleRow(s, agents) {
|
||||||
tr.append(td);
|
tr.append(td);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Actions cell — fire / edit / cancel-all. Glyph-only to fit a
|
// Actions cell — fire / pause-toggle / edit / cancel-all.
|
||||||
// compact column; the buttons keep their existing colour classes
|
// Glyph-only to fit a compact column; colour classes carry the
|
||||||
// so the visual cue (mauve = fire, yellow = edit, red = cancel)
|
// visual cue: mauve = fire, teal = pause/resume, yellow = edit,
|
||||||
// carries over from the card layout.
|
// red = cancel.
|
||||||
const actionsCell = el('td', { class: 'schedules-table-actions' });
|
const actionsCell = el('td', { class: 'schedules-table-actions' });
|
||||||
if (!cancelled) {
|
if (!cancelled) {
|
||||||
const activeTargets = (s.targets || []).filter((t) => !t.cancelled_at_unix);
|
const activeTargets = (s.targets || []).filter((t) => !t.cancelled_at_unix);
|
||||||
|
|
@ -738,14 +751,28 @@ function renderScheduleRow(s, agents) {
|
||||||
fireBtn.title = isOneShot
|
fireBtn.title = isOneShot
|
||||||
? 'fire once — one-shot, consumed after the manual fire'
|
? 'fire once — one-shot, consumed after the manual fire'
|
||||||
: 'fire once now — recurring; choose whether to reset the next-fire timer';
|
: 'fire once now — recurring; choose whether to reset the next-fire timer';
|
||||||
if (!activeTargets.length) {
|
if (!activeTargets.length || paused) {
|
||||||
fireBtn.disabled = true;
|
fireBtn.disabled = true;
|
||||||
fireBtn.title = 'every target is cancelled — nothing to fire';
|
fireBtn.title = paused
|
||||||
|
? 'resume the schedule first to use fire-now'
|
||||||
|
: 'every target is cancelled — nothing to fire';
|
||||||
}
|
}
|
||||||
fireBtn.addEventListener('click', () =>
|
fireBtn.addEventListener('click', () =>
|
||||||
fireScheduleNow(s.id, isOneShot, activeTargets.map((t) => t.target), fireBtn));
|
fireScheduleNow(s.id, isOneShot, activeTargets.map((t) => t.target), fireBtn));
|
||||||
actionsCell.append(fireBtn);
|
actionsCell.append(fireBtn);
|
||||||
|
|
||||||
|
// Pause / resume toggle — only meaningful for recurring schedules.
|
||||||
|
// One-shots can still be paused (to delay a one-time fire), so we
|
||||||
|
// show the button for both cases.
|
||||||
|
const pauseBtn = el('button', {
|
||||||
|
type: 'button',
|
||||||
|
class: 'btn btn-pause-schedule btn-inline-small' + (paused ? ' btn-resume-schedule' : ''),
|
||||||
|
}, paused ? '▶' : '⏸');
|
||||||
|
pauseBtn.title = paused ? 'resume schedule' : 'pause schedule';
|
||||||
|
pauseBtn.addEventListener('click', () =>
|
||||||
|
paused ? resumeSchedule(s.id) : pauseSchedule(s.id));
|
||||||
|
actionsCell.append(pauseBtn);
|
||||||
|
|
||||||
const editingThis = editingSchedules.has(s.id);
|
const editingThis = editingSchedules.has(s.id);
|
||||||
const editBtn = el('button', {
|
const editBtn = el('button', {
|
||||||
type: 'button',
|
type: 'button',
|
||||||
|
|
@ -1089,6 +1116,37 @@ async function postScheduleCancel(id, targets) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function pauseSchedule(id) {
|
||||||
|
try {
|
||||||
|
const resp = await fetch('/api/schedules/' + encodeURIComponent(id) + '/pause', {
|
||||||
|
method: 'POST',
|
||||||
|
});
|
||||||
|
if (!resp.ok) {
|
||||||
|
const text = await resp.text().catch(() => '');
|
||||||
|
themedToast('pause failed: http ' + resp.status + (text ? '\n\n' + text : ''), { type: 'error' });
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await refreshSchedules();
|
||||||
|
} catch (err) {
|
||||||
|
themedToast('pause failed: ' + err, { type: 'error' });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
async function resumeSchedule(id) {
|
||||||
|
try {
|
||||||
|
const resp = await fetch('/api/schedules/' + encodeURIComponent(id) + '/resume', {
|
||||||
|
method: 'POST',
|
||||||
|
});
|
||||||
|
if (!resp.ok) {
|
||||||
|
const text = await resp.text().catch(() => '');
|
||||||
|
themedToast('resume failed: http ' + resp.status + (text ? '\n\n' + text : ''), { type: 'error' });
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await refreshSchedules();
|
||||||
|
} catch (err) {
|
||||||
|
themedToast('resume failed: ' + err, { type: 'error' });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
export function applySchedulesChanged(ev) {
|
export function applySchedulesChanged(ev) {
|
||||||
schedulesState = (ev.schedules || []).slice();
|
schedulesState = (ev.schedules || []).slice();
|
||||||
renderSchedulesList();
|
renderSchedulesList();
|
||||||
|
|
|
||||||
|
|
@ -63,6 +63,19 @@ const AUTH_FAIL_MARKERS: &[&str] = &[
|
||||||
"Failed to authenticate. API Error: 401",
|
"Failed to authenticate. API Error: 401",
|
||||||
];
|
];
|
||||||
|
|
||||||
|
/// Substring claude-code emits when `--resume <id>` is handed a session id
|
||||||
|
/// that doesn't exist in this cwd's project (stale persisted id, or a crash
|
||||||
|
/// before the first turn ever completed a `.jsonl`). On a hit we clear the
|
||||||
|
/// persisted id so the NEXT turn starts a fresh session and re-captures —
|
||||||
|
/// the agent self-heals instead of failing `--resume` forever.
|
||||||
|
const SESSION_NOT_FOUND_MARKER: &str = "No conversation found with session ID";
|
||||||
|
|
||||||
|
/// Name of the harness-owned file under `paths::harness_dir()` that holds
|
||||||
|
/// the claude session id to resume. Written after every turn with the id
|
||||||
|
/// claude reported on its stream (the id can change across resume turns in
|
||||||
|
/// some claude-code versions, so we always rewrite with the last-seen value).
|
||||||
|
const CLAUDE_SESSION_ID_FILE: &str = "claude-session-id";
|
||||||
|
|
||||||
/// How long to sleep after detecting a rate-limit before re-entering the
|
/// How long to sleep after detecting a rate-limit before re-entering the
|
||||||
/// serve loop. Overridable via `HIVE_RATE_LIMIT_SLEEP_SECS`. Default is
|
/// serve loop. Overridable via `HIVE_RATE_LIMIT_SLEEP_SECS`. Default is
|
||||||
/// 5 minutes — enough for most short-lived throttles; the operator can
|
/// 5 minutes — enough for most short-lived throttles; the operator can
|
||||||
|
|
@ -552,8 +565,11 @@ fn session_refreshed(prev: DirSnapshot, now: DirSnapshot) -> bool {
|
||||||
/// Spawn `claude` for one turn and pump `stream-json` stdout into the
|
/// Spawn `claude` for one turn and pump `stream-json` stdout into the
|
||||||
/// live event bus. Prompt goes over stdin (variadic
|
/// live event bus. Prompt goes over stdin (variadic
|
||||||
/// `--allowedTools`/`--tools` would otherwise eat a trailing positional
|
/// `--allowedTools`/`--tools` would otherwise eat a trailing positional
|
||||||
/// prompt). The session is persistent across turns via `--continue` and
|
/// prompt). The session is persistent across turns via `--resume <id>`
|
||||||
/// claude's in-session auto-compact is disabled via the managed
|
/// against the harness's own captured session id (NOT bare `--continue`,
|
||||||
|
/// which resumes the *latest* session in this cwd and so lets a `choom`
|
||||||
|
/// session hijack the live harness context). claude's in-session
|
||||||
|
/// auto-compact is disabled via the managed
|
||||||
/// settings at `/etc/claude-code/managed-settings.json` so it doesn't
|
/// settings at `/etc/claude-code/managed-settings.json` so it doesn't
|
||||||
/// stall mid-turn — hyperhive owns compaction.
|
/// stall mid-turn — hyperhive owns compaction.
|
||||||
pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
|
pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
|
||||||
|
|
@ -624,16 +640,32 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
|
||||||
const STDERR_TAIL_LINES: usize = 20;
|
const STDERR_TAIL_LINES: usize = 20;
|
||||||
let model = bus.model();
|
let model = bus.model();
|
||||||
let effort = bus.effort();
|
let effort = bus.effort();
|
||||||
let resume = !bus.take_skip_continue();
|
// Resolve which claude session to resume. We NEVER pass bare
|
||||||
if !resume {
|
// `--continue`: that resumes the latest session in this cwd, which a
|
||||||
// Flag the fresh session so the bin loop mints a new `sessions`
|
// `choom` invocation (same cwd) can hijack, wiping the harness context.
|
||||||
// row + stamps its id onto this turn's stats (and subsequent
|
// Instead we `--resume <id>` against the id claude reported on a prior
|
||||||
// turns until the next fresh start).
|
// turn, persisted under `harness_dir()/claude-session-id`.
|
||||||
|
let persist_path = crate::paths::harness_dir().join(CLAUDE_SESSION_ID_FILE);
|
||||||
|
let resume_id: Option<String> = if bus.take_skip_continue() {
|
||||||
|
// Fresh session requested: mint a new one (no --resume). Flag it so
|
||||||
|
// the bin loop mints a new `sessions` row + stamps its id onto this
|
||||||
|
// turn's stats. Drop any stale persisted id — the new id claude
|
||||||
|
// reports this turn is captured + written below.
|
||||||
bus.mark_fresh_session();
|
bus.mark_fresh_session();
|
||||||
|
let _ = std::fs::remove_file(&persist_path);
|
||||||
bus.emit(LiveEvent::Note {
|
bus.emit(LiveEvent::Note {
|
||||||
text: "fresh session (--continue suppressed for this turn)".into(),
|
text: "fresh session (continue suppressed for this turn)".into(),
|
||||||
});
|
});
|
||||||
}
|
None
|
||||||
|
} else {
|
||||||
|
// Continue: resume OUR captured id. Absent (first turn / just
|
||||||
|
// self-healed from a stale id) → fall through to a fresh session
|
||||||
|
// and capture the new id below.
|
||||||
|
match std::fs::read_to_string(&persist_path) {
|
||||||
|
Ok(s) if !s.trim().is_empty() => Some(s.trim().to_string()),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
};
|
||||||
let mut cmd = Command::new("claude");
|
let mut cmd = Command::new("claude");
|
||||||
// Spawn inside the agent's state dir so relative paths in tool calls
|
// Spawn inside the agent's state dir so relative paths in tool calls
|
||||||
// (Read foo.md, Bash ls, Write notes.md) land in the durable dir
|
// (Read foo.md, Bash ls, Write notes.md) land in the durable dir
|
||||||
|
|
@ -651,8 +683,8 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
|
||||||
.arg(&model)
|
.arg(&model)
|
||||||
.arg("--effort")
|
.arg("--effort")
|
||||||
.arg(&effort);
|
.arg(&effort);
|
||||||
if resume {
|
if let Some(id) = &resume_id {
|
||||||
cmd.arg("--continue");
|
cmd.arg("--resume").arg(id);
|
||||||
}
|
}
|
||||||
cmd.arg("--system-prompt-file").arg(&files.system_prompt);
|
cmd.arg("--system-prompt-file").arg(&files.system_prompt);
|
||||||
cmd.arg("--mcp-config")
|
cmd.arg("--mcp-config")
|
||||||
|
|
@ -679,12 +711,21 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
|
||||||
let prompt_too_long = Arc::new(AtomicBool::new(false));
|
let prompt_too_long = Arc::new(AtomicBool::new(false));
|
||||||
let rate_limited = Arc::new(AtomicBool::new(false));
|
let rate_limited = Arc::new(AtomicBool::new(false));
|
||||||
let auth_failed = Arc::new(AtomicBool::new(false));
|
let auth_failed = Arc::new(AtomicBool::new(false));
|
||||||
|
// `--resume` against a stale/missing id: clear the persist file so the
|
||||||
|
// next turn self-heals into a fresh session.
|
||||||
|
let session_not_found = Arc::new(AtomicBool::new(false));
|
||||||
|
// Last `session_id` claude reported on its stream this turn; persisted
|
||||||
|
// after the child exits so the next turn `--resume`s it.
|
||||||
|
let session_id_seen = Arc::new(Mutex::new(None::<String>));
|
||||||
let flag_out = prompt_too_long.clone();
|
let flag_out = prompt_too_long.clone();
|
||||||
let flag_err = prompt_too_long.clone();
|
let flag_err = prompt_too_long.clone();
|
||||||
let rate_out = rate_limited.clone();
|
let rate_out = rate_limited.clone();
|
||||||
let rate_err = rate_limited.clone();
|
let rate_err = rate_limited.clone();
|
||||||
let auth_out = auth_failed.clone();
|
let auth_out = auth_failed.clone();
|
||||||
let auth_err = auth_failed.clone();
|
let auth_err = auth_failed.clone();
|
||||||
|
let notfound_out = session_not_found.clone();
|
||||||
|
let notfound_err = session_not_found.clone();
|
||||||
|
let session_id_out = session_id_seen.clone();
|
||||||
let bus_out = bus.clone();
|
let bus_out = bus.clone();
|
||||||
let bus_err = bus.clone();
|
let bus_err = bus.clone();
|
||||||
let pump_stdout = tokio::spawn(async move {
|
let pump_stdout = tokio::spawn(async move {
|
||||||
|
|
@ -712,7 +753,19 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
|
||||||
if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
|
if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
|
||||||
auth_out.store(true, Ordering::Relaxed);
|
auth_out.store(true, Ordering::Relaxed);
|
||||||
}
|
}
|
||||||
|
if line.contains(SESSION_NOT_FOUND_MARKER) {
|
||||||
|
notfound_out.store(true, Ordering::Relaxed);
|
||||||
|
}
|
||||||
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) {
|
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) {
|
||||||
|
// Track the session id claude reports (init + result events
|
||||||
|
// both carry it). Persisted after exit so the next turn
|
||||||
|
// `--resume`s it; re-captured each turn since the id can
|
||||||
|
// change across resumes in some claude-code versions.
|
||||||
|
if let Some(sid) = v.get("session_id").and_then(|s| s.as_str())
|
||||||
|
&& !sid.is_empty()
|
||||||
|
{
|
||||||
|
*session_id_out.lock().unwrap() = Some(sid.to_string());
|
||||||
|
}
|
||||||
// Rate-limit detection: only fire on JSON `error` events,
|
// Rate-limit detection: only fire on JSON `error` events,
|
||||||
// not on arbitrary text content. An agent discussing a past
|
// not on arbitrary text content. An agent discussing a past
|
||||||
// rate limit in its response would otherwise trigger a false
|
// rate limit in its response would otherwise trigger a false
|
||||||
|
|
@ -779,6 +832,9 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
|
||||||
if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
|
if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) {
|
||||||
auth_err.store(true, Ordering::Relaxed);
|
auth_err.store(true, Ordering::Relaxed);
|
||||||
}
|
}
|
||||||
|
if line.contains(SESSION_NOT_FOUND_MARKER) {
|
||||||
|
notfound_err.store(true, Ordering::Relaxed);
|
||||||
|
}
|
||||||
// Mirror to journald so post-mortems work without the web UI
|
// Mirror to journald so post-mortems work without the web UI
|
||||||
// or the events sqlite. The bus event is what the dashboard
|
// or the events sqlite. The bus event is what the dashboard
|
||||||
// renders; the tracing line is what `journalctl -M <c> -b`
|
// renders; the tracing line is what `journalctl -M <c> -b`
|
||||||
|
|
@ -801,6 +857,18 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
|
||||||
let too_long = prompt_too_long.load(Ordering::Relaxed);
|
let too_long = prompt_too_long.load(Ordering::Relaxed);
|
||||||
let is_rate_limited = rate_limited.load(Ordering::Relaxed);
|
let is_rate_limited = rate_limited.load(Ordering::Relaxed);
|
||||||
let is_auth_failed = auth_failed.load(Ordering::Relaxed);
|
let is_auth_failed = auth_failed.load(Ordering::Relaxed);
|
||||||
|
// Session-id bookkeeping. On a stale/missing `--resume` id, drop the
|
||||||
|
// persist file so the next turn starts fresh and self-heals. Otherwise
|
||||||
|
// rewrite it with the id claude reported this turn (handles the id
|
||||||
|
// changing across resumes in some claude-code versions).
|
||||||
|
if session_not_found.load(Ordering::Relaxed) {
|
||||||
|
let _ = std::fs::remove_file(&persist_path);
|
||||||
|
} else if let Some(sid) = session_id_seen.lock().unwrap().clone() {
|
||||||
|
if let Some(parent) = persist_path.parent() {
|
||||||
|
let _ = std::fs::create_dir_all(parent);
|
||||||
|
}
|
||||||
|
let _ = std::fs::write(&persist_path, sid);
|
||||||
|
}
|
||||||
if !status.success() && !too_long && !is_rate_limited && !is_auth_failed {
|
if !status.success() && !too_long && !is_rate_limited && !is_auth_failed {
|
||||||
let tail = stderr_tail.lock().unwrap();
|
let tail = stderr_tail.lock().unwrap();
|
||||||
if tail.is_empty() {
|
if tail.is_empty() {
|
||||||
|
|
|
||||||
|
|
@ -131,6 +131,14 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
||||||
"/api/schedules/{id}/cancel",
|
"/api/schedules/{id}/cancel",
|
||||||
post(schedules::post_schedule_cancel),
|
post(schedules::post_schedule_cancel),
|
||||||
)
|
)
|
||||||
|
.route(
|
||||||
|
"/api/schedules/{id}/pause",
|
||||||
|
post(schedules::post_schedule_pause),
|
||||||
|
)
|
||||||
|
.route(
|
||||||
|
"/api/schedules/{id}/resume",
|
||||||
|
post(schedules::post_schedule_resume),
|
||||||
|
)
|
||||||
.route(
|
.route(
|
||||||
"/api/schedules/{id}/fire-now",
|
"/api/schedules/{id}/fire-now",
|
||||||
post(schedules::post_schedule_fire_now),
|
post(schedules::post_schedule_fire_now),
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,8 @@ use axum::{
|
||||||
|
|
||||||
use problem_details::ProblemDetails;
|
use problem_details::ProblemDetails;
|
||||||
|
|
||||||
|
use crate::scheduled_prompts::ScheduleNotFoundOrCancelled;
|
||||||
|
|
||||||
use super::{AppState, error_problem, error_response};
|
use super::{AppState, error_problem, error_response};
|
||||||
|
|
||||||
/// `GET /api/schedules` — snapshot of every schedule for the
|
/// `GET /api/schedules` — snapshot of every schedule for the
|
||||||
|
|
@ -228,6 +230,50 @@ pub(super) async fn patch_schedule(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// `POST /api/schedules/{id}/pause` — pause a schedule so the worker
|
||||||
|
/// skips it until explicitly resumed. Idempotent; no-op on an already-
|
||||||
|
/// paused row. Returns 404 when the schedule is cancelled or not found.
|
||||||
|
pub(super) async fn post_schedule_pause(
|
||||||
|
State(state): State<AppState>,
|
||||||
|
AxumPath(id): AxumPath<i64>,
|
||||||
|
) -> Response {
|
||||||
|
match state.coord.scheduled_prompts.pause(id) {
|
||||||
|
Ok(()) => {
|
||||||
|
state.coord.emit_schedules_snapshot();
|
||||||
|
(StatusCode::OK, "ok").into_response()
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
if e.downcast_ref::<ScheduleNotFoundOrCancelled>().is_some() {
|
||||||
|
(StatusCode::NOT_FOUND, format!("{e}")).into_response()
|
||||||
|
} else {
|
||||||
|
error_response(&format!("pause schedule {id}: {e:#}"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `POST /api/schedules/{id}/resume` — resume a paused schedule.
|
||||||
|
/// Idempotent; no-op on an already-active row. Returns 404 when the
|
||||||
|
/// schedule is cancelled or not found.
|
||||||
|
pub(super) async fn post_schedule_resume(
|
||||||
|
State(state): State<AppState>,
|
||||||
|
AxumPath(id): AxumPath<i64>,
|
||||||
|
) -> Response {
|
||||||
|
match state.coord.scheduled_prompts.resume(id) {
|
||||||
|
Ok(()) => {
|
||||||
|
state.coord.emit_schedules_snapshot();
|
||||||
|
(StatusCode::OK, "ok").into_response()
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
if e.downcast_ref::<ScheduleNotFoundOrCancelled>().is_some() {
|
||||||
|
(StatusCode::NOT_FOUND, format!("{e}")).into_response()
|
||||||
|
} else {
|
||||||
|
error_response(&format!("resume schedule {id}: {e:#}"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// `POST /api/schedules/{id}/cancel` — operator-side cancel
|
/// `POST /api/schedules/{id}/cancel` — operator-side cancel
|
||||||
/// (whole schedule when no `targets` field, partial when one is
|
/// (whole schedule when no `targets` field, partial when one is
|
||||||
/// provided). Operator bypasses the topology check; the manager
|
/// provided). Operator bypasses the topology check; the manager
|
||||||
|
|
|
||||||
|
|
@ -139,12 +139,17 @@ fn agent_email(name: &str) -> String {
|
||||||
/// All Forgejo API calls that don't shell out to `forgejo admin` go
|
/// All Forgejo API calls that don't shell out to `forgejo admin` go
|
||||||
/// through here — one place for auth header, content-type, error
|
/// through here — one place for auth header, content-type, error
|
||||||
/// propagation, and the shared reqwest Client.
|
/// propagation, and the shared reqwest Client.
|
||||||
|
/// Returns the response status **and body**. The body lets callers log
|
||||||
|
/// *why* Forgejo rejected a request (e.g. the validation message on a
|
||||||
|
/// 422); status-only callers just bind `(status, _)`. Body read is
|
||||||
|
/// best-effort — a read error yields an empty string rather than
|
||||||
|
/// failing the whole call.
|
||||||
async fn forge_http(
|
async fn forge_http(
|
||||||
method: reqwest::Method,
|
method: reqwest::Method,
|
||||||
url: &str,
|
url: &str,
|
||||||
token: &str,
|
token: &str,
|
||||||
body: &str,
|
body: &str,
|
||||||
) -> Result<StatusCode> {
|
) -> Result<(StatusCode, String)> {
|
||||||
let client = reqwest::Client::new();
|
let client = reqwest::Client::new();
|
||||||
let resp = client
|
let resp = client
|
||||||
.request(method, url)
|
.request(method, url)
|
||||||
|
|
@ -154,7 +159,9 @@ async fn forge_http(
|
||||||
.send()
|
.send()
|
||||||
.await
|
.await
|
||||||
.with_context(|| format!("forge HTTP request to {url}"))?;
|
.with_context(|| format!("forge HTTP request to {url}"))?;
|
||||||
Ok(resp.status())
|
let status = resp.status();
|
||||||
|
let text = resp.text().await.unwrap_or_default();
|
||||||
|
Ok((status, text))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Ensure a forgejo user named `name` exists. Idempotent: forgejo
|
/// Ensure a forgejo user named `name` exists. Idempotent: forgejo
|
||||||
|
|
@ -251,14 +258,14 @@ async fn ensure_user_email(name: &str) {
|
||||||
let body = format!(r#"{{"email":"{email}","login_name":"{name}","source_id":0}}"#);
|
let body = format!(r#"{{"email":"{email}","login_name":"{name}","source_id":0}}"#);
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/admin/users/{name}");
|
let url = format!("{FORGE_HTTP}/api/v1/admin/users/{name}");
|
||||||
match forge_http(reqwest::Method::PATCH, &url, &token, &body).await {
|
match forge_http(reqwest::Method::PATCH, &url, &token, &body).await {
|
||||||
Ok(status) if status.is_success() => {
|
Ok((status, _)) if status.is_success() => {
|
||||||
if let Some(parent) = marker.parent() {
|
if let Some(parent) = marker.parent() {
|
||||||
std::fs::create_dir_all(parent).ok();
|
std::fs::create_dir_all(parent).ok();
|
||||||
}
|
}
|
||||||
std::fs::write(&marker, "").ok();
|
std::fs::write(&marker, "").ok();
|
||||||
tracing::info!(%name, %email, "forge: user email aligned");
|
tracing::info!(%name, %email, "forge: user email aligned");
|
||||||
}
|
}
|
||||||
Ok(status) if status == reqwest::StatusCode::FORBIDDEN => {
|
Ok((status, _)) if status == reqwest::StatusCode::FORBIDDEN => {
|
||||||
// Core token missing admin scope — see
|
// Core token missing admin scope — see
|
||||||
// `docs/forge.md::Token scopes` migration note.
|
// `docs/forge.md::Token scopes` migration note.
|
||||||
tracing::warn!(
|
tracing::warn!(
|
||||||
|
|
@ -267,7 +274,7 @@ async fn ensure_user_email(name: &str) {
|
||||||
Delete {CORE_TOKEN_PATH} and restart hive-c0re to re-mint with the new scopes."
|
Delete {CORE_TOKEN_PATH} and restart hive-c0re to re-mint with the new scopes."
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
Ok(status) => {
|
Ok((status, _)) => {
|
||||||
tracing::warn!(%name, %email, %status, "forge: PATCH user email returned non-success");
|
tracing::warn!(%name, %email, %status, "forge: PATCH user email returned non-success");
|
||||||
}
|
}
|
||||||
Err(e) => tracing::warn!(%name, error = %e, "forge: PATCH user email transport error"),
|
Err(e) => tracing::warn!(%name, error = %e, "forge: PATCH user email transport error"),
|
||||||
|
|
@ -302,21 +309,21 @@ async fn ensure_repo_creation_disabled(name: &str) {
|
||||||
let body = format!(r#"{{"login_name":"{name}","source_id":0,"max_repo_creation":0}}"#);
|
let body = format!(r#"{{"login_name":"{name}","source_id":0,"max_repo_creation":0}}"#);
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/admin/users/{name}");
|
let url = format!("{FORGE_HTTP}/api/v1/admin/users/{name}");
|
||||||
match forge_http(reqwest::Method::PATCH, &url, &token, &body).await {
|
match forge_http(reqwest::Method::PATCH, &url, &token, &body).await {
|
||||||
Ok(status) if status.is_success() => {
|
Ok((status, _)) if status.is_success() => {
|
||||||
if let Some(parent) = marker.parent() {
|
if let Some(parent) = marker.parent() {
|
||||||
std::fs::create_dir_all(parent).ok();
|
std::fs::create_dir_all(parent).ok();
|
||||||
}
|
}
|
||||||
std::fs::write(&marker, "").ok();
|
std::fs::write(&marker, "").ok();
|
||||||
tracing::info!(%name, "forge: disabled direct repo creation (max_repo_creation=0)");
|
tracing::info!(%name, "forge: disabled direct repo creation (max_repo_creation=0)");
|
||||||
}
|
}
|
||||||
Ok(status) if status == reqwest::StatusCode::FORBIDDEN => {
|
Ok((status, _)) if status == reqwest::StatusCode::FORBIDDEN => {
|
||||||
tracing::warn!(
|
tracing::warn!(
|
||||||
%name, %status,
|
%name, %status,
|
||||||
"forge: PATCH max_repo_creation forbidden — core token likely missing admin scope. \
|
"forge: PATCH max_repo_creation forbidden — core token likely missing admin scope. \
|
||||||
Delete {CORE_TOKEN_PATH} and restart hive-c0re to re-mint with the new scopes."
|
Delete {CORE_TOKEN_PATH} and restart hive-c0re to re-mint with the new scopes."
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
Ok(status) => {
|
Ok((status, _)) => {
|
||||||
tracing::warn!(%name, %status, "forge: PATCH max_repo_creation returned non-success");
|
tracing::warn!(%name, %status, "forge: PATCH max_repo_creation returned non-success");
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
|
@ -444,7 +451,7 @@ async fn ensure_core_avatar(token: &str) -> Result<()> {
|
||||||
base64::engine::general_purpose::STANDARD.encode(&png_bytes),
|
base64::engine::general_purpose::STANDARD.encode(&png_bytes),
|
||||||
);
|
);
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/admin/users/core/avatar");
|
let url = format!("{FORGE_HTTP}/api/v1/admin/users/core/avatar");
|
||||||
let status = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
let (status, _) = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
||||||
if !status.is_success() {
|
if !status.is_success() {
|
||||||
anyhow::bail!("set core avatar: HTTP {status}");
|
anyhow::bail!("set core avatar: HTTP {status}");
|
||||||
}
|
}
|
||||||
|
|
@ -475,7 +482,7 @@ async fn ensure_config_org_avatar(token: &str) -> Result<()> {
|
||||||
base64::engine::general_purpose::STANDARD.encode(&png_bytes),
|
base64::engine::general_purpose::STANDARD.encode(&png_bytes),
|
||||||
);
|
);
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/orgs/{CONFIG_ORG}/avatar");
|
let url = format!("{FORGE_HTTP}/api/v1/orgs/{CONFIG_ORG}/avatar");
|
||||||
let status = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
let (status, _) = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
||||||
if !status.is_success() {
|
if !status.is_success() {
|
||||||
anyhow::bail!("set {CONFIG_ORG} avatar: HTTP {status}");
|
anyhow::bail!("set {CONFIG_ORG} avatar: HTTP {status}");
|
||||||
}
|
}
|
||||||
|
|
@ -528,7 +535,7 @@ fn classify_core_token_status(status: StatusCode) -> CoreTokenCheck {
|
||||||
async fn check_core_token(token: &str) -> CoreTokenCheck {
|
async fn check_core_token(token: &str) -> CoreTokenCheck {
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/user");
|
let url = format!("{FORGE_HTTP}/api/v1/user");
|
||||||
match forge_http(reqwest::Method::GET, &url, token, "").await {
|
match forge_http(reqwest::Method::GET, &url, token, "").await {
|
||||||
Ok(status) => classify_core_token_status(status),
|
Ok((status, _)) => classify_core_token_status(status),
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::debug!(
|
tracing::debug!(
|
||||||
error = %e,
|
error = %e,
|
||||||
|
|
@ -590,7 +597,8 @@ fn repo_body_public(name: &str) -> String {
|
||||||
/// created as private on an older deployment.
|
/// created as private on an older deployment.
|
||||||
async fn set_repo_public(owner: &str, repo: &str, token: &str) -> Result<()> {
|
async fn set_repo_public(owner: &str, repo: &str, token: &str) -> Result<()> {
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/repos/{owner}/{repo}");
|
let url = format!("{FORGE_HTTP}/api/v1/repos/{owner}/{repo}");
|
||||||
let status = forge_http(reqwest::Method::PATCH, &url, token, r#"{"private":false}"#).await?;
|
let (status, _) =
|
||||||
|
forge_http(reqwest::Method::PATCH, &url, token, r#"{"private":false}"#).await?;
|
||||||
match status.as_u16() {
|
match status.as_u16() {
|
||||||
200 => {
|
200 => {
|
||||||
tracing::debug!(%owner, %repo, "forge: repo set to public");
|
tracing::debug!(%owner, %repo, "forge: repo set to public");
|
||||||
|
|
@ -615,7 +623,7 @@ async fn ensure_org_repo_public(org: &str, name: &str, token: &str) -> Result<()
|
||||||
/// (HTTP 409 / 422) into success. `label` is `<owner>/<name>` — purely
|
/// (HTTP 409 / 422) into success. `label` is `<owner>/<name>` — purely
|
||||||
/// for log + error context.
|
/// for log + error context.
|
||||||
async fn create_repo(url: &str, body: &str, token: &str, label: &str) -> Result<()> {
|
async fn create_repo(url: &str, body: &str, token: &str, label: &str) -> Result<()> {
|
||||||
let status = forge_http(reqwest::Method::POST, url, token, body).await?;
|
let (status, _) = forge_http(reqwest::Method::POST, url, token, body).await?;
|
||||||
match status.as_u16() {
|
match status.as_u16() {
|
||||||
201 => {
|
201 => {
|
||||||
tracing::info!(%label, "forge: created repo");
|
tracing::info!(%label, "forge: created repo");
|
||||||
|
|
@ -891,7 +899,7 @@ pub async fn push_config(name: &str) -> Result<()> {
|
||||||
async fn ensure_org(name: &str, admin_token: &str) -> Result<()> {
|
async fn ensure_org(name: &str, admin_token: &str) -> Result<()> {
|
||||||
let body = format!(r#"{{"username":"{name}"}}"#);
|
let body = format!(r#"{{"username":"{name}"}}"#);
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/orgs");
|
let url = format!("{FORGE_HTTP}/api/v1/orgs");
|
||||||
let status = forge_http(reqwest::Method::POST, &url, admin_token, &body).await?;
|
let (status, _) = forge_http(reqwest::Method::POST, &url, admin_token, &body).await?;
|
||||||
match status.as_u16() {
|
match status.as_u16() {
|
||||||
201 => {
|
201 => {
|
||||||
tracing::info!(%name, "forge: created org");
|
tracing::info!(%name, "forge: created org");
|
||||||
|
|
@ -926,7 +934,7 @@ async fn ensure_operators_team(token: &str) -> Result<()> {
|
||||||
let body = format!(
|
let body = format!(
|
||||||
r#"{{"name":"{OPERATORS_TEAM}","description":"hyperhive operators — merge gate for agent repos","permission":"write","includes_all_repositories":true,"can_create_org_repo":false}}"#
|
r#"{{"name":"{OPERATORS_TEAM}","description":"hyperhive operators — merge gate for agent repos","permission":"write","includes_all_repositories":true,"can_create_org_repo":false}}"#
|
||||||
);
|
);
|
||||||
let status = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
let (status, _) = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
||||||
match status.as_u16() {
|
match status.as_u16() {
|
||||||
201 => {
|
201 => {
|
||||||
tracing::info!("forge: created {OPERATORS_TEAM} team in {AGENTS_ORG}");
|
tracing::info!("forge: created {OPERATORS_TEAM} team in {AGENTS_ORG}");
|
||||||
|
|
@ -954,7 +962,7 @@ async fn add_collaborator(
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/repos/{owner}/{repo}/collaborators/{user}");
|
let url = format!("{FORGE_HTTP}/api/v1/repos/{owner}/{repo}/collaborators/{user}");
|
||||||
let body = format!(r#"{{"permission":"{permission}"}}"#);
|
let body = format!(r#"{{"permission":"{permission}"}}"#);
|
||||||
let status = forge_http(reqwest::Method::PUT, &url, token, &body).await?;
|
let (status, _) = forge_http(reqwest::Method::PUT, &url, token, &body).await?;
|
||||||
match status.as_u16() {
|
match status.as_u16() {
|
||||||
201 | 204 => {
|
201 | 204 => {
|
||||||
tracing::debug!(%owner, %repo, %user, %permission, "forge: collaborator set");
|
tracing::debug!(%owner, %repo, %user, %permission, "forge: collaborator set");
|
||||||
|
|
@ -976,7 +984,7 @@ async fn apply_operator_branch_protection(repo: &str, token: &str) -> Result<()>
|
||||||
let body = format!(
|
let body = format!(
|
||||||
r#"{{"branch_name":"main","enable_merge_whitelist":true,"merge_whitelist_teams":["{OPERATORS_TEAM}"],"enable_approvals_whitelist":true,"approvals_whitelist_teams":["{OPERATORS_TEAM}"],"required_approvals":1,"block_on_official_review_requests":true}}"#
|
r#"{{"branch_name":"main","enable_merge_whitelist":true,"merge_whitelist_teams":["{OPERATORS_TEAM}"],"enable_approvals_whitelist":true,"approvals_whitelist_teams":["{OPERATORS_TEAM}"],"required_approvals":1,"block_on_official_review_requests":true}}"#
|
||||||
);
|
);
|
||||||
let status = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
let (status, _) = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
||||||
match status.as_u16() {
|
match status.as_u16() {
|
||||||
201 => {
|
201 => {
|
||||||
tracing::info!(%repo, "forge: applied operator branch protection");
|
tracing::info!(%repo, "forge: applied operator branch protection");
|
||||||
|
|
@ -1013,19 +1021,30 @@ async fn apply_config_repo_branch_protection(repo: &str, token: &str) -> Result<
|
||||||
let body = format!(
|
let body = format!(
|
||||||
r#"{{"branch_name":"main","enable_push_whitelist":true,"push_whitelist_usernames":["core"],"enable_merge_whitelist":true,"merge_whitelist_usernames":["core"],"enable_approvals_whitelist":true,"approvals_whitelist_teams":["{OPERATORS_TEAM}"],"required_approvals":1,"block_on_official_review_requests":true,"allow_manual_merge":true,"enable_force_push":false}}"#
|
r#"{{"branch_name":"main","enable_push_whitelist":true,"push_whitelist_usernames":["core"],"enable_merge_whitelist":true,"merge_whitelist_usernames":["core"],"enable_approvals_whitelist":true,"approvals_whitelist_teams":["{OPERATORS_TEAM}"],"required_approvals":1,"block_on_official_review_requests":true,"allow_manual_merge":true,"enable_force_push":false}}"#
|
||||||
);
|
);
|
||||||
let status = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
let (status, resp_body) = forge_http(reqwest::Method::POST, &url, token, &body).await?;
|
||||||
match status.as_u16() {
|
if status.as_u16() == 201 {
|
||||||
201 => {
|
tracing::info!(%repo, "forge: applied config-repo branch protection");
|
||||||
tracing::info!(%repo, "forge: applied config-repo branch protection");
|
return Ok(());
|
||||||
Ok(())
|
}
|
||||||
}
|
// Non-201 is ambiguous: it can mean "rule already exists" (idempotent
|
||||||
200 | 409 | 422 => {
|
// success) OR a silent rejection — e.g. a 422 where Forgejo refused
|
||||||
tracing::debug!(%repo, "forge: config-repo branch protection already present");
|
// the request and created NO rule. The old code treated 200/409/422
|
||||||
Ok(())
|
// all as success, so a rejected POST left the repo unprotected with
|
||||||
}
|
// no error (the reported case: a new agent's config repo had no
|
||||||
other => {
|
// `main` rule and nothing was logged). Don't trust the status code:
|
||||||
anyhow::bail!("POST {CONFIG_ORG}/{repo}/branch_protections returned HTTP {other}")
|
// verify the `main` rule actually exists, and on failure surface the
|
||||||
}
|
// POST's response body so the real reason is in the journal.
|
||||||
|
let main_url = format!("{url}/main");
|
||||||
|
let (check, _) = forge_http(reqwest::Method::GET, &main_url, token, "").await?;
|
||||||
|
if check.as_u16() == 200 {
|
||||||
|
tracing::debug!(%repo, %status, "forge: config-repo branch protection already present");
|
||||||
|
Ok(())
|
||||||
|
} else {
|
||||||
|
anyhow::bail!(
|
||||||
|
"branch protection for {CONFIG_ORG}/{repo} not applied: POST -> HTTP {status} \
|
||||||
|
(body: {body}); GET main -> HTTP {check}, no `main` rule present",
|
||||||
|
body = resp_body.trim(),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -1412,7 +1431,7 @@ pub async fn mark_pr_merged(repo: &str, pr: u64, sha: &str) -> Result<(), ForgeM
|
||||||
.ok_or_else(|| ForgeMergeError::Other(anyhow::anyhow!("forge core token absent")))?;
|
.ok_or_else(|| ForgeMergeError::Other(anyhow::anyhow!("forge core token absent")))?;
|
||||||
let url = format!("{FORGE_HTTP}/api/v1/repos/{repo}/pulls/{pr}/merge");
|
let url = format!("{FORGE_HTTP}/api/v1/repos/{repo}/pulls/{pr}/merge");
|
||||||
let body = format!(r#"{{"Do":"manually-merged","MergeCommitID":"{sha}"}}"#);
|
let body = format!(r#"{{"Do":"manually-merged","MergeCommitID":"{sha}"}}"#);
|
||||||
let status = forge_http(reqwest::Method::POST, &url, &token, &body)
|
let (status, _) = forge_http(reqwest::Method::POST, &url, &token, &body)
|
||||||
.await
|
.await
|
||||||
.context("POST pulls/<pr>/merge (manually-merged)")?;
|
.context("POST pulls/<pr>/merge (manually-merged)")?;
|
||||||
if status.is_success() {
|
if status.is_success() {
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,20 @@ use anyhow::{Context, Result, bail};
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
use rusqlite::{Connection, OptionalExtension, params};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
/// Typed error returned by [`ScheduledPrompts::pause`] and
|
||||||
|
/// [`ScheduledPrompts::resume`] when the target row does not exist or
|
||||||
|
/// is already cancelled. Handlers downcast on this type to emit 404
|
||||||
|
/// rather than 500, avoiding brittle string-matching on the message.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub struct ScheduleNotFoundOrCancelled(pub i64);
|
||||||
|
|
||||||
|
impl std::fmt::Display for ScheduleNotFoundOrCancelled {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
write!(f, "schedule {} not found or is cancelled", self.0)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
impl std::error::Error for ScheduleNotFoundOrCancelled {}
|
||||||
|
|
||||||
const SCHEMA: &str = r"
|
const SCHEMA: &str = r"
|
||||||
CREATE TABLE IF NOT EXISTS scheduled_prompts (
|
CREATE TABLE IF NOT EXISTS scheduled_prompts (
|
||||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
|
@ -29,11 +43,12 @@ CREATE TABLE IF NOT EXISTS scheduled_prompts (
|
||||||
created_at_unix INTEGER NOT NULL,
|
created_at_unix INTEGER NOT NULL,
|
||||||
source TEXT NOT NULL,
|
source TEXT NOT NULL,
|
||||||
cancelled_at_unix INTEGER,
|
cancelled_at_unix INTEGER,
|
||||||
description TEXT
|
description TEXT,
|
||||||
|
paused_at_unix INTEGER
|
||||||
);
|
);
|
||||||
CREATE INDEX IF NOT EXISTS idx_scheduled_due
|
CREATE INDEX IF NOT EXISTS idx_scheduled_due
|
||||||
ON scheduled_prompts (next_fire_at_unix)
|
ON scheduled_prompts (next_fire_at_unix)
|
||||||
WHERE cancelled_at_unix IS NULL;
|
WHERE cancelled_at_unix IS NULL AND paused_at_unix IS NULL;
|
||||||
|
|
||||||
CREATE TABLE IF NOT EXISTS scheduled_prompt_targets (
|
CREATE TABLE IF NOT EXISTS scheduled_prompt_targets (
|
||||||
schedule_id INTEGER NOT NULL,
|
schedule_id INTEGER NOT NULL,
|
||||||
|
|
@ -68,6 +83,9 @@ pub struct Schedule {
|
||||||
/// next pass.
|
/// next pass.
|
||||||
pub cancelled_at_unix: Option<i64>,
|
pub cancelled_at_unix: Option<i64>,
|
||||||
pub description: Option<String>,
|
pub description: Option<String>,
|
||||||
|
/// Set while the schedule is paused. Worker skips rows where
|
||||||
|
/// `paused_at_unix IS NOT NULL`. Cleared by `resume()`.
|
||||||
|
pub paused_at_unix: Option<i64>,
|
||||||
pub targets: Vec<ScheduleTarget>,
|
pub targets: Vec<ScheduleTarget>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -173,6 +191,24 @@ impl ScheduledPrompts {
|
||||||
.context("enable foreign keys")?;
|
.context("enable foreign keys")?;
|
||||||
conn.execute_batch(SCHEMA)
|
conn.execute_batch(SCHEMA)
|
||||||
.context("apply scheduled_prompts schema")?;
|
.context("apply scheduled_prompts schema")?;
|
||||||
|
// Migration: add paused_at_unix to existing databases.
|
||||||
|
// Silently ignores "duplicate column name" errors so this is
|
||||||
|
// idempotent across daemon restarts on already-migrated DBs.
|
||||||
|
let _ = conn.execute(
|
||||||
|
"ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER",
|
||||||
|
[],
|
||||||
|
);
|
||||||
|
// Migration: recreate the due-rows index to also exclude paused
|
||||||
|
// rows. `CREATE INDEX IF NOT EXISTS` won't update an existing
|
||||||
|
// index's WHERE clause, so we drop + recreate on every open.
|
||||||
|
// The table is small and the op is cheap.
|
||||||
|
conn.execute_batch(
|
||||||
|
"DROP INDEX IF EXISTS idx_scheduled_due;
|
||||||
|
CREATE INDEX idx_scheduled_due
|
||||||
|
ON scheduled_prompts (next_fire_at_unix)
|
||||||
|
WHERE cancelled_at_unix IS NULL AND paused_at_unix IS NULL;",
|
||||||
|
)
|
||||||
|
.context("scheduled_prompts: recreate due-index for paused support")?;
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
conn: Mutex::new(conn),
|
conn: Mutex::new(conn),
|
||||||
})
|
})
|
||||||
|
|
@ -221,7 +257,8 @@ impl ScheduledPrompts {
|
||||||
let row = conn
|
let row = conn
|
||||||
.query_row(
|
.query_row(
|
||||||
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
||||||
created_at_unix, source, cancelled_at_unix, description
|
created_at_unix, source, cancelled_at_unix, description,
|
||||||
|
paused_at_unix
|
||||||
FROM scheduled_prompts WHERE id = ?1",
|
FROM scheduled_prompts WHERE id = ?1",
|
||||||
params![id],
|
params![id],
|
||||||
row_to_schedule_header,
|
row_to_schedule_header,
|
||||||
|
|
@ -242,7 +279,8 @@ impl ScheduledPrompts {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
let mut stmt = conn.prepare(
|
let mut stmt = conn.prepare(
|
||||||
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
||||||
created_at_unix, source, cancelled_at_unix, description
|
created_at_unix, source, cancelled_at_unix, description,
|
||||||
|
paused_at_unix
|
||||||
FROM scheduled_prompts
|
FROM scheduled_prompts
|
||||||
ORDER BY next_fire_at_unix ASC, id ASC",
|
ORDER BY next_fire_at_unix ASC, id ASC",
|
||||||
)?;
|
)?;
|
||||||
|
|
@ -264,9 +302,11 @@ impl ScheduledPrompts {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
let mut stmt = conn.prepare(
|
let mut stmt = conn.prepare(
|
||||||
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
||||||
created_at_unix, source, cancelled_at_unix, description
|
created_at_unix, source, cancelled_at_unix, description,
|
||||||
|
paused_at_unix
|
||||||
FROM scheduled_prompts
|
FROM scheduled_prompts
|
||||||
WHERE cancelled_at_unix IS NULL AND next_fire_at_unix <= ?1
|
WHERE cancelled_at_unix IS NULL AND paused_at_unix IS NULL
|
||||||
|
AND next_fire_at_unix <= ?1
|
||||||
ORDER BY next_fire_at_unix ASC, id ASC
|
ORDER BY next_fire_at_unix ASC, id ASC
|
||||||
LIMIT ?2",
|
LIMIT ?2",
|
||||||
)?;
|
)?;
|
||||||
|
|
@ -546,6 +586,45 @@ impl ScheduledPrompts {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Pause a schedule. Idempotent; no-op on an already-paused row.
|
||||||
|
/// The worker skips paused rows so no fires occur while paused;
|
||||||
|
/// `next_fire_at_unix` is preserved so the first resume fires at
|
||||||
|
/// the next intended cadence instant (no catch-up needed — a
|
||||||
|
/// paused schedule simply slips its upcoming fire). Returns
|
||||||
|
/// `Err(ScheduleNotFoundOrCancelled)` when the schedule is
|
||||||
|
/// cancelled or does not exist (handlers downcast to emit 404).
|
||||||
|
pub fn pause(&self, id: i64) -> Result<()> {
|
||||||
|
let conn = self.conn.lock().unwrap();
|
||||||
|
let now = now_unix();
|
||||||
|
let n = conn.execute(
|
||||||
|
"UPDATE scheduled_prompts
|
||||||
|
SET paused_at_unix = COALESCE(paused_at_unix, ?1)
|
||||||
|
WHERE id = ?2 AND cancelled_at_unix IS NULL",
|
||||||
|
params![now, id],
|
||||||
|
)?;
|
||||||
|
if n == 0 {
|
||||||
|
return Err(ScheduleNotFoundOrCancelled(id).into());
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Resume a paused schedule. Idempotent; no-op on an active row.
|
||||||
|
/// Returns `Err(ScheduleNotFoundOrCancelled)` when the schedule is
|
||||||
|
/// cancelled or does not exist (handlers downcast to emit 404).
|
||||||
|
pub fn resume(&self, id: i64) -> Result<()> {
|
||||||
|
let conn = self.conn.lock().unwrap();
|
||||||
|
let n = conn.execute(
|
||||||
|
"UPDATE scheduled_prompts
|
||||||
|
SET paused_at_unix = NULL
|
||||||
|
WHERE id = ?1 AND cancelled_at_unix IS NULL",
|
||||||
|
params![id],
|
||||||
|
)?;
|
||||||
|
if n == 0 {
|
||||||
|
return Err(ScheduleNotFoundOrCancelled(id).into());
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// Reap cancelled rows older than `older_than_unix`. Returns
|
/// Reap cancelled rows older than `older_than_unix`. Returns
|
||||||
/// the number of rows deleted. Called from the worker on each
|
/// the number of rows deleted. Called from the worker on each
|
||||||
/// tick so cancellations clear out of the dashboard without an
|
/// tick so cancellations clear out of the dashboard without an
|
||||||
|
|
@ -575,6 +654,7 @@ fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result<Schedule> {
|
||||||
source: ScheduleSource::from_db_string(&source_str),
|
source: ScheduleSource::from_db_string(&source_str),
|
||||||
cancelled_at_unix: row.get(7)?,
|
cancelled_at_unix: row.get(7)?,
|
||||||
description: row.get(8)?,
|
description: row.get(8)?,
|
||||||
|
paused_at_unix: row.get(9)?,
|
||||||
targets: Vec::new(),
|
targets: Vec::new(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -2021,6 +2021,7 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSc
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
cancelled_at_unix: s.cancelled_at_unix,
|
cancelled_at_unix: s.cancelled_at_unix,
|
||||||
|
paused_at_unix: s.paused_at_unix,
|
||||||
description: s.description,
|
description: s.description,
|
||||||
targets: s
|
targets: s
|
||||||
.targets
|
.targets
|
||||||
|
|
@ -2126,6 +2127,7 @@ mod tests {
|
||||||
created_at_unix: 0,
|
created_at_unix: 0,
|
||||||
source: hive_sh4re::WireScheduleSource::Operator,
|
source: hive_sh4re::WireScheduleSource::Operator,
|
||||||
cancelled_at_unix: None,
|
cancelled_at_unix: None,
|
||||||
|
paused_at_unix: None,
|
||||||
description: None,
|
description: None,
|
||||||
targets: targets.iter().map(|t| target(t)).collect(),
|
targets: targets.iter().map(|t| target(t)).collect(),
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1395,6 +1395,12 @@ pub struct WireSchedule {
|
||||||
pub source: WireScheduleSource,
|
pub source: WireScheduleSource,
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
pub cancelled_at_unix: Option<i64>,
|
pub cancelled_at_unix: Option<i64>,
|
||||||
|
/// Set while the schedule is paused. Worker skips paused rows;
|
||||||
|
/// they keep their `next_fire_at_unix` so resuming at any time
|
||||||
|
/// fires at the next intended instant (no catch-up clamp needed
|
||||||
|
/// — a paused schedule simply slips its next fire).
|
||||||
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
|
pub paused_at_unix: Option<i64>,
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
pub description: Option<String>,
|
pub description: Option<String>,
|
||||||
pub targets: Vec<WireScheduleTarget>,
|
pub targets: Vec<WireScheduleTarget>,
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,44 @@ let
|
||||||
# from `userName` to keep them coupled.
|
# from `userName` to keep them coupled.
|
||||||
userName = config.hyperhive.user.name;
|
userName = config.hyperhive.user.name;
|
||||||
homeDir = "/home/${userName}";
|
homeDir = "/home/${userName}";
|
||||||
|
# Hive-wide OpenTelemetry config (host-driven; baked in per-agent by
|
||||||
|
# meta.rs `otel_config`).
|
||||||
|
otelCfg = config.hyperhive.otel;
|
||||||
|
# Hive/swarm display names are forwarded into each agent's build by
|
||||||
|
# meta.rs as `environment.variables` (per-agent, build-time strings),
|
||||||
|
# so they can be baked into the resource attributes below without a
|
||||||
|
# runtime shell. Absent (option unset) → "unknown".
|
||||||
|
hiveDisplayName = config.environment.variables.HYPERHIVE_HIVE_NAME or "unknown";
|
||||||
|
swarmDisplayName = config.environment.variables.HYPERHIVE_SWARM_NAME or "unknown";
|
||||||
|
# OTEL environment Claude Code reads to export metrics/logs/traces.
|
||||||
|
# Shipped via the managed claude settings json (below), which claude
|
||||||
|
# auto-discovers for BOTH the harness turn-loop and `hivectl choom` —
|
||||||
|
# so telemetry parity is declarative, with no launch wrapper. The
|
||||||
|
# auth header (`otel.headersCredential`) is deliberately NOT included
|
||||||
|
# here: it's a secret and this file lives in the world-readable nix
|
||||||
|
# store. It's injected at *runtime* into the agent's `0600`
|
||||||
|
# `~/.claude/settings.json` by the `hive-otel-header` oneshot below
|
||||||
|
# (claude merges the `env` from the user settings on top of these
|
||||||
|
# managed ones), so the token is read from disk at start and never
|
||||||
|
# touches the store.
|
||||||
|
otelSettingsEnv = {
|
||||||
|
CLAUDE_CODE_ENABLE_TELEMETRY = "1";
|
||||||
|
OTEL_METRICS_EXPORTER = "otlp";
|
||||||
|
OTEL_LOGS_EXPORTER = "otlp";
|
||||||
|
OTEL_TRACES_EXPORTER = "otlp";
|
||||||
|
OTEL_EXPORTER_OTLP_PROTOCOL = otelCfg.protocol;
|
||||||
|
OTEL_EXPORTER_OTLP_ENDPOINT = otelCfg.endpoint;
|
||||||
|
# Force CUMULATIVE temporality — Claude Code defaults to DELTA,
|
||||||
|
# which Prometheus/Mimir-family backends (incl. grafana-lgtm)
|
||||||
|
# silently drop without a deltatocumulative processor.
|
||||||
|
OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE = "cumulative";
|
||||||
|
OTEL_RESOURCE_ATTRIBUTES =
|
||||||
|
"service.name=hyperhive-agent,agent=${userName},hive=${hiveDisplayName},swarm=${swarmDisplayName}"
|
||||||
|
+ lib.optionalString (otelCfg.extraResourceAttributes != "") ",${otelCfg.extraResourceAttributes}";
|
||||||
|
}
|
||||||
|
// lib.optionalAttrs (otelCfg.metricIntervalMs != null) {
|
||||||
|
OTEL_METRIC_EXPORT_INTERVAL = toString otelCfg.metricIntervalMs;
|
||||||
|
};
|
||||||
# Single source of truth for the default matrix homeserver URL, shared
|
# Single source of truth for the default matrix homeserver URL, shared
|
||||||
# by the `hyperhive.matrix.url` option default and the daemon-unit guard
|
# by the `hyperhive.matrix.url` option default and the daemon-unit guard
|
||||||
# that decides whether to set a unit-level HIVE_MATRIX_URL (so the two
|
# that decides whether to set a unit-level HIVE_MATRIX_URL (so the two
|
||||||
|
|
@ -231,11 +269,17 @@ in
|
||||||
description = ''
|
description = ''
|
||||||
Absolute path to an operator-provided secret file whose contents
|
Absolute path to an operator-provided secret file whose contents
|
||||||
become `OTEL_EXPORTER_OTLP_HEADERS` (e.g.
|
become `OTEL_EXPORTER_OTLP_HEADERS` (e.g.
|
||||||
`Authorization=Bearer <token>`). Loaded via systemd
|
`Authorization=Bearer <token>`). Host-driven via
|
||||||
`LoadCredential` into the unit-private credential store at
|
|
||||||
runtime, so the token is never copied into the nix store or
|
|
||||||
exposed in the process argv. Host-driven via
|
|
||||||
`services.hyperhive.otel.headersCredential`.
|
`services.hyperhive.otel.headersCredential`.
|
||||||
|
|
||||||
|
The rest of the OTEL config ships in the world-readable managed
|
||||||
|
claude settings json, but the header is a secret, so it's handled
|
||||||
|
separately: hive-c0re forwards this file into the container's
|
||||||
|
systemd credential store, and the `hive-otel-header` oneshot
|
||||||
|
reads it at runtime (`LoadCredential`) and writes it into the
|
||||||
|
agent's `0600` `~/.claude/settings.json` `env` block. The token
|
||||||
|
is read from disk at start and never copied into the nix store or
|
||||||
|
the world-readable settings file.
|
||||||
'';
|
'';
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
@ -1073,8 +1117,68 @@ in
|
||||||
# deliberately NOT shipped here: effort is controlled live via the
|
# deliberately NOT shipped here: effort is controlled live via the
|
||||||
# `--effort` CLI flag (HIVE_DEFAULT_EFFORT / the per-agent UI slider),
|
# `--effort` CLI flag (HIVE_DEFAULT_EFFORT / the per-agent UI slider),
|
||||||
# which managed scope would otherwise override and lock.
|
# which managed scope would otherwise override and lock.
|
||||||
|
# Base hive-enforced settings, plus — when OTEL is enabled — an `env`
|
||||||
|
# block so Claude Code exports telemetry natively from the settings
|
||||||
|
# file it reads in every context (harness turn-loop AND `hivectl
|
||||||
|
# choom`), with no launch wrapper. With OTEL off it's the shared
|
||||||
|
# static asset verbatim; with OTEL on it's a per-agent merge (the
|
||||||
|
# resource attributes carry the agent name) done at BUILD time via
|
||||||
|
# `jq` — not eval-time `readFile`, which would be import-from-
|
||||||
|
# derivation.
|
||||||
environment.etc."claude-code/managed-settings.json".source =
|
environment.etc."claude-code/managed-settings.json".source =
|
||||||
"${pkgs.hyperhive-assets}/share/hyperhive/prompts/claude-settings.json";
|
let
|
||||||
|
baseSettings = "${pkgs.hyperhive-assets}/share/hyperhive/prompts/claude-settings.json";
|
||||||
|
in
|
||||||
|
if !otelCfg.enable then
|
||||||
|
baseSettings
|
||||||
|
else
|
||||||
|
pkgs.runCommand "managed-settings.json" { nativeBuildInputs = [ pkgs.jq ]; } ''
|
||||||
|
jq --argjson env ${lib.escapeShellArg (builtins.toJSON otelSettingsEnv)} \
|
||||||
|
'. + { env: $env }' ${baseSettings} > "$out"
|
||||||
|
'';
|
||||||
|
|
||||||
|
# Inject the OTEL auth header (a secret) into the agent's *user*
|
||||||
|
# claude settings at runtime, keeping it out of the world-readable
|
||||||
|
# managed settings json above and out of the nix store entirely.
|
||||||
|
# hive-c0re forwards the operator's `headersCredential` file into
|
||||||
|
# this container's systemd credential store; this oneshot reads it
|
||||||
|
# via `LoadCredential` at start and merges `OTEL_EXPORTER_OTLP_HEADERS`
|
||||||
|
# into `~/.claude/settings.json` (0600, agent-owned). claude layers
|
||||||
|
# the user `env` on top of the managed one, so both the harness
|
||||||
|
# turn-loop and `hivectl choom` (same agent user) pick it up. Ordering
|
||||||
|
# is best-effort (`before`, not a hard dep): if it fails the harness
|
||||||
|
# still starts and telemetry just exports unauthenticated.
|
||||||
|
systemd.services.hive-otel-header =
|
||||||
|
lib.mkIf (config.hyperhive.otel.enable && config.hyperhive.otel.headersCredential != null)
|
||||||
|
{
|
||||||
|
description = "Inject the OTEL auth header into the agent's claude user settings";
|
||||||
|
wantedBy = [ "multi-user.target" ];
|
||||||
|
before = [ "hive-ag3nt.service" ];
|
||||||
|
serviceConfig = {
|
||||||
|
Type = "oneshot";
|
||||||
|
RemainAfterExit = true;
|
||||||
|
User = userName;
|
||||||
|
Group = userName;
|
||||||
|
LoadCredential = [ "otel-headers" ];
|
||||||
|
ExecStart = pkgs.writeShellScript "hive-otel-header" ''
|
||||||
|
set -eu
|
||||||
|
umask 077
|
||||||
|
hdr="$CREDENTIALS_DIRECTORY/otel-headers"
|
||||||
|
[ -r "$hdr" ] || exit 0
|
||||||
|
dir=${homeDir}/.claude
|
||||||
|
settings="$dir/settings.json"
|
||||||
|
mkdir -p "$dir"
|
||||||
|
base='{}'
|
||||||
|
[ -s "$settings" ] && base="$(cat "$settings")"
|
||||||
|
printf '%s' "$base" | ${pkgs.jq}/bin/jq \
|
||||||
|
--rawfile h "$hdr" \
|
||||||
|
'.env = ((.env // {}) + { OTEL_EXPORTER_OTLP_HEADERS: ($h | rtrimstr("\n")) })' \
|
||||||
|
> "$settings.tmp"
|
||||||
|
mv "$settings.tmp" "$settings"
|
||||||
|
chmod 0600 "$settings"
|
||||||
|
'';
|
||||||
|
};
|
||||||
|
};
|
||||||
|
|
||||||
# Merged frontend static tree. Base = `${frontend.dist}/agent/`,
|
# Merged frontend static tree. Base = `${frontend.dist}/agent/`,
|
||||||
# then each `extraFiles` entry is laid on top at its `target`
|
# then each `extraFiles` entry is laid on top at its `target`
|
||||||
|
|
@ -1707,49 +1811,11 @@ in
|
||||||
systemd.services.hive-ag3nt =
|
systemd.services.hive-ag3nt =
|
||||||
let
|
let
|
||||||
binary = "hive";
|
binary = "hive";
|
||||||
otel = config.hyperhive.otel;
|
# OTEL is shipped declaratively via the managed claude settings
|
||||||
# Claude Code's native OpenTelemetry is env-driven; the harness
|
# json (`environment.etc."claude-code/managed-settings.json"`,
|
||||||
# spawns `claude` as a child which inherits this unit's env, so
|
# `otelSettingsEnv` in the top-level let) — claude reads it for
|
||||||
# setting these here is all it takes to export per-agent stats.
|
# both the harness turn-loop and `hivectl choom`, so there's no
|
||||||
otelEnv = lib.optionalAttrs otel.enable (
|
# launch wrapper or per-unit OTEL env here anymore.
|
||||||
{
|
|
||||||
CLAUDE_CODE_ENABLE_TELEMETRY = "1";
|
|
||||||
OTEL_METRICS_EXPORTER = "otlp";
|
|
||||||
OTEL_LOGS_EXPORTER = "otlp";
|
|
||||||
# Route traces to OTLP too so any spans Claude Code emits land
|
|
||||||
# at the configured collector rather than a default exporter.
|
|
||||||
OTEL_TRACES_EXPORTER = "otlp";
|
|
||||||
OTEL_EXPORTER_OTLP_PROTOCOL = otel.protocol;
|
|
||||||
OTEL_EXPORTER_OTLP_ENDPOINT = otel.endpoint;
|
|
||||||
# Force CUMULATIVE metric temporality. Claude Code defaults to
|
|
||||||
# DELTA, which Prometheus/Mimir-family backends (the common case,
|
|
||||||
# incl. grafana-lgtm) silently drop unless a deltatocumulative
|
|
||||||
# processor is wired — so delta = "logs arrive, metrics vanish".
|
|
||||||
# Cumulative is what those backends ingest natively. Verified:
|
|
||||||
# a delta export never registers the metric name; cumulative does.
|
|
||||||
OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE = "cumulative";
|
|
||||||
}
|
|
||||||
// lib.optionalAttrs (otel.metricIntervalMs != null) {
|
|
||||||
OTEL_METRIC_EXPORT_INTERVAL = toString otel.metricIntervalMs;
|
|
||||||
}
|
|
||||||
);
|
|
||||||
# When OTEL is on, wrap the harness launch so (1) the bearer-token
|
|
||||||
# header is read from the systemd credential at start (never in the
|
|
||||||
# nix store or argv) and (2) the resource attributes are assembled
|
|
||||||
# from this agent's name (known at build time) plus the hive/swarm
|
|
||||||
# names (inherited HYPERHIVE_HIVE_NAME / HYPERHIVE_SWARM_NAME env,
|
|
||||||
# the same vars `identity::hive_name`/`swarm_name` read at runtime).
|
|
||||||
otelExecStart = pkgs.writeShellScript "hive-serve-otel" ''
|
|
||||||
set -eu
|
|
||||||
if [ -n "''${CREDENTIALS_DIRECTORY:-}" ] && [ -r "$CREDENTIALS_DIRECTORY/otel-headers" ]; then
|
|
||||||
OTEL_EXPORTER_OTLP_HEADERS="$(cat "$CREDENTIALS_DIRECTORY/otel-headers")"
|
|
||||||
export OTEL_EXPORTER_OTLP_HEADERS
|
|
||||||
fi
|
|
||||||
export OTEL_RESOURCE_ATTRIBUTES="service.name=hyperhive-agent,agent=${userName},hive=''${HYPERHIVE_HIVE_NAME:-unknown},swarm=''${HYPERHIVE_SWARM_NAME:-unknown}${
|
|
||||||
lib.optionalString (otel.extraResourceAttributes != "") ",${otel.extraResourceAttributes}"
|
|
||||||
}"
|
|
||||||
exec ${pkgs.hyperhive}/bin/${binary} serve
|
|
||||||
'';
|
|
||||||
in
|
in
|
||||||
{
|
{
|
||||||
description = "${binary} harness";
|
description = "${binary} harness";
|
||||||
|
|
@ -1773,7 +1839,6 @@ in
|
||||||
# bind-mounts and gateway upstream config stay in sync.
|
# bind-mounts and gateway upstream config stay in sync.
|
||||||
HIVE_WEB_SOCKET = "/run/hive-agent/${userName}/web.sock";
|
HIVE_WEB_SOCKET = "/run/hive-agent/${userName}/web.sock";
|
||||||
}
|
}
|
||||||
// otelEnv
|
|
||||||
// lib.optionalAttrs config.hyperhive.gui.enable {
|
// lib.optionalAttrs config.hyperhive.gui.enable {
|
||||||
# Tells the harness which fixed VNC port weston bound, and (by
|
# Tells the harness which fixed VNC port weston bound, and (by
|
||||||
# its presence) that gui is enabled — the harness `/screen/ws`
|
# its presence) that gui is enabled — the harness `/screen/ws`
|
||||||
|
|
@ -1784,13 +1849,9 @@ in
|
||||||
HIVE_GUI_VNC_PORT = toString config.hyperhive.gui.vncPort;
|
HIVE_GUI_VNC_PORT = toString config.hyperhive.gui.vncPort;
|
||||||
};
|
};
|
||||||
serviceConfig = {
|
serviceConfig = {
|
||||||
ExecStart = if otel.enable then "${otelExecStart}" else "${pkgs.hyperhive}/bin/${binary} serve";
|
ExecStart = "${pkgs.hyperhive}/bin/${binary} serve";
|
||||||
# Pin the journal identity to the binary name. Without this,
|
# Pin the journal identity to the binary name (otherwise systemd
|
||||||
# systemd derives SyslogIdentifier from the ExecStart basename —
|
# derives SyslogIdentifier from the ExecStart basename).
|
||||||
# which under OTEL is the wrapper script's store path
|
|
||||||
# (`<hash>-hive-serve-otel`), so every agent's harness logs showed
|
|
||||||
# that opaque name instead of `hive`. Set explicitly so the
|
|
||||||
# identity is stable across the otel / non-otel ExecStart branches.
|
|
||||||
SyslogIdentifier = binary;
|
SyslogIdentifier = binary;
|
||||||
Restart = "on-failure";
|
Restart = "on-failure";
|
||||||
RestartSec = 2;
|
RestartSec = 2;
|
||||||
|
|
@ -1801,15 +1862,6 @@ in
|
||||||
RuntimeDirectory = "hive-config";
|
RuntimeDirectory = "hive-config";
|
||||||
User = userName;
|
User = userName;
|
||||||
Group = userName;
|
Group = userName;
|
||||||
}
|
|
||||||
// lib.optionalAttrs (otel.enable && otel.headersCredential != null) {
|
|
||||||
# Inherit form (no `:path`): hive-c0re forwards the host file at
|
|
||||||
# `headersCredential` into this container's credential store via
|
|
||||||
# nspawn `--load-credential=otel-headers:<host path>` (see
|
|
||||||
# lifecycle.rs::hive_load_credentials). The path isn't reachable
|
|
||||||
# from inside the container, so we inherit the already-loaded
|
|
||||||
# credential by name rather than re-reading the host path here.
|
|
||||||
LoadCredential = [ "otel-headers" ];
|
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue