Skip to content

Commit fae5862

Browse files
zhangguiping-xydtzhangguiping-xydt
authored
Merge e331da8 into 02acb0b 
2 parents 02acb0b + e331da8 commit fae5862

3 files changed

Lines changed: 100 additions & 7 deletions

File tree

‎src/infra/heartbeat-runner.returns-default-unset.test.ts‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -706,6 +706,7 @@ describe("runHeartbeatOnce", () => {
706706
options?: {
707707
nowMs?: number;
708708
getReplyFromConfig?: HeartbeatDeps["getReplyFromConfig"];
709+
listActiveEmbeddedRunSessionKeys?: HeartbeatDeps["listActiveEmbeddedRunSessionKeys"];
709710
},
710711
): HeartbeatDeps => ({
711712
whatsapp: sendWhatsApp,
@@ -714,6 +715,9 @@ describe("runHeartbeatOnce", () => {
714715
webAuthExists: async () => true,
715716
hasActiveWebListener: () => true,
716717
...(options?.getReplyFromConfig ? { getReplyFromConfig: options.getReplyFromConfig } : null),
718+
...(options?.listActiveEmbeddedRunSessionKeys
719+
? { listActiveEmbeddedRunSessionKeys: options.listActiveEmbeddedRunSessionKeys }
720+
: null),
717721
});
718722

719723
it("skips when agent heartbeat is not enabled", async () => {
@@ -731,6 +735,33 @@ describe("runHeartbeatOnce", () => {
731735
}
732736
});
733737

738+
it.each([
739+
["the heartbeat main session", (cfg: OpenClawConfig) => resolveMainSessionKey(cfg)],
740+
["another session for the same agent", () => "agent:main:telegram:alerts"],
741+
])("retries instead of dispatching while %s has an embedded run", async (_name, activeKey) => {
742+
const cfg: OpenClawConfig = {
743+
agents: {
744+
defaults: {
745+
heartbeat: { every: "5m", target: "none" },
746+
},
747+
},
748+
};
749+
const replySpy = vi.fn().mockResolvedValue({ text: "heartbeat reply" });
750+
const sendWhatsApp = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
751+
752+
const res = await runHeartbeatOnce({
753+
cfg,
754+
deps: createHeartbeatDeps(sendWhatsApp, {
755+
getReplyFromConfig: replySpy,
756+
listActiveEmbeddedRunSessionKeys: () => [activeKey(cfg)],
757+
}),
758+
});
759+
760+
expect(res).toEqual({ status: "skipped", reason: "requests-in-flight" });
761+
expect(replySpy).not.toHaveBeenCalled();
762+
expect(sendWhatsApp).not.toHaveBeenCalled();
763+
});
764+
734765
it("skips outside active hours", async () => {
735766
const cfg: OpenClawConfig = {
736767
agents: {

‎src/infra/heartbeat-runner.skips-busy-session-lane.test.ts‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -360,6 +360,39 @@ describe("heartbeat runner skips when target session lane is busy", () => {
360360
});
361361
});
362362

363+
it("returns requests-in-flight when reply admission races after the preflight checks", async () => {
364+
await withTempHeartbeatSandbox(async ({ storePath }) => {
365+
const cfg = createHeartbeatTelegramConfig();
366+
const sessionKey = await seedHeartbeatTelegramSession(storePath, cfg);
367+
let operation: ReturnType<typeof createReplyOperation> | undefined;
368+
const replySpy = vi.fn(async () => {
369+
operation = createReplyOperation({
370+
sessionKey,
371+
sessionId: "racing-visible-session",
372+
resetTriggered: false,
373+
});
374+
operation.setPhase("running");
375+
return undefined;
376+
});
377+
378+
try {
379+
const result = await runHeartbeatOnce({
380+
cfg,
381+
deps: {
382+
getQueueSize: vi.fn((_lane?: string) => 0),
383+
nowMs: () => Date.now(),
384+
getReplyFromConfig: replySpy,
385+
} as HeartbeatDeps,
386+
});
387+
388+
expect(result).toEqual({ status: "skipped", reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT });
389+
expect(replySpy).toHaveBeenCalledOnce();
390+
} finally {
391+
operation?.complete();
392+
}
393+
});
394+
});
395+
363396
it("returns requests-in-flight when an isolated heartbeat reply run is still active", async () => {
364397
await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
365398
const cfg = createHeartbeatTelegramConfig();

‎src/infra/heartbeat-runner.ts‎

Lines changed: 36 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import {
2020
} from "../agents/agent-scope.js";
2121
import { appendCronStyleCurrentTimeLine } from "../agents/current-time.js";
2222
import { resolveEmbeddedSessionLane } from "../agents/embedded-agent-runner/lanes.js";
23+
import { listActiveEmbeddedRunSessionKeys } from "../agents/embedded-agent-runner/run-state.js";
2324
import { formatReasoningMessage } from "../agents/embedded-agent-utils.js";
2425
import { resolveAgentHarnessPolicy } from "../agents/harness/policy.js";
2526
import { resolveModelRefFromString, type ModelRef } from "../agents/model-selection.js";
@@ -155,6 +156,7 @@ export type HeartbeatDeps = OutboundSendDeps &
155156
getCommandLaneSnapshots?: () => readonly CommandLaneSnapshot[];
156157
isReplyRunActive?: (sessionKey: string) => boolean;
157158
listActiveReplyRunSessionKeys?: () => readonly string[];
159+
listActiveEmbeddedRunSessionKeys?: () => readonly string[];
158160
nowMs?: () => number;
159161
};
160162

@@ -229,17 +231,22 @@ function hasAgentOptInBusyLaneWork(
229231
return hasQueuedWorkInLaneSnapshots(getSnapshots(), (lane) => laneBelongsToAgent(lane, agentId));
230232
}
231233

232-
function hasActiveReplyRunForAgent(
233-
agentId: string,
234-
listSessionKeys: () => readonly string[],
235-
): boolean {
234+
function hasActiveRunForAgent(agentId: string, listSessionKeys: () => readonly string[]): boolean {
236235
const normalizedAgentId = normalizeAgentId(agentId);
237236
return listSessionKeys().some((sessionKey) => {
238237
const parsed = parseAgentSessionKey(sessionKey);
239238
return parsed ? normalizeAgentId(parsed.agentId) === normalizedAgentId : false;
240239
});
241240
}
242241

242+
function hasActiveRunForSession(
243+
sessionKey: string,
244+
listSessionKeys: () => readonly string[],
245+
): boolean {
246+
const normalizedSessionKey = sessionKey.trim();
247+
return Boolean(normalizedSessionKey) && listSessionKeys().includes(normalizedSessionKey);
248+
}
249+
243250
function resolveHeartbeatChannelPlugin(channel: string): ChannelPlugin | undefined {
244251
const activePlugin = getActivePluginChannelRegistry()?.channels.find(
245252
(entry) => entry.plugin.id === channel,
@@ -1358,10 +1365,16 @@ export async function runHeartbeatOnce(opts: {
13581365
const shouldHonorActiveReplyRuns = opts.intent !== "immediate" && opts.intent !== "manual";
13591366
const listActiveReplyRuns =
13601367
opts.deps?.listActiveReplyRunSessionKeys ?? listActiveReplyRunSessionKeys;
1368+
const listActiveEmbeddedRuns =
1369+
opts.deps?.listActiveEmbeddedRunSessionKeys ?? listActiveEmbeddedRunSessionKeys;
13611370
// Scheduled heartbeats are background work, so defer them when any session on
13621371
// the same agent is already replying; immediate/manual wakes keep their
13631372
// existing semantics for explicit user/system actions.
1364-
if (shouldHonorActiveReplyRuns && hasActiveReplyRunForAgent(agentId, listActiveReplyRuns)) {
1373+
if (
1374+
shouldHonorActiveReplyRuns &&
1375+
(hasActiveRunForAgent(agentId, listActiveReplyRuns) ||
1376+
hasActiveRunForAgent(agentId, listActiveEmbeddedRuns))
1377+
) {
13651378
emitHeartbeatEvent({
13661379
status: "skipped",
13671380
reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT,
@@ -1417,7 +1430,7 @@ export async function runHeartbeatOnce(opts: {
14171430
const { entry, sessionKey, storePath, suppressOriginatingContext } = preflight.session;
14181431
const isReplyRunActive =
14191432
opts.deps?.isReplyRunActive ?? ((key: string) => replyRunRegistry.isActive(key));
1420-
if (isReplyRunActive(sessionKey)) {
1433+
if (isReplyRunActive(sessionKey) || hasActiveRunForSession(sessionKey, listActiveEmbeddedRuns)) {
14211434
emitHeartbeatEvent({
14221435
status: "skipped",
14231436
reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT,
@@ -1576,7 +1589,10 @@ export async function runHeartbeatOnce(opts: {
15761589
isolatedSessionKey,
15771590
isolatedBaseSessionKey,
15781591
});
1579-
if (isReplyRunActive(isolatedSessionKey)) {
1592+
if (
1593+
isReplyRunActive(isolatedSessionKey) ||
1594+
hasActiveRunForSession(isolatedSessionKey, listActiveEmbeddedRuns)
1595+
) {
15801596
emitHeartbeatEvent({
15811597
status: "skipped",
15821598
reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT,
@@ -1800,6 +1816,19 @@ export async function runHeartbeatOnce(opts: {
18001816
const replyResult = await getReplyFromConfig(ctx, replyOpts, cfg);
18011817
const heartbeatToolResponse = resolveHeartbeatToolResponseFromReplyResult(replyResult);
18021818
const replyPayload = resolveHeartbeatReplyPayload(replyResult);
1819+
if (
1820+
!heartbeatToolResponse &&
1821+
(!replyPayload || !hasOutboundReplyContent(replyPayload)) &&
1822+
(isReplyRunActive(runSessionKey) ||
1823+
hasActiveRunForSession(runSessionKey, listActiveEmbeddedRuns))
1824+
) {
1825+
emitHeartbeatEvent({
1826+
status: "skipped",
1827+
reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT,
1828+
durationMs: Date.now() - startedAt,
1829+
});
1830+
return { status: "skipped", reason: HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT };
1831+
}
18031832
const includeReasoning = heartbeat?.includeReasoning === true;
18041833
const reasoningPayloads = includeReasoning
18051834
? resolveHeartbeatReasoningPayloads(replyResult).filter((payload) => payload !== replyPayload)

0 commit comments

Comments
 (0)