From 99bf4d635dd4b1456b76b9e531ecd3d09789dd9a Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 26 May 2026 16:14:47 +0200 Subject: [PATCH] scheduled prompts: add/remove targets on edit (#474 fast-follow) --- hive-ag3nt/prompts/manager.md | 2 +- hive-ag3nt/src/mcp.rs | 12 ++ hive-c0re/src/dashboard.rs | 13 +++ hive-c0re/src/manager_server.rs | 8 ++ hive-c0re/src/scheduled_prompts.rs | 174 +++++++++++++++++++++++++++++ hive-sh4re/src/lib.rs | 10 +- 6 files changed, 216 insertions(+), 3 deletions(-) diff --git a/hive-ag3nt/prompts/manager.md b/hive-ag3nt/prompts/manager.md index a089b7b9..9010f115 100644 --- a/hive-ag3nt/prompts/manager.md +++ b/hive-ag3nt/prompts/manager.md @@ -14,7 +14,7 @@ Tools (hyperhive surface): - `mcp__hyperhive__request_schedule_prompt(targets, body, first_fire_at_unix, interval_seconds?, description?)` — queue an approval for the operator to add a scheduled prompt. On approve hive-c0re inserts a schedule row and the worker fans `body` out to each agent in `targets` at `first_fire_at_unix` (recurring every `interval_seconds` if set, one-shot when absent). Even self-targeted schedules go through approval — the existing `remind` tool stays the quick no-approval self-wake path. Catch-up clamp: long downtime fires ONCE per recurring row on resume (skipped count surfaces in per-target `last_result`), not N stacked pulses. - `mcp__hyperhive__cancel_schedule(id, targets?)` — cancel a schedule. Omit `targets` / pass empty to cancel the whole schedule; pass a list to cancel just those recipients (the schedule keeps firing for any remaining active targets, auto-cancels when every target is gone). Authorization: you can cancel schedules you own OR any owned by a sub-agent in your subtree per topology.json. - `mcp__hyperhive__fire_schedule_now(id)` — fire a scheduled prompt out of band. Runs the per-target fan-out once immediately. Recurring schedules keep their cadence intact (the manual fire is additive); one-shot schedules are CONSUMED by the manual fire (cancelled afterwards). Same authorization as `cancel_schedule`. -- `mcp__hyperhive__edit_schedule(id, body?, description?, interval_seconds?, next_fire_at_unix?)` — partial-update a schedule's mutable fields (#474). Pass only the fields you want to change. Targets stay immutable (cancel + new schedule is the workaround). Refuses cancelled rows. Same authorization as `cancel_schedule`. Note: clearing fields (e.g. flipping recurring→one-shot) is operator-only via the dashboard PATCH — the agent surface only supports positive sets. +- `mcp__hyperhive__edit_schedule(id, body?, description?, interval_seconds?, next_fire_at_unix?, targets_add?, targets_remove?)` — partial-update a schedule's mutable fields (#474). Pass only the fields you want to change. `targets_add` / `targets_remove` mutate the recipient list in the same transaction; re-adding a previously-cancelled target drops the tombstone + history (operator intent: "fresh start"). Refuses cancelled rows. Same authorization as `cancel_schedule`. Note: clearing scalar fields (e.g. flipping recurring→one-shot) is operator-only via the dashboard PATCH — the agent surface only supports positive sets on `description` / `interval_seconds`. - `mcp__hyperhive__list_schedules()` — snapshot every schedule in the queue (active + cancelled-but-not-reaped). Returns id, owner, body, target set with per-target `last_fired_at` + `last_result`, `next_fire_at_unix`, recurring `interval_seconds`. Use to look up an id before cancelling, or to audit upcoming wake-ups across the swarm. - `mcp__hyperhive__get_logs(agent, lines?)` — fetch recent journal lines for a sub-agent container. Use to diagnose MCP-server registration failures, startup crashes, or harness issues you can't see from inside. Pass the plain logical agent name; `lines` defaults to 50 (capped at 500). - `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the operator (default, or `to: "operator"`) OR a sub-agent (`to: ""`). Returns immediately with a question id; the answer arrives later as a system `question_answered { id, question, answer, answerer }` event in your inbox. Options are advisory: the dashboard always lets the operator type a free-text answer in addition. Set `multi: true` to render options as checkboxes (operator can pick multiple); the answer comes back as `, `-separated. Set `ttl_seconds` to auto-cancel after a deadline (capped at 6h server-side) — on expiry the answer is `[expired]` and `answerer` is `"ttl-watchdog"`. Do not poll inside the same turn — finish the current work and react when the event lands. diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 227c7a66..c5912c5c 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -1006,6 +1006,16 @@ pub struct EditScheduleArgs { /// 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>, } #[derive(Debug, serde::Deserialize, schemars::JsonSchema)] @@ -1382,6 +1392,8 @@ impl ManagerServer { 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( diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index a3ba2b2b..48b537c1 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -1509,6 +1509,17 @@ struct EditScheduleForm { 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 @@ -1543,6 +1554,8 @@ async fn patch_schedule( 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:#}")); diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 19d19e50..303d69f1 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -356,6 +356,8 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp description, interval_seconds, next_fire_at_unix, + targets_add, + targets_remove, } => handle_edit_schedule( coord, hive_sh4re::MANAGER_AGENT, @@ -364,6 +366,8 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp description.clone(), *interval_seconds, *next_fire_at_unix, + targets_add.clone(), + targets_remove.clone(), ), ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() { Ok(schedules) => ManagerResponse::Schedules { @@ -895,6 +899,8 @@ fn handle_edit_schedule( description: Option>, interval_seconds: Option>, next_fire_at_unix: Option, + targets_add: Option>, + targets_remove: Option>, ) -> ManagerResponse { let schedule = match coord.scheduled_prompts.get(schedule_id) { Ok(Some(s)) => s, @@ -922,6 +928,8 @@ fn handle_edit_schedule( description, interval_seconds, next_fire_at_unix, + targets_add, + targets_remove, }; match coord.scheduled_prompts.update(schedule_id, patch) { Ok(()) => ManagerResponse::Ok, diff --git a/hive-c0re/src/scheduled_prompts.rs b/hive-c0re/src/scheduled_prompts.rs index 9de7f799..ccd41909 100644 --- a/hive-c0re/src/scheduled_prompts.rs +++ b/hive-c0re/src/scheduled_prompts.rs @@ -153,12 +153,24 @@ pub struct NewSchedule { /// `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 (#474 fast-follow): `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)] 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 { @@ -404,6 +416,59 @@ impl ScheduledPrompts { 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 = now_unix(); + 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(()) } @@ -846,6 +911,115 @@ mod tests { 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(); diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 8d53ddac..e5411178 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -954,8 +954,10 @@ pub enum ManagerRequest { /// doubly-wrapped so `Some(None)` (set explicit null) can flip /// a recurring schedule back to one-shot / clear the /// description, while plain `None` keeps the current value. - /// Targets stay immutable. Refuses cancelled rows. - /// Authorization mirrors `CancelSchedule`. + /// `targets_add` / `targets_remove` mutate the recipient list + /// in the same transaction; re-adding a previously-cancelled + /// target drops the tombstone (replace-on-conflict). Refuses + /// cancelled rows. Authorization mirrors `CancelSchedule`. EditSchedule { id: i64, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -966,6 +968,10 @@ pub enum ManagerRequest { 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>, }, }