diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 1d545efd..fa0b965a 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -5,7 +5,7 @@ use std::sync::Arc; -use anyhow::{Context as _, Result, bail}; +use anyhow::{Result, bail}; use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent, MANAGER_AGENT}; use crate::coordinator::{Coordinator, TransientKind}; @@ -90,15 +90,6 @@ 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 - } } } @@ -136,39 +127,6 @@ 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 @@ -343,7 +301,6 @@ 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 @@ -393,9 +350,9 @@ fn finish_approval( sha: approval.fetched_sha.clone(), tag: terminal_tag, }), - // UpdateMetaInputs / SchedulePrompt: ApprovalResolved already - // carries the result. No separate lifecycle event needed. - ApprovalKind::UpdateMetaInputs | ApprovalKind::SchedulePrompt => {} + // UpdateMetaInputs: ApprovalResolved already carries the result. + // No separate lifecycle event needed. + ApprovalKind::UpdateMetaInputs => {} } result } @@ -727,7 +684,6 @@ 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 279a8eff..9afc7090 100644 --- a/hive-c0re/src/approvals.rs +++ b/hive-c0re/src/approvals.rs @@ -286,7 +286,6 @@ 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, @@ -329,7 +328,6 @@ fn kind_to_str(kind: ApprovalKind) -> &'static str { ApprovalKind::Spawn => "spawn", ApprovalKind::InitConfig => "init_config", ApprovalKind::UpdateMetaInputs => "update_meta_inputs", - ApprovalKind::SchedulePrompt => "schedule_prompt", } } @@ -339,7 +337,6 @@ 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 6a7f46c2..357e9b44 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -37,10 +37,6 @@ 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, @@ -196,15 +192,12 @@ 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 f4e322a6..4f28b01f 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -74,8 +74,6 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { .route("/request-spawn", post(post_request_spawn)) .route("/op-send", post(post_op_send)) .route("/meta-update", post(post_meta_update)) - .route("/api/schedules", get(api_schedules).post(post_schedule_new)) - .route("/api/schedules/{id}/cancel", post(post_schedule_cancel)) .route("/dashboard/stream", get(dashboard_stream)) .route("/dashboard/history", get(dashboard_history)) // Anything not matched by the dynamic routes above falls @@ -603,7 +601,6 @@ 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, @@ -664,15 +661,6 @@ 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 @@ -1391,84 +1379,6 @@ async fn api_reminders(State(state): State) -> Response { } } -/// `GET /api/schedules` — snapshot of every schedule for the -/// scheduled-prompts tab (#444). Returns the wire shape directly -/// so the frontend can render without an extra translation layer. -async fn api_schedules(State(state): State) -> Response { - match state.coord.scheduled_prompts.list() { - Ok(rows) => axum::Json( - rows.into_iter() - .map(crate::manager_server::schedule_to_wire_public) - .collect::>(), - ) - .into_response(), - Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")), - } -} - -/// `POST /api/schedules` — operator-direct schedule creation -/// (mara: "user can add them manually"). Accepts the same -/// `SchedulePromptPayload` shape as the manager request flow but -/// skips the approval gate — the operator click *is* the -/// approval. The schedule lands directly with -/// `source = Operator` and the worker picks it up at fire time. -async fn post_schedule_new( - State(state): State, - axum::Json(payload): axum::Json, -) -> Response { - if payload.targets.is_empty() { - return error_response("schedule must have at least one target"); - } - if payload.body.trim().is_empty() { - return error_response("schedule body must be non-empty"); - } - if let Some(0) = payload.interval_seconds { - return error_response("interval_seconds must be > 0 (use None for one-shot)"); - } - let new = crate::scheduled_prompts::NewSchedule { - owner: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), - 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::Operator, - }; - match state.coord.scheduled_prompts.submit(new) { - Ok(id) => axum::Json(serde_json::json!({"id": id})).into_response(), - Err(e) => error_response(&format!("schedule submit: {e:#}")), - } -} - -#[derive(serde::Deserialize, Default)] -struct CancelScheduleForm { - /// `None` / absent / empty array → cancel whole schedule. - #[serde(default)] - targets: Option>, -} - -/// `POST /api/schedules/{id}/cancel` — operator-side cancel -/// (whole schedule when no `targets` field, partial when one is -/// provided). Operator bypasses the topology check; the manager -/// surface enforces it for agent callers. -async fn post_schedule_cancel( - State(state): State, - AxumPath(id): AxumPath, - body: Option>, -) -> Response { - let targets = body - .and_then(|axum::Json(b)| b.targets) - .filter(|t| !t.is_empty()); - let result = match targets.as_deref() { - Some(list) => state.coord.scheduled_prompts.cancel_targets(id, list), - None => state.coord.scheduled_prompts.cancel_all(id), - }; - match result { - Ok(()) => (StatusCode::OK, "ok").into_response(), - Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")), - } -} - /// Same-origin proxy that fetches the named agent's /// `GET /api/state` and forwards only the `links` field to the /// dashboard JS (issue #262). Lets the agent backend stay the diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 3b9f0e77..1a92787a 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -21,8 +21,6 @@ mod stats_vacuum; mod flake_check; mod forge; mod lifecycle; -mod scheduled_prompts; -mod scheduled_prompts_worker; mod limits; mod loose_ends; mod manager_server; @@ -237,10 +235,6 @@ 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 38331d50..c7258650 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -344,20 +344,6 @@ 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, @@ -706,179 +692,6 @@ 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. -/// Public alias `schedule_to_wire_public` re-exports for -/// `dashboard.rs::api_schedules` without crossing the module -/// boundary into the manager-server file. -pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule { - schedule_to_wire(s) -} - -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.rs b/hive-c0re/src/scheduled_prompts.rs deleted file mode 100644 index 9ed90fc6..00000000 --- a/hive-c0re/src/scheduled_prompts.rs +++ /dev/null @@ -1,680 +0,0 @@ -//! Scheduled prompts (closes #444). Persistent sqlite queue of -//! `(fire_at, targets, body)` rows that the worker fans out as -//! broker `Message`s to each target's inbox at fire time. Recurring -//! schedules carry `interval_seconds` and re-arm `next_fire_at` on -//! delivery; one-shots are reaped. -//! -//! ## Three submit paths -//! -//! - **Operator-direct** (`source = Operator`): the operator adds -//! a schedule through the dashboard form. Lands in the table -//! immediately, no approval gate. -//! - **Agent-requested** (`source = Approval { id }`): a sub-agent -//! (or the manager) submits a `RequestSchedulePrompt` through the -//! manager socket. An `ApprovalKind::SchedulePrompt` row is -//! queued; on approve, hive-c0re inserts the schedule row with -//! `source = Approval { id: approval_id }` so the audit trail -//! points back at the operator decision. -//! - **No self-target shortcut**: even agent-self schedules need -//! approval. The existing `remind` MCP tool stays the quick -//! self-wake path; this module is the bigger, multi-recipient, -//! operator-visible thing. -//! -//! ## Catch-up clamp (missed-while-down) -//! -//! When hive-c0re comes back from being down, the worker sees rows -//! whose `next_fire_at` is well in the past. For recurring rows -//! that would mean firing N delayed pulses in a row — spammy and -//! useless. Instead the worker fires ONCE per row and bumps -//! `next_fire_at` to the next interval slot ≥ `now`, recording how -//! many cycles were skipped in `last_result`. Operators see "fired -//! late, caught up from 17 skipped" instead of 17 wake-up storms. -//! -//! ## Per-target state -//! -//! `targets` is its own table so partial cancellation flips a -//! single row + so the dashboard can show last-fired / last-result -//! per recipient. Cancelling every target reaps the parent row on -//! the next worker pass. - -use std::path::Path; -use std::sync::Mutex; - -use anyhow::{Context, Result, bail}; -use rusqlite::{Connection, OptionalExtension, params}; -use serde::{Deserialize, Serialize}; - -const SCHEMA: &str = r" -CREATE TABLE IF NOT EXISTS scheduled_prompts ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - owner TEXT NOT NULL, - body TEXT NOT NULL, - interval_seconds INTEGER, - next_fire_at_unix INTEGER NOT NULL, - created_at_unix INTEGER NOT NULL, - source TEXT NOT NULL, - cancelled_at_unix INTEGER, - description TEXT -); -CREATE INDEX IF NOT EXISTS idx_scheduled_due - ON scheduled_prompts (next_fire_at_unix) - WHERE cancelled_at_unix IS NULL; - -CREATE TABLE IF NOT EXISTS scheduled_prompt_targets ( - schedule_id INTEGER NOT NULL, - target TEXT NOT NULL, - cancelled_at_unix INTEGER, - last_fired_at_unix INTEGER, - last_result TEXT, - PRIMARY KEY (schedule_id, target), - FOREIGN KEY (schedule_id) REFERENCES scheduled_prompts(id) ON DELETE CASCADE -); -"; - -/// One scheduled-prompt row + its current target set. Returned by -/// `list` / `get`; the per-target last-result blob is suitable for -/// rendering on the dashboard without a second query. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Schedule { - pub id: i64, - /// `"operator"` or an agent name — drives cancel-permission - /// checks. For approval-sourced rows this is the requesting - /// agent (the `Approval.agent` at submit time). - pub owner: String, - pub body: String, - /// `None` = one-shot, deleted after first fire. - /// `Some(n)` = recurring every `n` seconds. - pub interval_seconds: Option, - pub next_fire_at_unix: i64, - pub created_at_unix: i64, - pub source: ScheduleSource, - /// Set when the *entire* schedule was cancelled (all targets - /// flipped, or operator cancel-all). Worker reaps these on the - /// next pass. - pub cancelled_at_unix: Option, - pub description: Option, - pub targets: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ScheduleTarget { - pub target: String, - pub cancelled_at_unix: Option, - pub last_fired_at_unix: Option, - pub last_result: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum ScheduleSource { - Operator, - Approval { id: i64 }, -} - -impl ScheduleSource { - fn to_db_string(&self) -> String { - match self { - ScheduleSource::Operator => "operator".to_owned(), - ScheduleSource::Approval { id } => format!("approval:{id}"), - } - } - - fn from_db_string(s: &str) -> Self { - if let Some(rest) = s.strip_prefix("approval:") - && let Ok(id) = rest.parse::() - { - return ScheduleSource::Approval { id }; - } - // Unknown / "operator" / corrupt → operator (best the row - // can do without an unparseable-source variant). - ScheduleSource::Operator - } -} - -/// Submission payload — everything the caller knows at insert time. -/// Used by both the operator-direct path and the approval-flow -/// path; the latter sets `source = Approval { id }`. -#[derive(Debug, Clone)] -pub struct NewSchedule { - pub owner: String, - pub targets: Vec, - pub body: String, - pub first_fire_at_unix: i64, - pub interval_seconds: Option, - pub description: Option, - pub source: ScheduleSource, -} - -pub struct ScheduledPrompts { - conn: Mutex, -} - -impl ScheduledPrompts { - pub fn open(path: &Path) -> Result { - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent).with_context(|| { - format!("create scheduled_prompts db parent {}", parent.display()) - })?; - } - let conn = Connection::open(path) - .with_context(|| format!("open scheduled_prompts db {}", path.display()))?; - // Required for ON DELETE CASCADE to actually fire — sqlite - // ships with FKs disabled per connection by default. - conn.execute_batch("PRAGMA foreign_keys = ON;") - .context("enable foreign keys")?; - conn.execute_batch(SCHEMA) - .context("apply scheduled_prompts schema")?; - Ok(Self { - conn: Mutex::new(conn), - }) - } - - /// Insert a new schedule. Returns the new id. Empty `targets` is - /// rejected — a schedule with no recipients would silently - /// never fan out, masking caller bugs. - pub fn submit(&self, new: NewSchedule) -> Result { - if new.targets.is_empty() { - bail!("schedule must have at least one target"); - } - let mut conn = self.conn.lock().unwrap(); - let tx = conn.transaction()?; - tx.execute( - "INSERT INTO scheduled_prompts - (owner, body, interval_seconds, next_fire_at_unix, - created_at_unix, source, description) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", - params![ - new.owner, - new.body, - new.interval_seconds.map(i64::try_from).and_then(Result::ok), - new.first_fire_at_unix, - now_unix(), - new.source.to_db_string(), - new.description, - ], - )?; - let id = tx.last_insert_rowid(); - for target in &new.targets { - tx.execute( - "INSERT INTO scheduled_prompt_targets (schedule_id, target) - VALUES (?1, ?2)", - params![id, target], - )?; - } - tx.commit()?; - Ok(id) - } - - /// Fetch a single schedule by id (with its target rows). - /// `Ok(None)` for a non-existent / already-reaped id. - pub fn get(&self, id: i64) -> Result> { - let conn = self.conn.lock().unwrap(); - let row = conn - .query_row( - "SELECT id, owner, body, interval_seconds, next_fire_at_unix, - created_at_unix, source, cancelled_at_unix, description - FROM scheduled_prompts WHERE id = ?1", - params![id], - row_to_schedule_header, - ) - .optional()?; - let Some(mut s) = row else { - return Ok(None); - }; - s.targets = load_targets(&conn, id)?; - Ok(Some(s)) - } - - /// Every active (non-globally-cancelled) schedule in insert - /// order. Used by the dashboard list view + the cancel-auth - /// check (the latter only needs the header but list() is the - /// shared hot path). - pub fn list(&self) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT id, owner, body, interval_seconds, next_fire_at_unix, - created_at_unix, source, cancelled_at_unix, description - FROM scheduled_prompts - ORDER BY next_fire_at_unix ASC, id ASC", - )?; - let rows = stmt.query_map([], row_to_schedule_header)?; - let mut out = Vec::new(); - for row in rows { - let mut s = row?; - s.targets = load_targets(&conn, s.id)?; - out.push(s); - } - Ok(out) - } - - /// Pop the set of active rows that are due (`next_fire_at_unix <= now`) - /// up to `limit`. Read-only — the worker calls `mark_fired` - /// after each successful fan-out so the rows reappear on the - /// next tick when re-armed. - pub fn due(&self, now: i64, limit: u64) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT id, owner, body, interval_seconds, next_fire_at_unix, - created_at_unix, source, cancelled_at_unix, description - FROM scheduled_prompts - WHERE cancelled_at_unix IS NULL AND next_fire_at_unix <= ?1 - ORDER BY next_fire_at_unix ASC, id ASC - LIMIT ?2", - )?; - let rows = stmt.query_map(params![now, limit], row_to_schedule_header)?; - let mut out = Vec::new(); - for row in rows { - let mut s = row?; - s.targets = load_targets(&conn, s.id)?; - out.push(s); - } - Ok(out) - } - - /// Record a per-target fan-out result. `result` is "ok" or a - /// short error string; surfaces on the dashboard last-result - /// column. No-op for a target row that's already cancelled. - pub fn record_target_result( - &self, - schedule_id: i64, - target: &str, - fired_at_unix: i64, - result: &str, - ) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE scheduled_prompt_targets - SET last_fired_at_unix = ?1, last_result = ?2 - WHERE schedule_id = ?3 AND target = ?4 AND cancelled_at_unix IS NULL", - params![fired_at_unix, result, schedule_id, target], - )?; - Ok(()) - } - - /// Advance a recurring schedule's `next_fire_at` to the smallest - /// multiple-of-interval > `from`. Returns the count of skipped - /// cycles (≥ 0); the worker stamps that into the per-row - /// last_result so operators see "caught up from N missed". - /// - /// For one-shots (`interval_seconds IS NULL`) this is a no-op - /// at the SQL level; callers should `delete` them after fan-out - /// instead. - pub fn rearm(&self, id: i64, from_unix: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let row: Option<(Option, i64)> = conn - .query_row( - "SELECT interval_seconds, next_fire_at_unix - FROM scheduled_prompts WHERE id = ?1", - params![id], - |row| Ok((row.get(0)?, row.get(1)?)), - ) - .optional()?; - let Some((interval, current_next)) = row else { - return Ok(0); - }; - let Some(interval) = interval.filter(|&i| i > 0) else { - return Ok(0); - }; - // Smallest multiple-of-interval > `from_unix`. Catch-up - // semantics: when from_unix > current_next, every step in - // between gets counted as a skipped cycle. - let mut next = current_next + interval; - let mut skipped: u64 = 0; - while next <= from_unix { - next += interval; - skipped += 1; - } - conn.execute( - "UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2", - params![next, id], - )?; - Ok(skipped) - } - - /// Delete a one-shot row after its single fire. Cascades the - /// target rows via the FK constraint. Idempotent on a missing - /// id. - pub fn delete(&self, id: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "DELETE FROM scheduled_prompts WHERE id = ?1", - params![id], - )?; - Ok(()) - } - - /// Cancel the entire schedule (all targets flipped + the parent - /// row marked cancelled). Idempotent; safe to call on an - /// already-cancelled row. - pub fn cancel_all(&self, id: i64) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); - let now = now_unix(); - let tx = conn.transaction()?; - tx.execute( - "UPDATE scheduled_prompts - SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) - WHERE id = ?2", - params![now, id], - )?; - tx.execute( - "UPDATE scheduled_prompt_targets - SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) - WHERE schedule_id = ?2", - params![now, id], - )?; - tx.commit()?; - Ok(()) - } - - /// Cancel a subset of a schedule's targets. When the last - /// active target is cancelled, the parent row is auto-cancelled - /// too (so the worker reaps it). Unknown targets in the list - /// are silently skipped. - pub fn cancel_targets(&self, id: i64, targets: &[String]) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); - let now = now_unix(); - let tx = conn.transaction()?; - for target in targets { - tx.execute( - "UPDATE scheduled_prompt_targets - SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) - WHERE schedule_id = ?2 AND target = ?3", - params![now, id, target], - )?; - } - // If every target on this schedule is now cancelled, flip - // the parent so the worker stops scanning it. - let active_targets: i64 = tx.query_row( - "SELECT COUNT(*) FROM scheduled_prompt_targets - WHERE schedule_id = ?1 AND cancelled_at_unix IS NULL", - params![id], - |row| row.get(0), - )?; - if active_targets == 0 { - tx.execute( - "UPDATE scheduled_prompts - SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) - WHERE id = ?2", - params![now, id], - )?; - } - tx.commit()?; - Ok(()) - } - - /// Reap cancelled rows older than `older_than_unix`. Returns - /// the number of rows deleted. Called from the worker on each - /// tick so cancellations clear out of the dashboard without an - /// extra periodic vacuum task. - pub fn reap_cancelled(&self, older_than_unix: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let n = conn.execute( - "DELETE FROM scheduled_prompts - WHERE cancelled_at_unix IS NOT NULL - AND cancelled_at_unix <= ?1", - params![older_than_unix], - )?; - Ok(n) - } -} - -fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result { - let interval: Option = row.get(3)?; - let source_str: String = row.get(6)?; - Ok(Schedule { - id: row.get(0)?, - owner: row.get(1)?, - body: row.get(2)?, - interval_seconds: interval.and_then(|i| u64::try_from(i).ok()), - next_fire_at_unix: row.get(4)?, - created_at_unix: row.get(5)?, - source: ScheduleSource::from_db_string(&source_str), - cancelled_at_unix: row.get(7)?, - description: row.get(8)?, - targets: Vec::new(), - }) -} - -fn load_targets(conn: &Connection, schedule_id: i64) -> Result> { - let mut stmt = conn.prepare( - "SELECT target, cancelled_at_unix, last_fired_at_unix, last_result - FROM scheduled_prompt_targets - WHERE schedule_id = ?1 - ORDER BY target ASC", - )?; - let rows = stmt.query_map(params![schedule_id], |row| { - Ok(ScheduleTarget { - target: row.get(0)?, - cancelled_at_unix: row.get(1)?, - last_fired_at_unix: row.get(2)?, - last_result: row.get(3)?, - }) - })?; - let mut out = Vec::new(); - for row in rows { - out.push(row?); - } - Ok(out) -} - -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) -} - -#[cfg(test)] -mod tests { - use super::*; - use tempfile::TempDir; - - fn open() -> (TempDir, ScheduledPrompts) { - let dir = TempDir::new().expect("tempdir"); - let path = dir.path().join("schedules.sqlite"); - let db = ScheduledPrompts::open(&path).expect("open"); - (dir, db) - } - - fn submit_one_shot(db: &ScheduledPrompts, fire_at: i64, targets: &[&str]) -> i64 { - db.submit(NewSchedule { - owner: "operator".into(), - targets: targets.iter().map(|t| (*t).to_owned()).collect(), - body: "wake".into(), - first_fire_at_unix: fire_at, - interval_seconds: None, - description: None, - source: ScheduleSource::Operator, - }) - .expect("submit") - } - - #[test] - fn submit_and_get_round_trips_targets() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice", "bob"]); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.id, id); - assert_eq!(s.targets.len(), 2); - assert_eq!(s.targets[0].target, "alice"); - assert_eq!(s.targets[1].target, "bob"); - assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_none())); - } - - #[test] - fn submit_rejects_empty_targets() { - let (_dir, db) = open(); - let err = db - .submit(NewSchedule { - owner: "operator".into(), - targets: Vec::new(), - body: "wake".into(), - first_fire_at_unix: 100, - interval_seconds: None, - description: None, - source: ScheduleSource::Operator, - }) - .unwrap_err(); - assert!(format!("{err:#}").contains("at least one target")); - } - - #[test] - fn due_returns_only_past_active_rows() { - let (_dir, db) = open(); - let _future = submit_one_shot(&db, 1000, &["alice"]); - let past = submit_one_shot(&db, 50, &["alice"]); - let cancelled_past = submit_one_shot(&db, 50, &["bob"]); - db.cancel_all(cancelled_past).expect("cancel"); - let due = db.due(100, 10).expect("due"); - assert_eq!(due.len(), 1); - assert_eq!(due[0].id, past); - } - - #[test] - fn rearm_handles_catch_up() { - let (_dir, db) = open(); - // Recurring every 60s, last fire at t=100. - let id = db - .submit(NewSchedule { - owner: "operator".into(), - targets: vec!["alice".into()], - body: "wake".into(), - first_fire_at_unix: 100, - interval_seconds: Some(60), - description: None, - source: ScheduleSource::Operator, - }) - .expect("submit"); - // Worker comes back at t=400 — 5 missed cycles (160, 220, - // 280, 340, 400) → next should be 460, skipped = 5. - let skipped = db.rearm(id, 400).expect("rearm"); - assert_eq!(skipped, 5); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 460); - } - - #[test] - fn rearm_advances_one_step_when_caught_up() { - let (_dir, db) = open(); - let id = db - .submit(NewSchedule { - owner: "operator".into(), - targets: vec!["alice".into()], - body: "wake".into(), - first_fire_at_unix: 100, - interval_seconds: Some(60), - description: None, - source: ScheduleSource::Operator, - }) - .expect("submit"); - // Worker fires right at t=100 — single advance to t=160. - let skipped = db.rearm(id, 100).expect("rearm"); - assert_eq!(skipped, 0); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 160); - } - - #[test] - fn rearm_is_a_no_op_for_one_shots() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - let skipped = db.rearm(id, 1000).expect("rearm"); - assert_eq!(skipped, 0); - // next_fire_at unchanged — one-shots are reaped via delete(). - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 100); - } - - #[test] - fn cancel_targets_auto_cancels_parent_when_last_drops() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice", "bob"]); - db.cancel_targets(id, &["alice".to_owned()]).expect("cancel"); - let s = db.get(id).expect("get").expect("present"); - // Parent still active (bob remains). - assert!(s.cancelled_at_unix.is_none()); - db.cancel_targets(id, &["bob".to_owned()]).expect("cancel"); - let s = db.get(id).expect("get").expect("present"); - // Parent auto-cancels once every target is gone. - assert!(s.cancelled_at_unix.is_some()); - assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some())); - } - - #[test] - fn cancel_all_flips_parent_and_targets() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice", "bob"]); - db.cancel_all(id).expect("cancel"); - let s = db.get(id).expect("get").expect("present"); - assert!(s.cancelled_at_unix.is_some()); - assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some())); - } - - #[test] - fn delete_cascades_targets() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice", "bob"]); - db.delete(id).expect("delete"); - assert!(db.get(id).expect("get").is_none()); - } - - #[test] - fn record_target_result_skips_cancelled_targets() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice", "bob"]); - db.cancel_targets(id, &["alice".to_owned()]).expect("cancel"); - db.record_target_result(id, "alice", 200, "ok") - .expect("record alice"); - db.record_target_result(id, "bob", 200, "ok") - .expect("record bob"); - let s = db.get(id).expect("get").expect("present"); - let alice = s.targets.iter().find(|t| t.target == "alice").unwrap(); - let bob = s.targets.iter().find(|t| t.target == "bob").unwrap(); - // Cancelled targets do NOT get last-result writes. - assert!(alice.last_fired_at_unix.is_none()); - assert!(alice.last_result.is_none()); - assert_eq!(bob.last_fired_at_unix, Some(200)); - assert_eq!(bob.last_result.as_deref(), Some("ok")); - } - - #[test] - fn approval_source_round_trips() { - let (_dir, db) = open(); - let id = db - .submit(NewSchedule { - owner: "manager".into(), - targets: vec!["alice".into()], - body: "wake".into(), - first_fire_at_unix: 100, - interval_seconds: None, - description: None, - source: ScheduleSource::Approval { id: 42 }, - }) - .expect("submit"); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.source, ScheduleSource::Approval { id: 42 }); - } - - #[test] - fn reap_cancelled_removes_old_rows_only() { - let (_dir, db) = open(); - let stale = submit_one_shot(&db, 100, &["alice"]); - let recent = submit_one_shot(&db, 100, &["bob"]); - db.cancel_all(stale).expect("cancel stale"); - // Manually backdate the stale row. - { - let conn = db.conn.lock().unwrap(); - conn.execute( - "UPDATE scheduled_prompts SET cancelled_at_unix = ?1 WHERE id = ?2", - params![10_i64, stale], - ) - .expect("backdate"); - } - db.cancel_all(recent).expect("cancel recent"); - let n = db.reap_cancelled(100).expect("reap"); - assert_eq!(n, 1); - assert!(db.get(stale).expect("get stale").is_none()); - assert!(db.get(recent).expect("get recent").is_some()); - } -} diff --git a/hive-c0re/src/scheduled_prompts_worker.rs b/hive-c0re/src/scheduled_prompts_worker.rs deleted file mode 100644 index 8880e9ea..00000000 --- a/hive-c0re/src/scheduled_prompts_worker.rs +++ /dev/null @@ -1,247 +0,0 @@ -//! 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.) get -//! the per-target `last_result` annotated AND a `tracing::warn`, -//! but the post-fire bookkeeping treats the row the same way it -//! does on a clean fire: -//! - **recurring** rows re-arm — the next interval slot tries -//! the broker send again, so transient errors self-heal. -//! - **one-shots** delete unconditionally after their single -//! fan-out pass; a broker failure on a one-shot is NOT -//! retried (the operator advisory + last_result are the only -//! audit trail). - -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. -/// 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(); - 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 527f2cdb..66874491 100644 --- a/hive-c0re/src/topology.rs +++ b/hive-c0re/src/topology.rs @@ -62,32 +62,6 @@ 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 b29003ac..32170968 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -110,14 +110,6 @@ 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)] @@ -918,93 +910,6 @@ 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)] @@ -1039,12 +944,6 @@ 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.