From f675c5f411a8159a32cd1173be917fbbfe5c5863 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 21 Apr 2026 13:22:01 -0500 Subject: [PATCH] fix(ui): harden sandboxed parser async results --- ui/src/adapters/dynamic-loader.ts | 119 +++++++++++++----- ui/src/adapters/registry.ts | 4 +- .../adapters/sandboxed-parser-worker.test.ts | 6 + ui/src/adapters/sandboxed-parser-worker.ts | 6 +- 4 files changed, 101 insertions(+), 34 deletions(-) diff --git a/ui/src/adapters/dynamic-loader.ts b/ui/src/adapters/dynamic-loader.ts index fcf63e63a97..fc3f5132452 100644 --- a/ui/src/adapters/dynamic-loader.ts +++ b/ui/src/adapters/dynamic-loader.ts @@ -15,16 +15,14 @@ * Worker → Main: { type: "result", id, entries } * * Because the parse call is async (cross-thread postMessage), but the - * existing `parseStdoutLine` contract is synchronous, we maintain an - * internal queue and return results via the registry re-registration - * pattern that already exists for the lazy-loading bridge. + * existing `parseStdoutLine` contract is synchronous, we cache completed + * worker results and ask the adapter registry to recompute transcripts when + * a new result arrives. * * **Synchronous fast-path**: After init, parse requests are sent to the * worker which responds asynchronously. The `parseStdoutLine` wrapper - * collects lines and returns `[]` synchronously; when the worker responds, - * the results are merged into the transcript via the existing adapter - * change notification mechanism. In practice this adds ~1 frame of - * latency which is imperceptible. + * returns cached results synchronously on the next transcript recomputation. + * In practice this adds ~1 frame of latency which is imperceptible. * * Security: see `sandboxed-parser-worker.ts` for the full lockdown. */ @@ -64,6 +62,12 @@ const failedLoads = new Set(); /** In-flight init promises so concurrent callers share the same load. */ const loadPromises = new Map>(); +let resultNotifier: (() => void) | null = null; + +export function setDynamicParserResultNotifier(fn: (() => void) | null): void { + resultNotifier = fn; +} + // ── Internal helpers ──────────────────────────────────────────────────────── function sendToWorker(sandbox: SandboxedParser, msg: SandboxRequest): void { @@ -74,6 +78,18 @@ function nextRequestId(sandbox: SandboxedParser): number { return sandbox.nextId++; } +function lineCacheKey(line: string, ts: string): string { + return `${ts}\u0000${line}`; +} + +function statefulLineCacheKey(index: number, line: string, ts: string): string { + return `${index}\u0000${ts}\u0000${line}`; +} + +function notifyResultReady(): void { + resultNotifier?.(); +} + /** * Parse a single line synchronously by delegating to the worker. * Returns a Promise that resolves with the TranscriptEntry[] from the worker. @@ -98,6 +114,13 @@ function drainPendingRequests(sandbox: SandboxedParser): void { sandbox.pendingParserCreates.clear(); } +function drainPendingParserCreates(sandbox: SandboxedParser): void { + for (const resolver of sandbox.pendingParserCreates.values()) { + resolver(null); + } + sandbox.pendingParserCreates.clear(); +} + /** * Create a sandboxed worker, send the parser source, and wait for init. */ @@ -145,7 +168,7 @@ function initSandboxedWorker(source: string): Promise { } } else if (resp.type === "error") { console.error("[adapter-ui-loader] Worker reported error:", resp.message); - drainPendingRequests(sandbox); + drainPendingParserCreates(sandbox); } }; @@ -191,31 +214,39 @@ function initSandboxedWorker(source: string): Promise { * subsequent render shows the parsed entries. */ function buildParserModule(sandbox: SandboxedParser): DynamicParserModule { - // Accumulator for async results. The sync parseStdoutLine returns the - // accumulated buffer and clears it, so results appear on the next call. - const asyncBuffer: TranscriptEntry[] = []; + const parseCache = new Map(); + const pendingParseKeys = new Set(); const parseStdoutLine: StdoutLineParser = (line: string, ts: string) => { - // Drain any previously resolved async results first. - const buffered = asyncBuffer.splice(0, asyncBuffer.length); + const key = lineCacheKey(line, ts); + const cached = parseCache.get(key); + if (cached) return cached.slice(); - // Fire off the async parse; results will be buffered for next call. - parseLineAsync(sandbox, line, ts).then((entries) => { - if (entries.length > 0) { - asyncBuffer.push(...entries); - } - }); + if (!pendingParseKeys.has(key)) { + pendingParseKeys.add(key); + parseLineAsync(sandbox, line, ts).then((entries) => { + pendingParseKeys.delete(key); + parseCache.set(key, entries); + notifyResultReady(); + }); + } - return buffered; + return []; }; const mod: DynamicParserModule = { parseStdoutLine }; if (sandbox.hasFactory) { + const statefulCache = new Map(); + const statefulPendingKeys = new Set(); + mod.createStdoutParser = (): StatefulStdoutParser => { let parserId: number | null = null; let parserReady = false; - const parserBuffer: TranscriptEntry[] = []; + let lineIndex = 0; + const history: { key: string; line: string; ts: string }[] = []; + const sentKeys = new Set(); + let replaying = false; // Request a parser instance from the worker. const id = nextRequestId(sandbox); @@ -223,25 +254,51 @@ function buildParserModule(sandbox: SandboxedParser): DynamicParserModule { if (createdParserId !== null) { parserId = createdParserId; parserReady = true; + flushReplay(); } }); sendToWorker(sandbox, { type: "create_parser", id }); + const sendParserLine = (item: { key: string; line: string; ts: string }, cacheResult: boolean): void => { + if (parserId === null || sentKeys.has(item.key)) return; + sentKeys.add(item.key); + const reqId = nextRequestId(sandbox); + sandbox.pendingResolves.set(reqId, (entries) => { + if (cacheResult) { + statefulPendingKeys.delete(item.key); + statefulCache.set(item.key, entries); + notifyResultReady(); + } + }); + sendToWorker(sandbox, { type: "parser_parse", id: reqId, parserId, line: item.line, ts: item.ts }); + }; + + function flushReplay(): void { + if (!parserReady || parserId === null || replaying) return; + replaying = true; + try { + for (const item of history) { + sendParserLine(item, statefulPendingKeys.has(item.key)); + } + } finally { + replaying = false; + } + } + return { parseLine: (line: string, ts: string): TranscriptEntry[] => { - const buffered = parserBuffer.splice(0, parserBuffer.length); + const key = statefulLineCacheKey(lineIndex++, line, ts); + const cached = statefulCache.get(key); + history.push({ key, line, ts }); - if (parserReady && parserId !== null) { - const reqId = nextRequestId(sandbox); - sandbox.pendingResolves.set(reqId, (entries) => { - if (entries.length > 0) { - parserBuffer.push(...entries); - } - }); - sendToWorker(sandbox, { type: "parser_parse", id: reqId, parserId, line, ts }); + if (cached) return cached.slice(); + + if (!statefulPendingKeys.has(key)) { + statefulPendingKeys.add(key); + flushReplay(); } - return buffered; + return []; }, reset: () => { if (parserReady && parserId !== null) { diff --git a/ui/src/adapters/registry.ts b/ui/src/adapters/registry.ts index e418457ee44..e8e24eafed1 100644 --- a/ui/src/adapters/registry.ts +++ b/ui/src/adapters/registry.ts @@ -9,7 +9,7 @@ import { openClawGatewayUIAdapter } from "./openclaw-gateway"; import { hermesLocalUIAdapter } from "./hermes-local"; import { processUIAdapter } from "./process"; import { httpUIAdapter } from "./http"; -import { loadDynamicParser, invalidateDynamicParser } from "./dynamic-loader"; +import { loadDynamicParser, invalidateDynamicParser, setDynamicParserResultNotifier } from "./dynamic-loader"; import { SchemaConfigFields, buildSchemaAdapterConfig } from "./schema-config-fields"; const uiAdapters: UIAdapterModule[] = []; @@ -45,6 +45,8 @@ function notifyAdapterChange(): void { for (const fn of adapterChangeListeners) fn(); } +setDynamicParserResultNotifier(notifyAdapterChange); + function registerBuiltInUIAdapters() { for (const adapter of [ claudeLocalUIAdapter, diff --git a/ui/src/adapters/sandboxed-parser-worker.test.ts b/ui/src/adapters/sandboxed-parser-worker.test.ts index 4ba64835d01..cc071862ee3 100644 --- a/ui/src/adapters/sandboxed-parser-worker.test.ts +++ b/ui/src/adapters/sandboxed-parser-worker.test.ts @@ -9,10 +9,16 @@ describe("sandboxed parser worker bootstrap", () => { expect(source).toContain("self.Worker = _undefined"); expect(source).toContain("self.SharedWorker = _undefined"); expect(source).toContain("self.Blob = _undefined"); + expect(source).toContain("self.RTCPeerConnection = _undefined"); + expect(source).toContain("self.RTCDataChannel = _undefined"); expect(source).toContain('"createObjectURL"'); expect(source).toContain('"revokeObjectURL"'); }); + it("evaluates parser source in strict mode", () => { + expect(getWorkerBootstrapSource()).toContain('\\"use strict\\";\\n{\\n" + msg.source'); + }); + it("does not include the unused parse_batch protocol branch", () => { expect(getWorkerBootstrapSource()).not.toContain("parse_batch"); }); diff --git a/ui/src/adapters/sandboxed-parser-worker.ts b/ui/src/adapters/sandboxed-parser-worker.ts index 7351c370c40..fa8e054dd95 100644 --- a/ui/src/adapters/sandboxed-parser-worker.ts +++ b/ui/src/adapters/sandboxed-parser-worker.ts @@ -53,6 +53,8 @@ self.fetch = _undefined; self.XMLHttpRequest = _undefined; self.WebSocket = _undefined; self.EventSource = _undefined; +self.RTCPeerConnection = _undefined; +self.RTCDataChannel = _undefined; self.Request = _undefined; self.Response = _undefined; self.Headers = _undefined; @@ -112,7 +114,7 @@ self.onmessage = function (e) { const factory = new Function( "exports", "module", "self", "globalThis", // Wrap in a block to prevent hoisted declarations from leaking. - "{\\n" + msg.source + "\\n}" + "\\"use strict\\";\\n{\\n" + msg.source + "\\n}" ); factory(exports, module, _undefined, _undefined); @@ -211,7 +213,7 @@ export function createSandboxedWorker(): Worker { try { return new Worker(url); } finally { - // Revoke immediately — the Worker has already loaded the source. + // Revoke after construction; the Worker has already captured the Blob URL source. URL.revokeObjectURL(url); } }