admin/openfang
1
0
Fork 0
mirror of https://github.com/RightNow-AI/openfang.git synced 2026-08-14 08:52:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement "Run Now" for cron jobs

Browse Source 
Add POST /api/cron/jobs/{id}/run endpoint that triggers a cron job
immediately without waiting for its next scheduled fire time. The job
executes asynchronously in the background and its status can be polled
via the existing /status endpoint.

Key changes:
- Extract per-job execution logic from the inline cron tick loop into
  a reusable `cron_run_job()` method on OpenFangKernel, called by both
  the background scheduler and the new API endpoint
- Add `reserve_run()` on CronScheduler to pre-advance next_run for
  overdue jobs before spawning manual runs, preventing duplicate
  execution from the scheduler tick (only advances when next_run <= now
  to avoid skipping imminent scheduled runs)
- Fix dashboard scheduler.js to call the correct cron API endpoint
  instead of the legacy /api/schedules/ path
- Document all cron/scheduler endpoints in api-reference.md

Partially addresses upstream issue #634.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
vnz
2026-03-19 04:24:39 +01:00
co-authored by Claude Opus 4.6
parent 93ea832394
commit 19260945bd
6 changed files with 352 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/openfang-api/src/routes.rs
+63
View File
@@ -10120,6 +10120,69 @@ pub async fn cron_job_status(
}
}
// ---------------------------------------------------------------------------
// Run cron job on demand
// ---------------------------------------------------------------------------
/// POST /api/cron/jobs/{id}/run — Trigger a cron job immediately.
///
/// Returns `{"status": "triggered", "job_id": "..."}` and spawns the execution
/// in the background. The job's status can be polled via
/// `GET /api/cron/jobs/{id}/status`.
pub async fn run_cron_job(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> impl IntoResponse {
let uuid = match uuid::Uuid::parse_str(&id) {
Ok(u) => u,
Err(_) => {
return (
StatusCode::BAD_REQUEST,
Json(serde_json::json!({"status": "error", "error": "Invalid job ID"})),
);
}
};
let job_id = openfang_types::scheduler::CronJobId(uuid);
let job = match state.kernel.cron_scheduler.get_job(job_id) {
Some(j) => j,
None => {
return (
StatusCode::NOT_FOUND,
Json(serde_json::json!({"status": "error", "error": "Job not found"})),
);
}
};
if !job.enabled {
return (
StatusCode::BAD_REQUEST,
Json(serde_json::json!({"status": "error", "error": "Job is disabled"})),
);
}
// Pre-advance next_run so the background scheduler tick won't also fire
// this job while the manual run is in progress.
state.kernel.cron_scheduler.reserve_run(job_id);
// Spawn execution in the background so we don't block the HTTP response.
// `job` is already a clone from `get_job()`, so move it directly.
let kernel = Arc::clone(&state.kernel);
let job_name = job.name.clone();
tokio::spawn(async move {
match kernel.cron_run_job(&job).await {
Ok(_) => tracing::info!(job = %job_name, "On-demand cron job completed"),
Err(e) => tracing::warn!(job = %job_name, error = %e, "On-demand cron job failed"),
}
});
(
StatusCode::OK,
Json(serde_json::json!({
"status": "triggered",
"job_id": id,
})),
)
}
// ---------------------------------------------------------------------------
// Webhook trigger endpoints
// ---------------------------------------------------------------------------
crates/openfang-api/src/server.rs
+4
View File
@@ -586,6 +586,10 @@ pub async fn build_router(
"/api/cron/jobs/{id}/status",
axum::routing::get(routes::cron_job_status),
)
.route(
"/api/cron/jobs/{id}/run",
axum::routing::post(routes::run_cron_job),
)
// Webhook trigger endpoints (external event injection)
.route("/hooks/wake", axum::routing::post(routes::webhook_wake))
.route("/hooks/agent", axum::routing::post(routes::webhook_agent))
crates/openfang-api/static/js/pages/scheduler.js
+5 -5
View File
@@ -201,15 +201,15 @@ function schedulerPage() {
async runNow(job) {
this.runningJobId = job.id;
try {
var result = await OpenFangAPI.post('/api/schedules/' + job.id + '/run', {});
if (result.status === 'completed') {
OpenFangToast.success('Schedule "' + (job.name || 'job') + '" executed successfully');
var result = await OpenFangAPI.post('/api/cron/jobs/' + job.id + '/run', {});
if (result.status === 'triggered' || result.status === 'completed') {
OpenFangToast.success('Job "' + (job.name || 'job') + '" triggered');
job.last_run = new Date().toISOString();
} else {
OpenFangToast.error('Schedule run failed: ' + (result.error || 'Unknown error'));
OpenFangToast.error('Run failed: ' + (result.error || 'Unknown error'));
}
} catch(e) {
OpenFangToast.error('Run Now is not yet available for cron jobs');
OpenFangToast.error('Run failed: ' + (e.message || e));
}
this.runningJobId = '';
},
crates/openfang-kernel/src/cron.rs
+62
View File
@@ -306,6 +306,25 @@ impl CronScheduler {
due
}
/// Pre-advance a job's `next_run` so the background scheduler won't also
/// fire it while a manual (on-demand) run is in progress.
///
/// Only advances `next_run` when the job is already due (`next_run <= now`),
/// matching the same guard used by `due_jobs()`. This avoids silently
/// skipping an imminent scheduled run when the user triggers a manual run
/// on a job that isn't due yet.
///
/// Call this **before** spawning the manual execution task to close the race
/// window between the API handler and the next scheduler tick.
pub fn reserve_run(&self, id: CronJobId) {
if let Some(mut meta) = self.jobs.get_mut(&id) {
let now = Utc::now();
if meta.job.next_run.map(|t| t <= now).unwrap_or(false) {
meta.job.next_run = Some(compute_next_run_after(&meta.job.schedule, now));
}
}
}
// -- Outcome recording --------------------------------------------------
/// Record a successful execution for a job.
@@ -1207,4 +1226,47 @@ mod tests {
assert_eq!(sched.list_jobs(other).len(), 1);
}
}
// -- reserve_run ---------------------------------------------------------
#[test]
fn reserve_run_skips_not_yet_due_job() {
let (sched, _tmp) = make_scheduler(100);
let agent = AgentId::new();
let mut job = make_job(agent);
// Schedule is Every 3600s; next_run will be ~1 hour from now after add.
job.schedule = CronSchedule::Every { every_secs: 3600 };
let id = sched.add_job(job, false).unwrap();
let original_next_run = sched.get_job(id).unwrap().next_run;
assert!(original_next_run.is_some());
// Manual trigger on a not-yet-due job should NOT move next_run.
sched.reserve_run(id);
let after = sched.get_job(id).unwrap().next_run;
assert_eq!(original_next_run, after, "reserve_run must not move next_run for a job that is not yet due");
}
#[test]
fn reserve_run_advances_overdue_job() {
let (sched, _tmp) = make_scheduler(100);
let agent = AgentId::new();
let mut job = make_job(agent);
job.schedule = CronSchedule::Every { every_secs: 3600 };
let id = sched.add_job(job, false).unwrap();
// Force next_run into the past to simulate an overdue job.
if let Some(mut meta) = sched.jobs.get_mut(&id) {
meta.job.next_run = Some(Utc::now() - Duration::seconds(10));
}
let before = sched.get_job(id).unwrap().next_run.unwrap();
assert!(before < Utc::now(), "precondition: job should be overdue");
sched.reserve_run(id);
let after = sched.get_job(id).unwrap().next_run.unwrap();
assert!(after > Utc::now(), "reserve_run should advance next_run past now for overdue jobs");
}
}
crates/openfang-kernel/src/kernel.rs
+128 -141
View File
@@ -4016,149 +4016,14 @@ impl OpenFangKernel {
let due = kernel.cron_scheduler.due_jobs();
for job in due {
let job_id = job.id;
let agent_id = job.agent_id;
let job_name = job.name.clone();
match &job.action {
openfang_types::scheduler::CronAction::SystemEvent { text } => {
tracing::debug!(job = %job_name, "Cron: firing system event");
let payload_bytes = serde_json::to_vec(&serde_json::json!({
"type": format!("cron.{}", job_name),
"text": text,
"job_id": job_id.to_string(),
}))
.unwrap_or_default();
let event = Event::new(
AgentId::new(), // system-originated
EventTarget::Broadcast,
EventPayload::Custom(payload_bytes),
);
kernel.publish_event(event).await;
kernel.cron_scheduler.record_success(job_id);
tracing::debug!(job = %job_name, "Cron: firing scheduled job");
match kernel.cron_run_job(&job).await {
Ok(_) => {
tracing::info!(job = %job_name, "Cron job completed successfully");
}
openfang_types::scheduler::CronAction::AgentTurn {
message,
timeout_secs,
..
} => {
tracing::debug!(job = %job_name, agent = %agent_id, "Cron: firing agent turn");
let timeout_s = timeout_secs.unwrap_or(120);
let timeout = std::time::Duration::from_secs(timeout_s);
let delivery = job.delivery.clone();
let kh: std::sync::Arc<
dyn openfang_runtime::kernel_handle::KernelHandle,
> = kernel.clone();
match tokio::time::timeout(
timeout,
kernel.send_message_with_handle(
agent_id,
message,
Some(kh),
None,
None,
),
)
.await
{
Ok(Ok(result)) => {
match cron_deliver_response(
&kernel,
agent_id,
&result.response,
&delivery,
)
.await
{
Ok(()) => {
tracing::info!(job = %job_name, "Cron job completed successfully");
kernel.cron_scheduler.record_success(job_id);
}
Err(e) => {
tracing::warn!(job = %job_name, error = %e, "Cron job delivery failed");
kernel.cron_scheduler.record_failure(job_id, &e);
}
}
}
Ok(Err(e)) => {
let err_msg = format!("{e}");
tracing::warn!(job = %job_name, error = %err_msg, "Cron job failed");
kernel.cron_scheduler.record_failure(job_id, &err_msg);
}
Err(_) => {
tracing::warn!(job = %job_name, timeout_s, "Cron job timed out");
kernel.cron_scheduler.record_failure(
job_id,
&format!("timed out after {timeout_s}s"),
);
}
}
}
openfang_types::scheduler::CronAction::WorkflowRun {
workflow_id,
input,
timeout_secs,
} => {
tracing::debug!(job = %job_name, workflow = %workflow_id, "Cron: firing workflow run");
let wf_input = input.clone().unwrap_or_default();
let timeout_s = timeout_secs.unwrap_or(120);
let timeout = std::time::Duration::from_secs(timeout_s);
let delivery = job.delivery.clone();
// Resolve workflow: try UUID first, then name
let wf_id = match uuid::Uuid::parse_str(workflow_id) {
Ok(uuid) => crate::workflow::WorkflowId(uuid),
Err(_) => {
let all_wfs = kernel.workflows.list_workflows().await;
if let Some(wf) =
all_wfs.iter().find(|w| w.name == *workflow_id)
{
wf.id
} else {
let err_msg =
format!("workflow not found: {workflow_id}");
tracing::warn!(job = %job_name, %err_msg);
kernel.cron_scheduler.record_failure(job_id, &err_msg);
continue;
}
}
};
match tokio::time::timeout(
timeout,
kernel.run_workflow(wf_id, wf_input),
)
.await
{
Ok(Ok((_run_id, output))) => {
match cron_deliver_response(
&kernel, agent_id, &output, &delivery,
)
.await
{
Ok(()) => {
tracing::info!(job = %job_name, "Cron workflow completed");
kernel.cron_scheduler.record_success(job_id);
}
Err(e) => {
tracing::warn!(job = %job_name, error = %e, "Cron workflow delivery failed");
kernel.cron_scheduler.record_failure(job_id, &e);
}
}
}
Ok(Err(e)) => {
let err_msg = format!("{e}");
tracing::warn!(job = %job_name, error = %err_msg, "Cron workflow failed");
kernel.cron_scheduler.record_failure(job_id, &err_msg);
}
Err(_) => {
tracing::warn!(job = %job_name, timeout_s, "Cron workflow timed out");
kernel.cron_scheduler.record_failure(
job_id,
&format!("workflow timed out after {timeout_s}s"),
);
}
}
Err(e) => {
tracing::warn!(job = %job_name, error = %e, "Cron job failed");
}
}
}
@@ -5382,6 +5247,128 @@ impl OpenFangKernel {
}
context_parts.join("\n\n")
}
/// Execute a cron job on demand and deliver its result.
///
/// This is the same logic used by the background cron tick loop, extracted
/// so the API can trigger a job immediately via `POST /api/cron/jobs/{id}/run`.
/// Records success/failure on the job's metadata just like the scheduler does.
pub async fn cron_run_job(self: &Arc<Self>, job: &openfang_types::scheduler::CronJob) -> Result<String, String> {
use openfang_types::scheduler::CronAction;
let job_id = job.id;
let agent_id = job.agent_id;
let job_name = &job.name;
match &job.action {
CronAction::SystemEvent { text } => {
let payload_bytes = serde_json::to_vec(&serde_json::json!({
"type": format!("cron.{}", job_name),
"text": text,
"job_id": job_id.to_string(),
}))
.unwrap_or_default();
let event = Event::new(
AgentId::new(),
EventTarget::Broadcast,
EventPayload::Custom(payload_bytes),
);
self.publish_event(event).await;
self.cron_scheduler.record_success(job_id);
Ok("system event published".to_string())
}
CronAction::AgentTurn {
message,
timeout_secs,
..
} => {
let timeout_s = timeout_secs.unwrap_or(120);
let timeout = std::time::Duration::from_secs(timeout_s);
let delivery = job.delivery.clone();
let kh: Arc<dyn KernelHandle> = self.clone();
match tokio::time::timeout(
timeout,
self.send_message_with_handle(agent_id, message, Some(kh), None, None),
)
.await
{
Ok(Ok(result)) => {
match cron_deliver_response(self, agent_id, &result.response, &delivery)
.await
{
Ok(()) => {
self.cron_scheduler.record_success(job_id);
Ok(result.response)
}
Err(e) => {
self.cron_scheduler.record_failure(job_id, &e);
Err(e)
}
}
}
Ok(Err(e)) => {
let err_msg = format!("{e}");
self.cron_scheduler.record_failure(job_id, &err_msg);
Err(err_msg)
}
Err(_) => {
let err_msg = format!("timed out after {timeout_s}s");
self.cron_scheduler.record_failure(job_id, &err_msg);
Err(err_msg)
}
}
}
CronAction::WorkflowRun {
workflow_id,
input,
timeout_secs,
} => {
let wf_input = input.clone().unwrap_or_default();
let timeout_s = timeout_secs.unwrap_or(120);
let timeout = std::time::Duration::from_secs(timeout_s);
let delivery = job.delivery.clone();
let wf_id = match uuid::Uuid::parse_str(workflow_id) {
Ok(uuid) => crate::workflow::WorkflowId(uuid),
Err(_) => {
let all_wfs = self.workflows.list_workflows().await;
if let Some(wf) = all_wfs.iter().find(|w| w.name == *workflow_id) {
wf.id
} else {
let err_msg = format!("workflow not found: {workflow_id}");
self.cron_scheduler.record_failure(job_id, &err_msg);
return Err(err_msg);
}
}
};
match tokio::time::timeout(timeout, self.run_workflow(wf_id, wf_input)).await {
Ok(Ok((_run_id, output))) => {
match cron_deliver_response(self, agent_id, &output, &delivery).await {
Ok(()) => {
self.cron_scheduler.record_success(job_id);
Ok(output)
}
Err(e) => {
self.cron_scheduler.record_failure(job_id, &e);
Err(e)
}
}
}
Ok(Err(e)) => {
let err_msg = format!("{e}");
self.cron_scheduler.record_failure(job_id, &err_msg);
Err(err_msg)
}
Err(_) => {
let err_msg = format!("workflow timed out after {timeout_s}s");
self.cron_scheduler.record_failure(job_id, &err_msg);
Err(err_msg)
}
}
}
}
}
}
/// Convert a manifest's capability declarations into Capability enums.
docs/api-reference.md
+90
View File
@@ -23,6 +23,7 @@ All responses include security headers (CSP, X-Frame-Options, X-Content-Type-Opt
- [Usage & Analytics Endpoints](#usage--analytics-endpoints)
- [Migration Endpoints](#migration-endpoints)
- [Session Management Endpoints](#session-management-endpoints)
- [Cron/Scheduler Endpoints](#cronscheduler-endpoints)
- [WebSocket Protocol](#websocket-protocol)
- [SSE Streaming](#sse-streaming)
- [OpenAI-Compatible API](#openai-compatible-api)
@@ -1770,6 +1771,95 @@ Switch an agent's LLM model at runtime.
---
## Cron/Scheduler Endpoints
Manage recurring and one-shot scheduled jobs. Jobs can trigger agent turns, system events, or workflow runs on a schedule.
### GET /api/cron/jobs
List all cron jobs.
**Response** `200 OK`: Array of `CronJob` objects.
### POST /api/cron/jobs
Create a new cron job.
**Request Body**:
```json
{
"agent_id": "uuid",
"name": "daily-report",
"enabled": true,
"schedule": { "kind": "every", "every_secs": 3600 },
"action": {
"kind": "agent_turn",
"message": "Generate the daily report",
"timeout_secs": 120
},
"delivery": {
"kind": "channel",
"channel": "slack",
"to": "#reports"
}
}
```
**Response** `200 OK`: The created `CronJob` object with assigned `id`.
### DELETE /api/cron/jobs/{id}
Delete a cron job by ID.
**Response** `200 OK`:
```json
{ "status": "deleted" }
```
### PUT /api/cron/jobs/{id}/enable
Enable or disable a cron job.
**Request Body**:
```json
{ "enabled": false }
```
**Response** `200 OK`:
```json
{ "status": "updated", "enabled": false }
```
### GET /api/cron/jobs/{id}/status
Get job metadata including last run time, status, and error history.
**Response** `200 OK`: `JobMeta` object with fields like `last_run`, `last_status`, `consecutive_errors`.
### POST /api/cron/jobs/{id}/run
Trigger a cron job immediately. The job executes asynchronously in the background — this endpoint returns immediately without waiting for completion. Poll `GET /api/cron/jobs/{id}/status` to check the result.
**Response** `200 OK`:
```json
{
"status": "triggered",
"job_id": "550e8400-e29b-41d4-a716-446655440000"
}
```
**Error Responses**:
- `400 Bad Request` — Invalid job ID or job is disabled
- `404 Not Found` — Job not found
---
## WebSocket Protocol
### Connecting