vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:68560215bdae5ad4a6070c8a6e6b4a334dbc7493
Branches Tags
vinzenz:main
..
compare: vinzenz:b8627e7e9420e155336105e1f116af38fda1ed81
Branches Tags
vinzenz:main
9 changed files with 151 additions and 141 deletions
Download patch file Download diff file Expand all files Collapse all files

20
docs/approvals.md
View file

@ -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:<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
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: <name>"` 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

8
hive-c0re/src/actions.rs
View file

@ -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<Coordinator>, 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",

22
hive-c0re/src/coordinator.rs
View file

@ -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<String>,
summary: String,
source: Option<String>,
) -> 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<String>,
summary: String,
source: Option<String>,
) -> 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

4
hive-c0re/src/job_queue/exec.rs
View file

@ -152,7 +152,7 @@ async fn run_emit_rebuilt(coord: &Arc<Coordinator>, 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<Coordinator>, 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",

4
hive-c0re/src/server.rs
View file

@ -297,7 +297,7 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
// Bind the MCP listener now that the container is starting up.
// The harness connects to this socket on its first turn.
coord.register_agent(name)?;
let _ = coord
coord
.push_todo(
hive_sh4re::MANAGER_AGENT,
"core",
@ -312,7 +312,7 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
Err(e) => {
// 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",

2
hive-c0re/src/socket_server/lifecycle_handlers.rs
View file

@ -129,7 +129,7 @@ pub(super) async fn handle_kill(coord: &Arc<Coordinator>, agent: &str, name: &st
.await;
match result {
Ok(()) => {
let _ = coord
coord
.push_todo(
hive_sh4re::MANAGER_AGENT,
"core",

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

@ -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<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`,
/// coalescing with any existing undelivered one so that multiple moves
/// while the agent is offline collapse into a single message spanning

4
hive-c0re/src/workers/crash_watch.rs
View file

@ -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",

195
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();
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<Coordinator>) {
});
}
async fn tick(coord: &Arc<Coordinator>) {
fn tick(coord: &Arc<Coordinator>) {
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<Coordinator>) {
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<Coordinator>) {
/// 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<Coordinator>, schedule: &Schedule, now: i64) {
let known: std::collections::HashSet<String> = known_agents().await;
fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
let known: std::collections::HashSet<String> = 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<Coordinator>, 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 {
// 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,
"scheduled_prompts: skipping — same body already pending for target"
);
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(reason) => {
Err(e) => format!("broker send failed: {e:#}"),
};
if let Err(e) = result {
tracing::warn!(
schedule = schedule.id,
%target,
%reason,
"scheduled_prompts: delivery failed (will retry on next interval)"
error = ?e,
"scheduled_prompts: broker send failed (will retry on next interval)"
);
reason
}
};
if let Err(e) =
coord
.scheduled_prompts
@ -153,45 +176,44 @@ async fn fire_schedule(coord: &Arc<Coordinator>, 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<Coordinator>,
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<String> {
// `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<String> = 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
/// against an agent that no longer exists. Best-effort — failure
@ -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(()) => {
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;
"manual fire: ok".to_owned()
}
Err(reason) => {
} else {
report.failed += 1;
tracing::warn!(
schedule = schedule_id,
%target,
%reason,
"fire_now: delivery failed (no retry — manual fires don't loop)"
error = ?result.as_ref().err(),
"fire_now: broker send failed (no retry — manual fires don't loop)"
);
format!("manual fire: {reason}")
}
};
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<String> {
/// 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<String> {
use std::collections::HashSet;
let mut out: HashSet<String> = HashSet::new();
out.insert(hive_sh4re::MANAGER_AGENT.to_owned());