Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fd2eac99ce | ||
|
|
53b4e752ef | ||
|
|
674505fbe7 | ||
|
|
f5f06a5f14 | ||
|
|
a053d33184 |
15 changed files with 2229 additions and 2343 deletions
File diff suppressed because it is too large
Load diff
|
|
@ -15,7 +15,7 @@
|
|||
//!
|
||||
//! Same process-singleton handle pattern as `build_logs`: installed once
|
||||
//! at `Coordinator::open`, fetched via [`global`] so the recording sites
|
||||
//! (e.g. `agent_server::handle_restart_infra`) don't have to thread an
|
||||
//! (e.g. `socket_server::handle_restart_infra`) don't have to thread an
|
||||
//! `Arc<AuditLog>` through every call path. Recording is best-effort: a
|
||||
//! sqlite blip must never fail the underlying privileged action.
|
||||
|
||||
|
|
|
|||
|
|
@ -137,7 +137,7 @@ fn auth_failed_sentinel(name: &str) -> bool {
|
|||
|
||||
/// Read the agent's free-text status and the Unix timestamp when it was last set
|
||||
/// (derived from the file's mtime). Returns `(None, None)` when the file is absent
|
||||
/// or empty. `pub` so `agent_server` and `manager_server` can populate `AgentMeta`.
|
||||
/// or empty. `pub` so `socket_server` and `socket_server` can populate `AgentMeta`.
|
||||
///
|
||||
/// NB: callers building `AgentMeta` for a *stopped* container should
|
||||
/// clear the result — the on-disk status is a stale snapshot from
|
||||
|
|
|
|||
|
|
@ -10,12 +10,12 @@ use std::sync::{Arc, Mutex};
|
|||
use anyhow::{Context, Result};
|
||||
use tokio::sync::{broadcast, watch};
|
||||
|
||||
use crate::agent_server::{self, AgentSocket};
|
||||
use crate::approvals::Approvals;
|
||||
use crate::broker::Broker;
|
||||
use crate::container_view::{self, ContainerView};
|
||||
use crate::dashboard_events::DashboardEvent;
|
||||
use crate::operator_questions::OperatorQuestions;
|
||||
use crate::socket_server::{self, AgentSocket};
|
||||
|
||||
/// Capacity of the dashboard event channel. Slow browser subscribers
|
||||
/// (idle tab, throttled connection) drop frames past this — that's
|
||||
|
|
@ -126,7 +126,7 @@ pub struct Coordinator {
|
|||
/// `recent_crash_counts`, which prunes entries older than its window.
|
||||
recent_crashes: Mutex<HashMap<String, Vec<std::time::Instant>>>,
|
||||
/// Agents with a graceful stop in progress. Set by the `GracefulStop`
|
||||
/// orchestration; read by `agent_server::handle_recv`, which returns
|
||||
/// orchestration; read by `socket_server::handle_recv`, which returns
|
||||
/// `Response::GracefulStop` (instead of polling the broker) while an
|
||||
/// agent is in this set — the inbound fence. Cleared when the agent
|
||||
/// reports `GracefulStopComplete` or the container is stopped.
|
||||
|
|
@ -429,7 +429,7 @@ impl Coordinator {
|
|||
crate::build_logs::install(build_logs.clone());
|
||||
// Audit log shares the same db dir; install its process-wide
|
||||
// handle so privileged-action recording sites (e.g.
|
||||
// `agent_server::handle_restart_infra`) write without threading an
|
||||
// `socket_server::handle_restart_infra`) write without threading an
|
||||
// `Arc<AuditLog>` through the agent-request surface.
|
||||
let audit_log =
|
||||
Arc::new(crate::audit_log::AuditLog::open(build_logs_dir).context("open audit_log")?);
|
||||
|
|
@ -522,7 +522,7 @@ impl Coordinator {
|
|||
let mut schedules: Vec<hive_sh4re::WireSchedule> = match self.scheduled_prompts.list() {
|
||||
Ok(rows) => rows
|
||||
.into_iter()
|
||||
.map(crate::manager_server::schedule_to_wire_public)
|
||||
.map(crate::socket_server::schedule_to_wire_public)
|
||||
.collect(),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed");
|
||||
|
|
@ -534,7 +534,7 @@ impl Coordinator {
|
|||
// momentarily contended we emit unfiltered rather than block this
|
||||
// sync path — the next snapshot / page reload corrects it.
|
||||
if let Some(live) = self.live_container_names_blocking() {
|
||||
crate::manager_server::filter_ghost_schedule_targets(&mut schedules, &live);
|
||||
crate::socket_server::filter_ghost_schedule_targets(&mut schedules, &live);
|
||||
}
|
||||
self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
|
||||
seq: self.next_seq(),
|
||||
|
|
@ -1068,7 +1068,7 @@ impl Coordinator {
|
|||
// Hand the full Coordinator to the per-agent socket — it
|
||||
// needs broker + operator_questions to handle the agent-side
|
||||
// `ask` / `answer` tools, not just the broker.
|
||||
let socket = agent_server::start(name, &socket_path, self.clone())?;
|
||||
let socket = socket_server::start(name, &socket_path, self.clone())?;
|
||||
self.agents.lock().unwrap().insert(name.to_owned(), socket);
|
||||
Ok(agent_dir)
|
||||
}
|
||||
|
|
@ -1136,7 +1136,7 @@ impl Coordinator {
|
|||
}
|
||||
|
||||
/// Mark `name` as having a graceful stop in progress. While set,
|
||||
/// `agent_server::handle_recv` returns `Response::GracefulStop` for
|
||||
/// `socket_server::handle_recv` returns `Response::GracefulStop` for
|
||||
/// this agent instead of polling the broker (the inbound fence).
|
||||
pub fn mark_graceful_stop(&self, name: &str) {
|
||||
self.graceful_stop_pending
|
||||
|
|
@ -1330,7 +1330,7 @@ impl Coordinator {
|
|||
}
|
||||
|
||||
/// Ensure a runtime dir + (for sub-agents) per-agent socket exists. For
|
||||
/// the manager, `manager_server::start` owns the socket — just return
|
||||
/// the manager, `socket_server::start_manager` owns the socket — just return
|
||||
/// 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`.
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ mod topology;
|
|||
mod webhook;
|
||||
|
||||
// Pre-computed at approval-submit time by the manager-socket handler
|
||||
// (`manager_server.rs`) and embedded in the `ApprovalAdded` event, so
|
||||
// (`socket_server.rs`) and embedded in the `ApprovalAdded` event, so
|
||||
// re-exported at the module root to preserve the `crate::dashboard::approval_diff`
|
||||
// path across the submodule split.
|
||||
pub(crate) use approvals::approval_diff;
|
||||
|
|
|
|||
|
|
@ -76,7 +76,7 @@ pub(super) async fn post_kill(
|
|||
// host-side approval queue without the manager up, and
|
||||
// operator-driven meta-input updates work from the dashboard
|
||||
// either way. The MCP-surface self-kill guard in
|
||||
// `manager_server.rs::ManagerRequest::Kill` stays in place: a
|
||||
// `socket_server.rs::ManagerRequest::Kill` stays in place: a
|
||||
// manager calling Kill on its own container is self-suicide
|
||||
// mid-call, not a legitimate operator action.
|
||||
state.coord.rebuild_queue.enqueue(
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
|
|||
Ok(rows) => {
|
||||
let mut wire: Vec<hive_sh4re::WireSchedule> = rows
|
||||
.into_iter()
|
||||
.map(crate::manager_server::schedule_to_wire_public)
|
||||
.map(crate::socket_server::schedule_to_wire_public)
|
||||
.collect();
|
||||
// Drop ghost targets (agents that no longer exist) so the
|
||||
// table never shows dead columns. Uses the reliable async
|
||||
|
|
@ -33,7 +33,7 @@ pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
|
|||
.into_iter()
|
||||
.map(|c| c.name)
|
||||
.collect();
|
||||
crate::manager_server::filter_ghost_schedule_targets(&mut wire, &live);
|
||||
crate::socket_server::filter_ghost_schedule_targets(&mut wire, &live);
|
||||
axum::Json(wire).into_response()
|
||||
}
|
||||
Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
|
||||
|
|
@ -214,7 +214,7 @@ pub(super) async fn patch_schedule(
|
|||
}
|
||||
match state.coord.scheduled_prompts.get(id) {
|
||||
Ok(Some(s)) => {
|
||||
let wire = crate::manager_server::schedule_to_wire_public(s);
|
||||
let wire = crate::socket_server::schedule_to_wire_public(s);
|
||||
state.coord.emit_schedules_snapshot();
|
||||
axum::Json(wire).into_response()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,7 +13,6 @@
|
|||
//! surface beyond "this is where the modules live".
|
||||
|
||||
pub mod actions;
|
||||
pub mod agent_server;
|
||||
pub mod agent_sockets;
|
||||
pub mod approvals;
|
||||
pub mod audit_log;
|
||||
|
|
@ -39,7 +38,6 @@ pub mod knowledge;
|
|||
pub mod lifecycle;
|
||||
pub mod limits;
|
||||
pub mod loose_ends;
|
||||
pub mod manager_server;
|
||||
pub mod matrix;
|
||||
pub mod meta;
|
||||
pub mod migrate;
|
||||
|
|
@ -52,5 +50,6 @@ pub mod reminder_scheduler;
|
|||
pub mod scheduled_prompts;
|
||||
pub mod scheduled_prompts_worker;
|
||||
pub mod server;
|
||||
pub mod socket_server;
|
||||
pub mod tool_groups;
|
||||
pub mod topology;
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
//! to a state file and the path sent as the body.
|
||||
//!
|
||||
//! Reminders get a separate auto-file escape hatch (see
|
||||
//! `agent_server::handle_remind`) so callers don't have to think
|
||||
//! `socket_server::handle_remind`) so callers don't have to think
|
||||
//! about it — oversized reminder bodies get persisted to disk
|
||||
//! transparently and the inbox sees a pointer.
|
||||
|
||||
|
|
|
|||
|
|
@ -13,8 +13,8 @@ use hive_sh4re::{HostRequest, HostResponse};
|
|||
use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig};
|
||||
use hive_c0re::{
|
||||
agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard,
|
||||
dashboard_events, events_vacuum, forge, knowledge, manager_server, matrix, migrate,
|
||||
rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server,
|
||||
dashboard_events, events_vacuum, forge, knowledge, matrix, migrate, rebuild_queue,
|
||||
reminder_scheduler, scheduled_prompts_worker, server, socket_server,
|
||||
};
|
||||
|
||||
#[derive(Parser)]
|
||||
|
|
@ -265,7 +265,7 @@ async fn cmd_serve(
|
|||
// Copy value first for the dashboard + knowledge-webhook tasks.
|
||||
let dashboard_port = env.dashboard_port;
|
||||
let coord = Arc::new(Coordinator::open(&db, env, model_prices)?);
|
||||
manager_server::start(coord.clone())?;
|
||||
socket_server::start_manager(coord.clone())?;
|
||||
// Idempotent pre-flight: rewrite pre-meta-layout applied
|
||||
// repos, ensure proposed repos carry the `applied`
|
||||
// remote, bootstrap the meta repo, repoint containers at
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -21,7 +21,7 @@ use std::sync::Arc;
|
|||
use crate::approvals::kind_to_str;
|
||||
use crate::coordinator::Coordinator;
|
||||
use crate::limits;
|
||||
use crate::manager_server::spawn_question_watchdog;
|
||||
use crate::socket_server::spawn_question_watchdog;
|
||||
|
||||
/// Cap on how long an asker can demand an answer before the watchdog
|
||||
/// auto-resolves with `[expired]`. Six hours mirrors typical agent
|
||||
|
|
@ -141,26 +141,27 @@ pub fn handle_answer(
|
|||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle `CancelLooseEnd` from either surface. Dispatches by kind to
|
||||
/// the per-kind cancel, each of which does its own auth check
|
||||
/// (canceller == owner / asker, or `operator`, or `privileged`).
|
||||
/// `privileged` is `true` when the request arrived on the manager
|
||||
/// socket — privilege derives from the socket (the trust boundary),
|
||||
/// not from matching a hardcoded agent name. On question cancel, fires
|
||||
/// the `QuestionAnswered` event back to the asker so the harness
|
||||
/// loop can react (mirrors the operator-cancel dashboard path).
|
||||
/// Handle `CancelLooseEnd` from a per-agent socket. Dispatches by kind, each
|
||||
/// with its own auth check: question / reminder cancels are ownership-only
|
||||
/// (an agent cancels its own), and approval cancels require the `approvals`
|
||||
/// tool-group (the grantable capability) — no positional / hardcoded
|
||||
/// privilege. (The operator's cancel-anything path is a separate handler.)
|
||||
/// On question cancel, fires the `QuestionAnswered` event back to the asker
|
||||
/// so the harness loop can react (mirrors the operator-cancel dashboard path).
|
||||
pub fn handle_cancel_loose_end(
|
||||
coord: &Arc<Coordinator>,
|
||||
canceller: &str,
|
||||
privileged: bool,
|
||||
kind: hive_sh4re::CancelLooseEndKind,
|
||||
id: i64,
|
||||
) -> Result<(), String> {
|
||||
match kind {
|
||||
hive_sh4re::CancelLooseEndKind::Question => {
|
||||
// Agent-socket path: never privileged — an agent may only cancel
|
||||
// its own question (ownership). The operator's cancel-anything
|
||||
// path goes through a separate handler with `privileged = true`.
|
||||
let (question, asker, target) = coord
|
||||
.questions
|
||||
.cancel(id, canceller, privileged)
|
||||
.cancel(id, canceller, false)
|
||||
.map_err(|e| format!("{e:#}"))?;
|
||||
let sentinel = format!("[cancelled by {canceller}]");
|
||||
tracing::info!(%id, %canceller, %asker, "question cancelled");
|
||||
|
|
@ -184,19 +185,21 @@ pub fn handle_cancel_loose_end(
|
|||
Ok(())
|
||||
}
|
||||
hive_sh4re::CancelLooseEndKind::Reminder => {
|
||||
// Agent-socket path: ownership-only (cancel your own reminder).
|
||||
let owner = coord
|
||||
.broker
|
||||
.cancel_reminder_as(id, canceller, privileged)
|
||||
.cancel_reminder_as(id, canceller, false)
|
||||
.map_err(|e| format!("{e:#}"))?;
|
||||
tracing::info!(%id, %canceller, %owner, "reminder cancelled");
|
||||
coord.emit_reminders_snapshot();
|
||||
Ok(())
|
||||
}
|
||||
hive_sh4re::CancelLooseEndKind::Approval => {
|
||||
// Privileged-only: only a caller on the manager socket (which
|
||||
// is the sole approval submitter) may withdraw approvals.
|
||||
// Sub-agents have no pending approvals of their own anyway.
|
||||
check_can_cancel_approval(privileged)?;
|
||||
// Withdrawing an approval is a hive-wide orchestration action,
|
||||
// gated on the grantable `approvals` tool-group (held by the
|
||||
// orchestrator that submits approvals) — not on a positional /
|
||||
// hardcoded privilege.
|
||||
check_can_cancel_approval(canceller)?;
|
||||
let approval = coord
|
||||
.approvals
|
||||
.mark_cancelled(id, canceller)
|
||||
|
|
@ -220,21 +223,26 @@ pub fn handle_cancel_loose_end(
|
|||
}
|
||||
}
|
||||
|
||||
/// Privileged-only guard on the `Approval` cancel arm. Pulled out so
|
||||
/// the auth check has its own focused unit test — testing the full
|
||||
/// `handle_cancel_loose_end` flow would need a `Coordinator` fixture
|
||||
/// (broker + sqlite + in-memory questions), which we don't have
|
||||
/// today. Privilege is a property of the socket the request arrived
|
||||
/// on (the manager socket), threaded in as `privileged` — not a match
|
||||
/// against a hardcoded agent name.
|
||||
fn check_can_cancel_approval(privileged: bool) -> Result<(), String> {
|
||||
if !privileged {
|
||||
return Err(
|
||||
"cancel_loose_end: only a privileged (manager-socket) caller can cancel approval rows"
|
||||
/// Capability guard on the `Approval` cancel arm: the caller must hold the
|
||||
/// `approvals` tool-group (the grantable capability for approval-submitting
|
||||
/// orchestrators), checked server-side via `tool_groups::groups_for`. Pulled
|
||||
/// out so the auth check has its own focused unit test — exercising the full
|
||||
/// `handle_cancel_loose_end` flow would need a `Coordinator` fixture (broker +
|
||||
/// sqlite + in-memory questions) we don't have. Keys on a grantable capability,
|
||||
/// not a positional / hardcoded privilege.
|
||||
fn check_can_cancel_approval(canceller: &str) -> Result<(), String> {
|
||||
const APPROVALS_GROUP: &str = "approvals";
|
||||
if crate::tool_groups::groups_for(canceller)
|
||||
.iter()
|
||||
.any(|g| g == APPROVALS_GROUP)
|
||||
{
|
||||
Ok(())
|
||||
} else {
|
||||
Err(
|
||||
"cancel_loose_end: cancelling approval rows requires the `approvals` tool group"
|
||||
.to_owned(),
|
||||
);
|
||||
)
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
|
@ -242,18 +250,14 @@ mod tests {
|
|||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn approval_cancel_rejects_unprivileged_callers() {
|
||||
// A non-privileged caller (any regular agent socket) must not be
|
||||
// able to cancel approval rows even if it invents an id. The guard
|
||||
// is server-side so client cooperation is irrelevant — and it keys
|
||||
// on the socket-derived `privileged` flag, not on any agent name.
|
||||
let err = check_can_cancel_approval(false).unwrap_err();
|
||||
assert!(err.contains("only a privileged"), "{err}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn approval_cancel_allows_privileged() {
|
||||
check_can_cancel_approval(true).expect("a privileged caller must pass the guard");
|
||||
fn approval_cancel_rejects_callers_without_the_approvals_group() {
|
||||
// A caller that doesn't hold the `approvals` tool-group must not be
|
||||
// able to cancel approval rows even if it invents an id. The guard is
|
||||
// server-side so client cooperation is irrelevant — and it keys on a
|
||||
// grantable capability (the tool-group), not on any agent name.
|
||||
// `groups_for` of a name with no tool_groups.json entry is empty.
|
||||
let err = check_can_cancel_approval("nobody-with-no-groups").unwrap_err();
|
||||
assert!(err.contains("approvals` tool group"), "{err}");
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -130,7 +130,7 @@ fn inline_fallback(req_path: &str, reason: &str, message: &str) -> String {
|
|||
/// Persist `message` to `host_path` with the symlink-escape defenses
|
||||
/// described in the module docs. Returns `Ok(())` on success, or a
|
||||
/// human-readable reason string on any failure (caller logs +
|
||||
/// inline-falls-back). `pub` because `agent_server::handle_remind`
|
||||
/// inline-falls-back). `pub` because `socket_server::handle_remind`
|
||||
/// reuses it for the at-remind-time auto-file path.
|
||||
pub fn write_payload(agent: &str, host_path: &Path, message: &str) -> Result<(), String> {
|
||||
let Some(parent) = host_path.parent() else {
|
||||
|
|
@ -186,7 +186,7 @@ pub fn container_state_prefix(agent: &str) -> String {
|
|||
/// validating that it lives under the agent's own state subtree, has
|
||||
/// a non-empty relative tail, and doesn't try to traverse out via
|
||||
/// `..`. Returns the host `PathBuf` on success, or a human-readable
|
||||
/// reason string on rejection. `pub` so `agent_server::handle_remind`
|
||||
/// reason string on rejection. `pub` so `socket_server::handle_remind`
|
||||
/// can reuse it for the at-remind-time auto-file path.
|
||||
pub fn resolve_host_path(agent: &str, req_path: &str) -> Result<PathBuf, String> {
|
||||
let prefix = container_state_prefix(agent);
|
||||
|
|
|
|||
2161
hive-c0re/src/socket_server.rs
Normal file
2161
hive-c0re/src/socket_server.rs
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -140,7 +140,7 @@ pub fn resolve_recipient_in(
|
|||
/// `candidate` upward; the walk terminates at root or on a cycle
|
||||
/// (cycle defence: bounded to 32 hops, more than any plausible
|
||||
/// hive depth). Used by the cancel-authorization check in
|
||||
/// `manager_server::handle_cancel_schedule` to enforce
|
||||
/// `socket_server::handle_cancel_schedule` to enforce
|
||||
/// "managers can cancel anything their subtree owns."
|
||||
#[must_use]
|
||||
pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool {
|
||||
|
|
|
|||
Loading…
Reference in a new issue