From 903acdf0d428e18b34b5d2b7a17624e05f6a401a Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 31 Jul 2026 15:23:21 -0400 Subject: [PATCH] refactor(schema): move session pagination cursors to schema --- packages/protocol/src/groups/session.ts | 90 +------------ packages/protocol/test/session-cursor.test.ts | 18 --- .../schema/src/session-messages-cursor.ts | 48 +++++++ packages/schema/src/sessions-query.ts | 125 ++++++++++++++++++ packages/schema/test/sessions-query.test.ts | 33 +++++ packages/server/src/handlers/message.ts | 41 ++---- packages/server/src/handlers/session.ts | 36 +---- 7 files changed, 225 insertions(+), 166 deletions(-) delete mode 100644 packages/protocol/test/session-cursor.test.ts create mode 100644 packages/schema/src/session-messages-cursor.ts create mode 100644 packages/schema/src/sessions-query.ts create mode 100644 packages/schema/test/sessions-query.test.ts diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 2abe6c47c7..c147d4884d 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -4,7 +4,8 @@ import { PromptInput } from "@opencode-ai/schema/prompt-input" import { Session } from "@opencode-ai/schema/session" import { InstructionEntry } from "@opencode-ai/schema/instruction-entry" import { Project } from "@opencode-ai/schema/project" -import { AbsolutePath, PositiveInt, RelativePath, statics } from "@opencode-ai/schema/schema" +import { AbsolutePath, PositiveInt, RelativePath } from "@opencode-ai/schema/schema" +import { SessionsQuery } from "@opencode-ai/schema/sessions-query" import { Event } from "@opencode-ai/schema/event" import { Workspace } from "@opencode-ai/schema/workspace" import { Context, Effect, Encoding, Result, Schema, SchemaGetter, Struct } from "effect" @@ -29,76 +30,6 @@ import { Location } from "@opencode-ai/schema/location" import { SessionEvent } from "@opencode-ai/schema/session-event" import { EventLog } from "@opencode-ai/schema/event-log" -const ParentIDFilter = Schema.Union([ - Session.ID, - Schema.Null.pipe( - Schema.encodeTo(Schema.Literal("null"), { - decode: SchemaGetter.transform(() => null), - encode: SchemaGetter.transform(() => "null" as const), - }), - ), -]).annotate({ - description: "Filter by parent session. Use null to return only root sessions.", -}) - -const SessionsQueryFields = { - workspace: Workspace.ID.pipe(Schema.optional), - limit: Schema.NumberFromString.pipe(Schema.decodeTo(PositiveInt), Schema.optional).annotate({ - description: "Maximum number of sessions to return. Defaults to the newest 50 sessions.", - }), - order: Schema.optional(Schema.Union([Schema.Literal("asc"), Schema.Literal("desc")])).annotate({ - description: "Session order for the first page. Use desc for newest first or asc for oldest first.", - }), - search: Schema.optional(Schema.String), - parentID: ParentIDFilter.pipe(Schema.optional), -} - -const SessionsDirectoryQuery = Schema.Struct({ - ...SessionsQueryFields, - directory: AbsolutePath, -}) - -const SessionsProjectQuery = Schema.Struct({ - ...SessionsQueryFields, - project: Project.ID, - subpath: RelativePath.pipe(Schema.optional), -}) - -const SessionsAllQuery = Schema.Struct(SessionsQueryFields) - -const withCursor = (schema: Schema.Struct) => - schema.mapFields((fields) => ({ - ...Struct.omit(fields, ["limit"]), - anchor: Session.ListAnchor, - })) - -const SessionsCursorInput = Schema.Union([ - withCursor(SessionsDirectoryQuery), - withCursor(SessionsProjectQuery), - withCursor(SessionsAllQuery), -]) -const SessionsCursorJson = Schema.fromJsonString(SessionsCursorInput) -const encodeSessionsCursor = Schema.encodeSync(SessionsCursorJson) -const decodeSessionsCursor = Schema.decodeUnknownEffect(SessionsCursorJson) -const invalidCursor = "Invalid cursor" as const - -export const SessionsCursor = Schema.String.pipe( - Schema.brand("SessionsCursor"), - statics((schema) => { - const make = schema.make.bind(schema) - return { - make: (input: typeof SessionsCursorInput.Type) => make(Encoding.encodeBase64Url(encodeSessionsCursor(input))), - parse: (input: string) => - Effect.suspend(() => { - const result = Encoding.decodeBase64UrlString(input) - return Result.isFailure(result) - ? Effect.fail(invalidCursor) - : decodeSessionsCursor(result.success).pipe(Effect.mapError(() => invalidCursor)) - }), - } - }), -) -export type SessionsCursor = typeof SessionsCursor.Type const SessionActive = Schema.Struct({ type: Schema.Literal("running"), @@ -111,28 +42,17 @@ const BooleanFromString = Schema.Literals(["true", "false"]).pipe( }), ) -const SessionsQueryCursor = SessionsCursor.annotate({ - description: "Opaque pagination cursor returned as cursor.previous or cursor.next in the previous response.", -}) - -export const SessionsQuery = Schema.Struct({ - ...SessionsQueryFields, - directory: AbsolutePath.pipe(Schema.optional), - project: Project.ID.pipe(Schema.optional), - subpath: RelativePath.pipe(Schema.optional), - cursor: SessionsQueryCursor.pipe(Schema.optional), -}).annotate({ identifier: "SessionsQuery" }) export const makeSessionGroup = (sessionLocationMiddleware: Context.Key) => HttpApiGroup.make("server.session") .add( HttpApiEndpoint.get("session.list", "/api/session", { - query: SessionsQuery, + query: SessionsQuery.Query, success: Schema.Struct({ data: Schema.Array(Session.Info), cursor: Schema.Struct({ - previous: SessionsCursor.pipe(Schema.optional), - next: SessionsCursor.pipe(Schema.optional), + previous: SessionsQuery.Cursor.pipe(Schema.optional), + next: SessionsQuery.Cursor.pipe(Schema.optional), }), }).annotate({ identifier: "SessionsResponse" }), error: [InvalidCursorError, InvalidRequestError], diff --git a/packages/protocol/test/session-cursor.test.ts b/packages/protocol/test/session-cursor.test.ts deleted file mode 100644 index 86c78d9503..0000000000 --- a/packages/protocol/test/session-cursor.test.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { describe, expect, test } from "bun:test" -import { Effect } from "effect" -import { SessionsCursor } from "../src/groups/session.js" -import { Session } from "@opencode-ai/schema/session" - -describe("SessionsCursor", () => { - test("round trips without Node globals", async () => { - const input = { - workspace: undefined, - search: "protocol", - order: "desc" as const, - anchor: { id: Session.ID.make("ses_test"), time: 1, direction: "next" as const }, - } - const cursor = SessionsCursor.make(input) - - expect(await Effect.runPromise(SessionsCursor.parse(cursor))).toEqual(input) - }) -}) diff --git a/packages/schema/src/session-messages-cursor.ts b/packages/schema/src/session-messages-cursor.ts new file mode 100644 index 0000000000..326799a7fc --- /dev/null +++ b/packages/schema/src/session-messages-cursor.ts @@ -0,0 +1,48 @@ +export * as SessionMessagesCursor from "./session-messages-cursor.js" + +import { Effect, Encoding, Result, Schema } from "effect" +import { SessionMessage } from "./session-message.js" + +/** + * Shared session message pagination cursor. Lives in schema so the protocol + * message handler and the core plugin host paginate identically. The encoded + * cursor carries the page order, so a cursor cannot be combined with an + * explicit order. + */ + +export const DefaultLimit = 50 + +export const Order = Schema.Literals(["asc", "desc"]) +export type Order = typeof Order.Type + +const Payload = Schema.Struct({ + id: SessionMessage.ID, + order: Order, + direction: Schema.Literals(["previous", "next"]), +}) +export type Payload = typeof Payload.Type + +const PayloadJson = Schema.fromJsonString(Payload) +const encodePayload = Schema.encodeSync(PayloadJson) +const decodePayload = Schema.decodeUnknownEffect(PayloadJson) +const invalidCursor = "Invalid cursor" as const + +export const make = (input: Payload) => Encoding.encodeBase64Url(encodePayload(input)) + +export const parse = (input: string) => + Effect.suspend(() => { + const result = Encoding.decodeBase64UrlString(input) + return Result.isFailure(result) + ? Effect.fail(invalidCursor) + : decodePayload(result.success).pipe(Effect.mapError(() => invalidCursor)) + }) + +/** previous/next cursors for one returned page of messages. */ +export const page = (data: ReadonlyArray, order: Order) => { + const first = data[0] + const last = data.at(-1) + return { + previous: first ? make({ id: first.id, order, direction: "previous" }) : undefined, + next: last ? make({ id: last.id, order, direction: "next" }) : undefined, + } +} diff --git a/packages/schema/src/sessions-query.ts b/packages/schema/src/sessions-query.ts new file mode 100644 index 0000000000..6fb8861505 --- /dev/null +++ b/packages/schema/src/sessions-query.ts @@ -0,0 +1,125 @@ +export * as SessionsQuery from "./sessions-query.js" + +import { DateTime, Effect, Encoding, Result, Schema, SchemaGetter, Struct } from "effect" +import { Project } from "./project.js" +import { AbsolutePath, PositiveInt, RelativePath, statics } from "./schema.js" +import { Session } from "./session.js" +import { Workspace } from "./workspace.js" + +/** + * Shared session list query and cursor codec. Lives in schema so both the + * protocol session group and the core plugin host paginate identically. + */ + +export const DefaultLimit = 50 + +const ParentIDFilter = Schema.Union([ + Session.ID, + Schema.Null.pipe( + Schema.encodeTo(Schema.Literal("null"), { + decode: SchemaGetter.transform(() => null), + encode: SchemaGetter.transform(() => "null" as const), + }), + ), +]).annotate({ + description: "Filter by parent session. Use null to return only root sessions.", +}) + +export const Fields = { + workspace: Workspace.ID.pipe(Schema.optional), + limit: Schema.NumberFromString.pipe(Schema.decodeTo(PositiveInt), Schema.optional).annotate({ + description: "Maximum number of sessions to return. Defaults to the newest 50 sessions.", + }), + order: Schema.optional(Schema.Union([Schema.Literal("asc"), Schema.Literal("desc")])).annotate({ + description: "Session order for the first page. Use desc for newest first or asc for oldest first.", + }), + search: Schema.optional(Schema.String), + parentID: ParentIDFilter.pipe(Schema.optional), +} + +const DirectoryQuery = Schema.Struct({ + ...Fields, + directory: AbsolutePath, +}) + +const ProjectQuery = Schema.Struct({ + ...Fields, + project: Project.ID, + subpath: RelativePath.pipe(Schema.optional), +}) + +const AllQuery = Schema.Struct(Fields) + +const withAnchor = (schema: Schema.Struct) => + schema.mapFields((fields) => ({ + ...Struct.omit(fields, ["limit"]), + anchor: Session.ListAnchor, + })) + +const CursorInput = Schema.Union([withAnchor(DirectoryQuery), withAnchor(ProjectQuery), withAnchor(AllQuery)]) +const CursorJson = Schema.fromJsonString(CursorInput) +const encodeCursor = Schema.encodeSync(CursorJson) +const decodeCursor = Schema.decodeUnknownEffect(CursorJson) +const invalidCursor = "Invalid cursor" as const + +type PageQuery = Omit + +export const Cursor = Schema.String.pipe( + Schema.brand("SessionsCursor"), + statics((schema) => { + const make = schema.make.bind(schema) + const makeCursor = (input: typeof CursorInput.Type) => make(Encoding.encodeBase64Url(encodeCursor(input))) + return { + make: makeCursor, + parse: (input: string) => + Effect.suspend(() => { + const result = Encoding.decodeBase64UrlString(input) + return Result.isFailure(result) + ? Effect.fail(invalidCursor) + : decodeCursor(result.success).pipe(Effect.mapError(() => invalidCursor)) + }), + /** + * previous/next cursors for one returned page, re-anchoring the + * originating query at the page's first and last items. + */ + page: (query: PageQuery, data: ReadonlyArray) => { + const first = data[0] + const last = data.at(-1) + return { + previous: first + ? makeCursor({ + ...query, + anchor: { + id: first.id, + time: DateTime.toEpochMillis(first.time.updated), + direction: "previous", + }, + }) + : undefined, + next: last + ? makeCursor({ + ...query, + anchor: { + id: last.id, + time: DateTime.toEpochMillis(last.time.updated), + direction: "next", + }, + }) + : undefined, + } + }, + } + }), +) +export type Cursor = typeof Cursor.Type + +export const Query = Schema.Struct({ + ...Fields, + directory: AbsolutePath.pipe(Schema.optional), + project: Project.ID.pipe(Schema.optional), + subpath: RelativePath.pipe(Schema.optional), + cursor: Cursor.annotate({ + description: "Opaque pagination cursor returned as cursor.previous or cursor.next in the previous response.", + }).pipe(Schema.optional), +}).annotate({ identifier: "SessionsQuery" }) +export type Query = typeof Query.Type diff --git a/packages/schema/test/sessions-query.test.ts b/packages/schema/test/sessions-query.test.ts new file mode 100644 index 0000000000..483941bae2 --- /dev/null +++ b/packages/schema/test/sessions-query.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, test } from "bun:test" +import { Effect } from "effect" +import { SessionsQuery } from "../src/sessions-query.js" +import { SessionMessagesCursor } from "../src/session-messages-cursor.js" +import { SessionMessage } from "../src/session-message.js" +import { Session } from "../src/session.js" + +describe("SessionsQuery.Cursor", () => { + test("round trips without Node globals", async () => { + const input = { + workspace: undefined, + search: "protocol", + order: "desc" as const, + anchor: { id: Session.ID.make("ses_test"), time: 1, direction: "next" as const }, + } + const cursor = SessionsQuery.Cursor.make(input) + + expect(await Effect.runPromise(SessionsQuery.Cursor.parse(cursor))).toEqual(input) + }) +}) + +describe("SessionMessagesCursor", () => { + test("round trips", async () => { + const input = { + id: SessionMessage.ID.make("msg_test"), + order: "desc" as const, + direction: "next" as const, + } + const cursor = SessionMessagesCursor.make(input) + + expect(await Effect.runPromise(SessionMessagesCursor.parse(cursor))).toEqual(input) + }) +}) diff --git a/packages/server/src/handlers/message.ts b/packages/server/src/handlers/message.ts index 8bff2191b0..0f4f38fa3b 100644 --- a/packages/server/src/handlers/message.ts +++ b/packages/server/src/handlers/message.ts @@ -1,29 +1,10 @@ -import { SessionMessage } from "@opencode-ai/core/session/message" import { Session } from "@opencode-ai/core/session" -import { Effect, Schema } from "effect" +import { SessionMessagesCursor } from "@opencode-ai/schema/session-messages-cursor" +import { Effect } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { Api } from "../api" import { InvalidCursorError, SessionNotFoundError, UnknownError } from "@opencode-ai/protocol/errors" -const DefaultMessagesLimit = 50 - -const Cursor = Schema.Struct({ - id: SessionMessage.ID, - 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.Info, order: "asc" | "desc", direction: "previous" | "next") { - return Buffer.from(JSON.stringify({ id: message.id, order, direction })).toString("base64url") - }, - decode(input: string) { - return decodeCursor(JSON.parse(Buffer.from(input, "base64url").toString("utf8"))) - }, -} - export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handlers) => Effect.gen(function* () { const session = yield* Session.Service @@ -33,15 +14,16 @@ export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handl Effect.fn(function* (ctx) { if (ctx.query.cursor && ctx.query.order !== undefined) return yield* new InvalidCursorError({ message: "Cursor cannot be combined with order" }) - const decoded = yield* Effect.try({ - try: () => (ctx.query.cursor ? cursor.decode(ctx.query.cursor) : undefined), - catch: () => new InvalidCursorError({ message: "Invalid cursor" }), - }) + const decoded = ctx.query.cursor + ? yield* SessionMessagesCursor.parse(ctx.query.cursor).pipe( + Effect.mapError(() => new InvalidCursorError({ message: "Invalid cursor" })), + ) + : undefined const order = decoded?.order ?? ctx.query.order ?? "desc" const messages = yield* session .messages({ sessionID: ctx.params.sessionID, - limit: ctx.query.limit ?? DefaultMessagesLimit, + limit: ctx.query.limit ?? SessionMessagesCursor.DefaultLimit, order, cursor: decoded ? { id: decoded.id, direction: decoded.direction } : undefined, }) @@ -66,14 +48,9 @@ export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handl ) }), ) - const first = messages[0] - const last = messages.at(-1) return { data: messages, - cursor: { - previous: first ? cursor.encode(first, order, "previous") : undefined, - next: last ? cursor.encode(last, order, "next") : undefined, - }, + cursor: SessionMessagesCursor.page(messages, order), } }), ) diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index a62ccfcd6c..46223220da 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -3,7 +3,7 @@ import { InstructionEntry } from "@opencode-ai/core/session/instruction-entry" import { DateTime, Effect, Stream } from "effect" import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi" import { Api } from "../api" -import { SessionsCursor } from "@opencode-ai/protocol/groups/session" +import { SessionsQuery } from "@opencode-ai/schema/sessions-query" import { ConflictError, CommandEvaluationError, @@ -19,8 +19,6 @@ import { } from "@opencode-ai/protocol/errors" import { AbsolutePath } from "@opencode-ai/core/schema" -const DefaultSessionsLimit = 50 - export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handlers) => Effect.gen(function* () { const session = yield* Session.Service @@ -31,42 +29,18 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl Effect.fn(function* (ctx) { const query = ctx.query.cursor !== undefined - ? yield* SessionsCursor.parse(ctx.query.cursor).pipe( + ? yield* SessionsQuery.Cursor.parse(ctx.query.cursor).pipe( Effect.mapError(() => new InvalidCursorError({ message: "Invalid cursor" })), ) : ctx.query const page = yield* session.list({ ...query, workspaceID: query.workspace, - limit: ctx.query.limit ?? DefaultSessionsLimit, + limit: ctx.query.limit ?? SessionsQuery.DefaultLimit, }) - const sessions = page.data - const first = sessions[0] - const last = sessions.at(-1) return { - data: sessions, - cursor: { - previous: first - ? SessionsCursor.make({ - ...query, - anchor: { - id: first.id, - time: DateTime.toEpochMillis(first.time.updated), - direction: "previous", - }, - }) - : undefined, - next: last - ? SessionsCursor.make({ - ...query, - anchor: { - id: last.id, - time: DateTime.toEpochMillis(last.time.updated), - direction: "next", - }, - }) - : undefined, - }, + data: page.data, + cursor: SessionsQuery.Cursor.page(query, page.data), } }), )