manager: same agent loop, ManagerServer MCP surface
This commit is contained in:
müde
parent
accb1445e3
commit
09787659ab
6 changed files with 422 additions and 142 deletions
|
|
@ -1,5 +1,4 @@
|
|||
use std::path::{Path, PathBuf};
|
||||
use std::process::Stdio;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
|
|
@ -7,10 +6,8 @@ use anyhow::{Result, bail};
|
|||
use clap::{Parser, Subcommand};
|
||||
use hive_ag3nt::events::{Bus, LiveEvent};
|
||||
use hive_ag3nt::login::{self, LoginState};
|
||||
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, web_ui};
|
||||
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
||||
use hive_sh4re::{AgentRequest, AgentResponse};
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
use tokio::process::Command;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(name = "hive-ag3nt", about = "hyperhive sub-agent harness")]
|
||||
|
|
@ -97,7 +94,7 @@ async fn main() -> Result<()> {
|
|||
render(&resp)?;
|
||||
check(&resp)
|
||||
}
|
||||
Cmd::Mcp => mcp::serve_stdio(cli.socket).await,
|
||||
Cmd::Mcp => mcp::serve_agent_stdio(cli.socket).await,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -133,7 +130,7 @@ async fn serve(
|
|||
) -> Result<()> {
|
||||
tracing::info!(socket = %socket.display(), "hive-ag3nt serve");
|
||||
let _ = state; // reserved for future state transitions (turn-loop -> needs-login)
|
||||
let mcp_config = write_mcp_config(socket).await?;
|
||||
let mcp_config = turn::write_mcp_config(socket).await?;
|
||||
let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hive-ag3nt".into());
|
||||
loop {
|
||||
let recv: Result<AgentResponse> = client::request(socket, &AgentRequest::Recv).await;
|
||||
|
|
@ -145,7 +142,8 @@ async fn serve(
|
|||
body: body.clone(),
|
||||
});
|
||||
let prompt = format_wake_prompt(&label, &from, &body);
|
||||
let outcome = run_turn(&prompt, &mcp_config, &bus).await;
|
||||
let outcome =
|
||||
turn::run_turn(&prompt, &mcp_config, &bus, mcp::Flavor::Agent).await;
|
||||
match outcome {
|
||||
Ok(()) => {
|
||||
bus.emit(LiveEvent::TurnEnd {
|
||||
|
|
@ -202,93 +200,6 @@ fn format_wake_prompt(label: &str, from: &str, body: &str) -> String {
|
|||
)
|
||||
}
|
||||
|
||||
/// Spawn `claude` for one turn and stream its `stream-json` stdout into
|
||||
/// the live event bus. `--verbose` is required by claude-code when pairing
|
||||
/// `--print` with `--output-format stream-json`. Each stdout line is one
|
||||
/// JSON event; we broadcast the parsed value (or a `Note` fallback on
|
||||
/// parse error so the UI doesn't silently lose information). The tool
|
||||
/// whitelist is the same as before: omit WebFetch/WebSearch/Task; allow
|
||||
/// the hyperhive MCP surface auto-approved. Bash pattern allow-list is on
|
||||
/// the backlog (CLAUDE.md).
|
||||
async fn run_turn(prompt: &str, mcp_config: &Path, bus: &Bus) -> Result<()> {
|
||||
// Don't pass the prompt as a positional arg: `--allowedTools <tools...>`
|
||||
// and `--tools <tools...>` are variadic in claude-code, and the
|
||||
// trailing positional gets swallowed into one of them — claude then
|
||||
// errors with "Input must be provided either through stdin or as a
|
||||
// prompt argument when using --print". Pipe via stdin instead.
|
||||
let mut child = Command::new("claude")
|
||||
.arg("--print")
|
||||
.arg("--verbose")
|
||||
.arg("--output-format")
|
||||
.arg("stream-json")
|
||||
.arg("--model")
|
||||
.arg("haiku")
|
||||
.arg("--mcp-config")
|
||||
.arg(mcp_config)
|
||||
.arg("--tools")
|
||||
.arg(mcp::builtin_tools_arg())
|
||||
.arg("--allowedTools")
|
||||
.arg(mcp::allowed_tools_arg())
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()?;
|
||||
|
||||
if let Some(mut stdin) = child.stdin.take() {
|
||||
stdin.write_all(prompt.as_bytes()).await?;
|
||||
stdin.shutdown().await.ok();
|
||||
drop(stdin); // signal EOF to claude
|
||||
}
|
||||
let stdout = child.stdout.take().expect("piped stdout");
|
||||
let stderr = child.stderr.take().expect("piped stderr");
|
||||
|
||||
let bus_out = bus.clone();
|
||||
let bus_err = bus.clone();
|
||||
let pump_stdout = tokio::spawn(async move {
|
||||
let mut reader = BufReader::new(stdout).lines();
|
||||
while let Ok(Some(line)) = reader.next_line().await {
|
||||
match serde_json::from_str::<serde_json::Value>(&line) {
|
||||
Ok(v) => bus_out.emit(LiveEvent::Stream(v)),
|
||||
Err(_) => bus_out.emit(LiveEvent::Note(format!("(non-json) {line}"))),
|
||||
}
|
||||
}
|
||||
});
|
||||
let pump_stderr = tokio::spawn(async move {
|
||||
let mut reader = BufReader::new(stderr).lines();
|
||||
while let Ok(Some(line)) = reader.next_line().await {
|
||||
bus_err.emit(LiveEvent::Note(format!("stderr: {line}")));
|
||||
}
|
||||
});
|
||||
|
||||
let status = child.wait().await?;
|
||||
let _ = pump_stdout.await;
|
||||
let _ = pump_stderr.await;
|
||||
if !status.success() {
|
||||
bail!("claude exited {status}");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Drop the per-agent MCP config on disk so the turn loop can hand its path
|
||||
/// to `claude --mcp-config`. Lives under `/run/hive/` (the bind-mounted
|
||||
/// per-agent runtime dir) so it's ephemeral and isolated per container.
|
||||
/// Returns the config path.
|
||||
async fn write_mcp_config(socket: &Path) -> Result<PathBuf> {
|
||||
let parent = socket.parent().unwrap_or_else(|| Path::new("/run/hive"));
|
||||
tokio::fs::create_dir_all(parent).await.ok();
|
||||
let path = parent.join("claude-mcp-config.json");
|
||||
// `/proc/self/exe` resolves to the running hive-ag3nt binary's nix store
|
||||
// path, which the spawned child can re-invoke as the MCP server. Avoids
|
||||
// needing claude-code's $PATH to contain hive-ag3nt.
|
||||
let exe = std::env::current_exe()
|
||||
.ok()
|
||||
.map_or_else(|| "hive-ag3nt".into(), |p| p.display().to_string());
|
||||
let body = mcp::render_claude_config(&exe, socket);
|
||||
tokio::fs::write(&path, body).await?;
|
||||
tracing::info!(path = %path.display(), "wrote claude MCP config");
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
fn render(resp: &AgentResponse) -> Result<()> {
|
||||
println!("{}", serde_json::to_string_pretty(resp)?);
|
||||
Ok(())
|
||||
|
|
|
|||
|
|
@ -9,9 +9,9 @@ use std::time::Duration;
|
|||
|
||||
use anyhow::{Result, bail};
|
||||
use clap::{Parser, Subcommand};
|
||||
use hive_ag3nt::events::Bus;
|
||||
use hive_ag3nt::events::{Bus, LiveEvent};
|
||||
use hive_ag3nt::login::{self, LoginState};
|
||||
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, web_ui};
|
||||
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, turn, web_ui};
|
||||
use hive_sh4re::{HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER};
|
||||
|
||||
#[derive(Parser)]
|
||||
|
|
@ -43,6 +43,11 @@ enum Cmd {
|
|||
Kill { name: String },
|
||||
/// Submit a config commit on the agent's config repo for user approval.
|
||||
RequestApplyCommit { agent: String, commit_ref: String },
|
||||
/// Run the manager MCP server on stdio. Spawned by claude via
|
||||
/// `--mcp-config`; same shape as `hive-ag3nt mcp` but with the
|
||||
/// manager tool surface (`request_spawn`, `kill`,
|
||||
/// `request_apply_commit`).
|
||||
Mcp,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
|
|
@ -74,15 +79,16 @@ async fn main() -> Result<()> {
|
|||
tracing::error!(error = ?e, "web ui failed");
|
||||
}
|
||||
});
|
||||
let _ = bus; // manager turn loop not wired to events yet
|
||||
match initial {
|
||||
LoginState::Online => serve(&cli.socket, Duration::from_millis(poll_ms)).await,
|
||||
LoginState::Online => {
|
||||
serve(&cli.socket, Duration::from_millis(poll_ms), bus).await
|
||||
}
|
||||
LoginState::NeedsLogin => {
|
||||
tracing::warn!(
|
||||
claude_dir = %claude_dir.display(),
|
||||
"manager has no claude session — staying in partial-run mode"
|
||||
);
|
||||
needs_login_loop(&cli.socket, &claude_dir, login_state, poll_ms).await
|
||||
needs_login_loop(&cli.socket, &claude_dir, login_state, poll_ms, bus).await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -99,6 +105,7 @@ async fn main() -> Result<()> {
|
|||
)
|
||||
.await
|
||||
}
|
||||
Cmd::Mcp => mcp::serve_manager_stdio(cli.socket).await,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -118,6 +125,7 @@ async fn needs_login_loop(
|
|||
claude_dir: &Path,
|
||||
state: Arc<Mutex<LoginState>>,
|
||||
poll_ms: u64,
|
||||
bus: Bus,
|
||||
) -> Result<()> {
|
||||
let probe = Duration::from_millis(poll_ms.max(2000));
|
||||
loop {
|
||||
|
|
@ -125,25 +133,55 @@ async fn needs_login_loop(
|
|||
if login::has_session(claude_dir) {
|
||||
tracing::info!("manager claude session detected — entering inbox loop");
|
||||
*state.lock().unwrap() = LoginState::Online;
|
||||
return serve(socket, Duration::from_millis(poll_ms)).await;
|
||||
return serve(socket, Duration::from_millis(poll_ms), bus).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn serve(socket: &Path, interval: Duration) -> Result<()> {
|
||||
async fn serve(socket: &Path, interval: Duration, bus: Bus) -> Result<()> {
|
||||
tracing::info!(socket = %socket.display(), "hive-m1nd serve");
|
||||
let mcp_config = turn::write_mcp_config(socket).await?;
|
||||
let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hm1nd".into());
|
||||
loop {
|
||||
let recv: Result<ManagerResponse> = client::request(socket, &ManagerRequest::Recv).await;
|
||||
match recv {
|
||||
Ok(ManagerResponse::Message { from, body }) => {
|
||||
if from == SYSTEM_SENDER {
|
||||
if let Ok(event) = serde_json::from_str::<HelperEvent>(&body) {
|
||||
// Helper events (ApprovalResolved, etc.) — log + surface
|
||||
// in live view but don't burn a claude turn on them.
|
||||
let parsed = serde_json::from_str::<HelperEvent>(&body).ok();
|
||||
if let Some(event) = parsed {
|
||||
tracing::info!(?event, "helper event");
|
||||
} else {
|
||||
tracing::info!(%from, %body, "system message");
|
||||
}
|
||||
} else {
|
||||
tracing::info!(%from, %body, "manager inbox");
|
||||
bus.emit(LiveEvent::Note(format!("[system] {body}")));
|
||||
continue;
|
||||
}
|
||||
tracing::info!(%from, %body, "manager inbox");
|
||||
bus.emit(LiveEvent::TurnStart {
|
||||
from: from.clone(),
|
||||
body: body.clone(),
|
||||
});
|
||||
let prompt = format_wake_prompt(&label, &from, &body);
|
||||
let outcome =
|
||||
turn::run_turn(&prompt, &mcp_config, &bus, mcp::Flavor::Manager).await;
|
||||
match outcome {
|
||||
Ok(()) => {
|
||||
bus.emit(LiveEvent::TurnEnd {
|
||||
ok: true,
|
||||
note: None,
|
||||
});
|
||||
tracing::info!("manager turn finished");
|
||||
}
|
||||
Err(e) => {
|
||||
let note = format!("{e:#}");
|
||||
bus.emit(LiveEvent::TurnEnd {
|
||||
ok: false,
|
||||
note: Some(note.clone()),
|
||||
});
|
||||
tracing::warn!(error = %note, "manager turn failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(ManagerResponse::Empty) => {}
|
||||
|
|
@ -160,3 +198,35 @@ async fn serve(socket: &Path, interval: Duration) -> Result<()> {
|
|||
tokio::time::sleep(interval).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Manager-flavored wake prompt. Mentions the privileged tools the sub-agent
|
||||
/// prompt doesn't have access to, and points the manager at its own
|
||||
/// editable config repo for self-modification.
|
||||
fn format_wake_prompt(label: &str, from: &str, body: &str) -> String {
|
||||
format!(
|
||||
"You are the hyperhive manager `{label}` in a multi-agent system. You \
|
||||
coordinate sub-agents and relay between them and the operator.\n\
|
||||
\n\
|
||||
Incoming message from `{from}`:\n\
|
||||
---\n\
|
||||
{body}\n\
|
||||
---\n\
|
||||
\n\
|
||||
Tools (hyperhive surface):\n\
|
||||
- `mcp__hyperhive__recv()` — drain one more message from your inbox.\n\
|
||||
- `mcp__hyperhive__send(to, body)` — message an agent (by name), \
|
||||
another peer, or the operator (`operator` surfaces in the dashboard).\n\
|
||||
- `mcp__hyperhive__request_spawn(name)` — queue a brand-new sub-agent \
|
||||
for operator approval (≤9 char name).\n\
|
||||
- `mcp__hyperhive__kill(name)` — graceful stop on a sub-agent.\n\
|
||||
- `mcp__hyperhive__request_apply_commit(agent, commit_ref)` — submit \
|
||||
a config change for any agent (`hm1nd` for self) for operator \
|
||||
approval.\n\
|
||||
\n\
|
||||
Your own editable config lives at `/agents/hm1nd/config/agent.nix`; \
|
||||
every sub-agent's lives at `/agents/<name>/config/agent.nix`. Use \
|
||||
file/git tools to edit + commit, then `request_apply_commit`.\n\
|
||||
\n\
|
||||
Handle the inbox, then stop. Don't narrate intent — act."
|
||||
)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue