scheduled prompts: fire-now operator + manager surfaces (closes #467)
This commit is contained in:
parent
caa0c6ce4c
commit
3cd9240594
6 changed files with 251 additions and 1 deletions
|
|
@ -76,6 +76,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
|||
.route("/meta-update", post(post_meta_update))
|
||||
.route("/api/schedules", get(api_schedules).post(post_schedule_new))
|
||||
.route("/api/schedules/{id}/cancel", post(post_schedule_cancel))
|
||||
.route("/api/schedules/{id}/fire-now", post(post_schedule_fire_now))
|
||||
.route("/dashboard/stream", get(dashboard_stream))
|
||||
.route("/dashboard/history", get(dashboard_history))
|
||||
// Anything not matched by the dynamic routes above falls
|
||||
|
|
@ -1440,6 +1441,24 @@ async fn post_schedule_new(
|
|||
}
|
||||
}
|
||||
|
||||
/// `POST /api/schedules/{id}/fire-now` — operator-initiated
|
||||
/// out-of-band fire of a scheduled prompt (#467). Runs the
|
||||
/// per-target fan-out once immediately and reports per-target
|
||||
/// outcome counts. Does NOT touch `next_fire_at_unix` on
|
||||
/// recurring schedules (their cadence stays intact); one-shot
|
||||
/// schedules are consumed (cancelled) by a manual fire — the
|
||||
/// operator's intent is "send this now, the scheduled time was
|
||||
/// wrong."
|
||||
async fn post_schedule_fire_now(
|
||||
State(state): State<AppState>,
|
||||
AxumPath(id): AxumPath<i64>,
|
||||
) -> Response {
|
||||
match crate::scheduled_prompts_worker::fire_now(&state.coord, id).await {
|
||||
Ok(report) => axum::Json(report).into_response(),
|
||||
Err(e) => error_response(&format!("fire schedule {id} now: {e:#}")),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize, Default)]
|
||||
struct CancelScheduleForm {
|
||||
/// `None` / absent / empty array → cancel whole schedule.
|
||||
|
|
|
|||
|
|
@ -358,6 +358,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
|||
message: format!("list scheduled prompts: {e:#}"),
|
||||
},
|
||||
},
|
||||
ManagerRequest::FireScheduleNow { id } => {
|
||||
handle_fire_schedule_now(coord, hive_sh4re::MANAGER_AGENT, *id).await
|
||||
}
|
||||
ManagerRequest::Ask {
|
||||
question,
|
||||
options,
|
||||
|
|
@ -821,10 +824,50 @@ fn handle_cancel_schedule(
|
|||
}
|
||||
}
|
||||
|
||||
/// Authorize + dispatch a `FireScheduleNow` request from the
|
||||
/// manager surface. Same ownership rules as `CancelSchedule`:
|
||||
/// requester can fire its own schedules + any owned by an agent
|
||||
/// in its subtree. The actual fan-out lives in
|
||||
/// `scheduled_prompts_worker::fire_now`.
|
||||
async fn handle_fire_schedule_now(
|
||||
coord: &Arc<Coordinator>,
|
||||
requester: &str,
|
||||
schedule_id: i64,
|
||||
) -> ManagerResponse {
|
||||
let schedule = match coord.scheduled_prompts.get(schedule_id) {
|
||||
Ok(Some(s)) => s,
|
||||
Ok(None) => {
|
||||
return ManagerResponse::Err {
|
||||
message: format!("schedule {schedule_id} not found"),
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
return ManagerResponse::Err {
|
||||
message: format!("read schedule {schedule_id}: {e:#}"),
|
||||
}
|
||||
}
|
||||
};
|
||||
if !cancel_authorized(requester, &schedule.owner) {
|
||||
return ManagerResponse::Err {
|
||||
message: format!(
|
||||
"not authorized: {requester} cannot fire schedule owned by {owner}",
|
||||
owner = schedule.owner
|
||||
),
|
||||
};
|
||||
}
|
||||
match crate::scheduled_prompts_worker::fire_now(coord, schedule_id).await {
|
||||
Ok(_report) => ManagerResponse::Ok,
|
||||
Err(e) => ManagerResponse::Err {
|
||||
message: format!("fire schedule {schedule_id} now: {e:#}"),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Permission check for `CancelSchedule` on the manager surface.
|
||||
/// `requester` (always `hm1nd` here) can cancel its own schedules.
|
||||
/// Sub-agent ownership is delegated to topology — see
|
||||
/// `crate::topology::is_descendant_of`.
|
||||
/// `crate::topology::is_descendant_of`. Also reused by
|
||||
/// `handle_fire_schedule_now` — fire-auth follows the same shape.
|
||||
fn cancel_authorized(requester: &str, owner: &str) -> bool {
|
||||
if requester == owner {
|
||||
return true;
|
||||
|
|
|
|||
|
|
@ -245,3 +245,147 @@ fn now_unix() -> i64 {
|
|||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Per-target outcome counts for one `fire_now` invocation.
|
||||
/// Returned to the operator so the dashboard can render
|
||||
/// "fired to N (M failed, K missing)" without a follow-up GET.
|
||||
#[derive(Debug, Clone, serde::Serialize)]
|
||||
pub struct FireNowReport {
|
||||
/// Targets the broker accepted the message for.
|
||||
pub ok: u32,
|
||||
/// Targets where broker.send returned an error.
|
||||
pub failed: u32,
|
||||
/// Targets that didn't resolve to a known agent (and got the
|
||||
/// operator-advisory treatment).
|
||||
pub missing: u32,
|
||||
/// Whether the one-shot was consumed by this manual fire.
|
||||
/// `true` only when the schedule was a one-shot (recurring
|
||||
/// schedules never auto-cancel on manual fire — they keep
|
||||
/// their cadence).
|
||||
pub one_shot_consumed: bool,
|
||||
}
|
||||
|
||||
/// Manual / out-of-band fire of a scheduled prompt (#467 "fire
|
||||
/// now" button). Mirrors the per-target fan-out of `fire_schedule`
|
||||
/// but skips the rearm step entirely — manual fires don't disturb
|
||||
/// a recurring schedule's rhythm. For one-shots, a manual fire
|
||||
/// **consumes** the schedule (operator intent: "send this now,
|
||||
/// the scheduled time was wrong"); recurring schedules keep their
|
||||
/// `next_fire_at_unix` unchanged.
|
||||
///
|
||||
/// `last_result` is annotated with the `manual fire:` prefix so
|
||||
/// the dashboard's per-target last-result column can distinguish
|
||||
/// scheduled fires from operator-initiated ones at a glance.
|
||||
///
|
||||
/// Returns Err if the schedule is missing, cancelled, or fully
|
||||
/// drained of active targets — the dashboard can surface those
|
||||
/// as plain 4xxs instead of pretending to fire a phantom row.
|
||||
pub async fn fire_now(
|
||||
coord: &std::sync::Arc<Coordinator>,
|
||||
schedule_id: i64,
|
||||
) -> anyhow::Result<FireNowReport> {
|
||||
let now = now_unix();
|
||||
let schedule = coord
|
||||
.scheduled_prompts
|
||||
.get(schedule_id)?
|
||||
.ok_or_else(|| anyhow::anyhow!("schedule {schedule_id} not found"))?;
|
||||
if schedule.cancelled_at_unix.is_some() {
|
||||
anyhow::bail!("schedule {schedule_id} is already cancelled");
|
||||
}
|
||||
if !schedule.targets.iter().any(|t| t.cancelled_at_unix.is_none()) {
|
||||
anyhow::bail!("schedule {schedule_id} has no active targets");
|
||||
}
|
||||
let known = known_agents_async().await;
|
||||
let mut report = FireNowReport {
|
||||
ok: 0,
|
||||
failed: 0,
|
||||
missing: 0,
|
||||
one_shot_consumed: false,
|
||||
};
|
||||
for target_row in &schedule.targets {
|
||||
if target_row.cancelled_at_unix.is_some() {
|
||||
continue;
|
||||
}
|
||||
let target = &target_row.target;
|
||||
if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
|
||||
let reason = format!("manual fire: no such agent: {target}");
|
||||
if let Err(e) = coord.scheduled_prompts.record_target_result(
|
||||
schedule_id,
|
||||
target,
|
||||
now,
|
||||
&reason,
|
||||
) {
|
||||
tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
|
||||
}
|
||||
notify_operator_missing_target(coord, &schedule, target);
|
||||
report.missing += 1;
|
||||
continue;
|
||||
}
|
||||
let msg = Message {
|
||||
from: "scheduled".to_owned(),
|
||||
to: target.clone(),
|
||||
body: schedule.body.clone(),
|
||||
in_reply_to: None,
|
||||
};
|
||||
let result = coord.broker.send(&msg);
|
||||
let result_str = match &result {
|
||||
Ok(()) => "manual fire: ok".to_owned(),
|
||||
Err(e) => format!("manual fire: broker send failed: {e:#}"),
|
||||
};
|
||||
if result.is_ok() {
|
||||
report.ok += 1;
|
||||
} else {
|
||||
report.failed += 1;
|
||||
tracing::warn!(
|
||||
schedule = schedule_id,
|
||||
%target,
|
||||
error = ?result.as_ref().err(),
|
||||
"fire_now: broker send failed (no retry — manual fires don't loop)"
|
||||
);
|
||||
}
|
||||
if let Err(e) =
|
||||
coord
|
||||
.scheduled_prompts
|
||||
.record_target_result(schedule_id, target, now, &result_str)
|
||||
{
|
||||
tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
|
||||
}
|
||||
}
|
||||
if schedule.interval_seconds.is_none() {
|
||||
// One-shot is consumed by the manual fire. Recurring
|
||||
// schedules stay untouched — their cadence is the whole
|
||||
// point and a manual fire is meant to be additive.
|
||||
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");
|
||||
} else {
|
||||
report.one_shot_consumed = true;
|
||||
}
|
||||
}
|
||||
Ok(report)
|
||||
}
|
||||
|
||||
/// Async variant of `known_agents` for `fire_now`. Same logic +
|
||||
/// same fail-closed degradation; the difference is just that the
|
||||
/// dashboard handler is genuinely async so we `await` the
|
||||
/// `lifecycle::list` directly instead of going through the
|
||||
/// `block_in_place` shim.
|
||||
async fn known_agents_async() -> std::collections::HashSet<String> {
|
||||
use std::collections::HashSet;
|
||||
let mut out: HashSet<String> = HashSet::new();
|
||||
out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
|
||||
match crate::lifecycle::list().await {
|
||||
Ok(list) => {
|
||||
for raw in list {
|
||||
if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) {
|
||||
out.insert(name.to_owned());
|
||||
} else if raw == crate::lifecycle::MANAGER_NAME {
|
||||
out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = ?e, "fire_now: container listing failed");
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue