export * as SessionV2 from "./session" export * from "./session/schema" import { DateTime, Effect, Layer, Schema, Context, Stream } from "effect" import { ListAnchor } from "@opencode-ai/schema/session" import { and, asc, desc, eq, gt, like, lt, or, type SQL } from "drizzle-orm" import { ProjectV2 } from "./project" import { WorkspaceV2 } from "./workspace" import { ModelV2 } from "./model" import { Location } from "./location" import { SessionMessage } from "./session/message" import { Prompt } from "./session/prompt" import { PromptInput } from "@opencode-ai/schema/prompt-input" import { EventV2 } from "./event" import { Database } from "./database/database" import { SessionProjector } from "./session/projector" import { SessionMessageTable, SessionTable } from "./session/sql" import { SessionSchema } from "./session/schema" import { AbsolutePath, PositiveInt, RelativePath } from "./schema" import { AgentV2 } from "./agent" import { SessionV1 } from "./v1/session" import { InstallationVersion } from "./installation/version" import { Slug } from "./util/slug" import { ProjectTable } from "./project/sql" import path from "path" import { fromRow } from "./session/info" import { SessionRunner } from "./session/runner/index" import { SessionStore } from "./session/store" import { SessionExecution } from "./session/execution" import { makeGlobalNode } from "./effect/app-node" import { LocationServiceMap } from "./location-service-map" import { MessageDecodeError } from "./session/error" import { SessionEvent } from "./session/event" import { SessionInput } from "./session/input" import { Snapshot } from "./snapshot" import { SessionRevert } from "./session/revert" import { Revert } from "@opencode-ai/schema/revert" import { FSUtil } from "./fs-util" import { SessionDurable } from "@opencode-ai/schema/durable-event-manifest" export const RevertState = Revert.State export type RevertState = Revert.State // get project -> project.locations // // get all sessions // // - by project // - by subpath // - by workspace (home is special) export { ListAnchor } const ListInputBase = { workspaceID: WorkspaceV2.ID.pipe(Schema.optional), search: Schema.String.pipe(Schema.optional), limit: PositiveInt.pipe(Schema.optional), order: Schema.Literals(["asc", "desc"]).pipe(Schema.optional), anchor: ListAnchor.pipe(Schema.optional), } const ListDirectoryInput = Schema.Struct({ ...ListInputBase, directory: AbsolutePath, }) const ListProjectInput = Schema.Struct({ ...ListInputBase, project: ProjectV2.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 type CreateInput = { id?: SessionSchema.ID agent?: AgentV2.ID model?: ModelV2.Ref location: Location.Ref } type CompactInput = { sessionID: SessionSchema.ID prompt?: Prompt } export class NotFoundError extends Schema.TaggedErrorClass()("Session.NotFoundError", { sessionID: SessionSchema.ID, }) {} export class OperationUnavailableError extends Schema.TaggedErrorClass()( "Session.OperationUnavailableError", { operation: Schema.Literals(["move", "shell", "skill", "switchAgent", "compact", "wait"]), }, ) {} export { ContextSnapshotDecodeError, MessageDecodeError } from "./session/error" export class PromptConflictError extends Schema.TaggedErrorClass()("Session.PromptConflictError", { sessionID: SessionSchema.ID, messageID: SessionMessage.ID, }) {} export const MessageNotFoundError = SessionRevert.MessageNotFoundError export type MessageNotFoundError = SessionRevert.MessageNotFoundError export type Error = NotFoundError | MessageDecodeError | OperationUnavailableError | PromptConflictError export interface Interface { readonly list: (input?: ListInput) => Effect.Effect readonly create: (input: CreateInput) => Effect.Effect readonly get: (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 message: (input: { sessionID: SessionSchema.ID messageID: SessionMessage.ID }) => Effect.Effect readonly context: ( sessionID: SessionSchema.ID, ) => Effect.Effect readonly events: (input: { sessionID: SessionSchema.ID after?: number }) => Stream.Stream readonly history: (input: { sessionID: SessionSchema.ID after?: number limit: number }) => Effect.Effect<{ events: ReadonlyArray; hasMore: boolean }, NotFoundError> readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect readonly switchModel: (input: { sessionID: SessionSchema.ID model: ModelV2.Ref }) => Effect.Effect readonly prompt: (input: { id?: SessionMessage.ID sessionID: SessionSchema.ID prompt: PromptInput.Prompt delivery?: SessionInput.Delivery resume?: boolean }) => Effect.Effect readonly shell: (input: { id?: EventV2.ID sessionID: SessionSchema.ID command: string resume?: boolean }) => Effect.Effect readonly skill: (input: { id?: EventV2.ID sessionID: SessionSchema.ID skill: string resume?: boolean }) => Effect.Effect readonly compact: (input: CompactInput) => Effect.Effect readonly wait: (id: SessionSchema.ID) => Effect.Effect readonly active: Effect.Effect> readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect readonly revert: { readonly stage: (input: { sessionID: SessionSchema.ID messageID: SessionMessage.ID files?: boolean }) => Effect.Effect readonly clear: (sessionID: SessionSchema.ID) => Effect.Effect readonly commit: (sessionID: SessionSchema.ID) => Effect.Effect } } export class Service extends Context.Service()("@opencode/v2/Session") {} const layer = Layer.effect( Service, Effect.gen(function* () { const database = yield* Database.Service const db = database.db const events = yield* EventV2.Service const projects = yield* ProjectV2.Service const execution = yield* SessionExecution.Service const store = yield* SessionStore.Service const locations = yield* LocationServiceMap.Service const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message) const isDurableSessionEvent = Schema.is(SessionEvent.Durable) const decode = (row: typeof SessionMessageTable.$inferSelect) => decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe( Effect.mapError( () => new MessageDecodeError({ sessionID: SessionSchema.ID.make(row.session_id), messageID: SessionMessage.ID.make(row.id), }), ), ) const result = Service.of({ create: Effect.fn("V2Session.create")(function* (input) { const sessionID = input.id ?? SessionSchema.ID.create() const recorded = yield* store.get(sessionID) if (recorded) return recorded const project = yield* projects.resolve(input.location.directory) yield* db .insert(ProjectTable) .values({ id: project.id, worktree: project.directory, vcs: project.vcs?.type, sandboxes: [] }) .onConflictDoNothing() .run() .pipe(Effect.orDie) const now = Date.now() const info = SessionV1.SessionInfo.make({ id: sessionID, slug: Slug.create(), version: InstallationVersion, projectID: project.id, directory: input.location.directory, path: path.relative(project.directory, input.location.directory).replaceAll("\\", "/"), workspaceID: input.location.workspaceID ? WorkspaceV2.ID.make(input.location.workspaceID) : undefined, title: `New session - ${new Date(now).toISOString()}`, agent: input.agent, model: input.model ? { id: ModelV2.ID.make(input.model.id), providerID: input.model.providerID, variant: input.model.variant, } : undefined, cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, time: { created: now, updated: now }, }) const projected = yield* events .publish(SessionV1.Event.Created, { sessionID, info }, { location: input.location }) .pipe( Effect.as({ type: "created" } as const), Effect.catchDefect((defect) => { if (!(defect instanceof SessionProjector.SessionAlreadyProjected)) { return Effect.die(defect) } // Concurrent creation lost the projection race. The existing Session identity wins. return store .get(sessionID) .pipe( Effect.flatMap((session) => session ? Effect.succeed({ type: "existing", session } as const) : Effect.die(defect), ), ) }), ) if (projected.type === "existing") return projected.session // TODO: Restore recorded sessions onto replacement synchronized workspaces in a future API slice. return yield* result.get(sessionID).pipe(Effect.orDie) }), get: Effect.fn("V2Session.get")(function* (sessionID) { const session = yield* store.get(sessionID) if (!session) return yield* new NotFoundError({ sessionID }) return session }), list: Effect.fn("V2Session.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_created 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 (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) 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("V2Session.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, decode) }), message: Effect.fn("V2Session.message")(function* (input) { const stored = yield* store.message(input.messageID) return stored?.sessionID === input.sessionID ? stored.message : undefined }), context: Effect.fn("V2Session.context")(function* (sessionID) { yield* result.get(sessionID) return yield* store.context(sessionID) }), events: (input) => Stream.unwrap( result .get(input.sessionID) .pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))), ).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))), history: Effect.fn("V2Session.history")(function* (input) { yield* result.get(input.sessionID) return yield* EventV2.readAggregate(db, { ...input, aggregateID: input.sessionID, manifest: SessionDurable, }) }), prompt: Effect.fn("V2Session.prompt")((input) => Effect.uninterruptible( Effect.gen(function* () { yield* result.get(input.sessionID) const prompt = resolvePrompt(input.prompt) const messageID = input.id ?? SessionMessage.ID.create() const delivery = input.delivery ?? "steer" const expected = { sessionID: input.sessionID, messageID, prompt, delivery } const admitted = yield* SessionInput.admit(db, events, { id: messageID, sessionID: input.sessionID, prompt, delivery, }).pipe( Effect.catchDefect((defect) => defect instanceof SessionInput.LifecycleConflict ? new PromptConflictError({ sessionID: input.sessionID, messageID }) : Effect.die(defect), ), ) if (!SessionInput.equivalent(admitted, expected)) return yield* new PromptConflictError({ sessionID: input.sessionID, messageID }) if (input.resume !== false) yield* execution.wake(admitted.sessionID) return admitted }), ), ), shell: Effect.fn("V2Session.shell")(function* () { return yield* new OperationUnavailableError({ operation: "shell" }) }), skill: Effect.fn("V2Session.skill")(function* () { return yield* new OperationUnavailableError({ operation: "skill" }) }), switchAgent: Effect.fn("V2Session.switchAgent")(function* (input) { yield* result.get(input.sessionID) yield* events.publish(SessionEvent.AgentSwitched, { sessionID: input.sessionID, messageID: SessionMessage.ID.create(), timestamp: yield* DateTime.now, agent: input.agent, }) }), switchModel: Effect.fn("V2Session.switchModel")(function* (input) { const session = yield* result.get(input.sessionID) if ( session.model?.providerID === input.model.providerID && session.model.id === input.model.id && (session.model.variant ?? "default") === (input.model.variant ?? "default") ) return yield* events.publish(SessionEvent.ModelSwitched, { sessionID: input.sessionID, messageID: SessionMessage.ID.create(), timestamp: yield* DateTime.now, model: input.model, }) }), compact: Effect.fn("V2Session.compact")(function* (input) { yield* result.get(input.sessionID) return yield* new OperationUnavailableError({ operation: "compact" }) }), wait: Effect.fn("V2Session.wait")(function* (sessionID) { yield* result.get(sessionID) return yield* new OperationUnavailableError({ operation: "wait" }) }), active: execution.active, resume: Effect.fn("V2Session.resume")(function* (sessionID) { yield* result.get(sessionID) yield* execution.resume(sessionID) }), interrupt: Effect.fn("V2Session.interrupt")((sessionID) => Effect.uninterruptible(execution.interrupt(sessionID)), ), revert: { stage: Effect.fn("V2Session.revert.stage")(function* (input) { const session = yield* result.get(input.sessionID) return yield* SessionRevert.stage({ session, messageID: input.messageID, files: input.files }).pipe( Effect.provideService(Database.Service, database), Effect.provideService(EventV2.Service, events), Effect.provide(locations.get(session.location)), ) }), clear: Effect.fn("V2Session.revert.clear")(function* (sessionID) { const session = yield* result.get(sessionID) yield* SessionRevert.clear(session).pipe( Effect.provideService(EventV2.Service, events), Effect.provide(locations.get(session.location)), ) }), commit: Effect.fn("V2Session.revert.commit")(function* (sessionID) { const session = yield* result.get(sessionID) yield* SessionRevert.commit(session).pipe(Effect.provideService(EventV2.Service, events)) }), }, }) return result }), ) const resolvePrompt = (input: PromptInput.Prompt) => Prompt.make({ text: input.text, agents: input.agents, files: input.files?.map((file) => { const dataMime = file.uri.match(/^data:([^;,]+)[;,]/i)?.[1] const target = URL.canParse(file.uri) ? new URL(file.uri).pathname : (file.name ?? file.uri) return { ...file, mime: dataMime ?? (target.endsWith("/") ? "application/x-directory" : FSUtil.mimeType(target)), } }), }) export const node = makeGlobalNode({ service: Service, layer: layer.pipe(Layer.orDie), deps: [ Database.node, EventV2.node, ProjectV2.node, SessionExecution.node, SessionStore.node, LocationServiceMap.node, SessionProjector.node, ], })