diff --git a/packages/opencode/src/server/routes/instance/httpapi/groups/v2.ts b/packages/opencode/src/server/routes/instance/httpapi/groups/v2.ts index 6fa0d23a4f4..0cd768e0d9e 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/groups/v2.ts +++ b/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", diff --git a/packages/opencode/src/server/routes/instance/httpapi/groups/v2/event.ts b/packages/opencode/src/server/routes/instance/httpapi/groups/v2/event.ts new file mode 100644 index 00000000000..181ff38ffcd --- /dev/null +++ b/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 diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/v2.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/v2.ts index f168731f214..c6152f6ddd6 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/v2.ts +++ b/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), diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/v2/event.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/v2/event.ts new file mode 100644 index 00000000000..bde4bbb86da --- /dev/null +++ b/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", + }, + }, + ) + }), + ), +) diff --git a/packages/opencode/test/server/httpapi-v2-location.test.ts b/packages/opencode/test/server/httpapi-v2-location.test.ts index 6cb98952baf..481e05b1cbb 100644 --- a/packages/opencode/test/server/httpapi-v2-location.test.ts +++ b/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 -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) { + 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, 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() + }) })