diff --git a/packages/coding-agent/src/core/cron-jobs.ts b/packages/coding-agent/src/core/cron-jobs.ts index 95bf322bac..457b97e0e9 100644 --- a/packages/coding-agent/src/core/cron-jobs.ts +++ b/packages/coding-agent/src/core/cron-jobs.ts @@ -33,6 +33,7 @@ export interface AgentCronJob { updatedAt: string; nextRunAt?: string; lastRunAt?: string; + lastSkippedAt?: string; lastError?: string; runCount: number; } @@ -58,6 +59,12 @@ export interface AgentCronSchedulerHooks { onError?: (job: AgentCronJob, error: unknown) => void; } +export interface HeartbeatCronSessionActivity { + isStreaming: boolean; + isBashRunning: boolean; + pendingMessageCount: number; +} + interface CronJobsFile { jobs?: unknown; } @@ -483,7 +490,12 @@ export class AgentCronJobStore { return job; } const nextRunAt = nextRunAtForSchedule(job.schedule, now); - updated = { ...job, nextRunAt: nextRunAt?.toISOString(), updatedAt: now.toISOString() }; + updated = { + ...job, + nextRunAt: nextRunAt?.toISOString(), + lastSkippedAt: now.toISOString(), + updatedAt: now.toISOString(), + }; return updated; }); if (updated) { @@ -760,7 +772,8 @@ export function formatAgentCronJob(job: AgentCronJob): string { const preview = job.prompt.replace(/\s+/g, " ").slice(0, 80); const error = job.lastError ? ` error=${job.lastError}` : ""; const label = job.label ? ` label="${job.label}"` : ""; - return `${job.id} ${job.status}${label} next=${next} last=${last} runs=${job.runCount} schedule="${job.schedule.expression}" prompt="${preview}"${error}`; + const skipped = job.lastSkippedAt ? ` skipped=${new Date(job.lastSkippedAt).toLocaleString()}` : ""; + return `${job.id} ${job.status}${label} next=${next} last=${last}${skipped} runs=${job.runCount} schedule="${job.schedule.expression}" prompt="${preview}"${error}`; } export function createAgentHeartbeatToolDefinitions(controller: AgentCronToolController): ToolDefinition[] { @@ -814,6 +827,16 @@ function consumeLeadingEverySchedule(text: string): { interval: string; rest: st }; } +export function isHeartbeatCronJob(job: AgentCronJob): boolean { + return job.source === "heartbeat" || job.source === "rlm_heartbeat"; +} + +export function shouldDeferHeartbeatCronJob(job: AgentCronJob, activity: HeartbeatCronSessionActivity): boolean { + return ( + isHeartbeatCronJob(job) && (activity.isStreaming || activity.isBashRunning || activity.pendingMessageCount > 0) + ); +} + function nextCronRunAfter(expression: string, after: Date): Date { const fields = parseCronExpression(expression); const candidate = new Date(after.getTime()); diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index 0461285d2a..8e985845ae 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -22,7 +22,9 @@ import { type AgentHeartbeatUpdateAction, createAgentHeartbeatToolDefinitions, DEFAULT_HEARTBEAT_SCHEDULE, + isHeartbeatCronJob, normalizeHeartbeatSchedule, + shouldDeferHeartbeatCronJob, } from "../../core/cron-jobs.js"; import type { CreateRlmSubagentRuntimeOptions, @@ -471,18 +473,19 @@ export class AgentDaemon { if (!state) { return; } - const followUpQueueKey = isHeartbeatCronJob(job) ? `heartbeat:${job.id}` : undefined; - if (followUpQueueKey && (state.runtime.session.isStreaming || state.runtime.session.pendingMessageCount > 0)) { - const didQueue = await state.runtime.session.followUp(job.prompt, undefined, { queueKey: followUpQueueKey }); - return didQueue ? undefined : "skipped"; + if (shouldDeferHeartbeatCronJob(job, state.runtime.session)) { + return "skipped"; } - if (!followUpQueueKey && (state.runtime.session.isStreaming || state.runtime.session.pendingMessageCount > 0)) { + if ( + !isHeartbeatCronJob(job) && + (state.runtime.session.isStreaming || state.runtime.session.pendingMessageCount > 0) + ) { await state.runtime.session.followUp(job.prompt); return; } await state.runtime.session.prompt(job.prompt, { - streamingBehavior: state.runtime.session.isStreaming ? "followUp" : undefined, - followUpQueueKey, + streamingBehavior: "followUp", + followUpQueueKey: isHeartbeatCronJob(job) ? `heartbeat:${job.id}` : undefined, source: "rpc", }); } @@ -1670,10 +1673,6 @@ export class AgentDaemon { } } -function isHeartbeatCronJob(job: AgentCronJob): boolean { - return job.source === "heartbeat" || job.source === "rlm_heartbeat"; -} - function serializeSavedSessionInfo(session: SessionInfo): DaemonSavedSessionInfo { return { path: session.path, diff --git a/packages/coding-agent/test/cron-jobs.test.ts b/packages/coding-agent/test/cron-jobs.test.ts index 0f355051e3..fcb2d1b431 100644 --- a/packages/coding-agent/test/cron-jobs.test.ts +++ b/packages/coding-agent/test/cron-jobs.test.ts @@ -9,6 +9,7 @@ import { createAgentHeartbeatToolDefinitions, parseAgentCronSchedule, parseHeartbeatCommand, + shouldDeferHeartbeatCronJob, } from "../src/core/cron-jobs.js"; const start = new Date("2026-01-01T12:34:00.000Z"); @@ -702,6 +703,7 @@ describe("AgentCronScheduler", () => { id: job.id, status: "active", nextRunAt: "2026-01-01T12:45:00.000Z", + lastSkippedAt: "2026-01-01T12:40:00.000Z", runCount: 0, }); expect(store.getHeartbeat("active-1")).not.toHaveProperty("lastRunAt"); @@ -752,6 +754,69 @@ describe("AgentCronScheduler", () => { }); }); +describe("shouldDeferHeartbeatCronJob", () => { + const baseJob: AgentCronJob = { + id: "job-1", + status: "active", + activeSessionId: "active-1", + sessionId: "session-1", + sessionFile: "/tmp/session.jsonl", + cwd: "/tmp/project", + prompt: "check progress", + schedule: { kind: "interval", expression: "every 5m", intervalMs: 300_000 }, + createdAt: "2026-01-01T12:34:00.000Z", + updatedAt: "2026-01-01T12:34:00.000Z", + nextRunAt: "2026-01-01T12:39:00.000Z", + runCount: 0, + }; + + it("defers user and RLM heartbeats while the target session is working", () => { + for (const source of ["heartbeat", "rlm_heartbeat"] as const) { + const job = { ...baseJob, source }; + + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: true, + isBashRunning: false, + pendingMessageCount: 0, + }), + ).toBe(true); + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: false, + isBashRunning: true, + pendingMessageCount: 0, + }), + ).toBe(true); + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: false, + isBashRunning: false, + pendingMessageCount: 1, + }), + ).toBe(true); + } + }); + + it("allows heartbeats when the target session is idle", () => { + expect( + shouldDeferHeartbeatCronJob( + { ...baseJob, source: "heartbeat" }, + { isStreaming: false, isBashRunning: false, pendingMessageCount: 0 }, + ), + ).toBe(false); + }); + + it("does not defer ordinary cron jobs", () => { + expect( + shouldDeferHeartbeatCronJob( + { ...baseJob, source: "cron" }, + { isStreaming: true, isBashRunning: true, pendingMessageCount: 2 }, + ), + ).toBe(false); + }); +}); + describe("createAgentHeartbeatToolDefinitions", () => { it("exposes only read-only user heartbeat inspection to the model", () => { const tools = createAgentHeartbeatToolDefinitions({ diff --git a/packages/coding-agent/test/daemon-mode.test.ts b/packages/coding-agent/test/daemon-mode.test.ts index 28514dbe4b..87ba96faa7 100644 --- a/packages/coding-agent/test/daemon-mode.test.ts +++ b/packages/coding-agent/test/daemon-mode.test.ts @@ -181,7 +181,7 @@ describe("daemon mode helpers", () => { } }); - it("queues busy heartbeat cron jobs with a per-job coalescing key", async () => { + it("defers busy heartbeat cron jobs instead of queueing a follow-up", async () => { const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", @@ -197,6 +197,7 @@ describe("daemon mode helpers", () => { runtime: ActiveSessionState["runtime"] & { session: { isStreaming: boolean; + isBashRunning: boolean; pendingMessageCount: number; prompt: typeof prompt; followUp: typeof followUp; @@ -205,6 +206,7 @@ describe("daemon mode helpers", () => { }; state.runtime.session = { isStreaming: true, + isBashRunning: false, pendingMessageCount: 0, prompt, followUp, @@ -216,17 +218,20 @@ describe("daemon mode helpers", () => { ).sessions.set(state.activeSessionId, state); const runCronJob = ( daemon as unknown as { - runCronJob(job: AgentCronJob): Promise; + runCronJob(job: AgentCronJob): Promise<"skipped" | undefined>; } ).runCronJob.bind(daemon); - await runCronJob(makeCronJob({ id: "heartbeat-1", source: "heartbeat", activeSessionId: state.activeSessionId })); + const result = await runCronJob( + makeCronJob({ id: "heartbeat-1", source: "heartbeat", activeSessionId: state.activeSessionId }), + ); - expect(followUp).toHaveBeenCalledWith("heartbeat prompt", undefined, { queueKey: "heartbeat:heartbeat-1" }); + expect(result).toBe("skipped"); + expect(followUp).not.toHaveBeenCalled(); expect(prompt).not.toHaveBeenCalled(); }); - it("uses separate queue keys for separate RLM heartbeat cron jobs", async () => { + it("defers separate RLM heartbeat cron jobs while the session is busy", async () => { const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", @@ -242,6 +247,7 @@ describe("daemon mode helpers", () => { runtime: ActiveSessionState["runtime"] & { session: { isStreaming: boolean; + isBashRunning: boolean; pendingMessageCount: number; prompt: typeof prompt; followUp: typeof followUp; @@ -250,6 +256,7 @@ describe("daemon mode helpers", () => { }; state.runtime.session = { isStreaming: true, + isBashRunning: false, pendingMessageCount: 0, prompt, followUp, @@ -261,19 +268,24 @@ describe("daemon mode helpers", () => { ).sessions.set(state.activeSessionId, state); const runCronJob = ( daemon as unknown as { - runCronJob(job: AgentCronJob): Promise; + runCronJob(job: AgentCronJob): Promise<"skipped" | undefined>; } ).runCronJob.bind(daemon); - await runCronJob(makeCronJob({ id: "rlm-1", source: "rlm_heartbeat", activeSessionId: state.activeSessionId })); - await runCronJob(makeCronJob({ id: "rlm-2", source: "rlm_heartbeat", activeSessionId: state.activeSessionId })); + const first = await runCronJob( + makeCronJob({ id: "rlm-1", source: "rlm_heartbeat", activeSessionId: state.activeSessionId }), + ); + const second = await runCronJob( + makeCronJob({ id: "rlm-2", source: "rlm_heartbeat", activeSessionId: state.activeSessionId }), + ); - expect(followUp).toHaveBeenNthCalledWith(1, "heartbeat prompt", undefined, { queueKey: "heartbeat:rlm-1" }); - expect(followUp).toHaveBeenNthCalledWith(2, "heartbeat prompt", undefined, { queueKey: "heartbeat:rlm-2" }); + expect(first).toBe("skipped"); + expect(second).toBe("skipped"); + expect(followUp).not.toHaveBeenCalled(); expect(prompt).not.toHaveBeenCalled(); }); - it("skips duplicate queued heartbeat cron jobs", async () => { + it("does not enqueue another heartbeat when one is already pending", async () => { const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, createRuntime: async () => { @@ -284,6 +296,7 @@ describe("daemon mode helpers", () => { runtime: ActiveSessionState["runtime"] & { session: { isStreaming: boolean; + isBashRunning: boolean; pendingMessageCount: number; prompt: ReturnType; followUp: ReturnType; @@ -294,6 +307,7 @@ describe("daemon mode helpers", () => { const removeQueuedFollowUp = vi.fn(() => true); state.runtime.session = { isStreaming: true, + isBashRunning: false, pendingMessageCount: 1, prompt: vi.fn(), followUp, @@ -305,7 +319,7 @@ describe("daemon mode helpers", () => { ).runCronJob(makeCronJob({ id: "heartbeat-1", source: "heartbeat", activeSessionId: state.activeSessionId })); expect(result).toBe("skipped"); - expect(followUp).toHaveBeenCalledWith("heartbeat prompt", undefined, { queueKey: "heartbeat:heartbeat-1" }); + expect(followUp).not.toHaveBeenCalled(); expect(removeQueuedFollowUp).not.toHaveBeenCalled(); }); @@ -339,6 +353,88 @@ describe("daemon mode helpers", () => { expect(prompt).not.toHaveBeenCalled(); }); + it("prompts idle sessions with a followUp streaming behavior to survive a mid-call stream start", async () => { + const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { + defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + const prompt = vi.fn(async () => {}); + const followUp = vi.fn(async () => true); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: { + isStreaming: boolean; + isBashRunning: boolean; + pendingMessageCount: number; + prompt: typeof prompt; + followUp: typeof followUp; + }; + }; + }; + state.runtime.session = { + isStreaming: false, + isBashRunning: false, + pendingMessageCount: 0, + prompt, + followUp, + } as never; + (daemon as unknown as { sessions: Map }).sessions.set(state.activeSessionId, state); + + await (daemon as unknown as { runCronJob(job: AgentCronJob): Promise<"skipped" | undefined> }).runCronJob( + makeCronJob({ id: "heartbeat-1", source: "heartbeat", activeSessionId: state.activeSessionId }), + ); + + expect(prompt).toHaveBeenCalledWith("heartbeat prompt", { + streamingBehavior: "followUp", + followUpQueueKey: "heartbeat:heartbeat-1", + source: "rpc", + }); + expect(followUp).not.toHaveBeenCalled(); + }); + + it("prompts idle generic cron jobs without a heartbeat coalescing key", async () => { + const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { + defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + const prompt = vi.fn(async () => {}); + const followUp = vi.fn(async () => true); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: { + isStreaming: boolean; + isBashRunning: boolean; + pendingMessageCount: number; + prompt: typeof prompt; + followUp: typeof followUp; + }; + }; + }; + state.runtime.session = { + isStreaming: false, + isBashRunning: false, + pendingMessageCount: 0, + prompt, + followUp, + } as never; + (daemon as unknown as { sessions: Map }).sessions.set(state.activeSessionId, state); + + await (daemon as unknown as { runCronJob(job: AgentCronJob): Promise<"skipped" | undefined> }).runCronJob( + makeCronJob({ id: "cron-1", source: "cron", activeSessionId: state.activeSessionId }), + ); + + expect(prompt).toHaveBeenCalledWith("heartbeat prompt", { + streamingBehavior: "followUp", + followUpQueueKey: undefined, + source: "rpc", + }); + expect(followUp).not.toHaveBeenCalled(); + }); + it("removes queued heartbeat follow-ups when a heartbeat is cleared", async () => { const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-heartbeat-clear-")); try {