merge dev

packages/opencode/src/session/compaction.ts

@@ -1,24 +1,22 @@
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
-import { Session } from "."
+import * as Session from "./session"
 import { SessionID, MessageID, PartID } from "./schema"
-import { Instance } from "../project/instance"
-import { Provider } from "../provider/provider"
-import { ProviderTransform } from "../provider/transform"
+import { Provider, ProviderTransform } from "../provider"
 import { MessageV2 } from "./message-v2"
 import z from "zod"
-import { Token } from "../util/token"
-import { Log } from "../util/log"
+import { Token } from "../util"
+import { Log } from "../util"
 import { SessionProcessor } from "./processor"
-import { fn } from "@/util/fn"
 import { Agent } from "@/agent/agent"
 import { Plugin } from "@/plugin"
-import { Config } from "@/config/config"
-import { NotFoundError } from "@/storage/db"
+import { Config } from "@/config"
+import { NotFoundError } from "@/storage"
 import { ModelID, ProviderID } from "@/provider/schema"
-import { Effect, Layer, ServiceMap } from "effect"
+import { Effect, Layer, Context } from "effect"
+import { InstanceState } from "@/effect"
 import { makeRuntime } from "@/effect/run-service"
-import { InstanceState } from "@/effect/instance-state"
+import { fn } from "@/util/fn"
 import { isOverflow as overflow } from "./overflow"
 
 export namespace SessionCompaction {
@@ -99,7 +97,7 @@ export namespace SessionCompaction {
     }) => Effect.Effect<void>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionCompaction") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionCompaction") {}
 
   export const layer: Layer.Layer<
     Service,
@@ -182,7 +180,7 @@ export namespace SessionCompaction {
       // calls, then erases output of older tool calls to free context space
       const prune = Effect.fn("SessionCompaction.prune")(function* (input: { sessionID: SessionID }) {
         const cfg = yield* config.get()
-        if (cfg.compaction?.prune !== true) return
+        if (cfg.compaction?.prune === false) return
         log.info("pruning")
 
         const msgs = yield* session
@@ -416,31 +414,55 @@ When constructing the summary, try to stick to this template:
           }
 
           if (!replay) {
-            const continueMsg = yield* session.updateMessage({
-              id: MessageID.ascending(),
-              role: "user",
-              sessionID: input.sessionID,
-              time: { created: Date.now() },
-              agent: userMessage.agent,
-              model: userMessage.model,
-            })
-            const text =
-              (input.overflow
-                ? "The previous request exceeded the provider's size limit due to large media attachments. The conversation was compacted and media files were removed from context. If the user was asking about attached images or files, explain that the attachments were too large to process and suggest they try again with smaller or fewer files.\n\n"
-                : "") +
-              "Continue if you have next steps, or stop and ask for clarification if you are unsure how to proceed."
-            yield* session.updatePart({
-              id: PartID.ascending(),
-              messageID: continueMsg.id,
-              sessionID: input.sessionID,
-              type: "text",
-              synthetic: true,
-              text,
-              time: {
-                start: Date.now(),
-                end: Date.now(),
-              },
-            })
+            const info = yield* provider.getProvider(userMessage.model.providerID)
+            if (
+              (yield* plugin.trigger(
+                "experimental.compaction.autocontinue",
+                {
+                  sessionID: input.sessionID,
+                  agent: userMessage.agent,
+                  model: yield* provider.getModel(userMessage.model.providerID, userMessage.model.modelID),
+                  provider: {
+                    source: info.source,
+                    info,
+                    options: info.options,
+                  },
+                  message: userMessage,
+                  overflow: input.overflow === true,
+                },
+                { enabled: true },
+              )).enabled
+            ) {
+              const continueMsg = yield* session.updateMessage({
+                id: MessageID.ascending(),
+                role: "user",
+                sessionID: input.sessionID,
+                time: { created: Date.now() },
+                agent: userMessage.agent,
+                model: userMessage.model,
+              })
+              const text =
+                (input.overflow
+                  ? "The previous request exceeded the provider's size limit due to large media attachments. The conversation was compacted and media files were removed from context. If the user was asking about attached images or files, explain that the attachments were too large to process and suggest they try again with smaller or fewer files.\n\n"
+                  : "") +
+                "Continue if you have next steps, or stop and ask for clarification if you are unsure how to proceed."
+              yield* session.updatePart({
+                id: PartID.ascending(),
+                messageID: continueMsg.id,
+                sessionID: input.sessionID,
+                type: "text",
+                // Internal marker for auto-compaction followups so provider plugins
+                // can distinguish them from manual post-compaction user prompts.
+                // This is not a stable plugin contract and may change or disappear.
+                metadata: { compaction_continue: true },
+                synthetic: true,
+                text,
+                time: {
+                  start: Date.now(),
+                  end: Date.now(),
+                },
+              })
+            }
           }
         }
 

packages/opencode/src/session/index.ts

@@ -1,837 +1 @@
-import { Slug } from "@opencode-ai/util/slug"
-import path from "path"
-import { BusEvent } from "@/bus/bus-event"
-import { Bus } from "@/bus"
-import { Decimal } from "decimal.js"
-import z from "zod"
-import { type ProviderMetadata } from "ai"
-import { Flag } from "../flag/flag"
-import { Installation } from "../installation"
-
-import { Database, NotFoundError, eq, and, gte, isNull, desc, like, inArray, lt } from "../storage/db"
-import { SyncEvent } from "../sync"
-import type { SQL } from "../storage/db"
-import { PartTable, SessionTable } from "./session.sql"
-import { ProjectTable } from "../project/project.sql"
-import { Storage } from "@/storage/storage"
-import { Log } from "../util/log"
-import { updateSchema } from "../util/update-schema"
-import { MessageV2 } from "./message-v2"
-import { Instance } from "../project/instance"
-import { InstanceState } from "@/effect/instance-state"
-import { fn } from "@/util/fn"
-import { Snapshot } from "@/snapshot"
-import { ProjectID } from "../project/schema"
-import { WorkspaceID } from "../control-plane/schema"
-import { SessionID, MessageID, PartID } from "./schema"
-
-import type { Provider } from "@/provider/provider"
-import { Permission } from "@/permission"
-import { Global } from "@/global"
-import type { LanguageModelV2Usage } from "@ai-sdk/provider"
-import { Effect, Layer, ServiceMap } from "effect"
-import { makeRuntime } from "@/effect/run-service"
-
-export namespace Session {
-  const log = Log.create({ service: "session" })
-
-  const parentTitlePrefix = "New session - "
-  const childTitlePrefix = "Child session - "
-
-  function createDefaultTitle(isChild = false) {
-    return (isChild ? childTitlePrefix : parentTitlePrefix) + new Date().toISOString()
-  }
-
-  export function isDefaultTitle(title: string) {
-    return new RegExp(
-      `^(${parentTitlePrefix}|${childTitlePrefix})\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z$`,
-    ).test(title)
-  }
-
-  type SessionRow = typeof SessionTable.$inferSelect
-
-  export function fromRow(row: SessionRow): Info {
-    const summary =
-      row.summary_additions !== null || row.summary_deletions !== null || row.summary_files !== null
-        ? {
-            additions: row.summary_additions ?? 0,
-            deletions: row.summary_deletions ?? 0,
-            files: row.summary_files ?? 0,
-            diffs: row.summary_diffs ?? undefined,
-          }
-        : undefined
-    const share = row.share_url ? { url: row.share_url } : undefined
-    const revert = row.revert ?? undefined
-    return {
-      id: row.id,
-      slug: row.slug,
-      projectID: row.project_id,
-      workspaceID: row.workspace_id ?? undefined,
-      directory: row.directory,
-      parentID: row.parent_id ?? undefined,
-      title: row.title,
-      version: row.version,
-      summary,
-      share,
-      revert,
-      permission: row.permission ?? undefined,
-      time: {
-        created: row.time_created,
-        updated: row.time_updated,
-        compacting: row.time_compacting ?? undefined,
-        archived: row.time_archived ?? undefined,
-      },
-    }
-  }
-
-  export function toRow(info: Info) {
-    return {
-      id: info.id,
-      project_id: info.projectID,
-      workspace_id: info.workspaceID,
-      parent_id: info.parentID,
-      slug: info.slug,
-      directory: info.directory,
-      title: info.title,
-      version: info.version,
-      share_url: info.share?.url,
-      summary_additions: info.summary?.additions,
-      summary_deletions: info.summary?.deletions,
-      summary_files: info.summary?.files,
-      summary_diffs: info.summary?.diffs,
-      revert: info.revert ?? null,
-      permission: info.permission,
-      time_created: info.time.created,
-      time_updated: info.time.updated,
-      time_compacting: info.time.compacting,
-      time_archived: info.time.archived,
-    }
-  }
-
-  function getForkedTitle(title: string): string {
-    const match = title.match(/^(.+) \(fork #(\d+)\)$/)
-    if (match) {
-      const base = match[1]
-      const num = parseInt(match[2], 10)
-      return `${base} (fork #${num + 1})`
-    }
-    return `${title} (fork #1)`
-  }
-
-  export const Info = z
-    .object({
-      id: SessionID.zod,
-      slug: z.string(),
-      projectID: ProjectID.zod,
-      workspaceID: WorkspaceID.zod.optional(),
-      directory: z.string(),
-      parentID: SessionID.zod.optional(),
-      summary: z
-        .object({
-          additions: z.number(),
-          deletions: z.number(),
-          files: z.number(),
-          diffs: Snapshot.FileDiff.array().optional(),
-        })
-        .optional(),
-      share: z
-        .object({
-          url: z.string(),
-        })
-        .optional(),
-      title: z.string(),
-      version: z.string(),
-      time: z.object({
-        created: z.number(),
-        updated: z.number(),
-        compacting: z.number().optional(),
-        archived: z.number().optional(),
-      }),
-      permission: Permission.Ruleset.optional(),
-      revert: z
-        .object({
-          messageID: MessageID.zod,
-          partID: PartID.zod.optional(),
-          snapshot: z.string().optional(),
-          diff: z.string().optional(),
-        })
-        .optional(),
-    })
-    .meta({
-      ref: "Session",
-    })
-  export type Info = z.output<typeof Info>
-
-  export const ProjectInfo = z
-    .object({
-      id: ProjectID.zod,
-      name: z.string().optional(),
-      worktree: z.string(),
-    })
-    .meta({
-      ref: "ProjectSummary",
-    })
-  export type ProjectInfo = z.output<typeof ProjectInfo>
-
-  export const GlobalInfo = Info.extend({
-    project: ProjectInfo.nullable(),
-  }).meta({
-    ref: "GlobalSession",
-  })
-  export type GlobalInfo = z.output<typeof GlobalInfo>
-
-  export const Event = {
-    Created: SyncEvent.define({
-      type: "session.created",
-      version: 1,
-      aggregate: "sessionID",
-      schema: z.object({
-        sessionID: SessionID.zod,
-        info: Info,
-      }),
-    }),
-    Updated: SyncEvent.define({
-      type: "session.updated",
-      version: 1,
-      aggregate: "sessionID",
-      schema: z.object({
-        sessionID: SessionID.zod,
-        info: updateSchema(Info).extend({
-          share: updateSchema(Info.shape.share.unwrap()).optional(),
-          time: updateSchema(Info.shape.time).optional(),
-        }),
-      }),
-      busSchema: z.object({
-        sessionID: SessionID.zod,
-        info: Info,
-      }),
-    }),
-    Deleted: SyncEvent.define({
-      type: "session.deleted",
-      version: 1,
-      aggregate: "sessionID",
-      schema: z.object({
-        sessionID: SessionID.zod,
-        info: Info,
-      }),
-    }),
-    Diff: BusEvent.define(
-      "session.diff",
-      z.object({
-        sessionID: SessionID.zod,
-        diff: Snapshot.FileDiff.array(),
-      }),
-    ),
-    Error: BusEvent.define(
-      "session.error",
-      z.object({
-        sessionID: SessionID.zod.optional(),
-        error: MessageV2.Assistant.shape.error,
-      }),
-    ),
-  }
-
-  export function plan(input: { slug: string; time: { created: number } }) {
-    const base = Instance.project.vcs
-      ? path.join(Instance.worktree, ".opencode", "plans")
-      : path.join(Global.Path.data, "plans")
-    return path.join(base, [input.time.created, input.slug].join("-") + ".md")
-  }
-
-  export const getUsage = (input: {
-    model: Provider.Model
-    usage: LanguageModelV2Usage
-    metadata?: ProviderMetadata
-  }) => {
-    const safe = (value: number) => {
-      if (!Number.isFinite(value)) return 0
-      return value
-    }
-    const inputTokens = safe(input.usage.inputTokens ?? 0)
-    const outputTokens = safe(input.usage.outputTokens ?? 0)
-    const reasoningTokens = safe(input.usage.reasoningTokens ?? 0)
-
-    const cacheReadInputTokens = safe(input.usage.cachedInputTokens ?? 0)
-    const cacheWriteInputTokens = safe(
-      (input.metadata?.["anthropic"]?.["cacheCreationInputTokens"] ??
-        // google-vertex-anthropic returns metadata under "vertex" key
-        // (AnthropicMessagesLanguageModel custom provider key from 'vertex.anthropic.messages')
-        input.metadata?.["vertex"]?.["cacheCreationInputTokens"] ??
-        // @ts-expect-error
-        input.metadata?.["bedrock"]?.["usage"]?.["cacheWriteInputTokens"] ??
-        // @ts-expect-error
-        input.metadata?.["venice"]?.["usage"]?.["cacheCreationInputTokens"] ??
-        0) as number,
-    )
-
-    // AI SDK v6 normalized inputTokens to include cached tokens across all providers
-    // (including Anthropic/Bedrock which previously excluded them). Always subtract cache
-    // tokens to get the non-cached input count for separate cost calculation.
-    const adjustedInputTokens = safe(inputTokens - cacheReadInputTokens - cacheWriteInputTokens)
-
-    const total = input.usage.totalTokens
-
-    const tokens = {
-      total,
-      input: adjustedInputTokens,
-      output: outputTokens - reasoningTokens,
-      reasoning: reasoningTokens,
-      cache: {
-        write: cacheWriteInputTokens,
-        read: cacheReadInputTokens,
-      },
-    }
-
-    const costInfo =
-      input.model.cost?.experimentalOver200K && tokens.input + tokens.cache.read > 200_000
-        ? input.model.cost.experimentalOver200K
-        : input.model.cost
-    return {
-      cost: safe(
-        new Decimal(0)
-          .add(new Decimal(tokens.input).mul(costInfo?.input ?? 0).div(1_000_000))
-          .add(new Decimal(tokens.output).mul(costInfo?.output ?? 0).div(1_000_000))
-          .add(new Decimal(tokens.cache.read).mul(costInfo?.cache?.read ?? 0).div(1_000_000))
-          .add(new Decimal(tokens.cache.write).mul(costInfo?.cache?.write ?? 0).div(1_000_000))
-          // TODO: update models.dev to have better pricing model, for now:
-          // charge reasoning tokens at the same rate as output tokens
-          .add(new Decimal(tokens.reasoning).mul(costInfo?.output ?? 0).div(1_000_000))
-          .toNumber(),
-      ),
-      tokens,
-    }
-  }
-
-  export class BusyError extends Error {
-    constructor(public readonly sessionID: string) {
-      super(`Session ${sessionID} is busy`)
-    }
-  }
-
-  export interface Interface {
-    readonly create: (input?: {
-      parentID?: SessionID
-      title?: string
-      permission?: Permission.Ruleset
-      workspaceID?: WorkspaceID
-    }) => Effect.Effect<Info>
-    readonly fork: (input: { sessionID: SessionID; messageID?: MessageID }) => Effect.Effect<Info>
-    readonly touch: (sessionID: SessionID) => Effect.Effect<void>
-    readonly get: (id: SessionID) => Effect.Effect<Info>
-    readonly setTitle: (input: { sessionID: SessionID; title: string }) => Effect.Effect<void>
-    readonly setArchived: (input: { sessionID: SessionID; time?: number }) => Effect.Effect<void>
-    readonly setPermission: (input: { sessionID: SessionID; permission: Permission.Ruleset }) => Effect.Effect<void>
-    readonly setRevert: (input: {
-      sessionID: SessionID
-      revert: Info["revert"]
-      summary: Info["summary"]
-    }) => Effect.Effect<void>
-    readonly clearRevert: (sessionID: SessionID) => Effect.Effect<void>
-    readonly setSummary: (input: { sessionID: SessionID; summary: Info["summary"] }) => Effect.Effect<void>
-    readonly diff: (sessionID: SessionID) => Effect.Effect<Snapshot.FileDiff[]>
-    readonly messages: (input: { sessionID: SessionID; limit?: number }) => Effect.Effect<MessageV2.WithParts[]>
-    readonly children: (parentID: SessionID) => Effect.Effect<Info[]>
-    readonly remove: (sessionID: SessionID) => Effect.Effect<void>
-    readonly updateMessage: <T extends MessageV2.Info>(msg: T) => Effect.Effect<T>
-    readonly removeMessage: (input: { sessionID: SessionID; messageID: MessageID }) => Effect.Effect<MessageID>
-    readonly removePart: (input: {
-      sessionID: SessionID
-      messageID: MessageID
-      partID: PartID
-    }) => Effect.Effect<PartID>
-    readonly getPart: (input: {
-      sessionID: SessionID
-      messageID: MessageID
-      partID: PartID
-    }) => Effect.Effect<MessageV2.Part | undefined>
-    readonly updatePart: <T extends MessageV2.Part>(part: T) => Effect.Effect<T>
-    readonly updatePartDelta: (input: {
-      sessionID: SessionID
-      messageID: MessageID
-      partID: PartID
-      field: string
-      delta: string
-    }) => Effect.Effect<void>
-  }
-
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Session") {}
-
-  type Patch = z.infer<typeof Event.Updated.schema>["info"]
-
-  const db = <T>(fn: (d: Parameters<typeof Database.use>[0] extends (trx: infer D) => any ? D : never) => T) =>
-    Effect.sync(() => Database.use(fn))
-
-  export const layer: Layer.Layer<Service, never, Bus.Service> = Layer.effect(
-    Service,
-    Effect.gen(function* () {
-      const bus = yield* Bus.Service
-
-      const createNext = Effect.fn("Session.createNext")(function* (input: {
-        id?: SessionID
-        title?: string
-        parentID?: SessionID
-        workspaceID?: WorkspaceID
-        directory: string
-        permission?: Permission.Ruleset
-      }) {
-        const ctx = yield* InstanceState.context
-        const result: Info = {
-          id: SessionID.descending(input.id),
-          slug: Slug.create(),
-          version: Installation.VERSION,
-          projectID: ctx.project.id,
-          directory: input.directory,
-          workspaceID: input.workspaceID,
-          parentID: input.parentID,
-          title: input.title ?? createDefaultTitle(!!input.parentID),
-          permission: input.permission,
-          time: {
-            created: Date.now(),
-            updated: Date.now(),
-          },
-        }
-        log.info("created", result)
-
-        yield* Effect.sync(() => SyncEvent.run(Event.Created, { sessionID: result.id, info: result }))
-
-        if (!Flag.OPENCODE_EXPERIMENTAL_WORKSPACES) {
-          // This only exist for backwards compatibility. We should not be
-          // manually publishing this event; it is a sync event now
-          yield* bus.publish(Event.Updated, {
-            sessionID: result.id,
-            info: result,
-          })
-        }
-
-        return result
-      })
-
-      const get = Effect.fn("Session.get")(function* (id: SessionID) {
-        const row = yield* db((d) => d.select().from(SessionTable).where(eq(SessionTable.id, id)).get())
-        if (!row) throw new NotFoundError({ message: `Session not found: ${id}` })
-        return fromRow(row)
-      })
-
-      const children = Effect.fn("Session.children")(function* (parentID: SessionID) {
-        const ctx = yield* InstanceState.context
-        const rows = yield* db((d) =>
-          d
-            .select()
-            .from(SessionTable)
-            .where(and(eq(SessionTable.project_id, ctx.project.id), eq(SessionTable.parent_id, parentID)))
-            .all(),
-        )
-        return rows.map(fromRow)
-      })
-
-      const remove: (sessionID: SessionID) => Effect.Effect<void> = Effect.fnUntraced(function* (sessionID: SessionID) {
-        try {
-          const session = yield* get(sessionID)
-          const kids = yield* children(sessionID)
-          for (const child of kids) {
-            yield* remove(child.id)
-          }
-          yield* Effect.sync(() => {
-            SyncEvent.run(Event.Deleted, { sessionID, info: session })
-            SyncEvent.remove(sessionID)
-          })
-        } catch (e) {
-          log.error(e)
-        }
-      })
-
-      const updateMessage = <T extends MessageV2.Info>(msg: T): Effect.Effect<T> =>
-        Effect.gen(function* () {
-          yield* Effect.sync(() => SyncEvent.run(MessageV2.Event.Updated, { sessionID: msg.sessionID, info: msg }))
-          return msg
-        }).pipe(Effect.withSpan("Session.updateMessage"))
-
-      const updatePart = <T extends MessageV2.Part>(part: T): Effect.Effect<T> =>
-        Effect.gen(function* () {
-          yield* Effect.sync(() =>
-            SyncEvent.run(MessageV2.Event.PartUpdated, {
-              sessionID: part.sessionID,
-              part: structuredClone(part),
-              time: Date.now(),
-            }),
-          )
-          return part
-        }).pipe(Effect.withSpan("Session.updatePart"))
-
-      const getPart: Interface["getPart"] = Effect.fn("Session.getPart")(function* (input) {
-        const row = Database.use((db) =>
-          db
-            .select()
-            .from(PartTable)
-            .where(
-              and(
-                eq(PartTable.session_id, input.sessionID),
-                eq(PartTable.message_id, input.messageID),
-                eq(PartTable.id, input.partID),
-              ),
-            )
-            .get(),
-        )
-        if (!row) return
-        return {
-          ...row.data,
-          id: row.id,
-          sessionID: row.session_id,
-          messageID: row.message_id,
-        } as MessageV2.Part
-      })
-
-      const create = Effect.fn("Session.create")(function* (input?: {
-        parentID?: SessionID
-        title?: string
-        permission?: Permission.Ruleset
-        workspaceID?: WorkspaceID
-      }) {
-        const directory = yield* InstanceState.directory
-        return yield* createNext({
-          parentID: input?.parentID,
-          directory,
-          title: input?.title,
-          permission: input?.permission,
-          workspaceID: input?.workspaceID,
-        })
-      })
-
-      const fork = Effect.fn("Session.fork")(function* (input: { sessionID: SessionID; messageID?: MessageID }) {
-        const directory = yield* InstanceState.directory
-        const original = yield* get(input.sessionID)
-        const title = getForkedTitle(original.title)
-        const session = yield* createNext({
-          directory,
-          workspaceID: original.workspaceID,
-          title,
-        })
-        const msgs = yield* messages({ sessionID: input.sessionID })
-        const idMap = new Map<string, MessageID>()
-
-        for (const msg of msgs) {
-          if (input.messageID && msg.info.id >= input.messageID) break
-          const newID = MessageID.ascending()
-          idMap.set(msg.info.id, newID)
-
-          const parentID = msg.info.role === "assistant" && msg.info.parentID ? idMap.get(msg.info.parentID) : undefined
-          const cloned = yield* updateMessage({
-            ...msg.info,
-            sessionID: session.id,
-            id: newID,
-            ...(parentID && { parentID }),
-          })
-
-          for (const part of msg.parts) {
-            yield* updatePart({
-              ...part,
-              id: PartID.ascending(),
-              messageID: cloned.id,
-              sessionID: session.id,
-            })
-          }
-        }
-        return session
-      })
-
-      const patch = (sessionID: SessionID, info: Patch) =>
-        Effect.sync(() => SyncEvent.run(Event.Updated, { sessionID, info }))
-
-      const touch = Effect.fn("Session.touch")(function* (sessionID: SessionID) {
-        yield* patch(sessionID, { time: { updated: Date.now() } })
-      })
-
-      const setTitle = Effect.fn("Session.setTitle")(function* (input: { sessionID: SessionID; title: string }) {
-        yield* patch(input.sessionID, { title: input.title })
-      })
-
-      const setArchived = Effect.fn("Session.setArchived")(function* (input: { sessionID: SessionID; time?: number }) {
-        yield* patch(input.sessionID, { time: { archived: input.time } })
-      })
-
-      const setPermission = Effect.fn("Session.setPermission")(function* (input: {
-        sessionID: SessionID
-        permission: Permission.Ruleset
-      }) {
-        yield* patch(input.sessionID, { permission: input.permission, time: { updated: Date.now() } })
-      })
-
-      const setRevert = Effect.fn("Session.setRevert")(function* (input: {
-        sessionID: SessionID
-        revert: Info["revert"]
-        summary: Info["summary"]
-      }) {
-        yield* patch(input.sessionID, { summary: input.summary, time: { updated: Date.now() }, revert: input.revert })
-      })
-
-      const clearRevert = Effect.fn("Session.clearRevert")(function* (sessionID: SessionID) {
-        yield* patch(sessionID, { time: { updated: Date.now() }, revert: null })
-      })
-
-      const setSummary = Effect.fn("Session.setSummary")(function* (input: {
-        sessionID: SessionID
-        summary: Info["summary"]
-      }) {
-        yield* patch(input.sessionID, { time: { updated: Date.now() }, summary: input.summary })
-      })
-
-      const diff = Effect.fn("Session.diff")(function* (sessionID: SessionID) {
-        return yield* Effect.tryPromise(() => Storage.read<Snapshot.FileDiff[]>(["session_diff", sessionID])).pipe(
-          Effect.orElseSucceed((): Snapshot.FileDiff[] => []),
-        )
-      })
-
-      const messages = Effect.fn("Session.messages")(function* (input: { sessionID: SessionID; limit?: number }) {
-        if (input.limit) {
-          return MessageV2.page({ sessionID: input.sessionID, limit: input.limit }).items
-        }
-        return Array.from(MessageV2.stream(input.sessionID)).reverse()
-      })
-
-      const removeMessage = Effect.fn("Session.removeMessage")(function* (input: {
-        sessionID: SessionID
-        messageID: MessageID
-      }) {
-        yield* Effect.sync(() =>
-          SyncEvent.run(MessageV2.Event.Removed, {
-            sessionID: input.sessionID,
-            messageID: input.messageID,
-          }),
-        )
-        return input.messageID
-      })
-
-      const removePart = Effect.fn("Session.removePart")(function* (input: {
-        sessionID: SessionID
-        messageID: MessageID
-        partID: PartID
-      }) {
-        yield* Effect.sync(() =>
-          SyncEvent.run(MessageV2.Event.PartRemoved, {
-            sessionID: input.sessionID,
-            messageID: input.messageID,
-            partID: input.partID,
-          }),
-        )
-        return input.partID
-      })
-
-      const updatePartDelta = Effect.fn("Session.updatePartDelta")(function* (input: {
-        sessionID: SessionID
-        messageID: MessageID
-        partID: PartID
-        field: string
-        delta: string
-      }) {
-        yield* bus.publish(MessageV2.Event.PartDelta, input)
-      })
-
-      return Service.of({
-        create,
-        fork,
-        touch,
-        get,
-        setTitle,
-        setArchived,
-        setPermission,
-        setRevert,
-        clearRevert,
-        setSummary,
-        diff,
-        messages,
-        children,
-        remove,
-        updateMessage,
-        removeMessage,
-        removePart,
-        updatePart,
-        getPart,
-        updatePartDelta,
-      })
-    }),
-  )
-
-  export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
-
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
-  export const create = fn(
-    z
-      .object({
-        parentID: SessionID.zod.optional(),
-        title: z.string().optional(),
-        permission: Info.shape.permission,
-        workspaceID: WorkspaceID.zod.optional(),
-      })
-      .optional(),
-    (input) => runPromise((svc) => svc.create(input)),
-  )
-
-  export const fork = fn(z.object({ sessionID: SessionID.zod, messageID: MessageID.zod.optional() }), (input) =>
-    runPromise((svc) => svc.fork(input)),
-  )
-
-  export const get = fn(SessionID.zod, (id) => runPromise((svc) => svc.get(id)))
-
-  export const setTitle = fn(z.object({ sessionID: SessionID.zod, title: z.string() }), (input) =>
-    runPromise((svc) => svc.setTitle(input)),
-  )
-
-  export const setArchived = fn(z.object({ sessionID: SessionID.zod, time: z.number().optional() }), (input) =>
-    runPromise((svc) => svc.setArchived(input)),
-  )
-
-  export const setRevert = fn(
-    z.object({ sessionID: SessionID.zod, revert: Info.shape.revert, summary: Info.shape.summary }),
-    (input) =>
-      runPromise((svc) => svc.setRevert({ sessionID: input.sessionID, revert: input.revert, summary: input.summary })),
-  )
-
-  export const messages = fn(z.object({ sessionID: SessionID.zod, limit: z.number().optional() }), (input) =>
-    runPromise((svc) => svc.messages(input)),
-  )
-
-  export function* list(input?: {
-    directory?: string
-    workspaceID?: WorkspaceID
-    roots?: boolean
-    start?: number
-    search?: string
-    limit?: number
-  }) {
-    const project = Instance.project
-    const conditions = [eq(SessionTable.project_id, project.id)]
-
-    if (input?.workspaceID) {
-      conditions.push(eq(SessionTable.workspace_id, input.workspaceID))
-    }
-    if (input?.directory) {
-      conditions.push(eq(SessionTable.directory, input.directory))
-    }
-    if (input?.roots) {
-      conditions.push(isNull(SessionTable.parent_id))
-    }
-    if (input?.start) {
-      conditions.push(gte(SessionTable.time_updated, input.start))
-    }
-    if (input?.search) {
-      conditions.push(like(SessionTable.title, `%${input.search}%`))
-    }
-
-    const limit = input?.limit ?? 100
-
-    const rows = Database.use((db) =>
-      db
-        .select()
-        .from(SessionTable)
-        .where(and(...conditions))
-        .orderBy(desc(SessionTable.time_updated))
-        .limit(limit)
-        .all(),
-    )
-    for (const row of rows) {
-      yield fromRow(row)
-    }
-  }
-
-  export function* listGlobal(input?: {
-    directory?: string
-    roots?: boolean
-    start?: number
-    cursor?: number
-    search?: string
-    limit?: number
-    archived?: boolean
-  }) {
-    const conditions: SQL[] = []
-
-    if (input?.directory) {
-      conditions.push(eq(SessionTable.directory, input.directory))
-    }
-    if (input?.roots) {
-      conditions.push(isNull(SessionTable.parent_id))
-    }
-    if (input?.start) {
-      conditions.push(gte(SessionTable.time_updated, input.start))
-    }
-    if (input?.cursor) {
-      conditions.push(lt(SessionTable.time_updated, input.cursor))
-    }
-    if (input?.search) {
-      conditions.push(like(SessionTable.title, `%${input.search}%`))
-    }
-    if (!input?.archived) {
-      conditions.push(isNull(SessionTable.time_archived))
-    }
-
-    const limit = input?.limit ?? 100
-
-    const rows = Database.use((db) => {
-      const query =
-        conditions.length > 0
-          ? db
-              .select()
-              .from(SessionTable)
-              .where(and(...conditions))
-          : db.select().from(SessionTable)
-      return query.orderBy(desc(SessionTable.time_updated), desc(SessionTable.id)).limit(limit).all()
-    })
-
-    const ids = [...new Set(rows.map((row) => row.project_id))]
-    const projects = new Map<string, ProjectInfo>()
-
-    if (ids.length > 0) {
-      const items = Database.use((db) =>
-        db
-          .select({ id: ProjectTable.id, name: ProjectTable.name, worktree: ProjectTable.worktree })
-          .from(ProjectTable)
-          .where(inArray(ProjectTable.id, ids))
-          .all(),
-      )
-      for (const item of items) {
-        projects.set(item.id, {
-          id: item.id,
-          name: item.name ?? undefined,
-          worktree: item.worktree,
-        })
-      }
-    }
-
-    for (const row of rows) {
-      const project = projects.get(row.project_id) ?? null
-      yield { ...fromRow(row), project }
-    }
-  }
-
-  export const children = fn(SessionID.zod, (id) => runPromise((svc) => svc.children(id)))
-  export const remove = fn(SessionID.zod, (id) => runPromise((svc) => svc.remove(id)))
-  export async function updateMessage<T extends MessageV2.Info>(msg: T): Promise<T> {
-    MessageV2.Info.parse(msg)
-    return runPromise((svc) => svc.updateMessage(msg))
-  }
-
-  export const removeMessage = fn(z.object({ sessionID: SessionID.zod, messageID: MessageID.zod }), (input) =>
-    runPromise((svc) => svc.removeMessage(input)),
-  )
-
-  export const removePart = fn(
-    z.object({ sessionID: SessionID.zod, messageID: MessageID.zod, partID: PartID.zod }),
-    (input) => runPromise((svc) => svc.removePart(input)),
-  )
-
-  export async function updatePart<T extends MessageV2.Part>(part: T): Promise<T> {
-    MessageV2.Part.parse(part)
-    return runPromise((svc) => svc.updatePart(part))
-  }
-
-  export const updatePartDelta = fn(
-    z.object({
-      sessionID: SessionID.zod,
-      messageID: MessageID.zod,
-      partID: PartID.zod,
-      field: z.string(),
-      delta: z.string(),
-    }),
-    (input) => runPromise((svc) => svc.updatePartDelta(input)),
-  )
-}
+export * as Session from "./session"

packages/opencode/src/session/instruction.ts

@@ -1,16 +1,15 @@
 import os from "os"
 import path from "path"
-import { Effect, Layer, ServiceMap } from "effect"
+import { Effect, Layer, Context } from "effect"
 import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/unstable/http"
-import { Config } from "@/config/config"
-import { InstanceState } from "@/effect/instance-state"
-import { makeRuntime } from "@/effect/run-service"
+import { Config } from "@/config"
+import { InstanceState } from "@/effect"
 import { Flag } from "@/flag/flag"
-import { AppFileSystem } from "@/filesystem"
+import { AppFileSystem } from "@opencode-ai/shared/filesystem"
 import { withTransientReadRetry } from "@/util/effect-http-client"
 import { Global } from "../global"
 import { Instance } from "../project/instance"
-import { Log } from "../util/log"
+import { Log } from "../util"
 import type { MessageV2 } from "./message-v2"
 import type { MessageID } from "./schema"
 
@@ -64,7 +63,7 @@ export namespace Instruction {
     ) => Effect.Effect<{ filepath: string; content: string }[], AppFileSystem.Error>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Instruction") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/Instruction") {}
 
   export const layer: Layer.Layer<Service, never, AppFileSystem.Service | Config.Service | HttpClient.HttpClient> =
     Layer.effect(
@@ -238,21 +237,7 @@ export namespace Instruction {
     Layer.provide(FetchHttpClient.layer),
   )
 
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
-  export function clear(messageID: MessageID) {
-    return runPromise((svc) => svc.clear(messageID))
-  }
-
-  export async function systemPaths() {
-    return runPromise((svc) => svc.systemPaths())
-  }
-
   export function loaded(messages: MessageV2.WithParts[]) {
     return extract(messages)
   }
-
-  export async function resolve(messages: MessageV2.WithParts[], filepath: string, messageID: MessageID) {
-    return runPromise((svc) => svc.resolve(messages, filepath, messageID))
-  }
 }

packages/opencode/src/session/llm.ts

@@ -1,13 +1,12 @@
-import { Provider } from "@/provider/provider"
-import { Log } from "@/util/log"
-import { Cause, Effect, Layer, Record, ServiceMap } from "effect"
-import * as Queue from "effect/Queue"
+import { Provider } from "@/provider"
+import { Log } from "@/util"
+import { Context, Effect, Layer, Record } from "effect"
 import * as Stream from "effect/Stream"
 import { streamText, wrapLanguageModel, type ModelMessage, type Tool, tool, jsonSchema } from "ai"
 import { mergeDeep, pipe } from "remeda"
 import { GitLabWorkflowLanguageModel } from "gitlab-ai-provider"
-import { ProviderTransform } from "@/provider/transform"
-import { Config } from "@/config/config"
+import { ProviderTransform } from "@/provider"
+import { Config } from "@/config"
 import { Instance } from "@/project/instance"
 import type { Agent } from "@/agent/agent"
 import type { MessageV2 } from "./message-v2"
@@ -17,14 +16,19 @@ import { Flag } from "@/flag/flag"
 import { Permission } from "@/permission"
 import { PermissionID } from "@/permission/schema"
 import { Bus } from "@/bus"
-import { Wildcard } from "@/util/wildcard"
+import { Wildcard } from "@/util"
 import { SessionID } from "@/session/schema"
 import { Auth } from "@/auth"
 import { Installation } from "@/installation"
+import { InstallationVersion } from "@/installation/version"
+import { EffectBridge } from "@/effect"
+import * as Option from "effect/Option"
+import * as OtelTracer from "@effect/opentelemetry/Tracer"
 
 export namespace LLM {
   const log = Log.create({ service: "llm" })
   export const OUTPUT_TOKEN_MAX = ProviderTransform.OUTPUT_TOKEN_MAX
+  type Result = Awaited<ReturnType<typeof streamText>>
 
   export type StreamInput = {
     user: MessageV2.User
@@ -45,354 +49,387 @@ export namespace LLM {
     abort: AbortSignal
   }
 
-  export type Event = Awaited<ReturnType<typeof stream>>["fullStream"] extends AsyncIterable<infer T> ? T : never
+  export type Event = Result["fullStream"] extends AsyncIterable<infer T> ? T : never
 
   export interface Interface {
     readonly stream: (input: StreamInput) => Stream.Stream<Event, unknown>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/LLM") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/LLM") {}
 
-  export const layer = Layer.effect(
+  const live: Layer.Layer<
+    Service,
+    never,
+    Auth.Service | Config.Service | Provider.Service | Plugin.Service | Permission.Service
+  > = Layer.effect(
     Service,
     Effect.gen(function* () {
-      return Service.of({
-        stream(input) {
-          return Stream.scoped(
-            Stream.unwrap(
-              Effect.gen(function* () {
-                const ctrl = yield* Effect.acquireRelease(
-                  Effect.sync(() => new AbortController()),
-                  (ctrl) => Effect.sync(() => ctrl.abort()),
-                )
+      const auth = yield* Auth.Service
+      const config = yield* Config.Service
+      const provider = yield* Provider.Service
+      const plugin = yield* Plugin.Service
+      const perm = yield* Permission.Service
 
-                const result = yield* Effect.promise(() => LLM.stream({ ...input, abort: ctrl.signal }))
+      const run = Effect.fn("LLM.run")(function* (input: StreamRequest) {
+        const l = log
+          .clone()
+          .tag("providerID", input.model.providerID)
+          .tag("modelID", input.model.id)
+          .tag("sessionID", input.sessionID)
+          .tag("small", (input.small ?? false).toString())
+          .tag("agent", input.agent.name)
+          .tag("mode", input.agent.mode)
+        l.info("stream", {
+          modelID: input.model.id,
+          providerID: input.model.providerID,
+        })
 
-                return Stream.fromAsyncIterable(result.fullStream, (e) =>
-                  e instanceof Error ? e : new Error(String(e)),
-                )
+        const [language, cfg, item, info] = yield* Effect.all(
+          [
+            provider.getLanguage(input.model),
+            config.get(),
+            provider.getProvider(input.model.providerID),
+            auth.get(input.model.providerID),
+          ],
+          { concurrency: "unbounded" },
+        )
+
+        // TODO: move this to a proper hook
+        const isOpenaiOauth = item.id === "openai" && info?.type === "oauth"
+
+        const system: string[] = []
+        system.push(
+          [
+            // use agent prompt otherwise provider prompt
+            ...(input.agent.prompt ? [input.agent.prompt] : SystemPrompt.provider(input.model)),
+            // any custom prompt passed into this call
+            ...input.system,
+            // any custom prompt from last user message
+            ...(input.user.system ? [input.user.system] : []),
+          ]
+            .filter((x) => x)
+            .join("\n"),
+        )
+
+        const header = system[0]
+        yield* plugin.trigger(
+          "experimental.chat.system.transform",
+          { sessionID: input.sessionID, model: input.model },
+          { system },
+        )
+        // rejoin to maintain 2-part structure for caching if header unchanged
+        if (system.length > 2 && system[0] === header) {
+          const rest = system.slice(1)
+          system.length = 0
+          system.push(header, rest.join("\n"))
+        }
+
+        const variant =
+          !input.small && input.model.variants && input.user.model.variant
+            ? input.model.variants[input.user.model.variant]
+            : {}
+        const base = input.small
+          ? ProviderTransform.smallOptions(input.model)
+          : ProviderTransform.options({
+              model: input.model,
+              sessionID: input.sessionID,
+              providerOptions: item.options,
+            })
+        const options: Record<string, any> = pipe(
+          base,
+          mergeDeep(input.model.options),
+          mergeDeep(input.agent.options),
+          mergeDeep(variant),
+        )
+        if (isOpenaiOauth) {
+          options.instructions = system.join("\n")
+        }
+
+        const isWorkflow = language instanceof GitLabWorkflowLanguageModel
+        const messages = isOpenaiOauth
+          ? input.messages
+          : isWorkflow
+            ? input.messages
+            : [
+                ...system.map(
+                  (x): ModelMessage => ({
+                    role: "system",
+                    content: x,
+                  }),
+                ),
+                ...input.messages,
+              ]
+
+        const params = yield* plugin.trigger(
+          "chat.params",
+          {
+            sessionID: input.sessionID,
+            agent: input.agent.name,
+            model: input.model,
+            provider: item,
+            message: input.user,
+          },
+          {
+            temperature: input.model.capabilities.temperature
+              ? (input.agent.temperature ?? ProviderTransform.temperature(input.model))
+              : undefined,
+            topP: input.agent.topP ?? ProviderTransform.topP(input.model),
+            topK: ProviderTransform.topK(input.model),
+            maxOutputTokens: ProviderTransform.maxOutputTokens(input.model),
+            options,
+          },
+        )
+
+        const { headers } = yield* plugin.trigger(
+          "chat.headers",
+          {
+            sessionID: input.sessionID,
+            agent: input.agent.name,
+            model: input.model,
+            provider: item,
+            message: input.user,
+          },
+          {
+            headers: {},
+          },
+        )
+
+        const tools = resolveTools(input)
+
+        // LiteLLM and some Anthropic proxies require the tools parameter to be present
+        // when message history contains tool calls, even if no tools are being used.
+        // Add a dummy tool that is never called to satisfy this validation.
+        // This is enabled for:
+        // 1. Providers with "litellm" in their ID or API ID (auto-detected)
+        // 2. Providers with explicit "litellmProxy: true" option (opt-in for custom gateways)
+        const isLiteLLMProxy =
+          item.options?.["litellmProxy"] === true ||
+          input.model.providerID.toLowerCase().includes("litellm") ||
+          input.model.api.id.toLowerCase().includes("litellm")
+
+        // LiteLLM/Bedrock rejects requests where the message history contains tool
+        // calls but no tools param is present. When there are no active tools (e.g.
+        // during compaction), inject a stub tool to satisfy the validation requirement.
+        // The stub description explicitly tells the model not to call it.
+        if (
+          (isLiteLLMProxy || input.model.providerID.includes("github-copilot")) &&
+          Object.keys(tools).length === 0 &&
+          hasToolCalls(input.messages)
+        ) {
+          tools["_noop"] = tool({
+            description: "Do not call this tool. It exists only for API compatibility and must never be invoked.",
+            inputSchema: jsonSchema({
+              type: "object",
+              properties: {
+                reason: { type: "string", description: "Unused" },
+              },
+            }),
+            execute: async () => ({ output: "", title: "", metadata: {} }),
+          })
+        }
+
+        // Wire up toolExecutor for DWS workflow models so that tool calls
+        // from the workflow service are executed via opencode's tool system
+        // and results sent back over the WebSocket.
+        if (language instanceof GitLabWorkflowLanguageModel) {
+          const workflowModel = language as GitLabWorkflowLanguageModel & {
+            sessionID?: string
+            sessionPreapprovedTools?: string[]
+            approvalHandler?: (approvalTools: { name: string; args: string }[]) => Promise<{ approved: boolean }>
+          }
+          workflowModel.sessionID = input.sessionID
+          workflowModel.systemPrompt = system.join("\n")
+          workflowModel.toolExecutor = async (toolName, argsJson, _requestID) => {
+            const t = tools[toolName]
+            if (!t || !t.execute) {
+              return { result: "", error: `Unknown tool: ${toolName}` }
+            }
+            try {
+              const result = await t.execute!(JSON.parse(argsJson), {
+                toolCallId: _requestID,
+                messages: input.messages,
+                abortSignal: input.abort,
+              })
+              const output = typeof result === "string" ? result : (result?.output ?? JSON.stringify(result))
+              return {
+                result: output,
+                metadata: typeof result === "object" ? result?.metadata : undefined,
+                title: typeof result === "object" ? result?.title : undefined,
+              }
+            } catch (e: any) {
+              return { result: "", error: e.message ?? String(e) }
+            }
+          }
+
+          const ruleset = Permission.merge(input.agent.permission ?? [], input.permission ?? [])
+          workflowModel.sessionPreapprovedTools = Object.keys(tools).filter((name) => {
+            const match = ruleset.findLast((rule) => Wildcard.match(name, rule.permission))
+            return !match || match.action !== "ask"
+          })
+
+          const bridge = yield* EffectBridge.make()
+          const approvedToolsForSession = new Set<string>()
+          workflowModel.approvalHandler = Instance.bind(async (approvalTools) => {
+            const uniqueNames = [...new Set(approvalTools.map((t: { name: string }) => t.name))] as string[]
+            // Auto-approve tools that were already approved in this session
+            // (prevents infinite approval loops for server-side MCP tools)
+            if (uniqueNames.every((name) => approvedToolsForSession.has(name))) {
+              return { approved: true }
+            }
+
+            const id = PermissionID.ascending()
+            let unsub: (() => void) | undefined
+            try {
+              unsub = Bus.subscribe(Permission.Event.Replied, (evt) => {
+                if (evt.properties.requestID === id) void evt.properties.reply
+              })
+              const toolPatterns = approvalTools.map((t: { name: string; args: string }) => {
+                try {
+                  const parsed = JSON.parse(t.args) as Record<string, unknown>
+                  const title = (parsed?.title ?? parsed?.name ?? "") as string
+                  return title ? `${t.name}: ${title}` : t.name
+                } catch {
+                  return t.name
+                }
+              })
+              const uniquePatterns = [...new Set(toolPatterns)] as string[]
+              await bridge.promise(
+                perm.ask({
+                  id,
+                  sessionID: SessionID.make(input.sessionID),
+                  permission: "workflow_tool_approval",
+                  patterns: uniquePatterns,
+                  metadata: { tools: approvalTools },
+                  always: uniquePatterns,
+                  ruleset: [],
+                }),
+              )
+              for (const name of uniqueNames) approvedToolsForSession.add(name)
+              workflowModel.sessionPreapprovedTools = [...(workflowModel.sessionPreapprovedTools ?? []), ...uniqueNames]
+              return { approved: true }
+            } catch {
+              return { approved: false }
+            } finally {
+              unsub?.()
+            }
+          })
+        }
+
+        const tracer = cfg.experimental?.openTelemetry
+          ? Option.getOrUndefined(yield* Effect.serviceOption(OtelTracer.OtelTracer))
+          : undefined
+
+        return streamText({
+          onError(error) {
+            l.error("stream error", {
+              error,
+            })
+          },
+          async experimental_repairToolCall(failed) {
+            const lower = failed.toolCall.toolName.toLowerCase()
+            if (lower !== failed.toolCall.toolName && tools[lower]) {
+              l.info("repairing tool call", {
+                tool: failed.toolCall.toolName,
+                repaired: lower,
+              })
+              return {
+                ...failed.toolCall,
+                toolName: lower,
+              }
+            }
+            return {
+              ...failed.toolCall,
+              input: JSON.stringify({
+                tool: failed.toolCall.toolName,
+                error: failed.error.message,
               }),
-            ),
-          )
-        },
+              toolName: "invalid",
+            }
+          },
+          temperature: params.temperature,
+          topP: params.topP,
+          topK: params.topK,
+          providerOptions: ProviderTransform.providerOptions(input.model, params.options),
+          activeTools: Object.keys(tools).filter((x) => x !== "invalid"),
+          tools,
+          toolChoice: input.toolChoice,
+          maxOutputTokens: params.maxOutputTokens,
+          abortSignal: input.abort,
+          headers: {
+            ...(input.model.providerID.startsWith("opencode")
+              ? {
+                  "x-opencode-project": Instance.project.id,
+                  "x-opencode-session": input.sessionID,
+                  "x-opencode-request": input.user.id,
+                  "x-opencode-client": Flag.OPENCODE_CLIENT,
+                }
+              : {
+                  "x-session-affinity": input.sessionID,
+                  ...(input.parentSessionID ? { "x-parent-session-id": input.parentSessionID } : {}),
+                  "User-Agent": `opencode/${InstallationVersion}`,
+                }),
+            ...input.model.headers,
+            ...headers,
+          },
+          maxRetries: input.retries ?? 0,
+          messages,
+          model: wrapLanguageModel({
+            model: language,
+            middleware: [
+              {
+                specificationVersion: "v3" as const,
+                async transformParams(args) {
+                  if (args.type === "stream") {
+                    // @ts-expect-error
+                    args.params.prompt = ProviderTransform.message(args.params.prompt, input.model, options)
+                  }
+                  return args.params
+                },
+              },
+            ],
+          }),
+          experimental_telemetry: {
+            isEnabled: cfg.experimental?.openTelemetry,
+            functionId: "session.llm",
+            tracer,
+            metadata: {
+              userId: cfg.username ?? "unknown",
+              sessionId: input.sessionID,
+            },
+          },
+        })
       })
+
+      const stream: Interface["stream"] = (input) =>
+        Stream.scoped(
+          Stream.unwrap(
+            Effect.gen(function* () {
+              const ctrl = yield* Effect.acquireRelease(
+                Effect.sync(() => new AbortController()),
+                (ctrl) => Effect.sync(() => ctrl.abort()),
+              )
+
+              const result = yield* run({ ...input, abort: ctrl.signal })
+
+              return Stream.fromAsyncIterable(result.fullStream, (e) => (e instanceof Error ? e : new Error(String(e))))
+            }),
+          ),
+        )
+
+      return Service.of({ stream })
     }),
   )
 
-  export const defaultLayer = layer
+  export const layer = live.pipe(Layer.provide(Permission.defaultLayer))
 
-  export async function stream(input: StreamRequest) {
-    const l = log
-      .clone()
-      .tag("providerID", input.model.providerID)
-      .tag("modelID", input.model.id)
-      .tag("sessionID", input.sessionID)
-      .tag("small", (input.small ?? false).toString())
-      .tag("agent", input.agent.name)
-      .tag("mode", input.agent.mode)
-    l.info("stream", {
-      modelID: input.model.id,
-      providerID: input.model.providerID,
-    })
-    const [language, cfg, provider, auth] = await Promise.all([
-      Provider.getLanguage(input.model),
-      Config.get(),
-      Provider.getProvider(input.model.providerID),
-      Auth.get(input.model.providerID),
-    ])
-    // TODO: move this to a proper hook
-    const isOpenaiOauth = provider.id === "openai" && auth?.type === "oauth"
-
-    const system: string[] = []
-    system.push(
-      [
-        // use agent prompt otherwise provider prompt
-        ...(input.agent.prompt ? [input.agent.prompt] : SystemPrompt.provider(input.model)),
-        // any custom prompt passed into this call
-        ...input.system,
-        // any custom prompt from last user message
-        ...(input.user.system ? [input.user.system] : []),
-      ]
-        .filter((x) => x)
-        .join("\n"),
-    )
-
-    const header = system[0]
-    await Plugin.trigger(
-      "experimental.chat.system.transform",
-      { sessionID: input.sessionID, model: input.model },
-      { system },
-    )
-    // rejoin to maintain 2-part structure for caching if header unchanged
-    if (system.length > 2 && system[0] === header) {
-      const rest = system.slice(1)
-      system.length = 0
-      system.push(header, rest.join("\n"))
-    }
-
-    const variant =
-      !input.small && input.model.variants && input.user.model.variant
-        ? input.model.variants[input.user.model.variant]
-        : {}
-    const base = input.small
-      ? ProviderTransform.smallOptions(input.model)
-      : ProviderTransform.options({
-          model: input.model,
-          sessionID: input.sessionID,
-          providerOptions: provider.options,
-        })
-    const options: Record<string, any> = pipe(
-      base,
-      mergeDeep(input.model.options),
-      mergeDeep(input.agent.options),
-      mergeDeep(variant),
-    )
-    if (isOpenaiOauth) {
-      options.instructions = system.join("\n")
-    }
-
-    const isWorkflow = language instanceof GitLabWorkflowLanguageModel
-    const messages = isOpenaiOauth
-      ? input.messages
-      : isWorkflow
-        ? input.messages
-        : [
-            ...system.map(
-              (x): ModelMessage => ({
-                role: "system",
-                content: x,
-              }),
-            ),
-            ...input.messages,
-          ]
-
-    const params = await Plugin.trigger(
-      "chat.params",
-      {
-        sessionID: input.sessionID,
-        agent: input.agent.name,
-        model: input.model,
-        provider,
-        message: input.user,
-      },
-      {
-        temperature: input.model.capabilities.temperature
-          ? (input.agent.temperature ?? ProviderTransform.temperature(input.model))
-          : undefined,
-        topP: input.agent.topP ?? ProviderTransform.topP(input.model),
-        topK: ProviderTransform.topK(input.model),
-        maxOutputTokens: ProviderTransform.maxOutputTokens(input.model),
-        options,
-      },
-    )
-
-    const { headers } = await Plugin.trigger(
-      "chat.headers",
-      {
-        sessionID: input.sessionID,
-        agent: input.agent.name,
-        model: input.model,
-        provider,
-        message: input.user,
-      },
-      {
-        headers: {},
-      },
-    )
-
-    const tools = await resolveTools(input)
-
-    // LiteLLM and some Anthropic proxies require the tools parameter to be present
-    // when message history contains tool calls, even if no tools are being used.
-    // Add a dummy tool that is never called to satisfy this validation.
-    // This is enabled for:
-    // 1. Providers with "litellm" in their ID or API ID (auto-detected)
-    // 2. Providers with explicit "litellmProxy: true" option (opt-in for custom gateways)
-    const isLiteLLMProxy =
-      provider.options?.["litellmProxy"] === true ||
-      input.model.providerID.toLowerCase().includes("litellm") ||
-      input.model.api.id.toLowerCase().includes("litellm")
-
-    // LiteLLM/Bedrock rejects requests where the message history contains tool
-    // calls but no tools param is present. When there are no active tools (e.g.
-    // during compaction), inject a stub tool to satisfy the validation requirement.
-    // The stub description explicitly tells the model not to call it.
-    if (isLiteLLMProxy && Object.keys(tools).length === 0 && hasToolCalls(input.messages)) {
-      tools["_noop"] = tool({
-        description: "Do not call this tool. It exists only for API compatibility and must never be invoked.",
-        inputSchema: jsonSchema({
-          type: "object",
-          properties: {
-            reason: { type: "string", description: "Unused" },
-          },
-        }),
-        execute: async () => ({ output: "", title: "", metadata: {} }),
-      })
-    }
-
-    // Wire up toolExecutor for DWS workflow models so that tool calls
-    // from the workflow service are executed via opencode's tool system
-    // and results sent back over the WebSocket.
-    if (language instanceof GitLabWorkflowLanguageModel) {
-      const workflowModel = language as GitLabWorkflowLanguageModel & {
-        sessionID?: string
-        sessionPreapprovedTools?: string[]
-        approvalHandler?: (approvalTools: { name: string; args: string }[]) => Promise<{ approved: boolean }>
-      }
-      workflowModel.sessionID = input.sessionID
-      workflowModel.systemPrompt = system.join("\n")
-      workflowModel.toolExecutor = async (toolName, argsJson, _requestID) => {
-        const t = tools[toolName]
-        if (!t || !t.execute) {
-          return { result: "", error: `Unknown tool: ${toolName}` }
-        }
-        try {
-          const result = await t.execute!(JSON.parse(argsJson), {
-            toolCallId: _requestID,
-            messages: input.messages,
-            abortSignal: input.abort,
-          })
-          const output = typeof result === "string" ? result : (result?.output ?? JSON.stringify(result))
-          return {
-            result: output,
-            metadata: typeof result === "object" ? result?.metadata : undefined,
-            title: typeof result === "object" ? result?.title : undefined,
-          }
-        } catch (e: any) {
-          return { result: "", error: e.message ?? String(e) }
-        }
-      }
-
-      const ruleset = Permission.merge(input.agent.permission ?? [], input.permission ?? [])
-      workflowModel.sessionPreapprovedTools = Object.keys(tools).filter((name) => {
-        const match = ruleset.findLast((rule) => Wildcard.match(name, rule.permission))
-        return !match || match.action !== "ask"
-      })
-
-      const approvedToolsForSession = new Set<string>()
-      workflowModel.approvalHandler = Instance.bind(async (approvalTools) => {
-        const uniqueNames = [...new Set(approvalTools.map((t: { name: string }) => t.name))] as string[]
-        // Auto-approve tools that were already approved in this session
-        // (prevents infinite approval loops for server-side MCP tools)
-        if (uniqueNames.every((name) => approvedToolsForSession.has(name))) {
-          return { approved: true }
-        }
-
-        const id = PermissionID.ascending()
-        let reply: Permission.Reply | undefined
-        let unsub: (() => void) | undefined
-        try {
-          unsub = Bus.subscribe(Permission.Event.Replied, (evt) => {
-            if (evt.properties.requestID === id) reply = evt.properties.reply
-          })
-          const toolPatterns = approvalTools.map((t: { name: string; args: string }) => {
-            try {
-              const parsed = JSON.parse(t.args) as Record<string, unknown>
-              const title = (parsed?.title ?? parsed?.name ?? "") as string
-              return title ? `${t.name}: ${title}` : t.name
-            } catch {
-              return t.name
-            }
-          })
-          const uniquePatterns = [...new Set(toolPatterns)] as string[]
-          await Permission.ask({
-            id,
-            sessionID: SessionID.make(input.sessionID),
-            permission: "workflow_tool_approval",
-            patterns: uniquePatterns,
-            metadata: { tools: approvalTools },
-            always: uniquePatterns,
-            ruleset: [],
-          })
-          for (const name of uniqueNames) approvedToolsForSession.add(name)
-          workflowModel.sessionPreapprovedTools = [...(workflowModel.sessionPreapprovedTools ?? []), ...uniqueNames]
-          return { approved: true }
-        } catch {
-          return { approved: false }
-        } finally {
-          unsub?.()
-        }
-      })
-    }
-
-    return streamText({
-      onError(error) {
-        l.error("stream error", {
-          error,
-        })
-      },
-      async experimental_repairToolCall(failed) {
-        const lower = failed.toolCall.toolName.toLowerCase()
-        if (lower !== failed.toolCall.toolName && tools[lower]) {
-          l.info("repairing tool call", {
-            tool: failed.toolCall.toolName,
-            repaired: lower,
-          })
-          return {
-            ...failed.toolCall,
-            toolName: lower,
-          }
-        }
-        return {
-          ...failed.toolCall,
-          input: JSON.stringify({
-            tool: failed.toolCall.toolName,
-            error: failed.error.message,
-          }),
-          toolName: "invalid",
-        }
-      },
-      temperature: params.temperature,
-      topP: params.topP,
-      topK: params.topK,
-      providerOptions: ProviderTransform.providerOptions(input.model, params.options),
-      activeTools: Object.keys(tools).filter((x) => x !== "invalid"),
-      tools,
-      toolChoice: input.toolChoice,
-      maxOutputTokens: params.maxOutputTokens,
-      abortSignal: input.abort,
-      headers: {
-        ...(input.model.providerID.startsWith("opencode")
-          ? {
-              "x-opencode-project": Instance.project.id,
-              "x-opencode-session": input.sessionID,
-              "x-opencode-request": input.user.id,
-              "x-opencode-client": Flag.OPENCODE_CLIENT,
-            }
-          : {
-              "x-session-affinity": input.sessionID,
-              ...(input.parentSessionID ? { "x-parent-session-id": input.parentSessionID } : {}),
-              "User-Agent": `opencode/${Installation.VERSION}`,
-            }),
-        ...input.model.headers,
-        ...headers,
-      },
-      maxRetries: input.retries ?? 0,
-      messages,
-      model: wrapLanguageModel({
-        model: language,
-        middleware: [
-          {
-            specificationVersion: "v3" as const,
-            async transformParams(args) {
-              if (args.type === "stream") {
-                // @ts-expect-error
-                args.params.prompt = ProviderTransform.message(args.params.prompt, input.model, options)
-              }
-              return args.params
-            },
-          },
-        ],
-      }),
-      experimental_telemetry: {
-        isEnabled: cfg.experimental?.openTelemetry,
-        metadata: {
-          userId: cfg.username ?? "unknown",
-          sessionId: input.sessionID,
-        },
-      },
-    })
-  }
+  export const defaultLayer = Layer.suspend(() =>
+    layer.pipe(
+      Layer.provide(Auth.defaultLayer),
+      Layer.provide(Config.defaultLayer),
+      Layer.provide(Provider.defaultLayer),
+      Layer.provide(Plugin.defaultLayer),
+    ),
+  )
 
   function resolveTools(input: Pick<StreamInput, "tools" | "agent" | "permission" | "user">) {
     const disabled = Permission.disabled(

packages/opencode/src/session/message-v2.ts

@@ -1,20 +1,21 @@
 import { BusEvent } from "@/bus/bus-event"
 import { SessionID, MessageID, PartID } from "./schema"
 import z from "zod"
-import { NamedError } from "@opencode-ai/util/error"
+import { NamedError } from "@opencode-ai/shared/util/error"
 import { APICallError, convertToModelMessages, LoadAPIKeyError, type ModelMessage, type UIMessage } from "ai"
 import { LSP } from "../lsp"
 import { Snapshot } from "@/snapshot"
 import { SyncEvent } from "../sync"
-import { Database, NotFoundError, and, desc, eq, inArray, lt, or } from "@/storage/db"
+import { Database, NotFoundError, and, desc, eq, inArray, lt, or } from "@/storage"
 import { MessageTable, PartTable, SessionTable } from "./session.sql"
-import { ProviderError } from "@/provider/error"
+import { ProviderError } from "@/provider"
 import { iife } from "@/util/iife"
 import { errorMessage } from "@/util/error"
 import type { SystemError } from "bun"
-import type { Provider } from "@/provider/provider"
+import type { Provider } from "@/provider"
 import { ModelID, ProviderID } from "@/provider/schema"
 import { Effect } from "effect"
+import { EffectLogger } from "@/effect"
 
 /** Error shape thrown by Bun's fetch() when gzip/br decompression fails mid-stream */
 interface FetchDecompressionError extends Error {
@@ -24,6 +25,8 @@ interface FetchDecompressionError extends Error {
 }
 
 export namespace MessageV2 {
+  export const SYNTHETIC_ATTACHMENT_PROMPT = "Attached image(s) from tool result:"
+
   export function isMedia(mime: string) {
     return mime.startsWith("image/") || mime === "application/pdf"
   }
@@ -808,7 +811,7 @@ export namespace MessageV2 {
               parts: [
                 {
                   type: "text" as const,
-                  text: "Attached image(s) from tool result:",
+                  text: SYNTHETIC_ATTACHMENT_PROMPT,
                 },
                 ...media.map((attachment) => ({
                   type: "file" as const,
@@ -840,7 +843,7 @@ export namespace MessageV2 {
     model: Provider.Model,
     options?: { stripMedia?: boolean },
   ): Promise<ModelMessage[]> {
-    return Effect.runPromise(toModelMessagesEffect(input, model, options))
+    return Effect.runPromise(toModelMessagesEffect(input, model, options).pipe(Effect.provide(EffectLogger.layer)))
   }
 
   export function page(input: { sessionID: SessionID; limit: number; before?: string }) {

packages/opencode/src/session/message.ts

@@ -1,7 +1,7 @@
 import z from "zod"
 import { SessionID } from "./schema"
 import { ModelID, ProviderID } from "../provider/schema"
-import { NamedError } from "@opencode-ai/util/error"
+import { NamedError } from "@opencode-ai/shared/util/error"
 
 export namespace Message {
   export const OutputLengthError = NamedError.create("MessageOutputLengthError", z.object({}))

packages/opencode/src/session/overflow.ts

@@ -1,6 +1,6 @@
-import type { Config } from "@/config/config"
-import type { Provider } from "@/provider/provider"
-import { ProviderTransform } from "@/provider/transform"
+import type { Config } from "@/config"
+import type { Provider } from "@/provider"
+import { ProviderTransform } from "@/provider"
 import type { MessageV2 } from "./message-v2"
 
 const COMPACTION_BUFFER = 20_000

packages/opencode/src/session/processor.ts

@@ -1,13 +1,12 @@
-import { Cause, Deferred, Effect, Layer, ServiceMap } from "effect"
+import { Cause, Deferred, Effect, Layer, Context, Scope } from "effect"
 import * as Stream from "effect/Stream"
 import { Agent } from "@/agent/agent"
 import { Bus } from "@/bus"
-import { Config } from "@/config/config"
+import { Config } from "@/config"
 import { Permission } from "@/permission"
 import { Plugin } from "@/plugin"
 import { Snapshot } from "@/snapshot"
-import { Log } from "@/util/log"
-import { Session } from "."
+import * as Session from "./session"
 import { LLM } from "./llm"
 import { MessageV2 } from "./message-v2"
 import { isOverflow } from "./overflow"
@@ -16,9 +15,10 @@ import type { SessionID } from "./schema"
 import { SessionRetry } from "./retry"
 import { SessionStatus } from "./status"
 import { SessionSummary } from "./summary"
-import type { Provider } from "@/provider/provider"
+import type { Provider } from "@/provider"
 import { Question } from "@/question"
 import { errorMessage } from "@/util/error"
+import { Log } from "@/util"
 import { isRecord } from "@/util/record"
 
 export namespace SessionProcessor {
@@ -76,7 +76,7 @@ export namespace SessionProcessor {
 
   type StreamEvent = Event
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionProcessor") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionProcessor") {}
 
   export const layer: Layer.Layer<
     Service,
@@ -89,6 +89,7 @@ export namespace SessionProcessor {
     | LLM.Service
     | Permission.Service
     | Plugin.Service
+    | SessionSummary.Service
     | SessionStatus.Service
   > = Layer.effect(
     Service,
@@ -101,6 +102,8 @@ export namespace SessionProcessor {
       const llm = yield* LLM.Service
       const permission = yield* Permission.Service
       const plugin = yield* Plugin.Service
+      const summary = yield* SessionSummary.Service
+      const scope = yield* Scope.Scope
       const status = yield* SessionStatus.Service
 
       const create = Effect.fn("SessionProcessor.create")(function* (input: Input) {
@@ -121,6 +124,7 @@ export namespace SessionProcessor {
           reasoningMap: {},
         }
         let aborted = false
+        const slog = log.clone().tag("sessionID", input.sessionID).tag("messageID", input.assistantMessage.id)
 
         const parse = (e: unknown) =>
           MessageV2.fromError(e, {
@@ -245,7 +249,8 @@ export namespace SessionProcessor {
 
             case "reasoning-end":
               if (!(value.id in ctx.reasoningMap)) return
-              ctx.reasoningMap[value.id].text = ctx.reasoningMap[value.id].text.trimEnd()
+              // oxlint-disable-next-line no-self-assign -- reactivity trigger
+              ctx.reasoningMap[value.id].text = ctx.reasoningMap[value.id].text
               ctx.reasoningMap[value.id].time = { ...ctx.reasoningMap[value.id].time, end: Date.now() }
               if (value.providerMetadata) ctx.reasoningMap[value.id].metadata = value.providerMetadata
               yield* session.updatePart(ctx.reasoningMap[value.id])
@@ -384,10 +389,12 @@ export namespace SessionProcessor {
                 }
                 ctx.snapshot = undefined
               }
-              SessionSummary.summarize({
-                sessionID: ctx.sessionID,
-                messageID: ctx.assistantMessage.parentID,
-              })
+              yield* summary
+                .summarize({
+                  sessionID: ctx.sessionID,
+                  messageID: ctx.assistantMessage.parentID,
+                })
+                .pipe(Effect.ignore, Effect.forkIn(scope))
               if (
                 !ctx.assistantMessage.summary &&
                 isOverflow({ cfg: yield* config.get(), tokens: usage.tokens, model: ctx.model })
@@ -425,7 +432,8 @@ export namespace SessionProcessor {
 
             case "text-end":
               if (!ctx.currentText) return
-              ctx.currentText.text = ctx.currentText.text.trimEnd()
+              // oxlint-disable-next-line no-self-assign -- reactivity trigger
+              ctx.currentText.text = ctx.currentText.text
               ctx.currentText.text = (yield* plugin.trigger(
                 "experimental.text.complete",
                 {
@@ -448,7 +456,7 @@ export namespace SessionProcessor {
               return
 
             default:
-              log.info("unhandled", { ...value })
+              slog.info("unhandled", { event: value.type, value })
               return
           }
         })
@@ -514,7 +522,7 @@ export namespace SessionProcessor {
         })
 
         const halt = Effect.fn("SessionProcessor.halt")(function* (e: unknown) {
-          log.error("process", { error: e, stack: e instanceof Error ? e.stack : undefined })
+          slog.error("process", { error: errorMessage(e), stack: e instanceof Error ? e.stack : undefined })
           const error = parse(e)
           if (MessageV2.ContextOverflowError.isInstance(error)) {
             ctx.needsCompaction = true
@@ -530,7 +538,7 @@ export namespace SessionProcessor {
         })
 
         const process = Effect.fn("SessionProcessor.process")(function* (streamInput: LLM.StreamInput) {
-          log.info("process")
+          slog.info("process")
           ctx.needsCompaction = false
           ctx.shouldBreak = (yield* config.get()).experimental?.continue_loop_on_deny !== true
 
@@ -602,6 +610,7 @@ export namespace SessionProcessor {
       Layer.provide(LLM.defaultLayer),
       Layer.provide(Permission.defaultLayer),
       Layer.provide(Plugin.defaultLayer),
+      Layer.provide(SessionSummary.defaultLayer),
       Layer.provide(SessionStatus.defaultLayer),
       Layer.provide(Bus.layer),
       Layer.provide(Config.defaultLayer),

packages/opencode/src/session/projectors.ts

@@ -1,10 +1,9 @@
-import { NotFoundError, eq, and } from "../storage/db"
+import { NotFoundError, eq, and } from "../storage"
 import { SyncEvent } from "@/sync"
-import { Session } from "./index"
+import * as Session from "./session"
 import { MessageV2 } from "./message-v2"
 import { SessionTable, MessageTable, PartTable } from "./session.sql"
-import { ProjectTable } from "../project/project.sql"
-import { Log } from "../util/log"
+import { Log } from "../util"
 
 const log = Log.create({ service: "session.projector" })
 

packages/opencode/src/session/prompt.ts

@@ -3,23 +3,23 @@ import os from "os"
 import z from "zod"
 import { SessionID, MessageID, PartID } from "./schema"
 import { MessageV2 } from "./message-v2"
-import { Log } from "../util/log"
+import { Log } from "../util"
 import { SessionRevert } from "./revert"
-import { Session } from "."
+import * as Session from "./session"
 import { Agent } from "../agent/agent"
-import { Provider } from "../provider/provider"
+import { Provider } from "../provider"
 import { ModelID, ProviderID } from "../provider/schema"
 import { type Tool as AITool, tool, jsonSchema, type ToolExecutionOptions, asSchema } from "ai"
 import { SessionCompaction } from "./compaction"
 import { Bus } from "../bus"
-import { ProviderTransform } from "../provider/transform"
+import { ProviderTransform } from "../provider"
 import { SystemPrompt } from "./system"
 import { Instruction } from "./instruction"
 import { Plugin } from "../plugin"
 import PROMPT_PLAN from "../session/prompt/plan.txt"
 import BUILD_SWITCH from "../session/prompt/build-switch.txt"
 import MAX_STEPS from "../session/prompt/max-steps.txt"
-import { ToolRegistry } from "../tool/registry"
+import { ToolRegistry } from "../tool"
 import { MCP } from "../mcp"
 import { LSP } from "../lsp"
 import { FileTime } from "../file/time"
@@ -30,24 +30,25 @@ import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
 import * as Stream from "effect/Stream"
 import { Command } from "../command"
 import { pathToFileURL, fileURLToPath } from "url"
-import { ConfigMarkdown } from "../config/markdown"
+import { ConfigMarkdown } from "../config"
 import { SessionSummary } from "./summary"
-import { NamedError } from "@opencode-ai/util/error"
+import { NamedError } from "@opencode-ai/shared/util/error"
 import { SessionProcessor } from "./processor"
-import { Tool } from "@/tool/tool"
+import { Tool } from "@/tool"
 import { Permission } from "@/permission"
 import { SessionStatus } from "./status"
 import { LLM } from "./llm"
 import { Shell } from "@/shell/shell"
-import { AppFileSystem } from "@/filesystem"
-import { Truncate } from "@/tool/truncate"
+import { AppFileSystem } from "@opencode-ai/shared/filesystem"
+import { Truncate } from "@/tool"
 import { decodeDataUrl } from "@/util/data-url"
-import { Process } from "@/util/process"
-import { Cause, Effect, Exit, Layer, Option, Scope, ServiceMap } from "effect"
-import { InstanceState } from "@/effect/instance-state"
-import { makeRuntime } from "@/effect/run-service"
-import { TaskTool } from "@/tool/task"
+import { Process } from "@/util"
+import { Cause, Effect, Exit, Layer, Option, Scope, Context } from "effect"
+import { EffectLogger } from "@/effect"
+import { InstanceState } from "@/effect"
+import { TaskTool, type TaskPromptOps } from "@/tool/task"
 import { SessionRunState } from "./run-state"
+import { EffectBridge } from "@/effect"
 
 // @ts-ignore
 globalThis.AI_SDK_LOG_WARNINGS = false
@@ -64,6 +65,7 @@ const STRUCTURED_OUTPUT_SYSTEM_PROMPT = `IMPORTANT: The user has requested struc
 
 export namespace SessionPrompt {
   const log = Log.create({ service: "session.prompt" })
+  const elog = EffectLogger.create({ service: "session.prompt" })
 
   export interface Interface {
     readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
@@ -74,7 +76,7 @@ export namespace SessionPrompt {
     readonly resolvePromptParts: (template: string) => Effect.Effect<PromptInput["parts"]>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionPrompt") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionPrompt") {}
 
   export const layer = Layer.effect(
     Service,
@@ -100,9 +102,23 @@ export namespace SessionPrompt {
       const instruction = yield* Instruction.Service
       const state = yield* SessionRunState.Service
       const revert = yield* SessionRevert.Service
+      const summary = yield* SessionSummary.Service
+      const sys = yield* SystemPrompt.Service
+      const llm = yield* LLM.Service
+      const runner = Effect.fn("SessionPrompt.runner")(function* () {
+        return yield* EffectBridge.make()
+      })
+      const ops = Effect.fn("SessionPrompt.ops")(function* () {
+        const run = yield* runner()
+        return {
+          cancel: (sessionID: SessionID) => run.fork(cancel(sessionID)),
+          resolvePromptParts: (template: string) => resolvePromptParts(template),
+          prompt: (input: PromptInput) => prompt(input),
+        } satisfies TaskPromptOps
+      })
 
       const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) {
-        log.info("cancel", { sessionID })
+        yield* elog.info("cancel", { sessionID })
         yield* state.cancel(sessionID)
       })
 
@@ -172,21 +188,24 @@ export namespace SessionPrompt {
         const msgs = onlySubtasks
           ? [{ role: "user" as const, content: subtasks.map((p) => p.prompt).join("\n") }]
           : yield* MessageV2.toModelMessagesEffect(context, mdl)
-        const text = yield* Effect.promise(async (signal) => {
-          const result = await LLM.stream({
+        const text = yield* llm
+          .stream({
             agent: ag,
             user: firstInfo,
             system: [],
             small: true,
             tools: {},
             model: mdl,
-            abort: signal,
             sessionID: input.session.id,
             retries: 2,
             messages: [{ role: "user", content: "Generate a title for this conversation:\n" }, ...msgs],
           })
-          return result.text
-        })
+          .pipe(
+            Stream.filter((e): e is Extract<LLM.Event, { type: "text-delta" }> => e.type === "text-delta"),
+            Stream.map((e) => e.text),
+            Stream.mkString,
+            Effect.orDie,
+          )
         const cleaned = text
           .replace(/<think>[\s\S]*?<\/think>\s*/g, "")
           .split("\n")
@@ -196,11 +215,7 @@ export namespace SessionPrompt {
         const t = cleaned.length > 100 ? cleaned.substring(0, 97) + "..." : cleaned
         yield* sessions
           .setTitle({ sessionID: input.session.id, title: t })
-          .pipe(
-            Effect.catchCause((cause) =>
-              Effect.sync(() => log.error("failed to generate title", { error: Cause.squash(cause) })),
-            ),
-          )
+          .pipe(Effect.catchCause((cause) => elog.error("failed to generate title", { error: Cause.squash(cause) })))
       })
 
       const insertReminders = Effect.fn("SessionPrompt.insertReminders")(function* (input: {
@@ -245,8 +260,7 @@ export namespace SessionPrompt {
             messageID: userMessage.info.id,
             sessionID: userMessage.info.sessionID,
             type: "text",
-            text:
-              BUILD_SWITCH + "\n\n" + `A plan file exists at ${plan}. You should execute on the plan defined within it`,
+            text: `${BUILD_SWITCH}\n\nA plan file exists at ${plan}. You should execute on the plan defined within it`,
             synthetic: true,
           })
           userMessage.parts.push(part)
@@ -350,40 +364,40 @@ NOTE: At any point in time through this workflow you should feel free to ask the
       }) {
         using _ = log.time("resolveTools")
         const tools: Record<string, AITool> = {}
+        const run = yield* runner()
+        const promptOps = yield* ops()
 
         const context = (args: any, options: ToolExecutionOptions): Tool.Context => ({
           sessionID: input.session.id,
           abort: options.abortSignal!,
           messageID: input.processor.message.id,
           callID: options.toolCallId,
-          extra: { model: input.model, bypassAgentCheck: input.bypassAgentCheck },
+          extra: { model: input.model, bypassAgentCheck: input.bypassAgentCheck, promptOps },
           agent: input.agent.name,
           messages: input.messages,
           metadata: (val) =>
-            Effect.runPromise(
-              input.processor.updateToolCall(options.toolCallId, (match) => {
-                if (!["running", "pending"].includes(match.state.status)) return match
-                return {
-                  ...match,
-                  state: {
-                    title: val.title,
-                    metadata: val.metadata,
-                    status: "running",
-                    input: args,
-                    time: { start: Date.now() },
-                  },
-                }
-              }),
-            ),
+            input.processor.updateToolCall(options.toolCallId, (match) => {
+              if (!["running", "pending"].includes(match.state.status)) return match
+              return {
+                ...match,
+                state: {
+                  title: val.title,
+                  metadata: val.metadata,
+                  status: "running",
+                  input: args,
+                  time: { start: Date.now() },
+                },
+              }
+            }),
           ask: (req) =>
-            Effect.runPromise(
-              permission.ask({
+            permission
+              .ask({
                 ...req,
                 sessionID: input.session.id,
                 tool: { messageID: input.processor.message.id, callID: options.toolCallId },
                 ruleset: Permission.merge(input.agent.permission, input.session.permission ?? []),
-              }),
-            ),
+              })
+              .pipe(Effect.orDie),
         })
 
         for (const item of yield* registry.tools({
@@ -397,7 +411,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
             description: item.description,
             inputSchema: jsonSchema(schema as any),
             execute(args, options) {
-              return Effect.runPromise(
+              return run.promise(
                 Effect.gen(function* () {
                   const ctx = context(args, options)
                   yield* plugin.trigger(
@@ -405,7 +419,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
                     { tool: item.id, sessionID: ctx.sessionID, callID: ctx.callID },
                     { args },
                   )
-                  const result = yield* Effect.promise(() => item.execute(args, ctx))
+                  const result = yield* item.execute(args, ctx)
                   const output = {
                     ...result,
                     attachments: result.attachments?.map((attachment) => ({
@@ -438,7 +452,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
           const transformed = ProviderTransform.schema(input.model, schema)
           item.inputSchema = jsonSchema(transformed)
           item.execute = (args, opts) =>
-            Effect.runPromise(
+            run.promise(
               Effect.gen(function* () {
                 const ctx = context(args, opts)
                 yield* plugin.trigger(
@@ -446,7 +460,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
                   { tool: key, sessionID: ctx.sessionID, callID: opts.toolCallId },
                   { args },
                 )
-                yield* Effect.promise(() => ctx.ask({ permission: key, metadata: {}, patterns: ["*"], always: ["*"] }))
+                yield* ctx.ask({ permission: key, metadata: {}, patterns: ["*"], always: ["*"] })
                 const result: Awaited<ReturnType<NonNullable<typeof execute>>> = yield* Effect.promise(() =>
                   execute(args, opts),
                 )
@@ -482,7 +496,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
 
                 const truncated = yield* truncate.output(textParts.join("\n\n"), {}, input.agent)
                 const metadata = {
-                  ...(result.metadata ?? {}),
+                  ...result.metadata,
                   truncated: truncated.truncated,
                   ...(truncated.truncated && { outputPath: truncated.outputPath }),
                 }
@@ -521,6 +535,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
       }) {
         const { task, model, lastUser, sessionID, session, msgs } = input
         const ctx = yield* InstanceState.context
+        const promptOps = yield* ops()
         const { task: taskTool } = yield* registry.named()
         const taskModel = task.model ? yield* getModel(task.model.providerID, task.model.modelID, sessionID) : model
         const assistantMessage: MessageV2.Assistant = yield* sessions.updateMessage({
@@ -578,63 +593,61 @@ NOTE: At any point in time through this workflow you should feel free to ask the
         }
 
         let error: Error | undefined
-        const result = yield* Effect.promise((signal) =>
-          taskTool
-            .execute(taskArgs, {
-              agent: task.agent,
-              messageID: assistantMessage.id,
-              sessionID,
-              abort: signal,
-              callID: part.callID,
-              extra: { bypassAgentCheck: true },
-              messages: msgs,
-              metadata(val: { title?: string; metadata?: Record<string, any> }) {
-                return Effect.runPromise(
-                  Effect.gen(function* () {
-                    part = yield* sessions.updatePart({
-                      ...part,
-                      type: "tool",
-                      state: { ...part.state, ...val },
-                    } satisfies MessageV2.ToolPart)
-                  }),
-                )
-              },
-              ask(req: any) {
-                return Effect.runPromise(
-                  permission.ask({
-                    ...req,
-                    sessionID,
-                    ruleset: Permission.merge(taskAgent.permission, session.permission ?? []),
-                  }),
-                )
-              },
-            })
-            .catch((e) => {
-              error = e instanceof Error ? e : new Error(String(e))
-              log.error("subtask execution failed", { error, agent: task.agent, description: task.description })
-              return undefined
-            }),
-        ).pipe(
-          Effect.onInterrupt(() =>
-            Effect.gen(function* () {
-              assistantMessage.finish = "tool-calls"
-              assistantMessage.time.completed = Date.now()
-              yield* sessions.updateMessage(assistantMessage)
-              if (part.state.status === "running") {
-                yield* sessions.updatePart({
+        const taskAbort = new AbortController()
+        const result = yield* taskTool
+          .execute(taskArgs, {
+            agent: task.agent,
+            messageID: assistantMessage.id,
+            sessionID,
+            abort: taskAbort.signal,
+            callID: part.callID,
+            extra: { bypassAgentCheck: true, promptOps },
+            messages: msgs,
+            metadata: (val: { title?: string; metadata?: Record<string, any> }) =>
+              Effect.gen(function* () {
+                part = yield* sessions.updatePart({
                   ...part,
-                  state: {
-                    status: "error",
-                    error: "Cancelled",
-                    time: { start: part.state.time.start, end: Date.now() },
-                    metadata: part.state.metadata,
-                    input: part.state.input,
-                  },
+                  type: "tool",
+                  state: { ...part.state, ...val },
                 } satisfies MessageV2.ToolPart)
-              }
+              }),
+            ask: (req: any) =>
+              permission
+                .ask({
+                  ...req,
+                  sessionID,
+                  ruleset: Permission.merge(taskAgent.permission, session.permission ?? []),
+                })
+                .pipe(Effect.orDie),
+          })
+          .pipe(
+            Effect.catchCause((cause) => {
+              const defect = Cause.squash(cause)
+              error = defect instanceof Error ? defect : new Error(String(defect))
+              log.error("subtask execution failed", { error, agent: task.agent, description: task.description })
+              return Effect.void
             }),
-          ),
-        )
+            Effect.onInterrupt(() =>
+              Effect.gen(function* () {
+                taskAbort.abort()
+                assistantMessage.finish = "tool-calls"
+                assistantMessage.time.completed = Date.now()
+                yield* sessions.updateMessage(assistantMessage)
+                if (part.state.status === "running") {
+                  yield* sessions.updatePart({
+                    ...part,
+                    state: {
+                      status: "error",
+                      error: "Cancelled",
+                      time: { start: part.state.time.start, end: Date.now() },
+                      metadata: part.state.metadata,
+                      input: part.state.input,
+                    },
+                  } satisfies MessageV2.ToolPart)
+                }
+              }),
+            ),
+          )
 
         const attachments = result?.attachments?.map((attachment) => ({
           ...attachment,
@@ -707,6 +720,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
 
       const shellImpl = Effect.fn("SessionPrompt.shellImpl")(function* (input: ShellInput) {
         const ctx = yield* InstanceState.context
+        const run = yield* runner()
         const session = yield* sessions.get(input.sessionID)
         if (session.revert) {
           yield* revert.cleanup(session)
@@ -857,7 +871,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
               output += chunk
               if (part.state.status === "running") {
                 part.state.metadata = { output, description: "" }
-                void Effect.runFork(sessions.updatePart(part))
+                void run.fork(sessions.updatePart(part))
               }
             }),
           )
@@ -902,12 +916,8 @@ NOTE: At any point in time through this workflow you should feel free to ask the
       })
 
       const lastModel = Effect.fnUntraced(function* (sessionID: SessionID) {
-        const model = yield* Effect.promise(async () => {
-          for await (const item of MessageV2.stream(sessionID)) {
-            if (item.info.role === "user" && item.info.model) return item.info.model
-          }
-        })
-        if (model) return model
+        const match = yield* sessions.findMessage(sessionID, (m) => m.info.role === "user" && !!m.info.model)
+        if (Option.isSome(match) && match.value.info.role === "user") return match.value.info.model
         return yield* provider.defaultModel()
       })
 
@@ -1039,19 +1049,21 @@ NOTE: At any point in time through this workflow you should feel free to ask the
                 if (yield* fsys.isDir(filepath)) part.mime = "application/x-directory"
 
                 const { read } = yield* registry.named()
-                const execRead = (args: Parameters<typeof read.execute>[0], extra?: Tool.Context["extra"]) =>
-                  Effect.promise((signal: AbortSignal) =>
-                    read.execute(args, {
+                const execRead = (args: Parameters<typeof read.execute>[0], extra?: Tool.Context["extra"]) => {
+                  const controller = new AbortController()
+                  return read
+                    .execute(args, {
                       sessionID: input.sessionID,
-                      abort: signal,
+                      abort: controller.signal,
                       agent: input.agent!,
                       messageID: info.id,
                       extra: { bypassCwdCheck: true, ...extra },
                       messages: [],
-                      metadata: async () => {},
-                      ask: async () => {},
-                    }),
-                  )
+                      metadata: () => Effect.void,
+                      ask: () => Effect.void,
+                    })
+                    .pipe(Effect.onInterrupt(() => Effect.sync(() => controller.abort())))
+                }
 
                 if (part.mime === "text/plain") {
                   let offset: number | undefined
@@ -1288,27 +1300,25 @@ NOTE: At any point in time through this workflow you should feel free to ask the
         },
       )
 
-      const lastAssistant = (sessionID: SessionID) =>
-        Effect.promise(async () => {
-          let latest: MessageV2.WithParts | undefined
-          for await (const item of MessageV2.stream(sessionID)) {
-            latest ??= item
-            if (item.info.role !== "user") return item
-          }
-          if (latest) return latest
-          throw new Error("Impossible")
-        })
+      const lastAssistant = Effect.fnUntraced(function* (sessionID: SessionID) {
+        const match = yield* sessions.findMessage(sessionID, (m) => m.info.role !== "user")
+        if (Option.isSome(match)) return match.value
+        const msgs = yield* sessions.messages({ sessionID, limit: 1 })
+        if (msgs.length > 0) return msgs[0]
+        throw new Error("Impossible")
+      })
 
       const runLoop: (sessionID: SessionID) => Effect.Effect<MessageV2.WithParts> = Effect.fn("SessionPrompt.run")(
         function* (sessionID: SessionID) {
           const ctx = yield* InstanceState.context
+          const slog = elog.with({ sessionID })
           let structured: unknown | undefined
           let step = 0
           const session = yield* sessions.get(sessionID)
 
           while (true) {
             yield* status.set(sessionID, { type: "busy" })
-            log.info("loop", { step, sessionID })
+            yield* slog.info("loop", { step })
 
             let msgs = yield* MessageV2.filterCompactedEffect(sessionID)
 
@@ -1344,7 +1354,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
               !hasToolCalls &&
               lastUser.id < lastAssistant.id
             ) {
-              log.info("exiting loop", { sessionID })
+              yield* slog.info("exiting loop")
               break
             }
 
@@ -1443,7 +1453,10 @@ NOTE: At any point in time through this workflow you should feel free to ask the
                 })
               }
 
-              if (step === 1) SessionSummary.summarize({ sessionID, messageID: lastUser.id })
+              if (step === 1)
+                yield* summary
+                  .summarize({ sessionID, messageID: lastUser.id })
+                  .pipe(Effect.ignore, Effect.forkIn(scope))
 
               if (step > 1 && lastFinished) {
                 for (const m of msgs) {
@@ -1466,8 +1479,8 @@ NOTE: At any point in time through this workflow you should feel free to ask the
               yield* plugin.trigger("experimental.chat.messages.transform", {}, { messages: msgs })
 
               const [skills, env, instructions, modelMsgs] = yield* Effect.all([
-                Effect.promise(() => SystemPrompt.skills(agent)),
-                Effect.promise(() => SystemPrompt.environment(model)),
+                sys.skills(agent),
+                Effect.sync(() => sys.environment(model)),
                 instruction.system().pipe(Effect.orDie),
                 MessageV2.toModelMessagesEffect(msgs, model),
               ])
@@ -1540,7 +1553,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
       )
 
       const command = Effect.fn("SessionPrompt.command")(function* (input: CommandInput) {
-        log.info("command", input)
+        yield* elog.info("command", { sessionID: input.sessionID, command: input.command, agent: input.agent })
         const cmd = yield* commands.get(input.command)
         if (!cmd) {
           const available = (yield* commands.list()).map((c) => c.name)
@@ -1666,7 +1679,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
     }),
   )
 
-  const defaultLayer = Layer.suspend(() =>
+  export const defaultLayer = Layer.suspend(() =>
     layer.pipe(
       Layer.provide(SessionRunState.defaultLayer),
       Layer.provide(SessionStatus.defaultLayer),
@@ -1685,13 +1698,18 @@ NOTE: At any point in time through this workflow you should feel free to ask the
       Layer.provide(Plugin.defaultLayer),
       Layer.provide(Session.defaultLayer),
       Layer.provide(SessionRevert.defaultLayer),
-      Layer.provide(Agent.defaultLayer),
-      Layer.provide(Bus.layer),
-      Layer.provide(CrossSpawnSpawner.defaultLayer),
+      Layer.provide(SessionSummary.defaultLayer),
+      Layer.provide(
+        Layer.mergeAll(
+          Agent.defaultLayer,
+          SystemPrompt.defaultLayer,
+          LLM.defaultLayer,
+          Bus.layer,
+          CrossSpawnSpawner.defaultLayer,
+        ),
+      ),
     ),
   )
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
   export const PromptInput = z.object({
     sessionID: SessionID.zod,
     messageID: MessageID.zod.optional(),
@@ -1759,26 +1777,10 @@ NOTE: At any point in time through this workflow you should feel free to ask the
   })
   export type PromptInput = z.infer<typeof PromptInput>
 
-  export async function prompt(input: PromptInput) {
-    return runPromise((svc) => svc.prompt(PromptInput.parse(input)))
-  }
-
-  export async function resolvePromptParts(template: string) {
-    return runPromise((svc) => svc.resolvePromptParts(z.string().parse(template)))
-  }
-
-  export async function cancel(sessionID: SessionID) {
-    return runPromise((svc) => svc.cancel(SessionID.zod.parse(sessionID)))
-  }
-
   export const LoopInput = z.object({
     sessionID: SessionID.zod,
   })
 
-  export async function loop(input: z.infer<typeof LoopInput>) {
-    return runPromise((svc) => svc.loop(LoopInput.parse(input)))
-  }
-
   export const ShellInput = z.object({
     sessionID: SessionID.zod,
     messageID: MessageID.zod.optional(),
@@ -1793,10 +1795,6 @@ NOTE: At any point in time through this workflow you should feel free to ask the
   })
   export type ShellInput = z.infer<typeof ShellInput>
 
-  export async function shell(input: ShellInput) {
-    return runPromise((svc) => svc.shell(ShellInput.parse(input)))
-  }
-
   export const CommandInput = z.object({
     messageID: MessageID.zod.optional(),
     sessionID: SessionID.zod,
@@ -1820,17 +1818,13 @@ NOTE: At any point in time through this workflow you should feel free to ask the
   })
   export type CommandInput = z.infer<typeof CommandInput>
 
-  export async function command(input: CommandInput) {
-    return runPromise((svc) => svc.command(CommandInput.parse(input)))
-  }
-
   /** @internal Exported for testing */
   export function createStructuredOutputTool(input: {
     schema: Record<string, any>
     onSuccess: (output: unknown) => void
   }): AITool {
     // Remove $schema property if present (not needed for tool input)
-    const { $schema, ...toolSchema } = input.schema
+    const { $schema: _, ...toolSchema } = input.schema
 
     return tool({
       id: "StructuredOutput" as any,

packages/opencode/src/session/retry.ts

@@ -1,4 +1,4 @@
-import type { NamedError } from "@opencode-ai/util/error"
+import type { NamedError } from "@opencode-ai/shared/util/error"
 import { Cause, Clock, Duration, Effect, Schedule } from "effect"
 import { MessageV2 } from "./message-v2"
 import { iife } from "@/util/iife"
@@ -56,7 +56,10 @@ export namespace SessionRetry {
     // context overflow errors should not be retried
     if (MessageV2.ContextOverflowError.isInstance(error)) return undefined
     if (MessageV2.APIError.isInstance(error)) {
-      if (!error.data.isRetryable) return undefined
+      const status = error.data.statusCode
+      // 5xx errors are transient server failures and should always be retried,
+      // even when the provider SDK doesn't explicitly mark them as retryable.
+      if (!error.data.isRetryable && !(status !== undefined && status >= 500)) return undefined
       if (error.data.responseBody?.includes("FreeUsageLimitError")) return GO_UPSELL_MESSAGE
       return error.data.message.includes("Overloaded") ? "Provider is overloaded" : error.data.message
     }

packages/opencode/src/session/revert.ts

@@ -1,17 +1,15 @@
 import z from "zod"
-import { Effect, Layer, ServiceMap } from "effect"
-import { makeRuntime } from "@/effect/run-service"
+import { Effect, Layer, Context } from "effect"
 import { Bus } from "../bus"
 import { Snapshot } from "../snapshot"
-import { Storage } from "@/storage/storage"
+import { Storage } from "@/storage"
 import { SyncEvent } from "../sync"
-import { Log } from "../util/log"
-import { Session } from "."
+import { Log } from "../util"
+import * as Session from "./session"
 import { MessageV2 } from "./message-v2"
 import { SessionID, MessageID, PartID } from "./schema"
 import { SessionRunState } from "./run-state"
 import { SessionSummary } from "./summary"
-import { SessionStatus } from "./status"
 
 export namespace SessionRevert {
   const log = Log.create({ service: "session.revert" })
@@ -29,7 +27,7 @@ export namespace SessionRevert {
     readonly cleanup: (session: Session.Info) => Effect.Effect<void>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionRevert") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRevert") {}
 
   export const layer = Layer.effect(
     Service,
@@ -160,18 +158,4 @@ export namespace SessionRevert {
       Layer.provide(SessionSummary.defaultLayer),
     ),
   )
-
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
-  export async function revert(input: RevertInput) {
-    return runPromise((svc) => svc.revert(input))
-  }
-
-  export async function unrevert(input: { sessionID: SessionID }) {
-    return runPromise((svc) => svc.unrevert(input))
-  }
-
-  export async function cleanup(session: Session.Info) {
-    return runPromise((svc) => svc.cleanup(session))
-  }
 }

packages/opencode/src/session/run-state.ts

@@ -1,8 +1,7 @@
-import { InstanceState } from "@/effect/instance-state"
-import { Runner } from "@/effect/runner"
-import { makeRuntime } from "@/effect/run-service"
-import { Effect, Layer, Scope, ServiceMap } from "effect"
-import { Session } from "."
+import { InstanceState } from "@/effect"
+import { Runner } from "@/effect"
+import { Effect, Layer, Scope, Context } from "effect"
+import * as Session from "./session"
 import { MessageV2 } from "./message-v2"
 import { SessionID } from "./schema"
 import { SessionStatus } from "./status"
@@ -23,7 +22,7 @@ export namespace SessionRunState {
     ) => Effect.Effect<MessageV2.WithParts>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionRunState") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRunState") {}
 
   export const layer = Layer.effect(
     Service,
@@ -33,7 +32,7 @@ export namespace SessionRunState {
       const state = yield* InstanceState.make(
         Effect.fn("SessionRunState.state")(function* () {
           const scope = yield* Scope.Scope
-          const runners = new Map<SessionID, Runner<MessageV2.WithParts>>()
+          const runners = new Map<SessionID, Runner.Runner<MessageV2.WithParts>>()
           yield* Effect.addFinalizer(
             Effect.fnUntraced(function* () {
               yield* Effect.forEach(runners.values(), (runner) => runner.cancel, {
@@ -106,9 +105,4 @@ export namespace SessionRunState {
   )
 
   export const defaultLayer = layer.pipe(Layer.provide(SessionStatus.defaultLayer))
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
-  export async function assertNotBusy(sessionID: SessionID) {
-    return runPromise((svc) => svc.assertNotBusy(sessionID))
-  }
 }

packages/opencode/src/session/schema.ts

@@ -2,35 +2,33 @@ import { Schema } from "effect"
 import z from "zod"
 
 import { Identifier } from "@/id/id"
+import { ZodOverride } from "@/util/effect-zod"
 import { withStatics } from "@/util/schema"
 
-export const SessionID = Schema.String.pipe(
+export const SessionID = Schema.String.annotate({ [ZodOverride]: Identifier.schema("session") }).pipe(
   Schema.brand("SessionID"),
   withStatics((s) => ({
-    make: (id: string) => s.makeUnsafe(id),
-    descending: (id?: string) => s.makeUnsafe(Identifier.descending("session", id)),
+    descending: (id?: string) => s.make(Identifier.descending("session", id)),
     zod: Identifier.schema("session").pipe(z.custom<Schema.Schema.Type<typeof s>>()),
   })),
 )
 
 export type SessionID = Schema.Schema.Type<typeof SessionID>
 
-export const MessageID = Schema.String.pipe(
+export const MessageID = Schema.String.annotate({ [ZodOverride]: Identifier.schema("message") }).pipe(
   Schema.brand("MessageID"),
   withStatics((s) => ({
-    make: (id: string) => s.makeUnsafe(id),
-    ascending: (id?: string) => s.makeUnsafe(Identifier.ascending("message", id)),
+    ascending: (id?: string) => s.make(Identifier.ascending("message", id)),
     zod: Identifier.schema("message").pipe(z.custom<Schema.Schema.Type<typeof s>>()),
   })),
 )
 
 export type MessageID = Schema.Schema.Type<typeof MessageID>
 
-export const PartID = Schema.String.pipe(
+export const PartID = Schema.String.annotate({ [ZodOverride]: Identifier.schema("part") }).pipe(
   Schema.brand("PartID"),
   withStatics((s) => ({
-    make: (id: string) => s.makeUnsafe(id),
-    ascending: (id?: string) => s.makeUnsafe(Identifier.ascending("part", id)),
+    ascending: (id?: string) => s.make(Identifier.ascending("part", id)),
     zod: Identifier.schema("part").pipe(z.custom<Schema.Schema.Type<typeof s>>()),
   })),
 )

packages/opencode/src/session/session.sql.ts

@@ -1,6 +1,7 @@
 import { sqliteTable, text, integer, index, primaryKey } from "drizzle-orm/sqlite-core"
 import { ProjectTable } from "../project/project.sql"
 import type { MessageV2 } from "./message-v2"
+import type { SessionEntry } from "../v2/session-entry"
 import type { Snapshot } from "../snapshot"
 import type { Permission } from "../permission"
 import type { ProjectID } from "../project/schema"
@@ -94,6 +95,25 @@ export const TodoTable = sqliteTable(
   ],
 )
 
+export const SessionEntryTable = sqliteTable(
+  "session_entry",
+  {
+    id: text().$type<SessionEntry.ID>().primaryKey(),
+    session_id: text()
+      .$type<SessionID>()
+      .notNull()
+      .references(() => SessionTable.id, { onDelete: "cascade" }),
+    type: text().$type<SessionEntry.Type>().notNull(),
+    ...Timestamps,
+    data: text({ mode: "json" }).notNull().$type<Omit<SessionEntry.Entry, "type" | "id">>(),
+  },
+  (table) => [
+    index("session_entry_session_idx").on(table.session_id),
+    index("session_entry_session_type_idx").on(table.session_id, table.type),
+    index("session_entry_time_created_idx").on(table.time_created),
+  ],
+)
+
 export const PermissionTable = sqliteTable("permission", {
   project_id: text()
     .primaryKey()

packages/opencode/src/session/session.ts

@@ -0,0 +1,809 @@
+import { Slug } from "@opencode-ai/shared/util/slug"
+import path from "path"
+import { BusEvent } from "@/bus/bus-event"
+import { Bus } from "@/bus"
+import { Decimal } from "decimal.js"
+import z from "zod"
+import { type ProviderMetadata, type LanguageModelUsage } from "ai"
+import { Flag } from "../flag/flag"
+import { Installation } from "../installation"
+import { InstallationVersion } from "../installation/version"
+
+import { Database, NotFoundError, eq, and, gte, isNull, desc, like, inArray, lt } from "../storage"
+import { SyncEvent } from "../sync"
+import type { SQL } from "../storage"
+import { PartTable, SessionTable } from "./session.sql"
+import { ProjectTable } from "../project/project.sql"
+import { Storage } from "@/storage"
+import { Log } from "../util"
+import { updateSchema } from "../util/update-schema"
+import { MessageV2 } from "./message-v2"
+import { Instance } from "../project/instance"
+import { InstanceState } from "@/effect"
+import { Snapshot } from "@/snapshot"
+import { ProjectID } from "../project/schema"
+import { WorkspaceID } from "../control-plane/schema"
+import { SessionID, MessageID, PartID } from "./schema"
+
+import type { Provider } from "@/provider"
+import { Permission } from "@/permission"
+import { Global } from "@/global"
+import { Effect, Layer, Option, Context } from "effect"
+
+const log = Log.create({ service: "session" })
+
+const parentTitlePrefix = "New session - "
+const childTitlePrefix = "Child session - "
+
+function createDefaultTitle(isChild = false) {
+  return (isChild ? childTitlePrefix : parentTitlePrefix) + new Date().toISOString()
+}
+
+export function isDefaultTitle(title: string) {
+  return new RegExp(
+    `^(${parentTitlePrefix}|${childTitlePrefix})\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z$`,
+  ).test(title)
+}
+
+type SessionRow = typeof SessionTable.$inferSelect
+
+export function fromRow(row: SessionRow): Info {
+  const summary =
+    row.summary_additions !== null || row.summary_deletions !== null || row.summary_files !== null
+      ? {
+          additions: row.summary_additions ?? 0,
+          deletions: row.summary_deletions ?? 0,
+          files: row.summary_files ?? 0,
+          diffs: row.summary_diffs ?? undefined,
+        }
+      : undefined
+  const share = row.share_url ? { url: row.share_url } : undefined
+  const revert = row.revert ?? undefined
+  return {
+    id: row.id,
+    slug: row.slug,
+    projectID: row.project_id,
+    workspaceID: row.workspace_id ?? undefined,
+    directory: row.directory,
+    parentID: row.parent_id ?? undefined,
+    title: row.title,
+    version: row.version,
+    summary,
+    share,
+    revert,
+    permission: row.permission ?? undefined,
+    time: {
+      created: row.time_created,
+      updated: row.time_updated,
+      compacting: row.time_compacting ?? undefined,
+      archived: row.time_archived ?? undefined,
+    },
+  }
+}
+
+export function toRow(info: Info) {
+  return {
+    id: info.id,
+    project_id: info.projectID,
+    workspace_id: info.workspaceID,
+    parent_id: info.parentID,
+    slug: info.slug,
+    directory: info.directory,
+    title: info.title,
+    version: info.version,
+    share_url: info.share?.url,
+    summary_additions: info.summary?.additions,
+    summary_deletions: info.summary?.deletions,
+    summary_files: info.summary?.files,
+    summary_diffs: info.summary?.diffs,
+    revert: info.revert ?? null,
+    permission: info.permission,
+    time_created: info.time.created,
+    time_updated: info.time.updated,
+    time_compacting: info.time.compacting,
+    time_archived: info.time.archived,
+  }
+}
+
+function getForkedTitle(title: string): string {
+  const match = title.match(/^(.+) \(fork #(\d+)\)$/)
+  if (match) {
+    const base = match[1]
+    const num = parseInt(match[2], 10)
+    return `${base} (fork #${num + 1})`
+  }
+  return `${title} (fork #1)`
+}
+
+export const Info = z
+  .object({
+    id: SessionID.zod,
+    slug: z.string(),
+    projectID: ProjectID.zod,
+    workspaceID: WorkspaceID.zod.optional(),
+    directory: z.string(),
+    parentID: SessionID.zod.optional(),
+    summary: z
+      .object({
+        additions: z.number(),
+        deletions: z.number(),
+        files: z.number(),
+        diffs: Snapshot.FileDiff.array().optional(),
+      })
+      .optional(),
+    share: z
+      .object({
+        url: z.string(),
+      })
+      .optional(),
+    title: z.string(),
+    version: z.string(),
+    time: z.object({
+      created: z.number(),
+      updated: z.number(),
+      compacting: z.number().optional(),
+      archived: z.number().optional(),
+    }),
+    permission: Permission.Ruleset.zod.optional(),
+    revert: z
+      .object({
+        messageID: MessageID.zod,
+        partID: PartID.zod.optional(),
+        snapshot: z.string().optional(),
+        diff: z.string().optional(),
+      })
+      .optional(),
+  })
+  .meta({
+    ref: "Session",
+  })
+export type Info = z.output<typeof Info>
+
+export const ProjectInfo = z
+  .object({
+    id: ProjectID.zod,
+    name: z.string().optional(),
+    worktree: z.string(),
+  })
+  .meta({
+    ref: "ProjectSummary",
+  })
+export type ProjectInfo = z.output<typeof ProjectInfo>
+
+export const GlobalInfo = Info.extend({
+  project: ProjectInfo.nullable(),
+}).meta({
+  ref: "GlobalSession",
+})
+export type GlobalInfo = z.output<typeof GlobalInfo>
+
+export const CreateInput = z
+  .object({
+    parentID: SessionID.zod.optional(),
+    title: z.string().optional(),
+    permission: Info.shape.permission,
+    workspaceID: WorkspaceID.zod.optional(),
+  })
+  .optional()
+export type CreateInput = z.output<typeof CreateInput>
+
+export const ForkInput = z.object({ sessionID: SessionID.zod, messageID: MessageID.zod.optional() })
+export const GetInput = SessionID.zod
+export const ChildrenInput = SessionID.zod
+export const RemoveInput = SessionID.zod
+export const SetTitleInput = z.object({ sessionID: SessionID.zod, title: z.string() })
+export const SetArchivedInput = z.object({ sessionID: SessionID.zod, time: z.number().optional() })
+export const SetPermissionInput = z.object({ sessionID: SessionID.zod, permission: Permission.Ruleset.zod })
+export const SetRevertInput = z.object({
+  sessionID: SessionID.zod,
+  revert: Info.shape.revert,
+  summary: Info.shape.summary,
+})
+export const MessagesInput = z.object({ sessionID: SessionID.zod, limit: z.number().optional() })
+
+export const Event = {
+  Created: SyncEvent.define({
+    type: "session.created",
+    version: 1,
+    aggregate: "sessionID",
+    schema: z.object({
+      sessionID: SessionID.zod,
+      info: Info,
+    }),
+  }),
+  Updated: SyncEvent.define({
+    type: "session.updated",
+    version: 1,
+    aggregate: "sessionID",
+    schema: z.object({
+      sessionID: SessionID.zod,
+      info: updateSchema(Info).extend({
+        share: updateSchema(Info.shape.share.unwrap()).optional(),
+        time: updateSchema(Info.shape.time).optional(),
+      }),
+    }),
+    busSchema: z.object({
+      sessionID: SessionID.zod,
+      info: Info,
+    }),
+  }),
+  Deleted: SyncEvent.define({
+    type: "session.deleted",
+    version: 1,
+    aggregate: "sessionID",
+    schema: z.object({
+      sessionID: SessionID.zod,
+      info: Info,
+    }),
+  }),
+  Diff: BusEvent.define(
+    "session.diff",
+    z.object({
+      sessionID: SessionID.zod,
+      diff: Snapshot.FileDiff.array(),
+    }),
+  ),
+  Error: BusEvent.define(
+    "session.error",
+    z.object({
+      sessionID: SessionID.zod.optional(),
+      error: MessageV2.Assistant.shape.error,
+    }),
+  ),
+}
+
+export function plan(input: { slug: string; time: { created: number } }) {
+  const base = Instance.project.vcs
+    ? path.join(Instance.worktree, ".opencode", "plans")
+    : path.join(Global.Path.data, "plans")
+  return path.join(base, [input.time.created, input.slug].join("-") + ".md")
+}
+
+export const getUsage = (input: { model: Provider.Model; usage: LanguageModelUsage; metadata?: ProviderMetadata }) => {
+  const safe = (value: number) => {
+    if (!Number.isFinite(value)) return 0
+    return value
+  }
+  const inputTokens = safe(input.usage.inputTokens ?? 0)
+  const outputTokens = safe(input.usage.outputTokens ?? 0)
+  const reasoningTokens = safe(input.usage.outputTokenDetails?.reasoningTokens ?? input.usage.reasoningTokens ?? 0)
+
+  const cacheReadInputTokens = safe(
+    input.usage.inputTokenDetails?.cacheReadTokens ?? input.usage.cachedInputTokens ?? 0,
+  )
+  const cacheWriteInputTokens = safe(
+    (input.usage.inputTokenDetails?.cacheWriteTokens ??
+      input.metadata?.["anthropic"]?.["cacheCreationInputTokens"] ??
+      // google-vertex-anthropic returns metadata under "vertex" key
+      // (AnthropicMessagesLanguageModel custom provider key from 'vertex.anthropic.messages')
+      input.metadata?.["vertex"]?.["cacheCreationInputTokens"] ??
+      // @ts-expect-error
+      input.metadata?.["bedrock"]?.["usage"]?.["cacheWriteInputTokens"] ??
+      // @ts-expect-error
+      input.metadata?.["venice"]?.["usage"]?.["cacheCreationInputTokens"] ??
+      0) as number,
+  )
+
+  // AI SDK v6 normalized inputTokens to include cached tokens across all providers
+  // (including Anthropic/Bedrock which previously excluded them). Always subtract cache
+  // tokens to get the non-cached input count for separate cost calculation.
+  const adjustedInputTokens = safe(inputTokens - cacheReadInputTokens - cacheWriteInputTokens)
+
+  const total = input.usage.totalTokens
+
+  const tokens = {
+    total,
+    input: adjustedInputTokens,
+    output: safe(outputTokens - reasoningTokens),
+    reasoning: reasoningTokens,
+    cache: {
+      write: cacheWriteInputTokens,
+      read: cacheReadInputTokens,
+    },
+  }
+
+  const costInfo =
+    input.model.cost?.experimentalOver200K && tokens.input + tokens.cache.read > 200_000
+      ? input.model.cost.experimentalOver200K
+      : input.model.cost
+  return {
+    cost: safe(
+      new Decimal(0)
+        .add(new Decimal(tokens.input).mul(costInfo?.input ?? 0).div(1_000_000))
+        .add(new Decimal(tokens.output).mul(costInfo?.output ?? 0).div(1_000_000))
+        .add(new Decimal(tokens.cache.read).mul(costInfo?.cache?.read ?? 0).div(1_000_000))
+        .add(new Decimal(tokens.cache.write).mul(costInfo?.cache?.write ?? 0).div(1_000_000))
+        // TODO: update models.dev to have better pricing model, for now:
+        // charge reasoning tokens at the same rate as output tokens
+        .add(new Decimal(tokens.reasoning).mul(costInfo?.output ?? 0).div(1_000_000))
+        .toNumber(),
+    ),
+    tokens,
+  }
+}
+
+export class BusyError extends Error {
+  constructor(public readonly sessionID: string) {
+    super(`Session ${sessionID} is busy`)
+  }
+}
+
+export interface Interface {
+  readonly create: (input?: {
+    parentID?: SessionID
+    title?: string
+    permission?: Permission.Ruleset
+    workspaceID?: WorkspaceID
+  }) => Effect.Effect<Info>
+  readonly fork: (input: { sessionID: SessionID; messageID?: MessageID }) => Effect.Effect<Info>
+  readonly touch: (sessionID: SessionID) => Effect.Effect<void>
+  readonly get: (id: SessionID) => Effect.Effect<Info>
+  readonly setTitle: (input: { sessionID: SessionID; title: string }) => Effect.Effect<void>
+  readonly setArchived: (input: { sessionID: SessionID; time?: number }) => Effect.Effect<void>
+  readonly setPermission: (input: { sessionID: SessionID; permission: Permission.Ruleset }) => Effect.Effect<void>
+  readonly setRevert: (input: {
+    sessionID: SessionID
+    revert: Info["revert"]
+    summary: Info["summary"]
+  }) => Effect.Effect<void>
+  readonly clearRevert: (sessionID: SessionID) => Effect.Effect<void>
+  readonly setSummary: (input: { sessionID: SessionID; summary: Info["summary"] }) => Effect.Effect<void>
+  readonly diff: (sessionID: SessionID) => Effect.Effect<Snapshot.FileDiff[]>
+  readonly messages: (input: { sessionID: SessionID; limit?: number }) => Effect.Effect<MessageV2.WithParts[]>
+  readonly children: (parentID: SessionID) => Effect.Effect<Info[]>
+  readonly remove: (sessionID: SessionID) => Effect.Effect<void>
+  readonly updateMessage: <T extends MessageV2.Info>(msg: T) => Effect.Effect<T>
+  readonly removeMessage: (input: { sessionID: SessionID; messageID: MessageID }) => Effect.Effect<MessageID>
+  readonly removePart: (input: { sessionID: SessionID; messageID: MessageID; partID: PartID }) => Effect.Effect<PartID>
+  readonly getPart: (input: {
+    sessionID: SessionID
+    messageID: MessageID
+    partID: PartID
+  }) => Effect.Effect<MessageV2.Part | undefined>
+  readonly updatePart: <T extends MessageV2.Part>(part: T) => Effect.Effect<T>
+  readonly updatePartDelta: (input: {
+    sessionID: SessionID
+    messageID: MessageID
+    partID: PartID
+    field: string
+    delta: string
+  }) => Effect.Effect<void>
+  /** Finds the first message matching the predicate, searching newest-first. */
+  readonly findMessage: (
+    sessionID: SessionID,
+    predicate: (msg: MessageV2.WithParts) => boolean,
+  ) => Effect.Effect<Option.Option<MessageV2.WithParts>>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/Session") {}
+
+type Patch = z.infer<typeof Event.Updated.schema>["info"]
+
+const db = <T>(fn: (d: Parameters<typeof Database.use>[0] extends (trx: infer D) => any ? D : never) => T) =>
+  Effect.sync(() => Database.use(fn))
+
+export const layer: Layer.Layer<Service, never, Bus.Service | Storage.Service> = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    const bus = yield* Bus.Service
+    const storage = yield* Storage.Service
+
+    const createNext = Effect.fn("Session.createNext")(function* (input: {
+      id?: SessionID
+      title?: string
+      parentID?: SessionID
+      workspaceID?: WorkspaceID
+      directory: string
+      permission?: Permission.Ruleset
+    }) {
+      const ctx = yield* InstanceState.context
+      const result: Info = {
+        id: SessionID.descending(input.id),
+        slug: Slug.create(),
+        version: InstallationVersion,
+        projectID: ctx.project.id,
+        directory: input.directory,
+        workspaceID: input.workspaceID,
+        parentID: input.parentID,
+        title: input.title ?? createDefaultTitle(!!input.parentID),
+        permission: input.permission,
+        time: {
+          created: Date.now(),
+          updated: Date.now(),
+        },
+      }
+      log.info("created", result)
+
+      yield* Effect.sync(() => SyncEvent.run(Event.Created, { sessionID: result.id, info: result }))
+
+      if (!Flag.OPENCODE_EXPERIMENTAL_WORKSPACES) {
+        // This only exist for backwards compatibility. We should not be
+        // manually publishing this event; it is a sync event now
+        yield* bus.publish(Event.Updated, {
+          sessionID: result.id,
+          info: result,
+        })
+      }
+
+      return result
+    })
+
+    const get = Effect.fn("Session.get")(function* (id: SessionID) {
+      const row = yield* db((d) => d.select().from(SessionTable).where(eq(SessionTable.id, id)).get())
+      if (!row) throw new NotFoundError({ message: `Session not found: ${id}` })
+      return fromRow(row)
+    })
+
+    const children = Effect.fn("Session.children")(function* (parentID: SessionID) {
+      const rows = yield* db((d) =>
+        d
+          .select()
+          .from(SessionTable)
+          .where(and(eq(SessionTable.parent_id, parentID)))
+          .all(),
+      )
+      return rows.map(fromRow)
+    })
+
+    const remove: Interface["remove"] = Effect.fnUntraced(function* (sessionID: SessionID) {
+      try {
+        const session = yield* get(sessionID)
+        const kids = yield* children(sessionID)
+        for (const child of kids) {
+          yield* remove(child.id)
+        }
+
+        // `remove` needs to work in all cases, such as a broken
+        // sessions that run cleanup. In certain cases these will
+        // run without any instance state, so we need to turn off
+        // publishing of events in that case
+        const hasInstance = yield* InstanceState.directory.pipe(
+          Effect.as(true),
+          Effect.catchCause(() => Effect.succeed(false)),
+        )
+
+        yield* Effect.sync(() => {
+          SyncEvent.run(Event.Deleted, { sessionID, info: session }, { publish: hasInstance })
+          SyncEvent.remove(sessionID)
+        })
+      } catch (e) {
+        log.error(e)
+      }
+    })
+
+    const updateMessage = <T extends MessageV2.Info>(msg: T): Effect.Effect<T> =>
+      Effect.gen(function* () {
+        yield* Effect.sync(() => SyncEvent.run(MessageV2.Event.Updated, { sessionID: msg.sessionID, info: msg }))
+        return msg
+      }).pipe(Effect.withSpan("Session.updateMessage"))
+
+    const updatePart = <T extends MessageV2.Part>(part: T): Effect.Effect<T> =>
+      Effect.gen(function* () {
+        yield* Effect.sync(() =>
+          SyncEvent.run(MessageV2.Event.PartUpdated, {
+            sessionID: part.sessionID,
+            part: structuredClone(part),
+            time: Date.now(),
+          }),
+        )
+        return part
+      }).pipe(Effect.withSpan("Session.updatePart"))
+
+    const getPart: Interface["getPart"] = Effect.fn("Session.getPart")(function* (input) {
+      const row = Database.use((db) =>
+        db
+          .select()
+          .from(PartTable)
+          .where(
+            and(
+              eq(PartTable.session_id, input.sessionID),
+              eq(PartTable.message_id, input.messageID),
+              eq(PartTable.id, input.partID),
+            ),
+          )
+          .get(),
+      )
+      if (!row) return
+      return {
+        ...row.data,
+        id: row.id,
+        sessionID: row.session_id,
+        messageID: row.message_id,
+      } as MessageV2.Part
+    })
+
+    const create = Effect.fn("Session.create")(function* (input?: {
+      parentID?: SessionID
+      title?: string
+      permission?: Permission.Ruleset
+      workspaceID?: WorkspaceID
+    }) {
+      const directory = yield* InstanceState.directory
+      return yield* createNext({
+        parentID: input?.parentID,
+        directory,
+        title: input?.title,
+        permission: input?.permission,
+        workspaceID: input?.workspaceID,
+      })
+    })
+
+    const fork = Effect.fn("Session.fork")(function* (input: { sessionID: SessionID; messageID?: MessageID }) {
+      const directory = yield* InstanceState.directory
+      const original = yield* get(input.sessionID)
+      const title = getForkedTitle(original.title)
+      const session = yield* createNext({
+        directory,
+        workspaceID: original.workspaceID,
+        title,
+      })
+      const msgs = yield* messages({ sessionID: input.sessionID })
+      const idMap = new Map<string, MessageID>()
+
+      for (const msg of msgs) {
+        if (input.messageID && msg.info.id >= input.messageID) break
+        const newID = MessageID.ascending()
+        idMap.set(msg.info.id, newID)
+
+        const parentID = msg.info.role === "assistant" && msg.info.parentID ? idMap.get(msg.info.parentID) : undefined
+        const cloned = yield* updateMessage({
+          ...msg.info,
+          sessionID: session.id,
+          id: newID,
+          ...(parentID && { parentID }),
+        })
+
+        for (const part of msg.parts) {
+          yield* updatePart({
+            ...part,
+            id: PartID.ascending(),
+            messageID: cloned.id,
+            sessionID: session.id,
+          })
+        }
+      }
+      return session
+    })
+
+    const patch = (sessionID: SessionID, info: Patch) =>
+      Effect.sync(() => SyncEvent.run(Event.Updated, { sessionID, info }))
+
+    const touch = Effect.fn("Session.touch")(function* (sessionID: SessionID) {
+      yield* patch(sessionID, { time: { updated: Date.now() } })
+    })
+
+    const setTitle = Effect.fn("Session.setTitle")(function* (input: { sessionID: SessionID; title: string }) {
+      yield* patch(input.sessionID, { title: input.title })
+    })
+
+    const setArchived = Effect.fn("Session.setArchived")(function* (input: { sessionID: SessionID; time?: number }) {
+      yield* patch(input.sessionID, { time: { archived: input.time } })
+    })
+
+    const setPermission = Effect.fn("Session.setPermission")(function* (input: {
+      sessionID: SessionID
+      permission: Permission.Ruleset
+    }) {
+      yield* patch(input.sessionID, { permission: input.permission, time: { updated: Date.now() } })
+    })
+
+    const setRevert = Effect.fn("Session.setRevert")(function* (input: {
+      sessionID: SessionID
+      revert: Info["revert"]
+      summary: Info["summary"]
+    }) {
+      yield* patch(input.sessionID, { summary: input.summary, time: { updated: Date.now() }, revert: input.revert })
+    })
+
+    const clearRevert = Effect.fn("Session.clearRevert")(function* (sessionID: SessionID) {
+      yield* patch(sessionID, { time: { updated: Date.now() }, revert: null })
+    })
+
+    const setSummary = Effect.fn("Session.setSummary")(function* (input: {
+      sessionID: SessionID
+      summary: Info["summary"]
+    }) {
+      yield* patch(input.sessionID, { time: { updated: Date.now() }, summary: input.summary })
+    })
+
+    const diff = Effect.fn("Session.diff")(function* (sessionID: SessionID) {
+      return yield* storage
+        .read<Snapshot.FileDiff[]>(["session_diff", sessionID])
+        .pipe(Effect.orElseSucceed((): Snapshot.FileDiff[] => []))
+    })
+
+    const messages = Effect.fn("Session.messages")(function* (input: { sessionID: SessionID; limit?: number }) {
+      if (input.limit) {
+        return MessageV2.page({ sessionID: input.sessionID, limit: input.limit }).items
+      }
+      return Array.from(MessageV2.stream(input.sessionID)).reverse()
+    })
+
+    const removeMessage = Effect.fn("Session.removeMessage")(function* (input: {
+      sessionID: SessionID
+      messageID: MessageID
+    }) {
+      yield* Effect.sync(() =>
+        SyncEvent.run(MessageV2.Event.Removed, {
+          sessionID: input.sessionID,
+          messageID: input.messageID,
+        }),
+      )
+      return input.messageID
+    })
+
+    const removePart = Effect.fn("Session.removePart")(function* (input: {
+      sessionID: SessionID
+      messageID: MessageID
+      partID: PartID
+    }) {
+      yield* Effect.sync(() =>
+        SyncEvent.run(MessageV2.Event.PartRemoved, {
+          sessionID: input.sessionID,
+          messageID: input.messageID,
+          partID: input.partID,
+        }),
+      )
+      return input.partID
+    })
+
+    const updatePartDelta = Effect.fn("Session.updatePartDelta")(function* (input: {
+      sessionID: SessionID
+      messageID: MessageID
+      partID: PartID
+      field: string
+      delta: string
+    }) {
+      yield* bus.publish(MessageV2.Event.PartDelta, input)
+    })
+
+    /** Finds the first message matching the predicate, searching newest-first. */
+    const findMessage = Effect.fn("Session.findMessage")(function* (
+      sessionID: SessionID,
+      predicate: (msg: MessageV2.WithParts) => boolean,
+    ) {
+      for (const item of MessageV2.stream(sessionID)) {
+        if (predicate(item)) return Option.some(item)
+      }
+      return Option.none<MessageV2.WithParts>()
+    })
+
+    return Service.of({
+      create,
+      fork,
+      touch,
+      get,
+      setTitle,
+      setArchived,
+      setPermission,
+      setRevert,
+      clearRevert,
+      setSummary,
+      diff,
+      messages,
+      children,
+      remove,
+      updateMessage,
+      removeMessage,
+      removePart,
+      updatePart,
+      getPart,
+      updatePartDelta,
+      findMessage,
+    })
+  }),
+)
+
+export const defaultLayer = layer.pipe(Layer.provide(Bus.layer), Layer.provide(Storage.defaultLayer))
+
+export function* list(input?: {
+  directory?: string
+  workspaceID?: WorkspaceID
+  roots?: boolean
+  start?: number
+  search?: string
+  limit?: number
+}) {
+  const project = Instance.project
+  const conditions = [eq(SessionTable.project_id, project.id)]
+
+  if (input?.workspaceID) {
+    conditions.push(eq(SessionTable.workspace_id, input.workspaceID))
+  }
+  if (input?.directory) {
+    conditions.push(eq(SessionTable.directory, input.directory))
+  }
+  if (input?.roots) {
+    conditions.push(isNull(SessionTable.parent_id))
+  }
+  if (input?.start) {
+    conditions.push(gte(SessionTable.time_updated, input.start))
+  }
+  if (input?.search) {
+    conditions.push(like(SessionTable.title, `%${input.search}%`))
+  }
+
+  const limit = input?.limit ?? 100
+
+  const rows = Database.use((db) =>
+    db
+      .select()
+      .from(SessionTable)
+      .where(and(...conditions))
+      .orderBy(desc(SessionTable.time_updated))
+      .limit(limit)
+      .all(),
+  )
+  for (const row of rows) {
+    yield fromRow(row)
+  }
+}
+
+export function* listGlobal(input?: {
+  directory?: string
+  roots?: boolean
+  start?: number
+  cursor?: number
+  search?: string
+  limit?: number
+  archived?: boolean
+}) {
+  const conditions: SQL[] = []
+
+  if (input?.directory) {
+    conditions.push(eq(SessionTable.directory, input.directory))
+  }
+  if (input?.roots) {
+    conditions.push(isNull(SessionTable.parent_id))
+  }
+  if (input?.start) {
+    conditions.push(gte(SessionTable.time_updated, input.start))
+  }
+  if (input?.cursor) {
+    conditions.push(lt(SessionTable.time_updated, input.cursor))
+  }
+  if (input?.search) {
+    conditions.push(like(SessionTable.title, `%${input.search}%`))
+  }
+  if (!input?.archived) {
+    conditions.push(isNull(SessionTable.time_archived))
+  }
+
+  const limit = input?.limit ?? 100
+
+  const rows = Database.use((db) => {
+    const query =
+      conditions.length > 0
+        ? db
+            .select()
+            .from(SessionTable)
+            .where(and(...conditions))
+        : db.select().from(SessionTable)
+    return query.orderBy(desc(SessionTable.time_updated), desc(SessionTable.id)).limit(limit).all()
+  })
+
+  const ids = [...new Set(rows.map((row) => row.project_id))]
+  const projects = new Map<string, ProjectInfo>()
+
+  if (ids.length > 0) {
+    const items = Database.use((db) =>
+      db
+        .select({ id: ProjectTable.id, name: ProjectTable.name, worktree: ProjectTable.worktree })
+        .from(ProjectTable)
+        .where(inArray(ProjectTable.id, ids))
+        .all(),
+    )
+    for (const item of items) {
+      projects.set(item.id, {
+        id: item.id,
+        name: item.name ?? undefined,
+        worktree: item.worktree,
+      })
+    }
+  }
+
+  for (const row of rows) {
+    const project = projects.get(row.project_id) ?? null
+    yield { ...fromRow(row), project }
+  }
+}

packages/opencode/src/session/status.ts

@@ -1,9 +1,8 @@
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
-import { InstanceState } from "@/effect/instance-state"
-import { makeRuntime } from "@/effect/run-service"
+import { InstanceState } from "@/effect"
 import { SessionID } from "./schema"
-import { Effect, Layer, ServiceMap } from "effect"
+import { Effect, Layer, Context } from "effect"
 import z from "zod"
 
 export namespace SessionStatus {
@@ -50,7 +49,7 @@ export namespace SessionStatus {
     readonly set: (sessionID: SessionID, status: Info) => Effect.Effect<void>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionStatus") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionStatus") {}
 
   export const layer = Layer.effect(
     Service,
@@ -86,17 +85,4 @@ export namespace SessionStatus {
   )
 
   export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
-  export async function get(sessionID: SessionID) {
-    return runPromise((svc) => svc.get(sessionID))
-  }
-
-  export async function list() {
-    return runPromise((svc) => svc.list())
-  }
-
-  export async function set(sessionID: SessionID, status: Info) {
-    return runPromise((svc) => svc.set(sessionID, status))
-  }
 }

packages/opencode/src/session/summary.ts

@@ -1,10 +1,9 @@
 import z from "zod"
-import { Effect, Layer, ServiceMap } from "effect"
-import { makeRuntime } from "@/effect/run-service"
+import { Effect, Layer, Context } from "effect"
 import { Bus } from "@/bus"
 import { Snapshot } from "@/snapshot"
-import { Storage } from "@/storage/storage"
-import { Session } from "."
+import { Storage } from "@/storage"
+import * as Session from "./session"
 import { MessageV2 } from "./message-v2"
 import { SessionID, MessageID } from "./schema"
 
@@ -71,7 +70,7 @@ export namespace SessionSummary {
     readonly computeDiff: (input: { messages: MessageV2.WithParts[] }) => Effect.Effect<Snapshot.FileDiff[]>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionSummary") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionSummary") {}
 
   export const layer = Layer.effect(
     Service,
@@ -159,17 +158,8 @@ export namespace SessionSummary {
     ),
   )
 
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
-  export const summarize = (input: { sessionID: SessionID; messageID: MessageID }) =>
-    void runPromise((svc) => svc.summarize(input)).catch(() => {})
-
   export const DiffInput = z.object({
     sessionID: SessionID.zod,
     messageID: MessageID.zod.optional(),
   })
-
-  export async function diff(input: z.infer<typeof DiffInput>) {
-    return runPromise((svc) => svc.diff(input))
-  }
 }

packages/opencode/src/session/system.ts

@@ -1,4 +1,4 @@
-import { Ripgrep } from "../file/ripgrep"
+import { Context, Effect, Layer } from "effect"
 
 import { Instance } from "../project/instance"
 
@@ -11,7 +11,7 @@ import PROMPT_KIMI from "./prompt/kimi.txt"
 
 import PROMPT_CODEX from "./prompt/codex.txt"
 import PROMPT_TRINITY from "./prompt/trinity.txt"
-import type { Provider } from "@/provider/provider"
+import type { Provider } from "@/provider"
 import type { Agent } from "@/agent/agent"
 import { Permission } from "@/permission"
 import { Skill } from "@/skill"
@@ -33,44 +33,52 @@ export namespace SystemPrompt {
     return [PROMPT_DEFAULT]
   }
 
-  export async function environment(model: Provider.Model) {
-    const project = Instance.project
-    return [
-      [
-        `You are powered by the model named ${model.api.id}. The exact model ID is ${model.providerID}/${model.api.id}`,
-        `Here is some useful information about the environment you are running in:`,
-        `<env>`,
-        `  Working directory: ${Instance.directory}`,
-        `  Workspace root folder: ${Instance.worktree}`,
-        `  Is directory a git repo: ${project.vcs === "git" ? "yes" : "no"}`,
-        `  Platform: ${process.platform}`,
-        `  Today's date: ${new Date().toDateString()}`,
-        `</env>`,
-        `<directories>`,
-        `  ${
-          project.vcs === "git" && false
-            ? await Ripgrep.tree({
-                cwd: Instance.directory,
-                limit: 50,
-              })
-            : ""
-        }`,
-        `</directories>`,
-      ].join("\n"),
-    ]
+  export interface Interface {
+    readonly environment: (model: Provider.Model) => string[]
+    readonly skills: (agent: Agent.Info) => Effect.Effect<string | undefined>
   }
 
-  export async function skills(agent: Agent.Info) {
-    if (Permission.disabled(["skill"], agent.permission).has("skill")) return
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SystemPrompt") {}
 
-    const list = await Skill.available(agent)
+  export const layer = Layer.effect(
+    Service,
+    Effect.gen(function* () {
+      const skill = yield* Skill.Service
 
-    return [
-      "Skills provide specialized instructions and workflows for specific tasks.",
-      "Use the skill tool to load a skill when a task matches its description.",
-      // the agents seem to ingest the information about skills a bit better if we present a more verbose
-      // version of them here and a less verbose version in tool description, rather than vice versa.
-      Skill.fmt(list, { verbose: true }),
-    ].join("\n")
-  }
+      return Service.of({
+        environment(model) {
+          const project = Instance.project
+          return [
+            [
+              `You are powered by the model named ${model.api.id}. The exact model ID is ${model.providerID}/${model.api.id}`,
+              `Here is some useful information about the environment you are running in:`,
+              `<env>`,
+              `  Working directory: ${Instance.directory}`,
+              `  Workspace root folder: ${Instance.worktree}`,
+              `  Is directory a git repo: ${project.vcs === "git" ? "yes" : "no"}`,
+              `  Platform: ${process.platform}`,
+              `  Today's date: ${new Date().toDateString()}`,
+              `</env>`,
+            ].join("\n"),
+          ]
+        },
+
+        skills: Effect.fn("SystemPrompt.skills")(function* (agent: Agent.Info) {
+          if (Permission.disabled(["skill"], agent.permission).has("skill")) return
+
+          const list = yield* skill.available(agent)
+
+          return [
+            "Skills provide specialized instructions and workflows for specific tasks.",
+            "Use the skill tool to load a skill when a task matches its description.",
+            // the agents seem to ingest the information about skills a bit better if we present a more verbose
+            // version of them here and a less verbose version in tool description, rather than vice versa.
+            Skill.fmt(list, { verbose: true }),
+          ].join("\n")
+        }),
+      })
+    }),
+  )
+
+  export const defaultLayer = layer.pipe(Layer.provide(Skill.defaultLayer))
 }

packages/opencode/src/session/todo.ts

@@ -1,10 +1,9 @@
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
-import { makeRuntime } from "@/effect/run-service"
 import { SessionID } from "./schema"
-import { Effect, Layer, ServiceMap } from "effect"
+import { Effect, Layer, Context } from "effect"
 import z from "zod"
-import { Database, eq, asc } from "../storage/db"
+import { Database, eq, asc } from "../storage"
 import { TodoTable } from "./session.sql"
 
 export namespace Todo {
@@ -32,7 +31,7 @@ export namespace Todo {
     readonly get: (sessionID: SessionID) => Effect.Effect<Info[]>
   }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionTodo") {}
+  export class Service extends Context.Service<Service, Interface>()("@opencode/SessionTodo") {}
 
   export const layer = Layer.effect(
     Service,
@@ -83,9 +82,4 @@ export namespace Todo {
   )
 
   export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
-  const { runPromise } = makeRuntime(Service, defaultLayer)
-
-  export async function get(sessionID: SessionID) {
-    return runPromise((svc) => svc.get(sessionID))
-  }
 }