refactor(core): canonicalize pty service (#32182)

2026-06-14 16:16:39 +05:30
parent ca6a8dfccc
commit ccd64b6949
30 changed files with 1134 additions and 506 deletions
@@ -1,23 +1,22 @@
 import * as InstanceState from "@/effect/instance-state"
 import { registerDisposer } from "@/effect/instance-registry"
 import { InstanceRef, WorkspaceRef } from "@/effect/instance-ref"
-import { PtyPreparation } from "@/pty-preparation"
+import { Plugin } from "@/plugin"
 import { Pty } from "@opencode-ai/core/pty"
-import { handlePtyInput } from "@opencode-ai/core/pty/input"
+import { PtyProtocol } from "@opencode-ai/core/pty/protocol"
 import { PtyID } from "@opencode-ai/core/pty/schema"
 import { PtyTicket } from "@opencode-ai/core/pty/ticket"
 import { LocationServiceMap } from "@opencode-ai/core/location-layer"
 import { Location } from "@opencode-ai/core/location"
 import { AbsolutePath } from "@opencode-ai/core/schema"
-import { Shell } from "@/shell/shell"
-import { EffectBridge } from "@/effect/bridge"
-import { CorsConfig, isAllowedRequestOrigin, type CorsOptions } from "@/server/cors"
+import { Shell } from "@opencode-ai/core/shell"
+import { CorsConfig, isAllowedRequestOrigin, type CorsOptions } from "@opencode-ai/server/cors"
 import {
  PTY_CONNECT_TICKET_QUERY,
  PTY_CONNECT_TOKEN_HEADER,
  PTY_CONNECT_TOKEN_HEADER_VALUE,
 } from "@/server/shared/pty-ticket"
-import { Effect, Layer, Option, Schema } from "effect"
+import { Effect, Layer, Option, Queue, Schema } from "effect"
 import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
 import { HttpApiBuilder } from "effect/unstable/httpapi"
 import * as Socket from "effect/unstable/socket/Socket"
@@ -36,10 +35,14 @@ const ticketScope = Effect.gen(function* () {
  return { directory: instance?.directory, workspaceID }
 })

+// Legacy surface compatibility: before exited-session retention, sessions vanished the moment
+// their process exited. These routes preserve that observable behavior — exited sessions are
+// invisible here — while the canonical /api/pty surface exposes them until removal.
 export const ptyHandlers = HttpApiBuilder.group(InstanceHttpApi, "pty", (handlers) =>
  Effect.gen(function* () {
    const tickets = yield* PtyTicket.Service
    const cors = yield* CorsConfig
+    const plugin = yield* Plugin.Service
    const locations = yield* LocationServiceMap
    const unregister = registerDisposer((directory) =>
      Effect.runPromise(locations.invalidate(Location.Ref.make({ directory: AbsolutePath.make(directory) }))),
@@ -59,33 +62,42 @@ export const ptyHandlers = HttpApiBuilder.group(InstanceHttpApi, "pty", (handler
    })

    const list = Effect.fn("PtyHttpApi.list")(function* () {
-      return yield* pty(Pty.Service.use((service) => service.list()))
+      const sessions = yield* pty(Pty.Service.use((service) => service.list()))
+      return sessions.filter((info) => info.status === "running")
    })

    const create = Effect.fn("PtyHttpApi.create")(function* (ctx: { payload: typeof Pty.CreateInput.Type }) {
+      const cwd = ctx.payload.cwd || (yield* InstanceState.context).directory
+      const shell = yield* plugin.trigger("shell.env", { cwd }, { env: {} as Record<string, string> })
      return yield* pty(
        Pty.Service.use((service) =>
-          Effect.flatMap(
-            PtyPreparation.prepareCreate({
-              ...ctx.payload,
-              args: ctx.payload.args ? [...ctx.payload.args] : undefined,
-              env: ctx.payload.env ? { ...ctx.payload.env } : undefined,
-            }),
-            service.create,
-          ),
+          service.create({
+            ...ctx.payload,
+            args: ctx.payload.args ? [...ctx.payload.args] : undefined,
+            cwd,
+            env: { ...ctx.payload.env, ...shell.env },
+          }),
        ),
      )
    })

    const get = Effect.fn("PtyHttpApi.get")(function* (ctx: { params: { ptyID: PtyID } }) {
      return yield* pty(Pty.Service.use((service) => service.get(ctx.params.ptyID))).pipe(
-        Effect.catchTag("Pty.NotFoundError", (error) =>
-          Effect.fail(
+        Effect.catchTag(
+          "Pty.NotFoundError",
+          (error) =>
            new ApiError.PtyNotFoundError({
              ptyID: error.ptyID,
              message: `PTY session not found: ${error.ptyID}`,
            }),
-          ),
+        ),
+        Effect.flatMap((info) =>
+          info.status === "running"
+            ? Effect.succeed(info)
+            : new ApiError.PtyNotFoundError({
+                ptyID: ctx.params.ptyID,
+                message: `PTY session not found: ${ctx.params.ptyID}`,
+              }),
        ),
      )
    })
@@ -94,6 +106,7 @@ export const ptyHandlers = HttpApiBuilder.group(InstanceHttpApi, "pty", (handler
      params: { ptyID: PtyID }
      payload: typeof Pty.UpdateInput.Type
    }) {
+      yield* get(ctx)
      return yield* pty(
        Pty.Service.use((service) =>
          service.update(ctx.params.ptyID, {
@@ -102,26 +115,27 @@ export const ptyHandlers = HttpApiBuilder.group(InstanceHttpApi, "pty", (handler
          }),
        ),
      ).pipe(
-        Effect.catchTag("Pty.NotFoundError", (error) =>
-          Effect.fail(
+        Effect.catchTag(
+          "Pty.NotFoundError",
+          (error) =>
            new ApiError.PtyNotFoundError({
              ptyID: error.ptyID,
              message: `PTY session not found: ${error.ptyID}`,
            }),
-          ),
        ),
      )
    })

    const remove = Effect.fn("PtyHttpApi.remove")(function* (ctx: { params: { ptyID: PtyID } }) {
+      yield* get(ctx)
      yield* pty(Pty.Service.use((service) => service.remove(ctx.params.ptyID))).pipe(
-        Effect.catchTag("Pty.NotFoundError", (error) =>
-          Effect.fail(
+        Effect.catchTag(
+          "Pty.NotFoundError",
+          (error) =>
            new ApiError.PtyNotFoundError({
              ptyID: error.ptyID,
              message: `PTY session not found: ${error.ptyID}`,
            }),
-          ),
        ),
      )
      return true
@@ -131,16 +145,7 @@ export const ptyHandlers = HttpApiBuilder.group(InstanceHttpApi, "pty", (handler
      const request = yield* HttpServerRequest.HttpServerRequest
      if (request.headers[PTY_CONNECT_TOKEN_HEADER] !== PTY_CONNECT_TOKEN_HEADER_VALUE || !validOrigin(request, cors))
        return yield* new ApiError.PtyForbiddenError({ message: "Invalid PTY connect token request" })
-      yield* pty(Pty.Service.use((service) => service.get(ctx.params.ptyID))).pipe(
-        Effect.catchTag("Pty.NotFoundError", (error) =>
-          Effect.fail(
-            new ApiError.PtyNotFoundError({
-              ptyID: error.ptyID,
-              message: `PTY session not found: ${error.ptyID}`,
-            }),
-          ),
-        ),
-      )
+      yield* get(ctx)
      return yield* tickets.issue({ ptyID: ctx.params.ptyID, ...(yield* ticketScope) })
    })

@@ -180,7 +185,7 @@ export const ptyConnectHandlers = HttpApiBuilder.group(PtyConnectApi, "pty-conne
        request: HttpServerRequest.HttpServerRequest
      }) {
        const exists = yield* pty(Pty.Service.use((service) => service.get(ctx.params.ptyID))).pipe(
-          Effect.as(true),
+          Effect.map((info) => info.status === "running"),
          Effect.catchTag("Pty.NotFoundError", () => Effect.succeed(false)),
        )
        if (!exists) return HttpServerResponse.empty({ status: 404 })
@@ -214,48 +219,53 @@ export const ptyConnectHandlers = HttpApiBuilder.group(PtyConnectApi, "pty-conne
          yield* closeAccepted(WebSocketTracker.SERVER_CLOSING_EVENT())
          return HttpServerResponse.empty()
        }
-        const bridge = yield* EffectBridge.make()
-        const writeScoped = (effect: Effect.Effect<void, unknown>) => {
-          bridge.fork(effect.pipe(Effect.catch(() => Effect.void)))
-        }
-        let closed = false
-        const adapter = {
-          get readyState() {
-            return closed ? 3 : 1
-          },
-          send: (data: string | Uint8Array | ArrayBuffer) => {
-            if (closed) return
-            writeScoped(write(data instanceof ArrayBuffer ? new Uint8Array(data) : data))
-          },
-          close: (code?: number, reason?: string) => {
-            if (closed) return
-            closed = true
-            writeScoped(write(new Socket.CloseEvent(code, reason)))
-          },
-        }
-        const handler = yield* pty(
-          Pty.Service.use((service) => service.connect(ctx.params.ptyID, adapter, cursor)),
-        ).pipe(
-          Effect.catchTag("Pty.NotFoundError", () =>
-            closeAccepted(new Socket.CloseEvent(4404, "session not found")).pipe(Effect.as(undefined)),
-          ),
-        )
-        if (!handler) return HttpServerResponse.empty()

-        // The handshake runs inside `socket.runRaw`, after the input callback is
-        // registered, so the client cannot send frames before PTY input is wired.
-        yield* socket
-          .runRaw((message) => handlePtyInput(handler, message))
-          .pipe(
-            Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void),
-            Effect.ensuring(
-              Effect.sync(() => {
-                closed = true
-                handler.onClose()
-              }),
-            ),
-            Effect.orDie,
-          )
+        // Outbound frames flow through one queue drained by a single writer so replay, live
+        // output, and the close frame keep their order.
+        const outbox = yield* Queue.unbounded<string | Uint8Array | Socket.CloseEvent>()
+        const attachment = yield* pty(
+          Pty.Service.use((service) =>
+            service.attach(ctx.params.ptyID, {
+              cursor,
+              onData: (chunk) => Queue.offerUnsafe(outbox, chunk),
+              onEnd: () => Queue.offerUnsafe(outbox, new Socket.CloseEvent(1000)),
+            }),
+          ),
+        ).pipe(
+          Effect.catchTags({
+            "Pty.NotFoundError": () =>
+              closeAccepted(new Socket.CloseEvent(4404, "session not found")).pipe(Effect.as(undefined)),
+            "Pty.ExitedError": () =>
+              closeAccepted(new Socket.CloseEvent(4404, "session not found")).pipe(Effect.as(undefined)),
+          }),
+        )
+        if (!attachment) return HttpServerResponse.empty()
+
+        for (const chunk of PtyProtocol.chunks(attachment.replay)) Queue.offerUnsafe(outbox, chunk)
+        Queue.offerUnsafe(outbox, PtyProtocol.metaFrame(attachment.cursor))
+        attachment.activate()
+
+        const drain = Effect.gen(function* () {
+          while (true) {
+            const item = yield* Queue.take(outbox)
+            yield* write(item)
+            if (item instanceof Socket.CloseEvent) return
+          }
+        })
+
+        // The reader runs concurrently with the writer; whichever finishes first ends the
+        // connection and the attachment is always released.
+        yield* Effect.race(
+          drain,
+          socket.runRaw((message) => {
+            const decoded = PtyProtocol.decodeInput(message)
+            if (decoded !== undefined) attachment.write(decoded)
+          }),
+        ).pipe(
+          Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void),
+          Effect.ensuring(Effect.sync(() => attachment.detach())),
+          Effect.orDie,
+        )
        return HttpServerResponse.empty()
      }),
    )
@@ -61,7 +61,7 @@ import { PtyTicket } from "@opencode-ai/core/pty/ticket"
 import { Ripgrep } from "@opencode-ai/core/ripgrep"
 import { SessionProjector } from "@opencode-ai/core/session/projector"
 import { lazy } from "@/util/lazy"
-import { CorsConfig, isAllowedCorsOrigin, type CorsOptions } from "@/server/cors"
+import { CorsConfig, isAllowedCorsOrigin, type CorsOptions } from "@opencode-ai/server/cors"
 import { serveUIEffect } from "@/server/shared/ui"
 import { ServerAuth } from "@/server/auth"
 import { InstanceHttpApi, RootHttpApi } from "./api"