From f7c422d5815e7be03415a87b30558c7bcf2cf8e9 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Fri, 12 Jun 2026 16:55:49 -0700 Subject: [PATCH] Refactor MCP services into top-level modules - Move MCP session registry and preview broker out of `Layers/` and `Services/` - Update imports, tests, and server wiring to use the new module layout --- .../src/mcp/Layers/McpSessionRegistry.ts | 173 ---------- .../src/mcp/Layers/PreviewAutomationBroker.ts | 270 ---------------- .../mcp/{Layers => }/McpHttpServer.test.ts | 28 +- .../src/mcp/{Layers => }/McpHttpServer.ts | 49 +-- .../{Services => }/McpInvocationContext.ts | 2 +- .../mcp/{Services => }/McpProviderSession.ts | 6 - .../{Layers => }/McpSessionRegistry.test.ts | 6 +- apps/server/src/mcp/McpSessionRegistry.ts | 202 ++++++++++++ .../PreviewAutomationBroker.test.ts | 8 +- .../server/src/mcp/PreviewAutomationBroker.ts | 306 ++++++++++++++++++ .../src/mcp/Services/McpSessionRegistry.ts | 29 -- .../mcp/Services/PreviewAutomationBroker.ts | 40 --- .../src/mcp/toolkits/preview/handlers.ts | 28 +- apps/server/src/mcp/toolkits/preview/tools.ts | 32 +- .../src/preview/{Layers => }/Manager.test.ts | 35 +- .../src/preview/{Layers => }/Manager.ts | 189 ++++++----- .../preview/{Layers => }/PortScanner.test.ts | 15 +- .../src/preview/{Layers => }/PortScanner.ts | 165 ++++++---- apps/server/src/preview/Services/Manager.ts | 94 ------ .../src/preview/Services/PortScanner.ts | 46 --- .../src/provider/Layers/ClaudeAdapter.ts | 4 +- .../src/provider/Layers/CodexAdapter.ts | 4 +- .../src/provider/Layers/CursorAdapter.ts | 4 +- .../server/src/provider/Layers/GrokAdapter.ts | 4 +- .../src/provider/Layers/OpenCodeAdapter.ts | 4 +- .../src/provider/Layers/ProviderService.ts | 26 +- apps/server/src/server.test.ts | 8 +- apps/server/src/server.ts | 17 +- apps/server/src/terminal/Layers/Manager.ts | 4 +- apps/server/src/ws.ts | 12 +- 30 files changed, 867 insertions(+), 943 deletions(-) delete mode 100644 apps/server/src/mcp/Layers/McpSessionRegistry.ts delete mode 100644 apps/server/src/mcp/Layers/PreviewAutomationBroker.ts rename apps/server/src/mcp/{Layers => }/McpHttpServer.test.ts (84%) rename apps/server/src/mcp/{Layers => }/McpHttpServer.ts (79%) rename apps/server/src/mcp/{Services => }/McpInvocationContext.ts (95%) rename apps/server/src/mcp/{Services => }/McpProviderSession.ts (82%) rename apps/server/src/mcp/{Layers => }/McpSessionRegistry.test.ts (92%) create mode 100644 apps/server/src/mcp/McpSessionRegistry.ts rename apps/server/src/mcp/{Layers => }/PreviewAutomationBroker.test.ts (90%) create mode 100644 apps/server/src/mcp/PreviewAutomationBroker.ts delete mode 100644 apps/server/src/mcp/Services/McpSessionRegistry.ts delete mode 100644 apps/server/src/mcp/Services/PreviewAutomationBroker.ts rename apps/server/src/preview/{Layers => }/Manager.test.ts (89%) rename apps/server/src/preview/{Layers => }/Manager.ts (63%) rename apps/server/src/preview/{Layers => }/PortScanner.test.ts (94%) rename apps/server/src/preview/{Layers => }/PortScanner.ts (69%) delete mode 100644 apps/server/src/preview/Services/Manager.ts delete mode 100644 apps/server/src/preview/Services/PortScanner.ts diff --git a/apps/server/src/mcp/Layers/McpSessionRegistry.ts b/apps/server/src/mcp/Layers/McpSessionRegistry.ts deleted file mode 100644 index a6aa95072a6faa23b2a151601af1b2801e07ae42..0000000000000000000000000000000000000000 --- a/apps/server/src/mcp/Layers/McpSessionRegistry.ts +++ /dev/null @@ -1,173 +0,0 @@ -import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; -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 { HttpServer } from "effect/unstable/http"; - -import { ServerEnvironment } from "../../environment/Services/ServerEnvironment.ts"; -import type { McpInvocationScope } from "../Services/McpInvocationContext.ts"; -import { - McpSessionRegistry, - type McpCredentialRequest, - type McpIssuedCredential, - type McpSessionRegistryShape, -} from "../Services/McpSessionRegistry.ts"; - -interface CredentialRecord { - readonly tokenHash: string; - readonly scope: McpInvocationScope; - readonly lastUsedAt: number; -} - -interface RegistryState { - readonly records: ReadonlyMap; -} - -export interface McpSessionRegistryOptions { - readonly idleTimeoutMs?: number; - readonly maximumLifetimeMs?: number; - readonly now?: () => number; -} - -const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1_000; -const DEFAULT_MAXIMUM_LIFETIME_MS = 8 * 60 * 60 * 1_000; - -const bytesToHex = (bytes: Uint8Array): string => - Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); - -const tokenFromBytes = (bytes: Uint8Array): string => Buffer.from(bytes).toString("base64url"); - -export const makeMcpSessionRegistry = (options: McpSessionRegistryOptions = {}) => - Effect.gen(function* () { - 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 idleTimeoutMs = options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS; - const maximumLifetimeMs = options.maximumLifetimeMs ?? DEFAULT_MAXIMUM_LIFETIME_MS; - const endpoint = - httpServer.address._tag === "TcpAddress" - ? `http://127.0.0.1:${httpServer.address.port}/mcp` - : "http://127.0.0.1/mcp"; - - const hashToken = (token: string) => - crypto - .digest("SHA-256", new TextEncoder().encode(token)) - .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 issue: McpSessionRegistryShape["issue"] = (request) => - Effect.gen(function* () { - const issuedAt = now(); - 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); - const expiresAt = issuedAt + maximumLifetimeMs; - const scope: McpInvocationScope = { - environmentId, - threadId: ThreadId.make(request.threadId), - providerSessionId, - providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview"]), - issuedAt, - expiresAt, - }; - yield* Ref.update(state, ({ records }) => { - const next = new Map(pruneExpired(records, issuedAt)); - next.set(tokenHash, { tokenHash, scope, lastUsedAt: issuedAt }); - return { records: next }; - }); - return { - config: { - environmentId, - threadId: scope.threadId, - providerSessionId, - providerInstanceId: scope.providerInstanceId, - endpoint, - authorizationHeader: `Bearer ${rawToken}`, - }, - expiresAt, - }; - }); - - const resolve: McpSessionRegistryShape["resolve"] = (rawToken) => - Effect.gen(function* () { - if (rawToken.length === 0) return undefined; - const tokenHash = yield* hashToken(rawToken); - const timestamp = now(); - let resolved: McpInvocationScope | undefined; - yield* Ref.update(state, ({ records }) => { - const current = pruneExpired(records, timestamp); - const record = current.get(tokenHash); - if (!record) return { records: current }; - resolved = record.scope; - const next = new Map(current); - next.set(tokenHash, { ...record, lastUsedAt: timestamp }); - return { records: next }; - }); - return resolved; - }); - - const revokeWhere = (predicate: (record: CredentialRecord) => boolean) => - Ref.update(state, ({ records }) => ({ - records: new Map(Array.from(records).filter(([, record]) => !predicate(record))), - })); - - return McpSessionRegistry.of({ - issue, - resolve, - revokeProviderSession: (providerSessionId) => - revokeWhere((record) => record.scope.providerSessionId === providerSessionId), - revokeThread: (threadId) => revokeWhere((record) => record.scope.threadId === threadId), - revokeAll: Ref.set(state, { records: new Map() }), - }); - }); - -let activeMcpSessionRegistry: McpSessionRegistryShape | undefined; - -export const McpSessionRegistryLive: Layer.Layer< - McpSessionRegistry, - never, - Crypto.Crypto | ServerEnvironment | HttpServer.HttpServer -> = Layer.effect( - McpSessionRegistry, - makeMcpSessionRegistry().pipe( - Effect.tap((registry) => - Effect.sync(() => { - activeMcpSessionRegistry = registry; - }), - ), - ), -); - -export const issueActiveMcpCredential = ( - request: McpCredentialRequest, -): Effect.Effect => - activeMcpSessionRegistry - ? activeMcpSessionRegistry - .revokeThread(request.threadId) - .pipe(Effect.andThen(activeMcpSessionRegistry.issue(request))) - : Effect.sync((): McpIssuedCredential | undefined => undefined); - -export const revokeActiveMcpThread = (threadId: ThreadId): Effect.Effect => - activeMcpSessionRegistry ? activeMcpSessionRegistry.revokeThread(threadId) : Effect.void; - -export const revokeAllActiveMcpCredentials = (): Effect.Effect => - activeMcpSessionRegistry ? activeMcpSessionRegistry.revokeAll : Effect.void; diff --git a/apps/server/src/mcp/Layers/PreviewAutomationBroker.ts b/apps/server/src/mcp/Layers/PreviewAutomationBroker.ts deleted file mode 100644 index d32523375b1ecbbcaefc6220b8c4f1e0ac84df4b..0000000000000000000000000000000000000000 --- a/apps/server/src/mcp/Layers/PreviewAutomationBroker.ts +++ /dev/null @@ -1,270 +0,0 @@ -import { - PreviewAutomationControlInterruptedError, - PreviewAutomationExecutionError, - PreviewAutomationInvalidSelectorError, - PreviewAutomationNoFocusedOwnerError, - PreviewAutomationResultTooLargeError, - PreviewAutomationTabNotFoundError, - PreviewAutomationTimeoutError, - PreviewAutomationUnavailableError, - PreviewAutomationUnsupportedClientError, - type PreviewAutomationError, - type PreviewAutomationOwner, - type PreviewAutomationResponse, -} from "@t3tools/contracts"; -import * as Deferred from "effect/Deferred"; -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 { - PreviewAutomationBroker, - type PreviewAutomationBrokerShape, -} from "../Services/PreviewAutomationBroker.ts"; - -interface ClientConnection { - readonly clientId: string; - readonly queue: Queue.Queue< - Parameters[0] extends never - ? never - : import("@t3tools/contracts").PreviewAutomationRequest - >; -} - -interface PendingRequest { - readonly clientId: string; - readonly deferred: Deferred.Deferred; -} - -interface BrokerState { - readonly clients: ReadonlyMap; - readonly owners: ReadonlyMap; - readonly pending: ReadonlyMap; -} - -const makeResponseError = ( - error: NonNullable, -): PreviewAutomationError => { - switch (error._tag) { - case "PreviewAutomationNoFocusedOwnerError": - return new PreviewAutomationNoFocusedOwnerError({ message: error.message }); - case "PreviewAutomationUnsupportedClientError": - return new PreviewAutomationUnsupportedClientError({ message: error.message }); - case "PreviewAutomationTabNotFoundError": - return new PreviewAutomationTabNotFoundError({ message: error.message }); - case "PreviewAutomationTimeoutError": - return new PreviewAutomationTimeoutError({ message: error.message }); - case "PreviewAutomationControlInterruptedError": - return new PreviewAutomationControlInterruptedError({ message: error.message }); - case "PreviewAutomationInvalidSelectorError": { - const detail = - typeof error.detail === "object" && error.detail !== null ? error.detail : undefined; - return new PreviewAutomationInvalidSelectorError({ - message: error.message, - selector: - detail && "selector" in detail && typeof detail.selector === "string" - ? detail.selector - : "", - }); - } - case "PreviewAutomationResultTooLargeError": { - const detail = - typeof error.detail === "object" && error.detail !== null ? error.detail : undefined; - return new PreviewAutomationResultTooLargeError({ - message: error.message, - maximumBytes: - detail && "maximumBytes" in detail && typeof detail.maximumBytes === "number" - ? detail.maximumBytes - : 64_000, - }); - } - case "PreviewAutomationUnavailableError": - return new PreviewAutomationUnavailableError({ message: error.message }); - default: - return new PreviewAutomationExecutionError({ - message: error.message, - detail: error.detail, - }); - } -}; - -export const makePreviewAutomationBroker = (): PreviewAutomationBrokerShape => { - const state = Effect.runSync( - Ref.make({ - clients: new Map(), - owners: new Map(), - pending: new Map(), - }), - ); - let requestSequence = 0; - - const disconnect = (clientId: string, queue: ClientConnection["queue"]) => - Effect.gen(function* () { - const toFail: PendingRequest[] = []; - yield* Ref.update(state, (current) => { - if (current.clients.get(clientId)?.queue !== queue) return current; - const clients = new Map(current.clients); - const owners = new Map(current.owners); - const pending = new Map(current.pending); - clients.delete(clientId); - owners.delete(clientId); - for (const [requestId, entry] of pending) { - if (entry.clientId === clientId) { - pending.delete(requestId); - toFail.push(entry); - } - } - return { clients, owners, pending }; - }); - yield* Effect.forEach( - toFail, - ({ deferred }) => - Deferred.fail( - deferred, - new PreviewAutomationUnavailableError({ - message: "The preview automation client disconnected.", - }), - ), - { discard: true }, - ); - yield* Queue.shutdown(queue); - }); - - const connect: PreviewAutomationBrokerShape["connect"] = (clientId) => - Effect.gen(function* () { - const queue = yield* Queue.unbounded(); - let previous: ClientConnection | undefined; - yield* Ref.update(state, (current) => { - previous = current.clients.get(clientId); - const clients = new Map(current.clients); - clients.set(clientId, { clientId, queue }); - return { ...current, clients }; - }); - if (previous) yield* disconnect(clientId, previous.queue); - return Stream.fromQueue(queue).pipe(Stream.ensuring(disconnect(clientId, queue))); - }); - - const reportOwner: PreviewAutomationBrokerShape["reportOwner"] = (owner) => - Ref.update(state, (current) => { - const owners = new Map(current.owners); - owners.set(owner.clientId, owner); - return { ...current, owners }; - }); - - const clearOwner: PreviewAutomationBrokerShape["clearOwner"] = (clientId) => - Ref.update(state, (current) => { - const owners = new Map(current.owners); - owners.delete(clientId); - return { ...current, owners }; - }); - - const respond: PreviewAutomationBrokerShape["respond"] = (response) => - Effect.gen(function* () { - let pending: PendingRequest | undefined; - yield* Ref.update(state, (current) => { - pending = current.pending.get(response.requestId); - if (!pending) return current; - const next = new Map(current.pending); - next.delete(response.requestId); - return { ...current, pending: next }; - }); - if (!pending) return; - if (response.ok) { - yield* Deferred.succeed(pending.deferred, response.result); - } else { - yield* Deferred.fail( - pending.deferred, - response.error - ? makeResponseError(response.error) - : new PreviewAutomationExecutionError({ - message: "Preview automation failed without an error payload.", - }), - ); - } - }); - - const invoke = ( - input: Parameters[0], - ): Effect.Effect => - Effect.gen(function* () { - const current = yield* Ref.get(state); - const candidates = Array.from(current.owners.values()) - .filter( - (owner) => - owner.environmentId === input.scope.environmentId && - owner.threadId === input.scope.threadId && - owner.supportsAutomation, - ) - .sort((left, right) => right.focusedAt.localeCompare(left.focusedAt)); - const owner = candidates[0]; - if (!owner) { - return yield* new PreviewAutomationNoFocusedOwnerError({ - message: "No desktop browser host is available for this thread.", - }); - } - const connection = current.clients.get(owner.clientId); - if (!connection) { - return yield* new PreviewAutomationUnavailableError({ - message: "The browser host is not connected.", - }); - } - if ( - input.operation !== "open" && - input.operation !== "status" && - !owner.tabId && - !input.tabId - ) { - return yield* new PreviewAutomationTabNotFoundError({ - 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 pending = new Map(next.pending); - pending.set(requestId, { clientId: owner.clientId, deferred }); - return { ...next, pending }; - }); - 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 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), - }); - }); - - return PreviewAutomationBroker.of({ connect, reportOwner, clearOwner, respond, invoke }); -}; - -export const previewAutomationBroker = makePreviewAutomationBroker(); - -export const PreviewAutomationBrokerLive: Layer.Layer = Layer.succeed( - PreviewAutomationBroker, - previewAutomationBroker, -); diff --git a/apps/server/src/mcp/Layers/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts similarity index 84% rename from apps/server/src/mcp/Layers/McpHttpServer.test.ts rename to apps/server/src/mcp/McpHttpServer.test.ts index 374d147d8029c8667930297a159d03e67f1af052..ebf8bf00c1c0ca40e64489e83ad5b026c06d5852 100644 --- a/apps/server/src/mcp/Layers/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -6,9 +6,9 @@ import * as Stream from "effect/Stream"; import { McpSchema, McpServer } from "effect/unstable/ai"; import { HttpServerResponse } from "effect/unstable/http"; -import { McpInvocationContext } from "../Services/McpInvocationContext.ts"; -import { normalizeMcpHttpResponse, PreviewToolkitRegistrationLive } from "./McpHttpServer.ts"; -import { previewAutomationBroker } from "./PreviewAutomationBroker.ts"; +import * as McpHttpServer from "./McpHttpServer.ts"; +import * as McpInvocationContext from "./McpInvocationContext.ts"; +import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; const environmentId = EnvironmentId.make("environment-mcp-test"); const threadId = ThreadId.make("thread-mcp-test"); @@ -31,17 +31,18 @@ const client = McpSchema.McpServerClient.of({ }, getClient: Effect.die("unused"), }); -const TestLayer = PreviewToolkitRegistrationLive.pipe( +const TestLayer = McpHttpServer.PreviewToolkitRegistrationLive.pipe( Layer.provideMerge(McpServer.McpServer.layer), + Layer.provideMerge(PreviewAutomationBroker.layer), ); it("normalizes empty successful notification responses to accepted", () => { - const notificationResponse = normalizeMcpHttpResponse( + const notificationResponse = McpHttpServer.normalizeMcpHttpResponse( HttpServerResponse.text("", { status: 200, contentType: "application/json" }), ); expect(notificationResponse.status).toBe(202); - const resultResponse = normalizeMcpHttpResponse( + const resultResponse = McpHttpServer.normalizeMcpHttpResponse( HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: 1, result: {} }), ); expect(resultResponse.status).toBe(200); @@ -51,9 +52,10 @@ it.effect("registers annotated tools and preserves authenticated request context Effect.scoped( Effect.gen(function* () { const server = yield* McpServer.McpServer; - const requests = yield* previewAutomationBroker.connect("mcp-test-client"); + const broker = yield* PreviewAutomationBroker.PreviewAutomationBroker; + const requests = yield* broker.connect("mcp-test-client"); yield* Stream.runForEach(requests, (request) => - previewAutomationBroker.respond({ + broker.respond({ requestId: request.requestId, ok: true, result: @@ -88,7 +90,7 @@ it.effect("registers annotated tools and preserves authenticated request context }), ).pipe(Effect.forkScoped); yield* Effect.yieldNow; - yield* previewAutomationBroker.reportOwner({ + yield* broker.reportOwner({ clientId: "mcp-test-client", environmentId, threadId, @@ -120,7 +122,7 @@ it.effect("registers annotated tools and preserves authenticated request context const status = yield* server .callTool({ name: "preview_status", arguments: {} }) .pipe( - Effect.provideService(McpInvocationContext, invocation), + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), Effect.provideService(McpSchema.McpServerClient, client), ); expect(status.isError).toBe(false); @@ -132,7 +134,7 @@ it.effect("registers annotated tools and preserves authenticated request context const malformed = yield* server .callTool({ name: "preview_click", arguments: { selector: "" } }) .pipe( - Effect.provideService(McpInvocationContext, invocation), + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), Effect.provideService(McpSchema.McpServerClient, client), ); expect(malformed.isError).toBe(true); @@ -140,7 +142,7 @@ it.effect("registers annotated tools and preserves authenticated request context const snapshot = yield* server .callTool({ name: "preview_snapshot", arguments: {} }) .pipe( - Effect.provideService(McpInvocationContext, invocation), + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), Effect.provideService(McpSchema.McpServerClient, client), ); expect(snapshot.isError).toBe(false); @@ -152,7 +154,7 @@ it.effect("registers annotated tools and preserves authenticated request context const press = yield* server .callTool({ name: "preview_press", arguments: { key: "Enter" } }) .pipe( - Effect.provideService(McpInvocationContext, invocation), + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), Effect.provideService(McpSchema.McpServerClient, client), ); expect(press.isError).toBe(false); diff --git a/apps/server/src/mcp/Layers/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts similarity index 79% rename from apps/server/src/mcp/Layers/McpHttpServer.ts rename to apps/server/src/mcp/McpHttpServer.ts index 356e2eabbef3bee71d766e7073c8ab3b0187460b..d13eb00422214d9f476fdee3ac1df4ec002a78c5 100644 --- a/apps/server/src/mcp/Layers/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -8,11 +8,12 @@ import * as Stream from "effect/Stream"; import { McpSchema, McpServer, Tool } from "effect/unstable/ai"; import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"; -import packageJson from "../../../package.json" with { type: "json" }; -import { McpInvocationContext } from "../Services/McpInvocationContext.ts"; -import { McpSessionRegistry } from "../Services/McpSessionRegistry.ts"; -import { PreviewToolkitHandlersLive } from "../toolkits/preview/handlers.ts"; -import { PreviewToolkit } from "../toolkits/preview/tools.ts"; +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"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -41,25 +42,24 @@ export const normalizeMcpHttpResponse = ( }; const McpAuthMiddlewareLive = HttpRouter.middleware<{ - provides: McpInvocationContext; + provides: McpInvocationContext.McpInvocationContext; }>()( Effect.gen(function* () { - const registry = yield* McpSessionRegistry; - return (httpEffect) => - Effect.gen(function* () { - const request = yield* HttpServerRequest.HttpServerRequest; - const authorization = request.headers.authorization; - const token = - authorization?.startsWith("Bearer ") === true - ? authorization.slice("Bearer ".length).trim() - : ""; - const invocation = yield* registry.resolve(token); - if (!invocation) return unauthorized; - return yield* httpEffect.pipe( - Effect.provideService(McpInvocationContext, invocation), - Effect.map(normalizeMcpHttpResponse), - ); - }); + const registry = yield* McpSessionRegistry.McpSessionRegistry; + return Effect.fn("McpHttpServer.authenticateRequest")(function* (httpEffect) { + const request = yield* HttpServerRequest.HttpServerRequest; + const authorization = request.headers.authorization; + const token = + authorization?.startsWith("Bearer ") === true + ? authorization.slice("Bearer ".length).trim() + : ""; + const invocation = yield* registry.resolve(token); + if (!invocation) return unauthorized; + return yield* httpEffect.pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), + Effect.map(normalizeMcpHttpResponse), + ); + }); }), ).layer; @@ -79,7 +79,7 @@ export const PreviewToolkitRegistrationLive = Layer.effectDiscard( ) => Effect.Effect< Stream.Stream<{ readonly encodedResult: unknown }, Error>, Error, - McpInvocationContext + McpInvocationContext.McpInvocationContext >; for (const tool of Object.values(built.tools)) { yield* server.addTool({ @@ -160,6 +160,7 @@ export const PreviewToolkitRegistrationLive = Layer.effectDiscard( }), ).pipe(Layer.provide(PreviewToolkitHandlersLive)); -export const McpHttpServerLive = Layer.mergeAll(PreviewToolkitRegistrationLive).pipe( +export const layer = Layer.mergeAll(PreviewToolkitRegistrationLive).pipe( Layer.provideMerge(McpTransportLive), + Layer.provide(PreviewAutomationBroker.layer), ); diff --git a/apps/server/src/mcp/Services/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts similarity index 95% rename from apps/server/src/mcp/Services/McpInvocationContext.ts rename to apps/server/src/mcp/McpInvocationContext.ts index 89a3820479b654583f775b3ad04503c10c4aee72..0d3f84df42ce648c072247b2e377bf24deba578a 100644 --- a/apps/server/src/mcp/Services/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -18,7 +18,7 @@ export interface McpInvocationScope { export class McpInvocationContext extends Context.Service< McpInvocationContext, McpInvocationScope ->()("t3/mcp/Services/McpInvocationContext") {} +>()("t3/mcp/McpInvocationContext") {} export const requireMcpCapability = Effect.fn("mcp.requireCapability")(function* ( capability: McpCapability, diff --git a/apps/server/src/mcp/Services/McpProviderSession.ts b/apps/server/src/mcp/McpProviderSession.ts similarity index 82% rename from apps/server/src/mcp/Services/McpProviderSession.ts rename to apps/server/src/mcp/McpProviderSession.ts index b97cbb1e1c043ff115e2a64ac1b3826c915c2ca6..d5dc582046c1a69dbbd1898307c071377e6e70fb 100644 --- a/apps/server/src/mcp/Services/McpProviderSession.ts +++ b/apps/server/src/mcp/McpProviderSession.ts @@ -1,5 +1,4 @@ import type { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; -import * as Context from "effect/Context"; export interface McpProviderSessionConfig { readonly environmentId: EnvironmentId; @@ -10,11 +9,6 @@ export interface McpProviderSessionConfig { readonly authorizationHeader: string; } -export class McpProviderSession extends Context.Service< - McpProviderSession, - McpProviderSessionConfig ->()("t3/mcp/Services/McpProviderSession") {} - const sessionsByThread = new Map(); export function setMcpProviderSession(config: McpProviderSessionConfig): void { diff --git a/apps/server/src/mcp/Layers/McpSessionRegistry.test.ts b/apps/server/src/mcp/McpSessionRegistry.test.ts similarity index 92% rename from apps/server/src/mcp/Layers/McpSessionRegistry.test.ts rename to apps/server/src/mcp/McpSessionRegistry.test.ts index bbd27c63cac25e7045cb17c73309a03325f60c48..d345953caeb82fbca2f24937d44923d28b7e99dc 100644 --- a/apps/server/src/mcp/Layers/McpSessionRegistry.test.ts +++ b/apps/server/src/mcp/McpSessionRegistry.test.ts @@ -4,8 +4,8 @@ import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts" import * as Effect from "effect/Effect"; import { HttpServer } from "effect/unstable/http"; -import { ServerEnvironment } from "../../environment/Services/ServerEnvironment.ts"; -import { makeMcpSessionRegistry } from "./McpSessionRegistry.ts"; +import { ServerEnvironment } from "../environment/Services/ServerEnvironment.ts"; +import * as McpSessionRegistry from "./McpSessionRegistry.ts"; const environmentId = EnvironmentId.make("environment-1"); const fakeHttpServer = HttpServer.HttpServer.of({ @@ -18,7 +18,7 @@ const fakeEnvironment = ServerEnvironment.of({ }); const makeRegistry = (now: () => number) => - makeMcpSessionRegistry({ + McpSessionRegistry.makeForTest({ now, idleTimeoutMs: 100, maximumLifetimeMs: 1_000, diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts new file mode 100644 index 0000000000000000000000000000000000000000..c87abe88bb5022df614fae62262a318c27d8a0e7 --- /dev/null +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -0,0 +1,202 @@ +import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +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 { HttpServer } from "effect/unstable/http"; + +import { ServerEnvironment } from "../environment/Services/ServerEnvironment.ts"; +import * as McpInvocationContext from "./McpInvocationContext.ts"; +import * as McpProviderSession from "./McpProviderSession.ts"; + +export interface McpCredentialRequest { + readonly threadId: ThreadId; + readonly providerInstanceId: ProviderInstanceId; +} + +export interface McpIssuedCredential { + readonly config: McpProviderSession.McpProviderSessionConfig; + readonly expiresAt: number; +} + +export interface McpSessionRegistryShape { + readonly issue: (request: McpCredentialRequest) => Effect.Effect; + readonly resolve: ( + rawToken: string, + ) => Effect.Effect; + readonly revokeProviderSession: (providerSessionId: string) => Effect.Effect; + readonly revokeThread: (threadId: ThreadId) => Effect.Effect; + readonly revokeAll: Effect.Effect; +} + +export class McpSessionRegistry extends Context.Service< + McpSessionRegistry, + McpSessionRegistryShape +>()("t3/mcp/McpSessionRegistry") {} + +interface CredentialRecord { + readonly tokenHash: string; + readonly scope: McpInvocationContext.McpInvocationScope; + readonly lastUsedAt: number; +} + +interface RegistryState { + readonly records: ReadonlyMap; +} + +export interface McpSessionRegistryOptions { + readonly idleTimeoutMs?: number; + readonly maximumLifetimeMs?: number; + readonly now?: () => number; +} + +const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1_000; +const DEFAULT_MAXIMUM_LIFETIME_MS = 8 * 60 * 60 * 1_000; + +const bytesToHex = (bytes: Uint8Array): string => + Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); + +const tokenFromBytes = (bytes: Uint8Array): string => Buffer.from(bytes).toString("base64url"); + +const make = 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 idleTimeoutMs = options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS; + const maximumLifetimeMs = options.maximumLifetimeMs ?? DEFAULT_MAXIMUM_LIFETIME_MS; + const endpoint = + httpServer.address._tag === "TcpAddress" + ? `http://127.0.0.1:${httpServer.address.port}/mcp` + : "http://127.0.0.1/mcp"; + + const hashToken = (token: string) => + crypto + .digest("SHA-256", new TextEncoder().encode(token)) + .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 issue: McpSessionRegistryShape["issue"] = Effect.fn("McpSessionRegistry.issue")( + function* (request) { + const issuedAt = now(); + 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); + const expiresAt = issuedAt + maximumLifetimeMs; + const scope: McpInvocationContext.McpInvocationScope = { + environmentId, + threadId: ThreadId.make(request.threadId), + providerSessionId, + providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), + capabilities: new Set(["preview"]), + issuedAt, + expiresAt, + }; + yield* Ref.update(state, ({ records }) => { + const next = new Map(pruneExpired(records, issuedAt)); + next.set(tokenHash, { tokenHash, scope, lastUsedAt: issuedAt }); + return { records: next }; + }); + return { + config: { + environmentId, + threadId: scope.threadId, + providerSessionId, + providerInstanceId: scope.providerInstanceId, + endpoint, + authorizationHeader: `Bearer ${rawToken}`, + }, + expiresAt, + }; + }, + ); + + const resolve: McpSessionRegistryShape["resolve"] = Effect.fn("McpSessionRegistry.resolve")( + 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 current = pruneExpired(records, timestamp); + const record = current.get(tokenHash); + if (!record) return { records: current }; + resolved = record.scope; + const next = new Map(current); + next.set(tokenHash, { ...record, lastUsedAt: timestamp }); + return { records: next }; + }); + return resolved; + }, + ); + + const revokeWhere = (predicate: (record: CredentialRecord) => boolean) => + Ref.update(state, ({ records }) => ({ + records: new Map(Array.from(records).filter(([, record]) => !predicate(record))), + })); + + return McpSessionRegistry.of({ + issue, + resolve, + revokeProviderSession: Effect.fn("McpSessionRegistry.revokeProviderSession")( + function* (providerSessionId) { + yield* revokeWhere((record) => record.scope.providerSessionId === providerSessionId); + }, + ), + revokeThread: Effect.fn("McpSessionRegistry.revokeThread")(function* (threadId) { + yield* revokeWhere((record) => record.scope.threadId === threadId); + }), + revokeAll: Ref.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( + Effect.tap((registry) => + Effect.sync(() => { + activeMcpSessionRegistry = registry; + }), + ), + ), +); + +export const makeForTest = make; + +export const issueActiveMcpCredential = ( + request: McpCredentialRequest, +): Effect.Effect => + activeMcpSessionRegistry + ? activeMcpSessionRegistry + .revokeThread(request.threadId) + .pipe(Effect.andThen(activeMcpSessionRegistry.issue(request))) + : Effect.sync((): McpIssuedCredential | undefined => undefined); + +export const revokeActiveMcpThread = (threadId: ThreadId): Effect.Effect => + activeMcpSessionRegistry ? activeMcpSessionRegistry.revokeThread(threadId) : Effect.void; + +export const revokeAllActiveMcpCredentials = (): Effect.Effect => + activeMcpSessionRegistry ? activeMcpSessionRegistry.revokeAll : Effect.void; diff --git a/apps/server/src/mcp/Layers/PreviewAutomationBroker.test.ts b/apps/server/src/mcp/PreviewAutomationBroker.test.ts similarity index 90% rename from apps/server/src/mcp/Layers/PreviewAutomationBroker.test.ts rename to apps/server/src/mcp/PreviewAutomationBroker.test.ts index 7c97da39f2e40fdfe128481cc7eaa918a4e8d6f2..c5d316fd87fd9fe47ef3cf93c1882f33c16f8b14 100644 --- a/apps/server/src/mcp/Layers/PreviewAutomationBroker.test.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.test.ts @@ -8,7 +8,7 @@ import { import * as Effect from "effect/Effect"; import * as Stream from "effect/Stream"; -import { makePreviewAutomationBroker } from "./PreviewAutomationBroker.ts"; +import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; const scope = { environmentId: EnvironmentId.make("environment-1"), @@ -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 = makePreviewAutomationBroker(); + const broker = yield* PreviewAutomationBroker.makeForTest; 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 = makePreviewAutomationBroker(); + const broker = yield* PreviewAutomationBroker.makeForTest; 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 = makePreviewAutomationBroker(); + const broker = yield* PreviewAutomationBroker.makeForTest; 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 new file mode 100644 index 0000000000000000000000000000000000000000..68080bd93606b9dcddf40e43237f9ed4cd544bfc --- /dev/null +++ b/apps/server/src/mcp/PreviewAutomationBroker.ts @@ -0,0 +1,306 @@ +import { + PreviewAutomationControlInterruptedError, + PreviewAutomationExecutionError, + PreviewAutomationInvalidSelectorError, + PreviewAutomationNoFocusedOwnerError, + PreviewAutomationResultTooLargeError, + PreviewAutomationTabNotFoundError, + PreviewAutomationTimeoutError, + PreviewAutomationUnavailableError, + PreviewAutomationUnsupportedClientError, + type PreviewAutomationError, + type PreviewAutomationOperation, + type PreviewAutomationOwner, + type PreviewAutomationRequest, + type PreviewAutomationResponse, + type PreviewTabId, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +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 McpInvocationContext from "./McpInvocationContext.ts"; + +export interface PreviewAutomationInvokeInput { + readonly scope: McpInvocationContext.McpInvocationScope; + readonly operation: PreviewAutomationOperation; + readonly input: unknown; + readonly tabId?: PreviewTabId; + readonly timeoutMs?: number; +} + +export interface PreviewAutomationBrokerShape { + readonly connect: (clientId: string) => Effect.Effect>; + readonly reportOwner: ( + owner: PreviewAutomationOwner, + ) => Effect.Effect; + readonly clearOwner: (clientId: string) => Effect.Effect; + readonly respond: ( + response: PreviewAutomationResponse, + ) => Effect.Effect; + readonly invoke: ( + request: PreviewAutomationInvokeInput, + ) => Effect.Effect; +} + +export class PreviewAutomationBroker extends Context.Service< + PreviewAutomationBroker, + PreviewAutomationBrokerShape +>()("t3/mcp/PreviewAutomationBroker") {} + +interface ClientConnection { + readonly clientId: string; + readonly queue: Queue.Queue< + Parameters[0] extends never + ? never + : import("@t3tools/contracts").PreviewAutomationRequest + >; +} + +interface PendingRequest { + readonly clientId: string; + readonly deferred: Deferred.Deferred; +} + +interface BrokerState { + readonly clients: ReadonlyMap; + readonly owners: ReadonlyMap; + readonly pending: ReadonlyMap; +} + +const makeResponseError = ( + error: NonNullable, +): PreviewAutomationError => { + switch (error._tag) { + case "PreviewAutomationNoFocusedOwnerError": + return new PreviewAutomationNoFocusedOwnerError({ message: error.message }); + case "PreviewAutomationUnsupportedClientError": + return new PreviewAutomationUnsupportedClientError({ message: error.message }); + case "PreviewAutomationTabNotFoundError": + return new PreviewAutomationTabNotFoundError({ message: error.message }); + case "PreviewAutomationTimeoutError": + return new PreviewAutomationTimeoutError({ message: error.message }); + case "PreviewAutomationControlInterruptedError": + return new PreviewAutomationControlInterruptedError({ message: error.message }); + case "PreviewAutomationInvalidSelectorError": { + const detail = + typeof error.detail === "object" && error.detail !== null ? error.detail : undefined; + return new PreviewAutomationInvalidSelectorError({ + message: error.message, + selector: + detail && "selector" in detail && typeof detail.selector === "string" + ? detail.selector + : "", + }); + } + case "PreviewAutomationResultTooLargeError": { + const detail = + typeof error.detail === "object" && error.detail !== null ? error.detail : undefined; + return new PreviewAutomationResultTooLargeError({ + message: error.message, + maximumBytes: + detail && "maximumBytes" in detail && typeof detail.maximumBytes === "number" + ? detail.maximumBytes + : 64_000, + }); + } + case "PreviewAutomationUnavailableError": + return new PreviewAutomationUnavailableError({ message: error.message }); + default: + return new PreviewAutomationExecutionError({ + message: error.message, + detail: error.detail, + }); + } +}; + +const makeService = (): PreviewAutomationBrokerShape => { + const state = Effect.runSync( + Ref.make({ + clients: new Map(), + owners: new Map(), + pending: new Map(), + }), + ); + 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 clients = new Map(current.clients); + const owners = new Map(current.owners); + const pending = new Map(current.pending); + clients.delete(clientId); + owners.delete(clientId); + for (const [requestId, entry] of pending) { + if (entry.clientId === clientId) { + pending.delete(requestId); + toFail.push(entry); + } + } + return { clients, owners, pending }; + }); + yield* Effect.forEach( + toFail, + ({ deferred }) => + Deferred.fail( + deferred, + new PreviewAutomationUnavailableError({ + message: "The preview automation client disconnected.", + }), + ), + { discard: true }, + ); + yield* Queue.shutdown(queue); + }); + + const connect: PreviewAutomationBrokerShape["connect"] = Effect.fn( + "PreviewAutomationBroker.connect", + )(function* (clientId) { + const queue = yield* Queue.unbounded(); + let previous: ClientConnection | undefined; + yield* Ref.update(state, (current) => { + previous = current.clients.get(clientId); + const clients = new Map(current.clients); + clients.set(clientId, { clientId, queue }); + return { ...current, clients }; + }); + if (previous) yield* disconnect(clientId, previous.queue); + return Stream.fromQueue(queue).pipe(Stream.ensuring(disconnect(clientId, queue))); + }); + + const reportOwner: PreviewAutomationBrokerShape["reportOwner"] = Effect.fn( + "PreviewAutomationBroker.reportOwner", + )(function* (owner) { + yield* Ref.update(state, (current) => { + const owners = new Map(current.owners); + owners.set(owner.clientId, owner); + return { ...current, owners }; + }); + }); + + const clearOwner: PreviewAutomationBrokerShape["clearOwner"] = Effect.fn( + "PreviewAutomationBroker.clearOwner", + )(function* (clientId) { + yield* Ref.update(state, (current) => { + const owners = new Map(current.owners); + owners.delete(clientId); + return { ...current, owners }; + }); + }); + + 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 next = new Map(current.pending); + next.delete(response.requestId); + return { ...current, pending: next }; + }); + if (!pending) return; + if (response.ok) { + yield* Deferred.succeed(pending.deferred, response.result); + } else { + yield* Deferred.fail( + pending.deferred, + response.error + ? makeResponseError(response.error) + : new PreviewAutomationExecutionError({ + message: "Preview automation failed without an error payload.", + }), + ); + } + }); + + const invoke = Effect.fn("PreviewAutomationBroker.invoke")(function* ( + input: Parameters[0], + ): Effect.fn.Return { + const current = yield* Ref.get(state); + const candidates = Array.from(current.owners.values()) + .filter( + (owner) => + owner.environmentId === input.scope.environmentId && + owner.threadId === input.scope.threadId && + owner.supportsAutomation, + ) + .sort((left, right) => right.focusedAt.localeCompare(left.focusedAt)); + const owner = candidates[0]; + if (!owner) { + return yield* new PreviewAutomationNoFocusedOwnerError({ + message: "No desktop browser host is available for this thread.", + }); + } + const connection = current.clients.get(owner.clientId); + if (!connection) { + return yield* new PreviewAutomationUnavailableError({ + message: "The browser host is not connected.", + }); + } + if ( + input.operation !== "open" && + input.operation !== "status" && + !owner.tabId && + !input.tabId + ) { + return yield* new PreviewAutomationTabNotFoundError({ + 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 pending = new Map(next.pending); + pending.set(requestId, { clientId: owner.clientId, deferred }); + return { ...next, pending }; + }); + 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 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), + }); + }); + + return PreviewAutomationBroker.of({ connect, reportOwner, clearOwner, respond, invoke }); +}; + +const service = makeService(); +const make = Effect.succeed(service); +export const layer = Layer.effect(PreviewAutomationBroker, make); + +export const makeForTest = Effect.sync(makeService); diff --git a/apps/server/src/mcp/Services/McpSessionRegistry.ts b/apps/server/src/mcp/Services/McpSessionRegistry.ts deleted file mode 100644 index df2ca991271e6dbebed1c2bb80f2f503ce9ef4b8..0000000000000000000000000000000000000000 --- a/apps/server/src/mcp/Services/McpSessionRegistry.ts +++ /dev/null @@ -1,29 +0,0 @@ -import type { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; -import * as Context from "effect/Context"; -import type * as Effect from "effect/Effect"; - -import type { McpInvocationScope } from "./McpInvocationContext.ts"; -import type { McpProviderSessionConfig } from "./McpProviderSession.ts"; - -export interface McpCredentialRequest { - readonly threadId: ThreadId; - readonly providerInstanceId: ProviderInstanceId; -} - -export interface McpIssuedCredential { - readonly config: McpProviderSessionConfig; - readonly expiresAt: number; -} - -export interface McpSessionRegistryShape { - readonly issue: (request: McpCredentialRequest) => Effect.Effect; - readonly resolve: (rawToken: string) => Effect.Effect; - readonly revokeProviderSession: (providerSessionId: string) => Effect.Effect; - readonly revokeThread: (threadId: ThreadId) => Effect.Effect; - readonly revokeAll: Effect.Effect; -} - -export class McpSessionRegistry extends Context.Service< - McpSessionRegistry, - McpSessionRegistryShape ->()("t3/mcp/Services/McpSessionRegistry") {} diff --git a/apps/server/src/mcp/Services/PreviewAutomationBroker.ts b/apps/server/src/mcp/Services/PreviewAutomationBroker.ts deleted file mode 100644 index 1af1a3d24bfc72adc162f2da504142b611fefba1..0000000000000000000000000000000000000000 --- a/apps/server/src/mcp/Services/PreviewAutomationBroker.ts +++ /dev/null @@ -1,40 +0,0 @@ -import type { - PreviewAutomationError, - PreviewAutomationOperation, - PreviewAutomationOwner, - PreviewAutomationRequest, - PreviewAutomationResponse, - PreviewTabId, -} from "@t3tools/contracts"; -import * as Context from "effect/Context"; -import type * as Effect from "effect/Effect"; -import type * as Stream from "effect/Stream"; - -import type { McpInvocationScope } from "./McpInvocationContext.ts"; - -export interface PreviewAutomationInvokeInput { - readonly scope: McpInvocationScope; - readonly operation: PreviewAutomationOperation; - readonly input: unknown; - readonly tabId?: PreviewTabId; - readonly timeoutMs?: number; -} - -export interface PreviewAutomationBrokerShape { - readonly connect: (clientId: string) => Effect.Effect>; - readonly reportOwner: ( - owner: PreviewAutomationOwner, - ) => Effect.Effect; - readonly clearOwner: (clientId: string) => Effect.Effect; - readonly respond: ( - response: PreviewAutomationResponse, - ) => Effect.Effect; - readonly invoke: ( - request: PreviewAutomationInvokeInput, - ) => Effect.Effect; -} - -export class PreviewAutomationBroker extends Context.Service< - PreviewAutomationBroker, - PreviewAutomationBrokerShape ->()("t3/mcp/Services/PreviewAutomationBroker") {} diff --git a/apps/server/src/mcp/toolkits/preview/handlers.ts b/apps/server/src/mcp/toolkits/preview/handlers.ts index 62f9142134357e71b5de3c174e22462f72477c22..56c9a6d98e481fcc734b797a7f2f91667cb39b03 100644 --- a/apps/server/src/mcp/toolkits/preview/handlers.ts +++ b/apps/server/src/mcp/toolkits/preview/handlers.ts @@ -7,28 +7,28 @@ import type { PreviewAutomationStatus, } from "@t3tools/contracts"; -import { requireMcpCapability } from "../../Services/McpInvocationContext.ts"; -import { previewAutomationBroker } from "../../Layers/PreviewAutomationBroker.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts"; import { PreviewToolkit } from "./tools.ts"; -const invoke = ( +const invoke = Effect.fn("PreviewToolkit.invoke")(function* ( operation: PreviewAutomationOperation, input: unknown, timeoutMs?: number, -): Effect.Effect< +): Effect.fn.Return< A, import("@t3tools/contracts").PreviewAutomationError, - import("../../Services/McpInvocationContext.ts").McpInvocationContext -> => - Effect.gen(function* () { - const scope = yield* requireMcpCapability("preview"); - return yield* previewAutomationBroker.invoke({ - scope, - operation, - input, - ...(timeoutMs === undefined ? {} : { timeoutMs }), - }); + McpInvocationContext.McpInvocationContext | PreviewAutomationBroker.PreviewAutomationBroker +> { + const scope = yield* McpInvocationContext.requireMcpCapability("preview"); + const broker = yield* PreviewAutomationBroker.PreviewAutomationBroker; + return yield* broker.invoke({ + scope, + operation, + input, + ...(timeoutMs === undefined ? {} : { timeoutMs }), }); +}); export const PreviewToolkitHandlersLive = PreviewToolkit.toLayer({ preview_status: () => invoke("status", {}), diff --git a/apps/server/src/mcp/toolkits/preview/tools.ts b/apps/server/src/mcp/toolkits/preview/tools.ts index effd87fb7e434f2beedf9d9474f7c169cd746065..bcfe720a9504a3c4da80b3531ad40cf83ea2de97 100644 --- a/apps/server/src/mcp/toolkits/preview/tools.ts +++ b/apps/server/src/mcp/toolkits/preview/tools.ts @@ -16,7 +16,13 @@ import { import * as Schema from "effect/Schema"; import { Tool, Toolkit } from "effect/unstable/ai"; -import { McpInvocationContext } from "../../Services/McpInvocationContext.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts"; + +const dependencies = [ + McpInvocationContext.McpInvocationContext, + PreviewAutomationBroker.PreviewAutomationBroker, +]; const browserTool = (tool: T): T => tool.annotate(Tool.OpenWorld, true).annotate(Tool.Destructive, true) as T; @@ -32,7 +38,7 @@ export const PreviewStatusTool = Tool.make("preview_status", { "Report whether the scoped thread has an automation-capable desktop preview, including its active tab, URL, title, visibility, and loading state.", success: PreviewAutomationStatus, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }) .annotate(Tool.Title, "Get preview status") .annotate(Tool.Readonly, true) @@ -46,7 +52,7 @@ export const PreviewOpenTool = browserTool( parameters: PreviewAutomationOpenInput, success: PreviewAutomationStatus, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }) .annotate(Tool.Title, "Open browser preview") .annotate(Tool.Destructive, false), @@ -59,7 +65,7 @@ export const PreviewNavigateTool = safeBrowserTool( parameters: PreviewAutomationNavigateInput, success: PreviewAutomationStatus, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Navigate browser preview"), ); @@ -69,7 +75,7 @@ export const PreviewSnapshotTool = readonlyBrowserTool( "Inspect the current page before interacting. Returns URL/title/loading state, visible text, semantic interactive elements with reusable selectors and coordinates, accessibility data, recent console/network failures, action history, and a PNG screenshot.", success: PreviewAutomationSnapshot, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Inspect browser page"), ); @@ -79,7 +85,7 @@ export const PreviewClickTool = browserTool( "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, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Click preview page"), ); @@ -89,7 +95,7 @@ export const PreviewTypeTool = browserTool( "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, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Type into preview page"), ); @@ -99,7 +105,7 @@ export const PreviewPressTool = browserTool( "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, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Press key in preview page"), ); @@ -109,7 +115,7 @@ export const PreviewScrollTool = safeBrowserTool( "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, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Scroll preview page"), ); @@ -120,7 +126,7 @@ export const PreviewEvaluateTool = browserTool( parameters: PreviewAutomationEvaluateInput, success: Schema.Unknown, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Evaluate JavaScript in preview"), ); @@ -130,7 +136,7 @@ export const PreviewWaitForTool = readonlyBrowserTool( "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, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Wait for preview page condition"), ); @@ -140,7 +146,7 @@ export const PreviewRecordingStartTool = safeBrowserTool( "Start recording the active collaborative browser tab while keeping it interactive for both agent and human use.", success: PreviewAutomationRecordingStatus, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Start browser recording"), ); @@ -149,7 +155,7 @@ export const PreviewRecordingStopTool = safeBrowserTool( description: "Stop the active browser recording and save it as a local evidence artifact.", success: PreviewAutomationRecordingArtifact, failure: PreviewAutomationError, - dependencies: [McpInvocationContext], + dependencies, }).annotate(Tool.Title, "Stop browser recording"), ); diff --git a/apps/server/src/preview/Layers/Manager.test.ts b/apps/server/src/preview/Manager.test.ts similarity index 89% rename from apps/server/src/preview/Layers/Manager.test.ts rename to apps/server/src/preview/Manager.test.ts index 97a64ec844f8f3442c0806eaa6c041f8836ca631..a910e27470d2bab44bad93794f7110a3c8ea26aa 100644 --- a/apps/server/src/preview/Layers/Manager.test.ts +++ b/apps/server/src/preview/Manager.test.ts @@ -3,8 +3,7 @@ import { type PreviewEvent, ThreadId } from "@t3tools/contracts"; import { Effect, PubSub } from "effect"; import { expect } from "vite-plus/test"; -import { PreviewManager } from "../Services/Manager.ts"; -import { PreviewManagerLive } from "./Manager.ts"; +import * as PreviewManager from "./Manager.ts"; const DRAIN_LIMIT = 100; @@ -27,7 +26,7 @@ const freshThreadId = () => ThreadId.make(`thread-${++nextThreadId}`); * no event can land between subscribe and the consumer drain. */ const collectEvents = Effect.gen(function* () { - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const subscription = yield* manager.subscribeEvents; const collector: EventCollector = { drain: PubSub.takeUpTo(subscription, DRAIN_LIMIT), @@ -35,11 +34,11 @@ const collectEvents = Effect.gen(function* () { return collector; }).pipe(Effect.withSpan("preview.test.collectEvents")); -it.layer(PreviewManagerLive)("PreviewManager", (it) => { +it.layer(PreviewManager.layer)("PreviewManager", (it) => { it.effect("opens a session and emits opened with normalized URL", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const collector = yield* collectEvents; const snapshot = yield* manager.open({ threadId, url: "localhost:5173" }); @@ -61,7 +60,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("opens an Idle tab when no URL is supplied", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const snapshot = yield* manager.open({ threadId }); expect(snapshot.navStatus._tag).toBe("Idle"); }), @@ -70,7 +69,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("treats bare hosts as https", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const snapshot = yield* manager.open({ threadId, url: "example.com" }); if (snapshot.navStatus._tag === "Loading") { expect(snapshot.navStatus.url).toBe("https://example.com/"); @@ -81,7 +80,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("rejects empty URL with PreviewInvalidUrlError", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const error = yield* Effect.flip(manager.open({ threadId, url: " " })); expect(error._tag).toBe("PreviewInvalidUrlError"); }), @@ -90,7 +89,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("navigate updates snapshot and emits navigated", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const collector = yield* collectEvents; const opened = yield* manager.open({ threadId, url: "http://localhost:5173" }); @@ -114,7 +113,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("navigate fails for unknown tab", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const error = yield* Effect.flip( manager.navigate({ threadId, @@ -129,7 +128,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("reportStatus emits failed for LoadFailed nav", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const collector = yield* collectEvents; const opened = yield* manager.open({ threadId, url: "http://localhost:5173" }); @@ -160,7 +159,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("close removes the session and emits closed", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const collector = yield* collectEvents; yield* manager.open({ threadId, url: "http://localhost:5173" }); @@ -177,7 +176,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("close is idempotent for unknown threads", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; yield* manager.close({ threadId }); const result = yield* manager.list({ threadId }); expect(result.sessions).toHaveLength(0); @@ -187,7 +186,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("list returns every snapshot for the thread sorted by updatedAt", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const first = yield* manager.open({ threadId, url: "http://localhost:5173" }); const second = yield* manager.open({ threadId, url: "http://localhost:3000" }); const result = yield* manager.list({ threadId }); @@ -201,7 +200,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("open creates an independent tab on every call", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const collector = yield* collectEvents; const a = yield* manager.open({ threadId, url: "http://localhost:5173" }); @@ -219,7 +218,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("close with mismatching tabId is a no-op", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; yield* manager.open({ threadId, url: "http://localhost:5173" }); yield* manager.close({ threadId, tabId: "tab_missing" }); @@ -231,7 +230,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("close with explicit tabId removes only that tab", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const a = yield* manager.open({ threadId, url: "http://localhost:5173" }); const b = yield* manager.open({ threadId, url: "http://localhost:3000" }); @@ -245,7 +244,7 @@ it.layer(PreviewManagerLive)("PreviewManager", (it) => { it.effect("multiple subscribers receive every event independently", () => Effect.gen(function* () { const threadId = freshThreadId(); - const manager = yield* PreviewManager; + const manager = yield* PreviewManager.PreviewManager; const aSub = yield* manager.subscribeEvents; const bSub = yield* manager.subscribeEvents; diff --git a/apps/server/src/preview/Layers/Manager.ts b/apps/server/src/preview/Manager.ts similarity index 63% rename from apps/server/src/preview/Layers/Manager.ts rename to apps/server/src/preview/Manager.ts index dc58b3f9f13d511a59bdf51f12da768f2521bafe..8a52dbcc8c34000ce5c0ae60e0ca3bc6c2fff602 100644 --- a/apps/server/src/preview/Layers/Manager.ts +++ b/apps/server/src/preview/Manager.ts @@ -10,9 +10,16 @@ * fail an in-progress `navigate()`). */ import { + type PreviewCloseInput, type PreviewEvent, + type PreviewError, PreviewInvalidUrlError, + type PreviewListInput, type PreviewListResult, + type PreviewNavigateInput, + type PreviewOpenInput, + type PreviewRefreshInput, + type PreviewReportStatusInput, PreviewSessionLookupError, type PreviewSessionSnapshot, } from "@t3tools/contracts"; @@ -21,9 +28,33 @@ import { normalizePreviewUrl, PreviewUrlNormalizationError, } from "@t3tools/shared/preview"; -import { DateTime, Effect, Layer, PubSub, Stream, SynchronizedRef } from "effect"; +import { + Context, + DateTime, + Effect, + Layer, + PubSub, + type Scope, + Stream, + SynchronizedRef, +} from "effect"; -import { PreviewManager, type PreviewManagerShape } from "../Services/Manager.ts"; +export interface PreviewManagerShape { + readonly open: (input: PreviewOpenInput) => Effect.Effect; + readonly navigate: ( + input: PreviewNavigateInput, + ) => Effect.Effect; + readonly reportStatus: (input: PreviewReportStatusInput) => Effect.Effect; + readonly refresh: (input: PreviewRefreshInput) => Effect.Effect; + readonly close: (input: PreviewCloseInput) => Effect.Effect; + readonly list: (input: PreviewListInput) => Effect.Effect; + readonly events: Stream.Stream; + readonly subscribeEvents: Effect.Effect, never, Scope.Scope>; +} + +export class PreviewManager extends Context.Service()( + "t3/preview/Manager/PreviewManager", +) {} interface PreviewSessionState { readonly threadId: string; @@ -96,7 +127,7 @@ const buildIdleSnapshot = (input: { updatedAt: input.updatedAt, }); -export const makePreviewManager = Effect.gen(function* () { +const make = Effect.gen(function* () { 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 @@ -134,8 +165,8 @@ export const makePreviewManager = Effect.gen(function* () { ] as readonly [ModifyResult, ManagerState]); } return mutator(session).pipe( - Effect.flatMap(({ next, emit, result }) => - Effect.gen(function* () { + Effect.flatMap( + Effect.fn("PreviewManager.commitMutation")(function* ({ next, emit, result }) { if (emit) yield* PubSub.publish(eventsPubSub, emit); const sessions = new Map(state.sessions); sessions.set(compositeKey(threadId, tabId), next); @@ -153,43 +184,44 @@ export const makePreviewManager = Effect.gen(function* () { ); }; - const open: PreviewManagerShape["open"] = (input) => - Effect.gen(function* () { - const tabId = newPreviewTabId(); - const updatedAt = yield* currentIsoTimestamp; - const snapshot = input.url - ? buildLoadingSnapshot({ - threadId: input.threadId, - tabId, - url: yield* normalizeUrl(input.url), - title: "", - updatedAt, - }) - : buildIdleSnapshot({ threadId: input.threadId, tabId, updatedAt }); - yield* SynchronizedRef.update(stateRef, (state) => { - const sessions = new Map(state.sessions); - sessions.set(compositeKey(input.threadId, tabId), { + const open: PreviewManagerShape["open"] = Effect.fn("PreviewManager.open")(function* (input) { + const tabId = newPreviewTabId(); + const updatedAt = yield* currentIsoTimestamp; + const snapshot = input.url + ? buildLoadingSnapshot({ threadId: input.threadId, tabId, - snapshot, - }); - return { sessions }; - }); - yield* PubSub.publish(eventsPubSub, { - type: "opened", + url: yield* normalizeUrl(input.url), + title: "", + updatedAt, + }) + : buildIdleSnapshot({ threadId: input.threadId, tabId, updatedAt }); + yield* SynchronizedRef.update(stateRef, (state) => { + const sessions = new Map(state.sessions); + sessions.set(compositeKey(input.threadId, tabId), { threadId: input.threadId, tabId, - createdAt: snapshot.updatedAt, snapshot, }); - return snapshot; + return { sessions }; }); + yield* PubSub.publish(eventsPubSub, { + type: "opened", + threadId: input.threadId, + tabId, + createdAt: snapshot.updatedAt, + snapshot, + }); + return snapshot; + }); - const navigate: PreviewManagerShape["navigate"] = (input) => - Effect.gen(function* () { + const navigate: PreviewManagerShape["navigate"] = Effect.fn("PreviewManager.navigate")( + function* (input) { const url = yield* normalizeUrl(input.url); - return yield* mutateExistingSession(input.threadId, input.tabId, (session) => - Effect.gen(function* () { + return yield* mutateExistingSession( + input.threadId, + input.tabId, + Effect.fn("PreviewManager.navigateSession")(function* (session) { const updatedAt = yield* currentIsoTimestamp; const previousTitle = session.snapshot.navStatus._tag === "Idle" ? "" : session.snapshot.navStatus.title; @@ -215,11 +247,16 @@ export const makePreviewManager = Effect.gen(function* () { }; }), ); - }); + }, + ); - const reportStatus: PreviewManagerShape["reportStatus"] = (input) => - mutateExistingSession(input.threadId, input.tabId, (session) => - Effect.gen(function* () { + const reportStatus: PreviewManagerShape["reportStatus"] = Effect.fn( + "PreviewManager.reportStatus", + )(function* (input) { + yield* mutateExistingSession( + input.threadId, + input.tabId, + Effect.fn("PreviewManager.reportSessionStatus")(function* (session) { const updatedAt = yield* currentIsoTimestamp; const snapshot: PreviewSessionSnapshot = { threadId: session.threadId, @@ -255,48 +292,51 @@ export const makePreviewManager = Effect.gen(function* () { }; }), ); + }); - const refresh: PreviewManagerShape["refresh"] = (input) => - // Verify the session exists; the desktop bridge handles the actual reload - // and will report progress back via `reportStatus`. No event emitted. - mutateExistingSession(input.threadId, input.tabId, (session) => - Effect.succeed({ next: session, emit: null, result: undefined as void }), - ); + const refresh: PreviewManagerShape["refresh"] = Effect.fn("PreviewManager.refresh")( + function* (input) { + // Verify the session exists; the desktop bridge handles the actual reload + // and will report progress back via `reportStatus`. No event emitted. + yield* mutateExistingSession(input.threadId, input.tabId, (session) => + Effect.succeed({ next: session, emit: null, result: undefined as void }), + ); + }, + ); - const close: PreviewManagerShape["close"] = (input) => - Effect.gen(function* () { - const createdAt = yield* currentIsoTimestamp; - const events = yield* SynchronizedRef.modify(stateRef, (state) => { - const eventsToEmit: PreviewEvent[] = []; - const sessions = new Map(state.sessions); - const targets = input.tabId - ? [state.sessions.get(compositeKey(input.threadId, input.tabId))].filter( - (entry): entry is PreviewSessionState => entry !== undefined, - ) - : sessionsForThread(state, input.threadId); - for (const target of targets) { - sessions.delete(compositeKey(target.threadId, target.tabId)); - eventsToEmit.push({ - type: "closed", - threadId: target.threadId, - tabId: target.tabId, - createdAt, - }); - } - if (eventsToEmit.length === 0) { - return [eventsToEmit, state] as const; - } - return [eventsToEmit, { sessions }] as const; - }); - if (events.length > 0) { - yield* Effect.forEach(events, (event) => PubSub.publish(eventsPubSub, event), { - discard: true, + const close: PreviewManagerShape["close"] = Effect.fn("PreviewManager.close")(function* (input) { + const createdAt = yield* currentIsoTimestamp; + const events = yield* SynchronizedRef.modify(stateRef, (state) => { + const eventsToEmit: PreviewEvent[] = []; + const sessions = new Map(state.sessions); + const targets = input.tabId + ? [state.sessions.get(compositeKey(input.threadId, input.tabId))].filter( + (entry): entry is PreviewSessionState => entry !== undefined, + ) + : sessionsForThread(state, input.threadId); + for (const target of targets) { + sessions.delete(compositeKey(target.threadId, target.tabId)); + eventsToEmit.push({ + type: "closed", + threadId: target.threadId, + tabId: target.tabId, + createdAt, }); } + if (eventsToEmit.length === 0) { + return [eventsToEmit, state] as const; + } + return [eventsToEmit, { sessions }] as const; }); + if (events.length > 0) { + yield* Effect.forEach(events, (event) => PubSub.publish(eventsPubSub, event), { + discard: true, + }); + } + }); - const list: PreviewManagerShape["list"] = (input) => - SynchronizedRef.get(stateRef).pipe( + const list: PreviewManagerShape["list"] = Effect.fn("PreviewManager.list")(function* (input) { + return yield* SynchronizedRef.get(stateRef).pipe( Effect.map( (state): PreviewListResult => ({ sessions: sessionsForThread(state, input.threadId) @@ -305,6 +345,7 @@ export const makePreviewManager = Effect.gen(function* () { }), ), ); + }); return { open, @@ -318,4 +359,4 @@ export const makePreviewManager = Effect.gen(function* () { } satisfies PreviewManagerShape; }); -export const PreviewManagerLive = Layer.effect(PreviewManager, makePreviewManager); +export const layer = Layer.effect(PreviewManager, make); diff --git a/apps/server/src/preview/Layers/PortScanner.test.ts b/apps/server/src/preview/PortScanner.test.ts similarity index 94% rename from apps/server/src/preview/Layers/PortScanner.test.ts rename to apps/server/src/preview/PortScanner.test.ts index f0e78aeedbc7f281bf570f7feb15442ef451c3b6..0f75a538ec6e4d5898f5727d3f1f61753fdf3240 100644 --- a/apps/server/src/preview/Layers/PortScanner.test.ts +++ b/apps/server/src/preview/PortScanner.test.ts @@ -5,16 +5,15 @@ import { ThreadId } from "@t3tools/contracts"; import { Effect, Layer } from "effect"; import { afterEach, beforeEach, describe, expect, it } from "vite-plus/test"; -import { COMMON_DEV_PORTS, PortDiscovery } from "../Services/PortScanner.ts"; -import { ProcessRunner } from "../../processRunner.ts"; -import { __testing, PortDiscoveryLive } from "./PortScanner.ts"; +import { ProcessRunner } from "../processRunner.ts"; +import * as PortScanner from "./PortScanner.ts"; const { parseLsofOutput, parsePortFromLsofName, parseWindowsListenerOutput, serversEqual } = - __testing; + PortScanner.__testing; const TestProcessRunner = Layer.succeed(ProcessRunner, { run: () => Effect.die("ProcessRunner should not be used by Windows TCP probe tests"), }); -const TestPortDiscoveryLive = PortDiscoveryLive.pipe(Layer.provide(TestProcessRunner)); +const TestPortDiscoveryLive = PortScanner.layer.pipe(Layer.provide(TestProcessRunner)); describe("parsePortFromLsofName", () => { it("parses *:port", () => { @@ -189,7 +188,7 @@ describe("PortDiscovery integration (TCP probe fallback)", () => { beforeEach(async () => { originalPlatform = process.platform; Object.defineProperty(process, "platform", { value: "win32", configurable: true }); - for (const candidate of COMMON_DEV_PORTS) { + for (const candidate of PortScanner.COMMON_DEV_PORTS) { const candidateServer = net.createServer(); const listening = await new Promise((resolve) => { candidateServer.once("error", () => resolve(false)); @@ -215,7 +214,7 @@ describe("PortDiscovery integration (TCP probe fallback)", () => { effectIt.effect("scan() returns a server we just opened on a curated dev port", () => Effect.gen(function* () { - const scanner = yield* PortDiscovery; + const scanner = yield* PortScanner.PortDiscovery; const result = yield* scanner.scan(); const found = result.find((server) => server.port === port); expect(found).toBeDefined(); @@ -226,7 +225,7 @@ describe("PortDiscovery integration (TCP probe fallback)", () => { effectIt.effect("retain() drives an immediate broadcast to subscribers", () => { const received: number[] = []; return Effect.gen(function* () { - const scanner = yield* PortDiscovery; + const scanner = yield* PortScanner.PortDiscovery; const unsubscribe = yield* scanner.subscribe((servers) => Effect.sync(() => { for (const server of servers) received.push(server.port); diff --git a/apps/server/src/preview/Layers/PortScanner.ts b/apps/server/src/preview/PortScanner.ts similarity index 69% rename from apps/server/src/preview/Layers/PortScanner.ts rename to apps/server/src/preview/PortScanner.ts index 4e4d3b758a313c1b63e0284fef4bbf2292507615..008de8ae798539dfc5a326e16a4368b5753aa156 100644 --- a/apps/server/src/preview/Layers/PortScanner.ts +++ b/apps/server/src/preview/PortScanner.ts @@ -15,14 +15,34 @@ import * as net from "node:net"; import { ThreadId, type DiscoveredLocalServer } from "@t3tools/contracts"; import { LSOF_LOCAL_HOST_TOKENS } from "@t3tools/shared/preview"; -import { Cause, Duration, Effect, Layer, Ref, Schedule } from "effect"; - -import { ProcessRunner } from "../../processRunner.ts"; -import { - COMMON_DEV_PORTS, - PortDiscovery, - type PortDiscoveryShape, -} from "../Services/PortScanner.ts"; +import { Cause, Context, Duration, Effect, Layer, Ref, Schedule } from "effect"; + +import { ProcessRunner } from "../processRunner.ts"; + +export interface PortDiscoveryShape { + readonly scan: () => Effect.Effect>; + readonly subscribe: ( + listener: (servers: ReadonlyArray) => Effect.Effect, + ) => Effect.Effect<() => void>; + readonly retain: () => Effect.Effect<() => void>; + readonly registerTerminalProcesses: (input: { + readonly threadId: string; + readonly terminalId: string; + readonly processIds: ReadonlyArray; + }) => Effect.Effect; + readonly unregisterTerminal: (input: { + readonly threadId: string; + readonly terminalId: string; + }) => Effect.Effect; +} + +export class PortDiscovery extends Context.Service()( + "t3/preview/PortScanner/PortDiscovery", +) {} + +export const COMMON_DEV_PORTS: ReadonlyArray = Object.freeze([ + 3000, 3001, 3333, 4173, 4200, 4321, 5000, 5173, 5174, 5175, 5500, 8000, 8080, 8081, 8888, 9000, +]); const POLL_INTERVAL = Duration.seconds(3); const TCP_PROBE_TIMEOUT_MS = 200; @@ -186,7 +206,7 @@ const serversEqual = ( return true; }; -export const makePortDiscovery = Effect.gen(function* () { +const make = Effect.gen(function* () { const processRunner = yield* ProcessRunner; const stateRef = yield* Ref.make({ lastSnapshot: [], @@ -204,47 +224,46 @@ export const makePortDiscovery = Effect.gen(function* () { // be a synchronous side-effect-only function. let retainCount = 0; - const scanOnce = (): Effect.Effect> => - Effect.gen(function* () { - const terminalByProcessId = new Map(); - for (const registration of terminalProcesses.values()) { - for (const processId of registration.processIds) { - terminalByProcessId.set(processId, registration.owner); - } + const scanOnce = Effect.fn("PortDiscovery.scan")(function* () { + const terminalByProcessId = new Map(); + for (const registration of terminalProcesses.values()) { + for (const processId of registration.processIds) { + terminalByProcessId.set(processId, registration.owner); } - if (process.platform === "win32") { - const command = - 'Get-NetTCPConnection -State Listen -ErrorAction Stop | ForEach-Object { $processName = (Get-Process -Id $_.OwningProcess -ErrorAction SilentlyContinue).ProcessName; Write-Output "$($_.LocalAddress)|$($_.LocalPort)|$($_.OwningProcess)|$processName" }'; - const listeners = yield* processRunner - .run({ - command: "powershell.exe", - args: ["-NoProfile", "-NonInteractive", "-Command", command], - timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS), - maxOutputBytes: 1024 * 1024, - outputMode: "truncate", - }) - .pipe( - Effect.map((result) => parseWindowsListenerOutput(result.stdout, terminalByProcessId)), - Effect.catchCause(() => Effect.succeed(null)), - ); - if (listeners !== null) return listeners; - return yield* probeCommonPorts(); - } - const lsofResult = yield* processRunner + } + if (process.platform === "win32") { + const command = + 'Get-NetTCPConnection -State Listen -ErrorAction Stop | ForEach-Object { $processName = (Get-Process -Id $_.OwningProcess -ErrorAction SilentlyContinue).ProcessName; Write-Output "$($_.LocalAddress)|$($_.LocalPort)|$($_.OwningProcess)|$processName" }'; + const listeners = yield* processRunner .run({ - command: "lsof", - args: ["-iTCP", "-sTCP:LISTEN", "-P", "-n", "-F", "pcn"], - timeout: Duration.millis(LSOF_TIMEOUT_MS), + command: "powershell.exe", + args: ["-NoProfile", "-NonInteractive", "-Command", command], + timeout: Duration.millis(WINDOWS_LISTENER_TIMEOUT_MS), maxOutputBytes: 1024 * 1024, outputMode: "truncate", }) .pipe( - Effect.map((result) => parseLsofOutput(result.stdout, terminalByProcessId)), + Effect.map((result) => parseWindowsListenerOutput(result.stdout, terminalByProcessId)), Effect.catchCause(() => Effect.succeed(null)), ); - if (lsofResult !== null) return lsofResult; + if (listeners !== null) return listeners; return yield* probeCommonPorts(); - }); + } + const lsofResult = yield* processRunner + .run({ + command: "lsof", + args: ["-iTCP", "-sTCP:LISTEN", "-P", "-n", "-F", "pcn"], + timeout: Duration.millis(LSOF_TIMEOUT_MS), + maxOutputBytes: 1024 * 1024, + outputMode: "truncate", + }) + .pipe( + Effect.map((result) => parseLsofOutput(result.stdout, terminalByProcessId)), + Effect.catchCause(() => Effect.succeed(null)), + ); + if (lsofResult !== null) return lsofResult; + return yield* probeCommonPorts(); + }); const broadcast = (servers: ReadonlyArray): Effect.Effect => Effect.forEach(Array.from(listeners), (listener) => listener(servers), { discard: true }); @@ -266,33 +285,37 @@ export const makePortDiscovery = Effect.gen(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.gen(function* () { - const wasIdle = retainCount === 0; - 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.fn("PortDiscovery.retain")(function* () { + const wasIdle = retainCount === 0; + 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 subscribe: PortDiscoveryShape["subscribe"] = (listener) => - Effect.sync(() => { - listeners.add(listener); - return () => { - listeners.delete(listener); - }; - }); + const subscribe: PortDiscoveryShape["subscribe"] = Effect.fn("PortDiscovery.subscribe")( + function* (listener) { + return yield* Effect.sync(() => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }); + }, + ); - const registerTerminalProcesses: PortDiscoveryShape["registerTerminalProcesses"] = (input) => - Effect.sync(() => { + const registerTerminalProcesses: PortDiscoveryShape["registerTerminalProcesses"] = Effect.fn( + "PortDiscovery.registerTerminalProcesses", + )(function* (input) { + yield* Effect.sync(() => { const owner = { threadId: ThreadId.make(input.threadId), terminalId: input.terminalId, @@ -307,11 +330,15 @@ export const makePortDiscovery = Effect.gen(function* () { } terminalProcesses.set(key, { owner, processIds }); }); + }); - const unregisterTerminal: PortDiscoveryShape["unregisterTerminal"] = (input) => - Effect.sync(() => { + const unregisterTerminal: PortDiscoveryShape["unregisterTerminal"] = Effect.fn( + "PortDiscovery.unregisterTerminal", + )(function* (input) { + yield* Effect.sync(() => { terminalProcesses.delete(terminalOwnerKey(input)); }); + }); return { scan: scanOnce, @@ -322,7 +349,7 @@ export const makePortDiscovery = Effect.gen(function* () { } satisfies PortDiscoveryShape; }); -export const PortDiscoveryLive = Layer.effect(PortDiscovery, makePortDiscovery); +export const layer = Layer.effect(PortDiscovery, make); /** Exposed for tests. */ export const __testing = { diff --git a/apps/server/src/preview/Services/Manager.ts b/apps/server/src/preview/Services/Manager.ts deleted file mode 100644 index 7c80c21c542434d3e5c0a147f5013081d8631f64..0000000000000000000000000000000000000000 --- a/apps/server/src/preview/Services/Manager.ts +++ /dev/null @@ -1,94 +0,0 @@ -/** - * PreviewManager - Per-thread browser preview session orchestration. - * - * Owns metadata-only preview sessions keyed by `(threadId, tabId)`. A thread - * can host multiple concurrent tabs (browser-style). The actual Chromium - * lives in the desktop renderer; this service tracks the canonical - * URL/title/nav state per tab so sessions survive reconnects and are visible - * to multiple connected clients. - * - * Event delivery uses Effect's `PubSub` so listener failures (e.g. a - * disconnected WS subscriber's queue being closed) cannot propagate back - * into the mutating call's failure channel — matches the codebase - * convention used by `SessionCredentialService`. - * - * @module PreviewManager - */ -import { - type PreviewCloseInput, - type PreviewError, - type PreviewEvent, - type PreviewListInput, - type PreviewListResult, - type PreviewNavigateInput, - type PreviewOpenInput, - type PreviewRefreshInput, - type PreviewReportStatusInput, - type PreviewSessionSnapshot, -} from "@t3tools/contracts"; -import { Context, type Effect, type PubSub, type Scope, type Stream } from "effect"; - -export interface PreviewManagerShape { - /** - * Open a brand new preview tab for `threadId`. When `url` is omitted the - * tab starts in the `Idle` state so the user can type into the URL bar; - * otherwise it transitions straight to `Loading`. Always emits `opened`. - */ - readonly open: (input: PreviewOpenInput) => Effect.Effect; - - /** - * Update the session's URL/title from the renderer (after the - * resolved navigation, including redirects). Emits a `navigated` event. - */ - readonly navigate: ( - input: PreviewNavigateInput, - ) => Effect.Effect; - - /** - * Renderer reports current nav status (Loading/Success/LoadFailed) and - * back/forward availability. Drives `failed` and `navigated` events. - */ - readonly reportStatus: (input: PreviewReportStatusInput) => Effect.Effect; - - /** - * Renderer requested a reload. Server records intent (no state change) so - * subscribers can reconcile if needed; the desktop bridge actually reloads. - */ - readonly refresh: (input: PreviewRefreshInput) => Effect.Effect; - - /** - * Close the session (drop server-side state, emit `closed`). When `tabId` - * is omitted, closes every session for the thread. - */ - readonly close: (input: PreviewCloseInput) => Effect.Effect; - - /** - * List active preview sessions for a thread. Returns an empty array when - * the thread has no sessions. - */ - readonly list: (input: PreviewListInput) => Effect.Effect; - - /** - * Stream of preview events. Each subscriber gets its own buffered stream; - * subscriber failures are isolated and do not affect other subscribers - * or publishers. - * - * NOTE: `Stream.fromPubSub` defers `PubSub.subscribe` until the stream - * starts running, so `Stream.runForEach(...).pipe(Effect.forkScoped)` may - * miss a publish landing between fork and stream-start. Callers that - * cannot tolerate this gap should use `subscribeEvents` below. - */ - readonly events: Stream.Stream; - - /** - * Acquire a PubSub subscription synchronously in the caller's fiber so - * no publish can land in the narrow gap between subscribe and consumer - * loop start. The subscription's lifetime is bound to the provided - * `Scope`; release happens automatically on scope close. - */ - readonly subscribeEvents: Effect.Effect, never, Scope.Scope>; -} - -export class PreviewManager extends Context.Service()( - "t3/preview/Services/Manager/PreviewManager", -) {} diff --git a/apps/server/src/preview/Services/PortScanner.ts b/apps/server/src/preview/Services/PortScanner.ts deleted file mode 100644 index af69596ba5a752b7c6983aa8e945f56941daf7ae..0000000000000000000000000000000000000000 --- a/apps/server/src/preview/Services/PortScanner.ts +++ /dev/null @@ -1,46 +0,0 @@ -/** - * PortDiscovery - Discovers listening localhost ports and attributes them to - * registered terminal process families. - * - * Reference-counted polling: the scanner only runs when at least one client - * has called `retain()`. This keeps idle desktops from running `lsof` every - * 3 seconds for nothing. - * - * @module PortDiscovery - */ -import type { DiscoveredLocalServer } from "@t3tools/contracts"; -import { Context, type Effect } from "effect"; - -export interface PortDiscoveryShape { - /** One-shot snapshot of currently listening localhost ports. */ - readonly scan: () => Effect.Effect>; - /** Subscribe to changes. Listener invoked on every diff. Returns unsubscribe fn. */ - readonly subscribe: ( - listener: (servers: ReadonlyArray) => Effect.Effect, - ) => Effect.Effect<() => void>; - /** - * Hint that at least one client is interested → starts polling. Returns - * release fn. When release count hits 0, polling stops. - */ - readonly retain: () => Effect.Effect<() => void>; - /** Associate a terminal with its current process family for port attribution. */ - readonly registerTerminalProcesses: (input: { - readonly threadId: string; - readonly terminalId: string; - readonly processIds: ReadonlyArray; - }) => Effect.Effect; - /** Remove process attribution for a terminal that stopped or closed. */ - readonly unregisterTerminal: (input: { - readonly threadId: string; - readonly terminalId: string; - }) => Effect.Effect; -} - -export class PortDiscovery extends Context.Service()( - "t3/preview/Services/PortScanner/PortDiscovery", -) {} - -/** Curated list of common dev-server ports for the Windows TCP-probe fallback. */ -export const COMMON_DEV_PORTS: ReadonlyArray = Object.freeze([ - 3000, 3001, 3333, 4173, 4200, 4321, 5000, 5173, 5174, 5175, 5500, 8000, 8080, 8081, 8888, 9000, -]); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index b7cdd931ca632843cab3b36c2e3418db6a543407..c91f305b174c6565160f5eb795c96492a39f0ff2 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -69,7 +69,7 @@ import * as Stream from "effect/Stream"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; -import { readMcpProviderSession } from "../../mcp/Services/McpProviderSession.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { makeClaudeEnvironment } from "../Drivers/ClaudeHome.ts"; import { getClaudeModelCapabilities, @@ -3446,7 +3446,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(fastMode ? { fastMode: true } : {}), ...(ultracode ? { ultracode: true } : {}), }; - const mcpSession = readMcpProviderSession(input.threadId); + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); const queryOptions: ClaudeQueryOptions = { ...(input.cwd ? { cwd: input.cwd } : {}), ...(apiModelId ? { model: apiModelId } : {}), diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index edb2bccdbcf6f954772b75ad7a559d1ab3933f27..270126e934ba44327625d10c43c00d7bc0cf7d0e 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -39,7 +39,7 @@ import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { getCodexServiceTierOptionValue } from "../../codexModelOptions.ts"; -import { readMcpProviderSession } from "../../mcp/Services/McpProviderSession.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { ProviderAdapterRequestError, @@ -1386,7 +1386,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( input.modelSelection?.instanceId === boundInstanceId ? getCodexServiceTierOptionValue(input.modelSelection) : undefined; - const mcpSession = readMcpProviderSession(input.threadId); + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); const runtimeInput: CodexSessionRuntimeOptions = { threadId: input.threadId, providerInstanceId: boundInstanceId, diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index c9af2de35578adecaf9c22fd07dd238802d41b80..1560332ad7f8bcc21f574338409960928432428d 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -42,7 +42,7 @@ import type * as EffectAcpSchema from "effect-acp/schema"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; -import { readMcpProviderSession } from "../../mcp/Services/McpProviderSession.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { ProviderAdapterProcessError, ProviderAdapterRequestError, @@ -531,7 +531,7 @@ export function makeCursorAdapter( ? yield* options.resolveSettings : cursorSettings; - const mcpSession = readMcpProviderSession(input.threadId); + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); const acp = yield* makeCursorAcpRuntime({ cursorSettings: effectiveCursorSettings, ...(options?.environment ? { environment: options.environment } : {}), diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 35cf7fb98e9b06f586acddcf8fc4859d8db0b390..a21a2bb9fc7769c25c462753d8c5406a22f29c61 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -33,7 +33,7 @@ import type * as EffectAcpSchema from "effect-acp/schema"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; -import { readMcpProviderSession } from "../../mcp/Services/McpProviderSession.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { ProviderAdapterProcessError, ProviderAdapterRequestError, @@ -375,7 +375,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte threadId: input.threadId, }); - const mcpSession = readMcpProviderSession(input.threadId); + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); const acp = yield* makeGrokAcpRuntime({ grokSettings, ...(options?.environment ? { environment: options.environment } : {}), diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 7361638924d8d4f2eb871d660cfc11e783d6fe48..1eb6e47bc19fe916d3c05e2cbaee3cd16756ed23 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -26,7 +26,7 @@ import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; -import { readMcpProviderSession } from "../../mcp/Services/McpProviderSession.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; import { ProviderAdapterProcessError, @@ -1054,7 +1054,7 @@ export function makeOpenCodeAdapter( directory, ...(server.external && serverPassword ? { serverPassword } : {}), }); - const mcpSession = readMcpProviderSession(input.threadId); + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); if (mcpSession && !server.external) { yield* runOpenCodeSdk("mcp.add", () => client.mcp.add({ diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 7d653419b27366fdc297806c3ae822ff9dc9e176..ecb1dd2dbd3f3b635f277aebfcbcb9d3b03d00f0 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -56,16 +56,8 @@ import { import { type EventNdjsonLogger } from "./EventNdjsonLogger.ts"; import { ProviderEventLoggers } from "./ProviderEventLoggers.ts"; import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; -import { - issueActiveMcpCredential, - revokeActiveMcpThread, - revokeAllActiveMcpCredentials, -} from "../../mcp/Layers/McpSessionRegistry.ts"; -import { - clearAllMcpProviderSessions, - clearMcpProviderSession, - setMcpProviderSession, -} from "../../mcp/Services/McpProviderSession.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import * as McpSessionRegistry from "../../mcp/McpSessionRegistry.ts"; const isModelSelection = Schema.is(ModelSelection); /** @@ -223,14 +215,16 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const runtimeEventPubSub = yield* PubSub.unbounded(); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => - issueActiveMcpCredential({ threadId, providerInstanceId }).pipe( + McpSessionRegistry.issueActiveMcpCredential({ threadId, providerInstanceId }).pipe( Effect.tap((credential) => - credential ? Effect.sync(() => setMcpProviderSession(credential.config)) : Effect.void, + credential + ? Effect.sync(() => McpProviderSession.setMcpProviderSession(credential.config)) + : Effect.void, ), ); const clearMcpSession = (threadId: ThreadId) => - revokeActiveMcpThread(threadId).pipe( - Effect.tap(() => Effect.sync(() => clearMcpProviderSession(threadId))), + McpSessionRegistry.revokeActiveMcpThread(threadId).pipe( + Effect.tap(() => Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId))), ); const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => @@ -1027,8 +1021,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ).pipe(Effect.asVoid); yield* Effect.forEach(currentAdapters, ([, adapter]) => adapter.stopAll()).pipe(Effect.asVoid); - yield* revokeAllActiveMcpCredentials(); - clearAllMcpProviderSessions(); + yield* McpSessionRegistry.revokeAllActiveMcpCredentials(); + McpProviderSession.clearAllMcpProviderSessions(); const bindings = yield* directory.listBindings().pipe(Effect.orElseSucceed(() => [])); yield* Effect.forEach(bindings, (binding) => Effect.gen(function* () { diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index e6d0f35cea7e8bd369cc25024f83e8e704c745d5..b2166d6f0b224024de03b33b1371a9c4edee0c32 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -99,8 +99,8 @@ import { ServerLifecycleEvents, type ServerLifecycleEventsShape } from "./server import { ServerRuntimeStartup, type ServerRuntimeStartupShape } from "./serverRuntimeStartup.ts"; import { ServerSettingsService, type ServerSettingsShape } from "./serverSettings.ts"; import { TerminalManager, type TerminalManagerShape } from "./terminal/Services/Manager.ts"; -import { PreviewManager } from "./preview/Services/Manager.ts"; -import { PortDiscovery } from "./preview/Services/PortScanner.ts"; +import * as PreviewManager from "./preview/Manager.ts"; +import * as PortScanner from "./preview/PortScanner.ts"; import { BrowserTraceCollector, type BrowserTraceCollectorShape, @@ -669,7 +669,7 @@ const buildAppUnderTest = (options?: { ), Layer.provide( Layer.mergeAll( - Layer.mock(PreviewManager)({ + Layer.mock(PreviewManager.PreviewManager)({ open: () => Effect.die("PreviewManager not stubbed in this test"), navigate: () => Effect.die("PreviewManager not stubbed in this test"), reportStatus: () => Effect.void, @@ -681,7 +681,7 @@ const buildAppUnderTest = (options?: { PubSub.subscribe(pubsub), ), }), - Layer.mock(PortDiscovery)({ + Layer.mock(PortScanner.PortDiscovery)({ scan: () => Effect.succeed([]), subscribe: () => Effect.succeed(() => {}), retain: () => Effect.succeed(() => {}), diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index c1a1298a1aece658d6bbfa75eb39908a3738274f..c565c91f1a49ceb749434bc6c0f6c5ca3bfd3e51 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -35,10 +35,10 @@ import * as GitLabCli from "./sourceControl/GitLabCli.ts"; import * as TextGeneration from "./textGeneration/TextGeneration.ts"; import { ProviderInstanceRegistryHydrationLive } from "./provider/Layers/ProviderInstanceRegistryHydration.ts"; import { TerminalManagerLive } from "./terminal/Layers/Manager.ts"; -import { PreviewManagerLive } from "./preview/Layers/Manager.ts"; -import { PortDiscoveryLive } from "./preview/Layers/PortScanner.ts"; -import { McpHttpServerLive } from "./mcp/Layers/McpHttpServer.ts"; -import { McpSessionRegistryLive } from "./mcp/Layers/McpSessionRegistry.ts"; +import * as McpHttpServer from "./mcp/McpHttpServer.ts"; +import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts"; +import * as PreviewManager from "./preview/Manager.ts"; +import * as PortScanner from "./preview/PortScanner.ts"; import * as ProcessRunner from "./processRunner.ts"; import * as GitManager from "./git/GitManager.ts"; import { KeybindingsLive } from "./keybindings.ts"; @@ -236,14 +236,17 @@ const CheckpointingLayerLive = Layer.empty.pipe( Layer.provideMerge(CheckpointStoreLive.pipe(Layer.provide(VcsDriverRegistryLayerLive))), ); -const PortScannerLayerLive = PortDiscoveryLive.pipe(Layer.provide(ProcessRunner.layer)); +const PortScannerLayerLive = PortScanner.layer.pipe(Layer.provide(ProcessRunner.layer)); const TerminalLayerLive = TerminalManagerLive.pipe( Layer.provide(PtyAdapterLive), Layer.provide(PortScannerLayerLive), ); -const PreviewLayerLive = Layer.mergeAll(PreviewManagerLive, PortScannerLayerLive); +const PreviewLayerLive = Layer.empty.pipe( + Layer.provideMerge(PreviewManager.layer), + Layer.provideMerge(PortScannerLayerLive), +); const WorkspaceEntriesLayerLive = WorkspaceEntriesLive.pipe( Layer.provide(WorkspacePathsLive), @@ -353,7 +356,7 @@ export const makeRoutesLayer = Layer.mergeAll( staticAndDevRouteLayer, websocketRpcRouteLayer, ), - McpHttpServerLive.pipe(Layer.provide(McpSessionRegistryLive)), + McpHttpServer.layer.pipe(Layer.provide(McpSessionRegistry.layer)), ).pipe(Layer.provide(browserApiCorsLayer)); export const makeServerLayer = Layer.unwrap( diff --git a/apps/server/src/terminal/Layers/Manager.ts b/apps/server/src/terminal/Layers/Manager.ts index acde17fc85fc40750e897369fb9ef196bafde053..f2b466a43900da344eef3304b5a93db3da8e75c3 100644 --- a/apps/server/src/terminal/Layers/Manager.ts +++ b/apps/server/src/terminal/Layers/Manager.ts @@ -33,7 +33,7 @@ import { terminalSessionsTotal, } from "../../observability/Metrics.ts"; import * as ProcessRunner from "../../processRunner.ts"; -import { PortDiscovery } from "../../preview/Services/PortScanner.ts"; +import * as PortScanner from "../../preview/PortScanner.ts"; import { TerminalCwdError, TerminalHistoryError, @@ -996,7 +996,7 @@ interface TerminalManagerOptions { const makeTerminalManager = Effect.fn("makeTerminalManager")(function* () { const { terminalLogsDir } = yield* ServerConfig; const ptyAdapter = yield* PtyAdapter; - const portDiscovery = yield* PortDiscovery; + const portDiscovery = yield* PortScanner.PortDiscovery; return yield* makeTerminalManagerWithOptions({ logsDir: terminalLogsDir, ptyAdapter, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 04ac8f5a62b29aa1a7542ab2e811969ca887fbcd..c8c3c5de46e9c22312fa8f537015ab8c723e0e26 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -71,9 +71,9 @@ import { ServerLifecycleEvents } from "./serverLifecycleEvents.ts"; import { ServerRuntimeStartup } from "./serverRuntimeStartup.ts"; import { redactServerSettingsForClient, ServerSettingsService } from "./serverSettings.ts"; import { TerminalManager } from "./terminal/Services/Manager.ts"; -import { PreviewManager } from "./preview/Services/Manager.ts"; -import { PortDiscovery } from "./preview/Services/PortScanner.ts"; -import { previewAutomationBroker } from "./mcp/Layers/PreviewAutomationBroker.ts"; +import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; +import * as PreviewManager from "./preview/Manager.ts"; +import * as PortScanner from "./preview/PortScanner.ts"; import { WorkspaceEntries } from "./workspace/Services/WorkspaceEntries.ts"; import { WorkspaceFileSystem } from "./workspace/Services/WorkspaceFileSystem.ts"; import { WorkspacePathOutsideRootError } from "./workspace/Services/WorkspacePaths.ts"; @@ -256,8 +256,9 @@ const makeWsRpcLayer = (currentSession: AuthenticatedSession) => const vcsProvisioning = yield* VcsProvisioningService; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster; const terminalManager = yield* TerminalManager; - const previewManager = yield* PreviewManager; - const portDiscovery = yield* PortDiscovery; + const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker; + const previewManager = yield* PreviewManager.PreviewManager; + const portDiscovery = yield* PortScanner.PortDiscovery; const providerRegistry = yield* ProviderRegistry; const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const config = yield* ServerConfig; @@ -1573,6 +1574,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( Effect.provide( makeWsRpcLayer(session).pipe( Layer.provideMerge(RpcSerialization.layerJson), + Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ProviderMaintenanceRunner.layer), Layer.provide( SourceControlDiscoveryLayer.layer.pipe( -- 2.54.0