feat(schedules): make schedules pausable
Adds pause/resume support for scheduled prompts.
Backend:
- New paused_at_unix column on scheduled_prompts table (added via
ALTER TABLE migration so existing databases are upgraded on first
start). The due-rows index is dropped and recreated to also exclude
paused rows so the worker never fires them while paused.
- Worker's due() query gains AND paused_at_unix IS NULL filter.
- New pause(id) and resume(id) methods on ScheduledPrompts; both are
idempotent and refuse cancelled rows.
- New POST /api/schedules/{id}/pause and /api/schedules/{id}/resume
dashboard endpoints (operator-direct, no approval gate). Both emit
a schedules snapshot on success so the tab updates live.
- WireSchedule gains paused_at_unix: Option<i64> so the frontend can
render the state without an extra fetch.
Frontend:
- Paused rows render with a distinct row class + muted opacity.
- The next-fire cell shows a yellow pause glyph + tooltip with the
paused-since timestamp and the would-have-fired time.
- Actions column: pause/resume toggle button (⏸/▶) beside fire/edit/cancel.
Fire-now is disabled while paused (resume first).
- Sort order: active → paused → cancelled (paused slot keeps schedules
visible without mixing them into the active top section).
- pauseSchedule() / resumeSchedule() async functions POST to the new
endpoints and refresh the table on success.
This commit is contained in:
iris
mara
parent
3fedc102cc
commit
2bfa5bc1a8
7 changed files with 205 additions and 17 deletions
|
|
@ -1100,6 +1100,10 @@ footer .banner-thin {
|
|||
opacity: 0.75;
|
||||
}
|
||||
.schedules-table-row-cancelled td { opacity: 0.55; }
|
||||
.schedules-table-row-paused td { opacity: 0.75; }
|
||||
.sched-paused-label { color: var(--yellow); font-size: 0.9em; }
|
||||
.btn-pause-schedule { color: var(--teal); }
|
||||
.btn-resume-schedule { color: var(--green); }
|
||||
.schedules-table-body-cell {
|
||||
max-width: 30em;
|
||||
overflow: hidden;
|
||||
|
|
|
|||
|
|
@ -390,9 +390,10 @@ function renderSchedulesList() {
|
|||
table.append(renderSchedulesTableHead(agents));
|
||||
const tbody = el('tbody', {});
|
||||
const sorted = schedulesState.slice().sort((a, b) => {
|
||||
const aDone = a.cancelled_at_unix ? 1 : 0;
|
||||
const bDone = b.cancelled_at_unix ? 1 : 0;
|
||||
if (aDone !== bDone) return aDone - bDone;
|
||||
// Cancelled → bucket 2, paused → bucket 1, active → bucket 0.
|
||||
const aOrd = a.cancelled_at_unix ? 2 : a.paused_at_unix ? 1 : 0;
|
||||
const bOrd = b.cancelled_at_unix ? 2 : b.paused_at_unix ? 1 : 0;
|
||||
if (aOrd !== bOrd) return aOrd - bOrd;
|
||||
return a.next_fire_at_unix - b.next_fire_at_unix;
|
||||
});
|
||||
for (const s of sorted) {
|
||||
|
|
@ -637,8 +638,11 @@ function renderSchedulesTableHead(agents) {
|
|||
}
|
||||
function renderScheduleRow(s, agents) {
|
||||
const cancelled = !!s.cancelled_at_unix;
|
||||
const paused = !cancelled && !!s.paused_at_unix;
|
||||
const tr = el('tr', {
|
||||
class: 'schedules-table-row' + (cancelled ? ' schedules-table-row-cancelled' : ''),
|
||||
class: 'schedules-table-row'
|
||||
+ (cancelled ? ' schedules-table-row-cancelled' : '')
|
||||
+ (paused ? ' schedules-table-row-paused' : ''),
|
||||
});
|
||||
|
||||
tr.append(el('td', { class: 'meta schedules-table-id' }, '#' + s.id));
|
||||
|
|
@ -649,11 +653,20 @@ function renderScheduleRow(s, agents) {
|
|||
srcKind === 'approval' ? 'approval' : 'operator')));
|
||||
|
||||
// "next" cell — relative due-in for active schedules, "cancelled"
|
||||
// for cancelled ones. Both carry the absolute ISO in the title.
|
||||
// for cancelled ones, "paused" for paused ones. Both carry the
|
||||
// absolute ISO in the title.
|
||||
const nextCell = el('td', { class: 'meta schedules-table-next-col' });
|
||||
if (cancelled) {
|
||||
nextCell.title = 'cancelled ' + new Date(s.cancelled_at_unix * 1000).toISOString();
|
||||
nextCell.textContent = 'cancelled';
|
||||
} else if (paused) {
|
||||
nextCell.title = 'paused since '
|
||||
+ new Date(s.paused_at_unix * 1000).toISOString()
|
||||
+ '\nwould fire at '
|
||||
+ new Date(s.next_fire_at_unix * 1000).toISOString();
|
||||
nextCell.append(
|
||||
el('span', { class: 'sched-paused-label' }, '⏸ paused'),
|
||||
);
|
||||
} else {
|
||||
const dueIn = s.next_fire_at_unix - Math.floor(Date.now() / 1000);
|
||||
nextCell.title = new Date(s.next_fire_at_unix * 1000).toISOString();
|
||||
|
|
@ -723,10 +736,10 @@ function renderScheduleRow(s, agents) {
|
|||
tr.append(td);
|
||||
}
|
||||
|
||||
// Actions cell — fire / edit / cancel-all. Glyph-only to fit a
|
||||
// compact column; the buttons keep their existing colour classes
|
||||
// so the visual cue (mauve = fire, yellow = edit, red = cancel)
|
||||
// carries over from the card layout.
|
||||
// Actions cell — fire / pause-toggle / edit / cancel-all.
|
||||
// Glyph-only to fit a compact column; colour classes carry the
|
||||
// visual cue: mauve = fire, teal = pause/resume, yellow = edit,
|
||||
// red = cancel.
|
||||
const actionsCell = el('td', { class: 'schedules-table-actions' });
|
||||
if (!cancelled) {
|
||||
const activeTargets = (s.targets || []).filter((t) => !t.cancelled_at_unix);
|
||||
|
|
@ -738,14 +751,28 @@ function renderScheduleRow(s, agents) {
|
|||
fireBtn.title = isOneShot
|
||||
? 'fire once — one-shot, consumed after the manual fire'
|
||||
: 'fire once now — recurring; choose whether to reset the next-fire timer';
|
||||
if (!activeTargets.length) {
|
||||
if (!activeTargets.length || paused) {
|
||||
fireBtn.disabled = true;
|
||||
fireBtn.title = 'every target is cancelled — nothing to fire';
|
||||
fireBtn.title = paused
|
||||
? 'resume the schedule first to use fire-now'
|
||||
: 'every target is cancelled — nothing to fire';
|
||||
}
|
||||
fireBtn.addEventListener('click', () =>
|
||||
fireScheduleNow(s.id, isOneShot, activeTargets.map((t) => t.target), fireBtn));
|
||||
actionsCell.append(fireBtn);
|
||||
|
||||
// Pause / resume toggle — only meaningful for recurring schedules.
|
||||
// One-shots can still be paused (to delay a one-time fire), so we
|
||||
// show the button for both cases.
|
||||
const pauseBtn = el('button', {
|
||||
type: 'button',
|
||||
class: 'btn btn-pause-schedule btn-inline-small' + (paused ? ' btn-resume-schedule' : ''),
|
||||
}, paused ? '▶' : '⏸');
|
||||
pauseBtn.title = paused ? 'resume schedule' : 'pause schedule';
|
||||
pauseBtn.addEventListener('click', () =>
|
||||
paused ? resumeSchedule(s.id) : pauseSchedule(s.id));
|
||||
actionsCell.append(pauseBtn);
|
||||
|
||||
const editingThis = editingSchedules.has(s.id);
|
||||
const editBtn = el('button', {
|
||||
type: 'button',
|
||||
|
|
@ -1089,6 +1116,37 @@ async function postScheduleCancel(id, targets) {
|
|||
}
|
||||
}
|
||||
|
||||
async function pauseSchedule(id) {
|
||||
try {
|
||||
const resp = await fetch('/api/schedules/' + encodeURIComponent(id) + '/pause', {
|
||||
method: 'POST',
|
||||
});
|
||||
if (!resp.ok) {
|
||||
const text = await resp.text().catch(() => '');
|
||||
themedToast('pause failed: http ' + resp.status + (text ? '\n\n' + text : ''), { type: 'error' });
|
||||
return;
|
||||
}
|
||||
await refreshSchedules();
|
||||
} catch (err) {
|
||||
themedToast('pause failed: ' + err, { type: 'error' });
|
||||
}
|
||||
}
|
||||
async function resumeSchedule(id) {
|
||||
try {
|
||||
const resp = await fetch('/api/schedules/' + encodeURIComponent(id) + '/resume', {
|
||||
method: 'POST',
|
||||
});
|
||||
if (!resp.ok) {
|
||||
const text = await resp.text().catch(() => '');
|
||||
themedToast('resume failed: http ' + resp.status + (text ? '\n\n' + text : ''), { type: 'error' });
|
||||
return;
|
||||
}
|
||||
await refreshSchedules();
|
||||
} catch (err) {
|
||||
themedToast('resume failed: ' + err, { type: 'error' });
|
||||
}
|
||||
}
|
||||
|
||||
export function applySchedulesChanged(ev) {
|
||||
schedulesState = (ev.schedules || []).slice();
|
||||
renderSchedulesList();
|
||||
|
|
|
|||
|
|
@ -131,6 +131,14 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
|||
"/api/schedules/{id}/cancel",
|
||||
post(schedules::post_schedule_cancel),
|
||||
)
|
||||
.route(
|
||||
"/api/schedules/{id}/pause",
|
||||
post(schedules::post_schedule_pause),
|
||||
)
|
||||
.route(
|
||||
"/api/schedules/{id}/resume",
|
||||
post(schedules::post_schedule_resume),
|
||||
)
|
||||
.route(
|
||||
"/api/schedules/{id}/fire-now",
|
||||
post(schedules::post_schedule_fire_now),
|
||||
|
|
|
|||
|
|
@ -228,6 +228,52 @@ pub(super) async fn patch_schedule(
|
|||
}
|
||||
}
|
||||
|
||||
/// `POST /api/schedules/{id}/pause` — 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.
|
||||
pub(super) async fn post_schedule_pause(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
) -> Response {
|
||||
match state.coord.scheduled_prompts.pause(id) {
|
||||
Ok(()) => {
|
||||
state.coord.emit_schedules_snapshot();
|
||||
(StatusCode::OK, "ok").into_response()
|
||||
}
|
||||
Err(e) => {
|
||||
let msg = format!("{e:#}");
|
||||
if msg.contains("not found") || msg.contains("cancelled") {
|
||||
(StatusCode::NOT_FOUND, msg).into_response()
|
||||
} else {
|
||||
error_response(&format!("pause schedule {id}: {msg}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /api/schedules/{id}/resume` — resume a paused schedule.
|
||||
/// Idempotent; no-op on an already-active row. Returns 404 when the
|
||||
/// schedule is cancelled or not found.
|
||||
pub(super) async fn post_schedule_resume(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
) -> Response {
|
||||
match state.coord.scheduled_prompts.resume(id) {
|
||||
Ok(()) => {
|
||||
state.coord.emit_schedules_snapshot();
|
||||
(StatusCode::OK, "ok").into_response()
|
||||
}
|
||||
Err(e) => {
|
||||
let msg = format!("{e:#}");
|
||||
if msg.contains("not found") || msg.contains("cancelled") {
|
||||
(StatusCode::NOT_FOUND, msg).into_response()
|
||||
} else {
|
||||
error_response(&format!("resume schedule {id}: {msg}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /api/schedules/{id}/cancel` — operator-side cancel
|
||||
/// (whole schedule when no `targets` field, partial when one is
|
||||
/// provided). Operator bypasses the topology check; the manager
|
||||
|
|
|
|||
|
|
@ -29,11 +29,12 @@ CREATE TABLE IF NOT EXISTS scheduled_prompts (
|
|||
created_at_unix INTEGER NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
cancelled_at_unix INTEGER,
|
||||
description TEXT
|
||||
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;
|
||||
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,
|
||||
|
|
@ -68,6 +69,9 @@ pub struct Schedule {
|
|||
/// next pass.
|
||||
pub cancelled_at_unix: Option<i64>,
|
||||
pub description: Option<String>,
|
||||
/// Set while the schedule is paused. Worker skips rows where
|
||||
/// `paused_at_unix IS NOT NULL`. Cleared by `resume()`.
|
||||
pub paused_at_unix: Option<i64>,
|
||||
pub targets: Vec<ScheduleTarget>,
|
||||
}
|
||||
|
||||
|
|
@ -173,6 +177,24 @@ impl ScheduledPrompts {
|
|||
.context("enable foreign keys")?;
|
||||
conn.execute_batch(SCHEMA)
|
||||
.context("apply scheduled_prompts schema")?;
|
||||
// Migration: add paused_at_unix to existing databases.
|
||||
// Silently ignores "duplicate column name" errors so this is
|
||||
// idempotent across daemon restarts on already-migrated DBs.
|
||||
let _ = conn.execute(
|
||||
"ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER",
|
||||
[],
|
||||
);
|
||||
// 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),
|
||||
})
|
||||
|
|
@ -221,7 +243,8 @@ impl ScheduledPrompts {
|
|||
let row = conn
|
||||
.query_row(
|
||||
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
||||
created_at_unix, source, cancelled_at_unix, description
|
||||
created_at_unix, source, cancelled_at_unix, description,
|
||||
paused_at_unix
|
||||
FROM scheduled_prompts WHERE id = ?1",
|
||||
params![id],
|
||||
row_to_schedule_header,
|
||||
|
|
@ -242,7 +265,8 @@ impl ScheduledPrompts {
|
|||
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
|
||||
created_at_unix, source, cancelled_at_unix, description,
|
||||
paused_at_unix
|
||||
FROM scheduled_prompts
|
||||
ORDER BY next_fire_at_unix ASC, id ASC",
|
||||
)?;
|
||||
|
|
@ -264,9 +288,11 @@ impl ScheduledPrompts {
|
|||
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
|
||||
created_at_unix, source, cancelled_at_unix, description,
|
||||
paused_at_unix
|
||||
FROM scheduled_prompts
|
||||
WHERE cancelled_at_unix IS NULL AND next_fire_at_unix <= ?1
|
||||
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",
|
||||
)?;
|
||||
|
|
@ -546,6 +572,43 @@ impl ScheduledPrompts {
|
|||
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` when the schedule is cancelled or does not exist.
|
||||
pub fn pause(&self, id: i64) -> Result<()> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let now = now_unix();
|
||||
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 {
|
||||
bail!("schedule {id} not found or is cancelled");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resume a paused schedule. Idempotent; no-op on an active row.
|
||||
/// Returns `Err` when the schedule is cancelled or does not exist.
|
||||
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 {
|
||||
bail!("schedule {id} not found or is cancelled");
|
||||
}
|
||||
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
|
||||
|
|
@ -575,6 +638,7 @@ fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result<Schedule> {
|
|||
source: ScheduleSource::from_db_string(&source_str),
|
||||
cancelled_at_unix: row.get(7)?,
|
||||
description: row.get(8)?,
|
||||
paused_at_unix: row.get(9)?,
|
||||
targets: Vec::new(),
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2021,6 +2021,7 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSc
|
|||
}
|
||||
},
|
||||
cancelled_at_unix: s.cancelled_at_unix,
|
||||
paused_at_unix: s.paused_at_unix,
|
||||
description: s.description,
|
||||
targets: s
|
||||
.targets
|
||||
|
|
@ -2126,6 +2127,7 @@ mod tests {
|
|||
created_at_unix: 0,
|
||||
source: hive_sh4re::WireScheduleSource::Operator,
|
||||
cancelled_at_unix: None,
|
||||
paused_at_unix: None,
|
||||
description: None,
|
||||
targets: targets.iter().map(|t| target(t)).collect(),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1395,6 +1395,12 @@ pub struct WireSchedule {
|
|||
pub source: WireScheduleSource,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub cancelled_at_unix: Option<i64>,
|
||||
/// 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).
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub paused_at_unix: Option<i64>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub description: Option<String>,
|
||||
pub targets: Vec<WireScheduleTarget>,
|
||||
|
|
|
|||
Loading…
Reference in a new issue