hive-c0re: push HelperEvent::Spawned as a todo, not a broker message

Follow-up from #2955 (mara: 'make core able to give agent a todo').
First migration slice: Spawned was pure FYI-check-when-convenient
material, not something needing an immediate turn.

Coordinator::push_todo/push_todo_submitter do a best-effort live dial
of the target agent's own hive-agent-sock (hive_host_sock::agent_todo_
socket), sending the exact UpsertTodo request in-container producers
(matrix/bash/forge-notify) already send. Push, not queue: agent
offline (socket absent) or dial failure is a silent no-op, no retry,
no fallback delivery -- matches mara's 'not available if offline'
call exactly.

HelperEvent::Spawned removed entirely (enum variant + all 3 call
sites migrated: handle_spawn's two arms, finish_approval's Spawn
approval-kind arm) rather than kept alongside a translation layer --
per mara's correction on the first design attempt, migrating the
producer means deleting the old path, not bridging it.

Verified: cargo build/clippy/test -p hive-c0re -p hive-host-sock
-p hive-sh4re clean (318 tests), nix fmt clean.
This commit is contained in:
damocles 2026-08-02 22:47:47 +02:00 committed by mara
commit d77f81cd1e
8 changed files with 150 additions and 37 deletions

2
Cargo.lock generated
View file

@ -1632,11 +1632,13 @@ dependencies = [
"clap-markdown", "clap-markdown",
"clap_complete", "clap_complete",
"forgejo-api", "forgejo-api",
"hive-agent-sock",
"hive-core-agent-sock", "hive-core-agent-sock",
"hive-host-sock", "hive-host-sock",
"hive-jobq", "hive-jobq",
"hive-priv-sock", "hive-priv-sock",
"hive-sh4re", "hive-sh4re",
"hive-sock-client",
"hive-types", "hive-types",
"hmac 0.13.0", "hmac 0.13.0",
"indicatif", "indicatif",

View file

@ -570,15 +570,18 @@ Two more paths resolve a pending question with a sentinel answer:
`Coordinator::notify_submitter(approval_id, &HelperEvent)` routes the `Coordinator::notify_submitter(approval_id, &HelperEvent)` routes the
event to the agent that originally submitted the approval (looked up from event to the agent that originally submitted the approval (looked up from
the `submitter` column on the `approvals` table). The harness delivers it the `submitter` column on the `approvals` table). The harness delivers it
as a regular `system` inbox message so it drives a normal claude turn. as a regular `system` inbox message so it drives a normal claude turn. A
Legacy approval rows that predate the submitter column fall back to the "FYI, check when convenient" event doesn't need a message — those go
through `Coordinator::push_todo`/`push_todo_submitter` instead, a direct
live dial of the target agent's in-container todo socket (same
`UpsertTodo` request in-container producers use); the `Spawn` approval
uses this path, not a `HelperEvent`. Legacy approval rows that predate the
submitter column fall back to the
root agent. Variants (`hive_sh4re::HelperEvent`): root agent. Variants (`hive_sh4re::HelperEvent`):
- `ApprovalResolved { id, agent, commit_ref, status, note }` - `ApprovalResolved { id, agent, commit_ref, status, note }`
fired by `actions::approve` + `actions::deny` whenever an fired by `actions::approve` + `actions::deny` whenever an
approval transitions to its terminal state. approval transitions to its terminal state.
- `Spawned { agent, ok, note }` — the `Spawn` approval DAG +
admin `HostRequest::Spawn`.
- `Rebuilt { agent, ok, note }``auto_update::rebuild_agent` - `Rebuilt { agent, ok, note }``auto_update::rebuild_agent`
(covers startup scan + manual `/rebuild` from dashboard) + (covers startup scan + manual `/rebuild` from dashboard) +
the `MergeConfigPr` deploy. the `MergeConfigPr` deploy.

View file

@ -37,6 +37,8 @@ hive-sh4re.workspace = true
hive-host-sock.workspace = true hive-host-sock.workspace = true
hive-jobq.workspace = true hive-jobq.workspace = true
hive-priv-sock.workspace = true hive-priv-sock.workspace = true
hive-agent-sock.workspace = true
hive-sock-client.workspace = true
hive-types.workspace = true hive-types.workspace = true
libc.workspace = true libc.workspace = true
listenfd = "1" listenfd = "1"

View file

@ -498,7 +498,7 @@ async fn run_approval_schedule_prompt(
.context("insert scheduled prompt") .context("insert scheduled prompt")
} }
.await; .await;
finish_approval(coord, &approval, result, None) finish_approval(coord, &approval, result, None).await
} }
/// Resolve an approval row from how its DAG's work ended — the body of the /// Resolve an approval row from how its DAG's work ended — the body of the
@ -565,7 +565,7 @@ pub(crate) async fn resolve_approval_dag(
} }
_ => {} _ => {}
} }
if let Err(e) = finish_approval(coord, &approval, result, terminal_tag) { if let Err(e) = finish_approval(coord, &approval, result, terminal_tag).await {
tracing::warn!(approval_id, error = ?e, "approval dag resolved with failure"); tracing::warn!(approval_id, error = ?e, "approval dag resolved with failure");
} }
} }
@ -682,10 +682,10 @@ async fn run_approval_init_config(
{ {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed"); tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed");
} }
finish_approval(coord, &approval, result, None) finish_approval(coord, &approval, result, None).await
} }
fn finish_approval( async fn finish_approval(
coord: &Coordinator, coord: &Coordinator,
approval: &hive_sh4re::Approval, approval: &hive_sh4re::Approval,
result: Result<()>, result: Result<()>,
@ -747,14 +747,26 @@ fn finish_approval(
); );
} }
} }
ApprovalKind::Spawn => coord.notify_submitter( ApprovalKind::Spawn => {
approval.id, let summary = if ok {
&HelperEvent::Spawned { format!("agent '{}' spawned", approval.agent)
agent: approval.agent.to_string(), } else {
ok, format!(
note, "agent '{}' spawn FAILED: {}",
}, approval.agent,
), note.as_deref().unwrap_or("unknown error")
)
};
coord
.push_todo_submitter(
approval.id,
"core",
Some(format!("spawned:{}", approval.agent)),
summary,
None,
)
.await;
}
// MergeConfigPr ends in a container rebuild — surface a Rebuilt // MergeConfigPr ends in a container rebuild — surface a Rebuilt
// lifecycle event. (It is never a first spawn — the agent already // lifecycle event. (It is never a first spawn — the agent already
// exists — so it never needs the Spawned arm above.) // exists — so it never needs the Spawned arm above.)

View file

@ -1355,12 +1355,91 @@ impl Coordinator {
/// failure — fall back to the root agent, preserving the prior /// failure — fall back to the root agent, preserving the prior
/// always-root behaviour. /// always-root behaviour.
pub fn notify_submitter(&self, approval_id: i64, event: &hive_sh4re::HelperEvent) { pub fn notify_submitter(&self, approval_id: i64, event: &hive_sh4re::HelperEvent) {
let target = self let target = self.submitter_or_manager(approval_id);
.approvals self.notify_agent(&target, event);
}
/// Shared resolution for "which agent should hear about approval
/// `approval_id`" — the authenticated socket caller at submit time,
/// falling back to the manager for legacy rows with no recorded
/// submitter (or any lookup failure). Used by both `notify_submitter`
/// and `push_todo_submitter` so the fallback rule lives in one place.
fn submitter_or_manager(&self, approval_id: i64) -> String {
self.approvals
.submitter_of(approval_id) .submitter_of(approval_id)
.unwrap_or_default() .unwrap_or_default()
.unwrap_or_else(|| hive_sh4re::MANAGER_AGENT.to_owned()); .unwrap_or_else(|| hive_sh4re::MANAGER_AGENT.to_owned())
self.notify_agent(&target, event); }
/// Push a todo directly into `agent`'s in-container todo store — a
/// best-effort *live* dial of its `hive-agent-sock` socket
/// (`hive_host_sock::agent_todo_socket`), same `UpsertTodo` request
/// shape the in-container producers (matrix/bash/forge-notify)
/// already send. This is the migration target for `HelperEvent`
/// variants that are pure "FYI, check when convenient" notices: the
/// event stops being a broker `Message` (which always drives an
/// immediate turn) and becomes a todo instead, with the same
/// dedup-by-key semantics as any other producer.
///
/// Deliberately push, not queue: if the agent's container is down
/// (socket file absent) or the dial otherwise fails, this is a
/// silent no-op (logged at `debug`/`warn`) — no retry, no fallback
/// delivery. An agent that's down doesn't need a todo about
/// something it'll never see appear this way; whatever mechanism
/// resurfaces its state on the next boot is unrelated to this path.
pub async fn push_todo(
&self,
agent: &str,
subsystem: &str,
key: Option<String>,
summary: String,
source: Option<String>,
) {
let Ok(ident) = hive_types::Ident::parse(agent) else {
tracing::warn!(%agent, "push_todo: not a valid agent ident, skipping");
return;
};
let path = hive_host_sock::agent_todo_socket(&ident);
if !path.exists() {
tracing::debug!(%agent, path = %path.display(), "push_todo: agent socket not present (offline?), skipping");
return;
}
let req = hive_agent_sock::Request::UpsertTodo {
subsystem: subsystem.to_owned(),
key,
summary,
source,
};
match hive_sock_client::request::<_, hive_agent_sock::Response>(
&path,
&req,
hive_sock_client::Retry::None,
)
.await
{
Ok(hive_agent_sock::Response::Err { message }) => {
tracing::warn!(%agent, %message, "push_todo: agent rejected the todo");
}
Err(e) => {
tracing::warn!(%agent, error = ?e, "push_todo: dial failed");
}
Ok(_) => {}
}
}
/// `push_todo` to whichever agent submitted approval `approval_id` —
/// same resolution `notify_submitter` uses.
pub async fn push_todo_submitter(
&self,
approval_id: i64,
subsystem: &str,
key: Option<String>,
summary: String,
source: Option<String>,
) {
let target = self.submitter_or_manager(approval_id);
self.push_todo(&target, subsystem, key, summary, source)
.await;
} }
/// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded /// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded

View file

@ -297,22 +297,30 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
// Bind the MCP listener now that the container is starting up. // Bind the MCP listener now that the container is starting up.
// The harness connects to this socket on its first turn. // The harness connects to this socket on its first turn.
coord.register_agent(name)?; coord.register_agent(name)?;
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned { coord
agent: name.to_owned(), .push_todo(
ok: true, hive_sh4re::MANAGER_AGENT,
note: None, "core",
}); Some(format!("spawned:{name}")),
format!("agent '{name}' spawned"),
None,
)
.await;
// Update tmpfiles.d so the new agent's dirs survive a reboot. // Update tmpfiles.d so the new agent's dirs survive a reboot.
tokio::spawn(lifecycle::sync_tmpfiles()); tokio::spawn(lifecycle::sync_tmpfiles());
} }
Err(e) => { Err(e) => {
// Spawn failed: register_agent was never called, so there is // Spawn failed: register_agent was never called, so there is
// nothing to unregister. Notify the manager and propagate. // nothing to unregister. Notify the manager and propagate.
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned { coord
agent: name.to_owned(), .push_todo(
ok: false, hive_sh4re::MANAGER_AGENT,
note: Some(format!("{e:#}")), "core",
}); Some(format!("spawned:{name}")),
format!("agent '{name}' spawn FAILED: {e:#}"),
None,
)
.await;
return Err(e); return Err(e);
} }
} }

View file

@ -72,6 +72,20 @@ pub fn agent_web_socket(name: &Ident) -> PathBuf {
.join("web.sock") .join("web.sock")
} }
/// Per-agent in-container todo socket path —
/// `AGENT_SOCKET_DIR/<name>/agent.sock`. Same socket the harness's
/// `HIVE_AGENT_SOCKET` binds and in-container producers (matrix/bash/
/// forge-notify) already dial — `hive-c0re` uses this to push a todo
/// directly into a *running* agent's local store (see
/// `Coordinator::push_todo`); the container being down (socket absent)
/// means the push is simply skipped, not queued.
#[must_use]
pub fn agent_todo_socket(name: &Ident) -> PathBuf {
PathBuf::from(AGENT_SOCKET_DIR)
.join(name.as_str())
.join("agent.sock")
}
/// nspawn machine-name prefix for agent containers (`h-<name>`). A single /// nspawn machine-name prefix for agent containers (`h-<name>`). A single
/// `starts_with(AGENT_PREFIX)` filter enumerates managed containers. /// `starts_with(AGENT_PREFIX)` filter enumerates managed containers.
pub const AGENT_PREFIX: &str = "h-"; pub const AGENT_PREFIX: &str = "h-";

View file

@ -507,13 +507,6 @@ pub enum HelperEvent {
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
tag: Option<String>, tag: Option<String>,
}, },
/// A new container was spawned. `ok = false` = spawn failed.
Spawned {
agent: String,
ok: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
note: Option<String>,
},
/// A container was rebuilt (auto-update or manual). /// A container was rebuilt (auto-update or manual).
Rebuilt { Rebuilt {
agent: String, agent: String,