scheduled prompts: add/remove targets on edit (#474 fast-follow)
This commit is contained in:
parent
376b37fed7
commit
99bf4d635d
6 changed files with 216 additions and 3 deletions
|
|
@ -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: "<agent-name>"`). 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.
|
||||
|
|
|
|||
|
|
@ -1006,6 +1006,16 @@ pub struct EditScheduleArgs {
|
|||
/// leave the schedule on its current cadence.
|
||||
#[serde(default)]
|
||||
pub next_fire_at_unix: Option<i64>,
|
||||
/// 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<Vec<String>>,
|
||||
/// 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<Vec<String>>,
|
||||
}
|
||||
|
||||
#[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(
|
||||
|
|
|
|||
|
|
@ -1509,6 +1509,17 @@ struct EditScheduleForm {
|
|||
interval_seconds: Option<Option<u64>>,
|
||||
#[serde(default)]
|
||||
next_fire_at_unix: Option<i64>,
|
||||
/// 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<Vec<String>>,
|
||||
/// 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<Vec<String>>,
|
||||
}
|
||||
|
||||
/// 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:#}"));
|
||||
|
|
|
|||
|
|
@ -356,6 +356,8 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> 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<Coordinator>) -> 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<Option<String>>,
|
||||
interval_seconds: Option<Option<u64>>,
|
||||
next_fire_at_unix: Option<i64>,
|
||||
targets_add: Option<Vec<String>>,
|
||||
targets_remove: Option<Vec<String>>,
|
||||
) -> 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,
|
||||
|
|
|
|||
|
|
@ -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<String>,
|
||||
pub description: Option<Option<String>>,
|
||||
pub interval_seconds: Option<Option<u64>>,
|
||||
pub next_fire_at_unix: Option<i64>,
|
||||
pub targets_add: Option<Vec<String>>,
|
||||
pub targets_remove: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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<Option<u64>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
next_fire_at_unix: Option<i64>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
targets_add: Option<Vec<String>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
targets_remove: Option<Vec<String>>,
|
||||
},
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue