Skip to content

Commit cd98f19

Browse files
nxmxbbdnxmxbbd
authored
fix(gateway): ignore stale abort markers for fresh chat events (#91013)
Summary: - The branch stamps Gateway chat run registrations and abort markers with ordering metadata, uses freshness checks for chat projection suppression, and updates abort/restart/maintenance tests and related types. - PR surface: Source +79, Tests +103. Total +182 across 13 files. - Reproducibility: yes. source-level: on current main, seed abortedRuns for a client run id, register a same-k ... end; the presence-only checks suppress both projections. I did not execute tests in this read-only review. Automerge notes: - PR branch already contained follow-up commit before automerge: ci: re-trigger checks against current main - PR branch already contained follow-up commit before automerge: Merge upstream/main into stale-abort marker fix - PR branch already contained follow-up commit before automerge: Merge remote-tracking branch 'upstream/main' into nex/91013-conflict-… Validation: - ClawSweeper review passed for head 6f13d6f. - Required merge gates passed before the squash merge. Prepared head SHA: 6f13d6f Review: #91013 (comment) Co-authored-by: nxmxbbd <[email protected]>
1 parent be7d86e commit cd98f19

13 files changed

Lines changed: 221 additions & 39 deletions

‎src/gateway/chat-abort.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js";
1212
import { emitAgentEvent, getAgentEventLifecycleGeneration } from "../infra/agent-events.js";
1313
import { jsonUtf8Bytes } from "../infra/json-utf8-bytes.js";
1414
import { projectLiveAssistantBufferedText } from "./live-chat-projector.js";
15+
import { createChatAbortMarker, type ChatAbortMarker } from "./server-chat-state.js";
1516

1617
const DEFAULT_CHAT_RUN_ABORT_GRACE_MS = 60_000;
1718

@@ -337,7 +338,7 @@ export function boundInFlightRunSnapshotForChatHistory(params: {
337338
export type ChatAbortOps = {
338339
chatAbortControllers: Map<string, ChatAbortControllerEntry>;
339340
chatRunBuffers: Map<string, string>;
340-
chatAbortedRuns: Map<string, number>;
341+
chatAbortedRuns: Map<string, ChatAbortMarker>;
341342
clearChatRunState: (runId: string) => void;
342343
removeChatRun: (
343344
sessionId: string,
@@ -469,7 +470,7 @@ export function abortChatRunById(
469470

470471
const bufferedText = ops.chatRunBuffers.get(runId);
471472
const partialText = bufferedText && bufferedText.trim() ? bufferedText : undefined;
472-
ops.chatAbortedRuns.set(runId, Date.now());
473+
ops.chatAbortedRuns.set(runId, createChatAbortMarker());
473474
if (stopReason) {
474475
active.abortStopReason = stopReason;
475476
}

‎src/gateway/local-request-context.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import {
1111
} from "../plugins/runtime/gateway-request-scope.js";
1212
import { NodeRegistry } from "./node-registry.js";
1313
import type { ChannelRuntimeSnapshot } from "./server-channel-runtime.types.js";
14+
import { createChatRunEntry, type ChatRunEntry } from "./server-chat-state.js";
1415
import type { GatewayRequestContext } from "./server-methods/types.js";
1516

1617
// Embedded/local agent calls need enough GatewayRequestContext to reuse server
@@ -52,7 +53,7 @@ export function createLocalGatewayRequestContext(
5253
): GatewayRequestContext {
5354
const logGateway = createSubsystemLogger("gateway/local");
5455
const sessionEvents = new Set<string>();
55-
const chatRuns = new Map<string, { sessionKey: string; agentId?: string; clientRunId: string }>();
56+
const chatRuns = new Map<string, ChatRunEntry>();
5657
const chatRunBuffers: GatewayRequestContext["chatRunBuffers"] = new Map();
5758
const chatDeltaSentAt: GatewayRequestContext["chatDeltaSentAt"] = new Map();
5859
const chatDeltaLastBroadcastLen: GatewayRequestContext["chatDeltaLastBroadcastLen"] = new Map();
@@ -105,7 +106,7 @@ export function createLocalGatewayRequestContext(
105106
bufferedAgentEvents,
106107
clearChatRunState,
107108
addChatRun: (sessionId, entry) => {
108-
chatRuns.set(sessionId, entry);
109+
chatRuns.set(sessionId, createChatRunEntry(entry));
109110
},
110111
removeChatRun: (sessionId, clientRunId, sessionKey) => {
111112
const entry = chatRuns.get(sessionId);

‎src/gateway/server-chat-state.ts‎

Lines changed: 66 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,21 +10,79 @@ export type ChatRunTiming = {
1010
receivedAtMs: number;
1111
};
1212

13-
export type ChatRunEntry = {
13+
export type ChatRunRegistration = {
1414
sessionKey: string;
1515
agentId?: string;
1616
clientRunId: string;
1717
chatSendTiming?: ChatRunTiming;
1818
};
1919

20+
export type ChatRunEntry = ChatRunRegistration & {
21+
registeredAtMs: number;
22+
registeredSequence: number;
23+
};
24+
25+
export type ChatAbortMarker = number | { abortedAtMs: number; sequence: number };
26+
27+
let chatRunOrderingSequence = 0;
28+
29+
function nextChatRunOrderingSequence(): number {
30+
chatRunOrderingSequence += 1;
31+
return chatRunOrderingSequence;
32+
}
33+
34+
/** Stamp a chat run registration with the process-local ordering metadata used for abort freshness checks. */
35+
export function createChatRunEntry(entry: ChatRunRegistration): ChatRunEntry {
36+
return {
37+
...entry,
38+
registeredAtMs: Date.now(),
39+
registeredSequence: nextChatRunOrderingSequence(),
40+
};
41+
}
42+
43+
/** Create an abort marker ordered against chat run registrations, using a shared monotonic sequence. */
44+
export function createChatAbortMarker(now = Date.now()): ChatAbortMarker {
45+
return { abortedAtMs: now, sequence: nextChatRunOrderingSequence() };
46+
}
47+
48+
/** Return the wall-clock timestamp used by maintenance TTL pruning for both legacy and structured markers. */
49+
export function chatAbortMarkerTimestampMs(marker: ChatAbortMarker): number {
50+
return typeof marker === "number" ? marker : marker.abortedAtMs;
51+
}
52+
53+
/**
54+
* Return whether an abort marker should suppress events for the given chat run registration.
55+
* Structured markers compare the monotonic sequence first so same-millisecond aborts stay ordered;
56+
* legacy numeric markers fall back to timestamp comparison, and a missing entry preserves old suppress-on-presence behavior.
57+
*/
58+
export function isChatAbortMarkerCurrent(
59+
marker: ChatAbortMarker | undefined,
60+
entry?: Pick<ChatRunEntry, "registeredAtMs" | "registeredSequence">,
61+
): boolean {
62+
if (marker === undefined) {
63+
return false;
64+
}
65+
if (!entry) {
66+
return true;
67+
}
68+
if (typeof marker !== "number" && typeof entry.registeredSequence === "number") {
69+
return marker.sequence >= entry.registeredSequence;
70+
}
71+
if (typeof entry.registeredAtMs !== "number") {
72+
return true;
73+
}
74+
const abortedAtMs = typeof marker === "number" ? marker : marker.abortedAtMs;
75+
return abortedAtMs >= entry.registeredAtMs;
76+
}
77+
2078
export type BufferedAgentEvent = {
2179
sessionKey?: string;
2280
agentId?: string;
2381
payload: AgentEventPayload & { spawnedBy?: string };
2482
};
2583

2684
export type ChatRunRegistry = {
27-
add: (sessionId: string, entry: ChatRunEntry) => void;
85+
add: (sessionId: string, entry: ChatRunRegistration) => void;
2886
peek: (sessionId: string) => ChatRunEntry | undefined;
2987
shift: (sessionId: string) => ChatRunEntry | undefined;
3088
remove: (sessionId: string, clientRunId: string, sessionKey?: string) => ChatRunEntry | undefined;
@@ -35,12 +93,13 @@ export type ChatRunRegistry = {
3593
export function createChatRunRegistry(): ChatRunRegistry {
3694
const chatRunSessions = new Map<string, ChatRunEntry[]>();
3795

38-
const add = (sessionId: string, entry: ChatRunEntry) => {
96+
const add = (sessionId: string, entry: ChatRunRegistration) => {
97+
const registeredEntry = createChatRunEntry(entry);
3998
const queue = chatRunSessions.get(sessionId);
4099
if (queue) {
41-
queue.push(entry);
100+
queue.push(registeredEntry);
42101
} else {
43-
chatRunSessions.set(sessionId, [entry]);
102+
chatRunSessions.set(sessionId, [registeredEntry]);
44103
}
45104
};
46105

@@ -96,7 +155,7 @@ export type ChatRunState = {
96155
deltaLastBroadcastText: Map<string, string>;
97156
agentDeltaSentAt: Map<string, number>;
98157
bufferedAgentEvents: Map<string, BufferedAgentEvent>;
99-
abortedRuns: Map<string, number>;
158+
abortedRuns: Map<string, ChatAbortMarker>;
100159
clearRun: (runId: string) => void;
101160
clear: () => void;
102161
};
@@ -112,7 +171,7 @@ export function createChatRunState(): ChatRunState {
112171
const deltaLastBroadcastText = new Map<string, string>();
113172
const agentDeltaSentAt = new Map<string, number>();
114173
const bufferedAgentEvents = new Map<string, BufferedAgentEvent>();
115-
const abortedRuns = new Map<string, number>();
174+
const abortedRuns = new Map<string, ChatAbortMarker>();
116175

117176
const clearRun = (runId: string) => {
118177
rawBuffers.delete(runId);

‎src/gateway/server-chat.agent-events.test.ts‎

Lines changed: 79 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ import {
4444
createAgentEventHandler,
4545
createChatRunState,
4646
createSessionEventSubscriberRegistry,
47+
createChatAbortMarker,
4748
createSessionMessageSubscriberRegistry,
4849
createToolEventRecipientRegistry,
4950
type AgentEventHandlerOptions,
@@ -2762,7 +2763,7 @@ describe("agent event handler", () => {
27622763
sessionKey: "session-aborted",
27632764
clientRunId: "client-aborted",
27642765
});
2765-
chatRunState.abortedRuns.set("client-aborted", 1_000);
2766+
chatRunState.abortedRuns.set("client-aborted", createChatAbortMarker());
27662767

27672768
handler({
27682769
runId: "run-aborted",
@@ -2777,6 +2778,81 @@ describe("agent event handler", () => {
27772778
expect(chatBroadcastCalls(broadcast)).toHaveLength(0);
27782779
});
27792780

2781+
it.each([
2782+
{
2783+
name: "older timestamp",
2784+
marker: () => 1_000,
2785+
},
2786+
{
2787+
name: "same-millisecond older sequence",
2788+
marker: () => ({ abortedAtMs: 2_000, sequence: -1 }),
2789+
},
2790+
])(
2791+
"ignores stale aborted markers from older same-key runs for fresh chat lifecycle events ($name)",
2792+
({ marker }) => {
2793+
const { broadcast, nodeSendToSession, chatRunState, handler } = createHarness({ now: 2_000 });
2794+
chatRunState.abortedRuns.set("client-stale-abort", marker());
2795+
chatRunState.registry.add("run-stale-abort", {
2796+
sessionKey: "session-stale-abort",
2797+
clientRunId: "client-stale-abort",
2798+
});
2799+
2800+
handler({
2801+
runId: "run-stale-abort",
2802+
seq: 1,
2803+
stream: "assistant",
2804+
ts: 2_100,
2805+
data: { text: "Fresh output", delta: "Fresh output" },
2806+
});
2807+
handler({
2808+
runId: "run-stale-abort",
2809+
seq: 2,
2810+
stream: "lifecycle",
2811+
ts: 2_200,
2812+
data: { phase: "end" },
2813+
});
2814+
2815+
const chatCalls = chatBroadcastCalls(broadcast);
2816+
expect(chatCalls).toHaveLength(2);
2817+
const deltaPayload = chatCalls[0][1];
2818+
const finalPayload = chatCalls[1][1];
2819+
expect(deltaPayload.state).toBe("delta");
2820+
expect(finalPayload.state).toBe("final");
2821+
expect(sessionChatCalls(nodeSendToSession)).toHaveLength(2);
2822+
expect(chatRunState.abortedRuns.has("client-stale-abort")).toBe(true);
2823+
expect(chatRunState.registry.peek("run-stale-abort")).toBeUndefined();
2824+
},
2825+
);
2826+
2827+
it("honors same-millisecond abort markers from the current same-key run", () => {
2828+
const { broadcast, nodeSendToSession, chatRunState, handler } = createHarness({ now: 3_000 });
2829+
chatRunState.registry.add("run-current-abort", {
2830+
sessionKey: "session-current-abort",
2831+
clientRunId: "client-current-abort",
2832+
});
2833+
chatRunState.abortedRuns.set("client-current-abort", createChatAbortMarker());
2834+
2835+
handler({
2836+
runId: "run-current-abort",
2837+
seq: 1,
2838+
stream: "assistant",
2839+
ts: 3_100,
2840+
data: { text: "Suppressed output", delta: "Suppressed output" },
2841+
});
2842+
handler({
2843+
runId: "run-current-abort",
2844+
seq: 2,
2845+
stream: "lifecycle",
2846+
ts: 3_200,
2847+
data: { phase: "end", aborted: true, stopReason: "rpc" },
2848+
});
2849+
2850+
expect(chatBroadcastCalls(broadcast)).toHaveLength(0);
2851+
expect(sessionChatCalls(nodeSendToSession)).toHaveLength(0);
2852+
expect(chatRunState.abortedRuns.has("client-current-abort")).toBe(true);
2853+
expect(chatRunState.registry.peek("run-current-abort")).toBeUndefined();
2854+
});
2855+
27802856
it("keeps live session setting metadata at the top level for lifecycle updates", async () => {
27812857
vi.mocked(loadGatewaySessionRow).mockReturnValue({
27822858
key: "session-finished",
@@ -3166,7 +3242,7 @@ describe("agent event handler", () => {
31663242
data: { phase: "error", error: "provider failed" },
31673243
});
31683244

3169-
expect(chatRunState.registry.peek("run-fallback-retry")).toEqual({
3245+
expect(chatRunState.registry.peek("run-fallback-retry")).toMatchObject({
31703246
sessionKey: "session-fallback",
31713247
clientRunId: "run-fallback-client",
31723248
});
@@ -3189,7 +3265,7 @@ describe("agent event handler", () => {
31893265

31903266
vi.advanceTimersByTime(100);
31913267

3192-
expect(chatRunState.registry.peek("run-fallback-retry")).toEqual({
3268+
expect(chatRunState.registry.peek("run-fallback-retry")).toMatchObject({
31933269
sessionKey: "session-fallback",
31943270
clientRunId: "run-fallback-client",
31953271
});

‎src/gateway/server-chat.ts‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
resolveMergedAssistantText,
1919
shouldSuppressAssistantEventForLiveChat,
2020
} from "./live-chat-projector.js";
21+
import { isChatAbortMarkerCurrent } from "./server-chat-state.js";
2122
import type {
2223
BufferedAgentEvent,
2324
ChatRunEntry,
@@ -37,15 +38,18 @@ import { loadSessionEntry } from "./session-utils.js";
3738
import { formatForLog } from "./ws-log.js";
3839

3940
export {
41+
createChatAbortMarker,
4042
createChatRunRegistry,
4143
createChatRunState,
4244
createSessionEventSubscriberRegistry,
4345
createSessionMessageSubscriberRegistry,
4446
createToolEventRecipientRegistry,
4547
} from "./server-chat-state.js";
4648
export type {
49+
ChatAbortMarker,
4750
ChatRunEntry,
4851
ChatRunRegistry,
52+
ChatRunRegistration,
4953
ChatRunState,
5054
SessionEventSubscriberRegistry,
5155
SessionMessageSubscriberRegistry,
@@ -562,7 +566,8 @@ export function createAgentEventHandler({
562566
const clientRunId = chatLink?.clientRunId ?? evt.runId;
563567
const eventRunId = chatLink?.clientRunId ?? evt.runId;
564568
const isAborted =
565-
chatRunState.abortedRuns.has(clientRunId) || chatRunState.abortedRuns.has(evt.runId);
569+
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(clientRunId), chatLink) ||
570+
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(evt.runId), chatLink);
566571
const deliverySessionKey = sessionKey
567572
? resolveSessionDeliveryKey(sessionKey, sessionAgentId)
568573
: undefined;
@@ -1148,7 +1153,9 @@ export function createAgentEventHandler({
11481153
const eventRunId = chatLink?.clientRunId ?? evt.runId;
11491154
const eventForClients = chatLink ? { ...evt, runId: eventRunId } : evt;
11501155
const isAborted =
1151-
chatRunState.abortedRuns.has(clientRunId) || chatRunState.abortedRuns.has(evt.runId);
1156+
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(clientRunId), chatLink) ||
1157+
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(evt.runId), chatLink);
1158+
11521159
const restartRecoveryState = restartRecoverySessionKey
11531160
? resolveRestartRecoveryLifecycleState(restartRecoverySessionKey, restartRecoveryAgentId, evt)
11541161
: undefined;
@@ -1173,6 +1180,7 @@ export function createAgentEventHandler({
11731180
if (lifecyclePhase !== null && lifecyclePhase !== "error") {
11741181
clearPendingTerminalLifecycleError(evt.runId);
11751182
}
1183+
11761184
// Include sessionKey so Control UI can filter tool streams per session.
11771185
const spawnedBy = sessionKey ? resolveSpawnedBy(sessionKey) : null;
11781186
const agentPayload = sessionKey

0 commit comments

Comments
 (0)