ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-27 12:56:01 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(heartbeat): ignore ack-only pending delivery replay

Browse Source
This commit is contained in:
HCL
2026-05-08 13:40:18 +08:00
committed by Peter Steinberger
parent c14f4af2cc
commit 06d8cd1b23
3 changed files with 82 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/auto-reply/reply/agent-runner.ts
+18 -1
View File
@@ -42,6 +42,7 @@ import {
buildFallbackNotice,
resolveFallbackTransition,
} from "../fallback-state.js";
import { stripHeartbeatToken } from "../heartbeat.js";
import {
markReplyPayloadForSourceSuppressionDelivery,
setReplyPayloadMetadata,
@@ -832,6 +833,16 @@ function buildPendingFinalDeliveryText(payloads: ReplyPayload[]): string {
.join("\n\n");
}
function shouldSkipHeartbeatPendingFinalDelivery(params: {
isHeartbeat: boolean;
pendingText: string;
}): boolean {
if (!params.isHeartbeat) {
return false;
}
return stripHeartbeatToken(params.pendingText, { mode: "heartbeat" }).shouldSkip;
}
function enqueueCommitmentExtractionForTurn(params: {
cfg: OpenClawConfig;
commandBody: string;
@@ -1911,7 +1922,13 @@ export async function runReplyAgent(params: {
const pendingText = sourceReplyPolicy.suppressDelivery
? ""
: buildPendingFinalDeliveryText(finalPayloads);
if (pendingText) {
if (
pendingText &&
!shouldSkipHeartbeatPendingFinalDelivery({
isHeartbeat,
pendingText,
})
) {
await updateSessionStoreEntry({
storePath,
sessionKey,
src/auto-reply/reply/get-reply.ts
+56 -22
View File
@@ -19,6 +19,7 @@ import { createLazyImportLoader } from "../../shared/lazy-promise.js";
import { normalizeOptionalString } from "../../shared/string-coerce.js";
import { normalizeStringEntries } from "../../shared/string-normalization.js";
import type { GetReplyOptions } from "../get-reply-options.types.js";
import { stripHeartbeatToken } from "../heartbeat.js";
import type { ReplyPayload } from "../reply-payload.js";
import type { MsgContext } from "../templating.js";
import { normalizeVerboseLevel } from "../thinking.js";
@@ -51,6 +52,10 @@ import { createTypingController } from "./typing.js";
type ResetCommandAction = "new" | "reset";
function isHeartbeatPendingFinalDeliveryEffectivelyEmpty(text: string): boolean {
return stripHeartbeatToken(text, { mode: "heartbeat" }).shouldSkip;
}
const sessionResetModelRuntimeLoader = createLazyImportLoader(
() => import("./session-reset-model.runtime.js"),
);
@@ -371,29 +376,58 @@ export async function getReplyFromConfig(
// If it's a user message, we deliver the lost reply first, then continue.
// For now, let's just return the lost reply if it's a heartbeat.
if (opts?.isHeartbeat) {
const updatedAt = Date.now();
const attemptCount = (sessionEntry.pendingFinalDeliveryAttemptCount ?? 0) + 1;
sessionEntry.pendingFinalDeliveryLastAttemptAt = updatedAt;
sessionEntry.pendingFinalDeliveryAttemptCount = attemptCount;
sessionEntry.pendingFinalDeliveryLastError = null;
sessionEntry.updatedAt = updatedAt;
if (sessionKey && sessionStore) {
sessionStore[sessionKey] = sessionEntry;
if (isHeartbeatPendingFinalDeliveryEffectivelyEmpty(text)) {
sessionEntry.pendingFinalDelivery = undefined;
sessionEntry.pendingFinalDeliveryText = undefined;
sessionEntry.pendingFinalDeliveryCreatedAt = undefined;
sessionEntry.pendingFinalDeliveryLastAttemptAt = undefined;
sessionEntry.pendingFinalDeliveryAttemptCount = undefined;
sessionEntry.pendingFinalDeliveryLastError = undefined;
sessionEntry.pendingFinalDeliveryContext = undefined;
if (sessionKey && sessionStore) {
sessionStore[sessionKey] = sessionEntry;
}
if (sessionKey && storePath) {
const { updateSessionStoreEntry } = await import("../../config/sessions.js");
await updateSessionStoreEntry({
storePath,
sessionKey,
update: async () => ({
pendingFinalDelivery: undefined,
pendingFinalDeliveryText: undefined,
pendingFinalDeliveryCreatedAt: undefined,
pendingFinalDeliveryLastAttemptAt: undefined,
pendingFinalDeliveryAttemptCount: undefined,
pendingFinalDeliveryLastError: undefined,
pendingFinalDeliveryContext: undefined,
}),
});
}
} else {
const updatedAt = Date.now();
const attemptCount = (sessionEntry.pendingFinalDeliveryAttemptCount ?? 0) + 1;
sessionEntry.pendingFinalDeliveryLastAttemptAt = updatedAt;
sessionEntry.pendingFinalDeliveryAttemptCount = attemptCount;
sessionEntry.pendingFinalDeliveryLastError = null;
sessionEntry.updatedAt = updatedAt;
if (sessionKey && sessionStore) {
sessionStore[sessionKey] = sessionEntry;
}
if (sessionKey && storePath) {
const { updateSessionStoreEntry } = await import("../../config/sessions.js");
await updateSessionStoreEntry({
storePath,
sessionKey,
update: async () => ({
pendingFinalDeliveryLastAttemptAt: updatedAt,
pendingFinalDeliveryAttemptCount: attemptCount,
pendingFinalDeliveryLastError: null,
updatedAt,
}),
});
}
return { text };
}
if (sessionKey && storePath) {
const { updateSessionStoreEntry } = await import("../../config/sessions.js");
await updateSessionStoreEntry({
storePath,
sessionKey,
update: async () => ({
pendingFinalDeliveryLastAttemptAt: updatedAt,
pendingFinalDeliveryAttemptCount: attemptCount,
pendingFinalDeliveryLastError: null,
updatedAt,
}),
});
}
return { text };
}
}
src/infra/heartbeat-runner.ts
+8
View File
@@ -1230,8 +1230,16 @@ export async function runHeartbeatOnce(opts: {
opts.sessionKey,
);
const HEARTBEAT_DEFER_WINDOW_MS = 30_000;
const pendingFinalDeliveryText = recentSessionEntry?.pendingFinalDeliveryText;
const pendingFinalDeliveryIsHeartbeatAck =
typeof pendingFinalDeliveryText === "string" &&
stripHeartbeatToken(pendingFinalDeliveryText, {
mode: "heartbeat",
maxAckChars: resolveHeartbeatAckMaxChars(cfg, heartbeat),
}).shouldSkip;
if (
recentSessionEntry?.pendingFinalDelivery === true &&
!pendingFinalDeliveryIsHeartbeatAck &&
recentSessionEntry?.updatedAt &&
startedAt - recentSessionEntry.updatedAt < HEARTBEAT_DEFER_WINDOW_MS
) {