From 68560215bdae5ad4a6070c8a6e6b4a334dbc7493 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 3 Aug 2026 00:37:43 +0200 Subject: [PATCH] hive-c0re: deliver scheduled prompts via push_todo, not a broker message --- docs/approvals.md | 20 +- hive-c0re/src/stores/broker.rs | 19 -- .../src/workers/scheduled_prompts_worker.rs | 209 ++++++++---------- 3 files changed, 114 insertions(+), 134 deletions(-) diff --git a/docs/approvals.md b/docs/approvals.md index ccb8458f..8d4b4207 100644 --- a/docs/approvals.md +++ b/docs/approvals.md @@ -203,6 +203,22 @@ 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 @@ -211,7 +227,9 @@ 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. + 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. 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/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 76e7dd51..9a84bfde 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -354,25 +354,6 @@ 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/scheduled_prompts_worker.rs b/hive-c0re/src/workers/scheduled_prompts_worker.rs index b7dac081..1f1940a3 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); + tick(&coord).await; tokio::select! { () = tokio::time::sleep(POLL_INTERVAL) => {} _ = shutdown.changed() => { @@ -46,7 +46,7 @@ pub fn spawn(coord: Arc) { }); } -fn tick(coord: &Arc) { +async 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 @@ fn tick(coord: &Arc) { return; } for schedule in due { - fire_schedule(coord, &schedule, now); + fire_schedule(coord, &schedule, now).await; } 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,16 +78,28 @@ 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. -fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { - let known: std::collections::HashSet = known_agents(coord); +/// +/// 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; 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) — operator - // delivery uses the regular broker path; the dashboard - // mirrors `to == operator` into its own pane. + // `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. if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) { let reason = format!("no such agent: {target}"); if let Err(e) = @@ -100,53 +112,18 @@ fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { notify_operator_missing_target(coord, schedule, target); continue; } - // 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!( + let result_str = match deliver_to_target(coord, schedule.id, target, &schedule.body).await { + Ok(()) => "ok".to_owned(), + Err(reason) => { + tracing::warn!( schedule = schedule.id, %target, - "scheduled_prompts: skipping — same body already pending for target" + %reason, + "scheduled_prompts: delivery failed (will retry on next interval)" ); - let _ = coord.scheduled_prompts.record_target_result( - schedule.id, - target, - now, - "skipped: already pending", - ); - continue; + reason } - 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 @@ -176,43 +153,44 @@ fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { } } -/// 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"); - } +/// 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 } - out } /// Send the operator a one-line advisory when a schedule fires @@ -300,7 +278,7 @@ pub async fn fire_now( { anyhow::bail!("schedule {schedule_id} has no active targets"); } - let known = known_agents_async().await; + let known = known_agents().await; let mut report = FireNowReport { ok: 0, failed: 0, @@ -326,28 +304,22 @@ pub async fn fire_now( report.missing += 1; continue; } - let msg = Message { - from: hive_sh4re::trusted_sender("scheduled"), - to: target.clone(), - body: schedule.body.clone(), - in_reply_to: None, + 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 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 @@ -382,12 +354,21 @@ pub async fn fire_now( Ok(report) } -/// 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 { +/// 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 { use std::collections::HashSet; let mut out: HashSet = HashSet::new(); out.insert(hive_sh4re::MANAGER_AGENT.to_owned());