From 2196ea29304a2c79efd30fcfcafa6a8391667888 Mon Sep 17 00:00:00 2001 From: zhang-guiping Date: Tue, 16 Jun 2026 14:17:37 +0800 Subject: [PATCH] fix #85871: [Bug]: Heartbeat scheduler silently fails to fire on 5.20 and all 5.x versions (regression from 4.23) (#88970) * fix heartbeat deferral during active embedded runs * fix heartbeat admission busy retry * fix(heartbeat): bind retry to local admission --------- Co-authored-by: Vincent Koc <25068+vincentkoc@users.noreply.github.com> --- .../agent-runner.runreplyagent.e2e.test.ts | 33 +++++++++++-- src/auto-reply/reply/agent-runner.ts | 14 ++++++ .../reply/reply-operation-run-state.ts | 21 ++++++++ ...tbeat-runner.returns-default-unset.test.ts | 31 ++++++++++++ ...eat-runner.skips-busy-session-lane.test.ts | 39 +++++++++++++++ src/infra/heartbeat-runner.ts | 49 ++++++++++++++++--- 6 files changed, 176 insertions(+), 11 deletions(-) create mode 100644 src/auto-reply/reply/reply-operation-run-state.ts diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index abb4a3deadfb..b46e726e1ffa 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts @@ -15,6 +15,11 @@ import { type FollowupRun, type QueueSettings, } from "./queue.js"; +import { + REPLY_OPERATION_RUN_STATE, + type ReplyOperationRunState, + type ReplyOptionsWithOperationRunState, +} from "./reply-operation-run-state.js"; import { createReplyOperation, testing as replyRunTesting } from "./reply-run-registry.js"; import { consumeReplyUsageState } from "./reply-usage-state.js"; import { createMockTypingController } from "./test-helpers.js"; @@ -152,7 +157,7 @@ beforeEach(() => { }); function createMinimalRun(params?: { - opts?: GetReplyOptions; + opts?: GetReplyOptions & ReplyOptionsWithOperationRunState; resolvedVerboseLevel?: "off" | "on"; sessionStore?: Record; sessionEntry?: SessionEntry; @@ -245,13 +250,14 @@ function createMinimalRun(params?: { describe("runReplyAgent heartbeat followup guard", () => { it("drops heartbeat runs when reply-lane admission finds an active owner", async () => { + const runState: ReplyOperationRunState = {}; const active = createReplyOperation({ sessionKey: "main", sessionId: "active-session", resetTriggered: false, }); const { run, typing } = createMinimalRun({ - opts: { isHeartbeat: true }, + opts: { isHeartbeat: true, [REPLY_OPERATION_RUN_STATE]: runState }, isActive: false, shouldFollowup: false, }); @@ -261,9 +267,21 @@ describe("runReplyAgent heartbeat followup guard", () => { expect(result).toBeUndefined(); expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled(); expect(typing.cleanup).toHaveBeenCalledTimes(1); + expect(runState.admission).toEqual({ status: "skipped", reason: "active-run" }); active.complete(); }); + it("records the operation owned by an admitted heartbeat run", async () => { + const runState: ReplyOperationRunState = {}; + const { run } = createMinimalRun({ + opts: { isHeartbeat: true, [REPLY_OPERATION_RUN_STATE]: runState }, + }); + + await run(); + + expect(runState.admission).toEqual({ status: "owned" }); + }); + it("runs visible turns with the session id returned by admission", async () => { const active = createReplyOperation({ sessionKey: "main", @@ -315,8 +333,12 @@ describe("runReplyAgent heartbeat followup guard", () => { it("drops runs when reply-lane admission sees an already-aborted caller", async () => { const abortController = new AbortController(); abortController.abort(); + const runState: ReplyOperationRunState = {}; const { run, typing } = createMinimalRun({ - opts: { abortSignal: abortController.signal }, + opts: { + abortSignal: abortController.signal, + [REPLY_OPERATION_RUN_STATE]: runState, + }, isActive: false, shouldFollowup: false, }); @@ -326,11 +348,13 @@ describe("runReplyAgent heartbeat followup guard", () => { expect(result).toBeUndefined(); expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled(); expect(typing.cleanup).toHaveBeenCalledTimes(1); + expect(runState.admission).toEqual({ status: "skipped", reason: "aborted" }); }); it("drops heartbeat runs when another run is active", async () => { + const runState: ReplyOperationRunState = {}; const { run, typing } = createMinimalRun({ - opts: { isHeartbeat: true }, + opts: { isHeartbeat: true, [REPLY_OPERATION_RUN_STATE]: runState }, isActive: true, shouldFollowup: true, resolvedQueueMode: "collect", @@ -342,6 +366,7 @@ describe("runReplyAgent heartbeat followup guard", () => { expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled(); expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled(); expect(typing.cleanup).toHaveBeenCalledTimes(1); + expect(runState.admission).toEqual({ status: "skipped", reason: "active-run" }); }); it("drops heartbeat runs before steering active streams", async () => { diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index 89b78c637b52..639f7d9c44c0 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -118,6 +118,7 @@ import { type QueueSettings, } from "./queue.js"; import { createReplyMediaContext } from "./reply-media-paths.js"; +import { resolveReplyOperationRunState } from "./reply-operation-run-state.js"; import { replyRunRegistry, runAfterReplyOperationClear, @@ -1202,6 +1203,7 @@ export async function runReplyAgent(params: { const activeRunQueueMode = effectiveResetTriggered ? "interrupt" : resolvedQueue.mode; const isHeartbeat = opts?.isHeartbeat === true; + const replyOperationRunState = resolveReplyOperationRunState(opts); const traceAttributes = { provider: followupRun.run.provider, hasSessionKey: Boolean(sessionKey ?? followupRun.run.sessionKey), @@ -1295,6 +1297,9 @@ export async function runReplyAgent(params: { }); if (activeRunQueueAction === "drop") { + if (replyOperationRunState) { + replyOperationRunState.admission = { status: "skipped", reason: "active-run" }; + } typing.cleanup(); return undefined; } @@ -1405,6 +1410,9 @@ export async function runReplyAgent(params: { let replyOperation: ReplyOperation; if (providedReplyOperation) { replyOperation = providedReplyOperation; + if (replyOperationRunState) { + replyOperationRunState.admission = { status: "owned" }; + } } else { const replyTurnKind = resolveReplyTurnKind(opts); const admission = await admitReplyTurn({ @@ -1415,6 +1423,12 @@ export async function runReplyAgent(params: { routeThreadId: replyRouteThreadId, upstreamAbortSignal: opts?.abortSignal, }); + if (replyOperationRunState) { + replyOperationRunState.admission = + admission.status === "owned" + ? { status: "owned" } + : { status: "skipped", reason: admission.reason }; + } if (admission.status === "skipped") { typing.cleanup(); if (admission.reason !== "active-run" || replyTurnKind !== "visible") { diff --git a/src/auto-reply/reply/reply-operation-run-state.ts b/src/auto-reply/reply/reply-operation-run-state.ts new file mode 100644 index 000000000000..3a978166d66f --- /dev/null +++ b/src/auto-reply/reply/reply-operation-run-state.ts @@ -0,0 +1,21 @@ +export type ReplyOperationAdmissionSnapshot = + | { status: "owned" } + | { status: "skipped"; reason: "active-run" | "aborted" }; + +export type ReplyOperationRunState = { + admission?: ReplyOperationAdmissionSnapshot; +}; + +// Carries this invocation's admission decision through reply option spreads so +// heartbeat cleanup never infers it from whichever operation is active later. +export const REPLY_OPERATION_RUN_STATE = Symbol("openclaw.replyOperationRunState"); + +export type ReplyOptionsWithOperationRunState = { + [REPLY_OPERATION_RUN_STATE]?: ReplyOperationRunState; +}; + +export function resolveReplyOperationRunState( + options: object | undefined, +): ReplyOperationRunState | undefined { + return (options as ReplyOptionsWithOperationRunState | undefined)?.[REPLY_OPERATION_RUN_STATE]; +} diff --git a/src/infra/heartbeat-runner.returns-default-unset.test.ts b/src/infra/heartbeat-runner.returns-default-unset.test.ts index a11937c3d6eb..9341fe5569b8 100644 --- a/src/infra/heartbeat-runner.returns-default-unset.test.ts +++ b/src/infra/heartbeat-runner.returns-default-unset.test.ts @@ -706,6 +706,7 @@ describe("runHeartbeatOnce", () => { options?: { nowMs?: number; getReplyFromConfig?: HeartbeatDeps["getReplyFromConfig"]; + listActiveEmbeddedRunSessionKeys?: HeartbeatDeps["listActiveEmbeddedRunSessionKeys"]; }, ): HeartbeatDeps => ({ whatsapp: sendWhatsApp, @@ -714,6 +715,9 @@ describe("runHeartbeatOnce", () => { webAuthExists: async () => true, hasActiveWebListener: () => true, ...(options?.getReplyFromConfig ? { getReplyFromConfig: options.getReplyFromConfig } : null), + ...(options?.listActiveEmbeddedRunSessionKeys + ? { listActiveEmbeddedRunSessionKeys: options.listActiveEmbeddedRunSessionKeys } + : null), }); it("skips when agent heartbeat is not enabled", async () => { @@ -731,6 +735,33 @@ describe("runHeartbeatOnce", () => { } }); + it.each([ + ["the heartbeat main session", (cfg: OpenClawConfig) => resolveMainSessionKey(cfg)], + ["another session for the same agent", () => "agent:main:telegram:alerts"], + ])("retries instead of dispatching while %s has an embedded run", async (_name, activeKey) => { + const cfg: OpenClawConfig = { + agents: { + defaults: { + heartbeat: { every: "5m", target: "none" }, + }, + }, + }; + const replySpy = vi.fn().mockResolvedValue({ text: "heartbeat reply" }); + const sendWhatsApp = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" }); + + const res = await runHeartbeatOnce({ + cfg, + deps: createHeartbeatDeps(sendWhatsApp, { + getReplyFromConfig: replySpy, + listActiveEmbeddedRunSessionKeys: () => [activeKey(cfg)], + }), + }); + + expect(res).toEqual({ status: "skipped", reason: "requests-in-flight" }); + expect(replySpy).not.toHaveBeenCalled(); + expect(sendWhatsApp).not.toHaveBeenCalled(); + }); + it("skips outside active hours", async () => { const cfg: OpenClawConfig = { agents: { diff --git a/src/infra/heartbeat-runner.skips-busy-session-lane.test.ts b/src/infra/heartbeat-runner.skips-busy-session-lane.test.ts index d1067c0d22e4..553a1f281fe6 100644 --- a/src/infra/heartbeat-runner.skips-busy-session-lane.test.ts +++ b/src/infra/heartbeat-runner.skips-busy-session-lane.test.ts @@ -1,6 +1,7 @@ // Covers heartbeat skipping while session lanes or cron jobs are busy. import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { resolveNestedAgentLaneForSession } from "../agents/lanes.js"; +import { resolveReplyOperationRunState } from "../auto-reply/reply/reply-operation-run-state.js"; import { __testing as replyRunRegistryTesting, createReplyOperation, @@ -360,6 +361,44 @@ describe("heartbeat runner skips when target session lane is busy", () => { }); }); + it("does not infer admission rejection from a replacement run after an empty heartbeat", async () => { + await withTempHeartbeatSandbox(async ({ storePath }) => { + const cfg = createHeartbeatTelegramConfig(); + const sessionKey = await seedHeartbeatTelegramSession(storePath, cfg); + let operation: ReturnType | undefined; + const replySpy = vi.fn(async (_ctx, replyOptions) => { + const runState = resolveReplyOperationRunState(replyOptions); + if (!runState) { + throw new Error("expected heartbeat reply operation state"); + } + runState.admission = { status: "owned" }; + operation = createReplyOperation({ + sessionKey, + sessionId: "racing-visible-session", + resetTriggered: false, + }); + operation.setPhase("running"); + return undefined; + }); + + try { + const result = await runHeartbeatOnce({ + cfg, + deps: { + getQueueSize: vi.fn((_lane?: string) => 0), + nowMs: () => Date.now(), + getReplyFromConfig: replySpy, + } as HeartbeatDeps, + }); + + expect(result.status).toBe("ran"); + expect(replySpy).toHaveBeenCalledOnce(); + } finally { + operation?.complete(); + } + }); + }); + it("returns requests-in-flight when an isolated heartbeat reply run is still active", async () => { await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => { const cfg = createHeartbeatTelegramConfig(); diff --git a/src/infra/heartbeat-runner.ts b/src/infra/heartbeat-runner.ts index 5057881de09c..76e5a364eeb6 100644 --- a/src/infra/heartbeat-runner.ts +++ b/src/infra/heartbeat-runner.ts @@ -20,6 +20,7 @@ import { } from "../agents/agent-scope.js"; import { appendCronStyleCurrentTimeLine } from "../agents/current-time.js"; import { resolveEmbeddedSessionLane } from "../agents/embedded-agent-runner/lanes.js"; +import { listActiveEmbeddedRunSessionKeys } from "../agents/embedded-agent-runner/run-state.js"; import { formatReasoningMessage } from "../agents/embedded-agent-utils.js"; import { resolveAgentHarnessPolicy } from "../agents/harness/policy.js"; import { resolveModelRefFromString, type ModelRef } from "../agents/model-selection.js"; @@ -43,6 +44,10 @@ import { } from "../auto-reply/heartbeat.js"; import { replaceGenericExternalRunFailureText } from "../auto-reply/reply/agent-runner-failure-copy.js"; import { resolveDefaultModel } from "../auto-reply/reply/directive-handling.defaults.js"; +import { + REPLY_OPERATION_RUN_STATE, + type ReplyOperationRunState, +} from "../auto-reply/reply/reply-operation-run-state.js"; import { listActiveReplyRunSessionKeys, replyRunRegistry, @@ -155,6 +160,7 @@ export type HeartbeatDeps = OutboundSendDeps & getCommandLaneSnapshots?: () => readonly CommandLaneSnapshot[]; isReplyRunActive?: (sessionKey: string) => boolean; listActiveReplyRunSessionKeys?: () => readonly string[]; + listActiveEmbeddedRunSessionKeys?: () => readonly string[]; nowMs?: () => number; }; @@ -229,10 +235,7 @@ function hasAgentOptInBusyLaneWork( return hasQueuedWorkInLaneSnapshots(getSnapshots(), (lane) => laneBelongsToAgent(lane, agentId)); } -function hasActiveReplyRunForAgent( - agentId: string, - listSessionKeys: () => readonly string[], -): boolean { +function hasActiveRunForAgent(agentId: string, listSessionKeys: () => readonly string[]): boolean { const normalizedAgentId = normalizeAgentId(agentId); return listSessionKeys().some((sessionKey) => { const parsed = parseAgentSessionKey(sessionKey); @@ -240,6 +243,14 @@ function hasActiveReplyRunForAgent( }); } +function hasActiveRunForSession( + sessionKey: string, + listSessionKeys: () => readonly string[], +): boolean { + const normalizedSessionKey = sessionKey.trim(); + return Boolean(normalizedSessionKey) && listSessionKeys().includes(normalizedSessionKey); +} + function resolveHeartbeatChannelPlugin(channel: string): ChannelPlugin | undefined { const activePlugin = getActivePluginChannelRegistry()?.channels.find( (entry) => entry.plugin.id === channel, @@ -1358,10 +1369,16 @@ export async function runHeartbeatOnce(opts: { const shouldHonorActiveReplyRuns = opts.intent !== "immediate" && opts.intent !== "manual"; const listActiveReplyRuns = opts.deps?.listActiveReplyRunSessionKeys ?? listActiveReplyRunSessionKeys; + const listActiveEmbeddedRuns = + opts.deps?.listActiveEmbeddedRunSessionKeys ?? listActiveEmbeddedRunSessionKeys; // Scheduled heartbeats are background work, so defer them when any session on // the same agent is already replying; immediate/manual wakes keep their // existing semantics for explicit user/system actions. - if (shouldHonorActiveReplyRuns && hasActiveReplyRunForAgent(agentId, listActiveReplyRuns)) { + if ( + shouldHonorActiveReplyRuns && + (hasActiveRunForAgent(agentId, listActiveReplyRuns) || + hasActiveRunForAgent(agentId, listActiveEmbeddedRuns)) + ) { emitHeartbeatEvent({ status: "skipped", reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT, @@ -1417,7 +1434,7 @@ export async function runHeartbeatOnce(opts: { const { entry, sessionKey, storePath, suppressOriginatingContext } = preflight.session; const isReplyRunActive = opts.deps?.isReplyRunActive ?? ((key: string) => replyRunRegistry.isActive(key)); - if (isReplyRunActive(sessionKey)) { + if (isReplyRunActive(sessionKey) || hasActiveRunForSession(sessionKey, listActiveEmbeddedRuns)) { emitHeartbeatEvent({ status: "skipped", reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT, @@ -1576,7 +1593,10 @@ export async function runHeartbeatOnce(opts: { isolatedSessionKey, isolatedBaseSessionKey, }); - if (isReplyRunActive(isolatedSessionKey)) { + if ( + isReplyRunActive(isolatedSessionKey) || + hasActiveRunForSession(isolatedSessionKey, listActiveEmbeddedRuns) + ) { emitHeartbeatEvent({ status: "skipped", reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT, @@ -1781,8 +1801,10 @@ export async function runHeartbeatOnce(opts: { const timeoutOverrideSeconds = resolveHeartbeatTimeoutOverrideSeconds(cfg, heartbeat); const bootstrapContextMode: "lightweight" | undefined = heartbeat?.lightContext === true ? "lightweight" : undefined; + const replyOperationRunState: ReplyOperationRunState = {}; const replyOpts = { isHeartbeat: true, + [REPLY_OPERATION_RUN_STATE]: replyOperationRunState, ...(heartbeatModelOverride ? { heartbeatModelOverride } : {}), suppressToolErrorWarnings, ...(usesHeartbeatResponseTool ? { enableHeartbeatTool: true, forceHeartbeatTool: true } : {}), @@ -1800,6 +1822,19 @@ export async function runHeartbeatOnce(opts: { const replyResult = await getReplyFromConfig(ctx, replyOpts, cfg); const heartbeatToolResponse = resolveHeartbeatToolResponseFromReplyResult(replyResult); const replyPayload = resolveHeartbeatReplyPayload(replyResult); + if ( + !heartbeatToolResponse && + (!replyPayload || !hasOutboundReplyContent(replyPayload)) && + replyOperationRunState.admission?.status === "skipped" && + replyOperationRunState.admission.reason === "active-run" + ) { + emitHeartbeatEvent({ + status: "skipped", + reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT, + durationMs: Date.now() - startedAt, + }); + return { status: "skipped", reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT }; + } const includeReasoning = heartbeat?.includeReasoning === true; const reasoningPayloads = includeReasoning ? resolveHeartbeatReasoningPayloads(replyResult).filter((payload) => payload !== replyPayload)