Compare commits
10 changed files with 45 additions and 194 deletions
|
|
@ -159,9 +159,7 @@ pub async fn run_approval_apply_commit(
|
||||||
approval_id: i64,
|
approval_id: i64,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::ApplyCommit)?;
|
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::ApplyCommit)?;
|
||||||
// Runtime dir creation is handled inside lifecycle::rebuild_no_meta's
|
let agent_dir = coord.ensure_runtime(&approval.agent)?;
|
||||||
// spawn path (first-spawn) or is already present for rebuilds.
|
|
||||||
let agent_dir = Coordinator::agent_dir(&approval.agent);
|
|
||||||
let applied_dir = Coordinator::agent_applied_dir(&approval.agent);
|
let applied_dir = Coordinator::agent_applied_dir(&approval.agent);
|
||||||
coord.set_queue_step(queue_entry_id, "apply commit");
|
coord.set_queue_step(queue_entry_id, "apply commit");
|
||||||
let (result, terminal_tag, is_first_spawn) =
|
let (result, terminal_tag, is_first_spawn) =
|
||||||
|
|
@ -194,7 +192,7 @@ pub async fn run_approval_merge_config_pr(
|
||||||
approval_id: i64,
|
approval_id: i64,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::MergeConfigPr)?;
|
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::MergeConfigPr)?;
|
||||||
let agent_dir = Coordinator::agent_dir(&approval.agent);
|
let agent_dir = coord.ensure_runtime(&approval.agent)?;
|
||||||
let applied_dir = Coordinator::agent_applied_dir(&approval.agent);
|
let applied_dir = Coordinator::agent_applied_dir(&approval.agent);
|
||||||
coord.set_queue_step(queue_entry_id, "merge config pr");
|
coord.set_queue_step(queue_entry_id, "merge config pr");
|
||||||
let (result, terminal_tag) =
|
let (result, terminal_tag) =
|
||||||
|
|
|
||||||
|
|
@ -545,11 +545,14 @@ impl Coordinator {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Assemble the per-agent filesystem paths for `name`. `agent_dir`
|
/// Assemble the per-agent filesystem paths for `name`. The caller
|
||||||
/// is the runtime directory (`/run/hyperhive/agents/<name>`), obtained
|
/// must supply `agent_dir` (from `ensure_runtime`) since that
|
||||||
/// from `Coordinator::agent_dir(name)` (pure path) or from
|
/// creates the tmpfs entry on first call. All other paths are
|
||||||
/// `lifecycle::ensure_agent_runtime_dir` + `agent_dir` when the dir
|
/// derived statically from `name`.
|
||||||
/// must be created. All other paths are derived statically from `name`.
|
///
|
||||||
|
/// ```ignore
|
||||||
|
/// let paths = Coordinator::agent_paths(name, coord.ensure_runtime(name)?);
|
||||||
|
/// ```
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn agent_paths(name: &str, agent_dir: PathBuf) -> AgentPaths {
|
pub fn agent_paths(name: &str, agent_dir: PathBuf) -> AgentPaths {
|
||||||
AgentPaths {
|
AgentPaths {
|
||||||
|
|
@ -1450,16 +1453,11 @@ impl Coordinator {
|
||||||
Self::agent_dir(name).join("mcp.sock")
|
Self::agent_dir(name).join("mcp.sock")
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Ensure a runtime dir + (for sub-agents) per-agent socket exists.
|
/// Ensure a runtime dir + (for sub-agents) per-agent socket exists. For
|
||||||
///
|
/// the manager, `socket_server::start_manager` owns the socket — just return
|
||||||
/// **Prefer the split form (event-driven):**
|
/// the dir. For sub-agents this is `register_agent` (creates a fresh
|
||||||
/// - dir creation → `lifecycle::ensure_agent_runtime_dir(name)`
|
/// listener bound to `socket_path(name)`). Source directory of the
|
||||||
/// - listener registration → `register_agent(name)` at each explicit
|
/// `/run/hive/mcp.sock` bind that ends up in `set_nspawn_flags`.
|
||||||
/// lifecycle event (spawn, reconcile-start). `mcp_sockets::sync_on_start`
|
|
||||||
/// covers the daemon-restart case.
|
|
||||||
///
|
|
||||||
/// This method is kept as a convenience shim for any remaining callers
|
|
||||||
/// that need the combined semantics in one call.
|
|
||||||
pub fn ensure_runtime(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
|
pub fn ensure_runtime(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
|
||||||
if name == crate::lifecycle::MANAGER_NAME {
|
if name == crate::lifecycle::MANAGER_NAME {
|
||||||
let dir = Self::agent_dir(name);
|
let dir = Self::agent_dir(name);
|
||||||
|
|
|
||||||
|
|
@ -97,10 +97,9 @@ async fn run_prebuild(
|
||||||
relock: bool,
|
relock: bool,
|
||||||
) -> Result<NodeOutput> {
|
) -> Result<NodeOutput> {
|
||||||
let name = &claim.agent;
|
let name = &claim.agent;
|
||||||
// Prebuild runs while the agent is still up — the runtime dir and
|
let agent_dir = coord
|
||||||
// MCP listener already exist. Use the pure path accessor; no need
|
.ensure_runtime(name)
|
||||||
// to re-register the listener (event-driven: registered at start/create).
|
.with_context(|| format!("ensure_runtime {name}"))?;
|
||||||
let agent_dir = Coordinator::agent_dir(name);
|
|
||||||
let hive = coord.hive_env();
|
let hive = coord.hive_env();
|
||||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||||
crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?;
|
crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?;
|
||||||
|
|
@ -132,9 +131,7 @@ async fn run_prebuild(
|
||||||
/// `Reconcile` runs after this node terminal ok *or* fail.
|
/// `Reconcile` runs after this node terminal ok *or* fail.
|
||||||
async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
||||||
let name = &claim.agent;
|
let name = &claim.agent;
|
||||||
// Swap runs on an already-existing (stopped) container — runtime dir
|
let agent_dir = coord.ensure_runtime(name)?;
|
||||||
// and listener were created earlier. Pure path accessor suffices.
|
|
||||||
let agent_dir = Coordinator::agent_dir(name);
|
|
||||||
let hive = coord.hive_env();
|
let hive = coord.hive_env();
|
||||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||||
let result =
|
let result =
|
||||||
|
|
@ -181,7 +178,7 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
|
||||||
/// build+create — no prebuild needed).
|
/// build+create — no prebuild needed).
|
||||||
async fn run_create(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
async fn run_create(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
|
||||||
let name = &claim.agent;
|
let name = &claim.agent;
|
||||||
let agent_dir = Coordinator::agent_dir(name);
|
let agent_dir = coord.ensure_runtime(name)?;
|
||||||
let hive = coord.hive_env();
|
let hive = coord.hive_env();
|
||||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||||
ctx.step("nixos-container create");
|
ctx.step("nixos-container create");
|
||||||
|
|
@ -189,9 +186,6 @@ async fn run_create(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> R
|
||||||
// (sync_agents commit) before `nixos-container create` — hold the
|
// (sync_agents commit) before `nixos-container create` — hold the
|
||||||
// deploy-window gate so that commit can't land inside another
|
// deploy-window gate so that commit can't land inside another
|
||||||
// node's staged deploy window.
|
// node's staged deploy window.
|
||||||
// Runtime dir creation and MCP listener registration are deferred to
|
|
||||||
// the tail Reconcile (converge_start_preamble + register_agent) so this
|
|
||||||
// node stays purely "provision + create", not "create + start".
|
|
||||||
let _window = crate::meta::exclusive().await;
|
let _window = crate::meta::exclusive().await;
|
||||||
crate::lifecycle::create_container(name, &hive, &paths).await?;
|
crate::lifecycle::create_container(name, &hive, &paths).await?;
|
||||||
Ok(NodeOutput::default())
|
Ok(NodeOutput::default())
|
||||||
|
|
@ -251,22 +245,21 @@ async fn run_reconcile(
|
||||||
.transient
|
.transient
|
||||||
.is_none()
|
.is_none()
|
||||||
.then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting));
|
.then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting));
|
||||||
// Run the typed start preamble: ensures the runtime dir
|
// Converge the ephemeral host-side state before the start:
|
||||||
// exists and writes the nspawn/resource-limits drop-ins.
|
// `/run/hyperhive/agents/<name>` + `/run/hive-agent/<name>`
|
||||||
// The returned StartableAgent token is the only way to call
|
// (both nspawn bind sources — tmpfs, empty after a host
|
||||||
// start_with_fallback — omitting this becomes a compile error.
|
// reboot; nspawn refuses to start with a missing source)
|
||||||
let agent_dir = Coordinator::agent_dir(name);
|
// and the resource-limits drop-in under
|
||||||
|
// `/run/systemd/system/`. Rebuild DAGs get this from their
|
||||||
|
// Prebuild/Swap nodes; the bare-Reconcile templates (boot
|
||||||
|
// reconcile, plain start/restart) otherwise start with
|
||||||
|
// nothing under /run and fail.
|
||||||
|
let agent_dir = coord.ensure_runtime(name)?;
|
||||||
let hive = coord.hive_env();
|
let hive = coord.hive_env();
|
||||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||||
let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?;
|
crate::lifecycle::write_dropins(name, &hive, &paths).await?;
|
||||||
ctx.step("nixos-container start");
|
ctx.step("nixos-container start");
|
||||||
crate::lifecycle::start_with_fallback(token).await?;
|
crate::lifecycle::start_with_fallback(name).await?;
|
||||||
// Bind the MCP listener immediately after starting the container.
|
|
||||||
// The preamble created the runtime dir; the container is now
|
|
||||||
// coming up and will connect to this socket on its first turn.
|
|
||||||
// Event-driven (no background poll) — c0re owns the listener
|
|
||||||
// lifecycle, so register here rather than waiting for a sweep.
|
|
||||||
coord.register_agent(name)?;
|
|
||||||
coord.kick_agent(name, "container started");
|
coord.kick_agent(name, "container started");
|
||||||
coord.rescan_containers_and_emit().await;
|
coord.rescan_containers_and_emit().await;
|
||||||
}
|
}
|
||||||
|
|
@ -343,10 +336,7 @@ async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Re
|
||||||
/// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
|
/// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
|
||||||
async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||||||
let name = &claim.agent;
|
let name = &claim.agent;
|
||||||
// write_dropins only needs the path value to build AgentPaths; the
|
let agent_dir = coord.ensure_runtime(name)?;
|
||||||
// dir doesn't need to exist at this point (created by ensure_agent_runtime_dir
|
|
||||||
// on the upstream Prebuild/Start node).
|
|
||||||
let agent_dir = Coordinator::agent_dir(name);
|
|
||||||
let hive = coord.hive_env();
|
let hive = coord.hive_env();
|
||||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||||
crate::lifecycle::write_dropins(name, &hive, &paths).await?;
|
crate::lifecycle::write_dropins(name, &hive, &paths).await?;
|
||||||
|
|
|
||||||
|
|
@ -50,6 +50,6 @@ pub use stores::{
|
||||||
approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
|
approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
|
||||||
};
|
};
|
||||||
pub use workers::{
|
pub use workers::{
|
||||||
agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler,
|
agent_sockets, auto_update, crash_watch, knowledge, reminder_scheduler,
|
||||||
scheduled_prompts_worker,
|
scheduled_prompts_worker,
|
||||||
};
|
};
|
||||||
|
|
|
||||||
|
|
@ -275,9 +275,6 @@ async fn port_collision(self_name: &str) -> Option<String> {
|
||||||
|
|
||||||
pub async fn spawn(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
|
pub async fn spawn(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
|
||||||
create_container(name, hive, paths).await?;
|
create_container(name, hive, paths).await?;
|
||||||
// Runtime dir must exist before nixos-container start (nspawn bind-mount
|
|
||||||
// source). Create it here so callers don't need a separate preamble step.
|
|
||||||
ensure_agent_runtime_dir(name)?;
|
|
||||||
write_dropins(name, hive, paths).await?;
|
write_dropins(name, hive, paths).await?;
|
||||||
priv_run("start", name).await
|
priv_run("start", name).await
|
||||||
}
|
}
|
||||||
|
|
@ -423,59 +420,17 @@ pub async fn start(name: &str) -> Result<()> {
|
||||||
priv_run("start", name).await
|
priv_run("start", name).await
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Opaque token produced by [`converge_start_preamble`].
|
|
||||||
/// [`start_with_fallback`] requires this as proof that the pre-start
|
|
||||||
/// preamble (runtime dir + drop-ins) ran. Dropping the token without
|
|
||||||
/// calling `start_with_fallback` is a no-op.
|
|
||||||
#[must_use = "call lifecycle::start_with_fallback(token) to start the container"]
|
|
||||||
pub struct StartableAgent {
|
|
||||||
name: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Run the per-agent start preamble: ensure the runtime dir exists and write
|
|
||||||
/// the nspawn / resource-limits drop-ins. Returns a [`StartableAgent`] token
|
|
||||||
/// as typed proof that the preamble ran; pass it to [`start_with_fallback`].
|
|
||||||
/// Callers that omit this step cannot call `start_with_fallback` — the type
|
|
||||||
/// system makes forgetting the preamble a compile error.
|
|
||||||
///
|
|
||||||
/// # Errors
|
|
||||||
///
|
|
||||||
/// Returns an error if `ensure_agent_runtime_dir` or `write_dropins` fails.
|
|
||||||
pub async fn converge_start_preamble(
|
|
||||||
name: &str,
|
|
||||||
hive: &HiveEnv,
|
|
||||||
paths: &AgentPaths,
|
|
||||||
) -> Result<StartableAgent> {
|
|
||||||
ensure_agent_runtime_dir(name)?;
|
|
||||||
write_dropins(name, hive, paths).await?;
|
|
||||||
Ok(StartableAgent {
|
|
||||||
name: name.to_owned(),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Start with the cold-start fallback: when a plain start fails (the
|
/// Start with the cold-start fallback: when a plain start fails (the
|
||||||
/// activation-error shape), retry once via stop + kill + start before
|
/// activation-error shape), retry once via stop + kill + start before
|
||||||
/// giving up. Used by the queue's fast-lane `Start` handler and the
|
/// giving up. Used by the queue's fast-lane `Start` handler and the
|
||||||
/// inline start-after-rebuild path.
|
/// inline start-after-rebuild path.
|
||||||
/// See `docs/coordinator.md::Cold-start fallback`.
|
/// See `docs/coordinator.md::Cold-start fallback`.
|
||||||
///
|
///
|
||||||
/// Requires a [`StartableAgent`] token from [`converge_start_preamble`]
|
|
||||||
/// to prove the preamble ran. For internal use within this module (where
|
|
||||||
/// the preamble is already enforced structurally) call
|
|
||||||
/// `start_with_fallback_inner` directly.
|
|
||||||
///
|
|
||||||
/// # Errors
|
/// # Errors
|
||||||
///
|
///
|
||||||
/// Propagates the retry's start error (annotated with the original
|
/// Propagates the retry's start error (annotated with the original
|
||||||
/// failure) when the fallback also fails.
|
/// failure) when the fallback also fails.
|
||||||
pub async fn start_with_fallback(token: StartableAgent) -> Result<()> {
|
pub async fn start_with_fallback(name: &str) -> Result<()> {
|
||||||
start_with_fallback_inner(&token.name).await
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Internal implementation of the cold-start fallback. Used by
|
|
||||||
/// [`start_with_fallback`] (public, token-gated) and by
|
|
||||||
/// [`rebuild_no_meta`] where the preamble is already enforced structurally.
|
|
||||||
async fn start_with_fallback_inner(name: &str) -> Result<()> {
|
|
||||||
validate(name)?;
|
validate(name)?;
|
||||||
if let Err(start_err) = priv_run("start", name).await {
|
if let Err(start_err) = priv_run("start", name).await {
|
||||||
let container = container_name(name);
|
let container = container_name(name);
|
||||||
|
|
@ -622,9 +577,7 @@ pub async fn rebuild_no_meta(
|
||||||
return Ok(true);
|
return Ok(true);
|
||||||
}
|
}
|
||||||
on_step("nixos-container start");
|
on_step("nixos-container start");
|
||||||
// write_dropins was called above; use the inner fn directly
|
start_with_fallback(name).await?;
|
||||||
// since the preamble is enforced structurally in this path.
|
|
||||||
start_with_fallback_inner(name).await?;
|
|
||||||
}
|
}
|
||||||
Ok(false)
|
Ok(false)
|
||||||
} else {
|
} else {
|
||||||
|
|
@ -632,8 +585,6 @@ pub async fn rebuild_no_meta(
|
||||||
// See `docs/coordinator.md::Spawn path`.
|
// See `docs/coordinator.md::Spawn path`.
|
||||||
on_step("nixos-container create");
|
on_step("nixos-container create");
|
||||||
priv_run("create", name).await?;
|
priv_run("create", name).await?;
|
||||||
// Runtime dir must exist before nixos-container start.
|
|
||||||
ensure_agent_runtime_dir(name)?;
|
|
||||||
write_dropins(name, hive, paths).await?;
|
write_dropins(name, hive, paths).await?;
|
||||||
on_step("nixos-container start");
|
on_step("nixos-container start");
|
||||||
priv_run("start", name).await?;
|
priv_run("start", name).await?;
|
||||||
|
|
@ -798,23 +749,6 @@ pub async fn sync_tmpfiles() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Ensure the per-agent runtime directory `/run/hyperhive/agents/<name>`
|
|
||||||
/// exists. The directory is also written by `SyncAgentTmpfiles` (run at
|
|
||||||
/// boot + spawn/destroy), but explicit creation in start/spawn paths guards
|
|
||||||
/// against races where hive-c0re starts a container before tmpfiles.d has
|
|
||||||
/// applied the new entry.
|
|
||||||
///
|
|
||||||
/// Pure filesystem op — no `Coordinator` dependency — so callers that only
|
|
||||||
/// need the dir do not have to hold an `Arc<Coordinator>`.
|
|
||||||
///
|
|
||||||
/// # Errors
|
|
||||||
/// Returns an error if `create_dir_all` fails.
|
|
||||||
pub fn ensure_agent_runtime_dir(name: &str) -> Result<()> {
|
|
||||||
let dir = std::path::PathBuf::from(format!("/run/hyperhive/agents/{name}"));
|
|
||||||
std::fs::create_dir_all(&dir)
|
|
||||||
.with_context(|| format!("create agent runtime dir {}", dir.display()))
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Build the per-line callback for `create_container_streaming` /
|
/// Build the per-line callback for `create_container_streaming` /
|
||||||
/// `update_container_streaming`. Both ops share identical dispatch logic
|
/// `update_container_streaming`. Both ops share identical dispatch logic
|
||||||
/// (stdout → info + `append_stdout`, stderr → warn + `append_stderr`); this
|
/// (stdout → info + `append_stdout`, stderr → warn + `append_stderr`); this
|
||||||
|
|
|
||||||
|
|
@ -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, broker, client, crash_watch, dashboard, dashboard_events, forge,
|
agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events, forge,
|
||||||
job_queue, knowledge, matrix, mcp_sockets, migrate, reminder_scheduler,
|
job_queue, knowledge, matrix, migrate, reminder_scheduler, scheduled_prompts_worker, server,
|
||||||
scheduled_prompts_worker, server, socket_server,
|
socket_server,
|
||||||
};
|
};
|
||||||
|
|
||||||
#[derive(Parser)]
|
#[derive(Parser)]
|
||||||
|
|
@ -420,13 +420,6 @@ async fn cmd_serve(
|
||||||
// is one stat per agent per tick.
|
// is one stat per agent per tick.
|
||||||
// See `docs/gateway.md::Per-agent unix-socket upstream`.
|
// See `docs/gateway.md::Per-agent unix-socket upstream`.
|
||||||
agent_sockets::spawn_poll();
|
agent_sockets::spawn_poll();
|
||||||
// MCP socket listener startup sync: one-shot sweep that re-registers any
|
|
||||||
// running agent container whose MCP listener was lost when hive-c0re
|
|
||||||
// restarted (Coordinator starts empty; /run/hyperhive/agents/ is tmpfs).
|
|
||||||
// After this, listener registration is event-driven: run_create /
|
|
||||||
// run_reconcile call register_agent on start; kill/destroy call
|
|
||||||
// unregister_agent. No recurring poll needed — c0re owns the listeners.
|
|
||||||
mcp_sockets::sync_on_start(coord.clone()).await;
|
|
||||||
// Reminder scheduler: drains due reminders + handles
|
// Reminder scheduler: drains due reminders + handles
|
||||||
// file_path payload persistence. See reminder_scheduler.rs.
|
// file_path payload persistence. See reminder_scheduler.rs.
|
||||||
reminder_scheduler::spawn(coord.clone());
|
reminder_scheduler::spawn(coord.clone());
|
||||||
|
|
|
||||||
|
|
@ -202,21 +202,14 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
|
||||||
/// registration and notifying the manager on failure.
|
/// registration and notifying the manager on failure.
|
||||||
async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
|
async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
|
||||||
tracing::info!(%name, "spawn");
|
tracing::info!(%name, "spawn");
|
||||||
let agent_dir = Coordinator::agent_dir(name);
|
let agent_dir = coord.ensure_runtime(name)?;
|
||||||
let hive = coord.hive_env();
|
let hive = coord.hive_env();
|
||||||
let paths = Coordinator::agent_paths(name, agent_dir);
|
let paths = Coordinator::agent_paths(name, agent_dir);
|
||||||
// lifecycle::spawn creates the runtime dir internally before start.
|
|
||||||
// MCP listener registration is event-driven: bind immediately on
|
|
||||||
// success so the harness can connect on its first turn without
|
|
||||||
// waiting for any poll interval.
|
|
||||||
match lifecycle::spawn(name, &hive, &paths).await {
|
match lifecycle::spawn(name, &hive, &paths).await {
|
||||||
Ok(()) => {
|
Ok(()) => {
|
||||||
if let Err(e) = coord.power.set(name, crate::power::Wanted::Up) {
|
if let Err(e) = coord.power.set(name, crate::power::Wanted::Up) {
|
||||||
tracing::warn!(%name, error = ?e, "agent_power: set wanted=up failed");
|
tracing::warn!(%name, error = ?e, "agent_power: set wanted=up failed");
|
||||||
}
|
}
|
||||||
// Bind the MCP listener now that the container is starting up.
|
|
||||||
// The harness connects to this socket on its first turn.
|
|
||||||
coord.register_agent(name)?;
|
|
||||||
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
|
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
|
||||||
agent: name.to_owned(),
|
agent: name.to_owned(),
|
||||||
ok: true,
|
ok: true,
|
||||||
|
|
@ -227,8 +220,8 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
|
||||||
tokio::spawn(lifecycle::sync_tmpfiles());
|
tokio::spawn(lifecycle::sync_tmpfiles());
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
// Spawn failed: register_agent was never called, so there is
|
// Roll back socket registration if container creation failed.
|
||||||
// nothing to unregister. Notify the manager and propagate.
|
coord.unregister_agent(name);
|
||||||
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
|
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
|
||||||
agent: name.to_owned(),
|
agent: name.to_owned(),
|
||||||
ok: false,
|
ok: false,
|
||||||
|
|
|
||||||
|
|
@ -153,9 +153,7 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
tracing::info!("manager container missing — spawning");
|
tracing::info!("manager container missing — spawning");
|
||||||
// lifecycle::spawn creates the runtime dir internally; no manual
|
let runtime = coord.ensure_runtime(MANAGER_NAME)?;
|
||||||
// ensure_agent_runtime_dir needed here.
|
|
||||||
let runtime = Coordinator::agent_dir(MANAGER_NAME);
|
|
||||||
let hive = coord.hive_env();
|
let hive = coord.hive_env();
|
||||||
let paths = Coordinator::agent_paths(MANAGER_NAME, runtime);
|
let paths = Coordinator::agent_paths(MANAGER_NAME, runtime);
|
||||||
lifecycle::spawn(MANAGER_NAME, &hive, &paths).await?;
|
lifecycle::spawn(MANAGER_NAME, &hive, &paths).await?;
|
||||||
|
|
|
||||||
|
|
@ -1,52 +0,0 @@
|
||||||
//! MCP socket listener boot sync.
|
|
||||||
//!
|
|
||||||
//! On hive-c0re startup, any agent containers that survived the daemon restart
|
|
||||||
//! still have their bind-mount source dirs but no live MCP listener (the
|
|
||||||
//! `Coordinator` is freshly empty). `sync_on_start` does a one-shot sweep to
|
|
||||||
//! re-register all running agents.
|
|
||||||
//!
|
|
||||||
//! After startup, listeners are managed event-driven:
|
|
||||||
//! - `run_create` calls `register_agent` eagerly on first-spawn.
|
|
||||||
//! - `run_reconcile` calls `register_agent` immediately after `start_with_fallback`.
|
|
||||||
//! - `kill`/`destroy` paths call `unregister_agent`.
|
|
||||||
//!
|
|
||||||
//! No recurring poll is needed because c0re owns the listener lifecycle —
|
|
||||||
//! a listener can only disappear when c0re itself restarts, which is exactly
|
|
||||||
//! the case `sync_on_start` covers.
|
|
||||||
|
|
||||||
use std::sync::Arc;
|
|
||||||
|
|
||||||
use crate::coordinator::Coordinator;
|
|
||||||
|
|
||||||
/// One-shot MCP listener sync run at daemon startup.
|
|
||||||
///
|
|
||||||
/// Iterates all currently-running agent containers and calls `register_agent`
|
|
||||||
/// for any that have no live listener in the `Coordinator`. Safe to call
|
|
||||||
/// concurrently with the rest of startup — `register_agent` is idempotent
|
|
||||||
/// (drops and rebinds) and the coordinator lock serialises concurrent calls.
|
|
||||||
pub async fn sync_on_start(coord: Arc<Coordinator>) {
|
|
||||||
let running = match crate::lifecycle::list().await {
|
|
||||||
Ok(names) => names,
|
|
||||||
Err(e) => {
|
|
||||||
tracing::warn!(error = ?e, "mcp_sockets: startup sync failed to list agents; MCP listeners may be missing until next start");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
let registered = coord.list_agents();
|
|
||||||
for container in running {
|
|
||||||
let Some(name) = container.strip_prefix(crate::lifecycle::AGENT_PREFIX) else {
|
|
||||||
continue;
|
|
||||||
};
|
|
||||||
if !registered.contains(&name.to_owned()) {
|
|
||||||
if let Err(e) = coord.register_agent(name) {
|
|
||||||
tracing::warn!(
|
|
||||||
agent = %name,
|
|
||||||
error = ?e,
|
|
||||||
"mcp_sockets: startup register_agent failed"
|
|
||||||
);
|
|
||||||
} else {
|
|
||||||
tracing::debug!(agent = %name, "mcp_sockets: registered listener on startup");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,13 +1,12 @@
|
||||||
//! Background tasks and periodic sweeps: crash/login watcher, the
|
//! Background tasks and periodic sweeps: crash/login watcher, the
|
||||||
//! reminder and scheduled-prompt delivery loops, boot-time auto-update
|
//! reminder and scheduled-prompt delivery loops, boot-time auto-update
|
||||||
//! reconcile, the agent-sockets.json writer loop, the MCP socket listener
|
//! reconcile, the agent-sockets.json writer loop, and knowledge-repo
|
||||||
//! reconcile loop, and knowledge-repo sync. Each submodule is re-exported
|
//! sync. Each submodule is re-exported at the crate root, so
|
||||||
//! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged.
|
//! `crate::crash_watch::…` etc. keep working unchanged.
|
||||||
|
|
||||||
pub mod agent_sockets;
|
pub mod agent_sockets;
|
||||||
pub mod auto_update;
|
pub mod auto_update;
|
||||||
pub mod crash_watch;
|
pub mod crash_watch;
|
||||||
pub mod knowledge;
|
pub mod knowledge;
|
||||||
pub mod mcp_sockets;
|
|
||||||
pub mod reminder_scheduler;
|
pub mod reminder_scheduler;
|
||||||
pub mod scheduled_prompts_worker;
|
pub mod scheduled_prompts_worker;
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue