From aa7d8d9c9a7daaf575d24bec3e8b5835e700b62d Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 26 May 2026 01:11:36 +0200 Subject: [PATCH] c0re: schedule_prompt approval kind + worker + manager surface (#444 step 2) --- hive-c0re/src/actions.rs | 52 ++++- hive-c0re/src/approvals.rs | 3 + hive-c0re/src/coordinator.rs | 7 + hive-c0re/src/dashboard.rs | 10 + hive-c0re/src/main.rs | 5 + hive-c0re/src/manager_server.rs | 180 +++++++++++++++++ hive-c0re/src/scheduled_prompts_worker.rs | 232 ++++++++++++++++++++++ hive-c0re/src/topology.rs | 26 +++ hive-sh4re/src/lib.rs | 101 ++++++++++ 9 files changed, 612 insertions(+), 4 deletions(-) create mode 100644 hive-c0re/src/scheduled_prompts_worker.rs diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index fa0b965a..1d545efd 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -5,7 +5,7 @@ use std::sync::Arc; -use anyhow::{Result, bail}; +use anyhow::{Context as _, Result, bail}; use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent, MANAGER_AGENT}; use crate::coordinator::{Coordinator, TransientKind}; @@ -90,6 +90,15 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { coord.emit_rebuild_queue_snapshot(); Ok(()) } + ApprovalKind::SchedulePrompt => { + // No queue card for SchedulePrompt — the work is a single + // sqlite insert, the actual "running" lifetime lives on + // the scheduled-prompts surface itself (worker fires it + // at the scheduled time). Run inline + fire + // `ApprovalResolved` so the approval row leaves Pending + // immediately. + run_approval_schedule_prompt(&coord, approval).await + } } } @@ -127,6 +136,39 @@ pub async fn run_approval_apply_commit( finish_approval(coord, &approval, result, terminal_tag, is_first_spawn) } +/// Inline (non-queued) handler for `ApprovalKind::SchedulePrompt`. +/// On approve, decode the `SchedulePromptPayload` JSON from the +/// approval's `commit_ref`, insert a row into `scheduled_prompts` +/// (with `source = Approval { id }`), and fire `ApprovalResolved`. +/// The worker takes over from here — fan-out at fire time. +async fn run_approval_schedule_prompt( + coord: &Coordinator, + approval: hive_sh4re::Approval, +) -> Result<()> { + let result: Result<()> = async { + let payload: hive_sh4re::SchedulePromptPayload = + serde_json::from_str(&approval.commit_ref) + .context("decode SchedulePromptPayload from approval.commit_ref")?; + coord + .scheduled_prompts + .submit(crate::scheduled_prompts::NewSchedule { + owner: approval.agent.clone(), + targets: payload.targets, + body: payload.body, + first_fire_at_unix: payload.first_fire_at_unix, + interval_seconds: payload.interval_seconds, + description: payload.description, + source: crate::scheduled_prompts::ScheduleSource::Approval { + id: approval.id, + }, + }) + .map(|_| ()) + .context("insert scheduled prompt") + } + .await; + finish_approval(coord, &approval, result, None, false) +} + /// Worker entry point for `ApprovalKind::UpdateMetaInputs` queue /// entries. Inputs come from the approval row's `commit_ref` field /// (JSON-encoded by the manager submit path), not the queue entry's @@ -301,6 +343,7 @@ fn finish_approval( ApprovalKind::ApplyCommit => "apply_commit", ApprovalKind::InitConfig => "init_config", ApprovalKind::UpdateMetaInputs => "update_meta_inputs", + ApprovalKind::SchedulePrompt => "schedule_prompt", }; let sha_short = approval .fetched_sha @@ -350,9 +393,9 @@ fn finish_approval( sha: approval.fetched_sha.clone(), tag: terminal_tag, }), - // UpdateMetaInputs: ApprovalResolved already carries the result. - // No separate lifecycle event needed. - ApprovalKind::UpdateMetaInputs => {} + // UpdateMetaInputs / SchedulePrompt: ApprovalResolved already + // carries the result. No separate lifecycle event needed. + ApprovalKind::UpdateMetaInputs | ApprovalKind::SchedulePrompt => {} } result } @@ -684,6 +727,7 @@ pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<() ApprovalKind::ApplyCommit => "apply_commit", ApprovalKind::InitConfig => "init_config", ApprovalKind::UpdateMetaInputs => "update_meta_inputs", + ApprovalKind::SchedulePrompt => "schedule_prompt", }; let sha_short = sha.as_deref().map(|s| s[..s.len().min(12)].to_owned()); let description = a.description.clone(); diff --git a/hive-c0re/src/approvals.rs b/hive-c0re/src/approvals.rs index 9afc7090..279a8eff 100644 --- a/hive-c0re/src/approvals.rs +++ b/hive-c0re/src/approvals.rs @@ -286,6 +286,7 @@ fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result { "spawn" => ApprovalKind::Spawn, "init_config" => ApprovalKind::InitConfig, "update_meta_inputs" => ApprovalKind::UpdateMetaInputs, + "schedule_prompt" => ApprovalKind::SchedulePrompt, other => { return Err(rusqlite::Error::FromSqlConversionFailure( 2, @@ -328,6 +329,7 @@ fn kind_to_str(kind: ApprovalKind) -> &'static str { ApprovalKind::Spawn => "spawn", ApprovalKind::InitConfig => "init_config", ApprovalKind::UpdateMetaInputs => "update_meta_inputs", + ApprovalKind::SchedulePrompt => "schedule_prompt", } } @@ -337,6 +339,7 @@ fn kind_from_str(s: &str) -> Result { "spawn" => ApprovalKind::Spawn, "init_config" => ApprovalKind::InitConfig, "update_meta_inputs" => ApprovalKind::UpdateMetaInputs, + "schedule_prompt" => ApprovalKind::SchedulePrompt, other => bail!("unknown approval kind '{other}'"), }) } diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 357e9b44..6a7f46c2 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -37,6 +37,10 @@ pub struct Coordinator { pub broker: Arc, pub approvals: Arc, pub questions: Arc, + /// Scheduled-prompts queue (#444). One sqlite connection, + /// internal mutex; the worker drains due rows and the manager + /// handlers insert / cancel through the same handle. + pub scheduled_prompts: Arc, /// URL of the hyperhive flake (no fragment). Inlined into per-agent /// `flake.nix` files as `inputs.hyperhive.url`. pub hyperhive_flake: String, @@ -192,12 +196,15 @@ impl Coordinator { let broker = Broker::open(db_path).context("open broker")?; let approvals = Approvals::open(db_path).context("open approvals")?; let questions = OperatorQuestions::open(db_path).context("open operator_questions")?; + let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path) + .context("open scheduled_prompts")?; let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL); let (shutdown_tx, _) = watch::channel(false); Ok(Self { broker: Arc::new(broker), approvals: Arc::new(approvals), questions: Arc::new(questions), + scheduled_prompts: Arc::new(scheduled_prompts), hyperhive_flake, dashboard_port, operator_pronouns, diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 4f28b01f..b51f8750 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -601,6 +601,7 @@ fn history_view(a: Approval) -> ApprovalHistoryView { hive_sh4re::ApprovalKind::Spawn => "spawn", hive_sh4re::ApprovalKind::InitConfig => "init_config", hive_sh4re::ApprovalKind::UpdateMetaInputs => "update_meta_inputs", + hive_sh4re::ApprovalKind::SchedulePrompt => "schedule_prompt", }; ApprovalHistoryView { id: a.id, @@ -661,6 +662,15 @@ async fn build_approval_views(approvals: Vec) -> Vec { description: a.description, requested_at: a.requested_at, }, + hive_sh4re::ApprovalKind::SchedulePrompt => ApprovalView { + id: a.id, + agent: a.agent, + kind: "schedule_prompt", + sha_short: None, + diff: None, + description: a.description, + requested_at: a.requested_at, + }, }); } out diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 44384520..3b9f0e77 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -22,6 +22,7 @@ mod flake_check; mod forge; mod lifecycle; mod scheduled_prompts; +mod scheduled_prompts_worker; mod limits; mod loose_ends; mod manager_server; @@ -236,6 +237,10 @@ async fn cmd_serve( // Reminder scheduler: drains due reminders + handles // file_path payload persistence. See reminder_scheduler.rs. reminder_scheduler::spawn(coord.clone()); + // Scheduled-prompts worker: drains due scheduled_prompts rows + // and fans the body out to each active target's inbox. See + // scheduled_prompts_worker.rs (#444). + scheduled_prompts_worker::spawn(coord.clone()); // Rebuild-queue worker: drains the global rebuild/meta-update/ // spawn queue FIFO so hive-c0re never runs two heavyweight // container ops concurrently. Existing rebuild call sites diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index c7258650..2da1f79c 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -344,6 +344,20 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp ); ManagerResponse::Ok } + ManagerRequest::RequestSchedulePrompt(payload) => { + handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload).await + } + ManagerRequest::CancelSchedule { id, targets } => { + handle_cancel_schedule(coord, hive_sh4re::MANAGER_AGENT, *id, targets.as_deref()) + } + ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() { + Ok(schedules) => ManagerResponse::Schedules { + schedules: schedules.into_iter().map(schedule_to_wire).collect(), + }, + Err(e) => ManagerResponse::Err { + message: format!("list scheduled prompts: {e:#}"), + }, + }, ManagerRequest::Ask { question, options, @@ -692,6 +706,172 @@ async fn submit_apply_commit( Ok((id, sha)) } +/// Submit a `RequestSchedulePrompt` payload as an `ApprovalKind::SchedulePrompt` +/// row. Encodes the payload into the approval's `commit_ref` so the +/// approve handler can re-parse it without a side table. Validates +/// inputs (non-empty targets, non-empty body, sane interval) at +/// submit time — the operator should never see a malformed schedule +/// pending approval. +async fn handle_request_schedule_prompt( + coord: &Arc, + requester: &str, + payload: &hive_sh4re::SchedulePromptPayload, +) -> ManagerResponse { + if payload.targets.is_empty() { + return ManagerResponse::Err { + message: "schedule must have at least one target".into(), + }; + } + if payload.body.trim().is_empty() { + return ManagerResponse::Err { + message: "schedule body must be non-empty".into(), + }; + } + if let Some(0) = payload.interval_seconds { + return ManagerResponse::Err { + message: "interval_seconds must be > 0 (use None for one-shot)".into(), + }; + } + let commit_ref = match serde_json::to_string(payload) { + Ok(s) => s, + Err(e) => { + return ManagerResponse::Err { + message: format!("encode SchedulePromptPayload: {e:#}"), + } + } + }; + let id = match coord.approvals.submit_kind( + requester, + hive_sh4re::ApprovalKind::SchedulePrompt, + &commit_ref, + payload.description.as_deref(), + ) { + Ok(id) => id, + Err(e) => { + return ManagerResponse::Err { + message: format!("queue schedule_prompt approval: {e:#}"), + } + } + }; + tracing::info!( + %id, + requester, + targets = ?payload.targets, + first_fire_at = payload.first_fire_at_unix, + interval = ?payload.interval_seconds, + "schedule_prompt approval queued" + ); + coord.emit_approval_added( + id, + requester, + "schedule_prompt", + None, + None, + payload.description.clone(), + ); + ManagerResponse::Ok +} + +/// Cancel a schedule (whole or per-target). Manager-surface +/// authorization: a manager can cancel its own schedules + any +/// schedule whose owner is one of its sub-agents (topology-walked). +/// The operator surface bypasses this and can cancel anything; +/// agents reaching this path through the manager get the +/// topology-scoped check. +fn handle_cancel_schedule( + coord: &Arc, + requester: &str, + schedule_id: i64, + targets: Option<&[String]>, +) -> ManagerResponse { + let schedule = match coord.scheduled_prompts.get(schedule_id) { + Ok(Some(s)) => s, + Ok(None) => { + return ManagerResponse::Err { + message: format!("schedule {schedule_id} not found"), + } + } + Err(e) => { + return ManagerResponse::Err { + message: format!("read schedule {schedule_id}: {e:#}"), + } + } + }; + if !cancel_authorized(requester, &schedule.owner) { + return ManagerResponse::Err { + message: format!( + "not authorized: {requester} cannot cancel schedule owned by {owner}", + owner = schedule.owner + ), + }; + } + let result = match targets { + Some(list) if !list.is_empty() => coord + .scheduled_prompts + .cancel_targets(schedule_id, list) + .map_err(|e| format!("cancel targets: {e:#}")), + _ => coord + .scheduled_prompts + .cancel_all(schedule_id) + .map_err(|e| format!("cancel all: {e:#}")), + }; + match result { + Ok(()) => ManagerResponse::Ok, + Err(message) => ManagerResponse::Err { message }, + } +} + +/// Permission check for `CancelSchedule` on the manager surface. +/// `requester` (always `hm1nd` here) can cancel its own schedules. +/// Sub-agent ownership is delegated to topology — see +/// `crate::topology::is_descendant_of`. +fn cancel_authorized(requester: &str, owner: &str) -> bool { + if requester == owner { + return true; + } + if requester == hive_sh4re::OPERATOR_RECIPIENT { + return true; + } + // Manager can cancel anything owned by an agent in its subtree. + // For the current single-manager topology that covers everything, + // but the check stays correct as the tree grows. + crate::topology::is_descendant_of(owner, requester) +} + +/// Map a `scheduled_prompts::Schedule` to its public wire shape. +/// Field-by-field copy — the two types are intentionally identical; +/// the separation keeps hive-sh4re free of hive-c0re-internal types. +fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule { + hive_sh4re::WireSchedule { + id: s.id, + owner: s.owner, + body: s.body, + interval_seconds: s.interval_seconds, + next_fire_at_unix: s.next_fire_at_unix, + created_at_unix: s.created_at_unix, + source: match s.source { + crate::scheduled_prompts::ScheduleSource::Operator => { + hive_sh4re::WireScheduleSource::Operator + } + crate::scheduled_prompts::ScheduleSource::Approval { id } => { + hive_sh4re::WireScheduleSource::Approval { id } + } + }, + cancelled_at_unix: s.cancelled_at_unix, + description: s.description, + targets: s + .targets + .into_iter() + .map(|t| hive_sh4re::WireScheduleTarget { + target: t.target, + cancelled_at_unix: t.cancelled_at_unix, + last_fired_at_unix: t.last_fired_at_unix, + last_result: t.last_result, + }) + .collect(), + } +} + /// On `Ask { ttl_seconds: Some(n) }`, sleep n seconds and then try to /// resolve the question with `[expired]`. If the operator (or any /// other path) already answered it, `answer()` returns Err and we diff --git a/hive-c0re/src/scheduled_prompts_worker.rs b/hive-c0re/src/scheduled_prompts_worker.rs new file mode 100644 index 00000000..4377e8d1 --- /dev/null +++ b/hive-c0re/src/scheduled_prompts_worker.rs @@ -0,0 +1,232 @@ +//! Background loop that drains due `scheduled_prompts` rows +//! (#444) and fans the body out as inbox `Message`s to each +//! active target. Mirrors `reminder_scheduler::spawn` shape: +//! single `spawn(coord)` entry, 5s poll cadence, shutdown-aware. +//! +//! ## Catch-up semantics +//! +//! When hive-c0re comes back from being down, a recurring row +//! whose `next_fire_at` is well in the past would otherwise fire +//! N delayed pulses in a row. Instead we fire ONCE and let +//! `ScheduledPrompts::rearm` bump `next_fire_at` to the next +//! interval slot ≥ `now`, recording the skipped-cycle count in +//! the per-target `last_result` so operators see how many +//! firings were caught up rather than losing the signal. +//! +//! ## Missing-target failure +//! +//! When a target name doesn't resolve to a known agent (the +//! container has been destroyed, the operator typo'd a name, +//! etc.) the worker: +//! 1. records `last_result = "no such agent: "` against +//! the per-target row, +//! 2. sends a single advisory `Message` from `system` to +//! `operator` describing the schedule + target + reason, +//! 3. continues fanning out to the other (live) targets. +//! +//! Transient broker errors (sqlite lock contention, etc.) are +//! logged and retried on the next tick. + +use std::sync::Arc; +use std::time::Duration; + +use hive_sh4re::Message; + +use crate::coordinator::Coordinator; +use crate::scheduled_prompts::Schedule; + +/// Per-tick cap. Each schedule fires once per tick at most; +/// 100/tick × 5s tick = sustained throughput cap of ~20/sec, +/// matching `reminder_scheduler::REMINDER_BATCH_LIMIT`. Bump +/// together if real-world rates push past this. +const SCHEDULE_BATCH_LIMIT: u64 = 100; + +/// Poll interval. Same 5s as the reminder scheduler — picking +/// up freshly-due rows within at most one tick keeps the +/// dashboard's "next fire in ..." countdown honest without +/// burning CPU on empty sweeps. +const POLL_INTERVAL: Duration = Duration::from_secs(5); + +/// Reap cancelled schedules older than this from the table so +/// the dashboard list view doesn't accrue tombstones forever. +/// Cancelled rows live long enough that the operator can still +/// see what they cancelled in the recent past. +const CANCELLED_REAP_AGE: Duration = Duration::from_secs(3600); + +pub fn spawn(coord: Arc) { + let mut shutdown = coord.shutdown_rx(); + tokio::spawn(async move { + loop { + tick(&coord); + tokio::select! { + () = tokio::time::sleep(POLL_INTERVAL) => {} + _ = shutdown.changed() => { + tracing::info!("scheduled_prompts worker: shutdown signal received"); + break; + } + } + } + }); +} + +fn tick(coord: &Arc) { + let now = now_unix(); + let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) { + Ok(rows) => rows, + Err(e) => { + tracing::warn!(error = ?e, "scheduled_prompts: query due rows failed"); + return; + } + }; + if due.is_empty() { + // Periodic reaper still gets a chance even on empty ticks. + let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0); + if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) { + tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed"); + } + return; + } + for schedule in due { + 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) { + tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed"); + } +} + +/// 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); + 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. + if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) { + let reason = format!("no such agent: {target}"); + if let Err(e) = coord.scheduled_prompts.record_target_result( + schedule.id, + target, + now, + &reason, + ) { + tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed"); + } + notify_operator_missing_target(coord, schedule, target); + continue; + } + let msg = Message { + from: "scheduled".to_owned(), + 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 + .record_target_result(schedule.id, target, now, &result_str) + { + tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed"); + } + } + // Advance or reap. One-shots delete; recurring re-arm with + // catch-up clamp. + if schedule.interval_seconds.is_some() { + match coord.scheduled_prompts.rearm(schedule.id, now) { + Ok(0) => {} + Ok(skipped) => { + tracing::info!( + schedule = schedule.id, + skipped, + "scheduled_prompts: caught up missed cycles" + ); + } + Err(e) => { + tracing::warn!(error = ?e, schedule = schedule.id, "rearm failed"); + } + } + } else if let Err(e) = coord.scheduled_prompts.delete(schedule.id) { + tracing::warn!(error = ?e, schedule = schedule.id, "delete one-shot failed"); + } +} + +/// Snapshot of live container names for the missing-target check. +/// Returns an empty set on lifecycle errors — we fail-open then +/// (every target gets through) and the broker rejects unknown +/// recipients downstream. +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(); + 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()); + } else if raw == crate::lifecycle::MANAGER_NAME { + out.insert(hive_sh4re::MANAGER_AGENT.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 +/// to send just gets logged; the schedule continues firing. +fn notify_operator_missing_target(coord: &Coordinator, schedule: &Schedule, target: &str) { + let body = format!( + "scheduled prompt #{id} fired but target `{target}` is not a live agent. \ + body was:\n\n{body}", + id = schedule.id, + target = target, + body = schedule.body + ); + let msg = Message { + from: "scheduled".to_owned(), + to: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), + body, + in_reply_to: None, + }; + if let Err(e) = coord.broker.send(&msg) { + tracing::warn!(error = ?e, schedule = schedule.id, %target, "operator advisory send failed"); + } +} + +fn now_unix() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) + .unwrap_or(0) +} diff --git a/hive-c0re/src/topology.rs b/hive-c0re/src/topology.rs index 66874491..527f2cdb 100644 --- a/hive-c0re/src/topology.rs +++ b/hive-c0re/src/topology.rs @@ -62,6 +62,32 @@ pub fn parent_of(name: &str) -> Option { read().get(name).cloned().flatten() } +/// True when `candidate` is `ancestor` or any descendant of +/// `ancestor` per the current topology. Walks parents from +/// `candidate` upward; the walk terminates at root or on a cycle +/// (cycle defence: bounded to 32 hops, more than any plausible +/// hive depth). Used by the cancel-authorization check in +/// `manager_server::handle_cancel_schedule` to enforce +/// "managers can cancel anything their subtree owns." +#[must_use] +pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool { + if candidate == ancestor { + return true; + } + let topo = read(); + let mut cur = candidate.to_owned(); + for _ in 0..32 { + let Some(parent) = topo.get(&cur).cloned().flatten() else { + return false; + }; + if parent == ancestor { + return true; + } + cur = parent; + } + false +} + /// Persist the topology map. Sorted JSON output (BTreeMap is sorted by /// key) keeps git diffs minimal across re-writes. Best-effort — /// returns an `io::Error` so callers can decide whether a failure diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 32170968..b29003ac 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -110,6 +110,14 @@ pub enum ApprovalKind { /// JSON-encoded inputs array (`"[]"` = all inputs). Agent field is /// set to `hm1nd` (the requesting manager). UpdateMetaInputs, + /// Add a scheduled prompt (closes #444). On approval hive-c0re + /// inserts a row into `scheduled_prompts` with + /// `source = Approval { id }`; the worker fans the body out as + /// inbox messages to each target at the scheduled time, recurring + /// when `interval_seconds` is set. `commit_ref` stores the + /// JSON-encoded `SchedulePromptPayload` so the approval row carries + /// the full submission verbatim (target list, body, schedule). + SchedulePrompt, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -910,6 +918,93 @@ pub enum ManagerRequest { #[serde(default, skip_serializing_if = "Option::is_none")] description: Option, }, + /// Queue an approval to add a scheduled prompt (#444). The + /// requester (caller of this request) is recorded as the schedule + /// owner; on operator approval hive-c0re inserts the schedule and + /// the worker fans the body out at fire time. Even agent-self + /// schedules go through approval — the existing `remind` MCP tool + /// is the unapproved self-wake path. + RequestSchedulePrompt(SchedulePromptPayload), + /// Cancel a scheduled prompt (#444). `targets = None` cancels the + /// whole schedule; `Some(list)` cancels just those recipients, + /// auto-cancelling the parent when no active targets remain. + /// Authorization: manager can cancel its own schedules + any + /// sub-agent schedules (i.e. owner reachable via topology); the + /// operator surface bypasses this check. + CancelSchedule { + id: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + targets: Option>, + }, + /// List every schedule in the queue. Manager-side this is + /// unfiltered — the dashboard does the topology-filter for the + /// per-agent view. + ListSchedules, +} + +/// Submission payload for `RequestSchedulePrompt`. Lives outside the +/// enum so it can also serialize into the approval row's `commit_ref` +/// (the dispatcher re-parses it on approve and inserts the schedule). +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct SchedulePromptPayload { + /// Names of recipient agents. Operator + `hm1nd` allowed. + pub targets: Vec, + /// Message body delivered to each target's inbox at fire time. + /// Same size budget as `Send.body` — soft cap at the broker level. + pub body: String, + /// Absolute unix timestamp (seconds) for the FIRST fire. For + /// recurring schedules the worker then re-arms in + /// `interval_seconds` steps. + pub first_fire_at_unix: i64, + /// `None` = one-shot. `Some(n > 0)` = recurring every `n` seconds. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub interval_seconds: Option, + /// Optional description shown on the dashboard approval card AND + /// stored on the resulting schedule row for the operator's + /// "what is this?" reference later. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, +} + +/// Schedule row shape on the wire — mirror of +/// `scheduled_prompts::Schedule` but in the public crate so dashboard +/// + agent surfaces can deserialize without depending on +/// hive-c0re-internal types. Kept structurally identical to the +/// in-process type; the conversion is field-by-field in +/// `manager_server` / `dashboard`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WireSchedule { + pub id: i64, + pub owner: String, + pub body: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub interval_seconds: Option, + pub next_fire_at_unix: i64, + pub created_at_unix: i64, + pub source: WireScheduleSource, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cancelled_at_unix: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + pub targets: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum WireScheduleSource { + Operator, + Approval { id: i64 }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WireScheduleTarget { + pub target: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cancelled_at_unix: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_fired_at_unix: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_result: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -944,6 +1039,12 @@ pub enum ManagerResponse { Logs { content: String, }, + /// `ListSchedules` result (#444). Snapshot of every schedule + /// (active + cancelled-but-not-yet-reaped); the dashboard does the + /// per-agent topology filter on top. + Schedules { + schedules: Vec, + }, /// `GetLooseEnds` result: hive-wide loose ends (approvals + /// unanswered questions). Same `LooseEnd` variants as the /// agent surface; the manager's view is unfiltered.