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
|
||||
|
||||
- **State badge: compacting + napping states.** Idle/thinking already
|
||||
ship (driven from SSE turn_start/turn_end). Add `compacting 📦` and
|
||||
`napping 😴` once the `/compact` trigger and `nap` tool exist —
|
||||
both need a harness signal (an explicit `LiveEvent::StateChange`
|
||||
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.
|
||||
- **State badge: napping state.** Idle / thinking / compacting
|
||||
already ship from server-side `TurnState`. Add `napping 😴`
|
||||
once the `nap` tool exists — it just adds a new `TurnState`
|
||||
variant the harness flips into for the duration of the nap.
|
||||
- **Terminal: `/model` slash command.** Operator-typeable model
|
||||
override from the terminal. Depends on the model-override work
|
||||
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
|
||||
|
||||
- **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
|
||||
`HelperEvent::ContainerCrash` to the manager's inbox so the manager can
|
||||
react (restart, escalate, etc.).
|
||||
|
|
|
|||
|
|
@ -228,6 +228,11 @@ pre.diff {
|
|||
text-shadow: 0 0 6px rgba(250, 179, 135, 0.65);
|
||||
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 {
|
||||
animation: state-flash 600ms ease-out;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -291,10 +291,11 @@
|
|||
// each second so the "· 12s" suffix stays current. State changes
|
||||
// trigger a short flash animation via .state-just-changed.
|
||||
const STATE_LABELS = {
|
||||
loading: { glyph: '…', text: 'booting' },
|
||||
offline: { glyph: '○', text: 'offline' },
|
||||
idle: { glyph: '💤', text: 'idle' },
|
||||
thinking: { glyph: '🧠', text: 'thinking' },
|
||||
loading: { glyph: '…', text: 'booting' },
|
||||
offline: { glyph: '○', text: 'offline' },
|
||||
idle: { glyph: '💤', text: 'idle' },
|
||||
thinking: { glyph: '🧠', text: 'thinking' },
|
||||
compacting: { glyph: '📦', text: 'compacting' },
|
||||
};
|
||||
let stateName = 'loading';
|
||||
let stateSince = Date.now();
|
||||
|
|
@ -318,19 +319,22 @@
|
|||
if (cancelBtn) cancelBtn.hidden = stateName !== 'thinking';
|
||||
}
|
||||
function setState(next) {
|
||||
if (next === stateName) return;
|
||||
// Capture the just-ending state's duration when leaving 'thinking'
|
||||
// so the operator can eyeball turn length without scrolling the
|
||||
// terminal back.
|
||||
setStateAbs(next, Math.floor(Date.now() / 1000));
|
||||
}
|
||||
/// Set state with an authoritative since-unix from the server. Lets
|
||||
/// `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') {
|
||||
const elapsedMs = Date.now() - stateSince;
|
||||
renderLastTurn(elapsedMs);
|
||||
}
|
||||
const flashing = next !== stateName;
|
||||
stateName = next;
|
||||
stateSince = Date.now();
|
||||
stateSince = sinceUnix * 1000;
|
||||
const badge = $('state-badge');
|
||||
if (badge) {
|
||||
// Re-add the flash class so the animation replays.
|
||||
if (badge && flashing) {
|
||||
badge.classList.remove('state-just-changed');
|
||||
void badge.offsetWidth;
|
||||
badge.classList.add('state-just-changed');
|
||||
|
|
@ -411,11 +415,15 @@
|
|||
if (!headerSet) { setHeader(s.label, s.dashboard_port); headerSet = true; }
|
||||
renderTermInput(s.label, s.status === 'online');
|
||||
renderInbox(s.inbox || []);
|
||||
// Drive the state badge from the harness status. Live SSE events
|
||||
// override to 'thinking' / 'idle' as turns start/end; this only
|
||||
// kicks in for the not-online (offline) case and the initial seed.
|
||||
if (s.status !== 'online') setState('offline');
|
||||
else if (stateName === 'loading' || stateName === 'offline') setState('idle');
|
||||
// Authoritative state comes from the harness via /api/state.
|
||||
// Login-not-yet → 'offline'; otherwise use the server-reported
|
||||
// turn_state (idle / thinking / compacting). stateSince in
|
||||
// unix-seconds is converted to a client-side Date.now() anchor.
|
||||
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
|
||||
// common case is `online` polling itself — without this guard, the
|
||||
// operator's <input value> gets clobbered every cycle.
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ use std::time::Duration;
|
|||
|
||||
use anyhow::Result;
|
||||
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::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
||||
use hive_sh4re::{AgentRequest, AgentResponse};
|
||||
|
|
@ -126,6 +126,7 @@ async fn serve(
|
|||
body: body.clone(),
|
||||
unread,
|
||||
});
|
||||
bus.set_state(TurnState::Thinking);
|
||||
let prompt = format_wake_prompt(&from, &body, unread);
|
||||
let outcome = turn::drive_turn(
|
||||
&prompt,
|
||||
|
|
@ -137,6 +138,7 @@ async fn serve(
|
|||
)
|
||||
.await;
|
||||
turn::emit_turn_end(&bus, &outcome);
|
||||
bus.set_state(TurnState::Idle);
|
||||
}
|
||||
Ok(AgentResponse::Empty) => {}
|
||||
Ok(AgentResponse::Ok | AgentResponse::Status { .. } | AgentResponse::Recent { .. }) => {
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ use std::time::Duration;
|
|||
|
||||
use anyhow::Result;
|
||||
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::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
||||
use hive_sh4re::{HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER};
|
||||
|
|
@ -124,6 +124,7 @@ async fn serve(socket: &Path, interval: Duration, bus: Bus) -> Result<()> {
|
|||
unread,
|
||||
});
|
||||
let prompt = format_wake_prompt(&from, &body, unread);
|
||||
bus.set_state(TurnState::Thinking);
|
||||
let outcome = turn::drive_turn(
|
||||
&prompt,
|
||||
&mcp_config,
|
||||
|
|
@ -134,6 +135,7 @@ async fn serve(socket: &Path, interval: Duration, bus: Bus) -> Result<()> {
|
|||
)
|
||||
.await;
|
||||
turn::emit_turn_end(&bus, &outcome);
|
||||
bus.set_state(TurnState::Idle);
|
||||
}
|
||||
Ok(ManagerResponse::Empty) => {}
|
||||
Ok(
|
||||
|
|
|
|||
|
|
@ -24,6 +24,14 @@ const HISTORY_CAPACITY: usize = 2000;
|
|||
/// `HYPERHIVE_EVENTS_DB` env var (used in tests and one-shot tools).
|
||||
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 = "
|
||||
CREATE TABLE IF NOT EXISTS events (
|
||||
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)]
|
||||
pub struct Bus {
|
||||
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
|
||||
/// missing `/state/` mount in dev / test scenarios.
|
||||
store: Option<Arc<EventStore>>,
|
||||
/// Current turn-loop state + since-when (unix seconds).
|
||||
state: Arc<Mutex<(TurnState, i64)>>,
|
||||
}
|
||||
|
||||
impl Bus {
|
||||
|
|
@ -144,9 +170,26 @@ impl Bus {
|
|||
Self {
|
||||
tx: Arc::new(tx),
|
||||
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) {
|
||||
if let Some(store) = &self.store
|
||||
&& let Err(e) = store.append(&event)
|
||||
|
|
|
|||
|
|
@ -153,6 +153,11 @@ struct StateSnapshot {
|
|||
/// from the broker via the per-agent socket on each render.
|
||||
/// Empty on transport failure.
|
||||
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)]
|
||||
|
|
@ -187,12 +192,15 @@ async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
|
|||
.and_then(|s| s.parse::<u16>().ok())
|
||||
.unwrap_or(7000);
|
||||
let inbox = recent_inbox(&state.socket, state.flavor).await;
|
||||
let (turn_state, turn_state_since) = state.bus.state_snapshot();
|
||||
axum::Json(StateSnapshot {
|
||||
label: state.label.clone(),
|
||||
dashboard_port,
|
||||
status,
|
||||
session: session_view,
|
||||
inbox,
|
||||
turn_state,
|
||||
turn_state_since,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -359,7 +367,10 @@ async fn post_compact(State(state): State<AppState>) -> Response {
|
|||
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!(
|
||||
"/compact failed: {e:#}"
|
||||
)));
|
||||
|
|
|
|||
|
|
@ -161,11 +161,66 @@
|
|||
}
|
||||
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);
|
||||
}
|
||||
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) {
|
||||
const root = $('tombstones-section');
|
||||
root.innerHTML = '';
|
||||
|
|
|
|||
|
|
@ -143,6 +143,53 @@ a:hover {
|
|||
opacity: 0.85;
|
||||
}
|
||||
.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 {
|
||||
color: var(--amber);
|
||||
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("/cancel-question/{id}", post(post_cancel_question))
|
||||
.route("/purge-tombstone/{name}", post(post_purge_tombstone))
|
||||
.route("/api/journal/{name}", get(get_journal))
|
||||
.route("/request-spawn", post(post_request_spawn))
|
||||
.route("/messages/stream", get(messages_stream))
|
||||
.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(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(name): AxumPath<String>,
|
||||
|
|
|
|||
|
|
@ -45,14 +45,51 @@ const WEB_PORT_RANGE: u16 = 900;
|
|||
const DEFAULT_MEMORY_MAX: &str = "2G";
|
||||
const DEFAULT_CPU_QUOTA: &str = "50%";
|
||||
|
||||
/// Returns the per-agent web UI port. Same hash on both sides — manager,
|
||||
/// dashboard, and agent harness all agree. Manager is fixed at
|
||||
/// `MANAGER_PORT`.
|
||||
/// Returns the per-agent web UI port. Manager is fixed at `MANAGER_PORT`.
|
||||
/// For sub-agents the port is sticky once chosen: looked up from
|
||||
/// `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]
|
||||
pub fn agent_web_port(name: &str) -> u16 {
|
||||
if name == MANAGER_NAME {
|
||||
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;
|
||||
for b in name.bytes() {
|
||||
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)
|
||||
}
|
||||
|
||||
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]
|
||||
pub fn container_name(name: &str) -> String {
|
||||
if name == MANAGER_NAME {
|
||||
|
|
|
|||
Loading…
Reference in a new issue