feat: live SSE updates for the SCH3DUL3S tab
Add `SchedulesChanged` to the dashboard event channel so the
operator's schedule list updates in real time without requiring a
tab-activation or form-submit refresh.
Backend:
- `dashboard_events.rs`: new `SchedulesChanged { seq, schedules }`
variant carrying a full `Vec<WireSchedule>` snapshot (same
snapshot-over-diff rationale as `RebuildQueueChanged`).
- `coordinator.rs`: `emit_schedules_snapshot()` helper — queries the
scheduled_prompts list, converts to wire shape, broadcasts the event.
- `dashboard.rs`: call `emit_schedules_snapshot()` at the end of each
operator API handler that mutates a schedule:
`post_schedule_new`, `post_schedule_fire_now`,
`patch_schedule`, `post_schedule_cancel`.
- `scheduled_prompts_worker.rs`: call `emit_schedules_snapshot()`
after each tick that fires schedules, so `last_fired_at_unix`,
`next_fire_at_unix`, and reaped one-shots surface live.
Frontend:
- `tabs.js`: add `applySchedulesChanged(ev)` — replaces
`schedulesState` from the snapshot and calls `renderSchedulesList()`.
Registered in `MUTATION_HANDLERS` as `schedules_changed`.
Tab-activation re-fetch kept as safety net for approval-path
inserts and disconnect windows; comment updated to reflect this.
Docs:
- `docs/web-ui/dashboard.md`: document `schedules_changed` event.
- `CLAUDE.md`: add `SchedulesChanged` to the file-map entry.
This commit is contained in:
parent
b411a81c1a
commit
76c4a67b1c
7 changed files with 78 additions and 9 deletions
|
|
@ -82,7 +82,8 @@ hive-c0re/ host daemon + sibling operator CLI (lib + 2 bins)
|
||||||
(`ApprovalAdded` / `ApprovalResolved`,
|
(`ApprovalAdded` / `ApprovalResolved`,
|
||||||
`QuestionAdded` / `QuestionResolved`,
|
`QuestionAdded` / `QuestionResolved`,
|
||||||
`TransientSet` / `TransientCleared`,
|
`TransientSet` / `TransientCleared`,
|
||||||
`RebuildQueueChanged`). Each frame carries a
|
`RebuildQueueChanged`, `SchedulesChanged`).
|
||||||
|
Each frame carries a
|
||||||
monotonic per-process `seq` clients use to
|
monotonic per-process `seq` clients use to
|
||||||
dedupe against snapshot reads.
|
dedupe against snapshot reads.
|
||||||
src/approvals.rs sqlite Approval queue + kinds
|
src/approvals.rs sqlite Approval queue + kinds
|
||||||
|
|
|
||||||
|
|
@ -843,6 +843,15 @@ payload):
|
||||||
`meta_inputs_changed`: the list is small and the client's
|
`meta_inputs_changed`: the list is small and the client's
|
||||||
`parent_id` grouping is most naturally re-derived from the
|
`parent_id` grouping is most naturally re-derived from the
|
||||||
full list. Cold-loaded from `/api/state.rebuild_queue`.
|
full list. Cold-loaded from `/api/state.rebuild_queue`.
|
||||||
|
- `schedules_changed` (seq, schedules: `Vec<WireSchedule>`) —
|
||||||
|
full snapshot of all scheduled prompts. Emitted after every
|
||||||
|
operator mutation via the `/api/schedules` surface (new /
|
||||||
|
edit / cancel / fire-now) and after the worker fires or
|
||||||
|
rearms a row. Same snapshot-shape rationale as
|
||||||
|
`rebuild_queue_changed`. The SCH3DUL3S tab subscribes and
|
||||||
|
re-renders `schedulesState` on receipt; tab activation still
|
||||||
|
re-fetches as a safety net for approval-path inserts and
|
||||||
|
disconnect windows.
|
||||||
|
|
||||||
`/api/state` is **only fetched on cold-load and on the few
|
`/api/state` is **only fetched on cold-load and on the few
|
||||||
forms that mutate non-event-derived state** (PURG3 +
|
forms that mutate non-event-derived state** (PURG3 +
|
||||||
|
|
|
||||||
|
|
@ -223,6 +223,10 @@ window.marked = marked;
|
||||||
// See docs/web-ui.md::Container row for the badge taxonomy.
|
// See docs/web-ui.md::Container row for the badge taxonomy.
|
||||||
renderContainersFromState();
|
renderContainersFromState();
|
||||||
}
|
}
|
||||||
|
function applySchedulesChanged(ev) {
|
||||||
|
schedulesState = (ev.schedules || []).slice();
|
||||||
|
renderSchedulesList();
|
||||||
|
}
|
||||||
// Map from agent name → highest-priority in-flight queue entry
|
// Map from agent name → highest-priority in-flight queue entry
|
||||||
// (`running` beats `queued`). Used by the container row renderer
|
// (`running` beats `queued`). Used by the container row renderer
|
||||||
// to surface "building..." / "meta-updating..." badges on the
|
// to surface "building..." / "meta-updating..." badges on the
|
||||||
|
|
@ -2471,8 +2475,11 @@ window.marked = marked;
|
||||||
// ─── scheduled prompts ─────────────────────────────────────────────────
|
// ─── scheduled prompts ─────────────────────────────────────────────────
|
||||||
// Backend exposes `/api/schedules` (snapshot), `/api/schedules`
|
// Backend exposes `/api/schedules` (snapshot), `/api/schedules`
|
||||||
// (POST, operator-direct submit), `/api/schedules/{id}/cancel`
|
// (POST, operator-direct submit), `/api/schedules/{id}/cancel`
|
||||||
// (whole or per-target). No SSE channel for schedule mutations yet,
|
// (whole or per-target), `/api/schedules/{id}` (PATCH edit),
|
||||||
// so we refresh on tab activation + after every submit/cancel POST.
|
// `/api/schedules/{id}/fire-now` (POST). Mutations now emit a
|
||||||
|
// `schedules_changed` SSE event so the list updates live;
|
||||||
|
// `applySchedulesChanged` handles it. Tab-activation re-fetch kept as
|
||||||
|
// a safety net for approval-path inserts and disconnect windows.
|
||||||
// Local cache lets `refreshTabCounts` show the active count without
|
// Local cache lets `refreshTabCounts` show the active count without
|
||||||
// re-fetching every second.
|
// re-fetching every second.
|
||||||
let schedulesState = [];
|
let schedulesState = [];
|
||||||
|
|
@ -3596,6 +3603,7 @@ window.marked = marked;
|
||||||
meta_inputs_changed: applyMetaInputsChanged,
|
meta_inputs_changed: applyMetaInputsChanged,
|
||||||
meta_update_running: applyMetaUpdateRunning,
|
meta_update_running: applyMetaUpdateRunning,
|
||||||
rebuild_queue_changed: applyRebuildQueueChanged,
|
rebuild_queue_changed: applyRebuildQueueChanged,
|
||||||
|
schedules_changed: applySchedulesChanged,
|
||||||
};
|
};
|
||||||
(function bindDashboardStream() {
|
(function bindDashboardStream() {
|
||||||
// Route through the SharedWorker so all open hyperhive tabs share
|
// Route through the SharedWorker so all open hyperhive tabs share
|
||||||
|
|
@ -3650,8 +3658,9 @@ window.marked = marked;
|
||||||
renderSelectionBar(Array.from(containersState.values()));
|
renderSelectionBar(Array.from(containersState.values()));
|
||||||
// Keep overflow button active state in sync after tab change.
|
// Keep overflow button active state in sync after tab change.
|
||||||
updateTabbarOverflow();
|
updateTabbarOverflow();
|
||||||
// Schedules pane has no SSE channel for mutations, so re-fetch
|
// Re-fetch schedules on activation as a safety net (SSE covers
|
||||||
// on activation so the operator never lands on stale data.
|
// live mutations but re-sync ensures consistency after disconnect
|
||||||
|
// windows or approval-path inserts that don't yet emit).
|
||||||
if (target === 'schedules') refreshSchedules();
|
if (target === 'schedules') refreshSchedules();
|
||||||
// Permissions tables (capabilities + tool-groups) have no SSE channel;
|
// Permissions tables (capabilities + tool-groups) have no SSE channel;
|
||||||
// fetch both on each activation so the operator sees fresh data.
|
// fetch both on each activation so the operator sees fresh data.
|
||||||
|
|
|
||||||
|
|
@ -368,6 +368,27 @@ 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<Self>) {
|
||||||
|
let schedules = match self.scheduled_prompts.list() {
|
||||||
|
Ok(rows) => rows
|
||||||
|
.into_iter()
|
||||||
|
.map(crate::manager_server::schedule_to_wire_public)
|
||||||
|
.collect(),
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
|
||||||
|
seq: self.next_seq(),
|
||||||
|
schedules,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
/// Update the `step` label on a running queue entry and (if it
|
/// Update the `step` label on a running queue entry and (if it
|
||||||
/// actually changed) re-emit the queue snapshot so the dashboard
|
/// actually changed) re-emit the queue snapshot so the dashboard
|
||||||
/// renders the new phase. Returns `true` when the label was new
|
/// renders the new phase. Returns `true` when the label was new
|
||||||
|
|
|
||||||
|
|
@ -1964,7 +1964,10 @@ async fn post_schedule_new(
|
||||||
source: crate::scheduled_prompts::ScheduleSource::Operator,
|
source: crate::scheduled_prompts::ScheduleSource::Operator,
|
||||||
};
|
};
|
||||||
match state.coord.scheduled_prompts.submit(&new) {
|
match state.coord.scheduled_prompts.submit(&new) {
|
||||||
Ok(id) => axum::Json(serde_json::json!({"id": id})).into_response(),
|
Ok(id) => {
|
||||||
|
state.coord.emit_schedules_snapshot();
|
||||||
|
axum::Json(serde_json::json!({"id": id})).into_response()
|
||||||
|
}
|
||||||
Err(e) => error_response(&format!("schedule submit: {e:#}")),
|
Err(e) => error_response(&format!("schedule submit: {e:#}")),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -1982,7 +1985,10 @@ async fn post_schedule_fire_now(
|
||||||
AxumPath(id): AxumPath<i64>,
|
AxumPath(id): AxumPath<i64>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
match crate::scheduled_prompts_worker::fire_now(&state.coord, id).await {
|
match crate::scheduled_prompts_worker::fire_now(&state.coord, id).await {
|
||||||
Ok(report) => axum::Json(report).into_response(),
|
Ok(report) => {
|
||||||
|
state.coord.emit_schedules_snapshot();
|
||||||
|
axum::Json(report).into_response()
|
||||||
|
}
|
||||||
Err(e) => error_response(&format!("fire schedule {id} now: {e:#}")),
|
Err(e) => error_response(&format!("fire schedule {id} now: {e:#}")),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -2093,7 +2099,9 @@ async fn patch_schedule(
|
||||||
}
|
}
|
||||||
match state.coord.scheduled_prompts.get(id) {
|
match state.coord.scheduled_prompts.get(id) {
|
||||||
Ok(Some(s)) => {
|
Ok(Some(s)) => {
|
||||||
axum::Json(crate::manager_server::schedule_to_wire_public(s)).into_response()
|
let wire = crate::manager_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")),
|
Ok(None) => error_response(&format!("edit schedule {id}: row vanished post-update")),
|
||||||
Err(e) => error_response(&format!("re-read schedule {id}: {e:#}")),
|
Err(e) => error_response(&format!("re-read schedule {id}: {e:#}")),
|
||||||
|
|
@ -2117,7 +2125,10 @@ async fn post_schedule_cancel(
|
||||||
None => state.coord.scheduled_prompts.cancel_all(id),
|
None => state.coord.scheduled_prompts.cancel_all(id),
|
||||||
};
|
};
|
||||||
match result {
|
match result {
|
||||||
Ok(()) => (StatusCode::OK, "ok").into_response(),
|
Ok(()) => {
|
||||||
|
state.coord.emit_schedules_snapshot();
|
||||||
|
(StatusCode::OK, "ok").into_response()
|
||||||
|
}
|
||||||
Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")),
|
Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -192,6 +192,16 @@ pub enum DashboardEvent {
|
||||||
/// dashboard's grouping (`parent_id`) is most naturally re-derived
|
/// dashboard's grouping (`parent_id`) is most naturally re-derived
|
||||||
/// from the full list.
|
/// from the full list.
|
||||||
RebuildQueueChanged { seq: u64, queue: Vec<QueueEntry> },
|
RebuildQueueChanged { seq: u64, queue: Vec<QueueEntry> },
|
||||||
|
/// 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
|
||||||
|
/// `RebuildQueueChanged` — 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<hive_sh4re::WireSchedule>,
|
||||||
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
impl DashboardEvent {
|
impl DashboardEvent {
|
||||||
|
|
@ -222,6 +232,7 @@ impl DashboardEvent {
|
||||||
DashboardEvent::MetaInputsChanged { .. } => "meta_inputs_changed",
|
DashboardEvent::MetaInputsChanged { .. } => "meta_inputs_changed",
|
||||||
DashboardEvent::MetaUpdateRunning { .. } => "meta_update_running",
|
DashboardEvent::MetaUpdateRunning { .. } => "meta_update_running",
|
||||||
DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed",
|
DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed",
|
||||||
|
DashboardEvent::SchedulesChanged { .. } => "schedules_changed",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -340,6 +351,10 @@ mod tests {
|
||||||
seq: 1,
|
seq: 1,
|
||||||
queue: Vec::new(),
|
queue: Vec::new(),
|
||||||
},
|
},
|
||||||
|
DashboardEvent::SchedulesChanged {
|
||||||
|
seq: 1,
|
||||||
|
schedules: Vec::new(),
|
||||||
|
},
|
||||||
];
|
];
|
||||||
for ev in samples {
|
for ev in samples {
|
||||||
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
|
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
|
||||||
|
|
|
||||||
|
|
@ -69,6 +69,9 @@ fn tick(coord: &Arc<Coordinator>) {
|
||||||
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
|
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
|
||||||
tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
|
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
|
/// Fan out one schedule's body to every active target. Records
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue