diff --git a/Cargo.lock b/Cargo.lock index 399f5765..644146c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1632,11 +1632,13 @@ dependencies = [ "clap-markdown", "clap_complete", "forgejo-api", + "hive-agent-sock", "hive-core-agent-sock", "hive-host-sock", "hive-jobq", "hive-priv-sock", "hive-sh4re", + "hive-sock-client", "hive-types", "hmac 0.13.0", "indicatif", diff --git a/docs/approvals.md b/docs/approvals.md index 0dacc7c4..f0e9bd1d 100644 --- a/docs/approvals.md +++ b/docs/approvals.md @@ -570,15 +570,18 @@ Two more paths resolve a pending question with a sentinel answer: `Coordinator::notify_submitter(approval_id, &HelperEvent)` routes the event to the agent that originally submitted the approval (looked up from the `submitter` column on the `approvals` table). The harness delivers it -as a regular `system` inbox message so it drives a normal claude turn. -Legacy approval rows that predate the submitter column fall back to the +as a regular `system` inbox message so it drives a normal claude turn. A +"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`): - `ApprovalResolved { id, agent, commit_ref, status, note }` — fired by `actions::approve` + `actions::deny` whenever an 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` (covers startup scan + manual `/rebuild` from dashboard) + the `MergeConfigPr` deploy. diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index a4dae331..b87de707 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -37,6 +37,8 @@ hive-sh4re.workspace = true hive-host-sock.workspace = true hive-jobq.workspace = true hive-priv-sock.workspace = true +hive-agent-sock.workspace = true +hive-sock-client.workspace = true hive-types.workspace = true libc.workspace = true listenfd = "1" diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 207ecf95..c647d42d 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -498,7 +498,7 @@ async fn run_approval_schedule_prompt( .context("insert scheduled prompt") } .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 @@ -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"); } } @@ -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"); } - finish_approval(coord, &approval, result, None) + finish_approval(coord, &approval, result, None).await } -fn finish_approval( +async fn finish_approval( coord: &Coordinator, approval: &hive_sh4re::Approval, result: Result<()>, @@ -747,14 +747,26 @@ fn finish_approval( ); } } - ApprovalKind::Spawn => coord.notify_submitter( - approval.id, - &HelperEvent::Spawned { - agent: approval.agent.to_string(), - ok, - note, - }, - ), + ApprovalKind::Spawn => { + let summary = if ok { + format!("agent '{}' spawned", approval.agent) + } else { + format!( + "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 // lifecycle event. (It is never a first spawn — the agent already // exists — so it never needs the Spawned arm above.) diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 4b9e1366..67446c2a 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -1355,12 +1355,91 @@ impl Coordinator { /// failure — fall back to the root agent, preserving the prior /// always-root behaviour. pub fn notify_submitter(&self, approval_id: i64, event: &hive_sh4re::HelperEvent) { - let target = self - .approvals + let target = self.submitter_or_manager(approval_id); + 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) .unwrap_or_default() - .unwrap_or_else(|| hive_sh4re::MANAGER_AGENT.to_owned()); - self.notify_agent(&target, event); + .unwrap_or_else(|| hive_sh4re::MANAGER_AGENT.to_owned()) + } + + /// 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, + summary: String, + source: Option, + ) { + 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, + summary: String, + source: Option, + ) { + 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 diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index bec0cd75..826d59fc 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -297,22 +297,30 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result { // Spawn failed: register_agent was never called, so there is // nothing to unregister. Notify the manager and propagate. - coord.notify_manager(&hive_sh4re::HelperEvent::Spawned { - agent: name.to_owned(), - ok: false, - note: Some(format!("{e:#}")), - }); + coord + .push_todo( + hive_sh4re::MANAGER_AGENT, + "core", + Some(format!("spawned:{name}")), + format!("agent '{name}' spawn FAILED: {e:#}"), + None, + ) + .await; return Err(e); } } diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 5eb9419b..2363e3db 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -72,6 +72,20 @@ pub fn agent_web_socket(name: &Ident) -> PathBuf { .join("web.sock") } +/// Per-agent in-container todo socket path — +/// `AGENT_SOCKET_DIR//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-`). A single /// `starts_with(AGENT_PREFIX)` filter enumerates managed containers. pub const AGENT_PREFIX: &str = "h-"; diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 2d6a3301..8bcc7853 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -507,13 +507,6 @@ pub enum HelperEvent { #[serde(default, skip_serializing_if = "Option::is_none")] tag: Option, }, - /// A new container was spawned. `ok = false` = spawn failed. - Spawned { - agent: String, - ok: bool, - #[serde(default, skip_serializing_if = "Option::is_none")] - note: Option, - }, /// A container was rebuilt (auto-update or manual). Rebuilt { agent: String,