diff --git a/src/gateway/gateway-misc.test.ts b/src/gateway/gateway-misc.test.ts index afb58897a5eb..070a05bd5390 100644 --- a/src/gateway/gateway-misc.test.ts +++ b/src/gateway/gateway-misc.test.ts @@ -406,6 +406,32 @@ describe("gateway broadcaster", () => { expect(getBufferedAmount("c-admin")).toBeUndefined(); }); + it("closes a slow authoritative-session subscriber while delivering to healthy clients", () => { + const slowSocket = makeRecordingSocket(); + slowSocket.bufferedAmount = MAX_BUFFERED_BYTES + 1; + const healthySocket = makeRecordingSocket(); + const clients = makeOperatorWsClients([ + { connId: "slow-session", socket: slowSocket, scopes: ["operator.read"] }, + { connId: "healthy-session", socket: healthySocket, scopes: ["operator.read"] }, + ]); + const { broadcastToConnIds } = createGatewayBroadcaster({ clients }); + const payload = { + sessionKey: "agent:main:main", + messageId: "durable-user-1", + messageSeq: 1, + message: { + role: "user", + content: [{ type: "text", text: "shared durable prompt" }], + }, + }; + + broadcastToConnIds("session.message", payload, new Set(["slow-session", "healthy-session"])); + + expect(slowSocket.close).toHaveBeenCalledWith(1008, "slow consumer"); + expect(slowSocket.send).not.toHaveBeenCalled(); + expect(healthySocket.sent).toEqual([{ type: "event", event: "session.message", payload }]); + }); + it("keeps workers outside all generic and targeted gateway broadcasts", () => { const workerSocket = makeRecordingSocket(); const worker = makeGatewayWsClient("c-worker", workerSocket, { diff --git a/src/gateway/server-session-events.test.ts b/src/gateway/server-session-events.test.ts index eb46956a6030..170da497f84a 100644 --- a/src/gateway/server-session-events.test.ts +++ b/src/gateway/server-session-events.test.ts @@ -103,6 +103,29 @@ describe("createTranscriptUpdateBroadcastHandler", () => { sessionRow.thinkingLevel = "ultra"; }); + it("never silently drops an authoritative session message for a slow subscriber", async () => { + const { broadcastToConnIds, handler } = createHandler(false); + + handler({ + sessionFile: "/tmp/sess-main.jsonl", + sessionKey: "agent:main:main", + message: { role: "user", content: [{ type: "text", text: "shared durable prompt" }] }, + messageId: "durable-user-1", + messageSeq: 1, + }); + + await vi.waitFor(() => expect(broadcastToConnIds).toHaveBeenCalledTimes(1)); + expect(broadcastToConnIds).toHaveBeenCalledWith( + "session.message", + expect.objectContaining({ + sessionKey: "agent:main:main", + messageId: "durable-user-1", + messageSeq: 1, + }), + expect.any(Set), + ); + }); + it("keeps transcript snapshots active while plugin finalization delays the terminal event", async () => { // before_agent_finalize hooks run after the assistant transcript write but // before terminal delivery. The active-run registry remains authoritative diff --git a/src/gateway/server-session-events.ts b/src/gateway/server-session-events.ts index 8b670756febe..5496d5e1a84d 100644 --- a/src/gateway/server-session-events.ts +++ b/src/gateway/server-session-events.ts @@ -311,7 +311,6 @@ async function handleTranscriptUpdateBroadcast( ...sessionSnapshot, }, connIds, - { dropIfSlow: true }, ); return; } diff --git a/src/gateway/session-message-events.test.ts b/src/gateway/session-message-events.test.ts index 0e8a1a5f0d64..c6fa3f3ab31e 100644 --- a/src/gateway/session-message-events.test.ts +++ b/src/gateway/session-message-events.test.ts @@ -795,7 +795,12 @@ describe("session.message websocket events", () => { expect(subscription.ok).toBe(true); } - for (const [index, text] of ["Sent from the web.", "Sent from the TUI."].entries()) { + const sharedMessages = [ + "Sent from the web.", + "Sent from the TUI.", + ...Array.from({ length: 30 }, (_, index) => `Shared burst message ${index + 1}.`), + ]; + for (const [index, text] of sharedMessages.entries()) { const messageId = `shared-turn-${index + 1}`; const deliveries = [webWs, tuiWs].map((ws) => onceMessage( @@ -846,10 +851,12 @@ describe("session.message websocket events", () => { await connectOk(reconnectedTuiWs, { scopes: ["operator.read"] }); const history = await rpcReq(reconnectedTuiWs, "chat.history", { sessionKey }); expect(history.ok).toBe(true); - expect((history.payload as { messages?: unknown[] }).messages).toMatchObject([ - { content: [{ type: "text", text: "Sent from the web." }], role: "user" }, - { content: [{ type: "text", text: "Sent from the TUI." }], role: "user" }, - ]); + expect((history.payload as { messages?: unknown[] }).messages).toMatchObject( + sharedMessages.map((text) => ({ + content: [{ type: "text", text }], + role: "user", + })), + ); } finally { webWs.close(); tuiWs.close(); diff --git a/src/tui/components/chat-log.test.ts b/src/tui/components/chat-log.test.ts index 9f248445e791..1b8fe741064e 100644 --- a/src/tui/components/chat-log.test.ts +++ b/src/tui/components/chat-log.test.ts @@ -355,6 +355,234 @@ describe("ChatLog", () => { expect(chatLog.countPendingUsers()).toBe(1); }); + it("preserves live users when same-session history is rebuilt from a stale snapshot", () => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("Sent from the other client.", { + messageId: "shared-user", + runId: "shared-run", + }); + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addUser("Already persisted in history.", { messageId: "history-user" }); + chatLog.restoreLiveUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered).toContain("Sent from the other client."); + expect(rendered.indexOf("Already persisted in history.")).toBeLessThan( + rendered.indexOf("Sent from the other client."), + ); + expect(chatLog.children).toHaveLength(2); + }); + + it("does not resurrect historical users omitted by the authoritative history snapshot", () => { + const chatLog = new ChatLog(40); + + chatLog.addUser("Deleted historical prompt.", { + messageId: "deleted-history-user", + messageSeq: 1, + }); + chatLog.addLiveUser("New authoritative live prompt.", { + messageId: "live-user", + messageSeq: 3, + }); + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addUser("Current authoritative history.", { + messageId: "current-history-user", + messageSeq: 2, + }); + chatLog.restoreLiveUsers(4); + chatLog.finalizeAssistant("Current authoritative reply."); + chatLog.restoreLiveUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered).not.toContain("Deleted historical prompt."); + expect(rendered.match(/New authoritative live prompt\./g)).toHaveLength(1); + expect(rendered.indexOf("New authoritative live prompt.")).toBeLessThan( + rendered.indexOf("Current authoritative reply."), + ); + }); + + it("stops restoring live users after authoritative history adopts their identity", () => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("Adopted live prompt.", { + messageId: "adopted-user", + messageSeq: 1, + }); + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addUser("Adopted live prompt.", { + messageId: "adopted-user", + messageSeq: 1, + }); + chatLog.restoreLiveUsers(); + + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addUser("Replacement branch prompt.", { + messageId: "replacement-user", + messageSeq: 2, + }); + chatLog.restoreLiveUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered).toContain("Replacement branch prompt."); + expect(rendered).not.toContain("Adopted live prompt."); + }); + + it("restores a missing canonical user before its higher-sequence persisted reply", () => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("Authoritative shared prompt.", { + messageId: "shared-user", + messageSeq: 1, + runId: "shared-run", + }); + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addSystem("session agent:main:main"); + chatLog.restoreLiveUsers(2); + chatLog.finalizeAssistant("Already persisted reply."); + chatLog.restoreLiveUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered.match(/Authoritative shared prompt\./g)).toHaveLength(1); + expect(rendered.indexOf("Authoritative shared prompt.")).toBeLessThan( + rendered.indexOf("Already persisted reply."), + ); + }); + + it("restores live canonical users only before higher-sequence history rows", () => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("First missing prompt.", { + messageId: "shared-user-2", + messageSeq: 2, + }); + chatLog.addLiveUser("Second missing prompt.", { + messageId: "shared-user-4", + messageSeq: 4, + }); + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addUser("First persisted prompt.", { + messageId: "history-user-1", + messageSeq: 1, + }); + chatLog.restoreLiveUsers(3); + chatLog.addUser("Third persisted prompt.", { + messageId: "history-user-3", + messageSeq: 3, + }); + chatLog.restoreLiveUsers(5); + chatLog.finalizeAssistant("Fifth persisted reply."); + chatLog.restoreLiveUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + const messages = [ + "First persisted prompt.", + "First missing prompt.", + "Third persisted prompt.", + "Second missing prompt.", + "Fifth persisted reply.", + ]; + let previousMessage: string | undefined; + for (const message of messages) { + if (previousMessage !== undefined) { + expect(rendered.indexOf(previousMessage)).toBeLessThan(rendered.indexOf(message)); + } + previousMessage = message; + } + }); + + it("does not restore a live user already included in rebuilt authoritative history", () => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("Original live prompt.", { + messageId: "shared-user", + runId: "shared-run", + }); + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addUser("Authoritative persisted prompt.", { messageId: "shared-user" }); + chatLog.restoreLiveUsers(); + + let rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered).toContain("Authoritative persisted prompt."); + expect(rendered).not.toContain("Original live prompt."); + expect(chatLog.children).toHaveLength(1); + + chatLog.addLiveUser("Updated persisted prompt.", { + messageId: "shared-user", + runId: "shared-run", + }); + + rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered).toContain("Updated persisted prompt."); + expect(rendered).not.toContain("Authoritative persisted prompt."); + expect(chatLog.children).toHaveLength(1); + }); + + it("restores multiple live users in canonical event order", () => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("First shared prompt.", { + messageId: "shared-user-1", + runId: "shared-run-1", + }); + chatLog.addLiveUser("Second shared prompt.", { + messageId: "shared-user-2", + runId: "shared-run-2", + }); + chatLog.clearAll({ preserveLiveUsers: true }); + chatLog.addUser("Already persisted in history.", { messageId: "history-user" }); + chatLog.restoreLiveUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered.indexOf("Already persisted in history.")).toBeLessThan( + rendered.indexOf("First shared prompt."), + ); + expect(rendered.indexOf("First shared prompt.")).toBeLessThan( + rendered.indexOf("Second shared prompt."), + ); + expect(chatLog.children).toHaveLength(3); + }); + + it("restores both live and pending users across a same-session history rebuild", () => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("Sent from the other client.", { + messageId: "shared-user", + runId: "shared-run", + }); + chatLog.addPendingUser("local-run", "My pending prompt."); + chatLog.clearAll({ preserveLiveUsers: true, preservePendingUsers: true }); + chatLog.addUser("Already persisted in history.", { messageId: "history-user" }); + chatLog.restoreLiveUsers(); + chatLog.restorePendingUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered).toContain("Sent from the other client."); + expect(rendered).toContain("My pending prompt."); + expect(chatLog.countPendingUsers()).toBe(1); + expect(chatLog.children).toHaveLength(3); + }); + + it.each([ + { clear: "session switch", options: undefined }, + { clear: "pending-only rebuild", options: { preservePendingUsers: true } }, + ])("does not leak live users after a $clear", ({ options }) => { + const chatLog = new ChatLog(40); + + chatLog.addLiveUser("A previous session's prompt.", { + messageId: "previous-session-user", + runId: "previous-session-run", + }); + chatLog.clearAll(options); + chatLog.addUser("Current session history.", { messageId: "current-session-user" }); + chatLog.restoreLiveUsers(); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(rendered).toContain("Current session history."); + expect(rendered).not.toContain("A previous session's prompt."); + expect(chatLog.children).toHaveLength(1); + }); + it("does not append the same pending component twice when it is already mounted", () => { const chatLog = new ChatLog(40); @@ -419,6 +647,119 @@ describe("ChatLog", () => { ); }); + it("preserves a delayed shared prompt and its streaming reply at the scrollback limit", () => { + const chatLog = new ChatLog(20); + chatLog.startAssistant("Already streaming.", "shared-run"); + for (let index = 0; index < 19; index += 1) { + chatLog.addSystem(`notice-${index}`); + } + + chatLog.addLiveUser("Sent from the other client.", { + messageId: "shared-user", + runId: "shared-run", + }); + + let rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(chatLog.children).toHaveLength(20); + expect(rendered).toContain("Sent from the other client."); + expect(rendered.indexOf("Sent from the other client.")).toBeLessThan( + rendered.indexOf("Already streaming."), + ); + + chatLog.addLiveUser("Sent from the other client.", { + messageId: "shared-user", + runId: "shared-run", + }); + chatLog.updateAssistant("Still streaming.", "shared-run"); + + rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(chatLog.children).toHaveLength(20); + expect(rendered.match(/Sent from the other client\./g)).toHaveLength(1); + expect(rendered.indexOf("Sent from the other client.")).toBeLessThan( + rendered.indexOf("Still streaming."), + ); + }); + + it("evicts an unrelated older tool instead of a newer transcript row at full scrollback", () => { + const chatLog = new ChatLog(20); + chatLog.startTool("unrelated-old-tool", "read_file", { path: "unrelated-old.txt" }); + chatLog.startAssistant("Current streaming reply.", "shared-run"); + for (let index = 0; index < 18; index += 1) { + chatLog.addSystem(`newer-notice-${index}`); + } + + chatLog.addLiveUser("Current authoritative prompt.", { + messageId: "shared-user", + runId: "shared-run", + }); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(chatLog.children).toHaveLength(20); + expect(rendered).not.toContain("Read File"); + expect(rendered).toContain("newer-notice-0"); + expect(rendered).toContain("Current streaming reply."); + expect(rendered.indexOf("Current authoritative prompt.")).toBeLessThan( + rendered.indexOf("Current streaming reply."), + ); + }); + + it("preserves a delayed shared prompt, frozen reply, and tool at the scrollback limit", () => { + const chatLog = new ChatLog(20); + chatLog.startAssistant("Before the tool.", "shared-run"); + chatLog.startTool("shared-tool", "read_file", { path: "shared.txt" }); + for (let index = 0; index < 18; index += 1) { + chatLog.addSystem(`notice-${index}`); + } + + chatLog.addLiveUser("Sent from the other client.", { + messageId: "shared-user", + runId: "shared-run", + }); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(chatLog.children).toHaveLength(20); + expect(rendered).toContain("Sent from the other client."); + expect(rendered).toContain("Before the tool."); + expect(rendered).toContain("Read File"); + expect(rendered.indexOf("Sent from the other client.")).toBeLessThan( + rendered.indexOf("Before the tool."), + ); + expect(rendered.indexOf("Before the tool.")).toBeLessThan(rendered.indexOf("Read File")); + }); + + it("keeps scrollback bounded when every visible tool belongs to the delayed prompt's run", () => { + const chatLog = new ChatLog(20); + chatLog.startAssistant("Reply with many tools.", "shared-run"); + for (let index = 0; index < 19; index += 1) { + chatLog.startTool( + `shared-tool-${index}`, + "read_file", + { path: `shared-${index}.txt` }, + "shared-run", + ); + } + + chatLog.addLiveUser("Authoritative prompt before many tools.", { + messageId: "shared-user", + runId: "shared-run", + }); + + const rendered = normalizeTestText(chatLog.render(120).join("\n")); + expect(chatLog.children).toHaveLength(20); + expect(rendered.match(/Authoritative prompt before many tools\./g)).toHaveLength(1); + expect(rendered.indexOf("Authoritative prompt before many tools.")).toBeLessThan( + rendered.indexOf("Reply with many tools."), + ); + expect(rendered).not.toContain("shared-0.txt"); + expect(rendered).toContain("shared-18.txt"); + + chatLog.updateToolResult("shared-tool-0", { + content: [{ type: "text", text: "evicted tool must stay detached" }], + }); + expect(chatLog.children).toHaveLength(20); + expect(chatLog.render(120).join("\n")).not.toContain("evicted tool must stay detached"); + }); + it("deduplicates authoritative user events and adopts the matching pending prompt", () => { const chatLog = new ChatLog(40); chatLog.addPendingUser("shared-run", "Persisted prompt."); diff --git a/src/tui/components/chat-log.ts b/src/tui/components/chat-log.ts index ec47a01236df..38f4d52baafc 100644 --- a/src/tui/components/chat-log.ts +++ b/src/tui/components/chat-log.ts @@ -21,11 +21,14 @@ type RepeatableSystemMessage = { export class ChatLog extends Container { private readonly maxComponents: number; private toolById = new Map(); + private toolRunIds = new Map(); private streamingRuns = new Map(); private frozenAssistants = new Map>(); private committedAssistantText = new Map(); private latestAssistantText = new Map(); private liveUsers = new Map(); + private liveUserSequences = new Map(); + private liveEventUserIds = new Set(); private pendingUsers = new Map< string, { @@ -49,6 +52,7 @@ export class ChatLog extends Container { for (const [toolId, tool] of this.toolById.entries()) { if (tool === component) { this.toolById.delete(toolId); + this.toolRunIds.delete(toolId); } } for (const [runId, message] of this.streamingRuns.entries()) { @@ -72,6 +76,8 @@ export class ChatLog extends Container { for (const [messageId, user] of this.liveUsers.entries()) { if (user === component) { this.liveUsers.delete(messageId); + this.liveUserSequences.delete(messageId); + this.liveEventUserIds.delete(messageId); } } for (const [runId, entry] of this.pendingSystemNotices.entries()) { @@ -89,8 +95,11 @@ export class ChatLog extends Container { private pruneOverflow(protectedComponents?: ReadonlySet) { while (this.children.length > this.maxComponents) { + // Protect only the inserted prompt, its reply, and tools owned by that run. + // If owned tools fill the log, evict the oldest tool, never the prompt or reply. const oldest = protectedComponents - ? this.children.find((component) => !protectedComponents.has(component)) + ? (this.children.find((component) => !protectedComponents.has(component)) ?? + this.children.find((component) => component instanceof ToolExecutionComponent)) : this.children[0]; if (!oldest) { return; @@ -110,14 +119,28 @@ export class ChatLog extends Container { this.append(component); } - clearAll(opts?: { preservePendingUsers?: boolean }) { + clearAll(opts?: { preservePendingUsers?: boolean; preserveLiveUsers?: boolean }) { this.clear(); this.toolById.clear(); + this.toolRunIds.clear(); this.streamingRuns.clear(); this.frozenAssistants.clear(); this.committedAssistantText.clear(); this.latestAssistantText.clear(); - this.liveUsers.clear(); + if (opts?.preserveLiveUsers) { + // History rows are authoritative snapshots, not in-flight live events. + // Keeping them would resurrect deleted or switched-away transcript branches. + for (const messageId of this.liveUsers.keys()) { + if (!this.liveEventUserIds.has(messageId)) { + this.liveUsers.delete(messageId); + this.liveUserSequences.delete(messageId); + } + } + } else { + this.liveUsers.clear(); + this.liveUserSequences.clear(); + this.liveEventUserIds.clear(); + } this.pendingSystemNotices.clear(); this.btwMessage = null; this.repeatableSystemMessage = null; @@ -131,6 +154,29 @@ export class ChatLog extends Container { this.removeChild(tool); } this.toolById.clear(); + this.toolRunIds.clear(); + } + + restoreLiveUsers(beforeMessageSeq?: number) { + // Rebuilt history replaces matching IDs in addUser; only live prompts + // missing from a stale snapshot are restored before the next canonical row. + for (const messageId of this.liveEventUserIds) { + const component = this.liveUsers.get(messageId); + if (!component) { + this.liveEventUserIds.delete(messageId); + continue; + } + if (this.children.includes(component)) { + continue; + } + if (beforeMessageSeq !== undefined) { + const messageSeq = this.liveUserSequences.get(messageId); + if (messageSeq === undefined || messageSeq >= beforeMessageSeq) { + continue; + } + } + this.appendNonSystem(component); + } } restorePendingUsers() { @@ -203,15 +249,25 @@ export class ChatLog extends Container { return true; } - addUser(text: string, options?: { messageId?: string }) { + addUser(text: string, options?: { messageId?: string; messageSeq?: number }) { const component = new UserMessageComponent(text); if (options?.messageId) { this.liveUsers.set(options.messageId, component); + // Once authoritative history contains this identity it is no longer a + // missing live event and must not survive a later deletion or branch. + this.liveEventUserIds.delete(options.messageId); + if (options.messageSeq !== undefined) { + this.liveUserSequences.set(options.messageId, options.messageSeq); + } } this.appendNonSystem(component); } - addLiveUser(text: string, options: { messageId: string; runId?: string }) { + addLiveUser(text: string, options: { messageId: string; messageSeq?: number; runId?: string }) { + this.liveEventUserIds.add(options.messageId); + if (options.messageSeq !== undefined) { + this.liveUserSequences.set(options.messageId, options.messageSeq); + } const existing = this.liveUsers.get(options.messageId); if (existing) { existing.setText(text); @@ -239,7 +295,15 @@ export class ChatLog extends Container { // older components so the newly recovered prompt cannot disappear. this.repeatableSystemMessage = null; this.children.splice(assistantIndex, 0, component); - this.pruneOverflow(new Set([component, assistant])); + const protectedComponents = new Set([component, assistant]); + if (options.runId) { + for (const [toolId, tool] of this.toolById) { + if (this.toolRunIds.get(toolId) === options.runId) { + protectedComponents.add(tool); + } + } + } + this.pruneOverflow(protectedComponents); return component; } this.appendNonSystem(component); @@ -477,16 +541,21 @@ export class ChatLog extends Container { return this.btwMessage !== null; } - startTool(toolCallId: string, toolName: string, args: unknown) { + startTool(toolCallId: string, toolName: string, args: unknown, runId?: string) { const existing = this.toolById.get(toolCallId); if (existing) { existing.setArgs(args); return existing; } + const owningRunId = + runId ?? (this.streamingRuns.size === 1 ? this.streamingRuns.keys().next().value : undefined); this.freezeStreamingAssistants(); const component = new ToolExecutionComponent(toolName, args); component.setExpanded(this.toolsExpanded); this.toolById.set(toolCallId, component); + if (owningRunId) { + this.toolRunIds.set(toolCallId, owningRunId); + } this.appendNonSystem(component); return component; } diff --git a/src/tui/tui-event-handlers.test.ts b/src/tui/tui-event-handlers.test.ts index 4218dd06984a..ce53ed6549ea 100644 --- a/src/tui/tui-event-handlers.test.ts +++ b/src/tui/tui-event-handlers.test.ts @@ -239,7 +239,12 @@ describe("tui-event-handlers: handleAgentEvent", () => { handleAgentEvent(evt); - expect(chatLog.startTool).toHaveBeenCalledWith("tc1", "exec", { command: "echo hi" }); + expect(chatLog.startTool).toHaveBeenCalledWith( + "tc1", + "exec", + { command: "echo hi" }, + "run-123", + ); expect(tui.requestRender).toHaveBeenCalledTimes(1); }); @@ -968,7 +973,7 @@ describe("tui-event-handlers: handleAgentEvent", () => { handleAgentEvent(agentEvt); - expect(chatLog.startTool).toHaveBeenCalledWith("tc1", "exec", undefined); + expect(chatLog.startTool).toHaveBeenCalledWith("tc1", "exec", undefined, "run-42"); }); it("accepts chat events when session key is an alias of the active canonical key", () => { @@ -1647,7 +1652,12 @@ describe("tui-event-handlers: handleAgentEvent", () => { data: { phase: "start", toolCallId: "tc-final", name: "session_status" }, }); - expect(chatLog.startTool).toHaveBeenCalledWith("tc-final", "session_status", undefined); + expect(chatLog.startTool).toHaveBeenCalledWith( + "tc-final", + "session_status", + undefined, + "run-final", + ); expect(tui.requestRender).toHaveBeenCalled(); }); @@ -2506,6 +2516,7 @@ describe("tui-event-handlers: handleAgentEvent", () => { expect(chatLog.addLiveUser).toHaveBeenCalledWith("Sent from the other client.", { messageId: "shared-session-user", + messageSeq: 1, runId, }); expect(state.activeChatRunId).toBe(runId); diff --git a/src/tui/tui-event-handlers.ts b/src/tui/tui-event-handlers.ts index 51bf634a2f47..b340987a314f 100644 --- a/src/tui/tui-event-handlers.ts +++ b/src/tui/tui-event-handlers.ts @@ -7,7 +7,11 @@ import { sanitizeRenderableText, } from "./tui-formatters.js"; import { createTuiRunLifecycle } from "./tui-run-lifecycle.js"; -import { matchesSelectedTuiSession, readTuiSessionUserMessage } from "./tui-session-events.js"; +import { + matchesSelectedTuiSession, + readTuiSessionUserMessage, + readTuiTranscriptMessageSequence, +} from "./tui-session-events.js"; import { TuiSessionRunCoordinator } from "./tui-session-run-coordinator.js"; import { clearPendingSubmit, @@ -25,8 +29,11 @@ import type { } from "./tui-types.js"; type EventHandlerChatLog = { - addLiveUser: (text: string, options: { messageId: string; runId?: string }) => void; - startTool: (toolCallId: string, toolName: string, args: unknown) => void; + addLiveUser: ( + text: string, + options: { messageId: string; messageSeq?: number; runId?: string }, + ) => void; + startTool: (toolCallId: string, toolName: string, args: unknown, runId?: string) => void; updateToolResult: ( toolCallId: string, result: unknown, @@ -491,8 +498,16 @@ export function createEventHandlers(context: EventHandlerContext) { const liveUserMessage = readTuiSessionUserMessage(evt); if (liveUserMessage) { + const envelopeSequence = evt.messageSeq; + const messageSeq = + typeof envelopeSequence === "number" && + Number.isSafeInteger(envelopeSequence) && + envelopeSequence > 0 + ? envelopeSequence + : readTuiTranscriptMessageSequence(evt.message); chatLog.addLiveUser(liveUserMessage.text, { messageId: liveUserMessage.messageId, + ...(messageSeq !== undefined ? { messageSeq } : {}), ...(liveUserMessage.runId ? { runId: liveUserMessage.runId } : {}), }); tui.requestRender(); @@ -588,7 +603,7 @@ export function createEventHandlers(context: EventHandlerContext) { return; } if (phase === "start") { - chatLog.startTool(toolCallId, toolName, data.args); + chatLog.startTool(toolCallId, toolName, data.args, evt.runId); } else if (phase === "update") { if (!allowToolOutput) { return; diff --git a/src/tui/tui-formatters.test.ts b/src/tui/tui-formatters.test.ts index 3244cba31ce0..253a3ba67f52 100644 --- a/src/tui/tui-formatters.test.ts +++ b/src/tui/tui-formatters.test.ts @@ -60,6 +60,142 @@ describe("formatGoalFooter", () => { }); describe("extractTextFromMessage", () => { + it.each([ + { + name: "a browser image block", + content: [ + { + type: "image", + url: "/persisted-image.png", + source: { type: "url", url: "/persisted-image.png" }, + }, + ], + expected: "Attached image", + }, + { + name: "a persisted image block", + content: [{ type: "image", source: { type: "url", url: "/persisted-image.png" } }], + expected: "Attached image", + }, + { + name: "a browser document block", + content: [ + { + type: "attachment", + attachment: { + url: "/report.pdf", + kind: "document", + label: "report.pdf", + mimeType: "application/pdf", + }, + }, + ], + expected: "Attached file: report.pdf", + }, + { + name: "a browser audio block", + content: [ + { + type: "attachment", + attachment: { + url: "/voice.ogg", + kind: "audio", + label: "voice.ogg", + mimeType: "audio/ogg", + }, + }, + ], + expected: "Attached file: voice.ogg", + }, + { + name: "a browser file with the default label", + content: [ + { + type: "attachment", + attachment: { url: "/document", kind: "document", label: "Attached file" }, + }, + ], + expected: "Attached file", + }, + { + name: "multiple ordered browser attachments", + content: [ + { type: "image", source: { type: "url", url: "/image.png" } }, + { + type: "attachment", + attachment: { url: "/report.pdf", kind: "document", label: "report.pdf" }, + }, + ], + expected: "Attached image\nAttached file: report.pdf", + }, + ])("renders an attachment-only user turn containing $name", ({ content, expected }) => { + expect(extractTextFromMessage({ role: "user", content })).toBe(expected); + }); + + it.each([ + { + name: "image", + media: [{ path: "/media/inbound/generated-image.png", contentType: "image/png" }], + expected: "Attached image", + }, + { + name: "image inferred from its canonical path", + media: [{ path: "/media/inbound/media-only.png" }], + expected: "Attached image", + }, + { + name: "file", + media: [{ path: "/media/inbound/generated-report.pdf", contentType: "application/pdf" }], + expected: "Attached file", + }, + { + name: "ordered image and file", + media: [ + { path: "/media/inbound/generated-image.png", contentType: "image/png" }, + { path: "/media/inbound/generated-report.pdf", contentType: "application/pdf" }, + ], + expected: "Attached image\nAttached file", + }, + ])("renders an empty durable user turn with canonical $name media", ({ media, expected }) => { + expect( + extractTextFromMessage({ + role: "user", + content: "", + __openclaw: { media }, + }), + ).toBe(expected); + }); + + it("keeps an ordinary user prompt unchanged when it also has attachments", () => { + expect( + extractTextFromMessage({ + role: "user", + content: [ + { type: "text", text: "Describe this image" }, + { type: "image", source: { type: "url", url: "/image.png" } }, + ], + }), + ).toBe("Describe this image"); + }); + + it("sanitizes the display name of an attachment-only user turn", () => { + expect( + extractTextFromMessage({ + role: "user", + content: [ + { + type: "attachment", + attachment: { + url: "/report.pdf", + kind: "document", + label: "\u001b[31mreport.pdf\u001b[0m\u0000", + }, + }, + ], + }), + ).toBe("Attached file: report.pdf"); + }); + it("prefers final_answer text over commentary text for assistant messages", () => { const text = extractTextFromMessage({ role: "assistant", diff --git a/src/tui/tui-formatters.ts b/src/tui/tui-formatters.ts index 08158d6612e8..d80de0fb7586 100644 --- a/src/tui/tui-formatters.ts +++ b/src/tui/tui-formatters.ts @@ -4,6 +4,7 @@ import { splitTrailingAuthProfile } from "../agents/model-ref-profile.js"; import { stripLeadingInboundMetadata } from "../auto-reply/reply/strip-inbound-meta.js"; import type { SessionGoal } from "../config/sessions/types.js"; import { formatErrorMessage } from "../infra/errors.js"; +import { isImageMediaFact, readPersistedMediaFacts } from "../media/media-facts.js"; import { formatRawAssistantErrorForUi } from "../shared/assistant-error-format.js"; import { extractAssistantVisibleText } from "../shared/chat-message-content.js"; import { chunkTextByBreakResolver } from "../shared/text-chunking.js"; @@ -431,6 +432,37 @@ function extractTextBlocks(content: unknown, opts?: { includeThinking?: boolean }); } +function extractUserAttachmentText(record: Record): string { + const attachments: string[] = []; + if (Array.isArray(record.content)) { + for (const block of record.content) { + const entry = asMessageRecord(block); + if (entry?.type === "image") { + attachments.push("Attached image"); + } else if (entry?.type === "attachment") { + const attachment = asMessageRecord(entry.attachment); + const label = + typeof attachment?.label === "string" + ? sanitizeRenderableText(attachment.label).trim() + : ""; + attachments.push( + label && label !== "Attached file" ? `Attached file: ${label}` : "Attached file", + ); + } + } + } + if (attachments.length > 0) { + return attachments.join("\n"); + } + + // Gateway-persisted attachment-only turns keep blank content and carry + // their authoritative attachments in __openclaw.media instead. + return (readPersistedMediaFacts(record) ?? []) + .filter((fact) => fact.path || fact.url || fact.contentType || fact.kind) + .map((fact) => (isImageMediaFact(fact) ? "Attached image" : "Attached file")) + .join("\n"); +} + export function extractTextFromMessage( message: unknown, opts?: { includeThinking?: boolean }, @@ -454,6 +486,10 @@ export function extractTextFromMessage( return text; } + if (record.role === "user") { + return extractUserAttachmentText(record); + } + const errorText = formatAssistantErrorFromRecord(record); if (!errorText) { return ""; diff --git a/src/tui/tui-pty-harness.e2e.test.ts b/src/tui/tui-pty-harness.e2e.test.ts index ca8cde7a4029..c77b6f95aa1b 100644 --- a/src/tui/tui-pty-harness.e2e.test.ts +++ b/src/tui/tui-pty-harness.e2e.test.ts @@ -190,8 +190,8 @@ describe.sequential("TUI PTY harness", () => { STARTUP_TEST_TIMEOUT_MS, ); - it.each([{ failures: 1 }, { failures: 2 }])( - "bounds session subscription recovery after $failures startup failures", + it.each([{ failures: 1 }, { failures: 2 }, { failures: 3 }, { failures: 4 }])( + "recovers session subscription after $failures startup failures", async ({ failures }) => { const subscriptionFixture = await startTuiFixture({ env: { OPENCLAW_TUI_PTY_SUBSCRIBE_FAILURES: String(failures) }, @@ -200,7 +200,7 @@ describe.sequential("TUI PTY harness", () => { await subscriptionFixture.run.waitForOutput("local ready | idle", STARTUP_TIMEOUT_MS); const entries = await readFixtureLog(subscriptionFixture.logPath); expect(entries.filter((entry) => entry.method === "subscribeSessionEvents")).toHaveLength( - 2, + failures + 1, ); expect(entries.filter((entry) => entry.method === "subscribeSessionFailure")).toHaveLength( failures, @@ -218,6 +218,33 @@ describe.sequential("TUI PTY harness", () => { STARTUP_TEST_TIMEOUT_MS, ); + it( + "never reports ready after exhausting session subscription recovery", + async () => { + const subscriptionFixture = await startTuiFixture({ + env: { OPENCLAW_TUI_PTY_SUBSCRIBE_FAILURES: "5" }, + }); + try { + await subscriptionFixture.run.waitForOutput( + "session event subscribe failed", + STARTUP_TIMEOUT_MS, + ); + const entries = await readFixtureLog(subscriptionFixture.logPath); + expect(entries.filter((entry) => entry.method === "subscribeSessionEvents")).toHaveLength( + 5, + ); + expect(entries.filter((entry) => entry.method === "subscribeSessionFailure")).toHaveLength( + 5, + ); + expect(entries.some((entry) => entry.method === "loadHistory")).toBe(false); + expect(subscriptionFixture.run.visibleOutput()).not.toContain("local ready | idle"); + } finally { + await subscriptionFixture.cleanup(); + } + }, + STARTUP_TEST_TIMEOUT_MS, + ); + it("refreshes pending approvals before loading history", async () => { await fixture.waitForLogEntry((entry) => entry.method === "listPluginApprovals"); await fixture.waitForLogEntry((entry) => entry.method === "listTaskSuggestions"); diff --git a/src/tui/tui-session-actions.test.ts b/src/tui/tui-session-actions.test.ts index 2074c77c2712..d1cd1c386f7c 100644 --- a/src/tui/tui-session-actions.test.ts +++ b/src/tui/tui-session-actions.test.ts @@ -1,6 +1,6 @@ // Covers TUI session action routing and backend calls. import { describe, expect, it, vi } from "vitest"; -import type { ChatLog } from "./components/chat-log.js"; +import { ChatLog } from "./components/chat-log.js"; import type { TuiBackend } from "./tui-backend.js"; import { createSessionActions } from "./tui-session-actions.js"; import { TUI_SESSION_LOOKUP_LIMIT } from "./tui-session-list-policy.js"; @@ -44,6 +44,7 @@ describe("tui session actions", () => { addUser, finalizeAssistant: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue([]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), updateAssistant: vi.fn(), startTool: vi.fn(), @@ -87,6 +88,7 @@ describe("tui session actions", () => { clearPendingUsers: vi.fn(), clearAll: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue([]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), } as unknown as import("./components/chat-log.js").ChatLog, btw: createBtwPresenter(), @@ -538,6 +540,259 @@ describe("tui session actions", () => { }); }); + it("preserves an authoritative user received while same-session history is loading", async () => { + const deferredHistory = createDeferred(); + const chatLog = new ChatLog(); + const state = createBaseState({ currentSessionId: "session-main" }); + const { loadHistory } = createTestSessionActions({ + client: { + listSessions: vi.fn(), + loadHistory: vi.fn(() => deferredHistory.promise), + } as unknown as TuiBackend, + chatLog, + state, + }); + + const loading = loadHistory(); + chatLog.addLiveUser("Browser prompt received during refresh", { + messageId: "shared-user-1", + messageSeq: 1, + runId: "shared-run-1", + }); + deferredHistory.resolve({ + sessionId: "session-main", + sessionInfo: { key: "agent:main:main", sessionId: "session-main" }, + messages: [ + { + role: "assistant", + content: "History completed", + __openclaw: { id: "shared-assistant-1", seq: 2 }, + }, + ], + }); + + await expect(loading).resolves.toMatchObject({ loaded: true }); + const rendered = chatLog.render(120).join("\n"); + expect(rendered).toContain("Browser prompt received during refresh"); + expect(rendered.match(/Browser prompt received during refresh/g)).toHaveLength(1); + expect(rendered).toContain("History completed"); + expect(rendered.indexOf("Browser prompt received during refresh")).toBeLessThan( + rendered.indexOf("History completed"), + ); + }); + + it("does not restore deleted history while recovering an in-flight authoritative live user", async () => { + const deferredHistory = createDeferred(); + const chatLog = new ChatLog(); + chatLog.addUser("Deleted prompt from an earlier branch", { + messageId: "deleted-history-user", + messageSeq: 1, + }); + const state = createBaseState({ currentSessionId: "session-main" }); + const { loadHistory } = createTestSessionActions({ + client: { + listSessions: vi.fn(), + loadHistory: vi.fn(() => deferredHistory.promise), + } as unknown as TuiBackend, + chatLog, + state, + }); + + const loading = loadHistory(); + chatLog.addLiveUser("Browser prompt received during refresh", { + messageId: "live-user", + messageSeq: 3, + runId: "live-run", + }); + deferredHistory.resolve({ + sessionId: "session-main", + sessionInfo: { key: "agent:main:main", sessionId: "session-main" }, + messages: [ + { + role: "user", + content: "Current branch prompt", + __openclaw: { id: "current-history-user", seq: 2 }, + }, + { + role: "assistant", + content: "Current branch reply", + __openclaw: { id: "current-assistant", seq: 4 }, + }, + ], + }); + + await expect(loading).resolves.toMatchObject({ loaded: true }); + const rendered = chatLog.render(120).join("\n"); + expect(rendered).not.toContain("Deleted prompt from an earlier branch"); + expect(rendered.match(/Browser prompt received during refresh/g)).toHaveLength(1); + expect(rendered.indexOf("Browser prompt received during refresh")).toBeLessThan( + rendered.indexOf("Current branch reply"), + ); + }); + + it("deduplicates an authoritative live user already included in same-session history", async () => { + const deferredHistory = createDeferred(); + const chatLog = new ChatLog(); + const state = createBaseState({ currentSessionId: "session-main" }); + const { loadHistory } = createTestSessionActions({ + client: { + listSessions: vi.fn(), + loadHistory: vi.fn(() => deferredHistory.promise), + } as unknown as TuiBackend, + chatLog, + state, + }); + + const loading = loadHistory(); + chatLog.addLiveUser("Persisted browser prompt", { + messageId: "shared-user-2", + messageSeq: 1, + runId: "shared-run-2", + }); + deferredHistory.resolve({ + sessionId: "session-main", + sessionInfo: { key: "agent:main:main", sessionId: "session-main" }, + messages: [ + { + role: "user", + content: "Persisted browser prompt", + __openclaw: { id: "shared-user-2", seq: 1 }, + }, + { + role: "assistant", + content: "Persisted reply", + __openclaw: { id: "shared-assistant-2", seq: 2 }, + }, + ], + }); + + await expect(loading).resolves.toMatchObject({ loaded: true }); + const rendered = chatLog.render(120).join("\n"); + expect(rendered.match(/Persisted browser prompt/g)).toHaveLength(1); + expect(rendered).toContain("Persisted reply"); + }); + + it("preserves new-session live users without leaking the previous session during a switch", async () => { + const deferredHistory = createDeferred(); + const chatLog = new ChatLog(); + chatLog.addLiveUser("Private prompt from previous session", { + messageId: "previous-user", + runId: "previous-run", + }); + const state = createBaseState({ currentSessionId: "session-main" }); + const { setSession } = createTestSessionActions({ + client: { + listSessions: vi.fn(), + loadHistory: vi.fn(() => deferredHistory.promise), + } as unknown as TuiBackend, + chatLog, + state, + }); + + const switching = setSession("agent:main:other"); + chatLog.addLiveUser("Browser prompt in selected session", { + messageId: "other-user", + messageSeq: 1, + runId: "other-run", + }); + deferredHistory.resolve({ + sessionId: "session-other", + sessionInfo: { key: "agent:main:other", sessionId: "session-other" }, + messages: [ + { + role: "assistant", + content: "Other session reply", + __openclaw: { id: "other-assistant", seq: 2 }, + }, + ], + }); + + await switching; + const rendered = chatLog.render(120).join("\n"); + expect(rendered).toContain("Browser prompt in selected session"); + expect(rendered.match(/Browser prompt in selected session/g)).toHaveLength(1); + expect(rendered).toContain("Other session reply"); + expect(rendered.indexOf("Browser prompt in selected session")).toBeLessThan( + rendered.indexOf("Other session reply"), + ); + expect(rendered).not.toContain("Private prompt from previous session"); + }); + + it("keeps a recovered authoritative user ahead of its reply at full scrollback", async () => { + const deferredHistory = createDeferred(); + const chatLog = new ChatLog(20); + const state = createBaseState({ currentSessionId: "session-main" }); + const { loadHistory } = createTestSessionActions({ + client: { + listSessions: vi.fn(), + loadHistory: vi.fn(() => deferredHistory.promise), + } as unknown as TuiBackend, + chatLog, + state, + }); + + const loading = loadHistory(); + chatLog.addLiveUser("Browser prompt at the scrollback limit", { + messageId: "scrollback-user", + messageSeq: 19, + runId: "scrollback-run", + }); + deferredHistory.resolve({ + sessionId: "session-main", + sessionInfo: { key: "agent:main:main", sessionId: "session-main" }, + messages: [ + ...Array.from({ length: 18 }, (_, index) => ({ + role: "user", + content: `Earlier history ${index + 1}`, + __openclaw: { id: `history-user-${index + 1}`, seq: index + 1 }, + })), + { + role: "assistant", + content: "Reply at the scrollback limit", + __openclaw: { id: "scrollback-assistant", seq: 20 }, + }, + ], + }); + + await expect(loading).resolves.toMatchObject({ loaded: true }); + const rendered = chatLog.render(120).join("\n"); + expect(rendered.match(/Browser prompt at the scrollback limit/g)).toHaveLength(1); + expect(rendered.indexOf("Browser prompt at the scrollback limit")).toBeLessThan( + rendered.indexOf("Reply at the scrollback limit"), + ); + expect(chatLog.children).toHaveLength(20); + }); + + it("discards old live users when a same-key history response rotates the session id", async () => { + const deferredHistory = createDeferred(); + const chatLog = new ChatLog(); + const state = createBaseState({ currentSessionId: "session-before-reset" }); + const { loadHistory } = createTestSessionActions({ + client: { + listSessions: vi.fn(), + loadHistory: vi.fn(() => deferredHistory.promise), + } as unknown as TuiBackend, + chatLog, + state, + }); + + const loading = loadHistory(); + chatLog.addLiveUser("Private prompt from before reset", { + messageId: "before-reset-user", + runId: "before-reset-run", + }); + deferredHistory.resolve({ + sessionId: "session-after-reset", + sessionInfo: { key: "agent:main:main", sessionId: "session-after-reset" }, + messages: [{ role: "assistant", content: "Fresh session reply" }], + }); + + await expect(loading).resolves.toMatchObject({ loaded: true }); + const rendered = chatLog.render(120).join("\n"); + expect(rendered).toContain("Fresh session reply"); + expect(rendered).not.toContain("Private prompt from before reset"); + }); + it("accepts older session snapshots after switching session keys", async () => { const listSessions = vi.fn().mockResolvedValue({ ts: Date.now(), @@ -693,6 +948,7 @@ describe("tui session actions", () => { addUser: vi.fn(), finalizeAssistant: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue([]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), updateAssistant, startTool: vi.fn(), @@ -728,6 +984,7 @@ describe("tui session actions", () => { addUser: vi.fn(), finalizeAssistant: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue([]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), updateAssistant, startTool: vi.fn(), @@ -760,6 +1017,7 @@ describe("tui session actions", () => { addUser: vi.fn(), finalizeAssistant: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue([]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), updateAssistant, startTool: vi.fn(), @@ -1723,6 +1981,7 @@ describe("tui session actions", () => { clearAll: vi.fn(), clearPendingUsers: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue([]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), }; @@ -1736,7 +1995,11 @@ describe("tui session actions", () => { const result = await runLoadHistory(); - expect(chatLog.clearAll).toHaveBeenCalledWith({ preservePendingUsers: true }); + expect(chatLog.clearAll).toHaveBeenCalledWith({ + preservePendingUsers: true, + preserveLiveUsers: true, + }); + expect(chatLog.restoreLiveUsers).toHaveBeenCalledTimes(1); expect(chatLog.reconcilePendingUsers).toHaveBeenCalledWith([ { text: "persisted", timestamp: 2_000 }, ]); @@ -1755,6 +2018,7 @@ describe("tui session actions", () => { finalizeAssistant: vi.fn(), clearAll: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue(["run-pending"]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), }; const state = createBaseState({ @@ -1780,6 +2044,7 @@ describe("tui session actions", () => { finalizeAssistant: vi.fn(), clearAll: vi.fn(), reconcilePendingUsers: vi.fn().mockReturnValue([]), + restoreLiveUsers: vi.fn(), restorePendingUsers: vi.fn(), }; const state = createBaseState({ diff --git a/src/tui/tui-session-actions.ts b/src/tui/tui-session-actions.ts index f2315460606f..b781b2eebf4d 100644 --- a/src/tui/tui-session-actions.ts +++ b/src/tui/tui-session-actions.ts @@ -18,7 +18,10 @@ import { formatTuiErrorMessage, isCommandMessage, } from "./tui-formatters.js"; -import { readTuiSessionUserMessage } from "./tui-session-events.js"; +import { + readTuiSessionUserMessage, + readTuiTranscriptMessageSequence, +} from "./tui-session-events.js"; import { TUI_SESSION_LOOKUP_LIMIT } from "./tui-session-list-policy.js"; import * as submit from "./tui-submit-state.js"; import type { SessionInfo, TuiHistoryLoadResult, TuiOptions, TuiStateAccess } from "./tui-types.js"; @@ -484,6 +487,7 @@ export function createSessionActions(context: SessionActionContext) { // latest request may render, or a slow reload can replace a newer selection. const generation = ++historyLoadGeneration; const selection = captureSessionSelection(); + const previousSessionId = state.currentSessionId; const isCurrentLoad = () => generation === historyLoadGeneration && isCurrentSessionSelection(selection); try { @@ -544,7 +548,13 @@ export function createSessionActions(context: SessionActionContext) { } const showTools = (state.sessionInfo.verboseLevel ?? "off") !== "off"; const historyUsers: Array<{ text: string; timestamp?: number | null }> = []; - chatLog.clearAll({ preservePendingUsers: true }); + // An authoritative live prompt can arrive while history is in flight. + // Preserve it only while this response still owns the same session generation. + chatLog.clearAll({ + preservePendingUsers: true, + preserveLiveUsers: + previousSessionId === null || previousSessionId === state.currentSessionId, + }); btw.clear(); chatLog.addSystem(`session ${state.currentSessionKey}`); for (const entry of record.messages ?? []) { @@ -552,6 +562,10 @@ export function createSessionActions(context: SessionActionContext) { continue; } const message = entry as Record; + const messageSeq = readTuiTranscriptMessageSequence(message); + if (messageSeq !== undefined) { + chatLog.restoreLiveUsers(messageSeq); + } if (isCommandMessage(message)) { const text = extractTextFromMessage(message); if (text) { @@ -568,7 +582,10 @@ export function createSessionActions(context: SessionActionContext) { }); const liveUserMessage = readTuiSessionUserMessage({ message }); if (liveUserMessage) { - chatLog.addUser(text, { messageId: liveUserMessage.messageId }); + chatLog.addUser(text, { + messageId: liveUserMessage.messageId, + ...(messageSeq !== undefined ? { messageSeq } : {}), + }); } else { chatLog.addUser(text); } @@ -605,6 +622,7 @@ export function createSessionActions(context: SessionActionContext) { ); } } + chatLog.restoreLiveUsers(); submit.reconcilePendingSubmitHistory(state, chatLog.reconcilePendingUsers(historyUsers)); chatLog.restorePendingUsers(); // Restore a run still streaming for this session+agent that the gateway @@ -645,9 +663,11 @@ export function createSessionActions(context: SessionActionContext) { }; const setSession = async (rawKey: string) => { + const previousSelection = captureSessionSelection(); const nextKey = resolveSessionKey(rawKey); updateAgentFromSessionKey(nextKey); state.currentSessionKey = nextKey; + const selectionChanged = !isCurrentSessionSelection(previousSelection); state.activeChatRunId = null; submit.clearPendingSubmit(state); setActivityStatus("idle"); @@ -656,6 +676,10 @@ export function createSessionActions(context: SessionActionContext) { // so refresh data for the newly selected session isn't rejected as stale. state.sessionInfo.updatedAt = null; state.historyLoaded = false; + if (selectionChanged) { + // Live prompt identities belong to the old selection, not its pending successor. + chatLog.clearAll(); + } chatLog.clearPendingUsers(); clearLocalRunIds?.(); btw.clear(); diff --git a/src/tui/tui-session-events.test.ts b/src/tui/tui-session-events.test.ts index 0ec820bb6131..e41dbe0a173a 100644 --- a/src/tui/tui-session-events.test.ts +++ b/src/tui/tui-session-events.test.ts @@ -110,6 +110,53 @@ describe("matchesSelectedTuiSession", () => { }); describe("readTuiSessionUserMessage", () => { + it.each([ + { + name: "image block", + content: [{ type: "image", source: { type: "url", url: "/image.png" } }], + media: undefined, + expected: "Attached image", + }, + { + name: "document block", + content: [ + { + type: "attachment", + attachment: { url: "/report.pdf", kind: "document", label: "report.pdf" }, + }, + ], + media: undefined, + expected: "Attached file: report.pdf", + }, + { + name: "canonical persisted media", + content: "", + media: [{ path: "/media/inbound/image.png", contentType: "image/png" }], + expected: "Attached image", + }, + ])("accepts an authoritative attachment-only $name event", ({ content, media, expected }) => { + expect( + readTuiSessionUserMessage({ + sessionKey: "agent:main:main", + messageId: "attachment-user-1", + message: { + role: "user", + content, + __openclaw: { + id: "attachment-user-1", + idempotencyKey: "attachment-run-1:user", + seq: 1, + ...(media ? { media } : {}), + }, + }, + } satisfies SessionMessageEvent), + ).toEqual({ + messageId: "attachment-user-1", + runId: "attachment-run-1", + text: expected, + }); + }); + it("recovers the durable prompt identity and owning chat run", () => { expect( readTuiSessionUserMessage({ diff --git a/src/tui/tui-session-events.ts b/src/tui/tui-session-events.ts index a11ec387978e..f61885780cfc 100644 --- a/src/tui/tui-session-events.ts +++ b/src/tui/tui-session-events.ts @@ -9,6 +9,21 @@ type TuiSessionEvent = { agentId?: string; }; +/** Reads the monotonic transcript position shared by persisted and live messages. */ +export function readTuiTranscriptMessageSequence(message: unknown): number | undefined { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return undefined; + } + const marker = (message as Record)["__openclaw"]; + if (!marker || typeof marker !== "object" || Array.isArray(marker)) { + return undefined; + } + const sequence = (marker as Record).seq; + return typeof sequence === "number" && Number.isSafeInteger(sequence) && sequence > 0 + ? sequence + : undefined; +} + /** Reads the durable user identity without mistaking another run's prompt for this one. */ export function readTuiSessionUserMessage(event: SessionMessageEvent): { text: string; diff --git a/src/tui/tui.ts b/src/tui/tui.ts index 5f4936be12f1..9569b835f423 100644 --- a/src/tui/tui.ts +++ b/src/tui/tui.ts @@ -2,6 +2,7 @@ import { spawn } from "node:child_process"; import { existsSync } from "node:fs"; import path from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; import { fileURLToPath } from "node:url"; import { CombinedAutocompleteProvider, @@ -101,6 +102,8 @@ const DIST_ENTRY_MJS_PATH = fileURLToPath(new URL("../../dist/entry.mjs", import const OPENAI_CODEX_PROVIDER = "openai"; const CODEX_CLI_LOOKUP_TIMEOUT_MS = 5_000; +const SESSION_SUBSCRIPTION_MAX_ATTEMPTS = 5; +const SESSION_SUBSCRIPTION_RETRY_DELAY_MS = 25; type RunTuiOptions = TuiOptions & { backend?: TuiBackend; @@ -1674,7 +1677,7 @@ export async function runTui(opts: RunTuiOptions): Promise { setActivityStatus("starting up"); } void (async () => { - for (let attempt = 0; attempt < 2; attempt += 1) { + for (let attempt = 0; attempt < SESSION_SUBSCRIPTION_MAX_ATTEMPTS; attempt += 1) { try { await client.subscribeSessionEvents?.(); break; @@ -1682,12 +1685,21 @@ export async function runTui(opts: RunTuiOptions): Promise { if (!ownsConnection()) { return; } - // Subscription is idempotent; recover one transient Gateway failure - // without leaving this connected TUI permanently unsubscribed. - if (attempt === 0) { - continue; + if (attempt + 1 === SESSION_SUBSCRIPTION_MAX_ATTEMPTS) { + chatLog.addSystem(`session event subscribe failed: ${formatTuiErrorMessage(err)}`); + if (activityStatus === "starting up") { + setActivityStatus("idle"); + } + setConnectionStatus("session event subscription failed"); + tui.requestRender(); + return; + } + // A connected but unsubscribed TUI misses every peer's message. Wait + // between idempotent retries and abandon this generation on reconnect. + await delay(SESSION_SUBSCRIPTION_RETRY_DELAY_MS * (attempt + 1)); + if (!ownsConnection()) { + return; } - chatLog.addSystem(`session event subscribe failed: ${formatTuiErrorMessage(err)}`); } } if (!ownsConnection()) { diff --git a/ui/src/e2e/chat-flow.messaging.e2e.test.ts b/ui/src/e2e/chat-flow.messaging.e2e.test.ts index b0067cd05835..95bc6253b62e 100644 --- a/ui/src/e2e/chat-flow.messaging.e2e.test.ts +++ b/ui/src/e2e/chat-flow.messaging.e2e.test.ts @@ -292,104 +292,110 @@ suite.define(() => { }, ); - it("inserts a delayed persisted prompt ahead of an already-finalized reply", async () => { - const context = await suite.newBrowserContext({ - locale: "en-US", - serviceWorkers: "block", - viewport: { height: 900, width: 1280 }, - }); - const page = await context.newPage(); - const gateway = await installMockGateway(page, { historyMessages: [] }); - const runId = "shared-session-finalized-run"; - const prompt = "The persisted prompt arrived after the final."; - const finalText = "The reply was already finished."; - const userMessage = { - __openclaw: { id: "finalized-run-user", idempotencyKey: `${runId}:user`, seq: 1 }, - content: [{ text: prompt, type: "text" }], - role: "user", - timestamp: Date.now(), - }; - const assistantMessage = { - __openclaw: { id: "finalized-run-assistant", seq: 2 }, - content: [{ text: finalText, type: "text" }], - role: "assistant", - timestamp: Date.now(), - }; + it.each([ + { history: "up-to-date", includesPrompt: true }, + { history: "stale", includesPrompt: false }, + ])( + "preserves a delayed persisted prompt ahead of a finalized reply with $history history", + async ({ includesPrompt }) => { + const context = await suite.newBrowserContext({ + locale: "en-US", + serviceWorkers: "block", + viewport: { height: 900, width: 1280 }, + }); + const page = await context.newPage(); + const gateway = await installMockGateway(page, { historyMessages: [] }); + const runId = "shared-session-finalized-run"; + const prompt = "The persisted prompt arrived after the final."; + const finalText = "The reply was already finished."; + const userMessage = { + __openclaw: { id: "finalized-run-user", idempotencyKey: `${runId}:user`, seq: 1 }, + content: [{ text: prompt, type: "text" }], + role: "user", + timestamp: Date.now(), + }; + const assistantMessage = { + __openclaw: { id: "finalized-run-assistant", seq: 2 }, + content: [{ text: finalText, type: "text" }], + role: "assistant", + timestamp: Date.now(), + }; - try { - await page.goto(`${suite.server.baseUrl}chat`); - await gateway.waitForRequest("chat.startup"); - await gateway.emitGatewayEvent("chat", { - deltaText: finalText, - message: { - content: [{ text: finalText, type: "text" }], - role: "assistant", - timestamp: Date.now(), - }, - runId, - seq: 1, - sessionKey: "main", - state: "delta", - }); - await page.locator(".chat-bubble.streaming", { hasText: finalText }).waitFor({ - timeout: 10_000, - }); - await gateway.emitChatFinal({ runId, text: finalText }); - await page.locator(".chat-group.assistant .chat-text", { hasText: finalText }).waitFor({ - timeout: 10_000, - }); - await gateway.setHistoryMessages([userMessage, assistantMessage]); - await gateway.deferNext("chat.history"); - await gateway.emitGatewayEvent("session.message", { - activeRunIds: [], - clientRunId: runId, - hasActiveRun: false, - message: userMessage, - messageId: "finalized-run-user", - messageSeq: 1, - session: { + try { + await page.goto(`${suite.server.baseUrl}chat`); + await gateway.waitForRequest("chat.startup"); + await gateway.emitGatewayEvent("chat", { + deltaText: finalText, + message: { + content: [{ text: finalText, type: "text" }], + role: "assistant", + timestamp: Date.now(), + }, + runId, + seq: 1, + sessionKey: "main", + state: "delta", + }); + await page.locator(".chat-bubble.streaming", { hasText: finalText }).waitFor({ + timeout: 10_000, + }); + await gateway.emitChatFinal({ runId, text: finalText }); + await page.locator(".chat-group.assistant .chat-text", { hasText: finalText }).waitFor({ + timeout: 10_000, + }); + await gateway.setHistoryMessages([userMessage, assistantMessage]); + await gateway.deferNext("chat.history"); + await gateway.emitGatewayEvent("session.message", { activeRunIds: [], + clientRunId: runId, hasActiveRun: false, - key: "main", - kind: "direct", - status: "done", - updatedAt: Date.now(), - }, - sessionKey: "main", - }); + message: userMessage, + messageId: "finalized-run-user", + messageSeq: 1, + session: { + activeRunIds: [], + hasActiveRun: false, + key: "main", + kind: "direct", + status: "done", + updatedAt: Date.now(), + }, + sessionKey: "main", + }); - await expect - .poll(() => - page.locator(".chat-thread-inner").evaluate( - (thread, texts) => { - const user = Array.from(thread.querySelectorAll(".chat-group.user")).find((row) => - row.textContent?.includes(texts.prompt), - ); - const assistant = Array.from(thread.querySelectorAll(".chat-group.assistant")).find( - (row) => row.textContent?.includes(texts.finalText), - ); - return Boolean( - user && - assistant && - user.compareDocumentPosition(assistant) & Node.DOCUMENT_POSITION_FOLLOWING, - ); - }, - { finalText, prompt }, - ), - ) - .toBe(true); - await gateway.resolveDeferred("chat.history", { - messages: [userMessage, assistantMessage], - sessionId: "control-ui-e2e-session", - thinkingLevel: null, - }); - await expect - .poll(() => page.locator(".chat-group.user", { hasText: prompt }).count()) - .toBe(1); - } finally { - await suite.closeBrowserContext(context); - } - }); + await expect + .poll(() => + page.locator(".chat-thread-inner").evaluate( + (thread, texts) => { + const user = Array.from(thread.querySelectorAll(".chat-group.user")).find((row) => + row.textContent?.includes(texts.prompt), + ); + const assistant = Array.from(thread.querySelectorAll(".chat-group.assistant")).find( + (row) => row.textContent?.includes(texts.finalText), + ); + return Boolean( + user && + assistant && + user.compareDocumentPosition(assistant) & Node.DOCUMENT_POSITION_FOLLOWING, + ); + }, + { finalText, prompt }, + ), + ) + .toBe(true); + await gateway.resolveDeferred("chat.history", { + messages: includesPrompt ? [userMessage, assistantMessage] : [assistantMessage], + sessionId: "control-ui-e2e-session", + thinkingLevel: null, + }); + await expect + .poll(() => page.locator(".chat-group.user", { hasText: prompt }).count()) + .toBe(1); + } finally { + await suite.closeBrowserContext(context); + } + }, + ); it("keeps a browser-local prompt before a clock-skewed Gateway reply", async () => { const context = await suite.newBrowserContext({ diff --git a/ui/src/pages/chat/chat-history.ts b/ui/src/pages/chat/chat-history.ts index f12a2e58247c..4d765731c162 100644 --- a/ui/src/pages/chat/chat-history.ts +++ b/ui/src/pages/chat/chat-history.ts @@ -53,6 +53,7 @@ import { persistChatComposerState } from "./composer-persistence.ts"; import { isLocallyOptimisticHistoryMessage, messageDisplaySignature, + preserveLiveAuthoritativeUserMessages, preserveOptimisticTailMessages, readTranscriptSequence, } from "./history-merge.ts"; @@ -1186,6 +1187,7 @@ async function loadChatHistoryUncached( const previousMessages = state.chatMessages; const previousPagination = state.chatHistoryPagination; const previousSessionId = state.currentSessionId ?? null; + const previousDisplayedLeafEntryId = state.chatDisplayedLeafEntryId ?? null; const previousRunId = state.chatRunId; recordChatHistoryTiming(state, "start", startedAtMs, { requestSessionKey: sessionKey, @@ -1275,11 +1277,24 @@ async function loadChatHistoryUncached( (message) => !isPendingInitialUserMessage(state.initialUserMessage, state, sessionKey, message), ); - state.chatMessages = preserveOptimisticTailMessages( + const preservedOptimisticMessages = preserveOptimisticTailMessages( authoritativeMessages, reconciledTerminal.previousMessages, shouldHideHistoryMessage, ); + const nextDisplayedLeafEntryId = Object.hasOwn(res.sessionInfo ?? {}, "activeLeafEntryId") + ? res.sessionInfo?.activeLeafEntryId?.trim() || null + : previousDisplayedLeafEntryId; + const retainsTranscriptIdentity = + (!previousSessionId || !nextSessionId || previousSessionId === nextSessionId) && + (!previousDisplayedLeafEntryId || previousDisplayedLeafEntryId === nextDisplayedLeafEntryId); + state.chatMessages = retainsTranscriptIdentity + ? preserveLiveAuthoritativeUserMessages( + preservedOptimisticMessages, + reconciledTerminal.currentMessages, + shouldHideHistoryMessage, + ) + : preservedOptimisticMessages; if (Object.hasOwn(res.sessionInfo ?? {}, "activeLeafEntryId")) { state.chatDisplayedLeafEntryId = res.sessionInfo?.activeLeafEntryId?.trim() || null; } diff --git a/ui/src/pages/chat/chat-state-events.ts b/ui/src/pages/chat/chat-state-events.ts index f6d35df9dc29..eb1e115fec64 100644 --- a/ui/src/pages/chat/chat-state-events.ts +++ b/ui/src/pages/chat/chat-state-events.ts @@ -37,6 +37,7 @@ import type { ChatPageHost } from "./chat-state-host.ts"; import { requestChatPageUpdate } from "./chat-state-render.ts"; import { resolveChatAgentId, selectedChatSessionRow } from "./chat-state-route.ts"; import { handleBackgroundTasksEvent } from "./components/chat-background-tasks.ts"; +import { rememberLiveAuthoritativeUserMessage } from "./history-merge.ts"; import { reconcileChatRunFromCurrentSessionRow, reconcileChatRunFromSessionRow, @@ -114,7 +115,7 @@ function applyLiveUserMessage(state: ChatPageHost, payload: unknown): void { : null; const incoming: LiveUserMessageIdentity = { ...sourceIdentity, - id: eventId ?? sourceIdentity.id, + id: sourceIdentity.id ?? eventId, idempotencyKey: sourceIdentity.idempotencyKey ?? clientRunId, sequence: eventSequence ?? sourceIdentity.sequence, }; @@ -136,6 +137,7 @@ function applyLiveUserMessage(state: ChatPageHost, payload: unknown): void { ...(incoming.sequence !== null ? { seq: incoming.sequence } : {}), }, }; + rememberLiveAuthoritativeUserMessage(message); const incomingText = extractText(sourceMessage); const existingIndex = state.chatMessages.findIndex((candidate) => { const existing = readLiveUserMessageIdentity(candidate); diff --git a/ui/src/pages/chat/history-merge.test.ts b/ui/src/pages/chat/history-merge.test.ts index c28f39808e8c..c41065e9c06c 100644 --- a/ui/src/pages/chat/history-merge.test.ts +++ b/ui/src/pages/chat/history-merge.test.ts @@ -1,7 +1,11 @@ // @vitest-environment node // Control UI tests cover history merge behavior. import { describe, expect, it } from "vitest"; -import { preserveOptimisticTailMessages } from "./history-merge.ts"; +import { + preserveLiveAuthoritativeUserMessages, + preserveOptimisticTailMessages, + rememberLiveAuthoritativeUserMessage, +} from "./history-merge.ts"; function createHistoryMessage( role: "assistant" | "user", @@ -17,6 +21,75 @@ function createHistoryMessage( }; } +describe("preserveLiveAuthoritativeUserMessages", () => { + it("keeps a gateway-projected user ahead of a later stale-history reply", () => { + const liveUser = createHistoryMessage("user", "shared prompt", { + id: "shared-user", + seq: 1, + }); + const reply = createHistoryMessage("assistant", "shared reply", { + id: "shared-reply", + seq: 2, + }); + rememberLiveAuthoritativeUserMessage(liveUser); + + expect(preserveLiveAuthoritativeUserMessages([reply], [liveUser, reply])).toEqual([ + liveUser, + reply, + ]); + }); + + it("adopts the history projection without duplicating an authoritative user identity", () => { + const liveUser = createHistoryMessage("user", "live prompt", { + id: "shared-user", + seq: 1, + }); + const persistedUser = createHistoryMessage("user", "persisted prompt", { + id: "shared-user", + seq: 1, + }); + rememberLiveAuthoritativeUserMessage(liveUser); + + expect(preserveLiveAuthoritativeUserMessages([persistedUser], [liveUser])).toEqual([ + persistedUser, + ]); + }); + + it("does not revive an ordinary historical user absent from a new snapshot", () => { + const previousUser = createHistoryMessage("user", "removed prompt", { + id: "removed-user", + seq: 1, + }); + const reply = createHistoryMessage("assistant", "remaining reply", { + id: "remaining-reply", + seq: 2, + }); + + expect(preserveLiveAuthoritativeUserMessages([reply], [previousUser, reply])).toEqual([reply]); + }); + + it("uses the next authoritative row to place an id-only live prompt", () => { + const liveUser = createHistoryMessage("user", "shared prompt", { id: "shared-user" }); + const reply = createHistoryMessage("assistant", "shared reply", { id: "shared-reply" }); + rememberLiveAuthoritativeUserMessage(liveUser); + + expect(preserveLiveAuthoritativeUserMessages([reply], [liveUser, reply])).toEqual([ + liveUser, + reply, + ]); + }); + + it("does not restore a gateway message hidden from the selected transcript", () => { + const liveUser = createHistoryMessage("user", "hidden prompt", { + id: "hidden-user", + seq: 1, + }); + rememberLiveAuthoritativeUserMessage(liveUser); + + expect(preserveLiveAuthoritativeUserMessages([], [liveUser], () => true)).toEqual([]); + }); +}); + describe("preserveOptimisticTailMessages", () => { it("keeps optimistic tail messages while history is stale", () => { const persistedUser = createHistoryMessage("user", "first", { seq: 1 }); diff --git a/ui/src/pages/chat/history-merge.ts b/ui/src/pages/chat/history-merge.ts index a00ca2140334..bd8ba251534b 100644 --- a/ui/src/pages/chat/history-merge.ts +++ b/ui/src/pages/chat/history-merge.ts @@ -19,6 +19,8 @@ type IndexedHistoryMessage = { type HistoryMessageIndex = Map; +const liveAuthoritativeUserMessages = new WeakSet(); + function readTranscriptMetadata(message: unknown): Record | null { if (!message || typeof message !== "object" || Array.isArray(message)) { return null; @@ -224,6 +226,64 @@ function findTranscriptHistoryAnchor( return sameInstance ?? (entries.length === 1 ? (entries[0] ?? null) : null); } +export function rememberLiveAuthoritativeUserMessage(message: unknown): void { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return; + } + const identity = readTranscriptMessageIdentity(message); + if (identity.role === "user" && (identity.id !== null || identity.sequence !== null)) { + liveAuthoritativeUserMessages.add(message); + } +} + +export function preserveLiveAuthoritativeUserMessages( + historyMessages: unknown[], + currentMessages: unknown[], + shouldHideMessage: (message: unknown) => boolean = () => false, +): unknown[] { + let preservedMessages = historyMessages; + for (let currentIndex = 0; currentIndex < currentMessages.length; currentIndex += 1) { + const message = currentMessages[currentIndex]; + if ( + !message || + typeof message !== "object" || + !liveAuthoritativeUserMessages.has(message) || + shouldHideMessage(message) + ) { + continue; + } + const historyIndex = createHistoryMessageIndex(preservedMessages, shouldHideMessage); + if (findTranscriptHistoryAnchor(historyIndex, message)) { + continue; + } + const sequence = readTranscriptSequence(message); + let insertionIndex = + sequence === null + ? -1 + : preservedMessages.findIndex((candidate) => { + const candidateSequence = readTranscriptSequence(candidate); + return candidateSequence !== null && candidateSequence > sequence; + }); + if (insertionIndex < 0 && sequence === null) { + for (const nextMessage of currentMessages.slice(currentIndex + 1)) { + const anchor = findTranscriptHistoryAnchor(historyIndex, nextMessage); + if (anchor) { + insertionIndex = anchor.index; + break; + } + } + } + // Only a gateway-projected, identity-backed prompt may outlive a stale + // snapshot; transcript sequence or the next known row keeps its reply after it. + preservedMessages = preservedMessages.toSpliced( + insertionIndex < 0 ? preservedMessages.length : insertionIndex, + 0, + message, + ); + } + return preservedMessages; +} + function findOptimisticHistoryMatch( historyIndex: HistoryMessageIndex, identity: TranscriptMessageIdentity,