mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-14 03:56:55 +03:00
refactor(core): inline auxiliary model operations (#45974)
This commit is contained in:
Kit Langton
committed by
GitHub
Unverified
GitHub
parent
e8fa7daed5
commit
1f77f4a4ed
@@ -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<Settings>) => void
|
||||
}
|
||||
|
||||
type Dependencies = {
|
||||
readonly bus: Bus.Interface
|
||||
readonly llm: {
|
||||
readonly stream: (request: LLMRequest, options?: StreamOptions) => Stream.Stream<LLMEvent, AIError>
|
||||
}
|
||||
}
|
||||
|
||||
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<Settings, Draft>({
|
||||
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<SessionMessage.Assistant["tokens"]> } =>
|
||||
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<Settings, Draft>({
|
||||
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<SessionMessage.Assistant["tokens"]> } =>
|
||||
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,
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
+104
-126
@@ -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<LLMEvent, AIError>
|
||||
}
|
||||
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<void>
|
||||
@@ -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 })
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user