fix(ui): simplify sandboxed parser bridge

This commit is contained in:
Dotta committed 2026-04-21 13:27:57 -05:00
1 parent f675c5f411
commit 34c9da00e6
2 files changed
+20 -162

No files matched your search

+6 -111
View File
@@ -28,7 +28,7 @@
*/
import type { TranscriptEntry } from "@paperclipai/adapter-utils";
import type { StatefulStdoutParser, StdoutLineParser, StdoutParserFactory } from "./types";
import type { StdoutLineParser, StdoutParserFactory } from "./types";
import { createSandboxedWorker } from "./sandboxed-parser-worker";
import type { SandboxRequest, SandboxResponse } from "./sandboxed-parser-worker";
@@ -42,10 +42,8 @@ interface DynamicParserModule {
interface SandboxedParser {
worker: Worker;
ready: boolean;
hasFactory: boolean;
nextId: number;
pendingResolves: Map<number, (entries: TranscriptEntry[]) => void>;
pendingParserCreates: Map<number, (parserId: number | null) => void>;
}
// ── State ───────────────────────────────────────────────────────────────────
@@ -82,10 +80,6 @@ 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?.();
}
@@ -107,18 +101,6 @@ function drainPendingRequests(sandbox: SandboxedParser): void {
resolver([]);
}
sandbox.pendingResolves.clear();
for (const resolver of sandbox.pendingParserCreates.values()) {
resolver(null);
}
sandbox.pendingParserCreates.clear();
}
function drainPendingParserCreates(sandbox: SandboxedParser): void {
for (const resolver of sandbox.pendingParserCreates.values()) {
resolver(null);
}
sandbox.pendingParserCreates.clear();
}
/**
@@ -130,10 +112,8 @@ function initSandboxedWorker(source: string): Promise<SandboxedParser> {
const sandbox: SandboxedParser = {
worker,
ready: false,
hasFactory: false,
nextId: 1,
pendingResolves: new Map(),
pendingParserCreates: new Map(),
};
// Timeout if the worker doesn't respond within 5s
@@ -149,7 +129,6 @@ function initSandboxedWorker(source: string): Promise<SandboxedParser> {
if (msg.type === "ready") {
clearTimeout(timeout);
sandbox.ready = true;
sandbox.hasFactory = msg.hasFactory;
// Switch to the steady-state message handler.
worker.onmessage = (ev: MessageEvent<SandboxResponse>) => {
@@ -160,15 +139,9 @@ function initSandboxedWorker(source: string): Promise<SandboxedParser> {
sandbox.pendingResolves.delete(resp.id);
resolver(resp.entries as TranscriptEntry[]);
}
} else if (resp.type === "parser_created") {
const resolver = sandbox.pendingParserCreates.get(resp.id);
if (resolver) {
sandbox.pendingParserCreates.delete(resp.id);
resolver(resp.parserId);
}
} else if (resp.type === "error") {
console.error("[adapter-ui-loader] Worker reported error:", resp.message);
drainPendingParserCreates(sandbox);
drainPendingRequests(sandbox);
}
};
@@ -201,12 +174,9 @@ function initSandboxedWorker(source: string): Promise<SandboxedParser> {
* Build a DynamicParserModule that delegates all calls to the sandboxed worker.
*
* The parseStdoutLine wrapper is **synchronous** to match the existing contract.
* It sends the parse request to the worker and returns `[]` immediately.
* When the worker responds, the caller must handle the async result.
*
* For callers that need sync results (the current transcript renderer), we
* provide a sync wrapper that accumulates results. The registry's
* re-registration pattern will trigger a re-render when results arrive.
* Cache misses send a parse request to the worker and return `[]`; when the
* worker responds, the registry notification path recomputes transcripts and
* this wrapper returns the cached result synchronously.
*
* In practice, because the existing codebase already handles the "bridge"
* pattern where parseStdoutLine returns [] until the dynamic parser loads,
@@ -234,82 +204,7 @@ function buildParserModule(sandbox: SandboxedParser): DynamicParserModule {
return [];
};
const mod: DynamicParserModule = { parseStdoutLine };
if (sandbox.hasFactory) {
const statefulCache = new Map<string, TranscriptEntry[]>();
const statefulPendingKeys = new Set<string>();
mod.createStdoutParser = (): StatefulStdoutParser => {
let parserId: number | null = null;
let parserReady = false;
let lineIndex = 0;
const history: { key: string; line: string; ts: string }[] = [];
const sentKeys = new Set<string>();
let replaying = false;
// Request a parser instance from the worker.
const id = nextRequestId(sandbox);
sandbox.pendingParserCreates.set(id, (createdParserId) => {
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 key = statefulLineCacheKey(lineIndex++, line, ts);
const cached = statefulCache.get(key);
history.push({ key, line, ts });
if (cached) return cached.slice();
if (!statefulPendingKeys.has(key)) {
statefulPendingKeys.add(key);
flushReplay();
}
return [];
},
reset: () => {
if (parserReady && parserId !== null) {
sendToWorker(sandbox, { type: "parser_reset", parserId });
}
},
};
};
}
return mod;
return { parseStdoutLine };
}
// ── Public API ──────────────────────────────────────────────────────────────
+14 -51
View File
@@ -16,17 +16,13 @@
/** Messages sent from the main thread to the worker. */
export type SandboxRequest =
| { type: "init"; source: string }
| { type: "parse"; id: number; line: string; ts: string }
| { type: "create_parser"; id: number }
| { type: "parser_parse"; id: number; parserId: number; line: string; ts: string }
| { type: "parser_reset"; parserId: number };
| { type: "parse"; id: number; line: string; ts: string };
/** Messages sent from the worker back to the main thread. */
export type SandboxResponse =
| { type: "ready"; hasFactory: boolean }
| { type: "ready" }
| { type: "error"; message: string }
| { type: "result"; id: number; entries: unknown[] }
| { type: "parser_created"; id: number; parserId: number };
| { type: "result"; id: number; entries: unknown[] };
// ── Worker bootstrap source ─────────────────────────────────────────────────
@@ -37,8 +33,7 @@ export type SandboxResponse =
* no-ops or `undefined`.
* 2. Waits for an `init` message carrying the adapter's parser source.
* 3. Evaluates the source via `new Function()` and extracts exports.
* 4. Responds to `parse` / `create_parser` / `parser_parse` / `parser_reset`
* messages with `TranscriptEntry[]` results.
* 4. Responds to `parse` messages with `TranscriptEntry[]` results.
*/
const WORKER_BOOTSTRAP = `
"use strict";
@@ -88,8 +83,7 @@ self.IDBFactory = _undefined;
let parseStdoutLine = null;
let createStdoutParser = null;
let nextParserId = 1;
const parserInstances = new Map();
let fallbackParser = null;
// ── 3. Message handler ──────────────────────────────────────────────────────
@@ -129,13 +123,19 @@ self.onmessage = function (e) {
if (typeof resolved.createStdoutParser === "function") {
createStdoutParser = resolved.createStdoutParser;
}
if (!parseStdoutLine && createStdoutParser) {
fallbackParser = createStdoutParser();
if (fallbackParser && typeof fallbackParser.parseLine === "function") {
parseStdoutLine = (line, ts) => fallbackParser.parseLine(line, ts);
}
}
if (!parseStdoutLine && !createStdoutParser) {
self.postMessage({ type: "error", message: "Parser module exports neither parseStdoutLine nor createStdoutParser" });
if (!parseStdoutLine) {
self.postMessage({ type: "error", message: "Parser module exports no usable parseStdoutLine or createStdoutParser" });
return;
}
self.postMessage({ type: "ready", hasFactory: !!createStdoutParser });
self.postMessage({ type: "ready" });
} catch (err) {
self.postMessage({ type: "error", message: "Parser init failed: " + (err && err.message || String(err)) });
}
@@ -152,43 +152,6 @@ self.onmessage = function (e) {
return;
}
if (msg.type === "create_parser") {
try {
if (!createStdoutParser) {
self.postMessage({ type: "error", message: "No createStdoutParser factory available" });
return;
}
const id = nextParserId++;
parserInstances.set(id, createStdoutParser());
self.postMessage({ type: "parser_created", id: msg.id, parserId: id });
} catch (err) {
self.postMessage({ type: "error", message: "createStdoutParser failed: " + (err && err.message || String(err)) });
}
return;
}
if (msg.type === "parser_parse") {
try {
const parser = parserInstances.get(msg.parserId);
if (!parser) {
self.postMessage({ type: "result", id: msg.id, entries: [] });
return;
}
const entries = parser.parseLine(msg.line, msg.ts);
self.postMessage({ type: "result", id: msg.id, entries: entries || [] });
} catch (err) {
self.postMessage({ type: "result", id: msg.id, entries: [] });
}
return;
}
if (msg.type === "parser_reset") {
try {
const parser = parserInstances.get(msg.parserId);
if (parser) parser.reset();
} catch {}
return;
}
};
`;