Compare commits

...
Author SHA1 Message Date
atlas
fd2eac99ce refactor(#1865): drop stray issue-ref from start_manager doc (tracker-tag lint) 2026-06-22 13:58:52 +02:00
atlas
53b4e752ef refactor(#1865): replace the privileged flag with topology + capability gating
Per operator direction (no privileged mode; everything is perms /
capabilities), remove the socket-derived `privileged: bool` from the
unified dispatch and gate every verb on the caller's identity instead:

- serve/dispatch/dispatch_shared/dispatch_orchestration + all lifecycle
  handlers drop the `privileged` param.
- lifecycle (start/kill/restart/update/init_config/apply_commit) + get_logs
  gate on `topology::is_descendant_of` (a parent owns its whole subtree; the
  root covers every agent as a consequence, no positional privilege). The
  restart infra-branch stays InfraAdmin-gated (orthogonal).
- agent-state queries (loose-ends / reminder count + rollup): own subtree is
  free, other agents + the hive-wide `"*"` sweep require QueryAgentState.
  require_new_child + resolve_agent_state_target widened direct-child -> subtree.
- hive-wide orchestration verbs gate on the grantable tool-group via
  tool_groups::groups_for: schedules -> `scheduling`, meta-inputs +
  cancel-approval -> `approvals`. update_meta_inputs now attributes the
  approval to the caller, not a hardcoded MANAGER_AGENT.
- #1834 cancel-guard unwind: handle_cancel_loose_end drops `privileged`
  (agent path is never privileged); question/reminder cancels are
  ownership-only, approval cancel checks the `approvals` tool-group.

The manager socket stays as pure transport (serves agent=ruth, no authority
of its own); collapsing it into ruth's per-agent socket is the #1825
follow-up. No is_root here — root-identity primitives are #1825's.
2026-06-22 13:58:52 +02:00
atlas
674505fbe7 refactor(#1865): drop stray issue-ref from doc comment (tracker-tag lint) 2026-06-22 13:58:52 +02:00
atlas
f5f06a5f14 refactor(#1865): consolidate agent + manager socket servers into one
The per-agent and manager sockets ran two parallel dispatchers with
duplicated lifecycle handlers (agent-side topology-gated, manager-side
ungated) plus a manager-only handler set. Collapse to one parameterized
server in socket_server.rs:

- one serve() + dispatch(req, agent, privileged, coord); start() binds
  the per-agent sockets (privileged=false), start_manager() binds the
  manager socket (privileged=true).
- each lifecycle/config handler (start/restart/kill/update/init_config/
  apply_commit) merges its dual: the topology guard (require_child /
  require_new_child) runs only on the !privileged path; init_config
  records the requester as parent only when !privileged. restart keeps
  the orthogonal, capability-gated + audited infra-container branch.
- the agent-state queries (loose-ends / reminder count + rollup) branch
  on privileged: privileged keeps any-target + the "*" hive-wide sweep
  (query_agent_state-gated), non-privileged keeps the topology/cap gate.
- the privileged-only verbs (schedules / meta-inputs / get_logs) plus
  the submit/schedule/watchdog helpers move into socket_server; they are
  reached via dispatch_privileged_only(), which rejects the whole group
  on a non-privileged socket.
- delete manager_server.rs; repoint refs; merge the test modules.

No behavior change: the topology guard still applies on every
non-privileged lifecycle call, the privileged socket still acts on any
agent, and privileged-only verbs are still rejected on agent sockets.
2026-06-22 13:58:52 +02:00
atlas
a053d33184 refactor(#1865): rename agent_server module to socket_server
Pure rename ahead of the agent+manager server consolidation: the
per-agent socket dispatcher already hosts the shared dispatch and all
lifecycle handlers, and will absorb the manager-only handlers next, so
`agent_server` becomes a misnomer. No logic change — git mv plus a
mechanical `agent_server` -> `socket_server` rename across refs.
2026-06-22 13:58:52 +02:00
15 changed files with 2229 additions and 2343 deletions

File diff suppressed because it is too large Load diff

View file

@ -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.

View file

@ -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

View file

@ -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`.

View file

@ -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;

View file

@ -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(

View file

@ -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()
}

View file

@ -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;

View file

@ -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.

View file

@ -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

View file

@ -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}");
}
}

View file

@ -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);

File diff suppressed because it is too large Load diff

View file

@ -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 {