diff options
Diffstat (limited to 'packages/heartbeat')
| -rw-r--r-- | packages/heartbeat/package.json | 15 | ||||
| -rw-r--r-- | packages/heartbeat/src/config-store.test.ts | 152 | ||||
| -rw-r--r-- | packages/heartbeat/src/config-store.ts | 106 | ||||
| -rw-r--r-- | packages/heartbeat/src/extension.ts | 121 | ||||
| -rw-r--r-- | packages/heartbeat/src/heartbeat.test.ts | 781 | ||||
| -rw-r--r-- | packages/heartbeat/src/heartbeat.ts | 354 | ||||
| -rw-r--r-- | packages/heartbeat/src/index.ts | 17 | ||||
| -rw-r--r-- | packages/heartbeat/src/run-store.test.ts | 72 | ||||
| -rw-r--r-- | packages/heartbeat/src/run-store.ts | 95 | ||||
| -rw-r--r-- | packages/heartbeat/src/scheduler.test.ts | 343 | ||||
| -rw-r--r-- | packages/heartbeat/src/scheduler.ts | 193 | ||||
| -rw-r--r-- | packages/heartbeat/tsconfig.json | 12 |
12 files changed, 2261 insertions, 0 deletions
diff --git a/packages/heartbeat/package.json b/packages/heartbeat/package.json new file mode 100644 index 0000000..d83e631 --- /dev/null +++ b/packages/heartbeat/package.json @@ -0,0 +1,15 @@ +{ + "name": "@dispatch/heartbeat", + "version": "0.0.0", + "type": "module", + "private": true, + "main": "dist/index.js", + "types": "dist/index.d.ts", + "dependencies": { + "@dispatch/conversation-store": "workspace:*", + "@dispatch/kernel": "workspace:*", + "@dispatch/session-orchestrator": "workspace:*", + "@dispatch/system-prompt": "workspace:*", + "@dispatch/transport-contract": "workspace:*" + } +} diff --git a/packages/heartbeat/src/config-store.test.ts b/packages/heartbeat/src/config-store.test.ts new file mode 100644 index 0000000..9a6772d --- /dev/null +++ b/packages/heartbeat/src/config-store.test.ts @@ -0,0 +1,152 @@ +import type { StorageNamespace } from "@dispatch/kernel"; +import { describe, expect, it } from "vitest"; +import { + applyConfigUpdate, + createHeartbeatConfigStore, + DEFAULT_HEARTBEAT_CONFIG, +} from "./config-store.js"; + +function createMemoryStorage(): StorageNamespace { + const data = new Map<string, string>(); + return { + get: async (key) => data.get(key) ?? null, + set: async (key, value) => { + data.set(key, value); + }, + delete: async (key) => { + data.delete(key); + }, + has: async (key) => data.has(key), + keys: async (prefix) => { + const all = [...data.keys()]; + if (prefix === undefined) return all; + return all.filter((k) => k.startsWith(prefix)); + }, + }; +} + +describe("applyConfigUpdate (pure)", () => { + it("leaves omitted fields unchanged from the default", () => { + const next = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, {}); + expect(next).toEqual(DEFAULT_HEARTBEAT_CONFIG); + }); + + it("applies provided fields", () => { + const next = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { + enabled: true, + systemPrompt: "you are a monitor", + taskPrompt: "check for stuck chats", + model: "opencode/gpt-4o", + }); + expect(next.enabled).toBe(true); + expect(next.systemPrompt).toBe("you are a monitor"); + expect(next.taskPrompt).toBe("check for stuck chats"); + expect(next.model).toBe("opencode/gpt-4o"); + // Untouched fields keep defaults. + expect(next.intervalMinutes).toBe(30); + expect(next.reasoningEffort).toBeNull(); + }); + + it("clamps intervalMinutes to a minimum of 1", () => { + expect( + applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { intervalMinutes: 0 }).intervalMinutes, + ).toBe(1); + expect( + applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { intervalMinutes: -5 }).intervalMinutes, + ).toBe(1); + expect( + applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { intervalMinutes: 7 }).intervalMinutes, + ).toBe(7); + }); + + it("truncates a non-integer interval to an integer", () => { + expect( + applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { intervalMinutes: 12.9 }).intervalMinutes, + ).toBe(12); + }); + + it("clears reasoningEffort when null is passed", () => { + const withEffort = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { reasoningEffort: "high" }); + expect(withEffort.reasoningEffort).toBe("high"); + const cleared = applyConfigUpdate(withEffort, { reasoningEffort: null }); + expect(cleared.reasoningEffort).toBeNull(); + }); + + it("treats an absent reasoningEffort as unchanged (distinct from null)", () => { + const withEffort = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { reasoningEffort: "low" }); + const untouched = applyConfigUpdate(withEffort, { enabled: true }); + expect(untouched.reasoningEffort).toBe("low"); + }); + + it("defaults inactiveOnly to true (the heartbeat is quiet by default while the workspace is busy)", () => { + expect(DEFAULT_HEARTBEAT_CONFIG.inactiveOnly).toBe(true); + }); + + it("applies an inactiveOnly update", () => { + const next = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { inactiveOnly: false }); + expect(next.inactiveOnly).toBe(false); + }); + + it("leaves inactiveOnly unchanged when absent from the update", () => { + const off = applyConfigUpdate(DEFAULT_HEARTBEAT_CONFIG, { inactiveOnly: false }); + const untouched = applyConfigUpdate(off, { enabled: true }); + expect(untouched.inactiveOnly).toBe(false); + }); +}); + +describe("createHeartbeatConfigStore", () => { + it("returns the default config for an unknown workspace", async () => { + const store = createHeartbeatConfigStore(createMemoryStorage()); + expect(await store.get("ws-1")).toEqual(DEFAULT_HEARTBEAT_CONFIG); + }); + + it("persists and round-trips an update", async () => { + const storage = createMemoryStorage(); + const store = createHeartbeatConfigStore(storage); + const next = await store.update("ws-1", { enabled: true, intervalMinutes: 5 }); + expect(next.enabled).toBe(true); + expect(next.intervalMinutes).toBe(5); + // A fresh store over the same storage reads the persisted value. + const store2 = createHeartbeatConfigStore(storage); + expect(await store2.get("ws-1")).toEqual(next); + }); + + it("applies a partial update on top of existing config", async () => { + const storage = createMemoryStorage(); + const store = createHeartbeatConfigStore(storage); + await store.update("ws-1", { enabled: true, systemPrompt: "a", taskPrompt: "b" }); + const next = await store.update("ws-1", { taskPrompt: "c" }); + expect(next.enabled).toBe(true); + expect(next.systemPrompt).toBe("a"); + expect(next.taskPrompt).toBe("c"); + }); + + it("round-trips inactiveOnly through persistence", async () => { + const storage = createMemoryStorage(); + const store = createHeartbeatConfigStore(storage); + const next = await store.update("ws-1", { inactiveOnly: false }); + expect(next.inactiveOnly).toBe(false); + const store2 = createHeartbeatConfigStore(storage); + expect((await store2.get("ws-1")).inactiveOnly).toBe(false); + }); + + it("defaults inactiveOnly to true for legacy configs persisted before the field existed", async () => { + const storage = createMemoryStorage(); + // Simulate a config written by an older build that had no inactiveOnly + // field (a pre-inactive-only heartbeat config). + await storage.set("config:ws-1", JSON.stringify({ enabled: true, intervalMinutes: 5 })); + const store = createHeartbeatConfigStore(storage); + const config = await store.get("ws-1"); + expect(config.enabled).toBe(true); + expect(config.intervalMinutes).toBe(5); + // The missing field defaults ON (feature on by default) — never `undefined`. + expect(config.inactiveOnly).toBe(true); + }); + + it("lists persisted workspace ids", async () => { + const store = createHeartbeatConfigStore(createMemoryStorage()); + await store.update("ws-b", { enabled: true }); + await store.update("ws-a", { enabled: false }); + expect(await store.listWorkspaceIds()).toEqual(["ws-a", "ws-b"]); + }); +}); diff --git a/packages/heartbeat/src/config-store.ts b/packages/heartbeat/src/config-store.ts new file mode 100644 index 0000000..52aca3e --- /dev/null +++ b/packages/heartbeat/src/config-store.ts @@ -0,0 +1,106 @@ +import type { StorageNamespace } from "@dispatch/kernel"; +import type { HeartbeatConfig, UpdateHeartbeatRequest } from "@dispatch/transport-contract"; + +/** + * The default config returned for a workspace that has never configured a + * heartbeat. A heartbeat is OFF until explicitly enabled. + */ +export const DEFAULT_HEARTBEAT_CONFIG: HeartbeatConfig = { + enabled: false, + inactiveOnly: true, + systemPrompt: "", + taskPrompt: "", + intervalMinutes: 30, + model: "", + reasoningEffort: null, +}; + +/** Minimum scheduling interval, in minutes (a heartbeat can't fire faster). */ +export const MIN_INTERVAL_MINUTES = 1; + +/** Storage key for a workspace's heartbeat config. */ +function configKey(workspaceId: string): string { + return `config:${workspaceId}`; +} + +/** + * Pure: apply a partial update to a config, returning the new config. Clamps + * `intervalMinutes` to a minimum of 1 (a positive integer). Fields not present + * in `update` are left unchanged. `reasoningEffort: null` clears the override + * (back to inheriting the workspace default); `undefined` (absent) leaves it. + * + * Validation of `reasoningEffort` (rejecting unrecognized strings) is the + * transport layer's job (→ HTTP 400); this function trusts a caller that has + * already validated, but only ever produces a valid `HeartbeatConfig`. + */ +export function applyConfigUpdate( + current: HeartbeatConfig, + update: UpdateHeartbeatRequest, +): HeartbeatConfig { + const next: HeartbeatConfig = { + enabled: update.enabled !== undefined ? update.enabled : current.enabled, + inactiveOnly: update.inactiveOnly !== undefined ? update.inactiveOnly : current.inactiveOnly, + systemPrompt: update.systemPrompt !== undefined ? update.systemPrompt : current.systemPrompt, + taskPrompt: update.taskPrompt !== undefined ? update.taskPrompt : current.taskPrompt, + intervalMinutes: + update.intervalMinutes !== undefined + ? Math.max(MIN_INTERVAL_MINUTES, Math.trunc(update.intervalMinutes)) + : current.intervalMinutes, + model: update.model !== undefined ? update.model : current.model, + reasoningEffort: + update.reasoningEffort !== undefined ? update.reasoningEffort : current.reasoningEffort, + }; + return next; +} + +export interface HeartbeatConfigStore { + /** The config for a workspace (the default when never set). */ + readonly get: (workspaceId: string) => Promise<HeartbeatConfig>; + /** Apply a partial update and persist it; returns the new config. */ + readonly update: ( + workspaceId: string, + update: UpdateHeartbeatRequest, + ) => Promise<HeartbeatConfig>; + /** Every workspace id that has a persisted (non-default) config. */ + readonly listWorkspaceIds: () => Promise<readonly string[]>; +} + +export function createHeartbeatConfigStore(storage: StorageNamespace): HeartbeatConfigStore { + return { + async get(workspaceId: string): Promise<HeartbeatConfig> { + const raw = await storage.get(configKey(workspaceId)); + if (raw === null) return DEFAULT_HEARTBEAT_CONFIG; + try { + const parsed = JSON.parse(raw) as Partial<HeartbeatConfig>; + return { + enabled: typeof parsed.enabled === "boolean" ? parsed.enabled : false, + inactiveOnly: typeof parsed.inactiveOnly === "boolean" ? parsed.inactiveOnly : true, + systemPrompt: typeof parsed.systemPrompt === "string" ? parsed.systemPrompt : "", + taskPrompt: typeof parsed.taskPrompt === "string" ? parsed.taskPrompt : "", + intervalMinutes: + typeof parsed.intervalMinutes === "number" && parsed.intervalMinutes > 0 + ? Math.trunc(parsed.intervalMinutes) + : DEFAULT_HEARTBEAT_CONFIG.intervalMinutes, + model: typeof parsed.model === "string" ? parsed.model : "", + reasoningEffort: parsed.reasoningEffort ?? null, + }; + } catch { + return DEFAULT_HEARTBEAT_CONFIG; + } + }, + + async update(workspaceId: string, update: UpdateHeartbeatRequest): Promise<HeartbeatConfig> { + const current = await this.get(workspaceId); + const next = applyConfigUpdate(current, update); + await storage.set(configKey(workspaceId), JSON.stringify(next)); + return next; + }, + + async listWorkspaceIds(): Promise<readonly string[]> { + const keys = await storage.keys("config:"); + const ids = keys.map((k) => k.slice("config:".length)); + // De-duplicate + sort for a stable boot-scan order. + return [...new Set(ids)].sort(); + }, + }; +} diff --git a/packages/heartbeat/src/extension.ts b/packages/heartbeat/src/extension.ts new file mode 100644 index 0000000..b044619 --- /dev/null +++ b/packages/heartbeat/src/extension.ts @@ -0,0 +1,121 @@ +/** + * Heartbeat extension — manifest + activate(host). + * + * Wires the heartbeat service against the session-orchestrator + a storage + * namespace, registers the typed service handle, and arms every enabled + * workspace's scheduler on boot. Prompt templates (`systemPrompt` / + * `taskPrompt`) are resolved against the SAME variable catalog the global + * system-prompt template uses (via the system-prompt service's `resolveText`), + * so `[type:name]` placeholders reach the model substituted — not raw. An empty + * heartbeat `systemPrompt` inherits the global system prompt template (via + * `getTemplate`) before variable resolution runs — empty = inherit, not "no + * system prompt". + */ + +import type { ConversationStore } from "@dispatch/conversation-store"; +import { conversationStoreHandle } from "@dispatch/conversation-store"; +import type { Extension, HostAPI, Manifest } from "@dispatch/kernel"; +import { + type SessionOrchestrator, + sessionOrchestratorHandle, +} from "@dispatch/session-orchestrator"; +import type { SystemPromptService } from "@dispatch/system-prompt"; +import { systemPromptHandle } from "@dispatch/system-prompt"; +import { createHeartbeatService, heartbeatServiceHandle } from "./heartbeat.js"; + +export const manifest: Manifest = { + id: "heartbeat", + name: "Heartbeat", + version: "0.0.0", + apiVersion: "^0.1.0", + trust: "bundled", + // system-prompt provides `resolveText` (the variable resolver used to + // substitute [type:name] placeholders in heartbeat prompts); conversation- + // store resolves the workspace's default cwd (the resolver runs git / reads + // files against it, mirroring the global template). Both lookups are lazy + // (at fire time, not activation), but declaring them keeps the DAG honest. + dependsOn: ["session-orchestrator", "system-prompt", "conversation-store"], + activation: "eager", + contributes: { services: ["heartbeat"] }, +}; + +// Module-scoped store for deactivate (the extension object is created once). +const store: { service: { stopAll: () => void } | null } = { service: null }; + +export const extension: Extension = { + manifest, + async activate(host: HostAPI) { + const orchestrator = host.getService<SessionOrchestrator>(sessionOrchestratorHandle); + const storage = host.storage("heartbeat"); + const logger = host.logger; + + // Resolve [type:name] placeholders in heartbeat prompts via the + // system-prompt service (same resolver + variables as the global + // template). The cwd is the workspace's defaultCwd (resolved the same + // way the orchestrator resolves a new conversation's effective cwd); + // falling back to process.cwd() when the workspace has none. Both + // services are declared `dependsOn` (always activated before heartbeat). + const systemPromptService = host.getService<SystemPromptService>(systemPromptHandle); + const conversationStore = host.getService<ConversationStore>(conversationStoreHandle); + + const resolvePrompt = async ( + template: string, + ctx: { + readonly workspaceId: string; + readonly conversationId: string; + readonly model: string; + }, + ): Promise<string> => { + const workspace = await conversationStore.getWorkspace(ctx.workspaceId); + const cwd = workspace?.defaultCwd ?? process.cwd(); + return systemPromptService.resolveText(template, cwd, { + ...(ctx.conversationId !== "" ? { conversationId: ctx.conversationId } : {}), + ...(ctx.model !== "" ? { model: ctx.model } : {}), + workspaceId: ctx.workspaceId, + }); + }; + + const service = createHeartbeatService({ + storage, + orchestrator, + logger, + resolvePrompt, + // CR-HB-2: an empty heartbeat systemPrompt inherits the global + // system prompt template (GET /system-prompt / regular + // conversations resolve) before variable resolution runs. + getGlobalSystemPrompt: () => systemPromptService.getTemplate(), + // Pin the heartbeat turn's cwd to the CONFIGURED workspace's + // defaultCwd (not the heartbeat workspace's empty defaultCwd) so + // the turn's tools run where the prompt's [prompt:cwd] advertises. + // Lazy (resolved at fire time, mirroring resolvePrompt). + getWorkspaceCwd: async (wsId) => + (await conversationStore.getWorkspace(wsId))?.defaultCwd ?? null, + // inactiveOnly: the configured workspace is "busy" while any of its + // conversations are driving or queued for a turn. The orchestrator + // sets persisted status "active" on turn start and "idle" on settle, + // so a single store read (filtered to the configured workspaceId) is + // the live active-agent check. The heartbeat's own spawned conversation + // lives in the DEDICATED heartbeat workspace, so it never self-blocks. + hasActiveAgents: async (wsId) => { + const active = await conversationStore.listConversations({ + workspaceId: wsId, + status: ["active", "queued"], + }); + return active.length > 0; + }, + }); + + // Reconcile stale runs + arm enabled workspaces on boot. + await service.startAll(); + store.service = service; + + host.provideService(heartbeatServiceHandle, service); + + host.logger.info("heartbeat extension activated"); + }, + deactivate() { + // Stop schedulers so no new fires happen during shutdown. + store.service?.stopAll(); + store.service = null; + }, +}; diff --git a/packages/heartbeat/src/heartbeat.test.ts b/packages/heartbeat/src/heartbeat.test.ts new file mode 100644 index 0000000..a6f7ebb --- /dev/null +++ b/packages/heartbeat/src/heartbeat.test.ts @@ -0,0 +1,781 @@ +import type { StorageNamespace } from "@dispatch/kernel"; +import type { + EnqueueInput, + EnqueueResult, + SessionOrchestrator, + StartTurnResult, + TurnEventListener, +} from "@dispatch/session-orchestrator"; +import { describe, expect, it } from "vitest"; +import { createHeartbeatService } from "./heartbeat.js"; + +function createMemoryStorage(): StorageNamespace { + const data = new Map<string, string>(); + return { + get: async (key) => data.get(key) ?? null, + set: async (key, value) => { + data.set(key, value); + }, + delete: async (key) => { + data.delete(key); + }, + has: async (key) => data.has(key), + keys: async (prefix) => { + const all = [...data.keys()]; + if (prefix === undefined) return all; + return all.filter((k) => k.startsWith(prefix)); + }, + }; +} + +/** A controllable fake clock with an `advance` to move virtual time. */ +function createFakeTimers() { + let now = 0; + let nextId = 1; + const timers = new Map<number, { readonly fn: () => void; readonly firesAt: number }>(); + return { + timers: { + now: () => now, + setTimeout: (fn: () => void, ms: number) => { + const id = nextId++; + timers.set(id, { fn, firesAt: now + ms }); + return id as unknown as ReturnType<typeof setTimeout>; + }, + clearTimeout: (handle: ReturnType<typeof setTimeout> | undefined) => { + if (handle !== undefined) timers.delete(handle as unknown as number); + }, + }, + advance(ms: number): void { + now += ms; + const due = [...timers.entries()] + .filter(([, t]) => t.firesAt <= now) + .sort((a, b) => a[0] - b[0]); + for (const [id, t] of due) { + timers.delete(id); + t.fn(); + } + }, + }; +} + +interface PendingTurn { + readonly conversationId: string; + readonly text: string; + readonly systemPrompt?: string; + readonly modelName?: string; + readonly reasoningEffort?: unknown; + readonly workspaceId?: string; + readonly cwd?: string; + resolve: () => void; +} + +/** + * A fake orchestrator that records handleMessage calls and lets the test + * control when each turn seals. This is the injected edge — the service depends + * on the SessionOrchestrator interface, not its implementation. + */ +function createFakeOrchestrator(): SessionOrchestrator & { + readonly pending: readonly PendingTurn[]; + readonly stopped: readonly string[]; +} { + const pending: PendingTurn[] = []; + const stopped: string[] = []; + const turns = new Map<string, { resolve: () => void; reject: (e: unknown) => void }>(); + + const fake: SessionOrchestrator = { + startTurn(): StartTurnResult { + return { started: false, reason: "already-active" }; + }, + enqueue(_input: EnqueueInput): EnqueueResult { + return { startedTurn: false, queue: [] }; + }, + subscribe(_conversationId: string, _listener: TurnEventListener): () => void { + return () => {}; + }, + isActive(_conversationId: string): boolean { + return false; + }, + closeConversation(_conversationId: string): { abortedTurn: boolean } { + return { abortedTurn: false }; + }, + stopTurn(conversationId: string): { abortedTurn: boolean } { + stopped.push(conversationId); + const t = turns.get(conversationId); + if (t !== undefined) { + turns.delete(conversationId); + t.resolve(); + } + return { abortedTurn: true }; + }, + handleMessage(input): Promise<void> { + return new Promise<void>((resolve, reject) => { + const entry: PendingTurn = { + conversationId: input.conversationId, + text: input.text, + ...(input.systemPrompt !== undefined ? { systemPrompt: input.systemPrompt } : {}), + ...(input.modelName !== undefined ? { modelName: input.modelName } : {}), + ...(input.reasoningEffort !== undefined + ? { reasoningEffort: input.reasoningEffort } + : {}), + ...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}), + ...(input.cwd !== undefined ? { cwd: input.cwd } : {}), + resolve, + }; + pending.push(entry); + turns.set(input.conversationId, { resolve, reject }); + }); + }, + }; + return Object.assign(fake, { + get pending(): readonly PendingTurn[] { + return pending; + }, + get stopped(): readonly string[] { + return stopped; + }, + }); +} + +// Drain microtasks so async .finally handlers (run completion) run. `fire` has +// nested awaits (configStore.get → storage.get, runStore.create → storage.set) +// before it reaches handleMessage, so a single queueMicrotask isn't enough — +// setTimeout(0) schedules a macrotask, letting ALL pending microtasks drain. +const flush = async (): Promise<void> => { + await new Promise((r) => setTimeout(r, 0)); +}; + +function createService(opts: { + readonly orch: ReturnType<typeof createFakeOrchestrator>; + readonly storage?: StorageNamespace; + readonly resolvePrompt?: ( + template: string, + ctx: { + readonly workspaceId: string; + readonly conversationId: string; + readonly model: string; + }, + ) => Promise<string>; + readonly getGlobalSystemPrompt?: () => Promise<string>; + readonly getWorkspaceCwd?: (workspaceId: string) => Promise<string | null>; + readonly hasActiveAgents?: (workspaceId: string) => Promise<boolean>; +}) { + const fake = createFakeTimers(); + let id = 0; + const storage = opts.storage ?? createMemoryStorage(); + const svc = createHeartbeatService({ + storage, + orchestrator: opts.orch, + timers: fake.timers, + generateId: () => `id-${++id}`, + ...(opts.resolvePrompt !== undefined ? { resolvePrompt: opts.resolvePrompt } : {}), + ...(opts.getGlobalSystemPrompt !== undefined + ? { getGlobalSystemPrompt: opts.getGlobalSystemPrompt } + : {}), + ...(opts.getWorkspaceCwd !== undefined ? { getWorkspaceCwd: opts.getWorkspaceCwd } : {}), + ...(opts.hasActiveAgents !== undefined ? { hasActiveAgents: opts.hasActiveAgents } : {}), + }); + return { svc, advance: fake.advance, storage }; +} + +describe("createHeartbeatService", () => { + it("returns the default config for an unknown workspace", async () => { + const { svc } = createService({ orch: createFakeOrchestrator() }); + const cfg = await svc.getConfig("ws-1"); + expect(cfg.enabled).toBe(false); + expect(cfg.intervalMinutes).toBe(30); + // inactiveOnly defaults ON (the heartbeat is quiet by default while the + // workspace is busy). + expect(cfg.inactiveOnly).toBe(true); + }); + + describe("inactiveOnly (skip fire while the workspace has active agents)", () => { + it("skips the fire (records no run) when inactiveOnly is true and the workspace has active agents", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => Promise.resolve(true), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); // interval elapses → fire + await flush(); // let the async fire() reach the active-agent check + return + + expect(orch.pending).toHaveLength(0); // no turn started + expect(await svc.listRuns("ws-1")).toHaveLength(0); // no run recorded + }); + + it("still fires when inactiveOnly is true but the workspace has NO active agents", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => Promise.resolve(false), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); + await flush(); + expect(orch.pending).toHaveLength(1); + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("running"); + orch.pending[0]!.resolve(); + await flush(); + }); + + it("fires unconditionally when inactiveOnly is false, even with active agents", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + // Active agents reported, but the setting is OFF → must not block. + hasActiveAgents: () => Promise.resolve(true), + }); + await svc.updateConfig("ws-1", { + enabled: true, + inactiveOnly: false, + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + expect(orch.pending).toHaveLength(1); + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("running"); + orch.pending[0]!.resolve(); + await flush(); + }); + + it("re-arms and fires on the next interval after a skipped fire (the scheduler keeps ticking)", async () => { + const orch = createFakeOrchestrator(); + let busy = true; // workspace busy on the first fire, free on the next + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => Promise.resolve(busy), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); // first fire — skipped (busy) + await flush(); + expect(orch.pending).toHaveLength(0); + + busy = false; // workspace goes idle + advance(60_000); // next interval → fire again + await flush(); + expect(orch.pending).toHaveLength(1); // now it fires + orch.pending[0]!.resolve(); + await flush(); + }); + + it("does not consult hasActiveAgents when inactiveOnly is false (degrades off cleanly)", async () => { + const orch = createFakeOrchestrator(); + let consulted = false; + const { svc, advance } = createService({ + orch, + hasActiveAgents: () => { + consulted = true; + return Promise.resolve(true); + }, + }); + await svc.updateConfig("ws-1", { + enabled: true, + inactiveOnly: false, + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + expect(consulted).toBe(false); // never asked — setting is off + expect(orch.pending).toHaveLength(1); + orch.pending[0]!.resolve(); + await flush(); + }); + + it("checks active agents per configured workspace (only the configured workspace is consulted)", async () => { + const orch = createFakeOrchestrator(); + const busyWorkspaces = new Set<string>(["ws-busy"]); + const { svc, advance } = createService({ + orch, + hasActiveAgents: (wsId) => Promise.resolve(busyWorkspaces.has(wsId)), + }); + // ws-free is idle, ws-busy has active agents. + await svc.updateConfig("ws-free", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + await svc.updateConfig("ws-busy", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); // both fire + await flush(); + // ws-free fired (idle), ws-busy skipped (busy). + expect(orch.pending).toHaveLength(1); + expect((await svc.listRuns("ws-free"))[0]?.status).toBe("running"); + expect(await svc.listRuns("ws-busy")).toHaveLength(0); + orch.pending[0]!.resolve(); + await flush(); + }); + }); + + it("arming an enabled config does not fire immediately (waits for the interval)", () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + void svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go" }); + // Just arming → no turn yet. + expect(orch.pending).toHaveLength(0); + // Advancing just shy of the interval still no fire. + advance(59_999); + expect(orch.pending).toHaveLength(0); + }); + + it("fire sends the task prompt with the explicit system prompt + model + effort, and marks completed on seal", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { + enabled: true, + systemPrompt: "you are a monitor", + taskPrompt: "check stuck chats", + model: "opencode/gpt-4o", + reasoningEffort: "high", + intervalMinutes: 1, + }); + + advance(60_000); // 1-minute interval → fire + await flush(); // let the async fire() reach handleMessage + expect(orch.pending).toHaveLength(1); + const turn = orch.pending[0]!; + expect(turn.text).toBe("check stuck chats"); + expect(turn.systemPrompt).toBe("you are a monitor"); + expect(turn.modelName).toBe("opencode/gpt-4o"); + expect(turn.reasoningEffort).toBe("high"); + // Spawned conversations go to the DEDICATED heartbeat workspace (not + // the configured workspace), while the run stays tracked under ws-1. + expect(turn.workspaceId).toBe("heartbeat"); + + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("running"); + + turn.resolve(); + await flush(); + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("completed"); + }); + + it("heartbeat conversations always go to the dedicated heartbeat workspace, regardless of the configured workspace", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + // Two DIFFERENT workspaces each configure a heartbeat. + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + await svc.updateConfig("ws-2", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); // both fire (1-minute interval) + await flush(); + expect(orch.pending).toHaveLength(2); + // Every spawned conversation is filed in the heartbeat workspace — + // NOT ws-1 or ws-2 — so heartbeat runs don't clog either workspace's + // tabs. The run history, however, stays tracked per configured workspace. + for (const turn of orch.pending) { + expect(turn.workspaceId).toBe("heartbeat"); + } + // Run history is still per configured workspace. + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("running"); + expect((await svc.listRuns("ws-2"))[0]?.status).toBe("running"); + for (const turn of orch.pending) { + turn.resolve(); + } + await flush(); + }); + + it("the turn cwd is pinned to the CONFIGURED workspace's defaultCwd (not the heartbeat workspace's empty defaultCwd)", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + // The configured workspace has a defaultCwd. + getWorkspaceCwd: (wsId) => Promise.resolve(wsId === "ws-1" ? "/home/proj/ws-1" : null), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); + await flush(); + const turn = orch.pending[0]!; + // The conversation is filed in the heartbeat workspace … + expect(turn.workspaceId).toBe("heartbeat"); + // … but the turn's cwd is the CONFIGURED workspace's defaultCwd, so + // tools run where the prompt's [prompt:cwd] advertises (not the + // heartbeat workspace's empty defaultCwd → process.cwd()). + expect(turn.cwd).toBe("/home/proj/ws-1"); + turn.resolve(); + await flush(); + }); + + it("omits the cwd override when the configured workspace has no defaultCwd (orchestrator falls back to the server default)", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ + orch, + // The configured workspace has NO defaultCwd. + getWorkspaceCwd: () => Promise.resolve(null), + }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); + await flush(); + const turn = orch.pending[0]!; + expect(turn.workspaceId).toBe("heartbeat"); + // No cwd override sent — the orchestrator resolves the turn cwd from + // the heartbeat workspace (no defaultCwd → server default cwd). + expect(turn.cwd).toBeUndefined(); + turn.resolve(); + await flush(); + }); + + it("omits the cwd override when getWorkspaceCwd is not wired (resolution is optional)", async () => { + const orch = createFakeOrchestrator(); + // No getWorkspaceCwd → no cwd override (the default). + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + + advance(60_000); + await flush(); + const turn = orch.pending[0]!; + expect(turn.workspaceId).toBe("heartbeat"); + expect(turn.cwd).toBeUndefined(); + turn.resolve(); + await flush(); + }); + + it("omits modelName/reasoningEffort when empty/null (inherit defaults)", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { + enabled: true, + taskPrompt: "go", + model: "", + reasoningEffort: null, + intervalMinutes: 1, + }); + advance(60_000); + await flush(); + const turn = orch.pending[0]!; + expect(turn.modelName).toBeUndefined(); + expect(turn.reasoningEffort).toBeUndefined(); + turn.resolve(); + await flush(); + }); + + it("resolves [type:name] variables in both systemPrompt and taskPrompt before sending", async () => { + const orch = createFakeOrchestrator(); + // A fake resolver that mirrors the real resolver's contract: substitute + // known [type:name] placeholders, leave unknown text verbatim. + const resolvePrompt = async ( + template: string, + ctx: { + readonly workspaceId: string; + readonly conversationId: string; + readonly model: string; + }, + ): Promise<string> => { + return template + .replaceAll("[system:os]", "Linux (WSL)") + .replaceAll("[prompt:cwd]", "/repo") + .replaceAll("[prompt:workspace_id]", ctx.workspaceId) + .replaceAll("[prompt:conversation_id]", ctx.conversationId) + .replaceAll("[prompt:model]", ctx.model); + }; + const { svc, advance } = createService({ orch, resolvePrompt }); + await svc.updateConfig("ws-1", { + enabled: true, + systemPrompt: "You run on [system:os] in [prompt:cwd] (ws [prompt:workspace_id]).", + taskPrompt: "Check chats for [prompt:conversation_id] on [system:os].", + model: "opencode/gpt-4o", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + expect(orch.pending).toHaveLength(1); + const turn = orch.pending[0]!; + // Variables substituted — NOT left as literal [type:name] text. + expect(turn.systemPrompt).toBe("You run on Linux (WSL) in /repo (ws ws-1)."); + expect(turn.text).toBe(`Check chats for ${turn.conversationId} on Linux (WSL).`); + expect(turn.modelName).toBe("opencode/gpt-4o"); + turn.resolve(); + await flush(); + }); + + it("passes prompts through raw when no resolver is wired (resolution is optional)", async () => { + const orch = createFakeOrchestrator(); + // No resolvePrompt → raw pass-through (the default). + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { + enabled: true, + systemPrompt: "raw [system:os] prompt", + taskPrompt: "raw [system:date] task", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + const turn = orch.pending[0]!; + // Unresolved — literals reach the orchestrator verbatim. + expect(turn.systemPrompt).toBe("raw [system:os] prompt"); + expect(turn.text).toBe("raw [system:date] task"); + turn.resolve(); + await flush(); + }); + + it("an empty systemPrompt inherits the global system prompt template (CR-HB-2)", async () => { + const orch = createFakeOrchestrator(); + // The global getter returns the workspace's regular system prompt + // template (the same one GET /system-prompt returns). No resolver → the + // inherited template reaches the orchestrator verbatim (isolates CR-HB-2 + // from CR-HB-1). + const { svc, advance } = createService({ + orch, + getGlobalSystemPrompt: () => Promise.resolve("GLOBAL DEFAULT TEMPLATE"), + }); + await svc.updateConfig("ws-1", { + enabled: true, + // Empty systemPrompt = inherit the global default, NOT "no prompt". + systemPrompt: "", + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + expect(orch.pending).toHaveLength(1); + const turn = orch.pending[0]!; + expect(turn.systemPrompt).toBe("GLOBAL DEFAULT TEMPLATE"); + turn.resolve(); + await flush(); + }); + + it("CR-HB-2 composes with CR-HB-1: the inherited global template's [type:name] placeholders are resolved", async () => { + const orch = createFakeOrchestrator(); + const resolvePrompt = async ( + template: string, + ctx: { + readonly workspaceId: string; + readonly conversationId: string; + readonly model: string; + }, + ): Promise<string> => { + return template + .replaceAll("[system:os]", "Linux (WSL)") + .replaceAll("[prompt:cwd]", "/repo") + .replaceAll("[prompt:workspace_id]", ctx.workspaceId); + }; + // The global template carries [type:name] placeholders (like the real + // default template embeds [prompt:cwd] / [file:AGENTS.md]). + const { svc, advance } = createService({ + orch, + resolvePrompt, + getGlobalSystemPrompt: () => + Promise.resolve("You run on [system:os] in [prompt:cwd] (ws [prompt:workspace_id])."), + }); + await svc.updateConfig("ws-1", { + enabled: true, + // Empty → inherit the global template, THEN resolve its placeholders. + systemPrompt: "", + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + expect(orch.pending).toHaveLength(1); + const turn = orch.pending[0]!; + // The global template was inherited (not empty), then its [type:name] + // placeholders were substituted — NOT left literal, NOT empty. + expect(turn.systemPrompt).toBe("You run on Linux (WSL) in /repo (ws ws-1)."); + turn.resolve(); + await flush(); + }); + + it("a non-empty systemPrompt override bypasses the global template (only CR-HB-1 applies)", async () => { + const orch = createFakeOrchestrator(); + const resolvePrompt = async (template: string): Promise<string> => + template.replaceAll("[system:os]", "Linux (WSL)"); + const { svc, advance } = createService({ + orch, + resolvePrompt, + // A DISTINCT global template — must NOT be used when overriding. + getGlobalSystemPrompt: () => Promise.resolve("GLOBAL SHOULD NOT APPEAR [system:os]"), + }); + await svc.updateConfig("ws-1", { + enabled: true, + systemPrompt: "custom override on [system:os]", + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + const turn = orch.pending[0]!; + // The override is used (and resolved), not the global template. + expect(turn.systemPrompt).toBe("custom override on Linux (WSL)"); + turn.resolve(); + await flush(); + }); + + it("an empty systemPrompt stays empty when no global getter is wired (resolution is optional)", async () => { + const orch = createFakeOrchestrator(); + // No getGlobalSystemPrompt → empty stays empty (no system prompt). + // Mirrors the no-resolver pass-through default: both deps are optional. + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { + enabled: true, + systemPrompt: "", + taskPrompt: "go", + intervalMinutes: 1, + }); + + advance(60_000); + await flush(); + const turn = orch.pending[0]!; + expect(turn.systemPrompt).toBe(""); + turn.resolve(); + await flush(); + }); + + it("stopRun aborts the turn and marks the run stopped (not overwritten on completion)", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + advance(60_000); + await flush(); + const conversationId = orch.pending[0]?.conversationId; + const runId = (await svc.listRuns("ws-1"))[0]?.id; + + const res = await svc.stopRun("ws-1", runId); + expect(res).toEqual({ ok: true }); + expect(orch.stopped).toEqual([conversationId]); + + await flush(); // the aborted turn's handleMessage resolves + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("stopped"); + }); + + it("stopRun is idempotent for an already-finished run", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + advance(60_000); + await flush(); + orch.pending[0]?.resolve(); + await flush(); + + const runId = (await svc.listRuns("ws-1"))[0]?.id; + const res = await svc.stopRun("ws-1", runId); + expect(res).toEqual({ ok: true }); + expect(orch.stopped).toEqual([]); // no abort on a completed run + }); + + it("stopRun throws for an unknown run id", async () => { + const { svc } = createService({ orch: createFakeOrchestrator() }); + await expect(svc.stopRun("ws-1", "nope")).rejects.toThrow(); + }); + + it("disabling the config stops the schedule (no further fires)", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go" }); + await svc.updateConfig("ws-1", { enabled: false }); + advance(60_000); + expect(orch.pending).toHaveLength(0); + }); + + it("startAll sweeps stale running runs to stopped", async () => { + const storage = createMemoryStorage(); + await storage.set( + "run:ws-1:stale", + JSON.stringify({ + id: "stale", + conversationId: "c-stale", + triggeredAt: "2026-01-01T00:00:00.000Z", + status: "running", + }), + ); + await storage.set( + "config:ws-1", + JSON.stringify({ + enabled: false, + systemPrompt: "", + taskPrompt: "", + intervalMinutes: 30, + model: "", + reasoningEffort: null, + }), + ); + + const { svc } = createService({ orch: createFakeOrchestrator(), storage }); + await svc.startAll(); + expect((await svc.listRuns("ws-1"))[0]?.status).toBe("stopped"); + }); + + it("startAll arms enabled workspaces and skips disabled ones", async () => { + const orch = createFakeOrchestrator(); + const { svc: svc1, storage } = createService({ orch }); + await svc1.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + await svc1.updateConfig("ws-2", { enabled: false, taskPrompt: "no" }); + svc1.stopAll(); + + // Re-create a fresh service over the SAME storage (simulates a reboot), + // with new fake timers we can advance. + const fake = createFakeTimers(); + let id = 1000; + const svc2 = createHeartbeatService({ + storage, + orchestrator: orch, + timers: fake.timers, + generateId: () => `id-${++id}`, + }); + await svc2.startAll(); + fake.advance(60_000); + await flush(); + // ws-1 (enabled) fired → its spawned conversation is filed in the + // heartbeat workspace; ws-2 (disabled) never fired. + expect(orch.pending).toHaveLength(1); + expect(orch.pending[0]?.workspaceId).toBe("heartbeat"); + }); + + // ─── nextRunAt (CR-HB-3: server-authoritative next-run time) ──────────────── + + it("nextRunAt returns null when the heartbeat is disabled", async () => { + const { svc } = createService({ orch: createFakeOrchestrator() }); + // Default config → disabled → no schedule armed. + expect(await svc.nextRunAt("ws-1")).toBeNull(); + }); + + it("nextRunAt returns the ISO timestamp of the next fire when enabled", async () => { + const { svc } = createService({ orch: createFakeOrchestrator() }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + // now=0, interval=1m → next fire at epoch-ms 60_000 → its ISO form. + expect(await svc.nextRunAt("ws-1")).toBe(new Date(60_000).toISOString()); + }); + + it("nextRunAt returns null while a run is in progress, then the next fire after it completes", async () => { + const orch = createFakeOrchestrator(); + const { svc, advance } = createService({ orch }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + expect(await svc.nextRunAt("ws-1")).toBe(new Date(60_000).toISOString()); + + advance(60_000); // fire → run in progress + await flush(); + expect(orch.pending).toHaveLength(1); + // In flight → no next run queued yet. + expect(await svc.nextRunAt("ws-1")).toBeNull(); + + orch.pending[0]?.resolve(); + await flush(); + // Re-armed at completion-time (60_000) + interval (60_000) = 120_000. + expect(await svc.nextRunAt("ws-1")).toBe(new Date(120_000).toISOString()); + }); + + it("nextRunAt returns null after disabling the heartbeat", async () => { + const { svc } = createService({ orch: createFakeOrchestrator() }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + expect(await svc.nextRunAt("ws-1")).not.toBeNull(); + await svc.updateConfig("ws-1", { enabled: false }); + expect(await svc.nextRunAt("ws-1")).toBeNull(); + }); + + it("nextRunAt reflects a changed interval on the next re-arm", async () => { + const { svc } = createService({ orch: createFakeOrchestrator() }); + await svc.updateConfig("ws-1", { enabled: true, taskPrompt: "go", intervalMinutes: 1 }); + expect(await svc.nextRunAt("ws-1")).toBe(new Date(60_000).toISOString()); + // Re-arm with a 5-minute interval (not running) → recomputed fire time. + await svc.updateConfig("ws-1", { intervalMinutes: 5 }); + expect(await svc.nextRunAt("ws-1")).toBe(new Date(5 * 60_000).toISOString()); + }); +}); diff --git a/packages/heartbeat/src/heartbeat.ts b/packages/heartbeat/src/heartbeat.ts new file mode 100644 index 0000000..e04c538 --- /dev/null +++ b/packages/heartbeat/src/heartbeat.ts @@ -0,0 +1,354 @@ +import type { Logger, ServiceHandle } from "@dispatch/kernel"; +import { defineService } from "@dispatch/kernel"; +import type { SessionOrchestrator } from "@dispatch/session-orchestrator"; +import type { + HeartbeatConfig, + HeartbeatRun, + StopHeartbeatRunResponse, + UpdateHeartbeatRequest, +} from "@dispatch/transport-contract"; +import { createHeartbeatConfigStore, type HeartbeatConfigStore } from "./config-store.js"; +import { createHeartbeatRunStore, type HeartbeatRunStore } from "./run-store.js"; +import { HeartbeatScheduler, realTimers, type Timers } from "./scheduler.js"; + +/** + * The dedicated workspace heartbeat-spawned conversations are filed in (NOT the + * configured workspace). The config + run history stay per-workspace (tracked + * under the configured workspaceId); only the spawned conversation's PLACEMENT + * moves here, so heartbeat conversations don't clog the configured workspace's + * tabs. The orchestrator auto-creates this workspace on first fire (via + * `ensureWorkspace`), so it appears on the workspaces home page like any other. + */ +export const HEARTBEAT_WORKSPACE_ID = "heartbeat"; + +/** + * The heartbeat service surface — what transport-http consumes and what the + * extension wires into the host. + */ +export interface HeartbeatService { + /** The per-workspace heartbeat config (defaults when never set). */ + readonly getConfig: (workspaceId: string) => Promise<HeartbeatConfig>; + /** + * Apply a partial config update. Side effect: arms/disarms the scheduler + * for this workspace (enabled → schedule; disabled → stop). Returns the new + * config. + */ + readonly updateConfig: ( + workspaceId: string, + update: UpdateHeartbeatRequest, + ) => Promise<HeartbeatConfig>; + /** Heartbeat runs for a workspace, most-recent first. */ + readonly listRuns: (workspaceId: string) => Promise<readonly HeartbeatRun[]>; + /** + * The server-authoritative next-fire time for a workspace's heartbeat, as + * an ISO 8601 string — the moment the scheduler will fire the next run + * (the last run's completion + `intervalMinutes`, or the moment `enabled` + * was toggled on + `intervalMinutes` for the first run). `null` when the + * heartbeat is disabled/disarmed, or when a run is in flight and the next + * hasn't been queued yet (no countdown to show). A cheap read of the + * scheduler's pending fire time. + */ + readonly nextRunAt: (workspaceId: string) => Promise<string | null>; + /** + * Stop an in-flight run (abort its turn). Idempotent for an already-finished + * run. Throws when the run id is unknown (→ HTTP 404). + */ + readonly stopRun: (workspaceId: string, runId: string) => Promise<StopHeartbeatRunResponse>; + /** Boot: sweep stale runs + arm every enabled workspace's scheduler. */ + readonly startAll: () => Promise<void>; + /** Shutdown: stop every scheduler. */ + readonly stopAll: () => void; +} + +/** Typed service handle the heartbeat extension provides and transport consumes. */ +export const heartbeatServiceHandle: ServiceHandle<HeartbeatService> = + defineService<HeartbeatService>("heartbeat"); + +export interface HeartbeatServiceDeps { + /** Namespaced storage (from `host.storage("heartbeat")`). */ + readonly storage: import("@dispatch/kernel").StorageNamespace; + /** The session orchestrator (drives heartbeat turns). */ + readonly orchestrator: SessionOrchestrator; + readonly logger?: Logger; + /** Injectable timers (default: real). */ + readonly timers?: Timers; + /** Injectable id generator (default: crypto.randomUUID). */ + readonly generateId?: () => string; + /** + * Resolve `[type:name]` variable placeholders in a prompt template against + * the current environment — the SAME resolver + variable catalog the global + * system-prompt template uses (system/file/prompt/git groups). Applied once + * per run, when the turn is constructed (mirrors the global template's + * construct-once-per-conversation resolution). When omitted, templates pass + * through UNRESOLVED (raw) — the extension wires the real resolver; tests + * inject a fake. + */ + readonly resolvePrompt?: ( + template: string, + ctx: { + readonly workspaceId: string; + readonly conversationId: string; + readonly model: string; + }, + ) => Promise<string>; + /** + * Resolve an empty heartbeat `systemPrompt` to the GLOBAL system prompt + * template — the same one `GET /system-prompt` returns / that regular + * conversations resolve. Applied ONLY when the heartbeat's persisted + * `systemPrompt` is `""` (inherit), and BEFORE variable resolution + * (`resolvePrompt` / CR-HB-1) runs on the result, so both apply in order: + * empty ⇒ global template, then `[type:name]` placeholders resolved. A + * non-empty `systemPrompt` is an explicit override and bypasses this. + * When omitted, empty stays empty (no system prompt) — the extension + * wires the real getter; tests inject a fake. + */ + readonly getGlobalSystemPrompt?: () => Promise<string>; + /** + * The configured workspace's `defaultCwd` (or `null` when the workspace has + * none). Used to pin the heartbeat turn's cwd to the CONFIGURED workspace's + * directory — NOT the heartbeat workspace's (empty) defaultCwd — so the + * turn's tools run in the same directory the prompt's `[prompt:cwd]` + * variable advertises. Passed to the orchestrator as an explicit `cwd` + * override only when non-null; when `null` no override is sent and the + * orchestrator falls back to the server default cwd (matching the + * pre-heartbeat-workspace behavior for a workspace without a defaultCwd). + * When omitted, `null` (no override) — the extension wires the real getter + * (against `conversationStore.getWorkspace`); tests inject a fake. + */ + readonly getWorkspaceCwd?: (workspaceId: string) => Promise<string | null>; + /** + * Whether the configured workspace currently has any ACTIVE agents — + * conversations driving (or queued for) a turn. When `config.inactiveOnly` + * is `true`, the heartbeat SKIPS a fire while this returns `true` (the + * workspace is busy). The extension wires the real check against + * `conversationStore.listConversations({ workspaceId, status: ["active", + * "queued"] })` (the configured workspace's persisted statuses — the + * orchestrator sets `"active"` on turn start, `"idle"` on settle); tests + * inject a fake. When omitted, `false` (no active agents → never skip) so + * the inactive-only feature degrades off cleanly — the heartbeat fires + * unconditionally, matching the pre-inactive-only behavior. + */ + readonly hasActiveAgents?: (workspaceId: string) => Promise<boolean>; +} + +interface ActiveRun { + readonly conversationId: string; + readonly workspaceId: string; + stopped: boolean; +} + +export function createHeartbeatService(deps: HeartbeatServiceDeps): HeartbeatService { + const logger = deps.logger; + const timers = deps.timers ?? realTimers; + const generateId = deps.generateId ?? (() => crypto.randomUUID()); + const configStore: HeartbeatConfigStore = createHeartbeatConfigStore(deps.storage); + const runStore: HeartbeatRunStore = createHeartbeatRunStore(deps.storage); + const orchestrator = deps.orchestrator; + // Default: pass templates through UNRESOLVED (raw). The extension wires the + // real resolver so [type:name] placeholders are substituted like the global + // system-prompt template; tests inject a fake. + const resolvePrompt = deps.resolvePrompt ?? ((template: string) => Promise.resolve(template)); + // Default: an empty systemPrompt stays empty (no system prompt). The + // extension wires the real getter so empty INHERITS the global system + // prompt template (GET /system-prompt); tests inject a fake. + const getGlobalSystemPrompt = deps.getGlobalSystemPrompt ?? (() => Promise.resolve("")); + // Default: no cwd override (the orchestrator resolves the turn cwd from the + // conversation's workspace). The extension wires the real getter so the + // turn pins to the CONFIGURED workspace's defaultCwd; tests inject a fake. + const getWorkspaceCwd = deps.getWorkspaceCwd ?? (() => Promise.resolve(null)); + // Default: no active agents (never skip) — the inactive-only feature degrades + // off cleanly. The extension wires the real check (against the conversation + // store's persisted statuses); tests inject a fake. + const hasActiveAgents = deps.hasActiveAgents ?? (() => Promise.resolve(false)); + + // runId → active-run tracking (in-memory; the durable record lives in the + // run store). Used to (a) map a stop request to its conversation, and + // (b) keep a "stopped" flag so the turn's completion doesn't clobber a + // user-initiated stop. + const activeRuns = new Map<string, ActiveRun>(); + + const scheduler = new HeartbeatScheduler({ + timers, + fire: (workspaceId) => fire(workspaceId), + }); + + async function fire(workspaceId: string): Promise<void> { + const config = await configStore.get(workspaceId); + // Race: disabled/disarmed between the timer firing and now. + if (!config.enabled) return; + + // inactiveOnly: skip this fire when the configured workspace has active + // agents (a conversation driving or queued for a turn). The fire is + // silently skipped — no run is recorded — and the scheduler re-arms to + // try again at the next interval. The spawned heartbeat conversation lives + // in the DEDICATED heartbeat workspace, so it never self-blocks (a prior + // in-flight heartbeat run is NOT an active agent of the configured + // workspace). Disabled (inactiveOnly === false) fires unconditionally. + if (config.inactiveOnly && (await hasActiveAgents(workspaceId))) { + logger?.info("heartbeat: fire skipped — workspace has active agents", { workspaceId }); + return; + } + + const conversationId = generateId(); + const runId = generateId(); + const triggeredAt = new Date(timers.now()).toISOString(); + const run: HeartbeatRun = { + id: runId, + conversationId, + triggeredAt, + status: "running", + }; + await runStore.create(workspaceId, run); + activeRuns.set(runId, { conversationId, workspaceId, stopped: false }); + + logger?.info("heartbeat: run started", { workspaceId, runId, conversationId }); + + // Resolve the heartbeat's prompts ONCE, when the turn is constructed. + // + // CR-HB-2: an empty `systemPrompt` INHERITS the global system prompt + // template (the same one `GET /system-prompt` returns / that regular + // conversations resolve) — empty is an override-means-inherit flag, not + // "no system prompt". A non-empty `systemPrompt` is an explicit override. + // This step runs FIRST. + // + // CR-HB-1: then `[type:name]` variable placeholders in whichever prompt is + // in effect are resolved via the same resolver + variable catalog the + // global system-prompt template uses. The orchestrator sends an explicit + // systemPrompt override AS-IS (bypassing its own templated prompt), so + // resolution must happen HERE, before handleMessage. For prompts using + // stable variables (os/cwd/git) this yields a stable, cache-warm prompt + // across runs; time-bearing variables refresh per run (mirroring the + // global template's per-conversation resolution). + const baseSystemPrompt = + config.systemPrompt === "" ? await getGlobalSystemPrompt() : config.systemPrompt; + const resolveCtx = { workspaceId, conversationId, model: config.model }; + // resolveCtx uses the CONFIGURED workspaceId — the heartbeat operates + // ON BEHALF OF the configured workspace, so `[prompt:workspace_id]` and + // `[prompt:cwd]` refer to it (not the heartbeat workspace the spawned + // conversation is filed in). The turn's cwd is pinned to the SAME + // configured workspace's defaultCwd (below) so tools run where the + // prompt's `[prompt:cwd]` advertises. + const [systemPrompt, taskPrompt, configuredWorkspaceCwd] = await Promise.all([ + resolvePrompt(baseSystemPrompt, resolveCtx), + resolvePrompt(config.taskPrompt, resolveCtx), + getWorkspaceCwd(workspaceId), + ]); + + try { + await orchestrator.handleMessage({ + conversationId, + text: taskPrompt, + // Fire-and-forget: the heartbeat loop does not consume the + // streamed events (it only awaits turn completion to mark the + // run done). A no-op onEvent satisfies the required callback. + onEvent: () => {}, + // Always passed explicitly — bypasses the orchestrator's templated + // workspace prompt. Resolved above: an empty config systemPrompt + // inherited the global template (CR-HB-2), then [type:name] + // placeholders were substituted (CR-HB-1). Still "" when the global + // template itself is empty (no system prompt). + systemPrompt, + ...(config.model !== "" ? { modelName: config.model } : {}), + ...(config.reasoningEffort !== null ? { reasoningEffort: config.reasoningEffort } : {}), + // Pin the turn cwd to the CONFIGURED workspace's defaultCwd so + // the heartbeat's tools run in the same directory its prompt + // variables advertise — NOT the heartbeat workspace's (empty) + // defaultCwd → process.cwd(). Omitted when the configured + // workspace has no defaultCwd (the orchestrator then falls back + // to the server default cwd, matching the pre-heartbeat- + // workspace behavior for a workspace without one). + ...(configuredWorkspaceCwd !== null ? { cwd: configuredWorkspaceCwd } : {}), + // File the spawned conversation in the DEDICATED heartbeat + // workspace (not the configured workspace) so heartbeat + // conversations don't clog the configured workspace's tabs. The + // orchestrator auto-creates this workspace on first fire + // (ensureWorkspace), so it appears on the workspaces home page. + workspaceId: HEARTBEAT_WORKSPACE_ID, + }); + } finally { + const entry = activeRuns.get(runId); + activeRuns.delete(runId); + // If the user stopped it, stopRun already set "stopped"; don't + // clobber. Otherwise the turn sealed (normally or via abort) → done. + if (entry !== undefined && !entry.stopped) { + await runStore.setStatus(workspaceId, runId, "completed"); + logger?.info("heartbeat: run completed", { workspaceId, runId }); + } + } + } + + return { + async getConfig(workspaceId) { + return configStore.get(workspaceId); + }, + + async updateConfig(workspaceId, update) { + const next = await configStore.update(workspaceId, update); + // Arm/disarm from the new config. An in-progress run is left alone; + // the new interval takes effect on the next re-arm. + scheduler.arm(workspaceId, next); + logger?.info("heartbeat: config updated", { + workspaceId, + enabled: next.enabled, + intervalMinutes: next.intervalMinutes, + }); + return next; + }, + + async listRuns(workspaceId) { + return runStore.list(workspaceId); + }, + + async nextRunAt(workspaceId) { + const ms = scheduler.nextFireAt(workspaceId); + return ms === null ? null : new Date(ms).toISOString(); + }, + + async stopRun(workspaceId, runId) { + const run = await runStore.get(workspaceId, runId); + if (run === null) { + throw new Error("Heartbeat run not found"); + } + // Idempotent: an already-finished run is a no-op. + if (run.status !== "running") { + return { ok: true }; + } + const entry = activeRuns.get(runId); + const conversationId = entry?.conversationId ?? run.conversationId; + if (entry !== undefined) { + entry.stopped = true; + } + await runStore.setStatus(workspaceId, runId, "stopped"); + orchestrator.stopTurn(conversationId); + logger?.info("heartbeat: run stopped", { workspaceId, runId, conversationId }); + return { ok: true }; + }, + + async startAll() { + // Sweep stale "running" runs (orphaned by a prior crash/restart) → + // "stopped". Never leave the system showing an in-flight run that + // can never finish. + const workspaceIds = await configStore.listWorkspaceIds(); + for (const workspaceId of workspaceIds) { + const runs = await runStore.list(workspaceId); + for (const run of runs) { + if (run.status === "running") { + await runStore.setStatus(workspaceId, run.id, "stopped"); + } + } + const config = await configStore.get(workspaceId); + if (config.enabled) { + scheduler.arm(workspaceId, config); + logger?.info("heartbeat: scheduler armed on boot", { + workspaceId, + intervalMinutes: config.intervalMinutes, + }); + } + } + }, + + stopAll() { + scheduler.disarmAll(); + }, + }; +} diff --git a/packages/heartbeat/src/index.ts b/packages/heartbeat/src/index.ts new file mode 100644 index 0000000..d25e338 --- /dev/null +++ b/packages/heartbeat/src/index.ts @@ -0,0 +1,17 @@ +export { + applyConfigUpdate, + createHeartbeatConfigStore, + DEFAULT_HEARTBEAT_CONFIG, + type HeartbeatConfigStore, + MIN_INTERVAL_MINUTES, +} from "./config-store.js"; +export { extension, manifest } from "./extension.js"; +export { + createHeartbeatService, + HEARTBEAT_WORKSPACE_ID, + type HeartbeatService, + type HeartbeatServiceDeps, + heartbeatServiceHandle, +} from "./heartbeat.js"; +export { createHeartbeatRunStore, type HeartbeatRunStore } from "./run-store.js"; +export { HeartbeatScheduler, realTimers, type TimerHandle, type Timers } from "./scheduler.js"; diff --git a/packages/heartbeat/src/run-store.test.ts b/packages/heartbeat/src/run-store.test.ts new file mode 100644 index 0000000..8be21f5 --- /dev/null +++ b/packages/heartbeat/src/run-store.test.ts @@ -0,0 +1,72 @@ +import type { StorageNamespace } from "@dispatch/kernel"; +import type { HeartbeatRun } from "@dispatch/transport-contract"; +import { describe, expect, it } from "vitest"; +import { createHeartbeatRunStore } from "./run-store.js"; + +function createMemoryStorage(): StorageNamespace { + const data = new Map<string, string>(); + return { + get: async (key) => data.get(key) ?? null, + set: async (key, value) => { + data.set(key, value); + }, + delete: async (key) => { + data.delete(key); + }, + has: async (key) => data.has(key), + keys: async (prefix) => { + const all = [...data.keys()]; + if (prefix === undefined) return all; + return all.filter((k) => k.startsWith(prefix)); + }, + }; +} + +function run(id: string, conversationId: string, triggeredAt: string): HeartbeatRun { + return { id, conversationId, triggeredAt, status: "running" }; +} + +describe("createHeartbeatRunStore", () => { + it("creates and reads back a run", async () => { + const store = createHeartbeatRunStore(createMemoryStorage()); + const created = await store.create("ws-1", run("r1", "c1", "2026-01-01T00:00:00.000Z")); + expect(created.status).toBe("running"); + const read = await store.get("ws-1", "r1"); + expect(read).toEqual(created); + }); + + it("returns null for an unknown run", async () => { + const store = createHeartbeatRunStore(createMemoryStorage()); + expect(await store.get("ws-1", "nope")).toBeNull(); + }); + + it("updates the status of a run", async () => { + const store = createHeartbeatRunStore(createMemoryStorage()); + await store.create("ws-1", run("r1", "c1", "2026-01-01T00:00:00.000Z")); + const updated = await store.setStatus("ws-1", "r1", "completed"); + expect(updated?.status).toBe("completed"); + expect((await store.get("ws-1", "r1"))?.status).toBe("completed"); + }); + + it("setStatus is a no-op (returns null) for an unknown run", async () => { + const store = createHeartbeatRunStore(createMemoryStorage()); + expect(await store.setStatus("ws-1", "ghost", "stopped")).toBeNull(); + }); + + it("lists runs most-recent first by triggeredAt", async () => { + const store = createHeartbeatRunStore(createMemoryStorage()); + await store.create("ws-1", run("r1", "c1", "2026-01-01T00:00:00.000Z")); + await store.create("ws-1", run("r2", "c2", "2026-02-01T00:00:00.000Z")); + await store.create("ws-1", run("r3", "c3", "2026-01-15T00:00:00.000Z")); + const runs = await store.list("ws-1"); + expect(runs.map((r) => r.id)).toEqual(["r2", "r3", "r1"]); + }); + + it("scopes runs per workspace", async () => { + const store = createHeartbeatRunStore(createMemoryStorage()); + await store.create("ws-1", run("r1", "c1", "2026-01-01T00:00:00.000Z")); + await store.create("ws-2", run("r2", "c2", "2026-01-01T00:00:00.000Z")); + expect((await store.list("ws-1")).map((r) => r.id)).toEqual(["r1"]); + expect((await store.list("ws-2")).map((r) => r.id)).toEqual(["r2"]); + }); +}); diff --git a/packages/heartbeat/src/run-store.ts b/packages/heartbeat/src/run-store.ts new file mode 100644 index 0000000..9650181 --- /dev/null +++ b/packages/heartbeat/src/run-store.ts @@ -0,0 +1,95 @@ +import type { StorageNamespace } from "@dispatch/kernel"; +import type { HeartbeatRun, HeartbeatRunStatus } from "@dispatch/transport-contract"; + +/** Storage key for a single heartbeat run record. */ +function runKey(workspaceId: string, runId: string): string { + return `run:${workspaceId}:${runId}`; +} + +/** Prefix matching every run record for a workspace (for enumeration). */ +function runPrefix(workspaceId: string): string { + return `run:${workspaceId}:`; +} + +/** Extract the runId from a full `run:<workspaceId>:<runId>` key. */ +function parseRunId(key: string, workspaceId: string): string { + const prefix = runPrefix(workspaceId); + return key.startsWith(prefix) ? key.slice(prefix.length) : key; +} + +export interface HeartbeatRunStore { + /** Create a new run record (status `"running"`) and persist it. */ + readonly create: (workspaceId: string, run: HeartbeatRun) => Promise<HeartbeatRun>; + /** Update the status of an existing run. No-op if the run is unknown. */ + readonly setStatus: ( + workspaceId: string, + runId: string, + status: HeartbeatRunStatus, + ) => Promise<HeartbeatRun | null>; + /** A single run by id, or `null` when unknown. */ + readonly get: (workspaceId: string, runId: string) => Promise<HeartbeatRun | null>; + /** All runs for a workspace, most-recent first (by `triggeredAt`). */ + readonly list: (workspaceId: string) => Promise<readonly HeartbeatRun[]>; +} + +export function createHeartbeatRunStore(storage: StorageNamespace): HeartbeatRunStore { + async function readRun(workspaceId: string, runId: string): Promise<HeartbeatRun | null> { + const raw = await storage.get(runKey(workspaceId, runId)); + if (raw === null) return null; + try { + const parsed = JSON.parse(raw) as Partial<HeartbeatRun>; + if ( + typeof parsed.id !== "string" || + typeof parsed.conversationId !== "string" || + typeof parsed.triggeredAt !== "string" || + typeof parsed.status !== "string" + ) { + return null; + } + return { + id: parsed.id, + conversationId: parsed.conversationId, + triggeredAt: parsed.triggeredAt, + status: parsed.status as HeartbeatRunStatus, + }; + } catch { + return null; + } + } + + return { + async create(workspaceId: string, run: HeartbeatRun): Promise<HeartbeatRun> { + await storage.set(runKey(workspaceId, run.id), JSON.stringify(run)); + return run; + }, + + async setStatus( + workspaceId: string, + runId: string, + status: HeartbeatRunStatus, + ): Promise<HeartbeatRun | null> { + const run = await readRun(workspaceId, runId); + if (run === null) return null; + const updated: HeartbeatRun = { ...run, status }; + await storage.set(runKey(workspaceId, runId), JSON.stringify(updated)); + return updated; + }, + + async get(workspaceId: string, runId: string): Promise<HeartbeatRun | null> { + return readRun(workspaceId, runId); + }, + + async list(workspaceId: string): Promise<readonly HeartbeatRun[]> { + const keys = await storage.keys(runPrefix(workspaceId)); + const runs: HeartbeatRun[] = []; + for (const key of keys) { + const runId = parseRunId(key, workspaceId); + const run = await readRun(workspaceId, runId); + if (run !== null) runs.push(run); + } + // Most-recent first by triggeredAt (ISO-8601 sorts lexicographically). + runs.sort((a, b) => b.triggeredAt.localeCompare(a.triggeredAt)); + return runs; + }, + }; +} diff --git a/packages/heartbeat/src/scheduler.test.ts b/packages/heartbeat/src/scheduler.test.ts new file mode 100644 index 0000000..82e5c4d --- /dev/null +++ b/packages/heartbeat/src/scheduler.test.ts @@ -0,0 +1,343 @@ +import { describe, expect, it } from "vitest"; +import { HeartbeatScheduler, type Timers } from "./scheduler.js"; + +interface FakeTimer { + readonly fn: () => void; + readonly firesAt: number; +} + +/** + * A controllable fake clock: `advance(ms)` moves virtual time forward and runs + * any timers that became due. The injected `fire` returns a deferred the test + * resolves manually, so we can assert the "running" state and the re-arm that + * happens only AFTER the run completes. + */ +function createFakeTimers() { + let now = 0; + let nextId = 1; + const timers = new Map<number, FakeTimer>(); + const timersApi: Timers = { + now: () => now, + setTimeout: (fn, ms) => { + const id = nextId++; + timers.set(id, { fn, firesAt: now + ms }); + return id as unknown as ReturnType<typeof setTimeout>; + }, + clearTimeout: (handle) => { + if (handle !== undefined) timers.delete(handle as unknown as number); + }, + }; + return { + timers: timersApi, + advance(ms: number): number { + now += ms; + let fired = 0; + const due = [...timers.entries()] + .filter(([, t]) => t.firesAt <= now) + .sort((a, b) => a[0] - b[0]); + for (const [id, t] of due) { + timers.delete(id); + t.fn(); + fired++; + } + return fired; + }, + pendingCount(): number { + return timers.size; + }, + }; +} + +/** A deferred promise the test resolves to signal run completion. */ +function createDeferred(): { promise: Promise<void>; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise<void>((r) => { + resolve = r; + }); + return { promise, resolve }; +} + +// Flush pending microtasks (the scheduler's .finally re-arm runs as a microtask +// after the fire promise resolves). +const flush = async (): Promise<void> => { + await new Promise((r) => queueMicrotask(r)); +}; + +describe("HeartbeatScheduler", () => { + it("arms a timer that fires after intervalMinutes", () => { + const fake = createFakeTimers(); + const fires: string[] = []; + const deferreds: Array<{ resolve: () => void }> = []; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: (ws) => { + fires.push(ws); + const d = createDeferred(); + deferreds.push(d); + return d.promise; + }, + }); + + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + expect(scheduler.isArmed("ws-1")).toBe(true); + expect(fake.pendingCount()).toBe(1); + + // Just shy of the interval → no fire. + expect(fake.advance(59_999)).toBe(0); + expect(fires).toEqual([]); + + // Exactly the interval (1 minute = 60_000ms) → fires. + expect(fake.advance(1)).toBe(1); + expect(fires).toEqual(["ws-1"]); + // While the run is in progress: running, no pending timer. + expect(scheduler.isRunning("ws-1")).toBe(true); + expect(fake.pendingCount()).toBe(0); + }); + + it("re-arms only after the run completes (reset timer after each run)", async () => { + const fake = createFakeTimers(); + const fires: string[] = []; + const deferreds: Array<{ resolve: () => void }> = []; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: (ws) => { + fires.push(ws); + const d = createDeferred(); + deferreds.push(d); + return d.promise; + }, + }); + + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + fake.advance(60_000); // first fire + expect(fires).toEqual(["ws-1"]); + // Run in progress — no new timer yet. + expect(fake.pendingCount()).toBe(0); + + // Completing the run schedules the next fire. + deferreds[0]?.resolve(); + await flush(); + expect(scheduler.isRunning("ws-1")).toBe(false); + expect(fake.pendingCount()).toBe(1); + + // Next fire after another interval. + fake.advance(60_000); + expect(fires).toEqual(["ws-1", "ws-1"]); + }); + + it("uses a longer interval for a higher intervalMinutes", () => { + const fake = createFakeTimers(); + let fireCount = 0; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => { + fireCount++; + return createDeferred().promise; + }, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 30 }); + // 1 minute wouldn't fire a 30-minute schedule. + expect(fake.advance(60_000)).toBe(0); + // 28 more minutes (29 total) still wouldn't fire. + expect(fake.advance(28 * 60_000)).toBe(0); + // The remaining minute completes the 30 minutes → fires. + expect(fake.advance(60_000)).toBe(1); + expect(fireCount).toBe(1); + }); + + it("disarms (clears the pending timer) when disabled", () => { + const fake = createFakeTimers(); + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => createDeferred().promise, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + expect(fake.pendingCount()).toBe(1); + scheduler.arm("ws-1", { enabled: false, intervalMinutes: 1 }); + expect(scheduler.isArmed("ws-1")).toBe(false); + expect(fake.pendingCount()).toBe(0); + }); + + it("disarmAll stops every schedule", () => { + const fake = createFakeTimers(); + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => createDeferred().promise, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + scheduler.arm("ws-2", { enabled: true, intervalMinutes: 1 }); + expect(fake.pendingCount()).toBe(2); + scheduler.disarmAll(); + expect(fake.pendingCount()).toBe(0); + expect(scheduler.isArmed("ws-1")).toBe(false); + expect(scheduler.isArmed("ws-2")).toBe(false); + }); + + it("does not re-arm after a disarm during a run", async () => { + const fake = createFakeTimers(); + const deferreds: Array<{ resolve: () => void }> = []; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => { + const d = createDeferred(); + deferreds.push(d); + return d.promise; + }, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + fake.advance(60_000); // fire in progress + scheduler.disarm("ws-1"); // disarm mid-run + deferreds[0]?.resolve(); + await flush(); + // Disarmed → no re-arm scheduled. + expect(fake.pendingCount()).toBe(0); + expect(scheduler.isArmed("ws-1")).toBe(false); + }); + + it("a config update during a run applies the new interval on the next re-arm", async () => { + const fake = createFakeTimers(); + const deferreds: Array<{ resolve: () => void }> = []; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => { + const d = createDeferred(); + deferreds.push(d); + return d.promise; + }, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + fake.advance(60_000); // first fire (interval=1m) + // Update interval to 5m while the run is in progress. + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 5 }); + expect(scheduler.isRunning("ws-1")).toBe(true); + expect(fake.pendingCount()).toBe(0); // no new timer while running + deferreds[0]?.resolve(); + await flush(); + // Re-armed with the NEW 5-minute interval. + expect(fake.advance(60_000)).toBe(0); // 1 minute isn't enough now + expect(fake.advance(4 * 60_000)).toBe(1); // completes 5 minutes → fires + }); + + it("a thrown fire is swallowed and the loop continues", async () => { + const fake = createFakeTimers(); + let fireCount = 0; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => { + fireCount++; + return Promise.reject(new Error("boom")); + }, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + fake.advance(60_000); + await flush(); + expect(fireCount).toBe(1); + // Loop continues: next fire scheduled. + expect(fake.pendingCount()).toBe(1); + fake.advance(60_000); + expect(fireCount).toBe(2); + }); + + // ─── nextFireAt (CR-HB-3: server-authoritative next-run time) ────────────── + + it("nextFireAt returns null for a workspace with no schedule", () => { + const fake = createFakeTimers(); + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => createDeferred().promise, + }); + expect(scheduler.nextFireAt("unknown")).toBeNull(); + }); + + it("nextFireAt returns the absolute fire time when armed (now + intervalMs)", () => { + const fake = createFakeTimers(); + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => createDeferred().promise, + }); + // now=0, interval=1m → next fire at epoch-ms 60_000. + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + expect(scheduler.nextFireAt("ws-1")).toBe(60_000); + }); + + it("nextFireAt reflects a longer interval", () => { + const fake = createFakeTimers(); + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => createDeferred().promise, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 30 }); + expect(scheduler.nextFireAt("ws-1")).toBe(30 * 60_000); + }); + + it("nextFireAt reflects a new interval on re-arm (not running)", () => { + const fake = createFakeTimers(); + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => createDeferred().promise, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + expect(scheduler.nextFireAt("ws-1")).toBe(60_000); + // Re-arm with a 5-minute interval (armed, not running) → recomputed. + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 5 }); + expect(scheduler.nextFireAt("ws-1")).toBe(5 * 60_000); + }); + + it("nextFireAt returns null when disarmed", () => { + const fake = createFakeTimers(); + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => createDeferred().promise, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + expect(scheduler.nextFireAt("ws-1")).toBe(60_000); + scheduler.disarm("ws-1"); + expect(scheduler.nextFireAt("ws-1")).toBeNull(); + }); + + it("nextFireAt returns null while a run is in progress, then the next fire after it completes", async () => { + const fake = createFakeTimers(); + const deferreds: Array<{ resolve: () => void }> = []; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => { + const d = createDeferred(); + deferreds.push(d); + return d.promise; + }, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + expect(scheduler.nextFireAt("ws-1")).toBe(60_000); + + fake.advance(60_000); // fire → run in progress + expect(scheduler.isRunning("ws-1")).toBe(true); + // In flight → no next run queued yet. + expect(scheduler.nextFireAt("ws-1")).toBeNull(); + + deferreds[0]?.resolve(); + await flush(); + // Re-armed at completion-time (60_000) + interval (60_000) = 120_000. + expect(scheduler.nextFireAt("ws-1")).toBe(120_000); + }); + + it("nextFireAt returns null after disarming mid-run (no re-arm)", async () => { + const fake = createFakeTimers(); + const deferreds: Array<{ resolve: () => void }> = []; + const scheduler = new HeartbeatScheduler({ + timers: fake.timers, + fire: () => { + const d = createDeferred(); + deferreds.push(d); + return d.promise; + }, + }); + scheduler.arm("ws-1", { enabled: true, intervalMinutes: 1 }); + fake.advance(60_000); // fire in progress + expect(scheduler.nextFireAt("ws-1")).toBeNull(); // running + scheduler.disarm("ws-1"); // disarm mid-run + deferreds[0]?.resolve(); + await flush(); + // Disarmed → no re-arm, no fire time. + expect(scheduler.nextFireAt("ws-1")).toBeNull(); + }); +}); diff --git a/packages/heartbeat/src/scheduler.ts b/packages/heartbeat/src/scheduler.ts new file mode 100644 index 0000000..8b00b61 --- /dev/null +++ b/packages/heartbeat/src/scheduler.ts @@ -0,0 +1,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; + } +} diff --git a/packages/heartbeat/tsconfig.json b/packages/heartbeat/tsconfig.json new file mode 100644 index 0000000..b947b31 --- /dev/null +++ b/packages/heartbeat/tsconfig.json @@ -0,0 +1,12 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "rootDir": "src", "outDir": "dist", "composite": true }, + "include": ["src/**/*.ts"], + "references": [ + { "path": "../kernel" }, + { "path": "../transport-contract" }, + { "path": "../session-orchestrator" }, + { "path": "../system-prompt" }, + { "path": "../conversation-store" } + ] +} |