diff --git a/hive-agent-mcp/README.md b/hive-agent-mcp/README.md index 3f6f086a..f65388bb 100644 --- a/hive-agent-mcp/README.md +++ b/hive-agent-mcp/README.md @@ -9,8 +9,8 @@ would hit. HTTP is the sole transport; there is no stdio mode here. ## When to use it This is where the core hyperhive tool surface lives: `send`, `recv`, -`remind`, `get_loose_ends`, `set_status`, -`get_agent_meta`, scheduling, and the approval-request tools. Reach +`remind`, `get_loose_ends`, `set_status` +and `get_agent_meta`. Reach for this crate when you're adding or changing a built-in tool rather than an `extraMcpServers` add-on — those are separate stdio bridges (see `hive-bash-mcp`, `hive-matrix-mcp`) that dial the harness socket diff --git a/hive-agent-mcp/src/mcp/args.rs b/hive-agent-mcp/src/mcp/args.rs index de4e892e..3a7270ad 100644 --- a/hive-agent-mcp/src/mcp/args.rs +++ b/hive-agent-mcp/src/mcp/args.rs @@ -94,7 +94,7 @@ pub struct CompactArgs { } // ----------------------------------------------------------------------------- -// Privileged tool arg types (lifecycle, approvals, scheduling) +// Privileged tool arg types (lifecycle, approvals) // ----------------------------------------------------------------------------- #[derive(Debug, serde::Deserialize, schemars::JsonSchema)] @@ -123,89 +123,3 @@ pub struct CancelLooseEndArgs { /// Row id from the matching `get_loose_ends` entry. pub id: i64, } - -#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct RequestSchedulePromptArgs { - /// Recipient agents — one schedule fires to many inboxes at the - /// scheduled time. `operator` is not a valid target — she isn't a - /// container with a todo inbox to push a wake into, and this is - /// for fanning a prompt out to agents, not scheduling something - /// at her. - pub targets: Vec, - /// Message body delivered to each target's inbox at fire time. - /// Same size budget as `send` bodies. - pub body: String, - /// Absolute unix timestamp (seconds) for the FIRST fire. For - /// recurring schedules the worker re-arms in - /// `interval_seconds` steps from this point on. - pub first_fire_at_unix: i64, - /// `None` / absent = one-shot. `Some(n > 0)` = recurring every - /// `n` seconds. The worker clamps catch-up so a long downtime - /// fires ONCE on resume (skipped-cycle count surfaces in the - /// per-target `last_result`), not N delayed pulses in a row. - #[serde(default)] - pub interval_seconds: Option, - /// Optional description shown on the dashboard approval card + - /// preserved on the schedule row for later operator reference. - #[serde(default)] - pub description: Option, -} - -#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct FireScheduleNowArgs { - /// Schedule id to fire out of band. Get this from a prior - /// `list_schedules` call or the approval-resolved event for - /// the originating `request_schedule_prompt`. - pub id: i64, -} - -#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct CancelScheduleArgs { - /// Schedule id from a prior `list_schedules` call or the - /// approval-resolved event for a `request_schedule_prompt`. - pub id: i64, - /// Optional target list. `None` / empty = cancel the entire - /// schedule. `Some(["alice", "bob"])` = cancel just those - /// recipients (the schedule keeps firing for any remaining - /// active targets, and auto-cancels its parent row when every - /// target is gone). - #[serde(default)] - pub targets: Option>, -} - -#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct EditScheduleArgs { - /// Schedule id from a prior `list_schedules` call or the - /// approval-resolved event for a `request_schedule_prompt`. - pub id: i64, - /// New body text. Omit to keep the existing one. - #[serde(default)] - pub body: Option, - /// New description. Omit to keep the existing one. (To CLEAR - /// the description, use the dashboard PATCH endpoint - /// directly — the agent surface intentionally keeps the args - /// flat / non-nullable to dodge the doubly-wrapped Option - /// schemars quirk; clearing fields is rare and operator-side.) - #[serde(default)] - pub description: Option, - /// Recurring interval in seconds. Omit to keep the existing - /// cadence; pass an explicit value to set a new one. Toggling - /// recurring↔one-shot (clearing the interval) is operator-only - /// for the same reason as `description` above. - #[serde(default)] - pub interval_seconds: Option, - /// New absolute unix timestamp for the next fire. Omit to - /// leave the schedule on its current cadence. - #[serde(default)] - pub next_fire_at_unix: Option, - /// Names of new targets to add. Replace-on-conflict: re-adding - /// a previously cancelled target resets its history (operator - /// intent on re-add = "this target is active again"). - #[serde(default)] - pub targets_add: Option>, - /// Names of targets to cancel. Tombstones preserve per-target - /// audit; when no active targets remain the schedule - /// auto-cancels. - #[serde(default)] - pub targets_remove: Option>, -} diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 497c3506..11a966e9 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -24,9 +24,8 @@ mod args; mod render; pub use args::{ - AckUntilArgs, CancelLooseEndArgs, CancelScheduleArgs, CompactArgs, EditScheduleArgs, - FireScheduleNowArgs, GetAgentMetaArgs, MarkTodosDoneArgs, RecvArgs, RemindArgs, - RequestSchedulePromptArgs, SendArgs, SetStatusArgs, + AckUntilArgs, CancelLooseEndArgs, CompactArgs, GetAgentMetaArgs, MarkTodosDoneArgs, RecvArgs, + RemindArgs, SendArgs, SetStatusArgs, }; pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; @@ -519,159 +518,6 @@ impl AgentServer { }) .await } - - #[tool( - description = "Queue an approval to add a scheduled prompt — one body delivered to \ - N agent inboxes at a target time, optionally recurring every `interval_seconds`. \ - The operator approves; on approve hive-c0re inserts the schedule and the worker \ - fans it out. Even self-targeted schedules go through this flow (the operator pays \ - for the wake-up tokens); the existing `remind` MCP tool stays the quick \ - no-approval self-wake path. \n\n\ - Catch-up clamp: if hive-c0re is down across multiple intervals, only ONE delayed \ - fire happens on resume (per recurring schedule). The skipped-cycle count surfaces \ - in the per-target `last_result` for the operator's audit trail. \n\n\ - Per-target failure: a target name that doesn't resolve to a live agent → operator \ - gets a one-line advisory `Message` from `system`; the schedule keeps firing for \ - the other (live) targets." - )] - async fn request_schedule_prompt( - &self, - Parameters(args): Parameters, - ) -> String { - let log = format!("{args:?}"); - run_tool_envelope("request_schedule_prompt", log, async move { - let target_count = args.targets.len(); - let (resp, retries) = self - .dispatch(hive_core_agent_sock::Request::RequestSchedulePrompt( - hive_sh4re::manager::SchedulePromptPayload { - targets: args.targets, - body: args.body, - first_fire_at_unix: args.first_fire_at_unix, - interval_seconds: args.interval_seconds, - description: args.description, - }, - )) - .await; - annotate_retries( - format_ack( - resp, - "request_schedule_prompt", - format!("approval queued: {target_count} target(s)"), - ), - retries, - ) - }) - .await - } - - #[tool( - description = "Fire a scheduled prompt out of band — runs the per-target fan-out \ - once immediately without disturbing the schedule's cadence. Recurring schedules \ - keep their next_fire_at unchanged (the manual fire is additive). One-shot \ - schedules are CONSUMED by the manual fire (cancelled afterwards): the operator's \ - intent on a one-shot is 'send this now, the scheduled time was wrong'. \n\n\ - Authorization mirrors `cancel_schedule`: you can fire any schedule, whoever owns it." - )] - async fn fire_schedule_now(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("fire_schedule_now", log, async move { - let id = args.id; - let (resp, retries) = self - .dispatch(hive_core_agent_sock::Request::FireScheduleNow { id }) - .await; - annotate_retries( - format_ack(resp, "fire_schedule_now", format!("fired #{id} now")), - retries, - ) - }) - .await - } - - #[tool( - description = "Cancel a scheduled prompt. With no `targets` field, cancels the \ - whole schedule (all recipients flipped). With a non-empty `targets` list, cancels \ - just those recipients; the schedule keeps firing for any remaining active targets \ - and auto-cancels its parent row when every target is cancelled. \n\n\ - Authorization: you can cancel any schedule, whoever owns it." - )] - async fn cancel_schedule(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("cancel_schedule", log, async move { - let id = args.id; - let (resp, retries) = self - .dispatch(hive_core_agent_sock::Request::CancelSchedule { - id: args.id, - targets: args.targets, - }) - .await; - annotate_retries( - format_ack(resp, "cancel_schedule", format!("cancelled #{id}")), - retries, - ) - }) - .await - } - - #[tool( - description = "Edit an existing scheduled prompt's mutable fields. Pass only \ - the fields you want to change — anything omitted keeps its current value. Editable: \ - `body`, `description`, `interval_seconds` (positive only via this tool; flipping \ - recurring→one-shot is operator-only via the dashboard), `next_fire_at_unix`, and \ - the target set via `targets_add` / `targets_remove`. Both target lists are \ - applied in the same transaction with removes-before-adds, so a single edit can \ - swap a target atomically. Re-adding a previously-removed target starts a fresh \ - per-target history (drops the tombstone). Draining all targets auto-cancels the \ - parent schedule. \n\n\ - Authorization mirrors `cancel_schedule` / `fire_schedule_now`: you can edit any \ - schedule, whoever owns it. \ - Refuses cancelled schedules (the row's terminal — submit a fresh one)." - )] - async fn edit_schedule(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("edit_schedule", log, async move { - let id = args.id; - let (resp, retries) = self - .dispatch(hive_core_agent_sock::Request::EditSchedule { - id: args.id, - body: args.body, - description: args.description.map(Some), - interval_seconds: args.interval_seconds.map(Some), - next_fire_at_unix: args.next_fire_at_unix, - targets_add: args.targets_add, - targets_remove: args.targets_remove, - }) - .await; - annotate_retries( - format_ack(resp, "edit_schedule", format!("edited #{id}")), - retries, - ) - }) - .await - } - - #[tool( - description = "List every scheduled prompt in the hive, whoever owns it — active and \ - cancelled-but-not-yet-reaped. Returns schedule id, owner, body, target set with \ - per-target last_fired_at + last_result, next fire time, recurring interval. \ - Use this to look up an id before calling `cancel_schedule`, or to audit what \ - the hive is going to be woken up about next." - )] - async fn list_schedules(&self) -> String { - run_tool_envelope("list_schedules", String::new(), async move { - let (resp, retries) = self - .dispatch(hive_core_agent_sock::Request::ListSchedules) - .await; - let body = match resp { - Ok(hive_core_agent_sock::Response::Schedules { schedules }) => { - serde_json::to_string(&schedules) - .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")) - } - other => reply_err(other, "list_schedules"), - }; - annotate_retries(body, retries) - }) - .await - } } #[tool_handler( diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index 3a920f1a..2f384565 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -643,12 +643,12 @@ mod tests { Ok(hive_core_agent_sock::Response::OkWarn { warnings: vec!["name is reserved".to_owned(), "second thing".to_owned()], }), - "request_schedule_prompt", - "schedule approval queued".to_owned(), + "send", + "sent to iris".to_owned(), ); // The operation HAPPENED — dropping the success line would read as a - // failure and invite a retry that queues a second approval. - assert!(out.starts_with("schedule approval queued")); + // failure and invite a retry that sends a second message. + assert!(out.starts_with("sent to iris")); assert!(out.contains("⚠️ name is reserved")); assert!(out.contains("⚠️ second thing")); } @@ -659,10 +659,10 @@ mod tests { // warning marker would pass the test above. let out = format_ack( Ok(hive_core_agent_sock::Response::Ok), - "request_schedule_prompt", - "queued".to_owned(), + "send", + "sent".to_owned(), ); - assert_eq!(out, "queued"); + assert_eq!(out, "sent"); assert!(!out.contains('⚠')); } } diff --git a/hive-agent/prompts/system.md b/hive-agent/prompts/system.md index 973e671f..46f69969 100644 --- a/hive-agent/prompts/system.md +++ b/hive-agent/prompts/system.md @@ -4,7 +4,6 @@ Tools (hyperhive surface). Full signature + behavior for each comes from the too - **Inbox / messaging** (always available): `mcp__hyperhive__recv`, `mcp__hyperhive__ack_until`, `mcp__hyperhive__send`, `mcp__hyperhive__get_loose_ends`, `mcp__hyperhive__cancel_loose_end`, `mcp__hyperhive__mark_todos_done`, `mcp__hyperhive__remind`, `mcp__hyperhive__set_status`, `mcp__hyperhive__get_agent_meta`. One habit worth internalizing beyond the tool descriptions themselves: prefer ending the turn over repeatedly polling `recv` when idle (only turn-boundaries observe in-container todo wakes — bash-task completions, matrix unread, forge activity — and ending the turn is also your checkpoint). For a large todo backlog (`get_loose_ends` caps at 40 rows), clear reviewed ids in bulk with `mark_todos_done` rather than cancelling one at a time — there's no blind range-clear, only ids you've actually looked at. - **Extra MCP tools** (some agents only): `mcp____` — agent-specific (matrix client, scraper, db connector, etc.) declared in your `agent.nix` under `hyperhive.extraMcpServers`. First-class tools, already operator-approved at deploy time. -- **Scheduling** (_requires `scheduling` tool group_): `request_schedule_prompt` (queues an approval), `cancel_schedule`, `fire_schedule_now`, `edit_schedule`, `list_schedules` (these four don't need approval — you can manage any schedule, whoever owns it). Need new packages, env vars, or other NixOS config for yourself? You can't edit your own config directly — open a PR on your `agent-configs/` repo, or contact the operator directly. Config repos live at `/agents/{label}/config/` (read-only inside your container). All changes flow through operator-approved commits. @@ -16,7 +15,7 @@ Messages from sender `system` are hyperhive helper events (JSON body, `event` fi - `needs_update` — agent's flake rev is stale. Ask the operator to rebuild it. - `container_crash` — ask the operator to start it again; if it keeps crashing, say so with what you saw. -- `approval_resolved` — one of your own submitted approvals (a scheduled prompt, …) was approved, denied, or failed; the body carries the resolution. +- `approval_resolved` — one of your own submitted approvals was approved, denied, or failed; the body carries the resolution. Lifecycle notices that don't need an immediate turn — a new agent spawned, a container rebuilt/killed/destroyed, or its login state changing — surface as todos instead of messages now. Call `get_loose_ends` to see them. diff --git a/hive-agent/src/schedule_timer.rs b/hive-agent/src/schedule_timer.rs index 82e283f7..bada2353 100644 --- a/hive-agent/src/schedule_timer.rs +++ b/hive-agent/src/schedule_timer.rs @@ -4,9 +4,8 @@ //! `agent-schedules` KV bucket. This task reads that key every //! [`POLL_INTERVAL`] with the one subject the agent's own queue credential is //! granted there (`swarm_queue_client::agent_schedules::read_own`), and fires -//! each due schedule as the same `schedule`-subsystem todo, keyed -//! `schedule:`, that hive-c0re's schedule worker pushes. The wake path -//! is the todo store's. +//! each due schedule as a `schedule`-subsystem todo, keyed `schedule:`. +//! The wake path is the todo store's. //! //! **Fire times come from the schedule alone**: `anchor + k * interval` //! ([`Schedule::latest_slot`]). Rewriting the list, or changing a body, @@ -43,7 +42,7 @@ use crate::todos::Todos; /// longest a fire can lag its slot, and a change can take to reach the agent. const POLL_INTERVAL: Duration = Duration::from_secs(10); -/// The todo subsystem and source hive-c0re's schedule worker uses. +/// The todo subsystem and source a schedule fire carries. const SUBSYSTEM: &str = "schedule"; const SOURCE: &str = "scheduled"; @@ -161,7 +160,7 @@ fn tick( } } -/// Push `schedule` as the todo hive-c0re's worker pushes, and wake the turn +/// Push `schedule` as a todo, and wake the turn /// loop when it is new or reopened. fn push_todo(todos: &Todos, wake: &Notify, schedule: &Schedule) -> bool { let key = format!("schedule:{}", schedule.id); @@ -364,11 +363,10 @@ mod tests { assert_eq!(fired(&l, &slots, 1005), ["a body"]); } - /// The same todo hive-c0re's worker pushes, so the wake path does not - /// change: subsystem `schedule`, key `schedule:`, source - /// `scheduled`, and a reopen once the agent acked the previous fire. + /// A fire is subsystem `schedule`, key `schedule:`, source + /// `scheduled`, and reopens once the agent acked the previous fire. #[test] - fn a_fire_is_the_hive_c0re_schedule_todo() { + fn a_fire_is_the_schedule_todo() { let todos = Todos::open(Path::new(":memory:")).unwrap(); let wake = Notify::new(); let s = schedule("abcdefgh", 1000, Some(100)); diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index f8a37fe5..e3da65b7 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -40,8 +40,6 @@ hive-jobq.workspace = true hive-jobq-wire.workspace = true hive-log.workspace = true hive-priv-sock.workspace = true -hive-agent-sock.workspace = true -hive-sock-client.workspace = true hive-types.workspace = true libc.workspace = true listenfd = "1" diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 57a09c66..1d4fade0 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -21,7 +21,7 @@ use crate::lifecycle; /// /// Every queued kind resolves its approval row via /// [`resolve_approval_dag`] when the DAG settles terminal. -pub async fn approve(coord: Arc, id: i64) -> Result<()> { +pub fn approve(coord: &Arc, id: i64) -> Result<()> { let approval = coord.approvals.mark_approved(id)?; tracing::info!( %approval.id, @@ -50,19 +50,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. - let result = run_approval_schedule_prompt(&coord, approval).await; - if result.is_ok() { - coord.emit_schedules_snapshot(); - } - result - } } } @@ -162,37 +149,6 @@ pub(crate) async fn post_rev_deploy_failure_to_pr(agent: &str, rev: &str, err: & } } -/// 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::approvals::Approval, -) -> Result<()> { - let result: Result<()> = async { - let payload: hive_sh4re::manager::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.to_string(), - 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) -} - /// Resolve an approval row from how its DAG's work ended — the body of the /// [`NodeKind::ResolveApproval`] tail node. Every approval-carrying template /// resolves here: the pipeline is ordinary queue nodes, diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 5f03fa59..0e5fc88f 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -31,9 +31,9 @@ const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running"; pub struct Coordinator { pub broker: Arc, pub approvals: Arc, - /// Scheduled-prompts queue. One sqlite connection, - /// internal mutex; the worker drains due rows and the manager - /// handlers insert / cancel through the same handle. + /// Scheduled-prompt rows, read-only: the dashboard lists them so the + /// operator can recreate them with `swarmctl schedule`. Nothing on + /// the hive writes or fires them. pub scheduled_prompts: Arc, /// Full build-log capture. `lifecycle::run` / /// `lifecycle::prebuild_toplevel` `start()` a row per attempt, @@ -426,17 +426,6 @@ pub struct ApprovalResolved<'a> { pub description: Option, } -/// Field-named payload for [`Coordinator::emit_approval_added`]. -/// Mirrors the `ApprovalAdded` dashboard-event fields. `agent` -/// borrows from the caller; `approval_kind` is a compile-time -/// constant. -pub struct ApprovalAdded<'a> { - pub id: i64, - pub agent: &'a str, - pub approval_kind: &'static str, - pub description: Option, -} - /// The two ways [`Coordinator::set_paused_by_name`] can fail. See its own /// doc comment for why this is a distinct type rather than one /// `anyhow::Error` (`Write` already wraps one, `BadName` never needs to). @@ -587,42 +576,11 @@ impl Coordinator { }); } - /// Emit a `SchedulesChanged` snapshot event. Called from every - /// schedule mutation site (operator API handlers + the worker - /// after each tick that fires or rearms a row) so the dashboard's - /// scheduled-prompts tab updates live without polling. - pub fn emit_schedules_snapshot(self: &Arc) { - let mut schedules: Vec = - match self.scheduled_prompts.list() { - Ok(rows) => rows - .into_iter() - .map(crate::socket_server::schedule_to_wire_public) - .collect(), - Err(e) => { - tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed"); - return; - } - }; - // Strip ghost targets (destroyed agents) so the dashboard doesn't - // render dead columns. Best-effort: if the roster cache is - // momentarily contended we emit unfiltered rather than block this - // sync path — the next snapshot / page reload corrects it. - if let Some(live) = self.live_container_names_blocking() { - crate::socket_server::filter_ghost_schedule_targets(&mut schedules, &live); - } - self.emit_dashboard_event(DashboardEvent::SchedulesChanged { - seq: self.next_seq(), - schedules, - }); - } - /// Best-effort synchronous snapshot of live container names (the /// logical agent names from the last `nixos-container list` scan). /// Returns `None` when the roster cache lock is momentarily - /// contended, so a transient miss is treated as "roster unknown, - /// don't filter" rather than hiding live schedule targets. The async - /// `containers_snapshot` is the reliable path for request handlers; - /// this exists for the sync `emit_schedules_snapshot` SSE emit. + /// contended. The async `containers_snapshot` is the reliable path + /// for request handlers; this exists for the sync SSE snapshot emits. #[must_use] pub fn live_container_names_blocking(&self) -> Option> { self.last_containers @@ -772,24 +730,6 @@ impl Coordinator { self.meta_updates_active.load(Ordering::SeqCst) > 0 } - /// Emit `ApprovalAdded` immediately after the row is inserted in - /// sqlite. - pub fn emit_approval_added(&self, ev: ApprovalAdded<'_>) { - let ApprovalAdded { - id, - agent, - approval_kind, - description, - } = ev; - self.emit_dashboard_event(DashboardEvent::ApprovalAdded { - seq: self.next_seq(), - id, - agent: agent.to_owned(), - approval_kind, - description, - }); - } - /// Emit `ApprovalResolved` after `mark_approved` / `mark_denied` / /// `mark_failed` lands. `resolved_at` is stamped from the system /// clock here so call sites don't repeat the conversion; if you @@ -904,7 +844,7 @@ impl Coordinator { .with_context(|| format!("create agent dir {}", agent_dir.display()))?; let socket_path = Self::socket_path(name); // Hand the full Coordinator to the per-agent socket — it needs - // more than just the broker (approvals, scheduled_prompts, ...). + // more than just the broker (approvals, ...). let socket = socket_server::start(name, &socket_path, self.clone())?; self.agents.lock().unwrap().insert(name.to_owned(), socket); Ok(agent_dir) @@ -1176,77 +1116,6 @@ impl Coordinator { .unwrap_or_else(|| hive_sh4re::manager::MANAGER_AGENT.to_owned()) } - /// Push a todo directly into `agent`'s in-container todo store — a - /// best-effort *live* dial of its `hive-agent-sock` socket - /// (`hive_host_sock::agent_todo_socket`), same `UpsertTodo` request - /// shape the built-in in-container producers (matrix/bash/forge-notify) - /// already send — `subsystem` isn't restricted to that set, any - /// producer can use the same shape. This is the migration target for - /// `HelperEvent` variants that are pure "FYI, check when convenient" notices: the - /// event stops being a broker `Message` (which always drives an - /// immediate turn) and becomes a todo instead, with the same - /// dedup-by-key semantics as any other producer. - /// - /// Deliberately push, not queue: if the agent's container is down - /// (socket file absent) or the dial otherwise fails, this is a - /// silent no-op (logged at `debug`/`warn`) — no retry, no fallback - /// 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. - /// `reopen_if_acked` forwards to `UpsertTodo` (see its doc comment) — - /// `false` for every caller here except the scheduled-prompts worker. - pub async fn push_todo( - &self, - agent: &str, - subsystem: &str, - key: Option, - summary: String, - source: Option, - reopen_if_acked: bool, - ) -> 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")); - }; - 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?)")); - } - let req = hive_agent_sock::Request::UpsertTodo { - subsystem: subsystem.to_owned(), - key, - summary, - source, - reopen_if_acked, - }; - match hive_sock_client::request::<_, hive_agent_sock::Response>( - &path, - &req, - hive_sock_client::Retry::None, - ) - .await - { - 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(()), - } - } - /// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded /// the same way as `notify_manager` (sender = `SYSTEM_SENDER`, /// body = JSON-encoded event) — e.g. `ContainerCrash`, `NeedsUpdate`. diff --git a/hive-c0re/src/dashboard/approvals.rs b/hive-c0re/src/dashboard/approvals.rs index 6e0a8865..ced85ce2 100644 --- a/hive-c0re/src/dashboard/approvals.rs +++ b/hive-c0re/src/dashboard/approvals.rs @@ -31,7 +31,7 @@ pub(super) async fn post_approve( State(state): State, AxumPath(id): AxumPath, ) -> Response { - match actions::approve(state.coord.clone(), id).await { + match actions::approve(&state.coord, id) { // 200 instead of 303 — `actions::approve` fires // `ApprovalResolved` (success path) or the eventual failure // event, both of which the dashboard's derived store applies diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index 1b146b17..52dd7a5f 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -41,7 +41,7 @@ use crate::lifecycle; (name = "meta_inputs", description = "bulk flake-input update for the meta flake"), (name = "misc_api", description = "operator inbox, compose, spawn-request, hive stats"), (name = "permissions", description = "tool-group + capability assignment for agents"), - (name = "schedules", description = "scheduled-prompt + rebuild-queue CRUD"), + (name = "schedules", description = "read-only scheduled-prompt list + rebuild-queue cancel"), (name = "state_files", description = "proxied reads of allow-listed per-agent state files"), (name = "state_snapshot", description = "cold-load dashboard snapshot"), (name = "tombstones", description = "purge of retained state for destroyed agents"), @@ -112,8 +112,7 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { // GET-vs-GET or POST-vs-POST pairs across different paths do. Each // call below is therefore scoped to exactly one path — two handlers // in the same call only when they genuinely share a path with - // different methods (`schedules::api_schedules`/`post_schedule_new` - // on `/api/schedules`) — chained via + // different methods — chained via // repeated `.routes(...)` calls instead of one giant `routes!(...)` // with everything in it. let (router, api) = OpenApiRouter::::with_openapi(ApiDoc::openapi()) @@ -141,15 +140,7 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { .routes(routes!(permissions::post_permissions)) .routes(routes!(permissions::get_stale_permissions)) .routes(routes!(permissions::delete_agent_permissions)) - .routes(routes!( - schedules::api_schedules, - schedules::post_schedule_new - )) - .routes(routes!(schedules::patch_schedule)) - .routes(routes!(schedules::post_schedule_cancel)) - .routes(routes!(schedules::post_schedule_pause)) - .routes(routes!(schedules::post_schedule_resume)) - .routes(routes!(schedules::post_schedule_fire_now)) + .routes(routes!(schedules::api_schedules)) .routes(routes!(schedules::post_rebuild_queue_cancel)) .routes(routes!(approvals::post_approve)) .routes(routes!(approvals::post_deny)) @@ -367,15 +358,7 @@ mod router_build_probe { .routes(routes!(permissions::post_permissions)) .routes(routes!(permissions::get_stale_permissions)) .routes(routes!(permissions::delete_agent_permissions)) - .routes(routes!( - schedules::api_schedules, - schedules::post_schedule_new - )) - .routes(routes!(schedules::patch_schedule)) - .routes(routes!(schedules::post_schedule_cancel)) - .routes(routes!(schedules::post_schedule_pause)) - .routes(routes!(schedules::post_schedule_resume)) - .routes(routes!(schedules::post_schedule_fire_now)) + .routes(routes!(schedules::api_schedules)) .routes(routes!(schedules::post_rebuild_queue_cancel)) .routes(routes!(approvals::post_approve)) .routes(routes!(approvals::post_deny)) diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs index ec82bbd5..9c263d34 100644 --- a/hive-c0re/src/dashboard/schedules.rs +++ b/hive-c0re/src/dashboard/schedules.rs @@ -1,41 +1,31 @@ -//! Scheduled-prompt + rebuild-queue endpoints for the dashboard. +//! Scheduled-prompt list + rebuild-queue cancel endpoints for the dashboard. //! -//! Operator-direct schedule CRUD (`/api/schedules` GET/POST, `{id}` PATCH, -//! `{id}/cancel` + `{id}/fire-now`) — the operator click *is* the approval, -//! so these skip the manager approval gate. Also the co-located +//! `/api/schedules` is read-only: it lists the rows still in the hive's +//! `scheduled_prompts` table so the operator can recreate them with +//! `swarmctl schedule` (`docs/swarm/schedules.md`). Nothing on the hive +//! writes or fires those rows. Also the co-located //! `/api/rebuild-queue/{id}/cancel` endpoint. use axum::{ extract::{Path as AxumPath, State}, - http::StatusCode, response::{IntoResponse, Response}, }; -use problem_details::ProblemDetails; - -use crate::scheduled_prompts::ScheduleNotFoundOrCancelled; -use crate::scheduled_prompts_worker::FireNowReport; - -use super::{AppState, error_problem, error_response}; - -#[derive(serde::Serialize, utoipa::ToSchema)] -pub(super) struct NewScheduleBody { - id: i64, -} +use super::{AppState, error_response}; #[derive(serde::Serialize, utoipa::ToSchema)] pub(super) struct CancelResultBody { cancelled: bool, } -/// Snapshot of every schedule for the +/// Snapshot of every schedule row for the /// scheduled-prompts tab. /// /// Returns the wire shape directly so the frontend can render /// without an extra translation layer. // `hive_sh4re::schedule::WireSchedule` (the actual body) has no `ToSchema` — adding // one would pull `utoipa` into the wire-types crate for a single dashboard -// endpoint. `serde_json::Value` placeholder; see the batch report. +// endpoint. `serde_json::Value` placeholder. #[utoipa::path( get, path = "/api/schedules", @@ -48,13 +38,8 @@ pub(super) struct CancelResultBody { pub(super) async fn api_schedules(State(state): State) -> Response { match state.coord.scheduled_prompts.list() { Ok(rows) => { - let mut wire: Vec = rows - .into_iter() - .map(crate::socket_server::schedule_to_wire_public) - .collect(); - // Drop ghost targets (agents that no longer exist) so the - // table never shows dead columns. Uses the reliable async - // roster snapshot here on the request path. + let mut wire: Vec = + rows.into_iter().map(schedule_to_wire).collect(); let live: std::collections::HashSet = state .coord .containers_snapshot() @@ -62,117 +47,56 @@ pub(super) async fn api_schedules(State(state): State) -> Response { .into_iter() .map(|c| c.name) .collect(); - crate::socket_server::filter_ghost_schedule_targets(&mut wire, &live); + filter_ghost_schedule_targets(&mut wire, &live); axum::Json(wire).into_response() } Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")), } } -/// 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. -#[utoipa::path( - post, - path = "/api/schedules", - // `hive_sh4re::manager::SchedulePromptPayload` (the actual body) has no `ToSchema` — - // same reasoning as the `Vec` placeholder on - // `api_schedules` above. - request_body(content = serde_json::Value, description = "SchedulePromptPayload wire shape"), - responses( - (status = 200, description = "created; body carries the new row id", body = NewScheduleBody), - (status = 400, description = "no targets, empty body, or interval_seconds == 0"), - (status = 500, description = "submit failed"), - ), - tag = "schedules" -)] -pub(super) async fn post_schedule_new( - State(state): State, - axum::Json(payload): axum::Json, -) -> Result { - if payload.targets.is_empty() { - return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) - .with_detail("schedule must have at least one target")); - } - if payload - .targets - .iter() - .any(|t| t == hive_sh4re::manager::OPERATOR_RECIPIENT) - { - return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) - .with_detail("operator is not a valid schedule target")); - } - if payload.body.trim().is_empty() { - return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) - .with_detail("schedule body must be non-empty")); - } - if let Some(0) = payload.interval_seconds { - return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) - .with_detail("interval_seconds must be > 0 (use None for one-shot)")); - } - let new = crate::scheduled_prompts::NewSchedule { - owner: hive_sh4re::manager::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) => { - state.coord.emit_schedules_snapshot(); - Ok(axum::Json(NewScheduleBody { id }).into_response()) - } - Err(e) => Err(error_problem(&format!("schedule submit: {e:#}"))), +/// Drop schedule targets that point at agents which no longer exist, so +/// the schedule table doesn't render ghost columns for destroyed agents. +/// `live` is the set of logical agent names from the last +/// `nixos-container list` scan (stopped agents included, destroyed ones +/// absent). A view filter: the rows keep every target. +fn filter_ghost_schedule_targets( + schedules: &mut [hive_sh4re::schedule::WireSchedule], + live: &std::collections::HashSet, +) { + for s in schedules.iter_mut() { + s.targets.retain(|t| live.contains(&t.target)); } } -/// Optional JSON body for `fire-now`. Absent / empty body ⇒ -/// `reset_timer = false` (back-compat: cadence stays intact). -#[derive(serde::Deserialize, Default, utoipa::ToSchema)] -pub(super) struct FireNowBody { - #[serde(default)] - reset_timer: bool, -} - -/// Operator-initiated -/// out-of-band fire of a scheduled prompt. -/// -/// Runs the per-target fan-out once immediately and reports -/// per-target outcome counts. One-shot schedules are consumed -/// (cancelled) by a manual fire — the operator's intent is "send -/// this now, the scheduled time was wrong." For recurring schedules -/// the cadence stays intact unless the body carries -/// `{"reset_timer": true}`, in which case the countdown is re-armed -/// from now (`next_fire_at = now + interval`). -#[utoipa::path( - post, - path = "/api/schedules/{id}/fire-now", - params(("id" = i64, Path, description = "schedule row id")), - request_body(content = FireNowBody, description = "optional; absent body means reset_timer = false"), - responses( - (status = 200, description = "fired; per-target outcome counts", body = FireNowReport), - (status = 500, description = "schedule missing, cancelled, or fully drained"), - ), - tag = "schedules" -)] -pub(super) async fn post_schedule_fire_now( - State(state): State, - AxumPath(id): AxumPath, - body: Option>, -) -> Response { - let reset_timer = body.is_some_and(|axum::Json(b)| b.reset_timer); - match crate::scheduled_prompts_worker::fire_now(&state.coord, id, reset_timer).await { - Ok(report) => { - state.coord.emit_schedules_snapshot(); - axum::Json(report).into_response() - } - Err(e) => error_response(&format!("fire schedule {id} now: {e:#}")), +fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::schedule::WireSchedule { + hive_sh4re::schedule::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::schedule::WireScheduleSource::Operator + } + crate::scheduled_prompts::ScheduleSource::Approval { id } => { + hive_sh4re::schedule::WireScheduleSource::Approval { id } + } + }, + cancelled_at_unix: s.cancelled_at_unix, + paused_at_unix: s.paused_at_unix, + description: s.description, + targets: s + .targets + .into_iter() + .map(|t| hive_sh4re::schedule::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(), } } @@ -212,213 +136,55 @@ pub(super) async fn post_rebuild_queue_cancel( } } -#[derive(serde::Deserialize, Default, utoipa::ToSchema)] -pub(super) struct CancelScheduleForm { - /// `None` / absent / empty array → cancel whole schedule. - #[serde(default)] - targets: Option>, -} +#[cfg(test)] +mod tests { + use super::*; -#[derive(serde::Deserialize, Default, utoipa::ToSchema)] -#[allow( - clippy::option_option, - reason = "double-Option carries three-state PATCH semantics on the wire \ - (missing key = leave alone, JSON null = clear, value = set); \ - collapsing to a single Option would lose the 'clear' state" -)] -pub(super) struct EditScheduleForm { - #[serde(default)] - body: Option, - /// Double-`Option` semantics on the wire: missing key = leave - /// alone, explicit `null` = clear, value = set. serde's - /// `deserialize_with` trick to distinguish missing from null: - /// we wrap each editable field in its own helper. Simpler - /// here — keep them plain `Option>` and document - /// that the dashboard caller passes JSON `null` to clear. - #[serde(default, deserialize_with = "deserialize_some")] - description: Option>, - #[serde(default, deserialize_with = "deserialize_some")] - interval_seconds: Option>, - #[serde(default)] - next_fire_at_unix: Option, - /// New targets to add. Replace-on-conflict: re-adding a - /// previously-cancelled target drops the tombstone and the - /// target starts fresh (operator intent on re-add = "this - /// target is active again, fresh start"). - #[serde(default)] - targets_add: Option>, - /// Targets to cancel. Same path as `cancel_targets`: - /// tombstones preserve audit and the parent schedule - /// auto-cancels when no active targets remain. - #[serde(default)] - targets_remove: Option>, -} - -/// serde adaptor: turns missing-key into `None`, explicit-null -/// into `Some(None)`, value into `Some(Some(v))`. Standard trick -/// for distinguishing "field absent" from "field set to null" in -/// JSON PATCH bodies. -fn deserialize_some<'de, T, D>(deserializer: D) -> Result, D::Error> -where - T: serde::Deserialize<'de>, - D: serde::Deserializer<'de>, -{ - T::deserialize(deserializer).map(Some) -} - -/// Partial update of an existing -/// schedule. -/// -/// Mutable fields: `body`, `description`, `interval_seconds`, -/// `next_fire_at_unix`, plus the target set via `targets_add` / -/// `targets_remove`. Both target lists are applied in the same -/// transaction as the scalar fields with removes-before-adds; -/// re-adding a previously-removed target resets per-target history -/// (fresh start); draining all targets auto-cancels the parent -/// schedule. JSON body uses missing-key = "leave alone", explicit -/// null = "clear" for `description` + `interval_seconds`. Cancelled -/// schedules are refused — submit a new one instead. Returns the -/// updated `WireSchedule` so the caller's post-edit refresh has the -/// new state inline. -#[utoipa::path( - patch, - path = "/api/schedules/{id}", - params(("id" = i64, Path, description = "schedule row id")), - responses( - (status = 200, description = "updated, wire shape", body = serde_json::Value), - (status = 500, description = "update failed (cancelled, not found, ...)"), - ), - tag = "schedules" -)] -pub(super) async fn patch_schedule( - State(state): State, - AxumPath(id): AxumPath, - axum::Json(form): axum::Json, -) -> Response { - let patch = crate::scheduled_prompts::UpdateSchedule { - body: form.body, - description: form.description, - interval_seconds: form.interval_seconds, - next_fire_at_unix: form.next_fire_at_unix, - targets_add: form.targets_add, - targets_remove: form.targets_remove, - }; - if let Err(e) = state.coord.scheduled_prompts.update(id, patch) { - return error_response(&format!("edit schedule {id}: {e:#}")); - } - match state.coord.scheduled_prompts.get(id) { - Ok(Some(s)) => { - let wire = crate::socket_server::schedule_to_wire_public(s); - state.coord.emit_schedules_snapshot(); - axum::Json(wire).into_response() - } - Ok(None) => error_response(&format!("edit schedule {id}: row vanished post-update")), - Err(e) => error_response(&format!("re-read schedule {id}: {e:#}")), - } -} - -/// Pause a schedule so the worker -/// skips it until explicitly resumed. -/// -/// Idempotent; no-op on an already-paused row. Returns 404 when the -/// schedule is cancelled or not found. -#[utoipa::path( - post, - path = "/api/schedules/{id}/pause", - params(("id" = i64, Path, description = "schedule row id")), - responses( - (status = 200, description = "paused", body = String), - (status = 404, description = "schedule cancelled or not found"), - (status = 500, description = "pause failed"), - ), - tag = "schedules" -)] -pub(super) async fn post_schedule_pause( - State(state): State, - AxumPath(id): AxumPath, -) -> Response { - match state.coord.scheduled_prompts.pause(id) { - Ok(()) => { - state.coord.emit_schedules_snapshot(); - (StatusCode::OK, "ok").into_response() - } - Err(e) => { - if e.downcast_ref::().is_some() { - (StatusCode::NOT_FOUND, format!("{e}")).into_response() - } else { - error_response(&format!("pause schedule {id}: {e:#}")) - } + fn target(name: &str) -> hive_sh4re::schedule::WireScheduleTarget { + hive_sh4re::schedule::WireScheduleTarget { + target: name.to_owned(), + cancelled_at_unix: None, + last_fired_at_unix: None, + last_result: None, } } -} -/// Resume a paused schedule. -/// -/// Idempotent; no-op on an already-active row. Returns 404 when the -/// schedule is cancelled or not found. -#[utoipa::path( - post, - path = "/api/schedules/{id}/resume", - params(("id" = i64, Path, description = "schedule row id")), - responses( - (status = 200, description = "resumed", body = String), - (status = 404, description = "schedule cancelled or not found"), - (status = 500, description = "resume failed"), - ), - tag = "schedules" -)] -pub(super) async fn post_schedule_resume( - State(state): State, - AxumPath(id): AxumPath, -) -> Response { - match state.coord.scheduled_prompts.resume(id) { - Ok(()) => { - state.coord.emit_schedules_snapshot(); - (StatusCode::OK, "ok").into_response() - } - Err(e) => { - if e.downcast_ref::().is_some() { - (StatusCode::NOT_FOUND, format!("{e}")).into_response() - } else { - error_response(&format!("resume schedule {id}: {e:#}")) - } + fn schedule(targets: &[&str]) -> hive_sh4re::schedule::WireSchedule { + hive_sh4re::schedule::WireSchedule { + id: 1, + owner: "operator".to_owned(), + body: "ping".to_owned(), + interval_seconds: None, + next_fire_at_unix: hive_sh4re::wire_time::from_secs(0), + created_at_unix: hive_sh4re::wire_time::from_secs(0), + source: hive_sh4re::schedule::WireScheduleSource::Operator, + cancelled_at_unix: None, + paused_at_unix: None, + description: None, + targets: targets.iter().map(|t| target(t)).collect(), } } -} -/// 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. -#[utoipa::path( - post, - path = "/api/schedules/{id}/cancel", - params(("id" = i64, Path, description = "schedule row id")), - responses( - (status = 200, description = "cancelled (whole or partial)", body = String), - (status = 500, description = "cancel failed"), - ), - tag = "schedules" -)] -pub(super) 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(()) => { - state.coord.emit_schedules_snapshot(); - (StatusCode::OK, "ok").into_response() - } - Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")), + #[test] + fn ghost_filter_drops_dead_agents_keeps_live() { + let live: std::collections::HashSet = ["iris".to_owned(), "damocles".to_owned()] + .into_iter() + .collect(); + let mut schedules = vec![schedule(&["iris", "ghost", "damocles"])]; + filter_ghost_schedule_targets(&mut schedules, &live); + let kept: Vec<&str> = schedules[0] + .targets + .iter() + .map(|t| t.target.as_str()) + .collect(); + assert_eq!(kept, vec!["iris", "damocles"]); + } + + #[test] + fn ghost_filter_can_empty_targets_when_all_dead() { + let live: std::collections::HashSet = std::collections::HashSet::new(); + let mut schedules = vec![schedule(&["gone1", "gone2"])]; + filter_ghost_schedule_targets(&mut schedules, &live); + assert!(schedules[0].targets.is_empty()); } } diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 2b46bad0..56265f2b 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -152,8 +152,7 @@ struct ApprovalView { #[serde(skip_serializing_if = "Option::is_none")] description: Option, /// Raw `commit_ref` payload for `UpdateMetaInputs` (JSON-encoded - /// `Vec` of input names; `"[]"` = all inputs) and - /// `SchedulePrompt` (JSON-encoded `SchedulePromptPayload`). The + /// `Vec` of input names; `"[]"` = all inputs). The /// frontend parses this to render a human-readable card body. commit_ref: String, /// RFC 3339 UTC time the approval was queued. Rendered as a diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index 7af7d384..7be92dcf 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -48,24 +48,14 @@ pub enum DashboardEvent { #[serde(default, skip_serializing_if = "Vec::is_empty")] file_refs: Vec, }, - /// A new approval landed in the pending queue. Payload carries - /// enough to render the dashboard row without a `/api/state` - /// refetch. - /// - /// The approval's own kind (`"schedule_prompt"` / `"update_meta_inputs"`) lives - /// on `approval_kind` rather than `kind` because the latter is taken - /// by the serde tag identifying which `DashboardEvent` variant - /// this is. - ApprovalAdded { - seq: u64, - id: i64, - agent: String, - approval_kind: &'static str, - description: Option, - }, /// A pending approval transitioned to a terminal state /// (approved / denied / failed). Clients move the row out of the /// pending list and into history. + /// + /// The approval's own kind (`"update_meta_inputs"`) lives + /// on `approval_kind` rather than `kind` because the latter is taken + /// by the serde tag identifying which `DashboardEvent` variant + /// this is. ApprovalResolved { seq: u64, id: i64, @@ -170,17 +160,6 @@ pub enum DashboardEvent { /// (transients, container state, approvals), and a poll loop would be /// slower for the same information. RebuildQueueChanged { seq: u64 }, - /// Full snapshot of all scheduled prompts. Emitted after every - /// operator mutation (new / edit / cancel / fire-now) and after the - /// worker fires or rearms a row. Same snapshot-shape rationale as - /// `TombstonesChanged` / `MetaInputsChanged` — the list is small and the - /// client's - /// per-target `last_result` / `last_fired_at_unix` fields are most - /// naturally re-derived from the full list. - SchedulesChanged { - seq: u64, - schedules: Vec, - }, /// Full snapshot of capability grants (per-agent `Vec`). /// Emitted from the rebuild-queue worker after a `PermChange` /// `Capabilities` entry commits the JSON file. Lets the P3RM1SS10NS @@ -235,7 +214,6 @@ impl DashboardEvent { match self { DashboardEvent::Sent { .. } => "sent", DashboardEvent::Delivered { .. } => "delivered", - DashboardEvent::ApprovalAdded { .. } => "approval_added", DashboardEvent::ApprovalResolved { .. } => "approval_resolved", DashboardEvent::TransientSet { .. } => "transient_set", DashboardEvent::TransientCleared { .. } => "transient_cleared", @@ -245,7 +223,6 @@ impl DashboardEvent { DashboardEvent::MetaInputsChanged { .. } => "meta_inputs_changed", DashboardEvent::MetaUpdateRunning { .. } => "meta_update_running", DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed", - DashboardEvent::SchedulesChanged { .. } => "schedules_changed", DashboardEvent::CapabilitiesChanged { .. } => "capabilities_changed", DashboardEvent::ToolGroupsChanged { .. } => "tool_groups_changed", } @@ -294,18 +271,11 @@ mod tests { in_reply_to: None, file_refs: Vec::new(), }, - DashboardEvent::ApprovalAdded { - seq: 1, - id: 1, - agent: "x".into(), - approval_kind: "schedule_prompt", - description: None, - }, DashboardEvent::ApprovalResolved { seq: 1, id: 1, agent: "x".into(), - approval_kind: "schedule_prompt", + approval_kind: "update_meta_inputs", status: "approved", resolved_at: hive_sh4re::wire_time::from_secs(0), note: None, @@ -339,10 +309,6 @@ mod tests { running: false, }, DashboardEvent::RebuildQueueChanged { seq: 1 }, - DashboardEvent::SchedulesChanged { - seq: 1, - schedules: Vec::new(), - }, DashboardEvent::CapabilitiesChanged { seq: 1, caps: Vec::new(), diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 9749d7d3..710cfa98 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -383,10 +383,6 @@ async fn run_destroy_bookkeeping(coord: &Arc, agent: &str, purge: b // (gained one on destroy, lost one on purge — recompute either way). coord.rescan_containers_and_emit().await; crate::dashboard::emit_tombstones_snapshot(coord).await; - // Re-emit the schedules snapshot: the rescan above refreshed the live - // roster, so any schedule that still targets the just-destroyed agent now - // drops that ghost column live (no page reload needed). - coord.emit_schedules_snapshot(); } /// Emit this agent's rebuild-complete todo. `ok` is not computed — it is which diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 3911e305..9d1a6278 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -45,9 +45,7 @@ mod workers; pub(crate) use agent_config::{capabilities, limits, resource_limits, tool_groups, topology}; pub(crate) use stats::{container_stats, hive_stats, host_stats, otel_metrics, warnings}; pub(crate) use stores::{approvals, broker, build_logs, db, power, scheduled_prompts}; -pub(crate) use workers::{ - agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, scheduled_prompts_worker, -}; +pub(crate) use workers::{agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets}; use coordinator::{Coordinator, HiveEnv, ServeConfig}; @@ -240,7 +238,7 @@ fn submit_periodic_sweep( } /// Start the coordinator daemon: open the broker, run migrations, spawn -/// background tasks (auto-update, vacuums, crash-watcher, scheduled-prompts, +/// background tasks (auto-update, vacuums, crash-watcher, /// dashboard), then serve the admin socket until a signal arrives. #[allow( clippy::too_many_lines, @@ -433,10 +431,6 @@ async fn cmd_serve( // run_reconcile call register_agent on start; kill/destroy call // unregister_agent. No recurring poll needed — c0re owns the listeners. mcp_sockets::sync_on_start(coord.clone()).await; - // Scheduled-prompts worker: drains due scheduled_prompts rows - // and fans the body out to each active target's inbox. See - // scheduled_prompts_worker.rs. - scheduled_prompts_worker::spawn(coord.clone()); // Job-queue scheduler: drives the global DAG queue (rebuild / // meta-update / spawn / power ops). Concurrency comes from the // build-slot count + per-agent leases inside the queue, not from diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 729fbd46..ef17b7ba 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -186,7 +186,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { HostRequest::Urls => HostResponse::urls(hive_urls()), HostRequest::Pending => HostResponse::pending(coord.approvals.pending()?), HostRequest::Approve { id } => { - actions::approve(coord.clone(), *id).await?; + actions::approve(&coord, *id)?; HostResponse::success() } HostRequest::Deny { id } => { diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 572a2403..69bd38d9 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -3,8 +3,8 @@ //! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means //! you are `foo`; the manager socket simply serves as `ruth`. There is no //! privilege flag — both transports run the same [`serve`] / [`dispatch`] -//! code, and authority derives uniformly from the caller's identity: -//! tool-group membership for the orchestration verbs. An agent-targeting +//! code, and authority derives uniformly from the caller's identity +//! (tool-group membership for cancelling an approval). An agent-targeting //! verb may name any agent, so `ruth` reaches every agent exactly the way //! every other agent does, not via any hardcoded name match. @@ -21,16 +21,6 @@ use tokio::task::JoinHandle; use crate::coordinator::Coordinator; -mod schedules; - -pub(crate) use schedules::filter_ghost_schedule_targets; -pub use schedules::schedule_to_wire_public; - -use schedules::{ - EditSchedulePatch, handle_cancel_schedule, handle_edit_schedule, handle_fire_schedule_now, - handle_list_schedules, handle_request_schedule_prompt, -}; - pub struct AgentSocket { pub path: PathBuf, pub handle: JoinHandle<()>, @@ -275,21 +265,15 @@ pub(crate) fn recv_timeout(wait_seconds: Option) -> std::time::Duration { } } -/// Handle the subset of `Request` variants that are identical on both -/// the agent socket and the manager socket. Returns `Some(response)` for -/// every variant it handles; returns `None` for the remaining variants — -/// `GetLooseEnds` and the orchestration verbs (schedules), which need -/// per-verb tool-group gating — or for host-admin / unknown requests -/// invalid on either socket. -/// -/// The unified `dispatch` calls this first; the remaining arms (which gate -/// on topology / capabilities / tool-groups) are handled there. -pub(crate) async fn dispatch_shared( +/// Unified dispatch for every socket connection — per-agent sockets and the +/// (now pure-transport) manager socket alike. There is no privilege bit; +/// authority derives uniformly from the caller's identity. +async fn dispatch( req: &hive_core_agent_sock::Request, agent: &str, coord: &Arc, -) -> Option { - Some(match req { +) -> hive_core_agent_sock::Response { + match req { hive_core_agent_sock::Request::Send { to, body, @@ -331,9 +315,8 @@ pub(crate) async fn dispatch_shared( coord.clear_pause_pending(agent); hive_core_agent_sock::Response::Ok } - // Not a shared variant. - _ => return None, - }) + hive_core_agent_sock::Request::GetLooseEnds => handle_get_loose_ends(coord, agent), + } } /// `Recv` — long-poll the broker for up to `max` messages (capped at @@ -572,104 +555,6 @@ fn handle_requeue_inflight( } } -/// Unified dispatch for every socket connection — per-agent sockets and the -/// (now pure-transport) manager socket alike. There is no privilege bit; -/// authority derives uniformly from the caller's identity: hive-wide -/// orchestration verbs (schedules) require the matching tool-group (the -/// grantable capability). -async fn dispatch(req: &Request, agent: &str, coord: &Arc) -> Response { - if let Some(resp) = dispatch_shared(req, agent, coord).await { - return resp; - } - match req { - Request::GetLooseEnds => handle_get_loose_ends(coord, agent), - // Orchestration verbs — gated per-verb on tool-group membership - // (see `dispatch_orchestration`). - _ => dispatch_orchestration(req, agent, coord).await, - } -} - -/// Handle the hive-wide orchestration verbs (scheduling). No blanket socket -/// gate: each verb gates on the grantable capability that authorises it — -/// the matching tool-group (`scheduling`). Any other variant is a -/// host-admin / unknown request invalid on either socket. -async fn dispatch_orchestration(req: &Request, agent: &str, coord: &Arc) -> Response { - match req { - Request::RequestSchedulePrompt(payload) => { - if let Some(err) = require_group(agent, "scheduling", "schedule a prompt") { - return err; - } - handle_request_schedule_prompt(coord, agent, payload) - } - Request::CancelSchedule { id, targets } => { - if let Some(err) = require_group(agent, "scheduling", "cancel a schedule") { - return err; - } - handle_cancel_schedule(coord, agent, *id, targets.as_deref()) - } - Request::EditSchedule { - id, - body, - description, - interval_seconds, - next_fire_at_unix, - targets_add, - targets_remove, - } => { - if let Some(err) = require_group(agent, "scheduling", "edit a schedule") { - return err; - } - handle_edit_schedule( - coord, - agent, - *id, - EditSchedulePatch { - body: body.clone(), - description: description.clone(), - interval_seconds: *interval_seconds, - next_fire_at_unix: *next_fire_at_unix, - targets_add: targets_add.clone(), - targets_remove: targets_remove.clone(), - }, - ) - } - Request::ListSchedules => { - if let Some(err) = require_group(agent, "scheduling", "list schedules") { - return err; - } - handle_list_schedules(coord, agent) - } - Request::FireScheduleNow { id } => { - if let Some(err) = require_group(agent, "scheduling", "fire a schedule") { - return err; - } - handle_fire_schedule_now(coord, agent, *id).await - } - // Host-admin-only / unknown variants: never valid on either socket. - _ => Response::Err { - message: "request not handled on this socket".to_owned(), - }, - } -} - -/// Capability guard for the hive-wide orchestration verbs: the caller must -/// hold the given tool-group. The tool-group (c0re-owned `tool_groups.json`, -/// read server-side via [`crate::tool_groups::groups_for`]) is the grantable -/// capability — granting it to an orchestrator (e.g. the root) authorises -/// these verbs without any positional/hardcoded privilege. `action` is the -/// verb phrase for the message. -fn require_group(agent: &str, group: &str, action: &str) -> Option { - match crate::tool_groups::groups_for(agent) { - Ok(groups) if groups.iter().any(|g| g == group) => None, - Ok(_) => Some(Response::Err { - message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"), - }), - Err(e) => Some(Response::Err { - message: format!("agent `{agent}` cannot {action}: {e}"), - }), - } -} - /// `GetLooseEnds` — always the caller's own loose ends. fn handle_get_loose_ends(coord: &Arc, agent: &str) -> Response { match crate::loose_ends::for_agent(coord, agent) { @@ -767,10 +652,9 @@ fn check_can_cancel_approval(canceller: &str) -> Result<(), String> { } } -/// Common Send handler shared between dispatch arms. Applies the +/// `Send` handler. Applies the /// 4 KiB body cap, then routes broadcast (`to == "*"`) / unicast through /// their respective broker calls. -/// `pub(crate)` so `dispatch_shared` can use it across both socket paths. pub(crate) fn handle_send( coord: &Arc, agent: &str, @@ -837,6 +721,21 @@ pub(crate) fn handle_send( mod tests { use super::*; + /// A real `Coordinator` over a throwaway sqlite dir. `serve` takes an + /// `Arc`, so there is no lighter way in; `open` touches + /// nothing outside the db path it is handed. + fn coordinator() -> (tempfile::TempDir, Arc) { + let dir = tempfile::tempdir().expect("tempdir"); + let coord = Coordinator::open( + &dir.path().join("broker.sqlite"), + crate::coordinator::HiveEnv::default(), + crate::hive_stats::PriceTable::new(), + 1, + ) + .expect("open coordinator"); + (dir, Arc::new(coord)) + } + #[test] fn approval_cancel_rejects_callers_without_the_approvals_group() { // A caller that doesn't hold the `approvals` tool-group must not be @@ -869,7 +768,7 @@ mod tests { #[tokio::test] async fn serve_answers_an_over_long_line_with_an_error_and_closes() { - let (_dir, coord) = schedules::tests::coordinator(); + let (_dir, coord) = coordinator(); let (client, server) = UnixStream::pair().expect("socketpair"); let handler = tokio::spawn(serve(server, "iris".to_owned(), coord)); let (read, mut write) = client.into_split(); diff --git a/hive-c0re/src/socket_server/schedules.rs b/hive-c0re/src/socket_server/schedules.rs deleted file mode 100644 index 564e89c3..00000000 --- a/hive-c0re/src/socket_server/schedules.rs +++ /dev/null @@ -1,485 +0,0 @@ -//! Scheduled-prompt request handlers (`ListSchedules` / -//! `RequestSchedulePrompt` / `CancelSchedule` / `EditSchedule` / -//! `FireScheduleNow`) and the -//! schedule-to-wire mapping reused by the dashboard -//! (`schedule_to_wire_public` / `filter_ghost_schedule_targets`). - -use std::sync::Arc; - -use hive_core_agent_sock::Response; - -use crate::coordinator::Coordinator; - -/// `ListSchedules` — snapshot every scheduled prompt. Ownership no longer -/// narrows the view: `CancelSchedule`/`EditSchedule`/`FireScheduleNow` accept -/// any requester, so filtering the listing would only hide rows the requester -/// can act on anyway. -pub(super) fn handle_list_schedules(coord: &Arc, requester: &str) -> Response { - tracing::debug!(%requester, "list schedules"); - match coord.scheduled_prompts.list() { - Ok(schedules) => Response::Schedules { - schedules: schedules.into_iter().map(schedule_to_wire).collect(), - }, - Err(e) => Response::Err { - message: format!("list scheduled prompts: {e:#}"), - }, - } -} - -/// 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. -pub(super) fn handle_request_schedule_prompt( - coord: &Arc, - requester: &str, - payload: &hive_sh4re::manager::SchedulePromptPayload, -) -> Response { - if payload.targets.is_empty() { - return Response::Err { - message: "schedule must have at least one target".into(), - }; - } - if payload - .targets - .iter() - .any(|t| t == hive_sh4re::manager::OPERATOR_RECIPIENT) - { - return Response::Err { - message: "operator is not a valid schedule target".into(), - }; - } - if payload.body.trim().is_empty() { - return Response::Err { - message: "schedule body must be non-empty".into(), - }; - } - if let Some(0) = payload.interval_seconds { - return Response::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 Response::Err { - message: format!("encode SchedulePromptPayload: {e:#}"), - }; - } - }; - let id = match coord.approvals.submit_kind( - requester, - hive_sh4re::approvals::ApprovalKind::SchedulePrompt, - &commit_ref, - payload.description.as_deref(), - requester, - ) { - Ok(id) => id, - Err(e) => { - return Response::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(crate::coordinator::ApprovalAdded { - id, - agent: requester, - approval_kind: "schedule_prompt", - description: payload.description.clone(), - }); - Response::Ok -} - -/// Cancel a schedule (whole or per-target). Any requester may cancel any -/// schedule, whoever owns it. -pub(super) fn handle_cancel_schedule( - coord: &Arc, - requester: &str, - schedule_id: i64, - targets: Option<&[String]>, -) -> Response { - // Existence check only — it buys the caller a clean "not found" instead - // of whatever the cancel itself would say about an absent row. - if let Some(err) = require_schedule(coord, schedule_id) { - return err; - } - tracing::info!(%requester, %schedule_id, "cancel schedule"); - 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(()) => { - coord.emit_schedules_snapshot(); - Response::Ok - } - Err(message) => Response::Err { message }, - } -} - -/// Dispatch a `FireScheduleNow` request. Same rule as `CancelSchedule`: any -/// requester, any schedule. The actual fan-out lives in -/// `scheduled_prompts_worker::fire_now`. -pub(super) async fn handle_fire_schedule_now( - coord: &Arc, - requester: &str, - schedule_id: i64, -) -> Response { - if let Some(err) = require_schedule(coord, schedule_id) { - return err; - } - tracing::info!(%requester, %schedule_id, "fire schedule now"); - // MCP fire_schedule_now stays no-reset (cadence intact); the - // reset-timer option is a dashboard-dialog affordance. - match crate::scheduled_prompts_worker::fire_now(coord, schedule_id, false).await { - Ok(_report) => { - coord.emit_schedules_snapshot(); - Response::Ok - } - Err(e) => Response::Err { - message: format!("fire schedule {schedule_id} now: {e:#}"), - }, - } -} - -/// Field-named PATCH payload for [`handle_edit_schedule`]. Every -/// field is "leave alone" when `None`; the double-`Option` fields -/// additionally distinguish clear (`Some(None)`) from set -/// (`Some(Some(v))`). -#[allow( - clippy::option_option, - reason = "double-Option carries three-state PATCH semantics: outer None = \ - leave alone, Some(None) = clear, Some(Some(v)) = set" -)] -pub(super) struct EditSchedulePatch { - pub(super) body: Option, - pub(super) description: Option>, - pub(super) interval_seconds: Option>, - pub(super) next_fire_at_unix: Option, - pub(super) targets_add: Option>, - pub(super) targets_remove: Option>, -} - -/// Dispatch an `EditSchedule` patch. Same rule as `CancelSchedule`: any -/// requester, any schedule. Forwards the partial payload to -/// `ScheduledPrompts::update` which enforces the cancelled-row / -/// zero-interval validation. Returns `Ok` on a clean update; -/// `Err` with the underlying message on any validation -/// failure so the dashboard can surface it verbatim. -pub(super) fn handle_edit_schedule( - coord: &Arc, - requester: &str, - schedule_id: i64, - patch: EditSchedulePatch, -) -> Response { - let EditSchedulePatch { - body, - description, - interval_seconds, - next_fire_at_unix, - targets_add, - targets_remove, - } = patch; - if let Some(err) = require_schedule(coord, schedule_id) { - return err; - } - tracing::info!(%requester, %schedule_id, "edit schedule"); - let patch = crate::scheduled_prompts::UpdateSchedule { - body, - description, - interval_seconds, - next_fire_at_unix, - targets_add, - targets_remove, - }; - match coord.scheduled_prompts.update(schedule_id, patch) { - Ok(()) => { - coord.emit_schedules_snapshot(); - Response::Ok - } - Err(e) => Response::Err { - message: format!("edit schedule {schedule_id}: {e:#}"), - }, - } -} - -/// Shared existence guard for the three schedule-mutating verbs: `None` when -/// the row is there, `Some(Err)` to short-circuit when it isn't or the read -/// failed. Not a permission check — any requester may act on any schedule. -fn require_schedule(coord: &Arc, schedule_id: i64) -> Option { - match coord.scheduled_prompts.get(schedule_id) { - Ok(Some(_)) => None, - Ok(None) => Some(Response::Err { - message: format!("schedule {schedule_id} not found"), - }), - Err(e) => Some(Response::Err { - message: format!("read schedule {schedule_id}: {e:#}"), - }), - } -} - -/// 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 socket-server file. -pub fn schedule_to_wire_public( - s: crate::scheduled_prompts::Schedule, -) -> hive_sh4re::schedule::WireSchedule { - schedule_to_wire(s) -} - -/// Drop schedule targets that point at agents which no longer exist, so -/// the dashboard's schedule table doesn't render ghost columns for -/// destroyed agents. `live` is the set of logical agent names from the -/// last `nixos-container list` scan (stopped agents included, destroyed -/// ones absent). Applied only to the dashboard wire paths -/// (`api_schedules` + the `SchedulesChanged` SSE emit) — `list_schedules` -/// doesn't apply this filter (it still returns every live target, ghost -/// or not, on the schedules the requester is authorized to see) so an -/// agent can still see and cancel its own stale targets. This is a view -/// filter: the underlying schedule rows keep every target, so a -/// re-spawned agent's targets reappear on their own. -pub(crate) fn filter_ghost_schedule_targets( - schedules: &mut [hive_sh4re::schedule::WireSchedule], - live: &std::collections::HashSet, -) { - for s in schedules.iter_mut() { - s.targets.retain(|t| live.contains(&t.target)); - } -} - -fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::schedule::WireSchedule { - hive_sh4re::schedule::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::schedule::WireScheduleSource::Operator - } - crate::scheduled_prompts::ScheduleSource::Approval { id } => { - hive_sh4re::schedule::WireScheduleSource::Approval { id } - } - }, - cancelled_at_unix: s.cancelled_at_unix, - paused_at_unix: s.paused_at_unix, - description: s.description, - targets: s - .targets - .into_iter() - .map(|t| hive_sh4re::schedule::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(), - } -} - -#[cfg(test)] -pub(super) mod tests { - use super::*; - - fn target(name: &str) -> hive_sh4re::schedule::WireScheduleTarget { - hive_sh4re::schedule::WireScheduleTarget { - target: name.to_owned(), - cancelled_at_unix: None, - last_fired_at_unix: None, - last_result: None, - } - } - - fn schedule(targets: &[&str]) -> hive_sh4re::schedule::WireSchedule { - hive_sh4re::schedule::WireSchedule { - id: 1, - owner: "operator".to_owned(), - body: "ping".to_owned(), - interval_seconds: None, - next_fire_at_unix: hive_sh4re::wire_time::from_secs(0), - created_at_unix: hive_sh4re::wire_time::from_secs(0), - source: hive_sh4re::schedule::WireScheduleSource::Operator, - cancelled_at_unix: None, - paused_at_unix: None, - description: None, - targets: targets.iter().map(|t| target(t)).collect(), - } - } - - #[test] - fn ghost_filter_drops_dead_agents_keeps_live() { - let live: std::collections::HashSet = ["iris".to_owned(), "damocles".to_owned()] - .into_iter() - .collect(); - let mut schedules = vec![schedule(&["iris", "ghost", "damocles"])]; - filter_ghost_schedule_targets(&mut schedules, &live); - let kept: Vec<&str> = schedules[0] - .targets - .iter() - .map(|t| t.target.as_str()) - .collect(); - // `ghost` (destroyed) dropped; live agents kept. - assert_eq!(kept, vec!["iris", "damocles"]); - } - - #[test] - fn ghost_filter_can_empty_targets_when_all_dead() { - let live: std::collections::HashSet = std::collections::HashSet::new(); - let mut schedules = vec![schedule(&["gone1", "gone2"])]; - filter_ghost_schedule_targets(&mut schedules, &live); - assert!(schedules[0].targets.is_empty()); - } - - // ----------------------------------------------------------------------- - // The scheduling surface permits an unrelated requester - // - // These pin the *decision* the deleted `schedule_authorized` used to make, - // inverted: a requester with no relation to a schedule's owner reaches the - // handler's effect instead of a refusal. They live at the handler level - // because the predicate, its pure `_in` form and all three wrappers are - // gone — the handler is the lowest layer where the decision still exists - // to be observed. - // - // `OWNER` and `STRANGER` are names no hive has, so no `topology.json` on - // any machine running these tests can relate them. Any ancestry check - // reintroduced between the two — disk-reading or pure — therefore answers - // "unrelated", and every assertion below flips from the handler's effect - // to a refusal. - // ----------------------------------------------------------------------- - - /// Owner of the schedule under test. - const OWNER: &str = "schedule-owner-with-no-relations"; - /// Requester acting on it: not the owner, not the operator, and not - /// anybody's ancestor or descendant. - const STRANGER: &str = "requester-with-no-relations"; - - /// A real `Coordinator` over a throwaway sqlite dir. The socket-server - /// handlers take `&Arc`, so there is no lighter way in; - /// `open` touches nothing outside the db path it is handed. - pub(crate) fn coordinator() -> (tempfile::TempDir, Arc) { - let dir = tempfile::tempdir().expect("tempdir"); - let coord = Coordinator::open( - &dir.path().join("broker.sqlite"), - crate::coordinator::HiveEnv::default(), - crate::hive_stats::PriceTable::new(), - 1, - ) - .expect("open coordinator"); - (dir, Arc::new(coord)) - } - - /// Insert a schedule owned by [`OWNER`], returning its id. - fn submit_owned_by_other(coord: &Arc) -> i64 { - coord - .scheduled_prompts - .submit(&crate::scheduled_prompts::NewSchedule { - owner: OWNER.to_owned(), - targets: vec!["iris".to_owned()], - body: "ping".to_owned(), - first_fire_at_unix: 100, - interval_seconds: None, - description: None, - source: crate::scheduled_prompts::ScheduleSource::Operator, - }) - .expect("submit") - } - - #[test] - fn cancel_schedule_permits_unrelated_requester() { - let (_dir, coord) = coordinator(); - let id = submit_owned_by_other(&coord); - let resp = handle_cancel_schedule(&coord, STRANGER, id, None); - assert!( - matches!(resp, Response::Ok), - "unrelated requester must be allowed to cancel; got {resp:?}" - ); - // Ok alone could be a refusal that forgot to say so — check the row. - let row = coord - .scheduled_prompts - .get(id) - .expect("read back") - .expect("row present"); - assert!( - row.cancelled_at_unix.is_some(), - "cancel must have taken effect, not just answered Ok" - ); - } - - #[test] - fn edit_schedule_permits_unrelated_requester() { - let (_dir, coord) = coordinator(); - let id = submit_owned_by_other(&coord); - let resp = handle_edit_schedule( - &coord, - STRANGER, - id, - EditSchedulePatch { - body: Some("edited by a stranger".to_owned()), - description: None, - interval_seconds: None, - next_fire_at_unix: None, - targets_add: None, - targets_remove: None, - }, - ); - assert!( - matches!(resp, Response::Ok), - "unrelated requester must be allowed to edit; got {resp:?}" - ); - let row = coord - .scheduled_prompts - .get(id) - .expect("read back") - .expect("row present"); - assert_eq!(row.body, "edited by a stranger"); - } - - #[tokio::test] - async fn fire_schedule_now_permits_unrelated_requester() { - let (_dir, coord) = coordinator(); - let id = submit_owned_by_other(&coord); - let resp = handle_fire_schedule_now(&coord, STRANGER, id).await; - // The fan-out itself is allowed to find no live target (this test has - // no containers) — that is reported inside the report, not as an Err. - assert!( - matches!(resp, Response::Ok), - "unrelated requester must be allowed to fire; got {resp:?}" - ); - } - - #[test] - fn list_schedules_shows_a_schedule_owned_by_someone_else() { - let (_dir, coord) = coordinator(); - let id = submit_owned_by_other(&coord); - let Response::Schedules { schedules } = handle_list_schedules(&coord, STRANGER) else { - panic!("expected a Schedules response"); - }; - // The deleted filter dropped every row the requester was unrelated to, - // which for this requester is all of them. - assert!( - schedules.iter().any(|s| s.id == id && s.owner == OWNER), - "unrelated requester must see another agent's schedule; got {schedules:?}" - ); - } -} diff --git a/hive-c0re/src/stores/approvals.rs b/hive-c0re/src/stores/approvals.rs index 1d7dd361..c8b56c4e 100644 --- a/hive-c0re/src/stores/approvals.rs +++ b/hive-c0re/src/stores/approvals.rs @@ -1,10 +1,9 @@ -//! Approval queue. Requests are submitted by an agent -//! (`RequestSchedulePrompt`); the user approves/denies via the host admin -//! CLI; on approval the host runs the corresponding action. +//! Approval queue. The user approves/denies via the host admin CLI; on +//! approval the host runs the corresponding action. //! -//! `UpdateMetaInputs` rows are legacy: the MCP tool that queued them was -//! removed and nothing produces the kind any more. The variant and -//! its (de)serialization stay so pre-existing rows still read back. +//! Nothing submits approvals any more. `UpdateMetaInputs`, the only kind +//! left, is legacy: the MCP tool that queued it was removed. The variant +//! and its (de)serialization stay so pre-existing rows still read back. use std::path::Path; use std::sync::Mutex; @@ -64,6 +63,13 @@ const MIGRATIONS: &[Migration] = &[ sql: "DELETE FROM approvals WHERE kind = 'merge_config_pr'", adds_column: None, }, + // v6: drop `schedule_prompt` rows. Scheduled prompts are swarm-level, so + // no approval of that kind can be acted on, and `row_to_approval` rejects + // the kind. + Migration { + sql: "DELETE FROM approvals WHERE kind = 'schedule_prompt'", + adds_column: None, + }, ]; pub struct Approvals { @@ -81,7 +87,8 @@ impl Approvals { }) } - /// Insert a new pending approval row. + /// Test fixture: insert a new pending approval row. + #[cfg(test)] pub fn submit_kind( &self, agent: &str, @@ -354,7 +361,6 @@ fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result { let kind: String = row.get(2)?; let kind = match kind.as_str() { "update_meta_inputs" => ApprovalKind::UpdateMetaInputs, - "schedule_prompt" => ApprovalKind::SchedulePrompt, other => { return Err(rusqlite::Error::FromSqlConversionFailure( 2, @@ -404,7 +410,6 @@ fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result { fn kind_from_str(s: &str) -> Result { Ok(match s { "update_meta_inputs" => ApprovalKind::UpdateMetaInputs, - "schedule_prompt" => ApprovalKind::SchedulePrompt, other => bail!("unknown approval kind '{other}'"), }) } @@ -421,17 +426,6 @@ mod tests { (dir, path, db) } - #[test] - fn mixed_kinds_all_listed() { - let (_dir, _path, db) = open_temp(); - db.submit_kind("b", ApprovalKind::SchedulePrompt, "", None, "b") - .unwrap(); - db.submit_kind("c", ApprovalKind::UpdateMetaInputs, "[]", None, "c") - .unwrap(); - let pending = db.pending().expect("pending"); - assert_eq!(pending.len(), 2, "both kinds must be visible"); - } - #[test] fn mark_cancelled_transitions_pending_row() { // Manager withdraws a pending approval. Row leaves pending(), @@ -441,8 +435,8 @@ mod tests { let id = db .submit_kind( "bitburner", - ApprovalKind::SchedulePrompt, - "{}", + ApprovalKind::UpdateMetaInputs, + "[]", Some("test"), "bitburner", ) @@ -464,7 +458,7 @@ mod tests { // final — re-cancelling errors instead of silently overwriting. let (_dir, _path, db) = open_temp(); let id = db - .submit_kind("a", ApprovalKind::SchedulePrompt, "{}", None, "a") + .submit_kind("a", ApprovalKind::UpdateMetaInputs, "[]", None, "a") .unwrap(); db.mark_cancelled(id, "manager").expect("first cancel"); let err = db @@ -479,7 +473,7 @@ mod tests { // whole list — collect_lenient skips it instead of failing. let (_dir, path, db) = open_temp(); let good = db - .submit_kind("good", ApprovalKind::SchedulePrompt, "{}", None, "good") + .submit_kind("good", ApprovalKind::UpdateMetaInputs, "[]", None, "good") .unwrap(); let raw = Connection::open(&path).unwrap(); raw.execute( @@ -502,14 +496,20 @@ mod tests { // fall back to the root agent. let (_dir, path, db) = open_temp(); let id = db - .submit_kind("child", ApprovalKind::SchedulePrompt, "{}", None, "parent") + .submit_kind( + "child", + ApprovalKind::UpdateMetaInputs, + "[]", + None, + "parent", + ) .unwrap(); assert_eq!(db.submitter_of(id).unwrap().as_deref(), Some("parent")); let raw = Connection::open(&path).unwrap(); raw.execute( "INSERT INTO approvals (agent, kind, commit_ref, requested_at, status) - VALUES ('old', 'schedule_prompt', '', 0, 'pending')", + VALUES ('old', 'update_meta_inputs', '', 0, 'pending')", [], ) .unwrap(); @@ -524,7 +524,7 @@ mod tests { fn a_pre_v5_merge_config_pr_row_is_dropped_on_open() { let (_dir, path, db) = open_temp(); let kept = db - .submit_kind("iris", ApprovalKind::SchedulePrompt, "{}", None, "iris") + .submit_kind("iris", ApprovalKind::UpdateMetaInputs, "[]", None, "iris") .unwrap(); drop(db); let raw = Connection::open(&path).unwrap(); @@ -546,4 +546,30 @@ mod tests { assert!(db.recent_resolved(10).unwrap().is_empty()); assert!(db.get(old).unwrap().is_none()); } + + /// A database written before v6 still holds `schedule_prompt` rows. + /// Opening it drops them and keeps every other row. + #[test] + fn a_pre_v6_schedule_prompt_row_is_dropped_on_open() { + let (_dir, path, db) = open_temp(); + let kept = db + .submit_kind("iris", ApprovalKind::UpdateMetaInputs, "[]", None, "iris") + .unwrap(); + drop(db); + let raw = Connection::open(&path).unwrap(); + raw.execute_batch( + "INSERT INTO approvals (agent, kind, commit_ref, requested_at, status) + VALUES ('iris', 'schedule_prompt', '{}', 0, 'pending'); + UPDATE schema_versions SET version = 5 WHERE store = 'approvals';", + ) + .unwrap(); + let old: i64 = raw.last_insert_rowid(); + drop(raw); + + let db = Approvals::open(&path).expect("a pre-v6 db opens"); + let pending = db.pending().expect("pending"); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].id, kept); + assert!(db.get(old).unwrap().is_none()); + } } diff --git a/hive-c0re/src/stores/mod.rs b/hive-c0re/src/stores/mod.rs index fb05ed2a..000f3e03 100644 --- a/hive-c0re/src/stores/mod.rs +++ b/hive-c0re/src/stores/mod.rs @@ -1,5 +1,5 @@ -//! Sqlite-backed host-side stores (broker, approval / schedule queues, -//! build logs, power intent) plus the shared connection open/migration +//! Sqlite-backed host-side stores (broker, approval queue, read-only +//! scheduled prompts, build logs, power intent) plus the shared connection open/migration //! helper (`db`). Each submodule is re-exported at the crate root, so //! `crate::broker::…` etc. keep working unchanged. diff --git a/hive-c0re/src/stores/scheduled_prompts.rs b/hive-c0re/src/stores/scheduled_prompts.rs index 2e82488a..b65e1e9c 100644 --- a/hive-c0re/src/stores/scheduled_prompts.rs +++ b/hive-c0re/src/stores/scheduled_prompts.rs @@ -1,41 +1,21 @@ -//! Scheduled prompts: 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. +//! Scheduled prompts left in the hive's sqlite: the rows the dashboard's +//! read-only schedule list shows so the operator can recreate them with +//! `swarmctl schedule` (`docs/swarm/schedules.md`). Nothing on the hive +//! writes or fires these rows. //! -//! Schema + retention: `docs/agent-lifecycle/persistence.md::/var/lib/hyperhive/db/broker.sqlite` +//! Schema: `docs/agent-lifecycle/persistence.md::/var/lib/hyperhive/db/broker.sqlite` //! (the `scheduled_prompts` / `scheduled_prompt_targets` table bullets). -//! Submit paths (operator-direct vs `ApprovalKind::SchedulePrompt`, -//! plus why even agent-self schedules go through approval): -//! `docs/agent-lifecycle/approvals.md::Scheduled prompts (submit paths)`. -//! Catch-up clamp on resume + per-target tombstoning: -//! `docs/agent-lifecycle/approvals.md::Scheduled prompt worker (catch-up clamp)`. use std::path::Path; use std::sync::Mutex; -use anyhow::{Context, Result, bail}; +use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; -use rusqlite::{Connection, OptionalExtension, params}; +use rusqlite::{Connection, params}; use serde::{Deserialize, Serialize}; use crate::db::Migration; -/// Typed error returned by [`ScheduledPrompts::pause`] and -/// [`ScheduledPrompts::resume`] when the target row does not exist or -/// is already cancelled. Handlers downcast on this type to emit 404 -/// rather than 500, avoiding brittle string-matching on the message. -#[derive(Debug)] -pub struct ScheduleNotFoundOrCancelled(pub i64); - -impl std::fmt::Display for ScheduleNotFoundOrCancelled { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "schedule {} not found or is cancelled", self.0) - } -} -impl std::error::Error for ScheduleNotFoundOrCancelled {} - const SCHEMA: &str = r" CREATE TABLE IF NOT EXISTS scheduled_prompts ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -49,9 +29,6 @@ CREATE TABLE IF NOT EXISTS scheduled_prompts ( description TEXT, paused_at_unix INTEGER ); -CREATE INDEX IF NOT EXISTS idx_scheduled_due - ON scheduled_prompts (next_fire_at_unix) - WHERE cancelled_at_unix IS NULL AND paused_at_unix IS NULL; CREATE TABLE IF NOT EXISTS scheduled_prompt_targets ( schedule_id INTEGER NOT NULL, @@ -64,30 +41,22 @@ CREATE TABLE IF NOT EXISTS scheduled_prompt_targets ( ); "; -/// 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. +/// One scheduled-prompt row + its target set, as `list` returns it. #[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). + /// `"operator"` or the name of the agent whose approved request + /// created the row. pub owner: String, pub body: String, - /// `None` = one-shot, deleted after first fire. - /// `Some(n)` = recurring every `n` seconds. + /// `None` = one-shot. `Some(n)` = recurring every `n` seconds. pub interval_seconds: Option, pub next_fire_at_unix: DateTime, pub created_at_unix: DateTime, pub source: ScheduleSource, - /// Set when the *entire* schedule was cancelled (all targets - /// flipped, or operator cancel-all). Worker reaps these on the - /// next pass. + /// Set when the whole schedule was cancelled. pub cancelled_at_unix: Option>, pub description: Option, - /// Set while the schedule is paused. Worker skips rows where - /// `paused_at_unix IS NOT NULL`. Cleared by `resume()`. pub paused_at_unix: Option>, pub targets: Vec, } @@ -108,13 +77,6 @@ pub enum ScheduleSource { } 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::() @@ -127,54 +89,6 @@ impl ScheduleSource { } } -/// 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, -} - -/// Partial-update payload for `ScheduledPrompts::update`. -/// Every field is `Option<_>`; `None` keeps the existing value. -/// The doubly-wrapped `Option>` on `interval_seconds` -/// is intentional: outer `None` = "don't touch", outer -/// `Some(None)` = "set to NULL (toggle to one-shot)", outer -/// `Some(Some(n))` = "set to n seconds." Same encoding as JSON -/// "missing key" vs "explicit null" — the dashboard surface -/// preserves the distinction. -/// -/// Target add/remove: `targets_add` / `targets_remove` -/// run inside the same transaction as the scalar field updates so -/// "save my changes" is atomic. Remove delegates to the same -/// cancel-targets path used by `cancel_targets` (tombstoning, preserves -/// audit, auto-cancels parent when no active targets remain). Add uses -/// `INSERT OR REPLACE` so re-adding a previously-cancelled target -/// drops the tombstone and starts fresh (operator intent on re-add = -/// "this target is active again"; prior history was already visible -/// at cancel time). -#[derive(Debug, Clone, Default)] -#[allow( - clippy::option_option, - reason = "double-Option carries three-state PATCH semantics: outer None = \ - leave alone, Some(None) = clear, Some(Some(v)) = set. \ - collapsing to a single Option would lose the 'clear' state" -)] -pub struct UpdateSchedule { - pub body: Option, - pub description: Option>, - pub interval_seconds: Option>, - pub next_fire_at_unix: Option, - pub targets_add: Option>, - pub targets_remove: Option>, -} - pub struct ScheduledPrompts { conn: Mutex, } @@ -182,113 +96,27 @@ pub struct ScheduledPrompts { impl ScheduledPrompts { pub fn open(path: &Path) -> Result { let conn = crate::db::open(path, "scheduled_prompts")?; - // 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")?; - // Versioned migration: add paused_at_unix. The per-migration - // `adds_column` guard skips the ALTER on a legacy DB that already - // has the column (it was the only migration before versioning). + // The per-migration `adds_column` guard skips the ALTER on a legacy + // DB that already has the column. crate::db::apply_versioned_migrations( &conn, "scheduled_prompts", &[ - // v1: add paused_at_unix for per-schedule pause support. + // v1: add paused_at_unix. Migration { sql: "ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER", adds_column: Some(("scheduled_prompts", "paused_at_unix")), }, ], )?; - // Migration: recreate the due-rows index to also exclude paused - // rows. `CREATE INDEX IF NOT EXISTS` won't update an existing - // index's WHERE clause, so we drop + recreate on every open. - // The table is small and the op is cheap. - conn.execute_batch( - "DROP INDEX IF EXISTS idx_scheduled_due; - CREATE INDEX idx_scheduled_due - ON scheduled_prompts (next_fire_at_unix) - WHERE cancelled_at_unix IS NULL AND paused_at_unix IS NULL;", - ) - .context("scheduled_prompts: recreate due-index for paused support")?; 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. `operator` is rejected too - /// — it isn't a container with a todo inbox to push a wake into, - /// and scheduling a prompt at the operator herself doesn't fit the - /// "fan a prompt out to agent inboxes" shape this exists for. - pub fn submit(&self, new: &NewSchedule) -> Result { - if new.targets.is_empty() { - bail!("schedule must have at least one target"); - } - if new - .targets - .iter() - .any(|t| t == hive_sh4re::manager::OPERATOR_RECIPIENT) - { - bail!("operator is not a valid schedule 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, - Utc::now().timestamp(), - 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, - paused_at_unix - 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). + /// Every row, cancelled ones included, ordered by next fire time. pub fn list(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( @@ -307,358 +135,6 @@ impl ScheduledPrompts { } 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, - paused_at_unix - FROM scheduled_prompts - WHERE cancelled_at_unix IS NULL AND paused_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) - } - - /// Set a schedule's `next_fire_at` to an absolute timestamp. Unlike - /// [`rearm`](Self::rearm) (which advances along the existing cadence - /// with catch-up), this writes the value verbatim — used by the - /// manual fire-now "reset timer" path to re-arm a recurring schedule - /// from now (`now + interval`). - pub fn set_next_fire(&self, id: i64, next_unix: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2", - params![next_unix, id], - )?; - Ok(()) - } - - /// Partial-update an existing schedule's mutable fields. - /// Every scalar field is `Option<_>`; `None` means "leave the - /// existing value alone", `Some(_)` means "set it to this." - /// Refuses cancelled rows (no point editing a tombstone — - /// operator can just submit a new schedule). Refuses - /// `interval_seconds = Some(0)` — same rule as submit-time - /// validation. - /// - /// Targets are mutable via `targets_remove` + `targets_add`. - /// Both are processed in the same - /// transaction, with **removes before adds** so a single PATCH - /// can swap a target without ever leaving the schedule - /// target-less mid-tx. Remove writes a tombstone via - /// `COALESCE(cancelled_at_unix, ?1)` so it's idempotent. Add - /// uses `INSERT OR REPLACE` so re-adding a previously-removed - /// target drops the tombstone and resets per-target history - /// (fresh start — option C from the iris design thread). If - /// the post-tx active-target count is zero the parent schedule - /// is auto-cancelled — the worker would otherwise spin forever - /// firing nothing. - pub fn update(&self, id: i64, patch: UpdateSchedule) -> Result<()> { - // Pre-flight: row must exist and not be cancelled. - let mut conn = self.conn.lock().unwrap(); - let cancelled: Option = conn - .query_row( - "SELECT cancelled_at_unix FROM scheduled_prompts WHERE id = ?1", - params![id], - |row| row.get(0), - ) - .optional()? - .ok_or_else(|| anyhow::anyhow!("schedule {id} not found"))?; - if cancelled.is_some() { - bail!("schedule {id} is cancelled — submit a new one"); - } - if let Some(Some(0)) = patch.interval_seconds { - bail!("interval_seconds must be > 0 (use None for one-shot)"); - } - if patch.targets_add.as_deref().is_some_and(|list| { - list.iter() - .any(|t| t == hive_sh4re::manager::OPERATOR_RECIPIENT) - }) { - bail!("operator is not a valid schedule target"); - } - // Single transaction so a partial failure can't leave a - // half-updated row. - let tx = conn.transaction()?; - if let Some(body) = patch.body { - tx.execute( - "UPDATE scheduled_prompts SET body = ?1 WHERE id = ?2", - params![body, id], - )?; - } - if let Some(description) = patch.description { - tx.execute( - "UPDATE scheduled_prompts SET description = ?1 WHERE id = ?2", - params![description, id], - )?; - } - if let Some(interval) = patch.interval_seconds { - // `Some(Some(n))` = set to n. `Some(None)` = clear (one-shot). - // `None` is filtered out by the outer `if let`. - let value: Option = interval.map(i64::try_from).and_then(Result::ok); - tx.execute( - "UPDATE scheduled_prompts SET interval_seconds = ?1 WHERE id = ?2", - params![value, id], - )?; - } - if let Some(next) = patch.next_fire_at_unix { - tx.execute( - "UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2", - params![next, id], - )?; - } - // Target removals first, then adds. Order matters: an - // operator who sends `{add: ["alice"], remove: ["alice"]}` - // probably typo'd, but treating add-after-remove as the - // surviving intent means a clean re-arm rather than an - // orphaned tombstone. Plus removing-then-adding is the - // natural way to "restart history" on one target without - // a separate flow. - let now = Utc::now().timestamp(); - if let Some(remove) = patch.targets_remove.as_deref() { - for target in remove { - 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 let Some(add) = patch.targets_add.as_deref() { - for target in add { - // INSERT OR REPLACE drops any prior row (cancelled - // tombstone or otherwise) and inserts a fresh one - // with empty history. Operator intent on re-add = - // "this target is active again, fresh start." - tx.execute( - "INSERT OR REPLACE INTO scheduled_prompt_targets - (schedule_id, target, cancelled_at_unix, - last_fired_at_unix, last_result) - VALUES (?1, ?2, NULL, NULL, NULL)", - params![id, target], - )?; - } - } - // If the remove set drained the last active target (and - // the add set didn't bring any back), auto-cancel the - // parent so the worker stops scanning it. Mirrors - // `cancel_targets`'s behaviour for the same edge case. - if patch.targets_remove.is_some() || patch.targets_add.is_some() { - 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(()) - } - - /// 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 = Utc::now().timestamp(); - 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 = Utc::now().timestamp(); - 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(()) - } - - /// Pause a schedule. Idempotent; no-op on an already-paused row. - /// The worker skips paused rows so no fires occur while paused; - /// `next_fire_at_unix` is preserved so the first resume fires at - /// the next intended cadence instant (no catch-up needed — a - /// paused schedule simply slips its upcoming fire). Returns - /// `Err(ScheduleNotFoundOrCancelled)` when the schedule is - /// cancelled or does not exist (handlers downcast to emit 404). - pub fn pause(&self, id: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); - let now = Utc::now().timestamp(); - let n = conn.execute( - "UPDATE scheduled_prompts - SET paused_at_unix = COALESCE(paused_at_unix, ?1) - WHERE id = ?2 AND cancelled_at_unix IS NULL", - params![now, id], - )?; - if n == 0 { - return Err(ScheduleNotFoundOrCancelled(id).into()); - } - Ok(()) - } - - /// Resume a paused schedule. Idempotent; no-op on an active row. - /// Returns `Err(ScheduleNotFoundOrCancelled)` when the schedule is - /// cancelled or does not exist (handlers downcast to emit 404). - pub fn resume(&self, id: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); - let n = conn.execute( - "UPDATE scheduled_prompts - SET paused_at_unix = NULL - WHERE id = ?1 AND cancelled_at_unix IS NULL", - params![id], - )?; - if n == 0 { - return Err(ScheduleNotFoundOrCancelled(id).into()); - } - 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 { @@ -714,509 +190,36 @@ mod tests { use super::*; use tempfile::TempDir; - fn open() -> (TempDir, ScheduledPrompts) { + #[test] + fn list_returns_rows_with_their_targets() { 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 submit_rejects_operator_as_a_target() { - let (_dir, db) = open(); - let err = db - .submit(&NewSchedule { - owner: "operator".into(), - targets: vec!["operator".into()], - body: "wake".into(), - first_fire_at_unix: 100, - interval_seconds: None, - description: None, - source: ScheduleSource::Operator, - }) - .unwrap_err(); - assert!(format!("{err:#}").contains("operator is not a valid schedule target")); - } - - #[test] - fn update_rejects_operator_in_targets_add() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - let err = db - .update( - id, - UpdateSchedule { - targets_add: Some(vec!["operator".into()]), - ..Default::default() - }, - ) - .unwrap_err(); - assert!(format!("{err:#}").contains("operator is not a valid schedule 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.timestamp(), 460); - } - - #[test] - fn set_next_fire_writes_absolute_value() { - let (_dir, db) = open(); - // Recurring every 60s, first 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"); - // Unlike rearm (cadence-aligned), set_next_fire writes verbatim — - // the manual fire-now "reset timer" path uses now + interval. - db.set_next_fire(id, 1_234).expect("set_next_fire"); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix.timestamp(), 1_234); - } - - #[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.timestamp(), 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.timestamp(), 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.map(|dt| dt.timestamp()), Some(200)); - assert_eq!(bob.last_result.as_deref(), Some("ok")); - } - - #[test] - fn update_partial_only_touches_set_fields() { - let (_dir, db) = open(); - let id = db - .submit(&NewSchedule { - owner: "operator".into(), - targets: vec!["alice".into()], - body: "old body".into(), - first_fire_at_unix: 100, - interval_seconds: Some(60), - description: Some("old desc".into()), - source: ScheduleSource::Operator, - }) - .expect("submit"); - // Only change body — everything else must stay. - db.update( - id, - UpdateSchedule { - body: Some("new body".into()), - ..Default::default() - }, + let raw = Connection::open(&path).expect("raw open"); + raw.execute_batch( + "INSERT INTO scheduled_prompts + (id, owner, body, interval_seconds, next_fire_at_unix, + created_at_unix, source, description) + VALUES (1, 'operator', 'later', NULL, 200, 10, 'operator', NULL), + (2, 'iris', 'hourly', 3600, 100, 10, 'approval:7', 'tick'); + INSERT INTO scheduled_prompt_targets (schedule_id, target, cancelled_at_unix) + VALUES (1, 'bob', NULL), (2, 'iris', NULL), (2, 'alice', 50);", ) - .expect("update body"); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.body, "new body"); - assert_eq!(s.description.as_deref(), Some("old desc")); - assert_eq!(s.interval_seconds, Some(60)); - assert_eq!(s.next_fire_at_unix.timestamp(), 100); - } + .expect("seed"); - #[test] - fn update_interval_toggle_recurring_to_one_shot() { - let (_dir, db) = open(); - let id = db - .submit(&NewSchedule { - owner: "operator".into(), - targets: vec!["alice".into()], - body: "x".into(), - first_fire_at_unix: 100, - interval_seconds: Some(60), - description: None, - source: ScheduleSource::Operator, - }) - .expect("submit"); - db.update( - id, - UpdateSchedule { - interval_seconds: Some(None), - ..Default::default() - }, - ) - .expect("update"); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.interval_seconds, None); - } - - #[test] - fn update_interval_toggle_one_shot_to_recurring() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - db.update( - id, - UpdateSchedule { - interval_seconds: Some(Some(3600)), - ..Default::default() - }, - ) - .expect("update"); - let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.interval_seconds, Some(3600)); - } - - #[test] - fn update_rejects_cancelled_row() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - db.cancel_all(id).expect("cancel"); - let err = db - .update( - id, - UpdateSchedule { - body: Some("nope".into()), - ..Default::default() - }, - ) - .unwrap_err(); - assert!(format!("{err:#}").contains("cancelled")); - } - - #[test] - fn update_rejects_zero_interval() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - let err = db - .update( - id, - UpdateSchedule { - interval_seconds: Some(Some(0)), - ..Default::default() - }, - ) - .unwrap_err(); - assert!(format!("{err:#}").contains("interval_seconds")); - } - - #[test] - fn update_clears_description() { - let (_dir, db) = open(); - let id = db - .submit(&NewSchedule { - owner: "operator".into(), - targets: vec!["alice".into()], - body: "x".into(), - first_fire_at_unix: 100, - interval_seconds: None, - description: Some("first".into()), - source: ScheduleSource::Operator, - }) - .expect("submit"); - db.update( - id, - UpdateSchedule { - description: Some(None), - ..Default::default() - }, - ) - .expect("update"); - let s = db.get(id).expect("get").expect("present"); - assert!(s.description.is_none()); - } - - #[test] - fn update_adds_new_targets() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - db.update( - id, - UpdateSchedule { - targets_add: Some(vec!["bob".into(), "carol".into()]), - ..Default::default() - }, - ) - .expect("update"); - let s = db.get(id).expect("get").expect("present"); - let names: Vec<_> = s.targets.iter().map(|t| t.target.as_str()).collect(); - assert!(names.contains(&"alice")); - assert!(names.contains(&"bob")); - assert!(names.contains(&"carol")); - } - - #[test] - fn update_removes_targets_via_cancel_tombstone() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice", "bob"]); - db.update( - id, - UpdateSchedule { - targets_remove: Some(vec!["alice".into()]), - ..Default::default() - }, - ) - .expect("update"); - 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(); - // alice tombstoned, bob still active, parent still active. - assert!(alice.cancelled_at_unix.is_some()); - assert!(bob.cancelled_at_unix.is_none()); - assert!(s.cancelled_at_unix.is_none()); - } - - #[test] - fn update_removing_last_target_auto_cancels_parent() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - db.update( - id, - UpdateSchedule { - targets_remove: Some(vec!["alice".into()]), - ..Default::default() - }, - ) - .expect("update"); - let s = db.get(id).expect("get").expect("present"); - assert!(s.cancelled_at_unix.is_some()); - } - - #[test] - fn update_re_add_resurrects_cancelled_target_fresh() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice", "bob"]); - // Record some history on alice, then cancel her. - db.record_target_result(id, "alice", 50, "ok") - .expect("record"); - db.update( - id, - UpdateSchedule { - targets_remove: Some(vec!["alice".into()]), - ..Default::default() - }, - ) - .expect("cancel alice"); - // Re-add alice — should drop the tombstone + prior history. - db.update( - id, - UpdateSchedule { - targets_add: Some(vec!["alice".into()]), - ..Default::default() - }, - ) - .expect("re-add alice"); - let s = db.get(id).expect("get").expect("present"); - let alice = s.targets.iter().find(|t| t.target == "alice").unwrap(); - assert!(alice.cancelled_at_unix.is_none(), "fresh row, no tombstone"); - assert!(alice.last_fired_at_unix.is_none(), "history reset"); - assert!(alice.last_result.is_none(), "history reset"); - } - - #[test] - fn update_add_and_remove_in_one_transaction() { - let (_dir, db) = open(); - let id = submit_one_shot(&db, 100, &["alice"]); - db.update( - id, - UpdateSchedule { - targets_remove: Some(vec!["alice".into()]), - targets_add: Some(vec!["bob".into()]), - ..Default::default() - }, - ) - .expect("update"); - 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(); - assert!(alice.cancelled_at_unix.is_some()); - assert!(bob.cancelled_at_unix.is_none()); - // Bob arrived AFTER alice was tombstoned in the same tx — - // active-target count > 0, so parent stays alive. - assert!(s.cancelled_at_unix.is_none()); - } - - #[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()); + let rows = db.list().expect("list"); + assert_eq!(rows.iter().map(|s| s.id).collect::>(), vec![2, 1]); + let hourly = &rows[0]; + assert_eq!(hourly.interval_seconds, Some(3600)); + assert_eq!(hourly.source, ScheduleSource::Approval { id: 7 }); + assert_eq!(hourly.description.as_deref(), Some("tick")); + let targets: Vec<(&str, bool)> = hourly + .targets + .iter() + .map(|t| (t.target.as_str(), t.cancelled_at_unix.is_some())) + .collect(); + assert_eq!(targets, vec![("alice", true), ("iris", false)]); + assert_eq!(rows[1].source, ScheduleSource::Operator); + assert_eq!(rows[1].interval_seconds, None); } } diff --git a/hive-c0re/src/workers/mod.rs b/hive-c0re/src/workers/mod.rs index da62baaa..c3772e92 100644 --- a/hive-c0re/src/workers/mod.rs +++ b/hive-c0re/src/workers/mod.rs @@ -1,6 +1,5 @@ -//! Background tasks and periodic sweeps: crash/login watcher, the -//! scheduled-prompt delivery loop, boot-time auto-update -//! reconcile, the agent-sockets.json writer loop, the MCP socket listener +//! Background tasks and periodic sweeps: crash/login watcher, boot-time +//! auto-update reconcile, the agent-sockets.json writer loop, the MCP socket listener //! reconcile loop, knowledge-repo sync, and the swarm wanted-state pull. //! Each submodule is re-exported //! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged. @@ -10,5 +9,4 @@ pub mod auto_update; pub mod crash_watch; pub mod knowledge; pub mod mcp_sockets; -pub mod scheduled_prompts_worker; pub mod wanted; diff --git a/hive-c0re/src/workers/scheduled_prompts_worker.rs b/hive-c0re/src/workers/scheduled_prompts_worker.rs deleted file mode 100644 index ce988735..00000000 --- a/hive-c0re/src/workers/scheduled_prompts_worker.rs +++ /dev/null @@ -1,386 +0,0 @@ -//! Background loop that drains due `scheduled_prompts` rows and fans -//! the body to each active target. 5s poll cadence, shutdown-aware. -//! Catch-up clamp, missing-target handling, and broker-error retry -//! semantics: `docs/agent-lifecycle/approvals.md::Scheduled prompt worker`. - -use std::sync::Arc; -use std::time::Duration; - -use chrono::Utc; -use hive_sh4re::inbox::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_hours(1); - -pub fn spawn(coord: Arc) { - let mut shutdown = coord.shutdown_rx(); - tokio::spawn(async move { - loop { - tick(&coord).await; - tokio::select! { - () = tokio::time::sleep(POLL_INTERVAL) => {} - _ = shutdown.changed() => { - tracing::info!("scheduled_prompts worker: shutdown signal received"); - break; - } - } - } - }); -} - -async fn tick(coord: &Arc) { - let now = Utc::now().timestamp(); - 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).await; - } - let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0); - if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) { - tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed"); - } - // Emit after all fires + reaps so the dashboard reflects updated - // last_fired_at_unix, next_fire_at_unix, and any reaped one-shots. - coord.emit_schedules_snapshot(); -} - -/// 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, keyed on schedule identity rather than on -/// the body happening to be byte-identical. `deliver_to_target` also -/// passes `reopen_if_acked = true`, so this collapse only applies while -/// the previous fire is genuinely still sitting there un-reviewed — once -/// the target acks it, the *next* fire reopens the row and wakes again -/// even with an identical body, instead of silently staying quiet -/// forever — a recurring schedule with a static body used to wake its -/// target once, ever, and then look dead. -async fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { - let known: std::collections::HashSet = known_agents().await; - for target_row in &schedule.targets { - if target_row.cancelled_at_unix.is_some() { - continue; - } - let target = &target_row.target; - if !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 result_str = match deliver_to_target(coord, schedule.id, target, &schedule.body).await { - Ok(()) => "ok".to_owned(), - Err(reason) => { - tracing::warn!( - schedule = schedule.id, - %target, - %reason, - "scheduled_prompts: delivery failed (will retry on next interval)" - ); - reason - } - }; - 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"); - } -} - -/// Deliver `body` to a single already-known-live `target` via -/// `push_todo`, 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 — but `reopen_if_acked = true` means that collapse only holds -/// while unreviewed; once acked, the next fire reopens regardless of -/// whether the body changed. Shared by the periodic `fire_schedule` -/// tick and the manual `fire_now` dashboard action — the only -/// difference between them is what each caller does with the `Result` -/// (log/prefix and per-target `last_result`/`FireNowReport` -/// bookkeeping), not the delivery choice itself. -async fn deliver_to_target( - coord: &Arc, - schedule_id: i64, - target: &str, - body: &str, -) -> Result<(), String> { - coord - .push_todo( - target, - "schedule", - Some(format!("schedule:{schedule_id}")), - body.to_owned(), - Some("scheduled".to_owned()), - // Each fire is a distinct occurrence, not a restatement of - // a persisting condition — an agent that already acked the - // *previous* firing hasn't acked *this* one, so an acked - // row must reopen even when the body is byte-identical - // (the common case: most schedules don't vary their text - // per fire). See `Todos::upsert`'s doc comment for the - // full reconciler-vs-event rationale. - true, - ) - .await -} - -/// 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: hive_sh4re::manager::trusted_sender("scheduled"), - to: hive_sh4re::manager::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"); - } -} - -/// Per-target outcome counts for one `fire_now` invocation. -/// Returned to the operator so the dashboard can render -/// "fired to N (M failed, K missing)" without a follow-up GET. -#[derive(Debug, Clone, serde::Serialize, utoipa::ToSchema)] -pub struct FireNowReport { - /// Targets the broker accepted the message for. - pub ok: u32, - /// Targets where broker.send returned an error. - pub failed: u32, - /// Targets that didn't resolve to a known agent (and got the - /// operator-advisory treatment). - pub missing: u32, - /// Whether the one-shot was consumed by this manual fire. - /// `true` only when the schedule was a one-shot (recurring - /// schedules never auto-cancel on manual fire — they keep - /// their cadence). - pub one_shot_consumed: bool, - /// Whether this manual fire re-armed a recurring schedule's timer - /// (`next_fire_at = now + interval`). `true` only when the caller - /// passed `reset_timer` AND the schedule is recurring. - pub timer_reset: bool, -} - -/// Manual / out-of-band fire of a scheduled prompt ("fire now" -/// dashboard button). Mirrors the per-target fan-out of `fire_schedule` -/// but skips the rearm step entirely — manual fires don't disturb -/// a recurring schedule's rhythm. For one-shots, a manual fire -/// **consumes** the schedule (operator intent: "send this now, -/// the scheduled time was wrong"); recurring schedules keep their -/// `next_fire_at_unix` unchanged. -/// -/// `last_result` is annotated with the `manual fire:` prefix so -/// the dashboard's per-target last-result column can distinguish -/// scheduled fires from operator-initiated ones at a glance. -/// -/// `reset_timer` re-arms a *recurring* schedule's countdown from now -/// (`next_fire_at = now + interval`) after the fan-out — the dashboard -/// fire-now dialog's "reset timer" checkbox. It's a no-op for one-shots -/// (still consumed) and when `false` (today's default: cadence intact). -/// -/// Returns Err if the schedule is missing, cancelled, or fully -/// drained of active targets — the dashboard can surface those -/// as plain 4xxs instead of pretending to fire a phantom row. -pub async fn fire_now( - coord: &std::sync::Arc, - schedule_id: i64, - reset_timer: bool, -) -> anyhow::Result { - let now = Utc::now().timestamp(); - let schedule = coord - .scheduled_prompts - .get(schedule_id)? - .ok_or_else(|| anyhow::anyhow!("schedule {schedule_id} not found"))?; - if schedule.cancelled_at_unix.is_some() { - anyhow::bail!("schedule {schedule_id} is already cancelled"); - } - if !schedule - .targets - .iter() - .any(|t| t.cancelled_at_unix.is_none()) - { - anyhow::bail!("schedule {schedule_id} has no active targets"); - } - let known = known_agents().await; - let mut report = FireNowReport { - ok: 0, - failed: 0, - missing: 0, - one_shot_consumed: false, - timer_reset: false, - }; - for target_row in &schedule.targets { - if target_row.cancelled_at_unix.is_some() { - continue; - } - let target = &target_row.target; - if !known.contains(target) { - let reason = format!("manual fire: 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); - report.missing += 1; - continue; - } - let result_str = match deliver_to_target(coord, schedule_id, target, &schedule.body).await { - Ok(()) => { - report.ok += 1; - "manual fire: ok".to_owned() - } - Err(reason) => { - report.failed += 1; - tracing::warn!( - schedule = schedule_id, - %target, - %reason, - "fire_now: delivery failed (no retry — manual fires don't loop)" - ); - format!("manual fire: {reason}") - } - }; - 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"); - } - } - match schedule.interval_seconds { - None => { - // One-shot is consumed by the manual fire. (reset_timer is - // moot here — there's no recurring cadence to re-arm.) - if let Err(e) = coord.scheduled_prompts.cancel_all(schedule_id) { - tracing::warn!(error = ?e, schedule = schedule_id, "cancel_all after one-shot manual fire failed"); - } else { - report.one_shot_consumed = true; - } - } - Some(interval) if reset_timer => { - // Recurring + operator asked to reset: re-arm the countdown - // from now (now + interval), not along the existing cadence. - let next = now.saturating_add(i64::try_from(interval).unwrap_or(i64::MAX)); - if let Err(e) = coord.scheduled_prompts.set_next_fire(schedule_id, next) { - tracing::warn!(error = ?e, schedule = schedule_id, "set_next_fire after manual fire reset failed"); - } else { - report.timer_reset = true; - } - } - // Recurring without reset: cadence stays intact (additive fire). - Some(_) => {} - } - 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-manager target looks missing -/// this tick and gets the same treatment as a genuinely-destroyed -/// agent: operator advisory + per-target `last_result` annotation + -/// skipped delivery. Recurring -/// schedules recover automatically on the next tick (the lifecycle -/// listing usually works); one-shots that land on this window -/// lose their single delivery. Logged at `warn`, not propagated. -/// Shared by `fire_schedule` and `fire_now` — both are async now -/// (the sync `fire_schedule` used to need a `block_in_place` variant -/// of this before it started `push_todo`ing, which is itself async). -async fn known_agents() -> std::collections::HashSet { - use std::collections::HashSet; - let mut out: HashSet = HashSet::new(); - out.insert(hive_sh4re::manager::MANAGER_AGENT.to_owned()); - match crate::lifecycle::list().await { - 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, "fire_now: container listing failed"); - } - } - out -} diff --git a/hive-core-agent-sock/src/lib.rs b/hive-core-agent-sock/src/lib.rs index 6c047ed1..4e67629a 100644 --- a/hive-core-agent-sock/src/lib.rs +++ b/hive-core-agent-sock/src/lib.rs @@ -9,8 +9,6 @@ use hive_sh4re::container::MatrixIdentity; use hive_sh4re::inbox::{CancelLooseEndKind, DeliveredMessage, InboxRow, LooseEnd}; -use hive_sh4re::manager::SchedulePromptPayload; -use hive_sh4re::schedule::WireSchedule; use hive_types::Ident; use serde::{Deserialize, Serialize}; @@ -20,8 +18,7 @@ fn default_true() -> bool { } /// Unified request enum for both agent and manager sockets. The agent's -/// identity is the socket it arrived on. Privileged variants are marked -/// `*(privileged)*` — an agent socket returns `Err` for them server-side. +/// identity is the socket it arrived on. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "cmd", rename_all = "snake_case")] pub enum Request { @@ -111,40 +108,9 @@ pub enum Request { /// DAG's drain node resolve immediately instead of waiting out its /// timeout fallback — same shape as `GracefulStopComplete`. PauseAcknowledged, - - // ---- privileged (manager socket only for now) --------------------------- - /// *(privileged)* Queue an approval to add a scheduled prompt. - RequestSchedulePrompt(SchedulePromptPayload), - /// *(privileged)* Cancel a scheduled prompt. - CancelSchedule { - id: i64, - #[serde(default, skip_serializing_if = "Option::is_none")] - targets: Option>, - }, - /// List every schedule in the hive, whoever owns it. - ListSchedules, - /// *(privileged)* Fire a scheduled prompt out of band immediately. - FireScheduleNow { id: i64 }, - /// *(privileged)* Edit an existing schedule's mutable fields. - EditSchedule { - id: i64, - #[serde(default, skip_serializing_if = "Option::is_none")] - body: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - description: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - interval_seconds: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - next_fire_at_unix: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - targets_add: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - targets_remove: Option>, - }, } -/// Unified response enum for both agent and manager sockets. Privileged -/// variants (`Schedules`) are never returned on agent sockets. +/// Unified response enum for both agent and manager sockets. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum Response { @@ -211,10 +177,6 @@ pub enum Response { #[serde(default, skip_serializing_if = "Vec::is_empty")] matrix_accounts: Vec, }, - /// `ListSchedules` result. Snapshot of the schedules the requester is - /// authorized to see — see `ListSchedules`'s own doc comment. - /// Returned on the manager socket only. - Schedules { schedules: Vec }, /// `Recv` result when a graceful stop is pending for this agent /// (set by hive-c0re's `GracefulStop` orchestration). Returned in /// place of `Messages` — it doubles as the inbound fence: the harness diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 44bb1026..442cf788 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -76,20 +76,6 @@ pub fn agent_web_socket(name: &Ident) -> PathBuf { .join("web.sock") } -/// Per-agent in-container todo socket path — -/// `AGENT_SOCKET_DIR//agent.sock`. Same socket the harness's -/// `HIVE_AGENT_SOCKET` binds and in-container producers (matrix/bash/ -/// forge-notify) already dial — `hive-c0re` uses this to push a todo -/// directly into a *running* agent's local store (see -/// `Coordinator::push_todo`); the container being down (socket absent) -/// means the push is simply skipped, not queued. -#[must_use] -pub fn agent_todo_socket(name: &Ident) -> PathBuf { - PathBuf::from(AGENT_SOCKET_DIR) - .join(name.as_str()) - .join("agent.sock") -} - /// nspawn machine-name prefix for agent containers (`h-`). A single /// `starts_with(AGENT_PREFIX)` filter enumerates managed containers. pub const AGENT_PREFIX: &str = "h-"; diff --git a/hive-sh4re/src/approvals.rs b/hive-sh4re/src/approvals.rs index 14f75475..d35a4f79 100644 --- a/hive-sh4re/src/approvals.rs +++ b/hive-sh4re/src/approvals.rs @@ -15,8 +15,8 @@ pub struct Approval { pub id: i64, pub agent: Ident, pub kind: ApprovalKind, - /// Kind-specific payload (git sha / inputs array / schedule - /// payload / empty). See the Approval struct doc. + /// Kind-specific payload (git sha / inputs array / empty). See the + /// Approval struct doc. pub commit_ref: String, pub requested_at: DateTime, pub status: ApprovalStatus, @@ -40,8 +40,6 @@ pub enum ApprovalKind { /// Run `nix flake update [inputs...]` on the meta flake and commit /// the resulting lock changes. UpdateMetaInputs, - /// Add a scheduled prompt to the broker queue. - SchedulePrompt, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] diff --git a/hive-sh4re/src/manager.rs b/hive-sh4re/src/manager.rs index df82c18b..ccccf6ee 100644 --- a/hive-sh4re/src/manager.rs +++ b/hive-sh4re/src/manager.rs @@ -1,8 +1,7 @@ //! Manager socket — `/run/hyperhive/manager/mcp.sock` on the host, //! bind-mounted into the manager container at `/run/hive/mcp.sock`. -//! Reserved recipient names/senders, the out-of-band `HelperEvent` -//! payload hive-c0re pushes into the manager's inbox, and the -//! schedule-prompt submission payload. +//! Reserved recipient names/senders and the out-of-band `HelperEvent` +//! payload hive-c0re pushes into the manager's inbox. use hive_types::Ident; use serde::{Deserialize, Serialize}; @@ -66,30 +65,6 @@ pub enum HelperEvent { }, } -/// 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 + `root` 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, -} - #[cfg(test)] mod reserved_name_tests { use super::{MANAGER_AGENT, OPERATOR_RECIPIENT, SYSTEM_SENDER}; diff --git a/hive-sh4re/src/permissions.rs b/hive-sh4re/src/permissions.rs index 51781755..3e1213ff 100644 --- a/hive-sh4re/src/permissions.rs +++ b/hive-sh4re/src/permissions.rs @@ -162,9 +162,6 @@ pub enum ToolGroup { /// server-side gate: `cancel_loose_end`'s approval-cancel arm /// requires it (`socket_server::check_can_cancel_approval`). Approvals, - /// `request_schedule_prompt`, `fire_schedule_now`, `cancel_schedule`, - /// `edit_schedule`, `list_schedules` - *(privileged)* - Scheduling, /// Gates no tool today — `create_repo`, its only member, was removed /// with no replacement. Kept so existing `meta/capabilities.json` /// grants still parse; `tools()` returns `&[]`. @@ -230,13 +227,6 @@ impl ToolGroup { Self::Messaging => &["send", "recv", "ack_until"], Self::Meta => &["get_agent_meta"], Self::Inbox => &["get_loose_ends", "cancel_loose_end", "remind"], - Self::Scheduling => &[ - "request_schedule_prompt", - "fire_schedule_now", - "cancel_schedule", - "edit_schedule", - "list_schedules", - ], // All five empty, for different reasons — see each // variant's own doc comment above. `Execution` grants the // out-of-process `bash` MCP server @@ -308,7 +298,6 @@ impl ToolGroup { Self::Inbox, Self::Lifecycle, Self::Approvals, - Self::Scheduling, Self::Execution, ]; @@ -321,7 +310,6 @@ impl ToolGroup { Self::Inbox, Self::Lifecycle, Self::Approvals, - Self::Scheduling, Self::Forge, Self::Execution, Self::WebTools, @@ -340,9 +328,6 @@ impl ToolGroup { Self::Approvals => { "no tools — grants cancel_loose_end's approval-cancel arm (privileged)" } - Self::Scheduling => { - "request_schedule_prompt and related — operator-visible scheduled prompts (privileged)" - } Self::Forge => "no tools — vestigial since create_repo was removed", Self::Execution => { "run, status — run shell commands via mcp__bash__run / mcp__bash__status" diff --git a/hive-sh4re/src/schedule.rs b/hive-sh4re/src/schedule.rs index 7ba71128..7d23d1d3 100644 --- a/hive-sh4re/src/schedule.rs +++ b/hive-sh4re/src/schedule.rs @@ -1,15 +1,14 @@ -//! Scheduled-prompt row shape on the wire, shared by the dashboard and -//! agent surfaces (mirrors `hive-c0re`'s internal `scheduled_prompts::Schedule`). +//! Scheduled-prompt row shape on the wire, served by the dashboard's +//! read-only schedule list (mirrors `hive-c0re`'s internal `scheduled_prompts::Schedule`). use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; /// Schedule row shape on the wire — mirror of -/// `scheduled_prompts::Schedule` but in the public crate so the -/// dashboard and 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`. +/// `scheduled_prompts::Schedule` but in the public crate so a client +/// can deserialize it without depending on hive-c0re-internal types. +/// Kept structurally identical to the in-process type; the conversion +/// is field-by-field in the dashboard. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WireSchedule { pub id: i64, @@ -22,10 +21,7 @@ pub struct WireSchedule { pub source: WireScheduleSource, #[serde(default, skip_serializing_if = "Option::is_none")] pub cancelled_at_unix: Option>, - /// Set while the schedule is paused. Worker skips paused rows; - /// they keep their `next_fire_at_unix` so resuming at any time - /// fires at the next intended instant (no catch-up clamp needed - /// — a paused schedule simply slips its next fire). + /// Set if the row is paused. #[serde(default, skip_serializing_if = "Option::is_none")] pub paused_at_unix: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/hive-sh4re/src/stream_enrich.rs b/hive-sh4re/src/stream_enrich.rs index 0b5fbe70..0cb709e3 100644 --- a/hive-sh4re/src/stream_enrich.rs +++ b/hive-sh4re/src/stream_enrich.rs @@ -645,13 +645,6 @@ fn fmt_hyperhive_tool(name: &str, short: &str, input: &Value) -> String { format!("{short} ≤{up_to}") } "mcp__hyperhive__remind" => fmt_hyperhive_remind(short, input), - "mcp__hyperhive__list_schedules" - | "mcp__hyperhive__cancel_schedule" - | "mcp__hyperhive__fire_schedule_now" - | "mcp__hyperhive__edit_schedule" - | "mcp__hyperhive__request_schedule_prompt" => { - fmt_hyperhive_schedule_tool(name, short, input) - } "mcp__hyperhive__set_status" => { format!("{short} \"{}\"", trim_str(&sv(input, "text"), 60)) } @@ -739,91 +732,6 @@ fn fmt_hyperhive_remind(short: &str, input: &Value) -> String { format!("{short}{when_part}{msg_part}") } -fn fmt_hyperhive_edit_schedule(short: &str, input: &Value) -> String { - let id = input - .get("id") - .and_then(Value::as_u64) - .map_or_else(|| "?".to_owned(), |n| n.to_string()); - let mut parts = vec![format!("#{id}")]; - if input.get("body").is_some() { - parts.push("body".to_owned()); - } - if input.get("interval_seconds").is_some() { - parts.push("interval".to_owned()); - } - if input.get("next_fire_at_unix").is_some() { - parts.push("next".to_owned()); - } - if let Some(a) = input.get("targets_add").and_then(Value::as_array) - && !a.is_empty() - { - parts.push(format!("+{} tgt", a.len())); - } - if let Some(r) = input.get("targets_remove").and_then(Value::as_array) - && !r.is_empty() - { - parts.push(format!("-{} tgt", r.len())); - } - format!("{short} {}", parts.join(" · ")) -} - -/// Schedule-management hyperhive tools (list/cancel/fire/edit/request). -fn fmt_hyperhive_schedule_tool(name: &str, short: &str, input: &Value) -> String { - match name { - "mcp__hyperhive__list_schedules" => format!("{short}()"), - "mcp__hyperhive__cancel_schedule" => { - let id = input - .get("id") - .and_then(Value::as_u64) - .map_or_else(|| "?".to_owned(), |n| n.to_string()); - let tgts = match input.get("targets").and_then(Value::as_array) { - Some(arr) if !arr.is_empty() => { - let names: Vec<&str> = arr.iter().filter_map(Value::as_str).collect(); - format!(" [{}]", names.join(", ")) - } - _ => " all".to_owned(), - }; - format!("{short} #{id}{tgts}") - } - "mcp__hyperhive__fire_schedule_now" => { - let id = input - .get("id") - .and_then(Value::as_u64) - .map_or_else(|| "?".to_owned(), |n| n.to_string()); - format!("{short} #{id}") - } - "mcp__hyperhive__edit_schedule" => fmt_hyperhive_edit_schedule(short, input), - "mcp__hyperhive__request_schedule_prompt" => { - let tgts = match input.get("targets").and_then(Value::as_array) { - Some(arr) => arr - .iter() - .filter_map(Value::as_str) - .collect::>() - .join(", "), - None => "?".to_owned(), - }; - let when = input - .get("first_fire_at_unix") - .and_then(Value::as_u64) - .map_or_else( - || "?".to_owned(), - |ts| { - let h = (ts % 86_400) / 3_600; - let m = (ts % 3_600) / 60; - format!("{h:02}:{m:02}Z") - }, - ); - let recur = input - .get("interval_seconds") - .and_then(Value::as_u64) - .map(|s| format!(" +{s}s")) - .unwrap_or_default(); - format!("{short} → {tgts} at {when}{recur}") - } - _ => fmt_args_generic(short, input), - } -} - /// `mcp__bash__*` tools. fn fmt_bash_tool(name: &str, short: &str, input: &Value) -> String { match name {