feat(opencode): add v2 event stream

packages/opencode/src/server/routes/instance/httpapi/groups/v2.ts

@@ -7,6 +7,7 @@ import { PermissionGroup, PermissionSavedGroup, SessionPermissionGroup } from ".
 import { FileSystemGroup } from "./v2/fs"
 import { CommandGroup } from "./v2/command"
 import { SkillGroup } from "./v2/skill"
+import { EventGroup } from "./v2/event"
 
 export const V2Api = HttpApi.make("v2")
   .add(SessionGroup)
@@ -19,6 +20,7 @@ export const V2Api = HttpApi.make("v2")
   .add(FileSystemGroup)
   .add(CommandGroup)
   .add(SkillGroup)
+  .add(EventGroup)
   .annotateMerge(
     OpenApi.annotations({
       title: "opencode experimental HttpApi",

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

@@ -0,0 +1,36 @@
+import { EventV2 } from "@opencode-ai/core/event"
+import { Location } from "@opencode-ai/core/location"
+import { Schema } from "effect"
+import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/unstable/httpapi"
+import { V2Authorization } from "../../middleware/authorization"
+import { LocationQuery, locationQueryOpenApi, V2LocationMiddleware } from "./location"
+
+const Event = Schema.Struct({
+  id: EventV2.ID,
+  type: Schema.String,
+  location: Location.Info.pipe(Schema.optional),
+  metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(Schema.optional),
+  version: Schema.Number.pipe(Schema.optional),
+  data: Schema.Unknown,
+})
+
+export const EventGroup = HttpApiGroup.make("v2.event")
+  .add(
+    HttpApiEndpoint.get("events", "/api/event", {
+      query: LocationQuery,
+      success: Schema.String.pipe(HttpApiSchema.asText({ contentType: "text/event-stream" })),
+    })
+      .annotateMerge(locationQueryOpenApi)
+      .annotateMerge(
+        OpenApi.annotations({
+          identifier: "v2.event.subscribe",
+          summary: "Subscribe to v2 events",
+          description: "Subscribe to native EventV2 payloads for a location.",
+        }),
+      ),
+  )
+  .annotateMerge(OpenApi.annotations({ title: "v2 events", description: "Experimental v2 event stream route." }))
+  .middleware(V2LocationMiddleware)
+  .middleware(V2Authorization)
+
+export type Event = typeof Event.Type

packages/opencode/src/server/routes/instance/httpapi/handlers/v2.ts

@@ -11,6 +11,7 @@ import { permissionHandlers, savedPermissionHandlers, sessionPermissionHandlers
 import { fileSystemHandlers } from "./v2/fs"
 import { commandHandlers } from "./v2/command"
 import { skillHandlers } from "./v2/skill"
+import { eventHandlers } from "./v2/event"
 
 export const v2Handlers = Layer.mergeAll(
   sessionHandlers,
@@ -23,6 +24,7 @@ export const v2Handlers = Layer.mergeAll(
   fileSystemHandlers,
   commandHandlers,
   skillHandlers,
+  eventHandlers,
 ).pipe(
   Layer.provide(v2LocationLayer),
   Layer.provide(LocationServiceMap.layer),

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

@@ -0,0 +1,60 @@
+import { EventV2 } from "@opencode-ai/core/event"
+import { Location } from "@opencode-ai/core/location"
+import { EventV2Bridge } from "@/event-v2-bridge"
+import { Effect, Stream } from "effect"
+import { HttpServerResponse } from "effect/unstable/http"
+import { HttpApiBuilder } from "effect/unstable/httpapi"
+import * as Sse from "effect/unstable/encoding/Sse"
+import { InstanceHttpApi } from "../../api"
+
+function eventData(data: unknown): Sse.Event {
+  return {
+    _tag: "Event",
+    event: "message",
+    id: undefined,
+    data: JSON.stringify(data),
+  }
+}
+
+export const eventHandlers = HttpApiBuilder.group(InstanceHttpApi, "v2.event", (handlers) =>
+  handlers.handleRaw("events", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2Bridge.Service
+      const location = yield* Location.Service
+      const connected = {
+        id: EventV2.ID.create(),
+        type: "server.connected",
+        location: new Location.Info({
+          directory: location.directory,
+          workspaceID: location.workspaceID,
+          project: location.project,
+        }),
+        data: {},
+      }
+      return HttpServerResponse.stream(
+        Stream.make(connected).pipe(
+          Stream.concat(
+            events.all().pipe(
+              Stream.filter(
+                (event) =>
+                  event.location?.directory === location.directory &&
+                  event.location.workspaceID === location.workspaceID,
+              ),
+            ),
+          ),
+          Stream.map(eventData),
+          Stream.pipeThroughChannel(Sse.encode()),
+          Stream.encodeText,
+        ),
+        {
+          contentType: "text/event-stream",
+          headers: {
+            "Cache-Control": "no-cache, no-transform",
+            "X-Accel-Buffering": "no",
+            "X-Content-Type-Options": "nosniff",
+          },
+        },
+      )
+    }),
+  ),
+)

packages/opencode/test/server/httpapi-v2-location.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, describe, expect, test } from "bun:test"
-import { Context } from "effect"
+import { Context, Schema } from "effect"
 import { HttpApiApp } from "../../src/server/routes/instance/httpapi/server"
 import * as Log from "@opencode-ai/core/util/log"
 import { resetDatabase } from "../fixture/db"
@@ -9,17 +9,42 @@ void Log.init({ print: false })
 
 const context = Context.empty() as Context.Context<unknown>
 
-function request(route: string, directory: string) {
+function request(route: string, directory: string, init: RequestInit = {}) {
+  const headers = new Headers(init.headers)
+  headers.set("x-opencode-directory", directory)
   return HttpApiApp.webHandler().handler(
     new Request(`http://localhost${route}`, {
-      headers: {
-        "x-opencode-directory": directory,
-      },
+      ...init,
+      headers,
     }),
     context,
   )
 }
 
+const Event = Schema.Struct({
+  id: Schema.String,
+  type: Schema.String,
+  location: Schema.Struct({
+    directory: Schema.String,
+    project: Schema.Struct({ id: Schema.String, directory: Schema.String }),
+  }),
+  data: Schema.Unknown,
+})
+
+async function readEvent(reader: ReadableStreamDefaultReader<Uint8Array>) {
+  const value = await reader.read()
+  if (value.done) throw new Error("event stream closed")
+  return Schema.decodeUnknownSync(Event)(JSON.parse(new TextDecoder().decode(value.value).replace(/^data: /, "")))
+}
+
+async function readEventType(reader: ReadableStreamDefaultReader<Uint8Array>, type: string) {
+  for (let index = 0; index < 20; index++) {
+    const event = await readEvent(reader)
+    if (event.type === type) return event
+  }
+  throw new Error(`timed out waiting for ${type}`)
+}
+
 afterEach(async () => {
   await disposeAllInstances()
   await resetDatabase()
@@ -38,4 +63,20 @@ describe("v2 location HttpApi", () => {
       expect(body.location.project.id).toBeTruthy()
     }
   })
+
+  test("streams native EventV2 payloads with resolved locations", async () => {
+    await using tmp = await tmpdir({ git: true })
+    const response = await request("/api/event", tmp.path)
+    const reader = response.body!.getReader()
+    expect((await readEvent(reader)).type).toBe("server.connected")
+
+    const created = await request("/session", tmp.path, { method: "POST" })
+    expect(created.status).toBe(200)
+    expect(await readEventType(reader, "session.created")).toMatchObject({
+      type: "session.created",
+      location: { directory: tmp.path, project: { directory: tmp.path } },
+      data: { sessionID: expect.any(String) },
+    })
+    await reader.cancel()
+  })
 })