Refactor v2 session events as schemas (#24512)

2026-05-02 22:09:48 -04:00
parent 1409a0715c
commit a3bc5d35b0
62 changed files with 12801 additions and 6015 deletions
@@ -1,6 +1,7 @@
 import { Config } from "@/config/config"
 import { GlobalBus, type GlobalEvent as GlobalBusEvent } from "@/bus/global"
 import { EffectBridge } from "@/effect/bridge"
+import { Bus } from "@/bus"
 import { Installation } from "@/installation"
 import { disposeAllInstancesAndEmitGlobalDisposed } from "@/server/global-lifecycle"
 import { InstallationVersion } from "@opencode-ai/core/installation/version"
@@ -43,11 +44,11 @@ function eventResponse() {
  })
  const heartbeat = Stream.tick("10 seconds").pipe(
    Stream.drop(1),
-    Stream.map(() => ({ payload: { type: "server.heartbeat", properties: {} } })),
+    Stream.map(() => ({ payload: { id: Bus.createID(), type: "server.heartbeat", properties: {} } })),
  )

  return HttpServerResponse.stream(
-    Stream.make({ payload: { type: "server.connected", properties: {} } }).pipe(
+    Stream.make({ payload: { id: Bus.createID(), type: "server.connected", properties: {} } }).pipe(
      Stream.concat(events.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))),
      Stream.map(eventData),
      Stream.pipeThroughChannel(Sse.encode()),
@@ -0,0 +1,6 @@
+import { SessionV2 } from "@/v2/session"
+import { Layer } from "effect"
+import { messageHandlers } from "./v2/message"
+import { sessionHandlers } from "./v2/session"
+
+export const v2Handlers = Layer.mergeAll(sessionHandlers, messageHandlers).pipe(Layer.provide(SessionV2.defaultLayer))
@@ -0,0 +1,60 @@
+import { SessionMessage } from "@/v2/session-message"
+import { SessionV2 } from "@/v2/session"
+import { Effect, Schema } from "effect"
+import * as DateTime from "effect/DateTime"
+import { HttpApiBuilder, HttpApiError } from "effect/unstable/httpapi"
+import { InstanceHttpApi } from "../../api"
+
+const DefaultMessagesLimit = 50
+
+const Cursor = Schema.Struct({
+  id: SessionMessage.ID,
+  time: Schema.Finite,
+  order: Schema.Union([Schema.Literal("asc"), Schema.Literal("desc")]),
+  direction: Schema.Union([Schema.Literal("previous"), Schema.Literal("next")]),
+})
+
+const decodeCursor = Schema.decodeUnknownSync(Cursor)
+
+const cursor = {
+  encode(message: SessionMessage.Message, order: "asc" | "desc", direction: "previous" | "next") {
+    return Buffer.from(
+      JSON.stringify({ id: message.id, time: DateTime.toEpochMillis(message.time.created), order, direction }),
+    ).toString("base64url")
+  },
+  decode(input: string) {
+    return decodeCursor(JSON.parse(Buffer.from(input, "base64url").toString("utf8")))
+  },
+}
+
+export const messageHandlers = HttpApiBuilder.group(InstanceHttpApi, "v2.message", (handlers) =>
+  Effect.gen(function* () {
+    const session = yield* SessionV2.Service
+
+    return handlers.handle(
+      "messages",
+      Effect.fn(function* (ctx) {
+        const decoded = yield* Effect.try({
+          try: () => (ctx.query.cursor ? cursor.decode(ctx.query.cursor) : undefined),
+          catch: () => new HttpApiError.BadRequest({}),
+        })
+        const order = decoded?.order ?? ctx.query.order ?? "desc"
+        const messages = yield* session.messages({
+          sessionID: ctx.params.sessionID,
+          limit: ctx.query.limit ?? DefaultMessagesLimit,
+          order,
+          cursor: decoded ? { id: decoded.id, time: decoded.time, direction: decoded.direction } : undefined,
+        })
+        const first = messages[0]
+        const last = messages.at(-1)
+        return {
+          items: messages,
+          cursor: {
+            previous: first ? cursor.encode(first, order, "previous") : undefined,
+            next: last ? cursor.encode(last, order, "next") : undefined,
+          },
+        }
+      }),
+    )
+  }),
+)
@@ -0,0 +1,115 @@
+import { WorkspaceID } from "@/control-plane/schema"
+import { SessionV2 } from "@/v2/session"
+import { Effect, Schema } from "effect"
+import { HttpApiBuilder, HttpApiError, HttpApiSchema } from "effect/unstable/httpapi"
+import { InstanceHttpApi } from "../../api"
+
+const DefaultSessionsLimit = 50
+
+const SessionCursor = Schema.Struct({
+  id: SessionV2.Info.fields.id,
+  time: Schema.Finite,
+  order: Schema.Union([Schema.Literal("asc"), Schema.Literal("desc")]),
+  direction: Schema.Union([Schema.Literal("previous"), Schema.Literal("next")]),
+  directory: Schema.String.pipe(Schema.optional),
+  path: Schema.String.pipe(Schema.optional),
+  workspaceID: WorkspaceID.pipe(Schema.optional),
+  roots: Schema.Boolean.pipe(Schema.optional),
+  start: Schema.Finite.pipe(Schema.optional),
+  search: Schema.String.pipe(Schema.optional),
+})
+type SessionCursor = typeof SessionCursor.Type
+
+const decodeCursor = Schema.decodeUnknownSync(SessionCursor)
+
+const sessionCursor = {
+  encode(
+    session: SessionV2.Info,
+    order: "asc" | "desc",
+    direction: "previous" | "next",
+    filters: Pick<SessionCursor, "directory" | "path" | "workspaceID" | "roots" | "start" | "search">,
+  ) {
+    return Buffer.from(
+      JSON.stringify({ id: session.id, time: session.time.created, order, direction, ...filters }),
+    ).toString("base64url")
+  },
+  decode(input: string) {
+    return decodeCursor(JSON.parse(Buffer.from(input, "base64url").toString("utf8")))
+  },
+}
+
+export const sessionHandlers = HttpApiBuilder.group(InstanceHttpApi, "v2.session", (handlers) =>
+  Effect.gen(function* () {
+    const session = yield* SessionV2.Service
+
+    return handlers
+      .handle(
+        "sessions",
+        Effect.fn(function* (ctx) {
+          const decoded = yield* Effect.try({
+            try: () => (ctx.query.cursor ? sessionCursor.decode(ctx.query.cursor) : undefined),
+            catch: () => new HttpApiError.BadRequest({}),
+          })
+          const order = decoded?.order ?? ctx.query.order ?? "desc"
+          const filters = decoded ?? {
+            directory: ctx.query.directory,
+            path: ctx.query.path,
+            workspaceID: ctx.query.workspace ? WorkspaceID.make(ctx.query.workspace) : undefined,
+            roots: ctx.query.roots,
+            start: ctx.query.start,
+            search: ctx.query.search,
+          }
+          const sessions = yield* session.list({
+            limit: ctx.query.limit ?? DefaultSessionsLimit,
+            order,
+            directory: filters.directory,
+            path: filters.path,
+            workspaceID: filters.workspaceID,
+            roots: filters.roots,
+            start: filters.start,
+            search: filters.search,
+            cursor: decoded ? { id: decoded.id, time: decoded.time, direction: decoded.direction } : undefined,
+          })
+          const first = sessions[0]
+          const last = sessions.at(-1)
+          return {
+            items: sessions,
+            cursor: {
+              previous: first ? sessionCursor.encode(first, order, "previous", filters) : undefined,
+              next: last ? sessionCursor.encode(last, order, "next", filters) : undefined,
+            },
+          }
+        }),
+      )
+      .handle(
+        "prompt",
+        Effect.fn(function* (ctx) {
+          return yield* session.prompt({
+            sessionID: ctx.params.sessionID,
+            prompt: ctx.payload.prompt,
+            delivery: ctx.payload.delivery ?? SessionV2.DefaultDelivery,
+          })
+        }),
+      )
+      .handle(
+        "compact",
+        Effect.fn(function* (ctx) {
+          yield* session.compact(ctx.params.sessionID)
+          return HttpApiSchema.NoContent.make()
+        }),
+      )
+      .handle(
+        "wait",
+        Effect.fn(function* (ctx) {
+          yield* session.wait(ctx.params.sessionID)
+          return HttpApiSchema.NoContent.make()
+        }),
+      )
+      .handle(
+        "context",
+        Effect.fn(function* (ctx) {
+          return yield* session.context(ctx.params.sessionID)
+        }),
+      )
+  }),
+)