From bb21c89b191dacf627526239658b1679c37f2437 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 02:48:55 +0800 Subject: [PATCH 01/11] fix(ui): bound realtime Talk conversation text --- .../pages/chat/realtime-talk-conversation.ts | 33 +++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/ui/src/pages/chat/realtime-talk-conversation.ts b/ui/src/pages/chat/realtime-talk-conversation.ts index 0d83e80d84b8..87d4ef5f81a7 100644 --- a/ui/src/pages/chat/realtime-talk-conversation.ts +++ b/ui/src/pages/chat/realtime-talk-conversation.ts @@ -1,4 +1,6 @@ // Control UI chat module implements realtime talk conversation behavior. +import { sliceUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; + type RealtimeTalkConversationRole = "user" | "assistant"; export type RealtimeTalkConversationEntry = { @@ -25,6 +27,9 @@ type RealtimeTalkTranscriptUpdate = { }; const MAX_CONVERSATION_ENTRIES = 60; +const MAX_CONVERSATION_ENTRY_CHARS = 8_000; +const CONVERSATION_ENTRY_PREFIX_CHARS = 256; +const CONVERSATION_ENTRY_TRUNCATION_MARKER = "\nā€¦\n"; const USER_FINAL_REWRITE_GRACE_MS = 1_500; export function createRealtimeTalkConversationState(): RealtimeTalkConversationState { @@ -96,7 +101,12 @@ function upsertRealtimeConversationEntry( const id = `rt-${state.nextEntryId}`; const entries = [ ...state.entries, - { id, role, text: text.trimStart(), isStreaming: !isFinal }, + { + id, + role, + text: boundRealtimeConversationText(text.trimStart()), + isStreaming: !isFinal, + }, ].slice(-MAX_CONVERSATION_ENTRIES); return rememberRealtimeConversationEntry( { ...state, entries, nextEntryId: state.nextEntryId + 1 }, @@ -115,10 +125,11 @@ function upsertRealtimeConversationEntry( if (!entry) { return upsertRealtimeConversationEntry(state, role, null, text, isFinal, nowMs); } - const updatedText = + const mergedText = role === "assistant" ? mergeAssistantTranscriptText(entry.text, text, isFinal) : mergeRealtimeTranscriptText(entry.text, text, isFinal); + const updatedText = boundRealtimeConversationText(mergedText); const entries = entry.text === updatedText && entry.isStreaming === !isFinal ? state.entries @@ -263,6 +274,24 @@ function mergeRealtimeTranscriptText(existing: string, incoming: string, isFinal return `${existing}${separator}${suffix}`; } +function boundRealtimeConversationText(text: string): string { + if (text.length <= MAX_CONVERSATION_ENTRY_CHARS) { + return text; + } + // Keep the opening context for late full-final replacement detection and + // the newest tail for the visible conversation. Reuse the original prefix + // so repeated streaming deltas do not move the truncation boundary. + const markerIndex = text.indexOf(CONVERSATION_ENTRY_TRUNCATION_MARKER); + const prefix = + markerIndex > 0 + ? text.slice(0, markerIndex) + : sliceUtf16Safe(text, 0, CONVERSATION_ENTRY_PREFIX_CHARS); + const tailChars = + MAX_CONVERSATION_ENTRY_CHARS - prefix.length - CONVERSATION_ENTRY_TRUNCATION_MARKER.length; + const tail = sliceUtf16Safe(text, -tailChars); + return `${prefix}${CONVERSATION_ENTRY_TRUNCATION_MARKER}${tail}`; +} + function looksLikeTranscriptReplacement(existing: string, incoming: string): boolean { const existingWords = transcriptWords(existing); const incomingWords = transcriptWords(incoming); From 74858fa463069afd4ec12905690c57d5aad08c74 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 02:48:56 +0800 Subject: [PATCH 02/11] test(ui): cover bounded realtime Talk entries --- .../chat/realtime-talk-conversation.test.ts | 93 +++++++++++++++++++ 1 file changed, 93 insertions(+) diff --git a/ui/src/pages/chat/realtime-talk-conversation.test.ts b/ui/src/pages/chat/realtime-talk-conversation.test.ts index 9811a79342c7..02911429dc7a 100644 --- a/ui/src/pages/chat/realtime-talk-conversation.test.ts +++ b/ui/src/pages/chat/realtime-talk-conversation.test.ts @@ -128,6 +128,99 @@ describe("realtime Talk conversation", () => { ]); }); + it("bounds streamed assistant delta growth while retaining useful context", () => { + let state = createRealtimeTalkConversationState(); + const opening = "Opening context stays visible. "; + + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: `${opening}${"a".repeat(7_900)}`, + final: false, + nowMs: 1, + }); + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: "b".repeat(500), + final: false, + nowMs: 2, + }); + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: `${"c".repeat(500)}NEWEST`, + final: false, + nowMs: 3, + }); + + expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000); + expect(state.entries[0]?.text.startsWith(opening)).toBe(true); + expect(state.entries[0]?.text).toContain("\nā€¦\n"); + expect(state.entries[0]?.text.split("\nā€¦\n")).toHaveLength(2); + expect(state.entries[0]?.text.endsWith("NEWEST")).toBe(true); + }); + + it("replaces a bounded assistant stream with the authoritative final transcript", () => { + let state = createRealtimeTalkConversationState(); + const opening = "Original opening context. "; + + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: `${opening}${"draft ".repeat(1_600)}`, + final: false, + nowMs: 1, + }); + expect(state.entries[0]?.text).toContain("\nā€¦\n"); + + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: `${opening}corrected ${"final ".repeat(1_600)}DONE`, + final: true, + nowMs: 2, + }); + + expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000); + expect(state.entries[0]?.text.startsWith(`${opening}corrected `)).toBe(true); + expect(state.entries[0]?.text).not.toContain("draft "); + expect(state.entries[0]?.text.endsWith("DONE")).toBe(true); + expect(state.entries[0]?.isStreaming).toBe(false); + }); + + it("does not expose dangling surrogates at a bounded transcript edge", () => { + let state = createRealtimeTalkConversationState(); + const transcript = `${"a".repeat(8_000)}šŸš€${"b".repeat(7_740)}`; + + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: transcript, + final: true, + nowMs: 1, + }); + + const text = state.entries[0]?.text ?? ""; + expect(text.length).toBeLessThanOrEqual(8_000); + expect(text).not.toMatch( + /(?:[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(? { + let state = createRealtimeTalkConversationState(); + + state = updateRealtimeTalkConversation(state, { + role, + text: `Useful opening. ${"x".repeat(9_000)}NEWEST`, + final: true, + nowMs: 1, + }); + + expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000); + expect(state.entries[0]?.text.startsWith("Useful opening. ")).toBe(true); + expect(state.entries[0]?.text.endsWith("NEWEST")).toBe(true); + expect(state.entries[0]?.isStreaming).toBe(false); + }, + ); + it("keeps alternating realtime turns as separate bubbles", () => { let state = createRealtimeTalkConversationState(); From a7c52ef6c1806bedc0c9f4f1c482c4a2ac164c1b Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 03:02:21 +0800 Subject: [PATCH 03/11] fix(ui): harden Talk transcript marker bounds --- .../pages/chat/realtime-talk-conversation.test.ts | 15 +++++++++++++++ ui/src/pages/chat/realtime-talk-conversation.ts | 10 ++++++---- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/ui/src/pages/chat/realtime-talk-conversation.test.ts b/ui/src/pages/chat/realtime-talk-conversation.test.ts index 02911429dc7a..08a44cae000c 100644 --- a/ui/src/pages/chat/realtime-talk-conversation.test.ts +++ b/ui/src/pages/chat/realtime-talk-conversation.test.ts @@ -202,6 +202,21 @@ describe("realtime Talk conversation", () => { ); }); + it("does not trust a natural truncation marker outside the bounded prefix", () => { + let state = createRealtimeTalkConversationState(); + + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: `${"a".repeat(7_998)}\nā€¦\n${"b".repeat(500)}NEWEST`, + final: true, + nowMs: 1, + }); + + expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000); + expect(state.entries[0]?.text.startsWith("a".repeat(256))).toBe(true); + expect(state.entries[0]?.text.endsWith("NEWEST")).toBe(true); + }); + it.each(["user", "assistant"] as const)( "bounds oversized final %s entries while retaining the newest text", (role) => { diff --git a/ui/src/pages/chat/realtime-talk-conversation.ts b/ui/src/pages/chat/realtime-talk-conversation.ts index 87d4ef5f81a7..04fbd493b041 100644 --- a/ui/src/pages/chat/realtime-talk-conversation.ts +++ b/ui/src/pages/chat/realtime-talk-conversation.ts @@ -282,10 +282,12 @@ function boundRealtimeConversationText(text: string): string { // the newest tail for the visible conversation. Reuse the original prefix // so repeated streaming deltas do not move the truncation boundary. const markerIndex = text.indexOf(CONVERSATION_ENTRY_TRUNCATION_MARKER); - const prefix = - markerIndex > 0 - ? text.slice(0, markerIndex) - : sliceUtf16Safe(text, 0, CONVERSATION_ENTRY_PREFIX_CHARS); + const hasBoundedPrefix = + markerIndex >= CONVERSATION_ENTRY_PREFIX_CHARS - 1 && + markerIndex <= CONVERSATION_ENTRY_PREFIX_CHARS; + const prefix = hasBoundedPrefix + ? text.slice(0, markerIndex) + : sliceUtf16Safe(text, 0, CONVERSATION_ENTRY_PREFIX_CHARS); const tailChars = MAX_CONVERSATION_ENTRY_CHARS - prefix.length - CONVERSATION_ENTRY_TRUNCATION_MARKER.length; const tail = sliceUtf16Safe(text, -tailChars); From 07822aaefdb249c2678bea03cca6c8db4863a650 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 03:19:48 +0800 Subject: [PATCH 04/11] fix(ui): preserve Talk transcript surrogate bounds --- .../chat/realtime-talk-conversation.test.ts | 23 +++++++++++++++++++ .../pages/chat/realtime-talk-conversation.ts | 7 +++--- 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/ui/src/pages/chat/realtime-talk-conversation.test.ts b/ui/src/pages/chat/realtime-talk-conversation.test.ts index 08a44cae000c..49cb5e1ab315 100644 --- a/ui/src/pages/chat/realtime-talk-conversation.test.ts +++ b/ui/src/pages/chat/realtime-talk-conversation.test.ts @@ -217,6 +217,29 @@ describe("realtime Talk conversation", () => { expect(state.entries[0]?.text.endsWith("NEWEST")).toBe(true); }); + it.each([255, 256])( + "does not retain a lone high surrogate before a natural marker at offset %i", + (markerOffset) => { + let state = createRealtimeTalkConversationState(); + const retainedText = "a".repeat(markerOffset - 1); + + state = updateRealtimeTalkConversation(state, { + role: "assistant", + text: `${retainedText}\uD800\nā€¦\n${"b".repeat(8_000)}NEWEST`, + final: true, + nowMs: 1, + }); + + const text = state.entries[0]?.text ?? ""; + expect(text.length).toBeLessThanOrEqual(8_000); + expect(text.startsWith(`${retainedText}\nā€¦\n`)).toBe(true); + expect(text.endsWith("NEWEST")).toBe(true); + expect(text).not.toMatch( + /(?:[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(? { diff --git a/ui/src/pages/chat/realtime-talk-conversation.ts b/ui/src/pages/chat/realtime-talk-conversation.ts index 04fbd493b041..74c275db500c 100644 --- a/ui/src/pages/chat/realtime-talk-conversation.ts +++ b/ui/src/pages/chat/realtime-talk-conversation.ts @@ -285,9 +285,10 @@ function boundRealtimeConversationText(text: string): string { const hasBoundedPrefix = markerIndex >= CONVERSATION_ENTRY_PREFIX_CHARS - 1 && markerIndex <= CONVERSATION_ENTRY_PREFIX_CHARS; - const prefix = hasBoundedPrefix - ? text.slice(0, markerIndex) - : sliceUtf16Safe(text, 0, CONVERSATION_ENTRY_PREFIX_CHARS); + const prefixEnd = hasBoundedPrefix ? markerIndex : CONVERSATION_ENTRY_PREFIX_CHARS; + // A natural marker can follow malformed provider text ending in a lone high + // surrogate. Keep that code unit out of the retained truncation boundary. + const prefix = sliceUtf16Safe(text, 0, prefixEnd).replace(/[\uD800-\uDBFF]$/, ""); const tailChars = MAX_CONVERSATION_ENTRY_CHARS - prefix.length - CONVERSATION_ENTRY_TRUNCATION_MARKER.length; const tail = sliceUtf16Safe(text, -tailChars); From 165c968f951c8fc3a863409bf14f91256aee5321 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 1 Aug 2026 12:39:24 -0700 Subject: [PATCH 05/11] fix(memory): recover restored session freshness (#117548) --- .../memory/manager-session-sync-state.test.ts | 8 + .../src/memory/manager-session-sync-state.ts | 4 +- .../src/memory/manager-source-sync-ops.ts | 27 +- .../manager-sync-ops.startup-catchup.test.ts | 265 +++++++++++++++++- 4 files changed, 289 insertions(+), 15 deletions(-) diff --git a/extensions/memory-core/src/memory/manager-session-sync-state.test.ts b/extensions/memory-core/src/memory/manager-session-sync-state.test.ts index 80d674014a42..fd0846619efa 100644 --- a/extensions/memory-core/src/memory/manager-session-sync-state.test.ts +++ b/extensions/memory-core/src/memory/manager-session-sync-state.test.ts @@ -106,6 +106,12 @@ describe("memory session sync state", () => { mtimeMs: 250, size: 20, }, + { + absPath: "/tmp/sessions/rolled-back.jsonl", + path: "sessions/rolled-back.jsonl", + mtimeMs: 150, + size: 20, + }, { absPath: "/tmp/sessions/resized.jsonl", path: "sessions/resized.jsonl", @@ -124,6 +130,7 @@ describe("memory session sync state", () => { { path: "sessions/sub-ms-newer.jsonl", hash: "hash-sub-ms", mtime: 100.25, size: 10 }, { path: "sessions/invalidated.jsonl", hash: "", mtime: 200, size: 20 }, { path: "sessions/newer.jsonl", hash: "hash-newer", mtime: 200, size: 20 }, + { path: "sessions/rolled-back.jsonl", hash: "hash-rolled-back", mtime: 200, size: 20 }, { path: "sessions/resized.jsonl", hash: "hash-resized", mtime: 300, size: 30 }, ], }); @@ -132,6 +139,7 @@ describe("memory session sync state", () => { "/tmp/sessions/sub-ms-newer.jsonl", "/tmp/sessions/invalidated.jsonl", "/tmp/sessions/newer.jsonl", + "/tmp/sessions/rolled-back.jsonl", "/tmp/sessions/resized.jsonl", "/tmp/sessions/missing.jsonl", ]); diff --git a/extensions/memory-core/src/memory/manager-session-sync-state.ts b/extensions/memory-core/src/memory/manager-session-sync-state.ts index 7857762185c6..7be696a96892 100644 --- a/extensions/memory-core/src/memory/manager-session-sync-state.ts +++ b/extensions/memory-core/src/memory/manager-session-sync-state.ts @@ -26,7 +26,9 @@ export function resolveMemorySessionStartupDirtyFiles(params: { dirtyFiles.push(file.absPath); continue; } - if (file.size !== indexedSize || file.mtimeMs > indexedMtimeMs) { + // File mtimes and SQLite session updatedAt values can move backward after + // restore/reset. The downstream content-hash gate suppresses unchanged rewrites. + if (file.size !== indexedSize || file.mtimeMs !== indexedMtimeMs) { dirtyFiles.push(file.absPath); } } diff --git a/extensions/memory-core/src/memory/manager-source-sync-ops.ts b/extensions/memory-core/src/memory/manager-source-sync-ops.ts index 2d9bcad63a27..0f3b992a97fc 100644 --- a/extensions/memory-core/src/memory/manager-source-sync-ops.ts +++ b/extensions/memory-core/src/memory/manager-source-sync-ops.ts @@ -174,6 +174,29 @@ export abstract class MemoryManagerSourceSyncOps extends MemoryManagerSessionSyn const deleteChunksByPathAndSource = this.db.prepare( `DELETE FROM memory_index_chunks WHERE path = ? AND source = ?`, ); + const updateUnchangedSessionSourceMetadata = this.db.prepare( + `UPDATE memory_index_sources + SET mtime = ?, size = ? + WHERE path = ? AND source = 'sessions' AND hash = ?`, + ); + const refreshUnchangedSessionSourceMetadata = (entry: MemoryIndexEntry): boolean => { + // Hash equality preserves chunks and embeddings; only converge the source + // fingerprint so restored sessions do not repeat catch-up on every startup. + return ( + updateUnchangedSessionSourceMetadata.run(entry.mtimeMs, entry.size, entry.path, entry.hash) + .changes === 1 + ); + }; + const canSkipUnchangedSessionEntry = ( + entry: MemoryIndexEntry, + absPath: string, + existingHash: string | undefined, + ): boolean => { + if (params.needsFullReindex || existingHash !== entry.hash) { + return false; + } + return !this.sessionsDirtyFiles.has(absPath) || refreshUnchangedSessionSourceMetadata(entry); + }; const deleteFtsRowsByPathAndSource = this.fts.enabled && this.fts.available ? this.db.prepare(`DELETE FROM ${FTS_TABLE} WHERE path = ? AND source = ?`) @@ -340,7 +363,7 @@ export abstract class MemoryManagerSourceSyncOps extends MemoryManagerSessionSyn path: entry.path, existingHashes, }); - if (!params.needsFullReindex && existingHash === entry.hash) { + if (canSkipUnchangedSessionEntry(entry, absPath, existingHash)) { if (params.progress) { params.progress.completed += 1; params.progress.report({ @@ -412,7 +435,7 @@ export abstract class MemoryManagerSourceSyncOps extends MemoryManagerSessionSyn path: entry.path, existingHashes, }); - if (!params.needsFullReindex && existingHash === entry.hash) { + if (canSkipUnchangedSessionEntry(entry, absPath, existingHash)) { if (params.progress) { params.progress.completed += 1; params.progress.report({ diff --git a/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts b/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts index a1324deae121..968c2d8891d5 100644 --- a/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts +++ b/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts @@ -2,13 +2,16 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import type { DatabaseSync } from "node:sqlite"; +import { DatabaseSync } from "node:sqlite"; import { resolveSessionTranscriptsDirForAgent, type OpenClawConfig, type ResolvedMemorySearchConfig, } from "openclaw/plugin-sdk/memory-core-host-engine-foundation"; -import { statSessionEntrySync } from "openclaw/plugin-sdk/memory-core-host-engine-qmd"; +import { + buildSessionEntry, + statSessionEntrySync, +} from "openclaw/plugin-sdk/memory-core-host-engine-qmd"; import { MEMORY_CHUNKING_VERSION, type MemorySource, @@ -64,8 +67,43 @@ type MemorySessionTranscriptUpdate = { const originalStartupStateDir = process.env.OPENCLAW_STATE_DIR; const originalStartupConfigPath = process.env.OPENCLAW_CONFIG_PATH; let transcriptUpdateListener: ((update: MemorySessionTranscriptUpdate) => void) | undefined; +const startupHarnessDatabases = new Set(); type SourceStateRow = { path: string; hash: string; mtime: number; size: number }; + +function createStartupHarnessDatabase(sourceRows: SourceStateRow[]): DatabaseSync { + const db = new DatabaseSync(":memory:"); + db.exec(` + CREATE TABLE memory_index_sources ( + path TEXT NOT NULL, + source TEXT NOT NULL, + hash TEXT NOT NULL, + mtime REAL NOT NULL, + size INTEGER NOT NULL, + UNIQUE(path, source) + ); + CREATE TABLE memory_index_chunks ( + id TEXT PRIMARY KEY, + path TEXT NOT NULL, + source TEXT NOT NULL, + model TEXT NOT NULL + ); + CREATE TABLE memory_index_source_update_audit (path TEXT NOT NULL); + CREATE TRIGGER memory_index_source_update_audit_trigger + AFTER UPDATE ON memory_index_sources + BEGIN + INSERT INTO memory_index_source_update_audit (path) VALUES (NEW.path); + END; + `); + const insert = db.prepare( + `INSERT INTO memory_index_sources (path, source, hash, mtime, size) VALUES (?, 'sessions', ?, ?, ?)`, + ); + for (const row of sourceRows) { + insert.run(row.path, row.hash, row.mtime, row.size); + } + startupHarnessDatabases.add(db); + return db; +} function setStartupStateDir(stateDir: string): void { Reflect.set(process.env, "OPENCLAW_STATE_DIR", stateDir); } @@ -148,16 +186,37 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps { sourceRows: SourceStateRow[], private readonly indexSessionUpdates = false, private readonly subscribeToRealEvents = false, + private readonly deferSessionIndex = false, + database?: DatabaseSync, ) { super(); this.sources.add("sessions"); - this.db = { - prepare: () => ({ - all: () => sourceRows, - get: () => undefined, - run: () => undefined, - }), - } as unknown as DatabaseSync; + this.db = database ?? createStartupHarnessDatabase(sourceRows); + } + + restartForStartup(): SessionStartupCatchupHarness { + return new SessionStartupCatchupHarness( + [], + this.indexSessionUpdates, + false, + this.deferSessionIndex, + this.db, + ); + } + + getIndexedSourceState(pathname: string): SourceStateRow | undefined { + return this.db + .prepare( + `SELECT path, hash, mtime, size FROM memory_index_sources WHERE path = ? AND source = 'sessions'`, + ) + .get(pathname) as SourceStateRow | undefined; + } + + getSourceMetadataUpdateCount(): number { + const row = this.db + .prepare(`SELECT COUNT(*) AS count FROM memory_index_source_update_audit`) + .get() as { count: number }; + return row.count; } async catchUp(): Promise { @@ -172,6 +231,13 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps { await this.runSync(params); } + async runArchiveSyncForTest(): Promise { + await this.syncArchiveFiles({ + needsFullReindex: false, + deferIndex: this.deferSessionIndex, + }); + } + getDirtyArchiveFiles(): string[] { return Array.from(this.sessionsDirtyFiles); } @@ -273,7 +339,10 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps { protected async sync(params?: MemorySyncParams): Promise { this.syncCalls.push(params ?? {}); this.pendingSyncWork = this.indexSessionUpdates - ? this.syncArchiveFiles({ needsFullReindex: false }).then(() => undefined) + ? this.syncArchiveFiles({ + needsFullReindex: false, + deferIndex: this.deferSessionIndex, + }).then(() => undefined) : Promise.resolve(); await this.pendingSyncWork; } @@ -333,20 +402,29 @@ describe("session startup catch-up", () => { restoreStartupEnv(); clearRuntimeConfigSnapshot(); clearConfigCache(); + for (const database of startupHarnessDatabases) { + database.close(); + } + startupHarnessDatabases.clear(); closeOpenClawAgentDatabasesForTest(); await fs.rm(stateDir, { recursive: true, force: true }); }); async function writeSessionFile( name: string, + content = "startup catchup", + timestamp?: string, ): Promise<{ filePath: string; size: number; mtimeMs: number }> { const sessionsDir = resolveSessionTranscriptsDirForAgent("main"); await fs.mkdir(sessionsDir, { recursive: true }); const filePath = path.join(sessionsDir, name); await fs.writeFile( filePath, - JSON.stringify({ type: "message", message: { role: "user", content: "startup catchup" } }) + - "\n", + JSON.stringify({ + type: "message", + ...(timestamp ? { timestamp } : {}), + message: { role: "user", content }, + }) + "\n", "utf-8", ); const stat = await fs.stat(filePath); @@ -533,6 +611,169 @@ describe("session startup catch-up", () => { expect(harness.syncCalls).toEqual([]); }); + it("indexes a same-size file transcript whose mtime rolled back", async () => { + const archiveName = "thread.jsonl.deleted.2026-08-01T10-00-00.000Z"; + const original = await writeSessionFile(archiveName, "version before"); + const originalEntry = await buildSessionEntry(original.filePath); + if (!originalEntry) { + throw new Error("expected original file transcript entry"); + } + const replacement = await writeSessionFile(archiveName, "version after!"); + expect(replacement.size).toBe(original.size); + const rolledBackMtime = new Date(Math.max(1, original.mtimeMs - 60_000)); + await fs.utimes(replacement.filePath, rolledBackMtime, rolledBackMtime); + const rolledBack = await fs.stat(replacement.filePath); + expect(rolledBack.mtimeMs).toBeLessThan(original.mtimeMs); + + const harness = new SessionStartupCatchupHarness( + [ + { + path: originalEntry.path, + hash: originalEntry.hash, + mtime: original.mtimeMs, + size: original.size, + }, + ], + true, + ); + + await expect(harness.catchUp()).resolves.toEqual([replacement.filePath]); + await harness.waitForSessionSync(); + + expect(harness.syncCalls).toEqual([{ reason: "session-startup-catchup" }]); + expect(harness.indexedPaths).toEqual([`sessions/main/${archiveName}`]); + expect(harness.indexedContents).toEqual(["User: version after!"]); + }); + + it("converges an unchanged file mtime rollback after direct session sync", async () => { + const archiveName = "thread.jsonl.deleted.2026-08-01T11-00-00.000Z"; + const messageTimestamp = "2026-08-01T10:30:00.000Z"; + const original = await writeSessionFile(archiveName, "unchanged content", messageTimestamp); + const originalEntry = await buildSessionEntry(original.filePath); + if (!originalEntry) { + throw new Error("expected original file transcript entry"); + } + const rolledBackMtime = new Date(Math.max(1, original.mtimeMs - 60_000)); + await fs.utimes(original.filePath, rolledBackMtime, rolledBackMtime); + const restoredEntry = await buildSessionEntry(original.filePath); + if (!restoredEntry) { + throw new Error("expected restored file transcript entry"); + } + expect(restoredEntry.hash).toBe(originalEntry.hash); + + const harness = new SessionStartupCatchupHarness( + [ + { + path: originalEntry.path, + hash: originalEntry.hash, + mtime: original.mtimeMs, + size: original.size, + }, + ], + true, + ); + + await expect(harness.catchUp()).resolves.toEqual([original.filePath]); + await harness.waitForSessionSync(); + expect(harness.indexedPaths).toEqual([]); + expect(harness.getIndexedSourceState(originalEntry.path)).toEqual({ + path: originalEntry.path, + hash: originalEntry.hash, + mtime: restoredEntry.mtimeMs, + size: restoredEntry.size, + }); + expect(harness.getSourceMetadataUpdateCount()).toBe(1); + + const restarted = harness.restartForStartup(); + await expect(restarted.catchUp()).resolves.toEqual([]); + expect(restarted.syncCalls).toEqual([]); + expect(restarted.indexedPaths).toEqual([]); + }); + + it("indexes a SQLite transcript whose updatedAt rolled back", async () => { + const session = await writeSqliteSession({ + content: "SQLite rollback", + updatedAt: 10, + }); + const state = statSessionEntrySync(session.sessionKey, { + agentId: "main", + sessionId: session.sessionId, + storePath: session.storePath, + sessionKey: session.sessionKey, + updatedAtMs: 10, + }); + if (!state) { + throw new Error("expected SQLite transcript state"); + } + const harness = new SessionStartupCatchupHarness( + [ + { + path: state.path, + hash: "previous-hash", + mtime: 20, + size: state.size, + }, + ], + true, + ); + + await expect(harness.catchUp()).resolves.toEqual([session.sessionKey]); + await harness.waitForSessionSync(); + + expect(harness.syncCalls).toEqual([{ reason: "session-startup-catchup" }]); + expect(harness.indexedPaths).toEqual([session.corpusPath]); + expect(harness.indexedContents).toEqual(["User: SQLite rollback"]); + }); + + it("converges an unchanged SQLite updatedAt rollback after deferred session sync", async () => { + const session = await writeSqliteSession({ updatedAt: 10 }); + const entry = await buildSessionEntry(session.sessionKey, { + agentId: "main", + sessionId: session.sessionId, + storePath: session.storePath, + sessionKey: session.sessionKey, + updatedAtMs: 10, + sessionKind: "interactive", + }); + if (!entry) { + throw new Error("expected SQLite transcript entry"); + } + const harness = new SessionStartupCatchupHarness( + [ + { + path: entry.path, + hash: entry.hash, + mtime: 20, + size: entry.size, + }, + ], + true, + false, + true, + ); + + await expect(harness.catchUp()).resolves.toEqual([session.sessionKey]); + await harness.waitForSessionSync(); + + expect(harness.syncCalls).toEqual([{ reason: "session-startup-catchup" }]); + expect(harness.indexedPaths).toEqual([]); + expect(harness.indexedContents).toEqual([]); + expect(harness.getIndexedSourceState(entry.path)).toEqual({ + path: entry.path, + hash: entry.hash, + mtime: entry.mtimeMs, + size: entry.size, + }); + expect(harness.getSourceMetadataUpdateCount()).toBe(1); + + const restarted = harness.restartForStartup(); + await expect(restarted.catchUp()).resolves.toEqual([]); + expect(restarted.syncCalls).toEqual([]); + expect(restarted.indexedPaths).toEqual([]); + await restarted.runArchiveSyncForTest(); + expect(restarted.getSourceMetadataUpdateCount()).toBe(1); + }); + it("does not fall back to full session sync when identity targets normalize away", async () => { await writeSessionFile("thread.jsonl"); const harness = new SessionStartupCatchupHarness([]); From 26aa884c6914ab8ba65d743bbc09e0410b47b54e Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 1 Aug 2026 12:41:34 -0700 Subject: [PATCH 06/11] refactor(ui): consolidate cron form control rendering (#117576) --- ui/src/pages/cron/view.test.ts | 31 +- ui/src/pages/cron/view.ts | 1047 ++++++++++++-------------------- 2 files changed, 398 insertions(+), 680 deletions(-) diff --git a/ui/src/pages/cron/view.test.ts b/ui/src/pages/cron/view.test.ts index ed361c19095b..06eb8edd6ac8 100644 --- a/ui/src/pages/cron/view.test.ts +++ b/ui/src/pages/cron/view.test.ts @@ -602,19 +602,38 @@ describe("cron view editor", () => { expect(onClosePanel).toHaveBeenCalledTimes(1); }); - it("wires form changes from prompt and name inputs", () => { + it("wires shared text and select controls without changing their field ownership", () => { const onFormChange = vi.fn(); - const container = renderView({ createOpen: true, onFormChange }); + const container = renderView({ + createOpen: true, + channels: ["telegram"], + channelMeta: [{ id: "telegram", label: "", detailLabel: "Telegram" }], + channelLabels: { telegram: "Telegram fallback" }, + form: { ...DEFAULT_CRON_FORM, scheduleKind: "cron", failureAlertMode: "custom" }, + onFormChange, + }); const prompt = getElement(container, "#cron-payload-text", HTMLTextAreaElement); prompt.value = "do the thing"; prompt.dispatchEvent(new Event("input", { bubbles: true })); expect(onFormChange).toHaveBeenCalledWith({ payloadText: "do the thing" }); - const name = getElement(container, "#cron-name", HTMLInputElement); - name.value = "Thing"; - name.dispatchEvent(new Event("input", { bubbles: true })); - expect(onFormChange).toHaveBeenCalledWith({ name: "Thing" }); + for (const field of ["name", "sessionKey", "deliveryAccountId", "payloadModel"] as const) { + const id = `cron-${field.replace(/[A-Z]/g, (letter) => `-${letter.toLowerCase()}`)}`; + const input = getElement(container, `#${id}`, HTMLInputElement); + if (field === "sessionKey" || field === "deliveryAccountId") { + expect(input.placeholder).toBe(field === "sessionKey" ? "agent:main:main" : "default"); + } + input.value = field; + input.dispatchEvent(new Event("input", { bubbles: true })); + expect(onFormChange).toHaveBeenLastCalledWith({ [field]: field }); + } + + const channel = getElement(container, "#cron-failure-alert-channel", HTMLSelectElement); + channel.value = "telegram"; + expect(channel.selectedOptions[0]?.textContent).toBe("Telegram fallback"); + channel.dispatchEvent(new Event("change", { bubbles: true })); + expect(onFormChange).toHaveBeenLastCalledWith({ failureAlertChannel: "telegram" }); }); it("switches schedule inputs by segmented kind and wires kind changes", () => { diff --git a/ui/src/pages/cron/view.ts b/ui/src/pages/cron/view.ts index cd32fb7ffbfb..d61454156a3a 100644 --- a/ui/src/pages/cron/view.ts +++ b/ui/src/pages/cron/view.ts @@ -133,40 +133,24 @@ type CronProps = { // ā”€ā”€ Shared option helpers ā”€ā”€ function buildChannelOptions(props: CronProps): string[] { - const options = ["last", ...props.channels.filter(Boolean)]; const current = props.form.deliveryChannel?.trim(); - if (current && !options.includes(current)) { - options.push(current); - } - const seen = new Set(); - return options.filter((value) => { - if (seen.has(value)) { - return false; - } - seen.add(value); - return true; - }); + return uniqueStrings(["last", ...props.channels.filter(Boolean), ...(current ? [current] : [])]); } function resolveChannelLabel(props: CronProps, channel: string): string { - if (channel === "last") { - return "last"; - } - const meta = props.channelMeta?.find((entry) => entry.id === channel); - if (meta?.label) { - return meta.label; - } - return props.channelLabels?.[channel] ?? channel; + return channel === "last" + ? channel + : props.channelMeta?.find((entry) => entry.id === channel)?.label || + (props.channelLabels?.[channel] ?? channel); } function renderSuggestionList(id: string, options: string[]) { const clean = uniqueStrings(normalizeStringEntries(options)); - if (clean.length === 0) { - return nothing; - } - return html` - ${clean.map((value) => html` `)} - `; + return clean.length === 0 + ? nothing + : html` + ${clean.map((value) => html` `)} + `; } // ā”€ā”€ Validation summary helpers ā”€ā”€ @@ -178,45 +162,27 @@ type BlockingField = { inputId: string; }; +const CRON_FIELD_LABEL_KEYS: Record = { + name: "cron.form.fieldName", + scheduleAt: "cron.form.runAt", + everyAmount: "cron.form.every", + cronExpr: "cron.form.expression", + staggerAmount: "cron.form.staggerWindow", + payloadText: "cron.form.assistantTaskPrompt", + payloadModel: "cron.form.model", + payloadThinking: "cron.form.thinking", + timeoutSeconds: "cron.form.timeoutSeconds", + deliveryTo: "cron.form.to", + failureAlertAfter: "cron.form.failureAlertAfter", + failureAlertCooldownSeconds: "cron.form.failureAlertCooldown", +}; + function errorIdForField(key: CronFieldKey) { return `cron-error-${key}`; } -function inputIdForField(key: CronFieldKey) { - if (key === "name") { - return "cron-name"; - } - if (key === "scheduleAt") { - return "cron-schedule-at"; - } - if (key === "everyAmount") { - return "cron-every-amount"; - } - if (key === "cronExpr") { - return "cron-cron-expr"; - } - if (key === "staggerAmount") { - return "cron-stagger-amount"; - } - if (key === "payloadText") { - return "cron-payload-text"; - } - if (key === "payloadModel") { - return "cron-payload-model"; - } - if (key === "payloadThinking") { - return "cron-payload-thinking"; - } - if (key === "timeoutSeconds") { - return "cron-timeout-seconds"; - } - if (key === "failureAlertAfter") { - return "cron-failure-alert-after"; - } - if (key === "failureAlertCooldownSeconds") { - return "cron-failure-alert-cooldown-seconds"; - } - return "cron-delivery-to"; +function inputIdForField(key: string) { + return `cron-${key.replace(/[A-Z]/g, (letter) => `-${letter.toLowerCase()}`)}`; } function fieldLabelForKey( @@ -224,29 +190,13 @@ function fieldLabelForKey( form: CronFormState, deliveryMode: CronFormState["deliveryMode"], ) { - if (key === "payloadText") { - return form.payloadKind === "systemEvent" - ? t("cron.form.mainTimelineMessage") - : t("cron.form.assistantTaskPrompt"); + if (key === "payloadText" && form.payloadKind === "systemEvent") { + return t("cron.form.mainTimelineMessage"); } - if (key === "deliveryTo") { - return deliveryMode === "webhook" ? t("cron.form.webhookUrl") : t("cron.form.to"); + if (key === "deliveryTo" && deliveryMode === "webhook") { + return t("cron.form.webhookUrl"); } - const labels: Record = { - name: t("cron.form.fieldName"), - scheduleAt: t("cron.form.runAt"), - everyAmount: t("cron.form.every"), - cronExpr: t("cron.form.expression"), - staggerAmount: t("cron.form.staggerWindow"), - payloadText: t("cron.form.assistantTaskPrompt"), - payloadModel: t("cron.form.model"), - payloadThinking: t("cron.form.thinking"), - timeoutSeconds: t("cron.form.timeoutSeconds"), - deliveryTo: t("cron.form.to"), - failureAlertAfter: t("cron.form.failureAlertAfter"), - failureAlertCooldownSeconds: t("cron.form.failureAlertCooldown"), - }; - return labels[key]; + return t(CRON_FIELD_LABEL_KEYS[key]); } function collectBlockingFields( @@ -254,34 +204,19 @@ function collectBlockingFields( form: CronFormState, deliveryMode: CronFormState["deliveryMode"], ): BlockingField[] { - const orderedKeys: CronFieldKey[] = [ - "name", - "scheduleAt", - "everyAmount", - "cronExpr", - "staggerAmount", - "payloadText", - "payloadModel", - "payloadThinking", - "timeoutSeconds", - "deliveryTo", - "failureAlertAfter", - "failureAlertCooldownSeconds", - ]; - const fields: BlockingField[] = []; - for (const key of orderedKeys) { + return (Object.keys(CRON_FIELD_LABEL_KEYS) as CronFieldKey[]).flatMap((key) => { const message = errors[key]; - if (!message) { - continue; - } - fields.push({ - key, - label: fieldLabelForKey(key, form, deliveryMode), - message, - inputId: inputIdForField(key), - }); - } - return fields; + return message + ? [ + { + key, + label: fieldLabelForKey(key, form, deliveryMode), + message, + inputId: inputIdForField(key), + }, + ] + : []; + }); } function focusFormField(id: string) { @@ -347,19 +282,122 @@ function renderFieldRow(params: { `; } -function renderToggleRow(params: { +type CronStringFormField = { + [Field in keyof CronFormState]: CronFormState[Field] extends string ? Field : never; +}[keyof CronFormState]; + +type CronBooleanFormField = { + [Field in keyof CronFormState]: CronFormState[Field] extends boolean ? Field : never; +}[keyof CronFormState]; + +type CronInputOptions = { label: string; - checked: boolean; help?: string; + placeholder?: string; + list?: string; + type?: string; + required?: boolean; disabled?: boolean; - onChange: (checked: boolean) => void; -}) { + mono?: boolean; + errorKey?: CronFieldKey; + describeError?: boolean; +}; + +function renderCronInput(props: CronProps, field: CronStringFormField, options: CronInputOptions) { + const error = options.errorKey ? props.fieldErrors[options.errorKey] : undefined; + const describedBy = + error && options.errorKey && options.describeError !== false + ? errorIdForField(options.errorKey) + : undefined; + return html` + + props.onFormChange({ [field]: (event.currentTarget as HTMLInputElement).value })} + /> + `; +} + +function renderCronInputField( + props: CronProps, + field: CronStringFormField, + options: CronInputOptions, +) { + const errorKey = options.errorKey; + return renderFieldRow({ + label: options.label, + controlId: inputIdForField(field), + required: options.required, + help: options.help, + error: errorKey ? props.fieldErrors[errorKey] : undefined, + errorId: errorKey ? errorIdForField(errorKey) : undefined, + control: renderCronInput(props, field, options), + }); +} + +type CronSelectOption = { value: string; label: string }; + +type CronSelectOptions = { + label: string; + options: readonly CronSelectOption[]; + help?: string; + value?: string; + disabled?: boolean; + standalone?: boolean; +}; + +function renderCronSelect( + props: CronProps, + field: CronStringFormField, + options: CronSelectOptions, +) { + return html` + + `; +} + +function renderCronSelectField( + props: CronProps, + field: CronStringFormField, + options: CronSelectOptions, +) { + return renderFieldRow({ + label: options.label, + controlId: inputIdForField(field), + help: options.help, + control: renderCronSelect(props, field, options), + }); +} + +function renderToggleRow( + props: CronProps, + field: CronBooleanFormField, + params: { label: string; help?: string }, +) { return renderSettingsToggleRow({ title: params.label, description: params.help, - checked: params.checked, - disabled: params.disabled, - onChange: params.onChange, + checked: props.form[field], + onChange: (checked) => props.onFormChange({ [field]: checked }), }); } @@ -517,6 +555,32 @@ function renderToolbar(props: CronProps, hasAdvancedJobsFilters: boolean) { `; } +function renderJobsFilter( + props: CronProps, + field: keyof Parameters[0], + params: { + label: string; + value: string; + options: readonly CronSelectOption[]; + testId?: string; + }, +) { + return html` + + `; +} + function renderJobsFilterPopover(props: CronProps, active: boolean) { return html`