openAgent

1export * as SessionV2 from "./session"
2export * from "./session/schema"
3 
4import { DateTime, Effect, Layer, Schema, Context, Stream } from "effect"
5import { ListAnchor } from "@opencode-ai/schema/session"
6import { and, asc, desc, eq, gt, like, lt, or, type SQL } from "drizzle-orm"
7import { ProjectV2 } from "./project"
8import { WorkspaceV2 } from "./workspace"
9import { ModelV2 } from "./model"
10import { Location } from "./location"
11import { SessionMessage } from "./session/message"
12import { Prompt } from "./session/prompt"
13import { PromptInput } from "@opencode-ai/schema/prompt-input"
14import { EventV2 } from "./event"
15import { Database } from "./database/database"
16import { SessionProjector } from "./session/projector"
17import { SessionMessageTable, SessionTable } from "./session/sql"
18import { SessionSchema } from "./session/schema"
19import { AbsolutePath, PositiveInt, RelativePath } from "./schema"
20import { AgentV2 } from "./agent"
21import { SessionV1 } from "./v1/session"
22import { InstallationVersion } from "./installation/version"
23import { Slug } from "./util/slug"
24import { ProjectTable } from "./project/sql"
25import path from "path"
26import { fromRow } from "./session/info"
27import { SessionRunner } from "./session/runner/index"
28import { SessionStore } from "./session/store"
29import { SessionExecution } from "./session/execution"
30import { makeGlobalNode } from "./effect/app-node"
31import { LocationServiceMap } from "./location-service-map"
32import { MessageDecodeError } from "./session/error"
33import { SessionEvent } from "./session/event"
34import { SessionInput } from "./session/input"
35import { Snapshot } from "./snapshot"
36import { SessionRevert } from "./session/revert"
37import { Revert } from "@opencode-ai/schema/revert"
38import { FSUtil } from "./fs-util"
39import { SessionDurable } from "@opencode-ai/schema/durable-event-manifest"
40 
41export const RevertState = Revert.State
42export type RevertState = Revert.State
43 
44// get project -> project.locations
45//
46// get all sessions
47//
48 
49// - by project
50//   - by subpath
51// - by workspace (home is special)
52 
53export { ListAnchor }
54 
55const ListInputBase = {
56  workspaceID: WorkspaceV2.ID.pipe(Schema.optional),
57  search: Schema.String.pipe(Schema.optional),
58  limit: PositiveInt.pipe(Schema.optional),
59  order: Schema.Literals(["asc", "desc"]).pipe(Schema.optional),
60  anchor: ListAnchor.pipe(Schema.optional),
61}
62 
63const ListDirectoryInput = Schema.Struct({
64  ...ListInputBase,
65  directory: AbsolutePath,
66})
67 
68const ListProjectInput = Schema.Struct({
69  ...ListInputBase,
70  project: ProjectV2.ID,
71  subpath: RelativePath.pipe(Schema.optional),
72})
73 
74const ListAllInput = Schema.Struct(ListInputBase)
75 
76export const ListInput = Schema.Union([ListDirectoryInput, ListProjectInput, ListAllInput])
77export type ListInput = typeof ListInput.Type
78 
79type CreateInput = {
80  id?: SessionSchema.ID
81  agent?: AgentV2.ID
82  model?: ModelV2.Ref
83  location: Location.Ref
84}
85 
86type CompactInput = {
87  sessionID: SessionSchema.ID
88  prompt?: Prompt
89}
90 
91export class NotFoundError extends Schema.TaggedErrorClass<NotFoundError>()("Session.NotFoundError", {
92  sessionID: SessionSchema.ID,
93}) {}
94 
95export class OperationUnavailableError extends Schema.TaggedErrorClass<OperationUnavailableError>()(
96  "Session.OperationUnavailableError",
97  {
98    operation: Schema.Literals(["move", "shell", "skill", "switchAgent", "compact", "wait"]),
99  },
100) {}
101 
102export { ContextSnapshotDecodeError, MessageDecodeError } from "./session/error"
103 
104export class PromptConflictError extends Schema.TaggedErrorClass<PromptConflictError>()("Session.PromptConflictError", {
105  sessionID: SessionSchema.ID,
106  messageID: SessionMessage.ID,
107}) {}
108export const MessageNotFoundError = SessionRevert.MessageNotFoundError
109export type MessageNotFoundError = SessionRevert.MessageNotFoundError
110 
111export type Error = NotFoundError | MessageDecodeError | OperationUnavailableError | PromptConflictError
112 
113export interface Interface {
114  readonly list: (input?: ListInput) => Effect.Effect<SessionSchema.Info[]>
115  readonly create: (input: CreateInput) => Effect.Effect<SessionSchema.Info>
116  readonly get: (sessionID: SessionSchema.ID) => Effect.Effect<SessionSchema.Info, NotFoundError>
117  readonly messages: (input: {
118    sessionID: SessionSchema.ID
119    limit?: number
120    order?: "asc" | "desc"
121    cursor?: {
122      id: SessionMessage.ID
123      direction: "previous" | "next"
124    }
125  }) => Effect.Effect<SessionMessage.Message[], NotFoundError | MessageDecodeError>
126  readonly message: (input: {
127    sessionID: SessionSchema.ID
128    messageID: SessionMessage.ID
129  }) => Effect.Effect<SessionMessage.Message | undefined>
130  readonly context: (
131    sessionID: SessionSchema.ID,
132  ) => Effect.Effect<SessionMessage.Message[], NotFoundError | MessageDecodeError>
133  readonly events: (input: {
134    sessionID: SessionSchema.ID
135    after?: number
136  }) => Stream.Stream<SessionEvent.DurableEvent, NotFoundError>
137  readonly history: (input: {
138    sessionID: SessionSchema.ID
139    after?: number
140    limit: number
141  }) => Effect.Effect<{ events: ReadonlyArray<SessionEvent.DurableEvent>; hasMore: boolean }, NotFoundError>
142  readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect<void, NotFoundError>
143  readonly switchModel: (input: {
144    sessionID: SessionSchema.ID
145    model: ModelV2.Ref
146  }) => Effect.Effect<void, NotFoundError>
147  readonly prompt: (input: {
148    id?: SessionMessage.ID
149    sessionID: SessionSchema.ID
150    prompt: PromptInput.Prompt
151    delivery?: SessionInput.Delivery
152    resume?: boolean
153  }) => Effect.Effect<SessionInput.Admitted, NotFoundError | PromptConflictError>
154  readonly shell: (input: {
155    id?: EventV2.ID
156    sessionID: SessionSchema.ID
157    command: string
158    resume?: boolean
159  }) => Effect.Effect<void, OperationUnavailableError>
160  readonly skill: (input: {
161    id?: EventV2.ID
162    sessionID: SessionSchema.ID
163    skill: string
164    resume?: boolean
165  }) => Effect.Effect<void, OperationUnavailableError>
166  readonly compact: (input: CompactInput) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
167  readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
168  readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
169  readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
170  readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
171  readonly revert: {
172    readonly stage: (input: {
173      sessionID: SessionSchema.ID
174      messageID: SessionMessage.ID
175      files?: boolean
176    }) => Effect.Effect<Revert.State, NotFoundError | MessageNotFoundError | Snapshot.Error>
177    readonly clear: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | Snapshot.Error>
178    readonly commit: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
179  }
180}
181 
182export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Session") {}
183 
184const layer = Layer.effect(
185  Service,
186  Effect.gen(function* () {
187    const database = yield* Database.Service
188    const db = database.db
189    const events = yield* EventV2.Service
190    const projects = yield* ProjectV2.Service
191    const execution = yield* SessionExecution.Service
192    const store = yield* SessionStore.Service
193    const locations = yield* LocationServiceMap.Service
194    const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
195    const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
196    const decode = (row: typeof SessionMessageTable.$inferSelect) =>
197      decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
198        Effect.mapError(
199          () =>
200            new MessageDecodeError({
201              sessionID: SessionSchema.ID.make(row.session_id),
202              messageID: SessionMessage.ID.make(row.id),
203            }),
204        ),
205      )
206 
207    const result = Service.of({
208      create: Effect.fn("V2Session.create")(function* (input) {
209        const sessionID = input.id ?? SessionSchema.ID.create()
210        const recorded = yield* store.get(sessionID)
211        if (recorded) return recorded
212        const project = yield* projects.resolve(input.location.directory)
213        yield* db
214          .insert(ProjectTable)
215          .values({ id: project.id, worktree: project.directory, vcs: project.vcs?.type, sandboxes: [] })
216          .onConflictDoNothing()
217          .run()
218          .pipe(Effect.orDie)
219        const now = Date.now()
220        const info = SessionV1.SessionInfo.make({
221          id: sessionID,
222          slug: Slug.create(),
223          version: InstallationVersion,
224          projectID: project.id,
225          directory: input.location.directory,
226          path: path.relative(project.directory, input.location.directory).replaceAll("\\", "/"),
227          workspaceID: input.location.workspaceID ? WorkspaceV2.ID.make(input.location.workspaceID) : undefined,
228          title: `New session - ${new Date(now).toISOString()}`,
229          agent: input.agent,
230          model: input.model
231            ? {
232                id: ModelV2.ID.make(input.model.id),
233                providerID: input.model.providerID,
234                variant: input.model.variant,
235              }
236            : undefined,
237          cost: 0,
238          tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
239          time: { created: now, updated: now },
240        })
241        const projected = yield* events
242          .publish(SessionV1.Event.Created, { sessionID, info }, { location: input.location })
243          .pipe(
244            Effect.as({ type: "created" } as const),
245            Effect.catchDefect((defect) => {
246              if (!(defect instanceof SessionProjector.SessionAlreadyProjected)) {
247                return Effect.die(defect)
248              }
249              // Concurrent creation lost the projection race. The existing Session identity wins.
250              return store
251                .get(sessionID)
252                .pipe(
253                  Effect.flatMap((session) =>
254                    session ? Effect.succeed({ type: "existing", session } as const) : Effect.die(defect),
255                  ),
256                )
257            }),
258          )
259        if (projected.type === "existing") return projected.session
260        // TODO: Restore recorded sessions onto replacement synchronized workspaces in a future API slice.
261        return yield* result.get(sessionID).pipe(Effect.orDie)
262      }),
263      get: Effect.fn("V2Session.get")(function* (sessionID) {
264        const session = yield* store.get(sessionID)
265        if (!session) return yield* new NotFoundError({ sessionID })
266        return session
267      }),
268      list: Effect.fn("V2Session.list")(function* (input = {}) {
269        const direction = input.anchor?.direction ?? "next"
270        const requestedOrder = input.order ?? "desc"
271        const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder
272        const sortColumn = SessionTable.time_created
273        const conditions: SQL[] = []
274        if ("directory" in input) conditions.push(eq(SessionTable.directory, input.directory))
275        if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID))
276        if ("project" in input) conditions.push(eq(SessionTable.project_id, input.project))
277        if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`))
278        if (input.anchor) {
279          conditions.push(
280            order === "asc"
281              ? or(
282                  gt(sortColumn, input.anchor.time),
283                  and(eq(sortColumn, input.anchor.time), gt(SessionTable.id, input.anchor.id)),
284                )!
285              : or(
286                  lt(sortColumn, input.anchor.time),
287                  and(eq(sortColumn, input.anchor.time), lt(SessionTable.id, input.anchor.id)),
288                )!,
289          )
290        }
291        const query = db
292          .select()
293          .from(SessionTable)
294          .where(conditions.length > 0 ? and(...conditions) : undefined)
295          .orderBy(
296            order === "asc" ? asc(sortColumn) : desc(sortColumn),
297            order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id),
298          )
299        const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
300          Effect.orDie,
301        )
302        return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row))
303      }),
304      messages: Effect.fn("V2Session.messages")(function* (input) {
305        yield* result.get(input.sessionID)
306        const direction = input.cursor?.direction ?? "next"
307        const requestedOrder = input.order ?? "desc"
308        const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder
309        const anchor = input.cursor
310          ? yield* db
311              .select({ seq: SessionMessageTable.seq })
312              .from(SessionMessageTable)
313              .where(
314                and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.cursor.id)),
315              )
316              .get()
317              .pipe(Effect.orDie)
318          : undefined
319        if (input.cursor && !anchor) return []
320        const boundary = anchor
321          ? order === "asc"
322            ? gt(SessionMessageTable.seq, anchor.seq)
323            : lt(SessionMessageTable.seq, anchor.seq)
324          : undefined
325        const where = boundary
326          ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary)
327          : eq(SessionMessageTable.session_id, input.sessionID)
328        const query = db
329          .select()
330          .from(SessionMessageTable)
331          .where(where)
332          .orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq))
333        const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
334          Effect.orDie,
335        )
336        return yield* Effect.forEach(direction === "previous" ? rows.toReversed() : rows, decode)
337      }),
338      message: Effect.fn("V2Session.message")(function* (input) {
339        const stored = yield* store.message(input.messageID)
340        return stored?.sessionID === input.sessionID ? stored.message : undefined
341      }),
342      context: Effect.fn("V2Session.context")(function* (sessionID) {
343        yield* result.get(sessionID)
344        return yield* store.context(sessionID)
345      }),
346      events: (input) =>
347        Stream.unwrap(
348          result
349            .get(input.sessionID)
350            .pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
351        ).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))),
352      history: Effect.fn("V2Session.history")(function* (input) {
353        yield* result.get(input.sessionID)
354        return yield* EventV2.readAggregate(db, {
355          ...input,
356          aggregateID: input.sessionID,
357          manifest: SessionDurable,
358        })
359      }),
360      prompt: Effect.fn("V2Session.prompt")((input) =>
361        Effect.uninterruptible(
362          Effect.gen(function* () {
363            yield* result.get(input.sessionID)
364            const prompt = resolvePrompt(input.prompt)
365            const messageID = input.id ?? SessionMessage.ID.create()
366            const delivery = input.delivery ?? "steer"
367            const expected = { sessionID: input.sessionID, messageID, prompt, delivery }
368            const admitted = yield* SessionInput.admit(db, events, {
369              id: messageID,
370              sessionID: input.sessionID,
371              prompt,
372              delivery,
373            }).pipe(
374              Effect.catchDefect((defect) =>
375                defect instanceof SessionInput.LifecycleConflict
376                  ? new PromptConflictError({ sessionID: input.sessionID, messageID })
377                  : Effect.die(defect),
378              ),
379            )
380            if (!SessionInput.equivalent(admitted, expected))
381              return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
382            if (input.resume !== false) yield* execution.wake(admitted.sessionID)
383            return admitted
384          }),
385        ),
386      ),
387      shell: Effect.fn("V2Session.shell")(function* () {
388        return yield* new OperationUnavailableError({ operation: "shell" })
389      }),
390      skill: Effect.fn("V2Session.skill")(function* () {
391        return yield* new OperationUnavailableError({ operation: "skill" })
392      }),
393      switchAgent: Effect.fn("V2Session.switchAgent")(function* (input) {
394        yield* result.get(input.sessionID)
395        yield* events.publish(SessionEvent.AgentSwitched, {
396          sessionID: input.sessionID,
397          messageID: SessionMessage.ID.create(),
398          timestamp: yield* DateTime.now,
399          agent: input.agent,
400        })
401      }),
402      switchModel: Effect.fn("V2Session.switchModel")(function* (input) {
403        const session = yield* result.get(input.sessionID)
404        if (
405          session.model?.providerID === input.model.providerID &&
406          session.model.id === input.model.id &&
407          (session.model.variant ?? "default") === (input.model.variant ?? "default")
408        )
409          return
410        yield* events.publish(SessionEvent.ModelSwitched, {
411          sessionID: input.sessionID,
412          messageID: SessionMessage.ID.create(),
413          timestamp: yield* DateTime.now,
414          model: input.model,
415        })
416      }),
417      compact: Effect.fn("V2Session.compact")(function* (input) {
418        yield* result.get(input.sessionID)
419        return yield* new OperationUnavailableError({ operation: "compact" })
420      }),
421      wait: Effect.fn("V2Session.wait")(function* (sessionID) {
422        yield* result.get(sessionID)
423        return yield* new OperationUnavailableError({ operation: "wait" })
424      }),
425      active: execution.active,
426      resume: Effect.fn("V2Session.resume")(function* (sessionID) {
427        yield* result.get(sessionID)
428        yield* execution.resume(sessionID)
429      }),
430      interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
431        Effect.uninterruptible(execution.interrupt(sessionID)),
432      ),
433      revert: {
434        stage: Effect.fn("V2Session.revert.stage")(function* (input) {
435          const session = yield* result.get(input.sessionID)
436          return yield* SessionRevert.stage({ session, messageID: input.messageID, files: input.files }).pipe(
437            Effect.provideService(Database.Service, database),
438            Effect.provideService(EventV2.Service, events),
439            Effect.provide(locations.get(session.location)),
440          )
441        }),
442        clear: Effect.fn("V2Session.revert.clear")(function* (sessionID) {
443          const session = yield* result.get(sessionID)
444          yield* SessionRevert.clear(session).pipe(
445            Effect.provideService(EventV2.Service, events),
446            Effect.provide(locations.get(session.location)),
447          )
448        }),
449        commit: Effect.fn("V2Session.revert.commit")(function* (sessionID) {
450          const session = yield* result.get(sessionID)
451          yield* SessionRevert.commit(session).pipe(Effect.provideService(EventV2.Service, events))
452        }),
453      },
454    })
455 
456    return result
457  }),
458)
459 
460const resolvePrompt = (input: PromptInput.Prompt) =>
461  Prompt.make({
462    text: input.text,
463    agents: input.agents,
464    files: input.files?.map((file) => {
465      const dataMime = file.uri.match(/^data:([^;,]+)[;,]/i)?.[1]
466      const target = URL.canParse(file.uri) ? new URL(file.uri).pathname : (file.name ?? file.uri)
467      return {
468        ...file,
469        mime: dataMime ?? (target.endsWith("/") ? "application/x-directory" : FSUtil.mimeType(target)),
470      }
471    }),
472  })
473 
474export const node = makeGlobalNode({
475  service: Service,
476  layer: layer.pipe(Layer.orDie),
477  deps: [
478    Database.node,
479    EventV2.node,
480    ProjectV2.node,
481    SessionExecution.node,
482    SessionStore.node,
483    LocationServiceMap.node,
484    SessionProjector.node,
485  ],
486})
487