diff --git a/TODO.md b/TODO.md index a9ee2bee..a39b90d6 100644 --- a/TODO.md +++ b/TODO.md @@ -62,36 +62,6 @@ Pick anything from here when relevant. Cross-cutting design notes live in derived from the same config so the operator stays in control of what's exposed. -## Operational hygiene (post-meta-flake) - -- **Tag retention.** Every approval mints up to 5 tags in - `applied//.git` (`proposal/`, `approved/`, `building/`, - `deployed/`, plus `failed/` or `denied/`). Every successful - deploy adds one commit to `/var/lib/hyperhive/meta/.git`. - Both grow unbounded. A retention policy — keep all - `deployed/*` indefinitely, age-out `failed/` + `denied/` - after N days, drop `proposal/` + `approved/` + `building/` - once a terminal sibling lands — would keep the audit - trails browsable without forever-growth. - -- **Inert `nix flake lock` no-args call in `meta::sync_agents`.** - Still valid in current nix (resolves missing inputs without - bumping existing ones) but parallel to the deprecated - `--update-input` we just had to migrate. Worth keeping an - eye on; if it gets renamed too, sync_agents stops being - able to seed a fresh meta repo. - -- **Stream `nixos-container update` stdout into tracing.** - Today `lifecycle::run` calls `.output()` which buffers - until exit, so a 5-minute rebuild looks identical to a hang - in the journal — only the success/failure summary lands. Pipe - stdout + stderr line-by-line into tracing (or into the - per-agent live event bus when we know which agent) so the - operator sees `building '/nix/store/…drv'` / `copying path …` - in real time and can tell "slow" from "stuck." Same applies to - every other `nixos-container` shellout that takes - meaningful time (`create`, `update`). - ## Bugs - **Pending question doesn't always appear on the dashboard.** @@ -114,14 +84,12 @@ Pick anything from here when relevant. Cross-cutting design notes live in ## UI / UX -- **Web UI for config repos + meta deploy log.** Browse - per-agent proposed / applied tags - (`proposal/* / approved/* / building/* / deployed/* / - failed/* / denied/*`) plus the swarm-wide meta repo's git - log on the dashboard. Read-only log + diff + raw-file view - is enough — something lighter than a full forge. The meta - log already answers "what's deployed where + when"; this - surfaces it without an ssh-to-host detour. +- **Web UI for config repos.** Browse history, diffs, tags + (proposed + approval/* + applied/*) per agent, all from the + dashboard. Something lighter than a full forge — read-only + log + diff + raw-file view is enough. Pairs naturally with + the upcoming config-repo overhaul (tags become the audit + trail; UI surfaces them). - **xterm.js terminal** embedded per-agent, attached to a PTY exposed by the harness. Pairs well with the unprivileged-container work — would let diff --git a/hive-ag3nt/prompts/agent.md b/hive-ag3nt/prompts/agent.md index 66b02432..724d6739 100644 --- a/hive-ag3nt/prompts/agent.md +++ b/hive-ag3nt/prompts/agent.md @@ -4,10 +4,11 @@ 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__send(to, body)` — message a peer (by their name) or the operator (recipient `operator`, surfaces in the dashboard). -- `mcp__hyperhive__ask_operator(question, options?, multi?, ttl_seconds?)` — surface a question to the human operator on the dashboard. Returns immediately with a question id — do NOT wait inline. When the operator answers, a system message with event `operator_answered { id, question, answer }` lands in your inbox; handle it on a future turn. Use this for clarifications, permission for risky actions, or choice between options. `options` is advisory: a short fixed-choice list when applicable, otherwise leave empty for free text. `multi: true` lets the operator pick multiple (checkboxes), answer comes back comma-joined. `ttl_seconds` auto-cancels with answer `[expired]` when the decision becomes moot. 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. +Need to ask the human operator a question (clarification, permission, choice)? You don't have direct operator access — ask the manager to surface the question on your behalf ("please ask the operator: …"). The manager has a channel for this. + Durable knowledge: write to `/state/notes.md` (free-form) or any other path under `/state/`. That directory is bind-mounted from the host and persists across container destroy/recreate — claude's `--continue` session only carries short-term context, but `/state/` is forever. Read it back at the start of relevant turns to remember things across resets. Keep messages short — a few sentences each. For anything big (file listings, long diffs, transcripts, analysis): write the payload to `/state/` and `send` a short pointer ("dropped the cluster audit in /state/cluster-audit-2026-05.md, headline: 3 nodes over 80% mem"). The manager + operator can read your `/state/` from the host as `/agents/{label}/state/`. Sub-agent peers can't read each other's `/state/` directly — go through the manager if a payload needs to reach another sub-agent. diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index 3a1b2011..ee77d296 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -125,10 +125,7 @@ async fn serve( bus.set_state(TurnState::Idle); } Ok(AgentResponse::Empty) => {} - Ok(AgentResponse::Ok - | AgentResponse::Status { .. } - | AgentResponse::Recent { .. } - | AgentResponse::QuestionQueued { .. }) => { + Ok(AgentResponse::Ok | AgentResponse::Status { .. } | AgentResponse::Recent { .. }) => { tracing::warn!("recv produced unexpected response kind"); } Ok(AgentResponse::Err { message }) => { diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index ab0e749d..dcfe56e9 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -50,7 +50,6 @@ impl From for SocketReply { hive_sh4re::AgentResponse::Empty => Self::Empty, hive_sh4re::AgentResponse::Status { unread } => Self::Status(unread), hive_sh4re::AgentResponse::Recent { rows } => Self::Recent(rows), - hive_sh4re::AgentResponse::QuestionQueued { id } => Self::QuestionQueued(id), } } } @@ -164,45 +163,6 @@ impl AgentServer { .await } - #[tool( - description = "Surface a question to the operator on the dashboard. Returns immediately \ - with a question id — do NOT wait inline. When the operator answers, a system message \ - with event `operator_answered { id, question, answer }` lands in your inbox; handle it \ - on a future turn. Use this when a decision needs human signal (ambiguous scope, \ - permission to do something risky, choosing between options). `options` is advisory: \ - pass a short fixed-choice list when applicable, otherwise leave empty for free text. \ - Set `multi: true` to let the operator pick multiple options (checkboxes); the answer \ - comes back as a comma-separated string. Set `ttl_seconds` to auto-cancel a \ - no-longer-relevant question — on expiry the answer is `[expired]` and the same \ - `operator_answered` event fires." - )] - async fn ask_operator(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("ask_operator", log, async move { - let resp = client::request::<_, hive_sh4re::AgentResponse>( - &self.socket, - &hive_sh4re::AgentRequest::AskOperator { - question: args.question, - options: args.options, - multi: args.multi, - ttl_seconds: args.ttl_seconds, - }, - ) - .await - .map(SocketReply::from); - match resp { - Ok(SocketReply::QuestionQueued(id)) => format!( - "question queued (id={id}); operator's answer will arrive as a system \ - `operator_answered` event in your inbox" - ), - Ok(SocketReply::Err(m)) => format!("ask_operator failed: {m}"), - Ok(other) => format!("ask_operator unexpected response: {other:?}"), - Err(e) => format!("ask_operator transport error: {e:#}"), - } - }) - .await - } - #[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 \ @@ -567,7 +527,7 @@ pub enum Flavor { #[must_use] pub fn allowed_mcp_tools(flavor: Flavor) -> Vec { let names: &[&str] = match flavor { - Flavor::Agent => &["send", "recv", "ask_operator"], + Flavor::Agent => &["send", "recv"], Flavor::Manager => &[ "send", "recv", diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index ed5b7fb5..c47e5b45 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -11,18 +11,14 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::task::JoinHandle; -use crate::coordinator::Coordinator; +use crate::broker::Broker; pub struct AgentSocket { pub path: PathBuf, pub handle: JoinHandle<()>, } -pub fn start( - agent: &str, - socket_path: &Path, - coord: Arc, -) -> Result { +pub fn start(agent: &str, socket_path: &Path, broker: Arc) -> Result { let agent = agent.to_owned(); if let Some(parent) = socket_path.parent() { std::fs::create_dir_all(parent) @@ -41,9 +37,9 @@ pub fn start( match listener.accept().await { Ok((stream, _)) => { let agent = agent.clone(); - let coord = coord.clone(); + let broker = broker.clone(); tokio::spawn(async move { - if let Err(e) = serve(stream, agent, coord).await { + if let Err(e) = serve(stream, agent, broker).await { tracing::warn!(error = ?e, "agent connection failed"); } }); @@ -58,7 +54,7 @@ pub fn start( Ok(AgentSocket { path, handle }) } -async fn serve(stream: UnixStream, agent: String, coord: Arc) -> Result<()> { +async fn serve(stream: UnixStream, agent: String, broker: Arc) -> Result<()> { let (read, mut write) = stream.into_split(); let mut reader = BufReader::new(read); let mut line = String::new(); @@ -69,7 +65,7 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc) -> Re return Ok(()); } let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(&req, &agent, &coord).await, + Ok(req) => dispatch(&req, &agent, &broker).await, Err(e) => AgentResponse::Err { message: format!("parse error: {e}"), }, @@ -97,8 +93,7 @@ fn recv_timeout(wait_seconds: Option) -> std::time::Duration { } } -async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { - let broker = &coord.broker; +async fn dispatch(req: &AgentRequest, agent: &str, broker: &Broker) -> AgentResponse { match req { AgentRequest::Send { to, body } => { match broker.send(&Message { @@ -147,35 +142,5 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> message: format!("{e:#}"), }, }, - AgentRequest::AskOperator { - question, - options, - multi, - ttl_seconds, - } => { - let deadline_at = ttl_seconds.and_then(|s| { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .ok() - .and_then(|d| i64::try_from(d.as_secs()).ok()) - .unwrap_or(0); - i64::try_from(s).ok().map(|s| now + s) - }); - match coord - .questions - .submit(agent, question, options, *multi, deadline_at) - { - Ok(id) => { - tracing::info!(%id, %agent, ?deadline_at, "agent question queued"); - if let Some(ttl) = *ttl_seconds { - crate::manager_server::spawn_question_watchdog(coord, id, ttl); - } - AgentResponse::QuestionQueued { id } - } - Err(e) => AgentResponse::Err { - message: format!("{e:#}"), - }, - } - } } } diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 96b2d75c..93ef7f62 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -82,7 +82,7 @@ impl Coordinator { }) } - pub fn register_agent(self: &Arc, name: &str) -> Result { + pub fn register_agent(&self, name: &str) -> Result { // Idempotent: drop any existing listener so re-registration (e.g. on rebuild, // or after a hive-c0re restart cleared /run/hyperhive) gets a fresh socket. self.unregister_agent(name); @@ -90,10 +90,7 @@ impl Coordinator { std::fs::create_dir_all(&agent_dir) .with_context(|| format!("create agent dir {}", agent_dir.display()))?; let socket_path = Self::socket_path(name); - // Hand the full Coordinator to the per-agent socket — it - // needs broker + operator_questions to handle the agent-side - // `ask_operator` tool, not just the broker. - let socket = agent_server::start(name, &socket_path, self.clone())?; + let socket = agent_server::start(name, &socket_path, self.broker.clone())?; self.agents.lock().unwrap().insert(name.to_owned(), socket); Ok(agent_dir) } @@ -151,15 +148,6 @@ impl Coordinator { /// recognises the sender and parses the body. Best-effort: a serde or /// broker error is logged but does not propagate. pub fn notify_manager(&self, event: &hive_sh4re::HelperEvent) { - self.notify_agent(hive_sh4re::MANAGER_AGENT, event); - } - - /// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded - /// the same way as `notify_manager` (sender = `SYSTEM_SENDER`, - /// body = JSON-encoded event). Used to route `OperatorAnswered` - /// events back to the agent that called `ask_operator`, not just - /// the manager. - pub fn notify_agent(&self, agent: &str, event: &hive_sh4re::HelperEvent) { let body = match serde_json::to_string(event) { Ok(s) => s, Err(e) => { @@ -169,10 +157,10 @@ impl Coordinator { }; if let Err(e) = self.broker.send(&hive_sh4re::Message { from: hive_sh4re::SYSTEM_SENDER.to_owned(), - to: agent.to_owned(), + to: hive_sh4re::MANAGER_AGENT.to_owned(), body, }) { - tracing::warn!(error = ?e, target = %agent, "failed to push helper event"); + tracing::warn!(error = ?e, "failed to push helper event to manager"); } } @@ -197,7 +185,7 @@ impl Coordinator { /// the dir. For sub-agents this is `register_agent` (creates a fresh /// listener bound to `socket_path(name)`). Source directory of the /// `/run/hive/mcp.sock` bind that ends up in `set_nspawn_flags`. - pub fn ensure_runtime(self: &Arc, name: &str) -> Result { + pub fn ensure_runtime(&self, name: &str) -> Result { if name == crate::lifecycle::MANAGER_NAME { let dir = Self::manager_dir(); std::fs::create_dir_all(&dir) diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 0027465d..210b9766 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -528,16 +528,15 @@ async fn post_answer_question( return error_response("answer: required"); } match state.coord.questions.answer(id, answer) { - Ok((question, asker)) => { - tracing::info!(%id, %asker, "operator answered question"); - state.coord.notify_agent( - &asker, - &hive_sh4re::HelperEvent::OperatorAnswered { + Ok(question) => { + tracing::info!(%id, "operator answered question"); + state + .coord + .notify_manager(&hive_sh4re::HelperEvent::OperatorAnswered { id, question, answer: answer.to_owned(), - }, - ); + }); Redirect::to("/").into_response() } Err(e) => error_response(&format!("answer {id} failed: {e:#}")), @@ -556,16 +555,15 @@ async fn post_cancel_question( ) -> Response { const SENTINEL: &str = "[cancelled]"; match state.coord.questions.answer(id, SENTINEL) { - Ok((question, asker)) => { - tracing::info!(%id, %asker, "operator cancelled question"); - state.coord.notify_agent( - &asker, - &hive_sh4re::HelperEvent::OperatorAnswered { + Ok(question) => { + tracing::info!(%id, "operator cancelled question"); + state + .coord + .notify_manager(&hive_sh4re::HelperEvent::OperatorAnswered { id, question, answer: SENTINEL.to_owned(), - }, - ); + }); Redirect::to("/").into_response() } Err(e) => error_response(&format!("cancel-question {id} failed: {e:#}")), diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 6913d090..a15b338d 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -324,26 +324,6 @@ async fn submit_apply_commit( return Err(anyhow::anyhow!("git_fetch_to_tag: {e:#}")); } }; - // Reject proposals that touch flake.nix — that file is fixed - // boilerplate the meta flake depends on (it exports - // `nixosModules.default = import ./agent.nix`); a manager edit - // there silently breaks the nixosConfiguration. Prompt tells the - // manager not to; this is the host-side belt. - let proposal_ref = format!("refs/tags/{tag}"); - match proposal_modifies(&applied_dir, &proposal_ref, "flake.nix").await { - Ok(true) => { - let note = "proposal modifies flake.nix — that file is hive-c0re-owned \ - boilerplate; edit only agent.nix (and sibling modules)"; - let _ = coord.approvals.mark_failed(id, note); - return Err(anyhow::anyhow!(note)); - } - Ok(false) => {} - Err(e) => { - // Diff itself failed — log + continue. The build will - // surface concrete errors if flake.nix is actually borked. - tracing::warn!(error = ?e, %tag, "flake.nix tamper-check failed; allowing"); - } - } coord .approvals .set_fetched_sha(id, &sha) @@ -351,36 +331,6 @@ async fn submit_apply_commit( Ok((id, sha)) } -/// True iff `proposal_ref`'s tree differs from `refs/heads/main` at -/// `path`. Used to enforce that proposals don't touch hive-c0re-owned -/// files in the applied repo. -async fn proposal_modifies( - applied_dir: &std::path::Path, - proposal_ref: &str, - path: &str, -) -> anyhow::Result { - let out = crate::lifecycle::git_command() - .current_dir(applied_dir) - .args([ - "diff", - "--name-only", - &format!("refs/heads/main..{proposal_ref}"), - "--", - path, - ]) - .output() - .await - .map_err(|e| anyhow::anyhow!("spawn git diff: {e}"))?; - if !out.status.success() { - anyhow::bail!( - "git diff exited {}: {}", - out.status, - String::from_utf8_lossy(&out.stderr).trim() - ); - } - Ok(!out.stdout.iter().all(u8::is_ascii_whitespace)) -} - /// On `AskOperator { ttl_seconds: Some(n) }`, sleep n seconds and then /// try to resolve the question with `[expired]`. If the operator (or /// any other path) already answered it, `answer()` returns Err and @@ -388,23 +338,20 @@ async fn proposal_modifies( /// helper event so the manager sees a terminal state. const TTL_SENTINEL: &str = "[expired]"; -pub fn spawn_question_watchdog(coord: &Arc, id: i64, ttl_secs: u64) { +fn spawn_question_watchdog(coord: &Arc, id: i64, ttl_secs: u64) { let coord = coord.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(ttl_secs)).await; // `answer` returns Err if already resolved — that's the // normal path when the operator responded before the ttl // fired, so no-op silently. - if let Ok((question, asker)) = coord.questions.answer(id, TTL_SENTINEL) { - tracing::info!(%id, %asker, "operator question expired (ttl)"); - coord.notify_agent( - &asker, - &hive_sh4re::HelperEvent::OperatorAnswered { - id, - question, - answer: TTL_SENTINEL.to_owned(), - }, - ); + if let Ok(question) = coord.questions.answer(id, TTL_SENTINEL) { + tracing::info!(%id, "operator question expired (ttl)"); + coord.notify_manager(&hive_sh4re::HelperEvent::OperatorAnswered { + id, + question, + answer: TTL_SENTINEL.to_owned(), + }); } }); } diff --git a/hive-c0re/src/operator_questions.rs b/hive-c0re/src/operator_questions.rs index c04c3bd3..4ca652ed 100644 --- a/hive-c0re/src/operator_questions.rs +++ b/hive-c0re/src/operator_questions.rs @@ -117,20 +117,17 @@ impl OperatorQuestions { } /// Mark the question answered. Returns the original question text so the - /// Mark a pending question answered. Returns `(question, asker)` - /// so the caller can both echo the question back in a helper - /// event AND route that event to whichever agent originally - /// asked it. - pub fn answer(&self, id: i64, answer: &str) -> Result<(String, String)> { + /// caller can include it in any helper event it fires off. + pub fn answer(&self, id: i64, answer: &str) -> Result { let conn = self.conn.lock().unwrap(); - let row: Option<(String, String, Option)> = conn + let question: Option<(String, Option)> = conn .query_row( - "SELECT question, asker, answered_at FROM operator_questions WHERE id = ?1", + "SELECT question, answered_at FROM operator_questions WHERE id = ?1", params![id], - |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + |row| Ok((row.get(0)?, row.get(1)?)), ) .optional()?; - let Some((question, asker, answered_at)) = row else { + let Some((question, answered_at)) = question else { bail!("question {id} not found"); }; if answered_at.is_some() { @@ -140,7 +137,7 @@ impl OperatorQuestions { "UPDATE operator_questions SET answer = ?1, answered_at = ?2 WHERE id = ?3", params![answer, now_unix(), id], )?; - Ok((question, asker)) + Ok(question) } #[allow(dead_code)] diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 298d5add..5d4a4fba 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -195,19 +195,6 @@ pub enum AgentRequest { /// Non-mutating — pulls from the broker without delivering. The /// per-agent web UI uses this to render its own inbox section. Recent { limit: u64 }, - /// Surface a question to the operator on the dashboard. Same - /// shape as `ManagerRequest::AskOperator` — any agent can ask; - /// the answer routes back to the asker's inbox as a - /// `HelperEvent::OperatorAnswered`. - AskOperator { - question: String, - #[serde(default)] - options: Vec, - #[serde(default)] - multi: bool, - #[serde(default)] - ttl_seconds: Option, - }, } /// Responses on a per-agent socket. @@ -226,9 +213,6 @@ pub enum AgentResponse { Status { unread: u64 }, /// `Recent` result: newest-first inbox rows. Recent { rows: Vec }, - /// `AskOperator` result: the queued question id. The answer lands - /// later as `HelperEvent::OperatorAnswered` in this agent's inbox. - QuestionQueued { id: i64 }, } // -----------------------------------------------------------------------------