372 lines
15 KiB
TypeScript
372 lines
15 KiB
TypeScript
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<typeof Delivery>
|
|
|
|
export const DefaultDelivery = "immediate" satisfies Delivery
|
|
|
|
export class Info extends Schema.Class<Info>("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<NotFoundError>()("Session.NotFoundError", {
|
|
sessionID: SessionID,
|
|
}) {}
|
|
|
|
export class OperationUnavailableError extends Schema.TaggedErrorClass<OperationUnavailableError>()(
|
|
"Session.OperationUnavailableError",
|
|
{
|
|
operation: Schema.Literals(["prompt", "compact", "wait"]),
|
|
},
|
|
) {}
|
|
|
|
export class MessageDecodeError extends Schema.TaggedErrorClass<MessageDecodeError>()("Session.MessageDecodeError", {
|
|
sessionID: SessionID,
|
|
messageID: SessionMessage.ID,
|
|
}) {}
|
|
|
|
export interface Interface {
|
|
readonly create: (input?: {
|
|
agent?: string
|
|
model?: ModelV2.Ref
|
|
parentID?: SessionID
|
|
workspaceID?: WorkspaceID
|
|
}) => Effect.Effect<Info>
|
|
readonly get: (sessionID: SessionID) => Effect.Effect<Info, NotFoundError>
|
|
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<Info[], never>
|
|
readonly messages: (input: {
|
|
sessionID: SessionID
|
|
limit?: number
|
|
order?: "asc" | "desc"
|
|
cursor?: {
|
|
id: SessionMessage.ID
|
|
time: number
|
|
direction: "previous" | "next"
|
|
}
|
|
}) => Effect.Effect<SessionMessage.Message[], NotFoundError | MessageDecodeError>
|
|
readonly context: (
|
|
sessionID: SessionID,
|
|
) => Effect.Effect<SessionMessage.Message[], NotFoundError | MessageDecodeError>
|
|
readonly prompt: (input: {
|
|
id?: EventV2.ID
|
|
sessionID: SessionID
|
|
prompt: Prompt
|
|
delivery?: Delivery
|
|
}) => Effect.Effect<SessionMessage.User, NotFoundError | OperationUnavailableError>
|
|
readonly shell: (input: { id?: EventV2.ID; sessionID: SessionID; command: string }) => Effect.Effect<void, never>
|
|
readonly skill: (input: { id?: EventV2.ID; sessionID: SessionID; skill: string }) => Effect.Effect<void, never>
|
|
readonly subagent: (input: {
|
|
id?: EventV2.ID
|
|
parentID: SessionID
|
|
prompt: Prompt
|
|
agent: string
|
|
model?: ModelV2.Ref
|
|
}) => Effect.Effect<void, NotFoundError | OperationUnavailableError | MessageDecodeError>
|
|
readonly switchAgent: (input: { sessionID: SessionID; agent: string }) => Effect.Effect<void, never>
|
|
readonly switchModel: (input: { sessionID: SessionID; model: ModelV2.Ref }) => Effect.Effect<void, never>
|
|
readonly compact: (sessionID: SessionID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
|
|
readonly wait: (sessionID: SessionID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
|
|
}
|
|
|
|
export class Service extends Context.Service<Service, Interface>()("@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: ModelV2.VariantID.make(row.model.variant ?? "default"),
|
|
}
|
|
: 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"
|