Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
637085644d | ||
|
|
0385d96bf3 | ||
|
|
79a46f359a |
11 changed files with 349 additions and 46 deletions
28
TODO.md
28
TODO.md
|
|
@ -27,19 +27,10 @@ Pick anything from here when relevant. Cross-cutting design notes live in
|
||||||
|
|
||||||
## UI / UX
|
## UI / UX
|
||||||
|
|
||||||
- **State badge: compacting + napping states.** Idle/thinking already
|
- **State badge: napping state.** Idle / thinking / compacting
|
||||||
ship (driven from SSE turn_start/turn_end). Add `compacting 📦` and
|
already ship from server-side `TurnState`. Add `napping 😴`
|
||||||
`napping 😴` once the `/compact` trigger and `nap` tool exist —
|
once the `nap` tool exists — it just adds a new `TurnState`
|
||||||
both need a harness signal (an explicit `LiveEvent::StateChange`
|
variant the harness flips into for the duration of the nap.
|
||||||
variant or piggyback on Note).
|
|
||||||
- **Server-side state badge.** Today the badge is computed client-side
|
|
||||||
from `turn_start`/`turn_end` events. On page reload mid-turn the
|
|
||||||
history replay re-derives it, but with a `compacting` / `napping`
|
|
||||||
state coming and a non-trivial state machine it's better to track
|
|
||||||
authoritative state in the harness and expose it via
|
|
||||||
`GET /api/state` (`status: "thinking" | "idle" | "compacting" |
|
|
||||||
"napping"`). JS just renders. Drops the
|
|
||||||
derive-from-events-and-pray code path.
|
|
||||||
- **Terminal: `/model` slash command.** Operator-typeable model
|
- **Terminal: `/model` slash command.** Operator-typeable model
|
||||||
override from the terminal. Depends on the model-override work
|
override from the terminal. Depends on the model-override work
|
||||||
above; once an override mechanism exists, wire a `/model <name>`
|
above; once an override mechanism exists, wire a `/model <name>`
|
||||||
|
|
@ -107,17 +98,6 @@ Pick anything from here when relevant. Cross-cutting design notes live in
|
||||||
|
|
||||||
## Lifecycle / reliability
|
## Lifecycle / reliability
|
||||||
|
|
||||||
- **journald viewer per container in the dashboard.** Surface the
|
|
||||||
equivalent of `journalctl -M h-coder -b` in the dashboard so the
|
|
||||||
operator can see container logs without ssh-ing in. Optional
|
|
||||||
filter by hive-specific systemd unit (`hive-ag3nt.service`,
|
|
||||||
`hive-m1nd.service`). Implementation: backend shells out to
|
|
||||||
`journalctl -M <container> -b --output=short-iso --no-pager`
|
|
||||||
(optionally `-u <unit>`), streams or paginates the result over a
|
|
||||||
new dashboard endpoint. Could be a `<details>` per container row
|
|
||||||
or a dedicated page. Honest journalctl, not the in-container
|
|
||||||
events stream — those are different surfaces (events = claude turn
|
|
||||||
loop; journalctl = systemd-wide logs incl. boot, network, etc.).
|
|
||||||
- **Container crash events.** Watch `container@*.service` via D-Bus, push
|
- **Container crash events.** Watch `container@*.service` via D-Bus, push
|
||||||
`HelperEvent::ContainerCrash` to the manager's inbox so the manager can
|
`HelperEvent::ContainerCrash` to the manager's inbox so the manager can
|
||||||
react (restart, escalate, etc.).
|
react (restart, escalate, etc.).
|
||||||
|
|
|
||||||
|
|
@ -228,6 +228,11 @@ pre.diff {
|
||||||
text-shadow: 0 0 6px rgba(250, 179, 135, 0.65);
|
text-shadow: 0 0 6px rgba(250, 179, 135, 0.65);
|
||||||
animation: badge-pulse 1.8s ease-in-out infinite;
|
animation: badge-pulse 1.8s ease-in-out infinite;
|
||||||
}
|
}
|
||||||
|
.state-badge.state-compacting {
|
||||||
|
color: var(--purple); border-color: var(--purple);
|
||||||
|
text-shadow: 0 0 6px rgba(203, 166, 247, 0.65);
|
||||||
|
animation: badge-pulse 1.8s ease-in-out infinite;
|
||||||
|
}
|
||||||
.state-badge.state-just-changed {
|
.state-badge.state-just-changed {
|
||||||
animation: state-flash 600ms ease-out;
|
animation: state-flash 600ms ease-out;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -291,10 +291,11 @@
|
||||||
// each second so the "· 12s" suffix stays current. State changes
|
// each second so the "· 12s" suffix stays current. State changes
|
||||||
// trigger a short flash animation via .state-just-changed.
|
// trigger a short flash animation via .state-just-changed.
|
||||||
const STATE_LABELS = {
|
const STATE_LABELS = {
|
||||||
loading: { glyph: '…', text: 'booting' },
|
loading: { glyph: '…', text: 'booting' },
|
||||||
offline: { glyph: '○', text: 'offline' },
|
offline: { glyph: '○', text: 'offline' },
|
||||||
idle: { glyph: '💤', text: 'idle' },
|
idle: { glyph: '💤', text: 'idle' },
|
||||||
thinking: { glyph: '🧠', text: 'thinking' },
|
thinking: { glyph: '🧠', text: 'thinking' },
|
||||||
|
compacting: { glyph: '📦', text: 'compacting' },
|
||||||
};
|
};
|
||||||
let stateName = 'loading';
|
let stateName = 'loading';
|
||||||
let stateSince = Date.now();
|
let stateSince = Date.now();
|
||||||
|
|
@ -318,19 +319,22 @@
|
||||||
if (cancelBtn) cancelBtn.hidden = stateName !== 'thinking';
|
if (cancelBtn) cancelBtn.hidden = stateName !== 'thinking';
|
||||||
}
|
}
|
||||||
function setState(next) {
|
function setState(next) {
|
||||||
if (next === stateName) return;
|
setStateAbs(next, Math.floor(Date.now() / 1000));
|
||||||
// Capture the just-ending state's duration when leaving 'thinking'
|
}
|
||||||
// so the operator can eyeball turn length without scrolling the
|
/// Set state with an authoritative since-unix from the server. Lets
|
||||||
// terminal back.
|
/// `last turn` track the actual server-side duration rather than
|
||||||
|
/// whatever the client perceived between SSE events.
|
||||||
|
function setStateAbs(next, sinceUnix) {
|
||||||
|
if (next === stateName && sinceUnix * 1000 === stateSince) return;
|
||||||
if (stateName === 'thinking' && next !== 'thinking') {
|
if (stateName === 'thinking' && next !== 'thinking') {
|
||||||
const elapsedMs = Date.now() - stateSince;
|
const elapsedMs = Date.now() - stateSince;
|
||||||
renderLastTurn(elapsedMs);
|
renderLastTurn(elapsedMs);
|
||||||
}
|
}
|
||||||
|
const flashing = next !== stateName;
|
||||||
stateName = next;
|
stateName = next;
|
||||||
stateSince = Date.now();
|
stateSince = sinceUnix * 1000;
|
||||||
const badge = $('state-badge');
|
const badge = $('state-badge');
|
||||||
if (badge) {
|
if (badge && flashing) {
|
||||||
// Re-add the flash class so the animation replays.
|
|
||||||
badge.classList.remove('state-just-changed');
|
badge.classList.remove('state-just-changed');
|
||||||
void badge.offsetWidth;
|
void badge.offsetWidth;
|
||||||
badge.classList.add('state-just-changed');
|
badge.classList.add('state-just-changed');
|
||||||
|
|
@ -411,11 +415,15 @@
|
||||||
if (!headerSet) { setHeader(s.label, s.dashboard_port); headerSet = true; }
|
if (!headerSet) { setHeader(s.label, s.dashboard_port); headerSet = true; }
|
||||||
renderTermInput(s.label, s.status === 'online');
|
renderTermInput(s.label, s.status === 'online');
|
||||||
renderInbox(s.inbox || []);
|
renderInbox(s.inbox || []);
|
||||||
// Drive the state badge from the harness status. Live SSE events
|
// Authoritative state comes from the harness via /api/state.
|
||||||
// override to 'thinking' / 'idle' as turns start/end; this only
|
// Login-not-yet → 'offline'; otherwise use the server-reported
|
||||||
// kicks in for the not-online (offline) case and the initial seed.
|
// turn_state (idle / thinking / compacting). stateSince in
|
||||||
if (s.status !== 'online') setState('offline');
|
// unix-seconds is converted to a client-side Date.now() anchor.
|
||||||
else if (stateName === 'loading' || stateName === 'offline') setState('idle');
|
if (s.status !== 'online') {
|
||||||
|
setState('offline');
|
||||||
|
} else if (s.turn_state) {
|
||||||
|
setStateAbs(s.turn_state, s.turn_state_since);
|
||||||
|
}
|
||||||
// Skip the re-render if nothing structurally changed. The most
|
// Skip the re-render if nothing structurally changed. The most
|
||||||
// common case is `online` polling itself — without this guard, the
|
// common case is `online` polling itself — without this guard, the
|
||||||
// operator's <input value> gets clobbered every cycle.
|
// operator's <input value> gets clobbered every cycle.
|
||||||
|
|
|
||||||
|
|
@ -4,7 +4,7 @@ use std::time::Duration;
|
||||||
|
|
||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
use clap::{Parser, Subcommand};
|
use clap::{Parser, Subcommand};
|
||||||
use hive_ag3nt::events::{Bus, LiveEvent};
|
use hive_ag3nt::events::{Bus, LiveEvent, TurnState};
|
||||||
use hive_ag3nt::login::{self, LoginState};
|
use hive_ag3nt::login::{self, LoginState};
|
||||||
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
||||||
use hive_sh4re::{AgentRequest, AgentResponse};
|
use hive_sh4re::{AgentRequest, AgentResponse};
|
||||||
|
|
@ -126,6 +126,7 @@ async fn serve(
|
||||||
body: body.clone(),
|
body: body.clone(),
|
||||||
unread,
|
unread,
|
||||||
});
|
});
|
||||||
|
bus.set_state(TurnState::Thinking);
|
||||||
let prompt = format_wake_prompt(&from, &body, unread);
|
let prompt = format_wake_prompt(&from, &body, unread);
|
||||||
let outcome = turn::drive_turn(
|
let outcome = turn::drive_turn(
|
||||||
&prompt,
|
&prompt,
|
||||||
|
|
@ -137,6 +138,7 @@ async fn serve(
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
turn::emit_turn_end(&bus, &outcome);
|
turn::emit_turn_end(&bus, &outcome);
|
||||||
|
bus.set_state(TurnState::Idle);
|
||||||
}
|
}
|
||||||
Ok(AgentResponse::Empty) => {}
|
Ok(AgentResponse::Empty) => {}
|
||||||
Ok(AgentResponse::Ok | AgentResponse::Status { .. } | AgentResponse::Recent { .. }) => {
|
Ok(AgentResponse::Ok | AgentResponse::Status { .. } | AgentResponse::Recent { .. }) => {
|
||||||
|
|
|
||||||
|
|
@ -8,7 +8,7 @@ use std::time::Duration;
|
||||||
|
|
||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
use clap::{Parser, Subcommand};
|
use clap::{Parser, Subcommand};
|
||||||
use hive_ag3nt::events::{Bus, LiveEvent};
|
use hive_ag3nt::events::{Bus, LiveEvent, TurnState};
|
||||||
use hive_ag3nt::login::{self, LoginState};
|
use hive_ag3nt::login::{self, LoginState};
|
||||||
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
||||||
use hive_sh4re::{HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER};
|
use hive_sh4re::{HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER};
|
||||||
|
|
@ -124,6 +124,7 @@ async fn serve(socket: &Path, interval: Duration, bus: Bus) -> Result<()> {
|
||||||
unread,
|
unread,
|
||||||
});
|
});
|
||||||
let prompt = format_wake_prompt(&from, &body, unread);
|
let prompt = format_wake_prompt(&from, &body, unread);
|
||||||
|
bus.set_state(TurnState::Thinking);
|
||||||
let outcome = turn::drive_turn(
|
let outcome = turn::drive_turn(
|
||||||
&prompt,
|
&prompt,
|
||||||
&mcp_config,
|
&mcp_config,
|
||||||
|
|
@ -134,6 +135,7 @@ async fn serve(socket: &Path, interval: Duration, bus: Bus) -> Result<()> {
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
turn::emit_turn_end(&bus, &outcome);
|
turn::emit_turn_end(&bus, &outcome);
|
||||||
|
bus.set_state(TurnState::Idle);
|
||||||
}
|
}
|
||||||
Ok(ManagerResponse::Empty) => {}
|
Ok(ManagerResponse::Empty) => {}
|
||||||
Ok(
|
Ok(
|
||||||
|
|
|
||||||
|
|
@ -24,6 +24,14 @@ const HISTORY_CAPACITY: usize = 2000;
|
||||||
/// `HYPERHIVE_EVENTS_DB` env var (used in tests and one-shot tools).
|
/// `HYPERHIVE_EVENTS_DB` env var (used in tests and one-shot tools).
|
||||||
const DEFAULT_EVENTS_DB: &str = "/state/hyperhive-events.sqlite";
|
const DEFAULT_EVENTS_DB: &str = "/state/hyperhive-events.sqlite";
|
||||||
|
|
||||||
|
fn now_unix() -> i64 {
|
||||||
|
std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.ok()
|
||||||
|
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
||||||
|
.unwrap_or(0)
|
||||||
|
}
|
||||||
|
|
||||||
const SCHEMA: &str = "
|
const SCHEMA: &str = "
|
||||||
CREATE TABLE IF NOT EXISTS events (
|
CREATE TABLE IF NOT EXISTS events (
|
||||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
|
@ -116,6 +124,22 @@ impl EventStore {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Authoritative turn-loop state. The harness owns it; the web UI
|
||||||
|
/// reads via `/api/state` and renders. Lives alongside the bus
|
||||||
|
/// because everyone who has a `Bus` already has the right handle to
|
||||||
|
/// poke the state on transitions.
|
||||||
|
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
|
||||||
|
#[serde(rename_all = "snake_case")]
|
||||||
|
pub enum TurnState {
|
||||||
|
/// Inbox is empty / waiting on `Recv`.
|
||||||
|
Idle,
|
||||||
|
/// `claude --print` is running for a turn.
|
||||||
|
Thinking,
|
||||||
|
/// Operator-triggered `/compact` is running on the persistent
|
||||||
|
/// session.
|
||||||
|
Compacting,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct Bus {
|
pub struct Bus {
|
||||||
tx: Arc<broadcast::Sender<LiveEvent>>,
|
tx: Arc<broadcast::Sender<LiveEvent>>,
|
||||||
|
|
@ -123,6 +147,8 @@ pub struct Bus {
|
||||||
/// at construction — we keep going so the harness doesn't die on a
|
/// at construction — we keep going so the harness doesn't die on a
|
||||||
/// missing `/state/` mount in dev / test scenarios.
|
/// missing `/state/` mount in dev / test scenarios.
|
||||||
store: Option<Arc<EventStore>>,
|
store: Option<Arc<EventStore>>,
|
||||||
|
/// Current turn-loop state + since-when (unix seconds).
|
||||||
|
state: Arc<Mutex<(TurnState, i64)>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Bus {
|
impl Bus {
|
||||||
|
|
@ -144,9 +170,26 @@ impl Bus {
|
||||||
Self {
|
Self {
|
||||||
tx: Arc::new(tx),
|
tx: Arc::new(tx),
|
||||||
store,
|
store,
|
||||||
|
state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Update the harness's authoritative turn-loop state. Records
|
||||||
|
/// the transition time so `state_snapshot` can return a since-age.
|
||||||
|
pub fn set_state(&self, next: TurnState) {
|
||||||
|
let mut guard = self.state.lock().unwrap();
|
||||||
|
if guard.0 == next {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
*guard = (next, now_unix());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Current state + since-when (unix seconds). Snapshot copy, no lock held.
|
||||||
|
#[must_use]
|
||||||
|
pub fn state_snapshot(&self) -> (TurnState, i64) {
|
||||||
|
*self.state.lock().unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
pub fn emit(&self, event: LiveEvent) {
|
pub fn emit(&self, event: LiveEvent) {
|
||||||
if let Some(store) = &self.store
|
if let Some(store) = &self.store
|
||||||
&& let Err(e) = store.append(&event)
|
&& let Err(e) = store.append(&event)
|
||||||
|
|
|
||||||
|
|
@ -153,6 +153,11 @@ struct StateSnapshot {
|
||||||
/// from the broker via the per-agent socket on each render.
|
/// from the broker via the per-agent socket on each render.
|
||||||
/// Empty on transport failure.
|
/// Empty on transport failure.
|
||||||
inbox: Vec<hive_sh4re::InboxRow>,
|
inbox: Vec<hive_sh4re::InboxRow>,
|
||||||
|
/// Authoritative turn-loop state from the harness and the unix
|
||||||
|
/// timestamp the state was entered. The JS computes the age
|
||||||
|
/// client-side off this rather than tracking it from SSE events.
|
||||||
|
turn_state: crate::events::TurnState,
|
||||||
|
turn_state_since: i64,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Serialize)]
|
#[derive(Serialize)]
|
||||||
|
|
@ -187,12 +192,15 @@ async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
|
||||||
.and_then(|s| s.parse::<u16>().ok())
|
.and_then(|s| s.parse::<u16>().ok())
|
||||||
.unwrap_or(7000);
|
.unwrap_or(7000);
|
||||||
let inbox = recent_inbox(&state.socket, state.flavor).await;
|
let inbox = recent_inbox(&state.socket, state.flavor).await;
|
||||||
|
let (turn_state, turn_state_since) = state.bus.state_snapshot();
|
||||||
axum::Json(StateSnapshot {
|
axum::Json(StateSnapshot {
|
||||||
label: state.label.clone(),
|
label: state.label.clone(),
|
||||||
dashboard_port,
|
dashboard_port,
|
||||||
status,
|
status,
|
||||||
session: session_view,
|
session: session_view,
|
||||||
inbox,
|
inbox,
|
||||||
|
turn_state,
|
||||||
|
turn_state_since,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -359,7 +367,10 @@ async fn post_compact(State(state): State<AppState>) -> Response {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if let Err(e) = crate::turn::compact_session(&settings, &bus).await {
|
bus.set_state(crate::events::TurnState::Compacting);
|
||||||
|
let r = crate::turn::compact_session(&settings, &bus).await;
|
||||||
|
bus.set_state(crate::events::TurnState::Idle);
|
||||||
|
if let Err(e) = r {
|
||||||
bus.emit(crate::events::LiveEvent::Note(format!(
|
bus.emit(crate::events::LiveEvent::Note(format!(
|
||||||
"/compact failed: {e:#}"
|
"/compact failed: {e:#}"
|
||||||
)));
|
)));
|
||||||
|
|
|
||||||
|
|
@ -161,11 +161,66 @@
|
||||||
}
|
}
|
||||||
li.append(actions);
|
li.append(actions);
|
||||||
|
|
||||||
|
// Per-container journald viewer. Expand to fetch + render the
|
||||||
|
// last N lines; refresh button re-fetches; unit selector
|
||||||
|
// narrows to the harness service (or empty = full machine).
|
||||||
|
const journalUnit = c.is_manager ? 'hive-m1nd.service' : 'hive-ag3nt.service';
|
||||||
|
li.append(buildJournalDetails(c.container, journalUnit));
|
||||||
|
|
||||||
ul.append(li);
|
ul.append(li);
|
||||||
}
|
}
|
||||||
root.append(ul);
|
root.append(ul);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Build the per-container journald <details>. Lazy-fetches when the
|
||||||
|
// operator expands; refresh re-fetches; unit toggle switches
|
||||||
|
// between the harness service and the full machine journal.
|
||||||
|
function buildJournalDetails(containerName, defaultUnit) {
|
||||||
|
const details = el('details', { class: 'journal' });
|
||||||
|
const summary = el('summary', {}, '↳ logs · ' + containerName);
|
||||||
|
const body = el('div', { class: 'journal-body' });
|
||||||
|
const controls = el('div', { class: 'journal-controls' });
|
||||||
|
const unitSelect = el('select', { class: 'journal-unit' });
|
||||||
|
unitSelect.append(
|
||||||
|
el('option', { value: defaultUnit }, defaultUnit),
|
||||||
|
el('option', { value: '' }, '(full machine journal)'),
|
||||||
|
);
|
||||||
|
const refresh = el('button', { type: 'button', class: 'btn btn-restart journal-refresh' },
|
||||||
|
'↻ refresh');
|
||||||
|
const pre = el('pre', { class: 'journal-output' }, 'fetching…');
|
||||||
|
let fetching = false;
|
||||||
|
async function fetchLogs() {
|
||||||
|
if (fetching) return;
|
||||||
|
fetching = true;
|
||||||
|
pre.textContent = 'fetching…';
|
||||||
|
const unit = unitSelect.value;
|
||||||
|
const params = new URLSearchParams({ lines: '500' });
|
||||||
|
if (unit) params.set('unit', unit);
|
||||||
|
try {
|
||||||
|
const resp = await fetch('/api/journal/' + containerName + '?' + params);
|
||||||
|
const text = await resp.text();
|
||||||
|
if (!resp.ok) {
|
||||||
|
pre.textContent = 'error: ' + resp.status + '\n' + text;
|
||||||
|
} else {
|
||||||
|
pre.textContent = text || '(empty)';
|
||||||
|
// Auto-scroll to bottom on fresh fetch.
|
||||||
|
pre.scrollTop = pre.scrollHeight;
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
pre.textContent = 'fetch failed: ' + err;
|
||||||
|
} finally {
|
||||||
|
fetching = false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
details.addEventListener('toggle', () => { if (details.open) fetchLogs(); });
|
||||||
|
refresh.addEventListener('click', (e) => { e.preventDefault(); fetchLogs(); });
|
||||||
|
unitSelect.addEventListener('change', fetchLogs);
|
||||||
|
controls.append(unitSelect, refresh);
|
||||||
|
body.append(controls, pre);
|
||||||
|
details.append(summary, body);
|
||||||
|
return details;
|
||||||
|
}
|
||||||
|
|
||||||
function renderTombstones(s) {
|
function renderTombstones(s) {
|
||||||
const root = $('tombstones-section');
|
const root = $('tombstones-section');
|
||||||
root.innerHTML = '';
|
root.innerHTML = '';
|
||||||
|
|
|
||||||
|
|
@ -143,6 +143,53 @@ a:hover {
|
||||||
opacity: 0.85;
|
opacity: 0.85;
|
||||||
}
|
}
|
||||||
.container-row.tombstone .name { color: var(--muted); }
|
.container-row.tombstone .name { color: var(--muted); }
|
||||||
|
/* Per-container journald viewer: collapsed by default, fetches
|
||||||
|
lazily on expand. The output is in monospace inside a bordered
|
||||||
|
<pre>; controls (unit select + refresh) sit above. */
|
||||||
|
.journal {
|
||||||
|
margin-top: 0.5em;
|
||||||
|
font-size: 0.85em;
|
||||||
|
}
|
||||||
|
.journal > summary {
|
||||||
|
cursor: pointer;
|
||||||
|
color: var(--muted);
|
||||||
|
letter-spacing: 0.05em;
|
||||||
|
}
|
||||||
|
.journal > summary:hover { color: var(--cyan); }
|
||||||
|
.journal .journal-body {
|
||||||
|
margin-top: 0.4em;
|
||||||
|
padding-top: 0.4em;
|
||||||
|
border-top: 1px dashed var(--border);
|
||||||
|
}
|
||||||
|
.journal-controls {
|
||||||
|
display: flex;
|
||||||
|
gap: 0.5em;
|
||||||
|
margin-bottom: 0.4em;
|
||||||
|
align-items: center;
|
||||||
|
}
|
||||||
|
.journal-unit {
|
||||||
|
font-family: inherit;
|
||||||
|
font-size: 0.9em;
|
||||||
|
background: var(--bg-elev);
|
||||||
|
color: var(--fg);
|
||||||
|
border: 1px solid var(--border);
|
||||||
|
padding: 0.2em 0.4em;
|
||||||
|
}
|
||||||
|
.journal-refresh { font-size: 0.75em; padding: 0.15em 0.5em; }
|
||||||
|
.journal-output {
|
||||||
|
margin: 0;
|
||||||
|
background: #11111b;
|
||||||
|
color: var(--fg);
|
||||||
|
border: 1px solid var(--purple-dim);
|
||||||
|
padding: 0.5em 0.7em;
|
||||||
|
max-height: 24em;
|
||||||
|
overflow: auto;
|
||||||
|
font-size: 0.85em;
|
||||||
|
line-height: 1.4;
|
||||||
|
white-space: pre;
|
||||||
|
word-break: normal;
|
||||||
|
}
|
||||||
|
|
||||||
.pending-state {
|
.pending-state {
|
||||||
color: var(--amber);
|
color: var(--amber);
|
||||||
font-size: 0.85em;
|
font-size: 0.85em;
|
||||||
|
|
|
||||||
|
|
@ -53,6 +53,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
||||||
.route("/answer-question/{id}", post(post_answer_question))
|
.route("/answer-question/{id}", post(post_answer_question))
|
||||||
.route("/cancel-question/{id}", post(post_cancel_question))
|
.route("/cancel-question/{id}", post(post_cancel_question))
|
||||||
.route("/purge-tombstone/{name}", post(post_purge_tombstone))
|
.route("/purge-tombstone/{name}", post(post_purge_tombstone))
|
||||||
|
.route("/api/journal/{name}", get(get_journal))
|
||||||
.route("/request-spawn", post(post_request_spawn))
|
.route("/request-spawn", post(post_request_spawn))
|
||||||
.route("/messages/stream", get(messages_stream))
|
.route("/messages/stream", get(messages_stream))
|
||||||
.with_state(AppState { coord });
|
.with_state(AppState { coord });
|
||||||
|
|
@ -467,6 +468,76 @@ async fn post_cancel_question(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
struct JournalQuery {
|
||||||
|
/// Optional systemd unit filter — e.g. `hive-ag3nt.service`. When
|
||||||
|
/// omitted, returns the full machine journal.
|
||||||
|
#[serde(default)]
|
||||||
|
unit: Option<String>,
|
||||||
|
/// Number of trailing lines to return. Capped at 5000.
|
||||||
|
#[serde(default)]
|
||||||
|
lines: Option<u32>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Shell out to `journalctl -M <container> -b` and return its text
|
||||||
|
/// output. Operator-only by virtue of the dashboard being host-bound;
|
||||||
|
/// hive-c0re already runs as root in its systemd unit so journalctl
|
||||||
|
/// has the access it needs.
|
||||||
|
async fn get_journal(
|
||||||
|
AxumPath(name): AxumPath<String>,
|
||||||
|
axum::extract::Query(q): axum::extract::Query<JournalQuery>,
|
||||||
|
) -> Response {
|
||||||
|
// Validate the container name against the list of managed
|
||||||
|
// containers so we don't shell out with arbitrary input.
|
||||||
|
let container = strip_container_prefix(&name);
|
||||||
|
let prefixed = if container == lifecycle::MANAGER_NAME {
|
||||||
|
container.clone()
|
||||||
|
} else {
|
||||||
|
format!("{}{container}", lifecycle::AGENT_PREFIX)
|
||||||
|
};
|
||||||
|
let live = lifecycle::list().await.unwrap_or_default();
|
||||||
|
if !live.iter().any(|c| c == &prefixed) {
|
||||||
|
return error_response(&format!("journal: no managed container {prefixed:?}"));
|
||||||
|
}
|
||||||
|
let lines = q.lines.unwrap_or(500).min(5000);
|
||||||
|
let mut cmd = tokio::process::Command::new("journalctl");
|
||||||
|
cmd.args([
|
||||||
|
"-M",
|
||||||
|
&prefixed,
|
||||||
|
"-b",
|
||||||
|
"--no-pager",
|
||||||
|
"--output=short-iso",
|
||||||
|
"--lines",
|
||||||
|
])
|
||||||
|
.arg(lines.to_string());
|
||||||
|
if let Some(u) = q.unit.as_deref().filter(|s| !s.is_empty()) {
|
||||||
|
// accept hive-ag3nt[.service] / hive-m1nd[.service] — anything
|
||||||
|
// else we refuse, again to keep the shell-out tight.
|
||||||
|
let allowed = ["hive-ag3nt.service", "hive-m1nd.service"];
|
||||||
|
let unit = if u.ends_with(".service") {
|
||||||
|
u.to_owned()
|
||||||
|
} else {
|
||||||
|
format!("{u}.service")
|
||||||
|
};
|
||||||
|
if !allowed.contains(&unit.as_str()) {
|
||||||
|
return error_response(&format!("journal: unknown unit {unit:?}"));
|
||||||
|
}
|
||||||
|
cmd.args(["-u", &unit]);
|
||||||
|
}
|
||||||
|
match cmd.output().await {
|
||||||
|
Ok(out) => {
|
||||||
|
// Combine stdout + stderr — journalctl emits to both on errors.
|
||||||
|
let mut body = String::from_utf8_lossy(&out.stdout).into_owned();
|
||||||
|
if !out.status.success() {
|
||||||
|
body.push_str("\n--- stderr ---\n");
|
||||||
|
body.push_str(&String::from_utf8_lossy(&out.stderr));
|
||||||
|
}
|
||||||
|
([("content-type", "text/plain; charset=utf-8")], body).into_response()
|
||||||
|
}
|
||||||
|
Err(e) => error_response(&format!("journalctl spawn: {e}")),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async fn post_purge_tombstone(
|
async fn post_purge_tombstone(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
AxumPath(name): AxumPath<String>,
|
AxumPath(name): AxumPath<String>,
|
||||||
|
|
|
||||||
|
|
@ -45,14 +45,51 @@ const WEB_PORT_RANGE: u16 = 900;
|
||||||
const DEFAULT_MEMORY_MAX: &str = "2G";
|
const DEFAULT_MEMORY_MAX: &str = "2G";
|
||||||
const DEFAULT_CPU_QUOTA: &str = "50%";
|
const DEFAULT_CPU_QUOTA: &str = "50%";
|
||||||
|
|
||||||
/// Returns the per-agent web UI port. Same hash on both sides — manager,
|
/// Returns the per-agent web UI port. Manager is fixed at `MANAGER_PORT`.
|
||||||
/// dashboard, and agent harness all agree. Manager is fixed at
|
/// For sub-agents the port is sticky once chosen: looked up from
|
||||||
/// `MANAGER_PORT`.
|
/// `agent_state_root(name)/port` if present, otherwise derived from
|
||||||
|
/// the FNV-1a hash of the name and *probed forward* through the
|
||||||
|
/// allocated range to skip any port another sub-agent has already
|
||||||
|
/// claimed (birthday-paradox collisions are real even at 2–3
|
||||||
|
/// agents). The chosen port is written back so subsequent calls
|
||||||
|
/// resolve to the same value without re-probing.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn agent_web_port(name: &str) -> u16 {
|
pub fn agent_web_port(name: &str) -> u16 {
|
||||||
if name == MANAGER_NAME {
|
if name == MANAGER_NAME {
|
||||||
return MANAGER_PORT;
|
return MANAGER_PORT;
|
||||||
}
|
}
|
||||||
|
let state_root = crate::coordinator::Coordinator::agent_state_root(name);
|
||||||
|
let port_file = state_root.join("port");
|
||||||
|
if let Ok(s) = std::fs::read_to_string(&port_file)
|
||||||
|
&& let Ok(port) = s.trim().parse::<u16>()
|
||||||
|
&& (WEB_PORT_BASE..WEB_PORT_BASE + WEB_PORT_RANGE).contains(&port)
|
||||||
|
{
|
||||||
|
return port;
|
||||||
|
}
|
||||||
|
let taken = scan_taken_ports(name);
|
||||||
|
let start = port_hash(name);
|
||||||
|
let mut port = start;
|
||||||
|
for _ in 0..WEB_PORT_RANGE {
|
||||||
|
if !taken.contains(&port) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
port = next_port(port);
|
||||||
|
if port == start {
|
||||||
|
// Range fully exhausted (very unlikely — 900 slots) —
|
||||||
|
// give up and just use the hashed value; collisions are
|
||||||
|
// surfaced as bind errors by the harness retry loop.
|
||||||
|
tracing::warn!(%name, "agent_web_port: range exhausted, returning hash");
|
||||||
|
return start;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let _ = std::fs::create_dir_all(&state_root);
|
||||||
|
if let Err(e) = std::fs::write(&port_file, format!("{port}\n")) {
|
||||||
|
tracing::warn!(error = ?e, file = %port_file.display(), "persisting agent port failed");
|
||||||
|
}
|
||||||
|
port
|
||||||
|
}
|
||||||
|
|
||||||
|
fn port_hash(name: &str) -> u16 {
|
||||||
let mut hash: u32 = 2_166_136_261;
|
let mut hash: u32 = 2_166_136_261;
|
||||||
for b in name.bytes() {
|
for b in name.bytes() {
|
||||||
hash ^= u32::from(b);
|
hash ^= u32::from(b);
|
||||||
|
|
@ -62,6 +99,48 @@ pub fn agent_web_port(name: &str) -> u16 {
|
||||||
WEB_PORT_BASE + u16::try_from(hash % u32::from(WEB_PORT_RANGE)).unwrap_or(0)
|
WEB_PORT_BASE + u16::try_from(hash % u32::from(WEB_PORT_RANGE)).unwrap_or(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn next_port(port: u16) -> u16 {
|
||||||
|
let p = port + 1;
|
||||||
|
if p >= WEB_PORT_BASE + WEB_PORT_RANGE {
|
||||||
|
WEB_PORT_BASE
|
||||||
|
} else {
|
||||||
|
p
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Scan every other agent's effective web UI port: prefer the
|
||||||
|
/// persisted `port` file when present, fall back to the hashed
|
||||||
|
/// value for legacy agents that pre-date the port-file scheme. The
|
||||||
|
/// latter is important on existing deployments — without it, a new
|
||||||
|
/// agent's collision check wouldn't see incumbents that haven't
|
||||||
|
/// written their port file yet, and we'd re-emit the same
|
||||||
|
/// collision the operator just hit.
|
||||||
|
fn scan_taken_ports(name: &str) -> std::collections::HashSet<u16> {
|
||||||
|
let mut out = std::collections::HashSet::new();
|
||||||
|
let Ok(rd) = std::fs::read_dir("/var/lib/hyperhive/agents") else {
|
||||||
|
return out;
|
||||||
|
};
|
||||||
|
for entry in rd.flatten() {
|
||||||
|
let Ok(file_name) = entry.file_name().into_string() else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
if file_name == name || file_name == MANAGER_NAME {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let pf = entry.path().join("port");
|
||||||
|
if let Ok(s) = std::fs::read_to_string(&pf)
|
||||||
|
&& let Ok(port) = s.trim().parse::<u16>()
|
||||||
|
{
|
||||||
|
out.insert(port);
|
||||||
|
} else {
|
||||||
|
// Legacy: no port file yet → its effective port is the
|
||||||
|
// bare hash. Treat as taken so we don't collide with it.
|
||||||
|
out.insert(port_hash(&file_name));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn container_name(name: &str) -> String {
|
pub fn container_name(name: &str) -> String {
|
||||||
if name == MANAGER_NAME {
|
if name == MANAGER_NAME {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue