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.
This commit is contained in:
parent
a053d33184
commit
f5f06a5f14
11 changed files with 1034 additions and 1118 deletions
|
|
@ -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
|
/// 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
|
/// (derived from the file's mtime). Returns `(None, None)` when the file is absent
|
||||||
/// or empty. `pub` so `socket_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
|
/// NB: callers building `AgentMeta` for a *stopped* container should
|
||||||
/// clear the result — the on-disk status is a stale snapshot from
|
/// clear the result — the on-disk status is a stale snapshot from
|
||||||
|
|
|
||||||
|
|
@ -522,7 +522,7 @@ impl Coordinator {
|
||||||
let mut schedules: Vec<hive_sh4re::WireSchedule> = match self.scheduled_prompts.list() {
|
let mut schedules: Vec<hive_sh4re::WireSchedule> = match self.scheduled_prompts.list() {
|
||||||
Ok(rows) => rows
|
Ok(rows) => rows
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(crate::manager_server::schedule_to_wire_public)
|
.map(crate::socket_server::schedule_to_wire_public)
|
||||||
.collect(),
|
.collect(),
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed");
|
tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed");
|
||||||
|
|
@ -534,7 +534,7 @@ impl Coordinator {
|
||||||
// momentarily contended we emit unfiltered rather than block this
|
// momentarily contended we emit unfiltered rather than block this
|
||||||
// sync path — the next snapshot / page reload corrects it.
|
// sync path — the next snapshot / page reload corrects it.
|
||||||
if let Some(live) = self.live_container_names_blocking() {
|
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 {
|
self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
|
||||||
seq: self.next_seq(),
|
seq: self.next_seq(),
|
||||||
|
|
@ -1330,7 +1330,7 @@ impl Coordinator {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Ensure a runtime dir + (for sub-agents) per-agent socket exists. For
|
/// 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
|
/// the dir. For sub-agents this is `register_agent` (creates a fresh
|
||||||
/// listener bound to `socket_path(name)`). Source directory of the
|
/// listener bound to `socket_path(name)`). Source directory of the
|
||||||
/// `/run/hive/mcp.sock` bind that ends up in `set_nspawn_flags`.
|
/// `/run/hive/mcp.sock` bind that ends up in `set_nspawn_flags`.
|
||||||
|
|
|
||||||
|
|
@ -42,7 +42,7 @@ mod topology;
|
||||||
mod webhook;
|
mod webhook;
|
||||||
|
|
||||||
// Pre-computed at approval-submit time by the manager-socket handler
|
// 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`
|
// re-exported at the module root to preserve the `crate::dashboard::approval_diff`
|
||||||
// path across the submodule split.
|
// path across the submodule split.
|
||||||
pub(crate) use approvals::approval_diff;
|
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
|
// host-side approval queue without the manager up, and
|
||||||
// operator-driven meta-input updates work from the dashboard
|
// operator-driven meta-input updates work from the dashboard
|
||||||
// either way. The MCP-surface self-kill guard in
|
// 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
|
// manager calling Kill on its own container is self-suicide
|
||||||
// mid-call, not a legitimate operator action.
|
// mid-call, not a legitimate operator action.
|
||||||
state.coord.rebuild_queue.enqueue(
|
state.coord.rebuild_queue.enqueue(
|
||||||
|
|
|
||||||
|
|
@ -21,7 +21,7 @@ pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
|
||||||
Ok(rows) => {
|
Ok(rows) => {
|
||||||
let mut wire: Vec<hive_sh4re::WireSchedule> = rows
|
let mut wire: Vec<hive_sh4re::WireSchedule> = rows
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(crate::manager_server::schedule_to_wire_public)
|
.map(crate::socket_server::schedule_to_wire_public)
|
||||||
.collect();
|
.collect();
|
||||||
// Drop ghost targets (agents that no longer exist) so the
|
// Drop ghost targets (agents that no longer exist) so the
|
||||||
// table never shows dead columns. Uses the reliable async
|
// 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()
|
.into_iter()
|
||||||
.map(|c| c.name)
|
.map(|c| c.name)
|
||||||
.collect();
|
.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()
|
axum::Json(wire).into_response()
|
||||||
}
|
}
|
||||||
Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
|
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) {
|
match state.coord.scheduled_prompts.get(id) {
|
||||||
Ok(Some(s)) => {
|
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();
|
state.coord.emit_schedules_snapshot();
|
||||||
axum::Json(wire).into_response()
|
axum::Json(wire).into_response()
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -38,7 +38,6 @@ pub mod knowledge;
|
||||||
pub mod lifecycle;
|
pub mod lifecycle;
|
||||||
pub mod limits;
|
pub mod limits;
|
||||||
pub mod loose_ends;
|
pub mod loose_ends;
|
||||||
pub mod manager_server;
|
|
||||||
pub mod matrix;
|
pub mod matrix;
|
||||||
pub mod meta;
|
pub mod meta;
|
||||||
pub mod migrate;
|
pub mod migrate;
|
||||||
|
|
|
||||||
|
|
@ -13,8 +13,8 @@ use hive_sh4re::{HostRequest, HostResponse};
|
||||||
use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig};
|
use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig};
|
||||||
use hive_c0re::{
|
use hive_c0re::{
|
||||||
agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard,
|
agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard,
|
||||||
dashboard_events, events_vacuum, forge, knowledge, manager_server, matrix, migrate,
|
dashboard_events, events_vacuum, forge, knowledge, matrix, migrate, rebuild_queue,
|
||||||
rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server,
|
reminder_scheduler, scheduled_prompts_worker, server, socket_server,
|
||||||
};
|
};
|
||||||
|
|
||||||
#[derive(Parser)]
|
#[derive(Parser)]
|
||||||
|
|
@ -265,7 +265,7 @@ async fn cmd_serve(
|
||||||
// Copy value first for the dashboard + knowledge-webhook tasks.
|
// Copy value first for the dashboard + knowledge-webhook tasks.
|
||||||
let dashboard_port = env.dashboard_port;
|
let dashboard_port = env.dashboard_port;
|
||||||
let coord = Arc::new(Coordinator::open(&db, env, model_prices)?);
|
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
|
// Idempotent pre-flight: rewrite pre-meta-layout applied
|
||||||
// repos, ensure proposed repos carry the `applied`
|
// repos, ensure proposed repos carry the `applied`
|
||||||
// remote, bootstrap the meta repo, repoint containers at
|
// 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::approvals::kind_to_str;
|
||||||
use crate::coordinator::Coordinator;
|
use crate::coordinator::Coordinator;
|
||||||
use crate::limits;
|
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
|
/// Cap on how long an asker can demand an answer before the watchdog
|
||||||
/// auto-resolves with `[expired]`. Six hours mirrors typical agent
|
/// auto-resolves with `[expired]`. Six hours mirrors typical agent
|
||||||
|
|
|
||||||
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
|
/// `candidate` upward; the walk terminates at root or on a cycle
|
||||||
/// (cycle defence: bounded to 32 hops, more than any plausible
|
/// (cycle defence: bounded to 32 hops, more than any plausible
|
||||||
/// hive depth). Used by the cancel-authorization check in
|
/// 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."
|
/// "managers can cancel anything their subtree owns."
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool {
|
pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue