vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

hive-c0re: deliver scheduled prompts via push_todo, not a broker message

Browse source
This commit is contained in:
damocles 2026-08-03 00:37:43 +02:00 • committed by mara
commit 68560215bd
3 changed files with 114 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

20
docs/approvals.md
View file

@ -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. `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:<id>"` 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 ### Missing-target failure
When a target name doesn't resolve to a known agent (container 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: <name>"` on the 1. Records `last_result = "no such agent: <name>"` on the
per-target row. per-target row.
2. Sends a single advisory `Message` from `system` to `operator` 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. 3. Continues fanning out to the other live targets.
Transient broker errors (sqlite lock contention, etc.) get the same Transient broker errors (sqlite lock contention, etc.) get the same

19
hive-c0re/src/stores/broker.rs
View file

@ -354,25 +354,6 @@ impl Broker {
Ok(u64::try_from(n.max(0)).unwrap_or(0)) 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<bool> {
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`, /// Send a "your parent changed from X to Y" notification to `child`,
/// coalescing with any existing undelivered one so that multiple moves /// coalescing with any existing undelivered one so that multiple moves
/// while the agent is offline collapse into a single message spanning /// while the agent is offline collapse into a single message spanning

209
hive-c0re/src/workers/scheduled_prompts_worker.rs
View file

@ -34,7 +34,7 @@ pub fn spawn(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx(); let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move { tokio::spawn(async move {
loop { loop {
tick(&coord); tick(&coord).await;
tokio::select! { tokio::select! {
() = tokio::time::sleep(POLL_INTERVAL) => {} () = tokio::time::sleep(POLL_INTERVAL) => {}
_ = shutdown.changed() => { _ = shutdown.changed() => {
@ -46,7 +46,7 @@ pub fn spawn(coord: Arc<Coordinator>) {
}); });
} }
fn tick(coord: &Arc<Coordinator>) { async fn tick(coord: &Arc<Coordinator>) {
let now = Utc::now().timestamp(); let now = Utc::now().timestamp();
let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) { let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) {
Ok(rows) => rows, Ok(rows) => rows,
@ -64,7 +64,7 @@ fn tick(coord: &Arc<Coordinator>) {
return; return;
} }
for schedule in due { 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); let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) { if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
@ -78,16 +78,28 @@ fn tick(coord: &Arc<Coordinator>) {
/// Fan out one schedule's body to every active target. Records /// Fan out one schedule's body to every active target. Records
/// per-target `last_result`; advances or reaps the parent row at /// per-target `last_result`; advances or reaps the parent row at
/// the end depending on whether `interval_seconds` is set. /// the end depending on whether `interval_seconds` is set.
fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) { ///
let known: std::collections::HashSet<String> = 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<Coordinator>, schedule: &Schedule, now: i64) {
let known: std::collections::HashSet<String> = known_agents().await;
for target_row in &schedule.targets { for target_row in &schedule.targets {
if target_row.cancelled_at_unix.is_some() { if target_row.cancelled_at_unix.is_some() {
continue; continue;
} }
let target = &target_row.target; let target = &target_row.target;
// `operator` is a valid recipient (mara c4) — operator // `operator` is a valid recipient (mara c4) — the operator has
// delivery uses the regular broker path; the dashboard // no in-container todo inbox, so it keeps the regular broker
// mirrors `to == operator` into its own pane. // `Message` path; the dashboard mirrors `to == operator` into
// its own pane.
if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) { if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
let reason = format!("no such agent: {target}"); let reason = format!("no such agent: {target}");
if let Err(e) = if let Err(e) =
@ -100,53 +112,18 @@ fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
notify_operator_missing_target(coord, schedule, target); notify_operator_missing_target(coord, schedule, target);
continue; continue;
} }
// Skip delivery if there is already an unread message from let result_str = match deliver_to_target(coord, schedule.id, target, &schedule.body).await {
// "scheduled" waiting in this target's inbox. Prevents the same Ok(()) => "ok".to_owned(),
// scheduled prompt from stacking up when an agent is slow or Err(reason) => {
// briefly offline, while still allowing distinct scheduled tracing::warn!(
// messages (different body) to enqueue independently.
match coord
.broker
.has_pending_with_body(target, "scheduled", &schedule.body)
{
Ok(true) => {
tracing::debug!(
schedule = schedule.id, schedule = schedule.id,
%target, %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( reason
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) = if let Err(e) =
coord coord
.scheduled_prompts .scheduled_prompts
@ -176,43 +153,44 @@ fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
} }
} }
/// Snapshot of live container names for the missing-target check. /// Deliver `body` to a single already-known-live `target` — a broker
/// Always seeds the manager name (which is always reachable); /// `Message` when `target` is the operator (no in-container todo inbox
/// adds every live nspawn container that matches the `h-` prefix. /// to push into there), a `push_todo` otherwise, keyed on the
/// On `lifecycle::list` failure the set stays at just the manager /// schedule's own identity (`schedule:{schedule_id}`) so a re-fire
/// — fail-CLOSED, meaning every non-operator/non-manager target /// against a target that hasn't reviewed the last one collapses via
/// looks missing this tick and gets the same treatment as a /// `push_todo`'s own upsert-by-key dedup. Shared by the periodic
/// genuinely-destroyed agent: operator advisory + per-target /// `fire_schedule` tick and the manual `fire_now` dashboard action —
/// `last_result` annotation + skipped delivery. Recurring /// the only difference between them is what each caller does with the
/// schedules recover automatically on the next tick (the lifecycle /// `Result` (log/prefix and per-target `last_result`/`FireNowReport`
/// listing usually works); one-shots that land on this window /// bookkeeping), not the delivery choice itself.
/// lose their single delivery. Logged at `warn`, not propagated. async fn deliver_to_target(
fn known_agents(_coord: &Coordinator) -> std::collections::HashSet<String> { coord: &Arc<Coordinator>,
// `lifecycle::list` is async; the worker tick is sync. Use the schedule_id: i64,
// blocking variant via a small `tokio::runtime::Handle::block_on` target: &str,
// wrapper. The worker runs in its own tokio task so this is body: &str,
// safe (we're not in a `current_thread` runtime). ) -> Result<(), String> {
use std::collections::HashSet; if target == hive_sh4re::OPERATOR_RECIPIENT {
let mut out: HashSet<String> = HashSet::new(); let msg = Message {
// Manager is always a scheduled-prompt target (fail-safe: include from: hive_sh4re::trusted_sender("scheduled"),
// it even if `list()` fails so prompts to the manager never silently drop). to: target.to_owned(),
out.insert(hive_sh4re::MANAGER_AGENT.to_owned()); body: body.to_owned(),
let containers = tokio::task::block_in_place(|| { in_reply_to: None,
tokio::runtime::Handle::current().block_on(crate::lifecycle::list()) };
}); coord
match containers { .broker
Ok(list) => { .send(&msg)
for raw in list { .map_err(|e| format!("broker send failed: {e:#}"))
if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) { } else {
out.insert(name.to_owned()); coord
} .push_todo(
} target,
} "schedule",
Err(e) => { Some(format!("schedule:{schedule_id}")),
tracing::warn!(error = ?e, "scheduled_prompts: container listing failed"); body.to_owned(),
} Some("scheduled".to_owned()),
)
.await
} }
out
} }
/// Send the operator a one-line advisory when a schedule fires /// 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"); anyhow::bail!("schedule {schedule_id} has no active targets");
} }
let known = known_agents_async().await; let known = known_agents().await;
let mut report = FireNowReport { let mut report = FireNowReport {
ok: 0, ok: 0,
failed: 0, failed: 0,
@ -326,28 +304,22 @@ pub async fn fire_now(
report.missing += 1; report.missing += 1;
continue; continue;
} }
let msg = Message { let result_str = match deliver_to_target(coord, schedule_id, target, &schedule.body).await {
from: hive_sh4re::trusted_sender("scheduled"), Ok(()) => {
to: target.clone(), report.ok += 1;
body: schedule.body.clone(), "manual fire: ok".to_owned()
in_reply_to: None, }
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) = if let Err(e) =
coord coord
.scheduled_prompts .scheduled_prompts
@ -382,12 +354,21 @@ pub async fn fire_now(
Ok(report) Ok(report)
} }
/// Async variant of `known_agents` for `fire_now`. Same logic + /// Snapshot of live container names for the missing-target check.
/// same fail-closed degradation; the difference is just that the /// Always seeds the manager name (which is always reachable);
/// dashboard handler is genuinely async so we `await` the /// adds every live nspawn container that matches the `h-` prefix.
/// `lifecycle::list` directly instead of going through the /// On `lifecycle::list` failure the set stays at just the manager
/// `block_in_place` shim. /// — fail-CLOSED, meaning every non-operator/non-manager target
async fn known_agents_async() -> std::collections::HashSet<String> { /// 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<String> {
use std::collections::HashSet; use std::collections::HashSet;
let mut out: HashSet<String> = HashSet::new(); let mut out: HashSet<String> = HashSet::new();
out.insert(hive_sh4re::MANAGER_AGENT.to_owned()); out.insert(hive_sh4re::MANAGER_AGENT.to_owned());