fix: preserve steering stream boundaries

This commit is contained in:
Shakker
2026-08-21 23:04:35 +01:00
parent df8898fe26
commit 7f5ae9249f
3 changed files with 39 additions and 4 deletions
@@ -294,6 +294,22 @@ suite.define(() => {
sessionKey: "main",
state: "delta",
});
const streamingBubble = page.locator(".chat-bubble.streaming", {
hasText: "Terminal response paragraph 1.",
});
await streamingBubble.waitFor();
const streamingRow = streamingBubble.locator(
"xpath=ancestor::div[contains(@class, 'chat-virtual-row')]",
);
await streamingRow.waitFor();
const steerBubble = page.locator(".chat-group.user", { hasText: steerText }).last();
const [steerBounds, streamingBounds] = await Promise.all([
steerBubble.boundingBox(),
streamingBubble.boundingBox(),
]);
expect(steerBounds).not.toBeNull();
expect(streamingBounds).not.toBeNull();
expect(streamingBounds!.y).toBeGreaterThanOrEqual(steerBounds!.y + steerBounds!.height - 1);
await gateway.emitGatewayEvent("session.message", {
activeRunIds: [],
clientRunId: runId,
+12 -1
View File
@@ -24,7 +24,7 @@ import {
} from "./chat-state-refresh.ts";
import { resolveChatAvatarUrl, selectedChatSessionRow } from "./chat-state-route.ts";
import { buildChatItems } from "./chat-thread-build.ts";
import { getChatSessionProjection } from "./history-merge.ts";
import { getChatSessionProjection, reduceChatSessionProjection } from "./history-merge.ts";
import { scheduleControlUiAfterPaint } from "./performance.ts";
import { applySessionMessagePayload } from "./session-message-apply.ts";
@@ -269,6 +269,16 @@ describe("canonical session message recovery", () => {
]);
expect(state.chatRunId).toBe(activeRunId);
expect(state.chatQueue).toEqual([]);
reduceChatSessionProjection(state, {
type: "sendPending",
runId: steerRunId,
message: {
role: "user",
content: [{ type: "text", text: "Steer prompt" }],
timestamp: 50,
__openclaw: { idempotencyKey: `${steerRunId}:user` },
},
});
state.chatRunId = steerRunId;
const steerEvent = {
@@ -296,6 +306,7 @@ describe("canonical session message recovery", () => {
handlePageGatewayEvent(state, steerEvent);
expect(state.chatRunId).toBe(activeRunId);
const segmentsAfterRequestBoundary = state.chatStreamSegments;
expect(segmentsAfterRequestBoundary.at(-1)?.boundaryRunId).toBe(steerRunId);
expect(state.chatStreamSegments).toBe(segmentsAfterRequestBoundary);
expect(
state.chatMessages.filter((message) => extractText(message) === "Steer prompt"),
+11 -3
View File
@@ -12,7 +12,12 @@ import {
readChatSessionProjectionScope,
reduceChatSessionProjection,
} from "./history-merge.ts";
import { persistedSteerTargetRunId, rolloverChatStream } from "./stream-causal-boundary.ts";
import {
latestPersistedSteerBoundary,
latestStreamBoundaryRunId,
persistedSteerTargetRunId,
rolloverChatStream,
} from "./stream-causal-boundary.ts";
import { maybeResetToolStream } from "./stream-reconciliation.ts";
import { prunePersistedAssistantStreamSegments } from "./stream-segment-pruning.ts";
@@ -121,7 +126,6 @@ export function applySessionMessagePayload(
...(incoming.sequence !== null ? { seq: incoming.sequence } : {}),
},
};
const previousMessageCount = state.chatMessages.length;
const projection = reduceChatSessionProjection(
state,
{
@@ -141,13 +145,17 @@ export function applySessionMessagePayload(
}
const steerTargetRunId = persistedSteerTargetRunId(message);
const currentRunId = state.chatRunId;
const persistedSteerBoundary = steerTargetRunId
? latestPersistedSteerBoundary(projection.messages, steerTargetRunId)
: null;
if (
incoming.role === "user" &&
runActive === true &&
incoming.runId &&
steerTargetRunId &&
(!currentRunId || currentRunId === steerTargetRunId || currentRunId === incoming.runId) &&
projection.messages.length > previousMessageCount
persistedSteerBoundary?.runId === incoming.runId &&
latestStreamBoundaryRunId(state) !== incoming.runId
) {
state.chatRunId = steerTargetRunId;
rolloverChatStream(state, {