diff --git a/src/agents/bash-process-registry.test.ts b/src/agents/bash-process-registry.test.ts index 8549c39ebc77..8d2670b066a3 100644 --- a/src/agents/bash-process-registry.test.ts +++ b/src/agents/bash-process-registry.test.ts @@ -7,6 +7,7 @@ import type { ChildProcessWithoutNullStreams } from "node:child_process"; import { beforeEach, describe, expect, it, vi } from "vitest"; import type { ProcessSession } from "./bash-process-registry.js"; import { + acknowledgeNotifyOnExit, addSession, appendOutput, createSessionSlug, @@ -18,6 +19,8 @@ import { listRunningSessions, markBackgrounded, markExited, + markTerminalPollObserved, + recordNotifyOnExitRemoval, setJobTtlMs, tail, } from "./bash-process-registry.js"; @@ -55,6 +58,26 @@ describe("bash process registry", () => { resetProcessRegistryForTests(); }); + it("suppresses a notify-on-exit event when terminal poll wins the race", () => { + const session = createRegistrySession({ + id: "poll-first", + maxOutputChars: 10_000, + pendingMaxOutputChars: 30_000, + backgrounded: true, + }); + addSession(session); + markTerminalPollObserved(session); + markExited(session, 0, null, "completed"); + + const remove = vi.fn(() => true); + recordNotifyOnExitRemoval(session, remove); + + expect(remove).toHaveBeenCalledOnce(); + expect(getFinishedSession("poll-first")?.terminalPollObserved).toBe(true); + acknowledgeNotifyOnExit(getFinishedSession("poll-first") ?? {}); + expect(remove).toHaveBeenCalledOnce(); + }); + it("captures output and truncates", () => { const session = createRegistrySession({ maxOutputChars: 10, diff --git a/src/agents/bash-process-registry.ts b/src/agents/bash-process-registry.ts index 8d6ac38ea33b..e6b0c626886d 100644 --- a/src/agents/bash-process-registry.ts +++ b/src/agents/bash-process-registry.ts @@ -42,6 +42,9 @@ type SessionStdin = { writableFinished?: boolean; }; +/** Removes one queued notify-on-exit event, if it is still pending. */ +type NotifyOnExitRemoval = () => boolean; + /** Mutable session state for a running bash exec process. */ export interface ProcessSession { id: string; @@ -65,6 +68,9 @@ export interface ProcessSession { notifyOnExit?: boolean; notifyOnExitEmptySuccess?: boolean; exitNotified?: boolean; + /** Set when process poll observed the terminal result before notification. */ + terminalPollObserved?: boolean; + notifyOnExitRemoval?: NotifyOnExitRemoval; child?: ChildProcessWithoutNullStreams; stdin?: SessionStdin; pid?: number; @@ -111,6 +117,8 @@ interface FinishedSession { tail: string; truncated: boolean; totalOutputChars: number; + terminalPollObserved?: boolean; + notifyOnExitRemoval?: NotifyOnExitRemoval; } const runningSessions = new Map(); @@ -256,6 +264,43 @@ export function markBackgrounded(session: ProcessSession) { } } +/** Records that a terminal process poll consumed the process result. */ +export function markTerminalPollObserved(session: ProcessSession): void { + session.terminalPollObserved = true; + const finished = finishedSessions.get(session.id); + if (finished) { + finished.terminalPollObserved = true; + } +} + +/** Retains the precise event removal handle across the finished-session move. */ +export function recordNotifyOnExitRemoval( + session: ProcessSession, + remove: NotifyOnExitRemoval, +): void { + if (session.terminalPollObserved) { + remove(); + return; + } + session.notifyOnExitRemoval = remove; + const finished = finishedSessions.get(session.id); + if (finished) { + finished.notifyOnExitRemoval = remove; + } +} + +/** Acknowledges one completion event without touching unrelated queue entries. */ +export function acknowledgeNotifyOnExit(record: { + notifyOnExitRemoval?: NotifyOnExitRemoval; +}): void { + const remove = record.notifyOnExitRemoval; + if (!remove) { + return; + } + remove(); + record.notifyOnExitRemoval = undefined; +} + /** Returns the number of live background exec sessions without exposing process details. */ export function getActiveBackgroundExecSessionCount(): number { return activeBackgroundExecSessionIds.size; @@ -319,6 +364,8 @@ function moveToFinished(session: ProcessSession, status: ProcessStatus) { tail: session.tail, truncated: session.truncated, totalOutputChars: session.totalOutputChars, + ...(session.terminalPollObserved ? { terminalPollObserved: true } : {}), + ...(session.notifyOnExitRemoval ? { notifyOnExitRemoval: session.notifyOnExitRemoval } : {}), }); finishedSessionOutputChars += session.aggregated.length; while ( diff --git a/src/agents/bash-tools.exec-runtime.test.ts b/src/agents/bash-tools.exec-runtime.test.ts index b3a59819bae3..27ab5847133f 100644 --- a/src/agents/bash-tools.exec-runtime.test.ts +++ b/src/agents/bash-tools.exec-runtime.test.ts @@ -21,6 +21,7 @@ import type { BashSandboxConfig } from "./bash-tools.shared.js"; const requestHeartbeatMock = vi.hoisted(() => vi.fn()); const enqueueSystemEventMock = vi.hoisted(() => vi.fn()); +const consumeSelectedSystemEventEntriesMock = vi.hoisted(() => vi.fn(() => [])); const supervisorMock = vi.hoisted(() => ({ spawn: vi.fn(), })); @@ -31,6 +32,16 @@ vi.mock("../infra/heartbeat-wake.js", () => ({ vi.mock("../infra/system-events.js", () => ({ enqueueSystemEvent: enqueueSystemEventMock, + enqueueSystemEventEntry: (text: string, options: { deliveryContext?: unknown }) => { + enqueueSystemEventMock(text, options); + return { + text, + ts: Date.now(), + contextKey: null, + deliveryContext: options.deliveryContext, + }; + }, + consumeSelectedSystemEventEntries: consumeSelectedSystemEventEntriesMock, })); vi.mock("../process/supervisor/index.js", () => ({ @@ -66,6 +77,7 @@ beforeEach(() => { resetProcessRegistryForTests(); requestHeartbeatMock.mockClear(); enqueueSystemEventMock.mockClear(); + consumeSelectedSystemEventEntriesMock.mockClear(); supervisorMock.spawn.mockReset(); }); diff --git a/src/agents/bash-tools.exec-runtime.ts b/src/agents/bash-tools.exec-runtime.ts index d721afc9452f..3815d374aa1e 100644 --- a/src/agents/bash-tools.exec-runtime.ts +++ b/src/agents/bash-tools.exec-runtime.ts @@ -16,7 +16,10 @@ import { } from "../infra/exec-approvals.js"; import { requestHeartbeat } from "../infra/heartbeat-wake.js"; import { findPathKey, mergePathPrepend, removePathPrepend } from "../infra/path-prepend.js"; -import { enqueueSystemEvent } from "../infra/system-events.js"; +import { + consumeSelectedSystemEventEntries, + enqueueSystemEventEntry, +} from "../infra/system-events.js"; import { isSubagentSessionKey } from "../sessions/session-key-utils.js"; /** * Bash exec runtime. @@ -43,6 +46,7 @@ import { appendOutput, createSessionSlug, markExited, + recordNotifyOnExitRemoval, tail, } from "./bash-process-registry.js"; import { appendExecTimeoutRetryGuidance, renderExecUpdateText } from "./bash-tools.exec-output.js"; @@ -311,7 +315,12 @@ export function applyShellPath(env: Record, shellPath?: string | } function maybeNotifyOnExit(session: ProcessSession, status: "completed" | "failed") { - if (!session.backgrounded || !session.notifyOnExit || session.exitNotified) { + if ( + !session.backgrounded || + !session.notifyOnExit || + session.exitNotified || + session.terminalPollObserved + ) { return; } const sessionKey = session.sessionKey?.trim(); @@ -339,10 +348,17 @@ function maybeNotifyOnExit(session: ProcessSession, status: "completed" | "faile mainKey: session.mainKey, sessionScope: session.sessionScope, }; - enqueueSystemEvent(eventText, { - sessionKey: resolveEventSessionKeyForPolicy(sessionKey, eventRouting), + const eventSessionKey = resolveEventSessionKeyForPolicy(sessionKey, eventRouting); + const event = enqueueSystemEventEntry(eventText, { + sessionKey: eventSessionKey, deliveryContext: session.notifyDeliveryContext, }); + if (event) { + recordNotifyOnExitRemoval( + session, + () => consumeSelectedSystemEventEntries(eventSessionKey, [event]).length > 0, + ); + } // Subagent sessions receive exec results via process poll and announce flow; // the heartbeat would fall back to the main session and cause spurious wakes. if (!isSubagentSessionKey(sessionKey)) { diff --git a/src/agents/bash-tools.process.e2e.test.ts b/src/agents/bash-tools.process.e2e.test.ts index aed444f2f4b0..5294045da42a 100644 --- a/src/agents/bash-tools.process.e2e.test.ts +++ b/src/agents/bash-tools.process.e2e.test.ts @@ -1,4 +1,5 @@ import { afterEach, expect, test } from "vitest"; +import { peekSystemEventEntries, resetSystemEventsForTest } from "../infra/system-events.js"; import { getSession } from "./bash-process-registry.js"; import { resetProcessRegistryForTests } from "./bash-process-registry.test-support.js"; import { createExecTool } from "./bash-tools.exec-run.js"; @@ -6,6 +7,7 @@ import { createProcessTool } from "./bash-tools.process.js"; afterEach(() => { resetProcessRegistryForTests(); + resetSystemEventsForTest(); }); function shellQuote(value: string): string { @@ -22,6 +24,51 @@ function textContent(result: { content: Array<{ type: string; text?: string }> } return result.content.find((part) => part.type === "text")?.text ?? ""; } +test.skipIf(process.platform === "win32")( + "consumes a real notify-on-exit event when process poll returns the terminal result", + async () => { + const scopeKey = "agent:main:process-notify-poll"; + const execTool = createExecTool({ + host: "gateway", + security: "full", + ask: "off", + allowBackground: true, + backgroundMs: 0, + timeoutSec: 10, + notifyOnExit: true, + notifyOnExitEmptySuccess: true, + sessionKey: scopeKey, + scopeKey, + }); + const processTool = createProcessTool({ scopeKey }); + const marker = "REAL_NOTIFY_ON_EXIT"; + const started = await execTool.execute("process-notify-start", { + command: currentNodeEvalCommand(`process.stdout.write(${JSON.stringify(marker)});`), + background: true, + }); + expect(started.details).toMatchObject({ status: "running" }); + const sessionId = (started.details as { sessionId?: string }).sessionId; + expect(sessionId).toEqual(expect.any(String)); + if (!sessionId) { + throw new Error("exec did not return a background session id"); + } + + await expect + .poll(() => peekSystemEventEntries(scopeKey).some((event) => event.text.includes(marker)), { + timeout: 5_000, + interval: 25, + }) + .toBe(true); + + const poll = await processTool.execute("process-notify-poll", { + action: "poll", + sessionId, + }); + expect(poll.details).toMatchObject({ status: "completed", sessionId }); + expect(peekSystemEventEntries(scopeKey)).toHaveLength(0); + }, +); + test.skipIf(process.platform === "win32")( "controls one real interactive background child through exec and process tools", async () => { diff --git a/src/agents/bash-tools.process.ts b/src/agents/bash-tools.process.ts index 9685b32cc70c..bc0ef23a1efd 100644 --- a/src/agents/bash-tools.process.ts +++ b/src/agents/bash-tools.process.ts @@ -9,6 +9,7 @@ import { getDiagnosticSessionState } from "../logging/diagnostic-session-state.j import { killProcessTree } from "../process/kill-tree.js"; import { getProcessSupervisor } from "../process/supervisor/index.js"; import { + acknowledgeNotifyOnExit, type ProcessSession, deleteSession, drainSession, @@ -16,6 +17,7 @@ import { getSession, listFinishedSessions, listRunningSessions, + markTerminalPollObserved, markExited, setJobTtlMs, } from "./bash-process-registry.js"; @@ -391,6 +393,7 @@ export function createProcessTool( if (!scopedSession) { if (scopedFinished) { resetPollRetrySuggestion(params.sessionId); + acknowledgeNotifyOnExit(scopedFinished); // Finished polls render a bounded tail; disclose retained content so the // model can recover it through paged logs instead of treating it as complete. const retainedOutputNote = @@ -457,6 +460,8 @@ export function createProcessTool( const exitCode = scopedSession.exitCode ?? 0; const exitSignal = scopedSession.exitSignal ?? undefined; if (exited) { + markTerminalPollObserved(scopedSession); + acknowledgeNotifyOnExit(scopedSession); const status = exitCode === 0 && exitSignal == null ? "completed" : "failed"; markExited( scopedSession, diff --git a/src/agents/bash-tools.test.ts b/src/agents/bash-tools.test.ts index 9dd5623633c7..2fafc3a55ac6 100644 --- a/src/agents/bash-tools.test.ts +++ b/src/agents/bash-tools.test.ts @@ -406,17 +406,6 @@ function useCapturedEnv(keys: string[], afterCapture?: () => void) { }); } -async function waitForCompletion(sessionId: string) { - let status = PROCESS_STATUS_RUNNING; - await expect - .poll(async () => { - status = (await pollProcessSession({ tool: processTool, sessionId })).status; - return status; - }, BACKGROUND_POLL_OPTIONS) - .not.toBe(PROCESS_STATUS_RUNNING); - return status; -} - function requireSessionId(details: { sessionId?: string }): string { if (!details.sessionId) { throw new Error("expected sessionId in exec result details"); @@ -478,12 +467,6 @@ async function drainNotifyEvents(sessionKey = DEFAULT_NOTIFY_SESSION_KEY) { }); } -async function runBackgroundCommandToCompletion(tool: ExecToolInstance, command: string) { - const sessionId = await startBackgroundCommand(tool, command); - const status = await waitForCompletion(sessionId); - return { sessionId, status }; -} - type ProcessLogWindow = { offset?: number; limit?: number }; async function readProcessLog(sessionId: string, options: ProcessLogWindow = {}) { return executeProcessTool(processTool, { @@ -695,8 +678,10 @@ const runLongLogExpectationCase = async ({ const runNotifyNoopCase = async ({ label, defaults, expectNotification }: NotifyNoopCase) => { const tool = createNotifyOnExitExecTool(defaults); - const { sessionId, status } = await runBackgroundCommandToCompletion(tool, COMMAND_NOOP); - expect(status).toBe(PROCESS_STATUS_COMPLETED); + const sessionId = await startBackgroundCommand(tool, COMMAND_NOOP); + await expect + .poll(() => getFinishedSession(sessionId)?.status, BACKGROUND_POLL_OPTIONS) + .toBe(PROCESS_STATUS_COMPLETED); const events = peekSystemEvents(DEFAULT_NOTIFY_SESSION_KEY); expectNotifyNoopEvents(events, expectNotification, sessionId, label); }; @@ -857,6 +842,19 @@ describe("exec notifyOnExit", () => { expect(formatted).toBeUndefined(); }); + it("consumes only the polled completion event", async () => { + const tool = createNotifyOnExitExecTool(); + const unpolledSessionId = await startBackgroundCommand(tool, shellEcho("unpolled")); + await waitForNotifyEvent(unpolledSessionId); + const sessionId = await startBackgroundCommand(tool, shellEcho("polled")); + await waitForNotifyEvent(sessionId); + const poll = await pollProcessSession({ tool: processTool, sessionId }); + + expect(poll.status).toBe(PROCESS_STATUS_COMPLETED); + expect(hasNotifyEventForPrefix(sessionId.slice(0, 8))).toBe(false); + expect(hasNotifyEventForPrefix(unpolledSessionId.slice(0, 8))).toBe(true); + }); + it("preserves the origin delivery context on background exec completion events", async () => { const sessionKey = "agent:main:telegram:group:-1003774691294:topic:47"; const tool = createNotifyOnExitExecTool({ diff --git a/src/infra/heartbeat-runner-execution.ts b/src/infra/heartbeat-runner-execution.ts index f0cc37b6707e..bf3cc1c78a1d 100644 --- a/src/infra/heartbeat-runner-execution.ts +++ b/src/infra/heartbeat-runner-execution.ts @@ -71,6 +71,7 @@ import { resolveHeartbeatPreflight, resolveHeartbeatRunPrompt, selectSystemEventsConsumedByHeartbeat, + shouldPreflightExecEventWake, } from "./heartbeat-runner-prompt.js"; import { resolveHeartbeatSession, @@ -144,6 +145,8 @@ export type HeartbeatRunOptions = { intent?: HeartbeatWakeIntent; reason?: string; runScope?: HeartbeatRunScope; + /** Persisted monitor cadence carried by a coalesced scheduled wake. */ + scheduledEveryMs?: number; tasks?: readonly HeartbeatScheduledTask[]; /** Exact cron run marker whose own activity must not block this wake. */ owningCronJobMarker?: CronActiveJobMarker; @@ -194,6 +197,33 @@ export async function resolveHeartbeatWakeStage(opts: HeartbeatRunOptions) { return { kind: "skipped", reason: "quiet-hours" } as const; } + const shouldInspectExecWakeBeforeBusy = shouldPreflightExecEventWake( + wakeSource, + opts.scheduledEveryMs, + runScope, + scheduledTasks.length, + ); + const resolvePreflight = () => + resolveHeartbeatPreflight({ + ...opts, + cfg, + agentId, + heartbeat, + runScope, + source: wakeSource, + scheduledTasks, + nowMs: startedAt, + }); + let preflight = shouldInspectExecWakeBeforeBusy ? await resolvePreflight() : undefined; + if (preflight?.skipReason) { + emitHeartbeatEvent({ + status: "skipped", + reason: preflight.skipReason, + durationMs: Date.now() - startedAt, + }); + return { kind: "skipped", reason: preflight.skipReason } as const; + } + const getSize = opts.deps?.getQueueSize ?? getQueueSize; if (getSize(CommandLane.Main) > 0) { return { kind: "skipped", reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT } as const; @@ -313,17 +343,9 @@ export async function resolveHeartbeatWakeStage(opts: HeartbeatRunOptions) { } // Preflight centralizes trigger classification, event inspection, and monitor-scratch gating. - const preflight = await resolveHeartbeatPreflight({ - cfg, - agentId, - heartbeat, - runScope, - forcedSessionKey: opts.sessionKey, - source: wakeSource, - reason: opts.reason, - scheduledTasks, - nowMs: startedAt, - }); + if (!preflight) { + preflight = await resolvePreflight(); + } if (preflight.skipReason) { emitHeartbeatEvent({ status: "skipped", diff --git a/src/infra/heartbeat-runner-prompt.ts b/src/infra/heartbeat-runner-prompt.ts index 1d4710c5fdb3..2e65862986d1 100644 --- a/src/infra/heartbeat-runner-prompt.ts +++ b/src/infra/heartbeat-runner-prompt.ts @@ -30,7 +30,11 @@ import { resolveHeartbeatWakePayloadFlags, type HeartbeatWakePayloadFlags, } from "./heartbeat-wake-policy.js"; -import type { HeartbeatScheduledTask, HeartbeatWakeSource } from "./heartbeat-wake.js"; +import { + HEARTBEAT_SKIP_NO_PENDING_EVENT, + type HeartbeatScheduledTask, + type HeartbeatWakeSource, +} from "./heartbeat-wake.js"; import { peekSystemEventEntries, resolveSystemEventDeliveryContext, @@ -43,7 +47,7 @@ export function truncateHeartbeatPreview(value: string | undefined): string | un return value ? truncateUtf16Safe(value, 200) : undefined; } -type HeartbeatSkipReason = "empty-heartbeat-file"; +type HeartbeatSkipReason = "empty-heartbeat-file" | typeof HEARTBEAT_SKIP_NO_PENDING_EVENT; function buildCommitmentDeliveryKey(commitment: CommitmentRecord): string { return [ @@ -110,20 +114,40 @@ type HeartbeatPreflight = HeartbeatWakePayloadFlags & { dueCommitments: CommitmentRecord[]; hasTaggedCronEvents: boolean; shouldInspectPendingEvents: boolean; + authoritativeScheduledTick: boolean; skipReason?: HeartbeatSkipReason; scratchJobId?: string; scratchRevision?: number; heartbeatScratchContent?: string; }; +export function shouldPreflightExecEventWake( + source: HeartbeatWakeSource | undefined, + scheduledEveryMs: number | undefined, + runScope: HeartbeatRunScope, + scheduledTaskCount: number, +): boolean { + return ( + source === "exec-event" && + !( + typeof scheduledEveryMs === "number" && + Number.isSafeInteger(scheduledEveryMs) && + scheduledEveryMs > 0 + ) && + runScope !== "commitment-only" && + scheduledTaskCount === 0 + ); +} + export async function resolveHeartbeatPreflight(params: { cfg: OpenClawConfig; agentId: string; heartbeat?: HeartbeatConfig; runScope: HeartbeatRunScope; - forcedSessionKey?: string; + sessionKey?: string; reason?: string; source?: HeartbeatWakeSource; + scheduledEveryMs?: number; scheduledTasks?: readonly HeartbeatScheduledTask[]; nowMs?: number; }): Promise { @@ -135,7 +159,7 @@ export async function resolveHeartbeatPreflight(params: { params.cfg, params.agentId, params.heartbeat, - params.forcedSessionKey, + params.sessionKey, ); const pendingEventEntries = params.runScope === "commitment-only" ? [] : peekSystemEventEntries(session.sessionKey); @@ -200,6 +224,10 @@ export async function resolveHeartbeatPreflight(params: { dueCommitments, hasTaggedCronEvents, shouldInspectPendingEvents, + authoritativeScheduledTick: + typeof params.scheduledEveryMs === "number" && + Number.isSafeInteger(params.scheduledEveryMs) && + params.scheduledEveryMs > 0, ...(monitorScratch?.jobId ? { scratchJobId: monitorScratch.jobId, @@ -214,6 +242,20 @@ export async function resolveHeartbeatPreflight(params: { : {}), } satisfies Omit; + // The exec completion can be acknowledged by process poll after its wake is + // queued. Treat that stale wake as consumed without touching unrelated events. + if ( + wakeFlags.isExecEventWake && + !basePreflight.authoritativeScheduledTick && + !params.scheduledTasks?.length && + !hasTaggedCronEvents && + !pendingEventEntries.some((event) => isExecCompletionEvent(event.text)) + ) { + return { + ...basePreflight, + skipReason: HEARTBEAT_SKIP_NO_PENDING_EVENT, + }; + } if (shouldBypassFileGates) { return basePreflight; } @@ -385,5 +427,8 @@ export function selectSystemEventsConsumedByHeartbeat(params: { isCronSystemEvent(event.text), ); } + if (preflight.isExecEventWake && !params.hasExecCompletion) { + return []; + } return preflight.pendingEventEntries; } diff --git a/src/infra/heartbeat-runner-scheduler.ts b/src/infra/heartbeat-runner-scheduler.ts index 5cff2bc9cabe..abf9ad466233 100644 --- a/src/infra/heartbeat-runner-scheduler.ts +++ b/src/infra/heartbeat-runner-scheduler.ts @@ -32,6 +32,7 @@ import { } from "./heartbeat-wake-policy.js"; import { areHeartbeatsEnabled, + HEARTBEAT_SKIP_NO_PENDING_EVENT, type HeartbeatRunResult, type HeartbeatWakeHandler, type HeartbeatWakeIntent, @@ -129,15 +130,45 @@ export function startHeartbeatRunner(opts: { agent.nextDueMs = seekActiveSlotForAgent(agent, rawDueMs); }; + const applyScheduledCadence = ( + agent: HeartbeatAgentState, + intervalMs: number | undefined, + anchorMs: number | undefined, + ) => { + if (intervalMs === undefined) { + return; + } + agent.intervalMs = intervalMs; + agent.phaseMs = + anchorMs ?? + resolveHeartbeatPhaseMs({ + schedulerSeed: state.schedulerSeed, + agentId: agent.agentId, + intervalMs, + }); + agent.heartbeat = { + ...agent.heartbeat, + every: `${intervalMs}ms`, + }; + }; + const advanceStaleScheduleAfterDeferral = ( agent: HeartbeatAgentState, now: number, reason?: string, decision?: DeferDecision, + options: { authoritativeScheduledTick?: boolean; execEventWake?: boolean } = {}, ) => { - if (!decision?.defer || decision.reason === "not-due" || agent.nextDueMs > now) { + if ( + !decision?.defer || + decision.reason === "not-due" || + agent.nextDueMs > now || + (options.execEventWake && !options.authoritativeScheduledTick) + ) { return; } + // A stale exec wake can be retained by the wake layer after a guard + // deferral, but it never owns cadence unless a scheduled tick joined it. // Deferrals that do not have wake-layer retry ownership still move the due // slot forward so repeated event wakes cannot retry a stale interval. advanceAgentSchedule(agent, now, reason); @@ -271,6 +302,7 @@ export function startHeartbeatRunner(opts: { const reason = params.reason; const intent = params.intent; + const execEventWake = params.source === "exec-event"; const requestedAgentId = params.agentId ? normalizeAgentId(params.agentId) : undefined; const requestedSessionKey = normalizeOptionalString(params.sessionKey); const requestedHeartbeat = params.heartbeat; @@ -280,6 +312,7 @@ export function startHeartbeatRunner(opts: { params.scheduledEveryMs > 0 ? params.scheduledEveryMs : undefined; + const authoritativeScheduledTick = scheduledEveryMs !== undefined; const scheduledAnchorMs = typeof params.scheduledAnchorMs === "number" && Number.isSafeInteger(params.scheduledAnchorMs) && @@ -328,14 +361,17 @@ export function startHeartbeatRunner(opts: { }; const runOneAgent = async ( agent: HeartbeatAgentState, - authoritativeScheduledTick = false, + scheduledTickIsAuthoritative = false, ): Promise => { const deferral = evaluateWakeDeferral(agent, now, reason, intent, { - authoritativeScheduledTick, + authoritativeScheduledTick: scheduledTickIsAuthoritative, retainedWork, }); if (deferral.defer) { - advanceStaleScheduleAfterDeferral(agent, now, reason, deferral); + advanceStaleScheduleAfterDeferral(agent, now, reason, deferral, { + authoritativeScheduledTick: scheduledTickIsAuthoritative, + execEventWake, + }); return { ran: false, result: { @@ -355,6 +391,7 @@ export function startHeartbeatRunner(opts: { source: params.source, intent, reason, + ...(scheduledEveryMs !== undefined ? { scheduledEveryMs } : {}), runScope: "global", tasks: requestedTasks, deps: { runtime: state.runtime }, @@ -377,6 +414,15 @@ export function startHeartbeatRunner(opts: { // agent — its target runtime is busy and the wake layer retries. return { ran: false, retryableBusySkip: res }; } + if ( + params.source === "exec-event" && + res.status === "skipped" && + res.reason === HEARTBEAT_SKIP_NO_PENDING_EVENT + ) { + // Poll already acknowledged the exec completion. This wake owns no + // cadence or commitment work, so it must remain a true no-op. + return { ran: false, result: res }; + } // Non-retryable outcome — record bookkeeping for cooldown gates. recordRunBookkeeping(agent, now); advanceAgentSchedule(agent, now, reason); @@ -430,21 +476,8 @@ export function startHeartbeatRunner(opts: { const targetAgent = state.agents.get(targetAgentId); // Task intent wins scheduled-task coalescing, so the cadence payload—not // the final intent—proves that the persisted monitor tick joined this turn. - const authoritativeScheduledTick = - params.source === "interval" && scheduledEveryMs !== undefined; - if (targetAgent && scheduledEveryMs !== undefined && authoritativeScheduledTick) { - targetAgent.intervalMs = scheduledEveryMs; - targetAgent.phaseMs = - scheduledAnchorMs ?? - resolveHeartbeatPhaseMs({ - schedulerSeed: state.schedulerSeed, - agentId: targetAgent.agentId, - intervalMs: scheduledEveryMs, - }); - targetAgent.heartbeat = { - ...targetAgent.heartbeat, - every: `${scheduledEveryMs}ms`, - }; + if (targetAgent && authoritativeScheduledTick) { + applyScheduledCadence(targetAgent, scheduledEveryMs, scheduledAnchorMs); } // A user-present targeted event may wake an unscheduled agent once. It // must not enroll that agent in the recurring heartbeat scheduler. @@ -480,7 +513,10 @@ export function startHeartbeatRunner(opts: { retainedWork, }); if (deferral.defer) { - advanceStaleScheduleAfterDeferral(targetAgent, now, reason, deferral); + advanceStaleScheduleAfterDeferral(targetAgent, now, reason, deferral, { + authoritativeScheduledTick, + execEventWake, + }); return { status: "skipped", reason: deferral.reason, @@ -503,6 +539,7 @@ export function startHeartbeatRunner(opts: { source: params.source, intent, reason, + ...(scheduledEveryMs !== undefined ? { scheduledEveryMs } : {}), runScope: "global", sessionKey: requestedSessionKey, tasks: requestedTasks, @@ -515,6 +552,13 @@ export function startHeartbeatRunner(opts: { // because the cooldown would treat this skipped attempt as a real run. return res; } + if ( + params.source === "exec-event" && + res.status === "skipped" && + res.reason === HEARTBEAT_SKIP_NO_PENDING_EVENT + ) { + return res; + } // Non-retryable outcome (ran, disabled, failed-but-not-busy). Record // bookkeeping and move the due slot so scheduleNext() cannot hot-loop // on a stale past-due agent. @@ -539,8 +583,15 @@ export function startHeartbeatRunner(opts: { } } + if (authoritativeScheduledTick) { + for (const agent of state.agents.values()) { + applyScheduledCadence(agent, scheduledEveryMs, scheduledAnchorMs); + } + } const agentOutcomes = await Promise.all( - Array.from(state.agents.values()).map((agent) => runOneAgent(agent)), + Array.from(state.agents.values()).map((agent) => + runOneAgent(agent, authoritativeScheduledTick), + ), ); let firstRetryableBusy: HeartbeatRunResult | undefined; for (const outcome of agentOutcomes) { diff --git a/src/infra/heartbeat-runner.ghost-reminder.test.ts b/src/infra/heartbeat-runner.ghost-reminder.test.ts index 7d4e81866cd4..aeeac6e67e3e 100644 --- a/src/infra/heartbeat-runner.ghost-reminder.test.ts +++ b/src/infra/heartbeat-runner.ghost-reminder.test.ts @@ -13,7 +13,13 @@ import { withTempHeartbeatSandbox, } from "./heartbeat-runner.test-utils.js"; import { HEARTBEAT_SKIP_CRON_IN_PROGRESS } from "./heartbeat-wake.js"; -import { enqueueSystemEvent, peekSystemEvents, resetSystemEventsForTest } from "./system-events.js"; +import { + consumeSelectedSystemEventEntries, + enqueueSystemEvent, + enqueueSystemEventEntry, + peekSystemEvents, + resetSystemEventsForTest, +} from "./system-events.js"; beforeEach(() => { setupTelegramHeartbeatPluginRuntimeForTests(); @@ -757,6 +763,31 @@ describe("Ghost reminder bug (issue #13317)", () => { expect(peekSystemEvents(sessionKey)).toEqual(["Node connected"]); }); + it("ignores an acknowledged exec-event wake without consuming unrelated events", async () => { + const { result, sendTelegram, calledCtx, replyCallCount, sessionKey } = await runHeartbeatCase({ + tmpPrefix: "openclaw-exec-acknowledged-", + replyText: "Unexpected heartbeat", + reason: "exec-event", + enqueue: (key) => { + const completion = enqueueSystemEventEntry( + "Exec completed (abc12345, code 0) :: deploy succeeded", + { sessionKey: key }, + ); + if (!completion) { + throw new Error("expected exec completion event"); + } + expect(consumeSelectedSystemEventEntries(key, [completion])).toHaveLength(1); + enqueueSystemEvent("Node connected", { sessionKey: key }); + }, + }); + + expect(result).toEqual({ status: "skipped", reason: "no-pending-event" }); + expect(replyCallCount).toBe(0); + expect(calledCtx).toBeNull(); + expect(sendTelegram).not.toHaveBeenCalled(); + expect(peekSystemEvents(sessionKey)).toEqual(["Node connected"]); + }); + it("classifies hook:wake exec completions as exec-event prompts", async () => { const { result, sendTelegram, calledCtx } = await runHeartbeatCase({ tmpPrefix: "openclaw-hook-exec-", diff --git a/src/infra/heartbeat-runner.stale-exec.test.ts b/src/infra/heartbeat-runner.stale-exec.test.ts new file mode 100644 index 000000000000..b39aac7febae --- /dev/null +++ b/src/infra/heartbeat-runner.stale-exec.test.ts @@ -0,0 +1,537 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { seedCommitmentsForTest } from "../commitments/store.test-utils.js"; +import type { CommitmentRecord } from "../commitments/types.js"; +import { resetConfigRuntimeState } from "../config/config.js"; +import type { OpenClawConfig } from "../config/config.js"; +import { resetGatewayWorkAdmission } from "../process/gateway-work-admission.js"; +import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; +import { captureEnv, setTestEnvValue } from "../test-utils/env.js"; +import { resetHeartbeatEventsForTest } from "./heartbeat-events.js"; +import { + runHeartbeatOnce, + setHeartbeatsEnabled, + startHeartbeatRunner, +} from "./heartbeat-runner.js"; +import { + seedMainSessionStore, + setupTelegramHeartbeatPluginRuntimeForTests, + withTempHeartbeatSandbox, +} from "./heartbeat-runner.test-utils.js"; +import { resolveHeartbeatPhaseMs } from "./heartbeat-schedule.js"; +import { + HEARTBEAT_SKIP_NO_PENDING_EVENT, + requestHeartbeat, + setHeartbeatWakeHandler as setRuntimeHeartbeatWakeHandler, +} from "./heartbeat-wake.js"; +import { enqueueSystemEvent, peekSystemEvents, resetSystemEventsForTest } from "./system-events.js"; + +describe("stale exec heartbeat wakes", () => { + type WakeRequest = Parameters[0]; + type WakeHandler = Parameters[0]; + const schedulerSeed = "stale-exec-heartbeat-test"; + const envSnapshot = captureEnv(["OPENCLAW_STATE_DIR"]); + let currentHandlerDisposer: (() => void) | undefined; + + function setHeartbeatWakeHandler(handler: WakeHandler): void { + currentHandlerDisposer?.(); + currentHandlerDisposer = setRuntimeHeartbeatWakeHandler(handler); + } + + function heartbeatConfig(every = "30m"): OpenClawConfig { + return { + agents: { + defaults: { heartbeat: { every } }, + }, + } as OpenClawConfig; + } + + function buildDueCommitment(nowMs: number): CommitmentRecord { + return { + id: "cm_interview", + agentId: "main", + sessionKey: "agent:main:telegram:user-155462274", + channel: "telegram", + accountId: "primary", + to: "1", + kind: "event_check_in", + sensitivity: "routine", + source: "inferred_user_context", + status: "pending", + reason: "The user said they had an interview yesterday.", + suggestedText: "How did the interview go?", + dedupeKey: "interview:2026-04-28", + confidence: 0.92, + dueWindow: { + earliestMs: nowMs - 60_000, + latestMs: nowMs + 60 * 60_000, + timezone: "America/Los_Angeles", + }, + createdAtMs: nowMs - 24 * 60 * 60_000, + updatedAtMs: nowMs - 24 * 60 * 60_000, + attempts: 0, + }; + } + + beforeEach(() => { + setupTelegramHeartbeatPluginRuntimeForTests(); + resetSystemEventsForTest(); + resetGatewayWorkAdmission(); + }); + + afterEach(async () => { + currentHandlerDisposer?.(); + if (vi.isFakeTimers()) { + currentHandlerDisposer = setRuntimeHeartbeatWakeHandler(async () => ({ + status: "skipped", + reason: "disabled", + })); + await vi.runAllTimersAsync(); + } + currentHandlerDisposer?.(); + currentHandlerDisposer = undefined; + closeOpenClawStateDatabaseForTest(); + resetConfigRuntimeState(); + resetGatewayWorkAdmission(); + resetHeartbeatEventsForTest(); + resetSystemEventsForTest(); + setHeartbeatsEnabled(true); + envSnapshot.restore(); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("retires a stale exec event without retrying or dropping coalesced task work", async () => { + vi.useFakeTimers(); + vi.setSystemTime(2_000_000_000_000); + const handler = vi.fn(async (request: WakeRequest) => + request.intent === "event" + ? ({ status: "skipped", reason: HEARTBEAT_SKIP_NO_PENDING_EVENT } as const) + : ({ status: "ran", durationMs: 1 } as const), + ); + setHeartbeatWakeHandler(handler); + + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + agentId: "main", + coalesceMs: 0, + }); + requestHeartbeat({ + source: "interval", + intent: "task", + reason: "heartbeat-task:job-inbox", + agentId: "main", + tasks: [{ jobId: "job-inbox", name: "inbox", prompt: "Check inbox" }], + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + + expect(handler.mock.calls.map(([request]) => request.intent)).toEqual(["task", "event"]); + expect(handler.mock.calls[0]?.[0]).toMatchObject({ + intent: "task", + tasks: [{ jobId: "job-inbox", name: "inbox", prompt: "Check inbox" }], + }); + + await vi.advanceTimersByTimeAsync(60_000); + expect(handler).toHaveBeenCalledTimes(2); + }); + + it("preserves scheduled cadence when an exec wake joins the scheduled turn", async () => { + vi.useFakeTimers(); + const handler = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); + setHeartbeatWakeHandler(handler); + + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + agentId: "main", + scheduledEveryMs: 5 * 60_000, + scheduledAnchorMs: 42_000, + coalesceMs: 100, + }); + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + agentId: "main", + coalesceMs: 100, + }); + + await vi.advanceTimersByTimeAsync(100); + + expect(handler).toHaveBeenCalledOnce(); + expect(handler).toHaveBeenCalledWith({ + source: "exec-event", + intent: "event", + reason: "exec-event", + agentId: "main", + scheduledEveryMs: 5 * 60_000, + scheduledAnchorMs: 42_000, + }); + }); + + it("passes persisted cadence through a coalesced exec wake", async () => { + vi.useFakeTimers(); + const runSpy = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); + const runner = startHeartbeatRunner({ + cfg: heartbeatConfig(), + runOnce: runSpy, + stableSchedulerSeed: schedulerSeed, + }); + + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + agentId: "main", + scheduledEveryMs: 5 * 60_000, + coalesceMs: 100, + }); + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + agentId: "main", + coalesceMs: 100, + }); + await vi.advanceTimersByTimeAsync(100); + + expect(runSpy).toHaveBeenCalledWith( + expect.objectContaining({ + source: "exec-event", + scheduledEveryMs: 5 * 60_000, + }), + ); + runner.stop(); + }); + + it("keeps a scheduled turn alive when an acknowledged exec wake coalesces with it", async () => { + await withTempHeartbeatSandbox(async ({ tmpDir, storePath }) => { + setTestEnvValue("OPENCLAW_STATE_DIR", tmpDir); + const cfg: OpenClawConfig = { + agents: { + defaults: { + workspace: tmpDir, + heartbeat: { every: "5m", target: "telegram" }, + }, + }, + channels: { telegram: { allowFrom: ["*"] } }, + session: { store: storePath }, + }; + const sessionKey = await seedMainSessionStore(storePath, cfg, { + lastChannel: "telegram", + lastProvider: "telegram", + lastTo: "-100155462274", + }); + enqueueSystemEvent("Unrelated queued event", { sessionKey }); + + const getReplyFromConfig = vi.fn().mockResolvedValue({ text: "HEARTBEAT_OK" }); + const telegram = vi.fn().mockResolvedValue({ + messageId: "m1", + chatId: "155462274", + }); + const result = await runHeartbeatOnce({ + cfg, + agentId: "main", + source: "exec-event", + intent: "event", + reason: "exec-event", + scheduledEveryMs: 5 * 60_000, + deps: { + getReplyFromConfig, + telegram, + }, + }); + expect(result.status).toBe("ran"); + expect(getReplyFromConfig).toHaveBeenCalledOnce(); + expect(peekSystemEvents(sessionKey)).toEqual(["Unrelated queued event"]); + }); + }); + + it("keeps tagged cron work alive when an exec wake is coalesced", async () => { + await withTempHeartbeatSandbox(async ({ tmpDir, storePath }) => { + setTestEnvValue("OPENCLAW_STATE_DIR", tmpDir); + const cfg: OpenClawConfig = { + agents: { + defaults: { + workspace: tmpDir, + heartbeat: { every: "5m", target: "telegram" }, + }, + }, + channels: { telegram: { allowFrom: ["*"] } }, + session: { store: storePath }, + }; + const sessionKey = await seedMainSessionStore(storePath, cfg, { + lastChannel: "telegram", + lastProvider: "telegram", + lastTo: "-100155462274", + }); + enqueueSystemEvent("Reminder: Check the overnight report", { + sessionKey, + contextKey: "cron:overnight-report", + }); + + const getReplyFromConfig = vi.fn().mockResolvedValue({ text: "HEARTBEAT_OK" }); + const result = await runHeartbeatOnce({ + cfg, + agentId: "main", + source: "exec-event", + intent: "event", + reason: "exec-event", + deps: { getReplyFromConfig }, + }); + + expect(result.status).toBe("ran"); + expect(getReplyFromConfig).toHaveBeenCalledOnce(); + expect(peekSystemEvents(sessionKey)).toEqual([]); + }); + }); + + it("retires a stale exec wake before retryable busy gates", async () => { + await withTempHeartbeatSandbox(async ({ tmpDir, storePath }) => { + setTestEnvValue("OPENCLAW_STATE_DIR", tmpDir); + const cfg: OpenClawConfig = { + agents: { + defaults: { + workspace: tmpDir, + heartbeat: { every: "5m", target: "telegram" }, + }, + }, + channels: { telegram: { allowFrom: ["*"] } }, + session: { store: storePath }, + }; + await seedMainSessionStore(storePath, cfg, { + lastChannel: "telegram", + lastProvider: "telegram", + lastTo: "-100155462274", + }); + + const result = await runHeartbeatOnce({ + cfg, + agentId: "main", + source: "exec-event", + intent: "event", + reason: "exec-event", + deps: { getQueueSize: () => 1 }, + }); + + expect(result).toEqual({ status: "skipped", reason: HEARTBEAT_SKIP_NO_PENDING_EVENT }); + }); + }); + + it("passes persisted cadence through an unscoped coalesced exec wake", async () => { + vi.useFakeTimers(); + const runSpy = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); + const runner = startHeartbeatRunner({ + cfg: heartbeatConfig(), + runOnce: runSpy, + stableSchedulerSeed: schedulerSeed, + }); + + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + scheduledEveryMs: 5 * 60_000, + coalesceMs: 100, + }); + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + coalesceMs: 100, + }); + await vi.advanceTimersByTimeAsync(100); + + const [options] = runSpy.mock.calls[0] ?? []; + expect(options).toMatchObject({ + source: "exec-event", + scheduledEveryMs: 5 * 60_000, + heartbeat: { every: "300000ms" }, + }); + runner.stop(); + }); + + it("does not move cadence when a stale exec wake defers for min-spacing", async () => { + vi.useFakeTimers(); + const intervalMs = 5 * 60_000; + const phaseMs = resolveHeartbeatPhaseMs({ + schedulerSeed, + agentId: "main", + intervalMs, + }); + const updateAtMs = phaseMs === 0 ? intervalMs - 1 : phaseMs - 1; + const initialNowMs = updateAtMs - 100; + vi.setSystemTime(initialNowMs); + + const runSpy = vi + .fn() + .mockResolvedValueOnce({ status: "ran", durationMs: 1 }) + .mockResolvedValueOnce({ status: "skipped", reason: HEARTBEAT_SKIP_NO_PENDING_EVENT }) + .mockResolvedValue({ status: "ran", durationMs: 1 }); + const runner = startHeartbeatRunner({ + cfg: heartbeatConfig(), + runOnce: runSpy, + stableSchedulerSeed: schedulerSeed, + }); + + requestHeartbeat({ + source: "manual", + intent: "manual", + reason: "manual", + agentId: "main", + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + await vi.advanceTimersByTimeAsync(99); + runner.updateConfig(heartbeatConfig("5m")); + await vi.advanceTimersByTimeAsync(1); + + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + agentId: "main", + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + expect(runSpy).toHaveBeenCalledOnce(); + + await vi.advanceTimersByTimeAsync(30_000); + + expect(runSpy).toHaveBeenCalledTimes(2); + runner.stop(); + }); + + it("does not move cadence when a stale exec wake defers for flood", async () => { + vi.useFakeTimers(); + const intervalMs = 5 * 60_000; + const phaseMs = resolveHeartbeatPhaseMs({ + schedulerSeed, + agentId: "main", + intervalMs, + }); + const updateAtMs = phaseMs === 0 ? intervalMs - 1 : phaseMs - 1; + const initialNowMs = updateAtMs - 100; + vi.setSystemTime(initialNowMs); + + const runSpy = vi + .fn() + .mockResolvedValueOnce({ status: "ran", durationMs: 1 }) + .mockResolvedValueOnce({ status: "ran", durationMs: 1 }) + .mockResolvedValueOnce({ status: "ran", durationMs: 1 }) + .mockResolvedValueOnce({ status: "ran", durationMs: 1 }) + .mockResolvedValueOnce({ status: "ran", durationMs: 1 }) + .mockResolvedValueOnce({ status: "skipped", reason: HEARTBEAT_SKIP_NO_PENDING_EVENT }) + .mockResolvedValue({ status: "ran", durationMs: 1 }); + const runner = startHeartbeatRunner({ + cfg: heartbeatConfig(), + runOnce: runSpy, + stableSchedulerSeed: schedulerSeed, + }); + + for (let index = 0; index < 5; index += 1) { + requestHeartbeat({ + source: "manual", + intent: "manual", + reason: "manual", + agentId: "main", + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + } + await vi.advanceTimersByTimeAsync(95); + runner.updateConfig(heartbeatConfig("5m")); + await vi.advanceTimersByTimeAsync(1); + + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + agentId: "main", + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + expect(runSpy).toHaveBeenCalledTimes(5); + + await vi.advanceTimersByTimeAsync(60_000); + + expect(runSpy).toHaveBeenCalledTimes(6); + runner.stop(); + }); + + it("does not record cooldown bookkeeping for an acknowledged exec wake", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date(0)); + const runSpy = vi + .fn() + .mockResolvedValueOnce({ status: "skipped", reason: HEARTBEAT_SKIP_NO_PENDING_EVENT }) + .mockResolvedValue({ status: "ran", durationMs: 1 }); + const runner = startHeartbeatRunner({ + cfg: heartbeatConfig(), + runOnce: runSpy, + stableSchedulerSeed: schedulerSeed, + }); + + const requestExecWake = () => + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + sessionKey: "agent:main:main", + coalesceMs: 0, + }); + requestExecWake(); + await vi.advanceTimersByTimeAsync(1); + requestExecWake(); + await vi.advanceTimersByTimeAsync(1); + + expect(runSpy).toHaveBeenCalledTimes(2); + runner.stop(); + }); + + it("does not fan out due commitments for an acknowledged exec wake", async () => { + vi.useFakeTimers(); + const nowMs = Date.parse("2026-04-29T17:00:00.000Z"); + vi.setSystemTime(nowMs); + + await withTempHeartbeatSandbox(async ({ tmpDir, storePath }) => { + setTestEnvValue("OPENCLAW_STATE_DIR", tmpDir); + seedCommitmentsForTest([buildDueCommitment(nowMs)]); + const cfg: OpenClawConfig = { + agents: { + defaults: { + workspace: tmpDir, + heartbeat: { every: "5m", target: "last" }, + }, + }, + session: { store: storePath }, + }; + const runOnce = vi + .fn() + .mockResolvedValue({ status: "skipped", reason: HEARTBEAT_SKIP_NO_PENDING_EVENT }); + const runner = startHeartbeatRunner({ + cfg, + runOnce, + stableSchedulerSeed: "acknowledged-exec-no-commitment", + }); + + requestHeartbeat({ + source: "exec-event", + intent: "event", + reason: "exec-event", + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + runner.stop(); + + expect(runOnce).toHaveBeenCalledTimes(1); + expect(runOnce.mock.calls[0]?.[0]).toMatchObject({ + source: "exec-event", + runScope: "global", + }); + }); + }); +}); diff --git a/src/infra/heartbeat-wake.ts b/src/infra/heartbeat-wake.ts index 4d3af0944859..b388aa5ae120 100644 --- a/src/infra/heartbeat-wake.ts +++ b/src/infra/heartbeat-wake.ts @@ -36,6 +36,7 @@ export type { export const HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT = "requests-in-flight"; export const HEARTBEAT_SKIP_CRON_IN_PROGRESS = "cron-in-progress"; export const HEARTBEAT_SKIP_LANES_BUSY = "lanes-busy"; +export const HEARTBEAT_SKIP_NO_PENDING_EVENT = "no-pending-event"; const RETRYABLE_BUSY_SKIP_REASONS = new Set([ HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT, HEARTBEAT_SKIP_CRON_IN_PROGRESS,