1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
|
/**
* Heartbeat scheduler — the imperative timer loop.
*
* The scheduler owns NO knowledge of conversations, prompts, or the
* orchestrator. It only manages per-workspace timers: when armed (a workspace's
* heartbeat is `enabled`), it schedules a fire after `intervalMinutes`; when the
* fire's work completes, it re-arms (resets the timer). This is the "reset timer
* after each run" semantics — the interval is measured from run-completion to
* the next fire, not a fixed wall-clock schedule.
*
* The actual run work (create a conversation, send the task prompt, track the
* run) is the `fire(workspaceId)` callback, injected by the heartbeat service.
* This keeps the scheduler pure-ish (only I/O is the injected timers) and
* testable with fake timers + a fake fire.
*/
/** A handle returned by `setTimeout`, opaque to the scheduler. */
export type TimerHandle = ReturnType<typeof setTimeout>;
/** Injectable timers — the only I/O effect the scheduler touches. */
export interface Timers {
readonly now: () => number;
readonly setTimeout: (fn: () => void, ms: number) => TimerHandle;
readonly clearTimeout: (handle: TimerHandle | undefined) => void;
}
/** The real timers (used at runtime; overridable in tests). */
export const realTimers: Timers = {
now: () => Date.now(),
setTimeout: (fn, ms) => setTimeout(fn, ms),
clearTimeout: (handle) => {
if (handle !== undefined) clearTimeout(handle);
},
};
export interface SchedulerDeps {
readonly timers: Timers;
/** Do the run work for a workspace. Resolves when the turn seals. */
readonly fire: (workspaceId: string) => Promise<void>;
}
interface WorkspaceSchedule {
/** Current interval (minutes), updated by `arm` on config changes. */
intervalMinutes: number;
/** Pending wait timer (null while a fire is in progress). */
timer: TimerHandle | undefined;
/** A fire is in progress for this workspace. */
running: boolean;
/** Scheduling is active (the workspace's heartbeat is enabled). */
armed: boolean;
/**
* Absolute epoch-ms timestamp of the next scheduled fire, or null when no
* fire is pending — the schedule is disarmed, or a run is in progress
* (the next fire is scheduled only after the run completes). This is the
* server-authoritative next-run time the `/heartbeat/next-run` endpoint
* derives its response from.
*/
nextFireMs: number | null;
}
const MINUTE_MS = 60_000;
/**
* Manages per-workspace heartbeat timers. A single instance is owned by the
* heartbeat service for its lifetime.
*/
export class HeartbeatScheduler {
private readonly schedules = new Map<string, WorkspaceSchedule>();
private readonly timers: Timers;
private readonly fire: (workspaceId: string) => Promise<void>;
constructor(deps: SchedulerDeps) {
this.timers = deps.timers;
this.fire = deps.fire;
}
/**
* Arm (or re-arm) a workspace's schedule from its config. When `enabled`
* is false the schedule is disarmed. When a config update arrives while a
* run is in progress, the new `intervalMinutes` takes effect on the next
* re-arm (the in-progress run is never cancelled here).
*/
arm(
workspaceId: string,
config: {
readonly enabled: boolean;
readonly intervalMinutes: number;
},
): void {
if (!config.enabled) {
this.disarm(workspaceId);
return;
}
let schedule = this.schedules.get(workspaceId);
if (schedule === undefined) {
schedule = {
intervalMinutes: config.intervalMinutes,
timer: undefined,
running: false,
armed: true,
nextFireMs: null,
};
this.schedules.set(workspaceId, schedule);
} else {
schedule.intervalMinutes = config.intervalMinutes;
schedule.armed = true;
}
// If a fire is in progress, let it finish — it will re-arm with the
// (possibly new) interval. Otherwise schedule the next fire now.
if (!schedule.running) {
this.clearTimer(schedule);
this.scheduleNext(workspaceId, schedule);
}
}
/** Stop scheduling for a workspace (clears a pending timer; an in-progress run finishes on its own). */
disarm(workspaceId: string): void {
const schedule = this.schedules.get(workspaceId);
if (schedule === undefined) return;
schedule.armed = false;
this.clearTimer(schedule);
this.schedules.delete(workspaceId);
}
/** Stop all schedules (deactivate). */
disarmAll(): void {
for (const workspaceId of [...this.schedules.keys()]) {
this.disarm(workspaceId);
}
}
/** Whether a schedule is currently armed (enabled) for a workspace. */
isArmed(workspaceId: string): boolean {
return this.schedules.get(workspaceId)?.armed ?? false;
}
/** Whether a fire is currently in progress for a workspace. */
isRunning(workspaceId: string): boolean {
return this.schedules.get(workspaceId)?.running ?? false;
}
/**
* The absolute epoch-ms timestamp of the next scheduled fire for a workspace,
* or null when no fire is pending — the heartbeat is disabled/disarmed, or a
* run is in progress (the next fire is scheduled only after the run
* completes). This is the server-authoritative next-run time.
*/
nextFireAt(workspaceId: string): number | null {
return this.schedules.get(workspaceId)?.nextFireMs ?? null;
}
private scheduleNext(workspaceId: string, schedule: WorkspaceSchedule): void {
const ms = Math.max(MINUTE_MS, schedule.intervalMinutes * MINUTE_MS);
// Record the absolute fire time (now + delay) BEFORE arming the timer
// so `nextFireAt` reports it while the timer is pending.
schedule.nextFireMs = this.timers.now() + ms;
schedule.timer = this.timers.setTimeout(() => {
this.onTick(workspaceId);
}, ms);
}
private onTick(workspaceId: string): void {
const schedule = this.schedules.get(workspaceId);
// Race: the schedule was disarmed after the timer was queued.
if (schedule === undefined || !schedule.armed) return;
schedule.timer = undefined;
// A fire is now in progress — no next run is queued yet (it's scheduled
// only after the run completes), so report no pending fire time.
schedule.nextFireMs = null;
schedule.running = true;
this.fire(workspaceId)
.catch(() => {
// The service records the run outcome; a thrown fire is logged
// by the service. Swallow so the loop continues.
})
.finally(() => {
const current = this.schedules.get(workspaceId);
if (current === undefined) return;
current.running = false;
if (current.armed) {
this.scheduleNext(workspaceId, current);
}
});
}
private clearTimer(schedule: WorkspaceSchedule): void {
if (schedule.timer !== undefined) {
this.timers.clearTimeout(schedule.timer);
schedule.timer = undefined;
}
schedule.nextFireMs = null;
}
}
|
