diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 3fe9b48064a..fd39b5cd470 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -3,9 +3,8 @@ export * from "./session/schema.js" import { Cause, Effect, Layer, Schema, Context, RcMap, Stream, Scope } from "effect" import { ListAnchor } from "@opencode-ai/schema/session" -import { and, asc, desc, eq, gt, isNull, like, lt, or, type SQL } from "drizzle-orm" +import { and, desc, eq } from "drizzle-orm" import { Project } from "./project.js" -import { Workspace } from "@opencode-ai/schema/workspace" import { Model } from "@opencode-ai/schema/model" import { Location } from "./location.js" import { SessionMessage } from "./session/message.js" @@ -13,14 +12,13 @@ import { PromptInput } from "@opencode-ai/schema/prompt-input" import { Bus } from "./bus.js" import { Database } from "./database/database.js" import { SessionProjector } from "./session/projector.js" -import { SessionMessageTable, SessionTable } from "./session/sql.js" +import { SessionMessageTable } from "./session/sql.js" import { SessionSchema } from "./session/schema.js" -import { AbsolutePath, PositiveInt, RelativePath } from "./schema.js" +import { AbsolutePath, RelativePath } from "./schema.js" import { Agent } from "@opencode-ai/schema/agent" import { App } from "./app.js" import { Slug } from "./util/slug.js" import path from "path" -import { fromRow } from "./session/info.js" import { SessionRunner } from "./session/runner/index.js" import { SessionStore } from "./session/store.js" import { SessionExecution } from "./session/execution.js" @@ -58,7 +56,6 @@ import { Job } from "./job.js" import { Command } from "./command.js" import { Global } from "@opencode-ai/util/global" import { SessionEnvironment } from "./session/environment.js" -import { SessionHistory } from "./session/history.js" import { InstructionEntry } from "./session/instruction-entry.js" // get project -> project.locations @@ -72,30 +69,8 @@ import { InstructionEntry } from "./session/instruction-entry.js" export { ListAnchor } -const ListInputBase = { - workspaceID: Workspace.ID.pipe(Schema.optional), - search: Schema.String.pipe(Schema.optional), - limit: PositiveInt.pipe(Schema.optional), - order: Schema.Literals(["asc", "desc"]).pipe(Schema.optional), - parentID: Schema.NullOr(SessionSchema.ID).pipe(Schema.optional), - anchor: ListAnchor.pipe(Schema.optional), -} - -const ListDirectoryInput = Schema.Struct({ - ...ListInputBase, - directory: AbsolutePath, -}) - -const ListProjectInput = Schema.Struct({ - ...ListInputBase, - project: Project.ID, - subpath: RelativePath.pipe(Schema.optional), -}) - -const ListAllInput = Schema.Struct(ListInputBase) - -export const ListInput = Schema.Union([ListDirectoryInput, ListProjectInput, ListAllInput]) -export type ListInput = typeof ListInput.Type +export const ListInput = SessionStore.ListInput +export type ListInput = SessionStore.ListInput type CreateBaseInput = { id?: SessionSchema.ID @@ -161,15 +136,9 @@ export interface Interface { }) => Effect.Effect readonly view: (input: { sessionID: SessionSchema.ID; idle: number }) => Effect.Effect readonly remove: (sessionID: SessionSchema.ID) => Effect.Effect - readonly messages: (input: { - sessionID: SessionSchema.ID - limit?: number - order?: "asc" | "desc" - cursor?: { - id: SessionMessage.ID - direction: "previous" | "next" - } - }) => Effect.Effect + readonly messages: ( + input: SessionStore.MessagesInput, + ) => Effect.Effect readonly message: (input: { sessionID: SessionSchema.ID messageID: SessionMessage.ID @@ -409,83 +378,12 @@ const layer = Layer.effect( yield* bus.publish(SessionEvent.Deleted, { sessionID }) yield* bus.remove(sessionID) }), - list: Effect.fn("Session.list")(function* (input = {}) { - const direction = input.anchor?.direction ?? "next" - const requestedOrder = input.order ?? "desc" - const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder - const sortColumn = SessionTable.time_updated - const conditions: SQL[] = [] - if ("directory" in input) conditions.push(eq(SessionTable.directory, input.directory)) - if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) - if ("project" in input) conditions.push(eq(SessionTable.project_id, input.project)) - if ("project" in input && input.subpath !== undefined) conditions.push(eq(SessionTable.path, input.subpath)) - if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) - if (input.parentID !== undefined) - conditions.push( - input.parentID === null ? isNull(SessionTable.parent_id) : eq(SessionTable.parent_id, input.parentID), - ) - if (input.anchor) { - conditions.push( - order === "asc" - ? or( - gt(sortColumn, input.anchor.time), - and(eq(sortColumn, input.anchor.time), gt(SessionTable.id, input.anchor.id)), - )! - : or( - lt(sortColumn, input.anchor.time), - and(eq(sortColumn, input.anchor.time), lt(SessionTable.id, input.anchor.id)), - )!, - ) - } - const query = db - .select() - .from(SessionTable) - .where(conditions.length > 0 ? and(...conditions) : undefined) - .orderBy( - order === "asc" ? asc(sortColumn) : desc(sortColumn), - order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), - ) - const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( - Effect.orDie, - ) - return { data: (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row)) } + list: Effect.fn("Session.list")(function* (input) { + return { data: yield* store.list(input) } }), messages: Effect.fn("Session.messages")(function* (input) { yield* result.get(input.sessionID) - const direction = input.cursor?.direction ?? "next" - const requestedOrder = input.order ?? "desc" - const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder - const anchor = input.cursor - ? yield* db - .select({ seq: SessionMessageTable.seq }) - .from(SessionMessageTable) - .where( - and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.cursor.id)), - ) - .get() - .pipe(Effect.orDie) - : undefined - if (input.cursor && !anchor) return [] - const boundary = anchor - ? order === "asc" - ? gt(SessionMessageTable.seq, anchor.seq) - : lt(SessionMessageTable.seq, anchor.seq) - : undefined - const where = boundary - ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) - : eq(SessionMessageTable.session_id, input.sessionID) - const query = db - .select() - .from(SessionMessageTable) - .where(where) - .orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq)) - const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( - Effect.orDie, - ) - return yield* Effect.forEach( - direction === "previous" ? rows.toReversed() : rows, - SessionHistory.decodeMessageRow, - ) + return yield* store.messages(input) }), message: (input) => sessions.forSession(input.sessionID).message(input.messageID), updateMessage: (input) => sessions.forSession(input.sessionID).updateMessage(input), diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index f657c3a26c6..e10c16e47e5 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -1,7 +1,10 @@ export * as SessionStore from "./store.js" -import { and, eq, isNotNull, isNull, notInArray, sql } from "drizzle-orm" -import { Context, Effect, Layer } from "effect" +import { and, asc, desc, eq, gt, isNotNull, isNull, like, lt, notInArray, or, sql, type SQL } from "drizzle-orm" +import { Context, Effect, Layer, Schema } from "effect" +import { Project } from "@opencode-ai/schema/project" +import { Workspace } from "@opencode-ai/schema/workspace" +import { AbsolutePath, PositiveInt, RelativePath } from "@opencode-ai/schema/schema" import { Database } from "../database/database.js" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { SessionHistory } from "./history.js" @@ -11,8 +14,45 @@ import { Session } from "@opencode-ai/schema/session" import { SessionMessageTable, SessionTable } from "./sql.js" import { fromRow } from "./info.js" +const ListInputBase = { + workspaceID: Workspace.ID.pipe(Schema.optional), + search: Schema.String.pipe(Schema.optional), + limit: PositiveInt.pipe(Schema.optional), + order: Schema.Literals(["asc", "desc"]).pipe(Schema.optional), + parentID: Schema.NullOr(Session.ID).pipe(Schema.optional), + anchor: Session.ListAnchor.pipe(Schema.optional), +} + +const ListDirectoryInput = Schema.Struct({ + ...ListInputBase, + directory: AbsolutePath, +}) + +const ListProjectInput = Schema.Struct({ + ...ListInputBase, + project: Project.ID, + subpath: RelativePath.pipe(Schema.optional), +}) + +const ListAllInput = Schema.Struct(ListInputBase) + +export const ListInput = Schema.Union([ListDirectoryInput, ListProjectInput, ListAllInput]) +export type ListInput = typeof ListInput.Type + +export type MessagesInput = { + sessionID: Session.ID + limit?: number + order?: "asc" | "desc" + cursor?: { + id: SessionMessage.ID + direction: "previous" | "next" + } +} + export interface Interface { readonly get: (sessionID: Session.ID) => Effect.Effect + readonly list: (input?: ListInput) => Effect.Effect + readonly messages: (input: MessagesInput) => Effect.Effect readonly context: (sessionID: Session.ID) => Effect.Effect readonly message: ( messageID: SessionMessage.ID, @@ -55,6 +95,83 @@ const layer = Layer.effect( const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get().pipe(Effect.orDie) return row ? fromRow(row) : undefined }), + list: Effect.fn("SessionStore.list")(function* (input = {}) { + const direction = input.anchor?.direction ?? "next" + const requestedOrder = input.order ?? "desc" + const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder + const sortColumn = SessionTable.time_updated + const conditions: SQL[] = [] + if ("directory" in input) conditions.push(eq(SessionTable.directory, input.directory)) + if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) + if ("project" in input) conditions.push(eq(SessionTable.project_id, input.project)) + if ("project" in input && input.subpath !== undefined) conditions.push(eq(SessionTable.path, input.subpath)) + if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) + if (input.parentID !== undefined) + conditions.push( + input.parentID === null ? isNull(SessionTable.parent_id) : eq(SessionTable.parent_id, input.parentID), + ) + if (input.anchor) { + conditions.push( + order === "asc" + ? or( + gt(sortColumn, input.anchor.time), + and(eq(sortColumn, input.anchor.time), gt(SessionTable.id, input.anchor.id)), + )! + : or( + lt(sortColumn, input.anchor.time), + and(eq(sortColumn, input.anchor.time), lt(SessionTable.id, input.anchor.id)), + )!, + ) + } + const query = db + .select() + .from(SessionTable) + .where(conditions.length > 0 ? and(...conditions) : undefined) + .orderBy( + order === "asc" ? asc(sortColumn) : desc(sortColumn), + order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), + ) + const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( + Effect.orDie, + ) + return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row)) + }), + messages: Effect.fn("SessionStore.messages")(function* (input) { + const direction = input.cursor?.direction ?? "next" + const requestedOrder = input.order ?? "desc" + const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder + const anchor = input.cursor + ? yield* db + .select({ seq: SessionMessageTable.seq }) + .from(SessionMessageTable) + .where( + and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.cursor.id)), + ) + .get() + .pipe(Effect.orDie) + : undefined + if (input.cursor && !anchor) return [] + const boundary = anchor + ? order === "asc" + ? gt(SessionMessageTable.seq, anchor.seq) + : lt(SessionMessageTable.seq, anchor.seq) + : undefined + const where = boundary + ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) + : eq(SessionMessageTable.session_id, input.sessionID) + const query = db + .select() + .from(SessionMessageTable) + .where(where) + .orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq)) + const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( + Effect.orDie, + ) + return yield* Effect.forEach( + direction === "previous" ? rows.toReversed() : rows, + SessionHistory.decodeMessageRow, + ) + }), context: Effect.fn("SessionStore.context")((sessionID) => SessionHistory.load(db, sessionID)), message: Effect.fn("SessionStore.message")(function* (messageID) { const row = yield* db diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index a987ca217b7..7591856881c 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -21,6 +21,7 @@ import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" import { fromRow } from "@opencode-ai/core/session/info" import { SessionInbox } from "@opencode-ai/core/session/inbox" +import { SessionStore } from "@opencode-ai/core/session/store" import { Shell } from "@opencode-ai/schema/shell" import { InstructionStateTable, @@ -32,9 +33,10 @@ import { testEffect } from "./lib/effect" import { Snapshot } from "@opencode-ai/core/snapshot" const it = testEffect( - AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionInbox.node]), [ - [Bus.node, Bus.configured({ persist: true })], - ]), + AppNodeBuilder.build( + LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionInbox.node, SessionStore.node]), + [[Bus.node, Bus.configured({ persist: true })]], + ), ) const sessionsLayer = AppNodeBuilder.build(Session.node, [[SessionExecution.node, SessionExecution.noopLayer]]) const sessionID = Session.ID.make("ses_projector_test") @@ -278,7 +280,9 @@ describe("SessionProjector", () => { yield* db.run(sql`update session_message set data = '{"time":{"created":0}}' where id = ${messageID}`) const sessions = yield* Session.Service + const store = yield* SessionStore.Service const expected = { _tag: "Session.MessageDecodeError", sessionID, messageID } + expect(yield* store.messages({ sessionID }).pipe(Effect.flip)).toMatchObject(expected) expect(yield* sessions.messages({ sessionID }).pipe(Effect.flip)).toMatchObject(expected) expect(yield* sessions.context(sessionID).pipe(Effect.flip)).toMatchObject(expected) expect(yield* sessions.message({ sessionID, messageID }).pipe(Effect.catchDefect(Effect.succeed))).toMatchObject( @@ -287,6 +291,21 @@ describe("SessionProjector", () => { }).pipe(Effect.provide(sessionsLayer)), ) + it.effect("checks session existence before resolving a missing message cursor", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const missing = Session.ID.make("ses_missing") + expect( + yield* sessions + .messages({ + sessionID: missing, + cursor: { id: SessionMessage.ID.make("msg_missing"), direction: "next" }, + }) + .pipe(Effect.flip), + ).toEqual(new Session.NotFoundError({ sessionID: missing })) + }).pipe(Effect.provide(sessionsLayer)), + ) + it.effect("consumes the pending row and projects the message at promotion", () => Effect.gen(function* () { const db = yield* seedSession() diff --git a/packages/core/test/session-store.test.ts b/packages/core/test/session-store.test.ts new file mode 100644 index 00000000000..fcf472f37d2 --- /dev/null +++ b/packages/core/test/session-store.test.ts @@ -0,0 +1,190 @@ +import { describe, expect } from "bun:test" +import { Effect } from "effect" +import { Bus } from "@opencode-ai/core/bus" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionStore } from "@opencode-ai/core/session/store" +import { Event } from "@opencode-ai/schema/event" +import { Project } from "@opencode-ai/schema/project" +import { AbsolutePath } from "@opencode-ai/schema/schema" +import { Session } from "@opencode-ai/schema/session" +import { SessionEvent } from "@opencode-ai/schema/session-event" +import { SessionMessage } from "@opencode-ai/schema/session-message" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { testEffect } from "./lib/effect" + +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node]), [ + [Bus.node, Bus.configured({ persist: true })], + ]), +) + +const seedSessions = (rows: { id: string; updated: number }[]) => + Effect.gen(function* () { + const database = yield* Database.Service + const bus = yield* Bus.Service + const directory = AbsolutePath.make("/project") + yield* database.db.insert(ProjectTable).values({ id: Project.ID.global, worktree: directory, sandboxes: [] }).run() + yield* Effect.forEach(rows, (row) => + Effect.gen(function* () { + const sessionID = Session.ID.make(row.id) + yield* bus.publish(SessionEvent.Created, { + sessionID, + projectID: Project.ID.global, + location: { directory }, + slug: "store-test", + version: "test", + }) + yield* bus.replay({ + id: Event.ID.create(), + created: row.updated, + aggregateID: sessionID, + seq: 1, + type: Bus.versionedType(SessionEvent.Renamed.type, 1), + data: { sessionID, title: row.id }, + }) + }), + ) + return bus + }) + +describe("SessionStore", () => { + it.effect("lists by updated time and ID with exclusive two-item pages in either direction", () => + Effect.gen(function* () { + yield* seedSessions([ + { id: "ses_d", updated: 20 }, + { id: "ses_z", updated: 10 }, + { id: "ses_a", updated: 30 }, + { id: "ses_c", updated: 20 }, + { id: "ses_y", updated: 10 }, + { id: "ses_e", updated: 30 }, + { id: "ses_b", updated: 20 }, + ]) + const store = yield* SessionStore.Service + expect((yield* store.list()).map((session) => String(session.id))).toEqual([ + "ses_e", + "ses_a", + "ses_d", + "ses_c", + "ses_b", + "ses_z", + "ses_y", + ]) + expect((yield* store.list({ order: "asc" })).map((session) => String(session.id))).toEqual([ + "ses_y", + "ses_z", + "ses_b", + "ses_c", + "ses_d", + "ses_a", + "ses_e", + ]) + const pages: { order: "asc" | "desc"; direction: "next" | "previous"; ids: string[] }[] = [ + { order: "asc", direction: "next", ids: ["ses_d", "ses_a"] }, + { order: "asc", direction: "previous", ids: ["ses_z", "ses_b"] }, + { order: "desc", direction: "next", ids: ["ses_b", "ses_z"] }, + { order: "desc", direction: "previous", ids: ["ses_a", "ses_d"] }, + ] + yield* Effect.forEach(pages, (page) => + Effect.gen(function* () { + const sessions = yield* store.list({ + order: page.order, + limit: 2, + anchor: { id: Session.ID.make("ses_c"), time: 20, direction: page.direction }, + }) + expect(sessions.map((session) => String(session.id))).toEqual(page.ids) + }), + ) + }), + ) + + it.effect("pages messages by durable sequence, not timestamp or ID, and scopes cursor lookup", () => + Effect.gen(function* () { + const sessionID = Session.ID.make("ses_messages") + const foreignID = Session.ID.make("ses_foreign") + const bus = yield* seedSessions([ + { id: sessionID, updated: 0 }, + { id: foreignID, updated: 0 }, + ]) + const store = yield* SessionStore.Service + yield* Effect.forEach( + [ + { id: "evt_z", created: 300 }, + { id: "evt_b", created: 700 }, + { id: "evt_x", created: 100 }, + { id: "evt_c", created: 400 }, + { id: "evt_w", created: 200 }, + { id: "evt_a", created: 600 }, + { id: "evt_y", created: 500 }, + ], + (event, index) => + bus.replay({ + id: Event.ID.make(event.id), + created: event.created, + aggregateID: sessionID, + seq: index + 2, + type: Bus.versionedType(SessionEvent.Synthetic.type, 1), + data: { sessionID, text: event.id }, + }), + ) + yield* bus.publish( + SessionEvent.Synthetic, + { sessionID: foreignID, text: "foreign" }, + { + id: Event.ID.make("evt_foreign"), + }, + ) + expect((yield* store.messages({ sessionID })).map((message) => String(message.id))).toEqual([ + "msg_y", + "msg_a", + "msg_w", + "msg_c", + "msg_x", + "msg_b", + "msg_z", + ]) + expect((yield* store.messages({ sessionID, order: "asc" })).map((message) => String(message.id))).toEqual([ + "msg_z", + "msg_b", + "msg_x", + "msg_c", + "msg_w", + "msg_a", + "msg_y", + ]) + const pages: { order: "asc" | "desc"; direction: "next" | "previous"; ids: string[] }[] = [ + { order: "asc", direction: "next", ids: ["msg_w", "msg_a"] }, + { order: "asc", direction: "previous", ids: ["msg_b", "msg_x"] }, + { order: "desc", direction: "next", ids: ["msg_x", "msg_b"] }, + { order: "desc", direction: "previous", ids: ["msg_a", "msg_w"] }, + ] + yield* Effect.forEach(pages, (page) => + Effect.gen(function* () { + const messages = yield* store.messages({ + sessionID, + order: page.order, + limit: 2, + cursor: { id: SessionMessage.ID.make("msg_c"), direction: page.direction }, + }) + expect(messages.map((message) => String(message.id))).toEqual(page.ids) + }), + ) + expect(yield* store.messages({ sessionID: Session.ID.make("ses_missing") })).toEqual([]) + expect( + yield* store.messages({ + sessionID, + cursor: { id: SessionMessage.ID.make("msg_missing"), direction: "next" }, + }), + ).toEqual([]) + expect( + yield* store.messages({ + sessionID, + order: "asc", + cursor: { id: SessionMessage.ID.make("msg_foreign"), direction: "next" }, + }), + ).toEqual([]) + }), + ) +}) diff --git a/packages/core/test/shared-schema.test.ts b/packages/core/test/shared-schema.test.ts index a079cee6d5a..0fd3fd3c9dc 100644 --- a/packages/core/test/shared-schema.test.ts +++ b/packages/core/test/shared-schema.test.ts @@ -2,6 +2,7 @@ import { expect, test } from "bun:test" import { Schema } from "effect" import { Agent } from "@opencode-ai/core/agent" import { Session } from "@opencode-ai/core/session" +import { SessionStore } from "@opencode-ai/core/session/store" import { Location } from "@opencode-ai/schema/location" import { Model } from "@opencode-ai/schema/model" import { Provider } from "@opencode-ai/schema/provider" @@ -126,6 +127,7 @@ test("Core reuses the canonical shared schemas", async () => { [Session.ID, schemaSession.Session.ID], [Session.Info, schemaSession.Session.Info], [Session.ListAnchor, schemaSession.Session.ListAnchor], + [Session.ListInput, SessionStore.ListInput], [coreSessionInbox.Delivery, SessionInbox.Delivery], [coreSessionInbox.Item, SessionInbox.Item], [coreSessionInbox.User, SessionInbox.User],