Compare commits
10 changed files with 4 additions and 1395 deletions
|
|
@ -5,7 +5,7 @@
|
||||||
|
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use anyhow::{Context as _, Result, bail};
|
use anyhow::{Result, bail};
|
||||||
use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent, MANAGER_AGENT};
|
use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent, MANAGER_AGENT};
|
||||||
|
|
||||||
use crate::coordinator::{Coordinator, TransientKind};
|
use crate::coordinator::{Coordinator, TransientKind};
|
||||||
|
|
@ -90,15 +90,6 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
|
||||||
coord.emit_rebuild_queue_snapshot();
|
coord.emit_rebuild_queue_snapshot();
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
ApprovalKind::SchedulePrompt => {
|
|
||||||
// No queue card for SchedulePrompt — the work is a single
|
|
||||||
// sqlite insert, the actual "running" lifetime lives on
|
|
||||||
// the scheduled-prompts surface itself (worker fires it
|
|
||||||
// at the scheduled time). Run inline + fire
|
|
||||||
// `ApprovalResolved` so the approval row leaves Pending
|
|
||||||
// immediately.
|
|
||||||
run_approval_schedule_prompt(&coord, approval).await
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -136,39 +127,6 @@ pub async fn run_approval_apply_commit(
|
||||||
finish_approval(coord, &approval, result, terminal_tag, is_first_spawn)
|
finish_approval(coord, &approval, result, terminal_tag, is_first_spawn)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Inline (non-queued) handler for `ApprovalKind::SchedulePrompt`.
|
|
||||||
/// On approve, decode the `SchedulePromptPayload` JSON from the
|
|
||||||
/// approval's `commit_ref`, insert a row into `scheduled_prompts`
|
|
||||||
/// (with `source = Approval { id }`), and fire `ApprovalResolved`.
|
|
||||||
/// The worker takes over from here — fan-out at fire time.
|
|
||||||
async fn run_approval_schedule_prompt(
|
|
||||||
coord: &Coordinator,
|
|
||||||
approval: hive_sh4re::Approval,
|
|
||||||
) -> Result<()> {
|
|
||||||
let result: Result<()> = async {
|
|
||||||
let payload: hive_sh4re::SchedulePromptPayload =
|
|
||||||
serde_json::from_str(&approval.commit_ref)
|
|
||||||
.context("decode SchedulePromptPayload from approval.commit_ref")?;
|
|
||||||
coord
|
|
||||||
.scheduled_prompts
|
|
||||||
.submit(crate::scheduled_prompts::NewSchedule {
|
|
||||||
owner: approval.agent.clone(),
|
|
||||||
targets: payload.targets,
|
|
||||||
body: payload.body,
|
|
||||||
first_fire_at_unix: payload.first_fire_at_unix,
|
|
||||||
interval_seconds: payload.interval_seconds,
|
|
||||||
description: payload.description,
|
|
||||||
source: crate::scheduled_prompts::ScheduleSource::Approval {
|
|
||||||
id: approval.id,
|
|
||||||
},
|
|
||||||
})
|
|
||||||
.map(|_| ())
|
|
||||||
.context("insert scheduled prompt")
|
|
||||||
}
|
|
||||||
.await;
|
|
||||||
finish_approval(coord, &approval, result, None, false)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Worker entry point for `ApprovalKind::UpdateMetaInputs` queue
|
/// Worker entry point for `ApprovalKind::UpdateMetaInputs` queue
|
||||||
/// entries. Inputs come from the approval row's `commit_ref` field
|
/// entries. Inputs come from the approval row's `commit_ref` field
|
||||||
/// (JSON-encoded by the manager submit path), not the queue entry's
|
/// (JSON-encoded by the manager submit path), not the queue entry's
|
||||||
|
|
@ -343,7 +301,6 @@ fn finish_approval(
|
||||||
ApprovalKind::ApplyCommit => "apply_commit",
|
ApprovalKind::ApplyCommit => "apply_commit",
|
||||||
ApprovalKind::InitConfig => "init_config",
|
ApprovalKind::InitConfig => "init_config",
|
||||||
ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
||||||
ApprovalKind::SchedulePrompt => "schedule_prompt",
|
|
||||||
};
|
};
|
||||||
let sha_short = approval
|
let sha_short = approval
|
||||||
.fetched_sha
|
.fetched_sha
|
||||||
|
|
@ -393,9 +350,9 @@ fn finish_approval(
|
||||||
sha: approval.fetched_sha.clone(),
|
sha: approval.fetched_sha.clone(),
|
||||||
tag: terminal_tag,
|
tag: terminal_tag,
|
||||||
}),
|
}),
|
||||||
// UpdateMetaInputs / SchedulePrompt: ApprovalResolved already
|
// UpdateMetaInputs: ApprovalResolved already carries the result.
|
||||||
// carries the result. No separate lifecycle event needed.
|
// No separate lifecycle event needed.
|
||||||
ApprovalKind::UpdateMetaInputs | ApprovalKind::SchedulePrompt => {}
|
ApprovalKind::UpdateMetaInputs => {}
|
||||||
}
|
}
|
||||||
result
|
result
|
||||||
}
|
}
|
||||||
|
|
@ -727,7 +684,6 @@ pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()
|
||||||
ApprovalKind::ApplyCommit => "apply_commit",
|
ApprovalKind::ApplyCommit => "apply_commit",
|
||||||
ApprovalKind::InitConfig => "init_config",
|
ApprovalKind::InitConfig => "init_config",
|
||||||
ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
||||||
ApprovalKind::SchedulePrompt => "schedule_prompt",
|
|
||||||
};
|
};
|
||||||
let sha_short = sha.as_deref().map(|s| s[..s.len().min(12)].to_owned());
|
let sha_short = sha.as_deref().map(|s| s[..s.len().min(12)].to_owned());
|
||||||
let description = a.description.clone();
|
let description = a.description.clone();
|
||||||
|
|
|
||||||
|
|
@ -286,7 +286,6 @@ fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result<Approval> {
|
||||||
"spawn" => ApprovalKind::Spawn,
|
"spawn" => ApprovalKind::Spawn,
|
||||||
"init_config" => ApprovalKind::InitConfig,
|
"init_config" => ApprovalKind::InitConfig,
|
||||||
"update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
|
"update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
|
||||||
"schedule_prompt" => ApprovalKind::SchedulePrompt,
|
|
||||||
other => {
|
other => {
|
||||||
return Err(rusqlite::Error::FromSqlConversionFailure(
|
return Err(rusqlite::Error::FromSqlConversionFailure(
|
||||||
2,
|
2,
|
||||||
|
|
@ -329,7 +328,6 @@ fn kind_to_str(kind: ApprovalKind) -> &'static str {
|
||||||
ApprovalKind::Spawn => "spawn",
|
ApprovalKind::Spawn => "spawn",
|
||||||
ApprovalKind::InitConfig => "init_config",
|
ApprovalKind::InitConfig => "init_config",
|
||||||
ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
||||||
ApprovalKind::SchedulePrompt => "schedule_prompt",
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -339,7 +337,6 @@ fn kind_from_str(s: &str) -> Result<ApprovalKind> {
|
||||||
"spawn" => ApprovalKind::Spawn,
|
"spawn" => ApprovalKind::Spawn,
|
||||||
"init_config" => ApprovalKind::InitConfig,
|
"init_config" => ApprovalKind::InitConfig,
|
||||||
"update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
|
"update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
|
||||||
"schedule_prompt" => ApprovalKind::SchedulePrompt,
|
|
||||||
other => bail!("unknown approval kind '{other}'"),
|
other => bail!("unknown approval kind '{other}'"),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -37,10 +37,6 @@ pub struct Coordinator {
|
||||||
pub broker: Arc<Broker>,
|
pub broker: Arc<Broker>,
|
||||||
pub approvals: Arc<Approvals>,
|
pub approvals: Arc<Approvals>,
|
||||||
pub questions: Arc<OperatorQuestions>,
|
pub questions: Arc<OperatorQuestions>,
|
||||||
/// Scheduled-prompts queue (#444). One sqlite connection,
|
|
||||||
/// internal mutex; the worker drains due rows and the manager
|
|
||||||
/// handlers insert / cancel through the same handle.
|
|
||||||
pub scheduled_prompts: Arc<crate::scheduled_prompts::ScheduledPrompts>,
|
|
||||||
/// URL of the hyperhive flake (no fragment). Inlined into per-agent
|
/// URL of the hyperhive flake (no fragment). Inlined into per-agent
|
||||||
/// `flake.nix` files as `inputs.hyperhive.url`.
|
/// `flake.nix` files as `inputs.hyperhive.url`.
|
||||||
pub hyperhive_flake: String,
|
pub hyperhive_flake: String,
|
||||||
|
|
@ -196,15 +192,12 @@ impl Coordinator {
|
||||||
let broker = Broker::open(db_path).context("open broker")?;
|
let broker = Broker::open(db_path).context("open broker")?;
|
||||||
let approvals = Approvals::open(db_path).context("open approvals")?;
|
let approvals = Approvals::open(db_path).context("open approvals")?;
|
||||||
let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
|
let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
|
||||||
let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path)
|
|
||||||
.context("open scheduled_prompts")?;
|
|
||||||
let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
|
let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
|
||||||
let (shutdown_tx, _) = watch::channel(false);
|
let (shutdown_tx, _) = watch::channel(false);
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
broker: Arc::new(broker),
|
broker: Arc::new(broker),
|
||||||
approvals: Arc::new(approvals),
|
approvals: Arc::new(approvals),
|
||||||
questions: Arc::new(questions),
|
questions: Arc::new(questions),
|
||||||
scheduled_prompts: Arc::new(scheduled_prompts),
|
|
||||||
hyperhive_flake,
|
hyperhive_flake,
|
||||||
dashboard_port,
|
dashboard_port,
|
||||||
operator_pronouns,
|
operator_pronouns,
|
||||||
|
|
|
||||||
|
|
@ -74,8 +74,6 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
|
||||||
.route("/request-spawn", post(post_request_spawn))
|
.route("/request-spawn", post(post_request_spawn))
|
||||||
.route("/op-send", post(post_op_send))
|
.route("/op-send", post(post_op_send))
|
||||||
.route("/meta-update", post(post_meta_update))
|
.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("/dashboard/stream", get(dashboard_stream))
|
.route("/dashboard/stream", get(dashboard_stream))
|
||||||
.route("/dashboard/history", get(dashboard_history))
|
.route("/dashboard/history", get(dashboard_history))
|
||||||
// Anything not matched by the dynamic routes above falls
|
// Anything not matched by the dynamic routes above falls
|
||||||
|
|
@ -603,7 +601,6 @@ fn history_view(a: Approval) -> ApprovalHistoryView {
|
||||||
hive_sh4re::ApprovalKind::Spawn => "spawn",
|
hive_sh4re::ApprovalKind::Spawn => "spawn",
|
||||||
hive_sh4re::ApprovalKind::InitConfig => "init_config",
|
hive_sh4re::ApprovalKind::InitConfig => "init_config",
|
||||||
hive_sh4re::ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
hive_sh4re::ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
|
||||||
hive_sh4re::ApprovalKind::SchedulePrompt => "schedule_prompt",
|
|
||||||
};
|
};
|
||||||
ApprovalHistoryView {
|
ApprovalHistoryView {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
|
|
@ -664,15 +661,6 @@ async fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
|
||||||
description: a.description,
|
description: a.description,
|
||||||
requested_at: a.requested_at,
|
requested_at: a.requested_at,
|
||||||
},
|
},
|
||||||
hive_sh4re::ApprovalKind::SchedulePrompt => ApprovalView {
|
|
||||||
id: a.id,
|
|
||||||
agent: a.agent,
|
|
||||||
kind: "schedule_prompt",
|
|
||||||
sha_short: None,
|
|
||||||
diff: None,
|
|
||||||
description: a.description,
|
|
||||||
requested_at: a.requested_at,
|
|
||||||
},
|
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
out
|
out
|
||||||
|
|
@ -1391,84 +1379,6 @@ async fn api_reminders(State(state): State<AppState>) -> Response {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// `GET /api/schedules` — snapshot of every schedule for the
|
|
||||||
/// scheduled-prompts tab (#444). Returns the wire shape directly
|
|
||||||
/// so the frontend can render without an extra translation layer.
|
|
||||||
async fn api_schedules(State(state): State<AppState>) -> Response {
|
|
||||||
match state.coord.scheduled_prompts.list() {
|
|
||||||
Ok(rows) => axum::Json(
|
|
||||||
rows.into_iter()
|
|
||||||
.map(crate::manager_server::schedule_to_wire_public)
|
|
||||||
.collect::<Vec<_>>(),
|
|
||||||
)
|
|
||||||
.into_response(),
|
|
||||||
Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// `POST /api/schedules` — operator-direct schedule creation
|
|
||||||
/// (mara: "user can add them manually"). Accepts the same
|
|
||||||
/// `SchedulePromptPayload` shape as the manager request flow but
|
|
||||||
/// skips the approval gate — the operator click *is* the
|
|
||||||
/// approval. The schedule lands directly with
|
|
||||||
/// `source = Operator` and the worker picks it up at fire time.
|
|
||||||
async fn post_schedule_new(
|
|
||||||
State(state): State<AppState>,
|
|
||||||
axum::Json(payload): axum::Json<hive_sh4re::SchedulePromptPayload>,
|
|
||||||
) -> Response {
|
|
||||||
if payload.targets.is_empty() {
|
|
||||||
return error_response("schedule must have at least one target");
|
|
||||||
}
|
|
||||||
if payload.body.trim().is_empty() {
|
|
||||||
return error_response("schedule body must be non-empty");
|
|
||||||
}
|
|
||||||
if let Some(0) = payload.interval_seconds {
|
|
||||||
return error_response("interval_seconds must be > 0 (use None for one-shot)");
|
|
||||||
}
|
|
||||||
let new = crate::scheduled_prompts::NewSchedule {
|
|
||||||
owner: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
|
||||||
targets: payload.targets,
|
|
||||||
body: payload.body,
|
|
||||||
first_fire_at_unix: payload.first_fire_at_unix,
|
|
||||||
interval_seconds: payload.interval_seconds,
|
|
||||||
description: payload.description,
|
|
||||||
source: crate::scheduled_prompts::ScheduleSource::Operator,
|
|
||||||
};
|
|
||||||
match state.coord.scheduled_prompts.submit(new) {
|
|
||||||
Ok(id) => axum::Json(serde_json::json!({"id": id})).into_response(),
|
|
||||||
Err(e) => error_response(&format!("schedule submit: {e:#}")),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(serde::Deserialize, Default)]
|
|
||||||
struct CancelScheduleForm {
|
|
||||||
/// `None` / absent / empty array → cancel whole schedule.
|
|
||||||
#[serde(default)]
|
|
||||||
targets: Option<Vec<String>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// `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
|
|
||||||
/// surface enforces it for agent callers.
|
|
||||||
async fn post_schedule_cancel(
|
|
||||||
State(state): State<AppState>,
|
|
||||||
AxumPath(id): AxumPath<i64>,
|
|
||||||
body: Option<axum::Json<CancelScheduleForm>>,
|
|
||||||
) -> Response {
|
|
||||||
let targets = body
|
|
||||||
.and_then(|axum::Json(b)| b.targets)
|
|
||||||
.filter(|t| !t.is_empty());
|
|
||||||
let result = match targets.as_deref() {
|
|
||||||
Some(list) => state.coord.scheduled_prompts.cancel_targets(id, list),
|
|
||||||
None => state.coord.scheduled_prompts.cancel_all(id),
|
|
||||||
};
|
|
||||||
match result {
|
|
||||||
Ok(()) => (StatusCode::OK, "ok").into_response(),
|
|
||||||
Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Same-origin proxy that fetches the named agent's
|
/// Same-origin proxy that fetches the named agent's
|
||||||
/// `GET /api/state` and forwards only the `links` field to the
|
/// `GET /api/state` and forwards only the `links` field to the
|
||||||
/// dashboard JS (issue #262). Lets the agent backend stay the
|
/// dashboard JS (issue #262). Lets the agent backend stay the
|
||||||
|
|
|
||||||
|
|
@ -21,8 +21,6 @@ mod stats_vacuum;
|
||||||
mod flake_check;
|
mod flake_check;
|
||||||
mod forge;
|
mod forge;
|
||||||
mod lifecycle;
|
mod lifecycle;
|
||||||
mod scheduled_prompts;
|
|
||||||
mod scheduled_prompts_worker;
|
|
||||||
mod limits;
|
mod limits;
|
||||||
mod loose_ends;
|
mod loose_ends;
|
||||||
mod manager_server;
|
mod manager_server;
|
||||||
|
|
@ -237,10 +235,6 @@ async fn cmd_serve(
|
||||||
// Reminder scheduler: drains due reminders + handles
|
// Reminder scheduler: drains due reminders + handles
|
||||||
// file_path payload persistence. See reminder_scheduler.rs.
|
// file_path payload persistence. See reminder_scheduler.rs.
|
||||||
reminder_scheduler::spawn(coord.clone());
|
reminder_scheduler::spawn(coord.clone());
|
||||||
// Scheduled-prompts worker: drains due scheduled_prompts rows
|
|
||||||
// and fans the body out to each active target's inbox. See
|
|
||||||
// scheduled_prompts_worker.rs (#444).
|
|
||||||
scheduled_prompts_worker::spawn(coord.clone());
|
|
||||||
// Rebuild-queue worker: drains the global rebuild/meta-update/
|
// Rebuild-queue worker: drains the global rebuild/meta-update/
|
||||||
// spawn queue FIFO so hive-c0re never runs two heavyweight
|
// spawn queue FIFO so hive-c0re never runs two heavyweight
|
||||||
// container ops concurrently. Existing rebuild call sites
|
// container ops concurrently. Existing rebuild call sites
|
||||||
|
|
|
||||||
|
|
@ -344,20 +344,6 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
||||||
);
|
);
|
||||||
ManagerResponse::Ok
|
ManagerResponse::Ok
|
||||||
}
|
}
|
||||||
ManagerRequest::RequestSchedulePrompt(payload) => {
|
|
||||||
handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload).await
|
|
||||||
}
|
|
||||||
ManagerRequest::CancelSchedule { id, targets } => {
|
|
||||||
handle_cancel_schedule(coord, hive_sh4re::MANAGER_AGENT, *id, targets.as_deref())
|
|
||||||
}
|
|
||||||
ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() {
|
|
||||||
Ok(schedules) => ManagerResponse::Schedules {
|
|
||||||
schedules: schedules.into_iter().map(schedule_to_wire).collect(),
|
|
||||||
},
|
|
||||||
Err(e) => ManagerResponse::Err {
|
|
||||||
message: format!("list scheduled prompts: {e:#}"),
|
|
||||||
},
|
|
||||||
},
|
|
||||||
ManagerRequest::Ask {
|
ManagerRequest::Ask {
|
||||||
question,
|
question,
|
||||||
options,
|
options,
|
||||||
|
|
@ -706,179 +692,6 @@ async fn submit_apply_commit(
|
||||||
Ok((id, sha))
|
Ok((id, sha))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Submit a `RequestSchedulePrompt` payload as an `ApprovalKind::SchedulePrompt`
|
|
||||||
/// row. Encodes the payload into the approval's `commit_ref` so the
|
|
||||||
/// approve handler can re-parse it without a side table. Validates
|
|
||||||
/// inputs (non-empty targets, non-empty body, sane interval) at
|
|
||||||
/// submit time — the operator should never see a malformed schedule
|
|
||||||
/// pending approval.
|
|
||||||
async fn handle_request_schedule_prompt(
|
|
||||||
coord: &Arc<Coordinator>,
|
|
||||||
requester: &str,
|
|
||||||
payload: &hive_sh4re::SchedulePromptPayload,
|
|
||||||
) -> ManagerResponse {
|
|
||||||
if payload.targets.is_empty() {
|
|
||||||
return ManagerResponse::Err {
|
|
||||||
message: "schedule must have at least one target".into(),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
if payload.body.trim().is_empty() {
|
|
||||||
return ManagerResponse::Err {
|
|
||||||
message: "schedule body must be non-empty".into(),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
if let Some(0) = payload.interval_seconds {
|
|
||||||
return ManagerResponse::Err {
|
|
||||||
message: "interval_seconds must be > 0 (use None for one-shot)".into(),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
let commit_ref = match serde_json::to_string(payload) {
|
|
||||||
Ok(s) => s,
|
|
||||||
Err(e) => {
|
|
||||||
return ManagerResponse::Err {
|
|
||||||
message: format!("encode SchedulePromptPayload: {e:#}"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
};
|
|
||||||
let id = match coord.approvals.submit_kind(
|
|
||||||
requester,
|
|
||||||
hive_sh4re::ApprovalKind::SchedulePrompt,
|
|
||||||
&commit_ref,
|
|
||||||
payload.description.as_deref(),
|
|
||||||
) {
|
|
||||||
Ok(id) => id,
|
|
||||||
Err(e) => {
|
|
||||||
return ManagerResponse::Err {
|
|
||||||
message: format!("queue schedule_prompt approval: {e:#}"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
};
|
|
||||||
tracing::info!(
|
|
||||||
%id,
|
|
||||||
requester,
|
|
||||||
targets = ?payload.targets,
|
|
||||||
first_fire_at = payload.first_fire_at_unix,
|
|
||||||
interval = ?payload.interval_seconds,
|
|
||||||
"schedule_prompt approval queued"
|
|
||||||
);
|
|
||||||
coord.emit_approval_added(
|
|
||||||
id,
|
|
||||||
requester,
|
|
||||||
"schedule_prompt",
|
|
||||||
None,
|
|
||||||
None,
|
|
||||||
payload.description.clone(),
|
|
||||||
);
|
|
||||||
ManagerResponse::Ok
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Cancel a schedule (whole or per-target). Manager-surface
|
|
||||||
/// authorization: a manager can cancel its own schedules + any
|
|
||||||
/// schedule whose owner is one of its sub-agents (topology-walked).
|
|
||||||
/// The operator surface bypasses this and can cancel anything;
|
|
||||||
/// agents reaching this path through the manager get the
|
|
||||||
/// topology-scoped check.
|
|
||||||
fn handle_cancel_schedule(
|
|
||||||
coord: &Arc<Coordinator>,
|
|
||||||
requester: &str,
|
|
||||||
schedule_id: i64,
|
|
||||||
targets: Option<&[String]>,
|
|
||||||
) -> 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 cancel schedule owned by {owner}",
|
|
||||||
owner = schedule.owner
|
|
||||||
),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
let result = match targets {
|
|
||||||
Some(list) if !list.is_empty() => coord
|
|
||||||
.scheduled_prompts
|
|
||||||
.cancel_targets(schedule_id, list)
|
|
||||||
.map_err(|e| format!("cancel targets: {e:#}")),
|
|
||||||
_ => coord
|
|
||||||
.scheduled_prompts
|
|
||||||
.cancel_all(schedule_id)
|
|
||||||
.map_err(|e| format!("cancel all: {e:#}")),
|
|
||||||
};
|
|
||||||
match result {
|
|
||||||
Ok(()) => ManagerResponse::Ok,
|
|
||||||
Err(message) => ManagerResponse::Err { message },
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// 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`.
|
|
||||||
fn cancel_authorized(requester: &str, owner: &str) -> bool {
|
|
||||||
if requester == owner {
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
if requester == hive_sh4re::OPERATOR_RECIPIENT {
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
// Manager can cancel anything owned by an agent in its subtree.
|
|
||||||
// For the current single-manager topology that covers everything,
|
|
||||||
// but the check stays correct as the tree grows.
|
|
||||||
crate::topology::is_descendant_of(owner, requester)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Map a `scheduled_prompts::Schedule` to its public wire shape.
|
|
||||||
/// Field-by-field copy — the two types are intentionally identical;
|
|
||||||
/// the separation keeps hive-sh4re free of hive-c0re-internal types.
|
|
||||||
/// Public alias `schedule_to_wire_public` re-exports for
|
|
||||||
/// `dashboard.rs::api_schedules` without crossing the module
|
|
||||||
/// boundary into the manager-server file.
|
|
||||||
pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
|
|
||||||
schedule_to_wire(s)
|
|
||||||
}
|
|
||||||
|
|
||||||
fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
|
|
||||||
hive_sh4re::WireSchedule {
|
|
||||||
id: s.id,
|
|
||||||
owner: s.owner,
|
|
||||||
body: s.body,
|
|
||||||
interval_seconds: s.interval_seconds,
|
|
||||||
next_fire_at_unix: s.next_fire_at_unix,
|
|
||||||
created_at_unix: s.created_at_unix,
|
|
||||||
source: match s.source {
|
|
||||||
crate::scheduled_prompts::ScheduleSource::Operator => {
|
|
||||||
hive_sh4re::WireScheduleSource::Operator
|
|
||||||
}
|
|
||||||
crate::scheduled_prompts::ScheduleSource::Approval { id } => {
|
|
||||||
hive_sh4re::WireScheduleSource::Approval { id }
|
|
||||||
}
|
|
||||||
},
|
|
||||||
cancelled_at_unix: s.cancelled_at_unix,
|
|
||||||
description: s.description,
|
|
||||||
targets: s
|
|
||||||
.targets
|
|
||||||
.into_iter()
|
|
||||||
.map(|t| hive_sh4re::WireScheduleTarget {
|
|
||||||
target: t.target,
|
|
||||||
cancelled_at_unix: t.cancelled_at_unix,
|
|
||||||
last_fired_at_unix: t.last_fired_at_unix,
|
|
||||||
last_result: t.last_result,
|
|
||||||
})
|
|
||||||
.collect(),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// On `Ask { ttl_seconds: Some(n) }`, sleep n seconds and then try to
|
/// On `Ask { ttl_seconds: Some(n) }`, sleep n seconds and then try to
|
||||||
/// resolve the question with `[expired]`. If the operator (or any
|
/// resolve the question with `[expired]`. If the operator (or any
|
||||||
/// other path) already answered it, `answer()` returns Err and we
|
/// other path) already answered it, `answer()` returns Err and we
|
||||||
|
|
|
||||||
|
|
@ -1,680 +0,0 @@
|
||||||
//! Scheduled prompts (closes #444). Persistent sqlite queue of
|
|
||||||
//! `(fire_at, targets, body)` rows that the worker fans out as
|
|
||||||
//! broker `Message`s to each target's inbox at fire time. Recurring
|
|
||||||
//! schedules carry `interval_seconds` and re-arm `next_fire_at` on
|
|
||||||
//! delivery; one-shots are reaped.
|
|
||||||
//!
|
|
||||||
//! ## Three submit paths
|
|
||||||
//!
|
|
||||||
//! - **Operator-direct** (`source = Operator`): the operator adds
|
|
||||||
//! a schedule through the dashboard form. Lands in the table
|
|
||||||
//! immediately, no approval gate.
|
|
||||||
//! - **Agent-requested** (`source = Approval { id }`): a sub-agent
|
|
||||||
//! (or the manager) submits a `RequestSchedulePrompt` through the
|
|
||||||
//! manager socket. An `ApprovalKind::SchedulePrompt` row is
|
|
||||||
//! queued; on approve, hive-c0re inserts the schedule row with
|
|
||||||
//! `source = Approval { id: approval_id }` so the audit trail
|
|
||||||
//! points back at the operator decision.
|
|
||||||
//! - **No self-target shortcut**: even agent-self schedules need
|
|
||||||
//! approval. The existing `remind` MCP tool stays the quick
|
|
||||||
//! self-wake path; this module is the bigger, multi-recipient,
|
|
||||||
//! operator-visible thing.
|
|
||||||
//!
|
|
||||||
//! ## Catch-up clamp (missed-while-down)
|
|
||||||
//!
|
|
||||||
//! When hive-c0re comes back from being down, the worker sees rows
|
|
||||||
//! whose `next_fire_at` is well in the past. For recurring rows
|
|
||||||
//! that would mean firing N delayed pulses in a row — spammy and
|
|
||||||
//! useless. Instead the worker fires ONCE per row and bumps
|
|
||||||
//! `next_fire_at` to the next interval slot ≥ `now`, recording how
|
|
||||||
//! many cycles were skipped in `last_result`. Operators see "fired
|
|
||||||
//! late, caught up from 17 skipped" instead of 17 wake-up storms.
|
|
||||||
//!
|
|
||||||
//! ## Per-target state
|
|
||||||
//!
|
|
||||||
//! `targets` is its own table so partial cancellation flips a
|
|
||||||
//! single row + so the dashboard can show last-fired / last-result
|
|
||||||
//! per recipient. Cancelling every target reaps the parent row on
|
|
||||||
//! the next worker pass.
|
|
||||||
|
|
||||||
use std::path::Path;
|
|
||||||
use std::sync::Mutex;
|
|
||||||
|
|
||||||
use anyhow::{Context, Result, bail};
|
|
||||||
use rusqlite::{Connection, OptionalExtension, params};
|
|
||||||
use serde::{Deserialize, Serialize};
|
|
||||||
|
|
||||||
const SCHEMA: &str = r"
|
|
||||||
CREATE TABLE IF NOT EXISTS scheduled_prompts (
|
|
||||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
||||||
owner TEXT NOT NULL,
|
|
||||||
body TEXT NOT NULL,
|
|
||||||
interval_seconds INTEGER,
|
|
||||||
next_fire_at_unix INTEGER NOT NULL,
|
|
||||||
created_at_unix INTEGER NOT NULL,
|
|
||||||
source TEXT NOT NULL,
|
|
||||||
cancelled_at_unix INTEGER,
|
|
||||||
description TEXT
|
|
||||||
);
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_scheduled_due
|
|
||||||
ON scheduled_prompts (next_fire_at_unix)
|
|
||||||
WHERE cancelled_at_unix IS NULL;
|
|
||||||
|
|
||||||
CREATE TABLE IF NOT EXISTS scheduled_prompt_targets (
|
|
||||||
schedule_id INTEGER NOT NULL,
|
|
||||||
target TEXT NOT NULL,
|
|
||||||
cancelled_at_unix INTEGER,
|
|
||||||
last_fired_at_unix INTEGER,
|
|
||||||
last_result TEXT,
|
|
||||||
PRIMARY KEY (schedule_id, target),
|
|
||||||
FOREIGN KEY (schedule_id) REFERENCES scheduled_prompts(id) ON DELETE CASCADE
|
|
||||||
);
|
|
||||||
";
|
|
||||||
|
|
||||||
/// One scheduled-prompt row + its current target set. Returned by
|
|
||||||
/// `list` / `get`; the per-target last-result blob is suitable for
|
|
||||||
/// rendering on the dashboard without a second query.
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct Schedule {
|
|
||||||
pub id: i64,
|
|
||||||
/// `"operator"` or an agent name — drives cancel-permission
|
|
||||||
/// checks. For approval-sourced rows this is the requesting
|
|
||||||
/// agent (the `Approval.agent` at submit time).
|
|
||||||
pub owner: String,
|
|
||||||
pub body: String,
|
|
||||||
/// `None` = one-shot, deleted after first fire.
|
|
||||||
/// `Some(n)` = recurring every `n` seconds.
|
|
||||||
pub interval_seconds: Option<u64>,
|
|
||||||
pub next_fire_at_unix: i64,
|
|
||||||
pub created_at_unix: i64,
|
|
||||||
pub source: ScheduleSource,
|
|
||||||
/// Set when the *entire* schedule was cancelled (all targets
|
|
||||||
/// flipped, or operator cancel-all). Worker reaps these on the
|
|
||||||
/// next pass.
|
|
||||||
pub cancelled_at_unix: Option<i64>,
|
|
||||||
pub description: Option<String>,
|
|
||||||
pub targets: Vec<ScheduleTarget>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct ScheduleTarget {
|
|
||||||
pub target: String,
|
|
||||||
pub cancelled_at_unix: Option<i64>,
|
|
||||||
pub last_fired_at_unix: Option<i64>,
|
|
||||||
pub last_result: Option<String>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
|
||||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
|
||||||
pub enum ScheduleSource {
|
|
||||||
Operator,
|
|
||||||
Approval { id: i64 },
|
|
||||||
}
|
|
||||||
|
|
||||||
impl ScheduleSource {
|
|
||||||
fn to_db_string(&self) -> String {
|
|
||||||
match self {
|
|
||||||
ScheduleSource::Operator => "operator".to_owned(),
|
|
||||||
ScheduleSource::Approval { id } => format!("approval:{id}"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn from_db_string(s: &str) -> Self {
|
|
||||||
if let Some(rest) = s.strip_prefix("approval:")
|
|
||||||
&& let Ok(id) = rest.parse::<i64>()
|
|
||||||
{
|
|
||||||
return ScheduleSource::Approval { id };
|
|
||||||
}
|
|
||||||
// Unknown / "operator" / corrupt → operator (best the row
|
|
||||||
// can do without an unparseable-source variant).
|
|
||||||
ScheduleSource::Operator
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Submission payload — everything the caller knows at insert time.
|
|
||||||
/// Used by both the operator-direct path and the approval-flow
|
|
||||||
/// path; the latter sets `source = Approval { id }`.
|
|
||||||
#[derive(Debug, Clone)]
|
|
||||||
pub struct NewSchedule {
|
|
||||||
pub owner: String,
|
|
||||||
pub targets: Vec<String>,
|
|
||||||
pub body: String,
|
|
||||||
pub first_fire_at_unix: i64,
|
|
||||||
pub interval_seconds: Option<u64>,
|
|
||||||
pub description: Option<String>,
|
|
||||||
pub source: ScheduleSource,
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct ScheduledPrompts {
|
|
||||||
conn: Mutex<Connection>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl ScheduledPrompts {
|
|
||||||
pub fn open(path: &Path) -> Result<Self> {
|
|
||||||
if let Some(parent) = path.parent() {
|
|
||||||
std::fs::create_dir_all(parent).with_context(|| {
|
|
||||||
format!("create scheduled_prompts db parent {}", parent.display())
|
|
||||||
})?;
|
|
||||||
}
|
|
||||||
let conn = Connection::open(path)
|
|
||||||
.with_context(|| format!("open scheduled_prompts db {}", path.display()))?;
|
|
||||||
// Required for ON DELETE CASCADE to actually fire — sqlite
|
|
||||||
// ships with FKs disabled per connection by default.
|
|
||||||
conn.execute_batch("PRAGMA foreign_keys = ON;")
|
|
||||||
.context("enable foreign keys")?;
|
|
||||||
conn.execute_batch(SCHEMA)
|
|
||||||
.context("apply scheduled_prompts schema")?;
|
|
||||||
Ok(Self {
|
|
||||||
conn: Mutex::new(conn),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Insert a new schedule. Returns the new id. Empty `targets` is
|
|
||||||
/// rejected — a schedule with no recipients would silently
|
|
||||||
/// never fan out, masking caller bugs.
|
|
||||||
pub fn submit(&self, new: NewSchedule) -> Result<i64> {
|
|
||||||
if new.targets.is_empty() {
|
|
||||||
bail!("schedule must have at least one target");
|
|
||||||
}
|
|
||||||
let mut conn = self.conn.lock().unwrap();
|
|
||||||
let tx = conn.transaction()?;
|
|
||||||
tx.execute(
|
|
||||||
"INSERT INTO scheduled_prompts
|
|
||||||
(owner, body, interval_seconds, next_fire_at_unix,
|
|
||||||
created_at_unix, source, description)
|
|
||||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
|
|
||||||
params![
|
|
||||||
new.owner,
|
|
||||||
new.body,
|
|
||||||
new.interval_seconds.map(i64::try_from).and_then(Result::ok),
|
|
||||||
new.first_fire_at_unix,
|
|
||||||
now_unix(),
|
|
||||||
new.source.to_db_string(),
|
|
||||||
new.description,
|
|
||||||
],
|
|
||||||
)?;
|
|
||||||
let id = tx.last_insert_rowid();
|
|
||||||
for target in &new.targets {
|
|
||||||
tx.execute(
|
|
||||||
"INSERT INTO scheduled_prompt_targets (schedule_id, target)
|
|
||||||
VALUES (?1, ?2)",
|
|
||||||
params![id, target],
|
|
||||||
)?;
|
|
||||||
}
|
|
||||||
tx.commit()?;
|
|
||||||
Ok(id)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Fetch a single schedule by id (with its target rows).
|
|
||||||
/// `Ok(None)` for a non-existent / already-reaped id.
|
|
||||||
pub fn get(&self, id: i64) -> Result<Option<Schedule>> {
|
|
||||||
let conn = self.conn.lock().unwrap();
|
|
||||||
let row = conn
|
|
||||||
.query_row(
|
|
||||||
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
|
|
||||||
created_at_unix, source, cancelled_at_unix, description
|
|
||||||
FROM scheduled_prompts WHERE id = ?1",
|
|
||||||
params![id],
|
|
||||||
row_to_schedule_header,
|
|
||||||
)
|
|
||||||
.optional()?;
|
|
||||||
let Some(mut s) = row else {
|
|
||||||
return Ok(None);
|
|
||||||
};
|
|
||||||
s.targets = load_targets(&conn, id)?;
|
|
||||||
Ok(Some(s))
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Every active (non-globally-cancelled) schedule in insert
|
|
||||||
/// order. Used by the dashboard list view + the cancel-auth
|
|
||||||
/// check (the latter only needs the header but list() is the
|
|
||||||
/// shared hot path).
|
|
||||||
pub fn list(&self) -> Result<Vec<Schedule>> {
|
|
||||||
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
|
|
||||||
FROM scheduled_prompts
|
|
||||||
ORDER BY next_fire_at_unix ASC, id ASC",
|
|
||||||
)?;
|
|
||||||
let rows = stmt.query_map([], row_to_schedule_header)?;
|
|
||||||
let mut out = Vec::new();
|
|
||||||
for row in rows {
|
|
||||||
let mut s = row?;
|
|
||||||
s.targets = load_targets(&conn, s.id)?;
|
|
||||||
out.push(s);
|
|
||||||
}
|
|
||||||
Ok(out)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Pop the set of active rows that are due (`next_fire_at_unix <= now`)
|
|
||||||
/// up to `limit`. Read-only — the worker calls `mark_fired`
|
|
||||||
/// after each successful fan-out so the rows reappear on the
|
|
||||||
/// next tick when re-armed.
|
|
||||||
pub fn due(&self, now: i64, limit: u64) -> Result<Vec<Schedule>> {
|
|
||||||
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
|
|
||||||
FROM scheduled_prompts
|
|
||||||
WHERE cancelled_at_unix IS NULL AND next_fire_at_unix <= ?1
|
|
||||||
ORDER BY next_fire_at_unix ASC, id ASC
|
|
||||||
LIMIT ?2",
|
|
||||||
)?;
|
|
||||||
let rows = stmt.query_map(params![now, limit], row_to_schedule_header)?;
|
|
||||||
let mut out = Vec::new();
|
|
||||||
for row in rows {
|
|
||||||
let mut s = row?;
|
|
||||||
s.targets = load_targets(&conn, s.id)?;
|
|
||||||
out.push(s);
|
|
||||||
}
|
|
||||||
Ok(out)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Record a per-target fan-out result. `result` is "ok" or a
|
|
||||||
/// short error string; surfaces on the dashboard last-result
|
|
||||||
/// column. No-op for a target row that's already cancelled.
|
|
||||||
pub fn record_target_result(
|
|
||||||
&self,
|
|
||||||
schedule_id: i64,
|
|
||||||
target: &str,
|
|
||||||
fired_at_unix: i64,
|
|
||||||
result: &str,
|
|
||||||
) -> Result<()> {
|
|
||||||
let conn = self.conn.lock().unwrap();
|
|
||||||
conn.execute(
|
|
||||||
"UPDATE scheduled_prompt_targets
|
|
||||||
SET last_fired_at_unix = ?1, last_result = ?2
|
|
||||||
WHERE schedule_id = ?3 AND target = ?4 AND cancelled_at_unix IS NULL",
|
|
||||||
params![fired_at_unix, result, schedule_id, target],
|
|
||||||
)?;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Advance a recurring schedule's `next_fire_at` to the smallest
|
|
||||||
/// multiple-of-interval > `from`. Returns the count of skipped
|
|
||||||
/// cycles (≥ 0); the worker stamps that into the per-row
|
|
||||||
/// last_result so operators see "caught up from N missed".
|
|
||||||
///
|
|
||||||
/// For one-shots (`interval_seconds IS NULL`) this is a no-op
|
|
||||||
/// at the SQL level; callers should `delete` them after fan-out
|
|
||||||
/// instead.
|
|
||||||
pub fn rearm(&self, id: i64, from_unix: i64) -> Result<u64> {
|
|
||||||
let conn = self.conn.lock().unwrap();
|
|
||||||
let row: Option<(Option<i64>, i64)> = conn
|
|
||||||
.query_row(
|
|
||||||
"SELECT interval_seconds, next_fire_at_unix
|
|
||||||
FROM scheduled_prompts WHERE id = ?1",
|
|
||||||
params![id],
|
|
||||||
|row| Ok((row.get(0)?, row.get(1)?)),
|
|
||||||
)
|
|
||||||
.optional()?;
|
|
||||||
let Some((interval, current_next)) = row else {
|
|
||||||
return Ok(0);
|
|
||||||
};
|
|
||||||
let Some(interval) = interval.filter(|&i| i > 0) else {
|
|
||||||
return Ok(0);
|
|
||||||
};
|
|
||||||
// Smallest multiple-of-interval > `from_unix`. Catch-up
|
|
||||||
// semantics: when from_unix > current_next, every step in
|
|
||||||
// between gets counted as a skipped cycle.
|
|
||||||
let mut next = current_next + interval;
|
|
||||||
let mut skipped: u64 = 0;
|
|
||||||
while next <= from_unix {
|
|
||||||
next += interval;
|
|
||||||
skipped += 1;
|
|
||||||
}
|
|
||||||
conn.execute(
|
|
||||||
"UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2",
|
|
||||||
params![next, id],
|
|
||||||
)?;
|
|
||||||
Ok(skipped)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Delete a one-shot row after its single fire. Cascades the
|
|
||||||
/// target rows via the FK constraint. Idempotent on a missing
|
|
||||||
/// id.
|
|
||||||
pub fn delete(&self, id: i64) -> Result<()> {
|
|
||||||
let conn = self.conn.lock().unwrap();
|
|
||||||
conn.execute(
|
|
||||||
"DELETE FROM scheduled_prompts WHERE id = ?1",
|
|
||||||
params![id],
|
|
||||||
)?;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Cancel the entire schedule (all targets flipped + the parent
|
|
||||||
/// row marked cancelled). Idempotent; safe to call on an
|
|
||||||
/// already-cancelled row.
|
|
||||||
pub fn cancel_all(&self, id: i64) -> Result<()> {
|
|
||||||
let mut conn = self.conn.lock().unwrap();
|
|
||||||
let now = now_unix();
|
|
||||||
let tx = conn.transaction()?;
|
|
||||||
tx.execute(
|
|
||||||
"UPDATE scheduled_prompts
|
|
||||||
SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
|
|
||||||
WHERE id = ?2",
|
|
||||||
params![now, id],
|
|
||||||
)?;
|
|
||||||
tx.execute(
|
|
||||||
"UPDATE scheduled_prompt_targets
|
|
||||||
SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
|
|
||||||
WHERE schedule_id = ?2",
|
|
||||||
params![now, id],
|
|
||||||
)?;
|
|
||||||
tx.commit()?;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Cancel a subset of a schedule's targets. When the last
|
|
||||||
/// active target is cancelled, the parent row is auto-cancelled
|
|
||||||
/// too (so the worker reaps it). Unknown targets in the list
|
|
||||||
/// are silently skipped.
|
|
||||||
pub fn cancel_targets(&self, id: i64, targets: &[String]) -> Result<()> {
|
|
||||||
let mut conn = self.conn.lock().unwrap();
|
|
||||||
let now = now_unix();
|
|
||||||
let tx = conn.transaction()?;
|
|
||||||
for target in targets {
|
|
||||||
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 every target on this schedule is now cancelled, flip
|
|
||||||
// the parent so the worker stops scanning it.
|
|
||||||
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(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// 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
|
|
||||||
/// extra periodic vacuum task.
|
|
||||||
pub fn reap_cancelled(&self, older_than_unix: i64) -> Result<usize> {
|
|
||||||
let conn = self.conn.lock().unwrap();
|
|
||||||
let n = conn.execute(
|
|
||||||
"DELETE FROM scheduled_prompts
|
|
||||||
WHERE cancelled_at_unix IS NOT NULL
|
|
||||||
AND cancelled_at_unix <= ?1",
|
|
||||||
params![older_than_unix],
|
|
||||||
)?;
|
|
||||||
Ok(n)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result<Schedule> {
|
|
||||||
let interval: Option<i64> = row.get(3)?;
|
|
||||||
let source_str: String = row.get(6)?;
|
|
||||||
Ok(Schedule {
|
|
||||||
id: row.get(0)?,
|
|
||||||
owner: row.get(1)?,
|
|
||||||
body: row.get(2)?,
|
|
||||||
interval_seconds: interval.and_then(|i| u64::try_from(i).ok()),
|
|
||||||
next_fire_at_unix: row.get(4)?,
|
|
||||||
created_at_unix: row.get(5)?,
|
|
||||||
source: ScheduleSource::from_db_string(&source_str),
|
|
||||||
cancelled_at_unix: row.get(7)?,
|
|
||||||
description: row.get(8)?,
|
|
||||||
targets: Vec::new(),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
fn load_targets(conn: &Connection, schedule_id: i64) -> Result<Vec<ScheduleTarget>> {
|
|
||||||
let mut stmt = conn.prepare(
|
|
||||||
"SELECT target, cancelled_at_unix, last_fired_at_unix, last_result
|
|
||||||
FROM scheduled_prompt_targets
|
|
||||||
WHERE schedule_id = ?1
|
|
||||||
ORDER BY target ASC",
|
|
||||||
)?;
|
|
||||||
let rows = stmt.query_map(params![schedule_id], |row| {
|
|
||||||
Ok(ScheduleTarget {
|
|
||||||
target: row.get(0)?,
|
|
||||||
cancelled_at_unix: row.get(1)?,
|
|
||||||
last_fired_at_unix: row.get(2)?,
|
|
||||||
last_result: row.get(3)?,
|
|
||||||
})
|
|
||||||
})?;
|
|
||||||
let mut out = Vec::new();
|
|
||||||
for row in rows {
|
|
||||||
out.push(row?);
|
|
||||||
}
|
|
||||||
Ok(out)
|
|
||||||
}
|
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
|
||||||
mod tests {
|
|
||||||
use super::*;
|
|
||||||
use tempfile::TempDir;
|
|
||||||
|
|
||||||
fn open() -> (TempDir, ScheduledPrompts) {
|
|
||||||
let dir = TempDir::new().expect("tempdir");
|
|
||||||
let path = dir.path().join("schedules.sqlite");
|
|
||||||
let db = ScheduledPrompts::open(&path).expect("open");
|
|
||||||
(dir, db)
|
|
||||||
}
|
|
||||||
|
|
||||||
fn submit_one_shot(db: &ScheduledPrompts, fire_at: i64, targets: &[&str]) -> i64 {
|
|
||||||
db.submit(NewSchedule {
|
|
||||||
owner: "operator".into(),
|
|
||||||
targets: targets.iter().map(|t| (*t).to_owned()).collect(),
|
|
||||||
body: "wake".into(),
|
|
||||||
first_fire_at_unix: fire_at,
|
|
||||||
interval_seconds: None,
|
|
||||||
description: None,
|
|
||||||
source: ScheduleSource::Operator,
|
|
||||||
})
|
|
||||||
.expect("submit")
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn submit_and_get_round_trips_targets() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
assert_eq!(s.id, id);
|
|
||||||
assert_eq!(s.targets.len(), 2);
|
|
||||||
assert_eq!(s.targets[0].target, "alice");
|
|
||||||
assert_eq!(s.targets[1].target, "bob");
|
|
||||||
assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_none()));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn submit_rejects_empty_targets() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let err = db
|
|
||||||
.submit(NewSchedule {
|
|
||||||
owner: "operator".into(),
|
|
||||||
targets: Vec::new(),
|
|
||||||
body: "wake".into(),
|
|
||||||
first_fire_at_unix: 100,
|
|
||||||
interval_seconds: None,
|
|
||||||
description: None,
|
|
||||||
source: ScheduleSource::Operator,
|
|
||||||
})
|
|
||||||
.unwrap_err();
|
|
||||||
assert!(format!("{err:#}").contains("at least one target"));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn due_returns_only_past_active_rows() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let _future = submit_one_shot(&db, 1000, &["alice"]);
|
|
||||||
let past = submit_one_shot(&db, 50, &["alice"]);
|
|
||||||
let cancelled_past = submit_one_shot(&db, 50, &["bob"]);
|
|
||||||
db.cancel_all(cancelled_past).expect("cancel");
|
|
||||||
let due = db.due(100, 10).expect("due");
|
|
||||||
assert_eq!(due.len(), 1);
|
|
||||||
assert_eq!(due[0].id, past);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn rearm_handles_catch_up() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
// Recurring every 60s, last 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");
|
|
||||||
// Worker comes back at t=400 — 5 missed cycles (160, 220,
|
|
||||||
// 280, 340, 400) → next should be 460, skipped = 5.
|
|
||||||
let skipped = db.rearm(id, 400).expect("rearm");
|
|
||||||
assert_eq!(skipped, 5);
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
assert_eq!(s.next_fire_at_unix, 460);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn rearm_advances_one_step_when_caught_up() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
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");
|
|
||||||
// Worker fires right at t=100 — single advance to t=160.
|
|
||||||
let skipped = db.rearm(id, 100).expect("rearm");
|
|
||||||
assert_eq!(skipped, 0);
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
assert_eq!(s.next_fire_at_unix, 160);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn rearm_is_a_no_op_for_one_shots() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let id = submit_one_shot(&db, 100, &["alice"]);
|
|
||||||
let skipped = db.rearm(id, 1000).expect("rearm");
|
|
||||||
assert_eq!(skipped, 0);
|
|
||||||
// next_fire_at unchanged — one-shots are reaped via delete().
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
assert_eq!(s.next_fire_at_unix, 100);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn cancel_targets_auto_cancels_parent_when_last_drops() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
|
|
||||||
db.cancel_targets(id, &["alice".to_owned()]).expect("cancel");
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
// Parent still active (bob remains).
|
|
||||||
assert!(s.cancelled_at_unix.is_none());
|
|
||||||
db.cancel_targets(id, &["bob".to_owned()]).expect("cancel");
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
// Parent auto-cancels once every target is gone.
|
|
||||||
assert!(s.cancelled_at_unix.is_some());
|
|
||||||
assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some()));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn cancel_all_flips_parent_and_targets() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
|
|
||||||
db.cancel_all(id).expect("cancel");
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
assert!(s.cancelled_at_unix.is_some());
|
|
||||||
assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some()));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn delete_cascades_targets() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
|
|
||||||
db.delete(id).expect("delete");
|
|
||||||
assert!(db.get(id).expect("get").is_none());
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn record_target_result_skips_cancelled_targets() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let id = submit_one_shot(&db, 100, &["alice", "bob"]);
|
|
||||||
db.cancel_targets(id, &["alice".to_owned()]).expect("cancel");
|
|
||||||
db.record_target_result(id, "alice", 200, "ok")
|
|
||||||
.expect("record alice");
|
|
||||||
db.record_target_result(id, "bob", 200, "ok")
|
|
||||||
.expect("record bob");
|
|
||||||
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();
|
|
||||||
// Cancelled targets do NOT get last-result writes.
|
|
||||||
assert!(alice.last_fired_at_unix.is_none());
|
|
||||||
assert!(alice.last_result.is_none());
|
|
||||||
assert_eq!(bob.last_fired_at_unix, Some(200));
|
|
||||||
assert_eq!(bob.last_result.as_deref(), Some("ok"));
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn approval_source_round_trips() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let id = db
|
|
||||||
.submit(NewSchedule {
|
|
||||||
owner: "manager".into(),
|
|
||||||
targets: vec!["alice".into()],
|
|
||||||
body: "wake".into(),
|
|
||||||
first_fire_at_unix: 100,
|
|
||||||
interval_seconds: None,
|
|
||||||
description: None,
|
|
||||||
source: ScheduleSource::Approval { id: 42 },
|
|
||||||
})
|
|
||||||
.expect("submit");
|
|
||||||
let s = db.get(id).expect("get").expect("present");
|
|
||||||
assert_eq!(s.source, ScheduleSource::Approval { id: 42 });
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn reap_cancelled_removes_old_rows_only() {
|
|
||||||
let (_dir, db) = open();
|
|
||||||
let stale = submit_one_shot(&db, 100, &["alice"]);
|
|
||||||
let recent = submit_one_shot(&db, 100, &["bob"]);
|
|
||||||
db.cancel_all(stale).expect("cancel stale");
|
|
||||||
// Manually backdate the stale row.
|
|
||||||
{
|
|
||||||
let conn = db.conn.lock().unwrap();
|
|
||||||
conn.execute(
|
|
||||||
"UPDATE scheduled_prompts SET cancelled_at_unix = ?1 WHERE id = ?2",
|
|
||||||
params![10_i64, stale],
|
|
||||||
)
|
|
||||||
.expect("backdate");
|
|
||||||
}
|
|
||||||
db.cancel_all(recent).expect("cancel recent");
|
|
||||||
let n = db.reap_cancelled(100).expect("reap");
|
|
||||||
assert_eq!(n, 1);
|
|
||||||
assert!(db.get(stale).expect("get stale").is_none());
|
|
||||||
assert!(db.get(recent).expect("get recent").is_some());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,247 +0,0 @@
|
||||||
//! Background loop that drains due `scheduled_prompts` rows
|
|
||||||
//! (#444) and fans the body out as inbox `Message`s to each
|
|
||||||
//! active target. Mirrors `reminder_scheduler::spawn` shape:
|
|
||||||
//! single `spawn(coord)` entry, 5s poll cadence, shutdown-aware.
|
|
||||||
//!
|
|
||||||
//! ## Catch-up semantics
|
|
||||||
//!
|
|
||||||
//! When hive-c0re comes back from being down, a recurring row
|
|
||||||
//! whose `next_fire_at` is well in the past would otherwise fire
|
|
||||||
//! N delayed pulses in a row. Instead we fire ONCE and let
|
|
||||||
//! `ScheduledPrompts::rearm` bump `next_fire_at` to the next
|
|
||||||
//! interval slot ≥ `now`, recording the skipped-cycle count in
|
|
||||||
//! the per-target `last_result` so operators see how many
|
|
||||||
//! firings were caught up rather than losing the signal.
|
|
||||||
//!
|
|
||||||
//! ## Missing-target failure
|
|
||||||
//!
|
|
||||||
//! When a target name doesn't resolve to a known agent (the
|
|
||||||
//! container has been destroyed, the operator typo'd a name,
|
|
||||||
//! etc.) the worker:
|
|
||||||
//! 1. records `last_result = "no such agent: <name>"` against
|
|
||||||
//! the per-target row,
|
|
||||||
//! 2. sends a single advisory `Message` from `system` to
|
|
||||||
//! `operator` describing the schedule + target + reason,
|
|
||||||
//! 3. continues fanning out to the other (live) targets.
|
|
||||||
//!
|
|
||||||
//! Transient broker errors (sqlite lock contention, etc.) get
|
|
||||||
//! the per-target `last_result` annotated AND a `tracing::warn`,
|
|
||||||
//! but the post-fire bookkeeping treats the row the same way it
|
|
||||||
//! does on a clean fire:
|
|
||||||
//! - **recurring** rows re-arm — the next interval slot tries
|
|
||||||
//! the broker send again, so transient errors self-heal.
|
|
||||||
//! - **one-shots** delete unconditionally after their single
|
|
||||||
//! fan-out pass; a broker failure on a one-shot is NOT
|
|
||||||
//! retried (the operator advisory + last_result are the only
|
|
||||||
//! audit trail).
|
|
||||||
|
|
||||||
use std::sync::Arc;
|
|
||||||
use std::time::Duration;
|
|
||||||
|
|
||||||
use hive_sh4re::Message;
|
|
||||||
|
|
||||||
use crate::coordinator::Coordinator;
|
|
||||||
use crate::scheduled_prompts::Schedule;
|
|
||||||
|
|
||||||
/// Per-tick cap. Each schedule fires once per tick at most;
|
|
||||||
/// 100/tick × 5s tick = sustained throughput cap of ~20/sec,
|
|
||||||
/// matching `reminder_scheduler::REMINDER_BATCH_LIMIT`. Bump
|
|
||||||
/// together if real-world rates push past this.
|
|
||||||
const SCHEDULE_BATCH_LIMIT: u64 = 100;
|
|
||||||
|
|
||||||
/// Poll interval. Same 5s as the reminder scheduler — picking
|
|
||||||
/// up freshly-due rows within at most one tick keeps the
|
|
||||||
/// dashboard's "next fire in ..." countdown honest without
|
|
||||||
/// burning CPU on empty sweeps.
|
|
||||||
const POLL_INTERVAL: Duration = Duration::from_secs(5);
|
|
||||||
|
|
||||||
/// Reap cancelled schedules older than this from the table so
|
|
||||||
/// the dashboard list view doesn't accrue tombstones forever.
|
|
||||||
/// Cancelled rows live long enough that the operator can still
|
|
||||||
/// see what they cancelled in the recent past.
|
|
||||||
const CANCELLED_REAP_AGE: Duration = Duration::from_secs(3600);
|
|
||||||
|
|
||||||
pub fn spawn(coord: Arc<Coordinator>) {
|
|
||||||
let mut shutdown = coord.shutdown_rx();
|
|
||||||
tokio::spawn(async move {
|
|
||||||
loop {
|
|
||||||
tick(&coord);
|
|
||||||
tokio::select! {
|
|
||||||
() = tokio::time::sleep(POLL_INTERVAL) => {}
|
|
||||||
_ = shutdown.changed() => {
|
|
||||||
tracing::info!("scheduled_prompts worker: shutdown signal received");
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
fn tick(coord: &Arc<Coordinator>) {
|
|
||||||
let now = now_unix();
|
|
||||||
let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) {
|
|
||||||
Ok(rows) => rows,
|
|
||||||
Err(e) => {
|
|
||||||
tracing::warn!(error = ?e, "scheduled_prompts: query due rows failed");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
if due.is_empty() {
|
|
||||||
// Periodic reaper still gets a chance even on empty ticks.
|
|
||||||
let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
|
|
||||||
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
|
|
||||||
tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
|
|
||||||
}
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
for schedule in due {
|
|
||||||
fire_schedule(coord, &schedule, now);
|
|
||||||
}
|
|
||||||
let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
|
|
||||||
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
|
|
||||||
tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Fan out one schedule's body to every active target. Records
|
|
||||||
/// per-target last_result; advances or reaps the parent row at
|
|
||||||
/// the end depending on whether `interval_seconds` is set.
|
|
||||||
fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
|
|
||||||
let known: std::collections::HashSet<String> = known_agents(coord);
|
|
||||||
for target_row in &schedule.targets {
|
|
||||||
if target_row.cancelled_at_unix.is_some() {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
let target = &target_row.target;
|
|
||||||
// `operator` is a valid recipient (mara c4) — operator
|
|
||||||
// delivery uses the regular broker path; the dashboard
|
|
||||||
// mirrors `to == operator` into its own pane.
|
|
||||||
if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
|
|
||||||
let reason = format!("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);
|
|
||||||
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(()) => "ok".to_owned(),
|
|
||||||
Err(e) => format!("broker send failed: {e:#}"),
|
|
||||||
};
|
|
||||||
if let Err(e) = result {
|
|
||||||
tracing::warn!(
|
|
||||||
schedule = schedule.id,
|
|
||||||
%target,
|
|
||||||
error = ?e,
|
|
||||||
"scheduled_prompts: broker send failed (will retry on next interval)"
|
|
||||||
);
|
|
||||||
}
|
|
||||||
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");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Advance or reap. One-shots delete; recurring re-arm with
|
|
||||||
// catch-up clamp.
|
|
||||||
if schedule.interval_seconds.is_some() {
|
|
||||||
match coord.scheduled_prompts.rearm(schedule.id, now) {
|
|
||||||
Ok(0) => {}
|
|
||||||
Ok(skipped) => {
|
|
||||||
tracing::info!(
|
|
||||||
schedule = schedule.id,
|
|
||||||
skipped,
|
|
||||||
"scheduled_prompts: caught up missed cycles"
|
|
||||||
);
|
|
||||||
}
|
|
||||||
Err(e) => {
|
|
||||||
tracing::warn!(error = ?e, schedule = schedule.id, "rearm failed");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else if let Err(e) = coord.scheduled_prompts.delete(schedule.id) {
|
|
||||||
tracing::warn!(error = ?e, schedule = schedule.id, "delete one-shot failed");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Snapshot of live container names for the missing-target check.
|
|
||||||
/// Always seeds the manager name (which is always reachable);
|
|
||||||
/// adds every live nspawn container that matches the `h-` prefix.
|
|
||||||
/// On `lifecycle::list` failure the set stays at just the manager
|
|
||||||
/// — fail-CLOSED, meaning every non-operator/non-manager target
|
|
||||||
/// looks missing this tick and gets the same treatment as a
|
|
||||||
/// genuinely-destroyed agent: operator advisory + per-target
|
|
||||||
/// `last_result` annotation + skipped delivery. Recurring
|
|
||||||
/// schedules recover automatically on the next tick (the lifecycle
|
|
||||||
/// listing usually works); one-shots that land on this window
|
|
||||||
/// lose their single delivery. Logged at `warn`, not propagated.
|
|
||||||
fn known_agents(_coord: &Coordinator) -> std::collections::HashSet<String> {
|
|
||||||
// `lifecycle::list` is async; the worker tick is sync. Use the
|
|
||||||
// blocking variant via a small `tokio::runtime::Handle::block_on`
|
|
||||||
// wrapper. The worker runs in its own tokio task so this is
|
|
||||||
// safe (we're not in a `current_thread` runtime).
|
|
||||||
use std::collections::HashSet;
|
|
||||||
let mut out: HashSet<String> = HashSet::new();
|
|
||||||
out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
|
|
||||||
let containers = tokio::task::block_in_place(|| {
|
|
||||||
tokio::runtime::Handle::current().block_on(crate::lifecycle::list())
|
|
||||||
});
|
|
||||||
match containers {
|
|
||||||
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, "scheduled_prompts: container listing failed");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
out
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Send the operator a one-line advisory when a schedule fires
|
|
||||||
/// against an agent that no longer exists. Best-effort — failure
|
|
||||||
/// to send just gets logged; the schedule continues firing.
|
|
||||||
fn notify_operator_missing_target(coord: &Coordinator, schedule: &Schedule, target: &str) {
|
|
||||||
let body = format!(
|
|
||||||
"scheduled prompt #{id} fired but target `{target}` is not a live agent. \
|
|
||||||
body was:\n\n{body}",
|
|
||||||
id = schedule.id,
|
|
||||||
target = target,
|
|
||||||
body = schedule.body
|
|
||||||
);
|
|
||||||
let msg = Message {
|
|
||||||
from: "scheduled".to_owned(),
|
|
||||||
to: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
|
||||||
body,
|
|
||||||
in_reply_to: None,
|
|
||||||
};
|
|
||||||
if let Err(e) = coord.broker.send(&msg) {
|
|
||||||
tracing::warn!(error = ?e, schedule = schedule.id, %target, "operator advisory send failed");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn now_unix() -> i64 {
|
|
||||||
std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.ok()
|
|
||||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
||||||
.unwrap_or(0)
|
|
||||||
}
|
|
||||||
|
|
@ -62,32 +62,6 @@ pub fn parent_of(name: &str) -> Option<String> {
|
||||||
read().get(name).cloned().flatten()
|
read().get(name).cloned().flatten()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// True when `candidate` is `ancestor` or any descendant of
|
|
||||||
/// `ancestor` per the current topology. Walks parents from
|
|
||||||
/// `candidate` upward; the walk terminates at root or on a cycle
|
|
||||||
/// (cycle defence: bounded to 32 hops, more than any plausible
|
|
||||||
/// hive depth). Used by the cancel-authorization check in
|
|
||||||
/// `manager_server::handle_cancel_schedule` to enforce
|
|
||||||
/// "managers can cancel anything their subtree owns."
|
|
||||||
#[must_use]
|
|
||||||
pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool {
|
|
||||||
if candidate == ancestor {
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
let topo = read();
|
|
||||||
let mut cur = candidate.to_owned();
|
|
||||||
for _ in 0..32 {
|
|
||||||
let Some(parent) = topo.get(&cur).cloned().flatten() else {
|
|
||||||
return false;
|
|
||||||
};
|
|
||||||
if parent == ancestor {
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
cur = parent;
|
|
||||||
}
|
|
||||||
false
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Persist the topology map. Sorted JSON output (BTreeMap is sorted by
|
/// Persist the topology map. Sorted JSON output (BTreeMap is sorted by
|
||||||
/// key) keeps git diffs minimal across re-writes. Best-effort —
|
/// key) keeps git diffs minimal across re-writes. Best-effort —
|
||||||
/// returns an `io::Error` so callers can decide whether a failure
|
/// returns an `io::Error` so callers can decide whether a failure
|
||||||
|
|
|
||||||
|
|
@ -110,14 +110,6 @@ pub enum ApprovalKind {
|
||||||
/// JSON-encoded inputs array (`"[]"` = all inputs). Agent field is
|
/// JSON-encoded inputs array (`"[]"` = all inputs). Agent field is
|
||||||
/// set to `hm1nd` (the requesting manager).
|
/// set to `hm1nd` (the requesting manager).
|
||||||
UpdateMetaInputs,
|
UpdateMetaInputs,
|
||||||
/// Add a scheduled prompt (closes #444). On approval hive-c0re
|
|
||||||
/// inserts a row into `scheduled_prompts` with
|
|
||||||
/// `source = Approval { id }`; the worker fans the body out as
|
|
||||||
/// inbox messages to each target at the scheduled time, recurring
|
|
||||||
/// when `interval_seconds` is set. `commit_ref` stores the
|
|
||||||
/// JSON-encoded `SchedulePromptPayload` so the approval row carries
|
|
||||||
/// the full submission verbatim (target list, body, schedule).
|
|
||||||
SchedulePrompt,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
|
|
@ -918,93 +910,6 @@ pub enum ManagerRequest {
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
description: Option<String>,
|
description: Option<String>,
|
||||||
},
|
},
|
||||||
/// Queue an approval to add a scheduled prompt (#444). The
|
|
||||||
/// requester (caller of this request) is recorded as the schedule
|
|
||||||
/// owner; on operator approval hive-c0re inserts the schedule and
|
|
||||||
/// the worker fans the body out at fire time. Even agent-self
|
|
||||||
/// schedules go through approval — the existing `remind` MCP tool
|
|
||||||
/// is the unapproved self-wake path.
|
|
||||||
RequestSchedulePrompt(SchedulePromptPayload),
|
|
||||||
/// Cancel a scheduled prompt (#444). `targets = None` cancels the
|
|
||||||
/// whole schedule; `Some(list)` cancels just those recipients,
|
|
||||||
/// auto-cancelling the parent when no active targets remain.
|
|
||||||
/// Authorization: manager can cancel its own schedules + any
|
|
||||||
/// sub-agent schedules (i.e. owner reachable via topology); the
|
|
||||||
/// operator surface bypasses this check.
|
|
||||||
CancelSchedule {
|
|
||||||
id: i64,
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
targets: Option<Vec<String>>,
|
|
||||||
},
|
|
||||||
/// List every schedule in the queue. Manager-side this is
|
|
||||||
/// unfiltered — the dashboard does the topology-filter for the
|
|
||||||
/// per-agent view.
|
|
||||||
ListSchedules,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Submission payload for `RequestSchedulePrompt`. Lives outside the
|
|
||||||
/// enum so it can also serialize into the approval row's `commit_ref`
|
|
||||||
/// (the dispatcher re-parses it on approve and inserts the schedule).
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
|
||||||
pub struct SchedulePromptPayload {
|
|
||||||
/// Names of recipient agents. Operator + `hm1nd` allowed.
|
|
||||||
pub targets: Vec<String>,
|
|
||||||
/// Message body delivered to each target's inbox at fire time.
|
|
||||||
/// Same size budget as `Send.body` — soft cap at the broker level.
|
|
||||||
pub body: String,
|
|
||||||
/// Absolute unix timestamp (seconds) for the FIRST fire. For
|
|
||||||
/// recurring schedules the worker then re-arms in
|
|
||||||
/// `interval_seconds` steps.
|
|
||||||
pub first_fire_at_unix: i64,
|
|
||||||
/// `None` = one-shot. `Some(n > 0)` = recurring every `n` seconds.
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub interval_seconds: Option<u64>,
|
|
||||||
/// Optional description shown on the dashboard approval card AND
|
|
||||||
/// stored on the resulting schedule row for the operator's
|
|
||||||
/// "what is this?" reference later.
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub description: Option<String>,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Schedule row shape on the wire — mirror of
|
|
||||||
/// `scheduled_prompts::Schedule` but in the public crate so dashboard
|
|
||||||
/// + agent surfaces can deserialize without depending on
|
|
||||||
/// hive-c0re-internal types. Kept structurally identical to the
|
|
||||||
/// in-process type; the conversion is field-by-field in
|
|
||||||
/// `manager_server` / `dashboard`.
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct WireSchedule {
|
|
||||||
pub id: i64,
|
|
||||||
pub owner: String,
|
|
||||||
pub body: String,
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub interval_seconds: Option<u64>,
|
|
||||||
pub next_fire_at_unix: i64,
|
|
||||||
pub created_at_unix: i64,
|
|
||||||
pub source: WireScheduleSource,
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub cancelled_at_unix: Option<i64>,
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub description: Option<String>,
|
|
||||||
pub targets: Vec<WireScheduleTarget>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
|
||||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
|
||||||
pub enum WireScheduleSource {
|
|
||||||
Operator,
|
|
||||||
Approval { id: i64 },
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct WireScheduleTarget {
|
|
||||||
pub target: String,
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub cancelled_at_unix: Option<i64>,
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub last_fired_at_unix: Option<i64>,
|
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
||||||
pub last_result: Option<String>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
|
|
@ -1039,12 +944,6 @@ pub enum ManagerResponse {
|
||||||
Logs {
|
Logs {
|
||||||
content: String,
|
content: String,
|
||||||
},
|
},
|
||||||
/// `ListSchedules` result (#444). Snapshot of every schedule
|
|
||||||
/// (active + cancelled-but-not-yet-reaped); the dashboard does the
|
|
||||||
/// per-agent topology filter on top.
|
|
||||||
Schedules {
|
|
||||||
schedules: Vec<WireSchedule>,
|
|
||||||
},
|
|
||||||
/// `GetLooseEnds` result: hive-wide loose ends (approvals +
|
/// `GetLooseEnds` result: hive-wide loose ends (approvals +
|
||||||
/// unanswered questions). Same `LooseEnd` variants as the
|
/// unanswered questions). Same `LooseEnd` variants as the
|
||||||
/// agent surface; the manager's view is unfiltered.
|
/// agent surface; the manager's view is unfiltered.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue