hive-c0re: optional reset-timer on manual fire-now for recurring schedules
This commit is contained in:
parent
513c828ab2
commit
b6aeaf0b57
4 changed files with 89 additions and 17 deletions
|
|
@ -77,19 +77,29 @@ pub(super) async fn post_schedule_new(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Optional JSON body for `fire-now`. Absent / empty body ⇒
|
||||||
|
/// `reset_timer = false` (back-compat: cadence stays intact).
|
||||||
|
#[derive(serde::Deserialize, Default)]
|
||||||
|
pub(super) struct FireNowBody {
|
||||||
|
#[serde(default)]
|
||||||
|
reset_timer: bool,
|
||||||
|
}
|
||||||
|
|
||||||
/// `POST /api/schedules/{id}/fire-now` — operator-initiated
|
/// `POST /api/schedules/{id}/fire-now` — operator-initiated
|
||||||
/// out-of-band fire of a scheduled prompt. Runs the
|
/// out-of-band fire of a scheduled prompt. Runs the per-target
|
||||||
/// per-target fan-out once immediately and reports per-target
|
/// fan-out once immediately and reports per-target outcome counts.
|
||||||
/// outcome counts. Does NOT touch `next_fire_at_unix` on
|
/// One-shot schedules are consumed (cancelled) by a manual fire —
|
||||||
/// recurring schedules (their cadence stays intact); one-shot
|
/// the operator's intent is "send this now, the scheduled time was
|
||||||
/// schedules are consumed (cancelled) by a manual fire — the
|
/// wrong." For recurring schedules the cadence stays intact unless
|
||||||
/// operator's intent is "send this now, the scheduled time was
|
/// the body carries `{"reset_timer": true}`, in which case the
|
||||||
/// wrong."
|
/// countdown is re-armed from now (`next_fire_at = now + interval`).
|
||||||
pub(super) async fn post_schedule_fire_now(
|
pub(super) async fn post_schedule_fire_now(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
AxumPath(id): AxumPath<i64>,
|
AxumPath(id): AxumPath<i64>,
|
||||||
|
body: Option<axum::Json<FireNowBody>>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
match crate::scheduled_prompts_worker::fire_now(&state.coord, id).await {
|
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) => {
|
Ok(report) => {
|
||||||
state.coord.emit_schedules_snapshot();
|
state.coord.emit_schedules_snapshot();
|
||||||
axum::Json(report).into_response()
|
axum::Json(report).into_response()
|
||||||
|
|
|
||||||
|
|
@ -727,7 +727,9 @@ async fn handle_fire_schedule_now(
|
||||||
),
|
),
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
match crate::scheduled_prompts_worker::fire_now(coord, schedule_id).await {
|
// 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) => {
|
Ok(_report) => {
|
||||||
coord.emit_schedules_snapshot();
|
coord.emit_schedules_snapshot();
|
||||||
ManagerResponse::Ok
|
ManagerResponse::Ok
|
||||||
|
|
|
||||||
|
|
@ -340,6 +340,20 @@ impl ScheduledPrompts {
|
||||||
Ok(skipped)
|
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.
|
/// Partial-update an existing schedule's mutable fields.
|
||||||
/// Every scalar field is `Option<_>`; `None` means "leave the
|
/// Every scalar field is `Option<_>`; `None` means "leave the
|
||||||
/// existing value alone", `Some(_)` means "set it to this."
|
/// existing value alone", `Some(_)` means "set it to this."
|
||||||
|
|
@ -684,6 +698,28 @@ mod tests {
|
||||||
assert_eq!(s.next_fire_at_unix, 460);
|
assert_eq!(s.next_fire_at_unix, 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, 1_234);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn rearm_advances_one_step_when_caught_up() {
|
fn rearm_advances_one_step_when_caught_up() {
|
||||||
let (_dir, db) = open();
|
let (_dir, db) = open();
|
||||||
|
|
|
||||||
|
|
@ -261,6 +261,10 @@ pub struct FireNowReport {
|
||||||
/// schedules never auto-cancel on manual fire — they keep
|
/// schedules never auto-cancel on manual fire — they keep
|
||||||
/// their cadence).
|
/// their cadence).
|
||||||
pub one_shot_consumed: bool,
|
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"
|
/// Manual / out-of-band fire of a scheduled prompt ("fire now"
|
||||||
|
|
@ -275,12 +279,18 @@ pub struct FireNowReport {
|
||||||
/// the dashboard's per-target last-result column can distinguish
|
/// the dashboard's per-target last-result column can distinguish
|
||||||
/// scheduled fires from operator-initiated ones at a glance.
|
/// 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
|
/// Returns Err if the schedule is missing, cancelled, or fully
|
||||||
/// drained of active targets — the dashboard can surface those
|
/// drained of active targets — the dashboard can surface those
|
||||||
/// as plain 4xxs instead of pretending to fire a phantom row.
|
/// as plain 4xxs instead of pretending to fire a phantom row.
|
||||||
pub async fn fire_now(
|
pub async fn fire_now(
|
||||||
coord: &std::sync::Arc<Coordinator>,
|
coord: &std::sync::Arc<Coordinator>,
|
||||||
schedule_id: i64,
|
schedule_id: i64,
|
||||||
|
reset_timer: bool,
|
||||||
) -> anyhow::Result<FireNowReport> {
|
) -> anyhow::Result<FireNowReport> {
|
||||||
let now = now_unix();
|
let now = now_unix();
|
||||||
let schedule = coord
|
let schedule = coord
|
||||||
|
|
@ -303,6 +313,7 @@ pub async fn fire_now(
|
||||||
failed: 0,
|
failed: 0,
|
||||||
missing: 0,
|
missing: 0,
|
||||||
one_shot_consumed: false,
|
one_shot_consumed: false,
|
||||||
|
timer_reset: false,
|
||||||
};
|
};
|
||||||
for target_row in &schedule.targets {
|
for target_row in &schedule.targets {
|
||||||
if target_row.cancelled_at_unix.is_some() {
|
if target_row.cancelled_at_unix.is_some() {
|
||||||
|
|
@ -352,15 +363,28 @@ pub async fn fire_now(
|
||||||
tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
|
tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if schedule.interval_seconds.is_none() {
|
match schedule.interval_seconds {
|
||||||
// One-shot is consumed by the manual fire. Recurring
|
None => {
|
||||||
// schedules stay untouched — their cadence is the whole
|
// One-shot is consumed by the manual fire. (reset_timer is
|
||||||
// point and a manual fire is meant to be additive.
|
// moot here — there's no recurring cadence to re-arm.)
|
||||||
if let Err(e) = coord.scheduled_prompts.cancel_all(schedule_id) {
|
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");
|
tracing::warn!(error = ?e, schedule = schedule_id, "cancel_all after one-shot manual fire failed");
|
||||||
} else {
|
} else {
|
||||||
report.one_shot_consumed = true;
|
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)
|
Ok(report)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue