From d8fe3d72c82cc3cc6a6e9275494f5e5377b058a6 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Fri, 12 Jun 2026 18:19:01 -0700 Subject: [PATCH] Scope preview listeners and control sessions - Tie preview and debugger listeners to Effect scopes - Factor shared automation helpers for snapshot and input handling - Improve cleanup for browser preview sessions and port scanning --- apps/desktop/src/preview/Manager.ts | 1037 +++++++++-------- apps/server/src/mcp/McpHttpServer.test.ts | 2 +- apps/server/src/mcp/McpHttpServer.ts | 155 +-- apps/server/src/mcp/McpSessionRegistry.ts | 68 +- .../src/mcp/PreviewAutomationBroker.test.ts | 6 +- .../server/src/mcp/PreviewAutomationBroker.ts | 102 +- .../src/mcp/toolkits/preview/handlers.ts | 28 +- apps/server/src/mcp/toolkits/preview/tools.ts | 21 + apps/server/src/preview/Manager.ts | 6 +- apps/server/src/preview/PortScanner.test.ts | 8 +- apps/server/src/preview/PortScanner.ts | 129 +- apps/server/src/server.test.ts | 4 +- apps/server/src/ws.ts | 36 +- 13 files changed, 854 insertions(+), 748 deletions(-) diff --git a/apps/desktop/src/preview/Manager.ts b/apps/desktop/src/preview/Manager.ts index a55283fbebb4da38831b035ad46bd04460ce1f75..f82741d908ffd3eef09268af8750c8fb79e99a6d 100644 --- a/apps/desktop/src/preview/Manager.ts +++ b/apps/desktop/src/preview/Manager.ts @@ -48,7 +48,7 @@ import * as Path from "effect/Path"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; -import type * as Scope from "effect/Scope"; +import * as Scope from "effect/Scope"; import * as SynchronizedRef from "effect/SynchronizedRef"; import * as DesktopEnvironment from "../app/DesktopEnvironment.ts"; @@ -247,10 +247,7 @@ type PreviewInputSignal = | { readonly kind: "key"; readonly key: string; readonly code: string }; interface ManagedListeners { - navigate: () => void; - failed: (event: Event, code: number, description: string) => void; - humanInput: (_event: unknown, signal?: unknown) => void; - beforeInput: (event: Electron.Event, input: Electron.Input) => void; + readonly scope: Scope.Closeable; } interface PickSession { @@ -260,6 +257,7 @@ interface PickSession { interface BrowserControlSession { readonly webContentsId: number; readonly semaphore: Semaphore.Semaphore; + readonly scope: Scope.Closeable; readonly onMessage: ( event: Electron.Event, method: string, @@ -339,6 +337,7 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function ) { const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; + const parentScope = yield* Scope.Scope; const context = yield* Effect.context(); const runFork = Effect.runForkWith(context); const resolvedArtifactDirectory = path.resolve(artifactDirectory); @@ -622,16 +621,15 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function copy.delete(webContentsId); }), ]); + if (control) { + yield* Scope.close(control.scope, Exit.void).pipe(Effect.ignore); + return; + } yield* Ref.update(diagnosticsRef, (diagnostics) => replaceMap(diagnostics, (copy) => { copy.delete(webContentsId); }), ); - const wc = webContents.fromId(webContentsId); - if (!wc || wc.isDestroyed()) return; - if (control) wc.debugger.off("message", control.onMessage); - if (!wc.debugger.isAttached()) return; - yield* attempt("detachControlSession", () => wc.debugger.detach()).pipe(Effect.ignore); }); const ensureControlSession = Effect.fn("PreviewManager.ensureControlSession")(function* ( @@ -662,87 +660,103 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function ), ); } - return Effect.gen(function* () { + const createControlSession = Effect.fn("PreviewManager.createControlSession")(function* () { const semaphore = yield* Semaphore.make(1); + const scope = yield* Scope.fork(parentScope, "sequential"); + const handleDebuggerMessage = Effect.fn("PreviewManager.handleDebuggerMessage")(function* ( + method: string, + params: Record, + ) { + if (method === "Page.screencastFrame") { + const sessionId = params["sessionId"]; + if (typeof sessionId === "number") { + yield* attemptPromise("ackScreencastFrame", () => + wc.debugger.sendCommand("Page.screencastFrameAck", { sessionId }), + ).pipe(Effect.ignore); + } + const tabId = yield* tabIdForWebContents(wc.id); + const metadata = + typeof params["metadata"] === "object" && params["metadata"] !== null + ? (params["metadata"] as Record) + : {}; + if (tabId && typeof params["data"] === "string") { + const receivedAt = yield* currentIso; + const listeners = yield* Ref.get(recordingFrameListenersRef); + const frame: DesktopPreviewRecordingFrame = { + tabId, + data: params["data"], + width: typeof metadata["deviceWidth"] === "number" ? metadata["deviceWidth"] : 0, + height: typeof metadata["deviceHeight"] === "number" ? metadata["deviceHeight"] : 0, + receivedAt, + }; + yield* Effect.forEach( + listeners, + (listener) => Effect.sync(() => listener(frame)).pipe(Effect.ignore), + { discard: true }, + ); + } + } + yield* captureDiagnosticMessage(wc.id, method, params); + }); const onMessage: BrowserControlSession["onMessage"] = (_event, method, params) => { - runFork( - Effect.gen(function* () { - if (method === "Page.screencastFrame") { - const sessionId = params["sessionId"]; - if (typeof sessionId === "number") { - yield* attemptPromise("ackScreencastFrame", () => - wc.debugger.sendCommand("Page.screencastFrameAck", { sessionId }), - ).pipe(Effect.ignore); - } - const tabId = yield* tabIdForWebContents(wc.id); - const metadata = - typeof params["metadata"] === "object" && params["metadata"] !== null - ? (params["metadata"] as Record) - : {}; - if (tabId && typeof params["data"] === "string") { - const receivedAt = yield* currentIso; - const listeners = yield* Ref.get(recordingFrameListenersRef); - const frame: DesktopPreviewRecordingFrame = { - tabId, - data: params["data"], - width: - typeof metadata["deviceWidth"] === "number" ? metadata["deviceWidth"] : 0, - height: - typeof metadata["deviceHeight"] === "number" ? metadata["deviceHeight"] : 0, - receivedAt, - }; - yield* Effect.forEach( - listeners, - (listener) => Effect.sync(() => listener(frame)).pipe(Effect.ignore), - { discard: true }, - ); - } - } - yield* captureDiagnosticMessage(wc.id, method, params); - }), - ); + runFork(handleDebuggerMessage(method, params)); }; - const control: BrowserControlSession = { webContentsId: wc.id, semaphore, onMessage }; - yield* Ref.update(diagnosticsRef, (diagnostics) => - replaceMap(diagnostics, (copy) => { - copy.set(wc.id, { - consoleEntries: [], - networkEntries: [], - requests: new Map(), - }); - }), - ); - yield* attempt("attachDebuggerListeners", () => { - wc.debugger.on("message", onMessage); - wc.debugger.attach("1.3"); - }); - yield* Effect.all( - ["Runtime.enable", "Accessibility.enable", "Network.enable", "Log.enable"].map((method) => - attemptPromise("initializeDebugger", () => wc.debugger.sendCommand(method)), - ), - { concurrency: "unbounded", discard: true }, - ).pipe( - Effect.tapError(() => - Effect.all([ + yield* Scope.addFinalizer( + scope, + Effect.all( + [ Ref.update(diagnosticsRef, (diagnostics) => replaceMap(diagnostics, (copy) => { copy.delete(wc.id); }), ), - attempt("detachFailedDebugger", () => { + attempt("detachControlSession", () => { wc.debugger.off("message", onMessage); if (wc.debugger.isAttached()) wc.debugger.detach(); }).pipe(Effect.ignore), - ]).pipe(Effect.asVoid), + ], + { discard: true }, ), ); - return [ - control, - replaceMap(sessions, (copy) => { - copy.set(wc.id, control); - }), - ] as const; + const control: BrowserControlSession = { + webContentsId: wc.id, + semaphore, + scope, + onMessage, + }; + const initialize = Effect.fn("PreviewManager.initializeControlSession")(function* () { + yield* Ref.update(diagnosticsRef, (diagnostics) => + replaceMap(diagnostics, (copy) => { + copy.set(wc.id, { + consoleEntries: [], + networkEntries: [], + requests: new Map(), + }); + }), + ); + yield* attempt("attachDebuggerListeners", () => { + wc.debugger.on("message", onMessage); + wc.debugger.attach("1.3"); + }); + yield* Effect.all( + ["Runtime.enable", "Accessibility.enable", "Network.enable", "Log.enable"].map( + (method) => + attemptPromise("initializeDebugger", () => wc.debugger.sendCommand(method)), + ), + { concurrency: "unbounded", discard: true }, + ); + return [ + control, + replaceMap(sessions, (copy) => { + copy.set(wc.id, control); + }), + ] as const; + }); + return yield* initialize().pipe( + Effect.onError(() => Scope.close(scope, Exit.void).pipe(Effect.ignore)), + ); }); + return createControlSession(); }); }); @@ -906,25 +920,13 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function const detachListeners = Effect.fn("PreviewManager.detachListeners")(function* ( webContentsId: number, ) { - const handlers = yield* Ref.modify(attachedRef, (attached) => [ + const managed = yield* Ref.modify(attachedRef, (attached) => [ attached.get(webContentsId), replaceMap(attached, (copy) => { copy.delete(webContentsId); }), ]); - if (!handlers) return; - const wc = webContents.fromId(webContentsId); - if (!wc || wc.isDestroyed()) return; - yield* attempt("detachListeners", () => { - wc.off("did-navigate", handlers.navigate); - wc.off("did-navigate-in-page", handlers.navigate); - wc.off("page-title-updated", handlers.navigate); - wc.off("did-start-loading", handlers.navigate); - wc.off("did-stop-loading", handlers.navigate); - wc.off("did-fail-load", handlers.failed as never); - wc.off("before-input-event", handlers.beforeInput); - wc.ipc.off(HUMAN_INPUT_CHANNEL, handlers.humanInput); - }).pipe(Effect.ignore); + if (managed) yield* Scope.close(managed.scope, Exit.void).pipe(Effect.ignore); }); const isAppShortcut = (input: Electron.Input): boolean => @@ -987,17 +989,16 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function tabId: string, wc: Electron.WebContents, ) { - const sync = () => - runFork( - Effect.gen(function* () { - if (wc.isDestroyed()) return; - yield* update(tabId, { - navStatus: computeNavStatus(wc), - canGoBack: wc.navigationHistory.canGoBack(), - canGoForward: wc.navigationHistory.canGoForward(), - }); - }), - ); + const scope = yield* Scope.fork(parentScope, "sequential"); + const syncState = Effect.fn("PreviewManager.syncWebContentsState")(function* () { + if (wc.isDestroyed()) return; + yield* update(tabId, { + navStatus: computeNavStatus(wc), + canGoBack: wc.navigationHistory.canGoBack(), + canGoForward: wc.navigationHistory.canGoForward(), + }); + }); + const sync = () => runFork(syncState()); const failed = (_event: Event, code: number, description: string): void => { if (code === -3) return; runFork( @@ -1012,73 +1013,85 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function }), ); }; + const handleHumanInput = Effect.fn("PreviewManager.handleHumanInput")(function* ( + rawSignal?: unknown, + ) { + if (isPreviewInputSignal(rawSignal) && (yield* consumeExpectedAgentInput(tabId, rawSignal))) { + return; + } + yield* Ref.update(controlEpochRef, (epochs) => + replaceMap(epochs, (copy) => { + copy.set(tabId, (epochs.get(tabId) ?? 0) + 1); + }), + ); + yield* update(tabId, { controller: "human" }); + yield* Effect.sleep(750); + const tabs = yield* SynchronizedRef.get(tabsRef); + if (tabs.get(tabId)?.controller === "human") { + yield* update(tabId, { controller: "none" }); + } + }); const humanInput = (_event: unknown, rawSignal?: unknown): void => { - runFork( - Effect.gen(function* () { - if ( - isPreviewInputSignal(rawSignal) && - (yield* consumeExpectedAgentInput(tabId, rawSignal)) - ) { - return; - } - yield* Ref.update(controlEpochRef, (epochs) => - replaceMap(epochs, (copy) => { - copy.set(tabId, (epochs.get(tabId) ?? 0) + 1); - }), - ); - yield* update(tabId, { controller: "human" }); - yield* Effect.sleep(750); - const tabs = yield* SynchronizedRef.get(tabsRef); - if (tabs.get(tabId)?.controller === "human") { - yield* update(tabId, { controller: "none" }); - } - }), - ); + runFork(handleHumanInput(rawSignal)); }; + const forwardShortcut = Effect.fn("PreviewManager.forwardShortcut")(function* ( + event: Electron.Event, + input: Electron.Input, + ) { + const mainWindow = yield* Ref.get(mainWindowRef); + if (!isAppShortcut(input) || Option.isNone(mainWindow) || mainWindow.value.isDestroyed()) { + return; + } + event.preventDefault(); + mainWindow.value.webContents.sendInputEvent({ + type: "keyDown", + keyCode: input.key, + modifiers: [ + ...(input.meta ? (["meta"] as const) : []), + ...(input.shift ? (["shift"] as const) : []), + ...(input.control ? (["control"] as const) : []), + ...(input.alt ? (["alt"] as const) : []), + ], + }); + }); const beforeInput = (event: Electron.Event, input: Electron.Input): void => { - runFork( - Effect.gen(function* () { - const mainWindow = yield* Ref.get(mainWindowRef); - if ( - !isAppShortcut(input) || - Option.isNone(mainWindow) || - mainWindow.value.isDestroyed() - ) { - return; - } - event.preventDefault(); - mainWindow.value.webContents.sendInputEvent({ - type: "keyDown", - keyCode: input.key, - modifiers: [ - ...(input.meta ? (["meta"] as const) : []), - ...(input.shift ? (["shift"] as const) : []), - ...(input.control ? (["control"] as const) : []), - ...(input.alt ? (["alt"] as const) : []), - ], - }); - }), - ); + runFork(forwardShortcut(event, input)); }; - yield* attempt("attachListeners", () => { - wc.on("did-navigate", sync); - wc.on("did-navigate-in-page", sync); - wc.on("page-title-updated", sync); - wc.on("did-start-loading", sync); - wc.on("did-stop-loading", sync); - wc.on("did-fail-load", failed as never); - wc.ipc.on(HUMAN_INPUT_CHANNEL, humanInput); - wc.setWindowOpenHandler(({ url }) => { - runFork(attemptPromise("openPreviewWindow", () => wc.loadURL(url)).pipe(Effect.ignore)); - return { action: "deny" }; + yield* Scope.addFinalizer( + scope, + attempt("detachListeners", () => { + wc.off("did-navigate", sync); + wc.off("did-navigate-in-page", sync); + wc.off("page-title-updated", sync); + wc.off("did-start-loading", sync); + wc.off("did-stop-loading", sync); + wc.off("did-fail-load", failed as never); + wc.off("before-input-event", beforeInput); + wc.ipc.off(HUMAN_INPUT_CHANNEL, humanInput); + }).pipe(Effect.ignore), + ); + const install = Effect.fn("PreviewManager.installWebContentsListeners")(function* () { + yield* attempt("attachListeners", () => { + wc.on("did-navigate", sync); + wc.on("did-navigate-in-page", sync); + wc.on("page-title-updated", sync); + wc.on("did-start-loading", sync); + wc.on("did-stop-loading", sync); + wc.on("did-fail-load", failed as never); + wc.ipc.on(HUMAN_INPUT_CHANNEL, humanInput); + wc.setWindowOpenHandler(({ url }) => { + runFork(attemptPromise("openPreviewWindow", () => wc.loadURL(url)).pipe(Effect.ignore)); + return { action: "deny" }; + }); + wc.on("before-input-event", beforeInput); }); - wc.on("before-input-event", beforeInput); + yield* Ref.update(attachedRef, (attached) => + replaceMap(attached, (copy) => { + copy.set(wc.id, { scope }); + }), + ); }); - yield* Ref.update(attachedRef, (attached) => - replaceMap(attached, (copy) => { - copy.set(wc.id, { navigate: sync, failed, humanInput, beforeInput }); - }), - ); + yield* install().pipe(Effect.onError(() => Scope.close(scope, Exit.void).pipe(Effect.ignore))); }); const setMainWindow = Effect.fn("PreviewManager.setMainWindow")(function* ( @@ -1271,7 +1284,7 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function const annotationTheme = yield* Ref.get(annotationThemeRef); return yield* Effect.callback( (resume) => { - const cleanup = Effect.gen(function* () { + const cleanup = Effect.fn("PreviewManager.cleanupPickElement")(function* () { yield* attempt("pickElement.cleanup", () => { wc.ipc.removeListener(ELEMENT_PICKED_CHANNEL, onMessage); wc.off("destroyed", onDestroyed); @@ -1283,18 +1296,19 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function }), ); }); + const settlePick = Effect.fn("PreviewManager.settlePickElement")(function* ( + payload: PreviewAnnotationPayload | null, + ) { + const active = (yield* Ref.get(pickSessionsRef)).get(tabId); + if (!active || active.cancel !== cancel) return; + yield* cleanup(); + resume(Effect.succeed(payload)); + }); const settle = (payload: PreviewAnnotationPayload | null) => { - runFork( - Effect.gen(function* () { - const active = (yield* Ref.get(pickSessionsRef)).get(tabId); - if (!active || active.cancel !== cancel) return; - yield* cleanup; - resume(Effect.succeed(payload)); - }), - ); + runFork(settlePick(payload)); }; - const cancel = Effect.gen(function* () { - yield* cleanup; + const cancelPickSession = Effect.fn("PreviewManager.cancelPickSession")(function* () { + yield* cleanup(); const tabs = yield* SynchronizedRef.get(tabsRef); const activeTab = tabs.get(tabId); if (activeTab?.webContentsId != null) { @@ -1307,6 +1321,7 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function } resume(Effect.succeed(null)); }); + const cancel = cancelPickSession(); const onMessage = (_event: Electron.IpcMainEvent, ...args: unknown[]): void => { const payload = args[0]; if (!isPreviewAnnotationPayload(payload)) { @@ -1330,24 +1345,25 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function }; const onDestroyed = () => settle(null); const onNavigated = () => settle(null); + const registerPickElement = Effect.fn("PreviewManager.registerPickElement")(function* () { + yield* attempt("pickElement.register", () => { + wc.ipc.on(ELEMENT_PICKED_CHANNEL, onMessage); + wc.once("destroyed", onDestroyed); + wc.once("did-start-navigation", onNavigated); + if (!wc.isFocused()) wc.focus(); + wc.send(START_PICK_CHANNEL, annotationTheme); + }); + yield* Ref.update(pickSessionsRef, (sessions) => + replaceMap(sessions, (copy) => { + copy.set(tabId, { cancel }); + }), + ); + }); runFork( - Effect.gen(function* () { - yield* attempt("pickElement.register", () => { - wc.ipc.on(ELEMENT_PICKED_CHANNEL, onMessage); - wc.once("destroyed", onDestroyed); - wc.once("did-start-navigation", onNavigated); - if (!wc.isFocused()) wc.focus(); - wc.send(START_PICK_CHANNEL, annotationTheme); - }); - yield* Ref.update(pickSessionsRef, (sessions) => - replaceMap(sessions, (copy) => { - copy.set(tabId, { cancel }); - }), - ); - }).pipe( + registerPickElement().pipe( Effect.catch((error: PreviewManagerError) => { resume(Effect.fail(error)); - return cleanup; + return cleanup(); }), ), ); @@ -1401,6 +1417,19 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function }; }); + const startScreencast = Effect.fn("PreviewManager.startScreencast")(function* ( + send: SendCommand, + ) { + yield* send("Page.enable"); + yield* send("Page.startScreencast", { + format: "jpeg", + quality: 80, + maxWidth: 1600, + maxHeight: 1200, + everyNthFrame: 1, + }); + }); + const startRecording = Effect.fn("PreviewManager.startRecording")(function* (tabId: string) { const recordingTabId = yield* Ref.get(recordingTabIdRef); if (Option.isSome(recordingTabId) && recordingTabId.value !== tabId) { @@ -1410,18 +1439,7 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function ); } const wc = yield* requireWebContents(tabId); - yield* withControlSession(tabId, wc, "recording.start", (send) => - Effect.gen(function* () { - yield* send("Page.enable"); - yield* send("Page.startScreencast", { - format: "jpeg", - quality: 80, - maxWidth: 1600, - maxHeight: 1200, - everyNthFrame: 1, - }); - }), - ); + yield* withControlSession(tabId, wc, "recording.start", startScreencast); yield* Ref.set(recordingTabIdRef, Option.some(tabId)); }); @@ -1493,104 +1511,108 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function }; }); + const captureAutomationSnapshot = Effect.fn("PreviewManager.captureAutomationSnapshot")( + function* (tabId: string, wc: Electron.WebContents, send: SendCommand) { + yield* Effect.all([send("Runtime.enable"), send("Accessibility.enable")], { + concurrency: 2, + discard: true, + }); + const page = yield* evaluateWithDebugger<{ + url: string; + title: string; + loading: boolean; + visibleText: string; + interactiveElements: PreviewAutomationSnapshot["interactiveElements"]; + }>( + send, + `(() => { + const selectorFor = (element) => { + if (element.id) return "#" + CSS.escape(element.id); + for (const attribute of ["data-testid", "name"]) { + const value = element.getAttribute(attribute); + if (value) return element.tagName.toLowerCase() + "[" + attribute + "=" + JSON.stringify(value) + "]"; + } + const buildParts = (current, parts = []) => { + if (!current || current.nodeType !== Node.ELEMENT_NODE || parts.length >= 8) { + return parts; + } + const parent = current.parentElement; + const siblings = parent + ? Array.from(parent.children).filter((child) => child.tagName === current.tagName) + : []; + const base = current.tagName.toLowerCase(); + const part = siblings.length > 1 + ? base + ":nth-of-type(" + (siblings.indexOf(current) + 1) + ")" + : base; + return buildParts(parent, [part, ...parts]); + }; + return buildParts(element).join(" > "); + }; + const visible = (element) => { + const style = getComputedStyle(element); + const rect = element.getBoundingClientRect(); + return style.visibility !== "hidden" && style.display !== "none" && rect.width > 0 && rect.height > 0; + }; + const elements = Array.from(document.querySelectorAll( + "a[href],button,input,textarea,select,[role],[tabindex]" + )).filter(visible).slice(0, ${MAX_INTERACTIVE_ELEMENTS}).map((element) => { + const rect = element.getBoundingClientRect(); + return { + tag: element.tagName.toLowerCase(), + role: element.getAttribute("role"), + name: element.getAttribute("aria-label") || element.innerText || element.getAttribute("name") || "", + selector: selectorFor(element), + x: rect.x, + y: rect.y, + width: rect.width, + height: rect.height + }; + }); + return { + url: location.href, + title: document.title, + loading: document.readyState !== "complete", + visibleText: (document.body?.innerText || "").slice(0, ${MAX_VISIBLE_TEXT_LENGTH}), + interactiveElements: elements + }; + })()`, + true, + ); + const [accessibility, sourceImage, diagnostics, timelines] = yield* Effect.all([ + send("Accessibility.getFullAXTree"), + attemptPromise("automationSnapshot.capturePage", () => wc.capturePage()), + Ref.get(diagnosticsRef), + Ref.get(actionTimelineRef), + ]); + const sourceSize = sourceImage.getSize(); + const image = + sourceSize.width > MAX_SCREENSHOT_WIDTH + ? sourceImage.resize({ width: MAX_SCREENSHOT_WIDTH }) + : sourceImage; + const size = image.getSize(); + const browserDiagnostics = diagnostics.get(wc.id); + return { + ...page, + accessibilityTree: accessibility, + consoleEntries: [...(browserDiagnostics?.consoleEntries ?? [])], + networkEntries: [...(browserDiagnostics?.networkEntries ?? [])], + actionTimeline: [...(timelines.get(tabId) ?? [])], + screenshot: { + mimeType: "image/png" as const, + data: image.toPNG().toString("base64"), + width: size.width, + height: size.height, + }, + }; + }, + ); + const automationSnapshot = Effect.fn("PreviewManager.automationSnapshot")(function* ( tabId: string, ) { const wc = yield* requireWebContents(tabId); return yield* withControlSession(tabId, wc, "snapshot", (send) => - Effect.gen(function* () { - yield* Effect.all([send("Runtime.enable"), send("Accessibility.enable")], { - concurrency: 2, - discard: true, - }); - const page = yield* evaluateWithDebugger<{ - url: string; - title: string; - loading: boolean; - visibleText: string; - interactiveElements: PreviewAutomationSnapshot["interactiveElements"]; - }>( - send, - `(() => { - const selectorFor = (element) => { - if (element.id) return "#" + CSS.escape(element.id); - for (const attribute of ["data-testid", "name"]) { - const value = element.getAttribute(attribute); - if (value) return element.tagName.toLowerCase() + "[" + attribute + "=" + JSON.stringify(value) + "]"; - } - const buildParts = (current, parts = []) => { - if (!current || current.nodeType !== Node.ELEMENT_NODE || parts.length >= 8) { - return parts; - } - const parent = current.parentElement; - const siblings = parent - ? Array.from(parent.children).filter((child) => child.tagName === current.tagName) - : []; - const base = current.tagName.toLowerCase(); - const part = siblings.length > 1 - ? base + ":nth-of-type(" + (siblings.indexOf(current) + 1) + ")" - : base; - return buildParts(parent, [part, ...parts]); - }; - return buildParts(element).join(" > "); - }; - const visible = (element) => { - const style = getComputedStyle(element); - const rect = element.getBoundingClientRect(); - return style.visibility !== "hidden" && style.display !== "none" && rect.width > 0 && rect.height > 0; - }; - const elements = Array.from(document.querySelectorAll( - "a[href],button,input,textarea,select,[role],[tabindex]" - )).filter(visible).slice(0, ${MAX_INTERACTIVE_ELEMENTS}).map((element) => { - const rect = element.getBoundingClientRect(); - return { - tag: element.tagName.toLowerCase(), - role: element.getAttribute("role"), - name: element.getAttribute("aria-label") || element.innerText || element.getAttribute("name") || "", - selector: selectorFor(element), - x: rect.x, - y: rect.y, - width: rect.width, - height: rect.height - }; - }); - return { - url: location.href, - title: document.title, - loading: document.readyState !== "complete", - visibleText: (document.body?.innerText || "").slice(0, ${MAX_VISIBLE_TEXT_LENGTH}), - interactiveElements: elements - }; - })()`, - true, - ); - const [accessibility, sourceImage, diagnostics, timelines] = yield* Effect.all([ - send("Accessibility.getFullAXTree"), - attemptPromise("automationSnapshot.capturePage", () => wc.capturePage()), - Ref.get(diagnosticsRef), - Ref.get(actionTimelineRef), - ]); - const sourceSize = sourceImage.getSize(); - const image = - sourceSize.width > MAX_SCREENSHOT_WIDTH - ? sourceImage.resize({ width: MAX_SCREENSHOT_WIDTH }) - : sourceImage; - const size = image.getSize(); - const browserDiagnostics = diagnostics.get(wc.id); - return { - ...page, - accessibilityTree: accessibility, - consoleEntries: [...(browserDiagnostics?.consoleEntries ?? [])], - networkEntries: [...(browserDiagnostics?.networkEntries ?? [])], - actionTimeline: [...(timelines.get(tabId) ?? [])], - screenshot: { - mimeType: "image/png" as const, - data: image.toPNG().toString("base64"), - width: size.width, - height: size.height, - }, - }; - }), + captureAutomationSnapshot(tabId, wc, send), ); }); @@ -1657,66 +1679,72 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function ); }); + const performAutomationClick = Effect.fn("PreviewManager.performAutomationClick")(function* ( + tabId: string, + input: PreviewAutomationClickInput, + send: SendCommand, + ) { + yield* Effect.all( + [send("Runtime.enable"), send("Input.setIgnoreInputEvents", { ignore: false })], + { concurrency: 2, discard: true }, + ); + const point = yield* resolveClickPoint(send, input); + const viewport = yield* evaluateWithDebugger<{ width: number; height: number }>( + send, + "({ width: window.innerWidth, height: window.innerHeight })", + true, + ); + if (point.x < 0 || point.y < 0 || point.x > viewport.width || point.y > viewport.height) { + return yield* fail( + "automationClick", + automationError( + "PreviewAutomationExecutionError", + `Click coordinates (${point.x}, ${point.y}) are outside the preview viewport.`, + ), + ); + } + const moveSequence = yield* nextCounter(pointerSequenceRef); + const moveCreatedAt = yield* currentIso; + yield* emitPointerEvent({ + tabId, + phase: "move", + ...point, + sequence: moveSequence, + createdAt: moveCreatedAt, + }); + yield* Effect.sleep(AGENT_CURSOR_MOVE_MS); + const clickSequence = yield* nextCounter(pointerSequenceRef); + const clickCreatedAt = yield* currentIso; + yield* emitPointerEvent({ + tabId, + phase: "click", + ...point, + sequence: clickSequence, + createdAt: clickCreatedAt, + }); + yield* Effect.sleep(AGENT_CURSOR_CLICK_LEAD_MS); + yield* expectAgentInput(tabId, { kind: "pointer", ...point, button: 0 }); + yield* send("Input.dispatchMouseEvent", { + type: "mousePressed", + ...point, + button: "left", + clickCount: 1, + }); + yield* send("Input.dispatchMouseEvent", { + type: "mouseReleased", + ...point, + button: "left", + clickCount: 1, + }); + }); + const automationClick = Effect.fn("PreviewManager.automationClick")(function* ( tabId: string, input: PreviewAutomationClickInput, ) { const wc = yield* requireWebContents(tabId); yield* withControlSession(tabId, wc, "click", (send) => - Effect.gen(function* () { - yield* Effect.all( - [send("Runtime.enable"), send("Input.setIgnoreInputEvents", { ignore: false })], - { concurrency: 2, discard: true }, - ); - const point = yield* resolveClickPoint(send, input); - const viewport = yield* evaluateWithDebugger<{ width: number; height: number }>( - send, - "({ width: window.innerWidth, height: window.innerHeight })", - true, - ); - if (point.x < 0 || point.y < 0 || point.x > viewport.width || point.y > viewport.height) { - return yield* fail( - "automationClick", - automationError( - "PreviewAutomationExecutionError", - `Click coordinates (${point.x}, ${point.y}) are outside the preview viewport.`, - ), - ); - } - const moveSequence = yield* nextCounter(pointerSequenceRef); - const moveCreatedAt = yield* currentIso; - yield* emitPointerEvent({ - tabId, - phase: "move", - ...point, - sequence: moveSequence, - createdAt: moveCreatedAt, - }); - yield* Effect.sleep(AGENT_CURSOR_MOVE_MS); - const clickSequence = yield* nextCounter(pointerSequenceRef); - const clickCreatedAt = yield* currentIso; - yield* emitPointerEvent({ - tabId, - phase: "click", - ...point, - sequence: clickSequence, - createdAt: clickCreatedAt, - }); - yield* Effect.sleep(AGENT_CURSOR_CLICK_LEAD_MS); - yield* expectAgentInput(tabId, { kind: "pointer", ...point, button: 0 }); - yield* send("Input.dispatchMouseEvent", { - type: "mousePressed", - ...point, - button: "left", - clickCount: 1, - }); - yield* send("Input.dispatchMouseEvent", { - type: "mouseReleased", - ...point, - button: "left", - clickCount: 1, - }); - }), + performAutomationClick(tabId, input, send), ); }); @@ -1769,171 +1797,188 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function } }); + const performAutomationType = Effect.fn("PreviewManager.performAutomationType")(function* ( + tabId: string, + input: PreviewAutomationTypeInput, + send: SendCommand, + ) { + yield* send("Runtime.enable"); + yield* focusAutomationTarget(send, input); + yield* send("Input.insertText", { text: input.text }); + const textJson = yield* encodeJson("automationType.encodeText", input.text); + yield* evaluateWithDebugger( + send, + `(() => { + const element = document.activeElement; + element?.dispatchEvent(new InputEvent("input", { bubbles: true, inputType: "insertText", data: ${textJson} })); + element?.dispatchEvent(new Event("change", { bubbles: true })); + })()`, + false, + ); + }); + const automationType = Effect.fn("PreviewManager.automationType")(function* ( tabId: string, input: PreviewAutomationTypeInput, ) { const wc = yield* requireWebContents(tabId); yield* withControlSession(tabId, wc, "type", (send) => - Effect.gen(function* () { - yield* send("Runtime.enable"); - yield* focusAutomationTarget(send, input); - yield* send("Input.insertText", { text: input.text }); - const textJson = yield* encodeJson("automationType.encodeText", input.text); - yield* evaluateWithDebugger( - send, - `(() => { - const element = document.activeElement; - element?.dispatchEvent(new InputEvent("input", { bubbles: true, inputType: "insertText", data: ${textJson} })); - element?.dispatchEvent(new Event("change", { bubbles: true })); - })()`, - false, - ); - }), + performAutomationType(tabId, input, send), ); }); + const performAutomationPress = Effect.fn("PreviewManager.performAutomationPress")(function* ( + tabId: string, + input: PreviewAutomationPressInput, + send: SendCommand, + ) { + const modifiers = (input.modifiers ?? []).reduce((value, modifier) => { + switch (modifier) { + case "Alt": + return value | 1; + case "Control": + return value | 2; + case "Meta": + return value | 4; + case "Shift": + return value | 8; + } + }, 0); + const key = input.key; + const text = key.length === 1 ? key : undefined; + const params = { + key, + code: key.length === 1 ? `Key${key.toUpperCase()}` : key, + modifiers, + ...(text ? { text, unmodifiedText: text } : {}), + }; + yield* expectAgentInput(tabId, { kind: "key", key, code: params.code }); + yield* send("Input.dispatchKeyEvent", { type: "keyDown", ...params }); + yield* send("Input.dispatchKeyEvent", { type: "keyUp", ...params }); + }); + const automationPress = Effect.fn("PreviewManager.automationPress")(function* ( tabId: string, input: PreviewAutomationPressInput, ) { const wc = yield* requireWebContents(tabId); yield* withControlSession(tabId, wc, "press", (send) => - Effect.gen(function* () { - const modifiers = (input.modifiers ?? []).reduce((value, modifier) => { - switch (modifier) { - case "Alt": - return value | 1; - case "Control": - return value | 2; - case "Meta": - return value | 4; - case "Shift": - return value | 8; - } - }, 0); - const key = input.key; - const text = key.length === 1 ? key : undefined; - const params = { - key, - code: key.length === 1 ? `Key${key.toUpperCase()}` : key, - modifiers, - ...(text ? { text, unmodifiedText: text } : {}), - }; - yield* expectAgentInput(tabId, { kind: "key", key, code: params.code }); - yield* send("Input.dispatchKeyEvent", { type: "keyDown", ...params }); - yield* send("Input.dispatchKeyEvent", { type: "keyUp", ...params }); - }), + performAutomationPress(tabId, input, send), ); }); + const performAutomationScroll = Effect.fn("PreviewManager.performAutomationScroll")(function* ( + tabId: string, + input: PreviewAutomationScrollInput, + send: SendCommand, + ) { + yield* send("Runtime.enable"); + const locator = automationLocator(input); + if (locator) yield* ensurePlaywrightInjected(send); + const locatorJson = locator + ? yield* encodeJson("automationScroll.encodeLocator", locator) + : null; + const result = yield* evaluateWithDebugger< + { ok: true } | { invalidSelector: true; message: string } | { notFound: true } + >( + send, + `(() => { + try { + const target = ${locatorJson ? `(() => { const injected = globalThis.__t3PlaywrightInjected; return injected.querySelector(injected.parseSelector(${locatorJson}), document, true); })()` : "window"}; + if (!target) return { notFound: true }; + target.scrollBy({ left: ${input.deltaX ?? 0}, top: ${input.deltaY ?? 0}, behavior: "instant" }); + return { ok: true }; + } catch (error) { + return { invalidSelector: true, message: String(error) }; + } + })()`, + true, + ); + if ("invalidSelector" in result) { + return yield* fail( + "automationScroll", + automationError("PreviewAutomationInvalidSelectorError", result.message, { + selector: input.selector ?? "", + }), + ); + } + if ("notFound" in result) { + return yield* fail( + "automationScroll", + automationError( + "PreviewAutomationExecutionError", + `No element matches locator ${locator}.`, + ), + ); + } + }); + const automationScroll = Effect.fn("PreviewManager.automationScroll")(function* ( tabId: string, input: PreviewAutomationScrollInput, ) { const wc = yield* requireWebContents(tabId); yield* withControlSession(tabId, wc, "scroll", (send) => - Effect.gen(function* () { - yield* send("Runtime.enable"); - const locator = automationLocator(input); - if (locator) yield* ensurePlaywrightInjected(send); - const locatorJson = locator - ? yield* encodeJson("automationScroll.encodeLocator", locator) - : null; - const result = yield* evaluateWithDebugger< - { ok: true } | { invalidSelector: true; message: string } | { notFound: true } - >( - send, - `(() => { - try { - const target = ${locatorJson ? `(() => { const injected = globalThis.__t3PlaywrightInjected; return injected.querySelector(injected.parseSelector(${locatorJson}), document, true); })()` : "window"}; - if (!target) return { notFound: true }; - target.scrollBy({ left: ${input.deltaX ?? 0}, top: ${input.deltaY ?? 0}, behavior: "instant" }); - return { ok: true }; - } catch (error) { - return { invalidSelector: true, message: String(error) }; - } - })()`, - true, - ); - if ("invalidSelector" in result) { - return yield* fail( - "automationScroll", - automationError("PreviewAutomationInvalidSelectorError", result.message, { - selector: input.selector ?? "", - }), - ); - } - if ("notFound" in result) { - return yield* fail( - "automationScroll", - automationError( - "PreviewAutomationExecutionError", - `No element matches locator ${locator}.`, - ), - ); - } - }), + performAutomationScroll(tabId, input, send), ); }); + const performAutomationEvaluate = Effect.fn("PreviewManager.performAutomationEvaluate")( + function* (input: PreviewAutomationEvaluateInput, send: SendCommand) { + yield* send("Runtime.enable"); + const value = yield* evaluateWithDebugger( + send, + input.expression, + input.returnByValue ?? true, + input.awaitPromise ?? true, + ); + const serialized = yield* encodeJson("automationEvaluate.encodeResult", value); + if (Buffer.byteLength(serialized, "utf8") > MAX_EVALUATION_BYTES) { + return yield* fail( + "automationEvaluate", + automationError( + "PreviewAutomationResultTooLargeError", + `Evaluation result exceeds ${MAX_EVALUATION_BYTES} bytes.`, + { maximumBytes: MAX_EVALUATION_BYTES }, + ), + ); + } + return value; + }, + ); + const automationEvaluate = Effect.fn("PreviewManager.automationEvaluate")(function* ( tabId: string, input: PreviewAutomationEvaluateInput, ) { const wc = yield* requireWebContents(tabId); return yield* withControlSession(tabId, wc, "evaluate", (send) => - Effect.gen(function* () { - yield* send("Runtime.enable"); - const value = yield* evaluateWithDebugger( - send, - input.expression, - input.returnByValue ?? true, - input.awaitPromise ?? true, - ); - const serialized = yield* encodeJson("automationEvaluate.encodeResult", value); - if (Buffer.byteLength(serialized, "utf8") > MAX_EVALUATION_BYTES) { - return yield* fail( - "automationEvaluate", - automationError( - "PreviewAutomationResultTooLargeError", - `Evaluation result exceeds ${MAX_EVALUATION_BYTES} bytes.`, - { maximumBytes: MAX_EVALUATION_BYTES }, - ), - ); - } - return value; - }), + performAutomationEvaluate(input, send), ); }); - const automationWaitFor = Effect.fn("PreviewManager.automationWaitFor")(function* ( - tabId: string, + const performAutomationWaitFor = Effect.fn("PreviewManager.performAutomationWaitFor")(function* ( input: PreviewAutomationWaitForInput, + send: SendCommand, ) { - const wc = yield* requireWebContents(tabId); const timeoutMs = input.timeoutMs ?? 15_000; - yield* withControlSession(tabId, wc, "waitFor", (send) => - Effect.gen(function* () { - yield* send("Runtime.enable"); - const locator = automationLocator(input); - if (locator) yield* ensurePlaywrightInjected(send); - const [locatorJson, textJson, urlIncludesJson] = yield* Effect.all([ - locator ? encodeJson("automationWaitFor.encodeLocator", locator) : Effect.succeed(null), - input.text - ? encodeJson("automationWaitFor.encodeText", input.text) - : Effect.succeed(null), - input.urlIncludes - ? encodeJson("automationWaitFor.encodeUrl", input.urlIncludes) - : Effect.succeed(null), - ]); - const deadline = (yield* currentMillis) + timeoutMs; - while ((yield* currentMillis) <= deadline) { - const result = yield* evaluateWithDebugger< - { matched: boolean } | { invalidSelector: true; message: string } - >( - send, - `(() => { + yield* send("Runtime.enable"); + const locator = automationLocator(input); + if (locator) yield* ensurePlaywrightInjected(send); + const [locatorJson, textJson, urlIncludesJson] = yield* Effect.all([ + locator ? encodeJson("automationWaitFor.encodeLocator", locator) : Effect.succeed(null), + input.text ? encodeJson("automationWaitFor.encodeText", input.text) : Effect.succeed(null), + input.urlIncludes + ? encodeJson("automationWaitFor.encodeUrl", input.urlIncludes) + : Effect.succeed(null), + ]); + const deadline = (yield* currentMillis) + timeoutMs; + while ((yield* currentMillis) <= deadline) { + const result = yield* evaluateWithDebugger< + { matched: boolean } | { invalidSelector: true; message: string } + >( + send, + `(() => { try { const selectorMatched = ${locatorJson ? `(() => { const injected = globalThis.__t3PlaywrightInjected; return injected.querySelector(injected.parseSelector(${locatorJson}), document, false) !== null; })()` : "true"}; const textMatched = ${ @@ -1947,27 +1992,35 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function return { invalidSelector: true, message: String(error) }; } })()`, - true, - ); - if ("invalidSelector" in result) { - return yield* fail( - "automationWaitFor", - automationError("PreviewAutomationInvalidSelectorError", result.message, { - selector: input.selector ?? "", - }), - ); - } - if (result.matched) return; - yield* Effect.sleep(100); - } + true, + ); + if ("invalidSelector" in result) { return yield* fail( "automationWaitFor", - automationError( - "PreviewAutomationTimeoutError", - `Preview condition did not match within ${timeoutMs}ms.`, - ), + automationError("PreviewAutomationInvalidSelectorError", result.message, { + selector: input.selector ?? "", + }), ); - }), + } + if (result.matched) return; + yield* Effect.sleep(100); + } + return yield* fail( + "automationWaitFor", + automationError( + "PreviewAutomationTimeoutError", + `Preview condition did not match within ${timeoutMs}ms.`, + ), + ); + }); + + const automationWaitFor = Effect.fn("PreviewManager.automationWaitFor")(function* ( + tabId: string, + input: PreviewAutomationWaitForInput, + ) { + const wc = yield* requireWebContents(tabId); + yield* withControlSession(tabId, wc, "waitFor", (send) => + performAutomationWaitFor(input, send), ); }); diff --git a/apps/server/src/mcp/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index ebf8bf00c1c0ca40e64489e83ad5b026c06d5852..f60652609f5cf563a67078933796c76e6ea4035c 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -158,7 +158,7 @@ it.effect("registers annotated tools and preserves authenticated request context Effect.provideService(McpSchema.McpServerClient, client), ); expect(press.isError).toBe(false); - expect(press.structuredContent).toBeUndefined(); + expect(press.structuredContent).toBeNull(); expect(press.content).toEqual([{ type: "text", text: "null" }]); }), ).pipe(Effect.provide(TestLayer)), diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 3ae91bf8baf2250b94e8270cc0d25b5c8def3fb5..6cde2017a9e415be2fab024601444fca738a42be 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -13,8 +13,15 @@ import packageJson from "../../package.json" with { type: "json" }; import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as McpSessionRegistry from "./McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; -import { PreviewToolkitHandlersLive } from "./toolkits/preview/handlers.ts"; -import { PreviewToolkit } from "./toolkits/preview/tools.ts"; +import { + PreviewSnapshotToolkitHandlersLive, + PreviewStandardToolkitHandlersLive, +} from "./toolkits/preview/handlers.ts"; +import { + PreviewSnapshotTool, + PreviewSnapshotToolkit, + PreviewStandardToolkit, +} from "./toolkits/preview/tools.ts"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -81,97 +88,95 @@ const McpAuthMiddlewareLive = HttpRouter.middleware<{ provides: McpInvocationContext.McpInvocationContext; }>()(makeMcpAuthMiddleware).layer; -const registerPreviewToolkit = Effect.fn("McpHttpServer.registerPreviewToolkit")(function* () { +const registerPreviewSnapshot = Effect.fn("McpHttpServer.registerPreviewSnapshot")(function* () { const server = yield* McpServer.McpServer; - const built = yield* PreviewToolkit; - const handleTool = built.handle as unknown as ( - name: keyof typeof built.tools, - payload: unknown, - ) => Effect.Effect< - Stream.Stream<{ readonly encodedResult: unknown }, Error>, - Error, - McpInvocationContext.McpInvocationContext - >; - for (const tool of Object.values(built.tools)) { - yield* server.addTool({ - tool: new McpSchema.Tool({ - name: tool.name, - description: Tool.getDescription(tool), - inputSchema: Tool.getJsonSchema(tool), - annotations: { - ...Context.getOption(tool.annotations, Tool.Title).pipe( - Option.map((title) => ({ title })), - Option.getOrUndefined, - ), - readOnlyHint: Context.get(tool.annotations, Tool.Readonly), - destructiveHint: Context.get(tool.annotations, Tool.Destructive), - idempotentHint: Context.get(tool.annotations, Tool.Idempotent), - openWorldHint: Context.get(tool.annotations, Tool.OpenWorld), - }, - }), - annotations: tool.annotations, - handle: (payload) => - handleTool(tool.name as keyof typeof built.tools, payload).pipe( + const broker = yield* PreviewAutomationBroker.PreviewAutomationBroker; + const built = yield* PreviewSnapshotToolkit; + const tool = PreviewSnapshotTool; + yield* server.addTool({ + tool: new McpSchema.Tool({ + name: tool.name, + description: Tool.getDescription(tool), + inputSchema: Tool.getJsonSchema(tool), + annotations: { + ...Context.getOption(tool.annotations, Tool.Title).pipe( + Option.map((title) => ({ title })), + Option.getOrUndefined, + ), + readOnlyHint: Context.get(tool.annotations, Tool.Readonly), + destructiveHint: Context.get(tool.annotations, Tool.Destructive), + idempotentHint: Context.get(tool.annotations, Tool.Idempotent), + openWorldHint: Context.get(tool.annotations, Tool.OpenWorld), + }, + }), + annotations: tool.annotations, + handle: (payload) => + Effect.withFiber((fiber) => { + const invocation = Context.getUnsafe( + fiber.context, + McpInvocationContext.McpInvocationContext, + ); + return built.handle("preview_snapshot", payload).pipe( Stream.unwrap, Stream.run(Sink.last()), Effect.flatMap(Effect.fromOption), + Effect.provideService(PreviewAutomationBroker.PreviewAutomationBroker, broker), + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), Effect.matchCause({ onFailure: (cause) => new McpSchema.CallToolResult({ isError: true, content: [{ type: "text", text: Cause.pretty(cause) }], }), - onSuccess: (result) => { - if (tool.name === "preview_snapshot") { - const snapshot = result.encodedResult as { - readonly screenshot: { - readonly mimeType: "image/png"; - readonly data: string; - readonly width: number; - readonly height: number; - }; - readonly [key: string]: unknown; + onSuccess: ({ encodedResult }) => { + const snapshot = encodedResult as { + readonly screenshot: { + readonly mimeType: "image/png"; + readonly data: string; + readonly width: number; + readonly height: number; }; - const { screenshot, ...page } = snapshot; - const metadata = { - ...page, - screenshot: { - mimeType: screenshot.mimeType, - width: screenshot.width, - height: screenshot.height, - }, - }; - return new McpSchema.CallToolResult({ - isError: false, - structuredContent: metadata, - content: [ - { type: "text", text: JSON.stringify(metadata) }, - { - type: "image", - data: new Uint8Array(Buffer.from(screenshot.data, "base64")), - mimeType: screenshot.mimeType, - }, - ], - }); - } - const encodedResultText = JSON.stringify(result.encodedResult) ?? "null"; + readonly [key: string]: unknown; + }; + const { screenshot, ...page } = snapshot; + const metadata = { + ...page, + screenshot: { + mimeType: screenshot.mimeType, + width: screenshot.width, + height: screenshot.height, + }, + }; return new McpSchema.CallToolResult({ isError: false, - structuredContent: - result.encodedResult !== null && typeof result.encodedResult === "object" - ? result.encodedResult - : undefined, - content: [{ type: "text", text: encodedResultText }], + structuredContent: metadata, + content: [ + { type: "text", text: JSON.stringify(metadata) }, + { + type: "image", + data: new Uint8Array(Buffer.from(screenshot.data, "base64")), + mimeType: screenshot.mimeType, + }, + ], }); }, }), - ) as unknown as Effect.Effect, - }); - } + ); + }), + }); }); -export const PreviewToolkitRegistrationLive = Layer.effectDiscard(registerPreviewToolkit()).pipe( - Layer.provide(PreviewToolkitHandlersLive), +const PreviewStandardToolkitRegistrationLive = McpServer.toolkit(PreviewStandardToolkit).pipe( + Layer.provide(PreviewStandardToolkitHandlersLive), +); + +const PreviewSnapshotRegistrationLive = Layer.effectDiscard(registerPreviewSnapshot()).pipe( + Layer.provide(PreviewSnapshotToolkitHandlersLive), +); + +export const PreviewToolkitRegistrationLive = Layer.mergeAll( + PreviewStandardToolkitRegistrationLive, + PreviewSnapshotRegistrationLive, ); const McpTransportLive = McpServer.layerHttp({ diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index 161b560f7465bd89db43b61e06421dfdaf1cdb78..1ee7d278c62889d09bb02f4532a9773e7523698f 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -1,9 +1,10 @@ import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; -import * as Ref from "effect/Ref"; +import * as SynchronizedRef from "effect/SynchronizedRef"; import { HttpServer } from "effect/unstable/http"; import { ServerEnvironment } from "../environment/Services/ServerEnvironment.ts"; @@ -59,15 +60,15 @@ const bytesToHex = (bytes: Uint8Array): string => const tokenFromBytes = (bytes: Uint8Array): string => Buffer.from(bytes).toString("base64url"); -const make = Effect.fn("McpSessionRegistry.make")(function* ( +const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( options: McpSessionRegistryOptions = {}, ) { const crypto = yield* Crypto.Crypto; const environment = yield* ServerEnvironment; const environmentId = yield* environment.getEnvironmentId; const httpServer = yield* HttpServer.HttpServer; - const state = yield* Ref.make({ records: new Map() }); - const now = options.now ?? Date.now; + const state = yield* SynchronizedRef.make({ records: new Map() }); + const currentTimeMillis = options.now ? Effect.sync(options.now) : Clock.currentTimeMillis; const idleTimeoutMs = options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS; const maximumLifetimeMs = options.maximumLifetimeMs ?? DEFAULT_MAXIMUM_LIFETIME_MS; const endpoint = @@ -81,21 +82,18 @@ const make = Effect.fn("McpSessionRegistry.make")(function* ( .pipe(Effect.map(bytesToHex), Effect.orDie); const pruneExpired = (records: ReadonlyMap, timestamp: number) => { - let changed = false; - const next = new Map(); - for (const [hash, record] of records) { - if (timestamp <= record.scope.expiresAt && timestamp - record.lastUsedAt <= idleTimeoutMs) { - next.set(hash, record); - } else { - changed = true; - } - } - return changed ? next : records; + const next = new Map( + Array.from(records).filter( + ([, record]) => + timestamp <= record.scope.expiresAt && timestamp - record.lastUsedAt <= idleTimeoutMs, + ), + ); + return next.size === records.size ? records : next; }; const issue: McpSessionRegistryShape["issue"] = Effect.fn("McpSessionRegistry.issue")( function* (request) { - const issuedAt = now(); + const issuedAt = yield* currentTimeMillis; const providerSessionId = yield* crypto.randomUUIDv4.pipe(Effect.orDie); const rawToken = yield* crypto.randomBytes(32).pipe(Effect.map(tokenFromBytes), Effect.orDie); const tokenHash = yield* hashToken(rawToken); @@ -109,7 +107,7 @@ const make = Effect.fn("McpSessionRegistry.make")(function* ( issuedAt, expiresAt, }; - yield* Ref.update(state, ({ records }) => { + yield* SynchronizedRef.update(state, ({ records }) => { const next = new Map(pruneExpired(records, issuedAt)); next.set(tokenHash, { tokenHash, scope, lastUsedAt: issuedAt }); return { records: next }; @@ -132,23 +130,20 @@ const make = Effect.fn("McpSessionRegistry.make")(function* ( function* (rawToken) { if (rawToken.length === 0) return undefined; const tokenHash = yield* hashToken(rawToken); - const timestamp = now(); - let resolved: McpInvocationContext.McpInvocationScope | undefined; - yield* Ref.update(state, ({ records }) => { + const timestamp = yield* currentTimeMillis; + return yield* SynchronizedRef.modify(state, ({ records }) => { const current = pruneExpired(records, timestamp); const record = current.get(tokenHash); - if (!record) return { records: current }; - resolved = record.scope; + if (!record) return [undefined, { records: current }] as const; const next = new Map(current); next.set(tokenHash, { ...record, lastUsedAt: timestamp }); - return { records: next }; + return [record.scope, { records: next }] as const; }); - return resolved; }, ); const revokeWhere = (predicate: (record: CredentialRecord) => boolean) => - Ref.update(state, ({ records }) => ({ + SynchronizedRef.update(state, ({ records }) => ({ records: new Map(Array.from(records).filter(([, record]) => !predicate(record))), })); @@ -163,27 +158,34 @@ const make = Effect.fn("McpSessionRegistry.make")(function* ( revokeThread: Effect.fn("McpSessionRegistry.revokeThread")(function* (threadId) { yield* revokeWhere((record) => record.scope.threadId === threadId); }), - revokeAll: Ref.set(state, { records: new Map() }), + revokeAll: SynchronizedRef.set(state, { records: new Map() }), }); }); let activeMcpSessionRegistry: McpSessionRegistryShape | undefined; -export const layer: Layer.Layer< - McpSessionRegistry, - never, - Crypto.Crypto | ServerEnvironment | HttpServer.HttpServer -> = Layer.effect( - McpSessionRegistry, - make().pipe( +const make = Effect.acquireRelease( + makeWithOptions().pipe( Effect.tap((registry) => Effect.sync(() => { activeMcpSessionRegistry = registry; }), ), ), + (registry) => + Effect.sync(() => { + if (activeMcpSessionRegistry === registry) { + activeMcpSessionRegistry = undefined; + } + }), ); +export const layer: Layer.Layer< + McpSessionRegistry, + never, + Crypto.Crypto | ServerEnvironment | HttpServer.HttpServer +> = Layer.effect(McpSessionRegistry, make); + export const issueActiveMcpCredential = ( request: McpCredentialRequest, ): Effect.Effect => @@ -201,5 +203,5 @@ export const revokeAllActiveMcpCredentials = (): Effect.Effect => /** Exposed for tests. */ export const __testing = { - make, + make: makeWithOptions, }; diff --git a/apps/server/src/mcp/PreviewAutomationBroker.test.ts b/apps/server/src/mcp/PreviewAutomationBroker.test.ts index 7d702cd970bc9e7eb2447241eed63dcd009f247a..353353aaef2fb2bac79ada74187d459e2e520626 100644 --- a/apps/server/src/mcp/PreviewAutomationBroker.test.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.test.ts @@ -23,7 +23,7 @@ const scope = { it.effect("routes a request to the focused owner and correlates its response", () => Effect.scoped( Effect.gen(function* () { - const broker = yield* PreviewAutomationBroker.__testing.make(); + const broker = yield* PreviewAutomationBroker.__testing.make; const requests = yield* broker.connect("client-1"); yield* Stream.runForEach(requests, (request) => broker.respond({ @@ -56,7 +56,7 @@ it.effect("routes a request to the focused owner and correlates its response", ( it.effect("rejects calls when no focused owner exists", () => Effect.gen(function* () { - const broker = yield* PreviewAutomationBroker.__testing.make(); + const broker = yield* PreviewAutomationBroker.__testing.make; const error = yield* broker .invoke({ scope, operation: "status", input: {} }) .pipe(Effect.flip); @@ -67,7 +67,7 @@ it.effect("rejects calls when no focused owner exists", () => it.effect("routes interactive commands to a hidden durable browser host", () => Effect.scoped( Effect.gen(function* () { - const broker = yield* PreviewAutomationBroker.__testing.make(); + const broker = yield* PreviewAutomationBroker.__testing.make; const requests = yield* broker.connect("client-hidden"); yield* Stream.runForEach(requests, (request) => broker.respond({ requestId: request.requestId, ok: true }), diff --git a/apps/server/src/mcp/PreviewAutomationBroker.ts b/apps/server/src/mcp/PreviewAutomationBroker.ts index a8ae4a9eec4dec9da3ffcde8cc2fa7fde412b416..e0a7b0c9285c7e3c8269a4da4adbf9aa953af153 100644 --- a/apps/server/src/mcp/PreviewAutomationBroker.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.ts @@ -21,8 +21,8 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; -import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as SynchronizedRef from "effect/SynchronizedRef"; import * as McpInvocationContext from "./McpInvocationContext.ts"; @@ -71,6 +71,7 @@ interface BrokerState { readonly clients: ReadonlyMap; readonly owners: ReadonlyMap; readonly pending: ReadonlyMap; + readonly requestSequence: number; } const makeResponseError = ( @@ -119,33 +120,35 @@ const makeResponseError = ( } }; -const make = Effect.fn("PreviewAutomationBroker.make")(function* () { - const state = yield* Ref.make({ +const make = Effect.gen(function* PreviewAutomationBrokerMake() { + const state = yield* SynchronizedRef.make({ clients: new Map(), owners: new Map(), pending: new Map(), + requestSequence: 0, }); - let requestSequence = 0; const disconnect = Effect.fn("PreviewAutomationBroker.disconnect")(function* ( clientId: string, queue: ClientConnection["queue"], ) { - const toFail: PendingRequest[] = []; - yield* Ref.update(state, (current) => { - if (current.clients.get(clientId)?.queue !== queue) return current; + const toFail = yield* SynchronizedRef.modify(state, (current) => { + if (current.clients.get(clientId)?.queue !== queue) { + return [[] as ReadonlyArray, current] as const; + } const clients = new Map(current.clients); const owners = new Map(current.owners); const pending = new Map(current.pending); + const disconnected: PendingRequest[] = []; clients.delete(clientId); owners.delete(clientId); for (const [requestId, entry] of pending) { if (entry.clientId === clientId) { pending.delete(requestId); - toFail.push(entry); + disconnected.push(entry); } } - return { clients, owners, pending }; + return [disconnected, { ...current, clients, owners, pending }] as const; }); yield* Effect.forEach( toFail, @@ -165,12 +168,10 @@ const make = Effect.fn("PreviewAutomationBroker.make")(function* () { "PreviewAutomationBroker.connect", )(function* (clientId) { const queue = yield* Queue.unbounded(); - let previous: ClientConnection | undefined; - yield* Ref.update(state, (current) => { - previous = current.clients.get(clientId); + const previous = yield* SynchronizedRef.modify(state, (current) => { const clients = new Map(current.clients); clients.set(clientId, { clientId, queue }); - return { ...current, clients }; + return [current.clients.get(clientId), { ...current, clients }] as const; }); if (previous) yield* disconnect(clientId, previous.queue); return Stream.fromQueue(queue).pipe(Stream.ensuring(disconnect(clientId, queue))); @@ -179,7 +180,7 @@ const make = Effect.fn("PreviewAutomationBroker.make")(function* () { const reportOwner: PreviewAutomationBrokerShape["reportOwner"] = Effect.fn( "PreviewAutomationBroker.reportOwner", )(function* (owner) { - yield* Ref.update(state, (current) => { + yield* SynchronizedRef.update(state, (current) => { const owners = new Map(current.owners); owners.set(owner.clientId, owner); return { ...current, owners }; @@ -189,7 +190,7 @@ const make = Effect.fn("PreviewAutomationBroker.make")(function* () { const clearOwner: PreviewAutomationBrokerShape["clearOwner"] = Effect.fn( "PreviewAutomationBroker.clearOwner", )(function* (clientId) { - yield* Ref.update(state, (current) => { + yield* SynchronizedRef.update(state, (current) => { const owners = new Map(current.owners); owners.delete(clientId); return { ...current, owners }; @@ -199,13 +200,12 @@ const make = Effect.fn("PreviewAutomationBroker.make")(function* () { const respond: PreviewAutomationBrokerShape["respond"] = Effect.fn( "PreviewAutomationBroker.respond", )(function* (response) { - let pending: PendingRequest | undefined; - yield* Ref.update(state, (current) => { - pending = current.pending.get(response.requestId); - if (!pending) return current; + const pending = yield* SynchronizedRef.modify(state, (current) => { + const entry = current.pending.get(response.requestId); + if (!entry) return [undefined, current] as const; const next = new Map(current.pending); next.delete(response.requestId); - return { ...current, pending: next }; + return [entry, { ...current, pending: next }] as const; }); if (!pending) return; if (response.ok) { @@ -225,7 +225,7 @@ const make = Effect.fn("PreviewAutomationBroker.make")(function* () { const invoke = Effect.fn("PreviewAutomationBroker.invoke")(function* ( input: Parameters[0], ): Effect.fn.Return { - const current = yield* Ref.get(state); + const current = yield* SynchronizedRef.get(state); const candidates = Array.from(current.owners.values()) .filter( (owner) => @@ -256,48 +256,52 @@ const make = Effect.fn("PreviewAutomationBroker.make")(function* () { message: "The browser host does not have an active tab.", }); } - const requestId = `preview-${requestSequence++}`; const timeoutMs = input.timeoutMs ?? 15_000; const deferred = yield* Deferred.make(); - yield* Ref.update(state, (next) => { + const requestId = yield* SynchronizedRef.modify(state, (next) => { + const requestId = `preview-${next.requestSequence}`; const pending = new Map(next.pending); pending.set(requestId, { clientId: owner.clientId, deferred }); - return { ...next, pending }; + return [requestId, { ...next, pending, requestSequence: next.requestSequence + 1 }] as const; }); - const offered = yield* Queue.offer(connection.queue, { - requestId, - threadId: input.scope.threadId, - tabId: input.tabId ?? owner.tabId ?? undefined, - operation: input.operation, - input: input.input, - timeoutMs, - }); - if (!offered) { - return yield* new PreviewAutomationUnavailableError({ - message: "The preview automation client is no longer accepting requests.", - }); - } - const result = yield* Deferred.await(deferred).pipe(Effect.timeoutOption(timeoutMs)); - yield* Ref.update(state, (next) => { + const removePending = SynchronizedRef.update(state, (next) => { + if (!next.pending.has(requestId)) return next; const pending = new Map(next.pending); pending.delete(requestId); return { ...next, pending }; }); - return yield* Option.match(result, { - onNone: () => - Effect.fail( - new PreviewAutomationTimeoutError({ - message: `Preview automation timed out after ${timeoutMs}ms.`, - }), - ), - onSome: (value) => Effect.succeed(value as A), + const awaitResponse = Effect.fn("PreviewAutomationBroker.awaitResponse")(function* () { + const offered = yield* Queue.offer(connection.queue, { + requestId, + threadId: input.scope.threadId, + tabId: input.tabId ?? owner.tabId ?? undefined, + operation: input.operation, + input: input.input, + timeoutMs, + }); + if (!offered) { + return yield* new PreviewAutomationUnavailableError({ + message: "The preview automation client is no longer accepting requests.", + }); + } + const result = yield* Deferred.await(deferred).pipe(Effect.timeoutOption(timeoutMs)); + return yield* Option.match(result, { + onNone: () => + Effect.fail( + new PreviewAutomationTimeoutError({ + message: `Preview automation timed out after ${timeoutMs}ms.`, + }), + ), + onSome: (value) => Effect.succeed(value as A), + }); }); + return yield* awaitResponse().pipe(Effect.ensuring(removePending)); }); return PreviewAutomationBroker.of({ connect, reportOwner, clearOwner, respond, invoke }); -}); +}).pipe(Effect.withSpan("PreviewAutomationBroker.make")); -export const layer = Layer.effect(PreviewAutomationBroker, make()); +export const layer = Layer.effect(PreviewAutomationBroker, make); /** Exposed for tests. */ export const __testing = { diff --git a/apps/server/src/mcp/toolkits/preview/handlers.ts b/apps/server/src/mcp/toolkits/preview/handlers.ts index 56c9a6d98e481fcc734b797a7f2f91667cb39b03..6013b1cac9e9fc6eb4f746366687bc023ad98311 100644 --- a/apps/server/src/mcp/toolkits/preview/handlers.ts +++ b/apps/server/src/mcp/toolkits/preview/handlers.ts @@ -9,7 +9,7 @@ import type { import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts"; -import { PreviewToolkit } from "./tools.ts"; +import { PreviewSnapshotToolkit, PreviewStandardToolkit, PreviewToolkit } from "./tools.ts"; const invoke = Effect.fn("PreviewToolkit.invoke")(function* ( operation: PreviewAutomationOperation, @@ -30,7 +30,7 @@ const invoke = Effect.fn("PreviewToolkit.invoke")(function* ( }); }); -export const PreviewToolkitHandlersLive = PreviewToolkit.toLayer({ +const handlers = { preview_status: () => invoke("status", {}), preview_open: (input) => invoke("open", { @@ -40,12 +40,24 @@ export const PreviewToolkitHandlersLive = PreviewToolkit.toLayer({ }), preview_navigate: (input) => invoke("navigate", input, input.timeoutMs), preview_snapshot: () => invoke("snapshot", {}), - preview_click: (input) => invoke("click", input, input.timeoutMs), - preview_type: (input) => invoke("type", input, input.timeoutMs), - preview_press: (input) => invoke("press", input), - preview_scroll: (input) => invoke("scroll", input), - preview_evaluate: (input) => invoke("evaluate", input), - preview_wait_for: (input) => invoke("waitFor", input, input.timeoutMs), + preview_click: (input) => invoke("click", input, input.timeoutMs).pipe(Effect.as(null)), + preview_type: (input) => invoke("type", input, input.timeoutMs).pipe(Effect.as(null)), + preview_press: (input) => invoke("press", input).pipe(Effect.as(null)), + preview_scroll: (input) => invoke("scroll", input).pipe(Effect.as(null)), + preview_evaluate: (input) => + invoke("evaluate", input).pipe(Effect.map((result) => result ?? null)), + preview_wait_for: (input) => + invoke("waitFor", input, input.timeoutMs).pipe(Effect.as(null)), preview_recording_start: () => invoke("recordingStart", {}), preview_recording_stop: () => invoke("recordingStop", {}), +} satisfies Parameters[0]; + +const { preview_snapshot, ...standardHandlers } = handlers; + +export const PreviewStandardToolkitHandlersLive = PreviewStandardToolkit.toLayer(standardHandlers); + +export const PreviewSnapshotToolkitHandlersLive = PreviewSnapshotToolkit.toLayer({ + preview_snapshot, }); + +export const PreviewToolkitHandlersLive = PreviewToolkit.toLayer(handlers); diff --git a/apps/server/src/mcp/toolkits/preview/tools.ts b/apps/server/src/mcp/toolkits/preview/tools.ts index bcfe720a9504a3c4da80b3531ad40cf83ea2de97..fd2fedbb36968171b655a6bc15ff8dfb112564bb 100644 --- a/apps/server/src/mcp/toolkits/preview/tools.ts +++ b/apps/server/src/mcp/toolkits/preview/tools.ts @@ -84,6 +84,7 @@ export const PreviewClickTool = browserTool( description: "Click exactly one page target. Prefer locator with a Playwright selector such as role=button[name='Send']; selector accepts legacy CSS; x and y are viewport CSS pixels and must be supplied together. Call preview_snapshot first when the target is unknown.", parameters: PreviewAutomationClickInput, + success: Schema.Null, failure: PreviewAutomationError, dependencies, }).annotate(Tool.Title, "Click preview page"), @@ -94,6 +95,7 @@ export const PreviewTypeTool = browserTool( description: "Insert literal text into one input. Prefer locator with a Playwright role/text selector; selector accepts legacy CSS. If neither is supplied, types into the currently focused element. Set clear=true to replace existing text.", parameters: PreviewAutomationTypeInput, + success: Schema.Null, failure: PreviewAutomationError, dependencies, }).annotate(Tool.Title, "Type into preview page"), @@ -104,6 +106,7 @@ export const PreviewPressTool = browserTool( description: "Press one keyboard key in the active page, for example {key:'Enter'}, {key:'Escape'}, or {key:'a',modifiers:['Meta']}. This targets the page's current focus.", parameters: PreviewAutomationPressInput, + success: Schema.Null, failure: PreviewAutomationError, dependencies, }).annotate(Tool.Title, "Press key in preview page"), @@ -114,6 +117,7 @@ export const PreviewScrollTool = safeBrowserTool( description: "Scroll by CSS pixels. Positive deltaY scrolls down and positive deltaX scrolls right. Without locator/selector it scrolls the viewport; otherwise it scrolls that container. At least one delta is required.", parameters: PreviewAutomationScrollInput, + success: Schema.Null, failure: PreviewAutomationError, dependencies, }).annotate(Tool.Title, "Scroll preview page"), @@ -135,6 +139,7 @@ export const PreviewWaitForTool = readonlyBrowserTool( description: "Wait until all supplied conditions match: a Playwright locator, legacy CSS selector, visible-text substring, and/or URL substring. Provide at least one condition. Defaults to 15 seconds, maximum 60 seconds.", parameters: PreviewAutomationWaitForInput, + success: Schema.Null, failure: PreviewAutomationError, dependencies, }).annotate(Tool.Title, "Wait for preview page condition"), @@ -173,3 +178,19 @@ export const PreviewToolkit = Toolkit.make( PreviewRecordingStartTool, PreviewRecordingStopTool, ); + +export const PreviewStandardToolkit = Toolkit.make( + PreviewStatusTool, + PreviewOpenTool, + PreviewNavigateTool, + PreviewClickTool, + PreviewTypeTool, + PreviewPressTool, + PreviewScrollTool, + PreviewEvaluateTool, + PreviewWaitForTool, + PreviewRecordingStartTool, + PreviewRecordingStopTool, +); + +export const PreviewSnapshotToolkit = Toolkit.make(PreviewSnapshotTool); diff --git a/apps/server/src/preview/Manager.ts b/apps/server/src/preview/Manager.ts index e18bbdaf49353c46bc7723d7efefce41e345fdaf..8fa3a3668bfc51b7d7b75b47b480e48b6af1d1e8 100644 --- a/apps/server/src/preview/Manager.ts +++ b/apps/server/src/preview/Manager.ts @@ -127,7 +127,7 @@ const buildIdleSnapshot = (input: { updatedAt: input.updatedAt, }); -const make = Effect.fn("PreviewManager.make")(function* () { +const make = Effect.gen(function* PreviewManagerMake() { const stateRef = yield* SynchronizedRef.make(initialState); // Unbounded PubSub is fine here — events are tiny and we don't want to // block publishers if a subscriber is slow. WS clients backpressure on @@ -357,6 +357,6 @@ const make = Effect.fn("PreviewManager.make")(function* () { events, subscribeEvents: PubSub.subscribe(eventsPubSub), } satisfies PreviewManagerShape; -}); +}).pipe(Effect.withSpan("PreviewManager.make")); -export const layer = Layer.effect(PreviewManager, make()); +export const layer = Layer.effect(PreviewManager, make); diff --git a/apps/server/src/preview/PortScanner.test.ts b/apps/server/src/preview/PortScanner.test.ts index 350268acff6622f93c42fa1edd32a1c692397665..8b37e86d8a9396c08043d3754cd23e79f4bf29a0 100644 --- a/apps/server/src/preview/PortScanner.test.ts +++ b/apps/server/src/preview/PortScanner.test.ts @@ -249,20 +249,18 @@ effectIt.layer(TestPortDiscoveryLive)("PortDiscovery integration (TCP probe fall ); it.effect( - "retain() drives an immediate broadcast to subscribers", + "retain drives an immediate broadcast to subscribers", Effect.fn("PortScannerTest.retainBroadcastsImmediately")(function* () { yield* windowsPlatform; const { port } = yield* commonDevServer; const received: number[] = []; const scanner = yield* PortScanner.PortDiscovery; - const unsubscribe = yield* scanner.subscribe((servers) => + yield* scanner.subscribe((servers) => Effect.sync(() => { for (const server of servers) received.push(server.port); }), ); - const release = yield* scanner.retain(); - unsubscribe(); - release(); + yield* scanner.retain; expect(received).toContain(port); }), ); diff --git a/apps/server/src/preview/PortScanner.ts b/apps/server/src/preview/PortScanner.ts index d2ad4f43c3735c070fe17bb85fa4518c97e529d4..c8d9a051ed610ce5188aa1cc4da0dc88f1cb7ee1 100644 --- a/apps/server/src/preview/PortScanner.ts +++ b/apps/server/src/preview/PortScanner.ts @@ -8,13 +8,13 @@ * Windows / lsof missing: checks a curated list of common dev ports through * the shared Net service. * - * Polling is reference-counted via `retain()`. A single layer-scoped fiber + * Polling is reference-counted via scoped `retain`. A single layer-scoped fiber * polls forever, but each tick is a no-op when the retain count is zero. */ import { ThreadId, type DiscoveredLocalServer } from "@t3tools/contracts"; import * as Net from "@t3tools/shared/Net"; import { LSOF_LOCAL_HOST_TOKENS } from "@t3tools/shared/preview"; -import { Cause, Context, Duration, Effect, Layer, Ref, Schedule } from "effect"; +import { Cause, Context, Duration, Effect, Layer, Ref, Schedule, Scope } from "effect"; import { ProcessRunner } from "../processRunner.ts"; @@ -22,8 +22,8 @@ export interface PortDiscoveryShape { readonly scan: () => Effect.Effect>; readonly subscribe: ( listener: (servers: ReadonlyArray) => Effect.Effect, - ) => Effect.Effect<() => void>; - readonly retain: () => Effect.Effect<() => void>; + ) => Effect.Effect; + readonly retain: Effect.Effect; readonly registerTerminalProcesses: (input: { readonly threadId: string; readonly terminalId: string; @@ -51,6 +51,15 @@ type Listener = (servers: ReadonlyArray) => Effect.Effect interface ScannerState { readonly lastSnapshot: ReadonlyArray; + readonly listeners: ReadonlySet; + readonly terminalProcesses: ReadonlyMap< + string, + { + readonly owner: TerminalProcessOwner; + readonly processIds: ReadonlySet; + } + >; + readonly retainCount: number; } interface TerminalProcessOwner { @@ -170,24 +179,15 @@ const serversEqual = ( return true; }; -const make = Effect.fn("PortDiscovery.make")(function* () { +const make = Effect.gen(function* PortDiscoveryMake() { const net = yield* Net.NetService; const processRunner = yield* ProcessRunner; const stateRef = yield* Ref.make({ lastSnapshot: [], + listeners: new Set(), + terminalProcesses: new Map(), + retainCount: 0, }); - const listeners = new Set(); - const terminalProcesses = new Map< - string, - { - readonly owner: TerminalProcessOwner; - readonly processIds: ReadonlySet; - } - >(); - // Plain integer because the release callback returned by `retain()` runs - // outside any Effect context (the WS subscriber's release path) and must - // be a synchronous side-effect-only function. - let retainCount = 0; const probeCommonPorts = Effect.fn("PortDiscovery.probeCommonPorts")(function* () { const results = yield* Effect.forEach( @@ -214,8 +214,9 @@ const make = Effect.fn("PortDiscovery.make")(function* () { }); const scanOnce = Effect.fn("PortDiscovery.scan")(function* () { + const state = yield* Ref.get(stateRef); const terminalByProcessId = new Map(); - for (const registration of terminalProcesses.values()) { + for (const registration of state.terminalProcesses.values()) { for (const processId of registration.processIds) { terminalByProcessId.set(processId, registration.owner); } @@ -254,17 +255,23 @@ const make = Effect.fn("PortDiscovery.make")(function* () { return yield* probeCommonPorts(); }); - const broadcast = (servers: ReadonlyArray): Effect.Effect => - Effect.forEach(Array.from(listeners), (listener) => listener(servers), { discard: true }); + const broadcast = Effect.fn("PortDiscovery.broadcast")(function* ( + servers: ReadonlyArray, + ) { + const listeners = (yield* Ref.get(stateRef)).listeners; + yield* Effect.forEach(listeners, (listener) => listener(servers), { discard: true }); + }); const pollTick = Effect.fn("PortDiscovery.pollTick")( function* () { - if (retainCount <= 0) return; + if ((yield* Ref.get(stateRef)).retainCount <= 0) return; const next = yield* scanOnce(); - const state = yield* Ref.get(stateRef); - if (serversEqual(state.lastSnapshot, next)) return; - yield* Ref.update(stateRef, (s) => ({ ...s, lastSnapshot: next })); - yield* broadcast(next); + const changed = yield* Ref.modify(stateRef, (state) => + serversEqual(state.lastSnapshot, next) + ? [false, state] + : [true, { ...state, lastSnapshot: next }], + ); + if (changed) yield* broadcast(next); }, Effect.catchCause((cause: Cause.Cause) => Effect.logWarning("preview port scan failed", Cause.pretty(cause)), @@ -275,58 +282,70 @@ const make = Effect.fn("PortDiscovery.make")(function* () { // currently retained, so the cost is one Ref.get every POLL_INTERVAL. yield* Effect.forkScoped(pollTick().pipe(Effect.repeat(Schedule.spaced(POLL_INTERVAL)))); - const retain: PortDiscoveryShape["retain"] = Effect.fn("PortDiscovery.retain")(function* () { - const wasIdle = retainCount === 0; - retainCount += 1; + const acquireRetention = Effect.fn("PortDiscovery.retain")(function* () { + const wasIdle = yield* Ref.modify(stateRef, (state) => [ + state.retainCount === 0, + { ...state, retainCount: state.retainCount + 1 }, + ]); if (wasIdle) { // Run an immediate scan + broadcast so the new retainer doesn't have // to wait up to POLL_INTERVAL for the first emission. yield* pollTick(); } - let released = false; - return () => { - if (released) return; - released = true; - retainCount = Math.max(0, retainCount - 1); - }; }); + const retain: PortDiscoveryShape["retain"] = Effect.acquireRelease(acquireRetention(), () => + Ref.update(stateRef, (state) => ({ + ...state, + retainCount: Math.max(0, state.retainCount - 1), + })), + ); + const subscribe: PortDiscoveryShape["subscribe"] = Effect.fn("PortDiscovery.subscribe")( - function* (listener) { - return yield* Effect.sync(() => { - listeners.add(listener); - return () => { - listeners.delete(listener); - }; - }); - }, + (listener) => + Effect.acquireRelease( + Ref.update(stateRef, (state) => ({ + ...state, + listeners: new Set([...state.listeners, listener]), + })), + () => + Ref.update(stateRef, (state) => { + const listeners = new Set(state.listeners); + listeners.delete(listener); + return { ...state, listeners }; + }), + ), ); const registerTerminalProcesses: PortDiscoveryShape["registerTerminalProcesses"] = Effect.fn( "PortDiscovery.registerTerminalProcesses", )(function* (input) { - yield* Effect.sync(() => { - const owner = { - threadId: ThreadId.make(input.threadId), - terminalId: input.terminalId, - }; - const processIds = new Set( - input.processIds.filter((processId) => Number.isInteger(processId) && processId > 0), - ); + const owner = { + threadId: ThreadId.make(input.threadId), + terminalId: input.terminalId, + }; + const processIds = new Set( + input.processIds.filter((processId) => Number.isInteger(processId) && processId > 0), + ); + yield* Ref.update(stateRef, (state) => { + const terminalProcesses = new Map(state.terminalProcesses); const key = terminalOwnerKey(owner); if (processIds.size === 0) { terminalProcesses.delete(key); - return; + } else { + terminalProcesses.set(key, { owner, processIds }); } - terminalProcesses.set(key, { owner, processIds }); + return { ...state, terminalProcesses }; }); }); const unregisterTerminal: PortDiscoveryShape["unregisterTerminal"] = Effect.fn( "PortDiscovery.unregisterTerminal", )(function* (input) { - yield* Effect.sync(() => { + yield* Ref.update(stateRef, (state) => { + const terminalProcesses = new Map(state.terminalProcesses); terminalProcesses.delete(terminalOwnerKey(input)); + return { ...state, terminalProcesses }; }); }); @@ -337,9 +356,9 @@ const make = Effect.fn("PortDiscovery.make")(function* () { registerTerminalProcesses, unregisterTerminal, } satisfies PortDiscoveryShape; -}); +}).pipe(Effect.withSpan("PortDiscovery.make")); -export const layer = Layer.effect(PortDiscovery, make()); +export const layer = Layer.effect(PortDiscovery, make); /** Exposed for tests. */ export const __testing = { diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index b2166d6f0b224024de03b33b1371a9c4edee0c32..a1cd119a7a6d54770068b8d29bcbda011fdbefeb 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -683,8 +683,8 @@ const buildAppUnderTest = (options?: { }), Layer.mock(PortScanner.PortDiscovery)({ scan: () => Effect.succeed([]), - subscribe: () => Effect.succeed(() => {}), - retain: () => Effect.succeed(() => {}), + subscribe: () => Effect.void, + retain: Effect.void, registerTerminalProcesses: () => Effect.void, unregisterTerminal: () => Effect.void, }), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index c8c3c5de46e9c22312fa8f537015ab8c723e0e26..d35ab88fee6b366ed8458c027680544a67ba465a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1425,29 +1425,21 @@ const makeWsRpcLayer = (currentSession: AuthenticatedSession) => observeRpcStream( WS_METHODS.subscribeDiscoveredLocalServers, Stream.callback((queue) => - Effect.acquireRelease( - Effect.gen(function* () { - const release = yield* portDiscovery.retain(); - const initial = yield* portDiscovery.scan(); - const initialScannedAt = DateTime.formatIso(yield* DateTime.now); - yield* Queue.offer(queue, { - servers: initial, - scannedAt: initialScannedAt, - }); - const unsubscribe = yield* portDiscovery.subscribe((servers) => - Effect.gen(function* () { - const scannedAt = DateTime.formatIso(yield* DateTime.now); - yield* Queue.offer(queue, { servers, scannedAt }); - }), - ); - return { unsubscribe, release }; - }), - ({ unsubscribe, release }) => - Effect.sync(() => { - unsubscribe(); - release(); + Effect.gen(function* () { + yield* portDiscovery.retain; + const initial = yield* portDiscovery.scan(); + const initialScannedAt = DateTime.formatIso(yield* DateTime.now); + yield* Queue.offer(queue, { + servers: initial, + scannedAt: initialScannedAt, + }); + yield* portDiscovery.subscribe((servers) => + Effect.gen(function* () { + const scannedAt = DateTime.formatIso(yield* DateTime.now); + yield* Queue.offer(queue, { servers, scannedAt }); }), - ), + ); + }), ), { "rpc.aggregate": "preview" }, ), -- 2.54.0