Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions packages/coding-agent/src/core/cron-jobs.ts
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,12 @@ export interface AgentCronSchedulerHooks {
onError?: (job: AgentCronJob, error: unknown) => void;
}

export interface HeartbeatCronSessionActivity {
isStreaming: boolean;
isBashRunning: boolean;
pendingMessageCount: number;
}

interface CronJobsFile {
jobs?: unknown;
}
Expand Down Expand Up @@ -814,6 +820,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());
Expand Down
13 changes: 5 additions & 8 deletions packages/coding-agent/src/modes/daemon/daemon-mode.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,9 @@ import {
type AgentHeartbeatUpdateAction,
createAgentHeartbeatToolDefinitions,
DEFAULT_HEARTBEAT_SCHEDULE,
isHeartbeatCronJob,
normalizeHeartbeatSchedule,
shouldDeferHeartbeatCronJob,
} from "../../core/cron-jobs.js";
import type {
CreateRlmSubagentRuntimeOptions,
Expand Down Expand Up @@ -471,11 +473,10 @@ 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";
}
const followUpQueueKey = isHeartbeatCronJob(job) ? `heartbeat:${job.id}` : undefined;
if (!followUpQueueKey && (state.runtime.session.isStreaming || state.runtime.session.pendingMessageCount > 0)) {
await state.runtime.session.followUp(job.prompt);
return;
Expand Down Expand Up @@ -1670,10 +1671,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,
Expand Down
64 changes: 64 additions & 0 deletions packages/coding-agent/test/cron-jobs.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -752,6 +753,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({
Expand Down
38 changes: 26 additions & 12 deletions packages/coding-agent/test/daemon-mode.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -197,6 +197,7 @@ describe("daemon mode helpers", () => {
runtime: ActiveSessionState["runtime"] & {
session: {
isStreaming: boolean;
isBashRunning: boolean;
pendingMessageCount: number;
prompt: typeof prompt;
followUp: typeof followUp;
Expand All @@ -205,6 +206,7 @@ describe("daemon mode helpers", () => {
};
state.runtime.session = {
isStreaming: true,
isBashRunning: false,
pendingMessageCount: 0,
prompt,
followUp,
Expand All @@ -216,17 +218,20 @@ describe("daemon mode helpers", () => {
).sessions.set(state.activeSessionId, state);
const runCronJob = (
daemon as unknown as {
runCronJob(job: AgentCronJob): Promise<void>;
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",
Expand All @@ -242,6 +247,7 @@ describe("daemon mode helpers", () => {
runtime: ActiveSessionState["runtime"] & {
session: {
isStreaming: boolean;
isBashRunning: boolean;
pendingMessageCount: number;
prompt: typeof prompt;
followUp: typeof followUp;
Expand All @@ -250,6 +256,7 @@ describe("daemon mode helpers", () => {
};
state.runtime.session = {
isStreaming: true,
isBashRunning: false,
pendingMessageCount: 0,
prompt,
followUp,
Expand All @@ -261,19 +268,24 @@ describe("daemon mode helpers", () => {
).sessions.set(state.activeSessionId, state);
const runCronJob = (
daemon as unknown as {
runCronJob(job: AgentCronJob): Promise<void>;
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 () => {
Expand All @@ -284,6 +296,7 @@ describe("daemon mode helpers", () => {
runtime: ActiveSessionState["runtime"] & {
session: {
isStreaming: boolean;
isBashRunning: boolean;
pendingMessageCount: number;
prompt: ReturnType<typeof vi.fn>;
followUp: ReturnType<typeof vi.fn>;
Expand All @@ -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,
Expand All @@ -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();
});

Expand Down