import { SessionMessageTable, SessionTable } from "@/session/session.sql" import { SessionID } from "@/session/schema" import { WorkspaceID } from "@/control-plane/schema" import { and, asc, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" import * as Database from "@/storage/db" import { Context, DateTime, Effect, Layer, Schema } from "effect" import { SessionMessage } from "@opencode-ai/core/session-message" import type { Prompt } from "@opencode-ai/core/session-prompt" import { ProjectID } from "@/project/schema" import { SessionEvent } from "@opencode-ai/core/session-event" import { V2Schema } from "@opencode-ai/core/v2-schema" import { optionalOmitUndefined } from "@opencode-ai/core/schema" import { EventV2 } from "@opencode-ai/core/event" import { EventV2Bridge } from "@/event-v2-bridge" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" export const Delivery = Schema.Literals(["immediate", "deferred"]).annotate({ identifier: "Session.Delivery", }) export type Delivery = Schema.Schema.Type export const DefaultDelivery = "immediate" satisfies Delivery export class Info extends Schema.Class("Session.Info")({ id: SessionID, parentID: optionalOmitUndefined(SessionID), projectID: ProjectID, workspaceID: optionalOmitUndefined(WorkspaceID), path: optionalOmitUndefined(Schema.String), agent: optionalOmitUndefined(Schema.String), model: ModelV2.Ref.pipe(optionalOmitUndefined), cost: Schema.Finite, tokens: Schema.Struct({ input: Schema.Finite, output: Schema.Finite, reasoning: Schema.Finite, cache: Schema.Struct({ read: Schema.Finite, write: Schema.Finite, }), }), time: Schema.Struct({ created: V2Schema.DateTimeUtcFromMillis, updated: V2Schema.DateTimeUtcFromMillis, archived: optionalOmitUndefined(V2Schema.DateTimeUtcFromMillis), }), title: Schema.String, /* slug: Schema.String, directory: Schema.String, path: optionalOmitUndefined(Schema.String), parentID: optionalOmitUndefined(SessionID), summary: optionalOmitUndefined(Summary), share: optionalOmitUndefined(Share), title: Schema.String, version: Schema.String, time: Time, permission: optionalOmitUndefined(Permission.Ruleset), revert: optionalOmitUndefined(Revert), */ }) {} export class NotFoundError extends Schema.TaggedErrorClass()("Session.NotFoundError", { sessionID: SessionID, }) {} export class OperationUnavailableError extends Schema.TaggedErrorClass()( "Session.OperationUnavailableError", { operation: Schema.Literals(["prompt", "compact", "wait"]), }, ) {} export class MessageDecodeError extends Schema.TaggedErrorClass()("Session.MessageDecodeError", { sessionID: SessionID, messageID: SessionMessage.ID, }) {} export interface Interface { readonly create: (input?: { agent?: string model?: ModelV2.Ref parentID?: SessionID workspaceID?: WorkspaceID }) => Effect.Effect readonly get: (sessionID: SessionID) => Effect.Effect readonly list: (input: { limit?: number order?: "asc" | "desc" directory?: string path?: string workspaceID?: WorkspaceID roots?: boolean start?: number search?: string cursor?: { id: SessionID time: number direction: "previous" | "next" } }) => Effect.Effect readonly messages: (input: { sessionID: SessionID limit?: number order?: "asc" | "desc" cursor?: { id: SessionMessage.ID time: number direction: "previous" | "next" } }) => Effect.Effect readonly context: ( sessionID: SessionID, ) => Effect.Effect readonly prompt: (input: { id?: EventV2.ID sessionID: SessionID prompt: Prompt delivery?: Delivery }) => Effect.Effect readonly shell: (input: { id?: EventV2.ID; sessionID: SessionID; command: string }) => Effect.Effect readonly skill: (input: { id?: EventV2.ID; sessionID: SessionID; skill: string }) => Effect.Effect readonly subagent: (input: { id?: EventV2.ID parentID: SessionID prompt: Prompt agent: string model?: ModelV2.Ref }) => Effect.Effect readonly switchAgent: (input: { sessionID: SessionID; agent: string }) => Effect.Effect readonly switchModel: (input: { sessionID: SessionID; model: ModelV2.Ref }) => Effect.Effect readonly compact: (sessionID: SessionID) => Effect.Effect readonly wait: (sessionID: SessionID) => Effect.Effect } export class Service extends Context.Service()("@opencode/v2/Session") {} export const layer = Layer.effect( Service, Effect.gen(function* () { const events = yield* EventV2Bridge.Service const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message) const decode = (row: typeof SessionMessageTable.$inferSelect) => decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe( Effect.mapError( () => new MessageDecodeError({ sessionID: SessionID.make(row.session_id), messageID: SessionMessage.ID.make(row.id), }), ), ) function fromRow(row: typeof SessionTable.$inferSelect): Info { return new Info({ id: SessionID.make(row.id), projectID: ProjectID.make(row.project_id), workspaceID: row.workspace_id ? WorkspaceID.make(row.workspace_id) : undefined, title: row.title, parentID: row.parent_id ? SessionID.make(row.parent_id) : undefined, path: row.path ?? "", agent: row.agent ?? undefined, model: row.model ? { id: ModelV2.ID.make(row.model.id), providerID: ProviderV2.ID.make(row.model.providerID), variant: row.model.variant ? ModelV2.VariantID.make(row.model.variant) : undefined, } : undefined, cost: row.cost, tokens: { input: row.tokens_input, output: row.tokens_output, reasoning: row.tokens_reasoning, cache: { read: row.tokens_cache_read, write: row.tokens_cache_write, }, }, time: { created: DateTime.makeUnsafe(row.time_created), updated: DateTime.makeUnsafe(row.time_updated), archived: row.time_archived ? DateTime.makeUnsafe(row.time_archived) : undefined, }, }) } const result = Service.of({ create: Effect.fn("V2Session.create")(function* (_input) { return {} as any }), get: Effect.fn("V2Session.get")(function* (sessionID) { const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()) if (!row) return yield* new NotFoundError({ sessionID }) return fromRow(row) }), list: Effect.fn("V2Session.list")(function* (input) { const direction = input.cursor?.direction ?? "next" let order = input.order ?? "desc" // This is a load bearing sort, desktop relies on this const sortColumn = SessionTable.time_updated // Query the adjacent rows in reverse, then flip them back into the requested order below. if (direction === "previous" && order === "asc") order = "desc" if (direction === "previous" && order === "desc") order = "asc" const conditions: SQL[] = [] if (input.directory) conditions.push(eq(SessionTable.directory, input.directory)) if (input.path) conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!) if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) if (input.roots) conditions.push(isNull(SessionTable.parent_id)) if (input.start) conditions.push(gte(sortColumn, input.start)) if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) if (input.cursor) { conditions.push( order === "asc" ? or( gt(sortColumn, input.cursor.time), and(eq(sortColumn, input.cursor.time), gt(SessionTable.id, input.cursor.id)), )! : or( lt(sortColumn, input.cursor.time), and(eq(sortColumn, input.cursor.time), lt(SessionTable.id, input.cursor.id)), )!, ) } const query = Database.Client() .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 = input.limit === undefined ? query.all() : query.limit(input.limit).all() return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row)) }), messages: Effect.fn("V2Session.messages")(function* (input) { yield* result.get(input.sessionID) const direction = input.cursor?.direction ?? "next" let order = input.order ?? "desc" // Query the adjacent rows in reverse, then flip them back into the requested order below. if (direction === "previous" && order === "asc") order = "desc" if (direction === "previous" && order === "desc") order = "asc" const boundary = input.cursor ? order === "asc" ? or( gt(SessionMessageTable.time_created, input.cursor.time), and( eq(SessionMessageTable.time_created, input.cursor.time), gt(SessionMessageTable.id, input.cursor.id), ), ) : or( lt(SessionMessageTable.time_created, input.cursor.time), and( eq(SessionMessageTable.time_created, input.cursor.time), lt(SessionMessageTable.id, input.cursor.id), ), ) : undefined const where = boundary ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) : eq(SessionMessageTable.session_id, input.sessionID) const rows = Database.use((db) => { const query = db .select() .from(SessionMessageTable) .where(where) .orderBy( order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), ) const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() return direction === "previous" ? rows.toReversed() : rows }) return yield* Effect.forEach(rows, (row) => decode(row)) }), context: Effect.fn("V2Session.context")(function* (sessionID) { yield* result.get(sessionID) const rows = Database.use((db) => { const compaction = db .select() .from(SessionMessageTable) .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id)) .limit(1) .get() return db .select() .from(SessionMessageTable) .where( and( eq(SessionMessageTable.session_id, sessionID), compaction ? or( gt(SessionMessageTable.time_created, compaction.time_created), and( eq(SessionMessageTable.time_created, compaction.time_created), gte(SessionMessageTable.id, compaction.id), ), ) : undefined, ), ) .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)) .all() }) return yield* Effect.forEach(rows, (row) => decode(row)) }), prompt: Effect.fn("V2Session.prompt")(function* (input) { yield* result.get(input.sessionID) return yield* new OperationUnavailableError({ operation: "prompt" }) }), shell: Effect.fn("V2Session.shell")(function* (_input) {}), skill: Effect.fn("V2Session.skill")(function* (_input) {}), switchAgent: Effect.fn("V2Session.switchAgent")(function* (input) { yield* events.publish(SessionEvent.AgentSwitched, { sessionID: input.sessionID, timestamp: DateTime.makeUnsafe(Date.now()), agent: input.agent, }) }), switchModel: Effect.fn("V2Session.switchModel")(function* (input) { yield* events.publish(SessionEvent.ModelSwitched, { sessionID: input.sessionID, timestamp: DateTime.makeUnsafe(Date.now()), model: input.model, }) }), subagent: Effect.fn("V2Session.subagent")(function* (input) { const parent = yield* result.get(input.parentID) const child = yield* result.create({ agent: input.agent, model: input.model, parentID: input.parentID, workspaceID: parent.workspaceID, }) yield* result.prompt({ prompt: input.prompt, sessionID: child.id, }) yield* Effect.gen(function* () { yield* result.wait(child.id) const messages = yield* result.messages({ sessionID: child.id, order: "desc" }) const assistant = messages.find((msg) => msg.type === "assistant") if (!assistant) return const text = assistant.content.findLast((part) => part.type === "text") if (!text) return }).pipe(Effect.forkChild()) }), compact: Effect.fn("V2Session.compact")(function* (sessionID) { yield* result.get(sessionID) return yield* new OperationUnavailableError({ operation: "compact" }) }), wait: Effect.fn("V2Session.wait")(function* (sessionID) { yield* result.get(sessionID) return yield* new OperationUnavailableError({ operation: "wait" }) }), }) return result }), ) export const defaultLayer = layer.pipe(Layer.provide(EventV2Bridge.defaultLayer)) export * as SessionV2 from "./session"