diff --git a/TODO.md b/TODO.md index 684c0d68..a55d9170 100644 --- a/TODO.md +++ b/TODO.md @@ -16,32 +16,21 @@ Pick anything from here when relevant. Cross-cutting design notes live in claude-code's `--allowedTools` extended grammar. Likely lives in `agent.nix` so each agent can scope its own shell surface. -## Per-agent extension - -- **Custom per-agent MCP tools.** Today every sub-agent gets the - same fixed MCP surface (`send`, `recv`). To move bitburner-agent - (and anything else with rich domain tooling) into hyperhive, an - agent needs a way to ship its own tools alongside hyperhive's. - Sketch: `agent.nix` declares a list of extra MCP servers - (command + args + env), each registered into the agent's - `--mcp-config` blob at flake-render time. The harness MCP server - remains the hyperhive surface; new servers slot in as additional - entries under `mcpServers.` so claude sees them as - `mcp____`. Per-agent tool whitelist (`allowedTools`) - derived from the same config so the operator stays in control of - what's exposed. - ## Per-agent settings -- **Model override persistence.** `/model ` already switches - the model at runtime via `Bus::set_model`; the chip on the agent - page reflects the current value. Override is in-memory only and - resets on harness restart โ€” by design for now, but consider - optional persistence (`/state/model` file?) so an operator-set - model survives a rebuild. +- **Model override.** Hard-coded to `haiku` in the turn loop right now. + Surface as a per-agent override: operator via dashboard, manager via + `request_apply_commit` setting an attr on the agent's flake (most natural + place since the flake already carries per-agent env/identity). Pair with + a **model status** indicator on the agent page (active / queued / last + switched) once the override is in place. ## UI / UX +- **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 ` @@ -92,6 +81,14 @@ Pick anything from here when relevant. Cross-cutting design notes live in ## Loop substance +- **`nap` tool.** Agent-side MCP tool `mcp__hyperhive__nap(seconds)` that + parks the turn loop for a short while before next-message processing. + Use cases: agent decides it has nothing useful to do, or wants to + throttle itself between rapid wake events. Implementation: harness + records a "wake-not-before" timestamp; `recv_blocking` skips the long + poll until that ts; the state badge reads `napping ยท MM:SS` during. + Operator can cancel via the same `/cancel` slash command or a + dashboard button. - **Notes compaction.** `/state/` is bind-mounted persistently and agents are told (in the system prompt) to keep `/state/notes.md` for durable knowledge โ€” but we don't currently nudge them to compact when notes diff --git a/hive-ag3nt/assets/agent.css b/hive-ag3nt/assets/agent.css index b83c8029..aeb165a6 100644 --- a/hive-ag3nt/assets/agent.css +++ b/hive-ag3nt/assets/agent.css @@ -171,15 +171,6 @@ pre.diff { font-size: 0.8em; letter-spacing: 0.05em; } -.model-chip { - display: inline-block; - padding: 0.1em 0.6em; - border: 1px solid var(--purple-dim); - border-radius: 999px; - color: var(--cyan); - font-size: 0.78em; - letter-spacing: 0.04em; -} .btn-dashlink { color: var(--cyan); border: 1px solid var(--cyan); diff --git a/hive-ag3nt/assets/app.js b/hive-ag3nt/assets/app.js index f4670fd2..1e506797 100644 --- a/hive-ag3nt/assets/app.js +++ b/hive-ag3nt/assets/app.js @@ -174,31 +174,8 @@ { name: '/clear', desc: 'wipe the terminal panel (local-only)' }, { name: '/cancel', desc: 'SIGINT the in-flight claude turn' }, { name: '/compact', desc: 'compact the persistent claude session' }, - { name: '/model', desc: '/model โ€” switch claude model for future turns' }, ]; - async function postModel(name) { - try { - const resp = await fetch('/api/model', { - method: 'POST', - headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, - body: new URLSearchParams({ model: name }), - redirect: 'manual', - }); - const ok = resp.ok || resp.type === 'opaqueredirect' - || (resp.status >= 200 && resp.status < 400); - if (!ok && termAPI) { - const text = await resp.text().catch(() => ''); - termAPI.row('turn-end-fail', 'โœ— /model failed: ' + resp.status - + (text ? ' โ€” ' + text : '')); - } else { - refreshState(); - } - } catch (err) { - if (termAPI) termAPI.row('turn-end-fail', 'โœ— /model failed: ' + err); - } - } - async function postSimple(url, label) { try { const resp = await fetch(url, { method: 'POST', redirect: 'manual' }); @@ -236,16 +213,6 @@ case '/compact': postCompact(); return true; - case '/model': { - const parts = trimmed.split(/\s+/); - if (parts.length < 2 || !parts[1]) { - termAPI.row('turn-end-fail', - 'โœ— /model needs a name (e.g. /model haiku, /model sonnet, /model opus)'); - } else { - postModel(parts[1]); - } - return true; - } default: termAPI.row('turn-end-fail', 'โœ— unknown slash command: ' + cmd + ' โ€” try /help'); return true; @@ -398,13 +365,6 @@ list.append(li); } } - function renderModelChip(model) { - const el_ = $('model-chip'); - if (!el_) return; - if (!model) { el_.hidden = true; return; } - el_.hidden = false; - el_.textContent = 'model ยท ' + model; - } function renderLastTurn(ms) { const el_ = $('last-turn'); if (!el_) return; @@ -464,7 +424,6 @@ } else if (s.turn_state) { setStateAbs(s.turn_state, s.turn_state_since); } - renderModelChip(s.model); // Skip the re-render if nothing structurally changed. The most // common case is `online` polling itself โ€” without this guard, the // operator's gets clobbered every cycle. diff --git a/hive-ag3nt/assets/index.html b/hive-ag3nt/assets/index.html index bde7c6ac..9906176c 100644 --- a/hive-ag3nt/assets/index.html +++ b/hive-ag3nt/assets/index.html @@ -15,7 +15,6 @@
โ€ฆ booting -
diff --git a/hive-ag3nt/prompts/agent.md b/hive-ag3nt/prompts/agent.md index 724d6739..a722f086 100644 --- a/hive-ag3nt/prompts/agent.md +++ b/hive-ag3nt/prompts/agent.md @@ -2,7 +2,7 @@ You are hyperhive agent `{label}` in a multi-agent system. Tools (hyperhive surface): -- `mcp__hyperhive__recv(wait_seconds?)` โ€” drain one more message from your inbox (returns `(empty)` if nothing pending after the wait). Without `wait_seconds` it long-polls 30s. To **wait** for work when you have nothing else useful to do this turn, call with a long wait (e.g. `wait_seconds: 180`, the max) โ€” you'll be woken instantly when a message arrives, otherwise return after the timeout. That is strictly better than calling `recv` repeatedly with short waits: lower latency on new work, fewer turns, no busy-loop. Never use a fixed `sleep` shell command for the same purpose. +- `mcp__hyperhive__recv()` โ€” drain one more message from your inbox (returns `(empty)` if nothing pending). - `mcp__hyperhive__send(to, body)` โ€” message a peer (by their name) or the operator (recipient `operator`, surfaces in the dashboard). Need new packages, env vars, or other NixOS config for yourself? You can't edit your own config directly โ€” message the manager (recipient `manager`) describing what you need + why. The manager evaluates the request (it doesn't rubber-stamp), edits `/agents/{label}/config/agent.nix` on your behalf, commits, and submits an approval that the operator can accept on the dashboard; on approve hive-c0re rebuilds your container with the new config. diff --git a/hive-ag3nt/prompts/manager.md b/hive-ag3nt/prompts/manager.md index 3cf86dc0..ca96cae8 100644 --- a/hive-ag3nt/prompts/manager.md +++ b/hive-ag3nt/prompts/manager.md @@ -2,7 +2,7 @@ You are the hyperhive manager `{label}` in a multi-agent system. You coordinate Tools (hyperhive surface): -- `mcp__hyperhive__recv(wait_seconds?)` โ€” drain one more message from your inbox. Without `wait_seconds` it long-polls 30s. To **wait** when you have nothing else to do, call with a long wait (e.g. `wait_seconds: 180`, the max) โ€” you'll wake instantly on new work, otherwise return after the timeout. Use this instead of ending the turn or sleeping in a Bash command. +- `mcp__hyperhive__recv()` โ€” drain one more message from your inbox. - `mcp__hyperhive__send(to, body)` โ€” message an agent (by name), another peer, or the operator (`operator` surfaces in the dashboard). - `mcp__hyperhive__request_spawn(name)` โ€” queue a brand-new sub-agent for operator approval (โ‰ค9 char name). - `mcp__hyperhive__kill(name)` โ€” graceful stop on a sub-agent. No approval required. diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index 7c661a0f..ed9f32bc 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -116,8 +116,7 @@ async fn serve( let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hive-ag3nt".into()); let system_prompt = turn::write_system_prompt(socket, &label, mcp::Flavor::Agent).await?; loop { - let recv: Result = - client::request(socket, &AgentRequest::Recv { wait_seconds: None }).await; + let recv: Result = client::request(socket, &AgentRequest::Recv).await; match recv { Ok(AgentResponse::Message { from, body }) => { tracing::info!(%from, %body, "inbox"); diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index bf997162..7d3113cf 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -96,8 +96,7 @@ async fn serve(socket: &Path, interval: Duration, bus: Bus) -> Result<()> { let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hm1nd".into()); let system_prompt = turn::write_system_prompt(socket, &label, mcp::Flavor::Manager).await?; loop { - let recv: Result = - client::request(socket, &ManagerRequest::Recv { wait_seconds: None }).await; + let recv: Result = client::request(socket, &ManagerRequest::Recv).await; match recv { Ok(ManagerResponse::Message { from, body }) => { if from == SYSTEM_SENDER { diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 23b138f5..40251370 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -140,13 +140,6 @@ pub enum TurnState { Compacting, } -/// Default claude model when nothing's been set at runtime. The -/// operator can switch via `/model ` in the web terminal; the -/// chosen model lives in `Bus::model` for the rest of the harness -/// process's life (resets on restart, by design โ€” operator overrides -/// shouldn't survive accidentally). -pub const DEFAULT_MODEL: &str = "haiku"; - #[derive(Clone)] pub struct Bus { tx: Arc>, @@ -156,9 +149,6 @@ pub struct Bus { store: Option>, /// Current turn-loop state + since-when (unix seconds). state: Arc>, - /// Model name passed to `claude --model`. Default `haiku`; the - /// operator can override at runtime via `POST /api/model`. - model: Arc>, } impl Bus { @@ -181,23 +171,9 @@ impl Bus { tx: Arc::new(tx), store, state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))), - model: Arc::new(Mutex::new(DEFAULT_MODEL.to_owned())), } } - /// Currently-selected claude model name. Read on every turn so a - /// `/model ` flip takes effect on the next turn. - #[must_use] - pub fn model(&self) -> String { - self.model.lock().unwrap().clone() - } - - /// Switch the model for future turns. The current turn (if any) - /// keeps the model it was already running. - pub fn set_model(&self, name: impl Into) { - *self.model.lock().unwrap() = name.into(); - } - /// 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) { diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index bd38c058..d47a021a 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -116,14 +116,7 @@ pub struct SendArgs { } #[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct RecvArgs { - /// How long to long-poll for a new message before returning the - /// empty marker. Capped at 60s server-side. Default (None) is - /// 30s. Useful when an agent wants to throttle wakes without - /// actually napping โ€” pick a longer wait to coalesce bursts. - #[serde(default)] - pub wait_seconds: Option, -} +pub struct RecvArgs {} /// Per-agent tool surface. Holds the socket path so each tool call doesn't /// re-derive it; the socket itself is the per-container `/run/hive/mcp.sock`. @@ -165,20 +158,13 @@ impl AgentServer { #[tool( description = "Pop one message from this agent's inbox. Returns the sender and body, \ - or an empty marker if nothing is waiting. Optional `wait_seconds` long-polls \ - for that many seconds (capped at 180) before returning empty โ€” default 30. \ - Use a long wait_seconds (e.g. 120 or 180) when you have nothing else to do โ€” \ - it parks the turn until either a message arrives or the timeout fires, which \ - is strictly better than a fixed sleep because incoming work wakes you instantly." + or an empty marker if nothing is waiting." )] - async fn recv(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("recv", log, async move { + async fn recv(&self, Parameters(_args): Parameters) -> String { + run_tool_envelope("recv", String::new(), async move { let resp = client::request::<_, hive_sh4re::AgentResponse>( &self.socket, - &hive_sh4re::AgentRequest::Recv { - wait_seconds: args.wait_seconds, - }, + &hive_sh4re::AgentRequest::Recv, ) .await .map(SocketReply::from); @@ -317,19 +303,11 @@ impl ManagerServer { #[tool( description = "Pop one message from the manager inbox. Returns sender + body, or \ - empty. Optional `wait_seconds` long-polls (capped at 180, default 30) so the \ - manager can sit on Recv when there's nothing to do without burning turns โ€” \ - prefer a long wait (120 or 180) over ending a turn early; you'll wake \ - instantly when work arrives." + empty." )] - async fn recv(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("recv", log, async move { - let resp = self - .dispatch(hive_sh4re::ManagerRequest::Recv { - wait_seconds: args.wait_seconds, - }) - .await; + async fn recv(&self, Parameters(_args): Parameters) -> String { + run_tool_envelope("recv", String::new(), async move { + let resp = self.dispatch(hive_sh4re::ManagerRequest::Recv).await; format_recv(resp) }) .await diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 6dd2643e..b5ca8cc4 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -227,14 +227,13 @@ async fn run_claude( flavor: mcp::Flavor, mode: ClaudeMode, ) -> Result { - let model = bus.model(); let mut cmd = Command::new("claude"); cmd.arg("--print") .arg("--verbose") .arg("--output-format") .arg("stream-json") .arg("--model") - .arg(&model) + .arg("haiku") .arg("--continue") .arg("--settings") .arg(settings); diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index cb4acefc..e03c7409 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -81,7 +81,6 @@ pub async fn serve( .route("/login/cancel", post(post_login_cancel)) .route("/api/cancel", post(post_cancel_turn)) .route("/api/compact", post(post_compact)) - .route("/api/model", post(post_set_model)) .with_state(state); let addr = SocketAddr::from(([0, 0, 0, 0], port)); let listener = bind_with_retry(addr, "web UI").await?; @@ -94,17 +93,16 @@ pub async fn serve( // Static assets + state snapshot // --------------------------------------------------------------------------- -/// Bind a TCP listener with `SO_REUSEADDR` set, retrying on -/// `AddrInUse` for up to ~20s. nspawn restarts can race the previous -/// harness's socket release; `SO_REUSEADDR` lets us reclaim a port -/// still in `TIME_WAIT` from a clean previous exit, and the retry -/// covers the case where the previous process is genuinely still -/// alive (systemd restart-delay overlap). +/// Bind a TCP listener, retrying on `AddrInUse` for up to ~20s. +/// nspawn restarts can race the previous harness's socket release; +/// without retry the new harness fails to bind and systemd just +/// keeps restarting it. `SO_REUSEADDR` would be the proper fix but +/// would require socket2; retry is good enough here. async fn bind_with_retry(addr: SocketAddr, label: &str) -> Result { let mut delay_ms = 250u64; let mut attempts = 0u32; loop { - match try_bind(addr) { + match tokio::net::TcpListener::bind(addr).await { Ok(l) => return Ok(l), Err(e) if e.kind() == std::io::ErrorKind::AddrInUse && attempts < 12 => { tracing::warn!( @@ -122,16 +120,6 @@ async fn bind_with_retry(addr: SocketAddr, label: &str) -> Result std::io::Result { - let sock = match addr { - SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?, - SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?, - }; - sock.set_reuseaddr(true)?; - sock.bind(addr)?; - sock.listen(1024) -} - async fn serve_index() -> impl IntoResponse { ( [("content-type", "text/html; charset=utf-8")], @@ -170,10 +158,6 @@ struct StateSnapshot { /// client-side off this rather than tracking it from SSE events. turn_state: crate::events::TurnState, turn_state_since: i64, - /// Currently-active claude model name. Reflected on the page so - /// the operator can see what they just switched to (and what's - /// in flight). Mutable at runtime via `POST /api/model`. - model: String, } #[derive(Serialize)] @@ -209,7 +193,6 @@ async fn api_state(State(state): State) -> axum::Json { .unwrap_or(7000); let inbox = recent_inbox(&state.socket, state.flavor).await; let (turn_state, turn_state_since) = state.bus.state_snapshot(); - let model = state.bus.model(); axum::Json(StateSnapshot { label: state.label.clone(), dashboard_port, @@ -218,7 +201,6 @@ async fn api_state(State(state): State) -> axum::Json { inbox, turn_state, turn_state_since, - model, }) } @@ -369,30 +351,6 @@ async fn post_login_cancel(State(state): State) -> Response { /// the "/compact done" note) lands in the live event panel like any /// other turn. If a regular turn is in flight, claude's own session /// lock will reject this one and we surface the error as a Note. -#[derive(Deserialize)] -struct ModelForm { - model: String, -} - -/// Switch the model for future turns. The current turn (if any) -/// keeps its model; `/model ` applies starting with the next -/// `recv` cycle. Empty / whitespace-only inputs are rejected. No -/// claude-side validation โ€” we just hand the string through to -/// `claude --model `; an unknown model surfaces as a turn -/// failure in the live panel and the operator can revert. -async fn post_set_model(State(state): State, Form(form): Form) -> Response { - let name = form.model.trim(); - if name.is_empty() { - return error_response("model: name required"); - } - state.bus.set_model(name); - state.bus.emit(crate::events::LiveEvent::Note(format!( - "operator: /model โ€” claude model set to '{name}' for future turns" - ))); - tracing::info!(%name, "operator set model"); - Redirect::to("/").into_response() -} - async fn post_compact(State(state): State) -> Response { let bus = state.bus.clone(); let socket = state.socket.clone(); diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index c47e5b45..1dacf469 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -77,21 +77,9 @@ async fn serve(stream: UnixStream, agent: String, broker: Arc) -> Result } } -/// Default and max long-poll window for `Recv`. Caller can request a -/// shorter (or longer up to `RECV_LONG_POLL_MAX`) wait via the -/// `wait_seconds` field; values above the cap are clamped. 180s -/// max keeps us under typical TCP/proxy idle limits while letting -/// agents park their turn until a message lands instead of busy- -/// looping with short waits. -const RECV_LONG_POLL_DEFAULT: std::time::Duration = std::time::Duration::from_secs(30); -const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180); - -fn recv_timeout(wait_seconds: Option) -> std::time::Duration { - match wait_seconds { - Some(s) => std::time::Duration::from_secs(s).min(RECV_LONG_POLL_MAX), - None => RECV_LONG_POLL_DEFAULT, - } -} +/// How long the long-poll `Recv` holds a connection open waiting for new +/// mail. Set well below typical TCP/proxy idle limits. +const RECV_LONG_POLL: std::time::Duration = std::time::Duration::from_secs(30); async fn dispatch(req: &AgentRequest, agent: &str, broker: &Broker) -> AgentResponse { match req { @@ -107,10 +95,7 @@ async fn dispatch(req: &AgentRequest, agent: &str, broker: &Broker) -> AgentResp }, } } - AgentRequest::Recv { wait_seconds } => match broker - .recv_blocking(agent, recv_timeout(*wait_seconds)) - .await - { + AgentRequest::Recv => match broker.recv_blocking(agent, RECV_LONG_POLL).await { Ok(Some(msg)) => AgentResponse::Message { from: msg.from, body: msg.body, diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index ddf700d5..6430d7c6 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -72,13 +72,13 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { // `/messages/stream` for broker traffic. // --------------------------------------------------------------------------- -/// `SO_REUSEADDR` bind with retry. Mirrors the per-agent variant โ€” +/// Retry-on-AddrInUse bind. Same shape as the per-agent variant โ€” /// hive-c0re restarts also race the previous process's socket release. async fn bind_with_retry(addr: SocketAddr) -> Result { let mut delay_ms = 250u64; let mut attempts = 0u32; loop { - match try_bind(addr) { + match tokio::net::TcpListener::bind(addr).await { Ok(l) => return Ok(l), Err(e) if e.kind() == std::io::ErrorKind::AddrInUse && attempts < 12 => { tracing::warn!( @@ -96,16 +96,6 @@ async fn bind_with_retry(addr: SocketAddr) -> Result { } } -fn try_bind(addr: SocketAddr) -> std::io::Result { - let sock = match addr { - SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?, - SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?, - }; - sock.set_reuseaddr(true)?; - sock.bind(addr)?; - sock.listen(1024) -} - async fn serve_index() -> impl IntoResponse { Html(include_str!("../assets/index.html")) } diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs index 2ddf30c6..f6757351 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -46,18 +46,13 @@ const DEFAULT_MEMORY_MAX: &str = "2G"; const DEFAULT_CPU_QUOTA: &str = "50%"; /// Returns the per-agent web UI port. Manager is fixed at `MANAGER_PORT`. -/// For sub-agents the port is sticky once chosen: -/// -/// - **Port file present** (`state_root/port`): use it. End of story. -/// - **Port file absent, applied flake present**: this is a legacy -/// agent whose container is already bound to the bare -/// `port_hash(name)`. Don't probe; just migrate by writing that -/// value to the port file. The container stays where it is and -/// subsequent renders agree with it. -/// - **Port file absent, no applied flake**: this is a fresh spawn. -/// Probe forward from `port_hash(name)` to skip any port another -/// sub-agent has already claimed (via port file or legacy hash). -/// Write the chosen port back. +/// 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 { @@ -71,36 +66,27 @@ pub fn agent_web_port(name: &str) -> u16 { { return port; } - let applied_exists = crate::coordinator::Coordinator::agent_applied_dir(name).exists(); - let chosen = if applied_exists { - // Legacy agent โ€” container already running on the hashed - // port. Don't move it; just persist the value so future - // calls bypass this path. - port_hash(name) - } else { - 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 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"); - break; - } + 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 - }; + 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!("{chosen}\n")) { + if let Err(e) = std::fs::write(&port_file, format!("{port}\n")) { tracing::warn!(error = ?e, file = %port_file.display(), "persisting agent port failed"); } - chosen + port } fn port_hash(name: &str) -> u16 { diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 4050641c..d946b276 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -69,17 +69,7 @@ async fn serve(stream: UnixStream, coord: Arc) -> Result<()> { } } -/// Default and max long-poll window for manager `Recv`. Caller can -/// request a shorter or longer (up to MAX) wait via `wait_seconds`. -const MANAGER_RECV_LONG_POLL_DEFAULT: std::time::Duration = std::time::Duration::from_secs(30); -const MANAGER_RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180); - -fn manager_recv_timeout(wait_seconds: Option) -> std::time::Duration { - match wait_seconds { - Some(s) => std::time::Duration::from_secs(s).min(MANAGER_RECV_LONG_POLL_MAX), - None => MANAGER_RECV_LONG_POLL_DEFAULT, - } -} +const MANAGER_RECV_LONG_POLL: std::time::Duration = std::time::Duration::from_secs(30); #[allow(clippy::too_many_lines)] async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResponse { @@ -116,9 +106,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp message: format!("{e:#}"), }, }, - ManagerRequest::Recv { wait_seconds } => match coord + ManagerRequest::Recv => match coord .broker - .recv_blocking(MANAGER_AGENT, manager_recv_timeout(*wait_seconds)) + .recv_blocking(MANAGER_AGENT, MANAGER_RECV_LONG_POLL) .await { Ok(Some(msg)) => ManagerResponse::Message { diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 90949d82..52f80004 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -166,13 +166,8 @@ pub struct InboxRow { pub enum AgentRequest { /// Send a message to another agent. Send { to: String, body: String }, - /// Pop one pending message from this agent's inbox. Long-polls - /// up to `wait_seconds` (capped at 60s server-side, default 30s - /// when None) before returning `Empty`. - Recv { - #[serde(default)] - wait_seconds: Option, - }, + /// Pop one pending message from this agent's inbox. + Recv, /// Non-mutating: how many pending messages are addressed to me? /// Used by the harness to render a status line after each tool call. Status, @@ -279,12 +274,7 @@ pub enum ManagerRequest { to: String, body: String, }, - /// Same shape as `AgentRequest::Recv` โ€” caller-tunable long-poll - /// duration, capped at 60s server-side, default 30s when None. - Recv { - #[serde(default)] - wait_seconds: Option, - }, + Recv, /// Non-mutating: pending message count, used to render a status line /// after each MCP tool call (mirrors `AgentRequest::Status`). Status,