From fdc2d0df8fa5ffca8c1c0957cc8bbbe5f4a5304c Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Mon, 29 Jun 2026 01:51:03 +0900 Subject: feat(heartbeat): add inactiveOnly mode (skip fire while workspace has active agents) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A new per-workspace heartbeat setting `inactiveOnly` (default true): when on, the heartbeat SKIPS a scheduled fire whenever the configured workspace has any active agents — a conversation whose persisted status is "active" (driving a turn) or "queued" (waiting on the message queue). The fire is silently skipped (no run recorded); the scheduler re-arms and retries at the next interval. Set false to fire unconditionally (the prior behavior). The check is wired against conversationStore.listConversations({ workspaceId, status: ["active","queued"] }) at fire time — the orchestrator sets "active" on turn start and "idle" on settle, so it is the live busy/idle signal. The spawned heartbeat conversation lives in the dedicated heartbeat workspace, so a heartbeat run never counts as an active agent of the configured workspace (no self-block). - transport-contract: add inactiveOnly to HeartbeatConfig + UpdateHeartbeatRequest - heartbeat config-store: default true, apply/persist, legacy-parse default true - heartbeat service: injectable hasActiveAgents dep + skip logic in fire() - heartbeat extension: wire hasActiveAgents against the conversation store - transport-http: validate inactiveOnly (boolean) on PUT /workspaces/:id/heartbeat - tests: config-store (+4), heartbeat service (+7), transport-http (+3) - notes/heartbeat-setting-handoff.md: API contract for the frontend Verification: typecheck EXIT 0; tests 2013 passed | 6 skipped; biome EXIT 0. --- packages/heartbeat/src/config-store.test.ts | 37 ++++++++ packages/heartbeat/src/config-store.ts | 3 + packages/heartbeat/src/extension.ts | 13 +++ packages/heartbeat/src/heartbeat.test.ts | 127 ++++++++++++++++++++++++++++ packages/heartbeat/src/heartbeat.ts | 29 +++++++ 5 files changed, 209 insertions(+) (limited to 'packages/heartbeat') diff --git a/packages/heartbeat/src/config-store.test.ts b/packages/heartbeat/src/config-store.test.ts index e77d10a..9a6772d 100644 --- a/packages/heartbeat/src/config-store.test.ts +++ b/packages/heartbeat/src/config-store.test.ts @@ -77,6 +77,21 @@ describe("applyConfigUpdate (pure)", () => { 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", () => { @@ -106,6 +121,28 @@ describe("createHeartbeatConfigStore", () => { 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 }); diff --git a/packages/heartbeat/src/config-store.ts b/packages/heartbeat/src/config-store.ts index f06c9e8..52aca3e 100644 --- a/packages/heartbeat/src/config-store.ts +++ b/packages/heartbeat/src/config-store.ts @@ -7,6 +7,7 @@ import type { HeartbeatConfig, UpdateHeartbeatRequest } from "@dispatch/transpor */ export const DEFAULT_HEARTBEAT_CONFIG: HeartbeatConfig = { enabled: false, + inactiveOnly: true, systemPrompt: "", taskPrompt: "", intervalMinutes: 30, @@ -38,6 +39,7 @@ export function applyConfigUpdate( ): 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: @@ -72,6 +74,7 @@ export function createHeartbeatConfigStore(storage: StorageNamespace): Heartbeat const parsed = JSON.parse(raw) as Partial; 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: diff --git a/packages/heartbeat/src/extension.ts b/packages/heartbeat/src/extension.ts index f245b7e..b044619 100644 --- a/packages/heartbeat/src/extension.ts +++ b/packages/heartbeat/src/extension.ts @@ -90,6 +90,19 @@ export const extension: Extension = { // 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. diff --git a/packages/heartbeat/src/heartbeat.test.ts b/packages/heartbeat/src/heartbeat.test.ts index ff1ec10..a6f7ebb 100644 --- a/packages/heartbeat/src/heartbeat.test.ts +++ b/packages/heartbeat/src/heartbeat.test.ts @@ -157,6 +157,7 @@ function createService(opts: { ) => Promise; readonly getGlobalSystemPrompt?: () => Promise; readonly getWorkspaceCwd?: (workspaceId: string) => Promise; + readonly hasActiveAgents?: (workspaceId: string) => Promise; }) { const fake = createFakeTimers(); let id = 0; @@ -171,6 +172,7 @@ function createService(opts: { ? { getGlobalSystemPrompt: opts.getGlobalSystemPrompt } : {}), ...(opts.getWorkspaceCwd !== undefined ? { getWorkspaceCwd: opts.getWorkspaceCwd } : {}), + ...(opts.hasActiveAgents !== undefined ? { hasActiveAgents: opts.hasActiveAgents } : {}), }); return { svc, advance: fake.advance, storage }; } @@ -181,6 +183,131 @@ describe("createHeartbeatService", () => { 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(["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)", () => { diff --git a/packages/heartbeat/src/heartbeat.ts b/packages/heartbeat/src/heartbeat.ts index 12e70e3..e04c538 100644 --- a/packages/heartbeat/src/heartbeat.ts +++ b/packages/heartbeat/src/heartbeat.ts @@ -116,6 +116,19 @@ export interface HeartbeatServiceDeps { * (against `conversationStore.getWorkspace`); tests inject a fake. */ readonly getWorkspaceCwd?: (workspaceId: string) => Promise; + /** + * 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; } interface ActiveRun { @@ -143,6 +156,10 @@ export function createHeartbeatService(deps: HeartbeatServiceDeps): HeartbeatSer // 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 @@ -160,6 +177,18 @@ export function createHeartbeatService(deps: HeartbeatServiceDeps): HeartbeatSer // 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(); -- cgit v1.2.3