furyhawk/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-21 10:16:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve hidden chat run boundaries

Browse Source
This commit is contained in:
Shakker
2026-07-15 04:38:01 +01:00
committed by Shakker
parent a3f06fb119
commit 4755dee240
8 changed files with 264 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gateway/chat-display-projection.ts
+82 -18
View File
@@ -36,6 +36,17 @@ type RoleContentMessage = {
content?: unknown;
};
type ChatDisplayProjectionOptions = {
maxChars?: number;
stripEnvelope?: boolean;
turnBoundaryPending?: boolean;
};
type ChatDisplayProjectionResult = {
messages: Array<Record<string, unknown>>;
turnBoundaryPending: boolean;
};
type PendingMessageToolVisibleReply = {
toolCallId?: string;
text: string;
@@ -1597,6 +1608,34 @@ function shouldHideProjectedHistoryMessage(message: Record<string, unknown>): bo
return isHeartbeatOkResponse(roleContent);
}
/** Identifies the hidden native input that starts a heartbeat-driven turn. */
export function isHeartbeatHistoryTurnBoundaryMessage(message: unknown): boolean {
const record = readRecord(message);
if (!record || isSessionsSendInterSessionUserMessage(record)) {
return false;
}
const roleContent = asRoleContentMessage(record);
return roleContent?.role === "user" && isHeartbeatUserMessage(roleContent, HEARTBEAT_PROMPT);
}
function attachProjectedTurnBoundary(message: Record<string, unknown>): Record<string, unknown> {
const metadata = readRecord(message["__openclaw"]);
if (metadata?.turnBoundary === true) {
return message;
}
return {
...message,
__openclaw: {
...metadata,
turnBoundary: true,
},
};
}
function canCarryProjectedTurnBoundary(message: RoleContentMessage | null): boolean {
return Boolean(message && message.role !== "system" && message.role !== "custom");
}
function openclawAssistantModel(message: Record<string, unknown>): string | undefined {
return message.role === "assistant" &&
message.provider === "openclaw" &&
@@ -1672,10 +1711,15 @@ function toProjectedMessages(messages: unknown[]): Array<Record<string, unknown>
function filterVisibleProjectedHistoryMessages(
messages: Array<Record<string, unknown>>,
): Array<Record<string, unknown>> {
turnBoundaryPending = false,
): {
messages: Array<Record<string, unknown>>;
turnBoundaryPending: boolean;
} {
if (messages.length === 0) {
return messages;
return { messages, turnBoundaryPending };
}
let pendingTurnBoundary = turnBoundaryPending;
let changed = false;
const visible: Array<Record<string, unknown>> = [];
for (let i = 0; i < messages.length; i++) {
@@ -1695,11 +1739,13 @@ function filterVisibleProjectedHistoryMessages(
!isProjectedSessionsSendForwardedMessage(next)
) {
changed = true;
pendingTurnBoundary = true;
i++;
continue;
}
if (shouldHideProjectedHistoryMessage(current)) {
changed = true;
pendingTurnBoundary ||= isHeartbeatHistoryTurnBoundaryMessage(current);
continue;
}
if (
@@ -1709,9 +1755,18 @@ function filterVisibleProjectedHistoryMessages(
changed = true;
continue;
}
visible.push(current);
if (pendingTurnBoundary && canCarryProjectedTurnBoundary(currentRoleContent)) {
visible.push(attachProjectedTurnBoundary(current));
pendingTurnBoundary = false;
changed = true;
} else {
visible.push(current);
}
}
return changed ? visible : messages;
return {
messages: changed ? visible : messages,
turnBoundaryPending: pendingTurnBoundary,
};
}
function stripInterSessionPromptPrefixFromContent(content: unknown): unknown {
@@ -1882,24 +1937,33 @@ function projectEmptyAssistantErrorMessages(
return changed ? projected : messages;
}
export function projectChatDisplayMessages(
export function projectChatDisplayMessagesWithState(
messages: unknown[],
options?: { maxChars?: number; stripEnvelope?: boolean },
): Array<Record<string, unknown>> {
options?: ChatDisplayProjectionOptions,
): ChatDisplayProjectionResult {
const source = options?.stripEnvelope === false ? messages : stripEnvelopeFromMessages(messages);
const mirrored = mirrorMessageToolVisibleReplies(source);
const projectedErrors = projectEmptyAssistantErrorMessages(toProjectedMessages(mirrored));
const projectedForwarded = mergeTtsSupplementMessages(
filterVisibleProjectedHistoryMessages(
projectSessionsSendInterSessionMessages(
toProjectedMessages(sanitizeChatHistoryMessages(projectedErrors, Number.MAX_SAFE_INTEGER)),
),
const filtered = filterVisibleProjectedHistoryMessages(
projectSessionsSendInterSessionMessages(
toProjectedMessages(sanitizeChatHistoryMessages(projectedErrors, Number.MAX_SAFE_INTEGER)),
),
options?.turnBoundaryPending,
);
return sanitizeChatHistoryMessages(
projectedForwarded,
options?.maxChars ?? DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS,
) as Array<Record<string, unknown>>;
return {
messages: sanitizeChatHistoryMessages(
mergeTtsSupplementMessages(filtered.messages),
options?.maxChars ?? DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS,
) as Array<Record<string, unknown>>,
turnBoundaryPending: filtered.turnBoundaryPending,
};
}
export function projectChatDisplayMessages(
messages: unknown[],
options?: ChatDisplayProjectionOptions,
): Array<Record<string, unknown>> {
return projectChatDisplayMessagesWithState(messages, options).messages;
}
function limitChatDisplayMessages<T>(messages: T[], maxMessages?: number): T[] {
@@ -1916,7 +1980,7 @@ function limitChatDisplayMessages<T>(messages: T[], maxMessages?: number): T[] {
export function projectRecentChatDisplayMessages(
messages: unknown[],
options?: { maxChars?: number; maxMessages?: number; stripEnvelope?: boolean },
options?: ChatDisplayProjectionOptions & { maxMessages?: number },
): Array<Record<string, unknown>> {
return limitChatDisplayMessages(
projectChatDisplayMessages(messages, options),
@@ -1926,7 +1990,7 @@ export function projectRecentChatDisplayMessages(
export function projectChatDisplayMessage(
message: unknown,
options?: { maxChars?: number; stripEnvelope?: boolean },
options?: ChatDisplayProjectionOptions,
): Record<string, unknown> | undefined {
return projectChatDisplayMessages([message], options)[0];
}
src/gateway/server-methods/chat-history-budget.test.ts
+5 -1
View File
@@ -43,11 +43,15 @@ describe("enforceChatHistoryFinalBudget", () => {
role: "assistant",
timestamp: 1,
content: [{ type: "text", text: "y".repeat(4000) }],
__openclaw: { id: "abc", seq: 7 },
__openclaw: { id: "abc", seq: 7, turnBoundary: true },
};
const result = enforceChatHistoryFinalBudget({ messages: [last], maxBytes: 2_000 });
expect(result.messages).toHaveLength(1);
expect(firstText(result.messages)).toContain("chat.history omitted: message too large");
expect(
(result.messages[0] as { __openclaw?: { turnBoundary?: boolean } })["__openclaw"]
?.turnBoundary,
).toBe(true);
// The placeholder is a new object, not the oversized original.
expect(result.messages[0]).not.toBe(last);
});
src/gateway/server-methods/chat-history-budget.ts
+2
View File
@@ -40,6 +40,7 @@ function buildOversizedHistoryPlaceholder(message?: unknown): Record<string, unk
const metadataSeq = typeof metadata.seq === "number" ? metadata.seq : undefined;
const metadataIdempotencyKey =
typeof metadata.idempotencyKey === "string" ? metadata.idempotencyKey : undefined;
const turnBoundary = metadata.turnBoundary === true;
return {
role,
timestamp,
@@ -48,6 +49,7 @@ function buildOversizedHistoryPlaceholder(message?: unknown): Record<string, unk
...(metadataId ? { id: metadataId } : {}),
...(metadataSeq !== undefined ? { seq: metadataSeq } : {}),
...(metadataIdempotencyKey ? { idempotencyKey: metadataIdempotencyKey } : {}),
...(turnBoundary ? { turnBoundary: true } : {}),
truncated: true,
reason: "oversized",
},
src/gateway/server-methods/chat-history-pages.ts
+5
View File
@@ -1,6 +1,7 @@
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
import {
dropPreSessionStartAnnouncePairs,
isHeartbeatHistoryTurnBoundaryMessage,
projectChatDisplayMessages,
projectRecentChatDisplayMessages,
} from "../chat-display-projection.js";
@@ -303,9 +304,11 @@ export async function readChatHistoryPage(params: {
? projectRecentChatDisplayMessages(recencyFilteredMessages, {
maxChars: effectiveMaxChars,
maxMessages: max,
turnBoundaryPending: isHeartbeatHistoryTurnBoundaryMessage(overreadContextMessage),
})
: projectChatDisplayMessages(recencyFilteredMessages, {
maxChars: effectiveMaxChars,
turnBoundaryPending: isHeartbeatHistoryTurnBoundaryMessage(overreadContextMessage),
});
const windowed = messageId
? (capChatHistoryAroundMessage({
@@ -344,6 +347,7 @@ export async function readChatHistoryPage(params: {
});
const overreadContextMessage =
readPage.messages.length > rawHistoryWindow.maxMessages ? readPage.messages[0] : undefined;
const turnBoundaryPending = isHeartbeatHistoryTurnBoundaryMessage(overreadContextMessage);
const localMessagesWithBoundaryFilter = dropLocalHistoryOverreadContextMessage(
dropPreSessionStartAnnouncePairs(
readPage.messages,
@@ -413,6 +417,7 @@ export async function readChatHistoryPage(params: {
const displayMessages = projectRecentChatDisplayMessages(recencyFilteredMessages, {
maxChars: effectiveMaxChars,
maxMessages: max,
turnBoundaryPending,
});
return {
messages: augmentChatHistoryWithCanvasBlocks(displayMessages),
src/gateway/server-methods/server-methods.test.ts
+60
View File
@@ -1648,11 +1648,71 @@ describe("projectRecentChatDisplayMessages", () => {
sourceSessionKey: "agent:main:webchat:source",
sourceTool: "sessions_send",
},
__openclaw: { turnBoundary: true },
timestamp: 2,
},
]);
});
it("marks only the first visible message after each hidden heartbeat input", () => {
const result = projectRecentChatDisplayMessages([
{
role: "user",
content: [{ type: "text", text: HEARTBEAT_PROMPT }],
__openclaw: { seq: 1 },
},
{
role: "assistant",
content: [{ type: "text", text: "First run started." }],
__openclaw: { seq: 2 },
},
{
role: "assistant",
content: [{ type: "text", text: "First run finished." }],
__openclaw: { seq: 3 },
},
{
role: "user",
content: [{ type: "text", text: HEARTBEAT_PROMPT }],
__openclaw: { seq: 4 },
},
{
role: "system",
content: [{ type: "text", text: "Compaction" }],
__openclaw: { kind: "compaction", seq: 5 },
},
{
role: "assistant",
content: [{ type: "text", text: "Second run finished." }],
__openclaw: { seq: 6 },
},
]);
expect(
result.map((message) => ({
text: (message.content as Array<{ text?: string }> | undefined)?.[0]?.text,
metadata: message["__openclaw"],
})),
).toEqual([
{
text: "First run started.",
metadata: { seq: 2, turnBoundary: true },
},
{
text: "First run finished.",
metadata: { seq: 3 },
},
{
text: "Compaction",
metadata: { kind: "compaction", seq: 5 },
},
{
text: "Second run finished.",
metadata: { seq: 6, turnBoundary: true },
},
]);
});
it("does not project user-authored sessions_send envelope text without provenance", () => {
const result = projectRecentChatDisplayMessages([
{
src/gateway/server.chat.gateway-server-chat-b.test.ts
+52
View File
@@ -7,6 +7,7 @@ import { expectDefined } from "@openclaw/normalization-core";
import { afterAll, afterEach, beforeAll, describe, expect, test, vi } from "vitest";
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
import type { GetReplyOptions } from "../auto-reply/get-reply-options.types.js";
import { HEARTBEAT_PROMPT } from "../auto-reply/heartbeat.js";
import type { InternalGetReplyOptions } from "../auto-reply/reply/get-reply.types.js";
import { clearConfigCache, getRuntimeConfig } from "../config/config.js";
import { resolveSessionRoutingContract } from "../config/sessions/main-session.js";
@@ -4968,6 +4969,57 @@ describe("gateway server chat", () => {
});
});
test("chat.history offset pages preserve a hidden heartbeat boundary from overread context", async () => {
await withGatewayChatHarness(async ({ ws, createSessionDir }) => {
await connectOk(ws);
const sessionDir = await createSessionDir();
await writeSessionStore({
entries: {
main: {
sessionId: "sess-main",
updatedAt: Date.now(),
},
},
});
await writeMainSessionTranscript(sessionDir, [
JSON.stringify({
message: {
role: "user",
content: [{ type: "text", text: HEARTBEAT_PROMPT }],
},
}),
JSON.stringify({
message: {
role: "assistant",
content: [{ type: "text", text: "heartbeat run output" }],
},
}),
JSON.stringify({
message: {
role: "assistant",
content: [{ type: "text", text: "newest output" }],
},
}),
]);
const page = await rpcReq<{
messages?: Array<{
content?: Array<{ text?: string }>;
__openclaw?: { turnBoundary?: boolean };
}>;
}>(ws, "chat.history", {
sessionKey: "main",
limit: 1,
offset: 1,
});
expect(page.ok).toBe(true);
expect(page.payload?.messages).toHaveLength(1);
expect(page.payload?.messages?.[0]?.content?.[0]?.text).toBe("heartbeat run output");
expect(page.payload?.messages?.[0]?.["__openclaw"]?.turnBoundary).toBe(true);
});
});
test("chat.send does not force-disable block streaming", async () => {
await withGatewayChatHarness(async ({ ws, createSessionDir }) => {
const spy = getReplyFromConfig;
src/gateway/session-history-state.test.ts
+34 -1
View File
@@ -588,10 +588,43 @@ describe("SessionHistorySseState", () => {
],
});
expectOnlyAssistantText(snapshot, "Disk usage crossed 95 percent.", 4);
expect(snapshot.history.messages).toEqual([
{
...assistantTextMessage("Disk usage crossed 95 percent.", 4),
__openclaw: { seq: 4, turnBoundary: true },
},
]);
expect(snapshot.rawTranscriptSeq).toBe(4);
});
test("carries a hidden heartbeat boundary into the next visible SSE append", () => {
const state = newState([
assistantTextMessage("already visible", 1),
{
role: "user",
content: HEARTBEAT_PROMPT,
__openclaw: { seq: 2 },
},
]);
expect(appendAssistantText(state, "HEARTBEAT_OK", 3)).toBeNull();
const compaction = state.appendInlineMessage({
message: {
role: "system",
content: textContent("Compaction summary"),
},
messageSeq: 4,
});
expect(compaction?.message["__openclaw"]?.turnBoundary).toBeUndefined();
const appended = appendAssistantText(state, "Disk usage crossed 95 percent.", 5);
expect(appended?.message).toMatchObject({
role: "assistant",
__openclaw: { seq: 5, turnBoundary: true },
});
});
test("does not append heartbeat or internal-only SSE messages", () => {
const state = newState([assistantTextMessage("already visible", 1)]);
src/gateway/session-history-state.ts
+24 -8
View File
@@ -4,6 +4,7 @@ import { asPositiveSafeInteger } from "@openclaw/normalization-core/number-coerc
import {
DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS,
projectChatDisplayMessages,
projectChatDisplayMessagesWithState,
} from "./chat-display-projection.js";
import { resolveTranscriptPathForComparison } from "./session-transcript-path.js";
import {
@@ -34,6 +35,7 @@ type PaginatedSessionHistory = {
type SessionHistorySnapshot = {
history: PaginatedSessionHistory;
rawTranscriptSeq: number;
turnBoundaryPending: boolean;
};
type InlineSessionHistoryAppend = {
@@ -158,11 +160,10 @@ export function buildSessionHistorySnapshot(params: {
rawTranscriptSeq?: number;
totalRawMessages?: number;
}): SessionHistorySnapshot {
const visibleMessages = toSessionHistoryMessages(
projectChatDisplayMessages(params.rawMessages, {
maxChars: params.maxChars ?? DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS,
}),
);
const projected = projectChatDisplayMessagesWithState(params.rawMessages, {
maxChars: params.maxChars ?? DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS,
});
const visibleMessages = toSessionHistoryMessages(projected.messages);
const history = paginateSessionMessages(visibleMessages, params.limit, params.cursor);
if (
!params.cursor &&
@@ -183,6 +184,7 @@ export function buildSessionHistorySnapshot(params: {
params.rawTranscriptSeq ??
resolveMessageSeq(rawHistoryMessages.at(-1)) ??
rawHistoryMessages.length,
turnBoundaryPending: projected.turnBoundaryPending,
};
}
@@ -194,6 +196,7 @@ export class SessionHistorySseState {
private readonly cursor: string | undefined;
private sentHistory: PaginatedSessionHistory;
private rawTranscriptSeq: number;
private turnBoundaryPending: boolean;
private transcriptPath: string | undefined;
static fromRawSnapshot(params: {
@@ -243,6 +246,7 @@ export class SessionHistorySseState {
});
this.sentHistory = snapshot.history;
this.rawTranscriptSeq = snapshot.rawTranscriptSeq;
this.turnBoundaryPending = snapshot.turnBoundaryPending;
this.transcriptPath = normalizeTranscriptPathForComparison(params.transcriptPath);
}
@@ -273,6 +277,12 @@ export class SessionHistorySseState {
...(idempotencyKey ? { idempotencyKey } : {}),
seq: this.rawTranscriptSeq,
});
const hadPendingTurnBoundary = this.turnBoundaryPending;
const nextProjection = projectChatDisplayMessagesWithState([nextMessage], {
maxChars: this.maxChars,
turnBoundaryPending: hadPendingTurnBoundary,
});
this.turnBoundaryPending = nextProjection.turnBoundaryPending;
// Projection can split, drop, or rewrite raw transcript messages. When one
// raw append changes multiple visible rows, callers must refresh instead of
// emitting a misleading single SSE item.
@@ -283,6 +293,13 @@ export class SessionHistorySseState {
);
if (projectedMessages.length > this.sentHistory.messages.length) {
const addedMessages = projectedMessages.slice(this.sentHistory.messages.length);
if (hadPendingTurnBoundary && !this.turnBoundaryPending && addedMessages[0]) {
const firstAdded = attachOpenClawTranscriptMeta(addedMessages[0], {
turnBoundary: true,
}) as SessionHistoryMessage;
addedMessages[0] = firstAdded;
projectedMessages[this.sentHistory.messages.length] = firstAdded;
}
if (addedMessages.length > 1) {
this.sentHistory = buildPaginatedSessionHistory({
messages: projectedMessages,
@@ -310,9 +327,7 @@ export class SessionHistorySseState {
};
}
}
const [sanitizedMessage] = toSessionHistoryMessages(
projectChatDisplayMessages([nextMessage], { maxChars: this.maxChars }),
);
const [sanitizedMessage] = toSessionHistoryMessages(nextProjection.messages);
if (!sanitizedMessage) {
if (projectedMessages.length < this.sentHistory.messages.length) {
this.sentHistory = buildPaginatedSessionHistory({
@@ -351,6 +366,7 @@ export class SessionHistorySseState {
const rawSnapshot = await this.readRawSnapshotAsync();
const snapshot = this.buildSnapshot(rawSnapshot);
this.rawTranscriptSeq = snapshot.rawTranscriptSeq;
this.turnBoundaryPending = snapshot.turnBoundaryPending;
this.transcriptPath = normalizeTranscriptPathForComparison(rawSnapshot.transcriptPath);
this.sentHistory = snapshot.history;
return snapshot.history;