diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index aed56665326..09edc54bd4a 100644 --- a/packages/core/src/session/compaction.ts +++ b/packages/core/src/session/compaction.ts @@ -10,12 +10,13 @@ import { LLMRequest, Message, type ContentPart, + type ToolResultPart, type Usage, } from "@opencode/ai" import type { StreamOptions } from "@opencode/ai/route" import type { SessionCompactionResult } from "@opencode/plugin/effect/session" import { SessionError } from "@opencode/schema/session-error" -import { Context, Effect, Layer, Stream } from "effect" +import { Context, Effect, Layer, Ref, Stream } from "effect" import { Bus } from "../bus.js" import { Database } from "../database/database.js" import { makeLocationNode } from "@opencode/util/effect/app-node" @@ -170,11 +171,38 @@ export interface Interface extends State.Transformable { export class Service extends Context.Service()("@opencode/SessionCompaction") {} +const inputTokens = (tokens: SessionMessage.Assistant["tokens"] & {}) => + tokens.input + tokens.cache.read + tokens.cache.write + const hasInputUsage = (message: SessionMessage.Info) => - message.type === "assistant" && - !message.error && - message.tokens !== undefined && - message.tokens.input + message.tokens.cache.read + message.tokens.cache.write > 0 + message.type === "assistant" && !message.error && message.tokens !== undefined && inputTokens(message.tokens) > 0 + +const lastCheckpoint = (messages: readonly SessionMessage.Info[]) => + messages.findLast( + (message): message is SessionMessage.CompactionCompleted => + message.type === "compaction" && message.status === "completed", + ) + +/** Index of the oldest item in the newest run whose sizes total at most `budget`. */ +const fitNewest = (items: readonly T[], size: (item: T) => number, budget: number) => { + let total = 0 + let start = items.length + while (start > 0) { + const next = total + size(items[start - 1]) + if (next > budget) break + total = next + start-- + } + return start +} + +/** System prompt and tool definitions: sent with every request but outside the message history. */ +const estimateFixed = (system: ReadonlyArray<{ readonly text: string }>, tools: SessionContext.Loaded["tools"]) => + system.reduce((sum, part) => sum + Token.estimate(part.text), 0) + + tools.definitions.reduce( + (sum, tool) => sum + Token.estimate(tool.name + tool.description + JSON.stringify(tool.inputSchema)), + 0, + ) export const estimateTokens = (input: RequiredInput) => { const prompt = estimatePrompt(input) @@ -193,15 +221,7 @@ export const estimatePrompt = (input: RequiredInput) => { .filter((message) => message.role !== "assistant" || message.id !== last?.id) .reduce((sum, message) => sum + message.content.reduce((sum, part) => sum + estimatePart(part), 0), 0) if (last?.type === "assistant" && last.tokens) - return { - measured: - last.tokens.input + - last.tokens.cache.read + - last.tokens.cache.write + - last.tokens.output + - last.tokens.reasoning, - estimated: added, - } + return { measured: inputTokens(last.tokens) + last.tokens.output + last.tokens.reasoning, estimated: added } const transcript = SessionModelRequest.baseTranscript({ agent: input.context.agent.info, model: input.resolved, @@ -209,16 +229,7 @@ export const estimatePrompt = (input: RequiredInput) => { initial: input.context.initial, messages: [], }) - return { - measured: 0, - estimated: - added + - transcript.system.reduce((sum, part) => sum + Token.estimate(part.text), 0) + - input.context.tools.definitions.reduce( - (sum, tool) => sum + Token.estimate(tool.name + tool.description + JSON.stringify(tool.inputSchema)), - 0, - ), - } + return { measured: 0, estimated: added + estimateFixed(transcript.system, input.context.tools) } } const estimateMedia = (mime: string) => { @@ -238,9 +249,12 @@ const estimatePart = (part: ContentPart): number => { (sum, content) => sum + (content.type === "text" ? Token.estimate(content.text) : estimateMedia(content.mime)), 0, ) - return Token.estimate( - typeof part.result.value === "string" ? part.result.value : (JSON.stringify(part.result.value) ?? ""), - ) + return Token.estimate(toolResultText(part.result)) +} + +const toolResultText = (result: ToolResultPart["result"]) => { + if (result.type === "content") return serializeToolContent(result.value) + return typeof result.value === "string" ? result.value : (JSON.stringify(result.value) ?? "") } /** Keep whole, real user messages, never synthetic guidance or half an attachment/tool exchange. */ @@ -258,15 +272,8 @@ export const retainUsers = ( model.capabilities, ), ) - let tokens = 0 - let start = users.length - for (let index = users.length - 1; index >= 0; index--) { - const size = users[index].content.reduce((sum, part) => sum + estimatePart(part), 0) - if (tokens + size > keepTokens) break - tokens += size - start = index - } - return users.slice(start) + const size = (message: Message) => message.content.reduce((sum, part) => sum + estimatePart(part), 0) + return users.slice(fitNewest(users, size, keepTokens)) } export const truncateToolOutput = (value: string, maxChars = TOOL_OUTPUT_MAX_CHARS) => { @@ -283,7 +290,7 @@ export const truncateToolOutput = (value: string, maxChars = TOOL_OUTPUT_MAX_CHA return `${value.slice(0, end)}\n[truncated]` } -export const serializeToolContent = (content: SessionMessage.ToolStateCompleted["content"]) => +export const serializeToolContent = (content: ReadonlyArray) => content .map((item) => item.type === "text" ? item.text : `[Attached ${item.mime}${item.name === undefined ? "" : `: ${item.name}`}]`, @@ -313,14 +320,11 @@ const serializeRecentMessage = (message: SessionMessage.Info) => { if (part.type === "text") return [`[Assistant]: ${part.text}`] if (part.type === "reasoning") return part.text ? [`[Assistant reasoning]: ${part.text}`] : [] const input = typeof part.state.input === "string" ? part.state.input : JSON.stringify(part.state.input) + const call = `[Assistant tool call]: ${part.name}(${input})` if (part.state.status === "completed") - return [ - `[Assistant tool call]: ${part.name}(${input})`, - `[Tool result]: ${truncateToolOutput(serializeToolContent(part.state.content))}`, - ] - if (part.state.status === "error") - return [`[Assistant tool call]: ${part.name}(${input})`, `[Tool error]: ${part.state.error.message}`] - return [`[Assistant tool call]: ${part.name}(${input})`] + return [call, `[Tool result]: ${truncateToolOutput(serializeToolContent(part.state.content))}`] + if (part.state.status === "error") return [call, `[Tool error]: ${part.state.error.message}`] + return [call] }) .join("\n") } @@ -341,19 +345,8 @@ const serializeFallback = (messages: readonly Message[]) => if (part.type === "media") return [`[Attached ${part.media.mediaType}${part.filename ? `: ${part.filename}` : ""}; content omitted]`] if (part.type === "tool-call") return [`[Tool call ${part.name}(${JSON.stringify(part.input)})]`] - if (part.type === "tool-result") { - const value = - part.result.type === "content" - ? part.result.value - .map((item) => - item.type === "text" ? item.text : `[Attached ${item.mime}${item.name ? `: ${item.name}` : ""}]`, - ) - .join("\n") - : typeof part.result.value === "string" - ? part.result.value - : JSON.stringify(part.result.value) - return [`[Tool result ${part.name}]: ${truncateToolOutput(value ?? "", FALLBACK_TOOL_CHARS)}`] - } + if (part.type === "tool-result") + return [`[Tool result ${part.name}]: ${truncateToolOutput(toolResultText(part.result), FALLBACK_TOOL_CHARS)}`] if (part.type === "compaction" && part.text) return [part.text] return [] }) @@ -372,14 +365,7 @@ const fitFallback = (entries: ReturnType, budget: numb const header = previous ? `${previous.text}\n\n` : "" const allowance = budget - Token.estimate(header) if (allowance <= 0) return - let start = groups.length - let total = 0 - while (start > 0) { - const size = Token.estimate(groups[start - 1]) - if (total + size > allowance) break - total += size - start-- - } + const start = fitNewest(groups, Token.estimate, allowance) if (groups.length && start === groups.length) return return { text: `${header}${start ? `[${start} older exchanges omitted from this summary input]\n\n` : ""}${groups.slice(start).join("\n\n")}`, @@ -404,29 +390,20 @@ const findTailStart = (messages: readonly SessionMessage.Info[], keepTokens: num if (conversation.length === 0) return undefined // Keep at least the newest entry, even if it exceeds the allowance. - let total = 0 - let start = conversation.length - for (let index = conversation.length - 1; index >= 0; index--) { - const next = total + Token.estimate(conversation[index].text) - if (start < conversation.length && next > keepTokens) break - total = next - start = index - } - + const fitted = Math.min( + fitNewest(conversation, (item) => Token.estimate(item.text), keepTokens), + conversation.length - 1, + ) // Start at a user boundary so an assistant's tool calls and results stay together. - while (start > 0 && conversation[start].message.type !== "user") start-- + const start = conversation.findLastIndex((item, index) => index <= fitted && item.message.type === "user") if (start > 0) return conversation[start].index // If everything fits, retain only the latest exchange to leave an older prefix to summarize. const latestUser = conversation.findLastIndex((item) => item.message.type === "user") if (latestUser > 0) return conversation[latestUser].index - const previousSummary = messages.findLast( - (message): message is SessionMessage.CompactionCompleted => - message.type === "compaction" && message.status === "completed", - ) // Without an older retained tail to summarize, summarize everything and retain nothing. - return previousSummary?.recent ? conversation[0].index : messages.length + return lastCheckpoint(messages)?.recent ? conversation[0].index : messages.length } export const buildPrompt = (update: boolean, legacy = false) => { @@ -460,6 +437,19 @@ export const buildPrompt = (update: boolean, legacy = false) => { const hasSummarySection = (summary: string) => summary.split("\n").some((line) => SUMMARY_HEADINGS.includes(line.trim())) +type Envelope = Pick + +/** One summary attempt: text so far plus the first failure, if any, and whether it was a context overflow. */ +type Summary = { + readonly summary: string + readonly overflow: boolean + readonly failure?: SessionError.Error + readonly providerState?: SessionMessage.ProviderState +} + +const NUDGE = + "The previous response did not fill in the required summary template. Do not call tools. Return the summary as text using the exact section headings from the template." + export const layer = Layer.effect( Service, Effect.gen(function* () { @@ -481,34 +471,35 @@ export const layer = Layer.effect( }, }), }) - const failed = Effect.fnUntraced(function* (input: SessionEvent.Compaction.Failed["data"]) { - yield* bus.publish(SessionEvent.Compaction.Failed, input) - return { status: "failed" as const, error: input.error } + const envelope = (input: ExecuteInput): Envelope => ({ + sessionID: input.context.session.id, + reason: input.reason, + inputID: input.inputID, }) - const started = (input: ExecuteInput, recent: string) => - input.started - ? Effect.void - : bus.publish(SessionEvent.Compaction.Started, { - sessionID: input.context.session.id, - reason: input.reason, - recent, - inputID: input.inputID, - }) - const supplied = Effect.fn("SessionCompaction.supplied")(function* ( + const recordUsage = (sessionID: SessionSchema.ID, usage: SessionUsage.Recorded | undefined) => + usage ? bus.publish(SessionEvent.UsageRecorded, { sessionID, source: "compaction", ...usage }) : Effect.void + const failed = Effect.fnUntraced(function* ( + target: Envelope, + error: SessionError.Error, + usage?: SessionUsage.Recorded, + ) { + yield* recordUsage(target.sessionID, usage) + yield* bus.publish(SessionEvent.Compaction.Failed, { ...target, error, ...usage }) + return { status: "failed" as const, error } + }) + const ended = Effect.fnUntraced(function* ( input: ExecuteInput, - result: SessionCompactionResult, - recent: string, + result: { + readonly text: string + readonly recent: string + readonly providerState?: SessionMessage.ProviderState + readonly providerContext?: SessionProviderContext.Info + readonly usage?: SessionUsage.Recorded + readonly metadata?: Record + }, ) { const context = input.context - const usage = result.tokens - ? { tokens: result.tokens, cost: SessionUsage.calculateCost(context.model.cost, result.tokens) } - : undefined - if (usage) - yield* bus.publish(SessionEvent.UsageRecorded, { - sessionID: context.session.id, - source: "compaction", - ...usage, - }) + yield* recordUsage(context.session.id, result.usage) yield* bus.publish( SessionEvent.Compaction.Ended, { @@ -516,72 +507,99 @@ export const layer = Layer.effect( reason: input.reason, model: context.model.ref, providerState: result.providerState, - text: result.summary, - recent, - ...usage, + providerContext: result.providerContext, + text: result.text, + recent: result.recent, + ...result.usage, }, { metadata: result.metadata }, ) return { status: "completed" as const } }) + const started = (input: ExecuteInput, recent: string) => + input.started ? Effect.void : bus.publish(SessionEvent.Compaction.Started, { ...envelope(input), recent }) + /** A hook answered the request itself. */ + const supplied = (input: ExecuteInput, result: SessionCompactionResult, recent: string) => + ended(input, { + text: result.summary, + recent, + providerState: result.providerState, + usage: result.tokens && { + tokens: result.tokens, + cost: SessionUsage.calculateCost(input.context.model.cost, result.tokens), + }, + metadata: result.metadata, + }) // Manual controls settle through the inbox; only automatic work needs a durable interruption record. - const interrupted = (input: ExecuteInput) => - input.reason === "auto" - ? failed({ - sessionID: input.context.session.id, - reason: input.reason, - inputID: input.inputID, - error: { type: "compaction.interrupted", message: "Compaction was interrupted" }, - }).pipe(Effect.asVoid) - : Effect.void + const interrupted = (input: ExecuteInput, usage?: SessionUsage.Recorded) => + Effect.gen(function* () { + yield* recordUsage(input.context.session.id, usage) + if (input.reason !== "auto") return + yield* failed(envelope(input), { type: "compaction.interrupted", message: "Compaction was interrupted" }) + }) + const prepare = ( + input: ExecuteInput, + transcript: Pick, + inputTokens: { readonly measured: number; readonly estimated: number }, + webSocket?: "session", + ) => + input.prepare({ + session: input.context.session, + agent: input.context.agent.id, + model: input.context.model, + tools: input.context.tools, + system: transcript.system, + messages: transcript.messages, + webSocket, + inputTokens, + }) + const transcript = (input: ExecuteInput, messages: readonly SessionMessage.Info[]) => + SessionModelRequest.baseTranscript({ + agent: input.context.agent.info, + model: input.context.model, + tools: input.context.tools, + initial: input.context.initial, + messages, + }) const compactionRequest = ( input: ExecuteInput, messages: readonly SessionMessage.Info[], webSocket?: "session", summaryPrompt?: string, ) => { - const context = input.context - const transcript = SessionModelRequest.baseTranscript({ - agent: context.agent.info, - model: context.model, - tools: context.tools, - initial: context.initial, - messages, - }) - const prompt = estimatePrompt({ messages, resolved: context.model, context }) - return input.prepare({ - session: context.session, - agent: context.agent.id, - model: context.model, - tools: context.tools, - system: transcript.system, - messages: [ - ...transcript.messages, - ...(input.instructionUpdate ? [Message.system(input.instructionUpdate)] : []), - ], - webSocket, + const base = transcript(input, messages) + const prompt = estimatePrompt({ messages, resolved: input.context.model, context: input.context }) + return prepare( + input, + { + system: base.system, + messages: [...base.messages, ...(input.instructionUpdate ? [Message.system(input.instructionUpdate)] : [])], + }, // The instruction update and summary prompt are sent outside the history, so count them too. - inputTokens: { + { measured: prompt.measured, estimated: prompt.estimated + Token.estimate((input.instructionUpdate ?? "") + (summaryPrompt ?? "")), }, - }) + webSocket, + ) } + const retry = Effect.fnUntraced(function* (input: ExecuteInput, hook: SessionModelRequest.Prepared["retry"]) { + return SessionRunnerRetry.transient(yield* SessionRunnerRetry.policy(input.context.session.id), { + agent: input.context.agent.id, + model: input.context.model.ref, + hook, + }) + }) /** The durable transcript since the last local summary, re-expanding every native window. */ const original = (sessionID: SessionSchema.ID) => SessionHistory.load(db, sessionID, "local").pipe(Effect.orDie) const recoverLocally = (input: ExecuteInput) => original(input.context.session.id).pipe( - Effect.flatMap((messages) => execute({ ...input, context: { ...input.context, messages } })), + Effect.flatMap((messages) => summarize({ ...input, context: { ...input.context, messages } })), ) - const executeProvider = Effect.fn("SessionCompaction.executeProvider")(function* (input: ExecuteInput) { + + const native = Effect.fn("SessionCompaction.native")(function* (input: ExecuteInput) { const context = input.context - const reject = (message: string) => - failed({ - sessionID: context.session.id, - reason: input.reason, - inputID: input.inputID, - error: { type: "provider.unsupported-operation", message }, - }) + const reject = (message: string) => failed(envelope(input), { type: "provider.unsupported-operation", message }) const prepared = yield* compactionRequest(input, context.messages, "session") if (prepared.event.result) { yield* started(input, "") @@ -592,16 +610,12 @@ export const layer = Layer.effect( if (!provenance) return yield* reject("Provider compaction requires a stable, configured endpoint") // History is selected before request hooks. Until that interface can select on the final route, // require routing in the catalog; never install a checkpoint that the next request would skip. - if ( - !SessionProviderContext.compatible( - provenance, - SessionProviderContext.provenance({ model: request.model, ref: context.model.ref }), - ) - ) + const routed = SessionProviderContext.provenance({ model: request.model, ref: context.model.ref }) + if (!SessionProviderContext.compatible(provenance, routed)) return yield* reject( "Provider compaction requires the endpoint in provider/model settings, not a model.request rewrite", ) - const native = state + const strategy = state .get() .native.toReversed() .map((strategy) => @@ -614,39 +628,25 @@ export const layer = Layer.effect( }), ) .find((effect) => effect !== undefined) - if (!native) + if (!strategy) return yield* reject( `No plugin provides native compaction for ${request.model.provider}/${request.model.route.id}`, ) - const transient = SessionRunnerRetry.transient(yield* SessionRunnerRetry.policy(context.session.id), { - agent: context.agent.id, - model: context.model.ref, - hook: prepared.retry, - }) + const transient = yield* retry(input, prepared.retry) yield* started(input, "") + // Transient provider failures retry like any other request; only a known automatic overflow permits + // local recovery, and nothing is installed until the provider returns a checkpoint. return yield* Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - // Transient provider failures retry like any other request; only a known automatic overflow permits - // local recovery, and nothing is installed until the provider returns a checkpoint. - const result = yield* restore(native.pipe(transient)) - const usage = result.usage ? SessionUsage.record(result.usage, context.model.cost) : undefined - if (usage) - yield* bus.publish(SessionEvent.UsageRecorded, { - sessionID: context.session.id, - source: "compaction" as const, - ...usage, - }) - yield* bus.publish(SessionEvent.Compaction.Ended, { - sessionID: context.session.id, - reason: input.reason, - model: context.model.ref, - text: "", - recent: "", - providerContext: SessionProviderContext.encode(provenance, result.replacement), - ...usage, - }) - return { status: "completed" as const } - }), + restore(strategy.pipe(transient)).pipe( + Effect.flatMap((result) => + ended(input, { + text: "", + recent: "", + providerContext: SessionProviderContext.encode(provenance, result.replacement), + usage: result.usage && SessionUsage.record(result.usage, context.model.cost), + }), + ), + ), ).pipe( Effect.onInterrupt(() => interrupted(input)), Effect.catchTag( @@ -658,239 +658,175 @@ export const layer = Layer.effect( result.status === "completed" ? { ...result, recoveredOverflow: true } : result, ), ) - : failed({ - sessionID: context.session.id, - reason: input.reason, - inputID: input.inputID, - error: toSessionError(cause), - }), + : failed(envelope(input), toSessionError(cause)), ), ) }) - const execute = Effect.fn("SessionCompaction.execute")(function* (input: ExecuteInput) { + + /** One summary request. Usage accumulates in `usage` so an interruption can still account for it. */ + const stream = ( + input: ExecuteInput, + request: LLMRequest, + options: StreamOptions, + transient: ReturnType, + usage: Ref.Ref, + ) => { + const context = input.context + const key = context.model.model.route.providerMetadataKey ?? context.model.model.provider + return llm.stream(request, options).pipe( + Stream.runFoldEffect( + (): Summary => ({ summary: "", overflow: false }), + (acc, event): Effect.Effect => { + if (LLMEvent.is.providerError(event)) { + const overflow = event.classification === "context-overflow" + return Effect.succeed({ + ...acc, + overflow, + failure: { type: overflow ? "provider.invalid-request" : "provider.error", message: event.message }, + }) + } + if (LLMEvent.is.textDelta(event)) + return bus + .publish(SessionEvent.Compaction.Delta, { sessionID: context.session.id, text: event.text }) + .pipe(Effect.as({ ...acc, summary: acc.summary + event.text })) + if (LLMEvent.is.stepFinish(event)) + return Ref.update(usage, (total) => { + const step = SessionUsage.record(event.usage, context.model.cost) + return total ? SessionUsage.add(total, step) : step + }).pipe(Effect.as({ ...acc, providerState: event.providerMetadata?.[key] })) + if (!LLMEvent.is.finish(event)) return Effect.succeed(acc) + const reason = event.reason.normalized + if (reason === "unknown") + return Effect.fail( + new AIError({ + reason: new InvalidProviderOutputError({ + message: "The provider response ended with an unknown finish reason.", + classification: "incomplete-stream", + }), + }), + ) + if (reason === "error") + return Effect.fail( + new AIError({ reason: new UnknownProviderError({ message: "Compaction generation failed" }) }), + ) + if (reason === "length") + return Effect.succeed({ + ...acc, + failure: { type: "compaction.failed", message: "Compaction summary reached the output token limit" }, + }) + if (reason === "content-filter") + return Effect.succeed({ + ...acc, + failure: { type: "provider.content-filter", message: "Compaction summary was blocked by the provider" }, + }) + return Effect.succeed(acc) + }, + ), + transient, + Effect.catchTag("AI.Error", (error) => + Effect.succeed({ + summary: "", + overflow: isContextOverflowFailure(error), + failure: toSessionError(error), + }), + ), + Effect.onInterrupt(() => Ref.get(usage).pipe(Effect.flatMap((total) => interrupted(input, total)))), + ) + } + + const summarize = Effect.fn("SessionCompaction.summarize")(function* (input: ExecuteInput) { const context = input.context const history = splitHistory(context.messages, state.get().tokens) if (!history) - return yield* failed({ - sessionID: context.session.id, - reason: input.reason, - error: { type: "compaction.unavailable", message: "Nothing to compact yet" }, - inputID: input.inputID, - }) + return yield* failed(envelope(input), { type: "compaction.unavailable", message: "Nothing to compact yet" }) yield* started(input, history.recent) - - const chunks: string[] = [] - let failure: SessionError.Error | undefined - let usage: SessionUsage.Recorded | undefined - let providerState: SessionMessage.ProviderState | undefined - const recordUsage = Effect.suspend(() => - usage - ? bus.publish(SessionEvent.UsageRecorded, { - sessionID: context.session.id, - source: "compaction", - ...usage, - }) - : Effect.void, - ) - const previous = history.messages.findLast( - (message): message is SessionMessage.CompactionCompleted => - message.type === "compaction" && message.status === "completed", - ) + const previous = lastCheckpoint(history.messages) // Checkpoints from the previous template ran far longer than this one asks for; its catch-all heading identifies them. - const legacy = previous?.summary.includes(LEGACY_HEADING) ?? false - const prompt = buildPrompt(previous !== undefined, legacy) + const prompt = buildPrompt(previous !== undefined, previous?.summary.includes(LEGACY_HEADING) ?? false) const prepared = yield* compactionRequest(input, history.messages, undefined, prompt) if (prepared.event.result) return yield* supplied(input, prepared.event.result, history.recent) - // Hooks see the transcript alone; the summary prompt is appended after they run. - const first = LLMRequest.update(prepared.request, { - messages: [...prepared.request.messages, Message.user(prompt)], - }) - const usable = usableWindow(context.model.limit) - const fixed = - prepared.request.system.reduce((sum, part) => sum + Token.estimate(part.text), 0) + - context.tools.definitions.reduce( - (sum, tool) => sum + Token.estimate(tool.name + tool.description + JSON.stringify(tool.inputSchema)), - 0, - ) + - Token.estimate(prompt) - let entries: ReturnType | undefined - let budget = usable ? compactionCeiling(usable, state.get().buffer) - fixed : undefined - let fallback = - usable > 0 && - (budget ?? 0) > 0 && - estimateTokens({ messages: history.messages, resolved: context.model, context }) + - Token.estimate(prompt) > - usable - let retries = 0 - let last: string | undefined - let omitted = 0 // Both requests share the retry allowance; rejected output never enters the reminder request. - const transient = SessionRunnerRetry.transient(yield* SessionRunnerRetry.policy(context.session.id), { - agent: context.agent.id, - model: context.model.ref, - hook: prepared.retry, + const transient = yield* retry(input, prepared.retry) + const usage = yield* Ref.make(undefined) + // Hooks see the transcript alone; the summary prompt is appended after they run. + const generate = Effect.fnUntraced(function* (request: LLMRequest, options: StreamOptions) { + const prompted = LLMRequest.update(request, { messages: [...request.messages, Message.user(prompt)] }) + const first = yield* stream(input, prompted, options, transient, usage) + if (first.failure || hasSummarySection(first.summary)) return first + const nudged = LLMRequest.update(prompted, { messages: [...prompted.messages, Message.user(NUDGE)] }) + return yield* stream(input, nudged, options, transient, usage) }) - let overflow = false - while (true) { - if (fallback) { - entries ??= serializeFallback(prepared.request.messages) - budget ??= Token.estimate(entries.map((entry) => entry.text).join("\n\n")) - } - const fitted = fallback && entries && budget !== undefined ? fitFallback(entries, budget) : undefined - if (fallback && (!fitted || fitted.text === last)) { - failure = { - type: "compaction.failed", - message: "The summary input cannot be reduced further without losing the latest exchange or checkpoint", - } - break - } - if (fitted) { - last = fitted.text - omitted = fitted.omitted - retries++ - } - const selected = fitted - ? yield* input.prepare({ - session: context.session, - agent: context.agent.id, - model: context.model, - tools: context.tools, - system: SessionModelRequest.baseTranscript({ - agent: context.agent.info, - model: context.model, - tools: context.tools, - initial: context.initial, - messages: [], - }).system, - messages: [Message.user(fitted.text)], - inputTokens: { measured: 0, estimated: fixed + Token.estimate(fitted.text) }, - }) - : prepared - if (selected.event.result) { - yield* recordUsage - return yield* supplied(input, selected.event.result, history.recent) - } - const request = fallback - ? LLMRequest.update(selected.request, { - messages: [...selected.request.messages, Message.user(prompt)], - }) - : first - for (const attempt of [ - request, - LLMRequest.update(request, { - messages: [ - ...request.messages, - Message.user( - "The previous response did not fill in the required summary template. Do not call tools. Return the summary as text using the exact section headings from the template.", - ), - ], - }), - ]) { - yield* Stream.suspend(() => { - chunks.length = 0 - providerState = undefined - failure = undefined - overflow = false - return llm.stream(attempt, selected.options) - }).pipe( - Stream.runForEach((event) => { - if (LLMEvent.is.providerError(event)) { - overflow = event.classification === "context-overflow" - failure = { - type: 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: context.session.id, - text: event.text, - }) - } - if (LLMEvent.is.stepFinish(event)) { - providerState = - event.providerMetadata?.[ - context.model.model.route.providerMetadataKey ?? context.model.model.provider - ] - const step = SessionUsage.record(event.usage, context.model.cost) - usage = usage ? SessionUsage.add(usage, step) : step - } - if (LLMEvent.is.finish(event)) { - if (event.reason.normalized === "length") - failure = { type: "compaction.failed", message: "Compaction summary reached the output token limit" } - if (event.reason.normalized === "content-filter") - failure = { - type: "provider.content-filter", - message: "Compaction summary was blocked by the provider", - } - if (event.reason.normalized === "unknown") - return Effect.fail( - new AIError({ - reason: new InvalidProviderOutputError({ - message: "The provider response ended with an unknown finish reason.", - classification: "incomplete-stream", - }), - }), - ) - if (event.reason.normalized === "error") - return Effect.fail( - new AIError({ reason: new UnknownProviderError({ message: "Compaction generation failed" }) }), - ) - } - return Effect.void - }), - transient, - Effect.catchTag("AI.Error", (error) => - Effect.sync(() => { - overflow = isContextOverflowFailure(error) - failure = toSessionError(error) - }), - ), - Effect.onInterrupt(() => recordUsage.pipe(Effect.andThen(interrupted(input)))), + const finish = Effect.fnUntraced(function* (result: Summary, omitted: number) { + const total = yield* Ref.get(usage) + if (result.failure || !hasSummarySection(result.summary)) + return yield* failed( + envelope(input), + result.failure ?? { + type: "compaction.failed", + message: result.summary.trim() + ? "Compaction summary did not match the required template" + : "Compaction produced no summary", + }, + total, ) - if (failure || hasSummarySection(chunks.join(""))) break - } - if (!overflow) break - if (retries >= FALLBACK_OVERFLOW_RETRIES) break - fallback = true - if (retries && budget !== undefined) budget = Math.floor(budget / 2) - } - yield* recordUsage - const summary = chunks.join("") - if (failure || !hasSummarySection(summary)) { - const error = failure ?? { - type: "compaction.failed" as const, - message: summary.trim() - ? "Compaction summary did not match the required template" - : "Compaction produced no summary", - } - return yield* failed({ - sessionID: context.session.id, - reason: input.reason, - error, - inputID: input.inputID, - ...usage, + return yield* ended(input, { + text: omitted + ? `${result.summary}\n\n[${omitted} older exchanges were omitted from the summary input; original session history is retained.]` + : result.summary, + recent: history.recent, + providerState: result.providerState, + usage: total, }) - } - yield* bus.publish(SessionEvent.Compaction.Ended, { - sessionID: context.session.id, - reason: input.reason, - model: context.model.ref, - providerState, - text: omitted - ? `${summary}\n\n[${omitted} older exchanges were omitted from the summary input; original session history is retained.]` - : summary, - recent: history.recent, - ...usage, }) - return { status: "completed" as const } + + const usable = usableWindow(context.model.limit) + const fixed = estimateFixed(prepared.request.system, context.tools) + Token.estimate(prompt) + const ceiling = usable ? compactionCeiling(usable, state.get().buffer) - fixed : undefined + const oversized = + usable > 0 && + (ceiling ?? 0) > 0 && + estimateTokens({ messages: history.messages, resolved: context.model, context }) + Token.estimate(prompt) > + usable + const normal = oversized ? undefined : yield* generate(prepared.request, prepared.options) + if (normal && !normal.overflow) return yield* finish(normal, 0) + + // Flatten the history to text and drop the oldest exchanges until the provider accepts the request. + const entries = serializeFallback(prepared.request.messages) + const system = transcript(input, []).system + let budget = ceiling ?? Token.estimate(entries.map((entry) => entry.text).join("\n\n")) + let last: string | undefined + for (let attempt = 1; ; attempt++) { + const fitted = fitFallback(entries, budget) + if (!fitted || fitted.text === last) + return yield* failed( + envelope(input), + { + type: "compaction.failed", + message: "The summary input cannot be reduced further without losing the latest exchange or checkpoint", + }, + yield* Ref.get(usage), + ) + const reduced = yield* prepare( + input, + { system, messages: [Message.user(fitted.text)] }, + { measured: 0, estimated: fixed + Token.estimate(fitted.text) }, + ) + if (reduced.event.result) { + yield* recordUsage(context.session.id, yield* Ref.get(usage)) + return yield* supplied(input, reduced.event.result, history.recent) + } + const result = yield* generate(reduced.request, reduced.options) + if (!result.overflow || attempt === FALLBACK_OVERFLOW_RETRIES) return yield* finish(result, fitted.omitted) + budget = Math.floor(budget / 2) + last = fitted.text + } }) + + const run = (input: ExecuteInput) => + input.context.model.compaction?.type === "native" ? native(input) : summarize(input) const compact = Effect.fn("SessionCompaction.compact")(function* (input: AutoInput): Effect.fn.Return { const request = { ...input, reason: "auto" as const } - if (input.overflow) return yield* recoverLocally(request) - if (input.context.model.compaction?.type !== "native") return yield* execute(request) - return yield* executeProvider(request) + return yield* input.overflow ? recoverLocally(request) : run(request) }) const required = (input: RequiredInput) => { const config = state.get() @@ -905,38 +841,25 @@ export const layer = Layer.effect( ) return false const limit = input.resolved.limit - const context = limit.context - if (context <= 0) return false + if (limit.context <= 0) return false return estimateTokens(input) >= compactionCeiling(usableWindow(limit), config.buffer) } const compactManual = Effect.fn("SessionCompaction.compactManual")(function* (input: ManualInput) { + const target: Envelope = { sessionID: input.session.id, reason: "manual", inputID: input.inputID } if (findTailStart(input.messages, state.get().tokens) === undefined) - return yield* failed({ - sessionID: input.session.id, - reason: "manual", - error: { type: "compaction.unavailable", message: "Nothing to compact yet" }, - inputID: input.inputID, - }) + return yield* failed(target, { type: "compaction.unavailable", message: "Nothing to compact yet" }) return yield* input.resolveContext(input.session).pipe( Effect.matchEffect({ - onFailure: (cause) => - failed({ - sessionID: input.session.id, - reason: "manual", - error: toSessionError(cause), - inputID: input.inputID, - }), - onSuccess: (context) => { - const request = { + onFailure: (cause) => failed(target, toSessionError(cause)), + onSuccess: (context) => + run({ context, instructionUpdate: context.instructionUpdate, prepare: input.prepare, - reason: "manual" as const, + reason: "manual", inputID: input.inputID, started: input.started, - } - return context.model.compaction?.type === "native" ? executeProvider(request) : execute(request) - }, + }), }), ) })