From 1f77f4a4ed721f0098f45ee53dafbacdf2ed2d71 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 28 Aug 2026 12:34:01 -0400 Subject: [PATCH] refactor(core): inline auxiliary model operations (#45974) --- packages/core/src/session/compaction.ts | 379 ++++++++++++------------ packages/core/src/session/title.ts | 230 +++++++------- 2 files changed, 290 insertions(+), 319 deletions(-) diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index 60259d8ea6a..ee634cd05de 100644 --- a/packages/core/src/session/compaction.ts +++ b/packages/core/src/session/compaction.ts @@ -1,7 +1,6 @@ export * as SessionCompaction from "./compaction.js" -import { LLMClient, AIError, LLMEvent, Message, type LLMRequest } from "@opencode-ai/ai" -import type { StreamOptions } from "@opencode-ai/ai/route" +import { LLMClient, LLMEvent, Message } from "@opencode-ai/ai" import { SessionError } from "@opencode-ai/schema/session-error" import { Context, Effect, Layer, Stream } from "effect" import { Bus } from "../bus.js" @@ -65,13 +64,6 @@ export type Draft = { configure: (settings: Partial) => void } -type Dependencies = { - readonly bus: Bus.Interface - readonly llm: { - readonly stream: (request: LLMRequest, options?: StreamOptions) => Stream.Stream - } -} - export type AutoInput = { readonly session: SessionSchema.Info readonly messages: readonly SessionMessage.Info[] @@ -240,195 +232,196 @@ const planContent = (messages: readonly SessionMessage.Info[], tokens: number) = } } -const make = (dependencies: Dependencies) => { - const state = State.create({ - name: "session-compaction", - initial: () => ({ auto: true, buffer: DEFAULT_BUFFER, tokens: DEFAULT_KEEP_TOKENS }), - draft: (draft) => ({ - configure: (settings) => { - if (settings.auto !== undefined) draft.auto = settings.auto - if (settings.buffer !== undefined) draft.buffer = settings.buffer - if (settings.tokens !== undefined) draft.tokens = settings.tokens - }, - }), - }) - const failed = Effect.fnUntraced(function* (input: { - readonly sessionID: SessionSchema.ID - readonly reason: SessionMessage.Compaction["reason"] - readonly error: SessionError.Error - readonly inputID?: SessionMessage.ID - }) { - yield* dependencies.bus.publish(SessionEvent.Compaction.Failed, input) - return { status: "failed" as const, error: input.error } - }) - const execute = Effect.fn("SessionCompaction.execute")(function* (plan: Plan) { - if (!plan.started) - yield* dependencies.bus.publish(SessionEvent.Compaction.Started, { - sessionID: plan.session.id, - reason: plan.reason, - recent: plan.recent, - inputID: plan.inputID, - }) - - const chunks: string[] = [] - let failure: SessionError.Error | undefined - let usage: SessionUsage.Recorded | undefined - const recordUsage = Effect.suspend(() => - usage - ? dependencies.bus.publish(SessionEvent.UsageRecorded, { - sessionID: plan.session.id, - source: "compaction", - ...usage, - }) - : Effect.void, - ) - const prepared = yield* plan.prepare({ - scope: { session: plan.session, agentID: Agent.ID.make("compaction"), model: plan.resolved }, - transcript: { system: [], messages: [Message.user(plan.prompt)] }, - contextHooks: false, - }) - yield* dependencies.llm.stream(prepared.request, prepared.options).pipe( - Stream.runForEach((event) => { - if (LLMEvent.is.providerError(event)) - failure = { - type: event.classification === "context-overflow" ? "provider.invalid-request" : "provider.error", - message: event.message, - } - if (LLMEvent.is.textDelta(event)) { - chunks.push(event.text) - return dependencies.bus.publish(SessionEvent.Compaction.Delta, { - sessionID: plan.session.id, - text: event.text, - }) - } - if (LLMEvent.is.stepFinish(event)) { - const step = SessionUsage.record(event.usage, plan.resolved.cost) - usage = usage ? SessionUsage.add(usage, step) : step - } - return Effect.void - }), - Effect.catchTag("AI.Error", (error) => - Effect.sync(() => { - failure = toSessionError(error) - }), - ), - Effect.onInterrupt(() => - recordUsage.pipe( - Effect.andThen( - plan.reason === "auto" - ? failed({ - sessionID: plan.session.id, - reason: plan.reason, - error: { type: "compaction.interrupted", message: "Compaction was interrupted" }, - inputID: plan.inputID, - }).pipe(Effect.asVoid) - : Effect.void, - ), - ), - ), - ) - yield* recordUsage - const summary = chunks.join("") - if (failure || !summary.trim()) { - const error = failure ?? { type: "compaction.failed" as const, message: "Compaction produced no summary" } - return yield* failed({ - sessionID: plan.session.id, - reason: plan.reason, - error, - inputID: plan.inputID, - }) - } - yield* dependencies.bus.publish(SessionEvent.Compaction.Ended, { - sessionID: plan.session.id, - reason: plan.reason, - text: summary, - recent: plan.recent, - }) - return { status: "completed" as const } - }) - const compact = Effect.fn("SessionCompaction.compact")(function* (input: AutoInput) { - const content = planContent(input.messages, state.get().tokens) - if (content) - return yield* execute({ - session: input.session, - resolved: input.resolved, - prepare: input.prepare, - reason: "auto", - ...content, - }) - return yield* failed({ - sessionID: input.session.id, - reason: "auto", - error: { type: "compaction.unavailable", message: "Nothing to compact yet" }, - }) - }) - const required = (input: RequiredInput) => { - const config = state.get() - if (!config.auto) return false - const limit = input.resolved.limit - const context = limit.context - if (context <= 0) return false - const last = input.messages.findLast( - (message): message is SessionMessage.Assistant & { tokens: NonNullable } => - message.type === "assistant" && message.tokens !== undefined, - ) - if (!last) return false - const output = Math.min(limit.output, OUTPUT_TOKEN_MAX) - const promptCeiling = Math.min( - limit.input === undefined ? Number.POSITIVE_INFINITY : limit.input - config.buffer, - context - Math.max(output, config.buffer), - ) - const used = - last.tokens.input + last.tokens.output + last.tokens.reasoning + last.tokens.cache.read + last.tokens.cache.write - if (used <= 0) return false - return used >= promptCeiling - } - const compactManual = Effect.fn("SessionCompaction.compactManual")(function* (input: ManualInput) { - const content = planContent(input.messages, state.get().tokens) - if (!content) - return yield* failed({ - sessionID: input.session.id, - reason: "manual", - error: { type: "compaction.unavailable", message: "Nothing to compact yet" }, - inputID: input.inputID, - }) - const resolved = yield* input.resolveModel(input.session).pipe( - Effect.catch((cause) => - failed({ - sessionID: input.session.id, - reason: "manual", - error: toSessionError(cause), - inputID: input.inputID, - }), - ), - ) - if ("status" in resolved) return resolved - return yield* execute({ - session: input.session, - resolved, - prepare: input.prepare, - reason: "manual", - inputID: input.inputID, - started: input.started, - ...content, - }) - }) - return Service.of({ - transform: state.transform, - reload: state.reload, - enabled: () => state.get().auto, - required, - compact, - compactManual, - }) -} - export const layer = Layer.effect( Service, Effect.gen(function* () { const bus = yield* Bus.Service const llm = yield* LLMClient.Service - return make({ bus, llm }) + + const state = State.create({ + name: "session-compaction", + initial: () => ({ auto: true, buffer: DEFAULT_BUFFER, tokens: DEFAULT_KEEP_TOKENS }), + draft: (draft) => ({ + configure: (settings) => { + if (settings.auto !== undefined) draft.auto = settings.auto + if (settings.buffer !== undefined) draft.buffer = settings.buffer + if (settings.tokens !== undefined) draft.tokens = settings.tokens + }, + }), + }) + const failed = Effect.fnUntraced(function* (input: { + readonly sessionID: SessionSchema.ID + readonly reason: SessionMessage.Compaction["reason"] + readonly error: SessionError.Error + readonly inputID?: SessionMessage.ID + }) { + yield* bus.publish(SessionEvent.Compaction.Failed, input) + return { status: "failed" as const, error: input.error } + }) + const execute = Effect.fn("SessionCompaction.execute")(function* (plan: Plan) { + if (!plan.started) + yield* bus.publish(SessionEvent.Compaction.Started, { + sessionID: plan.session.id, + reason: plan.reason, + recent: plan.recent, + inputID: plan.inputID, + }) + + const chunks: string[] = [] + let failure: SessionError.Error | undefined + let usage: SessionUsage.Recorded | undefined + const recordUsage = Effect.suspend(() => + usage + ? bus.publish(SessionEvent.UsageRecorded, { + sessionID: plan.session.id, + source: "compaction", + ...usage, + }) + : Effect.void, + ) + const prepared = yield* plan.prepare({ + scope: { session: plan.session, agentID: Agent.ID.make("compaction"), model: plan.resolved }, + transcript: { system: [], messages: [Message.user(plan.prompt)] }, + contextHooks: false, + }) + yield* llm.stream(prepared.request, prepared.options).pipe( + Stream.runForEach((event) => { + if (LLMEvent.is.providerError(event)) + failure = { + type: event.classification === "context-overflow" ? "provider.invalid-request" : "provider.error", + message: event.message, + } + if (LLMEvent.is.textDelta(event)) { + chunks.push(event.text) + return bus.publish(SessionEvent.Compaction.Delta, { + sessionID: plan.session.id, + text: event.text, + }) + } + if (LLMEvent.is.stepFinish(event)) { + const step = SessionUsage.record(event.usage, plan.resolved.cost) + usage = usage ? SessionUsage.add(usage, step) : step + } + return Effect.void + }), + Effect.catchTag("AI.Error", (error) => + Effect.sync(() => { + failure = toSessionError(error) + }), + ), + Effect.onInterrupt(() => + recordUsage.pipe( + Effect.andThen( + plan.reason === "auto" + ? failed({ + sessionID: plan.session.id, + reason: plan.reason, + error: { type: "compaction.interrupted", message: "Compaction was interrupted" }, + inputID: plan.inputID, + }).pipe(Effect.asVoid) + : Effect.void, + ), + ), + ), + ) + yield* recordUsage + const summary = chunks.join("") + if (failure || !summary.trim()) { + const error = failure ?? { type: "compaction.failed" as const, message: "Compaction produced no summary" } + return yield* failed({ + sessionID: plan.session.id, + reason: plan.reason, + error, + inputID: plan.inputID, + }) + } + yield* bus.publish(SessionEvent.Compaction.Ended, { + sessionID: plan.session.id, + reason: plan.reason, + text: summary, + recent: plan.recent, + }) + return { status: "completed" as const } + }) + const compact = Effect.fn("SessionCompaction.compact")(function* (input: AutoInput) { + const content = planContent(input.messages, state.get().tokens) + if (content) + return yield* execute({ + session: input.session, + resolved: input.resolved, + prepare: input.prepare, + reason: "auto", + ...content, + }) + return yield* failed({ + sessionID: input.session.id, + reason: "auto", + error: { type: "compaction.unavailable", message: "Nothing to compact yet" }, + }) + }) + const required = (input: RequiredInput) => { + const config = state.get() + if (!config.auto) return false + const limit = input.resolved.limit + const context = limit.context + if (context <= 0) return false + const last = input.messages.findLast( + (message): message is SessionMessage.Assistant & { tokens: NonNullable } => + message.type === "assistant" && message.tokens !== undefined, + ) + if (!last) return false + const output = Math.min(limit.output, OUTPUT_TOKEN_MAX) + const promptCeiling = Math.min( + limit.input === undefined ? Number.POSITIVE_INFINITY : limit.input - config.buffer, + context - Math.max(output, config.buffer), + ) + const used = + last.tokens.input + + last.tokens.output + + last.tokens.reasoning + + last.tokens.cache.read + + last.tokens.cache.write + if (used <= 0) return false + return used >= promptCeiling + } + const compactManual = Effect.fn("SessionCompaction.compactManual")(function* (input: ManualInput) { + const content = planContent(input.messages, state.get().tokens) + if (!content) + return yield* failed({ + sessionID: input.session.id, + reason: "manual", + error: { type: "compaction.unavailable", message: "Nothing to compact yet" }, + inputID: input.inputID, + }) + const resolved = yield* input.resolveModel(input.session).pipe( + Effect.catch((cause) => + failed({ + sessionID: input.session.id, + reason: "manual", + error: toSessionError(cause), + inputID: input.inputID, + }), + ), + ) + if ("status" in resolved) return resolved + return yield* execute({ + session: input.session, + resolved, + prepare: input.prepare, + reason: "manual", + inputID: input.inputID, + started: input.started, + ...content, + }) + }) + return Service.of({ + transform: state.transform, + reload: state.reload, + enabled: () => state.get().auto, + required, + compact, + compactManual, + }) }), ) diff --git a/packages/core/src/session/title.ts b/packages/core/src/session/title.ts index 4117fe249f4..8bdd3338e2a 100644 --- a/packages/core/src/session/title.ts +++ b/packages/core/src/session/title.ts @@ -1,8 +1,7 @@ export * as SessionTitle from "./title.js" import { isDeepStrictEqual } from "node:util" -import { LLMClient, AIError, LLMEvent, Message, SystemPart, type LLMRequest } from "@opencode-ai/ai" -import type { StreamOptions } from "@opencode-ai/ai/route" +import { LLMClient, LLMEvent, Message, SystemPart } from "@opencode-ai/ai" import { Context, DateTime, Effect, Layer, Stream } from "effect" import type { Agent } from "../agent.js" import { Database } from "../database/database.js" @@ -23,15 +22,6 @@ const MAX_CONTEXT_LENGTH = 8_000 const MAX_FIRST_MESSAGE_LENGTH = 2_000 const titleChanged = Symbol("Session title changed") -type Dependencies = { - readonly bus: Bus.Interface - readonly llm: { - readonly stream: (request: LLMRequest, options?: StreamOptions) => Stream.Stream - } - readonly context: SessionContext.Interface - readonly store: SessionStore.Interface -} - export interface Interface { /** Generates an initial title or regenerates one from bounded conversation history. */ readonly generate: (sessionID: SessionSchema.ID) => Effect.Effect @@ -46,117 +36,6 @@ export const isUntitled = (session: SessionSchema.Info) => time: { created: DateTime.toEpochMillis(session.time.created) }, }) -const attempt = Effect.fn("SessionTitle.attempt")(function* ( - dependencies: Dependencies, - input: { - readonly session: SessionSchema.Info - readonly agent: Agent.Info - readonly text: string - readonly model: SessionRunnerModel.Resolved - }, -) { - const chunks: string[] = [] - let failed = false - let usage: SessionUsage.Recorded | undefined - const recordUsage = Effect.suspend(() => - usage - ? dependencies.bus.publish(SessionEvent.UsageRecorded, { - sessionID: input.session.id, - source: "title", - ...usage, - }) - : Effect.void, - ) - const prepared = yield* dependencies.context.prepare({ - scope: { session: input.session, agentID: input.agent.id, model: input.model }, - transcript: { - system: input.agent.system ? [SystemPart.make(input.agent.system)] : [], - messages: [Message.user(input.text)], - }, - contextHooks: false, - }) - yield* dependencies.llm.stream(prepared.request, prepared.options).pipe( - Stream.runForEach((event) => { - if (LLMEvent.is.providerError(event)) failed = true - if (LLMEvent.is.textDelta(event)) chunks.push(event.text) - if (LLMEvent.is.stepFinish(event)) { - const step = SessionUsage.record(event.usage, input.model.cost) - usage = usage ? SessionUsage.add(usage, step) : step - } - return Effect.void - }), - Effect.catchTag("AI.Error", () => - Effect.sync(() => { - failed = true - }), - ), - Effect.onInterrupt(() => recordUsage.pipe(Effect.asVoid)), - ) - yield* recordUsage - if (failed) return - return chunks - .join("") - .split("\n") - .map((line) => line.trim()) - .find((line) => line.length > 0) -}) - -const make = (dependencies: Dependencies) => { - const generate = Effect.fn("SessionTitle.generate")(function* ( - db: Database.Interface["db"], - sessionID: SessionSchema.ID, - ) { - const session = yield* dependencies.store.get(sessionID) - if (!session) return - const firstUser = yield* SessionHistory.firstUserMessage(db, session.id) - if (!firstUser) return - const text = !isUntitled(session) - ? yield* dependencies.store.context(session.id).pipe( - Effect.map((messages) => { - const original = `Original request:\n${firstUser.text.slice(0, MAX_FIRST_MESSAGE_LENGTH)}` - const recent = messages - .flatMap((message) => { - if (message.type === "user" && message.id !== firstUser.id) return [`User: ${message.text.trim()}`] - if (message.type !== "assistant") return [] - const text = message.content - .flatMap((part) => (part.type === "text" ? [part.text.trim()] : [])) - .filter(Boolean) - .join("\n") - return text ? [`Assistant: ${text}`] : [] - }) - .join("\n\n") - if (!recent) return original - const prefix = `${original}\n\nRecent conversation:\n` - return `${prefix}${recent.slice(-(MAX_CONTEXT_LENGTH - prefix.length))}` - }), - Effect.orElseSucceed(() => firstUser.text), - ) - : firstUser.text - const selection = yield* dependencies.context.selectTitle(session) - if (!selection) return - const title = - (yield* attempt(dependencies, { session, agent: selection.agent, text, model: selection.selected })) ?? - (selection.primary && !isDeepStrictEqual(selection.selected.ref, selection.primary.ref) - ? yield* attempt(dependencies, { session, agent: selection.agent, text, model: selection.primary }) - : undefined) - if (!title) return - const expectedSequence = (yield* Bus.latestSequence(db, sessionID)) + 1 - const current = yield* dependencies.store.get(sessionID) - if (!current || current.title !== session.title || current.title === truncate(title)) return - yield* dependencies.bus - .publish( - SessionEvent.Renamed, - { - sessionID: session.id, - title: truncate(title), - }, - { commit: (sequence) => (sequence === expectedSequence ? Effect.void : Effect.die(titleChanged)) }, - ) - .pipe(Effect.catchDefect((defect) => (defect === titleChanged ? Effect.void : Effect.die(defect)))) - }) - return { generate } -} - export const layer = Layer.effect( Service, Effect.gen(function* () { @@ -164,11 +43,110 @@ export const layer = Layer.effect( const llm = yield* LLMClient.Service const context = yield* SessionContext.Service const store = yield* SessionStore.Service - const database = yield* Database.Service - const title = make({ bus, llm, context, store }) - return Service.of({ - generate: (sessionID) => title.generate(database.db, sessionID), + const db = (yield* Database.Service).db + + const attempt = Effect.fn("SessionTitle.attempt")(function* (input: { + readonly session: SessionSchema.Info + readonly agent: Agent.Info + readonly text: string + readonly model: SessionRunnerModel.Resolved + }) { + const chunks: string[] = [] + let failed = false + let usage: SessionUsage.Recorded | undefined + const recordUsage = Effect.suspend(() => + usage + ? bus.publish(SessionEvent.UsageRecorded, { + sessionID: input.session.id, + source: "title", + ...usage, + }) + : Effect.void, + ) + const prepared = yield* context.prepare({ + scope: { session: input.session, agentID: input.agent.id, model: input.model }, + transcript: { + system: input.agent.system ? [SystemPart.make(input.agent.system)] : [], + messages: [Message.user(input.text)], + }, + contextHooks: false, + }) + yield* llm.stream(prepared.request, prepared.options).pipe( + Stream.runForEach((event) => { + if (LLMEvent.is.providerError(event)) failed = true + if (LLMEvent.is.textDelta(event)) chunks.push(event.text) + if (LLMEvent.is.stepFinish(event)) { + const step = SessionUsage.record(event.usage, input.model.cost) + usage = usage ? SessionUsage.add(usage, step) : step + } + return Effect.void + }), + Effect.catchTag("AI.Error", () => + Effect.sync(() => { + failed = true + }), + ), + Effect.onInterrupt(() => recordUsage.pipe(Effect.asVoid)), + ) + yield* recordUsage + if (failed) return + return chunks + .join("") + .split("\n") + .map((line) => line.trim()) + .find((line) => line.length > 0) }) + + const generate = Effect.fn("SessionTitle.generate")(function* (sessionID: SessionSchema.ID) { + const session = yield* store.get(sessionID) + if (!session) return + const firstUser = yield* SessionHistory.firstUserMessage(db, session.id) + if (!firstUser) return + const text = !isUntitled(session) + ? yield* store.context(session.id).pipe( + Effect.map((messages) => { + const original = `Original request:\n${firstUser.text.slice(0, MAX_FIRST_MESSAGE_LENGTH)}` + const recent = messages + .flatMap((message) => { + if (message.type === "user" && message.id !== firstUser.id) return [`User: ${message.text.trim()}`] + if (message.type !== "assistant") return [] + const text = message.content + .flatMap((part) => (part.type === "text" ? [part.text.trim()] : [])) + .filter(Boolean) + .join("\n") + return text ? [`Assistant: ${text}`] : [] + }) + .join("\n\n") + if (!recent) return original + const prefix = `${original}\n\nRecent conversation:\n` + return `${prefix}${recent.slice(-(MAX_CONTEXT_LENGTH - prefix.length))}` + }), + Effect.orElseSucceed(() => firstUser.text), + ) + : firstUser.text + const selection = yield* context.selectTitle(session) + if (!selection) return + const title = + (yield* attempt({ session, agent: selection.agent, text, model: selection.selected })) ?? + (selection.primary && !isDeepStrictEqual(selection.selected.ref, selection.primary.ref) + ? yield* attempt({ session, agent: selection.agent, text, model: selection.primary }) + : undefined) + if (!title) return + const expectedSequence = (yield* Bus.latestSequence(db, sessionID)) + 1 + const current = yield* store.get(sessionID) + if (!current || current.title !== session.title || current.title === truncate(title)) return + yield* bus + .publish( + SessionEvent.Renamed, + { + sessionID: session.id, + title: truncate(title), + }, + { commit: (sequence) => (sequence === expectedSequence ? Effect.void : Effect.die(titleChanged)) }, + ) + .pipe(Effect.catchDefect((defect) => (defect === titleChanged ? Effect.void : Effect.die(defect)))) + }) + return Service.of({ generate }) }), )