diff --git a/docs/approvals.md b/docs/approvals.md index 8d4b4207..ccb8458f 100644 --- a/docs/approvals.md +++ b/docs/approvals.md @@ -203,22 +203,6 @@ One-shot rows fire once (if past due, on the next worker pass) and are deleted b `targets` is its own table (`scheduled_prompt_targets`) so partial cancellation flips a single row and the dashboard can show last-fired / last-result per recipient. Cancelling every target reaps the parent row on the next worker pass. -### Scheduled prompt delivery: todo, not a broker message - -An agent target's delivery is `push_todo` (`Coordinator::push_todo`, -`docs/coordinator.md` covers the mechanism generally), not a broker -`Message` — a scheduled prompt wakes its target with a todo instead of -driving an immediate turn, by design. `key = "schedule:"` per -target gives `push_todo`'s own upsert-by-key dedup the job a -now-removed `has_pending_with_body` broker check used to do: a re-fire -of the *same schedule* against a target that hasn't reviewed the last -one collapses into that one todo instead of stacking up. - -**`operator` is the one exception** — it's a valid schedule target but -has no in-container todo inbox, so it keeps the original broker -`Message` path (the dashboard mirrors `to == operator` into its own -pane, same as before). - ### Missing-target failure When a target name doesn't resolve to a known agent (container @@ -227,9 +211,7 @@ destroyed, operator typo, etc.) the worker: 1. Records `last_result = "no such agent: "` on the per-target row. 2. Sends a single advisory `Message` from `system` to `operator` - naming the schedule, target, and reason. This one stays a `Message` - regardless of target type — it's a to-operator advisory about a - broken schedule, not the schedule's own delivery. + naming the schedule, target, and reason. 3. Continues fanning out to the other live targets. Transient broker errors (sqlite lock contention, etc.) get the same diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 62c04aaa..602b3990 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -739,7 +739,7 @@ async fn finish_approval( match approval.kind { ApprovalKind::InitConfig => { if ok { - let _ = coord + coord .push_todo_submitter( approval.id, "core", @@ -763,7 +763,7 @@ async fn finish_approval( note.as_deref().unwrap_or("unknown error") ) }; - let _ = coord + coord .push_todo_submitter( approval.id, "core", @@ -784,7 +784,7 @@ async fn finish_approval( approval.fetched_sha.as_deref(), terminal_tag.as_deref(), ); - let _ = coord + coord .push_todo_submitter( approval.id, "core", @@ -963,7 +963,7 @@ pub async fn destroy(coord: &Arc, name: &str, purge: bool) -> Resul tracing::warn!(%name, error = ?e, "agent_power: remove on destroy failed"); } drop(guard); - let _ = coord + coord .push_todo( hive_sh4re::MANAGER_AGENT, "core", diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 4998e441..80686842 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -1387,14 +1387,6 @@ impl Coordinator { /// 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. - /// - /// Returns `Ok(())` on a successful push and `Err(reason)` — a - /// short human-readable string, already logged at `debug`/`warn` - /// by this function — on any of the silent-no-op cases below. - /// Most callers are pure fire-and-forget notices and ignore it - /// (`let _ = coord.push_todo(...).await;`); callers that track a - /// per-target delivery outcome (e.g. the scheduled-prompts worker's - /// `last_result` column) use it instead of assuming success. pub async fn push_todo( &self, agent: &str, @@ -1402,15 +1394,15 @@ impl Coordinator { key: Option, summary: String, source: Option, - ) -> Result<(), String> { + ) { let Ok(ident) = hive_types::Ident::parse(agent) else { tracing::warn!(%agent, "push_todo: not a valid agent ident, skipping"); - return Err(format!("'{agent}' is not a valid agent ident")); + 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 Err(format!("agent '{agent}' socket not present (offline?)")); + return; } let req = hive_agent_sock::Request::UpsertTodo { subsystem: subsystem.to_owned(), @@ -1427,13 +1419,11 @@ impl Coordinator { { Ok(hive_agent_sock::Response::Err { message }) => { tracing::warn!(%agent, %message, "push_todo: agent rejected the todo"); - Err(format!("agent '{agent}' rejected the todo: {message}")) } Err(e) => { tracing::warn!(%agent, error = ?e, "push_todo: dial failed"); - Err(format!("push_todo dial to '{agent}' failed: {e:#}")) } - Ok(_) => Ok(()), + Ok(_) => {} } } @@ -1446,10 +1436,10 @@ impl Coordinator { key: Option, summary: String, source: Option, - ) -> Result<(), String> { + ) { let target = self.submitter_or_manager(approval_id); self.push_todo(&target, subsystem, key, summary, source) - .await + .await; } /// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index f6dc3560..5208a43f 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -152,7 +152,7 @@ async fn run_emit_rebuilt(coord: &Arc, agent: &str, dag_id: Option< .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag))) .flatten(); let summary = crate::coordinator::rebuilt_todo_summary(agent, ok, note.as_deref(), None, None); - let _ = coord + coord .push_todo( hive_sh4re::MANAGER_AGENT, "core", @@ -429,7 +429,7 @@ async fn run_stop(coord: &Arc, name: &str) -> Result<()> { // own kind now. crate::lifecycle::kill(name).await?; coord.unregister_agent(name); - let _ = coord + coord .push_todo( hive_sh4re::MANAGER_AGENT, "core", diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index b9caf19f..826d59fc 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -297,7 +297,7 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result, name: &str) -> Result { // Spawn failed: register_agent was never called, so there is // nothing to unregister. Notify the manager and propagate. - let _ = coord + coord .push_todo( hive_sh4re::MANAGER_AGENT, "core", diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index 46446c12..0b7449ad 100644 --- a/hive-c0re/src/socket_server/lifecycle_handlers.rs +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -129,7 +129,7 @@ pub(super) async fn handle_kill(coord: &Arc, agent: &str, name: &st .await; match result { Ok(()) => { - let _ = coord + coord .push_todo( hive_sh4re::MANAGER_AGENT, "core", diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 9a84bfde..76e7dd51 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -354,6 +354,25 @@ impl Broker { Ok(u64::try_from(n.max(0)).unwrap_or(0)) } + /// Returns true when the recipient already has at least one + /// undelivered message from `sender` with exactly `body` in the + /// broker. Used by the scheduler to skip re-delivery of the same + /// scheduled prompt without blocking distinct schedules whose + /// bodies differ. + pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result { + let conn = self.conn.lock().unwrap(); + // An `ack_until`-closed pending row is dead — it must not + // suppress a fresh scheduled delivery of the same body. + let n: i64 = conn.query_row( + "SELECT COUNT(*) FROM messages + WHERE recipient = ?1 AND sender = ?2 AND body = ?3 + AND delivered_at IS NULL AND acked_at IS NULL", + params![recipient, sender, body], + |row| row.get(0), + )?; + Ok(n > 0) + } + /// Send a "your parent changed from X to Y" notification to `child`, /// coalescing with any existing undelivered one so that multiple moves /// while the agent is offline collapse into a single message spanning diff --git a/hive-c0re/src/workers/crash_watch.rs b/hive-c0re/src/workers/crash_watch.rs index 3cd704ee..df41815f 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/hive-c0re/src/workers/crash_watch.rs @@ -132,7 +132,7 @@ async fn emit_login_transitions( ) { for agent in current.difference(prev) { tracing::info!(%agent, "agent logged in"); - let _ = coord + coord .push_todo( hive_sh4re::MANAGER_AGENT, "core", @@ -163,7 +163,7 @@ async fn emit_login_transitions( .collect(); for agent in current_needs.difference(&prev_needs) { tracing::info!(%agent, "agent needs login"); - let _ = coord + coord .push_todo( hive_sh4re::MANAGER_AGENT, "core", diff --git a/hive-c0re/src/workers/scheduled_prompts_worker.rs b/hive-c0re/src/workers/scheduled_prompts_worker.rs index 1f1940a3..b7dac081 100644 --- a/hive-c0re/src/workers/scheduled_prompts_worker.rs +++ b/hive-c0re/src/workers/scheduled_prompts_worker.rs @@ -34,7 +34,7 @@ pub fn spawn(coord: Arc) { let mut shutdown = coord.shutdown_rx(); tokio::spawn(async move { loop { - tick(&coord).await; + tick(&coord); tokio::select! { () = tokio::time::sleep(POLL_INTERVAL) => {} _ = shutdown.changed() => { @@ -46,7 +46,7 @@ pub fn spawn(coord: Arc) { }); } -async fn tick(coord: &Arc) { +fn tick(coord: &Arc) { let now = Utc::now().timestamp(); let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) { Ok(rows) => rows, @@ -64,7 +64,7 @@ async fn tick(coord: &Arc) { return; } for schedule in due { - fire_schedule(coord, &schedule, now).await; + fire_schedule(coord, &schedule, now); } let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0); if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) { @@ -78,28 +78,16 @@ async fn tick(coord: &Arc) { /// Fan out one schedule's body to every active target. Records /// per-target `last_result`; advances or reaps the parent row at /// the end depending on whether `interval_seconds` is set. -/// -/// Delivery is `push_todo`, not a broker `Message` — a scheduled -/// prompt now wakes its target with a todo instead of driving an -/// immediate turn, by design: mara confirmed on the tracking issue -/// that this is the intended behavior, not an incidental side effect. -/// `key = "schedule:{id}"` per target gives `push_todo`'s own -/// upsert-by-key dedup the same job a now-removed -/// `has_pending_with_body` broker check used to do — collapsing a -/// re-fire of the *same schedule* against a target that hasn't -/// reviewed the last one yet — and does it more precisely (keyed on -/// schedule identity, not on the body happening to be byte-identical). -async fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { - let known: std::collections::HashSet = known_agents().await; +fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { + let known: std::collections::HashSet = known_agents(coord); for target_row in &schedule.targets { if target_row.cancelled_at_unix.is_some() { continue; } let target = &target_row.target; - // `operator` is a valid recipient (mara c4) — the operator has - // no in-container todo inbox, so it keeps the regular broker - // `Message` path; the dashboard mirrors `to == operator` into - // its own pane. + // `operator` is a valid recipient (mara c4) — operator + // delivery uses the regular broker path; the dashboard + // mirrors `to == operator` into its own pane. if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) { let reason = format!("no such agent: {target}"); if let Err(e) = @@ -112,18 +100,53 @@ async fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) notify_operator_missing_target(coord, schedule, target); continue; } - let result_str = match deliver_to_target(coord, schedule.id, target, &schedule.body).await { - Ok(()) => "ok".to_owned(), - Err(reason) => { - tracing::warn!( + // Skip delivery if there is already an unread message from + // "scheduled" waiting in this target's inbox. Prevents the same + // scheduled prompt from stacking up when an agent is slow or + // briefly offline, while still allowing distinct scheduled + // messages (different body) to enqueue independently. + match coord + .broker + .has_pending_with_body(target, "scheduled", &schedule.body) + { + Ok(true) => { + tracing::debug!( schedule = schedule.id, %target, - %reason, - "scheduled_prompts: delivery failed (will retry on next interval)" + "scheduled_prompts: skipping — same body already pending for target" ); - reason + let _ = coord.scheduled_prompts.record_target_result( + schedule.id, + target, + now, + "skipped: already pending", + ); + continue; } + Err(e) => { + tracing::warn!(schedule = schedule.id, %target, error = ?e, "has_pending_with_body failed"); + } + Ok(false) => {} + } + let msg = Message { + from: hive_sh4re::trusted_sender("scheduled"), + to: target.clone(), + body: schedule.body.clone(), + in_reply_to: None, }; + let result = coord.broker.send(&msg); + let result_str = match &result { + Ok(()) => "ok".to_owned(), + Err(e) => format!("broker send failed: {e:#}"), + }; + if let Err(e) = result { + tracing::warn!( + schedule = schedule.id, + %target, + error = ?e, + "scheduled_prompts: broker send failed (will retry on next interval)" + ); + } if let Err(e) = coord .scheduled_prompts @@ -153,44 +176,43 @@ async fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) } } -/// Deliver `body` to a single already-known-live `target` — a broker -/// `Message` when `target` is the operator (no in-container todo inbox -/// to push into there), a `push_todo` otherwise, keyed on the -/// schedule's own identity (`schedule:{schedule_id}`) so a re-fire -/// against a target that hasn't reviewed the last one collapses via -/// `push_todo`'s own upsert-by-key dedup. Shared by the periodic -/// `fire_schedule` tick and the manual `fire_now` dashboard action — -/// the only difference between them is what each caller does with the -/// `Result` (log/prefix and per-target `last_result`/`FireNowReport` -/// bookkeeping), not the delivery choice itself. -async fn deliver_to_target( - coord: &Arc, - schedule_id: i64, - target: &str, - body: &str, -) -> Result<(), String> { - if target == hive_sh4re::OPERATOR_RECIPIENT { - let msg = Message { - from: hive_sh4re::trusted_sender("scheduled"), - to: target.to_owned(), - body: body.to_owned(), - in_reply_to: None, - }; - coord - .broker - .send(&msg) - .map_err(|e| format!("broker send failed: {e:#}")) - } else { - coord - .push_todo( - target, - "schedule", - Some(format!("schedule:{schedule_id}")), - body.to_owned(), - Some("scheduled".to_owned()), - ) - .await +/// Snapshot of live container names for the missing-target check. +/// Always seeds the manager name (which is always reachable); +/// adds every live nspawn container that matches the `h-` prefix. +/// On `lifecycle::list` failure the set stays at just the manager +/// — fail-CLOSED, meaning every non-operator/non-manager target +/// looks missing this tick and gets the same treatment as a +/// genuinely-destroyed agent: operator advisory + per-target +/// `last_result` annotation + skipped delivery. Recurring +/// schedules recover automatically on the next tick (the lifecycle +/// listing usually works); one-shots that land on this window +/// lose their single delivery. Logged at `warn`, not propagated. +fn known_agents(_coord: &Coordinator) -> std::collections::HashSet { + // `lifecycle::list` is async; the worker tick is sync. Use the + // blocking variant via a small `tokio::runtime::Handle::block_on` + // wrapper. The worker runs in its own tokio task so this is + // safe (we're not in a `current_thread` runtime). + use std::collections::HashSet; + let mut out: HashSet = HashSet::new(); + // Manager is always a scheduled-prompt target (fail-safe: include + // it even if `list()` fails so prompts to the manager never silently drop). + out.insert(hive_sh4re::MANAGER_AGENT.to_owned()); + let containers = tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(crate::lifecycle::list()) + }); + match containers { + Ok(list) => { + for raw in list { + if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) { + out.insert(name.to_owned()); + } + } + } + Err(e) => { + tracing::warn!(error = ?e, "scheduled_prompts: container listing failed"); + } } + out } /// Send the operator a one-line advisory when a schedule fires @@ -278,7 +300,7 @@ pub async fn fire_now( { anyhow::bail!("schedule {schedule_id} has no active targets"); } - let known = known_agents().await; + let known = known_agents_async().await; let mut report = FireNowReport { ok: 0, failed: 0, @@ -304,22 +326,28 @@ pub async fn fire_now( report.missing += 1; continue; } - let result_str = match deliver_to_target(coord, schedule_id, target, &schedule.body).await { - Ok(()) => { - report.ok += 1; - "manual fire: ok".to_owned() - } - Err(reason) => { - report.failed += 1; - tracing::warn!( - schedule = schedule_id, - %target, - %reason, - "fire_now: delivery failed (no retry — manual fires don't loop)" - ); - format!("manual fire: {reason}") - } + let msg = Message { + from: hive_sh4re::trusted_sender("scheduled"), + to: target.clone(), + body: schedule.body.clone(), + in_reply_to: None, }; + let result = coord.broker.send(&msg); + let result_str = match &result { + Ok(()) => "manual fire: ok".to_owned(), + Err(e) => format!("manual fire: broker send failed: {e:#}"), + }; + if result.is_ok() { + report.ok += 1; + } else { + report.failed += 1; + tracing::warn!( + schedule = schedule_id, + %target, + error = ?result.as_ref().err(), + "fire_now: broker send failed (no retry — manual fires don't loop)" + ); + } if let Err(e) = coord .scheduled_prompts @@ -354,21 +382,12 @@ pub async fn fire_now( Ok(report) } -/// Snapshot of live container names for the missing-target check. -/// Always seeds the manager name (which is always reachable); -/// adds every live nspawn container that matches the `h-` prefix. -/// On `lifecycle::list` failure the set stays at just the manager -/// — fail-CLOSED, meaning every non-operator/non-manager target -/// looks missing this tick and gets the same treatment as a -/// genuinely-destroyed agent: operator advisory + per-target -/// `last_result` annotation + skipped delivery. Recurring -/// schedules recover automatically on the next tick (the lifecycle -/// listing usually works); one-shots that land on this window -/// lose their single delivery. Logged at `warn`, not propagated. -/// Shared by `fire_schedule` and `fire_now` — both are async now -/// (the sync `fire_schedule` used to need a `block_in_place` variant -/// of this before it started `push_todo`ing, which is itself async). -async fn known_agents() -> std::collections::HashSet { +/// Async variant of `known_agents` for `fire_now`. Same logic + +/// same fail-closed degradation; the difference is just that the +/// dashboard handler is genuinely async so we `await` the +/// `lifecycle::list` directly instead of going through the +/// `block_in_place` shim. +async fn known_agents_async() -> std::collections::HashSet { use std::collections::HashSet; let mut out: HashSet = HashSet::new(); out.insert(hive_sh4re::MANAGER_AGENT.to_owned());