Apply PR #24853: Default dev and beta builds to HttpApi

Author: opencode-agent[bot]

packages/opencode/src/server/backend.ts

@@ -0,0 +1,40 @@
+import { Flag } from "@opencode-ai/core/flag/flag"
+import { InstallationChannel, InstallationVersion } from "@opencode-ai/core/installation/version"
+
+export type Backend = "effect-httpapi" | "hono"
+
+export type Selection = {
+  backend: Backend
+  reason: "env" | "channel" | "stable" | "explicit"
+}
+
+export type Attributes = ReturnType<typeof attributes>
+
+const channelDefaultsToHttpApi = () =>
+  InstallationChannel === "local" ||
+  InstallationChannel === "dev" ||
+  InstallationChannel === "beta" ||
+  InstallationVersion.includes("-dev") ||
+  InstallationVersion.includes("-beta")
+
+export function select(): Selection {
+  if (Flag.OPENCODE_EXPERIMENTAL_HTTPAPI) return { backend: "effect-httpapi", reason: "env" }
+  if (channelDefaultsToHttpApi()) return { backend: "effect-httpapi", reason: "channel" }
+  return { backend: "hono", reason: "stable" }
+}
+
+export function attributes(selection: Selection): Record<string, string> {
+  return {
+    "opencode.server.backend": selection.backend,
+    "opencode.server.backend.reason": selection.reason,
+    "opencode.installation.channel": InstallationChannel,
+    "opencode.installation.version": InstallationVersion,
+  }
+}
+
+export function force(selection: Selection, backend: Backend): Selection {
+  return {
+    backend,
+    reason: selection.backend === backend ? selection.reason : "explicit",
+  }
+}

packages/opencode/src/server/middleware.ts

@@ -10,6 +10,7 @@ import { Flag } from "@opencode-ai/core/flag/flag"
 import { basicAuth } from "hono/basic-auth"
 import { cors } from "hono/cors"
 import { compress } from "hono/compress"
+import * as ServerBackend from "./backend"
 
 const log = Log.create({ service: "server" })
 
@@ -49,20 +50,20 @@ export const AuthMiddleware: MiddlewareHandler = (c, next) => {
   return basicAuth({ username, password })(c, next)
 }
 
-export const LoggerMiddleware: MiddlewareHandler = async (c, next) => {
-  const skip = c.req.path === "/log"
-  if (!skip) {
-    log.info("request", {
+export function LoggerMiddleware(backendAttributes: ServerBackend.Attributes): MiddlewareHandler {
+  return async (c, next) => {
+    const skip = c.req.path === "/log"
+    if (skip) return next()
+    const attributes = {
       method: c.req.method,
       path: c.req.path,
-    })
+      ...backendAttributes,
+    }
+    log.info("request", attributes)
+    const timer = log.time("request", attributes)
+    await next()
+    timer.stop()
   }
-  const timer = log.time("request", {
-    method: c.req.method,
-    path: c.req.path,
-  })
-  await next()
-  if (!skip) timer.stop()
 }
 
 export function CorsMiddleware(opts?: { cors?: string[] }): MiddlewareHandler {

packages/opencode/src/server/routes/instance/httpapi/event.ts

@@ -4,6 +4,7 @@ import { Effect, Schema } from "effect"
 import * as Stream from "effect/Stream"
 import { HttpRouter, HttpServerResponse } from "effect/unstable/http"
 import { HttpApi, HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
+import * as Sse from "effect/unstable/encoding/Sse"
 
 const log = Log.create({ service: "server" })
 
@@ -27,8 +28,13 @@ export const EventApi = HttpApi.make("event").add(
     .annotateMerge(OpenApi.annotations({ title: "event", description: "Instance event stream route." })),
 )
 
-function eventData(data: unknown) {
-  return `data: ${JSON.stringify(data)}\n\n`
+function eventData(data: unknown): Sse.Event {
+  return {
+    _tag: "Event",
+    event: "message",
+    id: undefined,
+    data: JSON.stringify(data),
+  }
 }
 
 export const eventRoute = HttpRouter.add(
@@ -47,6 +53,7 @@ export const eventRoute = HttpRouter.add(
       Stream.make({ type: "server.connected", properties: {} }).pipe(
         Stream.concat(events.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))),
         Stream.map(eventData),
+        Stream.pipeThroughChannel(Sse.encode()),
         Stream.encodeText,
         Stream.ensuring(Effect.sync(() => log.info("event disconnected"))),
       ),

packages/opencode/src/server/routes/instance/httpapi/global.ts

@@ -4,9 +4,11 @@ import { Installation } from "@/installation"
 import { Instance } from "@/project/instance"
 import { InstallationVersion } from "@opencode-ai/core/installation/version"
 import * as Log from "@opencode-ai/core/util/log"
-import { Effect, Schema } from "effect"
+import { Effect, Queue, Schema } from "effect"
+import * as Stream from "effect/Stream"
 import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
 import { HttpApi, HttpApiBuilder, HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
+import * as Sse from "effect/unstable/encoding/Sse"
 
 const log = Log.create({ service: "server" })
 
@@ -108,8 +110,13 @@ export const GlobalApi = HttpApi.make("global").add(
     .annotateMerge(OpenApi.annotations({ title: "global", description: "Global server routes." })),
 )
 
-function eventData(data: unknown) {
-  return `data: ${JSON.stringify(data)}\n\n`
+function eventData(data: unknown): Sse.Event {
+  return {
+    _tag: "Event",
+    event: "message",
+    id: undefined,
+    data: JSON.stringify(data),
+  }
 }
 
 function parseBody(body: string) {
@@ -121,49 +128,35 @@ function parseBody(body: string) {
 }
 
 function eventResponse() {
-  const encoder = new TextEncoder()
-  let heartbeat: ReturnType<typeof setInterval> | undefined
-  let unsubscribe = () => {}
-  let done = false
-
-  const cleanup = () => {
-    if (done) return
-    done = true
-    if (heartbeat) clearInterval(heartbeat)
-    unsubscribe()
-    log.info("global event disconnected")
-  }
-
   log.info("global event connected")
-  return HttpServerResponse.raw(
-    new Response(
-      new ReadableStream<Uint8Array>({
-        start(controller) {
-          const write = (data: unknown) => {
-            if (done) return
-            try {
-              controller.enqueue(encoder.encode(eventData(data)))
-            } catch {
-              cleanup()
-            }
-          }
-          const handler = (event: GlobalBusEvent) => write(event)
-          unsubscribe = () => GlobalBus.off("event", handler)
-          GlobalBus.on("event", handler)
-          write({ payload: { type: "server.connected", properties: {} } })
-          heartbeat = setInterval(() => write({ payload: { type: "server.heartbeat", properties: {} } }), 10_000)
-        },
-        cancel: cleanup,
-      }),
-      {
-        headers: {
-          "Cache-Control": "no-cache, no-transform",
-          "Content-Type": "text/event-stream",
-          "X-Accel-Buffering": "no",
-          "X-Content-Type-Options": "nosniff",
-        },
-      },
+  const events = Stream.callback<GlobalBusEvent>((queue) => {
+    const handler = (event: GlobalBusEvent) => Queue.offerUnsafe(queue, event)
+    return Effect.acquireRelease(
+      Effect.sync(() => GlobalBus.on("event", handler)),
+      () => Effect.sync(() => GlobalBus.off("event", handler)),
+    )
+  })
+  const heartbeat = Stream.tick("10 seconds").pipe(
+    Stream.drop(1),
+    Stream.map(() => ({ payload: { type: "server.heartbeat", properties: {} } })),
+  )
+
+  return HttpServerResponse.stream(
+    Stream.make({ payload: { type: "server.connected", properties: {} } }).pipe(
+      Stream.concat(events.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))),
+      Stream.map(eventData),
+      Stream.pipeThroughChannel(Sse.encode()),
+      Stream.encodeText,
+      Stream.ensuring(Effect.sync(() => log.info("global event disconnected"))),
     ),
+    {
+      contentType: "text/event-stream",
+      headers: {
+        "Cache-Control": "no-cache, no-transform",
+        "X-Accel-Buffering": "no",
+        "X-Content-Type-Options": "nosniff",
+      },
+    },
   )
 }
 

packages/opencode/src/server/routes/instance/httpapi/lifecycle.ts

@@ -3,7 +3,6 @@ import { Effect } from "effect"
 import { HttpEffect, HttpMiddleware, HttpServerRequest } from "effect/unstable/http"
 
 const disposeAfterResponse = new WeakMap<object, InstanceContext>()
-const reloadAfterResponse = new WeakMap<object, InstanceContext & { next: Parameters<typeof Instance.reload>[0] }>()
 
 export const markInstanceForDisposal = (ctx: InstanceContext) =>
   HttpEffect.appendPreResponseHandler((request, response) =>
@@ -14,27 +13,17 @@ export const markInstanceForDisposal = (ctx: InstanceContext) =>
   )
 
 export const markInstanceForReload = (ctx: InstanceContext, next: Parameters<typeof Instance.reload>[0]) =>
-  HttpEffect.appendPreResponseHandler((request, response) =>
-    Effect.sync(() => {
-      reloadAfterResponse.set(request.source, { ...ctx, next })
-      return response
-    }),
+  HttpEffect.appendPreResponseHandler((_request, response) =>
+    Effect.as(Effect.uninterruptible(Effect.promise(() => Instance.restore(ctx, () => Instance.reload(next)))), response),
   )
 
 export const disposeMiddleware: HttpMiddleware.HttpMiddleware = (effect) =>
   Effect.gen(function* () {
     const response = yield* effect
     const request = yield* HttpServerRequest.HttpServerRequest
-    const reload = reloadAfterResponse.get(request.source)
-    if (reload) {
-      reloadAfterResponse.delete(request.source)
-      yield* Effect.promise(() => Instance.restore(reload, () => Instance.reload(reload.next)))
-      return response
-    }
-
     const ctx = disposeAfterResponse.get(request.source)
     if (!ctx) return response
     disposeAfterResponse.delete(request.source)
-    yield* Effect.promise(() => Instance.restore(ctx, () => Instance.dispose()))
+    yield* Effect.uninterruptible(Effect.promise(() => Instance.restore(ctx, () => Instance.dispose())))
     return response
   })