diff --git a/extensions/codex/src/app-server/context-compaction-activity.test.ts b/extensions/codex/src/app-server/context-compaction-activity.test.ts index 19e7a03d5c02..12d3708f13e8 100644 --- a/extensions/codex/src/app-server/context-compaction-activity.test.ts +++ b/extensions/codex/src/app-server/context-compaction-activity.test.ts @@ -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(); diff --git a/extensions/codex/src/app-server/context-compaction-activity.ts b/extensions/codex/src/app-server/context-compaction-activity.ts index 7b55635809d6..a928adcecc21 100644 --- a/extensions/codex/src/app-server/context-compaction-activity.ts +++ b/extensions/codex/src/app-server/context-compaction-activity.ts @@ -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, }; diff --git a/extensions/codex/src/app-server/run-attempt-active-turn.ts b/extensions/codex/src/app-server/run-attempt-active-turn.ts index ebed8585a1f8..0029963dd593 100644 --- a/extensions/codex/src/app-server/run-attempt-active-turn.ts +++ b/extensions/codex/src/app-server/run-attempt-active-turn.ts @@ -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, }); } diff --git a/extensions/codex/src/app-server/settled-turn-finalizer.ts b/extensions/codex/src/app-server/settled-turn-finalizer.ts index c2ca70b20203..609f6309ce47 100644 --- a/extensions/codex/src/app-server/settled-turn-finalizer.ts +++ b/extensions/codex/src/app-server/settled-turn-finalizer.ts @@ -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, }); diff --git a/extensions/codex/src/app-server/transcript-mirror-attestation.ts b/extensions/codex/src/app-server/transcript-mirror-attestation.ts index 9f27e4528b79..d9c560d2bf77 100644 --- a/extensions/codex/src/app-server/transcript-mirror-attestation.ts +++ b/extensions/codex/src/app-server/transcript-mirror-attestation.ts @@ -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; @@ -29,6 +30,24 @@ export function attachCodexMirrorAttestation( return attested; } +export function attachCodexMirrorRunId( + 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)) { diff --git a/extensions/codex/src/app-server/transcript-mirror.test.ts b/extensions/codex/src/app-server/transcript-mirror.test.ts index b3db829d8652..bbabfe1e077a 100644 --- a/extensions/codex/src/app-server/transcript-mirror.test.ts +++ b/extensions/codex/src/app-server/transcript-mirror.test.ts @@ -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([]); }); diff --git a/extensions/codex/src/app-server/transcript-mirror.ts b/extensions/codex/src/app-server/transcript-mirror.ts index 02a94feba24c..1a8238d6217a 100644 --- a/extensions/codex/src/app-server/transcript-mirror.ts +++ b/extensions/codex/src/app-server/transcript-mirror.ts @@ -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, diff --git a/packages/gateway-client/src/session-projection-message-identity.ts b/packages/gateway-client/src/session-projection-message-identity.ts new file mode 100644 index 000000000000..7dd56c57b9f6 --- /dev/null +++ b/packages/gateway-client/src/session-projection-message-identity.ts @@ -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, + }; +} diff --git a/packages/gateway-client/src/session-projection.test.ts b/packages/gateway-client/src/session-projection.test.ts index c3f1f999cd6a..a3a420411b24 100644 --- a/packages/gateway-client/src/session-projection.test.ts +++ b/packages/gateway-client/src/session-projection.test.ts @@ -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", diff --git a/packages/gateway-client/src/session-projection.ts b/packages/gateway-client/src/session-projection.ts index 0847057c2959..36cb152c9b85 100644 --- a/packages/gateway-client/src/session-projection.ts +++ b/packages/gateway-client/src/session-projection.ts @@ -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; diff --git a/src/agents/tools/message-tool.internal-source-reply.integration.test.ts b/src/agents/tools/message-tool.internal-source-reply.integration.test.ts index 86c6c07eae6c..92fd2f2f5e83 100644 --- a/src/agents/tools/message-tool.internal-source-reply.integration.test.ts +++ b/src/agents/tools/message-tool.internal-source-reply.integration.test.ts @@ -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> = []; + const unsubscribe = onSessionTranscriptUpdate((update) => { + updates.push(update); + const content = + update.message && typeof update.message === "object" + ? (update.message as { content?: Array> }).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> } )?.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({ diff --git a/src/config/sessions/transcript.ts b/src/config/sessions/transcript.ts index e735c7fc8c9e..e081172b6f45 100644 --- a/src/config/sessions/transcript.ts +++ b/src/config/sessions/transcript.ts @@ -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 ? { diff --git a/src/gateway/internal-source-reply-persistence.ts b/src/gateway/internal-source-reply-persistence.ts index f03d4b0a1bd4..c758e8c951c0 100644 --- a/src/gateway/internal-source-reply-persistence.ts +++ b/src/gateway/internal-source-reply-persistence.ts @@ -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(); diff --git a/src/gateway/managed-image-attachments.ts b/src/gateway/managed-image-attachments.ts index 8368415f7ff3..61265ba741c8 100644 --- a/src/gateway/managed-image-attachments.ts +++ b/src/gateway/managed-image-attachments.ts @@ -787,6 +787,22 @@ export async function cleanupManagedOutgoingMediaRecords(params?: { return { deletedRecordCount, deletedFileCount, retainedCount }; } +export async function removeManagedOutgoingMediaBlocks(params: { + blocks: readonly Record[]; + messageId: string; + stateDir?: string; +}): Promise { + 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, diff --git a/ui/src/e2e/chat-active-turn-recovery.e2e.test.ts b/ui/src/e2e/chat-active-turn-recovery.e2e.test.ts index c618561ad4f4..f0076ce9b33b 100644 --- a/ui/src/e2e/chat-active-turn-recovery.e2e.test.ts +++ b/ui/src/e2e/chat-active-turn-recovery.e2e.test.ts @@ -158,7 +158,9 @@ async function installActiveRunSnapshot( } async function assertActiveTurnVisible(page: Page, streamText: string): Promise { - 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(".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(".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(".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, diff --git a/ui/src/e2e/chat-agent-run-transcript.e2e.test.ts b/ui/src/e2e/chat-agent-run-transcript.e2e.test.ts new file mode 100644 index 000000000000..58259432ed3d --- /dev/null +++ b/ui/src/e2e/chat-agent-run-transcript.e2e.test.ts @@ -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", + "I’ll 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. I’ll 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", + "I’ll 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: "I’ll 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("I’ll 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("I’ll 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(); + }); +}); diff --git a/ui/src/e2e/chat-flow.active-run-follow-ups.e2e.test.ts b/ui/src/e2e/chat-flow.active-run-follow-ups.e2e.test.ts index df7245426e68..b5809e81d3fb 100644 --- a/ui/src/e2e/chat-flow.active-run-follow-ups.e2e.test.ts +++ b/ui/src/e2e/chat-flow.active-run-follow-ups.e2e.test.ts @@ -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) => 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 }); diff --git a/ui/src/e2e/chat-flow.session-start.e2e.test.ts b/ui/src/e2e/chat-flow.session-start.e2e.test.ts index 654055ffce6b..9f1afc4bb185 100644 --- a/ui/src/e2e/chat-flow.session-start.e2e.test.ts +++ b/ui/src/e2e/chat-flow.session-start.e2e.test.ts @@ -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); diff --git a/ui/src/e2e/chat-flow.streaming.e2e.test.ts b/ui/src/e2e/chat-flow.streaming.e2e.test.ts index 9db2aa7e4887..3df461d10945 100644 --- a/ui/src/e2e/chat-flow.streaming.e2e.test.ts +++ b/ui/src/e2e/chat-flow.streaming.e2e.test.ts @@ -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); } diff --git a/ui/src/e2e/chat-run-lifecycle.e2e.test.ts b/ui/src/e2e/chat-run-lifecycle.e2e.test.ts index 73f50c3428cf..19071af0f11c 100644 --- a/ui/src/e2e/chat-run-lifecycle.e2e.test.ts +++ b/ui/src/e2e/chat-run-lifecycle.e2e.test.ts @@ -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"; diff --git a/ui/src/e2e/chat-tool-turn-outcome.e2e.test.ts b/ui/src/e2e/chat-tool-turn-outcome.e2e.test.ts index 494a107ce1cf..be4aa6ff152e 100644 --- a/ui/src/e2e/chat-tool-turn-outcome.e2e.test.ts +++ b/ui/src/e2e/chat-tool-turn-outcome.e2e.test.ts @@ -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( diff --git a/ui/src/e2e/session-placement.move.e2e.test.ts b/ui/src/e2e/session-placement.move.e2e.test.ts index 8edc9c7f46ae..a6c950636c5f 100644 --- a/ui/src/e2e/session-placement.move.e2e.test.ts +++ b/ui/src/e2e/session-placement.move.e2e.test.ts @@ -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); } diff --git a/ui/src/lib/chat/chat-types.ts b/ui/src/lib/chat/chat-types.ts index 621722839891..349727d1a6ed 100644 --- a/ui/src/lib/chat/chat-types.ts +++ b/ui/src/lib/chat/chat-types.ts @@ -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 */ diff --git a/ui/src/pages/chat/chat-agent-run-grouping.test.ts b/ui/src/pages/chat/chat-agent-run-grouping.test.ts new file mode 100644 index 000000000000..4bafbede5573 --- /dev/null +++ b/ui/src/pages/chat/chat-agent-run-grouping.test.ts @@ -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 = {}, +): 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[number], + { kind: "agent-run-frame" } +>; + +function requireFrame( + value: ReturnType[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); + }); +}); diff --git a/ui/src/pages/chat/chat-agent-run-grouping.ts b/ui/src/pages/chat/chat-agent-run-grouping.ts new file mode 100644 index 000000000000..4a8ce49618e2 --- /dev/null +++ b/ui/src/pages/chat/chat-agent-run-grouping.ts @@ -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 { + if (opts.searchActive) { + return items; + } + const result: Array = []; + 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; +} diff --git a/ui/src/pages/chat/chat-gateway.ts b/ui/src/pages/chat/chat-gateway.ts index 4bb8a5ff52ac..00e3bd13f806 100644 --- a/ui/src/pages/chat/chat-gateway.ts +++ b/ui/src/pages/chat/chat-gateway.ts @@ -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 diff --git a/ui/src/pages/chat/chat-progress.test.ts b/ui/src/pages/chat/chat-progress.test.ts index bd78e1f0f540..46eee157b1d5 100644 --- a/ui/src/pages/chat/chat-progress.test.ts +++ b/ui/src/pages/chat/chat-progress.test.ts @@ -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, diff --git a/ui/src/pages/chat/chat-progress.ts b/ui/src/pages/chat/chat-progress.ts index 868cafd27074..71a384b7acdb 100644 --- a/ui/src/pages/chat/chat-progress.ts +++ b/ui/src/pages/chat/chat-progress.ts @@ -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(); let anonymousWorkingProgressId = 0; +export function buildGuardianNoticeItem( + notice: ChatGuardianNotice, +): Extract { + 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, 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 | 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 | 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 { diff --git a/ui/src/pages/chat/chat-responsive.browser.test.ts b/ui/src/pages/chat/chat-responsive.browser.test.ts index 760446f9107f..4ca2b9a9c472 100644 --- a/ui/src/pages/chat/chat-responsive.browser.test.ts +++ b/ui/src/pages/chat/chat-responsive.browser.test.ts @@ -1384,6 +1384,11 @@ describeBrowserLayout.concurrent("chat responsive browser layout", () => {
Activity
+
+
+
Framed activity
+
+
Pull requests
@@ -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); diff --git a/ui/src/pages/chat/chat-thread-build.ts b/ui/src/pages/chat/chat-thread-build.ts index 26060726fea1..319bb3f5f7dd 100644 --- a/ui/src/pages/chat/chat-thread-build.ts +++ b/ui/src/pages/chat/chat-thread-build.ts @@ -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 { - 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 { let items: ChatItem[] = []; const tools = props.toolMessages.filter((message) => asRecord(message) !== null); @@ -504,7 +417,7 @@ export function buildChatItems(props: BuildChatItemsProps): Array 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 0; -} - function stampReplyAttribution( items: Array, ): Array { @@ -78,6 +76,8 @@ export function groupMessages(items: ChatItem[]): Array 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 !currentGroup || startsProjectedTurn || currentGroup.role !== role || + currentGroup.runId !== runId || splitsAssistantCommentary || splitsRuntimeActivity || (shouldSplitBySender && @@ -114,6 +115,7 @@ export function groupMessages(items: ChatItem[]): Array 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 } 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 - >; + runId?: string; + boundaryId?: string; + parts: Array>; }; - export function coalesceStreamRuns( items: RenderChatItem[], ): Array { const result: Array = []; 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; } diff --git a/ui/src/pages/chat/chat-thread-items.ts b/ui/src/pages/chat/chat-thread-items.ts index d0331244b2c7..0653fbcdfc46 100644 --- a/ui/src/pages/chat/chat-thread-items.ts +++ b/ui/src/pages/chat/chat-thread-items.ts @@ -557,6 +557,7 @@ export function queuedSendThreadMessage(item: ChatQueueItem): Record 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, +): string | undefined { + return part.kind === "question" ? undefined : part.runId; +} + +export function streamPartBoundaryId( + part: Extract, +): 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; +} diff --git a/ui/src/pages/chat/chat-thread.test.ts b/ui/src/pages/chat/chat-thread.test.ts index 93919ac98e0f..eb4d46fa34e7 100644 --- a/ui/src/pages/chat/chat-thread.test.ts +++ b/ui/src/pages/chat/chat-thread.test.ts @@ -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( diff --git a/ui/src/pages/chat/chat-thread.ts b/ui/src/pages/chat/chat-thread.ts index baf8197f9299..8901f415b39f 100644 --- a/ui/src/pages/chat/chat-thread.ts +++ b/ui/src/pages/chat/chat-thread.ts @@ -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) ) { diff --git a/ui/src/pages/chat/chat-view.test.ts b/ui/src/pages/chat/chat-view.test.ts index 7abd7b77f4c3..0eb0af65e126 100644 --- a/ui/src/pages/chat/chat-view.test.ts +++ b/ui/src/pages/chat/chat-view.test.ts @@ -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) => { + vi.mocked(chatThread.buildCachedChatItems).mockReturnValue(items); + renderChatInto(container, { transcript, messages: items }); + }; + + renderItems([user] as ReturnType); + 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); + + 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); + + 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); + + 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); + 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); + 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); diff --git a/ui/src/pages/chat/components/chat-agent-run-frame.ts b/ui/src/pages/chat/components/chat-agent-run-frame.ts new file mode 100644 index 000000000000..b1633a53a6db --- /dev/null +++ b/ui/src/pages/chat/components/chat-agent-run-frame.ts @@ -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[1]; + +type AgentRunFrameOptions = { + questionPrompts: ReadonlyMap; + 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, + }); +} diff --git a/ui/src/pages/chat/components/chat-message-group.ts b/ui/src/pages/chat/components/chat-message-group.ts index 30c8b402ee25..4e22b5adf6bd 100644 --- a/ui/src/pages/chat/components/chat-message-group.ts +++ b/ui/src/pages/chat/components/chat-message-group.ts @@ -108,6 +108,8 @@ type RenderMessageGroupOptions = { rewindDisabled?: boolean; activeContinuation?: ActiveContinuation; turnRecap?: TurnRecap; + frameContent?: unknown; + frameActionOwner?: MessageGroup["messages"][number] | null; }; type GroupedMessageRenderOptions = Parameters[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` -
-
-
- -
- ${activityExpanded - ? groups.map((group) => - group.messages.map((item, index) => - renderGroupedMessage( - item.message, - item.key, - buildGroupedMessageRenderOptions(group, item, index, opts), - opts.onOpenSidebar, - ), - ), - ) - : nothing} -
-
+ + ${reviewOutcome + ? html`${reviewOutcome === "denied" + ? icons.shieldX + : reviewOutcome === "reviewing" + ? icons.shieldQuestion + : icons.shieldCheck}` + : nothing} + + +
+ ${activityExpanded + ? groups.map((group) => + group.messages.map((item, index) => + renderGroupedMessage( + item.message, + item.key, + buildGroupedMessageRenderOptions(group, item, index, opts), + opts.onOpenSidebar, + ), + ), + ) + : nothing}
`; + return presentation === "continuation" + ? content + : html` +
+
${content}
+
+ `; } 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`
${footerActionDetails?.replyTarget && opts.onReply ? renderReplyButton(footerActionDetails.replyTarget, opts.onReply) @@ -492,7 +556,8 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup
` : 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`
${renderMessageActionButtons(actionDetails, opts)} @@ -541,7 +606,7 @@ export function renderMessageGroup(group: MessageGroup, opts: RenderMessageGroup ? html`
${renderMessageActionButtons(footerActionDetails, opts)}
diff --git a/ui/src/pages/chat/components/chat-message-stream.ts b/ui/src/pages/chat/components/chat-message-stream.ts index 7a6dff31dfd2..b0976275303f 100644 --- a/ui/src/pages/chat/components/chat-message-stream.ts +++ b/ui/src/pages/chat/components/chat-message-stream.ts @@ -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` -
-
-
- - -
-
+ const content = html` +
+ +
`; + return opts.presentation === "continuation" + ? content + : html` +
+
${content}
+
+ `; } diff --git a/ui/src/pages/chat/components/chat-message.ts b/ui/src/pages/chat/components/chat-message.ts index 87a97f555c50..2d88271eaa2b 100644 --- a/ui/src/pages/chat/components/chat-message.ts +++ b/ui/src/pages/chat/components/chat-message.ts @@ -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"; diff --git a/ui/src/pages/chat/components/chat-thread-interactions.ts b/ui/src/pages/chat/components/chat-thread-interactions.ts index 187b3a885ca8..8d2e9305bce3 100644 --- a/ui/src/pages/chat/components/chat-thread-interactions.ts +++ b/ui/src/pages/chat/components/chat-thread-interactions.ts @@ -460,7 +460,8 @@ export function handleTranscriptContextMenu(event: MouseEvent, props: Transcript (element) => element.dataset.messageActionsFor === messageId, ); const copyButton = actionOwner?.querySelector(".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"; diff --git a/ui/src/pages/chat/components/chat-transcript-announcement.ts b/ui/src/pages/chat/components/chat-transcript-announcement.ts new file mode 100644 index 000000000000..01c8c31f12d1 --- /dev/null +++ b/ui/src/pages/chat/components/chat-transcript-announcement.ts @@ -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[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; +} diff --git a/ui/src/pages/chat/components/chat-transcript-projection.ts b/ui/src/pages/chat/components/chat-transcript-projection.ts index a83a41b333df..95666d081f4c 100644 --- a/ui/src/pages/chat/components/chat-transcript-projection.ts +++ b/ui/src/pages/chat/components/chat-transcript-projection.ts @@ -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[number]; -const CHAT_TRANSCRIPT_ANNOUNCEMENT_MAX_CHARS = 500; +type ChatRenderItem = ReturnType[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 diff --git a/ui/src/pages/chat/components/chat-transcript-render-guard.ts b/ui/src/pages/chat/components/chat-transcript-render-guard.ts new file mode 100644 index 000000000000..bcb39566415e --- /dev/null +++ b/ui/src/pages/chat/components/chat-transcript-render-guard.ts @@ -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[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), + ); +} diff --git a/ui/src/pages/chat/terminal-message-identity.ts b/ui/src/pages/chat/terminal-message-identity.ts index b28f331af306..5cd0804c06b2 100644 --- a/ui/src/pages/chat/terminal-message-identity.ts +++ b/ui/src/pages/chat/terminal-message-identity.ts @@ -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(); const authoritativeTerminals = new WeakMap(); @@ -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; diff --git a/ui/src/styles/chat/grouped.css b/ui/src/styles/chat/grouped.css index f6c1d0c76405..4f340b4343fc 100644 --- a/ui/src/styles/chat/grouped.css +++ b/ui/src/styles/chat/grouped.css @@ -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; diff --git a/ui/src/styles/chat/tool-cards.css b/ui/src/styles/chat/tool-cards.css index 735d8d904d70..63864e941eec 100644 --- a/ui/src/styles/chat/tool-cards.css +++ b/ui/src/styles/chat/tool-cards.css @@ -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 {