Skip to content

Commit 80c9630

Browse files
alari76claude
andauthored
feat(workflows): trigger engine core — pre-dispatch gates, trigger ledger, heartbeat (#602)
Phase A0+A of the Joe value-expansion plan: scheduled workflow dispatch now passes through gates before creating any run or session, every decision is ledgered, and the dispatch loop is observable. - Single cron module (server/cron.ts) — parser/matcher/validator extracted from workflow-engine and workflow-routes; both re-export for compatibility - SHA-based change detection: HEAD sha recorded on successful runs only (lastReviewedSha); an unchanged repo holds the dispatch instead of running. Fixes the sliding sinceTimestamp window that advanced on skips/failures - Single-flight per repo+kind with a short retry, replacing unbounded stacking - Catch-up policy per schedule (collapse | skip) for fires missed in downtime - Per-tick dispatch cap staggers backlogs instead of thundering-herding - Trigger ledger table + GET /api/workflows/trigger-ledger; last hold surfaced on schedules and in the WorkflowRow UI - Engine heartbeat on every tick + GET /api/workflows/engine-health; the orchestrator monitor alerts Joe once per stall, re-arming on recovery - Dispatch tick wrapped so one bad schedule can never kill the interval Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 7089e8d commit 80c9630

9 files changed

Lines changed: 737 additions & 137 deletions

File tree

docs/WORKFLOWS.md

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,20 @@ Workflow runs are triggered by cron schedules configured per-repo via the workfl
1717

1818
---
1919

20+
## Trigger Dispatch
21+
22+
Scheduled runs pass through pre-dispatch gates before any run row or session is created. Each decision — fired or held, with its reason — is recorded in a trigger ledger (`GET /api/workflows/trigger-ledger`, optionally filtered by `scheduleId`), and the most recent hold is surfaced on the schedule (`lastHeldAt` / `lastHeldReason` / `heldCount`).
23+
24+
The gates, in order:
25+
26+
1. **Catch-up policy** — each schedule has `catchUp: 'collapse' | 'skip'` (default `collapse`, settable via `PATCH /api/workflows/schedules/:id`). `collapse` fires once no matter how many slots were missed during downtime; `skip` abandons a fire time missed by more than 10 minutes and waits for the next natural slot.
27+
2. **Single-flight** — a schedule whose previous run (same kind + repo) is still active is held and retried a few minutes later, never stacked.
28+
3. **Change detection** — the engine records the repo's HEAD sha when a run *succeeds* (`lastReviewedSha`). A scheduled fire where HEAD hasn't moved since is held: nothing new to review. Failed and skipped runs do not advance the anchor, so their commits are re-examined on the next fire. Manual triggers (`POST /api/workflows/schedules/:id/trigger`) bypass all gates but are still ledgered.
29+
30+
Dispatches are capped per 60-second tick (backlog staggers across ticks instead of stampeding after downtime), and the loop writes a heartbeat on every tick — `GET /api/workflows/engine-health` returns `{ lastTickAt, tickCount, stale }`, and the orchestrator monitor raises an alert notification if the heartbeat goes stale.
31+
32+
---
33+
2034
## MD File Format
2135

2236
Workflow definitions are Markdown files with YAML frontmatter. The frontmatter contains configuration metadata; the body is the prompt sent verbatim to Claude.

server/cron.test.ts

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/** Tests for the shared cron module — nextCronMatch (matching/validation are covered via the engine and routes re-exports). */
2+
import { describe, it, expect } from 'vitest'
3+
import { nextCronMatch, cronMatchesDate, isValidCron } from './cron.js'
4+
5+
describe('nextCronMatch', () => {
6+
it('finds the next matching minute', () => {
7+
const after = new Date('2026-08-29T10:30:45Z')
8+
const next = nextCronMatch('* * * * *', after)
9+
expect(next.getTime()).toBe(new Date('2026-08-29T10:31:00Z').getTime())
10+
})
11+
12+
it('rolls forward to the next daily slot when the time has passed today', () => {
13+
const after = new Date(2026, 7, 29, 10, 0, 0) // local 10:00
14+
const next = nextCronMatch('0 9 * * *', after) // daily at 09:00 local
15+
expect(next.getHours()).toBe(9)
16+
expect(next.getMinutes()).toBe(0)
17+
expect(next.getDate()).toBe(30)
18+
})
19+
20+
it('falls back to +24h for an expression that never matches', () => {
21+
const after = new Date('2026-08-29T10:30:00Z')
22+
const next = nextCronMatch('*/0 * * * *', after) // invalid step — matches nothing
23+
expect(next.getTime()).toBe(after.getTime() + 86400000)
24+
})
25+
})
26+
27+
describe('re-exported helpers', () => {
28+
it('cronMatchesDate and isValidCron agree on a valid expression', () => {
29+
expect(isValidCron('0 9 * * 1')).toBe(true)
30+
const monday9am = new Date(2026, 8, 7, 9, 0) // Mon Sep 7 2026, 09:00 local
31+
expect(cronMatchesDate('0 9 * * 1', monday9am)).toBe(true)
32+
})
33+
})

server/cron.ts

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
/**
2+
* Single cron implementation for the trigger engine.
3+
*
4+
* Standard 5-field cron (minute hour dom month dow). This module is the one
5+
* place cron expressions are parsed, matched, and validated — the engine's
6+
* scheduler, the route-level validation, and anything Joe schedules must all
7+
* agree on edge cases, so they all import from here.
8+
*/
9+
10+
function parseCronField(field: string, min: number, max: number): number[] {
11+
const values: number[] = []
12+
for (const part of field.split(',')) {
13+
const stepMatch = part.match(/^(.+)\/(\d+)$/)
14+
const step = stepMatch ? parseInt(stepMatch[2], 10) : 1
15+
// Defensive guard — `step <= 0` would make the loops below never advance
16+
// (or run backwards), pinning the scheduler. Any caller that reaches this
17+
// branch with a zero/negative step has bypassed isValidCron, so refuse loudly.
18+
if (!Number.isFinite(step) || step <= 0) {
19+
throw new Error(`Invalid cron step value: ${stepMatch?.[2]} (must be > 0)`)
20+
}
21+
const range = stepMatch ? stepMatch[1] : part
22+
23+
if (range === '*') {
24+
for (let i = min; i <= max; i += step) values.push(i)
25+
} else if (range.includes('-')) {
26+
const [start, end] = range.split('-').map(Number)
27+
for (let i = start; i <= end; i += step) values.push(i)
28+
} else {
29+
values.push(parseInt(range, 10))
30+
}
31+
}
32+
return values
33+
}
34+
35+
export function cronMatchesDate(expression: string, date: Date): boolean {
36+
const parts = expression.trim().split(/\s+/)
37+
if (parts.length !== 5) return false
38+
39+
try {
40+
const [minF, hourF, domF, monF, dowF] = parts
41+
const minute = parseCronField(minF, 0, 59)
42+
const hour = parseCronField(hourF, 0, 23)
43+
const dom = parseCronField(domF, 1, 31)
44+
const month = parseCronField(monF, 1, 12)
45+
const dow = parseCronField(dowF, 0, 6)
46+
47+
return (
48+
minute.includes(date.getMinutes()) &&
49+
hour.includes(date.getHours()) &&
50+
dom.includes(date.getDate()) &&
51+
month.includes(date.getMonth() + 1) &&
52+
dow.includes(date.getDay())
53+
)
54+
} catch {
55+
// Malformed expression (e.g. step 0). Treat as never-matching so a bad
56+
// legacy schedule cannot pin the scheduler in a tight loop.
57+
return false
58+
}
59+
}
60+
61+
/** Compute the next matching minute for a cron expression after `after`. */
62+
export function nextCronMatch(expression: string, after: Date): Date {
63+
const d = new Date(after)
64+
d.setSeconds(0, 0)
65+
d.setMinutes(d.getMinutes() + 1)
66+
// Search up to 366 days ahead
67+
for (let i = 0; i < 366 * 24 * 60; i++) {
68+
if (cronMatchesDate(expression, d)) return d
69+
d.setMinutes(d.getMinutes() + 1)
70+
}
71+
// Fallback: 24h from now
72+
return new Date(after.getTime() + 86400000)
73+
}
74+
75+
/** Validate a 5-field cron expression. Returns true if the format is valid. */
76+
export function isValidCron(expr: string): boolean {
77+
const parts = expr.trim().split(/\s+/)
78+
if (parts.length !== 5) return false
79+
const ranges = [
80+
[0, 59], // minute
81+
[0, 23], // hour
82+
[1, 31], // day of month
83+
[1, 12], // month
84+
[0, 6], // day of week
85+
]
86+
return parts.every((part, i) => {
87+
const [min, max] = ranges[i]
88+
return part.split(',').every(segment => {
89+
const stepMatch = segment.match(/^(.+)\/(\d+)$/)
90+
if (stepMatch) {
91+
const step = parseInt(stepMatch[2], 10)
92+
if (isNaN(step) || step < 1) return false
93+
}
94+
const range = stepMatch ? stepMatch[1] : segment
95+
if (range === '*') return true
96+
if (range.includes('-')) {
97+
const [a, b] = range.split('-').map(Number)
98+
return !isNaN(a) && !isNaN(b) && a >= min && b <= max && a <= b
99+
}
100+
const n = parseInt(range, 10)
101+
return !isNaN(n) && n >= min && n <= max
102+
})
103+
})
104+
}

server/orchestrator-monitor.ts

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,13 +50,17 @@ export class OrchestratorMonitor {
5050
/** (runId:status) pairs already notified — see handleGoalRunEvent. */
5151
private notedGoalRunStates = new Set<string>()
5252
private memory: OrchestratorMemory | null = null
53+
private engine: WorkflowEngine | null = null
54+
/** True once a stall alert has been sent — reset when the heartbeat recovers. */
55+
private engineStallNotified = false
5356

5457
constructor(sessions: SessionManager) {
5558
this.sessions = sessions
5659
}
5760

5861
/** Connect to the workflow engine for event-driven notifications. */
5962
setEngine(engine: WorkflowEngine): void {
63+
this.engine = engine
6064
engine.on('workflow_event', (event: WorkflowEvent) => {
6165
this.handleWorkflowEvent(event)
6266
})
@@ -172,7 +176,37 @@ export class OrchestratorMonitor {
172176
// Check for passive repos
173177
this.checkPassiveRepos(repoPaths)
174178

175-
// initial scan complete
179+
// Check the trigger engine's heartbeat
180+
this.checkEngineHeartbeat()
181+
}
182+
183+
/**
184+
* Watchdog for the workflow dispatch loop. The engine writes a heartbeat row on
185+
* every 60s tick; if it hasn't for 5+ minutes the interval has died or the
186+
* process is wedged — something the engine cannot report about itself.
187+
* Alerts once per stall, re-arming when the heartbeat recovers.
188+
*/
189+
private checkEngineHeartbeat(): void {
190+
if (!this.engine) return
191+
try {
192+
const health = this.engine.getEngineHealth()
193+
// A null heartbeat means the scheduler was never started (e.g. disabled) — not a stall.
194+
if (!health.lastTickAt) return
195+
196+
const stale = Date.now() - new Date(health.lastTickAt).getTime() > 5 * 60_000
197+
if (stale && !this.engineStallNotified) {
198+
this.engineStallNotified = true
199+
this.addNotification({
200+
severity: 'alert',
201+
title: 'Workflow trigger engine stalled',
202+
body: `The workflow dispatch loop last ticked at ${health.lastTickAt}. Scheduled workflows are not firing — the server likely needs attention.`,
203+
})
204+
} else if (!stale) {
205+
this.engineStallNotified = false
206+
}
207+
} catch (err) {
208+
console.error('[orchestrator-monitor] Engine heartbeat check error:', err)
209+
}
176210
}
177211

178212
/** Check for reports we haven't seen before. */

0 commit comments

Comments
 (0)