fix(ui): keep one agent run in one transcript response (#126278)

* fix(ui): compose agent run transcript responses

* fix(ui): preserve explicit run transcript boundaries

* fix(ui): preserve run transcript status

* fix(ui): preserve live agent run content

* fix(ui): preserve semantic agent run frames

* fix(ui): complete agent run transcript ownership

* fix(ui): stabilize agent run transcript lifecycle

* fix(ui): complete agent run frame ownership

* fix(ui): close transcript ownership races

* test(gateway-client): drop non-regressing projection case

---------

Co-authored-by: RoboClaw <309084314+roboclaw-bot@users.noreply.github.com>
This commit is contained in:
Vyctor H. Brzezowski
2026-08-23 00:13:27 -03:00
committed by GitHub
parent 32bb08dbbf
commit 2a87667bb2
47 changed files with 2582 additions and 481 deletions
@@ -55,6 +55,7 @@ describe("persistCodexContextCompactionActivity", () => {
display: true,
excludeFromContext: true,
idempotencyKey: "codex-context-compaction:thread-1:turn-1:compact-1",
__openclaw: { runId: "run-1" },
},
});
expect(publishUpdate).toHaveBeenCalledOnce();
@@ -39,6 +39,7 @@ export async function persistCodexContextCompactionActivity(params: {
itemId: params.itemId,
...(params.runId ? { runId: params.runId } : {}),
},
...(params.runId ? { __openclaw: { runId: params.runId } } : {}),
timestamp: params.timestamp,
idempotencyKey: activityId,
};
@@ -245,6 +245,8 @@ export async function activateCodexAttemptTurn(
cwd: effectiveCwd,
messages,
idempotencyScope: `codex-app-server:${resourceState.thread.threadId}`,
runId: params.runId,
runMirrorIdentityPrefix: `${activeTurnId}:`,
config: params.config,
});
}
@@ -114,6 +114,8 @@ export async function runCodexSettledTurnFinalization(
cwd: attempt.workspaceDir,
messages: [assistant],
idempotencyScope: `codex-settled-finalizer:${attempt.runId}`,
runId: attempt.runId,
terminalAssistantOwner: { mirrorIdentity, runId: attempt.runId },
config: attempt.config,
skipBeforeMessageWriteHooks: true,
});
@@ -1,5 +1,6 @@
import { createHash } from "node:crypto";
import type { AgentMessage } from "openclaw/plugin-sdk/agent-harness-runtime";
import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
import { readUpstreamUserText } from "./upstream-prompt-provenance.js";
type MirroredAgentMessage = Extract<AgentMessage, { role: "user" | "assistant" | "toolResult" }>;
@@ -29,6 +30,24 @@ export function attachCodexMirrorAttestation(
return attested;
}
export function attachCodexMirrorRunId<T extends AgentMessage>(
message: T,
runId: string,
terminal = false,
): T {
const existing = CODEX_META_KEY in message ? message[CODEX_META_KEY] : undefined;
const metadata = asOptionalRecord(existing) ?? {};
const { runTerminal: _staleTerminal, ...current } = metadata;
return {
...message,
[CODEX_META_KEY]: {
...current,
runId,
...(terminal ? { runTerminal: true } : {}),
},
} as T; // SAFETY: AgentMessage variants permit provider metadata at runtime; preserve T.
}
export function readCodexMirrorSourceFingerprint(message: AgentMessage): string | undefined {
const meta = CODEX_META_KEY in message ? message[CODEX_META_KEY] : undefined;
if (!meta || typeof meta !== "object" || Array.isArray(meta)) {
@@ -20,6 +20,7 @@ import {
import { afterEach, describe, expect, it, vi } from "vitest";
import type { CodexThread } from "./protocol.js";
import { readCodexMirroredSessionHistoryMessages } from "./session-history.js";
import { attachCodexMirrorRunId } from "./transcript-mirror-attestation.js";
import {
buildCodexUserPromptMessage,
codexTranscriptMirrorRuntime,
@@ -793,6 +794,17 @@ describe("projectBoundedCodexThreadHistory", () => {
});
describe("mirrorCodexAppServerTranscript", () => {
it("clears terminal ownership when a mirrored message becomes non-terminal", () => {
const message = makeAgentAssistantMessage({
content: [{ type: "text", text: "intermediate narration" }],
timestamp: Date.now(),
});
const terminal = attachCodexMirrorRunId(message, "run-1", true);
const intermediate = attachCodexMirrorRunId(terminal, "run-1");
expect(intermediate).toMatchObject({ __openclaw: { runId: "run-1" } });
expect(intermediate).not.toHaveProperty("__openclaw.runTerminal");
});
it("hides current memory-maintenance messages without hiding replayed turns", async () => {
initializeGlobalHookRunner(
createMockPluginRegistry([
@@ -1061,6 +1073,8 @@ describe("mirrorCodexAppServerTranscript", () => {
),
],
idempotencyScope: "codex-app-server:thread-1",
runId: "openclaw-run-1",
runMirrorIdentityPrefix: "turn-1:",
terminalAssistantOwner: {
mirrorIdentity: "turn-1:assistant",
runId: "openclaw-run-1",
@@ -1072,6 +1086,22 @@ describe("mirrorCodexAppServerTranscript", () => {
);
expect(updates.map((update) => update.update?.messageSeq)).toEqual([1, 2]);
expect(updates.map((update) => update.update?.runId)).toEqual([undefined, "openclaw-run-1"]);
expect(
updates.map(
(update) =>
(update.update?.message as { __openclaw?: { runId?: string } } | undefined)?.[
"__openclaw"
]?.runId,
),
).toEqual(["openclaw-run-1", "openclaw-run-1"]);
expect(
updates.map(
(update) =>
(update.update?.message as { __openclaw?: { runTerminal?: boolean } } | undefined)?.[
"__openclaw"
]?.runTerminal,
),
).toEqual([undefined, true]);
expect(
updates.map((update) => {
const message = update.update?.message as { role?: string } | undefined;
@@ -1483,7 +1513,7 @@ describe("mirrorCodexAppServerTranscript", () => {
turnId: "turn-1",
});
expect(mirrorOutcome.assistantTranscriptOwned).toBe(true);
expect(mirrorOutcome.assistantTranscriptOwned).toBe(false);
expect(mirrorOutcome.mirroredMessages).toEqual([]);
});
@@ -23,6 +23,7 @@ import {
} from "./transcript-history-projection.js";
import {
attachCodexMirrorAttestation,
attachCodexMirrorRunId,
fingerprintCodexMirrorSourceMessage,
readCodexMirrorSourceFingerprint,
} from "./transcript-mirror-attestation.js";
@@ -134,6 +135,8 @@ async function mirrorBestEffort(params: {
// identity (not via the scope). Dropping `turnId` from the scope here is
// what lets a re-emitted prior-turn entry collide with its existing key.
idempotencyScope: `codex-app-server:${params.threadId}`,
runId: params.params.runId,
runMirrorIdentityPrefix: `${params.turnId}:`,
terminalAssistantOwner: {
mirrorIdentity: `${params.turnId}:assistant`,
runId: params.params.runId,
@@ -166,11 +169,13 @@ async function mirrorBestEffort(params: {
);
});
const assistantMirrorIdentity = `${params.turnId}:assistant`;
const assistantTranscriptOwned =
mirrorResult.assistantMirrorIdentitiesOwned.includes(assistantMirrorIdentity);
const assistantTranscriptMessage = assistantTranscriptOwned
? mirroredMessages.find((message) => readMirrorIdentity(message) === assistantMirrorIdentity)
: undefined;
const assistantTranscriptMessage = mirroredMessages.find(
(message) => readMirrorIdentity(message) === assistantMirrorIdentity,
);
const assistantTranscriptOwned = Boolean(
assistantTranscriptMessage &&
mirrorResult.assistantMirrorIdentitiesOwned.includes(assistantMirrorIdentity),
);
const assistantTranscriptIdempotencyKey = normalizeOptionalString(
(assistantTranscriptMessage as { idempotencyKey?: unknown } | undefined)?.idempotencyKey,
);
@@ -283,6 +288,8 @@ export async function mirrorPromptAtTurnStartBestEffort(params: {
cwd: params.cwd,
messages: [userPromptMessage],
idempotencyScope: `codex-app-server:${params.threadId}`,
runId: params.params.runId,
runMirrorIdentityPrefix: `${params.turnId}:`,
config: params.params.config,
});
for (const receipt of mirrorResult.userMessageReceipts) {
@@ -327,6 +334,8 @@ async function mirror(params: {
storePath?: string;
messages: AgentMessage[];
idempotencyScope?: string;
runId?: string;
runMirrorIdentityPrefix?: string;
terminalAssistantOwner?: { mirrorIdentity: string; runId: string };
config?: SessionTranscriptWriteLockParams["config"];
skipBeforeMessageWriteHooks?: boolean;
@@ -377,8 +386,22 @@ async function mirror(params: {
idempotencyKeys: candidateIdempotencyKeys,
});
for (const { dedupeIdentity, idempotencyKey, message, sourceFingerprint } of candidates) {
const mirrorIdentity = readMirrorIdentity(message);
const ownsRun = Boolean(
params.runId &&
(!params.runMirrorIdentityPrefix ||
mirrorIdentity?.startsWith(params.runMirrorIdentityPrefix)),
);
const terminalOwner = params.terminalAssistantOwner;
const ownsTerminal = Boolean(
ownsRun && terminalOwner && mirrorIdentity === terminalOwner.mirrorIdentity,
);
const ownedMessage =
ownsRun && params.runId
? attachCodexMirrorRunId(message, params.runId, ownsTerminal)
: message;
const transcriptMessage = {
...attachCodexMirrorAttestation(message, sourceFingerprint),
...attachCodexMirrorAttestation(ownedMessage, sourceFingerprint),
...(idempotencyKey ? { idempotencyKey } : {}),
} as AgentMessage;
if (idempotencyKey && mirrorFacts.existingIdempotencyKeys.has(idempotencyKey)) {
@@ -428,12 +451,14 @@ async function mirror(params: {
}
: attachCodexMirrorAttestation(nextMessage, sourceFingerprint)
) as AgentMessage;
const mirrorIdentity = readMirrorIdentity(message);
if (mirrorIdentity) {
// Hooks may replace the whole message. Restore the provider-owned
// identity so retries cannot turn a stale idempotency hit into evidence.
messageToAppend = attachCodexMirrorIdentity(messageToAppend, mirrorIdentity);
}
if (ownsRun && params.runId) {
messageToAppend = attachCodexMirrorRunId(messageToAppend, params.runId, ownsTerminal);
}
messageToAppend = projectAgentHarnessTranscriptMessageForDisplay({
hidden: (message as { display?: boolean }).display === false,
message: messageToAppend,
@@ -0,0 +1,99 @@
import { asNullableRecord as readRecord } from "@openclaw/normalization-core/record-coerce";
export type SessionMessageEnvelope = {
messageId?: unknown;
messageSeq?: unknown;
clientRunId?: unknown;
runId?: unknown;
idempotencyKey?: unknown;
};
export type SessionMessageIdentity = {
role: string;
id: string | null;
sequence: number | null;
idempotencyKey: string | null;
runId: string | null;
isImported: boolean;
externalSource: string | null;
};
export function readSessionProjectionString(value: unknown): string | null {
return typeof value === "string" ? value.trim() || null : null;
}
function readPositiveSafeInteger(value: unknown): number | null {
return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : null;
}
/** History and status markers carry transcript order even when they have no chat role. */
export function readSessionMessageSequence(
message: unknown,
envelope?: SessionMessageEnvelope,
): number | null {
const metadata = readRecord(readRecord(message)?.["__openclaw"]);
return readPositiveSafeInteger(metadata?.seq) ?? readPositiveSafeInteger(envelope?.messageSeq);
}
/** Run ownership normalizes a user-turn suffix without changing its persisted send key. */
export function normalizeSessionProjectionRunId(value: unknown): string | null {
const runId = readSessionProjectionString(value);
return runId?.endsWith(":user") ? runId.slice(0, -":user".length) || null : runId;
}
/** Persisted row facts win; assistant run ownership comes from its authoritative producer. */
export function readSessionMessageIdentity(
message: unknown,
envelope?: SessionMessageEnvelope,
): SessionMessageIdentity | null {
const record = readRecord(message);
const role = readSessionProjectionString(record?.role)?.toLowerCase();
if (!record || !role) {
return null;
}
const metadata = readRecord(record["__openclaw"]);
const importedFrom = readSessionProjectionString(metadata?.importedFrom);
const cliSessionId = readSessionProjectionString(metadata?.cliSessionId);
const externalId = readSessionProjectionString(metadata?.externalId);
const idempotencyKey =
readSessionProjectionString(metadata?.idempotencyKey) ??
readSessionProjectionString(record.idempotencyKey) ??
readSessionProjectionString(envelope?.idempotencyKey) ??
readSessionProjectionString(envelope?.clientRunId);
const persistedRunId = normalizeSessionProjectionRunId(idempotencyKey);
const envelopeRunId = normalizeSessionProjectionRunId(envelope?.runId);
const metadataRunId = normalizeSessionProjectionRunId(metadata?.runId);
const mirroredMessage = readSessionProjectionString(metadata?.mirrorOrigin) !== null;
// CLI persistence namespaces assistant send keys; the suffix is the
// originating Gateway run identity consumed by every projection layer.
const isCliAssistant =
role === "assistant" && readSessionProjectionString(record.api)?.toLowerCase() === "cli";
const canonicalPersistedRunId =
isCliAssistant && persistedRunId?.startsWith("cli-assistant:")
? readSessionProjectionString(persistedRunId.slice("cli-assistant:".length))
: persistedRunId;
const optimisticRunId =
metadata && Object.keys(metadata).every((key) => key === "idempotencyKey")
? canonicalPersistedRunId
: null;
return {
role,
id:
readSessionProjectionString(metadata?.id) ?? readSessionProjectionString(envelope?.messageId),
sequence: readSessionMessageSequence(message, envelope),
idempotencyKey,
runId:
role === "assistant"
? (metadataRunId ??
envelopeRunId ??
(isCliAssistant || !mirroredMessage ? canonicalPersistedRunId : null) ??
optimisticRunId)
: (metadataRunId ?? canonicalPersistedRunId ?? envelopeRunId),
isImported: Boolean(importedFrom || cliSessionId || externalId),
// Imported IDs belong to their provider and CLI session, never the native ID namespace.
externalSource:
importedFrom && cliSessionId && externalId
? JSON.stringify([importedFrom, cliSessionId, externalId])
: null,
};
}
@@ -106,6 +106,42 @@ describe("readSessionMessageIdentity", () => {
expect(normalizeSessionProjectionRunId(input)).toBe(expected);
});
it("recovers the originating run from a persisted CLI assistant send key", () => {
expect(
readSessionMessageIdentity({
role: "assistant",
api: "cli",
content: "Done",
idempotencyKey: "cli-assistant:run-cli-1",
}),
).toMatchObject({
idempotencyKey: "cli-assistant:run-cli-1",
runId: "run-cli-1",
});
});
it("keeps assistant dedupe identity separate from producer-owned run identity", () => {
expect(
readSessionMessageIdentity({
role: "assistant",
content: "Commentary",
idempotencyKey: "codex-app-server:thread-1:turn-1:commentary:item-1",
__openclaw: { mirrorOrigin: "codex-app-server", runId: "run-1" },
}),
).toMatchObject({
idempotencyKey: "codex-app-server:thread-1:turn-1:commentary:item-1",
runId: "run-1",
});
expect(
readSessionMessageIdentity({
role: "assistant",
content: "Imported history",
idempotencyKey: "codex-app-server:thread-1:history:turn-1:assistant",
__openclaw: { mirrorOrigin: "codex-app-server" },
}),
).toHaveProperty("runId", null);
});
it("requires every imported source component before claiming provider identity", () => {
const identity = readSessionMessageIdentity(
createMessage("user", "imported", {
@@ -202,6 +238,22 @@ describe("session transcript projection", () => {
expect(state.messages).toEqual([persisted]);
});
it("does not promote a provisional final into same-run Codex commentary", () => {
const commentary = createMessage("assistant", "commentary", {
id: "commentary-1",
mirrorOrigin: "codex-app-server",
runId: "run-1",
});
const final = createMessage("assistant", "final answer");
let state = projectLiveSessionMessage(createSessionProjection(primaryScope), commentary, {
runId: "run-1",
});
state = projectLiveSessionMessage(state, final, { runId: "run-1" });
expect(state.messages).toEqual([commentary, final]);
});
it("keeps the durable assistant identity when its run's terminal projection replays", () => {
const persisted = createMessage("assistant", "persisted final", {
id: "assistant-final",
@@ -1,25 +1,24 @@
/** Browser-safe identity and replay rules shared by Gateway conversation clients. */
import { asNullableRecord as readRecord } from "@openclaw/normalization-core/record-coerce";
import {
normalizeSessionProjectionRunId,
readSessionMessageIdentity,
readSessionProjectionString as readNonemptyString,
type SessionMessageEnvelope,
type SessionMessageIdentity,
} from "./session-projection-message-identity.js";
import { reduceSessionProjectionRunEventImpl } from "./session-projection-run-event.js";
export type SessionMessageEnvelope = {
messageId?: unknown;
messageSeq?: unknown;
clientRunId?: unknown;
runId?: unknown;
idempotencyKey?: unknown;
};
export type SessionMessageIdentity = {
role: string;
id: string | null;
sequence: number | null;
idempotencyKey: string | null;
runId: string | null;
isImported: boolean;
externalSource: string | null;
};
export {
normalizeSessionProjectionRunId,
readSessionMessageIdentity,
readSessionMessageSequence,
} from "./session-projection-message-identity.js";
export type {
SessionMessageEnvelope,
SessionMessageIdentity,
} from "./session-projection-message-identity.js";
export type SessionProjectionScope = {
sessionKey?: string;
@@ -121,66 +120,6 @@ export type SessionProjectionEvent = ScopedSessionProjectionEvent &
| { type: "reconnected" }
);
function readNonemptyString(value: unknown): string | null {
return typeof value === "string" ? value.trim() || null : null;
}
function readPositiveSafeInteger(value: unknown): number | null {
return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : null;
}
/** History and status markers carry transcript order even when they have no chat role. */
export function readSessionMessageSequence(
message: unknown,
envelope?: SessionMessageEnvelope,
): number | null {
const metadata = readRecord(readRecord(message)?.["__openclaw"]);
return readPositiveSafeInteger(metadata?.seq) ?? readPositiveSafeInteger(envelope?.messageSeq);
}
/** Run ownership normalizes a user-turn suffix without changing its persisted send key. */
export function normalizeSessionProjectionRunId(value: unknown): string | null {
const runId = readNonemptyString(value);
return runId?.endsWith(":user") ? runId.slice(0, -":user".length) || null : runId;
}
/** Persisted row facts win; assistant run ownership comes from its authoritative producer. */
export function readSessionMessageIdentity(
message: unknown,
envelope?: SessionMessageEnvelope,
): SessionMessageIdentity | null {
const record = readRecord(message);
const role = readNonemptyString(record?.role)?.toLowerCase();
if (!record || !role) {
return null;
}
const metadata = readRecord(record["__openclaw"]);
const importedFrom = readNonemptyString(metadata?.importedFrom);
const cliSessionId = readNonemptyString(metadata?.cliSessionId);
const externalId = readNonemptyString(metadata?.externalId);
const idempotencyKey =
readNonemptyString(metadata?.idempotencyKey) ??
readNonemptyString(record.idempotencyKey) ??
readNonemptyString(envelope?.idempotencyKey) ??
readNonemptyString(envelope?.clientRunId);
return {
role,
id: readNonemptyString(metadata?.id) ?? readNonemptyString(envelope?.messageId),
sequence: readSessionMessageSequence(message, envelope),
idempotencyKey,
runId:
(role === "assistant" ? normalizeSessionProjectionRunId(envelope?.runId) : null) ??
normalizeSessionProjectionRunId(idempotencyKey) ??
normalizeSessionProjectionRunId(envelope?.runId),
isImported: Boolean(importedFrom || cliSessionId || externalId),
// Imported IDs belong to their provider and CLI session, never the native ID namespace.
externalSource:
importedFrom && cliSessionId && externalId
? JSON.stringify([importedFrom, cliSessionId, externalId])
: null,
};
}
/** Local turns have no durable transcript metadata beyond their own optional send key. */
export function isLocallyOptimisticSessionMessage(message: unknown): boolean {
const identity = readSessionMessageIdentity(message);
@@ -298,6 +237,7 @@ function entryMatches(
}
const durableEntry = left.identity?.id ? left : right.identity?.id ? right : null;
const provisionalEntry = durableEntry === left ? right : durableEntry === right ? left : null;
const durableMetadata = readRecord(readRecord(durableEntry?.message)?.["__openclaw"]);
if (
durableEntry?.live &&
provisionalEntry?.live &&
@@ -307,12 +247,15 @@ function entryMatches(
!provisionalEntry.identity.isImported &&
!provisionalEntry.identity.id &&
durableEntry.identity.runId &&
durableEntry.identity.runId === provisionalEntry.identity.runId
durableEntry.identity.runId === provisionalEntry.identity.runId &&
(readNonemptyString(durableMetadata?.mirrorOrigin) === null ||
durableMetadata?.runTerminal === true)
) {
return true;
}
const persisted = left.identity;
const observed = right.identity;
const persistedMetadata = readRecord(readRecord(left.message)?.["__openclaw"]);
if (
allowSnapshotPromotion &&
right.live &&
@@ -327,7 +270,9 @@ function entryMatches(
(persisted.role === "assistant" &&
observed.sequence === null &&
persisted.runId !== null &&
persisted.runId === observed.runId))
persisted.runId === observed.runId &&
(readNonemptyString(persistedMetadata?.mirrorOrigin) === null ||
persistedMetadata?.runTerminal === true)))
) {
// Only current-scope history can promote an observed native sequence or assistant run.
return true;
@@ -198,7 +198,24 @@ describe("WebChat message tool internal source reply", () => {
mediaUrls: imagePaths,
};
const updates: SessionTranscriptUpdate[] = [];
const unsubscribe = onSessionTranscriptUpdate((update) => updates.push(update));
const publishedDownloads: Array<Promise<unknown>> = [];
const unsubscribe = onSessionTranscriptUpdate((update) => {
updates.push(update);
const content =
update.message && typeof update.message === "object"
? (update.message as { content?: Array<Record<string, unknown>> }).content
: undefined;
for (const block of content?.filter((entry) => entry.type === "image") ?? []) {
publishedDownloads.push(
resolveManagedOutgoingMediaArtifactDownload({
sessionKey,
agentId: "main",
artifactId: String(block.artifactId),
stateDir,
}),
);
}
});
const [toolResult, overlappingResult] = await Promise.all([
tool.execute("restart-proof-call", sendParams),
tool.execute("restart-proof-call", sendParams),
@@ -272,6 +289,10 @@ describe("WebChat message tool internal source reply", () => {
published?.message as { content?: Array<Record<string, unknown>> }
)?.content;
expect(publishedContent?.filter((block) => block.type === "image")).toHaveLength(2);
await expect(Promise.all(publishedDownloads)).resolves.toEqual([
expect.objectContaining({ type: "image" }),
expect.objectContaining({ type: "image" }),
]);
for (const block of content.filter((entry) => entry.type === "image")) {
await expect(
resolveManagedOutgoingMediaArtifactDownload({
+6 -2
View File
@@ -390,6 +390,7 @@ export async function appendAssistantMessageToSessionTranscript(params: {
text?: string;
mediaUrls?: string[];
content?: SessionTranscriptAssistantMessage["content"];
eventId?: string;
idempotencyKey?: string;
runId?: string;
deliveryMirror?: InternalSessionTranscriptDeliveryMirror;
@@ -429,8 +430,9 @@ export async function appendAssistantMessageToSessionTranscript(params: {
? { sessionLifecyclePatch: params.sessionLifecyclePatch }
: {}),
storePath: params.storePath,
idempotencyKey: params.idempotencyKey,
runId: params.runId,
...(params.eventId ? { eventId: params.eventId } : {}),
...(params.idempotencyKey ? { idempotencyKey: params.idempotencyKey } : {}),
...(params.runId ? { runId: params.runId } : {}),
updateMode: params.updateMode,
config: params.config,
...(params.beforeMessageWrite ? { beforeMessageWrite: params.beforeMessageWrite } : {}),
@@ -470,6 +472,7 @@ export async function appendExactAssistantMessageToSessionTranscript(params: {
expectedSessionState?: SessionTranscriptTurnExpectedState;
sessionLifecyclePatch?: SessionTranscriptTurnLifecyclePatch;
message: SessionTranscriptAssistantMessage;
eventId?: string;
idempotencyKey?: string;
runId?: string;
storePath?: string;
@@ -598,6 +601,7 @@ export async function appendExactAssistantMessageToSessionTranscript(params: {
messages: [
{
message: preparedUnkeyedMessage,
...(params.eventId ? { eventId: params.eventId } : {}),
...(explicitIdempotencyKey ? { idempotencyLookup: "scan" } : {}),
...(explicitIdempotencyKey && params.beforeMessageWrite
? {
@@ -1,3 +1,4 @@
import { randomUUID } from "node:crypto";
import type { ReplyPayload } from "../auto-reply/reply-payload.js";
import { appendAssistantMessageToSessionTranscript } from "../config/sessions.js";
import { resolveSessionStorePathCore } from "../config/sessions/paths.js";
@@ -11,8 +12,8 @@ import { getAgentScopedMediaLocalRootsForSources } from "../media/local-roots.js
import { createKeyedFifoLeaseRegistry } from "../shared/keyed-fifo-lease.js";
import { isOpenClawDeliveryMirrorAssistantMessage } from "../shared/transcript-only-openclaw-assistant.js";
import {
attachManagedOutgoingMediaToMessage,
createManagedOutgoingMediaBlocks,
removeManagedOutgoingMediaBlocks,
} from "./managed-image-attachments.js";
import { prepareGatewayInjectedAssistantContent } from "./server-methods/chat-transcript-inject.js";
@@ -95,10 +96,12 @@ export async function persistInternalSourceReply(params: {
return;
}
const mediaUrls = collectSourceReplyMediaUrls(params.payload);
const messageId = randomUUID();
const mediaBlocks = await createManagedOutgoingMediaBlocks({
sessionKey: params.sessionKey,
agentId: params.agentId,
mediaUrls,
messageId,
localRoots: getAgentScopedMediaLocalRootsForSources({
cfg: params.cfg,
agentId: params.agentId,
@@ -119,6 +122,7 @@ export async function persistInternalSourceReply(params: {
: {}),
...(writerFence ? { expectedWriterRunId: writerFence.expectedWriterRunId } : {}),
content: prepareGatewayInjectedAssistantContent(content),
eventId: messageId,
idempotencyKey: params.idempotencyKey,
runId: params.runId,
...(params.sourceReplyFinal !== undefined
@@ -134,16 +138,11 @@ export async function persistInternalSourceReply(params: {
config: params.cfg,
});
if (!appended.ok) {
await removeManagedOutgoingMediaBlocks({ blocks: mediaBlocks, messageId });
throw new Error(`Internal source reply persistence failed: ${appended.reason}`);
}
if (
mediaBlocks.length > 0 &&
!attachManagedOutgoingMediaToMessage({
messageId: appended.messageId,
blocks: mediaBlocks,
})
) {
throw new Error("Internal source reply media ownership could not be persisted");
if (appended.messageId !== messageId) {
await removeManagedOutgoingMediaBlocks({ blocks: mediaBlocks, messageId });
}
} finally {
lease?.release();
+16
View File
@@ -787,6 +787,22 @@ export async function cleanupManagedOutgoingMediaRecords(params?: {
return { deletedRecordCount, deletedFileCount, retainedCount };
}
export async function removeManagedOutgoingMediaBlocks(params: {
blocks: readonly Record<string, unknown>[];
messageId: string;
stateDir?: string;
}): Promise<void> {
const stateDir = params.stateDir ?? resolveStateDir();
await Promise.all(
collectManagedOutgoingAttachmentRefs(params.blocks).map(async ({ attachmentId }) => {
const record = readManagedImageRecord(attachmentId, stateDir);
if (record?.messageId === params.messageId) {
await deleteManagedImageRecordArtifacts(record, stateDir);
}
}),
);
}
function resolveManagedSessionOwnerAgentId(
sessionKey: string,
explicitAgentId?: string,
@@ -158,7 +158,9 @@ async function installActiveRunSnapshot(
}
async function assertActiveTurnVisible(page: Page, streamText: string): Promise<void> {
await expect(page.getByText(streamText, { exact: true })).toHaveCount(1, { timeout: 10_000 });
await expect(
page.locator(".chat-thread-inner").getByText(streamText, { exact: true }),
).toHaveCount(1, { timeout: 10_000 });
await page.locator(".chat-tool-row--running").waitFor({ timeout: 10_000 });
await page.getByRole("button", { name: "Stop generating" }).waitFor({ timeout: 10_000 });
await expect
@@ -250,14 +252,14 @@ async function assertSteeredRecoveryOrder(
await expect(page.locator(".chat-working-indicator")).toHaveCount(1, { timeout: 10_000 });
const order = await thread.evaluate((element, expected) => {
const groups = Array.from(element.querySelectorAll<HTMLElement>(".chat-group"));
const groupWithText = (text: string) =>
groups.find((group) => (group.textContent ?? "").includes(text));
const original = groupWithText(expected.original);
const beforeSteer = groupWithText(expected.beforeSteer);
const steer = groupWithText(expected.steer);
const visibleText = Array.from(element.querySelectorAll<HTMLElement>(".chat-bubble"));
const bubbleWithText = (text: string) =>
visibleText.find((bubble) => (bubble.textContent ?? "").includes(text));
const original = bubbleWithText(expected.original);
const beforeSteer = bubbleWithText(expected.beforeSteer);
const steer = bubbleWithText(expected.steer);
const tool = element.querySelector<HTMLElement>(".chat-tool-row--running");
const afterSteer = groupWithText(expected.afterSteer);
const afterSteer = bubbleWithText(expected.afterSteer);
const precedes = (upper: Element | undefined | null, lower: Element | undefined | null) =>
Boolean(
upper && lower && upper.compareDocumentPosition(lower) & Node.DOCUMENT_POSITION_FOLLOWING,
@@ -0,0 +1,296 @@
import fs from "node:fs/promises";
import path from "node:path";
import { expect, it } from "vitest";
import { installMockGateway } from "../test-helpers/control-ui-e2e.ts";
import { createControlUiE2eSuite } from "./control-ui-e2e-suite.test-support.ts";
const suite = createControlUiE2eSuite({
name: "Control UI agent run transcript",
startServerBeforeBrowser: true,
});
function transcriptMessage(
role: "assistant" | "toolResult" | "user",
content: unknown,
runId: string,
id: string,
seq: number,
) {
return {
role,
content,
timestamp: Date.UTC(2026, 7, 19, 12, 0, seq),
__openclaw: { id, idempotencyKey: runId, seq },
};
}
suite.define(() => {
it("renders each run as one linear response with actions only on its terminal text", async () => {
const context = await suite.browser.newContext({ viewport: { height: 900, width: 1200 } });
const page = await context.newPage();
const firstRunId = "run-composed-first";
const secondRunId = "run-composed-second";
const toolOnlyRunId = "run-tool-only";
const commentaryToolRunId = "run-commentary-tool-only";
await installMockGateway(page, {
historyMessages: [
transcriptMessage("user", "Create the launch card.", `${firstRunId}:user`, "user-1", 1),
transcriptMessage(
"assistant",
"Ill create the launch card and check the existing style first.",
firstRunId,
"assistant-1",
2,
),
{
...transcriptMessage(
"assistant",
[
{
type: "toolCall",
id: "call-read",
name: "read",
arguments: { path: "ui/src/styles/chat.css" },
},
],
firstRunId,
"tool-call-1",
3,
),
},
{
...transcriptMessage(
"toolResult",
[{ type: "text", text: "Existing card styles loaded." }],
firstRunId,
"tool-result-1",
4,
),
toolCallId: "call-read",
toolName: "read",
runId: firstRunId,
},
transcriptMessage(
"assistant",
"The first draft matches the transcript rhythm. Ill render the asset now.",
firstRunId,
"assistant-2",
5,
),
{
...transcriptMessage(
"assistant",
[
{
type: "toolCall",
id: "call-render",
name: "exec",
arguments: { command: "render launch-card.svg" },
},
],
firstRunId,
"tool-call-2",
6,
),
},
{
...transcriptMessage(
"toolResult",
[{ type: "text", text: "Rendered launch-card.svg" }],
firstRunId,
"tool-result-2",
7,
),
toolCallId: "call-render",
toolName: "exec",
runId: firstRunId,
},
transcriptMessage(
"assistant",
"The launch card is ready: MEDIA:./launch-card.svg",
firstRunId,
"assistant-3",
8,
),
transcriptMessage("user", "Now write the caption.", `${secondRunId}:user`, "user-2", 9),
transcriptMessage(
"assistant",
"Caption ready for the second run.",
secondRunId,
"assistant-4",
10,
),
transcriptMessage("user", "Check without replying.", `${toolOnlyRunId}:user`, "user-3", 11),
{
...transcriptMessage(
"toolResult",
[{ type: "text", text: "Tool-only result" }],
toolOnlyRunId,
"tool-result-3",
12,
),
toolCallId: "call-tool-only",
toolName: "read",
runId: toolOnlyRunId,
},
transcriptMessage(
"user",
"Inspect and stop after the tool.",
`${commentaryToolRunId}:user`,
"user-4",
13,
),
transcriptMessage(
"assistant",
"Ill inspect the current state first.",
commentaryToolRunId,
"assistant-5",
14,
),
{
...transcriptMessage(
"toolResult",
[{ type: "text", text: "Commentary-led tool-only result" }],
commentaryToolRunId,
"tool-result-4",
15,
),
toolCallId: "call-commentary-tool-only",
toolName: "read",
runId: commentaryToolRunId,
},
],
});
await page.goto(`${suite.server.baseUrl}chat`);
const transcript = page.locator(".chat-thread-inner");
await transcript.getByText("Caption ready for the second run.", { exact: true }).waitFor();
const artifactDir = process.env.OPENCLAW_CONTROL_UI_E2E_ARTIFACT_DIR?.trim();
if (artifactDir) {
await fs.mkdir(artifactDir, { recursive: true });
await page.screenshot({
path: path.join(artifactDir, "agent-run-transcript.png"),
fullPage: true,
});
}
const assistantGroups = page.locator(".chat-group.assistant");
expect(await assistantGroups.count()).toBe(4);
const firstRun = assistantGroups.filter({
hasText: "Ill create the launch card and check the existing style first.",
});
expect(await firstRun.count()).toBe(1);
expect(await firstRun.locator(".chat-sender-name").count()).toBe(1);
expect(await firstRun.locator(".chat-group-footer-actions").count()).toBe(1);
expect(await firstRun.locator(".chat-message-actions-row").count()).toBe(0);
expect(await firstRun.locator(".chat-group-footer-actions button").count()).toBe(2);
expect(
await firstRun
.locator(".chat-group-footer-actions button")
.evaluateAll((buttons) => buttons.map((button) => button.getAttribute("aria-label"))),
).toEqual(["Reply to message", "Copy as markdown"]);
const orderedContent = await firstRun.locator(".chat-bubble").evaluateAll((bubbles) =>
bubbles.map((bubble) => ({
messageId: bubble.getAttribute("data-message-id"),
text: bubble.textContent?.replace(/\s+/gu, " ").trim(),
})),
);
expect(orderedContent).toEqual([
expect.objectContaining({ text: expect.stringContaining("Ill create the launch card") }),
expect.objectContaining({ text: expect.stringContaining("Read") }),
expect.objectContaining({ text: expect.stringContaining("The first draft matches") }),
expect.objectContaining({ text: expect.stringContaining("render launch-card.svg") }),
expect.objectContaining({ text: expect.stringContaining("The launch card is ready") }),
]);
expect(
await firstRun.getByText("Caption ready for the second run.", { exact: true }).count(),
).toBe(0);
const toolOnlyRun = page.locator(
`.chat-group.assistant[data-chat-row-key*="${toolOnlyRunId}"]`,
);
expect(await toolOnlyRun.count()).toBe(1);
expect(await toolOnlyRun.locator(".chat-group-footer-actions").count()).toBe(0);
const commentaryToolRun = page.locator(
`.chat-group.assistant[data-chat-row-key*="${commentaryToolRunId}"]`,
);
expect(await commentaryToolRun.count()).toBe(1);
expect(
await commentaryToolRun
.getByText("Ill inspect the current state first.", { exact: true })
.count(),
).toBe(1);
expect(await commentaryToolRun.locator(".chat-group-footer-actions").count()).toBe(0);
await context.close();
});
it("keeps the run row identity when a hidden heartbeat boundary reaches history", async () => {
const context = await suite.browser.newContext({ viewport: { height: 900, width: 1200 } });
const page = await context.newPage();
const runId = "run-heartbeat-browser-handoff";
const gateway = await installMockGateway(page, {
historyMessages: [],
inFlightRun: { runId, text: "" },
sessionInfo: {
activeRunIds: [runId],
hasActiveRun: true,
key: "main",
},
});
await page.goto(`${suite.server.baseUrl}chat`);
const liveRow = page.locator(".chat-virtual-row", {
has: page.locator(".chat-reading-indicator"),
});
await liveRow.waitFor();
const liveKey = await liveRow.getAttribute("data-virtual-row-key");
expect(liveKey).not.toBeNull();
const finalText = "Heartbeat handoff complete.";
const persistedMessage = {
role: "assistant",
api: "cli",
content: finalText,
idempotencyKey: `cli-assistant:${runId}`,
timestamp: Date.UTC(2026, 7, 19, 12, 1),
__openclaw: {
id: "assistant-after-hidden-heartbeat",
seq: 1,
turnBoundary: true,
},
};
await gateway.setHistoryMessages([persistedMessage]);
const historyRequestsBeforeFinal = (await gateway.getRequests("chat.history")).length;
await gateway.emitGatewayEvent("session.message", {
activeRunIds: [],
clientRunId: runId,
hasActiveRun: false,
message: persistedMessage,
messageId: "assistant-after-hidden-heartbeat",
messageSeq: 1,
session: {
activeRunIds: [],
hasActiveRun: false,
key: "main",
kind: "direct",
status: "done",
updatedAt: Date.now(),
},
sessionKey: "main",
});
await expect
.poll(async () => (await gateway.getRequests("chat.history")).length)
.toBeGreaterThan(historyRequestsBeforeFinal);
const settledRow = page.locator(".chat-virtual-row", {
has: page.getByText(finalText, { exact: true }),
});
await settledRow.waitFor();
await expect.poll(() => settledRow.getAttribute("data-virtual-row-key")).toBe(liveKey);
expect(await settledRow.count()).toBe(1);
await context.close();
});
});
@@ -197,7 +197,8 @@ suite.define(() => {
stream: "item",
ts: Date.now(),
});
await page.getByText(commentaryText, { exact: true }).waitFor();
const transcript = page.locator(".chat-thread-inner");
await transcript.getByText(commentaryText, { exact: true }).waitFor();
const emitTool = (data: Record<string, unknown>) =>
gateway.emitGatewayEvent("agent", {
data,
@@ -227,7 +228,9 @@ suite.define(() => {
steerParams.idempotencyKey,
"steer chat send idempotency key",
);
await expect.poll(() => page.getByText(commentaryText, { exact: true }).count()).toBe(1);
await expect
.poll(() => transcript.getByText(commentaryText, { exact: true }).count())
.toBe(1);
await gateway.resolveDeferred("chat.send", { runId: steerRunId, status: "started" });
const steerUser = {
__openclaw: {
@@ -278,7 +281,7 @@ suite.define(() => {
toolCallId: "callProcess",
});
const workingRowKey = await page
.locator("[data-virtual-row-key^='stream-run:']")
.locator("[data-virtual-row-key^='agent-run:']")
.last()
.getAttribute("data-virtual-row-key");
const finalText = Array.from(
@@ -381,11 +384,7 @@ suite.define(() => {
});
await expect
.poll(() =>
page
.locator(
"[data-virtual-row-key^='stream-run:'] .chat-group.assistant:not(.chat-group--working)",
)
.count(),
page.locator("[data-virtual-row-key^='agent-run:'] .chat-bubble.streaming").count(),
)
.toBe(0);
await gateway.emitChatFinal({ runId, text: finalText });
@@ -137,9 +137,14 @@ suite.define(() => {
sessionKey: "global",
state: "delta",
});
await page.getByText("First token visible.").waitFor({ timeout: 10_000 });
const transcript = page.locator(".chat-thread-inner");
await transcript.getByText("First token visible.", { exact: true }).waitFor({
timeout: 10_000,
});
await page.locator(".chat-thread").getByText(prompt).waitFor({ timeout: 10_000 });
await page.getByText("First token visible.").waitFor({ timeout: 10_000 });
await transcript.getByText("First token visible.", { exact: true }).waitFor({
timeout: 10_000,
});
await expect
.poll(() => page.locator('[data-chat-model-option="openai/startup-model"]').count())
.toBe(1);
+17 -10
View File
@@ -713,7 +713,7 @@ suite.define(() => {
state: "delta",
});
await page.getByText(response).waitFor({ timeout: 10_000 });
await page.locator(".chat-thread-inner").getByText(response).waitFor({ timeout: 10_000 });
await indicator.waitFor({ timeout: 10_000 });
const streamingLayout = await pendingRow.evaluate(
(row, visibleResponse) => ({
@@ -947,7 +947,8 @@ suite.define(() => {
sessionKey: "main",
state: "delta",
});
await page.getByText("I will inspect the file.").waitFor({ timeout: 10_000 });
const transcript = page.locator(".chat-thread-inner");
await transcript.getByText("I will inspect the file.").waitFor({ timeout: 10_000 });
await gateway.emitGatewayEvent("agent", {
data: {
@@ -981,20 +982,26 @@ suite.define(() => {
.poll(() => page.locator(".chat-bubble.streaming code.language-ts").textContent())
.toContain("const answer = 42;");
const visibleOrder = await page.locator(".chat-thread").evaluate((thread: Element) => {
return Array.from(thread.querySelectorAll(".chat-group")).flatMap((group: Element) => {
const text = group.textContent ?? "";
if (text.includes("I will inspect the file.")) {
const composedGroup = transcript
.locator(".chat-group.assistant")
.filter({ hasText: "I will inspect the file." });
expect(await composedGroup.count()).toBe(1);
const visibleOrder = await composedGroup.evaluate((group: Element) =>
Array.from(group.querySelectorAll(".chat-bubble")).flatMap((bubble: Element) => {
if ((bubble.textContent ?? "").includes("I will inspect the file.")) {
return ["assistant stream"];
}
if (group.querySelector('[data-message-id^="tool:assistant:call-read"]')) {
if (bubble.matches('[data-message-id^="tool:assistant:call-read"]')) {
return ["tool card"];
}
if ((bubble.textContent ?? "").includes("const answer = 42;")) {
return ["assistant continuation"];
}
return [];
});
});
}),
);
expect(visibleOrder).toEqual(["assistant stream", "tool card"]);
expect(visibleOrder).toEqual(["assistant stream", "tool card", "assistant continuation"]);
} finally {
await suite.closeBrowserContext(context);
}
+34
View File
@@ -59,6 +59,40 @@ suite.define(() => {
});
});
it("keeps a different active run in its own status row", async () => {
const context = await suite.browser.newContext({ viewport: { height: 800, width: 1200 } });
const currentPage = await context.newPage();
page = currentPage;
await installMockGateway(currentPage, {
historyMessages: [
{
role: "assistant",
content: "Older run result.",
timestamp: Date.now() - 1_000,
__openclaw: { id: "older-result", idempotencyKey: "older-run" },
},
],
inFlightRun: { runId: "newer-run", text: "" },
sessionInfo: {
activeRunIds: ["newer-run"],
hasActiveRun: true,
key: "main",
},
});
await currentPage.goto(`${suite.server?.baseUrl ?? ""}chat`);
await currentPage.getByText("Older run result.", { exact: true }).waitFor();
await currentPage.locator(".chat-reading-indicator").waitFor();
expect(await currentPage.locator(".chat-group.assistant").count()).toBe(2);
expect(
await currentPage
.locator(".chat-group.assistant", { hasText: "Older run result." })
.locator(".chat-working-indicator--continuation")
.count(),
).toBe(0);
});
it("restores only the unpersisted assistant response after reconnecting", async () => {
const artifactDir = path.resolve(".artifacts/control-ui-e2e/chat-inflight-reconnect");
const captureProof = process.env.OPENCLAW_CAPTURE_UI_PROOF === "1";
@@ -657,7 +657,7 @@ suite.define(() => {
sessionKey: "main",
state: "delta",
});
await page.getByText("Working on it.").waitFor();
await page.locator(".chat-thread-inner").getByText("Working on it.").waitFor();
const runningRow = page.locator(".chat-tool-row--running");
await runningRow.waitFor();
@@ -882,7 +882,9 @@ suite.define(() => {
});
}
const activity = page.locator(".chat-group--activity");
const activity = page.locator(".chat-activity-group", {
has: page.locator(`.chat-activity-group__review-status[data-outcome="${groupOutcome}"]`),
});
const summary = activity.locator(".chat-activity-group__summary");
await summary.waitFor();
const status = activity.locator(
+19 -6
View File
@@ -319,7 +319,11 @@ suite.define(() => {
const panes = page.locator("openclaw-chat-pane.chat-split-view__pane");
await expect.poll(() => panes.count()).toBe(2);
for (const pane of await panes.all()) {
await expect.poll(() => pane.getByText(partialText, { exact: true }).count()).toBe(1);
await expect
.poll(() =>
pane.locator(".chat-thread-inner").getByText(partialText, { exact: true }).count(),
)
.toBe(1);
}
await gateway.deferNext("sessions.move");
@@ -383,7 +387,10 @@ suite.define(() => {
.poll(() => page.getByRole("button", { name: "Device offline" }).count())
.toBe(0);
for (const pane of await panes.all()) {
await expect.poll(() => pane.getByText(partialText, { exact: true }).count()).toBe(1);
const transcript = pane.locator(".chat-thread-inner");
await expect
.poll(() => transcript.getByText(partialText, { exact: true }).count())
.toBe(1);
await expect
.poll(() => pane.locator(`[data-entry-id="${abandonedPartialIdentity.id}"]`).count())
.toBe(1);
@@ -453,8 +460,11 @@ suite.define(() => {
await gateway.emitChatFinal({ runId: localRunId, sessionKey, text: finalText });
for (const pane of await panes.all()) {
await expect.poll(() => pane.getByText(partialText, { exact: true }).count()).toBe(1);
await expect.poll(() => pane.getByText(finalText, { exact: true }).count()).toBe(1);
const transcript = pane.locator(".chat-thread-inner");
await expect
.poll(() => transcript.getByText(partialText, { exact: true }).count())
.toBe(1);
await expect.poll(() => transcript.getByText(finalText, { exact: true }).count()).toBe(1);
expect(await pane.locator(".chat-duplicate-count").count()).toBe(0);
expect(await pane.locator(`[data-entry-id="${localFinalIdentity.id}"]`).count()).toBe(1);
}
@@ -471,8 +481,11 @@ suite.define(() => {
const reloadedPanes = page.locator("openclaw-chat-pane.chat-split-view__pane");
await expect.poll(() => reloadedPanes.count()).toBe(2);
for (const pane of await reloadedPanes.all()) {
await expect.poll(() => pane.getByText(partialText, { exact: true }).count()).toBe(1);
await expect.poll(() => pane.getByText(finalText, { exact: true }).count()).toBe(1);
const transcript = pane.locator(".chat-thread-inner");
await expect
.poll(() => transcript.getByText(partialText, { exact: true }).count())
.toBe(1);
await expect.poll(() => transcript.getByText(finalText, { exact: true }).count()).toBe(1);
expect(await pane.locator(".chat-duplicate-count").count()).toBe(0);
expect(await pane.locator(`[data-entry-id="${localFinalIdentity.id}"]`).count()).toBe(1);
}
+17 -2
View File
@@ -115,8 +115,22 @@ export type ChatItem =
action?: { kind: "session-checkpoints"; label: string };
timestamp: number;
}
| { kind: "stream"; key: string; text: string; startedAt: number; isStreaming: boolean }
| { kind: "reading-indicator"; key: string; startedAt: number }
| {
kind: "stream";
key: string;
text: string;
startedAt: number;
isStreaming: boolean;
runId?: string;
boundaryId?: string;
}
| {
kind: "reading-indicator";
key: string;
startedAt: number;
runId?: string;
boundaryId?: string;
}
| { kind: "question"; key: string; questionId: string; startedAt: number };
export type ChatStreamSegment = {
@@ -177,6 +191,7 @@ export type MessageGroup = {
messages: Array<{ message: unknown; key: string; duplicateCount?: number }>;
timestamp: number;
isStreaming: boolean;
runId?: string;
};
/** Content item types in a normalized message */
@@ -0,0 +1,390 @@
import { describe, expect, it } from "vitest";
import type { MessageGroup } from "../../lib/chat/chat-types.ts";
import { coalesceAgentRunFrames } from "./chat-agent-run-grouping.ts";
import type {
ActivityRunRenderItem,
StreamRunRenderItem,
WorkGroupRenderItem,
} from "./chat-thread-grouping.ts";
function group(
role: "assistant" | "tool" | "user",
key: string,
runId: string | undefined,
overrides: Record<string, unknown> = {},
): MessageGroup {
return {
kind: "group",
key: `group:${key}`,
role,
messages: [
{
key,
message: {
role: role === "tool" ? "toolResult" : role,
content: key,
timestamp: 1,
...overrides,
},
},
],
timestamp: 1,
isStreaming: false,
...(runId ? { runId } : {}),
};
}
function userBoundary(sendId = "send-1"): MessageGroup {
return group("user", `user:${sendId}`, undefined, {
__openclaw: { id: `user:${sendId}`, idempotencyKey: `${sendId}:user` },
});
}
type AgentRunFrameRenderItem = Extract<
ReturnType<typeof coalesceAgentRunFrames>[number],
{ kind: "agent-run-frame" }
>;
function requireFrame(
value: ReturnType<typeof coalesceAgentRunFrames>[number] | undefined,
): AgentRunFrameRenderItem {
if (value?.kind !== "agent-run-frame") {
throw new Error("expected an agent run frame");
}
return value;
}
describe("coalesceAgentRunFrames", () => {
it("keeps one lifecycle-stable frame key while preserving semantic part keys", () => {
const runId = "run-1";
const stream: StreamRunRenderItem = {
kind: "stream-run",
key: "stream-run:run-1",
runId,
boundaryId: "send:send-1",
parts: [
{
kind: "stream",
key: "stream:run-1",
text: "Working on it.",
startedAt: 1,
isStreaming: true,
runId,
boundaryId: "send:send-1",
},
],
};
const tool = group("tool", "tool:run-1", runId);
const activity: ActivityRunRenderItem = {
kind: "activity-run",
key: "activity:tool:run-1",
groups: [tool],
};
const final = group("assistant", "assistant:run-1", runId);
const work: WorkGroupRenderItem = {
kind: "work-group",
key: "work:assistant:run-1",
groups: [tool],
durationMs: 1,
};
const boundary = userBoundary();
const streaming = requireFrame(coalesceAgentRunFrames([boundary, stream])[1]);
const tooling = requireFrame(coalesceAgentRunFrames([boundary, stream, activity])[1]);
const history = requireFrame(coalesceAgentRunFrames([boundary, work, final])[1]);
expect(streaming.key).toBe(tooling.key);
expect(tooling.key).toBe(history.key);
expect(history.key).toContain(JSON.stringify([runId, "send:send-1"]));
expect(tooling.parts.map((part) => part.key)).toEqual([stream.key, activity.key]);
expect(history.parts.map((part) => part.key)).toEqual([work.key, final.key]);
});
it("keeps the live send frame identity when a hidden boundary materializes in history", () => {
const runId = "run-heartbeat-handoff";
const stream: StreamRunRenderItem = {
kind: "stream-run",
key: "stream-run:heartbeat-handoff",
runId,
boundaryId: `send:${runId}`,
parts: [
{
kind: "reading-indicator",
key: "reading:heartbeat-handoff",
startedAt: 1,
runId,
boundaryId: `send:${runId}`,
},
],
};
const live = requireFrame(coalesceAgentRunFrames([userBoundary(runId), stream])[1]);
const persistedBoundary = group("assistant", "persisted-after-heartbeat", runId, {
api: "cli",
idempotencyKey: `cli-assistant:${runId}`,
__openclaw: {
id: "persisted-after-heartbeat",
turnBoundary: true,
},
});
const history = requireFrame(coalesceAgentRunFrames([persistedBoundary])[0]);
expect(history.boundaryId).toBe(`send:${runId}`);
expect(history.key).toBe(live.key);
});
it("remounts a large live stream after steer without destabilizing ordinary frames", () => {
const runId = "run-steered";
const boundaryId = "send:steer-run";
const working: StreamRunRenderItem = {
kind: "stream-run",
key: "stream-run:working",
runId,
boundaryId,
parts: [{ kind: "reading-indicator", key: "working", startedAt: 1, runId, boundaryId }],
};
const streamed: StreamRunRenderItem = {
kind: "stream-run",
key: "stream-run:stream-after-steer",
runId,
boundaryId,
parts: [
{
kind: "stream",
key: "working:after:steer-run",
text: "Large terminal response",
startedAt: 2,
isStreaming: true,
runId,
boundaryId,
},
],
};
expect(
requireFrame(coalesceAgentRunFrames([userBoundary("steer-run"), working])[1]).key,
).not.toBe(requireFrame(coalesceAgentRunFrames([userBoundary("steer-run"), streamed])[1]).key);
});
it("keeps different and missing run identities outside the same frame", () => {
const first = group("assistant", "first", "run-1");
const second = group("assistant", "second", "run-2");
const unowned = group("assistant", "unowned", undefined);
const items = coalesceAgentRunFrames([userBoundary(), first, second, unowned]);
expect(items.map((item) => item.kind)).toEqual([
"group",
"agent-run-frame",
"agent-run-frame",
"group",
]);
expect(requireFrame(items[1]).runId).toBe("run-1");
expect(requireFrame(items[2]).runId).toBe("run-2");
});
it("does not compose across forwarded sessions_send input", () => {
const boundary = group("assistant", "forwarded", "run-1", {
provenance: { kind: "inter_session", sourceTool: "sessions_send" },
});
const items = coalesceAgentRunFrames([
userBoundary(),
group("assistant", "before", "run-1"),
boundary,
group("assistant", "after", "run-1"),
]);
expect(items.filter((item) => item.kind === "agent-run-frame")).toHaveLength(1);
expect(items).toContain(boundary);
expect(items.at(-1)).toMatchObject({ kind: "group", key: "group:after" });
});
it("starts a new frame at an authoritative projected turn boundary", () => {
const projected = group("assistant", "steer-output", "run-1", {
__openclaw: { id: "steer-entry", turnBoundary: true },
});
const items = coalesceAgentRunFrames([
userBoundary(),
group("assistant", "before", "run-1"),
projected,
]);
const frames = items.filter(
(item): item is AgentRunFrameRenderItem => item.kind === "agent-run-frame",
);
expect(frames).toHaveLength(2);
expect(frames.map((frame) => frame.boundaryId)).toEqual(["send:send-1", "entry:steer-entry"]);
});
it("treats notices and dividers as hard boundaries", () => {
const notice = { kind: "notice" as const, key: "notice", text: "Notice", timestamp: 2 };
const divider = { kind: "divider" as const, key: "divider", label: "Reset", timestamp: 3 };
const items = coalesceAgentRunFrames([
userBoundary(),
group("assistant", "before", "run-1"),
notice,
group("assistant", "between", "run-1"),
divider,
group("assistant", "after", "run-1"),
]);
expect(items.filter((item) => item.kind === "agent-run-frame")).toHaveLength(1);
expect(items).toContain(notice);
expect(items).toContain(divider);
});
it("gives a restored run segment a unique key after a hard boundary", () => {
const runId = "run-1";
const notice = { kind: "notice" as const, key: "notice", text: "Notice", timestamp: 2 };
const restoredStream: StreamRunRenderItem = {
kind: "stream-run",
key: "stream-run:restored",
runId,
boundaryId: "send:send-1",
parts: [
{
kind: "reading-indicator",
key: "reading:restored",
startedAt: 3,
runId,
boundaryId: "send:send-1",
},
],
};
const items = coalesceAgentRunFrames([
userBoundary(),
group("assistant", "before", runId),
notice,
restoredStream,
]);
const frames = items.filter(
(item): item is AgentRunFrameRenderItem => item.kind === "agent-run-frame",
);
expect(frames).toHaveLength(2);
expect(frames[0]?.key).not.toBe(frames[1]?.key);
expect(frames[1]?.key).toContain("notice");
});
it("marks active frames active and tool-only terminal frames terminal", () => {
const runId = "run-1";
const activeStream: StreamRunRenderItem = {
kind: "stream-run",
key: "stream-run:active",
runId,
boundaryId: "send:send-1",
parts: [
{
kind: "reading-indicator",
key: "reading",
startedAt: 1,
runId,
boundaryId: "send:send-1",
},
],
};
const active = requireFrame(coalesceAgentRunFrames([userBoundary(), activeStream])[1]);
const toolOnly = requireFrame(
coalesceAgentRunFrames([userBoundary(), group("tool", "tool-only", runId)])[1],
);
expect(active.outcome).toEqual({ kind: "active" });
expect(toolOnly.outcome).toEqual({ kind: "completed", actionOwner: null });
expect(toolOnly.parts.at(-1)).toMatchObject({ role: "tool" });
});
it.each([
{
name: "tool-only completion",
parts: [group("tool", "tool-only", "run-1")],
outcome: { kind: "completed", actionOwner: null },
},
{
name: "tool-use commentary",
parts: [
group("assistant", "commentary-tool", "run-1", {
stopReason: "toolUse",
content: [
{ type: "text", text: "I will inspect it." },
{ type: "tool_call", id: "call-1", name: "read", args: {} },
{ type: "tool_result", id: "call-1", name: "read", text: "done" },
],
}),
],
outcome: { kind: "completed", actionOwner: null },
},
{
name: "persisted keyed commentary",
parts: [
group("assistant", "commentary-stop", "run-1", {
stopReason: "stop",
openclawStreamFallback: {
replacementText: "I will inspect it.",
source: "segment",
itemId: "commentary-1",
},
}),
group("tool", "commentary-tool", "run-1"),
],
outcome: { kind: "completed", actionOwner: null },
},
{
name: "Codex reasoning mirror",
parts: [
group("assistant", "reasoning", "run-1", {
stopReason: "stop",
__openclaw: { mirrorOrigin: "codex-app-server", runId: "run-1" },
}),
group("tool", "reasoning-tool", "run-1"),
],
outcome: { kind: "completed", actionOwner: null },
},
{
name: "explicit final followed by work",
parts: [
group("assistant", "final", "run-1", {
phase: "final_answer",
content: "Finished.",
}),
group("tool", "trailing-tool", "run-1"),
],
outcome: { kind: "completed", actionOwner: { key: "final" } },
},
])("records $name without deriving completion from the last part", ({ parts, outcome }) => {
const frame = requireFrame(coalesceAgentRunFrames([userBoundary(), ...parts])[1]);
expect(frame).toMatchObject({ outcome });
});
it("marks preceding commentary failed when an error closes the run", () => {
const error = group("assistant", "error", "run-1", { stopReason: "error" });
const items = coalesceAgentRunFrames([
userBoundary(),
group("assistant", "commentary", "run-1", { phase: "commentary" }),
error,
]);
expect(requireFrame(items[1])).toMatchObject({
outcome: { kind: "failed" },
parts: [{ key: "group:commentary" }, { key: "group:error" }],
});
});
it.each([
{ name: "placement abort", terminal: { stopReason: "stop", openclawAbort: { aborted: true } } },
{ name: "timeout", terminal: { stopReason: "timeout" } },
])("marks an interrupted partial failed for $name", ({ terminal }) => {
const frame = requireFrame(
coalesceAgentRunFrames([
userBoundary(),
group("assistant", "partial", "run-1", { content: "Partial answer", ...terminal }),
])[1],
);
expect(frame.outcome).toEqual({ kind: "failed" });
});
it("leaves active search projections uncomposed", () => {
const input = [userBoundary(), group("assistant", "match", "run-1")];
expect(coalesceAgentRunFrames(input, { searchActive: true })).toBe(input);
});
});
@@ -0,0 +1,283 @@
import { readSessionMessageIdentity } from "@openclaw/gateway-client/browser";
import { asNullableRecord as asRecord } from "@openclaw/normalization-core/record-coerce";
import { resolveAssistantMessagePhase } from "../../../../src/shared/chat-message-content.js";
import type { MessageGroup } from "../../lib/chat/chat-types.ts";
import { extractTextCached } from "../../lib/chat/message-extract.ts";
import type {
ActivityRunRenderItem,
CompletedTurnRenderItem,
StreamRunRenderItem,
WorkGroupRenderItem,
} from "./chat-thread-grouping.ts";
import { isKeyedAssistantStreamFallbackMessage } from "./chat-thread-run-identity.ts";
import { assistantGroupIsForwardedBoundary, chatItemStartsUserTurn } from "./chat-turn-boundary.ts";
import { readLiveTerminalDisposition } from "./terminal-message-identity.ts";
type AgentRunFramePart =
| MessageGroup
| WorkGroupRenderItem
| ActivityRunRenderItem
| StreamRunRenderItem;
export type AgentRunFrameRenderItem = {
kind: "agent-run-frame";
key: string;
runId: string;
boundaryId: string;
outcome:
| { kind: "active" }
| { kind: "completed"; actionOwner: MessageGroup["messages"][number] | null }
| { kind: "failed" };
parts: AgentRunFramePart[];
};
type AgentRunFrameInput = CompletedTurnRenderItem | ActivityRunRenderItem;
function itemGroups(item: AgentRunFramePart): MessageGroup[] {
if (item.kind === "group") {
return [item];
}
if (item.kind === "work-group" || item.kind === "activity-run") {
return item.groups;
}
return [];
}
function itemRunId(item: AgentRunFramePart): string | undefined {
if (item.kind === "stream-run") {
return item.runId;
}
const runIds = itemGroups(item).map((group) => group.runId);
const uniqueRunIds = new Set(runIds.filter((value) => value !== undefined));
return runIds.length > 0 && uniqueRunIds.size === 1 && runIds.every(Boolean)
? uniqueRunIds.values().next().value
: undefined;
}
function messageIsInterrupted(message: unknown): boolean {
const record = asRecord(message);
const stopReason = typeof record?.stopReason === "string" ? record.stopReason.toLowerCase() : "";
return (
readLiveTerminalDisposition(message) !== null ||
asRecord(record?.openclawAbort)?.aborted === true ||
["aborted", "cancelled", "canceled", "timeout", "timed_out"].includes(stopReason)
);
}
function itemFailsFrame(item: AgentRunFramePart): boolean {
return itemGroups(item).some((group) =>
group.messages.some(
({ message }) => messageIsInterrupted(message) || asRecord(message)?.stopReason === "error",
),
);
}
function itemIsActive(item: AgentRunFramePart): boolean {
if (item.kind === "stream-run") {
return item.parts.some(
(part) => part.kind === "reading-indicator" || (part.kind === "stream" && part.isStreaming),
);
}
return itemGroups(item).some((group) => group.isStreaming);
}
function itemBoundaryId(item: AgentRunFramePart): string | undefined {
return item.kind === "stream-run" ? item.boundaryId : undefined;
}
function groupBoundaryId(group: MessageGroup): string | undefined {
const firstMessage = group.messages[0]?.message;
const identity = readSessionMessageIdentity(firstMessage);
if (!chatItemStartsUserTurn(group)) {
return undefined;
}
const runId = identity?.runId;
if (runId) {
return `send:${runId}`;
}
return identity?.id ? `entry:${identity.id}` : undefined;
}
function isExternalBoundary(group: MessageGroup): boolean {
return group.role === "user" || assistantGroupIsForwardedBoundary(group);
}
function itemBoundaryGroup(item: AgentRunFramePart): MessageGroup | undefined {
const first = itemGroups(item)[0];
return first && chatItemStartsUserTurn(first) ? first : undefined;
}
function frameKey(runId: string, boundaryId: string, segmentId: string | undefined): string {
return `agent-run:${JSON.stringify(segmentId ? [runId, boundaryId, segmentId] : [runId, boundaryId])}`;
}
function frameSegmentId(
parts: AgentRunFramePart[],
hardBoundaryId: string | undefined,
): string | undefined {
return (
hardBoundaryId ??
parts
.flatMap((part) => (part.kind === "stream-run" ? part.parts : []))
.find((part) => part.kind === "stream" && part.key.includes(":after:"))?.key
);
}
export function agentRunFrameGroups(frame: AgentRunFrameRenderItem): MessageGroup[] {
return frame.parts.flatMap(itemGroups);
}
function messageCanOwnCompletedFrame(message: unknown, explicitOnly: boolean): boolean {
const record = asRecord(message);
const phase = resolveAssistantMessagePhase(message);
const stopReason = record?.stopReason;
const metadata = asRecord(record?.["__openclaw"]);
if (
!extractTextCached(message)?.trim() ||
isKeyedAssistantStreamFallbackMessage(message) ||
messageIsInterrupted(message) ||
phase === "commentary" ||
stopReason === "toolUse" ||
stopReason === "error" ||
metadata?.runtimeActivityKind === "context_compaction" ||
(metadata?.mirrorOrigin === "codex-app-server" && metadata.runTerminal !== true)
) {
return false;
}
return !explicitOnly || phase === "final_answer" || stopReason === "stop";
}
function completedFrameActionOwner(
parts: AgentRunFramePart[],
): MessageGroup["messages"][number] | null {
const messages = parts
.flatMap(itemGroups)
.flatMap((group) => (group.role === "assistant" ? group.messages : []));
const explicit = messages.findLast(({ message }) => messageCanOwnCompletedFrame(message, true));
if (explicit) {
return explicit;
}
const lastPart = parts.at(-1);
if (lastPart?.kind !== "group" || lastPart.role !== "assistant") {
return null;
}
const lastMessage = lastPart.messages.at(-1);
return lastMessage
? messageCanOwnCompletedFrame(lastMessage.message, false)
? lastMessage
: null
: null;
}
export function agentRunFrameActiveStatusParts(
frame: AgentRunFrameRenderItem,
): StreamRunRenderItem["parts"] | undefined {
if (frame.outcome.kind !== "active") {
return undefined;
}
const parts = frame.parts.flatMap((part) => (part.kind === "stream-run" ? part.parts : []));
return parts.length > 0 &&
frame.parts.every(
(part) =>
part.kind === "stream-run" &&
part.parts.every((streamPart) => streamPart.kind === "reading-indicator"),
)
? parts
: undefined;
}
function isAgentRunFramePart(item: AgentRunFrameInput): item is AgentRunFramePart {
return (
item.kind === "group" ||
item.kind === "work-group" ||
item.kind === "activity-run" ||
item.kind === "stream-run"
);
}
/** Wrap semantic work/activity rows in one run-owned presentation frame. */
export function coalesceAgentRunFrames(
items: AgentRunFrameInput[],
opts: { searchActive?: boolean } = {},
): Array<AgentRunFrameInput | AgentRunFrameRenderItem> {
if (opts.searchActive) {
return items;
}
const result: Array<AgentRunFrameInput | AgentRunFrameRenderItem> = [];
let boundaryId: string | undefined;
let segmentId: string | undefined;
let runId: string | undefined;
let parts: AgentRunFramePart[] = [];
const flush = (failed = false) => {
if (!runId || !boundaryId || parts.length === 0) {
return;
}
result.push({
kind: "agent-run-frame",
key: frameKey(runId, boundaryId, frameSegmentId(parts, segmentId)),
runId,
boundaryId,
outcome: failed
? { kind: "failed" }
: parts.some(itemIsActive)
? { kind: "active" }
: { kind: "completed", actionOwner: completedFrameActionOwner(parts) },
parts,
});
parts = [];
runId = undefined;
};
for (const item of items) {
if (!isAgentRunFramePart(item)) {
flush();
result.push(item);
boundaryId = undefined;
segmentId = item.key;
continue;
}
const candidate = item;
const boundaryGroup = itemBoundaryGroup(candidate);
if (boundaryGroup) {
flush();
segmentId = undefined;
const nextBoundaryId = groupBoundaryId(boundaryGroup);
if (isExternalBoundary(boundaryGroup) || !nextBoundaryId) {
result.push(item);
boundaryId = nextBoundaryId;
continue;
}
boundaryId = nextBoundaryId;
}
const candidateBoundaryId = itemBoundaryId(candidate);
if (candidateBoundaryId && candidateBoundaryId !== boundaryId) {
flush();
boundaryId = candidateBoundaryId;
}
const candidateRunId = itemRunId(candidate);
if (boundaryId && candidateRunId && itemFailsFrame(candidate)) {
if (runId && runId !== candidateRunId) {
flush();
}
runId = candidateRunId;
parts.push(candidate);
flush(true);
boundaryId = undefined;
segmentId = item.key;
continue;
}
if (!boundaryId || !candidateRunId) {
flush();
result.push(item);
boundaryId = undefined;
segmentId = item.key;
continue;
}
if (runId && runId !== candidateRunId) {
flush();
}
runId = candidateRunId;
parts.push(candidate);
}
flush();
return result;
}
+21 -1
View File
@@ -17,6 +17,7 @@ import {
type ChatEventPayload,
type ChatState,
} from "./chat-history.ts";
import { transcriptRunId } from "./chat-thread-run-identity.ts";
import {
getChatSessionProjection,
publishChatSessionProjectionMessages,
@@ -446,6 +447,7 @@ function handleChatEvent(state: ChatState, payload?: ChatEventPayload) {
normalizedMessage,
terminalRunId,
terminalAfterBoundaryRunId,
"aborted",
);
publishVisibleTerminal(
normalizedMessage,
@@ -491,14 +493,32 @@ function handleChatEvent(state: ChatState, payload?: ChatEventPayload) {
visiblePayloadMessage,
terminalRunId,
terminalAfterBoundaryRunId,
projectedRun?.currentRun?.status === "timeout" ? "timeout" : "error",
),
);
} else {
state.chatMessages = materializeVisibleStream({ includeCurrent: true });
state.chatMessages = [...state.chatMessages, visiblePayloadMessage];
state.chatMessages = [
...state.chatMessages,
rememberLiveTerminalRun(
visiblePayloadMessage,
terminalRunId,
terminalAfterBoundaryRunId,
projectedRun?.currentRun?.status === "timeout" ? "timeout" : "error",
),
];
}
} else {
state.chatMessages = materializeVisibleStream({ includeCurrent: true });
const materialized = state.chatMessages.findLast(
(message) => transcriptRunId(message) === terminalRunId,
);
rememberLiveTerminalRun(
materialized,
terminalRunId,
terminalAfterBoundaryRunId,
projectedRun?.currentRun?.status === "timeout" ? "timeout" : "error",
);
}
}
// The shared Gateway projection owns timeout classification; preserve it
+28 -1
View File
@@ -1,10 +1,37 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { resetWorkingProgress, resolveTurnRecap } from "./chat-progress.ts";
import { resetWorkingProgress, resolveTurnRecap, resolveWorkingProgress } from "./chat-progress.ts";
const SESSION = "agent:main:main";
const PREVIOUS_ENDED_AT = 900_000;
const RUN_ENDED_AT = 1_000_000;
describe("resolveWorkingProgress", () => {
beforeEach(() => resetWorkingProgress());
afterEach(() => resetWorkingProgress());
it("prefers observed stream identity over a future queued send", () => {
expect(
resolveWorkingProgress(
SESSION,
null,
1_000,
[
{
id: "future-send",
text: "Run next",
createdAt: 2_000,
sendRunId: "future-run",
sendState: "waiting-reconnect",
sendAttempts: 1,
},
],
[{ ts: 1_000, runId: "active-run" }],
[],
),
).toMatchObject({ runId: "active-run" });
});
});
const doneRow = (endedAt: number, runtimeMs = 51_000, outputTokens?: number) => ({
status: "done",
endedAt,
+72 -14
View File
@@ -1,29 +1,39 @@
import { asNullableRecord as asRecord } from "@openclaw/normalization-core/record-coerce";
import { t } from "../../i18n/index.ts";
import type { ChatItem, ChatQueueItem } from "../../lib/chat/chat-types.ts";
import type { ChatGuardianNotice, ChatItem, ChatQueueItem } from "../../lib/chat/chat-types.ts";
import { formatCompactTokenCount } from "../../lib/format.ts";
type WorkingProgress = {
key: string;
runId: string | null;
startedAt: number;
};
type WorkingProgressCache = WorkingProgress & {
runId: string | null;
};
type WorkingProgressCache = WorkingProgress;
const CONTEXT_COMPACTION_CUSTOM_TYPE = "openclaw.context-compaction";
export function isContextCompactionActivity(message: unknown): boolean {
return asRecord(asRecord(message)?.["__openclaw"])?.runtimeActivityKind === "context_compaction";
}
export function projectContextCompactionActivity(message: unknown): unknown {
const record = asRecord(message);
if (record?.role !== "custom" || record.customType !== CONTEXT_COMPACTION_CUSTOM_TYPE) {
return message;
}
const metadata = asRecord(record["__openclaw"]);
const details = asRecord(record.details);
const { idempotencyKey: _activityId, ...activity } = record;
return {
...record,
...activity,
role: "assistant",
content: [{ type: "text", text: t("chat.composer.contextCompacted") }],
...(typeof metadata?.runId === "string"
? { runId: metadata.runId }
: typeof details?.runId === "string"
? { runId: details.runId }
: {}),
__openclaw: {
...metadata,
runtimeActivityKind: "context_compaction",
@@ -34,6 +44,46 @@ export function projectContextCompactionActivity(message: unknown): unknown {
const workingProgressBySession = new Map<string, WorkingProgressCache>();
let anonymousWorkingProgressId = 0;
export function buildGuardianNoticeItem(
notice: ChatGuardianNotice,
): Extract<ChatItem, { kind: "notice" }> {
const action = notice.command ?? t("chat.systemNotice.guardian.requestedAction");
if (notice.kind === "approved") {
return {
kind: "notice",
key: notice.key,
icon: "shieldCheck",
label: t("chat.systemNotice.guardian.approvedSummary", { action }),
text: "",
timestamp: notice.timestamp,
};
}
if (notice.kind === "warning") {
return {
kind: "notice",
key: notice.key,
icon: "shieldCheck",
label: t("chat.systemNotice.guardian.warningLabel"),
text: notice.message ?? t("chat.systemNotice.guardian.warningFallback"),
timestamp: notice.timestamp,
tone: "danger",
};
}
return {
kind: "notice",
key: notice.key,
icon: "shieldCheck",
label: t("chat.systemNotice.guardian.deniedLabel"),
text: t("chat.systemNotice.guardian.deniedSummary", {
action,
risk: notice.riskLevel ?? t("chat.systemNotice.guardian.unknownRisk"),
rationale: notice.rationale ?? t("chat.systemNotice.guardian.noRationale"),
}),
timestamp: notice.timestamp,
tone: "danger",
};
}
export function buildCompactionDividerItem(
marker: Record<string, unknown>,
timestamp: number,
@@ -104,18 +154,26 @@ export function resolveWorkingProgress(
runId: string | null,
streamStartedAt: number | null,
queue: ChatQueueItem[],
streamSegments: Array<{ ts: number }>,
streamSegments: Array<{ ts: number; runId?: string }>,
toolMessages: unknown[],
): WorkingProgress {
const queuedRunId =
queue.find((item) => item.sendState === "sending" && shouldRenderQueuedSendInThread(item))
?.sendRunId ?? queue.find(shouldRenderQueuedSendInThread)?.sendRunId;
const toolRunId = toolMessages
.map((message) => (message as Record<string, unknown> | null)?.runId)
.find(
const queuedProgress =
queue.find((item) => item.sendState === "sending" && shouldRenderQueuedSendInThread(item)) ??
queue.find(shouldRenderQueuedSendInThread);
const queuedRunId = queuedProgress?.sendRunId ?? queuedProgress?.pendingRunId;
const segmentRunId = streamSegments
.map((segment) => segment.runId)
.findLast(
(candidate): candidate is string => typeof candidate === "string" && candidate.length > 0,
);
const explicitRunId = queuedRunId ?? runId ?? toolRunId;
const toolRunId = toolMessages
.map((message) => (message as Record<string, unknown> | null)?.runId)
.findLast(
(candidate): candidate is string => typeof candidate === "string" && candidate.length > 0,
);
// Stream and tool facts describe work already observed in this row. Queue
// identity is only a pre-run fallback and must not claim an active tail.
const explicitRunId = runId ?? segmentRunId ?? toolRunId ?? queuedRunId;
const cached = workingProgressBySession.get(sessionKey);
const compatibleCached =
cached && (!explicitRunId || !cached.runId || cached.runId === explicitRunId) ? cached : null;
@@ -146,7 +204,7 @@ export function resolveWorkingProgress(
runId: explicitRunId ?? compatibleCached?.runId ?? null,
startedAt,
});
return { key, startedAt };
return { key, runId: explicitRunId ?? compatibleCached?.runId ?? null, startedAt };
}
export function clearWorkingProgress(sessionKey: string): void {
@@ -1384,6 +1384,11 @@ describeBrowserLayout.concurrent("chat responsive browser layout", () => {
<div class="chat-activity-group">Activity</div>
</div>
</div>
<div class="chat-group assistant chat-group--with-footer">
<div class="chat-group-messages" data-frame-lane>
<div class="chat-activity-group">Framed activity</div>
</div>
</div>
</div>
</div>
<div class="chat-prs" data-chat-prs>Pull requests</div>
@@ -1398,10 +1403,11 @@ describeBrowserLayout.concurrent("chat responsive browser layout", () => {
return { center: bounds.x + bounds.width / 2, width: bounds.width };
};
return {
activity: rect("[data-activity-lane]"),
activity: rect("[data-activity-lane] .chat-activity-group"),
composer: rect("[data-composer]"),
prs: rect("[data-chat-prs]"),
shell: rect("[data-tool-shell]"),
framedActivity: rect("[data-frame-lane] .chat-activity-group"),
thread: rect(".chat-thread-inner"),
tool: rect("[data-tool-lane]"),
};
@@ -1416,9 +1422,10 @@ describeBrowserLayout.concurrent("chat responsive browser layout", () => {
expect(defaults.tool.width).toBeCloseTo(defaults.thread.width, 0);
expect(defaults.shell.width).toBeCloseTo(760, 0);
expect(defaults.activity.width).toBeCloseTo(760, 0);
expect(defaults.framedActivity.width).toBeCloseTo(defaults.activity.width, 0);
const configured = await renderFixture(true);
for (const key of ["activity", "shell", "tool"] as const) {
for (const key of ["activity", "framedActivity", "shell", "tool"] as const) {
expect(configured[key].width).toBeCloseTo(configured.thread.width, 0);
}
expect(configured.composer.width).toBeCloseTo(configured.prs.width, 0);
+27 -99
View File
@@ -22,6 +22,7 @@ import { normalizeRoleForGrouping } from "../../lib/chat/message-normalizer.ts";
import { areUiSessionKeysEquivalent } from "../../lib/sessions/session-key.ts";
import {
buildCompactionDividerItem,
buildGuardianNoticeItem,
buildResetDividerItem,
clearWorkingProgress,
projectContextCompactionActivity,
@@ -29,11 +30,7 @@ import {
shouldRenderQueuedSendInThread,
} from "./chat-progress.ts";
import { chatMessagesContainQueuedSend } from "./chat-send-support.ts";
import {
coalesceToolActivityMessages,
groupMessages,
isKeyedAssistantStreamFallbackMessage,
} from "./chat-thread-grouping.ts";
import { coalesceToolActivityMessages, groupMessages } from "./chat-thread-grouping.ts";
import {
appendCanvasBlockToAssistantMessage,
buildMessageKeys,
@@ -54,9 +51,16 @@ import {
timestampAfterVisibleItems,
transcriptPositionTimestamp,
turnHasMatchingAssistant,
userTurnSendIdentity,
type TurnInsertionBounds,
} from "./chat-thread-items.ts";
import {
findCurrentTurnBounds,
findRunTurnBounds,
isKeyedAssistantStreamFallbackMessage,
optionalBoundaryIdentity,
optionalRunIdentity,
resolveRunInsertionBounds,
} from "./chat-thread-run-identity.ts";
import { safeNormalizeMessage } from "./chat-turn-boundary.ts";
import { resolveSystemNoticeKind } from "./system-notice-kinds.ts";
import { isLiveTerminalForRun } from "./terminal-message-identity.ts";
@@ -92,97 +96,6 @@ export type BuildChatItemsProps = {
searchQuery?: string;
};
function guardianNoticeItem(notice: ChatGuardianNotice): Extract<ChatItem, { kind: "notice" }> {
const action = notice.command ?? t("chat.systemNotice.guardian.requestedAction");
if (notice.kind === "approved") {
return {
kind: "notice",
key: notice.key,
icon: "shieldCheck",
label: t("chat.systemNotice.guardian.approvedSummary", { action }),
text: "",
timestamp: notice.timestamp,
};
}
if (notice.kind === "warning") {
return {
kind: "notice",
key: notice.key,
icon: "shieldCheck",
label: t("chat.systemNotice.guardian.warningLabel"),
text: notice.message ?? t("chat.systemNotice.guardian.warningFallback"),
timestamp: notice.timestamp,
tone: "danger",
};
}
return {
kind: "notice",
key: notice.key,
icon: "shieldCheck",
label: t("chat.systemNotice.guardian.deniedLabel"),
text: t("chat.systemNotice.guardian.deniedSummary", {
action,
risk: notice.riskLevel ?? t("chat.systemNotice.guardian.unknownRisk"),
rationale: notice.rationale ?? t("chat.systemNotice.guardian.noRationale"),
}),
timestamp: notice.timestamp,
tone: "danger",
};
}
function isUserChatItem(item: ChatItem): boolean {
if (item.kind !== "message") {
return false;
}
const normalized = safeNormalizeMessage(item.message);
return normalized ? normalizeRoleForGrouping(normalized.role).toLowerCase() === "user" : false;
}
function findCurrentTurnBounds(items: ChatItem[]): TurnInsertionBounds | null {
const index = items.findLastIndex(isUserChatItem);
const item = items[index];
return index >= 0 && item ? { afterKey: item.key } : null;
}
function findRunTurnBounds(items: ChatItem[], runId: string): TurnInsertionBounds | null {
const sendIdentity = `send:${runId}`;
const index = items.findIndex(
(item) =>
item.kind === "message" &&
isUserChatItem(item) &&
userTurnSendIdentity(item.message) === sendIdentity,
);
const item = items[index];
if (index < 0 || !item) {
return null;
}
const nextUser = items.slice(index + 1).find(isUserChatItem);
return { afterKey: item.key, ...(nextUser ? { beforeKey: nextUser.key } : {}) };
}
function resolveRunInsertionBounds(
items: ChatItem[],
runId: unknown,
currentRunId: string | null | undefined,
currentTurnBounds: TurnInsertionBounds | null,
): TurnInsertionBounds | null {
if (typeof runId !== "string" || !runId.trim()) {
return currentRunId != null ? currentTurnBounds : null;
}
const runBounds = findRunTurnBounds(items, runId);
if (runId === currentRunId) {
// Active runs can span steers: the original prompt is a floor, not a ceiling.
return runBounds ? { afterKey: runBounds.afterKey } : currentTurnBounds;
}
if (runBounds || currentRunId == null) {
return runBounds;
}
// Legacy rows may lack the user-run identity needed for exact bounds. Keep
// their timestamp ordering across historical turns, but never cross the
// current prompt and become current-run output.
return currentTurnBounds?.afterKey ? { beforeKey: currentTurnBounds.afterKey } : null;
}
export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | MessageGroup> {
let items: ChatItem[] = [];
const tools = props.toolMessages.filter((message) => asRecord(message) !== null);
@@ -504,7 +417,7 @@ export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | Mes
};
if (!searchFiltering) {
for (const notice of props.guardianNotices ?? []) {
const item = guardianNoticeItem(notice);
const item = buildGuardianNoticeItem(notice);
timestampedProjectionItems.push(item);
applyRunBounds(item.key, notice.runId);
}
@@ -571,6 +484,8 @@ export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | Mes
text: visibleText,
startedAt: segment.ts,
isStreaming: false,
...optionalRunIdentity(segment.runId),
...optionalBoundaryIdentity(afterBoundaryBySegment.get(segment) ?? segment.runId),
};
timestampedProjectionItems.push(streamItem);
applyRunBounds(
@@ -639,6 +554,8 @@ export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | Mes
text,
startedAt: segment.ts,
isStreaming: false,
...optionalRunIdentity(segment.runId),
...optionalBoundaryIdentity(afterBoundaryBySegment.get(segment) ?? segment.runId),
};
timestampedProjectionItems.push(commentaryItem);
applyRunBounds(
@@ -712,6 +629,7 @@ export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | Mes
const visibleText = trimAccumulatedStreamPrefix(text, previousAccumulatedStreamText);
if (visibleText.length > 0 && !stripHeartbeatTokenForDisplay(visibleText).shouldSkip) {
const liveProgress = resolveProgress();
const liveRunId = props.runId ?? liveProgress.runId;
const liveStreamItem: ChatItem = {
kind: "stream",
key: latestBoundaryRunId
@@ -720,6 +638,8 @@ export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | Mes
text: visibleText,
startedAt: timestampAfterVisibleItems(items, props.streamStartedAt ?? Date.now()),
isStreaming: true,
...optionalRunIdentity(liveRunId),
...optionalBoundaryIdentity(latestBoundaryRunId ?? liveRunId),
};
const liveTurnRunId = latestBoundaryRunId ?? normalizeOptionalString(props.runId);
const liveTurnBounds = liveTurnRunId ? findRunTurnBounds(items, liveTurnRunId) : null;
@@ -732,7 +652,15 @@ export function buildChatItems(props: BuildChatItemsProps): Array<ChatItem | Mes
}
}
if (showWorkingIndicator) {
items.push({ kind: "reading-indicator", ...resolveProgress() });
const workingProgress = resolveProgress();
const workingRunId = props.runId ?? workingProgress.runId;
items.push({
kind: "reading-indicator",
key: workingProgress.key,
startedAt: workingProgress.startedAt,
...optionalRunIdentity(workingRunId),
...optionalBoundaryIdentity(latestBoundaryRunId ?? workingRunId),
});
}
// Future queued turns are a causal ceiling for every current-run projection.
// Append them after tools, streams, progress, and prompts so none can cross the
+38 -25
View File
@@ -9,7 +9,14 @@ import { extractTextCached } from "../../lib/chat/message-extract.ts";
import { normalizeMessage, normalizeRoleForGrouping } from "../../lib/chat/message-normalizer.ts";
import { senderIdentityKey } from "../../lib/chat/sender-label.ts";
import { extractToolCardsCached } from "../../lib/chat/tool-cards.ts";
import { isContextCompactionActivity } from "./chat-progress.ts";
import { resolveMessageToolUseId, resolveToolBlockId } from "./chat-thread-items.ts";
import {
isKeyedAssistantStreamFallbackMessage,
streamPartBoundaryId,
streamPartRunId,
transcriptRunId,
} from "./chat-thread-run-identity.ts";
import {
assistantGroupIsForwardedBoundary,
chatItemStartsUserTurn,
@@ -17,15 +24,6 @@ import {
} from "./chat-turn-boundary.ts";
import { indexTurnContinuations } from "./stream-causal-boundary.ts";
export function isKeyedAssistantStreamFallbackMessage(message: unknown): boolean {
const record = asRecord(message);
if (normalizeLowercaseStringOrEmpty(record?.role) !== "assistant") {
return false;
}
const fallback = asRecord(record?.openclawStreamFallback);
return typeof fallback?.itemId === "string" && fallback.itemId.trim().length > 0;
}
function stampReplyAttribution(
items: Array<ChatItem | MessageGroup>,
): Array<ChatItem | MessageGroup> {
@@ -78,6 +76,8 @@ export function groupMessages(items: ChatItem[]): Array<ChatItem | MessageGroup>
role === "user" || role === "assistant" ? (normalized.senderLabel ?? null) : null;
const sender = role === "user" ? normalized.sender : undefined;
const timestamp = normalized.timestamp || Date.now();
const runId =
role === "assistant" || role === "tool" ? transcriptRunId(item.message) : undefined;
const shouldSplitBySender = role === "user" || role === "assistant";
const startsProjectedTurn =
asRecord(asRecord(item.message)?.["__openclaw"])?.turnBoundary === true;
@@ -96,6 +96,7 @@ export function groupMessages(items: ChatItem[]): Array<ChatItem | MessageGroup>
!currentGroup ||
startsProjectedTurn ||
currentGroup.role !== role ||
currentGroup.runId !== runId ||
splitsAssistantCommentary ||
splitsRuntimeActivity ||
(shouldSplitBySender &&
@@ -114,6 +115,7 @@ export function groupMessages(items: ChatItem[]): Array<ChatItem | MessageGroup>
messages: [{ message: item.message, key: item.key, duplicateCount: item.duplicateCount }],
timestamp,
isStreaming: false,
...(runId ? { runId } : {}),
};
} else {
currentGroup.messages.push({
@@ -129,7 +131,6 @@ export function groupMessages(items: ChatItem[]): Array<ChatItem | MessageGroup>
}
return stampReplyAttribution(result);
}
function mergeToolCallResultPair(callItem: ChatItem, resultItem: ChatItem): ChatItem | null {
if (callItem.kind !== "message" || resultItem.kind !== "message") {
return null;
@@ -468,30 +469,42 @@ export function coalesceToolActivityMessages(items: ChatItem[]): ChatItem[] {
}
type RenderChatItem = ChatItem | MessageGroup;
type StreamRunRenderItem = {
export type StreamRunRenderItem = {
kind: "stream-run";
key: string;
parts: Array<
Extract<ChatItem, { kind: "stream" } | { kind: "reading-indicator" } | { kind: "question" }>
>;
runId?: string;
boundaryId?: string;
parts: Array<Extract<ChatItem, { kind: "stream" | "reading-indicator" | "question" }>>;
};
export function coalesceStreamRuns(
items: RenderChatItem[],
): Array<RenderChatItem | StreamRunRenderItem> {
const result: Array<RenderChatItem | StreamRunRenderItem> = [];
let run: StreamRunRenderItem["parts"] = [];
// Contiguous in-flight stream and reading-indicator items render under one
// assistant avatar; messages, groups, and dividers intentionally break the run.
const flush = () => {
const [first] = run;
if (first) {
result.push({ kind: "stream-run", key: `stream-run:${first.key}`, parts: run });
const runId = streamPartRunId(first);
const boundaryId = streamPartBoundaryId(first);
result.push({
kind: "stream-run",
key: `stream-run:${first.key}`,
parts: run,
...(runId ? { runId } : {}),
...(boundaryId ? { boundaryId } : {}),
});
run = [];
}
};
for (const item of items) {
if (item.kind === "stream" || item.kind === "reading-indicator") {
const first = run[0];
if (
first &&
(streamPartRunId(first) !== item.runId || streamPartBoundaryId(first) !== item.boundaryId)
) {
flush();
}
run.push(item);
continue;
}
@@ -501,15 +514,16 @@ export function coalesceStreamRuns(
flush();
return result;
}
/** Collapsed rollup of a completed turn's intermediate work (tools, commentary). */
type WorkGroupRenderItem = {
export type WorkGroupRenderItem = {
kind: "work-group";
key: string;
groups: MessageGroup[];
durationMs: number | null;
};
type ActivityRunRenderItem = {
export type ActivityRunRenderItem = {
kind: "activity-run";
key: string;
groups: MessageGroup[];
@@ -553,10 +567,6 @@ export function assistantGroupCanOwnActiveRunStatus(group: MessageGroup): boolea
);
}
function isContextCompactionActivity(message: unknown): boolean {
return asRecord(asRecord(message)?.["__openclaw"])?.runtimeActivityKind === "context_compaction";
}
// History carries no final-vs-commentary marker (commentary exists only as
// live stream segments), so the last assistant group with visible content
// stands in for the final reply. Turns whose last content is commentary
@@ -718,7 +728,7 @@ export function collapseCompletedTurnWork(
return result;
}
type CompletedTurnRenderItem = TurnRenderItem | WorkGroupRenderItem;
export type CompletedTurnRenderItem = TurnRenderItem | WorkGroupRenderItem;
/** Presentation-only rollup for tool groups separated by projected turn boundaries. */
export function coalesceActivityRuns(
@@ -742,6 +752,9 @@ export function coalesceActivityRuns(
};
for (const item of items) {
if (item.kind === "group" && item.role.toLowerCase() === "tool") {
if (groups.length > 0 && groups[0]?.runId !== item.runId) {
flush();
}
groups.push(item);
continue;
}
+2
View File
@@ -557,6 +557,7 @@ export function queuedSendThreadMessage(item: ChatQueueItem): Record<string, unk
if (content.length === 0) {
return null;
}
const runId = item.sendRunId ?? item.pendingRunId;
return {
role: "user",
content,
@@ -565,6 +566,7 @@ export function queuedSendThreadMessage(item: ChatQueueItem): Record<string, unk
kind: "pending-send",
id: item.id,
state: item.sendState,
...(runId ? { idempotencyKey: `${runId}:user` } : {}),
...(item.replyToId ? { replyToId: item.replyToId } : {}),
...(item.sender?.id ? { senderId: item.sender.id } : {}),
...(item.sender?.name ? { senderName: item.sender.name } : {}),
@@ -0,0 +1,107 @@
import { readSessionMessageIdentity } from "@openclaw/gateway-client/browser";
import { asNullableRecord as asRecord } from "@openclaw/normalization-core/record-coerce";
import {
normalizeLowercaseStringOrEmpty,
normalizeOptionalString,
} from "@openclaw/normalization-core/string-coerce";
import type { ChatItem } from "../../lib/chat/chat-types.ts";
import { normalizeRoleForGrouping } from "../../lib/chat/message-normalizer.ts";
import { userTurnSendIdentity, type TurnInsertionBounds } from "./chat-thread-items.ts";
import { safeNormalizeMessage } from "./chat-turn-boundary.ts";
import { readLiveTerminalRunId } from "./terminal-message-identity.ts";
export function transcriptRunId(message: unknown): string | undefined {
const identity = readSessionMessageIdentity(message);
if (identity?.runId) {
return identity.runId;
}
const record = asRecord(message);
return (
readLiveTerminalRunId(message) ??
normalizeOptionalString(record?.runId) ??
normalizeOptionalString(asRecord(record?.openclawStreamFallback)?.runId)
);
}
export function isKeyedAssistantStreamFallbackMessage(message: unknown): boolean {
const record = asRecord(message);
if (normalizeLowercaseStringOrEmpty(record?.role) !== "assistant") {
return false;
}
const fallback = asRecord(record?.openclawStreamFallback);
return typeof fallback?.itemId === "string" && fallback.itemId.trim().length > 0;
}
export function optionalRunIdentity(value: unknown): { runId: string } | undefined {
const runId = normalizeOptionalString(value);
return runId ? { runId } : undefined;
}
export function optionalBoundaryIdentity(value: unknown): { boundaryId: string } | undefined {
const runId = normalizeOptionalString(value);
return runId ? { boundaryId: `send:${runId}` } : undefined;
}
export function streamPartRunId(
part: Extract<ChatItem, { kind: "stream" | "reading-indicator" | "question" }>,
): string | undefined {
return part.kind === "question" ? undefined : part.runId;
}
export function streamPartBoundaryId(
part: Extract<ChatItem, { kind: "stream" | "reading-indicator" | "question" }>,
): string | undefined {
return part.kind === "question" ? undefined : part.boundaryId;
}
function isUserChatItem(item: ChatItem): boolean {
if (item.kind !== "message") {
return false;
}
const normalized = safeNormalizeMessage(item.message);
return normalized ? normalizeRoleForGrouping(normalized.role).toLowerCase() === "user" : false;
}
export function findCurrentTurnBounds(items: ChatItem[]): TurnInsertionBounds | null {
const index = items.findLastIndex(isUserChatItem);
const item = items[index];
return index >= 0 && item ? { afterKey: item.key } : null;
}
export function findRunTurnBounds(items: ChatItem[], runId: string): TurnInsertionBounds | null {
const sendIdentity = `send:${runId}`;
const index = items.findIndex(
(item) =>
item.kind === "message" &&
isUserChatItem(item) &&
userTurnSendIdentity(item.message) === sendIdentity,
);
const item = items[index];
if (index < 0 || !item) {
return null;
}
const nextUser = items.slice(index + 1).find(isUserChatItem);
return { afterKey: item.key, ...(nextUser ? { beforeKey: nextUser.key } : {}) };
}
export function resolveRunInsertionBounds(
items: ChatItem[],
runId: unknown,
currentRunId: string | null | undefined,
currentTurnBounds: TurnInsertionBounds | null,
): TurnInsertionBounds | null {
if (typeof runId !== "string" || !runId.trim()) {
return currentRunId != null ? currentTurnBounds : null;
}
const runBounds = findRunTurnBounds(items, runId);
if (runId === currentRunId) {
// Active runs can span steers: the original prompt is a floor, not a ceiling.
return runBounds ? { afterKey: runBounds.afterKey } : currentTurnBounds;
}
if (runBounds || currentRunId == null) {
return runBounds;
}
// Legacy rows may lack the user-run identity needed for exact bounds. Keep
// them ordered before the current prompt instead of attaching them to it.
return currentTurnBounds?.afterKey ? { beforeKey: currentTurnBounds.afterKey } : null;
}
+55
View File
@@ -6,6 +6,7 @@ import { describe, expect, it, vi } from "vitest";
import { markInboundContextLabel } from "../../../../src/auto-reply/reply/inbound-context-marker.js";
import type { MessageGroup } from "../../lib/chat/chat-types.ts";
import * as toolCards from "../../lib/chat/tool-cards.ts";
import { coalesceAgentRunFrames } from "./chat-agent-run-grouping.ts";
import {
assistantGroupCanOwnActiveRunStatus,
buildCachedChatItems,
@@ -767,6 +768,8 @@ describe("collapseCompletedTurnWork", () => {
content: "Context compacted",
display: true,
excludeFromContext: true,
details: { runId: "run-1" },
idempotencyKey: "codex-context-compaction:thread:turn:item",
timestamp: 2_000,
},
assistantMessage("All done.", 3_000),
@@ -779,8 +782,10 @@ describe("collapseCompletedTurnWork", () => {
expect(work.groups[0]?.messages[0]?.message).toMatchObject({
role: "assistant",
content: [{ type: "text", text: "Context compacted" }],
runId: "run-1",
__openclaw: { runtimeActivityKind: "context_compaction" },
});
expect(work.groups[0]?.messages[0]?.message).not.toHaveProperty("idempotencyKey");
expect(requireGroup(items[2]).messages[0]?.message).toMatchObject({
content: "All done.",
});
@@ -1165,6 +1170,14 @@ describe("coalesceActivityRuns", () => {
expect(appended.key).toBe(initial.key);
});
it("keeps adjacent tool activity from different runs separate", () => {
const groups = projectedToolGroups();
const first = { ...groups[0]!, runId: "run-1" };
const second = { ...groups[1]!, runId: "run-2" };
expect(coalesceActivityRuns([first, second])).toEqual([first, second]);
});
it("treats every non-tool item as a hard presentation boundary", () => {
const groups = projectedToolGroups();
const userBoundary: MessageGroup = {
@@ -1562,6 +1575,12 @@ describe("buildCachedChatItems working spark", () => {
coalesceStreamRuns(pendingItems).find((item) => item.kind === "stream-run"),
"pending stream run",
);
const pendingFrame = expectDefined(
coalesceAgentRunFrames(coalesceStreamRuns(pendingItems)).find(
(item) => item.kind === "agent-run-frame",
),
"pending agent run frame",
);
const acknowledgedItems = buildCachedChatItems(
createProps({
@@ -1580,12 +1599,19 @@ describe("buildCachedChatItems working spark", () => {
coalesceStreamRuns(acknowledgedItems).find((item) => item.kind === "stream-run"),
"acknowledged stream run",
);
const acknowledgedFrame = expectDefined(
coalesceAgentRunFrames(coalesceStreamRuns(acknowledgedItems)).find(
(item) => item.kind === "agent-run-frame",
),
"acknowledged agent run frame",
);
expect(acknowledgedIndicator).toMatchObject({
key: pendingIndicator.key,
startedAt: pendingIndicator.startedAt,
});
expect(acknowledgedRun.key).toBe(pendingRun.key);
expect(acknowledgedFrame.key).toBe(pendingFrame.key);
const streamingItems = buildCachedChatItems(
createProps({
@@ -1641,6 +1667,35 @@ describe("buildCachedChatItems working spark", () => {
expect(otherSessionIndicator.key).not.toBe(pendingIndicator.key);
});
it("keeps a future queued send from replacing the active stream run identity", () => {
const items = buildCachedChatItems(
createProps({
sessionKey: "agent:main:active-with-future-queue",
runWorking: true,
stream: "Current run output.",
streamSegments: [{ text: "", ts: 1_000, runId: "active-run", boundaryMarker: true }],
queue: [
{
id: "future-send",
text: "Run this next.",
createdAt: 2_000,
sendRunId: "future-run",
sendState: "waiting-reconnect",
sendSubmittedAtMs: 1,
sendAttempts: 1,
},
],
}),
);
expect(items.find((item) => item.kind === "stream" && item.isStreaming)).toMatchObject({
runId: "active-run",
});
expect(items.find((item) => item.kind === "reading-indicator")).toMatchObject({
runId: "active-run",
});
});
it("keeps client and engine run identities separate", () => {
const sessionKey = "agent:main:elapsed-run-namespaces";
buildCachedChatItems(
+12 -2
View File
@@ -28,6 +28,7 @@ export {
coalesceStreamRuns,
collapseCompletedTurnWork,
} from "./chat-thread-grouping.ts";
export { agentRunFrameGroups, coalesceAgentRunFrames } from "./chat-agent-run-grouping.ts";
type CachedChatItems = {
input: BuildChatItemsProps | null;
@@ -78,6 +79,7 @@ function sameMessageGroup(previous: MessageGroup, next: MessageGroup): boolean {
senderIdentityKey(previous.sender) === senderIdentityKey(next.sender) &&
senderIdentityKey(previous.replyToSender) === senderIdentityKey(next.replyToSender) &&
previous.isStreaming === next.isStreaming &&
previous.runId === next.runId &&
previous.messages.length === next.messages.length &&
previous.messages.every((entry, index) => {
const candidate = next.messages[index];
@@ -127,10 +129,17 @@ function sameChatItem(previous: RenderChatItem, next: RenderChatItem): boolean {
previous.kind === "stream" &&
previous.text === next.text &&
previous.startedAt === next.startedAt &&
previous.isStreaming === next.isStreaming
previous.isStreaming === next.isStreaming &&
previous.runId === next.runId &&
previous.boundaryId === next.boundaryId
);
case "reading-indicator":
return previous.kind === "reading-indicator" && previous.startedAt === next.startedAt;
return (
previous.kind === "reading-indicator" &&
previous.startedAt === next.startedAt &&
previous.runId === next.runId &&
previous.boundaryId === next.boundaryId
);
case "question":
return (
previous.kind === "question" &&
@@ -182,6 +191,7 @@ function stabilizeChatItems(
!prior ||
claimedGroupKeys.has(prior.key) ||
prior.role !== item.role ||
prior.runId !== item.runId ||
prior.senderLabel !== item.senderLabel ||
senderIdentityKey(prior.sender) !== senderIdentityKey(item.sender)
) {
+302
View File
@@ -1621,6 +1621,137 @@ describe("chat transcript rendering", () => {
);
});
it("announces a run preamble and its later terminal answer separately", () => {
const transcript = createTestTranscript();
const container = document.createElement("div");
const user = {
kind: "group",
key: "group:user:announcement",
role: "user",
messages: [
{
key: "message:user:announcement",
message: {
role: "user",
content: "Start",
__openclaw: { id: "user:announcement", idempotencyKey: "run-announcement:user" },
},
},
],
timestamp: 1,
isStreaming: false,
};
const renderItems = (items: ReturnType<typeof chatThread.buildCachedChatItems>) => {
vi.mocked(chatThread.buildCachedChatItems).mockReturnValue(items);
renderChatInto(container, { transcript, messages: items });
};
renderItems([user] as ReturnType<typeof chatThread.buildCachedChatItems>);
const stream = {
kind: "stream" as const,
key: "stream:announcement",
text: "Latest streamed narration",
startedAt: 2,
isStreaming: true,
runId: "run-announcement",
boundaryId: "send:run-announcement",
};
renderItems([
user,
stream,
{
kind: "reading-indicator",
key: "reading:announcement",
startedAt: 2,
runId: "run-announcement",
boundaryId: "send:run-announcement",
},
] as ReturnType<typeof chatThread.buildCachedChatItems>);
expect(container.querySelector(".chat-transcript-announcement")?.textContent).toBe(
"Latest streamed narration",
);
renderItems([
user,
{
kind: "group",
key: "group:assistant:persisted-announcement",
role: "assistant",
messages: [
{
key: "assistant:persisted-announcement",
message: {
role: "assistant",
content: "Persisted narration while the run continues",
runId: "run-announcement",
},
},
],
timestamp: 3,
isStreaming: false,
runId: "run-announcement",
},
{
kind: "reading-indicator",
key: "reading:persisted-announcement",
startedAt: 4,
runId: "run-announcement",
boundaryId: "send:run-announcement",
},
] as ReturnType<typeof chatThread.buildCachedChatItems>);
expect(container.querySelector(".chat-transcript-announcement")?.textContent).toBe(
"Persisted narration while the run continues",
);
renderItems([
user,
{ ...stream, isStreaming: false },
{
kind: "group",
key: "group:tool:announcement",
role: "tool",
messages: [
{
key: "tool:announcement",
message: {
role: "toolResult",
content: "Tool output",
runId: "run-announcement",
},
},
],
timestamp: 3,
isStreaming: false,
runId: "run-announcement",
},
{
kind: "group",
key: "group:assistant:announcement",
role: "assistant",
messages: [
{
key: "assistant:announcement",
message: {
role: "assistant",
phase: "final_answer",
content: "Terminal answer",
runId: "run-announcement",
},
},
],
timestamp: 4,
isStreaming: false,
runId: "run-announcement",
},
] as ReturnType<typeof chatThread.buildCachedChatItems>);
expect(container.querySelector(".chat-transcript-announcement")?.textContent).toBe(
"Terminal answer",
);
});
it("does not announce appended rows from an inactive split pane", () => {
const transcript = createTestTranscript();
const container = document.createElement("div");
@@ -2605,6 +2736,160 @@ describe("chat loading skeleton", () => {
).toBe(7_200);
});
it("keeps multi-part run usage current when only output tokens change", () => {
const runId = "run-composed";
const user = {
kind: "group",
key: "group:user:run-composed",
role: "user",
messages: [
{
key: "message:user:run-composed",
message: {
role: "user",
content: "Start the work.",
timestamp: 0,
__openclaw: { id: "user:run-composed", idempotencyKey: `${runId}:user` },
},
},
],
timestamp: 0,
isStreaming: false,
};
const assistant = {
kind: "group",
key: "group:assistant:run-start",
role: "assistant",
messages: [
{
key: "message:assistant:run-start",
message: { role: "assistant", content: "Starting the work.", timestamp: 1 },
},
],
timestamp: 1,
isStreaming: false,
runId,
};
const tool = {
kind: "group",
key: "group:tool:run-work",
role: "tool",
messages: [
{
key: "message:tool:run-work",
message: { role: "toolResult", content: "Tool complete.", timestamp: 2 },
},
],
timestamp: 2,
isStreaming: false,
runId,
};
const reading = {
kind: "reading-indicator",
key: "reading:run-composed",
startedAt: 1,
runId,
};
vi.mocked(chatThread.buildCachedChatItems).mockReturnValue([
user,
assistant,
tool,
reading,
] as ReturnType<typeof chatThread.buildCachedChatItems>);
const container = document.createElement("div");
const streamPartsSpy = vi.spyOn(chatMessage, "renderStreamGroupParts");
renderChatInto(container, { canAbort: true, runId, runOutputTokens: 5_500, stream: null });
streamPartsSpy.mockClear();
renderChatInto(container, { canAbort: true, runId, runOutputTokens: 7_200, stream: null });
expect(streamPartsSpy.mock.calls.at(-1)?.[1].runOutputTokens).toBe(7_200);
});
it("keeps the completed recap on one composed multi-part run", () => {
const runId = "run-composed";
vi.mocked(chatThread.buildCachedChatItems).mockReturnValue([
{
kind: "group",
key: "group:user:run-composed",
role: "user",
messages: [
{
key: "message:user:run-composed",
message: {
role: "user",
content: "Start the work.",
timestamp: 0,
__openclaw: { id: "user:run-composed", idempotencyKey: `${runId}:user` },
},
},
],
timestamp: 0,
isStreaming: false,
},
{
kind: "group",
key: "group:assistant:run-start",
role: "assistant",
messages: [
{
key: "message:assistant:run-start",
message: { role: "assistant", content: "Starting the work.", timestamp: 1 },
},
],
timestamp: 1,
isStreaming: false,
runId,
},
{
kind: "group",
key: "group:tool:run-work",
role: "tool",
messages: [
{
key: "message:tool:run-work",
message: { role: "toolResult", content: "Tool complete.", timestamp: 2 },
},
],
timestamp: 2,
isStreaming: false,
runId,
},
{
kind: "group",
key: "group:assistant:run-finish",
role: "assistant",
messages: [
{
key: "message:assistant:run-finish",
message: { role: "assistant", content: "Finished the work.", timestamp: 3 },
},
],
timestamp: 3,
isStreaming: false,
runId,
},
] as ReturnType<typeof chatThread.buildCachedChatItems>);
vi.spyOn(chatProgress, "resolveTurnRecap").mockReturnValue({
runtimeMs: 5_000,
outputTokens: 42,
});
const container = renderChatView();
const frameCall = renderMessageGroupMock.mock.calls.find(([group]) =>
group.key.startsWith("agent-run:"),
);
expect(frameCall).toBeDefined();
expect(frameCall?.[0].messages).toHaveLength(1);
expect(frameCall?.[1].frameContent).toBeDefined();
expect(frameCall?.[1].turnRecap).toEqual({
runtimeMs: 5_000,
outputTokens: 42,
});
expect(container.querySelector(".chat-turn-recap")).toBeNull();
});
it("releases the embedded recap when a later reply becomes the settled turn", () => {
const firstReply = {
kind: "group",
@@ -7173,6 +7458,23 @@ describe("right-click Reply", () => {
expect(onCopy).toHaveBeenCalledOnce();
});
it("offers Reply only for the bubble that owns the frame actions", () => {
const onSetReply = vi.fn();
const { bubble, group } = renderChatBubble(
{ onSetReply },
{ messageId: "commentary", text: "Intermediate commentary" },
);
const actionOwner = document.createElement("div");
actionOwner.dataset.messageActionsFor = "terminal";
group.append(actionOwner);
group.dataset.chatRowKey = 'agent-run:["run-1","send:send-1"]';
const event = dispatchContextMenu(bubble);
expect(event.defaultPrevented).toBe(false);
expect(document.querySelector(".chat-reply-context-menu")).toBeNull();
});
it("dismisses an inline confirmation before opening the reply context menu", () => {
const container = renderChatView({ onSetReply: vi.fn() });
document.body.appendChild(container);
@@ -0,0 +1,90 @@
import { html, nothing } from "lit";
import type { QuestionPrompt } from "../../../app/question-prompt.ts";
import type { MessageGroup } from "../../../lib/chat/chat-types.ts";
import {
agentRunFrameActiveStatusParts,
agentRunFrameGroups,
type AgentRunFrameRenderItem,
} from "../chat-agent-run-grouping.ts";
import type { TurnRecap } from "../chat-progress.ts";
import {
renderActivityGroup,
renderMessageGroup,
renderMessageGroupContent,
renderStreamGroup,
renderStreamGroupParts,
renderWorkGroupSummary,
type StreamGroupOptions,
} from "./chat-message.ts";
type MessageGroupRenderOptions = Parameters<typeof renderMessageGroup>[1];
type AgentRunFrameOptions = {
questionPrompts: ReadonlyMap<string, QuestionPrompt>;
streamOptions: StreamGroupOptions;
renderGroupOptions: (group: MessageGroup) => MessageGroupRenderOptions;
isWorkExpanded: (key: string) => boolean;
onToggleWork: (key: string, expanded: boolean) => void;
turnRecap?: TurnRecap;
};
export function renderAgentRunFrame(frame: AgentRunFrameRenderItem, opts: AgentRunFrameOptions) {
const statusParts = agentRunFrameActiveStatusParts(frame);
if (statusParts) {
return renderStreamGroup(statusParts, {
...opts.streamOptions,
questionPrompts: opts.questionPrompts,
});
}
const groups = agentRunFrameGroups(frame);
const firstAssistant = groups.find((group) => group.role === "assistant");
const actionOwner = frame.outcome.kind === "completed" ? frame.outcome.actionOwner : null;
const representative = firstAssistant ?? groups[0];
const streamStarts = frame.parts.flatMap((part) =>
part.kind === "stream-run" ? part.parts.map((streamPart) => streamPart.startedAt) : [],
);
const shell: MessageGroup = {
key: frame.key,
kind: "group",
role: "assistant",
senderLabel: firstAssistant?.senderLabel,
replyToSender: firstAssistant?.replyToSender,
messages: representative?.messages ?? [],
timestamp: Math.min(...groups.map((group) => group.timestamp), ...streamStarts, Date.now()),
isStreaming: frame.outcome.kind === "active",
runId: frame.runId,
};
const renderFrameGroup = (group: MessageGroup) =>
renderMessageGroupContent(group, opts.renderGroupOptions(group));
const frameContent = frame.parts.map((part) => {
if (part.kind === "stream-run") {
// The frame owns layout continuity; the indicator stays standalone so
// its visible claw remains present alongside streamed text.
return renderStreamGroupParts(part.parts, opts.streamOptions, "standalone");
}
if (part.kind === "work-group") {
const expanded = opts.isWorkExpanded(part.key);
return html`
${renderWorkGroupSummary(part, {
expanded,
onToggle: () => opts.onToggleWork(part.key, expanded),
presentation: "continuation",
})}
${expanded ? part.groups.map(renderFrameGroup) : nothing}
`;
}
if (part.kind === "activity-run") {
const firstGroup = part.groups[0];
return firstGroup
? renderActivityGroup(part.groups, opts.renderGroupOptions(firstGroup), "continuation")
: nothing;
}
return renderFrameGroup(part);
});
return renderMessageGroup(shell, {
...opts.renderGroupOptions(shell),
frameContent,
frameActionOwner: actionOwner,
turnRecap: opts.turnRecap,
});
}
@@ -108,6 +108,8 @@ type RenderMessageGroupOptions = {
rewindDisabled?: boolean;
activeContinuation?: ActiveContinuation;
turnRecap?: TurnRecap;
frameContent?: unknown;
frameActionOwner?: MessageGroup["messages"][number] | null;
};
type GroupedMessageRenderOptions = Parameters<typeof renderGroupedMessage>[2];
@@ -233,6 +235,7 @@ function shouldAnimateUserTurnEntry(messageKey: string, message: unknown): boole
export function renderActivityGroup(
groups: readonly MessageGroup[],
opts: RenderMessageGroupOptions,
presentation: "standalone" | "continuation" = "standalone",
) {
const firstGroup = groups[0];
if (!firstGroup || opts.showToolCalls === false) {
@@ -268,65 +271,68 @@ export function renderActivityGroup(
reviewer,
})
: "";
return html`
<div
class="chat-group tool chat-group--activity chat-group--with-footer"
data-chat-row-key=${firstGroup.key}
>
<div class="chat-group-messages">
<div class="chat-activity-group ${activityExpanded ? "is-open" : ""}">
<button
class="chat-inline-disclosure chat-activity-group__summary"
type="button"
aria-expanded=${String(activityExpanded)}
aria-controls=${activityBodyId}
@pointerenter=${syncToolDisclosureOverflow}
@focus=${syncToolDisclosureOverflow}
@click=${(event: MouseEvent) => {
if (shouldToggleSelectableDisclosure(event)) {
opts.onToggleToolMessageExpanded?.(activityDisclosureId, activityExpanded);
}
}}
const content = html`
<div class="chat-activity-group ${activityExpanded ? "is-open" : ""}">
<button
class="chat-inline-disclosure chat-activity-group__summary"
type="button"
aria-expanded=${String(activityExpanded)}
aria-controls=${activityBodyId}
@pointerenter=${syncToolDisclosureOverflow}
@focus=${syncToolDisclosureOverflow}
@click=${(event: MouseEvent) => {
if (shouldToggleSelectableDisclosure(event)) {
opts.onToggleToolMessageExpanded?.(activityDisclosureId, activityExpanded);
}
}}
>
<span class="chat-activity-group__icon">${icons.listTree}</span>
<span class="chat-tool-disclosure__content">
<span class="chat-activity-group__label" title=${groupSummaryLabel}
>${groupSummaryLabel}</span
>
<span class="chat-activity-group__icon">${icons.listTree}</span>
<span class="chat-tool-disclosure__content">
<span class="chat-activity-group__label" title=${groupSummaryLabel}
>${groupSummaryLabel}</span
>
</span>
${reviewOutcome
? html`<span
class="chat-activity-group__review-status"
data-outcome=${reviewOutcome}
role="img"
aria-label=${reviewAriaLabel}
>${reviewOutcome === "denied"
? icons.shieldX
: reviewOutcome === "reviewing"
? icons.shieldQuestion
: icons.shieldCheck}</span
>`
: nothing}
<span class="chat-tool-row__chevron" aria-hidden="true">${icons.chevronRight}</span>
</button>
<div class="chat-activity-group__body" id=${activityBodyId} ?hidden=${!activityExpanded}>
${activityExpanded
? groups.map((group) =>
group.messages.map((item, index) =>
renderGroupedMessage(
item.message,
item.key,
buildGroupedMessageRenderOptions(group, item, index, opts),
opts.onOpenSidebar,
),
),
)
: nothing}
</div>
</div>
</span>
${reviewOutcome
? html`<span
class="chat-activity-group__review-status"
data-outcome=${reviewOutcome}
role="img"
aria-label=${reviewAriaLabel}
>${reviewOutcome === "denied"
? icons.shieldX
: reviewOutcome === "reviewing"
? icons.shieldQuestion
: icons.shieldCheck}</span
>`
: nothing}
<span class="chat-tool-row__chevron" aria-hidden="true">${icons.chevronRight}</span>
</button>
<div class="chat-activity-group__body" id=${activityBodyId} ?hidden=${!activityExpanded}>
${activityExpanded
? groups.map((group) =>
group.messages.map((item, index) =>
renderGroupedMessage(
item.message,
item.key,
buildGroupedMessageRenderOptions(group, item, index, opts),
opts.onOpenSidebar,
),
),
)
: nothing}
</div>
</div>
`;
return presentation === "continuation"
? content
: html`
<div
class="chat-group tool chat-group--activity chat-group--with-footer"
data-chat-row-key=${firstGroup.key}
>
<div class="chat-group-messages">${content}</div>
</div>
`;
}
export function resolveMessageGroupSenderLabel(
@@ -357,6 +363,50 @@ export function resolveMessageGroupSenderLabel(
: normalizedRole;
}
export function renderMessageGroupContent(group: MessageGroup, opts: RenderMessageGroupOptions) {
if (normalizeRoleForGrouping(group.role) === "tool") {
const cards = group.messages.flatMap((item) => extractToolCardsCached(item.message, item.key));
if (
group.messages.length > 1 ||
cards.length > 1 ||
cards.some((card) => readToolApprovalReviews(card.details).length > 0)
) {
return renderActivityGroup([group], opts, "continuation");
}
}
const who = resolveMessageGroupSenderLabel(group, opts);
return group.messages.map((item, index) => {
const actionDetails = resolveMessageActionDetails({
message: item.message,
messageId: item.key,
canFetchFullMessage: Boolean(opts.loadFullAssistantMessage && opts.sessionKey),
getAssistantMessageExpansion: opts.getAssistantMessageExpansion,
onReply: opts.onReply,
senderLabel: who,
});
if (
actionDetails?.shouldFetchFullMessage &&
actionDetails.messageId &&
opts.loadFullAssistantMessage &&
opts.onToggleAssistantMessageExpanded
) {
const expansion = opts.getAssistantMessageExpansion?.(actionDetails.messageId);
if (
!expansion ||
(expansion.status === "error" && expansion.revision < FULL_MESSAGE_RETRY_REVISION_LIMIT)
) {
opts.onToggleAssistantMessageExpanded(actionDetails.messageId);
}
}
return renderGroupedMessage(
item.message,
item.key,
buildGroupedMessageRenderOptions(group, item, index, opts, actionDetails),
opts.onOpenSidebar,
);
});
}
export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroupOptions) {
const normalizedRole = normalizeRoleForGrouping(group.role);
const isWorkspaceConflict = group.messages.every((item) =>
@@ -399,7 +449,13 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup
return renderActivityGroup([group], opts);
}
const messageActionDetails = group.messages.map((item) =>
const ownsRunFrame = opts.frameContent !== undefined;
const actionOwners = ownsRunFrame
? opts.frameActionOwner
? [opts.frameActionOwner]
: []
: group.messages;
const messageActionDetails = actionOwners.map((item) =>
resolveMessageActionDetails({
message: item.message,
messageId: item.key,
@@ -425,7 +481,15 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup
}
}
const lastMessageIndex = group.messages.length - 1;
const footerActionDetails = messageActionDetails[lastMessageIndex] ?? null;
const runFrameActive = ownsRunFrame && Boolean(group.isStreaming || opts.activeContinuation);
const footerActionDetails = runFrameActive
? null
: ownsRunFrame
? (messageActionDetails[0] ?? null)
: (messageActionDetails[lastMessageIndex] ?? null);
const footerActionMessageKey = ownsRunFrame
? opts.frameActionOwner?.key
: group.messages[lastMessageIndex]?.key;
const hasUserFooterActions =
normalizedRole === "user" &&
Boolean((footerActionDetails?.replyTarget && opts.onReply) || opts.onRewind);
@@ -433,7 +497,7 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup
? html`
<div
class="chat-group-footer-actions"
data-message-actions-for=${group.messages[lastMessageIndex]?.key ?? nothing}
data-message-actions-for=${footerActionMessageKey ?? nothing}
>
${footerActionDetails?.replyTarget && opts.onReply
? renderReplyButton(footerActionDetails.replyTarget, opts.onReply)
@@ -492,7 +556,8 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup
</div>
`
: nothing}
${group.messages.map((item, index) => {
${opts.frameContent ??
group.messages.map((item, index) => {
const actionDetails = messageActionDetails[index];
return html`
${renderGroupedMessage(
@@ -501,7 +566,7 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup
buildGroupedMessageRenderOptions(group, item, index, opts, actionDetails),
opts.onOpenSidebar,
)}
${actionDetails && index < lastMessageIndex
${actionDetails && index < lastMessageIndex && !ownsRunFrame
? html`
<div class="chat-message-actions-row" data-message-actions-for=${item.key}>
${renderMessageActionButtons(actionDetails, opts)}
@@ -541,7 +606,7 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup
? html`
<div
class="chat-group-footer-actions"
data-message-actions-for=${group.messages[lastMessageIndex]?.key ?? nothing}
data-message-actions-for=${footerActionMessageKey ?? nothing}
>
${renderMessageActionButtons(footerActionDetails, opts)}
</div>
@@ -158,34 +158,37 @@ export function renderStreamGroup(parts: StreamGroupPart[], opts: StreamGroupOpt
*/
export function renderWorkGroupSummary(
item: { key: string; durationMs: number | null },
opts: { expanded: boolean; onToggle: () => void },
opts: { expanded: boolean; onToggle: () => void; presentation?: "standalone" | "continuation" },
) {
const duration = formatDurationCompact(item.durationMs);
const label = duration ? t("chat.workRun.workedFor", { duration }) : t("chat.workRun.worked");
return html`
<div class="chat-group tool chat-group--work" data-chat-row-key=${item.key}>
<div class="chat-group-messages">
<div class="chat-activity-group chat-work-group ${opts.expanded ? "is-open" : ""}">
<button
class="chat-inline-disclosure chat-activity-group__summary"
type="button"
aria-expanded=${String(opts.expanded)}
@pointerenter=${syncToolDisclosureOverflow}
@focus=${syncToolDisclosureOverflow}
@click=${(event: MouseEvent) => {
if (shouldToggleSelectableDisclosure(event)) {
opts.onToggle();
}
}}
>
<span class="chat-tool-disclosure__content">
<span class="chat-activity-group__label" title=${label}>${label}</span>
</span>
<span class="chat-tool-row__chevron" aria-hidden="true">${icons.chevronRight}</span>
</button>
<div class="chat-work-group__separator" aria-hidden="true"></div>
</div>
</div>
const content = html`
<div class="chat-activity-group chat-work-group ${opts.expanded ? "is-open" : ""}">
<button
class="chat-inline-disclosure chat-activity-group__summary"
type="button"
aria-expanded=${String(opts.expanded)}
@pointerenter=${syncToolDisclosureOverflow}
@focus=${syncToolDisclosureOverflow}
@click=${(event: MouseEvent) => {
if (shouldToggleSelectableDisclosure(event)) {
opts.onToggle();
}
}}
>
<span class="chat-tool-disclosure__content">
<span class="chat-activity-group__label" title=${label}>${label}</span>
</span>
<span class="chat-tool-row__chevron" aria-hidden="true">${icons.chevronRight}</span>
</button>
<div class="chat-work-group__separator" aria-hidden="true"></div>
</div>
`;
return opts.presentation === "continuation"
? content
: html`
<div class="chat-group tool chat-group--work" data-chat-row-key=${item.key}>
<div class="chat-group-messages">${content}</div>
</div>
`;
}
+10 -2
View File
@@ -6,7 +6,15 @@ export {
dismissConfirmedActionPopovers,
openChatRewindConfirmation,
} from "./chat-message-confirmation.ts";
export { renderActivityGroup, renderMessageGroup } from "./chat-message-group.ts";
export {
renderActivityGroup,
renderMessageGroup,
renderMessageGroupContent,
} from "./chat-message-group.ts";
export type { MessageReplyTarget } from "./chat-message-markdown.ts";
export { renderStreamGroup, renderWorkGroupSummary } from "./chat-message-stream.ts";
export {
renderStreamGroup,
renderStreamGroupParts,
renderWorkGroupSummary,
} from "./chat-message-stream.ts";
export type { StreamGroupOptions, StreamGroupPart } from "./chat-message-stream.ts";
@@ -460,7 +460,8 @@ export function handleTranscriptContextMenu(event: MouseEvent, props: Transcript
(element) => element.dataset.messageActionsFor === messageId,
);
const copyButton = actionOwner?.querySelector<HTMLButtonElement>(".chat-copy-btn");
const canReply = Boolean(text && props.onSetReply);
const ownsRunFrame = group.dataset.chatRowKey?.startsWith("agent-run:") === true;
const canReply = Boolean(text && props.onSetReply && (!ownsRunFrame || actionOwner));
const canRewind = isUserMessage && typeof props.onRewindMessage === "function";
const canCopy = Boolean(copyButton);
const canFork = isUserMessage && typeof props.onForkMessage === "function";
@@ -0,0 +1,88 @@
import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import type { MessageGroup } from "../../../lib/chat/chat-types.ts";
import { extractTextCached } from "../../../lib/chat/message-extract.ts";
import type { coalesceAgentRunFrames } from "../chat-agent-run-grouping.ts";
import type { TranscriptAnnouncement } from "./chat-transcript-controller.ts";
type ChatRenderItem = ReturnType<typeof coalesceAgentRunFrames>[number];
const ANNOUNCEMENT_MAX_CHARS = 500;
function assistantGroupAnnouncementSource(
group: MessageGroup,
): { key: string; text: string } | null {
if (group.role.toLowerCase() !== "assistant") {
return null;
}
for (let index = group.messages.length - 1; index >= 0; index -= 1) {
const source = group.messages[index];
const text = extractTextCached(source?.message)?.trim();
if (text) {
return { key: source?.key ?? group.key, text };
}
}
return null;
}
export function latestTranscriptAnnouncement(
items: readonly ChatRenderItem[],
): TranscriptAnnouncement | null {
const announcement = (key: string, text: string): TranscriptAnnouncement => ({
key,
text: truncateUtf16Safe(text, ANNOUNCEMENT_MAX_CHARS),
});
for (let itemIndex = items.length - 1; itemIndex >= 0; itemIndex -= 1) {
const item = items[itemIndex];
if (!item) {
continue;
}
if (item.kind === "agent-run-frame") {
if (item.outcome.kind === "completed") {
const owner = item.outcome.actionOwner;
const text = owner ? extractTextCached(owner.message)?.trim() : null;
if (owner && text) {
return announcement(owner.key, text);
}
continue;
}
if (item.outcome.kind === "failed") {
continue;
}
for (let partIndex = item.parts.length - 1; partIndex >= 0; partIndex -= 1) {
const part = item.parts[partIndex];
if (!part) {
continue;
}
if (part.kind === "stream-run") {
const text = part.parts.findLast(
(streamPart) => streamPart.kind === "stream" && streamPart.text.trim(),
);
if (text?.kind === "stream") {
return announcement(text.key, text.text.trim());
}
continue;
}
const groups = part.kind === "group" ? [part] : part.groups.toReversed();
for (const group of groups) {
const source = assistantGroupAnnouncementSource(group);
if (source) {
return announcement(source.key, source.text);
}
}
}
continue;
}
const groups =
item.kind === "group"
? [item]
: item.kind === "work-group" || item.kind === "activity-run"
? item.groups.toReversed()
: [];
for (const group of groups) {
const source = assistantGroupAnnouncementSource(group);
if (source) {
return announcement(source.key, source.text);
}
}
}
return null;
}
@@ -1,7 +1,5 @@
// Chat-item projection, expansion, reply hydration, and guarded row rendering.
import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import { nothing, type TemplateResult } from "lit";
import { guard } from "lit/directives/guard.js";
import { classifySessionKind } from "../../../../../src/sessions/classify-session-kind.js";
import { i18n } from "../../../i18n/index.ts";
import type { MessageGroup } from "../../../lib/chat/chat-types.ts";
@@ -13,11 +11,14 @@ import {
parseAgentSessionKey,
resolveUiGlobalAliasAgentId,
} from "../../../lib/sessions/session-key.ts";
import { agentRunFrameActiveStatusParts } from "../chat-agent-run-grouping.ts";
import { resolveTurnRecap, type TurnRecap } from "../chat-progress.ts";
import {
assistantGroupCanOwnActiveRunStatus,
agentRunFrameGroups,
assistantMessageExpansionSignature,
buildCachedChatItems,
coalesceAgentRunFrames,
coalesceActivityRuns,
coalesceStreamRuns,
collapseCompletedTurnWork,
@@ -30,6 +31,7 @@ import {
syncToolCardExpansionState,
} from "../chat-thread.ts";
import { getToolTitlesVersion } from "../tool-titles.ts";
import { renderAgentRunFrame } from "./chat-agent-run-frame.ts";
import { renderBackgroundTasksStatusRow } from "./chat-background-tasks-status.ts";
import { renderChatDivider, renderChatNotice } from "./chat-divider.ts";
import { resolveMessageGroupSenderLabel } from "./chat-message-group.ts";
@@ -49,13 +51,13 @@ import {
closeTranscriptSearch,
getTranscriptState,
type ChatThreadProps,
type ChatThreadState,
} from "./chat-thread-interactions.ts";
import type {
ChatTranscriptSession,
TranscriptAnnouncement,
TranscriptRow,
} from "./chat-transcript-controller.ts";
import { latestTranscriptAnnouncement } from "./chat-transcript-announcement.ts";
import type { ChatTranscriptSession, TranscriptRow } from "./chat-transcript-controller.ts";
import {
guardChatRenderItems,
trackTranscriptRenderDependencies,
} from "./chat-transcript-render-guard.ts";
import { renderChatTypingIndicator } from "./chat-typing-indicator.ts";
import { resolveAssistantDisplayAvatar } from "./chat-welcome.ts";
import { renderTurnRecapRow } from "./chat-working-indicator.ts";
@@ -68,8 +70,7 @@ type ChatTranscriptProjection = {
renderRows: (overlay?: unknown) => TemplateResult;
};
type ChatRenderItem = ReturnType<typeof coalesceActivityRuns>[number];
const CHAT_TRANSCRIPT_ANNOUNCEMENT_MAX_CHARS = 500;
type ChatRenderItem = ReturnType<typeof coalesceAgentRunFrames>[number];
type LoadedReplySource = {
rowKey: string;
@@ -105,76 +106,6 @@ function projectResolvedReplyPreview(
};
}
function latestTranscriptAnnouncement(
items: readonly ChatRenderItem[],
): TranscriptAnnouncement | null {
for (let itemIndex = items.length - 1; itemIndex >= 0; itemIndex -= 1) {
const item = items[itemIndex];
if (!item || item.kind !== "group" || item.role.toLowerCase() !== "assistant") {
continue;
}
for (let messageIndex = item.messages.length - 1; messageIndex >= 0; messageIndex -= 1) {
const message = item.messages[messageIndex]?.message;
const text = extractTextCached(message)?.trim();
if (text) {
return {
key: item.key,
text: truncateUtf16Safe(text, CHAT_TRANSCRIPT_ANNOUNCEMENT_MAX_CHARS),
};
}
}
}
return null;
}
function chatRenderItemGuardDependencies(item: ChatRenderItem): readonly unknown[] {
if (item.kind === "stream-run") {
return [item.key, ...item.parts];
}
if (item.kind === "work-group") {
return [item.key, item.durationMs, ...item.groups];
}
if (item.kind === "activity-run") {
return [item.key, ...item.groups];
}
return [item];
}
function trackTranscriptRenderDependencies(
state: ChatThreadState,
dependencies: unknown[],
): unknown[] {
const previous = state.transcriptRenderDependencies;
const nextLength = dependencies.length - 1;
let changed = previous.length !== nextLength;
for (let index = 0; !changed && index < nextLength; index += 1) {
changed = !Object.is(previous[index], dependencies[index + 1]);
}
if (changed) {
// The first dependency is chatItems. Keep the shared context stable when
// only the live row changes, but invalidate every row for presentation changes.
state.transcriptRenderDependencies = dependencies.slice(1);
state.transcriptRenderContext = {};
}
return dependencies;
}
function guardChatRenderItems(
state: ChatThreadState,
// Live run status is not derivable from a row's own item identity: ownership
// is decided by sibling rows, and the usage counter ticks on run patches that
// touch nothing else. Rows showing status must re-render on both, or the
// memoized copy stacks a second claw row or freezes the token count.
liveStatus: (item: ChatRenderItem) => string,
render: (item: ChatRenderItem) => unknown,
) {
return (item: ChatRenderItem) =>
guard(
[...chatRenderItemGuardDependencies(item), state.transcriptRenderContext, liveStatus(item)],
() => render(item),
);
}
export function projectChatTranscript(
props: ChatThreadProps,
transcript: ChatTranscriptSession,
@@ -427,6 +358,17 @@ export function projectChatTranscript(
// memoizing across usage patches.
const workingUsageKey = `usage:${props.runOutputTokens ?? ""}`;
const liveStatusSignature = (item: ChatRenderItem): string => {
if (item.kind === "agent-run-frame") {
const hasWorkingIndicator = item.parts.some(
(part) =>
part.kind === "stream-run" &&
part.parts.some((streamPart) => streamPart.kind === "reading-indicator"),
);
const recap = turnRecapByGroupKey.get(item.key);
return `${hasWorkingIndicator ? workingUsageKey : ""}|${
recap ? `${recap.runtimeMs}:${recap.outputTokens ?? ""}` : ""
}`;
}
if (item.kind === "stream-run") {
return item.parts.some((part) => part.kind === "reading-indicator") ? workingUsageKey : "";
}
@@ -480,6 +422,25 @@ export function projectChatTranscript(
}
return renderActivityGroup(item.groups, renderGroupOptions(firstGroup));
}
if (item.kind === "agent-run-frame") {
return renderAgentRunFrame(item, {
questionPrompts,
streamOptions: {
...streamGroupOptions,
questionPrompts,
startupPhase: props.startupStatus?.phase,
waitingApproval: props.waitingApproval,
runOutputTokens: props.runOutputTokens,
},
renderGroupOptions,
isWorkExpanded: (key) => expandedToolCards.get(key) ?? false,
onToggleWork: (key, expanded) => {
setExpansionState(expandedToolCards, key, !expanded);
requestUpdate();
},
turnRecap: turnRecapByGroupKey.get(item.key),
});
}
if (item.kind === "group") {
return renderGroupItem(item);
}
@@ -490,7 +451,7 @@ export function projectChatTranscript(
}
return nothing;
});
const collapsedItems = coalesceActivityRuns(
const semanticItems = coalesceActivityRuns(
collapseCompletedTurnWork(coalesceStreamRuns(chatItems), {
sessionKey: props.sessionKey,
runWorking: Boolean(props.runWorking),
@@ -498,6 +459,7 @@ export function projectChatTranscript(
}),
{ searchActive: searchFiltering },
);
const collapsedItems = coalesceAgentRunFrames(semanticItems, { searchActive: searchFiltering });
// Watch/settle on actual indicator visibility (not runWorking): queued
// sends show the claw before the run starts, and the recap must never
// stack under a visible working row.
@@ -511,22 +473,29 @@ export function projectChatTranscript(
props.runOutputTokens ?? null,
);
const transcriptItems = collapsedItems.filter((item, index) => {
if (item.kind !== "stream-run") {
return true;
}
const previous = collapsedItems[index - 1];
const isActiveStatusRun = item.parts.every((part) => part.kind === "reading-indicator");
const activeStatusParts =
item.kind === "stream-run" && item.parts.every((part) => part.kind === "reading-indicator")
? item.parts
: item.kind === "agent-run-frame"
? agentRunFrameActiveStatusParts(item)
: undefined;
const activeStatusRunId =
item.kind === "stream-run" || item.kind === "agent-run-frame" ? item.runId : undefined;
if (
previous?.kind !== "group" ||
!isActiveStatusRun ||
!assistantGroupCanOwnActiveRunStatus(previous)
!activeStatusParts ||
!assistantGroupCanOwnActiveRunStatus(previous) ||
(previous.runId !== undefined &&
activeStatusRunId !== undefined &&
previous.runId !== activeStatusRunId)
) {
return true;
}
// A reply and its still-running state are one turn-level presentation.
// Keeping the status in the reply avoids a second claw/assistant row.
activeContinuationByGroupKey.set(previous.key, {
parts: item.parts,
parts: activeStatusParts,
options: {
...streamGroupOptions,
startupPhase: props.startupStatus?.phase,
@@ -537,28 +506,37 @@ export function projectChatTranscript(
return false;
});
for (const item of transcriptItems) {
if (item.kind !== "group") {
const groups =
item.kind === "agent-run-frame"
? agentRunFrameGroups(item)
: item.kind === "group"
? [item]
: [];
const firstGroup = groups.find((group) => group.role === "assistant") ?? groups[0];
if (!firstGroup) {
continue;
}
const senderLabel = resolveMessageGroupSenderLabel(item, {
const senderLabel = resolveMessageGroupSenderLabel(firstGroup, {
assistantName: props.assistantName,
userId: props.userId,
userName: props.userName,
userAvatar: props.userAvatar,
});
for (const source of item.messages) {
const sourceMessageId = persistedMessageEntryId(source.message);
const text = resolveMessageReplyText(source.message);
if (sourceMessageId && text) {
loadedReplySources.set(sourceMessageId, {
rowKey: item.key,
preview: {
messageId: source.key,
sourceMessageId,
senderLabel,
text,
},
});
for (const group of groups) {
for (const source of group.messages) {
const sourceMessageId = persistedMessageEntryId(source.message);
const text = resolveMessageReplyText(source.message);
if (sourceMessageId && text) {
loadedReplySources.set(sourceMessageId, {
rowKey: item.key,
preview: {
messageId: source.key,
sourceMessageId,
senderLabel,
text,
},
});
}
}
}
}
@@ -571,6 +549,13 @@ export function projectChatTranscript(
if (lastItem?.kind === "group" && assistantGroupCanOwnActiveRunStatus(lastItem)) {
turnRecapByGroupKey.set(lastItem.key, turnRecap);
turnRecapOwnerKey = lastItem.key;
} else if (
lastItem?.kind === "agent-run-frame" &&
lastItem.outcome.kind === "completed" &&
lastItem.outcome.actionOwner
) {
turnRecapByGroupKey.set(lastItem.key, turnRecap);
turnRecapOwnerKey = lastItem.key;
}
}
// New row keys measure expanded work immediately; existing keys keep their
@@ -0,0 +1,53 @@
import { guard } from "lit/directives/guard.js";
import type { coalesceAgentRunFrames } from "../chat-agent-run-grouping.ts";
import type { ChatThreadState } from "./chat-thread-interactions.ts";
type ChatRenderItem = ReturnType<typeof coalesceAgentRunFrames>[number];
function itemDependencies(item: ChatRenderItem): readonly unknown[] {
if (item.kind === "stream-run") {
return [item.key, ...item.parts];
}
if (item.kind === "work-group") {
return [item.key, item.durationMs, ...item.groups];
}
if (item.kind === "activity-run") {
return [item.key, ...item.groups];
}
if (item.kind === "agent-run-frame") {
return [item.key, item.outcome, ...item.parts];
}
return [item];
}
export function trackTranscriptRenderDependencies(
state: ChatThreadState,
dependencies: unknown[],
): unknown[] {
const previous = state.transcriptRenderDependencies;
const nextLength = dependencies.length - 1;
let changed = previous.length !== nextLength;
for (let index = 0; !changed && index < nextLength; index += 1) {
changed = !Object.is(previous[index], dependencies[index + 1]);
}
if (changed) {
// The first dependency is chatItems. Keep the shared context stable when
// only the live row changes, but invalidate every row for presentation changes.
state.transcriptRenderDependencies = dependencies.slice(1);
state.transcriptRenderContext = {};
}
return dependencies;
}
export function guardChatRenderItems(
state: ChatThreadState,
// Live status ownership depends on sibling rows, while usage patches can
// update a visible indicator without changing the row itself.
liveStatus: (item: ChatRenderItem) => string,
render: (item: ChatRenderItem) => unknown,
) {
return (item: ChatRenderItem) =>
guard([...itemDependencies(item), state.transcriptRenderContext, liveStatus(item)], () =>
render(item),
);
}
+15 -1
View File
@@ -2,7 +2,11 @@ import { readSessionMessageIdentity } from "@openclaw/gateway-client/browser";
import { asNullableRecord } from "@openclaw/normalization-core/record-coerce";
import { areUiSessionKeysEquivalent } from "../../lib/sessions/session-key.ts";
type LiveTerminalIdentity = { runId: string; afterBoundaryRunId?: string };
type LiveTerminalIdentity = {
runId: string;
afterBoundaryRunId?: string;
disposition?: "aborted" | "error" | "timeout";
};
const liveTerminalIdentities = new WeakMap<object, LiveTerminalIdentity>();
const authoritativeTerminals = new WeakMap<object, AuthoritativeTerminal>();
@@ -19,11 +23,13 @@ export function rememberLiveTerminalRun(
message: unknown,
runId: string | null | undefined,
afterBoundaryRunId?: string,
disposition?: LiveTerminalIdentity["disposition"],
): unknown {
if (runId && message && typeof message === "object") {
liveTerminalIdentities.set(message, {
runId,
...(afterBoundaryRunId ? { afterBoundaryRunId } : {}),
...(disposition ? { disposition } : {}),
});
}
return message;
@@ -47,6 +53,14 @@ export function readLiveTerminalAfterBoundaryRunId(message: unknown): string | n
: null;
}
export function readLiveTerminalDisposition(
message: unknown,
): LiveTerminalIdentity["disposition"] | null {
return message && typeof message === "object"
? (liveTerminalIdentities.get(message)?.disposition ?? null)
: null;
}
export function rememberAuthoritativeTerminal(options: {
event: {
clientRunId?: string | null;
+9
View File
@@ -90,6 +90,7 @@
}
.chat-group-messages {
--chat-message-type-boundary-gap: var(--space-1);
position: relative; /* anchors message-local popovers */
display: flex;
flex-direction: column;
@@ -101,6 +102,14 @@
min-width: 0;
}
.chat-group-messages
> :is(
.chat-bubble--tool-shell + .chat-bubble:not(.chat-bubble--tool-shell),
.chat-bubble:not(.chat-bubble--tool-shell) + .chat-bubble--tool-shell
) {
margin-block-start: var(--chat-message-type-boundary-gap);
}
/* User messages align content right */
.chat-group.user .chat-group-messages {
align-items: flex-end;
+1 -7
View File
@@ -1113,16 +1113,10 @@
}
/* ── Activity group (a turn's run of tool rows) ── */
/* Keep the activity summary, call bubbles, and flat result rows on one width.
Include the role classes so this wins over the generic tool-group rule. */
.chat-group.tool.chat-group--activity .chat-group-messages {
max-width: var(--chat-message-max-width, min(760px, 100%));
}
.chat-activity-group {
width: 100%;
min-width: 0;
max-width: 100%;
max-width: var(--chat-message-max-width, min(760px, 100%));
}
.chat-activity-group__summary {